From dfb0b4181edf262652e1c1faa0a88227bd627767 Mon Sep 17 00:00:00 2001 From: CTO Date: Sun, 16 Aug 2026 14:24:44 +0000 Subject: [PATCH] fix(heartbeat): bound stale-lock re-adoption of a swept parked retry (BLO-22060) `sweepStaleIssueLocks` clears `executionRunId`/`executionLockedAt` on an expired pre-claim lock but deliberately leaves the run parked, so the park can still fire. That post-sweep state is exactly the precondition for re-adoption in `enqueueWakeup`: the legacy-run SELECT matched any run in `EXECUTION_PATH_HEARTBEAT_RUN_STATUSES` with no exclusion of the run that just lost the lock, and `cancelStaleScheduledRetry` declines to cancel a park owned by the issue's own assignee. The same dead holder could therefore re-acquire a fresh 6h lock on every wake, indefinitely. Bound it with a persisted release counter: `sweepStaleIssueLocks` increments `heartbeat_runs.issue_lock_release_count` when it releases a holder, and the legacy-run SELECT filters on `issueLockReleaseCount < MAX_SWEPT_ISSUE_LOCK_RELEASES`. A released holder is passed over exactly once, so total lock time attributable to one parked run is bounded regardless of how many wakes arrive. The bound applies to every status the fallback selects -- `queued` (BLO-18995) and silent-`running` (BLO-19941) holders are released and counted too. Also makes the sweep's optimistic-concurrency skip observable: the pinned UPDATE at `recovery/service.ts` matching zero rows is now logged and counted rather than silently deferring to the next tick. Records a decision on `deferRunForK8sIsolationConflict`: its uncapped re-queue attempts are intentional. That path leaves `executionLockedAt` untouched, so it never resets the stale-lock clock, and the conflict it waits on is transient by construction with the delay already capped at 5 min. Rationale is at the site. The column landed separately on master as `0219` via #1371, so this carries no migration -- which is also why this branch is now squash-linearized: it had accumulated 7 merge commits, leaving `rebaseable=false` against a REBASE merge queue that could never build it. Refs: BLO-22060 --- .../recovery-stale-issue-lock-sweep.test.ts | 339 ++++++++++++++++++ server/src/services/heartbeat.ts | 136 ++++++- server/src/services/recovery/service.ts | 63 +++- 3 files changed, 532 insertions(+), 6 deletions(-) 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;