diff --git a/runner/recovery.ts b/runner/recovery.ts new file mode 100644 index 0000000..ff182b0 --- /dev/null +++ b/runner/recovery.ts @@ -0,0 +1,224 @@ +import { closeSync, constants, fstatSync, lstatSync, mkdirSync, openSync, readdirSync, realpathSync, rmSync, statSync, statfsSync, writeFileSync } from 'node:fs'; +import { homedir } from 'node:os'; +import { dirname, join, resolve } from 'node:path'; +import { execFileSync } from 'node:child_process'; +import { createRequire } from 'node:module'; +import type { DatabaseSync } from 'node:sqlite'; +import type { Store } from './store.ts'; +import { WRITABLE_KINDS, isUuidV4 } from './lifecycle.ts'; + +/** + * Startup recovery and the single-runner lock. See docs/implementation/runner-lifecycle.md, + * "Startup recovery" and decision 1. D's recovery, export and removal are injected until #51 provides them. + */ +export class LockHeld extends Error { constructor() { super('Another codeboost runner is using this database.'); } } +export class RecoveryBlocked extends Error { + readonly items: string[]; + constructor(message: string, items: string[]) { super(`${message}: ${items.join(', ')}`); this.items = items; } +} + +/** Linux statfs magic numbers for network filesystems, where POSIX locks are unreliable. */ +const NETWORK_FILESYSTEMS = new Set([0x6969 /* NFS */, 0x517b /* SMB */, 0xff534d42 /* CIFS */, 0xfe534d42 /* SMB2 */, 0x65735546 /* FUSE */]); +function assertOwnerOnly(path: string, what: string): void { + const st = statSync(path); + if (!st.isDirectory() || (process.getuid && st.uid !== process.getuid()) || (st.mode & 0o022) !== 0) + throw new Error(`${what} ${path} must be a directory owned by you and not writable by group or others.`); +} +function assertLocal(path: string): void { + if (process.platform !== 'linux') return; // macOS statfs types are not stable enough to classify; see the PR notes. + if (NETWORK_FILESYSTEMS.has(Number(statfsSync(path).type))) throw new Error(`${path} is on a network filesystem; the runner lock needs a local filesystem.`); +} + +/** + * Strong references to every held lock connection. Without this, a caller that drops the returned RunnerLock lets the + * connection be garbage-collected, and closing it silently releases the OS lock while the runner is still alive. + */ +const heldLocks = new Set(); +export interface RunnerLock { + readonly file: { dev: bigint; ino: bigint }; + /** Step 4: the database path still names the locked file. Call after the Store opens. */ + verify(): void; + release(): void; +} +/** Steps 0–2: owner-only parent, identify (or create) the file with no-follow, then take the OS lock by device and inode. */ +export function acquireRunnerLock(databasePath: string, options: { lockRoot?: string } = {}): RunnerLock { + const absolute = resolve(databasePath), parent = realpathSync(dirname(absolute)), path = join(parent, absolute.slice(dirname(absolute).length + 1)); + assertOwnerOnly(parent, 'The database directory'); + assertLocal(parent); + const link = lstatSync(path, { throwIfNoEntry: false }); + if (link?.isSymbolicLink()) throw new Error('The database path must not be a symlink.'); + let fd: number; + try { fd = openSync(path, constants.O_RDWR | constants.O_NOFOLLOW); } + catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; + try { fd = openSync(path, constants.O_RDWR | constants.O_CREAT | constants.O_EXCL | constants.O_NOFOLLOW, 0o600); } + catch (race) { if ((race as NodeJS.ErrnoException).code !== 'EEXIST') throw race; fd = openSync(path, constants.O_RDWR | constants.O_NOFOLLOW); } + } + let db: DatabaseSync | undefined; + try { + const st = fstatSync(fd, { bigint: true }); + if (!st.isFile() || st.nlink !== 1n) throw new Error('The database must be a regular file with exactly one name (no hard links).'); + const lockRoot = options.lockRoot ?? join(homedir(), '.codeboost', 'locks'); + mkdirSync(lockRoot, { recursive: true, mode: 0o700 }); + assertOwnerOnly(lockRoot, 'The lock directory'); + assertLocal(lockRoot); + const { DatabaseSync } = createRequire(import.meta.url)('node:sqlite') as typeof import('node:sqlite'); + db = new DatabaseSync(join(lockRoot, `${st.dev}-${st.ino}.runner-lock`), { timeout: 0 }); + try { + // EXCLUSIVE locking mode keeps the file lock after the first write until the connection closes; the OS drops it on exit or crash. + db.exec('PRAGMA locking_mode=EXCLUSIVE; CREATE TABLE IF NOT EXISTS holder (id INTEGER PRIMARY KEY); INSERT OR REPLACE INTO holder VALUES (1);'); + } catch (error) { + if (/locked|busy/i.test((error as Error).message)) throw new LockHeld(); + throw error; + } + const file = { dev: st.dev, ino: st.ino }, heldDb = db; + heldLocks.add(heldDb); + let released = false; + return { + file, + verify() { + const now = lstatSync(path, { bigint: true }); + if (now.isSymbolicLink() || !now.isFile() || now.dev !== file.dev || now.ino !== file.ino || now.nlink !== 1n) + throw new Error('The database path changed while opening. Refusing to start.'); + }, + release() { if (released) return; released = true; heldLocks.delete(heldDb); heldDb.close(); closeSync(fd); }, + }; + } catch (error) { db?.close(); closeSync(fd); throw error; } +} + +export interface ProcessControl { + /** The group leader is alive and started at the recorded time (guards against PID reuse). */ + isAlive(pgid: number, startedAt: number): boolean; + /** SIGTERM the group, SIGKILL after the grace period, and resolve only once it has exited. */ + terminate(pgid: number, graceMs: number): Promise; +} +export const hostProcesses: ProcessControl = { + isAlive(pgid, startedAt) { + try { process.kill(-pgid, 0); } catch { return false; } + try { + const started = Date.parse(execFileSync('ps', ['-o', 'lstart=', '-p', String(pgid)], { encoding: 'utf8' }).trim()); + return Number.isFinite(started) && Math.abs(started - startedAt) < 2_000; + } catch { return true; } // Alive but unreadable: treat as ours and stop it (fail closed). + }, + async terminate(pgid, graceMs) { + const alive = () => { try { process.kill(-pgid, 0); return true; } catch { return false; } }; + try { process.kill(-pgid, 'SIGTERM'); } catch { return; } + const until = Date.now() + graceMs; + while (alive() && Date.now() < until) await new Promise(resolve => setTimeout(resolve, 50)); + if (alive()) { try { process.kill(-pgid, 'SIGKILL'); } catch {} } + while (alive()) await new Promise(resolve => setTimeout(resolve, 50)); + }, +}; + +export interface RecoveredStorage { readonly attemptId: string; readonly allocationId: string; readonly handle: unknown } +export interface RecoveryDeps { + /** D (#51): stop leftover agent containers, proxies and networks for this owner; keep task storage and return authenticated handles. */ + recoverLeftovers(runnerOwner: string): Promise<{ storage: RecoveredStorage[]; unowned: string[] }>; + /** D (#51, before F2): bounded diff of a recovered task volume; must stop its own work when the signal aborts. */ + exportTaskDiff(handle: unknown, maxBytes: number, signal: AbortSignal): Promise; + removeTaskFilesystems(handle: unknown): Promise; + /** F3: abort an interrupted rebase. Until F3 exists no rebase is ever recorded. */ + abortRebase?(planKey: string, marker: unknown): Promise; + processes?: ProcessControl; +} +export interface RecoveryOptions { + store: Store; runnerOwner: string; runnerRoot: string; diagnosticsDir: string; deps: RecoveryDeps; + now?: () => number; exportDeadlineMs?: number; graceMs?: number; +} +export interface RecoveryReport { + finalized: { attemptId: string; planKey: string; state: string; requeued: boolean }[]; + requeue: string[]; removedDirectories: string[]; unknownEntries: string[]; unmatchedStorage: string[]; repairedMerges: string[]; +} +const EXPORT_LIMIT = 1024 * 1024; + +/** + * Startup steps 2b–7 after the lock (step 1) and the Store open (2a). Throws on any step that must fail closed; + * the caller then closes the Store, releases the lock and exits without opening the coordinator. + */ +export async function recoverStartup(o: RecoveryOptions): Promise { + const now = o.now ?? Date.now, processes = o.deps.processes ?? hostProcesses; + if (!/^[0-9a-f]{32}$/.test(o.runnerOwner)) throw new Error('Invalid runner owner token.'); + const interrupted = o.store.interruptedAttempts(); + // Step 7's input is taken before finalization: an interrupted attempt that started preparation but never saved its group. + const unowned = interrupted.filter(a => a.preparationStartedAt !== null && a.preparationPgid === null).map(a => a.id); + // 2b. Stop leftover preparation before D's recovery or any storage work. + for (const a of interrupted) if (a.preparationPgid !== null && a.preparationStartedAt !== null && processes.isAlive(a.preparationPgid, a.preparationStartedAt)) + await processes.terminate(a.preparationPgid, o.graceMs ?? 5_000); + // 2c/2d. D's recovery; a rejection propagates and stops startup. + const recovered = await o.deps.recoverLeftovers(o.runnerOwner); + if (recovered.unowned.length) throw new RecoveryBlocked('Unlabelled codeboost resources from an older build must be removed by hand (see --list-unowned-agent-resources)', recovered.unowned); + const matched = recovered.storage.filter(s => isUuidV4(s.attemptId) && o.store.attemptOwner(s.attemptId) !== null); + const unmatchedStorage = recovered.storage.filter(s => !matched.includes(s)).map(s => s.attemptId); + // 3. Export phase: stopped writable attempts only, outside any transaction, fixed names, bounded deadline. + const exports: Record = {}; + mkdirSync(o.diagnosticsDir, { recursive: true, mode: 0o700 }); + for (const storage of matched) { + const attempt = interrupted.find(a => a.id === storage.attemptId); + if (!attempt || !WRITABLE_KINDS.includes(attempt.kind)) continue; + const controller = new AbortController(), timer = setTimeout(() => controller.abort(new Error('Export timed out.')), o.exportDeadlineMs ?? 60_000); + try { + const diff = await Promise.race([o.deps.exportTaskDiff(storage.handle, EXPORT_LIMIT, controller.signal), + new Promise((_, reject) => controller.signal.addEventListener('abort', () => reject(controller.signal.reason), { once: true }))]); + const file = join(o.diagnosticsDir, `${storage.attemptId}.diff`); + writeFileSync(file, diff.subarray(0, EXPORT_LIMIT), { mode: 0o600 }); + exports[storage.attemptId] = { diagnosticRef: file }; + } catch (error) { exports[storage.attemptId] = { failure: error instanceof Error ? error.message : String(error) }; } + finally { clearTimeout(timer); } + } + // 3. Finalization phase: one transaction; a failure stops startup. + const finalized = o.store.recoverInterrupted(now(), exports); + // 4. Rebases, storage removal (every matched handle), then attempt directories. + for (const rebase of o.store.rebasesInProgress()) { + if (!o.deps.abortRebase) throw new RecoveryBlocked('An interrupted rebase needs F3 to abort it', [rebase.planKey]); + await o.deps.abortRebase(rebase.planKey, rebase.marker); + } + for (const storage of matched) await o.deps.removeTaskFilesystems(storage.handle); + const removedDirectories: string[] = [], unknownEntries: string[] = []; + const attemptsDir = join(o.runnerRoot, o.runnerOwner, 'attempts'); + const rootDev = lstatSync(o.runnerRoot, { throwIfNoEntry: false })?.dev; + for (const name of lstatSync(attemptsDir, { throwIfNoEntry: false })?.isDirectory() ? readdirSync(attemptsDir) : []) { + const entry = join(attemptsDir, name), st = lstatSync(entry); + const owned = st.isDirectory() && !st.isSymbolicLink() && st.dev === rootDev && isUuidV4(name) && o.store.attemptOwner(name) !== null; + if (!owned) { unknownEntries.push(entry); continue; } + if (unowned.includes(name)) continue; // step 7 keeps it for --release-preparation + rmSync(entry, { recursive: true, force: true }); removedDirectories.push(entry); + } + // 5. Confirmed merges get their closed status and task-closed event. + const repairedMerges = o.store.reconcileMergedTasks(); + // 7. An unidentifiable preparation child may exist: fail closed until the user releases it. + const blocked = [...unowned, ...o.store.unownedPreparations().filter(id => !unowned.includes(id))]; + if (blocked.length) throw new RecoveryBlocked('Preparation started but its process was never recorded; stop it, then run --release-preparation', blocked); + return { finalized, requeue: finalized.filter(f => f.requeued).map(f => f.planKey), removedDirectories, unknownEntries, unmatchedStorage, repairedMerges }; +} + +/** --release-preparation: remove an attempt directory only when no process has a file open or a working directory in it. */ +export function releasePreparation(o: { store: Store; runnerRoot: string; runnerOwner: string; attemptId: string; openFiles?: (dir: string) => string[] }): void { + if (!isUuidV4(o.attemptId) || o.store.attemptOwner(o.attemptId) === null) throw new Error('Unknown attempt.'); + const dir = join(o.runnerRoot, o.runnerOwner, 'attempts', o.attemptId), st = lstatSync(dir, { throwIfNoEntry: false }); + if (st && (!st.isDirectory() || st.isSymbolicLink())) throw new Error('The attempt path is not a plain directory.'); + if (st) { + const users = (o.openFiles ?? hostOpenFiles)(dir); + if (users.length) throw new RecoveryBlocked('A process is still using the attempt directory', users); + rmSync(dir, { recursive: true, force: true }); + } + if (!o.store.clearPreparationMarker(o.attemptId)) throw new Error('The attempt is not waiting for release.'); +} +/** Processes with a file open or a working directory under dir. Throws if the check cannot run (fail closed). */ +export function hostOpenFiles(dir: string): string[] { + if (process.platform === 'linux') { + const users: string[] = []; + for (const pid of readdirSync('/proc').filter(name => /^\d+$/.test(name))) { + const links: string[] = []; + try { links.push(realpathSync(`/proc/${pid}/cwd`)); } catch {} + try { for (const fd of readdirSync(`/proc/${pid}/fd`)) { try { links.push(realpathSync(`/proc/${pid}/fd/${fd}`)); } catch {} } } catch {} + if (links.some(link => link === dir || link.startsWith(`${dir}/`))) users.push(pid); + } + return users; + } + try { return execFileSync('lsof', ['-t', '+D', dir], { encoding: 'utf8' }).split('\n').filter(Boolean); } + catch (error) { + const e = error as { status?: number; stdout?: string }; + if (e.status === 1 && !e.stdout) return []; // lsof exits 1 when nothing matches + throw new Error('Could not check which processes use the attempt directory. Refusing to release it.'); + } +} diff --git a/runner/store.ts b/runner/store.ts index fb20574..89279d6 100644 --- a/runner/store.ts +++ b/runner/store.ts @@ -610,7 +610,7 @@ export class Store { /** Admission, including retry: status, state version, requeue claim, active attempt and captured context are checked in one transaction. */ admitAttempt(identity: PlanIdentity, input: { expectedStateVersion: number; kind: AttemptKind; item?: string | null; expectedContext: InvocationContext; - deadline: number; budgetMs?: number; retryOf?: string; now?: number; + deadline: number; budgetMs?: number; retryOf?: string; now?: number; claimRequeue?: boolean; }): AttemptRecord { const now = input.now ?? Date.now(), budgetMs = input.budgetMs ?? DEFAULT_TASK_BUDGET_MS; if (!(input.kind in ATTEMPT_PHASES)) throw new GuardRefusal('Unknown attempt kind.'); @@ -622,7 +622,9 @@ export class Store { const task = this.#task(key); if (task.status !== 'running' && task.status !== 'queued') throw new GuardRefusal(`The task is ${task.status}; it cannot start work.`); if (task.state_version !== input.expectedStateVersion) throw new GuardRefusal('Stale task state. Reload before writing.'); - if (task.requeue_pending === 1) throw new GuardRefusal('Recovery is requeueing this task.'); + // Requeue claim: exactly one path (I3's requeue, or the user's Resume) clears it, by CAS in this admitting transaction. + if (task.requeue_pending === 1 && !input.claimRequeue) throw new GuardRefusal('Recovery is requeueing this task.'); + if (input.claimRequeue && task.requeue_pending !== 1) throw new GuardRefusal('The requeue was already claimed.'); if (task.cancel_requested !== null) throw new GuardRefusal('The task is being cancelled.'); if (this.#activeAttempt(key)) throw new GuardRefusal('An attempt is already active for this task.'); const current = this.#contextOf(key); @@ -637,7 +639,7 @@ export class Store { const id = randomUUID(), created = new Date(now).toISOString(); this.#run(`INSERT INTO attempts (id,plan_key,kind,phase,item,state,context,deadline,created_at) VALUES (?,?,?,?,?,'pending',?,?,?)`, id, key, input.kind, ATTEMPT_PHASES[input.kind], input.item ?? null, encode(current), input.deadline, created); - this.#run(`UPDATE tasks SET current_attempt_id=?, status='running', budget_deadline=COALESCE(budget_deadline, ?) WHERE plan_key=?`, id, now + budgetMs, key); + this.#run(`UPDATE tasks SET current_attempt_id=?, status='running', requeue_pending=0, budget_deadline=COALESCE(budget_deadline, ?) WHERE plan_key=?`, id, now + budgetMs, key); this.#touch(key); return this.getAttempt(identity, id); }); @@ -784,4 +786,123 @@ export class Store { })); } + // ---- F1d: startup recovery (runner-lifecycle.md, "Startup recovery") ---- + /** + * The per-database runner owner token, tied to the database file's device and inode. + * A stored value that is malformed is refused; a copied database (different file identity) gets a new token. + */ + runnerOwnerToken(file: { dev: number | bigint; ino: number | bigint }): string { + const identity = { dev: String(file.dev), ino: String(file.ino) }; + return this.#transaction(() => { + const row = this.#get("SELECT value FROM app_settings WHERE key='runner_owner'"); + if (row) { + let stored: { token?: unknown; dev?: unknown; ino?: unknown }; + try { stored = decode(row.value); } catch { throw new Error('Stored runner owner token is malformed. Refusing to start.'); } + if (typeof stored.token !== 'string' || !/^[0-9a-f]{32}$/.test(stored.token)) throw new Error('Stored runner owner token is malformed. Refusing to start.'); + if (stored.dev === identity.dev && stored.ino === identity.ino) return stored.token; + } + const token = randomUUID().replace(/-/g, ''); + this.#run("INSERT INTO app_settings VALUES ('runner_owner',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value", encode({ token, ...identity })); + return token; + }); + } + /** "Preparation starting": saved before any preparation subprocess is spawned. */ + markPreparationStarting(identity: PlanIdentity, id: string, startedAt: number): void { + const key = identityKey(identity); + if (this.#run(`UPDATE attempts SET preparation_started_at=? WHERE plan_key=? AND id=? AND state='pending'`, startedAt, key, id).changes !== 1) + throw new GuardRefusal('Only a pending attempt can start preparation.'); + } + /** The spawn failed synchronously: no child exists, so the "starting" marker must not block the next startup. */ + cancelPreparationStart(identity: PlanIdentity, id: string): void { + this.#run(`UPDATE attempts SET preparation_started_at=NULL WHERE plan_key=? AND id=? AND preparation_pgid IS NULL AND state='pending'`, identityKey(identity), id); + } + /** Saved in the same synchronous turn as the spawn. */ + recordPreparationGroup(identity: PlanIdentity, id: string, pgid: number): void { + if (!Number.isSafeInteger(pgid) || pgid < 2) throw new GuardRefusal('Invalid process group.'); + if (this.#run(`UPDATE attempts SET preparation_pgid=? WHERE plan_key=? AND id=? AND preparation_started_at IS NOT NULL`, pgid, identityKey(identity), id).changes !== 1) + throw new GuardRefusal('Preparation was not marked as starting.'); + } + /** F chooses the allocation ID and saves it before the asynchronous allocation starts. */ + recordAllocation(identity: PlanIdentity, id: string, allocationId: string): void { + assertUuidV4(allocationId, 'Allocation ID'); + if (this.#run(`UPDATE attempts SET allocation_id=? WHERE plan_key=? AND id=? AND state='pending' AND allocation_id IS NULL`, allocationId, identityKey(identity), id).changes !== 1) + throw new GuardRefusal('Allocation can be recorded once, for a pending attempt.'); + } + /** Every non-terminal attempt, across all plans, with the fields recovery needs. */ + interruptedAttempts(): (AttemptRecord & { planKey: string; preparationPgid: number | null; preparationStartedAt: number | null; allocationId: string | null })[] { + return this.#db.prepare("SELECT * FROM attempts WHERE state IN ('pending','running') ORDER BY rowid").all().map(row => ({ + ...this.#attemptRecord(row), planKey: row.plan_key as string, preparationPgid: row.preparation_pgid as number | null, + preparationStartedAt: row.preparation_started_at as number | null, allocationId: row.allocation_id as string | null, + })); + } + /** + * Startup recovery step 3, finalization phase: one transaction. Applies the settlement precedence to every + * leftover attempt, keeps the closed-task and pending-cancel guards, and sets the requeue claim. + */ + recoverInterrupted(now: number, exports: Readonly> = {}): { attemptId: string; planKey: string; state: string; requeued: boolean }[] { + const gated = ['needs human', 'needs amendment', 'needs approval', 'possibly already fixed']; + return this.#transaction(() => this.#db.prepare("SELECT * FROM attempts WHERE state IN ('pending','running') ORDER BY rowid").all().map(row => { + const key = row.plan_key as string, task = this.#task(key); + const contextCurrent = sameContext(decode(row.context), this.#contextOf(key)); + let firstReason = row.first_reason as FirstReason | null; + if (firstReason === null && contextCurrent && task.budget_deadline !== null && now >= (task.budget_deadline as number)) firstReason = 'time-limit'; + const deadlinePassed = firstReason === null && now >= (row.deadline as number); + const outcome = classifySettlement({ + firstReason, contextCurrent, exitCode: null, valid: false, + stopReason: (row.stop_reason as StopReason | null) ?? (deadlinePassed ? 'timeout' : undefined), + detail: 'Interrupted: codeboost stopped while this was running', + }); + const exported = exports[row.id as string]; + const diagnostic = exported?.failure ? `${outcome.reason ?? ''} Partial output could not be exported: ${exported.failure}`.trim() : outcome.reason; + this.#run(`UPDATE attempts SET state=?, first_reason=?, diagnostic=?, diagnostic_ref=COALESCE(?, diagnostic_ref), settled_at=? WHERE id=?`, + outcome.state, firstReason, bounded(diagnostic ?? ''), exported?.diagnosticRef ?? null, new Date(now).toISOString(), row.id!); + let requeued = false; + if (task.cancel_requested !== null && !this.#closed(task.status)) this.#closeTask(key, 'cancelled', task.cancel_requested as string); + else { + if (outcome.timeLimit && !this.#closed(task.status)) this.#run(`UPDATE tasks SET status='needs human' WHERE plan_key=?`, key); + const status = this.#task(key).status as string; + const interrupted = outcome.state === 'failed' && (outcome.reason ?? '').startsWith('Interrupted'); + const shutdown = outcome.state === 'cancelled' && firstReason === 'shutdown'; + if (!this.#closed(status) && !gated.includes(status) && (interrupted || shutdown)) { + this.#run('UPDATE tasks SET requeue_pending=1 WHERE plan_key=?', key); requeued = true; + } + this.#touch(key); + } + return { attemptId: row.id as string, planKey: key, state: outcome.state, requeued }; + })); + } + /** Tasks whose confirmed merge lacks its closed status or task-closed event (recovery step 5). */ + reconcileMergedTasks(): string[] { + return this.#transaction(() => { + const repaired: string[] = []; + for (const task of this.#db.prepare('SELECT plan_key, status FROM tasks').all()) { + const key = task.plan_key as string; + const latest = this.#get('SELECT id,data FROM merge_attempts WHERE key=? ORDER BY rowid DESC LIMIT 1', key); + if (!latest || decode(latest.data).state !== 'merged') continue; + const hasEvent = this.#get("SELECT 1 FROM feedback_events WHERE plan_key=? AND kind='task-closed'", key); + if (task.status === 'merged' && hasEvent) continue; + this.#closeTask(key, 'merged', latest.id as string); repaired.push(key); + } + return repaired; + }); + } + + /** Which plan owns an attempt ID, across all plans; null if none. */ + attemptOwner(attemptId: string): string | null { + return (this.#get('SELECT plan_key FROM attempts WHERE id=?', attemptId)?.plan_key as string | undefined) ?? null; + } + /** Interrupted rebases recorded by F3 (none exist before F3). */ + rebasesInProgress(): { planKey: string; marker: unknown }[] { + return this.#db.prepare('SELECT plan_key, rebase_in_progress FROM tasks WHERE rebase_in_progress IS NOT NULL').all() + .map(row => ({ planKey: row.plan_key as string, marker: decode(row.rebase_in_progress) })); + } + /** After --release-preparation verified the directory is unused: clear the "starting" marker of a terminal attempt. */ + clearPreparationMarker(attemptId: string): boolean { + return this.#run(`UPDATE attempts SET preparation_started_at=NULL WHERE id=? AND preparation_pgid IS NULL AND state NOT IN ('pending','running')`, attemptId).changes === 1; + } + /** Terminal attempts whose preparation started but whose process group was never saved. */ + unownedPreparations(): string[] { + return this.#db.prepare(`SELECT id FROM attempts WHERE preparation_started_at IS NOT NULL AND preparation_pgid IS NULL AND state NOT IN ('pending','running')`).all().map(row => row.id as string); + } + } diff --git a/test/runner-recovery.test.ts b/test/runner-recovery.test.ts new file mode 100644 index 0000000..a711486 --- /dev/null +++ b/test/runner-recovery.test.ts @@ -0,0 +1,259 @@ +import { chmodSync, copyFileSync, existsSync, linkSync, mkdirSync, mkdtempSync, readFileSync, renameSync, rmSync, statSync, symlinkSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { spawn } from 'node:child_process'; +import { once } from 'node:events'; +import { randomUUID } from 'node:crypto'; +import { fileURLToPath } from 'node:url'; +import { DatabaseSync } from 'node:sqlite'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { Store } from '../runner/store.ts'; +import { LockHeld, RecoveryBlocked, acquireRunnerLock, recoverStartup, releasePreparation, type RecoveryDeps, type RunnerLock } from '../runner/recovery.ts'; +import type { PlanIdentity } from '../core/identity.ts'; +import type { Plan, PlanContext } from '../core/plan.ts'; + +const oid = (n: number) => n.toString(16).padStart(40, '0'); +const plan = (summary = 'Example'): Plan => ({ schema_version: 1, revision: 1, issue: 1, summary, questions: [], items: [{ id: 'P1', title: 'Change', intent: 'Improve', files: [{ path: 'a', kind: 'edit', renamed_from: null, change: 'Change' }], acceptance: [{ type: 'check', text: 'Works' }], depends_on: [] }] }); +const ctx = (identity: PlanIdentity): PlanContext => ({ identity, issue: 1, baseEntries: [{ path: 'a', kind: 'file' }], pathKey: p => p, allowedCommands: [] }); +const id = (n: number): PlanIdentity => ({ repositoryId: 'repo', taskId: `task-${n}`, planId: 'plan' }); +const dirs: string[] = [], locks: RunnerLock[] = [], stores: Store[] = [], children: ReturnType[] = []; +afterEach(() => { + // A child holding a lock must never outlive its test, even when an assertion fails first. + for (const child of children.splice(0)) if (child.exitCode === null) child.kill('SIGKILL'); + for (const lock of locks.splice(0)) lock.release(); + for (const store of stores.splice(0)) { try { store.close(); } catch {} } + for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }); + vi.restoreAllMocks(); +}); +function dir() { const d = mkdtempSync(join(tmpdir(), 'codeboost-recovery-')); dirs.push(d); return d; } +function lock(path: string, lockRoot: string) { const l = acquireRunnerLock(path, { lockRoot }); locks.push(l); return l; } + +describe('runner lock', () => { + it('lets one holder at a time in this process, releases, and never deletes the lock file', () => { + const d = dir(), locksDir = join(d, 'locks'), path = join(d, 'db.sqlite'); + const first = lock(path, locksDir); + expect(statSync(path).mode & 0o777).toBe(0o600); + expect(() => acquireRunnerLock(path, { lockRoot: locksDir })).toThrow(LockHeld); + first.release(); locks.splice(locks.indexOf(first), 1); + const again = lock(path, locksDir); + expect(again.file).toEqual(first.file); + expect(existsSync(join(locksDir, `${first.file.dev}-${first.file.ino}.runner-lock`))).toBe(true); + }); + it('keeps holding the lock when the caller drops the returned object and garbage collection runs', async () => { + const d = dir(), locksDir = join(d, 'locks'), path = join(d, 'db.sqlite'); + const recovery = fileURLToPath(new URL('../runner/recovery.ts', import.meta.url)); + const child = spawn(process.execPath, ['--expose-gc', '-e', `import(${JSON.stringify(recovery)}).then(m => { m.acquireRunnerLock(${JSON.stringify(path)}, { lockRoot: ${JSON.stringify(locksDir)} }); for (let i = 0; i < 5; i++) globalThis.gc(); setTimeout(() => { globalThis.gc(); console.log('held'); }, 50); setInterval(() => {}, 1000); })`], { stdio: ['ignore', 'pipe', 'inherit'] }); + children.push(child); + await new Promise(resolve => child.stdout!.on('data', chunk => { if (String(chunk).includes('held')) resolve(); })); + expect(() => acquireRunnerLock(path, { lockRoot: locksDir })).toThrow(LockHeld); + }); + it('is released by the OS when the holding process is killed', async () => { + const d = dir(), locksDir = join(d, 'locks'), path = join(d, 'db.sqlite'); + const recovery = fileURLToPath(new URL('../runner/recovery.ts', import.meta.url)); + const child = spawn(process.execPath, ['-e', `import(${JSON.stringify(recovery)}).then(m => { m.acquireRunnerLock(${JSON.stringify(path)}, { lockRoot: ${JSON.stringify(locksDir)} }); console.log('held'); setInterval(() => {}, 1000); })`], { stdio: ['ignore', 'pipe', 'inherit'] }); + children.push(child); + await new Promise(resolve => child.stdout!.on('data', chunk => { if (String(chunk).includes('held')) resolve(); })); + expect(() => acquireRunnerLock(path, { lockRoot: locksDir })).toThrow(LockHeld); + child.kill('SIGKILL'); await once(child, 'exit'); + expect(() => lock(path, locksDir)).not.toThrow(); + }); + it('keys the lock by file identity, so a renamed or moved database meets the same lock', () => { + const d = dir(), locksDir = join(d, 'locks'), path = join(d, 'db.sqlite'); + lock(path, locksDir); + const renamed = join(d, 'renamed.sqlite'); renameSync(path, renamed); + expect(() => acquireRunnerLock(renamed, { lockRoot: locksDir })).toThrow(LockHeld); + const other = join(d, 'other'); mkdirSync(other, { mode: 0o700 }); + const moved = join(other, 'moved.sqlite'); renameSync(renamed, moved); + expect(() => acquireRunnerLock(moved, { lockRoot: locksDir })).toThrow(LockHeld); + }); + it('treats a copy as a different database with its own lock', () => { + const d = dir(), locksDir = join(d, 'locks'), path = join(d, 'db.sqlite'); + const original = lock(path, locksDir); + const copy = join(d, 'copy.sqlite'); copyFileSync(path, copy); + const copied = lock(copy, locksDir); + expect(copied.file.ino).not.toBe(original.file.ino); + }); + it('refuses hard links, symlinks and an unsafe parent directory before taking the lock', () => { + const d = dir(), locksDir = join(d, 'locks'), path = join(d, 'db.sqlite'); + writeFileSync(path, ''); chmodSync(path, 0o600); + linkSync(path, join(d, 'alias.sqlite')); + expect(() => acquireRunnerLock(path, { lockRoot: locksDir })).toThrow(/no hard links/); + const d2 = dir(), target = join(d2, 'real.sqlite'); writeFileSync(target, ''); + symlinkSync(target, join(d2, 'link.sqlite')); + expect(() => acquireRunnerLock(join(d2, 'link.sqlite'), { lockRoot: locksDir })).toThrow(/symlink/); + const d3 = dir(); chmodSync(d3, 0o777); + expect(() => acquireRunnerLock(join(d3, 'db.sqlite'), { lockRoot: locksDir })).toThrow(/not writable by group or others/); + expect(existsSync(join(d3, 'db.sqlite'))).toBe(false); + }); + it('detects a path swap after opening', () => { + const d = dir(), path = join(d, 'db.sqlite'), held = lock(path, join(d, 'locks')); + renameSync(path, join(d, 'moved.sqlite')); writeFileSync(path, ''); + expect(() => held.verify()).toThrow(/path changed/); + }); +}); + +function fixture(count = 1) { + const d = dir(), path = join(d, 'state.sqlite'), store = new Store(path); stores.push(store); + for (let n = 1; n <= count; n++) { + store.createPlan(JSON.stringify(plan()), 'json', ctx(id(n)), oid(1), oid(2)); + store.transitionTask(id(n), store.getTask(id(n)).stateVersion, 'queued'); + } + const raw = (sql: string) => { const db = new DatabaseSync(path); db.exec(sql); db.close(); }; + const admit = (identity: PlanIdentity, extra: Record = {}) => store.admitAttempt(identity, { + expectedStateVersion: store.getTask(identity).stateVersion, kind: 'execute', expectedContext: store.currentContext(identity), deadline: Date.now() + 60_000, ...extra, + }); + return { d, path, store, raw, admit }; +} + +describe('runner owner token', () => { + it('is stable for one file, new for a copy, and refused when malformed', () => { + const { store, raw } = fixture(); + const token = store.runnerOwnerToken({ dev: 1n, ino: 2n }); + expect(token).toMatch(/^[0-9a-f]{32}$/); + expect(store.runnerOwnerToken({ dev: 1n, ino: 2n })).toBe(token); + expect(store.runnerOwnerToken({ dev: 1n, ino: 3n })).not.toBe(token); + raw(`UPDATE app_settings SET value='{"token":"../x","dev":"1","ino":"3"}' WHERE key='runner_owner'`); + expect(() => store.runnerOwnerToken({ dev: 1n, ino: 3n })).toThrow(/malformed/); + }); +}); + +describe('finalizing interrupted attempts', () => { + it('applies the settlement precedence and the requeue rules to leftovers', () => { + const { store, raw, admit } = fixture(7); + const now = Date.now(); + const a = [1, 2, 3, 4, 5, 6, 7].map(n => { const attempt = admit(id(n)); store.markRunning(id(n), attempt.id); return attempt; }); + store.recordFirstReason(id(1), a[0]!.id, 'cancelled'); + store.recordFirstReason(id(3), a[2]!.id, 'shutdown'); raw(`UPDATE attempts SET stop_reason='timeout' WHERE id='${a[2]!.id}'`); + store.recordFirstReason(id(4), a[3]!.id, 'shutdown'); + raw(`UPDATE tasks SET budget_deadline=${now - 1} WHERE plan_key IN ('${store.getTask(id(5)).planKey}','${store.getTask(id(6)).planKey}')`); + store.setAssignment(id(6), store.getTask(id(6)).stateVersion, 'changed', 'hash'); + raw(`UPDATE attempts SET deadline=${now - 1} WHERE id='${a[6]!.id}'`); + const report = store.recoverInterrupted(now); + const by = (n: number) => report.find(r => r.attemptId === a[n - 1]!.id)!; + expect(by(1)).toMatchObject({ state: 'cancelled', requeued: false }); + expect(by(2)).toMatchObject({ state: 'failed', requeued: true }); + expect(store.getAttempt(id(2), a[1]!.id).diagnostic).toBe('Interrupted: codeboost stopped while this was running'); + expect(by(3)).toMatchObject({ state: 'failed', requeued: false }); + expect(by(4)).toMatchObject({ state: 'cancelled', requeued: true }); + expect(by(5)).toMatchObject({ state: 'cancelled', requeued: false }); + expect(store.getTask(id(5)).status).toBe('needs human'); + expect(by(6)).toMatchObject({ state: 'stale', requeued: false }); + expect(by(7)).toMatchObject({ state: 'failed', requeued: false }); + expect(store.getAttempt(id(7), a[6]!.id).diagnostic).toBe('Timed out.'); + expect(store.getTask(id(2)).requeuePending).toBe(true); + expect(store.interruptedAttempts()).toHaveLength(0); + }); + it('lets a pending cancel task win over a time limit after a crash', () => { + const { store, admit } = fixture(); + const attempt = admit(id(1)); store.markRunning(id(1), attempt.id); + store.recordFirstReason(id(1), attempt.id, 'time-limit'); + const cancelId = randomUUID(); store.cancelTask(id(1), store.getTask(id(1)).stateVersion, cancelId); + store.recoverInterrupted(Date.now()); + expect(store.getTask(id(1)).status).toBe('cancelled'); + expect(store.feedbackEvents(id(1)).filter(e => e.kind === 'task-closed')).toMatchObject([{ actionId: cancelId }]); + }); + it('holds the requeue claim until exactly one admission claims it', () => { + const { store, admit } = fixture(); + const attempt = admit(id(1)); store.markRunning(id(1), attempt.id); + store.recoverInterrupted(Date.now()); + expect(() => admit(id(1), { retryOf: attempt.id })).toThrow(/requeueing/); + const resumed = admit(id(1), { claimRequeue: true }); + expect(store.getTask(id(1)).requeuePending).toBe(false); + store.recordFirstReason(id(1), resumed.id, 'cancelled'); store.settleAttempt(id(1), resumed.id, { firstReason: null, exitCode: 0, valid: true }); + expect(() => admit(id(1), { claimRequeue: true })).toThrow(/already claimed/); + }); + it('repairs a confirmed merge whose task status or event is missing', () => { + const { store, raw } = fixture(); + const state = { revision: 1, snapshotId: store.getSnapshot(id(1)).id, reviewVersion: store.reviewVersion(id(1)) }; + const merge = store.beginMergeAttempt(id(1), state, oid(2), null, 'direct'); + store.finishMergeAttempt(id(1), merge.id, { state: 'merged' }); + raw(`UPDATE tasks SET status='in review'; DELETE FROM feedback_events;`); + expect(store.reconcileMergedTasks()).toEqual([store.getTask(id(1)).planKey]); + expect(store.getTask(id(1)).status).toBe('merged'); + expect(store.reconcileMergedTasks()).toEqual([]); + }); +}); + +describe('startup recovery sequence', () => { + const token = 'a'.repeat(32); + function deps(over: Partial = {}) { + const calls: string[] = []; + const d: RecoveryDeps = { + recoverLeftovers: async () => { calls.push('recover'); return { storage: [], unowned: [] }; }, + exportTaskDiff: async () => { calls.push('export'); return Buffer.from('diff'); }, + removeTaskFilesystems: async handle => { calls.push(`remove:${String(handle)}`); }, + processes: { isAlive: () => true, terminate: async pgid => { calls.push(`terminate:${pgid}`); } }, + ...over, + }; + return { d, calls }; + } + it('stops preparation first, then D recovery, export, finalization and removal, in that order', async () => { + const { d: root, store, admit } = fixture(2); + const writable = admit(id(1)); store.markRunning(id(1), writable.id); + const readOnly = admit(id(2), { kind: 'review' }); + store.markPreparationStarting(id(2), readOnly.id, Date.now()); store.recordPreparationGroup(id(2), readOnly.id, 4242); + const { d, calls } = deps({ + recoverLeftovers: async () => { calls.push('recover'); return { storage: [{ attemptId: writable.id, allocationId: randomUUID(), handle: 'w' }, { attemptId: readOnly.id, allocationId: randomUUID(), handle: 'r' }], unowned: [] }; }, + exportTaskDiff: async () => { calls.push(`export:${store.getAttempt(id(1), writable.id).state}`); return Buffer.from('partial'); }, + }); + const report = await recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'runner'), diagnosticsDir: join(root, 'diag'), deps: d }); + expect(calls).toEqual(['terminate:4242', 'recover', 'export:running', 'remove:w', 'remove:r']); + const finalized = store.getAttempt(id(1), writable.id); + expect(finalized).toMatchObject({ state: 'failed', diagnosticRef: join(root, 'diag', `${writable.id}.diff`) }); + expect(readFileSync(finalized.diagnosticRef!, 'utf8')).toBe('partial'); + expect(report.requeue).toContain(store.getTask(id(1)).planKey); + }); + it('stops before finalizing anything when D recovery rejects or reports unowned resources', async () => { + for (const recoverLeftovers of [async () => { throw new Error('docker down'); }, async () => ({ storage: [], unowned: ['container legacy'] })]) { + const { d: root, store, admit } = fixture(); + const attempt = admit(id(1)); store.markRunning(id(1), attempt.id); + await expect(recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'r'), diagnosticsDir: join(root, 'd'), deps: deps({ recoverLeftovers }).d })).rejects.toThrow(/docker down|older build/); + expect(store.getAttempt(id(1), attempt.id).state).toBe('running'); + } + }); + it('records an export timeout as a diagnostic and still finalizes and removes the storage', async () => { + const { d: root, store, admit } = fixture(); + const attempt = admit(id(1)); store.markRunning(id(1), attempt.id); + const { d, calls } = deps({ + recoverLeftovers: async () => ({ storage: [{ attemptId: attempt.id, allocationId: randomUUID(), handle: 'h' }], unowned: [] }), + exportTaskDiff: () => new Promise(() => undefined), + }); + await recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'r'), diagnosticsDir: join(root, 'd'), deps: d, exportDeadlineMs: 30 }); + expect(store.getAttempt(id(1), attempt.id)).toMatchObject({ state: 'failed', diagnosticRef: null }); + expect(store.getAttempt(id(1), attempt.id).diagnostic).toMatch(/Partial output could not be exported: Export timed out/); + expect(calls).toContain('remove:h'); + }); + it('removes nothing when the finalization transaction fails', async () => { + const { d: root, store, admit } = fixture(); + const attempt = admit(id(1)); store.markRunning(id(1), attempt.id); + vi.spyOn(store, 'recoverInterrupted').mockImplementation(() => { throw Object.assign(new Error('disk'), { code: 'ERR_SQLITE_ERROR' }); }); + const { d, calls } = deps({ recoverLeftovers: async () => ({ storage: [{ attemptId: attempt.id, allocationId: randomUUID(), handle: 'h' }], unowned: [] }) }); + await expect(recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'r'), diagnosticsDir: join(root, 'd'), deps: d })).rejects.toThrow(/disk/); + expect(calls.some(c => c.startsWith('remove'))).toBe(false); + }); + it('keeps storage that matches no attempt, sweeps owned attempt directories, and reports unknown entries', async () => { + const { d: root, store, admit } = fixture(); + const attempt = admit(id(1)); store.markRunning(id(1), attempt.id); + const attempts = join(root, 'r', token, 'attempts'); mkdirSync(join(attempts, attempt.id), { recursive: true }); + mkdirSync(join(attempts, 'stray')); symlinkSync(root, join(attempts, randomUUID())); + const { d, calls } = deps({ recoverLeftovers: async () => ({ storage: [{ attemptId: randomUUID(), allocationId: randomUUID(), handle: 'orphan' }], unowned: [] }) }); + const report = await recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'r'), diagnosticsDir: join(root, 'd'), deps: d }); + expect(calls).not.toContain('remove:orphan'); + expect(report.unmatchedStorage).toHaveLength(1); + expect(existsSync(join(attempts, attempt.id))).toBe(false); + expect(report.unknownEntries).toHaveLength(2); + expect(existsSync(root)).toBe(true); + }); + it('fails closed on an unrecorded preparation until it is released, and refuses release while a process uses it', async () => { + const { d: root, store, admit } = fixture(); + const attempt = admit(id(1)); store.markPreparationStarting(id(1), attempt.id, Date.now()); + const runnerRoot = join(root, 'r'), dirPath = join(runnerRoot, token, 'attempts', attempt.id); mkdirSync(dirPath, { recursive: true }); + const run = () => recoverStartup({ store, runnerOwner: token, runnerRoot, diagnosticsDir: join(root, 'd'), deps: deps().d }); + await expect(run()).rejects.toBeInstanceOf(RecoveryBlocked); + expect(existsSync(dirPath)).toBe(true); + expect(() => releasePreparation({ store, runnerRoot, runnerOwner: token, attemptId: attempt.id, openFiles: () => ['4242'] })).toThrow(/still using/); + releasePreparation({ store, runnerRoot, runnerOwner: token, attemptId: attempt.id, openFiles: () => [] }); + expect(existsSync(dirPath)).toBe(false); + await expect(run()).resolves.toMatchObject({ finalized: [] }); + }); +}); diff --git a/web/cli.ts b/web/cli.ts index bce49e5..f7442f3 100644 --- a/web/cli.ts +++ b/web/cli.ts @@ -4,6 +4,7 @@ import { parseArgs } from 'node:util'; import { startServer } from './server.ts'; import { createDemo } from '../scripts/demo.ts'; import { requireSupportedNode } from '../runner/store.ts'; +import { acquireRunnerLock } from '../runner/recovery.ts'; requireSupportedNode(); const { values } = parseArgs({ options: { demo: { type:'boolean' }, directory:{type:'string'}, config:{type:'string'}, port:{type:'string'}, help:{type:'boolean'} } }); if (values.help || (!values.demo && !values.config)) { @@ -12,8 +13,16 @@ if (values.help || (!values.demo && !values.config)) { const port = Number(values.port ?? '4318'); if (!Number.isInteger(port) || port < 0 || port > 65535) throw new Error('Invalid port.'); const config = values.demo ? createDemo(values.directory ?? '.codeboost-local/demo') : JSON.parse(readFileSync(resolve(values.config!), 'utf8')); - const app = await startServer(config, port); + // Decision 1: one runner per database, held as an OS lock keyed by the database file's device and inode. + let lock: ReturnType; + try { lock = acquireRunnerLock(config.database); } + catch (error) { console.error(error instanceof Error ? error.message : String(error)); process.exit(1); } + let app: Awaited>; + try { app = await startServer(config, port); } + catch (error) { lock.release(); throw error; } + try { lock.verify(); } + catch (error) { await app.close(); lock.release(); throw error; } console.log(`Review ready: ${app.url}\nRepository: ${config.repository}\nDatabase: ${config.database}\nSource files are read-only. Press Ctrl+C to stop.`); let stopping=false; - for(const signal of ['SIGINT','SIGTERM'] as const) process.on(signal,()=>{if(!stopping){stopping=true;void app.close().then(()=>process.exit(0));}}); + for(const signal of ['SIGINT','SIGTERM'] as const) process.on(signal,()=>{if(!stopping){stopping=true;void app.close().then(()=>{lock.release();process.exit(0);});}}); }