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" }); + }); +});