From 04bca8051bcd0bf9b12d311c25298d53b7e773ee Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 25 Sep 2026 03:08:52 +0000 Subject: [PATCH] fix(daemon): Retry a busy store instead of failing live sessions The event loop persisted each event in a deferred BEGIN that read before it wrote. A commit from another connection in between made the upgrade fail at once with SQLITE_BUSY_SNAPSHOT, which busy_timeout never retries. The catch then marked the session failed and left the loop while the harness kept running, so later events and usage were lost. Persist with BEGIN IMMEDIATE, and on any SQLITE_BUSY retry the same event with capped backoff instead of ending the loop. Other persistence errors still fail the session. classifyFailure now maps "database is locked" to STORE_BUSY with infra blame instead of UNKNOWN harness. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01R6o1wJTTJWcHm9mZMGvDTn --- .specs/features/store-busy-healing/spec.md | 34 +++++ src/core/errors.ts | 19 +++ src/daemon/daemon.ts | 53 ++++++-- tests/daemon-store-busy.test.ts | 138 +++++++++++++++++++++ 4 files changed, 231 insertions(+), 13 deletions(-) create mode 100644 .specs/features/store-busy-healing/spec.md create mode 100644 tests/daemon-store-busy.test.ts diff --git a/.specs/features/store-busy-healing/spec.md b/.specs/features/store-busy-healing/spec.md new file mode 100644 index 0000000..3222ad4 --- /dev/null +++ b/.specs/features/store-busy-healing/spec.md @@ -0,0 +1,34 @@ +# Store Busy Healing Specification + +## Problem Statement + +`attachDriverEvents` persistia cada evento num `BEGIN` deferred: `events.append` lê (dedupe por `source_key`, `MAX(sequence)`) antes do INSERT. Se outra conexão commita nesse intervalo, o upgrade leitura→escrita recebe `SQLITE_BUSY_SNAPSHOT` (errcode 517) na hora, sem `busy_timeout`. O `catch` tratava isso como falha do harness: gravava `session.failed` (`UNKNOWN`, `blame: harness`), marcava a linha `failed` e saía do loop, enquanto o harness seguia vivo e seus eventos deixavam de ser persistidos (sessão `df94`: tokens do segundo turno perdidos, `wait` com estado terminal falso). + +A causa do segundo escritor (N daemons) é tratada em `daemon-single-instance`. Esta spec torna o loop robusto a qualquer escritor concorrente (ex.: `codedeck usage backfill`, lock mantido > 5s). + +## Goals + +- [x] `BEGIN IMMEDIATE` no loop de eventos: o lock de escrita é pego antes da leitura, `busy_timeout` volta a valer +- [x] `SQLITE_BUSY` (errcode base 5, incluindo 517) não encerra o loop nem marca a sessão: o mesmo evento é retentado com backoff (25ms dobrando, teto 2s) enquanto o daemon roda +- [x] Erros de persistência não-busy mantêm o comportamento atual (`failed`) +- [x] `classifyFailure("database is locked")` → `STORE_BUSY`, `blame: infra`, `retryable: true` +- [ ] (próxima fase) Reconciliador: sessão `failed` com `STORE_BUSY`/`UNKNOWN` de lock e PID vivo com a mesma identidade é reatachada a partir dos offsets persistidos + +## Decisions + +| Decisão | Escolha | Racional | +| ------- | ------- | -------- | +| Retry sem limite | Sim, enquanto não `shuttingDown` | O harness continua vivo e a saída está no log em disco; desistir recria o bug. Um lock eterno é problema do daemon, logado uma vez em `daemon.log` | +| Idempotência do retry | Transação inteira refeita (evento + offsets + status) | ROLLBACK desfaz tudo; dedupe por `sourceKey` protege replays | +| Shutdown durante o retry | Sai do loop sem gravar nada | O drain de shutdown é dono do desfecho (`interrupted`) | + +## Acceptance Criteria + +1. WHEN outro processo commita entre a leitura e a escrita da transação do evento THEN todos os eventos SHALL ser persistidos e a sessão SHALL terminar `completed` +2. WHEN `append` lança `SQLITE_BUSY`/`BUSY_SNAPSHOT` THEN o daemon SHALL retentar o mesmo evento e nenhum `session.failed` SHALL ser gravado +3. WHEN `append` lança um erro não-busy THEN a sessão SHALL ser marcada `failed` (sem regressão) +4. WHEN o texto do erro é `database is locked` THEN `classifyFailure` SHALL retornar `STORE_BUSY` com `blame: infra` + +## Validation + +`tests/daemon-store-busy.test.ts` (AC 1-4). AC 1 reproduz a intercalação real com uma segunda `DatabaseSync` e falhava antes do fix com só `session.failed` no log de eventos. diff --git a/src/core/errors.ts b/src/core/errors.ts index 1eef1e0..af55dbc 100644 --- a/src/core/errors.ts +++ b/src/core/errors.ts @@ -84,6 +84,7 @@ export type FailureCode = | "SPAWN_FAILED" // could not start the harness binary | "TIMEOUT" | "SHUTDOWN" // daemon shut down gracefully; session left interrupted + | "STORE_BUSY" // the daemon's SQLite store was locked; the harness may still run | "UNKNOWN"; export interface FailureInfo { @@ -119,6 +120,19 @@ const SIGNAL_EXIT_CODES: Record = { 139: "SIGSEGV", }; +// SQLITE_BUSY (5) and its extended codes, e.g. SQLITE_BUSY_SNAPSHOT (517), +// which a WAL read snapshot gets when another writer commits before it +// upgrades to write. busy_timeout never retries that one. +const SQLITE_BUSY = 5; +const STORE_BUSY_PATTERN = /database is locked|SQLITE_BUSY|database is busy/i; + +export function isStoreBusy(error: unknown): boolean { + const errcode = (error as { errcode?: unknown } | null)?.errcode; + if (typeof errcode === "number") return (errcode & 0xff) === SQLITE_BUSY; + const message = error instanceof Error ? error.message : String(error); + return STORE_BUSY_PATTERN.test(message); +} + export function classifyFailure( text: string, exitCode?: number | null, @@ -126,6 +140,11 @@ export function classifyFailure( ): FailureInfo { const clean = (text || "").trim(); const detail = clean.slice(0, 300) || undefined; + // The daemon's own store, not the harness or the task: checked first so + // no harness signature claims it. + if (STORE_BUSY_PATTERN.test(clean)) { + return { code: "STORE_BUSY", blame: "infra", retryable: true, detail }; + } for (const [pattern, reason] of HARNESS_CRASH_PATTERNS) { if (pattern.test(clean)) { return { code: "HARNESS_CRASH", blame: "harness", retryable: true, reason, detail }; diff --git a/src/daemon/daemon.ts b/src/daemon/daemon.ts index 7ac4e16..ea309f7 100644 --- a/src/daemon/daemon.ts +++ b/src/daemon/daemon.ts @@ -24,7 +24,7 @@ import { readSessionProcessMetadata } from "../drivers/session-runtime.js"; import type { AgentEvent } from "../core/events.js"; import { loadConfig, resolveDefaultSandbox } from "../config/config.js"; import { invalidWebPortMessage, resolveWebPort } from "../config/web-port.js"; -import { classifyFailure, RunAgentError, type FailureInfo } from "../core/errors.js"; +import { classifyFailure, isStoreBusy, RunAgentError, type FailureInfo } from "../core/errors.js"; import { parseRole } from "../core/roles.js"; import { getCachedOrDiscoverModels, type HarnessModels } from "../core/models.js"; import { aggregateRunUsage } from "../core/run-usage.js"; @@ -108,6 +108,11 @@ export interface DaemonOptions { spawnWebChild?: WebSupervisorOptions["spawnChild"]; } +// Backoff for a busy store in the event loop; capped, never given up while +// the daemon runs (see persistDriverEvent). +const STORE_BUSY_RETRY_BASE_MS = 25; +const STORE_BUSY_RETRY_MAX_MS = 2000; + function appendDaemonLog(line: string): void { try { fs.appendFileSync(getPaths().daemonLog, `[${new Date().toISOString()}] ${line}\n`); @@ -1662,19 +1667,21 @@ class Daemon { } } - private async attachDriverEvents(sessionId: string, driver: AgentDriver, drvSession: DriverSession): Promise { - try { - for await (const ev of driver.events(drvSession)) { - // A stop operation owns the terminal outcome. A terminal frame already - // buffered by the harness must not race it into the event log. - if (this.sessionLocks.has(sessionId) && (ev.type === "session.completed" || ev.type === "session.failed")) { - continue; - } - const db = this.db.getHandle(); - db.exec("BEGIN"); - let inserted = 0; + // Commits one event with its cursor. A busy store is the daemon's problem, + // not the session's: the harness keeps running and its output stays in the + // log file, so retry the same event until the store frees up instead of + // failing a live session. Returns undefined when shutdown interrupts the + // wait (the drain owns the outcome); any other error propagates. + private async persistDriverEvent(sessionId: string, driver: AgentDriver, ev: AgentEvent): Promise { + const db = this.db.getHandle(); + for (let attempt = 0; ; attempt += 1) { + try { + // IMMEDIATE takes the write lock up front, so busy_timeout applies. + // A deferred BEGIN read first and then got SQLITE_BUSY_SNAPSHOT on + // the upgrade, which no busy handler retries. + db.exec("BEGIN IMMEDIATE"); try { - inserted = this.events.append(sessionId, ev, ev.raw); + const inserted = this.events.append(sessionId, ev, ev.raw); const offsets = driver.getOffsets?.(sessionId); if (offsets) { this.sessions.update(sessionId, { logOffset: offsets.log, stderrOffset: offsets.stderr }); @@ -1684,10 +1691,30 @@ class Daemon { this.updateSessionFromEvent(sessionId, ev, inserted); } db.exec("COMMIT"); + return inserted; } catch (error) { try { db.exec("ROLLBACK"); } catch {} throw error; } + } catch (error) { + if (!isStoreBusy(error)) throw error; + if (this.shuttingDown) return undefined; + if (attempt === 0) appendDaemonLog(`store busy persisting ${ev.type} for ${sessionId}; retrying`); + await sleep(Math.min(STORE_BUSY_RETRY_BASE_MS * 2 ** attempt, STORE_BUSY_RETRY_MAX_MS)); + } + } + } + + private async attachDriverEvents(sessionId: string, driver: AgentDriver, drvSession: DriverSession): Promise { + try { + for await (const ev of driver.events(drvSession)) { + // A stop operation owns the terminal outcome. A terminal frame already + // buffered by the harness must not race it into the event log. + if (this.sessionLocks.has(sessionId) && (ev.type === "session.completed" || ev.type === "session.failed")) { + continue; + } + const inserted = await this.persistDriverEvent(sessionId, driver, ev); + if (inserted === undefined) return; // Broadcast only after the durable event+cursor commit, and never // broadcast a replay that was already in the event store. if (inserted !== 0) this.broadcast(sessionId, ev); diff --git a/tests/daemon-store-busy.test.ts b/tests/daemon-store-busy.test.ts new file mode 100644 index 0000000..0c50786 --- /dev/null +++ b/tests/daemon-store-busy.test.ts @@ -0,0 +1,138 @@ +import path from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { Daemon } from "../src/daemon/daemon.js"; +import { classifyFailure } from "../src/core/errors.js"; +import type { AgentEvent } from "../src/core/events.js"; +import { makeTempDir, removeTempDir, seam, seed } from "./helpers/daemon-seam.js"; + +// A persistence error is the daemon's problem, not the session's: while the +// harness runs, a busy SQLite must never end the event loop or mark the +// session failed (the "database is locked" df94 incident). + +let dir: string; +let daemon: Daemon | undefined; + +beforeEach(() => { + dir = makeTempDir("store-busy-"); + process.env.RUN_AGENT_DIR = dir; +}); + +afterEach(() => { + try { if (daemon) seam(daemon).db.close(); } catch {} + daemon = undefined; + delete process.env.RUN_AGENT_DIR; + removeTempDir(dir); +}); + +function turn(sessionId: string): AgentEvent[] { + const ts = () => new Date().toISOString(); + return [ + { type: "message", sessionId, timestamp: ts(), content: "working on it", sourceKey: "log:10:0" }, + { type: "usage.updated", sessionId, timestamp: ts(), usage: { inputTokens: 1200, outputTokens: 80 }, sourceKey: "log:20:0" }, + { type: "session.completed", sessionId, timestamp: ts(), reason: "completed", sourceKey: "log:30:0" }, + ] as AgentEvent[]; +} + +function driverFor(events: AgentEvent[]) { + return { + async *events() { + for (const ev of events) yield ev; + }, + }; +} + +function busyError(errcode: number): Error { + return Object.assign(new Error("database is locked"), { code: "ERR_SQLITE_ERROR", errcode }); +} + +async function run(sessionId: string, events: AgentEvent[]): Promise { + await (daemon as any).attachDriverEvents(sessionId, driverFor(events), { id: sessionId }); +} + +describe("event loop under a busy store", () => { + it("survives another process committing mid-transaction (SQLITE_BUSY_SNAPSHOT)", async () => { + daemon = new Daemon(); + seed(daemon, "s-snap"); + const store = seam(daemon).events; + const handle = seam(daemon).db.getHandle(); + // A second writer on the same file: what a stray daemon was. + const other = new DatabaseSync(path.join(dir, "run-agent.db")); + other.exec("PRAGMA busy_timeout = 0; CREATE TABLE other_writer (n INTEGER);"); + const original = store.append.bind(store); + let interleaved = false; + store.append = (sessionId, event, raw) => { + if (!interleaved && event.type === "message") { + interleaved = true; + // Open this transaction's read snapshot, then let the other writer + // commit before our INSERT, exactly the production interleaving. + handle.prepare("SELECT COUNT(*) FROM events").get(); + try { + other.exec("INSERT INTO other_writer VALUES (1)"); + } catch (error) { + // An IMMEDIATE transaction already holds the write lock, so the + // other writer is the one refused. That is the fixed behaviour. + if (!/database is locked/.test((error as Error).message)) throw error; + } + } + return original(sessionId, event, raw); + }; + + try { + await run("s-snap", turn("s-snap")); + } finally { + other.close(); + } + + const types = store.list("s-snap").map((e) => e.type); + expect(types).toEqual(["message", "usage.updated", "session.completed"]); + expect(seam(daemon).sessions.get("s-snap")?.status).toBe("completed"); + }); + + it("retries a busy append instead of failing the session", async () => { + daemon = new Daemon(); + seed(daemon, "s-busy"); + const store = seam(daemon).events; + const original = store.append.bind(store); + let failures = 0; + store.append = (sessionId, event, raw) => { + // Busy for longer than busy_timeout covers: plain SQLITE_BUSY, then + // BUSY_SNAPSHOT, then the store frees up. + if (event.type === "usage.updated" && failures < 2) { + failures += 1; + throw busyError(failures === 1 ? 5 : 517); + } + return original(sessionId, event, raw); + }; + + await run("s-busy", turn("s-busy")); + + expect(failures).toBe(2); + const events = store.list("s-busy"); + expect(events.map((e) => e.type)).toEqual(["message", "usage.updated", "session.completed"]); + expect(events.some((e) => e.type === "session.failed")).toBe(false); + expect(seam(daemon).sessions.get("s-busy")?.status).toBe("completed"); + }); + + it("still fails the session on a non-busy persistence error", async () => { + daemon = new Daemon(); + seed(daemon, "s-bad"); + const store = seam(daemon).events; + const original = store.append.bind(store); + store.append = (sessionId, event, raw) => { + if (event.type === "usage.updated") throw new Error("disk I/O error"); + return original(sessionId, event, raw); + }; + + await run("s-bad", turn("s-bad")); + + expect(seam(daemon).sessions.get("s-bad")?.status).toBe("failed"); + }); +}); + +describe("classifyFailure for store errors", () => { + it("blames infra, not the harness, for a locked database", () => { + expect(classifyFailure("database is locked")).toMatchObject({ code: "STORE_BUSY", blame: "infra", retryable: true }); + expect(classifyFailure("SQLITE_BUSY: database is busy")).toMatchObject({ code: "STORE_BUSY", blame: "infra" }); + }); +});