From 5faaff7d101709ee343f2067c5274fed236395fd Mon Sep 17 00:00:00 2001 From: t Date: Mon, 7 Sep 2026 01:45:52 +0900 Subject: [PATCH] fix(storage): publish cleanup manifests atomically [skip ci] Preserve complete recovery records across handled publication failures and keep existing partial-purge restoration boundaries. Refs #3778. Local checks deferred to final stack CI by maintainer instruction. --- .../content/docs/reference/management-api.md | 4 + src/storage/cleanup.ts | 84 +++++++----- structure/02_config-and-codex-home.md | 6 + tests/storage/storage-cleanup.test.ts | 123 ++++++++++++++++++ 4 files changed, 182 insertions(+), 35 deletions(-) diff --git a/docs-site/src/content/docs/reference/management-api.md b/docs-site/src/content/docs/reference/management-api.md index a12303cfd7..2cd23eb4bd 100644 --- a/docs-site/src/content/docs/reference/management-api.md +++ b/docs-site/src/content/docs/reference/management-api.md @@ -231,6 +231,10 @@ Storage cleanup endpoints can move or permanently remove archived session data. first and submit the returned digest. Prefer quarantine when recovery may be needed. ::: +Cleanup recovery manifests are published atomically, preserving the previous complete record +if a replacement fails before publication. This does not reverse a permanent purge: restore +can still fail when a recorded session has no surviving rollout file. + ### Models and catalog | Method and path | Purpose | Notable errors | diff --git a/src/storage/cleanup.ts b/src/storage/cleanup.ts index c39bbeedf1..e44f7d7b5c 100644 --- a/src/storage/cleanup.ts +++ b/src/storage/cleanup.ts @@ -30,11 +30,11 @@ import { writeSync, chmodSync, } from "node:fs"; -import { basename, isAbsolute, join, relative, resolve, sep } from "node:path"; +import { basename, dirname, isAbsolute, join, relative, resolve, sep } from "node:path"; import { Database } from "bun:sqlite"; import { resolveCodexHomeDir } from "../codex/home"; import { readThreadFieldsFromRollout } from "../codex/history-provider"; -import { renameAtomicFile } from "../config"; +import { renameAtomicFile } from "../lib/windows-atomic-replace"; export const ARCHIVED_SESSIONS_DIR = "archived_sessions"; export const TRASH_DIR = ".trash"; @@ -115,9 +115,35 @@ function chmodPrivatePath(path: string, mode: number): void { try { chmodSync(path, mode); } catch { /* best-effort (e.g. Windows ACLs) */ } } -function writePrivateFile(path: string, content: string): void { - writeFileSync(path, content, "utf8"); - chmodPrivatePath(path, 0o600); +/** Publish complete stage metadata without truncating the last recovery record. */ +function writePrivateFile( + path: string, + content: string, + beforeRename?: (temporaryPath: string, targetPath: string) => void, +): void { + const temporaryPath = `${path}.${process.pid}.${randomUUID()}.tmp`; + let descriptor: number | undefined; + let created = false; + try { + descriptor = openSync(temporaryPath, "wx", 0o600); + created = true; + writeFileSync(descriptor, content, "utf8"); + fsyncSync(descriptor); + closeSync(descriptor); + descriptor = undefined; + chmodPrivatePath(temporaryPath, 0o600); + beforeRename?.(temporaryPath, path); + renameAtomicFile(temporaryPath, path, undefined, "storage-cleanup"); + chmodPrivatePath(path, 0o600); + fsyncDirectoryBestEffort(dirname(path)); + } finally { + if (descriptor !== undefined) { + try { closeSync(descriptor); } catch { /* preserve publication failure */ } + } + if (created) { + try { unlinkSync(temporaryPath); } catch { /* renamed or cleanup unavailable */ } + } + } } function chunkIds(ids: string[], chunkSize: number): string[][] { @@ -812,7 +838,6 @@ interface ReconcileTestHooks { const SATELLITE_BACKUP_FILE = "satellite-backup.json"; /** Marks an incomplete restore so retries can accept dest files and resume metadata. */ const RESTORE_PENDING_FILE = "restore-pending.json"; -let _satelliteBackupSeq = 0; type StagedFile = { from: string; to: string; relPath: string }; @@ -1070,34 +1095,11 @@ function writeSatelliteBackup( if (options?.failWrite) throw new Error("test_fail_satellite_backup_write"); const dest = join(stageDir, SATELLITE_BACKUP_FILE); const replacing = existsSync(dest); - const tmp = join(stageDir, `${SATELLITE_BACKUP_FILE}.${process.pid}.${++_satelliteBackupSeq}.tmp`); - const payload = Buffer.from(JSON.stringify(backup), "utf8"); - const fd = openSync(tmp, "w", 0o600); - try { - let offset = 0; - while (offset < payload.length) { - offset += writeSync(fd, payload, offset, payload.length - offset, null); + writePrivateFile(dest, JSON.stringify(backup), () => { + if (options?.failReplaceBeforeRename && replacing) { + throw new Error("test_fail_satellite_backup_replace"); } - fsyncSync(fd); - } catch (error) { - try { closeSync(fd); } catch { /* */ } - try { unlinkSync(tmp); } catch { /* */ } - throw error; - } - closeSync(fd); - chmodPrivatePath(tmp, 0o600); - if (options?.failReplaceBeforeRename && replacing) { - try { unlinkSync(tmp); } catch { /* */ } - throw new Error("test_fail_satellite_backup_replace"); - } - try { - renameAtomicFile(tmp, dest, undefined, "storage-cleanup"); - } catch (error) { - try { unlinkSync(tmp); } catch { /* */ } - throw error; - } - chmodPrivatePath(dest, 0o600); - fsyncDirectoryBestEffort(stageDir); + }); } function clearSatelliteBackup(stageDir: string): void { @@ -1734,6 +1736,12 @@ export interface ExecuteCleanupOptions { /** Test-only failure injection for atomicity regressions. */ _test?: { failManifestWrite?: boolean; + /** Observe the complete temp and prior destination before publication. Never serialized. */ + beforeManifestReplace?: ( + temporaryPath: string, + targetPath: string, + phase: "staging" | "pre-commit" | "purge-incomplete", + ) => void; failPurgeBasenames?: string[]; failRollbackBasenames?: string[]; blockStageDestBasenames?: string[]; @@ -1752,14 +1760,14 @@ export interface ExecuteCleanupOptions { /** Serializable cleanup test hooks allowed on the management API wire. */ export type CleanupWireTestHooks = Omit< NonNullable, - "afterSatelliteMutations" | "beforeReconcileLock" + "afterSatelliteMutations" | "beforeReconcileLock" | "beforeManifestReplace" >; function isStringArray(v: unknown): v is string[] { return Array.isArray(v) && v.every(e => typeof e === "string"); } -/** Pick only allowlisted serializable hooks; drops function hooks (afterSatelliteMutations, beforeReconcileLock) and unknown keys. */ +/** Pick only allowlisted serializable hooks; drops all function hooks and unknown keys. */ export function pickWireCleanupTestHooks(raw: unknown): CleanupWireTestHooks | undefined { if (!raw || typeof raw !== "object") return undefined; const o = raw as Record; @@ -1911,6 +1919,9 @@ export function executeArchivedCleanup(options: ExecuteCleanupOptions): CleanupR entries: manifestEntries, ...extra, }, null, 2), + (temporaryPath, targetPath) => options._test?.beforeManifestReplace?.( + temporaryPath, targetPath, extra.staging ? "staging" : "pre-commit", + ), ); }; @@ -1999,6 +2010,9 @@ export function executeArchivedCleanup(options: ExecuteCleanupOptions): CleanupR })) .filter(entry => entry.physicalRelPaths.length > 0), }, null, 2), + (temporaryPath, targetPath) => options._test?.beforeManifestReplace?.( + temporaryPath, targetPath, "purge-incomplete", + ), ); } catch { /* best-effort: the pre-commit manifest is still on disk */ } return { diff --git a/structure/02_config-and-codex-home.md b/structure/02_config-and-codex-home.md index bb8ec5630f..66a729b4fe 100644 --- a/structure/02_config-and-codex-home.md +++ b/structure/02_config-and-codex-home.md @@ -127,6 +127,12 @@ Worker cannot restore unrelated API keys or provider settings from a snapshot re If that metadata write is unavailable after cleanup has already completed, the job retains the cleanup outcome and exposes a bounded persistence error instead of relabeling the run as a Worker failure. +Cleanup manifests and satellite backups share the stage-local atomic publisher: an exclusive +private temporary file is fully written and file-synced before the existing Windows-tolerant +rename replaces the destination. Handled publication failures retain the previous record; +directory syncing remains best-effort. This does not make a partial permanent purge reversible: +restore still fails closed when a recorded logical entry has no surviving file. + Windows secret-file hardening resolves the effective token SID through an absolute, trusted PowerShell path before granting the owner and removing inherited broad ACL entries. The normal path obtains System32 from `GetSystemDirectoryW`. Windows ARM64 Bun builds that cannot execute diff --git a/tests/storage/storage-cleanup.test.ts b/tests/storage/storage-cleanup.test.ts index 421950368c..31cbcd6d21 100644 --- a/tests/storage/storage-cleanup.test.ts +++ b/tests/storage/storage-cleanup.test.ts @@ -8,6 +8,7 @@ import { readFileSync, renameSync, rmSync, + statSync, unlinkSync, utimesSync, writeFileSync, @@ -20,6 +21,7 @@ import { listArchivedCandidates, listTrashEntries, normalizeArchivedRolloutPath, + pickWireCleanupTestHooks, previewArchivedCleanup, previewExactArchivedCleanup, restoreTrashEntry, @@ -648,6 +650,127 @@ describe("executeArchivedCleanup", () => { expect(ids).toContain("told"); }); + test("initial manifest publication failure preserves originals and removes its private temp", () => { + home = buildHome(); + const observed: Array<{ priorExists: boolean; next: string; mode: number }> = []; + const result = runWithDigest(50, "quarantine", home, { + now: 881, + _test: { + beforeManifestReplace: (temporaryPath, targetPath, phase) => { + if (phase !== "staging") return; + observed.push({ + priorExists: existsSync(targetPath), + next: readFileSync(temporaryPath, "utf8"), + mode: statSync(temporaryPath).mode & 0o777, + }); + throw new Error("injected_manifest_publication_failure"); + }, + }, + }); + // Assert outside the production catch: an assertion inside the hook could be swallowed. + expect(observed).toHaveLength(1); + expect(observed[0]!.priorExists).toBe(false); + expect(JSON.parse(observed[0]!.next).staging).toBe(true); + if (process.platform !== "win32") expect(observed[0]!.mode).toBe(0o600); + expect(result.error).toBe("fs_failed"); + expect(existsSync(join(home, ".trash", "881"))).toBe(false); + expect(readFileSync(join(home, "archived_sessions", "rollout-old.jsonl"), "utf8")).toBe("OLD".repeat(10)); + const db = new Database(join(home, "state_5.sqlite"), { readonly: true }); + expect(db.query("SELECT id FROM threads WHERE id = 'told'").get()).toBeTruthy(); + db.close(); + expect(pickWireCleanupTestHooks({ + beforeManifestReplace: () => {}, failManifestWrite: true, + })).toEqual({ failManifestWrite: true }); + }, STORE_BUDGET_MS); + + test("failed pre-delete replacement leaves the prior manifest intact during publication and restorable", () => { + home = buildHome(); + let stagingBytes = ""; + const observed: Array<{ prior: string; next: string }> = []; + const result = runWithDigest(50, "quarantine", home, { + now: 882, + _test: { + failRollbackBasenames: ["rollout-old.jsonl"], + beforeManifestReplace: (temporaryPath, targetPath, phase) => { + if (phase === "staging") stagingBytes = readFileSync(temporaryPath, "utf8"); + if (phase !== "pre-commit") return; + observed.push({ prior: readFileSync(targetPath, "utf8"), next: readFileSync(temporaryPath, "utf8") }); + throw new Error("injected_manifest_publication_failure"); + }, + }, + }); + expect(observed).toHaveLength(1); + expect(observed[0]!.prior).toBe(stagingBytes); + expect(JSON.parse(observed[0]!.prior).staging).toBe(true); + expect(JSON.parse(observed[0]!.next).staging).toBeUndefined(); + expect(result.error).toBe("fs_failed"); + expect(result.trashDir).toBe(".trash/882"); + const stage = join(home, ".trash", "882"); + expect(readFileSync(join(stage, "manifest.json"), "utf8")).toBe(stagingBytes); + expect(readFileSync(join(stage, "rollout-old.jsonl"), "utf8")).toBe("OLD".repeat(10)); + expect(readdirSync(stage).filter(name => name.endsWith(".tmp"))).toEqual([]); + const db = new Database(join(home, "state_5.sqlite"), { readonly: true }); + expect(db.query("SELECT id FROM threads WHERE id = 'told'").get()).toBeTruthy(); + db.close(); + const restored = restoreTrashEntry(".trash/882", { codexHome: home }); + expect(restored.ok).toBe(true); + expect(restored.count).toBe(1); + expect(readFileSync(join(home, "archived_sessions", "rollout-old.jsonl"), "utf8")).toBe("OLD".repeat(10)); + }, STORE_BUDGET_MS); + + test.each([false, true])("failed post-purge manifest replacement preserves prior bytes (partial=%s)", partial => { + home = buildHome(); + let preCommitBytes = ""; + const observed: Array<{ prior: string; next: string }> = []; + const result = runWithDigest(100, "permanent", home, { + now: 883, + _test: { + failPurgeBasenames: partial + ? ["rollout-mid.jsonl"] + : ["rollout-old.jsonl", "rollout-mid.jsonl", "rollout-new.jsonl"], + beforeManifestReplace: (temporaryPath, targetPath, phase) => { + if (phase === "pre-commit") preCommitBytes = readFileSync(temporaryPath, "utf8"); + if (phase !== "purge-incomplete") return; + observed.push({ prior: readFileSync(targetPath, "utf8"), next: readFileSync(temporaryPath, "utf8") }); + throw new Error("injected_manifest_publication_failure"); + }, + }, + }); + expect(observed).toHaveLength(1); + expect(observed[0]!.prior).toBe(preCommitBytes); + expect(JSON.parse(observed[0]!.next).purgeIncomplete).toBe(true); + expect(JSON.parse(observed[0]!.next).entries).toHaveLength(partial ? 1 : 3); + expect(result.error).toBe("fs_failed"); + const stage = join(home, ".trash", "883"); + expect(readFileSync(join(stage, "manifest.json"), "utf8")).toBe(preCommitBytes); + expect(readdirSync(stage).filter(name => name.endsWith(".tmp"))).toEqual([]); + const dbBefore = new Database(join(home, "state_5.sqlite"), { readonly: true }); + const rowsBefore = dbBefore.query("SELECT id FROM threads ORDER BY id").all(); + dbBefore.close(); + expect(rowsBefore).toEqual([{ id: "active" }]); + const stageBefore = readdirSync(stage).sort(); + const restored = restoreTrashEntry(".trash/883", { codexHome: home }); + if (partial) { + // A wholly purged old entry still fails closed; valid JSON is not full recovery. + expect(restored.error).toBe("fs_failed"); + expect(restored.restoredPaths).toEqual([]); + expect(readdirSync(stage).sort()).toEqual(stageBefore); + expect(readFileSync(join(stage, "rollout-mid.jsonl"), "utf8")).toBe("MID".repeat(20)); + expect(existsSync(join(home, "archived_sessions", "rollout-old.jsonl"))).toBe(false); + const dbAfter = new Database(join(home, "state_5.sqlite"), { readonly: true }); + expect(dbAfter.query("SELECT id FROM threads ORDER BY id").all()).toEqual(rowsBefore); + dbAfter.close(); + } else { + expect(restored.ok).toBe(true); + expect(restored.count).toBe(3); + const dbAfter = new Database(join(home, "state_5.sqlite"), { readonly: true }); + expect(dbAfter.query("SELECT id FROM threads ORDER BY id").all()).toEqual([ + { id: "active" }, { id: "tmid" }, { id: "tnew" }, { id: "told" }, + ]); + dbAfter.close(); + } + }, { timeout: STORE_BUDGET_MS }); + test("rename-back failure keeps staged file and reports relative trashDir", () => { home = buildHome(); const db = new Database(join(home, "state_5.sqlite"));