From 3e4506c98b21a7a68764591f334b7ce6425a1948 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sun, 2 Aug 2026 13:25:56 +0300 Subject: [PATCH 1/3] fix(scheduler): reap a runner's own expired lease once its slot has moved past MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A hosted loop run stuck in `status: "running"` with a long-expired lease blocks its own loop forever under `overlap: "skip"`. Observed on the fleet as `agent-chief-finance-coordination-10m` and `agent-chief-planning-coordination-10m`: two 10-minute seat loops that had fired exactly once, ever, with leases expired 17.6h and 10.8h. Root cause is the sweep's TRIGGER and SELF-EXCLUSION, not its classification. `claimRuns` passed `excludeClaimedBy: runner.id`, protecting every run the polling runner owned on the assumption a later poll would reclaim it via same-slot takeover. That assumption holds only while the run's slot is still due. Under `catchUp: "latest"` — what every seat loop uses — `dueSlots` returns only the newest slot, so once wall time moves past a wedged run's own slot the takeover can never happen again: `overlap: "skip"` refuses the new slot because a `running` run exists, and the sweep skips that run because this runner owns it. Neither path can fire. The blanket exclusion is replaced with `protectRunIds`, scoped to the intent it actually served: runs belonging to loops this poll never examined because claim capacity ran out. Those are the runs a runner is genuinely about to take over. A run whose loop WAS examined and could not be claimed is no longer protected. Same-slot takeover is unaffected — it happens in the claim pass and re-leases the run, so the sweep stops selecting it. Coverage note: the existing "reclaims an expired overlap-skip lease" test never reached this because it uses `catchUp: "all"`, which keeps the original slot in the due list and so always permits the takeover. Production uses "latest". Verification: - bun test src/api/index.test.ts -> 61 pass, 0 fail (new test fails on the parent commit with the run still `running` and its lease expired) - bun test src/lib/storage/postgres-loop-storage.test.ts against a disposable PostgreSQL 16 -> 52 pass, 0 fail (unset, this suite reports 53 skip / rc=0) - bun test src/lib/storage/sqlite.test.ts src/lib/hygiene.test.ts -> 23 pass - bun run typecheck, bun run build -> rc=0 - staged secrets scan -> clean Agent: fabius --- src/api/index.test.ts | 92 +++++++++++++++++++ src/api/index.ts | 47 ++++++++-- src/lib/storage/postgres-loop-storage.test.ts | 25 +++++ src/lib/storage/postgres-loop-storage.ts | 2 + src/lib/store.ts | 11 +++ 5 files changed, 168 insertions(+), 9 deletions(-) diff --git a/src/api/index.test.ts b/src/api/index.test.ts index 16ee823..57f13e3 100644 --- a/src/api/index.test.ts +++ b/src/api/index.test.ts @@ -3007,6 +3007,98 @@ describe("loops-api foundation", () => { } }); + test("runner claim reaps its own expired lease once the due slot has moved past it", async () => { + // Regression for the wedged-run defect: a `catchUp: "latest"` + `overlap: "skip"` + // loop (the shape every agent-*-coordination-10m seat loop uses: 10m interval, + // 9m lease) whose run outlives its lease is never recovered by the runner that + // owns it. + // + // `claimRuns` passes `excludeClaimedBy: runner.id` to the sweep so that a runner + // which merely ran out of claim capacity can still take its own slot over on a + // later poll — that intent is correct and is covered by the "claim capacity" + // test above. But `dueSlots` under `catchUp: "latest"` returns ONLY the latest + // slot, so once wall time has moved past the wedged run's own slot the same-slot + // takeover it is being preserved for can never happen again: `overlap: "skip"` + // refuses the new slot because a `running` run exists, and the sweep skips that + // run because this runner owns it. Neither path can fire, so the loop is blocked + // for as long as the process lives. + // + // The existing "reclaims an expired overlap-skip lease" test does not reach this + // because it uses `catchUp: "all"`, which keeps the original slot in the due list + // and so always permits the same-slot takeover. + const mod = await import("./index.js"); + const storage = createSqliteLoopStorage(":memory:"); + let now = new Date("2026-01-01T00:00:00.000Z"); + const server = createTestServer( + mod, + { host: "127.0.0.1", port: 0, storage, now: () => now, random: () => 0.5 }, + runnerPrincipal("runner-a"), + ); + + try { + const loop = await storage.createLoop( + { + name: "api-own-expired-stale-slot", + schedule: { type: "interval", everyMs: 600_000 }, + target: { type: "command", command: "true" }, + catchUp: "latest", + overlap: "skip", + leaseMs: 540_000, + }, + now, + ); + const originalNextRunAt = loop.nextRunAt; + + // createLoop schedules the first slot one interval out, so move to it before + // the loop is due at all. + now = new Date("2026-01-01T00:10:00.000Z"); + + const first = await fetch(apiUrl(server, "/v1/runners/claim"), { + method: "POST", + headers: jsonHeaders, + body: JSON.stringify({ runnerId: "runner-a", maxClaims: 5 }), + }); + expect(first.status).toBe(200); + const firstBody = (await first.json()) as { + claims: Array<{ run: { id: string; status: string } }>; + }; + expect(firstBody.claims).toHaveLength(1); + const wedgedRunId = firstBody.claims[0]!.run.id; + + // The runner dies here: it never heartbeats, never completes. Wall time moves + // two hours on, far past both the 9m lease and this run's own 10m slot. + now = new Date("2026-01-01T02:00:00.000Z"); + + // Poll repeatedly with ample capacity — this is emphatically not the + // capacity-exhaustion case. A healthy scheduler recovers on the first of these. + for (let i = 0; i < 3; i += 1) { + const poll = await fetch(apiUrl(server, "/v1/runners/claim"), { + method: "POST", + headers: jsonHeaders, + body: JSON.stringify({ runnerId: "runner-a", maxClaims: 5 }), + }); + expect(poll.status).toBe(200); + } + + // No phantom may survive: the run is either abandoned, or genuinely taken over + // with a lease in the future. What must not persist is `running` with a lease + // that expired in the past — that is the state which blocks `overlap: "skip"`. + const wedged = await storage.getRun(wedgedRunId); + expect(wedged).toBeTruthy(); + const leaseStillExpired = wedged!.status === "running" + && (!wedged!.leaseExpiresAt || new Date(wedged!.leaseExpiresAt).getTime() <= now.getTime()); + expect(leaseStillExpired).toBe(false); + + // And the loop must have made progress rather than sitting on a permanently + // past nextRunAt. + const after = await storage.getLoop(loop.id); + expect(new Date(after!.nextRunAt!).getTime()).toBeGreaterThan(new Date(originalNextRunAt!).getTime()); + } finally { + server.stop(true); + await storage.close(); + } + }); + test("runner claim reaps an expired lease owned by an ineligible runner", async () => { const mod = await import("./index.js"); const storage = createSqliteLoopStorage(":memory:"); diff --git a/src/api/index.ts b/src/api/index.ts index 5b38f6a..074c557 100644 --- a/src/api/index.ts +++ b/src/api/index.ts @@ -1294,8 +1294,15 @@ async function claimRuns( }, ): Promise>> { const claims: Array> = []; - for (const loop of await storage.dueLoops(opts.now)) { - if (claims.length >= opts.maxClaims) break; + const dueLoopsForPoll = await storage.dueLoops(opts.now); + // Loops this poll never got to look at because claim capacity ran out first. + // Their runs are the ones the sweep must not touch — see `protectRunIds` below. + let unexaminedLoops: typeof dueLoopsForPoll = []; + for (const [loopIndex, loop] of dueLoopsForPoll.entries()) { + if (claims.length >= opts.maxClaims) { + unexaminedLoops = dueLoopsForPoll.slice(loopIndex); + break; + } if (!runnerMatchesLoop(loop.machine, runner)) continue; const workflow = loop.target.type === "workflow" ? await storage.getWorkflow(loop.target.workflowId) @@ -1323,14 +1330,36 @@ async function claimRuns( } // Runner polling is the hosted scheduler tick. After this runner has had a - // chance to take over an eligible expired slot through claimRun, reap every - // other expired lease in the tenant. Running the sweep after claim selection - // preserves same-slot takeover while ensuring a run owned by a missing or - // ineligible machine cannot remain `running` forever. Keep this pass bounded - // to the storage recovery batch and advance only rows recovered by this poll - // (the operator maintenance route owns historical replay). + // chance to take over an eligible expired slot through claimRun, reap the + // remaining expired leases in the tenant. Running the sweep after claim + // selection preserves same-slot takeover while ensuring a run owned by a + // missing or ineligible machine cannot remain `running` forever. Keep this + // pass bounded to the storage recovery batch and advance only rows recovered + // by this poll (the operator maintenance route owns historical replay). + // + // Protect exactly what the old `excludeClaimedBy: runner.id` was FOR: a run of + // this runner's own that belongs to a loop this poll never examined because + // claim capacity ran out first. Reaping one of those would pull a slot out from + // under a runner that is about to take it over on its next poll. + // + // What that blanket exclusion also did, and must not, is protect a run + // belonging to a loop this poll DID examine and could not claim. That is the + // wedge: under `overlap: "skip"` the new slot is refused precisely because this + // `running` run exists, while the sweep skipped it precisely because this + // runner owns it. Neither path can fire, so the run sits in `running` with a + // long-dead lease and blocks the loop indefinitely — observed on the fleet as + // `agent-*-coordination-10m` seat loops that had fired once, ever. Same-slot + // takeover (the legitimate reason to hold a run) still works: it happens in + // the claim pass above, and it re-leases the run so the sweep no longer + // selects it at all. + const protectRunIds: string[] = []; + for (const loop of unexaminedLoops) { + for (const run of await storage.listRuns({ loopId: loop.id, status: "running" })) { + if (run.claimedBy === runner.id) protectRunIds.push(run.id); + } + } const recovered = await storage.recoverExpiredRunLeasesDetailed(opts.now, { - excludeClaimedBy: runner.id, + protectRunIds, }); const advancementDeferred = await advanceRecoveredRuns(storage, recovered.abandoned, { random: opts.random, diff --git a/src/lib/storage/postgres-loop-storage.test.ts b/src/lib/storage/postgres-loop-storage.test.ts index b66e799..08fe5b7 100644 --- a/src/lib/storage/postgres-loop-storage.test.ts +++ b/src/lib/storage/postgres-loop-storage.test.ts @@ -1368,6 +1368,31 @@ suite("PostgresLoopStorage (live)", () => { expect(result.deferred.length).toBe(0); }); + test("recoverExpiredRunLeasesDetailed honours protectRunIds", async () => { + // The hosted control plane is the production path for this sweep, so the + // option the API relies on to avoid reaping a slot a runner is about to take + // over must be verified here and not only against sqlite. Both states are + // exercised deliberately: protected leaves the run alone, unprotected reaps + // it — a one-sided assertion here could pass on a backend that ignored the + // option entirely. + const protectedLoop = await storage.createLoop(loopInput("protect-kept", { leaseMs: 1 })); + const reapedLoop = await storage.createLoop(loopInput("protect-reaped", { leaseMs: 1 })); + const slot = "2026-07-06T12:00:00.000Z"; + const past = new Date(Date.now() - 60_000); + const keptClaim = await storage.claimRun(protectedLoop, slot, "runner-x", past); + const reapedClaim = await storage.claimRun(reapedLoop, slot, "runner-x", past); + expect(keptClaim).toBeTruthy(); + expect(reapedClaim).toBeTruthy(); + + const result = await storage.recoverExpiredRunLeasesDetailed(new Date(), { + protectRunIds: [keptClaim!.run.id], + }); + + expect(result.abandoned.map((run) => run.id)).toEqual([reapedClaim!.run.id]); + expect((await storage.getRun(keptClaim!.run.id))!.status).toBe("running"); + expect((await storage.getRun(reapedClaim!.run.id))!.status).toBe("abandoned"); + }); + test("paginates an immutable tenant-scoped recovered-row snapshot", async () => { for (const id of ["a", "z"]) { const loop = await storage.createLoop(loopInput(`recovered-keyset-pages-${id}`)); diff --git a/src/lib/storage/postgres-loop-storage.ts b/src/lib/storage/postgres-loop-storage.ts index 138e43d..1c6a8b4 100644 --- a/src/lib/storage/postgres-loop-storage.ts +++ b/src/lib/storage/postgres-loop-storage.ts @@ -1601,8 +1601,10 @@ export class PostgresLoopStorage implements LoopStorageContract { [finished, opts.runId ?? null, opts.excludeClaimedBy ?? null, scanLimit], ); const recovered: LoopRun[] = []; + const protectedRunIds = new Set(opts.protectRunIds ?? []); for (const row of rows) { if (recovered.length >= limit) break; + if (protectedRunIds.has(row.id)) continue; const run = await this.client.transaction(async (c) => { const res = await c.query( `UPDATE loop_runs SET status='abandoned', finished_at=$2, lease_expires_at=NULL, diff --git a/src/lib/store.ts b/src/lib/store.ts index 9a32440..60810de 100644 --- a/src/lib/store.ts +++ b/src/lib/store.ts @@ -5007,6 +5007,15 @@ export class Store { scanLimit?: number; runId?: string; excludeClaimedBy?: string; + /** + * Run ids to leave untouched. Unlike `excludeClaimedBy`, which protects + * everything a runner owns, this protects an explicit set the caller has + * established is still recoverable by other means (for Loops: still + * same-slot claimable). A runner's own run whose slot has already moved + * past is NOT recoverable by takeover, so blanket-excluding it strands it + * in `running` forever and, under `overlap: "skip"`, blocks the loop. + */ + protectRunIds?: readonly string[]; /** Leave every currently live process untouched, even after the daemon recovery grace ceiling. */ preserveLiveProcesses?: boolean; } = {}, @@ -5032,8 +5041,10 @@ export class Store { ); const recovered: LoopRun[] = []; const deferred: LoopRun[] = []; + const protectedRunIds = new Set(opts.protectRunIds ?? []); for (const row of rows) { if (recovered.length >= limit) break; + if (protectedRunIds.has(row.id)) continue; const looksAlive = isRecordedProcessAlive(row.pid, row.process_started_at) || this.hasLiveWorkflowStepProcesses(row.id); // "Looks alive" only buys a BOUNDED grace. Past the ceiling the run is // abandoned regardless: an expired lease that keeps failing to renew is From c5b44ec9f300053d957272a6c665ecbd4f6ba709 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sun, 2 Aug 2026 23:52:57 +0300 Subject: [PATCH 2/3] fix(scheduler): protect capacity-unexamined runs in the query, not after the scan MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses both NO_GO reviews on #184. The protection this PR added was built by enumerating run ids with `listRuns` per unexamined loop and discarding them in application code after the recovery scan had already been truncated. Three consequences, all introduced by the diff and all on the hosted scheduler's hottest path: - `listRuns` defaults to 100 rows on both backends, so a loop holding more than one page of running runs had the remainder silently unprotected and reaped. - Protected rows were filtered AFTER `LIMIT`, so a large protected set crowded the scan window and starved an unrelated, genuinely reapable run. Because the caller rebuilds the same protected set on every poll, that starvation was stable — the same "can never be reaped" class this PR exists to close, reintroduced through the fix. `excludeClaimedBy` did not have this property: it filtered inside the WHERE clause, before LIMIT. - One sequential `listRuns` round-trip per unexamined loop. The shipped runner polls with `maxClaims: 1`, so a single claim makes every remaining due loop unexamined, bounded at ~499 round-trips per claiming poll. `protectRunIds` is replaced by `protectClaimedByInLoops: { claimedBy, loopIds }`, applied inside the scan query on both backends. Expressing the protection as the predicate it actually is removes the enumeration entirely, so all three consequences go with it. `protectRunIds` had exactly one call site and was introduced by this PR, so it is removed rather than left as a trap. Also corrects the mechanism asserted in the `claimRuns` comment. `overlap: "skip"` does not refuse a slot because a `running` row exists — it refuses only while a run holds a LIVE lease or a live process (sqlite `hasBlockingRunningRunForOtherSlot`; the Postgres predicate is strictly more permissive). An expired lease never blocked the gate. What the blanket exclusion actually caused is an unreapable orphan row and a loop cursor that advanced only if some later run happened to finalize, never through recovery. The behaviour is unchanged; the false causal claim is not, because it was on its way into a third artefact of record. Regressions, each confirmed failing on the parent commit for its stated reason before the fix: - api: every own run in an unexamined loop stays protected past one page (101 rows; was 100). - api: protection does not consume the recovery scan window (520 protected rows plus one later-expiring reapable run; the reapable run was previously never reached). - api: protection cost does not scale with unexamined loop count (was 1:1). - store and live Postgres: protection honoured, scoped to the claiming runner, and applied before LIMIT. Verified: `bun test --timeout 60000` 1126 pass, 56 skip, 0 fail, exit 0. Live PostgreSQL 16.14 (disposable container) 54 pass, 0 fail. Typecheck and build exit 0. Staged secrets scan clean, with a positive control proving the pattern fires. Agent: Augustus --- src/api/index.test.ts | 253 ++++++++++++++++++ src/api/index.ts | 44 +-- src/lib/storage/postgres-loop-storage.test.ts | 63 ++++- src/lib/storage/postgres-loop-storage.ts | 13 +- src/lib/store.test.ts | 107 ++++++++ src/lib/store.ts | 41 ++- 6 files changed, 488 insertions(+), 33 deletions(-) diff --git a/src/api/index.test.ts b/src/api/index.test.ts index 57f13e3..afaa2b5 100644 --- a/src/api/index.test.ts +++ b/src/api/index.test.ts @@ -3099,6 +3099,259 @@ describe("loops-api foundation", () => { } }); + test("runner claim protects EVERY own run in a capacity-unexamined loop, not just the first page", async () => { + // The capacity protection must not be built by enumerating runs: `listRuns` + // defaults to 100 rows on both backends, so a loop holding more than one + // page of running runs would have the remainder silently unprotected and + // reaped out from under the runner that is about to take it over. + // + // `overlap: "allow"` with `catchUp: "all"` is a supported configuration and + // `catchUpLimit`/`maxClaims` both permit far more than 100 concurrent runs + // on one loop, so this is reachable rather than theoretical. + const mod = await import("./index.js"); + const storage = createSqliteLoopStorage(":memory:"); + let now = new Date("2026-01-01T00:00:00.000Z"); + const server = createTestServer( + mod, + { host: "127.0.0.1", port: 0, storage, now: () => now, random: () => 0.5 }, + runnerPrincipal("runner-a"), + ); + + try { + // Earliest nextRunAt, so `dueLoops` returns it first: it consumes the one + // claim this poll is allowed, which is what leaves the second loop + // unexamined. + await storage.createLoop( + { + name: "api-capacity-consumer", + schedule: { type: "interval", everyMs: 1_000 }, + target: { type: "command", command: "true" }, + machine: { id: "runner-a" }, + }, + new Date("2025-12-31T23:59:00.000Z"), + ); + + const unexamined = await storage.createLoop( + { + name: "api-unexamined-many-own-runs", + schedule: { type: "interval", everyMs: 1_000 }, + target: { type: "command", command: "true" }, + machine: { id: "runner-a" }, + overlap: "allow", + leaseMs: 1_000, + }, + new Date("2025-12-31T23:59:30.000Z"), + ); + + // One `listRuns` page is 100 rows. Cross it. + const OWNED = 101; + const ownedRunIds: string[] = []; + for (let i = 0; i < OWNED; i += 1) { + const slot = new Date(Date.parse("2026-01-01T00:00:00.000Z") + i * 1_000).toISOString(); + const claim = await storage.claimRun(unexamined, slot, "runner-a", now); + expect(claim).toBeTruthy(); + ownedRunIds.push(claim!.run.id); + } + + // Every one of those leases is now long expired. + now = new Date("2026-01-01T01:00:00.000Z"); + + const poll = await fetch(apiUrl(server, "/v1/runners/claim"), { + method: "POST", + headers: jsonHeaders, + body: JSON.stringify({ runnerId: "runner-a", maxClaims: 1 }), + }); + expect(poll.status).toBe(200); + + const statuses = await Promise.all( + ownedRunIds.map(async (id) => (await storage.getRun(id))!.status), + ); + expect(statuses.filter((status) => status === "running").length).toBe(OWNED); + } finally { + server.stop(true); + await storage.close(); + } + }); + + test("runner claim protection does not consume the recovery scan window", async () => { + // Protection must be expressed in the recovery QUERY, before its LIMIT. + // Filtering protected rows out in application code after the scan has + // already been truncated means a large protected set can crowd the window + // and starve an unrelated, genuinely reapable run — and because the same + // protected set is rebuilt on every poll, that starvation is stable rather + // than transient. That is the same "can never be reaped" class this PR + // exists to remove, reintroduced through the fix. + // + // The default recovery scan window is 100 * 5 = 500 rows, so the protected + // set here is deliberately larger, and the reapable run's lease expires + // LATER so it sorts behind them under `ORDER BY lease_expires_at ASC`. + const mod = await import("./index.js"); + const storage = createSqliteLoopStorage(":memory:"); + let now = new Date("2026-01-01T00:00:00.000Z"); + const server = createTestServer( + mod, + { host: "127.0.0.1", port: 0, storage, now: () => now, random: () => 0.5 }, + runnerPrincipal("runner-a"), + ); + + try { + await storage.createLoop( + { + name: "api-scanwindow-capacity-consumer", + schedule: { type: "interval", everyMs: 1_000 }, + target: { type: "command", command: "true" }, + machine: { id: "runner-a" }, + }, + new Date("2025-12-31T23:59:00.000Z"), + ); + + const unexamined = await storage.createLoop( + { + name: "api-scanwindow-protected", + schedule: { type: "interval", everyMs: 1_000 }, + target: { type: "command", command: "true" }, + machine: { id: "runner-a" }, + overlap: "allow", + leaseMs: 1_000, + }, + new Date("2025-12-31T23:59:30.000Z"), + ); + + const PROTECTED = 520; + const protectedRunIds: string[] = []; + for (let i = 0; i < PROTECTED; i += 1) { + const slot = new Date(Date.parse("2026-01-01T00:00:00.000Z") + i * 1_000).toISOString(); + const claim = await storage.claimRun(unexamined, slot, "runner-a", now); + expect(claim).toBeTruthy(); + protectedRunIds.push(claim!.run.id); + } + + // A run this runner does NOT own, on another loop, whose lease expires + // after every protected row above. Nothing protects it, so the sweep must + // reach and reap it. + const ghostLoop = await storage.createLoop( + { + name: "api-scanwindow-reapable", + schedule: { type: "interval", everyMs: 1_000 }, + target: { type: "command", command: "true" }, + machine: { id: "runner-ghost" }, + overlap: "skip", + leaseMs: 1_000, + }, + new Date("2025-12-31T23:59:45.000Z"), + ); + const ghostClaim = await storage.claimRun( + ghostLoop, + "2026-01-01T00:30:00.000Z", + "runner-ghost", + new Date("2026-01-01T00:30:00.000Z"), + ); + expect(ghostClaim).toBeTruthy(); + + now = new Date("2026-01-01T01:00:00.000Z"); + + const poll = await fetch(apiUrl(server, "/v1/runners/claim"), { + method: "POST", + headers: jsonHeaders, + body: JSON.stringify({ runnerId: "runner-a", maxClaims: 1 }), + }); + expect(poll.status).toBe(200); + + // Two-sided: the reapable run is reaped AND the protected set is intact. + // Asserting only the first would pass on an implementation that protects + // nothing at all. + expect((await storage.getRun(ghostClaim!.run.id))!.status).toBe("abandoned"); + const protectedStatuses = await Promise.all( + protectedRunIds.map(async (id) => (await storage.getRun(id))!.status), + ); + expect(protectedStatuses.filter((status) => status === "running").length).toBe(PROTECTED); + } finally { + server.stop(true); + await storage.close(); + } + }); + + test("runner claim protection costs a bounded number of storage reads, not one per unexamined loop", async () => { + // The claim endpoint is the hosted scheduler's tick and its hottest path. + // Building the protection set with one `listRuns` per unexamined loop is an + // unbatched N+1: the shipped runner polls with `maxClaims: 1`, so a single + // claim makes EVERY remaining due loop unexamined, and `dueLoops` returns up + // to 500 of them. + // + // The invariant under test is that the cost does not scale with the number + // of unexamined loops — not any particular call count, so that adding a + // legitimate constant read later does not fail this test spuriously. + const mod = await import("./index.js"); + + const measure = async (unexaminedLoopCount: number): Promise => { + const inner = createSqliteLoopStorage(":memory:"); + const counts: Record = {}; + const storage = new Proxy(inner, { + get(target, prop, receiver) { + const value = Reflect.get(target, prop, receiver) as unknown; + if (typeof value !== "function") return value; + return (...args: unknown[]) => { + counts[String(prop)] = (counts[String(prop)] ?? 0) + 1; + return (value as (...a: unknown[]) => unknown).apply(target, args); + }; + }, + }) as unknown as LoopStorageContract; + + let now = new Date("2026-01-01T00:00:00.000Z"); + const server = createTestServer( + mod, + { host: "127.0.0.1", port: 0, storage, now: () => now, random: () => 0.5 }, + runnerPrincipal("runner-a"), + ); + try { + await inner.createLoop( + { + name: "api-nplusone-capacity-consumer", + schedule: { type: "interval", everyMs: 1_000 }, + target: { type: "command", command: "true" }, + machine: { id: "runner-a" }, + }, + new Date("2025-12-31T23:59:00.000Z"), + ); + + for (let i = 0; i < unexaminedLoopCount; i += 1) { + const loop = await inner.createLoop( + { + name: `api-nplusone-unexamined-${i}`, + schedule: { type: "interval", everyMs: 1_000 }, + target: { type: "command", command: "true" }, + machine: { id: "runner-a" }, + overlap: "allow", + leaseMs: 1_000, + }, + new Date(Date.parse("2025-12-31T23:59:30.000Z") + i), + ); + const claim = await inner.claimRun(loop, "2026-01-01T00:00:00.000Z", "runner-a", now); + expect(claim).toBeTruthy(); + } + + now = new Date("2026-01-01T01:00:00.000Z"); + // Counting starts here so loop/run construction above is excluded. + for (const key of Object.keys(counts)) delete counts[key]; + + const poll = await fetch(apiUrl(server, "/v1/runners/claim"), { + method: "POST", + headers: jsonHeaders, + body: JSON.stringify({ runnerId: "runner-a", maxClaims: 1 }), + }); + expect(poll.status).toBe(200); + return counts.listRuns ?? 0; + } finally { + server.stop(true); + await inner.close(); + } + }; + + const few = await measure(3); + const many = await measure(30); + expect(many).toBe(few); + }); + test("runner claim reaps an expired lease owned by an ineligible runner", async () => { const mod = await import("./index.js"); const storage = createSqliteLoopStorage(":memory:"); diff --git a/src/api/index.ts b/src/api/index.ts index 074c557..577960d 100644 --- a/src/api/index.ts +++ b/src/api/index.ts @@ -1343,23 +1343,35 @@ async function claimRuns( // under a runner that is about to take it over on its next poll. // // What that blanket exclusion also did, and must not, is protect a run - // belonging to a loop this poll DID examine and could not claim. That is the - // wedge: under `overlap: "skip"` the new slot is refused precisely because this - // `running` run exists, while the sweep skipped it precisely because this - // runner owns it. Neither path can fire, so the run sits in `running` with a - // long-dead lease and blocks the loop indefinitely — observed on the fleet as - // `agent-*-coordination-10m` seat loops that had fired once, ever. Same-slot - // takeover (the legitimate reason to hold a run) still works: it happens in - // the claim pass above, and it re-leases the run so the sweep no longer - // selects it at all. - const protectRunIds: string[] = []; - for (const loop of unexaminedLoops) { - for (const run of await storage.listRuns({ loopId: loop.id, status: "running" })) { - if (run.claimedBy === runner.id) protectRunIds.push(run.id); - } - } + // belonging to a loop this poll DID examine and could not claim. Under + // `catchUp: "latest"` the due list holds only the newest slot, so once wall + // time moves past a run's own slot the same-slot takeover it was being held + // for can never happen again — and the sweep skipped it precisely because + // this runner owned it. The one runner able to finalize the run was the one + // runner forbidden from reaping it, so it stayed `running` with a long-dead + // lease indefinitely and the loop's cursor advanced only if some later run + // happened to finalize, never through recovery. + // + // Be precise about what that state does and does not do, because the + // imprecise version sends the next reader to the wrong place: an expired + // lease does NOT block `overlap: "skip"`. That gate refuses a new slot only + // while a run holds a LIVE lease or a live process (sqlite + // `hasBlockingRunningRunForOtherSlot`; the Postgres predicate is strictly + // more permissive still). So what this fixes is an unreapable orphan row and + // a recovery path that could not advance the loop — not a wedged scheduler. + // + // Same-slot takeover (the legitimate reason to hold a run) is unaffected: it + // happens in the claim pass above and re-leases the run, so the sweep stops + // selecting it at all. + // + // Passed as a loop-id set, never an enumerated run-id list: enumerating runs + // costs one query per unexamined loop on the scheduler's hottest path and + // silently truncates at one `listRuns` page. const recovered = await storage.recoverExpiredRunLeasesDetailed(opts.now, { - protectRunIds, + protectClaimedByInLoops: { + claimedBy: runner.id, + loopIds: unexaminedLoops.map((loop) => loop.id), + }, }); const advancementDeferred = await advanceRecoveredRuns(storage, recovered.abandoned, { random: opts.random, diff --git a/src/lib/storage/postgres-loop-storage.test.ts b/src/lib/storage/postgres-loop-storage.test.ts index 08fe5b7..67c60ba 100644 --- a/src/lib/storage/postgres-loop-storage.test.ts +++ b/src/lib/storage/postgres-loop-storage.test.ts @@ -1368,7 +1368,7 @@ suite("PostgresLoopStorage (live)", () => { expect(result.deferred.length).toBe(0); }); - test("recoverExpiredRunLeasesDetailed honours protectRunIds", async () => { + test("recoverExpiredRunLeasesDetailed honours protectClaimedByInLoops", async () => { // The hosted control plane is the production path for this sweep, so the // option the API relies on to avoid reaping a slot a runner is about to take // over must be verified here and not only against sqlite. Both states are @@ -1385,7 +1385,7 @@ suite("PostgresLoopStorage (live)", () => { expect(reapedClaim).toBeTruthy(); const result = await storage.recoverExpiredRunLeasesDetailed(new Date(), { - protectRunIds: [keptClaim!.run.id], + protectClaimedByInLoops: { claimedBy: "runner-x", loopIds: [protectedLoop.id] }, }); expect(result.abandoned.map((run) => run.id)).toEqual([reapedClaim!.run.id]); @@ -1393,6 +1393,65 @@ suite("PostgresLoopStorage (live)", () => { expect((await storage.getRun(reapedClaim!.run.id))!.status).toBe("abandoned"); }); + test("protectClaimedByInLoops protects one runner's runs, not the whole loop", async () => { + // Scoped to the claiming runner: another runner's orphan on a protected + // loop must still be reaped, otherwise naming a loop would shelter every + // runner's dead leases on it. + const loop = await storage.createLoop(loopInput("protect-scoped-to-runner", { leaseMs: 1 })); + const past = new Date(Date.now() - 60_000); + const otherClaim = await storage.claimRun(loop, "2026-07-06T13:00:00.000Z", "runner-other", past); + expect(otherClaim).toBeTruthy(); + + const result = await storage.recoverExpiredRunLeasesDetailed(new Date(), { + protectClaimedByInLoops: { claimedBy: "runner-x", loopIds: [loop.id] }, + }); + + expect(result.abandoned.map((run) => run.id)).toEqual([otherClaim!.run.id]); + }); + + test("protected runs do not consume the recovery scan window", async () => { + // Regression for the select-then-filter ordering: discarding protected rows + // in application code AFTER the scan `LIMIT` lets a large protected set + // crowd the window and starve an unrelated, genuinely reapable run. Because + // the caller rebuilds the same protected set on every poll, that starvation + // is stable rather than transient — the same "can never be reaped" class the + // protection itself exists to remove. + // + // `scanLimit` is pinned small so the window is crossed with three rows + // rather than the five hundred the default would need. + const protectedLoop = await storage.createLoop( + loopInput("scanwindow-protected", { leaseMs: 1, overlap: "allow" }), + ); + const reapableLoop = await storage.createLoop( + loopInput("scanwindow-reapable", { leaseMs: 1, overlap: "allow" }), + ); + const early = new Date(Date.now() - 600_000); + const late = new Date(Date.now() - 60_000); + + const protectedIds: string[] = []; + for (let i = 0; i < 3; i += 1) { + const claim = await storage.claimRun(protectedLoop, `2026-07-06T14:0${i}:00.000Z`, "runner-x", early); + expect(claim).toBeTruthy(); + protectedIds.push(claim!.run.id); + } + // Expires later, so it sorts behind every protected row under + // `ORDER BY lease_expires_at ASC` and is only reached if they never + // occupied the window. + const reapable = await storage.claimRun(reapableLoop, "2026-07-06T15:00:00.000Z", "runner-y", late); + expect(reapable).toBeTruthy(); + + const result = await storage.recoverExpiredRunLeasesDetailed(new Date(), { + limit: 1, + scanLimit: 3, + protectClaimedByInLoops: { claimedBy: "runner-x", loopIds: [protectedLoop.id] }, + }); + + expect(result.abandoned.map((run) => run.id)).toEqual([reapable!.run.id]); + for (const id of protectedIds) { + expect((await storage.getRun(id))!.status).toBe("running"); + } + }); + test("paginates an immutable tenant-scoped recovered-row snapshot", async () => { for (const id of ["a", "z"]) { const loop = await storage.createLoop(loopInput(`recovered-keyset-pages-${id}`)); diff --git a/src/lib/storage/postgres-loop-storage.ts b/src/lib/storage/postgres-loop-storage.ts index 1c6a8b4..7196e1a 100644 --- a/src/lib/storage/postgres-loop-storage.ts +++ b/src/lib/storage/postgres-loop-storage.ts @@ -1592,19 +1592,24 @@ export class PostgresLoopStorage implements LoopStorageContract { Math.min(5_000, Math.floor(opts.scanLimit ?? limit * DEFAULT_RECOVERY_SCAN_MULTIPLIER)), ); const finished = now.toISOString(); + // Applied inside the query, before LIMIT — see the sqlite implementation + // and the `protectClaimedByInLoops` contract note for why a post-scan + // filter starves unrelated reapable runs. + const protect = opts.protectClaimedByInLoops; + const protectLoopIds = protect ? [...new Set(protect.loopIds)] : []; + const protectClaimedBy = protectLoopIds.length > 0 ? protect!.claimedBy : null; const rows = await this.client.many( `SELECT * FROM loop_runs WHERE tenant_id = open_loops_current_tenant_id() AND status='running' AND lease_expires_at <= $1 AND ($2::text IS NULL OR id = $2) AND ($3::text IS NULL OR claimed_by IS DISTINCT FROM $3) - ORDER BY lease_expires_at ASC LIMIT $4`, - [finished, opts.runId ?? null, opts.excludeClaimedBy ?? null, scanLimit], + AND ($4::text IS NULL OR claimed_by IS NULL OR claimed_by <> $4 OR NOT (loop_id = ANY($5::text[]))) + ORDER BY lease_expires_at ASC LIMIT $6`, + [finished, opts.runId ?? null, opts.excludeClaimedBy ?? null, protectClaimedBy, protectLoopIds, scanLimit], ); const recovered: LoopRun[] = []; - const protectedRunIds = new Set(opts.protectRunIds ?? []); for (const row of rows) { if (recovered.length >= limit) break; - if (protectedRunIds.has(row.id)) continue; const run = await this.client.transaction(async (c) => { const res = await c.query( `UPDATE loop_runs SET status='abandoned', finished_at=$2, lease_expires_at=NULL, diff --git a/src/lib/store.test.ts b/src/lib/store.test.ts index c76414a..5315a6a 100644 --- a/src/lib/store.test.ts +++ b/src/lib/store.test.ts @@ -2701,6 +2701,113 @@ exit 0 } }); + test("lease recovery honours protectClaimedByInLoops and scopes it to the claiming runner", () => { + const store = new Store(":memory:"); + try { + const protectedLoop = store.createLoop( + { + name: "protect-kept", + schedule: { type: "interval", everyMs: 60_000 }, + target: { type: "command", command: "true" }, + overlap: "allow", + leaseMs: 10, + }, + new Date("2025-12-31T00:00:00Z"), + ); + const otherLoop = store.createLoop( + { + name: "protect-reaped", + schedule: { type: "interval", everyMs: 60_000 }, + target: { type: "command", command: "true" }, + leaseMs: 10, + }, + new Date("2025-12-31T00:00:00Z"), + ); + const at = new Date("2026-01-01T00:00:00Z"); + const kept = store.claimRun(protectedLoop, "2026-01-01T00:00:00.000Z", "runner-x", at); + // Same protected loop, different runner: the protection is per-runner, so + // this one must still be reaped. + const otherRunnerSameLoop = store.claimRun(protectedLoop, "2026-01-01T00:01:00.000Z", "runner-y", at); + const reaped = store.claimRun(otherLoop, "2026-01-01T00:00:00.000Z", "runner-x", at); + expect(kept).toBeTruthy(); + expect(otherRunnerSameLoop).toBeTruthy(); + expect(reaped).toBeTruthy(); + + const result = store.recoverExpiredRunLeasesDetailed(new Date("2026-01-01T00:02:00Z"), { + protectClaimedByInLoops: { claimedBy: "runner-x", loopIds: [protectedLoop.id] }, + }); + + expect(result.abandoned.map((run) => run.id).sort()).toEqual( + [otherRunnerSameLoop!.run.id, reaped!.run.id].sort(), + ); + expect(store.getRun(kept!.run.id)?.status).toBe("running"); + } finally { + store.close(); + } + }); + + test("protected runs do not consume the lease-recovery scan window", () => { + // Regression for select-then-filter ordering: protected rows discarded in + // application code after the scan `LIMIT` crowd the window and starve an + // unrelated reapable run. The caller rebuilds the same protected set every + // poll, so the starvation is permanent rather than transient. `scanLimit` + // is pinned small so three rows cross the window instead of five hundred. + const store = new Store(":memory:"); + try { + const protectedLoop = store.createLoop( + { + name: "scanwindow-protected", + schedule: { type: "interval", everyMs: 60_000 }, + target: { type: "command", command: "true" }, + overlap: "allow", + leaseMs: 10, + }, + new Date("2025-12-31T00:00:00Z"), + ); + const reapableLoop = store.createLoop( + { + name: "scanwindow-reapable", + schedule: { type: "interval", everyMs: 60_000 }, + target: { type: "command", command: "true" }, + leaseMs: 10, + }, + new Date("2025-12-31T00:00:00Z"), + ); + + const protectedIds: string[] = []; + for (let i = 0; i < 3; i += 1) { + const claim = store.claimRun( + protectedLoop, + `2026-01-01T00:0${i}:00.000Z`, + "runner-x", + new Date("2026-01-01T00:00:00Z"), + ); + expect(claim).toBeTruthy(); + protectedIds.push(claim!.run.id); + } + // Claimed later, so its lease expires last and it sorts behind every + // protected row under `ORDER BY lease_expires_at ASC`. + const reapable = store.claimRun( + reapableLoop, + "2026-01-01T00:10:00.000Z", + "runner-y", + new Date("2026-01-01T00:10:00Z"), + ); + expect(reapable).toBeTruthy(); + + const result = store.recoverExpiredRunLeasesDetailed(new Date("2026-01-01T01:00:00Z"), { + limit: 1, + scanLimit: 3, + protectClaimedByInLoops: { claimedBy: "runner-x", loopIds: [protectedLoop.id] }, + }); + + expect(result.abandoned.map((run) => run.id)).toEqual([reapable!.run.id]); + for (const id of protectedIds) expect(store.getRun(id)?.status).toBe("running"); + } finally { + store.close(); + } + }); + test("lease recovery abandons runs whose live pid fails the start-time fingerprint", () => { const store = new Store(":memory:"); try { diff --git a/src/lib/store.ts b/src/lib/store.ts index 60810de..16080dd 100644 --- a/src/lib/store.ts +++ b/src/lib/store.ts @@ -5008,26 +5008,46 @@ export class Store { runId?: string; excludeClaimedBy?: string; /** - * Run ids to leave untouched. Unlike `excludeClaimedBy`, which protects - * everything a runner owns, this protects an explicit set the caller has - * established is still recoverable by other means (for Loops: still - * same-slot claimable). A runner's own run whose slot has already moved - * past is NOT recoverable by takeover, so blanket-excluding it strands it - * in `running` forever and, under `overlap: "skip"`, blocks the loop. + * Leave one runner's runs untouched, but only within an explicit set of + * loops. Unlike `excludeClaimedBy`, which protects everything a runner + * owns unconditionally, this protects only the loops the caller has + * established are still recoverable by other means (for Loops: loops a + * poll never examined, whose runs that runner is about to take over on + * its next poll). A runner's own run on a loop that WAS examined is not + * recoverable by takeover, so blanket-excluding it strands it in + * `running` with a dead lease indefinitely. + * + * Expressed as a loop-id set rather than a run-id set on purpose: a + * run-id set has to be enumerated by the caller, which both caps + * silently at one page and costs a query per loop. It is applied inside + * the scan query, BEFORE `LIMIT`, so protected rows never consume the + * scan window and starve an unrelated reapable run. */ - protectRunIds?: readonly string[]; + protectClaimedByInLoops?: { claimedBy: string; loopIds: readonly string[] }; /** Leave every currently live process untouched, even after the daemon recovery grace ceiling. */ preserveLiveProcesses?: boolean; } = {}, ): RecoverExpiredRunLeasesResult { const limit = Math.max(1, Math.min(1_000, Math.floor(opts.limit ?? DEFAULT_RECOVERY_BATCH_LIMIT))); const scanLimit = Math.max(limit, Math.min(5_000, Math.floor(opts.scanLimit ?? limit * DEFAULT_RECOVERY_SCAN_MULTIPLIER))); + // Capacity protection is part of the QUERY, not a post-scan filter: rows + // discarded after `LIMIT` have already consumed the scan window, so a large + // protected set would crowd out an unrelated expired run and — because the + // caller rebuilds the same protected set on every poll — starve it + // permanently rather than transiently. + const protect = opts.protectClaimedByInLoops; + const protectLoopIds = protect ? [...new Set(protect.loopIds)] : []; + // `claimed_by IS NULL` first: an unclaimed row must stay reapable, and a + // bare `claimed_by <> ?` is NULL (not true) for those rows. + const protectClause = protectLoopIds.length > 0 + ? ` AND (claimed_by IS NULL OR claimed_by <> ? OR loop_id NOT IN (${protectLoopIds.map(() => "?").join(",")}))` + : ""; const rows = this.db - .query( + .query>( `SELECT * FROM loop_runs WHERE status = 'running' AND lease_expires_at <= ? AND (? IS NULL OR id = ?) - AND (? IS NULL OR claimed_by IS NULL OR claimed_by <> ?) + AND (? IS NULL OR claimed_by IS NULL OR claimed_by <> ?)${protectClause} ORDER BY lease_expires_at ASC LIMIT ?`, ) @@ -5037,14 +5057,13 @@ export class Store { opts.runId ?? null, opts.excludeClaimedBy ?? null, opts.excludeClaimedBy ?? null, + ...(protectLoopIds.length > 0 ? [protect!.claimedBy, ...protectLoopIds] : []), scanLimit, ); const recovered: LoopRun[] = []; const deferred: LoopRun[] = []; - const protectedRunIds = new Set(opts.protectRunIds ?? []); for (const row of rows) { if (recovered.length >= limit) break; - if (protectedRunIds.has(row.id)) continue; const looksAlive = isRecordedProcessAlive(row.pid, row.process_started_at) || this.hasLiveWorkflowStepProcesses(row.id); // "Looks alive" only buys a BOUNDED grace. Past the ceiling the run is // abandoned regardless: an expired lease that keeps failing to renew is From c659c2e865461c8f31e118b49b90ce7fdab7dc70 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Mon, 3 Aug 2026 00:23:35 +0300 Subject: [PATCH 3/3] fix(scheduler): protect partial-loop capacity Agent: unresolved-account001 --- src/api/index.test.ts | 52 +++++++++++++++++++++++++++++++++++++++++++ src/api/index.ts | 12 ++++++++-- 2 files changed, 62 insertions(+), 2 deletions(-) diff --git a/src/api/index.test.ts b/src/api/index.test.ts index afaa2b5..c02ed48 100644 --- a/src/api/index.test.ts +++ b/src/api/index.test.ts @@ -3007,6 +3007,58 @@ describe("loops-api foundation", () => { } }); + test("runner claim capacity protects later unexamined slots in the partially examined loop", async () => { + const mod = await import("./index.js"); + const storage = createSqliteLoopStorage(":memory:"); + let now = new Date("2026-01-01T00:00:00.000Z"); + const server = createTestServer( + mod, + { host: "127.0.0.1", port: 0, storage, now: () => now, random: () => 0.5 }, + runnerPrincipal("runner-a"), + ); + + try { + const loop = await storage.createLoop( + { + name: "api-partially-examined-capacity", + schedule: { type: "interval", everyMs: 1_000 }, + target: { type: "command", command: "true" }, + catchUp: "all", + catchUpLimit: 10, + overlap: "allow", + maxAttempts: 1, + leaseMs: 1_000, + }, + now, + ); + const firstSlot = loop.nextRunAt!; + const secondSlot = new Date(new Date(firstSlot).getTime() + 1_000).toISOString(); + const first = await storage.claimRun(loop, firstSlot, "runner-a", new Date(firstSlot)); + const second = await storage.claimRun(loop, secondSlot, "runner-a", new Date(firstSlot)); + expect(first).toBeTruthy(); + expect(second).toBeTruthy(); + + now = new Date("2026-01-01T00:00:10.000Z"); + const poll = await fetch(apiUrl(server, "/v1/runners/claim"), { + method: "POST", + headers: jsonHeaders, + body: JSON.stringify({ runnerId: "runner-a", maxClaims: 1 }), + }); + expect(poll.status).toBe(200); + const body = (await poll.json()) as { claims: Array<{ run: { id: string } }> }; + expect(body.claims.map((claim) => claim.run.id)).toEqual([first!.run.id]); + + expect(await storage.getRun(second!.run.id)).toMatchObject({ + status: "running", + claimedBy: "runner-a", + attempt: 1, + }); + } finally { + server.stop(true); + await storage.close(); + } + }); + test("runner claim reaps its own expired lease once the due slot has moved past it", async () => { // Regression for the wedged-run defect: a `catchUp: "latest"` + `overlap: "skip"` // loop (the shape every agent-*-coordination-10m seat loop uses: 10m interval, diff --git a/src/api/index.ts b/src/api/index.ts index 577960d..6e9ffb0 100644 --- a/src/api/index.ts +++ b/src/api/index.ts @@ -1296,8 +1296,10 @@ async function claimRuns( const claims: Array> = []; const dueLoopsForPoll = await storage.dueLoops(opts.now); // Loops this poll never got to look at because claim capacity ran out first. - // Their runs are the ones the sweep must not touch — see `protectRunIds` below. + // Their runs are the ones the sweep must not touch — see + // `protectClaimedByInLoops` below. let unexaminedLoops: typeof dueLoopsForPoll = []; + pollDueLoops: for (const [loopIndex, loop] of dueLoopsForPoll.entries()) { if (claims.length >= opts.maxClaims) { unexaminedLoops = dueLoopsForPoll.slice(loopIndex); @@ -1309,7 +1311,13 @@ async function claimRuns( : undefined; if (loop.target.type === "workflow" && !workflow) continue; for (const slot of dueSlots(loop, opts.now).slots) { - if (claims.length >= opts.maxClaims) break; + if (claims.length >= opts.maxClaims) { + // Capacity can be consumed inside one catch-up plan. In that case this + // loop still has unexamined slots, so protect it along with every later + // loop rather than letting the recovery sweep abandon those slots. + unexaminedLoops = dueLoopsForPoll.slice(loopIndex); + break pollDueLoops; + } const claim = await storage.claimRun(loop, slot, runner.id, opts.now); if (!claim) continue; const run = await storage.heartbeatRunLease(