Skip to content

Commit b1c6b1d

Browse files
mchwangclaude
andauthored
F1d: startup recovery and the single-runner lock (#59)
* F1d: startup recovery and the single-runner lock acquireRunnerLock: owner-only parent directory, no-follow identify or create, refuse hard links and symlinks, then an OS lock on ~/.codeboost/locks/<dev>-<ino>.runner-lock (SQLite exclusive locking) that the OS releases on exit or crash; verify() after the Store opens. The CLI takes the lock before starting and releases it after shutdown. recoverStartup: stop leftover preparation groups, D recovery (injected until transaction (settlement precedence, cancel and time-limit guards, requeue claim), storage removal, attempt-directory sweep, merged-task repair, and a fail-closed stop for unrecorded preparation. Store: runner owner token, preparation and allocation markers, requeue claim at admission, and releasePreparation with an open-files check. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Keep held runner locks referenced so GC cannot release them A caller that dropped the returned RunnerLock let its SQLite connection be garbage-collected, which released the OS lock while the runner was still alive. Found by a CI failure of the cross-process lock test on #60. The new test drops the lock and forces GC in a child, and fails without the fix. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Start the merge-repair recovery test from a task in review main (#62, #53) lets only a task in review begin a merge attempt, so the test puts its queued task in review before it records the merge. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Keep recovery fail-closed: unowned directories, unreadable start times, symlinked roots, unstopped exports - The attempt-directory sweep skips every unowned preparation, including one an earlier startup already finalized. Before, the second startup deleted its directory while step 7 still blocked on it. - ps runs with LC_ALL=C, and a start time Date.parse cannot read counts as ours, so a live preparation group is still stopped. - hostOpenFiles compares against the resolved directory. On Linux a symlink in the runner root made an in-use directory look unused. - A timed-out export is awaited, for up to the grace period, before its task storage is removed. An export that does not stop blocks startup. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1 parent b091bbd commit b1c6b1d

4 files changed

Lines changed: 673 additions & 5 deletions

File tree

‎runner/recovery.ts‎

Lines changed: 242 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,242 @@
1+
import { closeSync, constants, fstatSync, lstatSync, mkdirSync, openSync, readdirSync, realpathSync, rmSync, statSync, statfsSync, writeFileSync } from 'node:fs';
2+
import { homedir } from 'node:os';
3+
import { dirname, join, resolve } from 'node:path';
4+
import { execFileSync } from 'node:child_process';
5+
import { createRequire } from 'node:module';
6+
import type { DatabaseSync } from 'node:sqlite';
7+
import type { Store } from './store.ts';
8+
import { WRITABLE_KINDS, isUuidV4 } from './lifecycle.ts';
9+
10+
/**
11+
* Startup recovery and the single-runner lock. See docs/implementation/runner-lifecycle.md,
12+
* "Startup recovery" and decision 1. D's recovery, export and removal are injected until #51 provides them.
13+
*/
14+
export class LockHeld extends Error { constructor() { super('Another codeboost runner is using this database.'); } }
15+
export class RecoveryBlocked extends Error {
16+
readonly items: string[];
17+
constructor(message: string, items: string[]) { super(`${message}: ${items.join(', ')}`); this.items = items; }
18+
}
19+
20+
/** Linux statfs magic numbers for network filesystems, where POSIX locks are unreliable. */
21+
const NETWORK_FILESYSTEMS = new Set([0x6969 /* NFS */, 0x517b /* SMB */, 0xff534d42 /* CIFS */, 0xfe534d42 /* SMB2 */, 0x65735546 /* FUSE */]);
22+
function assertOwnerOnly(path: string, what: string): void {
23+
const st = statSync(path);
24+
if (!st.isDirectory() || (process.getuid && st.uid !== process.getuid()) || (st.mode & 0o022) !== 0)
25+
throw new Error(`${what} ${path} must be a directory owned by you and not writable by group or others.`);
26+
}
27+
function assertLocal(path: string): void {
28+
if (process.platform !== 'linux') return; // macOS statfs types are not stable enough to classify; see the PR notes.
29+
if (NETWORK_FILESYSTEMS.has(Number(statfsSync(path).type))) throw new Error(`${path} is on a network filesystem; the runner lock needs a local filesystem.`);
30+
}
31+
32+
/**
33+
* Strong references to every held lock connection. Without this, a caller that drops the returned RunnerLock lets the
34+
* connection be garbage-collected, and closing it silently releases the OS lock while the runner is still alive.
35+
*/
36+
const heldLocks = new Set<DatabaseSync>();
37+
export interface RunnerLock {
38+
readonly file: { dev: bigint; ino: bigint };
39+
/** Step 4: the database path still names the locked file. Call after the Store opens. */
40+
verify(): void;
41+
release(): void;
42+
}
43+
/** Steps 0–2: owner-only parent, identify (or create) the file with no-follow, then take the OS lock by device and inode. */
44+
export function acquireRunnerLock(databasePath: string, options: { lockRoot?: string } = {}): RunnerLock {
45+
const absolute = resolve(databasePath), parent = realpathSync(dirname(absolute)), path = join(parent, absolute.slice(dirname(absolute).length + 1));
46+
assertOwnerOnly(parent, 'The database directory');
47+
assertLocal(parent);
48+
const link = lstatSync(path, { throwIfNoEntry: false });
49+
if (link?.isSymbolicLink()) throw new Error('The database path must not be a symlink.');
50+
let fd: number;
51+
try { fd = openSync(path, constants.O_RDWR | constants.O_NOFOLLOW); }
52+
catch (error) {
53+
if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error;
54+
try { fd = openSync(path, constants.O_RDWR | constants.O_CREAT | constants.O_EXCL | constants.O_NOFOLLOW, 0o600); }
55+
catch (race) { if ((race as NodeJS.ErrnoException).code !== 'EEXIST') throw race; fd = openSync(path, constants.O_RDWR | constants.O_NOFOLLOW); }
56+
}
57+
let db: DatabaseSync | undefined;
58+
try {
59+
const st = fstatSync(fd, { bigint: true });
60+
if (!st.isFile() || st.nlink !== 1n) throw new Error('The database must be a regular file with exactly one name (no hard links).');
61+
const lockRoot = options.lockRoot ?? join(homedir(), '.codeboost', 'locks');
62+
mkdirSync(lockRoot, { recursive: true, mode: 0o700 });
63+
assertOwnerOnly(lockRoot, 'The lock directory');
64+
assertLocal(lockRoot);
65+
const { DatabaseSync } = createRequire(import.meta.url)('node:sqlite') as typeof import('node:sqlite');
66+
db = new DatabaseSync(join(lockRoot, `${st.dev}-${st.ino}.runner-lock`), { timeout: 0 });
67+
try {
68+
// EXCLUSIVE locking mode keeps the file lock after the first write until the connection closes; the OS drops it on exit or crash.
69+
db.exec('PRAGMA locking_mode=EXCLUSIVE; CREATE TABLE IF NOT EXISTS holder (id INTEGER PRIMARY KEY); INSERT OR REPLACE INTO holder VALUES (1);');
70+
} catch (error) {
71+
if (/locked|busy/i.test((error as Error).message)) throw new LockHeld();
72+
throw error;
73+
}
74+
const file = { dev: st.dev, ino: st.ino }, heldDb = db;
75+
heldLocks.add(heldDb);
76+
let released = false;
77+
return {
78+
file,
79+
verify() {
80+
const now = lstatSync(path, { bigint: true });
81+
if (now.isSymbolicLink() || !now.isFile() || now.dev !== file.dev || now.ino !== file.ino || now.nlink !== 1n)
82+
throw new Error('The database path changed while opening. Refusing to start.');
83+
},
84+
release() { if (released) return; released = true; heldLocks.delete(heldDb); heldDb.close(); closeSync(fd); },
85+
};
86+
} catch (error) { db?.close(); closeSync(fd); throw error; }
87+
}
88+
89+
export interface ProcessControl {
90+
/** The group leader is alive and started at the recorded time (guards against PID reuse). */
91+
isAlive(pgid: number, startedAt: number): boolean;
92+
/** SIGTERM the group, SIGKILL after the grace period, and resolve only once it has exited. */
93+
terminate(pgid: number, graceMs: number): Promise<void>;
94+
}
95+
export const hostProcesses: ProcessControl = {
96+
isAlive(pgid, startedAt) {
97+
try { process.kill(-pgid, 0); } catch { return false; }
98+
try { return startTimeMatches(execFileSync('ps', ['-o', 'lstart=', '-p', String(pgid)], { encoding: 'utf8', env: { ...process.env, LC_ALL: 'C' } }), startedAt); }
99+
catch { return true; } // Alive but unreadable: treat as ours and stop it (fail closed).
100+
},
101+
async terminate(pgid, graceMs) {
102+
const alive = () => { try { process.kill(-pgid, 0); return true; } catch { return false; } };
103+
try { process.kill(-pgid, 'SIGTERM'); } catch { return; }
104+
const until = Date.now() + graceMs;
105+
while (alive() && Date.now() < until) await new Promise(resolve => setTimeout(resolve, 50));
106+
if (alive()) { try { process.kill(-pgid, 'SIGKILL'); } catch {} }
107+
while (alive()) await new Promise(resolve => setTimeout(resolve, 50));
108+
},
109+
};
110+
111+
/** Whether `ps -o lstart=` output names the recorded start time. An unreadable time counts as a match (fail closed). */
112+
export function startTimeMatches(lstart: string, startedAt: number): boolean {
113+
const started = Date.parse(lstart.trim());
114+
return !Number.isFinite(started) || Math.abs(started - startedAt) < 2_000;
115+
}
116+
117+
export interface RecoveredStorage { readonly attemptId: string; readonly allocationId: string; readonly handle: unknown }
118+
export interface RecoveryDeps {
119+
/** D (#51): stop leftover agent containers, proxies and networks for this owner; keep task storage and return authenticated handles. */
120+
recoverLeftovers(runnerOwner: string): Promise<{ storage: RecoveredStorage[]; unowned: string[] }>;
121+
/** D (#51, before F2): bounded diff of a recovered task volume; must stop its own work when the signal aborts. */
122+
exportTaskDiff(handle: unknown, maxBytes: number, signal: AbortSignal): Promise<Buffer>;
123+
removeTaskFilesystems(handle: unknown): Promise<void>;
124+
/** F3: abort an interrupted rebase. Until F3 exists no rebase is ever recorded. */
125+
abortRebase?(planKey: string, marker: unknown): Promise<void>;
126+
processes?: ProcessControl;
127+
}
128+
export interface RecoveryOptions {
129+
store: Store; runnerOwner: string; runnerRoot: string; diagnosticsDir: string; deps: RecoveryDeps;
130+
now?: () => number; exportDeadlineMs?: number; graceMs?: number;
131+
}
132+
export interface RecoveryReport {
133+
finalized: { attemptId: string; planKey: string; state: string; requeued: boolean }[];
134+
requeue: string[]; removedDirectories: string[]; unknownEntries: string[]; unmatchedStorage: string[]; repairedMerges: string[];
135+
}
136+
const EXPORT_LIMIT = 1024 * 1024;
137+
138+
/**
139+
* Startup steps 2b–7 after the lock (step 1) and the Store open (2a). Throws on any step that must fail closed;
140+
* the caller then closes the Store, releases the lock and exits without opening the coordinator.
141+
*/
142+
export async function recoverStartup(o: RecoveryOptions): Promise<RecoveryReport> {
143+
const now = o.now ?? Date.now, processes = o.deps.processes ?? hostProcesses;
144+
if (!/^[0-9a-f]{32}$/.test(o.runnerOwner)) throw new Error('Invalid runner owner token.');
145+
const interrupted = o.store.interruptedAttempts();
146+
// Step 7's input is taken before finalization: an interrupted attempt that started preparation but never saved its group.
147+
const unowned = interrupted.filter(a => a.preparationStartedAt !== null && a.preparationPgid === null).map(a => a.id);
148+
// 2b. Stop leftover preparation before D's recovery or any storage work.
149+
for (const a of interrupted) if (a.preparationPgid !== null && a.preparationStartedAt !== null && processes.isAlive(a.preparationPgid, a.preparationStartedAt))
150+
await processes.terminate(a.preparationPgid, o.graceMs ?? 5_000);
151+
// 2c/2d. D's recovery; a rejection propagates and stops startup.
152+
const recovered = await o.deps.recoverLeftovers(o.runnerOwner);
153+
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);
154+
const matched = recovered.storage.filter(s => isUuidV4(s.attemptId) && o.store.attemptOwner(s.attemptId) !== null);
155+
const unmatchedStorage = recovered.storage.filter(s => !matched.includes(s)).map(s => s.attemptId);
156+
// 3. Export phase: stopped writable attempts only, outside any transaction, fixed names, bounded deadline.
157+
const exports: Record<string, { diagnosticRef?: string; failure?: string }> = {};
158+
mkdirSync(o.diagnosticsDir, { recursive: true, mode: 0o700 });
159+
for (const storage of matched) {
160+
const attempt = interrupted.find(a => a.id === storage.attemptId);
161+
if (!attempt || !WRITABLE_KINDS.includes(attempt.kind)) continue;
162+
const controller = new AbortController(), timer = setTimeout(() => controller.abort(new Error('Export timed out.')), o.exportDeadlineMs ?? 60_000);
163+
const exporting = o.deps.exportTaskDiff(storage.handle, EXPORT_LIMIT, controller.signal);
164+
try {
165+
const diff = await Promise.race([exporting,
166+
new Promise<never>((_, reject) => controller.signal.addEventListener('abort', () => reject(controller.signal.reason), { once: true }))]);
167+
const file = join(o.diagnosticsDir, `${storage.attemptId}.diff`);
168+
writeFileSync(file, diff.subarray(0, EXPORT_LIMIT), { mode: 0o600 });
169+
exports[storage.attemptId] = { diagnosticRef: file };
170+
} catch (error) { exports[storage.attemptId] = { failure: error instanceof Error ? error.message : String(error) }; }
171+
finally { clearTimeout(timer); }
172+
// A timed-out export must stop before step 4 removes its storage. D stops on abort; if it does not, fail closed.
173+
if (controller.signal.aborted) await stopped(exporting, o.graceMs ?? 5_000, storage.attemptId);
174+
}
175+
// 3. Finalization phase: one transaction; a failure stops startup.
176+
const finalized = o.store.recoverInterrupted(now(), exports);
177+
// 4. Rebases, storage removal (every matched handle), then attempt directories.
178+
for (const rebase of o.store.rebasesInProgress()) {
179+
if (!o.deps.abortRebase) throw new RecoveryBlocked('An interrupted rebase needs F3 to abort it', [rebase.planKey]);
180+
await o.deps.abortRebase(rebase.planKey, rebase.marker);
181+
}
182+
for (const storage of matched) await o.deps.removeTaskFilesystems(storage.handle);
183+
// Step 7's set, read after finalization so it also holds preparations an earlier startup already finalized.
184+
const blocked = [...new Set([...unowned, ...o.store.unownedPreparations()])];
185+
const removedDirectories: string[] = [], unknownEntries: string[] = [];
186+
const attemptsDir = join(o.runnerRoot, o.runnerOwner, 'attempts');
187+
const rootDev = lstatSync(o.runnerRoot, { throwIfNoEntry: false })?.dev;
188+
for (const name of lstatSync(attemptsDir, { throwIfNoEntry: false })?.isDirectory() ? readdirSync(attemptsDir) : []) {
189+
const entry = join(attemptsDir, name), st = lstatSync(entry);
190+
const owned = st.isDirectory() && !st.isSymbolicLink() && st.dev === rootDev && isUuidV4(name) && o.store.attemptOwner(name) !== null;
191+
if (!owned) { unknownEntries.push(entry); continue; }
192+
if (blocked.includes(name)) continue; // step 7 keeps it for --release-preparation
193+
rmSync(entry, { recursive: true, force: true }); removedDirectories.push(entry);
194+
}
195+
// 5. Confirmed merges get their closed status and task-closed event.
196+
const repairedMerges = o.store.reconcileMergedTasks();
197+
// 7. An unidentifiable preparation child may exist: fail closed until the user releases it.
198+
if (blocked.length) throw new RecoveryBlocked('Preparation started but its process was never recorded; stop it, then run --release-preparation', blocked);
199+
return { finalized, requeue: finalized.filter(f => f.requeued).map(f => f.planKey), removedDirectories, unknownEntries, unmatchedStorage, repairedMerges };
200+
}
201+
202+
/** Wait for an aborted export to settle, up to graceMs; otherwise stop startup before its storage is touched. */
203+
async function stopped(exporting: Promise<unknown>, graceMs: number, attemptId: string): Promise<void> {
204+
let timer: ReturnType<typeof setTimeout> | undefined;
205+
const settled = await Promise.race([exporting.then(() => true, () => true), new Promise<false>(resolve => { timer = setTimeout(() => resolve(false), graceMs); })]);
206+
clearTimeout(timer);
207+
if (!settled) throw new RecoveryBlocked('A partial-output export did not stop after its deadline; its task storage was left in place', [attemptId]);
208+
}
209+
210+
/** --release-preparation: remove an attempt directory only when no process has a file open or a working directory in it. */
211+
export function releasePreparation(o: { store: Store; runnerRoot: string; runnerOwner: string; attemptId: string; openFiles?: (dir: string) => string[] }): void {
212+
if (!isUuidV4(o.attemptId) || o.store.attemptOwner(o.attemptId) === null) throw new Error('Unknown attempt.');
213+
const dir = join(o.runnerRoot, o.runnerOwner, 'attempts', o.attemptId), st = lstatSync(dir, { throwIfNoEntry: false });
214+
if (st && (!st.isDirectory() || st.isSymbolicLink())) throw new Error('The attempt path is not a plain directory.');
215+
if (st) {
216+
const users = (o.openFiles ?? hostOpenFiles)(dir);
217+
if (users.length) throw new RecoveryBlocked('A process is still using the attempt directory', users);
218+
rmSync(dir, { recursive: true, force: true });
219+
}
220+
if (!o.store.clearPreparationMarker(o.attemptId)) throw new Error('The attempt is not waiting for release.');
221+
}
222+
/** Processes with a file open or a working directory under dir. Throws if the check cannot run (fail closed). */
223+
export function hostOpenFiles(dir: string): string[] {
224+
// /proc links and lsof report resolved paths, so compare against the resolved directory.
225+
dir = realpathSync(dir);
226+
if (process.platform === 'linux') {
227+
const users: string[] = [];
228+
for (const pid of readdirSync('/proc').filter(name => /^\d+$/.test(name))) {
229+
const links: string[] = [];
230+
try { links.push(realpathSync(`/proc/${pid}/cwd`)); } catch {}
231+
try { for (const fd of readdirSync(`/proc/${pid}/fd`)) { try { links.push(realpathSync(`/proc/${pid}/fd/${fd}`)); } catch {} } } catch {}
232+
if (links.some(link => link === dir || link.startsWith(`${dir}/`))) users.push(pid);
233+
}
234+
return users;
235+
}
236+
try { return execFileSync('lsof', ['-t', '+D', dir], { encoding: 'utf8' }).split('\n').filter(Boolean); }
237+
catch (error) {
238+
const e = error as { status?: number; stdout?: string };
239+
if (e.status === 1 && !e.stdout) return []; // lsof exits 1 when nothing matches
240+
throw new Error('Could not check which processes use the attempt directory. Refusing to release it.');
241+
}
242+
}

0 commit comments

Comments
 (0)