From cc4b24bc16e3ef41873f19669c9dec4118b8a3f3 Mon Sep 17 00:00:00 2001 From: Agnik47 <140933190+Agnik47@users.noreply.github.com> Date: Sat, 15 Aug 2026 02:24:33 +0530 Subject: [PATCH] fix(checkpoint): report save failures instead of dropping or crashing on them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `save()` is declared `void` but does asynchronous work, and nobody observed the resulting promise. `_performSave` ends with `throw lastError` when all retries fail or the error is non-retriable — ENOSPC, EACCES, a read-only volume, a bad path — and that rejection had no handler: `nextQueue` was never caught, and `nextQueue.finally(...)` created a second unhandled rejected promise. Under Bun/Node that surfaces as an `unhandledRejection`, which by default terminates the process mid-run with no indication that a checkpoint write, rather than a provider, was the cause. It also meant any `save()` issued while a rejected promise sat in `saveLock` chained onto it with `.then()` and never called `_performSave` at all, silently dropping those updates. Terminate the chain with a `.catch` that records the failure and logs it. The promise stored in `saveLock` therefore never rejects, which fixes both halves at once: no unhandled rejection, and a failed write no longer stops the saves queued behind it. `flush()` previously rethrew whatever the queue happened to be holding, so a checkpoint write failure arrived at `Orchestrator.run` as an opaque rejection after every phase had finished, and `runBenchmark` wrote `status: "failed"` over a run whose work was complete. It now reports accumulated failures deliberately, naming checkpoint persistence and the affected run, and consumes them so a later flush reports new failures rather than repeating one already surfaced. Throwing remains correct: if the checkpoint cannot be written then neither can `updateStatus`, so the on-disk record is stale either way — the fix is that the reason is legible. Finally, serialise the checkpoint in `save()` rather than in `_performSave`. The object is mutated concurrently by in-flight tasks (`updatePhase` does `Object.assign`), so a queued write used to persist whatever state existed when it eventually ran. The queue now provides the snapshot guarantee its shape implies. Fixes #71 --- src/orchestrator/checkpoint.test.ts | 201 ++++++++++++++++++++++++++++ src/orchestrator/checkpoint.ts | 85 ++++++++++-- 2 files changed, 272 insertions(+), 14 deletions(-) create mode 100644 src/orchestrator/checkpoint.test.ts diff --git a/src/orchestrator/checkpoint.test.ts b/src/orchestrator/checkpoint.test.ts new file mode 100644 index 0000000..80aaf21 --- /dev/null +++ b/src/orchestrator/checkpoint.test.ts @@ -0,0 +1,201 @@ +import { afterEach, describe, expect, test } from "bun:test" +import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "fs" +import { tmpdir } from "os" +import { join } from "path" +import { CheckpointManager } from "./checkpoint" +import type { RunCheckpoint } from "../types/checkpoint" + +const tempDirs: string[] = [] + +function makeManager(): CheckpointManager { + const dir = mkdtempSync(join(tmpdir(), "memorybench-checkpoint-")) + tempDirs.push(dir) + return new CheckpointManager(dir) +} + +function makeCheckpoint(runId: string): RunCheckpoint { + return { + runId, + provider: "rag", + benchmark: "longmemeval", + judge: "gpt-4o", + answeringModel: "gpt-4o", + status: "running", + createdAt: new Date(0).toISOString(), + updatedAt: new Date(0).toISOString(), + questions: {}, + } as RunCheckpoint +} + +/** Read/write the manager's private basePath, so a test can break and repair writes. */ +function basePathOf(manager: CheckpointManager): { get(): string; set(p: string): void } { + const target = manager as unknown as { basePath: string } + return { get: () => target.basePath, set: (p: string) => (target.basePath = p) } +} + +/** + * Force every write for this run to fail, the way ENOSPC/EACCES or a read-only + * volume would. Rooting the run directory under a regular *file* makes both the + * mkdir and the write fail with ENOTDIR. Returns the original base path. + */ +function breakWrites(manager: CheckpointManager): string { + const basePath = basePathOf(manager) + const original = basePath.get() + const wall = join(original, "wall") + writeFileSync(wall, "a file where a directory is required") + basePath.set(join(wall, "nested")) + return original +} + +afterEach(() => { + for (const dir of tempDirs.splice(0)) { + try { + rmSync(dir, { recursive: true, force: true }) + } catch { + /* best effort */ + } + } +}) + +describe("CheckpointManager.save", () => { + test("persists the checkpoint", async () => { + const manager = makeManager() + const checkpoint = makeCheckpoint("run-ok") + + manager.save(checkpoint) + await manager.flush("run-ok") + + expect(existsSync(manager.getCheckpointPath("run-ok"))).toBe(true) + expect(manager.load("run-ok")?.runId).toBe("run-ok") + }) + + test("snapshots the checkpoint at call time, not at write time", async () => { + const manager = makeManager() + const checkpoint = makeCheckpoint("run-snapshot") + + checkpoint.status = "running" + manager.save(checkpoint) + // Mutate immediately, while the first write is still queued — as concurrent + // updatePhase calls do. The queued write must not pick this up. + checkpoint.status = "completed" + + await manager.flush("run-snapshot") + + expect(manager.load("run-snapshot")?.status).toBe("running") + }) + + test("a failing save does not become an unhandled rejection", async () => { + const manager = makeManager() + const checkpoint = makeCheckpoint("run-unhandled") + breakWrites(manager) + + const rejections: unknown[] = [] + const onRejection = (e: unknown) => rejections.push(e) + process.on("unhandledRejection", onRejection) + + manager.save(checkpoint) + await manager.flush("run-unhandled").catch(() => {}) + // Give the microtask queue a chance to report an unhandled rejection. + await new Promise((resolve) => setTimeout(resolve, 50)) + + process.off("unhandledRejection", onRejection) + expect(rejections).toEqual([]) + }) + + test("a failed save does not stop later saves from running", async () => { + const manager = makeManager() + const checkpoint = makeCheckpoint("run-chained") + + const original = breakWrites(manager) + manager.save(checkpoint) + await manager.flush("run-chained").catch(() => {}) + + // Repair the destination and save again: the second write must actually run + // rather than chaining onto a rejected promise and being skipped. + basePathOf(manager).set(original) + checkpoint.status = "completed" + manager.save(checkpoint) + await manager.flush("run-chained") + + expect(manager.load("run-chained")?.status).toBe("completed") + }) +}) + +describe("CheckpointManager.flush", () => { + test("resolves quietly when every save succeeded", async () => { + const manager = makeManager() + manager.save(makeCheckpoint("run-quiet")) + + await manager.flush("run-quiet") + expect(manager.hasSaveError("run-quiet")).toBe(false) + }) + + test("reports a write failure as a checkpoint-persistence error", async () => { + const manager = makeManager() + breakWrites(manager) + manager.save(makeCheckpoint("run-failing")) + + // The message must name checkpoint persistence — the whole point is that an + // opaque rejection used to surface as an unrelated run failure. + await expect(manager.flush("run-failing")).rejects.toThrow(/Checkpoint could not be persisted/) + }) + + test("names the affected run", async () => { + const manager = makeManager() + breakWrites(manager) + manager.save(makeCheckpoint("run-named")) + + await expect(manager.flush("run-named")).rejects.toThrow(/run-named/) + }) + + test("consumes the error so a later clean flush succeeds", async () => { + const manager = makeManager() + const checkpoint = makeCheckpoint("run-consumed") + breakWrites(manager) + + manager.save(checkpoint) + await expect(manager.flush("run-consumed")).rejects.toThrow() + + // Same failure must not be re-reported once it has been surfaced. + await manager.flush("run-consumed") + expect(manager.hasSaveError("run-consumed")).toBe(false) + }) + + test("flushing every run surfaces a failure from any of them", async () => { + const manager = makeManager() + breakWrites(manager) + manager.save(makeCheckpoint("run-all")) + + await expect(manager.flush()).rejects.toThrow(/Checkpoint could not be persisted/) + }) + + test("deleting a run clears its pending error", async () => { + const manager = makeManager() + breakWrites(manager) + manager.save(makeCheckpoint("run-deleted")) + await manager.flush("run-deleted").catch(() => {}) + + manager.save(makeCheckpoint("run-deleted")) + await new Promise((resolve) => setTimeout(resolve, 20)) + manager.delete("run-deleted") + + expect(manager.hasSaveError("run-deleted")).toBe(false) + await manager.flush() + }) +}) + +describe("CheckpointManager save ordering", () => { + test("queued saves are applied in call order", async () => { + const manager = makeManager() + const checkpoint = makeCheckpoint("run-order") + + for (const status of ["initializing", "running", "completed"] as const) { + checkpoint.status = status + manager.save(checkpoint) + } + await manager.flush("run-order") + + const written = JSON.parse(readFileSync(manager.getCheckpointPath("run-order"), "utf8")) + expect(written.status).toBe("completed") + }) +}) diff --git a/src/orchestrator/checkpoint.ts b/src/orchestrator/checkpoint.ts index aa00835..41b137e 100644 --- a/src/orchestrator/checkpoint.ts +++ b/src/orchestrator/checkpoint.ts @@ -27,6 +27,8 @@ const RUNS_DIR = "./data/runs" export class CheckpointManager { private basePath: string private saveLock = new Map>() + /** Last save failure per run, recorded by save() and surfaced by flush(). */ + private saveErrors = new Map() constructor(basePath: string = RUNS_DIR) { this.basePath = basePath @@ -62,34 +64,52 @@ export class CheckpointManager { } save(checkpoint: RunCheckpoint): void { - const currentQueue = this.saveLock.get(checkpoint.runId) || Promise.resolve() - const nextQueue = currentQueue.then(() => this._performSave(checkpoint)) - this.saveLock.set(checkpoint.runId, nextQueue) + const runId = checkpoint.runId - nextQueue.finally(() => { - if (this.saveLock.get(checkpoint.runId) === nextQueue) { - this.saveLock.delete(checkpoint.runId) + // Serialise here rather than at write time. The checkpoint object is mutated + // concurrently by in-flight tasks (updatePhase does Object.assign), so a queued + // save would otherwise persist whatever state existed when the write finally + // ran instead of the state that was current when save() was called. + checkpoint.updatedAt = new Date().toISOString() + const serialized = JSON.stringify(checkpoint, null, 2) + + const currentQueue = this.saveLock.get(runId) || Promise.resolve() + + // The stored promise always ends in a .catch, so it never rejects: a failed + // write can neither become an unhandled rejection nor stop the saves queued + // behind it from running. + const nextQueue = currentQueue + .then(() => this._performSave(runId, serialized)) + .catch((e: unknown) => { + const error = e instanceof Error ? e : new Error(String(e)) + this.saveErrors.set(runId, error) + logger.error(`Checkpoint save failed for ${runId}: ${error.message}`) + }) + + this.saveLock.set(runId, nextQueue) + + void nextQueue.finally(() => { + if (this.saveLock.get(runId) === nextQueue) { + this.saveLock.delete(runId) } }) } - private async _performSave(checkpoint: RunCheckpoint): Promise { - const runPath = this.getRunPath(checkpoint.runId) - const path = this.getCheckpointPath(checkpoint.runId) + private async _performSave(runId: string, serialized: string): Promise { + const runPath = this.getRunPath(runId) + const path = this.getCheckpointPath(runId) const tempPath = path + ".tmp" if (!existsSync(runPath)) { mkdirSync(runPath, { recursive: true }) } - checkpoint.updatedAt = new Date().toISOString() - let lastError: any // Windows often locks files briefly (EPERM/EBUSY), so we retry a few times for (let attempt = 0; attempt < 5; attempt++) { try { - writeFileSync(tempPath, JSON.stringify(checkpoint, null, 2)) + writeFileSync(tempPath, serialized) renameSync(tempPath, path) return // Success } catch (e: any) { @@ -109,12 +129,47 @@ export class CheckpointManager { throw lastError } + /** + * Wait for queued saves to settle. + * + * Throws if any save failed since the last flush. A failure means the on-disk + * checkpoint is stale, so the caller must not go on to record the run as + * durably persisted — but the error now names checkpoint persistence as the + * cause instead of arriving as an opaque rejection from an unrelated await. + */ async flush(runId?: string): Promise { if (runId) { await this.saveLock.get(runId) - } else { - await Promise.all(Array.from(this.saveLock.values())) + this.throwIfSaveFailed([runId]) + return } + + await Promise.all(Array.from(this.saveLock.values())) + this.throwIfSaveFailed(Array.from(this.saveErrors.keys())) + } + + /** True if a save for this run has failed and not yet been reported by flush(). */ + hasSaveError(runId: string): boolean { + return this.saveErrors.has(runId) + } + + private throwIfSaveFailed(runIds: string[]): void { + const failures: string[] = [] + + for (const runId of runIds) { + const error = this.saveErrors.get(runId) + if (!error) continue + // Consume it, so a later flush reports new failures rather than this one again. + this.saveErrors.delete(runId) + failures.push(`${runId}: ${error.message}`) + } + + if (failures.length === 0) return + + throw new Error( + `Checkpoint could not be persisted (${failures.join("; ")}). ` + + `The on-disk state for the affected run(s) is stale.` + ) } create( @@ -169,6 +224,8 @@ export class CheckpointManager { rmSync(runPath, { recursive: true }) logger.info(`Deleted run: ${runPath}`) } + // The run is gone; a stale write failure for it must not fail a later flush. + this.saveErrors.delete(runId) } updateStatus(checkpoint: RunCheckpoint, status: RunStatus): void {