Skip to content

Commit ad41511

Browse files
goyameghgoyamegh
committed
fix(runs): codex-review hardening for checkpoint-resume
- runLivenessAgeMs uses max(heartbeat, resumed, created) — a fresh resume claim counts as liveness immediately; previously the dead server's stale heartbeatAt left a just-resumed run looking orphaned (double-resume window) - resume claim: stamp heartbeatAt + a resumeToken nonce, re-read and abort with an SSE error if another server won the claim race (no cross-server CAS primitive exists; write-then-verify shrinks the window to ~one refresh) - partial source re-resolution: only reset test cases that will actually execute; ids the sources no longer resolve keep their failed-with-note entry and are surfaced as missingCount/missingTestCaseIds on started - boot recovery: two-phase scan-then-mutate — offset pagination over the shrinking status:running result set skipped stale runs beyond page 1 - PUT upsert route: full-document replace via create() (update() is now a doc-merge; omitted nested results keys would have survived an import) - file adapter: per-run write serialization for update/updateResult - tests: liveness max-semantics regression, double-resume 409 integration Signed-off-by: Megha Goyal <goyamegh@amazon.com>
1 parent 9dd8031 commit ad41511

6 files changed

Lines changed: 187 additions & 63 deletions

File tree

‎server/adapters/file/StorageModule.ts‎

Lines changed: 29 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -825,6 +825,21 @@ export class FileSessionMetadataOperations implements ISessionMetadataOperations
825825

826826
class FileEvaluationRunOperations implements IEvaluationRunOperations {
827827
private readonly dir: string;
828+
/**
829+
* Per-run write serialization. update()/updateResult() are
830+
* read-modify-write with an `await` between read and write — with
831+
* run.concurrency > 1 two in-flight updates could interleave at that
832+
* boundary and the second write clobbers the first (lost per-test-case
833+
* results). Chain writes per run id so they apply in order.
834+
*/
835+
private writeQueues = new Map<string, Promise<unknown>>();
836+
837+
private serialized<T>(id: string, op: () => Promise<T>): Promise<T> {
838+
const prev = this.writeQueues.get(id) ?? Promise.resolve();
839+
const next = prev.then(op, op);
840+
this.writeQueues.set(id, next.catch(() => {}));
841+
return next;
842+
}
828843

829844
constructor(baseDir: string) {
830845
// Stored in same directory as benchmarks (same "index" concept)
@@ -855,11 +870,13 @@ class FileEvaluationRunOperations implements IEvaluationRunOperations {
855870
}
856871

857872
async update(id: string, updates: Partial<EvaluationRun>): Promise<EvaluationRun> {
858-
const existing = await this.getById(id);
859-
if (!existing) throw new Error(`Evaluation run ${id} not found`);
860-
const updated = { ...existing, ...updates } as EvaluationRun;
861-
writeJsonFile(path.join(this.dir, `${id}.json`), updated);
862-
return updated;
873+
return this.serialized(id, async () => {
874+
const existing = await this.getById(id);
875+
if (!existing) throw new Error(`Evaluation run ${id} not found`);
876+
const updated = { ...existing, ...updates } as EvaluationRun;
877+
writeJsonFile(path.join(this.dir, `${id}.json`), updated);
878+
return updated;
879+
});
863880
}
864881

865882
async delete(id: string): Promise<{ deleted: boolean }> {
@@ -910,11 +927,13 @@ class FileEvaluationRunOperations implements IEvaluationRunOperations {
910927
status: RunResultStatus;
911928
error?: string;
912929
}): Promise<boolean> {
913-
const existing = await this.getById(runId);
914-
if (!existing) return false;
915-
existing.results[testCaseId] = result;
916-
writeJsonFile(path.join(this.dir, `${runId}.json`), existing);
917-
return true;
930+
return this.serialized(runId, async () => {
931+
const existing = await this.getById(runId);
932+
if (!existing) return false;
933+
existing.results[testCaseId] = result;
934+
writeJsonFile(path.join(this.dir, `${runId}.json`), existing);
935+
return true;
936+
});
918937
}
919938
}
920939

‎server/routes/storage/evaluationRuns.ts‎

Lines changed: 54 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -61,9 +61,18 @@ export function runStaleAfterMs(): number {
6161
return Number.isFinite(parsed) && parsed > 0 ? parsed : 60 * 60 * 1000; // 1h
6262
}
6363

64-
/** Age of the run's most recent liveness signal (heartbeat > resumed > created). */
64+
/**
65+
* Age of the run's most recent liveness signal. Uses the MOST RECENT of
66+
* heartbeat/resumed/created (not a priority order): after a resume claims an
67+
* orphan, `resumedAt` is newer than the dead server's last `heartbeatAt`, and
68+
* the claim must count as liveness immediately.
69+
*/
6570
export function runLivenessAgeMs(run: Pick<EvaluationRun, 'createdAt' | 'resumedAt' | 'heartbeatAt'>, now = Date.now()): number {
66-
const last = new Date(run.heartbeatAt || run.resumedAt || run.createdAt || 0).getTime();
71+
const last = Math.max(
72+
...[run.heartbeatAt, run.resumedAt, run.createdAt]
73+
.map((t) => (t ? new Date(t).getTime() : NaN))
74+
.filter((t) => Number.isFinite(t) && t > 0)
75+
);
6776
return Number.isFinite(last) && last > 0 ? now - last : Infinity;
6877
}
6978

@@ -396,29 +405,62 @@ router.post('/api/storage/evaluation-runs/:id/resume', async (req: Request, res:
396405
return;
397406
}
398407

399-
// Reset the resumable results to pending and flip the run back to running.
408+
// Reset ONLY the test cases we will actually execute. Resumable ids the
409+
// sources no longer resolve (test case deleted, benchmark membership
410+
// changed) keep their existing failed-with-note entry instead of being
411+
// flipped to an eternally-pending state on a "completed" run.
412+
const executableIds = new Set(testCases.map((tc) => tc.id));
413+
const missingIds = resumableIds.filter((tcId) => !executableIds.has(tcId));
414+
415+
// Claim the run. There is no cross-server CAS primitive in the storage
416+
// interface, so we use a claim token: write it, re-read, and abort if
417+
// another claimer overwrote ours in the window. Same-process double
418+
// resumes are already excluded by activeCancellationTokens above.
400419
const now = new Date().toISOString();
420+
const resumeToken = `${process.pid}-${Math.random().toString(36).slice(2, 10)}`;
401421
const results = { ...(run.results || {}) };
402422
for (const tcId of resumableIds) {
403-
results[tcId] = { reportId: '', status: 'pending' };
423+
if (executableIds.has(tcId)) {
424+
results[tcId] = { reportId: '', status: 'pending' };
425+
}
404426
}
405427
await storage.evaluationRuns.update(id, {
406428
status: 'running',
407429
error: '',
408430
results,
409431
resumedAt: now,
410-
});
432+
// The claim itself is a liveness signal — without this, the dead
433+
// server's stale heartbeatAt would leave the freshly-resumed run
434+
// looking orphaned until the first 60s heartbeat tick.
435+
heartbeatAt: now,
436+
resumeToken,
437+
} as Partial<EvaluationRun>);
438+
const claimed = await storage.evaluationRuns.getById(id);
439+
if ((claimed as any)?.resumeToken !== resumeToken) {
440+
sendSSE(res, 'error', {
441+
error: 'Another server claimed this run for resume at the same time — aborting this attempt',
442+
runId: id,
443+
});
444+
res.end();
445+
return;
446+
}
411447
run.status = 'running';
412448
run.results = results;
413449
run.resumedAt = now;
450+
run.heartbeatAt = now;
414451
delete run.error;
415452

416453
sendSSE(res, 'started', {
417454
runId: id,
418455
resumed: true,
419456
testCases: run.testCaseSnapshots,
420457
pendingCount: testCases.length,
421-
skippedCount: (run.testCaseSnapshots?.length || 0) - testCases.length,
458+
skippedCount: (run.testCaseSnapshots?.length || 0) - resumableIds.length,
459+
// Resumable ids the run's sources no longer resolve — left as failed,
460+
// not re-executed. Surfaced so callers can warn instead of silently
461+
// "completing" past them.
462+
missingCount: missingIds.length,
463+
...(missingIds.length > 0 ? { missingTestCaseIds: missingIds } : {}),
422464
});
423465

424466
const cancellationToken = createCancellationToken();
@@ -492,8 +534,12 @@ router.put('/api/storage/evaluation-runs/:id', async (req: Request, res: Respons
492534
const run = { ...req.body, id, docType: 'evaluation-run' as const };
493535
const existing = await storage.evaluationRuns.getById(id);
494536
if (existing) {
495-
const updated = await storage.evaluationRuns.update(id, run);
496-
res.json(updated);
537+
// Full-document REPLACE, not merge: `update()` doc-merges partial
538+
// updates (so omitted nested keys — e.g. removed results entries —
539+
// would survive). This route's contract is upsert-with-replace, so
540+
// re-create the doc wholesale.
541+
await storage.evaluationRuns.create(run);
542+
res.json(run);
497543
} else {
498544
await storage.evaluationRuns.create(run);
499545
res.status(201).json(run);

‎server/services/evaluationRunRecoveryOnBoot.ts‎

Lines changed: 48 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,10 @@ export async function recoverOrphanEvaluationRuns(storage: IStorageModule): Prom
7777
const maxPages = envInt('EVALUATION_RUN_RECOVERY_MAX_PAGES', 50);
7878
const now = Date.now();
7979

80+
// Two-phase: SCAN everything first, MUTATE afterwards. The list query
81+
// filters on status:'running' — mutating docs to 'failed' while paging
82+
// with from/size shrinks the result set under the cursor and skips runs.
83+
const candidates: EvaluationRun[] = [];
8084
let from = 0;
8185
for (let page = 0; page < maxPages; page++) {
8286
let runs: EvaluationRun[];
@@ -89,54 +93,54 @@ export async function recoverOrphanEvaluationRuns(storage: IStorageModule): Prom
8993
break;
9094
}
9195
if (!runs || runs.length === 0) break;
96+
candidates.push(...runs);
97+
if (runs.length < pageSize) break;
98+
from += pageSize;
99+
}
92100

93-
for (const run of runs) {
94-
stat.scannedRuns++;
95-
if (run.status !== 'running') continue;
96-
97-
// Age from the most recent liveness signal (heartbeat > resumed > created).
98-
// Executing servers stamp `heartbeatAt` every minute, so a run on a
99-
// sibling server sharing this storage cluster never looks stale.
100-
const ageMs = runLivenessAgeMs(run, now);
101-
if (ageMs < staleAfterMs) continue;
102-
103-
if (isEvaluationRunActiveInThisProcess(run.id)) continue;
104-
105-
stat.staleRuns++;
106-
const reason =
107-
`Evaluation runner did not complete this test case ` +
108-
`(stale 'running' run found during boot recovery; original process likely died). ` +
109-
`Use Resume to re-execute the unfinished test cases.`;
110-
111-
const newResults: Record<string, any> = {};
112-
for (const [tcId, res] of Object.entries(run.results || {})) {
113-
const r: any = res;
114-
const isUnfinished = (r?.status === 'pending' || r?.status === 'running') && !r?.reportId;
115-
if (isUnfinished) {
116-
newResults[tcId] = { reportId: '', status: 'failed', error: reason };
117-
stat.resultsMarkedFailed++;
118-
} else {
119-
newResults[tcId] = r;
120-
}
121-
}
122-
123-
try {
124-
await storage.evaluationRuns.update(run.id, {
125-
status: 'failed',
126-
error: 'Run interrupted (server restarted mid-run). Completed test cases are preserved — use Resume to finish the rest.',
127-
results: newResults,
128-
completedAt: new Date().toISOString(),
129-
});
130-
stat.runsMarkedFailed++;
131-
console.log(`[evaluationRunRecovery] Marked stale run ${run.id} as failed (resumable)`);
132-
} catch (err: any) {
133-
stat.errors++;
134-
console.warn(`[evaluationRunRecovery] Failed to update run ${run.id}: ${err?.message || err}`);
101+
for (const run of candidates) {
102+
stat.scannedRuns++;
103+
if (run.status !== 'running') continue;
104+
105+
// Age from the most recent liveness signal (heartbeat / resumed / created).
106+
// Executing servers stamp `heartbeatAt` every minute, so a run on a
107+
// sibling server sharing this storage cluster never looks stale.
108+
const ageMs = runLivenessAgeMs(run, now);
109+
if (ageMs < staleAfterMs) continue;
110+
111+
if (isEvaluationRunActiveInThisProcess(run.id)) continue;
112+
113+
stat.staleRuns++;
114+
const reason =
115+
`Evaluation runner did not complete this test case ` +
116+
`(stale 'running' run found during boot recovery; original process likely died). ` +
117+
`Use Resume to re-execute the unfinished test cases.`;
118+
119+
const newResults: Record<string, any> = {};
120+
for (const [tcId, res] of Object.entries(run.results || {})) {
121+
const r: any = res;
122+
const isUnfinished = (r?.status === 'pending' || r?.status === 'running') && !r?.reportId;
123+
if (isUnfinished) {
124+
newResults[tcId] = { reportId: '', status: 'failed', error: reason };
125+
stat.resultsMarkedFailed++;
126+
} else {
127+
newResults[tcId] = r;
135128
}
136129
}
137130

138-
if (runs.length < pageSize) break;
139-
from += pageSize;
131+
try {
132+
await storage.evaluationRuns.update(run.id, {
133+
status: 'failed',
134+
error: 'Run interrupted (server restarted mid-run). Completed test cases are preserved — use Resume to finish the rest.',
135+
results: newResults,
136+
completedAt: new Date().toISOString(),
137+
});
138+
stat.runsMarkedFailed++;
139+
console.log(`[evaluationRunRecovery] Marked stale run ${run.id} as failed (resumable)`);
140+
} catch (err: any) {
141+
stat.errors++;
142+
console.warn(`[evaluationRunRecovery] Failed to update run ${run.id}: ${err?.message || err}`);
143+
}
140144
}
141145

142146
stat.durationMs = Date.now() - startedAt;

‎tests/integration/server/routes/evaluationRunResume.integration.test.ts‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -196,4 +196,41 @@ describe('POST /api/storage/evaluation-runs/:id/resume — checkpoint resume', (
196196
const res = await httpJson('POST', `${BASE_URL}/api/storage/evaluation-runs/does-not-exist-xyz/resume`);
197197
expect(res.status).toBe(404);
198198
}, TEST_TIMEOUT);
199+
200+
it('409s a second resume while the first is still executing (codex #2 — double-resume guard)', async () => {
201+
if (!backendAvailable) return;
202+
203+
// Seed a second interrupted run over the same test cases.
204+
const raceRunId = `eval-run-resume-race-${Date.now()}`;
205+
const seeded = await httpJson<any>('PUT', `${BASE_URL}/api/storage/evaluation-runs/${raceRunId}`, {
206+
name: 'resume-race-run',
207+
sources: [{ type: 'test-case-ids', ids: createdTestCaseIds }],
208+
agentKey: 'demo',
209+
modelId: 'demo-model',
210+
judgeModelId: 'demo-model',
211+
trigger: 'api',
212+
status: 'failed',
213+
createdAt: new Date().toISOString(),
214+
testCaseSnapshots: createdTestCaseIds.map((id, i) => ({ id, version: 1, name: `race-tc${i + 1}` })),
215+
results: {},
216+
});
217+
expect(seeded.status).toBeLessThan(300);
218+
219+
// Fire the first resume WITHOUT awaiting completion, then a second one.
220+
const first = httpJson<any>('POST', `${BASE_URL}/api/storage/evaluation-runs/${raceRunId}/resume`);
221+
await new Promise((r) => setTimeout(r, 1500)); // let the first claim + start
222+
const second = await httpJson<any>(`POST`, `${BASE_URL}/api/storage/evaluation-runs/${raceRunId}/resume`);
223+
expect(second.status).toBe(409);
224+
expect(second.body.error).toMatch(/currently executing/i);
225+
226+
// First resume runs to completion; collect its reports for cleanup.
227+
const firstRes = await first;
228+
expect(firstRes.status).toBe(200);
229+
const completed = parseSSE(firstRes.raw).find((e) => e.event === 'completed');
230+
expect(completed).toBeDefined();
231+
for (const v of Object.values<any>(completed!.data.results || {})) {
232+
if (v.reportId) createdReportIds.push(v.reportId);
233+
}
234+
await httpJson('DELETE', `${BASE_URL}/api/storage/evaluation-runs/${raceRunId}`).catch(() => {});
235+
}, TEST_TIMEOUT);
199236
});

‎tests/unit/server/routes/storage/evaluationRunResume.test.ts‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -74,7 +74,7 @@ describe('computeResumableTestCaseIds', () => {
7474
describe('run liveness (shared-cluster safety)', () => {
7575
const T0 = Date.parse('2026-01-01T00:00:00Z');
7676

77-
it('prefers heartbeatAt over resumedAt over createdAt', () => {
77+
it('uses the most recent of heartbeat/resumed/created', () => {
7878
const run = {
7979
createdAt: new Date(T0 - 3_600_000).toISOString(),
8080
resumedAt: new Date(T0 - 600_000).toISOString(),
@@ -85,6 +85,18 @@ describe('run liveness (shared-cluster safety)', () => {
8585
expect(runLivenessAgeMs({ createdAt: run.createdAt }, T0)).toBe(3_600_000);
8686
});
8787

88+
it('a fresh resume claim counts as liveness even when the dead server\'s heartbeat is stale (codex #1)', () => {
89+
// After claiming an orphan, resumedAt is NEWER than the dead server's
90+
// last heartbeatAt. A priority order (heartbeat first) would leave the
91+
// just-resumed run looking stale — max() must win here.
92+
const run = {
93+
createdAt: new Date(T0 - 7_200_000).toISOString(),
94+
heartbeatAt: new Date(T0 - 3_600_000).toISOString(), // dead server, 1h ago
95+
resumedAt: new Date(T0 - 1_000).toISOString(), // claimed 1s ago
96+
};
97+
expect(runLivenessAgeMs(run, T0)).toBe(1_000);
98+
});
99+
88100
it('treats missing/invalid timestamps as infinitely stale', () => {
89101
expect(runLivenessAgeMs({} as any, T0)).toBe(Infinity);
90102
expect(runLivenessAgeMs({ createdAt: 'not-a-date' } as any, T0)).toBe(Infinity);

‎types/index.ts‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1113,6 +1113,12 @@ export interface EvaluationRun {
11131113
* heartbeat stops.
11141114
*/
11151115
heartbeatAt?: string;
1116+
/**
1117+
* Claim token written by the server that most recently claimed this run
1118+
* for resume. Written-then-re-read to detect two servers racing to resume
1119+
* the same orphan (the storage interface has no cross-server CAS).
1120+
*/
1121+
resumeToken?: string;
11161122
status: BenchmarkRunStatus;
11171123
error?: string;
11181124

0 commit comments

Comments
 (0)