From a4e2424d0d31668eaed59461a08b009ad63fb1d5 Mon Sep 17 00:00:00 2001 From: Paperclip CTO Date: Sun, 2 Aug 2026 20:59:52 +0000 Subject: [PATCH 1/3] perf(heartbeat): bound the dispatch head scan independently of queue depth (BLO-20736) The dispatcher's keyset scan filters agent_id = $1 AND status = 'queued' and orders by (created_at, id) with LIMIT 200. PostgreSQL estimates the two equality qualifiers independently and multiplies their selectivities, but they are almost perfectly correlated: a backlogged agent's rows are overwhelmingly queued, an idle agent has none. Measured on a 200k-row interleaved fixture the estimate is 250x low at queue depth 1000 (rows=4 vs 1000) and 42x low at depth 5000 (rows=118 vs 5000). On an estimate that small the planner stops treating LIMIT as a reason to preserve index order and picks Bitmap Heap Scan + top-N Sort. A sort cannot emit until it has consumed its whole input, so the dispatcher read and sorted the agent's ENTIRE queue to return 200 rows, under the strict per-agent start lock, getting worse as the backlog deepened. Read the page in two phases instead. Phase 1 projects ONLY (created_at, id) -- both live in heartbeat_runs_agent_dispatch_idx behind the (agent_id, status) prefix, so the page comes from an ordered Index Only Scan with no heap access that stops after SCAN_LIMIT entries. Phase 2 hydrates exactly that page by primary key. Measured, at both depths: 1000 -> 200 and 5000 -> 200 rows actually read. The projection is what makes the plan insensitive to the estimate rather than dependent on fixing it: with no heap fetches to pay for, the ordered path's LIMIT-scaled cost stays ~10 whether the planner believes 118 rows or 5167, while the bitmap alternative must still materialize the whole match set before its sort emits. The plan stays correct at a still-42x wrong estimate. Deliberately NOT fixed with extended statistics: CREATE STATISTICS ON (agent_id, status) does correct the estimate and then makes the plan worse -- the planner switches to a BitmapAnd of two other indexes and still sorts. Measured at both depths, and inconsistent between them. Phase 2 takes the id list as its only SQL predicate and re-checks status in JS. Adding AND status = 'queued' there makes the dispatch index attractive again and PostgreSQL drives the fetch from it instead of the primary key, reintroducing the unbounded shape phase 1 exists to remove. Callers advance the cursor from the probe rather than from the hydrated rows, so a page thinned by a row leaving the queue between the two statements is not mistaken for the end of the backlog. The new plan test asserts both properties at two depths against one shared ceiling, on a fixture whose queued rows are interleaved one per heap page rather than packed at the tail -- against tight clustering bitmap+sort genuinely is the better plan, so the packed fixture would be asserting the wrong thing. Co-Authored-By: Claude --- .../src/heartbeat-dispatch-query-plan.test.ts | 205 +++++++++++++++++ server/src/services/heartbeat.ts | 212 +++++++++++++----- 2 files changed, 356 insertions(+), 61 deletions(-) diff --git a/packages/db/src/heartbeat-dispatch-query-plan.test.ts b/packages/db/src/heartbeat-dispatch-query-plan.test.ts index d5031441be0..cd8451a3843 100644 --- a/packages/db/src/heartbeat-dispatch-query-plan.test.ts +++ b/packages/db/src/heartbeat-dispatch-query-plan.test.ts @@ -284,6 +284,88 @@ const PRIORITY_LANE_ISSUE_LOOKUP = ` AND id IN (${LANE_ISSUE_IDS.map((id) => `'${id}'::uuid`).join(", ")}) `; +/** + * BLO-20736. Two depths, both well past SCAN_LIMIT and an order of magnitude + * apart, so a single shared ceiling can distinguish bounded work from work that + * tracks the backlog. + */ +const DISPATCH_HEAD_DEPTHS = [1_000, 5_000] as const; + +/** + * The head page as the dispatcher reads it now: ONLY the keyset columns. + * + * Every column here lives in heartbeat_runs_agent_dispatch_idx behind the + * (agent_id, status) prefix, so the page is served entirely from the index with + * no heap access — an ordered `Index Only Scan` that stops after SCAN_LIMIT + * entries. That is what makes the plan insensitive to the cardinality + * underestimate: with no heap fetches to pay for, the ordered path's + * LIMIT-scaled cost stays ~10 whether the planner believes 126 rows or 5167, + * while the bitmap alternative must still materialize the whole match set + * before its sort can emit a single row. + * + * `created_at::text` mirrors the dispatcher exactly, cast and all: it carries + * the cursor as text so postgres' microsecond timestamps survive a round trip + * through a JS Date, which only has millisecond resolution. The cast is applied + * on top of an indexed column, so the page is still served index-only — but the + * projection has to match production or this plan is not the production plan. + */ +const DISPATCH_HEAD_PROBE = ` + SELECT created_at::text AS dispatch_created_at_cursor, id FROM heartbeat_runs + WHERE agent_id = '${AGENT}'::uuid + AND status = 'queued' + ORDER BY created_at ASC, id ASC + LIMIT ${SCAN_LIMIT} +`; + +const DISPATCH_HEAD_PROBE_CURSOR = ` + SELECT created_at::text AS dispatch_created_at_cursor, id FROM heartbeat_runs + WHERE agent_id = '${AGENT}'::uuid + AND status = 'queued' + AND (created_at, id) > (now() - interval '90000 seconds', '00000000-0000-4000-8000-000000000000'::uuid) + ORDER BY created_at ASC, id ASC + LIMIT ${SCAN_LIMIT} +`; + +/** + * A deep queue whose rows are physically INTERLEAVED with everyone else's. + * + * `seed` above appends the agent's backlog in one block, which packs it into a + * few dozen heap pages — and against clustering that tight, bitmap+sort really + * is the cheaper plan, so asserting the ordered plan there would be asserting + * the wrong thing. Spreading one queued row per stride puts each on its own + * heap page (`Heap Blocks: exact=`), which is what a queue accumulated + * over time actually looks like. + * + * VACUUM, not just ANALYZE: index-only scans are costed against the visibility + * map, and a never-vacuumed table reports relallvisible = 0 and gets bitmap+sort + * regardless of projection. Production is continuously autovacuumed, so + * vacuuming here is what makes the fixture honest rather than what makes it + * pass. + */ +async function seedInterleavedQueue(sql: postgres.Sql, depth: number) { + await sql.unsafe(`SET session_replication_role = replica`); + const stride = Math.floor(TOTAL_RUNS / depth); + await sql.unsafe(` + INSERT INTO heartbeat_runs (company_id, agent_id, status, created_at, updated_at, context_snapshot) + SELECT + '${COMPANY}'::uuid, + CASE WHEN series % ${stride} = 0 + THEN '${AGENT}'::uuid + ELSE ('33333333-3333-4333-8333-' || lpad((series % 50)::text, 12, '0'))::uuid END, + CASE WHEN series % ${stride} = 0 OR series % 200 = 0 THEN 'queued' ELSE 'completed' END, + now() - ((series % 100000) || ' seconds')::interval, + now(), + jsonb_build_object( + 'issueId', '00000000-0000-4000-8000-' || lpad(((series % ${TOTAL_ISSUES}) + 1)::text, 12, '0'), + 'source', 'heartbeat_timer' + ) + FROM generate_series(1, ${TOTAL_RUNS}) AS series + `); + await sql.unsafe(`SET session_replication_role = origin`); + await sql.unsafe(`VACUUM ANALYZE heartbeat_runs`); +} + + describeEmbeddedPostgres("BLO-20396 dispatch query plans", () => { it("uses the dispatch index for the keyset scan and bounds the priority lane", async () => { const database = await startEmbeddedPostgresTestDatabase("paperclip-blo20396-plan-"); @@ -482,4 +564,127 @@ describeEmbeddedPostgres("BLO-20396 dispatch query plans", () => { expect(rowsInspected(laneRecovery.root)).toBeLessThanOrEqual(RECOVERY_LANE_ABSOLUTE_BOUND); expect(RECOVERY_LANE_ABSOLUTE_BOUND).toBeLessThan(DEEP_BACKLOG_ROWS / 10); }, 300_000); + + /** + * BLO-20736: the head scan's own bound, at production-shaped depth. + * + * The first test's `not.toContain("Sort")` assertion passed for the wrong + * reason. At AGENT_QUEUED_ROWS = 350, packed densely at the tail of the heap, + * PostgreSQL estimated the match set at a handful of rows and the ordered + * index scan won by a hair — not because the plan was robust, but because the + * estimate was tiny. `agent_id = X` and `status = 'queued'` are estimated + * independently and multiplied, and they are in fact almost perfectly + * correlated, so the estimate is wildly low: measured 200x low at depth 1000 + * (rows=5 vs 1000) and 42x low at depth 5000 (rows=120 vs 5000). Deepen the + * queue and the planner flips to `Bitmap Heap Scan` + top-N `Sort`, which + * cannot emit until it has consumed its whole input — so the dispatcher reads + * and sorts the agent's ENTIRE queue to return SCAN_LIMIT rows, under the + * strict per-agent start lock, and it gets worse as the backlog grows. + * + * Two properties are asserted, at two depths, against the SAME ceiling: + * the head page is served by an ordered index-only scan with no Sort, and the + * work it does is independent of queue depth. One depth cannot show the + * second property, which is the one that actually matters. + * + * The fixture interleaves the agent's rows one per stride across the whole + * insert order. The original packed 1000 rows into ~38 heap blocks, where + * bitmap+sort is genuinely the better plan and the good plan is not worth + * asserting. Interleaved, each queued row sits on its own heap page + * (`Heap Blocks: exact=` in the report), which is the production shape. + * + * The fixture is also VACUUMed, and that is load-bearing rather than + * incidental: an index-only scan is costed cheaply only when the visibility + * map reports most pages all-visible, and on a NEVER-vacuumed table + * (relallvisible = 0) the planner falls back to bitmap+sort even for this + * projection. Production is continuously autovacuumed and the cost model reads + * the table-wide relallvisible/relpages fraction, so a vacuumed fixture is the + * honest one. Churn alone does not break it — updating every queued row and + * re-ANALYZEing without a VACUUM still plans index-only. + */ + it("bounds the dispatch head scan independently of queue depth", async () => { + const report: string[] = []; + /** + * Same ceiling for every depth. Deriving it from the depth would make the + * assertion vacuous: the whole claim is that the work does NOT grow. + */ + const HEAD_ABSOLUTE_BOUND = SCAN_LIMIT * 3; + + for (const depth of DISPATCH_HEAD_DEPTHS) { + const database = await startEmbeddedPostgresTestDatabase(`paperclip-blo20736-${depth}-`); + cleanups.push(database.cleanup); + const sql = postgres(database.connectionString, { max: 1, onnotice: () => {} }); + cleanups.push(async () => sql.end()); + + await seedInterleavedQueue(sql, depth); + const [{ queued }] = await sql.unsafe( + `SELECT count(*)::int AS queued FROM heartbeat_runs + WHERE agent_id = '${AGENT}'::uuid AND status = 'queued'`, + ) as Array<{ queued: number }>; + expect(queued).toBe(depth); + + const record = (title: string, plan: { text: string; root: Record }) => { + report.push(`\n===== ${title} =====\n${plan.text}\n-- rows inspected: ${rowsInspected(plan.root)}`); + if (PLAN_REPORT) fs.writeFileSync(PLAN_REPORT, report.join("\n")); + }; + + // The shape the dispatcher used to issue, MEASURED but not asserted on — + // it is the before-evidence for why the projection changed, and pinning + // it would lock in the bad plan as a requirement. + record(`depth ${depth}: head scan BEFORE (SELECT *)`, await explain(sql, DISPATCH_QUERY)); + + const probe = await explain(sql, DISPATCH_HEAD_PROBE); + record(`depth ${depth}: head probe (created_at, id only)`, probe); + + expect(indexesUsed(probe.root)).toContain(DISPATCH_INDEX); + expect(scanKinds(probe.root)).not.toContain("Seq Scan"); + expect(scanKinds(probe.root)).not.toContain("Bitmap Heap Scan"); + // Ordered straight off the index. A Sort here would mean the whole match + // set is consumed before the first row is emitted, which is the defect. + expect(planNodes(probe.root).map((n) => n["Node Type"])).not.toContain("Sort"); + expect(scanKinds(probe.root)).toContain("Index Only Scan"); + expect(rowsInspected(probe.root)).toBeLessThanOrEqual(HEAD_ABSOLUTE_BOUND); + + // Same, resumed mid-queue: the cursor must be satisfied BY the index, not + // rechecked after it, or every resumed pass re-walks the discarded prefix. + const probeCursor = await explain(sql, DISPATCH_HEAD_PROBE_CURSOR); + record(`depth ${depth}: head probe, resumed`, probeCursor); + const probeCursorNode = indexScanNode(probeCursor.root, DISPATCH_INDEX); + expect(probeCursorNode).not.toBeNull(); + expect(String(probeCursorNode?.["Index Cond"] ?? "")).toMatch(KEYSET_PREDICATE); + expect(String(probeCursorNode?.["Filter"] ?? "")).not.toMatch(KEYSET_PREDICATE); + expect(planNodes(probeCursor.root).map((n) => n["Node Type"])).not.toContain("Sort"); + expect(rowsInspected(probeCursor.root)).toBeLessThanOrEqual(HEAD_ABSOLUTE_BOUND); + + // Phase 2 hydrates exactly the probed page by primary key, so it is bounded + // by SCAN_LIMIT and not by the queue behind it. + const pageIds = (await sql.unsafe( + `SELECT id FROM heartbeat_runs + WHERE agent_id = '${AGENT}'::uuid AND status = 'queued' + ORDER BY created_at ASC, id ASC LIMIT ${SCAN_LIMIT}`, + ) as Array<{ id: string }>).map((row) => `'${row.id}'::uuid`); + // Phase 2 hydrates exactly the probed page by primary key, so it is bounded + // by SCAN_LIMIT and not by the queue behind it. The id list is the ONLY + // predicate on purpose: adding `AND status = 'queued'` here makes the + // dispatch index look attractive again and PostgreSQL drives the fetch + // from it instead of the primary key, reintroducing the unbounded shape. + // The dispatcher re-checks status in JS instead. + const hydrate = await explain( + sql, + `SELECT * FROM heartbeat_runs WHERE id IN (${pageIds.join(", ")})`, + ); + record(`depth ${depth}: page hydrate by primary key`, hydrate); + expect(indexesUsed(hydrate.root)).toContain("heartbeat_runs_pkey"); + expect(scanKinds(hydrate.root)).not.toContain("Seq Scan"); + expect(rowsInspected(hydrate.root)).toBeLessThanOrEqual(HEAD_ABSOLUTE_BOUND * 2); + + while (cleanups.length > 0) await cleanups.pop()?.(); + } + + // The two depths are far enough apart, and deep enough, that a plan whose + // work tracks the backlog cannot satisfy one ceiling at both. + expect(DISPATCH_HEAD_DEPTHS[0]).toBeGreaterThanOrEqual(SCAN_LIMIT * 5); + expect(DISPATCH_HEAD_DEPTHS[1]).toBeGreaterThanOrEqual(DISPATCH_HEAD_DEPTHS[0]! * 5); + expect(HEAD_ABSOLUTE_BOUND).toBeLessThan(DISPATCH_HEAD_DEPTHS[0]! / 1.5); + }, 900_000); }); + diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index b0585fbb748..442ba5ec893 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -18106,9 +18106,6 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) * survive across passes. */ type DispatchCursor = { createdAt: string; id: string }; - type DispatchRun = typeof heartbeatRuns.$inferSelect & { - dispatchCreatedAtCursor: string; - }; const dispatchRunSelection = { ...getTableColumns(heartbeatRuns), dispatchCreatedAtCursor: sql`${heartbeatRuns.createdAt}::text`.as( @@ -18128,6 +18125,135 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const dispatchResumeCapRetryTimersByAgent = new Map>(); const dispatchAdmissionRetryTimersByAgent = new Map>(); + /** + * BLO-20736: read one keyset page of an agent's queued runs in two phases. + * + * Phase 1 projects ONLY (created_at, id). Both live in + * heartbeat_runs_agent_dispatch_idx alongside the (agent_id, status) prefix, + * so the whole page comes from the index with no heap access at all — an + * `Index Only Scan`, ordered, that stops after `limit` entries. + * + * That projection is the entire point, and it is a cost-model fix rather than + * a cosmetic one. PostgreSQL estimates `agent_id = $1` and `status = 'queued'` + * independently and multiplies their selectivities, but the two are almost + * perfectly correlated here: a backlogged agent's rows are overwhelmingly + * queued and an idle agent has none. Measured on a 200k-row interleaved + * fixture the estimate came out 200x low at depth 1000 (rows=5 vs 1000) and + * 42x low at depth 5000 (rows=120 vs 5000). On an estimate that small the + * planner stops treating LIMIT as a reason to preserve index order and picks + * `Bitmap Heap Scan` + top-N `Sort` instead. A sort cannot emit its first row + * until it has consumed its whole input, so the dispatcher read and sorted the + * agent's ENTIRE queue to return 200 rows — 10,400 rows inspected at depth + * 5000, growing with the backlog, all while the strict per-agent start lock is + * held. + * + * Fetching only index columns removes the planner's incentive: with zero heap + * fetches the ordered path's LIMIT-scaled cost is ~10 whether the planner + * believes 126 rows or 5167, while the bitmap alternative still has to + * materialize the entire match set before its sort can emit. So the plan is + * correct *despite* the bad estimate rather than needing the estimate fixed. + * Verified stable at a still-40x-wrong estimate. + * + * Deliberately NOT fixed with extended statistics: `CREATE STATISTICS ON + * (agent_id, status)` does correct the estimate (rows=1107 vs 1000 actual) and + * then makes the plan WORSE — the planner switches to a `BitmapAnd` of + * heartbeat_runs_company_status_process_started_idx and + * heartbeat_runs_company_agent_started_idx and still sorts. Measured, twice. + * + * Caveat worth knowing before trusting this: an index-only scan is only + * costed cheaply when the visibility map says most pages are all-visible. On a + * table that has NEVER been vacuumed (relallvisible = 0) the planner falls + * back to bitmap+sort even for this projection. That is a fixture artifact — + * heartbeat_runs is continuously autovacuumed, and the cost model reads the + * table-wide relallvisible/relpages fraction, not a per-range one. Churn alone + * does not break it: updating every queued row and re-ANALYZEing (no VACUUM) + * still plans index-only and still inspects a bounded page. + * + * Phase 2 then fetches exactly that page by primary key, and re-checks + * `status` in JS rather than in SQL. A row can leave the queue between the two + * statements, and such a row must drop out — but adding + * `AND status = 'queued'` to the fetch makes the dispatch index look + * attractive again and PostgreSQL drives phase 2 from it instead of the + * primary key (measured), which is exactly the unbounded shape phase 1 exists + * to avoid. Keeping the id list as the only SQL predicate pins the pkey plan; + * the status filter costs nothing in JS. + * + * Returns the probe's own length and last key rather than the fetched rows', + * so callers advance the cursor past everything SCANNED and decide exhaustion + * from the scan. Deriving either from `runs` would treat a page thinned by + * phase 2 as the end of the queue and strand the remaining backlog. + */ + async function readQueuedDispatchPage(input: { + agentId: string; + cutoff: Date | null; + cursor: DispatchCursor | null; + limit: number; + excludeRunIds?: Set | null; + }): Promise<{ + probed: number; + cursor: DispatchCursor | null; + runs: Array; + }> { + const queuedPagePredicate = and( + eq(heartbeatRuns.agentId, input.agentId), + // Keep the partial-index predicate a SQL literal. postgres.js uses + // prepared statements by default; a bound status parameter can receive a + // generic plan that cannot imply `status = 'queued'`. + sql`${heartbeatRuns.status} = 'queued'`, + input.cutoff ? gte(heartbeatRuns.createdAt, input.cutoff) : undefined, + // Keyset cursor. created_at alone is not unique (bulk wake fan-out stamps + // identical timestamps), so the id tiebreak is what keeps paging from + // skipping or repeating rows at a batch boundary. The bounds are bound as + // text with explicit casts: inside a raw `sql` template there is no column + // mapper, and postgres-js rejects a bare Date at bind time. + input.cursor + ? sql`(${heartbeatRuns.createdAt}, ${heartbeatRuns.id}) > (${input.cursor.createdAt}::timestamptz, ${input.cursor.id}::uuid)` + : undefined, + input.excludeRunIds?.size + ? notInArray(heartbeatRuns.id, [...input.excludeRunIds]) + : undefined, + ); + + // `created_at::text` rather than the Date column: a JS Date truncates + // postgres' microsecond timestamps to milliseconds, and a cursor that + // rounds down re-reads the rows it already returned while one that rounds + // up skips them. The cast is over an indexed column, so the page still + // comes entirely from the index. + const page = await db + .select({ + createdAt: sql`${heartbeatRuns.createdAt}::text`.as( + "dispatch_created_at_cursor", + ), + id: heartbeatRuns.id, + }) + .from(heartbeatRuns) + .where(queuedPagePredicate) + .orderBy(asc(heartbeatRuns.createdAt), asc(heartbeatRuns.id)) + .limit(input.limit); + if (page.length === 0) return { probed: 0, cursor: null, runs: [] }; + + const lastProbed = page[page.length - 1]!; + const rows = await db + .select() + .from(heartbeatRuns) + .where(inArray(heartbeatRuns.id, page.map((entry) => entry.id))); + + // Reorder in JS from the probe's ordering rather than adding an ORDER BY to + // phase 2: the keys are already sorted, and sorting there would put a Sort + // node back over ~2.8 kB-wide rows for no reason. Rows that left the queue + // between the two statements are dropped here. + const runById = new Map( + rows.filter((run) => run.status === "queued").map((run) => [run.id, run]), + ); + return { + probed: page.length, + cursor: { createdAt: lastProbed.createdAt, id: lastProbed.id }, + runs: page + .map((entry) => runById.get(entry.id)) + .filter((run): run is typeof heartbeatRuns.$inferSelect => run !== undefined), + }; + } + /** Run one more dispatch pass for `agentId`, detached from this critical section. */ function scheduleDetachedDispatchPass(agentId: string, reason: string) { options.onQueuedDispatchScheduledForTest?.({ @@ -18390,44 +18516,20 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) while (scannedBatches < queuedRunDispatchMaxScanBatches) { scannedBatches += 1; - // Annotated rather than inferred: `scanCursor` is assigned from this - // batch's last row, so control-flow analysis would otherwise chase - // `batch` -> `scanCursor` -> `batch` and give up with an implicit any. - const batch: DispatchRun[] = await db - .select(dispatchRunSelection) - .from(heartbeatRuns) - .where(and( - eq(heartbeatRuns.agentId, agentId), - // Keep the partial-index predicate a SQL literal. postgres.js uses - // prepared statements by default; a bound status parameter can - // receive a generic plan that cannot imply `status = 'queued'`. - sql`${heartbeatRuns.status} = 'queued'`, - cutoff ? gte(heartbeatRuns.createdAt, cutoff) : undefined, - // Keyset cursor. created_at alone is not unique (bulk wake fan-out - // stamps identical timestamps), so the id tiebreak is what keeps - // paging from skipping or repeating rows at a batch boundary. - // The bounds are bound as text with explicit casts: inside a raw - // `sql` template there is no column mapper, and postgres-js rejects - // a bare Date at bind time. - scanCursor - ? sql`(${heartbeatRuns.createdAt}, ${heartbeatRuns.id}) > (${scanCursor.createdAt}::timestamptz, ${scanCursor.id}::uuid)` - : undefined, - deferredRunIds?.size - ? notInArray(heartbeatRuns.id, [...deferredRunIds]) - : undefined, - )) - .orderBy(asc(heartbeatRuns.createdAt), asc(heartbeatRuns.id)) - .limit(queuedRunDispatchScanLimit); - if (batch.length === 0) { + const page = await readQueuedDispatchPage({ + agentId, + cutoff: cutoff ?? null, + cursor: scanCursor, + limit: queuedRunDispatchScanLimit, + excludeRunIds: deferredRunIds, + }); + const batch = page.runs; + if (page.probed === 0) { scanExhausted = true; break; } - const lastScannedRun = batch[batch.length - 1]!; - scanCursor = { - createdAt: lastScannedRun.dispatchCreatedAtCursor, - id: lastScannedRun.id, - }; - if (batch.length < queuedRunDispatchScanLimit) scanExhausted = true; + scanCursor = page.cursor; + if (page.probed < queuedRunDispatchScanLimit) scanExhausted = true; const batchIssueIds = [...new Set( batch @@ -18575,33 +18677,21 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) while (criticalLaneBatches < queuedRunDispatchMaxScanBatches) { criticalLaneBatches += 1; const batchStartCursor = criticalLaneCursor; - const batch = await db - .select(dispatchRunSelection) - .from(heartbeatRuns) - .where(and( - eq(heartbeatRuns.agentId, agentId), - sql`${heartbeatRuns.status} = 'queued'`, - cutoff ? gte(heartbeatRuns.createdAt, cutoff) : undefined, - criticalLaneCursor - ? sql`(${heartbeatRuns.createdAt}, ${heartbeatRuns.id}) > (${criticalLaneCursor.createdAt}::timestamptz, ${criticalLaneCursor.id}::uuid)` - : undefined, - deferredRunIds?.size - ? notInArray(heartbeatRuns.id, [...deferredRunIds]) - : undefined, - )) - .orderBy(asc(heartbeatRuns.createdAt), asc(heartbeatRuns.id)) - .limit(queuedRunDispatchScanLimit); - if (batch.length === 0) { + const criticalPage = await readQueuedDispatchPage({ + agentId, + cutoff: cutoff ?? null, + cursor: criticalLaneCursor, + limit: queuedRunDispatchScanLimit, + excludeRunIds: deferredRunIds, + }); + const batch = criticalPage.runs; + if (criticalPage.probed === 0) { criticalLaneExhausted = true; break; } - const lastCritical = batch[batch.length - 1]!; - criticalLaneCursor = { - createdAt: lastCritical.dispatchCreatedAtCursor, - id: lastCritical.id, - }; - if (batch.length < queuedRunDispatchScanLimit) criticalLaneExhausted = true; + criticalLaneCursor = criticalPage.cursor; + if (criticalPage.probed < queuedRunDispatchScanLimit) criticalLaneExhausted = true; const batchIssueIds = [...new Set( batch From ada633d8825b70e285870d25091e916cd883207a Mon Sep 17 00:00:00 2001 From: Omar Ramadan Date: Wed, 5 Aug 2026 19:07:02 -0700 Subject: [PATCH 2/3] fix(heartbeat): keep deferred ids out of dispatch probe --- .../src/heartbeat-dispatch-query-plan.test.ts | 50 ++++++++++++++++++- server/src/services/heartbeat.ts | 32 ++++++++---- 2 files changed, 72 insertions(+), 10 deletions(-) diff --git a/packages/db/src/heartbeat-dispatch-query-plan.test.ts b/packages/db/src/heartbeat-dispatch-query-plan.test.ts index cd8451a3843..602314f06ed 100644 --- a/packages/db/src/heartbeat-dispatch-query-plan.test.ts +++ b/packages/db/src/heartbeat-dispatch-query-plan.test.ts @@ -326,6 +326,17 @@ const DISPATCH_HEAD_PROBE_CURSOR = ` LIMIT ${SCAN_LIMIT} `; +function dispatchHeadProbeQuery(cursor: { createdAt: string; id: string } | null = null) { + return ` + SELECT created_at::text AS dispatch_created_at_cursor, id FROM heartbeat_runs + WHERE agent_id = '${AGENT}'::uuid + AND status = 'queued' + ${cursor ? `AND (created_at, id) > ('${cursor.createdAt}'::timestamptz, '${cursor.id}'::uuid)` : ""} + ORDER BY created_at ASC, id ASC + LIMIT ${SCAN_LIMIT} + `; +} + /** * A deep queue whose rows are physically INTERLEAVED with everyone else's. * @@ -655,6 +666,44 @@ describeEmbeddedPostgres("BLO-20396 dispatch query plans", () => { expect(planNodes(probeCursor.root).map((n) => n["Node Type"])).not.toContain("Sort"); expect(rowsInspected(probeCursor.root)).toBeLessThanOrEqual(HEAD_ABSOLUTE_BOUND); + // Deferred emergency-admission refusals are filtered *after* the raw + // cursor-bearing probe in production. If they were pushed into SQL as + // `NOT IN (...)`, PostgreSQL would have to keep walking the ordered index + // until it found SCAN_LIMIT non-deferred rows or proved none remained. + // Simulate several pages of deferred ids: each raw probe remains bounded, + // advances from the unfiltered page, and liveness reaches the next + // non-deferred page. + const deferredRows = await sql.unsafe( + `SELECT created_at::text AS created_at_text, id::text AS id FROM heartbeat_runs + WHERE agent_id = '${AGENT}'::uuid AND status = 'queued' + ORDER BY created_at ASC, id ASC + LIMIT ${SCAN_LIMIT * 3}`, + ) as Array<{ created_at_text: string; id: string }>; + expect(deferredRows).toHaveLength(SCAN_LIMIT * 3); + const deferredIds = new Set(deferredRows.map((row) => row.id)); + let deferredCursor: { createdAt: string; id: string } | null = null; + for (let pass = 0; pass < 3; pass += 1) { + const deferredProbeQuery = dispatchHeadProbeQuery(deferredCursor); + const deferredProbe = await explain(sql, deferredProbeQuery); + record(`depth ${depth}: deferred raw probe pass ${pass + 1}`, deferredProbe); + expect(rowsInspected(deferredProbe.root)).toBeLessThanOrEqual(HEAD_ABSOLUTE_BOUND); + + const rawPage = await sql.unsafe(deferredProbeQuery) as Array<{ + dispatch_created_at_cursor: string; + id: string; + }>; + expect(rawPage).toHaveLength(SCAN_LIMIT); + expect(rawPage.every((row) => deferredIds.has(row.id))).toBe(true); + const last = rawPage[rawPage.length - 1]!; + deferredCursor = { createdAt: last.dispatch_created_at_cursor, id: last.id }; + } + const afterDeferredRows = await sql.unsafe(dispatchHeadProbeQuery(deferredCursor)) as Array<{ + dispatch_created_at_cursor: string; + id: string; + }>; + expect(afterDeferredRows).toHaveLength(SCAN_LIMIT); + expect(afterDeferredRows.every((row) => !deferredIds.has(row.id))).toBe(true); + // Phase 2 hydrates exactly the probed page by primary key, so it is bounded // by SCAN_LIMIT and not by the queue behind it. const pageIds = (await sql.unsafe( @@ -687,4 +736,3 @@ describeEmbeddedPostgres("BLO-20396 dispatch query plans", () => { expect(HEAD_ABSOLUTE_BOUND).toBeLessThan(DISPATCH_HEAD_DEPTHS[0]! / 1.5); }, 900_000); }); - diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 442ba5ec893..bb1d2de88a7 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -18178,10 +18178,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) * to avoid. Keeping the id list as the only SQL predicate pins the pkey plan; * the status filter costs nothing in JS. * - * Returns the probe's own length and last key rather than the fetched rows', - * so callers advance the cursor past everything SCANNED and decide exhaustion - * from the scan. Deriving either from `runs` would treat a page thinned by - * phase 2 as the end of the queue and strand the remaining backlog. + * Returns the raw probe's own length and last key rather than the fetched + * rows', so callers advance the cursor past everything SCANNED and decide + * exhaustion from the scan. Deriving either from `runs` would treat a page + * thinned by phase 2 — including rows filtered by `excludeRunIds` after the + * probe — as the end of the queue and strand the remaining backlog. */ async function readQueuedDispatchPage(input: { agentId: string; @@ -18209,9 +18210,6 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) input.cursor ? sql`(${heartbeatRuns.createdAt}, ${heartbeatRuns.id}) > (${input.cursor.createdAt}::timestamptz, ${input.cursor.id}::uuid)` : undefined, - input.excludeRunIds?.size - ? notInArray(heartbeatRuns.id, [...input.excludeRunIds]) - : undefined, ); // `created_at::text` rather than the Date column: a JS Date truncates @@ -18233,10 +18231,26 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) if (page.length === 0) return { probed: 0, cursor: null, runs: [] }; const lastProbed = page[page.length - 1]!; + // Do not push `excludeRunIds` into phase 1. Emergency-admission refusals can + // accumulate many deferred rows near the head; a SQL `NOT IN (...)` would + // make PostgreSQL keep walking the ordered index until it finds enough + // non-deferred rows (or proves none remain), moving the start-lock work back + // toward queue depth. The raw probe is the bound. Filter after capturing its + // cursor so a page made entirely of deferred rows still makes progress. + const pageToHydrate = input.excludeRunIds?.size + ? page.filter((entry) => !input.excludeRunIds!.has(entry.id)) + : page; + if (pageToHydrate.length === 0) { + return { + probed: page.length, + cursor: { createdAt: lastProbed.createdAt, id: lastProbed.id }, + runs: [], + }; + } const rows = await db .select() .from(heartbeatRuns) - .where(inArray(heartbeatRuns.id, page.map((entry) => entry.id))); + .where(inArray(heartbeatRuns.id, pageToHydrate.map((entry) => entry.id))); // Reorder in JS from the probe's ordering rather than adding an ORDER BY to // phase 2: the keys are already sorted, and sorting there would put a Sort @@ -18248,7 +18262,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return { probed: page.length, cursor: { createdAt: lastProbed.createdAt, id: lastProbed.id }, - runs: page + runs: pageToHydrate .map((entry) => runById.get(entry.id)) .filter((run): run is typeof heartbeatRuns.$inferSelect => run !== undefined), }; From 8cadea2dfc1559536a8f380cacc4017b106b1222 Mon Sep 17 00:00:00 2001 From: Omar Ramadan Date: Thu, 6 Aug 2026 13:00:25 -0700 Subject: [PATCH 3/3] fix(heartbeat): bound deferred recovery lane paging --- .../heartbeat-dispatch-priority-sort.test.ts | 162 ++++++++++++++++++ server/src/services/heartbeat.ts | 20 ++- 2 files changed, 175 insertions(+), 7 deletions(-) diff --git a/server/src/__tests__/heartbeat-dispatch-priority-sort.test.ts b/server/src/__tests__/heartbeat-dispatch-priority-sort.test.ts index ae0678167f8..5595d76af33 100644 --- a/server/src/__tests__/heartbeat-dispatch-priority-sort.test.ts +++ b/server/src/__tests__/heartbeat-dispatch-priority-sort.test.ts @@ -2562,6 +2562,168 @@ describeEmbeddedPostgres("heartbeat dispatch priority sort (BLO-12990)", () => { 180_000, ); + it("advances the recovery lane through fully deferred admission-refusal pages", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const staleSlotRunId = randomUUID(); + const issuePrefix = `D${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + const baseCreatedAt = new Date(Date.now() - 5 * 60 * 1000); + const recoveryRunIds = Array.from({ length: 5 }, () => randomUUID()); + const recoveryIssueIds = Array.from({ length: recoveryRunIds.length }, () => randomUUID()); + let recoveryContinuationSchedules = 0; + let refusalStatusReads = 0; + let releasedStaleSlot = false; + let sawFullyDeferredRecoveryPage = false; + + const boundedHeartbeat = heartbeatService(db, { + penstockGate: allowPenstockGate, + queuedRunDispatchBounds: { scanLimit: 2, maxScanBatches: 1, maxResumePasses: 12 }, + beforeQueuedDispatchRefusalStatusReadForTest: async () => { + refusalStatusReads += 1; + }, + afterQueuedDispatchContinuationScheduledForTest: async (event) => { + if (event.reason !== "resume_recovery_lane" || releasedStaleSlot) return; + recoveryContinuationSchedules += 1; + if (recoveryContinuationSchedules <= refusalStatusReads) return; + sawFullyDeferredRecoveryPage = true; + releasedStaleSlot = true; + await db + .update(externalRuntimeReservations) + .set({ + state: "released", + releasedAt: new Date(), + releaseReason: "test_slot_released_after_deferred_page", + updatedAt: new Date(), + }) + .where(eq(externalRuntimeReservations.runId, staleSlotRunId)); + }, + }); + + await db.insert(companies).values({ + id: companyId, + name: "RecoveryLaneDeferredPagesCo", + issuePrefix, + requireBoardApprovalForNewAgents: false, + defaultResponsibleUserId: "responsible-user", + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "RecoveryLaneDeferredPagesAgent", + role: "engineer", + status: "idle", + adapterType: "claude_k8s", + adapterConfig: {}, + runtimeConfig: { heartbeat: { enabled: true, wakeOnDemand: true, maxConcurrentRuns: 1 } }, + permissions: {}, + }); + await db.insert(issues).values( + recoveryIssueIds.map((issueId, index) => ({ + id: issueId, + companyId, + title: `Recovery lane candidate ${index + 1}`, + status: "todo" as const, + priority: "medium" as const, + assigneeAgentId: agentId, + issueNumber: index + 1, + identifier: `${issuePrefix}-${index + 1}`, + })), + ); + + await db.insert(heartbeatRuns).values({ + id: staleSlotRunId, + companyId, + agentId, + invocationSource: "assignment", + triggerDetail: "system", + status: "succeeded", + contextSnapshot: {}, + startedAt: new Date(baseCreatedAt.getTime() - 60_000), + finishedAt: new Date(baseCreatedAt.getTime() - 30_000), + createdAt: new Date(baseCreatedAt.getTime() - 60_000), + updatedAt: new Date(baseCreatedAt.getTime() - 30_000), + }); + await db.insert(externalRuntimeReservations).values({ + companyId, + agentId, + runId: staleSlotRunId, + slotId: 0, + state: "launched", + jobName: `paperclip-agent-${staleSlotRunId}`, + jobUid: randomUUID(), + isolationMode: "run", + isolationKey: `run:${staleSlotRunId}`, + isolationBoundAt: new Date(baseCreatedAt.getTime() - 60_000), + reservedAt: new Date(baseCreatedAt.getTime() - 60_000), + launchingAt: new Date(baseCreatedAt.getTime() - 60_000), + launchedAt: new Date(baseCreatedAt.getTime() - 30_000), + createdAt: new Date(baseCreatedAt.getTime() - 60_000), + updatedAt: new Date(baseCreatedAt.getTime() - 30_000), + }); + + for (const [index, runId] of recoveryRunIds.entries()) { + const wakeId = randomUUID(); + const recoveryActionId = randomUUID(); + const issueId = recoveryIssueIds[index]!; + const createdAt = new Date(baseCreatedAt.getTime() + index); + await db.insert(agentWakeupRequests).values({ + id: wakeId, + companyId, + agentId, + source: "assignment", + triggerDetail: "system", + reason: "source_scoped_recovery_action", + payload: { issueId, recoveryActionId }, + status: "queued", + runId, + requestedAt: createdAt, + updatedAt: createdAt, + }); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + invocationSource: "assignment", + triggerDetail: "system", + status: "queued", + wakeupRequestId: wakeId, + contextSnapshot: { + issueId, + wakeReason: "source_scoped_recovery_action", + source: "issue_recovery_action", + recoveryActionId, + }, + createdAt, + updatedAt: createdAt, + }); + } + + const dispatchedRunIds: string[] = []; + mockAdapterExecute.mockImplementation(async (args: { runId: string }) => { + dispatchedRunIds.push(args.runId); + return { + exitCode: 0, + signal: null as string | null, + timedOut: false, + errorMessage: null as string | null, + resultJson: { exitCode: 0 }, + provider: "test", + model: "test-model", + }; + }); + + await boundedHeartbeat.resumeQueuedRuns(); + await waitForRunToSettle(boundedHeartbeat, recoveryRunIds[2]!, 60_000); + + expect(sawFullyDeferredRecoveryPage).toBe(true); + expect(releasedStaleSlot).toBe(true); + expect(refusalStatusReads).toBe(2); + expect(dispatchedRunIds[0]).toBe(recoveryRunIds[2]); + expect((await boundedHeartbeat.getRun(recoveryRunIds[0]!))?.status).toBe("queued"); + expect((await boundedHeartbeat.getRun(recoveryRunIds[1]!))?.status).toBe("queued"); + await boundedHeartbeat.drainInFlightExecutions(60_000); + }, 180_000); + it("retries a lone emergency run after atomic admission refusal", async () => { const companyId = randomUUID(); const agentId = randomUUID(); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index bb1d2de88a7..5c4f1745e12 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -18797,7 +18797,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) while (recoveryLaneBatches < queuedRunDispatchMaxScanBatches) { recoveryLaneBatches += 1; const batchStartCursor = recoveryLaneCursor; - const batch = await db + const rawBatch = await db .select(dispatchRunSelection) .from(heartbeatRuns) .where(and( @@ -18809,22 +18809,28 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) recoveryLaneCursor ? sql`(${heartbeatRuns.createdAt}, ${heartbeatRuns.id}) > (${recoveryLaneCursor.createdAt}::timestamptz, ${recoveryLaneCursor.id}::uuid)` : undefined, - deferredRunIds?.size - ? notInArray(heartbeatRuns.id, [...deferredRunIds]) - : undefined, )) .orderBy(asc(heartbeatRuns.createdAt), asc(heartbeatRuns.id)) .limit(queuedRunDispatchScanLimit); - if (batch.length === 0) { + if (rawBatch.length === 0) { recoveryLaneExhausted = true; break; } - const lastRun = batch[batch.length - 1]!; + const lastRun = rawBatch[rawBatch.length - 1]!; recoveryLaneCursor = { createdAt: lastRun.dispatchCreatedAtCursor, id: lastRun.id, }; - if (batch.length < queuedRunDispatchScanLimit) recoveryLaneExhausted = true; + if (rawBatch.length < queuedRunDispatchScanLimit) recoveryLaneExhausted = true; + + // Match the main/critical dispatch scans: admission-refused emergency + // rows are filtered after the raw cursor-bearing probe. Keeping them out + // of SQL with NOT IN makes PostgreSQL walk the deferred prefix under the + // per-agent start lock; capturing the raw cursor first lets even a fully + // deferred page advance in bounded work. + const batch = deferredRunIds?.size + ? rawBatch.filter((run) => !deferredRunIds.has(run.id)) + : rawBatch; const issueIdsToLoad = new Set(); for (const run of batch) {