diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 2ab0bbb301c..c3cf5eddf7c 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -352,6 +352,13 @@ by company and normalized session. Migration 0432 builds both indexes concurrently. A poll therefore reads a few index entries, not a run's full event history. +The attention feed finds retry-exhausted runs through the partial index +`heartbeat_run_events_retry_exhausted_idx`. The index covers only `lifecycle` +events whose message starts with `Bounded retry exhausted`, keyed by company, +run, and newest event first. The query predicate must stay textually equal to +the index predicate, or the scan reads every company event again. Migration +0454 builds the index concurrently. + A subtree pause release finds the runs that a dispatch gate refused for that pause through the partial index `heartbeat_runs_tree_hold_suppression_idx`. The index covers only cancelled runs with a `treeHoldSuppression` marker, keyed by company and diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index d6062cd9818..2633c213f01 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -616,6 +616,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0447_legacy_controller_process.sql", "0450_petite_frightful_four.sql", "0453_curious_ben_parker.sql", + "0454_lumpy_ronan.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0454_lumpy_ronan.sql b/packages/db/src/migrations/0454_lumpy_ronan.sql new file mode 100644 index 00000000000..53377f78787 --- /dev/null +++ b/packages/db/src/migrations/0454_lumpy_ronan.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "heartbeat_run_events_retry_exhausted_idx" ON "heartbeat_run_events" USING btree ("company_id","run_id","id" DESC NULLS LAST) WHERE "heartbeat_run_events"."event_type" = 'lifecycle' and "heartbeat_run_events"."message" like 'Bounded retry exhausted%'; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0449_snapshot.json b/packages/db/src/migrations/meta/0454_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0449_snapshot.json rename to packages/db/src/migrations/meta/0454_snapshot.json index 115776e68c6..4e0bb003c30 100644 --- a/packages/db/src/migrations/meta/0449_snapshot.json +++ b/packages/db/src/migrations/meta/0454_snapshot.json @@ -1,6 +1,6 @@ { - "id": "4bd9d289-9cb8-4262-9dd7-04ce2b6020c4", - "prevId": "360fef2e-1ac3-4ae8-92a2-245e544b61da", + "id": "f98af245-4a2e-4c3d-b89e-6ed3e81a074c", + "prevId": "d54bfeb4-dedb-4f8b-9195-a4d4aa67d71f", "version": "7", "dialect": "postgresql", "tables": { @@ -4102,6 +4102,28 @@ "method": "btree", "with": {} }, + "agent_wakeup_requests_issue_blocker_stalled_uq": { + "name": "agent_wakeup_requests_issue_blocker_stalled_uq", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "idempotency_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "\"agent_wakeup_requests\".\"idempotency_key\" LIKE 'issue_blocker_stalled:%' AND \"agent_wakeup_requests\".\"status\" IN ('queued', 'deferred_issue_execution', 'claimed', 'coalesced', 'completed')", + "concurrently": true, + "method": "btree", + "with": {} + }, "agent_wakeup_requests_company_payload_issue_idx": { "name": "agent_wakeup_requests_company_payload_issue_idx", "columns": [ @@ -7860,6 +7882,150 @@ }, "isRLSEnabled": false }, + "public.budget_resume_claims": { + "name": "budget_resume_claims", + "schema": "", + "columns": { + "run_id": { + "name": "run_id", + "type": "uuid", + "primaryKey": true, + "notNull": true + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "incident_id": { + "name": "incident_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "actor_user_id": { + "name": "actor_user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "actor_source": { + "name": "actor_source", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "state": { + "name": "state", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'pending'" + }, + "claimed_at": { + "name": "claimed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "settled_at": { + "name": "settled_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "reason": { + "name": "reason", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "request_id": { + "name": "request_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "budget_resume_claims_pending_idx": { + "name": "budget_resume_claims_pending_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"budget_resume_claims\".\"state\" = 'pending'", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "budget_resume_claims_run_id_heartbeat_runs_id_fk": { + "name": "budget_resume_claims_run_id_heartbeat_runs_id_fk", + "tableFrom": "budget_resume_claims", + "tableTo": "heartbeat_runs", + "columnsFrom": [ + "run_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "budget_resume_claims_company_id_companies_id_fk": { + "name": "budget_resume_claims_company_id_companies_id_fk", + "tableFrom": "budget_resume_claims", + "tableTo": "companies", + "columnsFrom": [ + "company_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "budget_resume_claims_incident_id_budget_incidents_id_fk": { + "name": "budget_resume_claims_incident_id_budget_incidents_id_fk", + "tableFrom": "budget_resume_claims", + "tableTo": "budget_incidents", + "columnsFrom": [ + "incident_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.built_in_managed_resources": { "name": "built_in_managed_resources", "schema": "", @@ -17233,7 +17399,7 @@ "columnsTo": [ "id" ], - "onDelete": "set null", + "onDelete": "cascade", "onUpdate": "no action" }, "company_secret_proposals_secret_proposal_id_company_secret_proposals_id_fk": { @@ -34386,6 +34552,34 @@ "concurrently": false, "method": "btree", "with": {} + }, + "heartbeat_run_events_retry_exhausted_idx": { + "name": "heartbeat_run_events_retry_exhausted_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_run_events\".\"event_type\" = 'lifecycle' and \"heartbeat_run_events\".\"message\" like 'Bounded retry exhausted%'", + "concurrently": true, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -35396,6 +35590,27 @@ "method": "btree", "with": {} }, + "heartbeat_runs_company_native_workspace_idx": { + "name": "heartbeat_runs_company_native_workspace_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"runner_profile_json\"->'nativeWorkspaceSync'->>'workspaceId')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": true, + "method": "btree", + "with": {} + }, "heartbeat_runs_company_issue_created_idx": { "name": "heartbeat_runs_company_issue_created_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 2b28994fd41..29424d35338 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -3165,6 +3165,13 @@ "when": 1794260223713, "tag": "0453_curious_ben_parker", "breakpoints": true + }, + { + "idx": 454, + "version": "7", + "when": 1794266236755, + "tag": "0454_lumpy_ronan", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/heartbeat_run_events.ts b/packages/db/src/schema/heartbeat_run_events.ts index 9b17e9bf47d..d260bf53518 100644 --- a/packages/db/src/schema/heartbeat_run_events.ts +++ b/packages/db/src/schema/heartbeat_run_events.ts @@ -54,5 +54,12 @@ export const heartbeatRunEvents = pgTable( compactionSessionIdx: index("heartbeat_run_events_compaction_session_idx") .on(table.companyId, sql`(${table.payload} -> 'prpEvent' ->> 'normalizedSessionId')`) .where(sql`${table.eventType} = 'context.compacted'`), + // Retry-exhaustion receipts for the attention feed: a few rows per company. + // The predicate must stay textually equal to the query in + // server/src/services/attention-exhausted-runs.ts so the planner can use it. + retryExhaustedIdx: index("heartbeat_run_events_retry_exhausted_idx") + .on(table.companyId, table.runId, table.id.desc()) + .where(sql`${table.eventType} = 'lifecycle' and ${table.message} like 'Bounded retry exhausted%'`) + .concurrently(), }), ); diff --git a/scripts/benchmark-exhausted-run-attention.ts b/scripts/benchmark-exhausted-run-attention.ts new file mode 100644 index 00000000000..82d83520942 --- /dev/null +++ b/scripts/benchmark-exhausted-run-attention.ts @@ -0,0 +1,192 @@ +// Measures the attention scan for retry-exhausted runs with and without the +// partial index from migration 0454. This is an index-only change: both sides +// run the same SQL that the service generates. +// Usage (from the repository root): +// pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-exhausted-run-attention.ts \ +// [--runs=30000] [--events-per-run=100] [--samples=9] [--plans] +// The script starts a disposable embedded PostgreSQL and loads a synthetic +// company plus a decoy company. It times the query without the index, times the +// migration's CREATE INDEX CONCURRENTLY, and times the query again. It repeats +// the cycle in reverse order to expose cache bias, asserts identical positional +// rows, and prints medians, spread, build times, and plan summaries as JSON. +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { promises as fs } from "node:fs"; +import { fileURLToPath } from "node:url"; +import { createDb, startEmbeddedPostgresTestDatabase, type Db } from "../packages/db/src/index.js"; +import { listAttentionExhaustedRuns } from "../server/src/services/attention-exhausted-runs.js"; + +type Query = { sql: string; params: unknown[] }; +const arg = (name: string, fallback: string) => + process.argv.find((value) => value.startsWith(`--${name}=`))?.slice(name.length + 3) ?? fallback; +const runCount = Number(arg("runs", "30000")); +const eventsPerRun = Number(arg("events-per-run", "100")); +const sampleCount = Number(arg("samples", "9")); +const revision = execFileSync("git", ["rev-parse", "HEAD"], { encoding: "utf8" }).trim(); +const migrationPath = "packages/db/src/migrations/0454_lumpy_ronan.sql"; +const indexName = "heartbeat_run_events_retry_exhausted_idx"; +const keepAlive = setInterval(() => {}, 1000); +let database: Awaited> | undefined; +let db: Db | undefined; + +async function seedCompany(client: Db["$client"], companyId: string, prefix: string, runs: number) { + await client.unsafe(`INSERT INTO companies (id, name, issue_prefix) VALUES ($1, $2, $3)`, [companyId, `Synthetic ${prefix}`, prefix]); + // 40 agents; two are terminated so their exhausted runs must stay hidden. + await client.unsafe(`INSERT INTO agents (id, company_id, name, adapter_type, status) + SELECT gen_random_uuid(), $1::uuid, 'Agent ' || a, 'codex_local', CASE WHEN a <= 2 THEN 'terminated' ELSE 'idle' END + FROM generate_series(1, 40) a`, [companyId]); + await client.unsafe(`WITH numbered AS ( + SELECT n, gen_random_uuid() AS id, + (SELECT id FROM agents WHERE company_id = $1::uuid ORDER BY name OFFSET (n % 40) LIMIT 1) AS agent_id + FROM generate_series(1, $2::int) n) + INSERT INTO bench_runs SELECT $1::uuid, n, id, agent_id FROM numbered`, [companyId, runs]); + // Status mix: about 10% failed, 3% timed out, the rest succeeded or cancelled. + // Pairs of runs share created_at so the event-id tie-breaker is exercised. + await client.unsafe(`INSERT INTO heartbeat_runs (id, company_id, agent_id, status, error, error_code, created_at, updated_at, finished_at, context_snapshot) + SELECT r.id, r.company_id, r.agent_id, + CASE WHEN r.n % 10 = 0 THEN 'failed' WHEN r.n % 33 = 0 THEN 'timed_out' WHEN r.n % 50 = 0 THEN 'cancelled' ELSE 'succeeded' END, + CASE WHEN r.n % 10 = 0 THEN 'adapter exited with code 1' END, + CASE WHEN r.n % 10 = 0 THEN 'adapter_failed' END, + timestamptz '2026-10-01T00:00:00Z' + (r.n / 2) * interval '1 second', + timestamptz '2026-10-01T00:00:00Z' + (r.n / 2) * interval '1 second' + interval '5 minutes', + timestamptz '2026-10-01T00:00:00Z' + (r.n / 2) * interval '1 second' + interval '5 minutes', + CASE r.n % 4 + WHEN 0 THEN jsonb_build_object('issueId', 'issue-' || r.n, 'prompt', repeat(md5(r.n::text), 64)) + WHEN 1 THEN jsonb_build_object('taskId', 'task-' || r.n) + WHEN 2 THEN jsonb_build_object('issueId', null, 'taskId', 'task-' || r.n) + ELSE '{}'::jsonb END + FROM bench_runs r WHERE r.company_id = $1::uuid`, [companyId]); + // Run output: interleave events from groups of 20 concurrent runs, as live + // heartbeats do. About 5% are lifecycle events. + await client.unsafe(`INSERT INTO heartbeat_run_events (company_id, run_id, agent_id, seq, event_type, stream, message, payload, created_at) + SELECT r.company_id, r.id, r.agent_id, s, + CASE WHEN s % 20 = 1 THEN 'lifecycle' ELSE 'item.completed' END, + CASE WHEN s % 20 = 1 THEN 'system' ELSE 'stdout' END, + CASE WHEN s % 20 = 1 THEN 'Run step ' || s ELSE md5(r.n::text || ':' || s) || md5(s::text) END, + jsonb_build_object('chunk', md5(r.id::text || s) || md5(s::text || r.n)), + timestamptz '2026-10-01T00:00:00Z' + (r.n / 2) * interval '1 second' + s * interval '1 millisecond' + FROM bench_runs r CROSS JOIN generate_series(1, $2::int) s + WHERE r.company_id = $1::uuid + ORDER BY r.n / 20, s, r.n`, [companyId, eventsPerRun]); + // Retry-exhaustion receipts on 1 in 23 runs (most of those are not failed), + // a second receipt on some, plus an output decoy that must not match. + await client.unsafe(`INSERT INTO heartbeat_run_events (company_id, run_id, agent_id, seq, event_type, stream, message, created_at) + SELECT r.company_id, r.id, r.agent_id, $2::int + x.k, x.event_type, 'system', x.message || ' (' || r.n || ')', now() + FROM bench_runs r + CROSS JOIN LATERAL (VALUES + (1, 'lifecycle', 'Bounded retry exhausted after 3 scheduled attempts; no further automatic retry will be queued'), + (2, 'stdout', 'Bounded retry exhausted quoted in output'), + (3, 'lifecycle', 'Bounded retry exhausted after 4 scheduled attempts; second receipt') + ) AS x(k, event_type, message) + WHERE r.company_id = $1::uuid AND r.n % 23 = 0 AND (x.k = 1 OR r.n % 46 = 0)`, [companyId, eventsPerRun]); +} + +async function timeOnce(query: Query) { + const start = performance.now(); + const rows = await db!.$client.unsafe(query.sql, query.params as never[]).values(); + return { ms: performance.now() - start, rows }; +} + +async function planSummary(query: Query) { + const [plan] = await db!.$client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${query.sql}`, query.params as never[]); + const root = (plan["QUERY PLAN"] as Array>)[0]; + return { + executionMs: root["Execution Time"], + sharedHitBlocks: root.Plan["Shared Hit Blocks"], + sharedReadBlocks: root.Plan["Shared Read Blocks"], + plan: root.Plan, + }; +} + +const stats = (samples: number[]) => { + const sorted = [...samples].sort((a, b) => a - b); + return { + medianMs: sorted[Math.floor(sorted.length / 2)], + minMs: sorted[0], + maxMs: sorted[sorted.length - 1], + samplesMs: samples.map((value) => Math.round(value * 10) / 10), + }; +}; + +try { + database = await startEmbeddedPostgresTestDatabase("exhausted-run-attention-benchmark-"); + db = createDb(database.connectionString); + const client = db.$client; + const createIndexSql = (await fs.readFile(fileURLToPath(new URL(`../${migrationPath}`, import.meta.url)), "utf8")).trim(); + const dropIndex = () => client.unsafe(`DROP INDEX IF EXISTS "${indexName}"`); + const buildIndex = async () => { + const start = performance.now(); + await client.unsafe(createIndexSql); + const ms = performance.now() - start; + await client.unsafe("ANALYZE heartbeat_run_events"); + return ms; + }; + + const companyId = randomUUID(); + const decoyCompanyId = randomUUID(); + const seedStart = performance.now(); + await dropIndex(); + // A regular table: pooled connections do not share temporary tables. + await client.unsafe(`CREATE TABLE bench_runs (company_id uuid, n int, id uuid, agent_id uuid)`); + await seedCompany(client, companyId, "BENCH", runCount); + await seedCompany(client, decoyCompanyId, "DECOY", Math.max(1, Math.floor(runCount / 6))); + await client.unsafe(`DROP TABLE bench_runs`); + await client.unsafe("ANALYZE companies, agents, heartbeat_runs, heartbeat_run_events"); + const seedMs = performance.now() - seedStart; + const [{ events, totalEvents, failed, tableBytes }] = await client.unsafe(`SELECT + (SELECT count(*) FROM heartbeat_run_events WHERE company_id = $1::uuid)::int AS events, + (SELECT count(*) FROM heartbeat_run_events)::int AS "totalEvents", + pg_total_relation_size('heartbeat_run_events')::bigint AS "tableBytes", + (SELECT count(*) FROM heartbeat_runs WHERE company_id = $1::uuid AND status IN ('failed', 'timed_out'))::int AS failed`, [companyId]); + const [{ server_version: version }] = await client.unsafe("SHOW server_version"); + + const query = listAttentionExhaustedRuns(db, companyId).toSQL() as Query; + const measure = async () => { + for (let warm = 0; warm < 2; warm++) await timeOnce(query); + const samples: number[] = []; + let rows: unknown[] = []; + for (let sample = 0; sample < sampleCount; sample++) { + const result = await timeOnce(query); + samples.push(result.ms); + rows = result.rows; + } + return { samples, rows, plan: await planSummary(query) }; + }; + + // Cycle 1: without the index, then with it. Cycle 2: with it, then without. + const without1 = await measure(); + const build1Ms = await buildIndex(); + const with1 = await measure(); + const with2 = await measure(); + await dropIndex(); + const without2 = await measure(); + const build2Ms = await buildIndex(); + const [{ indexBytes }] = await client.unsafe(`SELECT pg_relation_size('${indexName}')::bigint AS "indexBytes"`); + + assert.ok(without1.rows.length > 0, "fixture must return exhausted runs"); + for (const result of [with1, with2, without2]) assert.deepEqual(result.rows, without1.rows); + assert.ok(JSON.stringify(with1.plan.plan).includes(indexName), "query must use the partial index"); + const before = stats([...without1.samples, ...without2.samples]); + const after = stats([...with1.samples, ...with2.samples]); + const verbose = process.argv.includes("--plans"); + const brief = (plan: Awaited>) => ({ ...plan, plan: verbose ? plan.plan : undefined }); + console.log(JSON.stringify({ + revision, + postgresVersion: version, + fixture: { + runs: runCount, eventsPerRun, companyEvents: events, totalEvents, tableBytes: Number(tableBytes), + failedOrTimedOutRuns: failed, seedMs: Math.round(seedMs), + }, + equalRows: without1.rows.length, + withoutIndex: { ...before, explain: brief(without1.plan) }, + withIndex: { ...after, explain: brief(with1.plan) }, + speedup: before.medianMs / after.medianMs, + indexBuild: { firstMs: Math.round(build1Ms), secondMs: Math.round(build2Ms), indexBytes: Number(indexBytes) }, + }, null, 2)); +} finally { + try { + await db?.$client.end({ timeout: 0 }); + await database?.cleanup(); + } finally { clearInterval(keepAlive); } +} diff --git a/server/src/__tests__/attention-service.test.ts b/server/src/__tests__/attention-service.test.ts index fdf4044fefb..2ab01f77b2d 100644 --- a/server/src/__tests__/attention-service.test.ts +++ b/server/src/__tests__/attention-service.test.ts @@ -1040,6 +1040,57 @@ describeEmbeddedPostgres("attention service", () => { expect(failures.find((item) => item.subject.id === timedOutId)?.relatedIssue?.id).toBe(taskId); }); + it("orders exhausted runs and reads receipts through the partial index", async () => { + const { companyId, workerId } = await seedCompany("IDX"); + const [olderId, tiedLowId, tiedHighId] = Array.from({ length: 3 }, () => randomUUID()); + const older = new Date("2026-07-09T11:00:00.000Z"); + const tied = new Date("2026-07-09T12:00:00.000Z"); + await db.insert(heartbeatRuns).values([ + { id: olderId, companyId, agentId: workerId, status: "failed", createdAt: older, updatedAt: older }, + { id: tiedLowId, companyId, agentId: workerId, status: "timed_out", createdAt: tied, updatedAt: tied }, + { id: tiedHighId, companyId, agentId: workerId, status: "failed", createdAt: tied, updatedAt: tied }, + ]); + // Insert order sets event ids: a tie on created_at falls back to the newest receipt. + const receipts = [ + [tiedHighId, "Bounded retry exhausted first"], + [olderId, "Bounded retry exhausted older"], + [tiedLowId, "Bounded retry exhausted tied low"], + [tiedHighId, "Bounded retry exhausted tied high"], + ] as const; + for (const [index, [runId, message]] of receipts.entries()) { + await db.insert(heartbeatRunEvents).values({ + companyId, agentId: workerId, runId, seq: index + 1, eventType: "lifecycle", message, + }); + } + + const rows = await listAttentionExhaustedRuns(db, companyId); + expect(rows.map((row) => [row.id, row.exhaustionMessage])).toEqual([ + [tiedHighId, "Bounded retry exhausted tied high"], + [tiedLowId, "Bounded retry exhausted tied low"], + [olderId, "Bounded retry exhausted older"], + ]); + + // The partial index predicate must stay implied by the query predicate; + // otherwise the scan silently falls back to reading all company events. + // Check a forced generic plan too: prepared statements reuse it, and only + // literal predicates (not bound parameters) let it use a partial index. + const query = listAttentionExhaustedRuns(db, companyId).toSQL(); + const literal = (value: unknown) => `'${String(value).replaceAll("'", "''")}'`; + const plans = await db.$client.begin(async (tx) => { + await tx.unsafe("SET LOCAL enable_seqscan = off"); + const fresh = await tx.unsafe(`EXPLAIN (FORMAT JSON) ${query.sql}`, query.params as never[]); + await tx.unsafe("SET LOCAL plan_cache_mode = force_generic_plan"); + await tx.unsafe(`PREPARE attention_exhausted_runs AS ${query.sql}`); + const generic = await tx.unsafe( + `EXPLAIN (FORMAT JSON) EXECUTE attention_exhausted_runs(${query.params.map(literal).join(", ")})`, + ); + await tx.unsafe("DEALLOCATE attention_exhausted_runs"); + return { fresh, generic }; + }); + expect(JSON.stringify(plans.fresh)).toContain("heartbeat_run_events_retry_exhausted_idx"); + expect(JSON.stringify(plans.generic)).toContain("heartbeat_run_events_retry_exhausted_idx"); + }); + it("suppresses failed-run attention after a newer run for the same issue", async () => { const { companyId, workerId } = await seedCompany("ATN"); const issueId = await insertIssue({ diff --git a/server/src/services/attention-exhausted-runs.ts b/server/src/services/attention-exhausted-runs.ts index 29f3978dd22..fee63d1a25a 100644 --- a/server/src/services/attention-exhausted-runs.ts +++ b/server/src/services/attention-exhausted-runs.ts @@ -11,9 +11,12 @@ export function listAttentionExhaustedRuns(db: Db, companyId: string) { message: heartbeatRunEvents.message, }) .from(heartbeatRunEvents) + // Keep both receipt filters as SQL literals. They must match the partial + // index heartbeat_run_events_retry_exhausted_idx; a bound parameter would + // stop a generic prepared plan from using it. .where(and( eq(heartbeatRunEvents.companyId, companyId), - eq(heartbeatRunEvents.eventType, "lifecycle"), + sql`${heartbeatRunEvents.eventType} = 'lifecycle'`, sql`${heartbeatRunEvents.message} like 'Bounded retry exhausted%'`, )) .orderBy(asc(heartbeatRunEvents.runId), desc(heartbeatRunEvents.id))