diff --git a/docs/API_REFERENCE.md b/docs/API_REFERENCE.md index f843c58..d7c9022 100644 --- a/docs/API_REFERENCE.md +++ b/docs/API_REFERENCE.md @@ -186,7 +186,7 @@ Session helpers (read-only): `markStale(state, sessionId)`, `clearStale(state, s | `class StateStore` | `constructor(projectDir:string)` | `dir`, `file`, private `lockPath` | | `.withLock(fn)` | `(fn:()=>T)=>T` | Exclusive-create `.state.lock` (`@`); LOCK_TIMEOUT_MS=5s; locks older than LOCK_STALE_MS=15s broken. R5 tradeoff documented in ARCHITECTURE.md | | `.load()` | `()=>State` | validateState gate (schema_version, shapes, key/name agreement) → reject to fresh + error; backfillState for legacy files | -| `.save(state)` | `(state:State)=>void` | Atomic tmp+rename; blocking-sleep retry; direct-write fallback | +| `.save(state)` | `(state:State)=>void` | Flushed tmp+rename; bounded retries; preserves previous file on failure | | `.update(fn)` | `(mutate:(state:State)=>void)=>State` | `withLock(load → mutate → save)` | | `LOCK_TIMEOUT_MS` / `LOCK_STALE_MS` | `5000` / `15000` | Exported for tests | | `stateDirFor` / `isInsideStateDir` | path helpers | | diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index b90f5f1..5123d68 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -168,7 +168,7 @@ What exists instead, honestly: `stale` (offline proxy, host-verified), `delivery - **Locking:** every load→mutate→save runs under `withLock(fn)` (`src/core/store.ts` withLock) — exclusive-create `.state.lock` carrying `@`; acquisition retries (10ms busy waits) until `LOCK_TIMEOUT_MS=5s`, locks older than `LOCK_STALE_MS=15s` are presumed abandoned and broken. Bare reads stay lock-free because saves are atomic renames. - *Known tradeoff (R5):* a live-but-slow holder whose critical section exceeds 15s can get its lock broken mid-fn, briefly allowing concurrent mutation. All current critical sections are ms-scale synchronous blocks, so this is accepted rather than heartbeat-managed; revisit if async work ever moves inside the lock. - `load()` (`src/core/store.ts`): if missing → `emptyState()`. Parses then runs `validateState`: wrong `schema_version`, non-object roots, malformed member rows (bad role label / non-string role_prompt), channel key≠name aliasing, malformed messages/queues/delivered_to — ALL rejected to `emptyState()` + recovery error. Survivors run `backfillState` (`src/core/store.ts`): clamp `max_members∈[2,8]`, default timers for pre-timer files, migrate legacy ROLE-keyed timers to session-id keys (remapping the active role's elapsed onto its member). -- `save(state)` (`src/core/store.ts`): `mkdirSync(recursive)`, `writeFileSync(tmp)` where tmp = `.state...tmp`, `renameSync(tmp, file)`. On error: blocking 50ms sleep + retry rename; on second failure direct `writeFileSync(file)`; on third throw with combined message. +- `save(state)` (`src/core/store.ts`): `mkdirSync(recursive)`, `writeFileSync(tmp, payload, { flush: true })` where tmp = `.state...tmp`, `renameSync(tmp, file)`. On replacement failure: retry after bounded 50ms and 100ms blocking waits. If all three attempts fail, throw a persistence error with the original cause and preserve the previous file. Temporary files are cleaned up best-effort. Both stores share `replaceStateFile` in `src/core/atomic-file.ts`. - `update(fn)` (`src/core/store.ts`): `withLock(load → mutate → save)`. ## Hooks & Tools (plugin.ts) diff --git a/src/core/atomic-file.ts b/src/core/atomic-file.ts new file mode 100644 index 0000000..9f1cf37 --- /dev/null +++ b/src/core/atomic-file.ts @@ -0,0 +1,27 @@ +import { renameSync, rmSync, writeFileSync } from "node:fs" + +/** Keep the last valid file intact if an atomic replacement is unavailable. */ +export function replaceStateFile(file: string, temporary: string, payload: string): void { + try { + writeFileSync(temporary, payload, { encoding: "utf8", flush: true }) + const pause = new Int32Array(new SharedArrayBuffer(4)) + for (const delay of [0, 50, 100]) { + // Persistence is synchronous; only this bounded failure path blocks. + if (delay) Atomics.wait(pause, 0, 0, delay) + try { + renameSync(temporary, file) + return + } catch (error) { + if (delay === 100) throw error + } + } + } catch (cause) { + throw new Error(`OpenComms: failed to persist state at ${file}`, { cause }) + } finally { + try { + rmSync(temporary, { force: true }) + } catch { + // Cleanup is best-effort and must not hide the persistence error. + } + } +} diff --git a/src/core/store.ts b/src/core/store.ts index e972b91..752fbd5 100644 --- a/src/core/store.ts +++ b/src/core/store.ts @@ -5,7 +5,7 @@ * atomic: we serialize to a temp file in the same directory, flush it, then * rename over the target. On Windows, `rename` over an existing file is * supported by Node's fs.rename (it maps to MoveFileEx with REPLACE_EXISTING), - * but we defensively retry once after a short delay because antivirus or + * but we defensively retry with bounded delays because antivirus or * OneDrive can briefly hold a handle. * * Concurrency: multiple host sessions share one state.json. A bare @@ -19,7 +19,6 @@ import { mkdirSync, readFileSync, - renameSync, writeFileSync, existsSync, copyFileSync, @@ -30,6 +29,7 @@ import { type PathOrFileDescriptor, } from "node:fs" import { dirname, join } from "node:path" +import { replaceStateFile } from "./atomic-file.js" import { randomBytes } from "node:crypto" import { DEFAULT_MAX_MEMBERS, @@ -541,30 +541,7 @@ export class StateStore { mkdirSync(this.dir, { recursive: true }) const tmp = join(this.dir, `.state.${process.pid}.${randomBytes(4).toString("hex")}.tmp`) const payload = JSON.stringify(state, null, 2) - writeFileSync(tmp, payload, "utf8") - try { - renameSync(tmp, this.file) - } catch (error) { - // Windows: retry once after a short yielding pause (AV/OneDrive races). - // A blocking 50ms stall is acceptable here: the write already happened, - // and this path is rare; the atomic-rename guarantee is what matters. - try { - // save() is intentionally synchronous (callers rely on durability - // when it returns), so this rare retry path uses a bounded 50ms - // blocking wait rather than async. LOCK acquisition — the path that - // can wait seconds — uses yielding sleepAsync instead. - Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 50) - renameSync(tmp, this.file) - } catch (second) { - try { - writeFileSync(this.file, payload, "utf8") - } catch { - throw new Error( - `OpenComms: failed to persist state (${(error as Error).message}; ${(second as Error).message})`, - ) - } - } - } + replaceStateFile(this.file, tmp, payload) } /** Convenience: load, mutate, save — all under the cross-process lock. */ diff --git a/src/orchestrator/state.ts b/src/orchestrator/state.ts index aacd583..3f4a7df 100644 --- a/src/orchestrator/state.ts +++ b/src/orchestrator/state.ts @@ -20,18 +20,9 @@ * from agent-facing tool arguments. */ -import { - existsSync, - mkdirSync, - readFileSync, - renameSync, - writeFileSync, - openSync, - closeSync, - unlinkSync, - statSync, -} from "node:fs" +import { existsSync, mkdirSync, readFileSync, writeFileSync, openSync, closeSync, unlinkSync, statSync } from "node:fs" import { join } from "node:path" +import { replaceStateFile } from "../core/atomic-file.js" import { randomBytes, createHash } from "node:crypto" export const ORCHESTRATOR_FILE = "orchestrator.json" @@ -475,23 +466,7 @@ export class OrchestratorStore { save(state: OrchestratorState): void { mkdirSync(this.dir, { recursive: true }) const tmp = join(this.dir, `.orchestrator.${process.pid}.${randomBytes(4).toString("hex")}.tmp`) - writeFileSync(tmp, JSON.stringify(state, null, 2), "utf8") - try { - renameSync(tmp, this.file) - } catch { - // Windows AV/OneDrive races: bounded blocking retry (same discipline as - // StateStore.save), then direct write as a last resort. - try { - Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 50) - renameSync(tmp, this.file) - } catch (second) { - try { - writeFileSync(this.file, JSON.stringify(state, null, 2), "utf8") - } catch { - throw new Error(`OpenComms: failed to persist orchestrator state (${(second as Error).message})`) - } - } - } + replaceStateFile(this.file, tmp, JSON.stringify(state, null, 2)) } /** Load, mutate, save under the lock. The mutation runs synchronously. */ diff --git a/test/unit/atomic-save.test.ts b/test/unit/atomic-save.test.ts new file mode 100644 index 0000000..3c8c399 --- /dev/null +++ b/test/unit/atomic-save.test.ts @@ -0,0 +1,115 @@ +import test from "node:test" +import assert from "node:assert/strict" +import fs from "node:fs" +import { syncBuiltinESMExports } from "node:module" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { StateStore, emptyState } from "../../src/core/store.js" +import { OrchestratorStore, emptyOrchestratorState } from "../../src/orchestrator/state.js" + +const stores = [ + { + name: "StateStore", + create(dir: string) { + const store = new StateStore(dir) + return { + file: store.file, + dir: store.dir, + save(revision: number) { + const state = { ...emptyState(), revision } + store.save(state) + }, + } + }, + }, + { + name: "OrchestratorStore", + create(dir: string) { + const store = new OrchestratorStore(dir) + return { + file: store.file, + dir: store.dir, + save(revision: number) { + const state = { ...emptyOrchestratorState(dir), revision } + store.save(state) + }, + } + }, + }, +] + +for (const fixture of stores) { + for (const succeeds of [true, false]) { + test(`${fixture.name}: rename ${succeeds ? "retry succeeds" : "failure preserves previous state"}`, (t) => { + const dir = fs.mkdtempSync(join(tmpdir(), "opencomms-atomic-")) + try { + const store = fixture.create(dir) + store.save(1) + const previous = fs.readFileSync(store.file, "utf8") + const rename = fs.renameSync + const error = Object.assign(new Error("replacement denied"), { code: "EPERM" }) + let attempts = 0 + t.mock.method(fs, "renameSync", (from: fs.PathLike, to: fs.PathLike) => { + attempts++ + if (succeeds && attempts === 2) return rename(from, to) + throw error + }) + syncBuiltinESMExports() + if (succeeds) { + store.save(2) + assert.equal(JSON.parse(fs.readFileSync(store.file, "utf8")).revision, 2) + assert.equal(attempts, 2) + } else { + assert.throws(() => store.save(2), /failed to persist/) + assert.equal(fs.readFileSync(store.file, "utf8"), previous) + assert.ok(attempts >= 2 && attempts <= 3) + } + assert.deepEqual( + fs.readdirSync(store.dir).filter((name) => name.endsWith(".tmp")), + [], + ) + } finally { + t.mock.restoreAll() + syncBuiltinESMExports() + fs.rmSync(dir, { recursive: true, force: true }) + } + }) + } +} +for (const fixture of stores) { + test(`${fixture.name}: partial temporary write is cleaned up without touching state`, (t) => { + const dir = fs.mkdtempSync(join(tmpdir(), "opencomms-atomic-")) + try { + const store = fixture.create(dir) + store.save(1) + const previous = fs.readFileSync(store.file, "utf8") + const write = fs.writeFileSync + const error = Object.assign(new Error("disk full"), { code: "ENOSPC" }) + t.mock.method(fs, "writeFileSync", (file: fs.PathOrFileDescriptor) => { + write(file, "partial") + throw error + }) + const rename = t.mock.method(fs, "renameSync") + syncBuiltinESMExports() + assert.throws( + () => store.save(2), + (caught: unknown) => { + assert.ok(caught instanceof Error) + assert.match(caught.message, /failed to persist/) + assert.equal(caught.cause, error) + return true + }, + ) + assert.equal(rename.mock.callCount(), 0) + assert.equal(fs.readFileSync(store.file, "utf8"), previous) + assert.deepEqual( + fs.readdirSync(store.dir).filter((name) => name.endsWith(".tmp")), + [], + ) + } finally { + t.mock.restoreAll() + syncBuiltinESMExports() + fs.rmSync(dir, { recursive: true, force: true }) + } + }) +}