|
| 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 | +export interface RunnerLock { |
| 33 | + readonly file: { dev: bigint; ino: bigint }; |
| 34 | + /** Step 4: the database path still names the locked file. Call after the Store opens. */ |
| 35 | + verify(): void; |
| 36 | + release(): void; |
| 37 | +} |
| 38 | +/** Steps 0–2: owner-only parent, identify (or create) the file with no-follow, then take the OS lock by device and inode. */ |
| 39 | +export function acquireRunnerLock(databasePath: string, options: { lockRoot?: string } = {}): RunnerLock { |
| 40 | + const absolute = resolve(databasePath), parent = realpathSync(dirname(absolute)), path = join(parent, absolute.slice(dirname(absolute).length + 1)); |
| 41 | + assertOwnerOnly(parent, 'The database directory'); |
| 42 | + assertLocal(parent); |
| 43 | + const link = lstatSync(path, { throwIfNoEntry: false }); |
| 44 | + if (link?.isSymbolicLink()) throw new Error('The database path must not be a symlink.'); |
| 45 | + let fd: number; |
| 46 | + try { fd = openSync(path, constants.O_RDWR | constants.O_NOFOLLOW); } |
| 47 | + catch (error) { |
| 48 | + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; |
| 49 | + try { fd = openSync(path, constants.O_RDWR | constants.O_CREAT | constants.O_EXCL | constants.O_NOFOLLOW, 0o600); } |
| 50 | + catch (race) { if ((race as NodeJS.ErrnoException).code !== 'EEXIST') throw race; fd = openSync(path, constants.O_RDWR | constants.O_NOFOLLOW); } |
| 51 | + } |
| 52 | + let db: DatabaseSync | undefined; |
| 53 | + try { |
| 54 | + const st = fstatSync(fd, { bigint: true }); |
| 55 | + if (!st.isFile() || st.nlink !== 1n) throw new Error('The database must be a regular file with exactly one name (no hard links).'); |
| 56 | + const lockRoot = options.lockRoot ?? join(homedir(), '.codeboost', 'locks'); |
| 57 | + mkdirSync(lockRoot, { recursive: true, mode: 0o700 }); |
| 58 | + assertOwnerOnly(lockRoot, 'The lock directory'); |
| 59 | + assertLocal(lockRoot); |
| 60 | + const { DatabaseSync } = createRequire(import.meta.url)('node:sqlite') as typeof import('node:sqlite'); |
| 61 | + db = new DatabaseSync(join(lockRoot, `${st.dev}-${st.ino}.runner-lock`), { timeout: 0 }); |
| 62 | + try { |
| 63 | + // EXCLUSIVE locking mode keeps the file lock after the first write until the connection closes; the OS drops it on exit or crash. |
| 64 | + db.exec('PRAGMA locking_mode=EXCLUSIVE; CREATE TABLE IF NOT EXISTS holder (id INTEGER PRIMARY KEY); INSERT OR REPLACE INTO holder VALUES (1);'); |
| 65 | + } catch (error) { |
| 66 | + if (/locked|busy/i.test((error as Error).message)) throw new LockHeld(); |
| 67 | + throw error; |
| 68 | + } |
| 69 | + const file = { dev: st.dev, ino: st.ino }, heldDb = db; |
| 70 | + let released = false; |
| 71 | + return { |
| 72 | + file, |
| 73 | + verify() { |
| 74 | + const now = lstatSync(path, { bigint: true }); |
| 75 | + if (now.isSymbolicLink() || !now.isFile() || now.dev !== file.dev || now.ino !== file.ino || now.nlink !== 1n) |
| 76 | + throw new Error('The database path changed while opening. Refusing to start.'); |
| 77 | + }, |
| 78 | + release() { if (released) return; released = true; heldDb.close(); closeSync(fd); }, |
| 79 | + }; |
| 80 | + } catch (error) { db?.close(); closeSync(fd); throw error; } |
| 81 | +} |
| 82 | + |
| 83 | +export interface ProcessControl { |
| 84 | + /** The group leader is alive and started at the recorded time (guards against PID reuse). */ |
| 85 | + isAlive(pgid: number, startedAt: number): boolean; |
| 86 | + /** SIGTERM the group, SIGKILL after the grace period, and resolve only once it has exited. */ |
| 87 | + terminate(pgid: number, graceMs: number): Promise<void>; |
| 88 | +} |
| 89 | +export const hostProcesses: ProcessControl = { |
| 90 | + isAlive(pgid, startedAt) { |
| 91 | + try { process.kill(-pgid, 0); } catch { return false; } |
| 92 | + try { |
| 93 | + const started = Date.parse(execFileSync('ps', ['-o', 'lstart=', '-p', String(pgid)], { encoding: 'utf8' }).trim()); |
| 94 | + return Number.isFinite(started) && Math.abs(started - startedAt) < 2_000; |
| 95 | + } catch { return true; } // Alive but unreadable: treat as ours and stop it (fail closed). |
| 96 | + }, |
| 97 | + async terminate(pgid, graceMs) { |
| 98 | + const alive = () => { try { process.kill(-pgid, 0); return true; } catch { return false; } }; |
| 99 | + try { process.kill(-pgid, 'SIGTERM'); } catch { return; } |
| 100 | + const until = Date.now() + graceMs; |
| 101 | + while (alive() && Date.now() < until) await new Promise(resolve => setTimeout(resolve, 50)); |
| 102 | + if (alive()) { try { process.kill(-pgid, 'SIGKILL'); } catch {} } |
| 103 | + while (alive()) await new Promise(resolve => setTimeout(resolve, 50)); |
| 104 | + }, |
| 105 | +}; |
| 106 | + |
| 107 | +export interface RecoveredStorage { readonly attemptId: string; readonly allocationId: string; readonly handle: unknown } |
| 108 | +export interface RecoveryDeps { |
| 109 | + /** D (#51): stop leftover agent containers, proxies and networks for this owner; keep task storage and return authenticated handles. */ |
| 110 | + recoverLeftovers(runnerOwner: string): Promise<{ storage: RecoveredStorage[]; unowned: string[] }>; |
| 111 | + /** D (#51, before F2): bounded diff of a recovered task volume; must stop its own work when the signal aborts. */ |
| 112 | + exportTaskDiff(handle: unknown, maxBytes: number, signal: AbortSignal): Promise<Buffer>; |
| 113 | + removeTaskFilesystems(handle: unknown): Promise<void>; |
| 114 | + /** F3: abort an interrupted rebase. Until F3 exists no rebase is ever recorded. */ |
| 115 | + abortRebase?(planKey: string, marker: unknown): Promise<void>; |
| 116 | + processes?: ProcessControl; |
| 117 | +} |
| 118 | +export interface RecoveryOptions { |
| 119 | + store: Store; runnerOwner: string; runnerRoot: string; diagnosticsDir: string; deps: RecoveryDeps; |
| 120 | + now?: () => number; exportDeadlineMs?: number; graceMs?: number; |
| 121 | +} |
| 122 | +export interface RecoveryReport { |
| 123 | + finalized: { attemptId: string; planKey: string; state: string; requeued: boolean }[]; |
| 124 | + requeue: string[]; removedDirectories: string[]; unknownEntries: string[]; unmatchedStorage: string[]; repairedMerges: string[]; |
| 125 | +} |
| 126 | +const EXPORT_LIMIT = 1024 * 1024; |
| 127 | + |
| 128 | +/** |
| 129 | + * Startup steps 2b–7 after the lock (step 1) and the Store open (2a). Throws on any step that must fail closed; |
| 130 | + * the caller then closes the Store, releases the lock and exits without opening the coordinator. |
| 131 | + */ |
| 132 | +export async function recoverStartup(o: RecoveryOptions): Promise<RecoveryReport> { |
| 133 | + const now = o.now ?? Date.now, processes = o.deps.processes ?? hostProcesses; |
| 134 | + if (!/^[0-9a-f]{32}$/.test(o.runnerOwner)) throw new Error('Invalid runner owner token.'); |
| 135 | + const interrupted = o.store.interruptedAttempts(); |
| 136 | + // Step 7's input is taken before finalization: an interrupted attempt that started preparation but never saved its group. |
| 137 | + const unowned = interrupted.filter(a => a.preparationStartedAt !== null && a.preparationPgid === null).map(a => a.id); |
| 138 | + // 2b. Stop leftover preparation before D's recovery or any storage work. |
| 139 | + for (const a of interrupted) if (a.preparationPgid !== null && a.preparationStartedAt !== null && processes.isAlive(a.preparationPgid, a.preparationStartedAt)) |
| 140 | + await processes.terminate(a.preparationPgid, o.graceMs ?? 5_000); |
| 141 | + // 2c/2d. D's recovery; a rejection propagates and stops startup. |
| 142 | + const recovered = await o.deps.recoverLeftovers(o.runnerOwner); |
| 143 | + 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); |
| 144 | + const matched = recovered.storage.filter(s => isUuidV4(s.attemptId) && o.store.attemptOwner(s.attemptId) !== null); |
| 145 | + const unmatchedStorage = recovered.storage.filter(s => !matched.includes(s)).map(s => s.attemptId); |
| 146 | + // 3. Export phase: stopped writable attempts only, outside any transaction, fixed names, bounded deadline. |
| 147 | + const exports: Record<string, { diagnosticRef?: string; failure?: string }> = {}; |
| 148 | + mkdirSync(o.diagnosticsDir, { recursive: true, mode: 0o700 }); |
| 149 | + for (const storage of matched) { |
| 150 | + const attempt = interrupted.find(a => a.id === storage.attemptId); |
| 151 | + if (!attempt || !WRITABLE_KINDS.includes(attempt.kind)) continue; |
| 152 | + const controller = new AbortController(), timer = setTimeout(() => controller.abort(new Error('Export timed out.')), o.exportDeadlineMs ?? 60_000); |
| 153 | + try { |
| 154 | + const diff = await Promise.race([o.deps.exportTaskDiff(storage.handle, EXPORT_LIMIT, controller.signal), |
| 155 | + new Promise<never>((_, reject) => controller.signal.addEventListener('abort', () => reject(controller.signal.reason), { once: true }))]); |
| 156 | + const file = join(o.diagnosticsDir, `${storage.attemptId}.diff`); |
| 157 | + writeFileSync(file, diff.subarray(0, EXPORT_LIMIT), { mode: 0o600 }); |
| 158 | + exports[storage.attemptId] = { diagnosticRef: file }; |
| 159 | + } catch (error) { exports[storage.attemptId] = { failure: error instanceof Error ? error.message : String(error) }; } |
| 160 | + finally { clearTimeout(timer); } |
| 161 | + } |
| 162 | + // 3. Finalization phase: one transaction; a failure stops startup. |
| 163 | + const finalized = o.store.recoverInterrupted(now(), exports); |
| 164 | + // 4. Rebases, storage removal (every matched handle), then attempt directories. |
| 165 | + for (const rebase of o.store.rebasesInProgress()) { |
| 166 | + if (!o.deps.abortRebase) throw new RecoveryBlocked('An interrupted rebase needs F3 to abort it', [rebase.planKey]); |
| 167 | + await o.deps.abortRebase(rebase.planKey, rebase.marker); |
| 168 | + } |
| 169 | + for (const storage of matched) await o.deps.removeTaskFilesystems(storage.handle); |
| 170 | + const removedDirectories: string[] = [], unknownEntries: string[] = []; |
| 171 | + const attemptsDir = join(o.runnerRoot, o.runnerOwner, 'attempts'); |
| 172 | + const rootDev = lstatSync(o.runnerRoot, { throwIfNoEntry: false })?.dev; |
| 173 | + for (const name of lstatSync(attemptsDir, { throwIfNoEntry: false })?.isDirectory() ? readdirSync(attemptsDir) : []) { |
| 174 | + const entry = join(attemptsDir, name), st = lstatSync(entry); |
| 175 | + const owned = st.isDirectory() && !st.isSymbolicLink() && st.dev === rootDev && isUuidV4(name) && o.store.attemptOwner(name) !== null; |
| 176 | + if (!owned) { unknownEntries.push(entry); continue; } |
| 177 | + if (unowned.includes(name)) continue; // step 7 keeps it for --release-preparation |
| 178 | + rmSync(entry, { recursive: true, force: true }); removedDirectories.push(entry); |
| 179 | + } |
| 180 | + // 5. Confirmed merges get their closed status and task-closed event. |
| 181 | + const repairedMerges = o.store.reconcileMergedTasks(); |
| 182 | + // 7. An unidentifiable preparation child may exist: fail closed until the user releases it. |
| 183 | + const blocked = [...unowned, ...o.store.unownedPreparations().filter(id => !unowned.includes(id))]; |
| 184 | + if (blocked.length) throw new RecoveryBlocked('Preparation started but its process was never recorded; stop it, then run --release-preparation', blocked); |
| 185 | + return { finalized, requeue: finalized.filter(f => f.requeued).map(f => f.planKey), removedDirectories, unknownEntries, unmatchedStorage, repairedMerges }; |
| 186 | +} |
| 187 | + |
| 188 | +/** --release-preparation: remove an attempt directory only when no process has a file open or a working directory in it. */ |
| 189 | +export function releasePreparation(o: { store: Store; runnerRoot: string; runnerOwner: string; attemptId: string; openFiles?: (dir: string) => string[] }): void { |
| 190 | + if (!isUuidV4(o.attemptId) || o.store.attemptOwner(o.attemptId) === null) throw new Error('Unknown attempt.'); |
| 191 | + const dir = join(o.runnerRoot, o.runnerOwner, 'attempts', o.attemptId), st = lstatSync(dir, { throwIfNoEntry: false }); |
| 192 | + if (st && (!st.isDirectory() || st.isSymbolicLink())) throw new Error('The attempt path is not a plain directory.'); |
| 193 | + if (st) { |
| 194 | + const users = (o.openFiles ?? hostOpenFiles)(dir); |
| 195 | + if (users.length) throw new RecoveryBlocked('A process is still using the attempt directory', users); |
| 196 | + rmSync(dir, { recursive: true, force: true }); |
| 197 | + } |
| 198 | + if (!o.store.clearPreparationMarker(o.attemptId)) throw new Error('The attempt is not waiting for release.'); |
| 199 | +} |
| 200 | +/** Processes with a file open or a working directory under dir. Throws if the check cannot run (fail closed). */ |
| 201 | +export function hostOpenFiles(dir: string): string[] { |
| 202 | + if (process.platform === 'linux') { |
| 203 | + const users: string[] = []; |
| 204 | + for (const pid of readdirSync('/proc').filter(name => /^\d+$/.test(name))) { |
| 205 | + const links: string[] = []; |
| 206 | + try { links.push(realpathSync(`/proc/${pid}/cwd`)); } catch {} |
| 207 | + try { for (const fd of readdirSync(`/proc/${pid}/fd`)) { try { links.push(realpathSync(`/proc/${pid}/fd/${fd}`)); } catch {} } } catch {} |
| 208 | + if (links.some(link => link === dir || link.startsWith(`${dir}/`))) users.push(pid); |
| 209 | + } |
| 210 | + return users; |
| 211 | + } |
| 212 | + try { return execFileSync('lsof', ['-t', '+D', dir], { encoding: 'utf8' }).split('\n').filter(Boolean); } |
| 213 | + catch (error) { |
| 214 | + const e = error as { status?: number; stdout?: string }; |
| 215 | + if (e.status === 1 && !e.stdout) return []; // lsof exits 1 when nothing matches |
| 216 | + throw new Error('Could not check which processes use the attempt directory. Refusing to release it.'); |
| 217 | + } |
| 218 | +} |
0 commit comments