Skip to content

Commit 9f8434d

Browse files
mchwangclaude
andcommitted
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>
1 parent 6b9bb4d commit 9f8434d

2 files changed

Lines changed: 63 additions & 10 deletions

File tree

‎runner/recovery.ts‎

Lines changed: 25 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -95,10 +95,8 @@ export interface ProcessControl {
9595
export const hostProcesses: ProcessControl = {
9696
isAlive(pgid, startedAt) {
9797
try { process.kill(-pgid, 0); } catch { return false; }
98-
try {
99-
const started = Date.parse(execFileSync('ps', ['-o', 'lstart=', '-p', String(pgid)], { encoding: 'utf8' }).trim());
100-
return Number.isFinite(started) && Math.abs(started - startedAt) < 2_000;
101-
} catch { return true; } // Alive but unreadable: treat as ours and stop it (fail closed).
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).
102100
},
103101
async terminate(pgid, graceMs) {
104102
const alive = () => { try { process.kill(-pgid, 0); return true; } catch { return false; } };
@@ -110,6 +108,12 @@ export const hostProcesses: ProcessControl = {
110108
},
111109
};
112110

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+
113117
export interface RecoveredStorage { readonly attemptId: string; readonly allocationId: string; readonly handle: unknown }
114118
export interface RecoveryDeps {
115119
/** D (#51): stop leftover agent containers, proxies and networks for this owner; keep task storage and return authenticated handles. */
@@ -156,14 +160,17 @@ export async function recoverStartup(o: RecoveryOptions): Promise<RecoveryReport
156160
const attempt = interrupted.find(a => a.id === storage.attemptId);
157161
if (!attempt || !WRITABLE_KINDS.includes(attempt.kind)) continue;
158162
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);
159164
try {
160-
const diff = await Promise.race([o.deps.exportTaskDiff(storage.handle, EXPORT_LIMIT, controller.signal),
165+
const diff = await Promise.race([exporting,
161166
new Promise<never>((_, reject) => controller.signal.addEventListener('abort', () => reject(controller.signal.reason), { once: true }))]);
162167
const file = join(o.diagnosticsDir, `${storage.attemptId}.diff`);
163168
writeFileSync(file, diff.subarray(0, EXPORT_LIMIT), { mode: 0o600 });
164169
exports[storage.attemptId] = { diagnosticRef: file };
165170
} catch (error) { exports[storage.attemptId] = { failure: error instanceof Error ? error.message : String(error) }; }
166171
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);
167174
}
168175
// 3. Finalization phase: one transaction; a failure stops startup.
169176
const finalized = o.store.recoverInterrupted(now(), exports);
@@ -173,24 +180,33 @@ export async function recoverStartup(o: RecoveryOptions): Promise<RecoveryReport
173180
await o.deps.abortRebase(rebase.planKey, rebase.marker);
174181
}
175182
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()])];
176185
const removedDirectories: string[] = [], unknownEntries: string[] = [];
177186
const attemptsDir = join(o.runnerRoot, o.runnerOwner, 'attempts');
178187
const rootDev = lstatSync(o.runnerRoot, { throwIfNoEntry: false })?.dev;
179188
for (const name of lstatSync(attemptsDir, { throwIfNoEntry: false })?.isDirectory() ? readdirSync(attemptsDir) : []) {
180189
const entry = join(attemptsDir, name), st = lstatSync(entry);
181190
const owned = st.isDirectory() && !st.isSymbolicLink() && st.dev === rootDev && isUuidV4(name) && o.store.attemptOwner(name) !== null;
182191
if (!owned) { unknownEntries.push(entry); continue; }
183-
if (unowned.includes(name)) continue; // step 7 keeps it for --release-preparation
192+
if (blocked.includes(name)) continue; // step 7 keeps it for --release-preparation
184193
rmSync(entry, { recursive: true, force: true }); removedDirectories.push(entry);
185194
}
186195
// 5. Confirmed merges get their closed status and task-closed event.
187196
const repairedMerges = o.store.reconcileMergedTasks();
188197
// 7. An unidentifiable preparation child may exist: fail closed until the user releases it.
189-
const blocked = [...unowned, ...o.store.unownedPreparations().filter(id => !unowned.includes(id))];
190198
if (blocked.length) throw new RecoveryBlocked('Preparation started but its process was never recorded; stop it, then run --release-preparation', blocked);
191199
return { finalized, requeue: finalized.filter(f => f.requeued).map(f => f.planKey), removedDirectories, unknownEntries, unmatchedStorage, repairedMerges };
192200
}
193201

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+
194210
/** --release-preparation: remove an attempt directory only when no process has a file open or a working directory in it. */
195211
export function releasePreparation(o: { store: Store; runnerRoot: string; runnerOwner: string; attemptId: string; openFiles?: (dir: string) => string[] }): void {
196212
if (!isUuidV4(o.attemptId) || o.store.attemptOwner(o.attemptId) === null) throw new Error('Unknown attempt.');
@@ -205,6 +221,8 @@ export function releasePreparation(o: { store: Store; runnerRoot: string; runner
205221
}
206222
/** Processes with a file open or a working directory under dir. Throws if the check cannot run (fail closed). */
207223
export function hostOpenFiles(dir: string): string[] {
224+
// /proc links and lsof report resolved paths, so compare against the resolved directory.
225+
dir = realpathSync(dir);
208226
if (process.platform === 'linux') {
209227
const users: string[] = [];
210228
for (const pid of readdirSync('/proc').filter(name => /^\d+$/.test(name))) {

‎test/runner-recovery.test.ts‎

Lines changed: 38 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ import { fileURLToPath } from 'node:url';
88
import { DatabaseSync } from 'node:sqlite';
99
import { afterEach, describe, expect, it, vi } from 'vitest';
1010
import { Store } from '../runner/store.ts';
11-
import { LockHeld, RecoveryBlocked, acquireRunnerLock, recoverStartup, releasePreparation, type RecoveryDeps, type RunnerLock } from '../runner/recovery.ts';
11+
import { LockHeld, RecoveryBlocked, acquireRunnerLock, hostOpenFiles, recoverStartup, releasePreparation, startTimeMatches, type RecoveryDeps, type RunnerLock } from '../runner/recovery.ts';
1212
import type { PlanIdentity } from '../core/identity.ts';
1313
import type { Plan, PlanContext } from '../core/plan.ts';
1414

@@ -218,12 +218,27 @@ describe('startup recovery sequence', () => {
218218
const attempt = admit(id(1)); store.markRunning(id(1), attempt.id);
219219
const { d, calls } = deps({
220220
recoverLeftovers: async () => ({ storage: [{ attemptId: attempt.id, allocationId: randomUUID(), handle: 'h' }], unowned: [] }),
221-
exportTaskDiff: () => new Promise(() => undefined),
221+
// D's export stops a moment after its signal aborts.
222+
exportTaskDiff: (_h, _m, signal) => new Promise((_, reject) => signal.addEventListener('abort', () => setTimeout(() => { calls.push('export-stopped'); reject(signal.reason); }, 40), { once: true })),
222223
});
223224
await recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'r'), diagnosticsDir: join(root, 'd'), deps: d, exportDeadlineMs: 30 });
224225
expect(store.getAttempt(id(1), attempt.id)).toMatchObject({ state: 'failed', diagnosticRef: null });
225226
expect(store.getAttempt(id(1), attempt.id).diagnostic).toMatch(/Partial output could not be exported: Export timed out/);
226-
expect(calls).toContain('remove:h');
227+
// The storage is removed only after the export stopped using it.
228+
expect(calls.indexOf('export-stopped')).toBeGreaterThan(-1);
229+
expect(calls.indexOf('remove:h')).toBeGreaterThan(calls.indexOf('export-stopped'));
230+
});
231+
it('stops startup, leaving storage and the attempt untouched, when a timed-out export does not stop', async () => {
232+
const { d: root, store, admit } = fixture();
233+
const attempt = admit(id(1)); store.markRunning(id(1), attempt.id);
234+
const { d, calls } = deps({
235+
recoverLeftovers: async () => ({ storage: [{ attemptId: attempt.id, allocationId: randomUUID(), handle: 'h' }], unowned: [] }),
236+
exportTaskDiff: () => new Promise(() => undefined),
237+
});
238+
await expect(recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'r'), diagnosticsDir: join(root, 'd'), deps: d, exportDeadlineMs: 30, graceMs: 30 }))
239+
.rejects.toThrow(/did not stop/);
240+
expect(calls.some(c => c.startsWith('remove'))).toBe(false);
241+
expect(store.getAttempt(id(1), attempt.id).state).toBe('running');
227242
});
228243
it('removes nothing when the finalization transaction fails', async () => {
229244
const { d: root, store, admit } = fixture();
@@ -253,9 +268,29 @@ describe('startup recovery sequence', () => {
253268
const run = () => recoverStartup({ store, runnerOwner: token, runnerRoot, diagnosticsDir: join(root, 'd'), deps: deps().d });
254269
await expect(run()).rejects.toBeInstanceOf(RecoveryBlocked);
255270
expect(existsSync(dirPath)).toBe(true);
271+
// A second startup finds the attempt already finalized, and still keeps its directory for the release check.
272+
await expect(run()).rejects.toBeInstanceOf(RecoveryBlocked);
273+
expect(existsSync(dirPath)).toBe(true);
256274
expect(() => releasePreparation({ store, runnerRoot, runnerOwner: token, attemptId: attempt.id, openFiles: () => ['4242'] })).toThrow(/still using/);
257275
releasePreparation({ store, runnerRoot, runnerOwner: token, attemptId: attempt.id, openFiles: () => [] });
258276
expect(existsSync(dirPath)).toBe(false);
259277
await expect(run()).resolves.toMatchObject({ finalized: [] });
260278
});
261279
});
280+
281+
describe('host process checks', () => {
282+
it('treats a start time it cannot read as a match, so a live group is still stopped', () => {
283+
const at = Date.parse('Sat Sep 26 02:01:04 2026');
284+
expect(startTimeMatches('Sat Sep 26 02:01:04 2026\n', at)).toBe(true);
285+
expect(startTimeMatches('Sat Sep 26 02:01:04 2026', at + 60_000)).toBe(false);
286+
// procps under a Japanese locale; Date.parse cannot read it.
287+
expect(startTimeMatches('土 9月 26 02:01:04 2026', at + 60_000)).toBe(true);
288+
});
289+
it('finds a process working in the attempt directory when the path goes through a symlink', async () => {
290+
const root = dir(), real = join(root, 'real'), attempt = join(real, 'attempt');
291+
mkdirSync(attempt, { recursive: true }); symlinkSync(real, join(root, 'link'));
292+
const child = spawn('sleep', ['30'], { cwd: attempt, stdio: 'ignore' }); children.push(child);
293+
await once(child, 'spawn');
294+
expect(hostOpenFiles(join(root, 'link', 'attempt'))).toContain(String(child.pid));
295+
});
296+
});

0 commit comments

Comments
 (0)