diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index 5ddaad0ab08..160e8e23ef8 100644 --- a/tests/restart-chaos/README.md +++ b/tests/restart-chaos/README.md @@ -549,8 +549,10 @@ runner_event_commit_stalled ... last cause: Error`). events and many usage reports. L1 is opt-in (not a default workload). **`--db-pressure `** (1–8) starts harness-owned writers against the wave's own database. -Each one runs `update heartbeat_runs set updated_at = updated_at` on the wave company's `running` -rows back to back. No value changes, but every statement writes a new row version, which is what a +Each one runs `update heartbeat_runs set updated_at = updated_at` on each of the wave company's +`running` rows, one row per statement, back to back. One row per statement matches a run-log +append; a statement over every running row locked several rows at once and deadlocked with +product transactions (40P01), which a real stream never does. No value changes, but every statement writes a new row version, which is what a fast stream does to a run row. Locally the race window between a snapshot and its row lock is well under a millisecond, so without pressure the race almost never fires; the writers stand in for a hosted database whose round trip widens that window. The writers stop before the R6 sweep; @@ -570,10 +572,22 @@ At rest the harness reads each L1 task's runs from the wave database, and the T- `contention-db.test.ts` checks the knob itself on real PostgreSQL: with four writers, the pre-#20385 repeatable-read receipt write exhausted its four retries with 40001 in 8 of 8 attempts -locally, and the current READ COMMITTED writer (`persistUsageReceipt`) committed all 8. A live L1 -wave needs the runner stall (20 failed deliveries over at least 45 s) to show -`native_event_replay_conflict`; run it against the parent of the #20385 merge (red) and against -master (green) with the command above. +locally, and the current READ COMMITTED writer (`persistUsageReceipt`) committed all 8. + +Live waves on 2026-11-10 (command above, codex and claude, four tasks each): + +| Wave | Build | Receipt write 40001 in the server log | Tasks passed | Failures | +| --- | --- | --- | --- | --- | +| `w-20261110T040531-s20385` | #20385 parent `bf818280f2` | 117 | 1 of 4 | `adapter_failed`: 40001 exhausted in `createRunUsageRecorder`, `reserveRunBudget` and native finalization | +| `w-20261110T044338-s20385` | master `8004ad069e` + harness | 2 | 1 of 4 | `adapter_failed`: 40001 in `createRunUsageRecorder`; deadlocks with the old multi-row writers | +| `w-20261110T050818-s20386` | same build, one-row writers | 0 | 0 of 4 | every first run failed in `reserveRunBudget` before spawn (the log keeps only the failed SQL, not its cause); each accepted Retry then waited on `process_identity_missing` and never started a run | + +No wave reached the runner's commit stall (20 failed deliveries over at least 45 s), so none showed +`native_event_replay_conflict`: on the parent a receipt write failed 117 times but a later delivery +committed. #20385 removed the receipt-write failures; the other repeatable-read accounting writers +that lock the run row still fail under the same contention on master. The contention check +(`contention_produced`) fired only on tasks whose runs failed before any usage was reported; a run +that finished produced 208 log events and 20 receipts, well above the 40 and 6 thresholds. ## Universal checks (U1–U4) @@ -582,7 +596,7 @@ its workload. They catch a stuck or misleading task that no scenario anticipated | Rule | Fails when | Bound (env override) | | --- | --- | --- | -| U1 liveness | a non-terminal task has neither a live owner (a running run with recent evidence, a run queued within the queue bound, a retry timer, a system recovery action) nor a visible, actionable item (Retry, a question or `userAction`, a failed notice, a board recovery action, an execution blocker, `blocked` or `in_review`) for longer than the bound, counting only time with the server ready | 3 min ownerless (`RESTART_CHAOS_U1_ORPHAN_SECONDS`); a run counts as an owner for 5 min of queueing (`RESTART_CHAOS_U1_QUEUED_SECONDS`) | +| U1 liveness | a non-terminal task has neither a live owner (a running run with recent evidence, a run queued within the queue bound, a scheduled retry, an accepted Retry that waits on the system within the queue bound, a system recovery action) nor a visible, actionable item (Retry, a question or `userAction`, a failed notice, a board recovery action, an execution blocker, `blocked` or `in_review`; none of these count while an accepted Retry waits on the system) for longer than the bound, counting only time with the server ready | 3 min ownerless (`RESTART_CHAOS_U1_ORPHAN_SECONDS`); a queued run or an accepted Retry's wait counts as an owner for 5 min (`RESTART_CHAOS_U1_QUEUED_SECONDS`) | | U2 retry loop | more than N consecutive failed runs with the same status and code while the issue status stays the same | 3 (`RESTART_CHAOS_U2_SAME_FAILURE_RUNS`) | | U3 diagnosable | a failed run has no code or a generic one (`adapter_failed`, `unknown`, `error`, ...), its stored error is empty or names only a bare cause (`last cause: Error`), or a recovery action's cause or next action is empty or bare | none | | U4 stale notice | a Retry or failed notice stays on a task after its latest run succeeded, longer than the bound or at rest | 60 s (`RESTART_CHAOS_U4_STALE_SECONDS`) | @@ -592,8 +606,8 @@ U1, U2 and U4 fail the task (signature `U:...`). U3 is a warning: it is repor never fails a task. U3 reads run error text and recovery-action causes from the wave database at wave end, because the board's run list hides most error text. -The rules and bounds come from an audit of the saved evidence of 74 waves (1,592 tasks, 1,465 of -them passing R1–R6 and T; 28 waves where every task passed), run with: +The rules and bounds come from an audit of the saved evidence of 79 waves (1,620 tasks, 1,483 of +them passing R1–R6 and T; 30 waves where every task passed), run with: ```sh node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/universal-audit.ts [...] @@ -601,17 +615,18 @@ node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/universal-audit.ts ; expect(row?.n).toBe(RECEIPTS); }, 60_000); + + it("reads a task's accounting evidence from the wave database", async () => { + const f = await runningRun(); + const sourceId = randomUUID(); + const journal = (sequence: number, sourceEventId: string, eventType: string) => db.$client.unsafe( + `insert into run_usage_receipts (id, company_id, run_id, source_id, sequence, receipt_hash, receipt_json, complete, received_at) + values ($1, $2, $3, $4, $5, $6, $7::jsonb, false, now())`, + [randomUUID(), f.companyId, f.runId, sourceId, sequence, `hash-${sequence}`, + JSON.stringify({ producerEvent: { sourceEventId, eventType } })]); + await journal(1, "evt-1", "turn.started"); + await journal(2, "evt-2", "usage.reported"); + // The same producer event journaled twice for one source: the oracle's exactly-once check must see it. + await journal(3, "evt-2", "usage.reported"); + await journal(4, "evt-3", "turn.completed"); + await db.$client.unsafe( + `update heartbeat_runs set status = 'succeeded', error_code = null, log_bytes = 4096, cost_accounted_at = now(), + usage_json = usage_json || '{"accountingReceiptOwner":"native_producer","costUsd":0.25}'::jsonb where id = $1`, [f.runId]); + // Agent cost events name their agent; `agentOf` is the run whose agent is charged. + const cost = (companyId: string, runId: string | null, agentOf: string, cents: number) => db.$client.unsafe( + `insert into cost_events (company_id, agent_id, heartbeat_run_id, provider, model, cost_cents, occurred_at) + values ($1, (select agent_id from heartbeat_runs where id = $3), $2, 'openai', 'test-model', $4, now())`, + [companyId, runId, agentOf, cents]); + await cost(f.companyId, f.runId, f.runId, 10); + await cost(f.companyId, f.runId, f.runId, 15); + // A cost of the same company outside this run counts toward the company total only. + await cost(f.companyId, null, f.runId, 5); + // Another company's cost counts toward neither. + const elsewhere = await runningRun(); + await cost(elsewhere.companyId, elsewhere.runId, elsewhere.runId, 1000); + const evidence = await collectContentionEvidence(db, f.companyId, [f.runId], { spendCents: 30, pendingRunCount: 0 }); + expect(evidence.error).toBeNull(); + expect(evidence.runs).toHaveLength(1); + expect(evidence.runs[0]).toMatchObject({ id: f.runId, status: "succeeded", receipts: 4, duplicateReceipts: [`${sourceId}:evt-2`], + terminalReceipt: true, nativeProducer: true, costAccounted: true, recordedCostCents: 25, costEventCents: 25, logBytes: 4096, logEvents: 0 }); + expect(evidence.companyCostEventCents).toBe(30); + // A run of another company is never read, even when its id is passed. + const other = await runningRun(); + expect((await collectContentionEvidence(db, f.companyId, [other.runId], null)).runs).toEqual([]); + }, 60_000); }); diff --git a/tests/restart-chaos/contention.ts b/tests/restart-chaos/contention.ts index 688785a8be8..62e70309c13 100644 --- a/tests/restart-chaos/contention.ts +++ b/tests/restart-chaos/contention.ts @@ -174,6 +174,10 @@ export interface PressureStats { writers: number; updates: number; rowsTouched: * to back. Each statement keeps every value (`updated_at = updated_at`) but * still writes a new row version, which is exactly what invalidates a * repeatable-read snapshot that later locks the row. + * + * Each statement updates one row, like a run-log append. A statement over + * every running row locked several rows at once and deadlocked with product + * transactions (40P01 in the first live L1 wave), which the stream never does. */ export class RunRowPressure { readonly stats: PressureStats; @@ -194,12 +198,17 @@ export class RunRowPressure { async #loop() { while (!this.#stopped) { try { - const rows = await this.#db.$client.unsafe( - "update heartbeat_runs set updated_at = updated_at where company_id::text = $1 and status = 'running' returning id", [this.companyId]); - this.stats.updates += 1; - this.stats.rowsTouched += rows.length; - // Nothing running: idle briefly instead of spinning on an empty update. - if (!rows.length) await delay(50); + const running = await this.#db.$client.unsafe( + "select id::text as id from heartbeat_runs where company_id::text = $1 and status = 'running'", [this.companyId]) as Array<{ id: string }>; + // Nothing running: idle briefly instead of spinning on an empty read. + if (!running.length) { await delay(50); continue; } + for (const { id } of running) { + if (this.#stopped) break; + const rows = await this.#db.$client.unsafe( + "update heartbeat_runs set updated_at = updated_at where id::text = $1 and status = 'running' returning id", [id]); + this.stats.updates += 1; + this.stats.rowsTouched += rows.length; + } } catch (error) { // The server (and its database) restart during faults; keep going once it is back. this.stats.errors += 1; diff --git a/tests/restart-chaos/universal.test.ts b/tests/restart-chaos/universal.test.ts index ecac50768a3..c7a6eadbe90 100644 --- a/tests/restart-chaos/universal.test.ts +++ b/tests/restart-chaos/universal.test.ts @@ -70,6 +70,22 @@ describe("U1 liveness", () => { expect(gradeU1(record(pending, { restAt: null }), BOUNDS).verdict).toBe("pass"); }); + it("fails an accepted Retry that waits on the system past the bound with no run (w-20261110T050818-s20386)", () => { + const waiting = { ...retryable, retryWait: true }; + const failedRun = run("r1", "failed", 0, SEC, { errorCode: "adapter_failed" }); + const samples = polls(0, 9 * 60 * SEC, at => sample(at, "blocked", [failedRun], { interruption: waiting, executionBlocker: true })); + expect(gradeU1(record(samples, { restAt: null }), BOUNDS)).toMatchObject({ verdict: "fail", + detail: expect.stringMatching(/accepted Retry waiting on the system/) }); + // A wait that names a person's action (confirm the previous run stopped) is actionable. + const confirm = polls(0, 9 * 60 * SEC, at => sample(at, "blocked", [failedRun], + { interruption: { ...waiting, userAction: "confirm_run_stopped" } })); + expect(gradeU1(record(confirm, { restAt: null }), BOUNDS).verdict).toBe("pass"); + // A short wait that ends in a dispatched run is fine. + const short = [...polls(0, 60 * SEC, at => sample(at, "blocked", [failedRun], { interruption: waiting })), + sample(63 * SEC, "in_progress", [failedRun, run("r2", "running", 62 * SEC, null, { lastEvidenceAt: iso(63 * SEC) })])]; + expect(gradeU1(record(short, { restAt: null }), BOUNDS).verdict).toBe("pass"); + }); + it("fails a silent running run once it stops counting as an owner", () => { const samples = polls(0, 7 * 60 * SEC, at => sample(at, "in_progress", [run("r1", "running", 0)])); expect(gradeU1(record(samples, { restAt: null }), BOUNDS).verdict).toBe("fail"); diff --git a/tests/restart-chaos/universal.ts b/tests/restart-chaos/universal.ts index 435a9327b0f..83bf32c0af2 100644 --- a/tests/restart-chaos/universal.ts +++ b/tests/restart-chaos/universal.ts @@ -34,7 +34,7 @@ export interface UniversalBounds { } /** - * Defaults from the 2026-11-09 audit of 74 saved waves (1,592 tasks; README + * Defaults from the 2026-11-09 audit of 79 saved waves (1,620 tasks; README * "Universal checks"). Override with RESTART_CHAOS_U1_ORPHAN_SECONDS, * RESTART_CHAOS_U1_QUEUED_SECONDS, RESTART_CHAOS_U2_SAME_FAILURE_RUNS and * RESTART_CHAOS_U4_STALE_SECONDS. @@ -106,7 +106,14 @@ function finalRuns(record: TaskRecord): Map, bounds: UniversalBounds): string | null { +/** A retry timer or an accepted Retry that waits on the system, with no run dispatched yet. */ +export function retryTimer(sample: TaskSample): boolean { + const dispatched = (sample.runs ?? []).some(run => ["queued", "running", "scheduled_retry"].includes(run.status)); + return !dispatched && Boolean(sample.scheduledRetry || sample.interruption?.retryWait || sample.interruption?.state === "retry_scheduled"); +} + +/** `timerFor`: ready-server time an accepted Retry has waited on the system with no run dispatched. */ +export function liveOwner(sample: TaskSample, queuedFor: ReadonlyMap, bounds: UniversalBounds, timerFor = 0): string | null { for (const run of sample.runs ?? []) { if (run.status === "running") { if (run.toolInFlight || run.processAlive) return `run ${run.id} running (tool or process alive)`; @@ -118,8 +125,9 @@ export function liveOwner(sample: TaskSample, queuedFor: ReadonlyMap ["queued", "running", "scheduled_retry"].includes(run.status)); - if (!dispatched && (sample.scheduledRetry || sample.interruption?.retryWait || sample.interruption?.state === "retry_scheduled")) return "retry timer"; + // A scheduled retry is a timer (samples carry no due time, so it is not bounded here). An accepted Retry + // that waits on the system (`retryWait`) is an owner only within the queue bound. + if (retryTimer(sample) && (!sample.interruption?.retryWait || timerFor <= bounds.queuedMs)) return "retry timer"; if (sample.activeRecoveryAction && sample.recoveryOwner !== "board") return "system recovery action"; return null; } @@ -127,8 +135,12 @@ export function liveOwner(sample: TaskSample, queuedFor: ReadonlyMap(); let orphanMs = 0, worstOrphan = 0, orphanFrom: number | null = null, lastAt: number | null = null; let worstQueued = 0, worstQueuedRun = ""; + let timerFor = 0, worstTimer = 0; for (const sample of record.samples) { // Outages pause the clock: a down server is the fault's time, not the task's. if (!ready(sample)) { lastAt = null; continue; } @@ -153,8 +166,10 @@ export function gradeU1(record: TaskRecord, bounds: UniversalBounds): Omit worstQueued) { worstQueued = waited; worstQueuedRun = run.id; } } } + timerFor = retryTimer(sample) && sample.interruption?.retryWait ? timerFor + (lastAt !== null ? sample.at - lastAt : 0) : 0; + worstTimer = Math.max(worstTimer, timerFor); if (TERMINAL_ISSUE.has(sample.issueStatus ?? "")) { orphanMs = 0; orphanFrom = null; lastAt = sample.at; continue; } - const owned = liveOwner(sample, queuedFor, bounds) ?? actionableItem(sample); + const owned = liveOwner(sample, queuedFor, bounds, timerFor) ?? actionableItem(sample); if (owned) { orphanMs = 0; orphanFrom = null; } else { if (orphanFrom === null) orphanFrom = sample.at; @@ -162,14 +177,15 @@ export function gradeU1(record: TaskRecord, bounds: UniversalBounds): Omit bounds.orphanMs) { const queued = (sample.runs ?? []).filter(run => run.status === "queued" || run.status === "scheduled_retry"); - return { verdict: "fail", metrics: { ownerlessMs: orphanMs, queuedMs: worstQueued }, detail: `no_owner_no_action ${seconds(orphanMs)} in ${sample.issueStatus ?? "unknown"}` + return { verdict: "fail", metrics: { ownerlessMs: orphanMs, queuedMs: worstQueued, retryTimerMs: worstTimer }, detail: `no_owner_no_action ${seconds(orphanMs)} in ${sample.issueStatus ?? "unknown"}` + (queued.length ? ` with ${queued.map(run => `${run.id} queued ${seconds(queuedFor.get(run.id) ?? 0)}`).join(",")}` : "") + + (timerFor ? ` after a retry wait of ${seconds(timerFor)}${sample.interruption?.retryWait ? " (accepted Retry waiting on the system)" : ""}` : "") + ` (interruption ${sample.interruption?.state ?? "none"})` }; } } lastAt = sample.at; } - return { verdict: "pass", metrics: { ownerlessMs: worstOrphan, queuedMs: worstQueued }, detail: `longest ownerless ${seconds(worstOrphan)}; longest queued ${seconds(worstQueued)}${worstQueuedRun ? ` (${worstQueuedRun})` : ""}` }; + return { verdict: "pass", metrics: { ownerlessMs: worstOrphan, queuedMs: worstQueued, retryTimerMs: worstTimer }, detail: `longest ownerless ${seconds(worstOrphan)}; longest queued ${seconds(worstQueued)}${worstQueuedRun ? ` (${worstQueuedRun})` : ""}` }; } /** The issue status seen between two failures (both finish polls included) was not one value throughout. */