diff --git a/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts b/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts index dfb40f504845..25e08a35b458 100644 --- a/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts +++ b/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts @@ -610,6 +610,11 @@ describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => { const result = await sweepPromise!; expect(result.cleared).toBe(0); + // BLO-22060: a bump landing on the sweep's own 30s cadence can starve the + // clear indefinitely, and `cleared: 0` alone reads identically to a quiet + // pass. Count the bailout so the starvation is observable. + expect(result.skippedByConcurrentLockChange).toBe(1); + expect(result.skippedByConcurrentLockChangeIssueIds).toEqual([issueId]); const row = await db .select({ executionRunId: issues.executionRunId, @@ -1626,6 +1631,10 @@ describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => { scheduledRetryAt: Date; scheduledRetryReason?: string | null; sameRunHoldsCheckout?: boolean; + // Only the wake-driven case needs this: enqueueWakeup resolves a + // responsible user before it can seed a run, and throws 422 without one. + // Left null by default so the sweep-only cases keep their existing shape. + responsibleUserId?: string | null; }) { const wedgedRunId = randomUUID(); const issueId = randomUUID(); @@ -1649,6 +1658,7 @@ describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => { status: "in_progress", priority: "critical", assigneeAgentId: input.agentId, + responsibleUserId: input.responsibleUserId ?? null, checkoutRunId: input.sameRunHoldsCheckout === false ? null : wedgedRunId, executionRunId: wedgedRunId, executionLockedAt: input.lockedAt, @@ -1714,6 +1724,335 @@ describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => { expect(row?.executionRunId).toBeNull(); }); + // BLO-22060: the cap the test above proves is renewable unless the release is + // recorded on the run. The sweep deliberately leaves the parked run alive, and + // enqueueWakeup's legacy-run fallback re-selected exactly that run — + // cancelStaleScheduledRetry declines to cancel a park owned by the issue's own + // assignee — then re-stamped executionLockedAt = now(). One wake restored the + // full 6h window, so a capacity park deadlined days out kept the issue out of + // service for its assignee indefinitely, in 6h slices rather than one block. + it("does not let a wake re-adopt a parked retry whose lock the sweep already released (BLO-22060)", async () => { + const { companyId, agentId } = await seed(); + const { issueId, wedgedRunId } = await seedWedgedScheduledRetryIssue({ + companyId, + agentId, + lockedAt: new Date(Date.now() - 13 * 60 * 60 * 1000), + // Capacity parks take their horizon from the provider's reset, so the + // deadline is routinely days out — the whole window this bug covers. + scheduledRetryAt: new Date(Date.now() + 3 * 24 * 60 * 60 * 1000), + sameRunHoldsCheckout: false, + responsibleUserId: "responsible-user", + }); + + const heartbeat = heartbeatService(db, { skipQueuedRunDispatch: true }); + + const sweep = await heartbeat.sweepStaleIssueLocks(); + expect(sweep.cleared).toBe(1); + expect(sweep.issueIds).toContain(issueId); + + // The release is recorded on the run, because the issue columns it was + // recorded on are exactly what the sweep just nulled. + const releasedRun = await db + .select({ + status: heartbeatRuns.status, + issueLockReleaseCount: heartbeatRuns.issueLockReleaseCount, + }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, wedgedRunId)) + .then((rows) => rows[0]); + expect(releasedRun?.issueLockReleaseCount).toBe(1); + // The park itself survives — the sweep releases the lock, it does not cancel + // the retry, so the run still fires when its deadline arrives. + expect(releasedRun?.status).toBe("scheduled_retry"); + + // `manual` is user-initiated and so bypasses the ccrotate availability gate, + // keeping this hermetic. The adoption path under test is shared by every + // wake source. + const wake = await heartbeat.enqueueWakeup(agentId, { + source: "manual", + reason: "issue_assigned", + contextSnapshot: { issueId }, + payload: { issueId }, + }); + + // The assertion that survives master's 8446c1011. That commit stopped the + // legacy fallback from re-stamping `executionLockedAt` for a non-`running` + // holder, which fixes the visible renewal — but it still assigns the park to + // `activeExecutionRun`, so a same-agent wake is *coalesced into a run that + // will not execute until its deadline* and produces nothing. Observed in + // production on BLO-22438: two comments, both absorbed, zero runs. + // + // enqueueWakeup returns the run row itself, and the coalesce branch returns + // the run it merged into — so absorption is exactly `wake.id === wedgedRunId` + // with the park's own status. Asserting the fresh `queued` run positively + // keeps this non-vacuous: a suppressed wake returns null, which would + // satisfy any `not.toBe` on its own. + expect(wake).not.toBeNull(); + expect(wake?.id).not.toBe(wedgedRunId); + expect(wake?.status).toBe("queued"); + + const afterWake = await db + .select({ + executionRunId: issues.executionRunId, + executionLockedAt: issues.executionLockedAt, + }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0]); + + // The core assertion: the burnt-out park is not the holder again. Before the + // fix this was `wedgedRunId` with a brand-new executionLockedAt. + expect(afterWake?.executionRunId).not.toBe(wedgedRunId); + // And the sweep's release stays effective: nothing re-armed a 6h window on + // behalf of the run that just lost one. + const reReleasedRun = await db + .select({ issueLockReleaseCount: heartbeatRuns.issueLockReleaseCount }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, wedgedRunId)) + .then((rows) => rows[0]); + expect(reReleasedRun?.issueLockReleaseCount).toBe(1); + + // Idempotent under wake volume: the bound is on the run, not on the wake, so + // repeat wakes cannot walk it back. + const secondWake = await heartbeat.enqueueWakeup(agentId, { + source: "manual", + reason: "issue_assigned", + contextSnapshot: { issueId }, + payload: { issueId }, + }); + const afterSecondWake = await db + .select({ executionRunId: issues.executionRunId }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0]); + expect(afterSecondWake?.executionRunId).not.toBe(wedgedRunId); + expect(secondWake?.run?.id).not.toBe(wedgedRunId); + }); + + // BLO-22060 review follow-up: the bound must not be status-scoped. The sweep + // releases — and counts — three shapes of holder, and enqueueWakeup's + // legacy-run fallback selects all three via + // EXECUTION_PATH_HEARTBEAT_RUN_STATUSES. Scoping the predicate to + // `scheduled_retry` left the *original* renewable shape (BLO-18995's never- + // claimed `queued` lock) and BLO-19941's silent-`running` lock re-adoptable on + // every wake, and let a released park evade the bound outright by being + // promoted — promotion flips the run to `queued`, carrying its release count + // across, so a status-scoped predicate stopped applying to the same row. + async function seedSweptAdoptableIssue(input: { + companyId: string; + agentId: string; + status: "queued" | "running" | "scheduled_retry"; + lockedAt: Date; + // `running` holders are bounded on their own most-recent activity, never on + // the lock timestamp — see runningLockStaleBasis. + lastSignalAt?: Date; + scheduledRetryAt?: Date; + }) { + const wedgedRunId = randomUUID(); + const issueId = randomUUID(); + const runningSignal = input.status === "running" ? (input.lastSignalAt ?? null) : null; + await db.insert(heartbeatRuns).values({ + id: wedgedRunId, + companyId: input.companyId, + agentId: input.agentId, + status: input.status, + invocationSource: "automation", + startedAt: runningSignal, + lastOutputAt: runningSignal, + lastUsefulActionAt: runningSignal, + ...(input.status === "scheduled_retry" + ? { + scheduledRetryAt: input.scheduledRetryAt ?? new Date(Date.now() - 60_000), + scheduledRetryAttempt: 1, + // Deliberately not one of the reasons in + // SCHEDULED_RETRY_REASONS_REQUIRING_CONTINUOUS_ISSUE_LOCK, which the + // sweep refuses to release at all. + scheduledRetryReason: "ccrotate_capacity", + } + : {}), + // The fallback matches candidates on this. A run without it is not an + // adoption candidate at all, which would make the assertions vacuous. + contextSnapshot: { issueId, taskId: issueId }, + }); + await db.insert(issues).values({ + id: issueId, + companyId: input.companyId, + title: `Lock held by a run at ${input.status}`, + status: "in_progress", + priority: "high", + assigneeAgentId: input.agentId, + // enqueueWakeup resolves a responsible user before it can seed a run. + responsibleUserId: "responsible-user", + checkoutRunId: null, + executionRunId: wedgedRunId, + executionLockedAt: input.lockedAt, + }); + return { wedgedRunId, issueId }; + } + + it("does not let a wake re-adopt a never-claimed queued run whose lock the sweep released (BLO-22060)", async () => { + // The BLO-18995 shape: four enqueue paths stamp the lock at enqueue time + // alongside a freshly-inserted `queued` run. If that run is never claimed the + // sweep is the only thing that releases it — and re-adoption here restored + // the full 6h window, so the issue was never actually freed for its assignee. + const { companyId, agentId } = await seed(); + const { issueId, wedgedRunId } = await seedSweptAdoptableIssue({ + companyId, + agentId, + status: "queued", + lockedAt: new Date(Date.now() - 13 * 60 * 60 * 1000), + }); + + const heartbeat = heartbeatService(db, { skipQueuedRunDispatch: true }); + + const sweep = await heartbeat.sweepStaleIssueLocks(); + expect(sweep.cleared).toBe(1); + expect(sweep.issueIds).toContain(issueId); + + // Non-vacuity: the release is counted for a `queued` holder too, and the run + // is still alive and therefore still selectable by the fallback. + const released = await db + .select({ + status: heartbeatRuns.status, + issueLockReleaseCount: heartbeatRuns.issueLockReleaseCount, + }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, wedgedRunId)) + .then((rows) => rows[0]); + expect(released?.status).toBe("queued"); + expect(released?.issueLockReleaseCount).toBe(1); + + const wake = await heartbeat.enqueueWakeup(agentId, { + source: "manual", + reason: "issue_assigned", + contextSnapshot: { issueId }, + payload: { issueId }, + }); + + const afterWake = await db + .select({ executionRunId: issues.executionRunId }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0]); + expect(afterWake?.executionRunId).not.toBe(wedgedRunId); + // Not just un-stamped — un-absorbed. A released holder must not swallow the + // wake as a coalesce target either (see the scheduled_retry case above). + // The positive `queued` assertion keeps this non-vacuous. + expect(wake).not.toBeNull(); + expect(wake?.id).not.toBe(wedgedRunId); + expect(wake?.status).toBe("queued"); + }); + + it("does not let a wake re-adopt a silent running run whose lock the sweep released (BLO-22060)", async () => { + // BLO-19941's shape. Re-adopting a holder the sweep has already declared + // silent re-wedges the issue behind a run nothing is driving. + const { companyId, agentId } = await seed(); + const { issueId, wedgedRunId } = await seedSweptAdoptableIssue({ + companyId, + agentId, + status: "running", + lockedAt: new Date(Date.now() - 9 * 60 * 60 * 1000), + // Every activity stamp well past STALE_RUNNING_ISSUE_LOCK_MS (2h). + lastSignalAt: new Date(Date.now() - 5 * 60 * 60 * 1000), + }); + + const heartbeat = heartbeatService(db, { skipQueuedRunDispatch: true }); + + const sweep = await heartbeat.sweepStaleIssueLocks(); + expect(sweep.cleared).toBe(1); + expect(sweep.issueIds).toContain(issueId); + + const released = await db + .select({ + status: heartbeatRuns.status, + issueLockReleaseCount: heartbeatRuns.issueLockReleaseCount, + }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, wedgedRunId)) + .then((rows) => rows[0]); + expect(released?.status).toBe("running"); + expect(released?.issueLockReleaseCount).toBe(1); + + const wake = await heartbeat.enqueueWakeup(agentId, { + source: "manual", + reason: "issue_assigned", + contextSnapshot: { issueId }, + payload: { issueId }, + }); + + const afterWake = await db + .select({ executionRunId: issues.executionRunId }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0]); + expect(afterWake?.executionRunId).not.toBe(wedgedRunId); + // Not just un-stamped — un-absorbed. A released holder must not swallow the + // wake as a coalesce target either (see the scheduled_retry case above). + // The positive `queued` assertion keeps this non-vacuous. + expect(wake).not.toBeNull(); + expect(wake?.id).not.toBe(wedgedRunId); + expect(wake?.status).toBe("queued"); + }); + + it("does not let a released park evade the bound by being promoted to queued (BLO-22060)", async () => { + // The status-transition hole. The park is released and counted while it is + // `scheduled_retry`, then promoteDueScheduledRetries flips the same row to + // `queued`. A predicate keyed on status stopped applying at that point, so + // the next wake re-adopted the run and re-stamped executionLockedAt — the + // bound was one promotion away from being renewable again. + const { companyId, agentId } = await seed(); + const { issueId, wedgedRunId } = await seedSweptAdoptableIssue({ + companyId, + agentId, + status: "scheduled_retry", + lockedAt: new Date(Date.now() - 13 * 60 * 60 * 1000), + // Due, so promotion below is real rather than simulated by a status poke. + scheduledRetryAt: new Date(Date.now() - 60_000), + }); + + const heartbeat = heartbeatService(db, { skipQueuedRunDispatch: true }); + + const sweep = await heartbeat.sweepStaleIssueLocks(); + expect(sweep.cleared).toBe(1); + expect(sweep.issueIds).toContain(issueId); + + const promotion = await heartbeat.promoteDueScheduledRetries(new Date()); + expect(promotion.runIds).toContain(wedgedRunId); + + // The release count survives the status transition — that is what lets the + // bound keep applying to a row that is no longer a `scheduled_retry`. + const promoted = await db + .select({ + status: heartbeatRuns.status, + issueLockReleaseCount: heartbeatRuns.issueLockReleaseCount, + }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, wedgedRunId)) + .then((rows) => rows[0]); + expect(promoted?.status).toBe("queued"); + expect(promoted?.issueLockReleaseCount).toBe(1); + + const wake = await heartbeat.enqueueWakeup(agentId, { + source: "manual", + reason: "issue_assigned", + contextSnapshot: { issueId }, + payload: { issueId }, + }); + + const afterWake = await db + .select({ executionRunId: issues.executionRunId }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0]); + expect(afterWake?.executionRunId).not.toBe(wedgedRunId); + // Not just un-stamped — un-absorbed. A released holder must not swallow the + // wake as a coalesce target either (see the scheduled_retry case above). + // The positive `queued` assertion keeps this non-vacuous. + expect(wake).not.toBeNull(); + expect(wake?.id).not.toBe(wedgedRunId); + expect(wake?.status).toBe("queued"); + }); + it.each([ "max_turns_continuation", "capacity_blocked", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index c06876047751..0800365fdf3a 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -547,6 +547,58 @@ const execFile = promisify(execFileCallback); // terminalized still owns its issue lock); sourced from there so the two cannot // drift apart again. See issue-execution-lock.ts for the drift this closed. const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ISSUE_EXECUTION_LOCK_HOLDING_RUN_STATUSES; +// BLO-22060: how many times a run whose issue lock the stale-lock sweep has +// already released may be re-adopted as that issue's executionRunId by the +// legacy-run fallback below. +// +// sweepStaleIssueLocks bounds a pre-claim lock at STALE_PRE_CLAIM_ISSUE_LOCK_MS +// from executionLockedAt, and on expiry clears the issue columns but leaves the +// run alive so a park can still fire. Without this bound the fallback simply +// re-selected that same released run — cancelStaleScheduledRetry declines to +// cancel a park owned by the issue's own assignee — and re-stamped +// executionLockedAt = now(). The cap was real but renewable: a capacity park +// deadlined days out could re-acquire a fresh 6h lock on every wake. +// +// Deliberately NOT scoped to `scheduled_retry`. The sweep releases, and counts, +// three shapes of holder, and every one of them is selectable by the fallback +// via EXECUTION_PATH_HEARTBEAT_RUN_STATUSES (now every non-terminal status, +// sourced from ISSUE_EXECUTION_LOCK_HOLDING_RUN_STATUSES, so the candidate set +// is a superset of the three below and the bound still covers all of them): +// - `queued` — a run stamped onto the lock at enqueue time and never +// claimed (BLO-18995). This is the *original* renewable shape. +// - `running` — a silent run past STALE_RUNNING_ISSUE_LOCK_MS (BLO-19941). +// - `scheduled_retry` — a park (BLO-21309). +// A status-scoped bound also let a released park evade it by simply being +// promoted: promoteDueScheduledRetry flips the run to `queued` and the run +// carries its release count across, so a `scheduled_retry`-only predicate +// stopped applying to the very same row. The count is a property of the run's +// history, not of the status it happens to hold at select time. +// +// At 1, the first release is final: the run keeps the one lock window it had +// already been granted and is never re-adopted afterwards, so total lock time +// attributable to a single run is bounded regardless of wake volume. +// +// Still required after master's 8446c1011, which narrowed the fallback's issue +// -lock stamp to `running` runs only. That removes the *renewal* (a swept park +// no longer re-stamps executionLockedAt) but not the *absorption*: the park is +// still assigned to `activeExecutionRun`, so the wake is consumed by a run that +// will not execute until its deadline and no live run is produced. Observed in +// production on BLO-22438, where two comments were absorbed by a park deadlined +// six days out. This bound removes the released run from the candidate set +// entirely, so the wake falls through and a real run is created. +// +// Declining adoption strands nothing, because this fallback is not how a live +// run acquires the lock in the first place. promoteDueScheduledRetry's UPDATE +// is conditioned only on the run row (`status='scheduled_retry' and +// scheduledRetryAt <= now`) and never reads the issue lock, and claimQueuedRun +// re-stamps under `or(isNull(executionRunId), eq(executionRunId, claimed.id))` +// — so a park still fires and a `queued` run still claims, both re-acquiring if +// the issue is free. If a fresh run has taken the issue by then the retry +// declines to claim and is cancelled, which is the correct outcome: live work +// outranks a days-old continuation. For a swept `running` holder, declining is +// the whole point — the sweep has already declared it silent, and re-adopting +// it would re-wedge the issue behind a run nothing is driving. +const MAX_SWEPT_ISSUE_LOCK_RELEASES = 1; const TASK_SCOPE_COALESCIBLE_RUN_STATUSES = ["queued", "scheduled_retry"] as const; const CANCELLABLE_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const; const HEARTBEAT_RUN_TERMINAL_STATUSES = ["succeeded", "interrupted", "failed", "cancelled", "timed_out"] as const; @@ -21963,6 +22015,32 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) // has already been created" case -- if a refactor moves isolation binding // to after Job creation, or lets a bound reservation re-enter the throw // path, that test fails before this function can strand a live Job. + // BLO-22060 decision — the uncapped re-queue attempt count here is + // INTENTIONAL, and deliberately not bounded the way the parked-retry + // re-adoption above now is. Recorded so it stays a decision rather than an + // omission: + // + // - This path never resets the stale-lock clock by itself. It leaves + // issues.executionLockedAt untouched, and the sweep's `queued` branch + // measures from that column, so a run stuck deferring is still swept 6h + // after its original lock. The clock only moves when the run is genuinely + // re-claimed (claimQueuedRun re-stamps executionLockedAt) — i.e. when it + // made real progress toward dispatch, not while it sits parked. + // - The failure mode it protects against is transient by construction: the + // conflict is raised only while another live run holds the same mutable + // isolation scope, so the correct response is to wait for that run, and + // the wait is already capped at K8S_ISOLATION_RETRY_MAX_DELAY_MS (5 min) + // per attempt. + // - Capping attempts would convert "the neighbour is taking a long time" + // into a failed run. That trades a bounded wait for lost work, on a + // condition we cannot distinguish from a slow-but-healthy neighbour. + // - A genuine live-lock is therefore a bug in reservation binding, not + // something a retry cap should paper over — and it is already observable + // without one: retryAttempt is stamped on the run event and the log line + // at the end of this function, so a climbing count is visible per run. + // + // Revisit if that log ever shows attempts climbing without a distinct + // conflictingRunId, which would mean the scope is never actually released. async function deferRunForK8sIsolationConflict( run: typeof heartbeatRuns.$inferSelect, conflict: ExternalRuntimeIsolationConflictError, @@ -27807,6 +27885,18 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) eq(heartbeatRuns.companyId, issue.companyId), inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]), sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issue.id}`, + // BLO-22060: a run whose issue lock the stale-lock sweep has + // already released is not an adoption candidate any more. + // Excluded in SQL rather than after the fact so a *different* + // eligible run for this issue is still found and adopted + // normally — only the burnt-out holder is passed over. Applies + // to every status this fallback selects, not just + // `scheduled_retry`: the sweep releases and counts `queued` + // (BLO-18995) and silent-`running` (BLO-19941) holders too, and + // a released park that is later promoted arrives here as + // `queued` carrying its count. See + // MAX_SWEPT_ISSUE_LOCK_RELEASES. + lt(heartbeatRuns.issueLockReleaseCount, MAX_SWEPT_ISSUE_LOCK_RELEASES), ), ) .orderBy( @@ -27820,6 +27910,13 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) if (await cancelStaleScheduledRetry(legacyRun)) { activeExecutionRun = null; } else { + // Master (8446c1011) narrowed the issue-lock stamp to `running` + // legacy runs: a `queued`/`scheduled_retry` holder is adopted as + // the in-memory `activeExecutionRun` for this wake but no longer + // re-stamps `executionLockedAt`. That kills the *renewal* half of + // BLO-22060; the release bound in the SELECT above kills the + // *absorption* half, so a swept park cannot silently swallow the + // wake either. activeExecutionRun = legacyRun; if (legacyRun.status === "running") { const legacyAgent = await tx @@ -27827,7 +27924,20 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) .from(agents) .where(eq(agents.id, legacyRun.agentId)) .then((rows) => rows[0] ?? null); - await tx + // BLO-22060: guard the adoption on the lock still being free. + // The enclosing transaction DOES hold this issue row `for + // update` (taken at the top, before the issue is read), so a + // concurrent writer that goes through the same path cannot + // interleave here — this predicate is not load-bearing against + // that class of race. It is kept as a cheap assertion of the + // invariant the block is written against: every branch above + // either left executionRunId null or cleared it, so adoption + // must never overwrite a live holder. If the row lock is ever + // narrowed, or a writer that does not take it is added, this + // fails closed instead of silently stamping over the holder — + // and the fallback below then adopts whoever actually holds the + // issue. + const adopted = await tx .update(issues) .set({ executionRunId: legacyRun.id, @@ -27835,7 +27945,24 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) executionLockedAt: new Date(), updatedAt: new Date(), }) - .where(eq(issues.id, issue.id)); + .where(and(eq(issues.id, issue.id), isNull(issues.executionRunId))) + .returning({ id: issues.id }) + .then((rows) => rows[0] ?? null); + + if (!adopted) { + const currentHolderId = await tx + .select({ executionRunId: issues.executionRunId }) + .from(issues) + .where(eq(issues.id, issue.id)) + .then((rows) => rows[0]?.executionRunId ?? null); + activeExecutionRun = currentHolderId + ? await tx + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, currentHolderId)) + .then((rows) => rows[0] ?? null) + : null; + } } } } @@ -30779,6 +30906,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) sweepStaleIssueLocks, + // BLO-22060: already the injected dependency behind recoveryService, + // productivity-review and task-watchdogs; exposed here so the issue-lock + // adoption path can be driven directly from a test rather than only through + // a plugin-host or route wrapper. + enqueueWakeup, reconcileDetachedQueuedRuns, buildIssueGraphLivenessAutoRecoveryPreview, diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index bc48ac128851..ca2a0c77043e 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -153,6 +153,13 @@ function recoveryActionBoundsAtCreation(now: Date): { maxAttempts: number; timeo // reset and is routinely days away; letting that horizon set the lock lifetime // took the issue out of service for the whole park, for its own assignee. export const STALE_PRE_CLAIM_ISSUE_LOCK_MS = 6 * 60 * 60 * 1000; +// BLO-22060: sentinel return from the sweep transaction meaning "the lock moved +// between the pre-transaction scan and the FOR UPDATE revalidation, so this pass +// declined to clear it". Distinct from `null`, which means "revalidated and the +// lock is legitimately not stale". Only the former can starve the clear when a +// renewal keeps landing on the sweep's cadence, so only the former is counted +// and logged. +const LOCK_CHANGED_UNDER_SWEEP: unique symbol = Symbol("staleIssueLockSweep.lockChangedUnderSweep"); export const ISSUE_ASSIGNMENT_RECOVERY_PER_AGENT_SWEEP_LIMIT = 5; const ASSIGNMENT_RECOVERY_CAPACITY_RESERVATION_STATUS = "assignment_recovery_capacity_reserved"; // Enqueue normally finishes in seconds; an hour-old reservation has lost its owning process. @@ -9451,6 +9458,15 @@ export function recoveryService( const result = { cleared: 0, issueIds: [] as string[], + // BLO-22060: the sweep revalidates every candidate under FOR UPDATE and + // bails out silently when the lock moved between the scan and the + // transaction. That is correct — but it was also invisible, so a renewal + // landing repeatedly on the sweep's own cadence (30s) could starve the + // clear indefinitely and look identical to "nothing was stale". Count the + // bailouts so the starvation is observable in the sweep result and in the + // log line below. + skippedByConcurrentLockChange: 0, + skippedByConcurrentLockChangeIssueIds: [] as string[], }; const candidates = await db @@ -9665,10 +9681,15 @@ export function recoveryService( .then((rows) => rows[0] ?? null); if (!currentIssue) return null; - if (currentIssue.checkoutRunId !== issue.checkoutRunId) return null; - if (currentIssue.executionRunId !== issue.executionRunId) return null; + // BLO-22060: concurrent-bump bailouts — the lock moved between the + // pre-transaction scan and this revalidation. Report them distinctly + // from "revalidated and found not stale" (plain null below) so a lock + // that keeps being renewed under the sweep is visible rather than + // indistinguishable from a quiet pass. + if (currentIssue.checkoutRunId !== issue.checkoutRunId) return LOCK_CHANGED_UNDER_SWEEP; + if (currentIssue.executionRunId !== issue.executionRunId) return LOCK_CHANGED_UNDER_SWEEP; if ((currentIssue.executionLockedAt?.getTime() ?? null) !== (issue.executionLockedAt?.getTime() ?? null)) { - return null; + return LOCK_CHANGED_UNDER_SWEEP; } const currentReferencedRunIds = [ @@ -9799,7 +9820,26 @@ export function recoveryService( }) .then((rows) => rows[0] ?? null); - if (!updated) return null; + if (!updated) return LOCK_CHANGED_UNDER_SWEEP; + + // BLO-22060: the release must outlive the issue row we just nulled. + // The run itself is deliberately left alive (a `scheduled_retry` park + // still has to fire at its deadline, and a `queued` run must stay + // claimable), and enqueueWakeup's legacy-run fallback would otherwise + // re-adopt that same run and re-stamp executionLockedAt, restarting the + // 6h clock on every wake. Recording the release on the run is what lets + // adoption decline. Counted for whatever status held the lock — the + // fallback can select `queued`, `running` and `scheduled_retry` alike, + // and a released park that is later promoted reaches it as `queued`. + if (currentIssue.executionRunId) { + await tx + .update(heartbeatRuns) + .set({ + issueLockReleaseCount: sql`${heartbeatRuns.issueLockReleaseCount} + 1`, + updatedAt: clearedAt, + }) + .where(eq(heartbeatRuns.id, currentIssue.executionRunId)); + } // BLO-21621: clearing a stale pre-claim lock is the only positive // evidence that a queued row previously owned, and then lost, this @@ -10084,6 +10124,21 @@ export function recoveryService( } }); + if (sweepOutcome === LOCK_CHANGED_UNDER_SWEEP) { + result.skippedByConcurrentLockChange += 1; + result.skippedByConcurrentLockChangeIssueIds.push(issue.id); + logger.warn( + { + issueId: issue.id, + companyId: issue.companyId, + executionRunId: issue.executionRunId, + checkoutRunId: issue.checkoutRunId, + scannedExecutionLockedAt: issue.executionLockedAt?.toISOString() ?? null, + }, + "stale issue lock sweep skipped: lock changed between scan and clear", + ); + continue; + } if (!sweepOutcome) continue; const { updated } = sweepOutcome;