mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
perf(db): index retry-exhaustion receipts for the attention scan (#20386)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The attention feed shows runs that failed after automatic retries stopped > - To find these runs, the feed reads every `heartbeat_run_events` row of the company and filters the rows by a message prefix > - Run output makes this table very large, so each feed request does a large scan > - On a hosted test database, this one statement used the most total database time: 923 calls at a mean of 5 s (4,709 s in total) > - Request authentication waits behind this background work, so all users feel the load > - This pull request adds a small partial index that holds only the retry-exhaustion receipts > - The benefit is that the same query reads a few index entries instead of all company events ## Linked Issues or Issue Description **What existing behavior does this improve?** The attention feed query that finds retry-exhausted runs (`listAttentionExhaustedRuns`). **Subsystem affected** Server attention feed and the `heartbeat_run_events` table schema. **Current behavior** The query reads all company rows in `heartbeat_run_events`, filters them by `event_type = 'lifecycle'` and the `Bounded retry exhausted` message prefix, and then joins the few matches to failed or timed-out runs. The cost grows with all run output. It does not grow with the number of exhausted runs. **Proposed behavior** A partial index holds only the matching receipts. The query plan reads the index and joins a few rows. The query text and the result do not change. **Reason and benefit** The scan was the most expensive statement by total time on a loaded hosted database. A local reproduction shows a 54x faster query and about 116x fewer buffer reads. **Breaking changes** None. The API, the feed rows, and their order do not change. ## What Changed - Add the partial index `heartbeat_run_events_retry_exhausted_idx` on `(company_id, run_id, id DESC)`. The index covers only rows where `event_type = 'lifecycle'` and `message like 'Bounded retry exhausted%'`. - Add migration `0454`. It uses `CREATE INDEX CONCURRENTLY IF NOT EXISTS`. The migration runner runs it as a top-level statement, the same as migrations 0432, 0436 and 0450. - Add a regression test. It pins the row order (newest run first, then the newest receipt on a tie) and the latest receipt per run. It also checks that the query plan can use the new index. If the query predicate changes so that the index no longer applies, the test fails. - Add `scripts/benchmark-exhausted-run-attention.ts`. It measures the same query with and without the index, and the time to build the index. - Document the index in `doc/DATABASE.md`. ## Verification Cause and thesis. The scan cost comes from reading all company events, not from the few exhausted runs. Thesis: an index that holds only the receipts makes the query cost depend on the number of receipts, and keeps the result identical. The benchmark below confirms the thesis. Rejected alternative without a migration. I first tried a LATERAL lookup that starts from the failed and timed-out runs. On the same fixture it was about 2x slower (median 379 ms against 194 ms). It reads the full event history of each failed run with random I/O, and that costs more than one sequential scan. Local SQL benchmark (the SQL that the service generates; same SQL with and without the index): ```sh pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-exhausted-run-attention.ts --runs=30000 --events-per-run=100 --samples=9 ``` Fixture (synthetic): one company with 30,000 runs (3,819 failed or timed out) and 3.0 M events, plus a decoy company with 0.5 M events. The table is 1.4 GB. About 5% of events are `lifecycle` events. Events of 20 concurrent runs are interleaved. 1 in 23 runs has a receipt, some runs have two receipts, and output decoys contain the prefix. Two agents are terminated. Embedded PostgreSQL 18.1 on an Apple laptop. The hosted database is Neon with different compute, so the absolute times are local SQL measurements only. The script runs two cycles (without index, then with index; then with index, then without index) to expose cache bias. Each measurement has 2 warmups and 9 samples. | Query | Without index | With index | Samples and spread | Output check | | --- | --- | --- | --- | --- | | `listAttentionExhaustedRuns` | median 157 ms (96–1,271 ms) | median 3.3 ms (2.2–5.7 ms) | 18 each | 133 rows, positional rows equal in all 4 measurements | | Buffers (EXPLAIN ANALYZE) | 127,743 blocks | 5,580 blocks | | plan uses the new index (asserted) | Speedup: 48x (median), on head `ae6037b2a0` (the same SQL and index as the final head). The machine was under heavy load from other work, so the spread without the index is wide. An earlier run on the first head gave 248 ms against 4.6 ms (54x). Index build (`CREATE INDEX CONCURRENTLY` from the migration file): 1.7 s and 2.5 s for the 1.4 GB table. Index size: 152 KiB. Review fix: the query now writes `event_type = 'lifecycle'` as a SQL literal, not a bound parameter. With prepared statements, PostgreSQL can reuse a generic plan. A generic plan cannot prove that a bound parameter matches the index predicate, so it would not use the index. Tests: - `cd server && npx vitest run src/__tests__/attention-service.test.ts`: 44 passed. The new test checks a fresh plan and a forced generic plan (`plan_cache_mode = force_generic_plan` with `PREPARE`). It failed with the bound `event_type` parameter and with a changed message prefix (confirmed red), and it passes with the fix. Locally, the file's `afterAll` database cleanup hook timed out on a heavily loaded machine (load average about 28). The timeout also occurs when the new test is excluded. - `pnpm --filter @paperclipai/db check:migrations`: passed. - `pnpm --filter @paperclipai/db typecheck` and `pnpm --filter @paperclipai/server typecheck`: passed. - `pnpm verify`: see the CI results for this head. ## Risks - Migration time. A concurrent index build reads the full `heartbeat_run_events` table and does not block writes. Locally, the build took 1.2–2.5 s for 1.4 GB. The time grows linearly with the table size. I did not time the build on a production-size clone. On a database with a 5 s full company scan, I estimate the build at tens of seconds, which is in the 1–2 minute budget. This estimate is not proven. - If the build stops, PostgreSQL can leave an invalid index. The migration runner repairs invalid concurrent indexes when it runs again, and `IF NOT EXISTS` makes a rerun safe. - The index predicate must stay textually equal to the query predicate. If someone changes the message prefix or binds a predicate value as a parameter, the planner cannot use the index. The new test catches both cases, for fresh and generic plans. - Each new receipt writes one more small index entry. Receipts are rare, so the write cost is very small. - The hosted improvement is not verified until deployment. After deployment, compare a new query-statistics window for this statement. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use (shell, file edits, local PostgreSQL). Standard context window and reasoning mode. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
c57cd1bdb9
commit
c985541a57
9 files changed
+488
-4
No files matched your search
@@ -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
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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%';
|
||||
+218
-3
@@ -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": [
|
||||
+7
@@ -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
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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(),
|
||||
}),
|
||||
);
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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<Record<string, any>>)[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<ReturnType<typeof planSummary>>) => ({ ...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); }
|
||||
}
|
||||
@@ -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({
|
||||
|
||||
@@ -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))
|
||||
|
||||
Reference in new issue
Block a user