fix(persist): fence async commits against late-landing stale writes (#373)

src/persist/store.ts

@@ -158,6 +158,12 @@ export class StateStore<T> {
     private readonly writeChains = new Map<string, Promise<void>>();
     /** id → absolute path, populated by writes and loadAll. */
     private readonly discovered = new Map<string, string>();
+    /** 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<string, number>();
     /** id → cumulative write-failure count, for rate-limited alerting. */
     private readonly failCounts = new Map<string, number>();
     /** Monotonic counter for unique temp filenames within a process. */
@@ -222,7 +228,9 @@ export class StateStore<T> {
     /** 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<T> {
         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<T> {
             }
         }
         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<T> {
     }
 
     private async writeInner(id: string, build: () => T): Promise<void> {
+        // 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<T> {
             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<void>();
+                    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<T> {
                 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<T> {
         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<T> {
                 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<T> {
         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));

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 });
+    }
+});