From f3df3eb3b4e9a30b5dbaa2ddcbb7fa406aa471df Mon Sep 17 00:00:00 2001 From: ranxianglei Date: Fri, 25 Sep 2026 14:19:42 +0800 Subject: [PATCH] fix(persist): fence async commits against late-landing stale writes (#373) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit writeInner snapshots the payload, then commits via async rename. A flushSync writing a NEWER payload could return, and the stale in-flight rename could land afterward and clobber it — the flushSync success contract (disk holds the newest payload on return) was violated, and callers that drop in-memory state on success could lose data. Root fix: per-id commit fence. Every committed write (sync or async) bumps a seq. writeInner captures the seq BEFORE build() and re-checks it inside a SYNCHRONOUS commit block (fence check + renameSync), so the event loop cannot interleave a flushSync between check and rename. The spill fallback gets the same fenced synchronous write (its slot is shared per id, same clobber shape). Superseded writes settle quietly — no failure recorded, no throw into the chain; a newer payload owns disk. Regression test uses a flushSync re-entering from inside the async builder's build(), making the interleave deterministic: fails on the old code (disk holds the stale snapshot), passes on the fence. Spill and retry-path tests migrate their failure-injection seam from the async fsp.rename surface to the new synchronous commit surface; their intent is unchanged. --- src/persist/store.ts | 59 ++++++++++++++++++++++++++-- tests/persist.test.ts | 89 ++++++++++++++++++++++++++++++++++++++----- 2 files changed, 134 insertions(+), 14 deletions(-) diff --git a/src/persist/store.ts b/src/persist/store.ts index f839a14..f9f3725 100644 --- a/src/persist/store.ts +++ b/src/persist/store.ts @@ -158,6 +158,12 @@ export class StateStore { private readonly writeChains = new Map>(); /** id → absolute path, populated by writes and loadAll. */ private readonly discovered = new Map(); + /** id → seq of the last COMMITTED write (sync or async). Async writers + * capture the seq at snapshot time and re-check it inside their + * synchronous commit block; a flushSync (or a newer committed write) + * bumping the seq makes the stale in-flight write abort its commit + * instead of clobbering the newer on-disk payload. */ + private readonly commitSeq = new Map(); /** id → cumulative write-failure count, for rate-limited alerting. */ private readonly failCounts = new Map(); /** Monotonic counter for unique temp filenames within a process. */ @@ -222,7 +228,9 @@ export class StateStore { /** Synchronous flush for one id. Used where the caller cannot await * (memory eviction, sync shutdown paths). Cancels any pending debounce * timer. Returns true on success, false on failure — callers that use - * the result to drop in-memory state must NOT drop it on failure. */ + * the result to drop in-memory state must NOT drop it on failure. + * Committing bumps the per-id fence so a stale in-flight async write + * cannot land its older payload after this returns. */ flushSync(id: string, build: () => T): boolean { if (!this.enabled) return true; const timer = this.timers.get(id); @@ -266,6 +274,7 @@ export class StateStore { if (!lastErr) { this.discovered.set(id, file); this.clearFailure(id); + this.bumpCommit(id); this.removeSpillSync(file); return true; } @@ -286,6 +295,7 @@ export class StateStore { } } if (spillPath) this.discovered.set(id, spillPath); + if (spillPath) this.bumpCommit(id); this.recordFailure(id, lastErr, spillPath); return spillPath !== null; } @@ -393,6 +403,10 @@ export class StateStore { } private async writeInner(id: string, build: () => T): Promise { + // Fence point, captured BEFORE build(): from here on, any write that + // commits (a flushSync re-entering from inside build(), or a newer + // chained write) bumps the seq and this snapshot is stale. + const baseSeq = this.commitSeq.get(id) ?? 0; let payload: T; try { payload = build(); @@ -414,9 +428,28 @@ export class StateStore { try { await fsp.mkdir(path.dirname(file), { recursive: true }); await fsp.writeFile(tmp, data); - await fsp.rename(tmp, file); + // Commit block: fence check + rename in one synchronous run + // — the event loop cannot interleave a flushSync between the + // check and the rename, so a stale payload can never land + // after a newer committed write. + if ((this.commitSeq.get(id) ?? 0) !== baseSeq) { + await fsp.unlink(tmp).catch(() => {}); + return; + } + try { + fs.renameSync(tmp, file); + } catch (e) { + lastErr = e; + await fsp.unlink(tmp).catch(() => {}); + if (!isTransientFsError(e) || attempt === this.retryAttempts - 1) break; + const { promise, resolve } = Promise.withResolvers(); + setTimeout(resolve, this.backoffMs(attempt)); + await promise; + continue; + } this.discovered.set(id, file); this.clearFailure(id); + this.commitSeq.set(id, baseSeq + 1); await this.removeSpill(file); return; } catch (e) { @@ -430,6 +463,10 @@ export class StateStore { await promise; } } + // Fenced off before spilling: a newer committed write already owns + // the disk record for this id — the stale payload is dropped, not + // spilled (its spill would clobber the newer spill on the same slot). + if ((this.commitSeq.get(id) ?? 0) !== baseSeq) return; // The canonical write outlived the retry window (Windows lock held // by AV/indexer/SMB, or a sweeper keeps deleting the fresh temp). // Spill to a side file so the data still lands on disk instead of @@ -442,7 +479,11 @@ export class StateStore { for (let attempt = 0; attempt < this.retryAttempts && spillPath === null; attempt++) { try { await fsp.mkdir(path.dirname(spill), { recursive: true }); - await fsp.writeFile(spill, data); + // Fence + write synchronously: the spill slot is shared per + // id, so a late-landing stale spill would overwrite a newer + // spill exactly like the canonical case above. + if ((this.commitSeq.get(id) ?? 0) !== baseSeq) return; + fs.writeFileSync(spill, data); spillPath = spill; } catch (e) { if (!isTransientFsError(e) || attempt === this.retryAttempts - 1) break; @@ -451,7 +492,10 @@ export class StateStore { await promise; } } - if (spillPath) this.discovered.set(id, spillPath); + if (spillPath) { + this.discovered.set(id, spillPath); + this.commitSeq.set(id, baseSeq + 1); + } this.recordFailure(id, lastErr, spillPath); throw lastErr; } @@ -513,6 +557,13 @@ export class StateStore { await fsp.unlink(this.spillPathFor(file)).catch(() => {}); } + /** Advance the per-id commit fence after a successful synchronous + * commit, fencing off any in-flight async write holding an older + * snapshot. */ + private bumpCommit(id: string): void { + this.commitSeq.set(id, (this.commitSeq.get(id) ?? 0) + 1); + } + private removeSpillSync(file: string): void { try { fs.unlinkSync(this.spillPathFor(file)); diff --git a/tests/persist.test.ts b/tests/persist.test.ts index d7e681a..ffcd91b 100644 --- a/tests/persist.test.ts +++ b/tests/persist.test.ts @@ -493,7 +493,7 @@ test("rename failure spills to a side file instead of dropping data", async (t) const dir = tmpDir(); try { const s = store(dir, { retryAttempts: 2, retryBaseMs: 1, retryMaxMs: 2 }); - t.mock.method(fsp, "rename", async () => { + t.mock.method(fs, "renameSync", () => { throw eperm(); }); let threw = false; @@ -561,7 +561,7 @@ test("write failures alert on the first and power-of-two counts, not every write retryMaxMs: 2, log: (level, msg) => logs.push({ level, msg }), }); - t.mock.method(fsp, "rename", async () => { + t.mock.method(fs, "renameSync", () => { throw eperm(); }); for (let i = 0; i < 5; i++) { @@ -613,13 +613,13 @@ test("a successful write clears the failure counter so alerting restarts", async retryMaxMs: 2, log: (level, msg) => logs.push({ level, msg }), }); - t.mock.method(fsp, "rename", async () => { + t.mock.method(fs, "renameSync", () => { throw eperm(); }); await s.writeNow("sid-reset", () => ({ label: "a", count: 0 })).catch(() => {}); t.mock.restoreAll(); await s.writeNow("sid-reset", () => ({ label: "b", count: 1 })); - t.mock.method(fsp, "rename", async () => { + t.mock.method(fs, "renameSync", () => { throw eperm(); }); await s.writeNow("sid-reset", () => ({ label: "c", count: 2 })).catch(() => {}); @@ -668,13 +668,13 @@ test("a temp file vanishing before rename (ENOENT) is healed by whole-cycle retr const dir = tmpDir(); try { const s = store(dir, { retryAttempts: 3, retryBaseMs: 1, retryMaxMs: 1 }); - const realRename = fsp.rename.bind(fsp); + const realRenameSync = fs.renameSync.bind(fs); let calls = 0; - t.mock.method(fsp, "rename", (async (src: string, dest: string) => { + t.mock.method(fs, "renameSync", ((src: string, dest: string) => { calls++; if (calls === 1) throw enoent(); - await realRename(src, dest); - }) as typeof fsp.rename); + realRenameSync(src, dest); + }) as typeof fs.renameSync); await s.writeNow("sid-sweep", () => ({ label: "healed", count: 3 })); const canonical = path.join(dir, flatFileNameFor("sid-sweep")); const env = JSON.parse(readFileSync(canonical, "utf8")) as { id: string; payload: Payload }; @@ -713,7 +713,7 @@ test("persistent ENOENT still spills so data is never dropped", async (t) => { const dir = tmpDir(); try { const s = store(dir, { retryAttempts: 2, retryBaseMs: 1, retryMaxMs: 2 }); - t.mock.method(fsp, "rename", (async () => { + t.mock.method(fs, "renameSync", (() => { throw enoent(); }) as typeof fsp.rename); let threw = false; @@ -809,7 +809,7 @@ test("codec applies to spill writes as well", async (t) => { const dir = tmpDir(); try { const s = store(dir, { codec: xorCodec(), retryAttempts: 2, retryBaseMs: 1, retryMaxMs: 2 }); - t.mock.method(fsp, "rename", async () => { + t.mock.method(fs, "renameSync", () => { throw eperm(); }); let threw = false; @@ -883,3 +883,72 @@ test("mixed tree: unencoded legacy files and encoded files both load", async () rmSync(dir, { recursive: true, force: true }); } }); + +test("#373: a flushSync landing while an async write is mid-flight cannot be clobbered by the stale rename", async () => { + const dir = tmpDir(); + try { + const s = store(dir); + // Reentrant interleave: while writeInner is between its build() and + // its commit, a flushSync commits a NEWER payload. Deterministic — + // the reentrant call runs inside build() itself. + const inFlight = s.writeNow("sid-fence", () => { + const snap: Payload = { label: "stale", count: 1 }; + const ok = s.flushSync("sid-fence", () => ({ label: "fresh", count: 2 })); + assert.equal(ok, true, "reentrant flushSync must succeed"); + return snap; + }); + await inFlight; + const loaded = s.loadSync("sid-fence"); + assert.ok(loaded, "record must exist"); + assert.deepEqual( + loaded.payload, + { label: "fresh", count: 2 }, + "disk must hold the flushSync payload, not the stale in-flight snapshot", + ); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test("#373: fenced async writes that lose the race settle quietly and never throw into the chain", async () => { + const dir = tmpDir(); + try { + const s = store(dir); + let observed: Payload | null = null; + const p1 = s.writeNow("sid-chain", () => { + const snap: Payload = { label: "older", count: 1 }; + s.flushSync("sid-chain", () => ({ label: "newer", count: 5 })); + return snap; + }); + await p1; + // A queued write built AFTER the flushSync commits is NOT fenced — + // it snapshots later state and must land. + await s.writeNow("sid-chain", () => { + observed = { label: "latest", count: 9 }; + return observed; + }); + const loaded = s.loadSync("sid-chain"); + assert.ok(loaded); + assert.deepEqual(loaded.payload, { label: "latest", count: 9 }); + // The fenced older write left no tmp orphans behind. + const leftovers = readdirSync(dir).filter((f) => f.startsWith(".tmp-")); + assert.deepEqual(leftovers, []); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test("#373: flushSync after a fully settled async write simply supersedes it", async () => { + const dir = tmpDir(); + try { + const s = store(dir); + await s.writeNow("sid-order", () => ({ label: "async-first", count: 1 })); + const ok = s.flushSync("sid-order", () => ({ label: "sync-second", count: 2 })); + assert.equal(ok, true); + const loaded = s.loadSync("sid-order"); + assert.ok(loaded); + assert.deepEqual(loaded.payload, { label: "sync-second", count: 2 }); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +});