fix(responses): stop rewriting an unchanged snapshot every two seconds

docs-site/src/content/docs/troubleshooting/disk-usage-temp-files.md

@@ -50,6 +50,25 @@ running process can own those.
 The safety rules are unchanged: a file younger than 15 minutes is never removed,
 and the proxy never removes a file it is writing itself.
 
+## How often the snapshot is written
+
+Writes are debounced, and the debounce is derived from the size of the **last
+snapshot actually written**: while that file is small the next write is scheduled
+about two seconds after a change, and once it is near the 24 MB bound the wait
+stretches to at most thirty seconds. A cache that has only just grown therefore
+still takes the short wait once — the longer cadence applies from the write after
+it. A flush that would reproduce the existing file byte-for-byte is skipped
+entirely.
+
+Together these keep the write rate roughly flat as the cache grows, instead of
+re-serializing and replacing the whole file every two seconds.
+
+A graceful shutdown flushes immediately rather than waiting out the timer, so the
+longer wait mainly widens the window in which a hard kill loses the most recent
+continuation entries — which are cache, as above. That flush is still a disk
+write and can fail like any other, so a shutdown on a full or read-only volume
+can lose the same entries.
+
 ## Reclaiming files that already accumulated
 
 If the proxy runs, this happens automatically within a minute or two.

src/responses/state.ts

@@ -18,6 +18,12 @@ import {
 const MAX_STORED_RESPONSES = 1_000;
 const RESPONSE_TTL_MS = 60 * 60 * 1_000;
 const SNAPSHOT_DEBOUNCE_MS = 2_000;
+/** Snapshot size below which the debounce stays at its base value. */
+const SNAPSHOT_DEBOUNCE_SCALE_FROM_BYTES = 1 * 1024 * 1024;
+/** Ceiling for the stretched debounce. Continuation state is only read after a
+ *  restart, and a graceful shutdown flushes, so the exposure a longer debounce adds
+ *  is bounded by a hard kill — paid against rewriting the whole snapshot every 2 s. */
+const SNAPSHOT_DEBOUNCE_MAX_MS = 30_000;
 /** In-memory high-water byte cap across all entries. Forced store:false retention (kiro/cursor
  * continuation chains) stores the full expanded input each turn — ~quadratic bytes per chain —
  * so a count cap alone cannot bound memory. Oldest-first eviction applies past this mark. */
@@ -90,6 +96,32 @@ let oldestResidentId: string | undefined;
 let oldestResidentAt: number | null = null;
 let byteCapOverride: number | null = null;
 let stateRevision = 0;
+/** Byte length and digest of the last snapshot actually written, for the
+ *  identical-payload skip and the size-scaled debounce. The payload itself is not
+ *  retained: at the 24 MiB bound that would double the snapshot's memory cost. */
+let lastSnapshotBytes = 0;
+let lastSnapshotDigest: string | null = null;
+// The resolved file the digest above describes. Keeping it means a config-dir
+// change or a retargeted symlink is a miss rather than a false "unchanged".
+let lastSnapshotTarget: string | null = null;
+
+/**
+ * Is the snapshot on disk still byte-for-byte what we last wrote?
+ *
+ * The cached digest proves what this process wrote, not what is there now. Size is
+ * checked first so the common mismatch costs a `stat`, and the content comparison
+ * only runs when the size already agrees. Any read failure answers "no" and the
+ * caller rewrites — the safe direction.
+ */
+async function snapshotOnDiskMatches(path: string, payload: string, payloadBytes: number): Promise<boolean> {
+  try {
+    const file = Bun.file(path);
+    if (file.size !== payloadBytes) return false;
+    return await file.text() === payload;
+  } catch {
+    return false;
+  }
+}
 const spillCounters = { writes: 0, writeFailures: 0, readFailures: 0 };
 /**
  * Admission-boundary observability (test-visible). directSpills: oversized
@@ -788,9 +820,38 @@ async function writeBoundedSnapshot(path: string): Promise<SnapshotWriteOutcome>
         entries.push(persistEntry);
       }
       entries.reverse();
-      mkdirSync(dirname(path), { recursive: true, mode: 0o700 });
-      try { chmodSync(dirname(path), 0o700); } catch { /* best-effort (e.g. Windows) */ }
-      await atomicWriteFileAsync(path, JSON.stringify({ version: 2, states: entries }));
+      const payload = JSON.stringify({ version: 2, states: entries });
+      const payloadBytes = Buffer.byteLength(payload, "utf8");
+      const payloadDigest = Bun.hash(payload).toString(36);
+      // A mutation does not always change what gets persisted: entries past the
+      // per-entry or total byte bound are dropped from the selection, and spill
+      // demotion moves bytes out of it. Re-writing a byte-identical 24 MiB file
+      // buys nothing, so compare first — but the cached digest describes what THIS
+      // process last wrote, which is not the same claim as "that is what is on disk
+      // now". A second proxy sharing the home, or anything that rewrites the file
+      // in place, leaves the digest describing bytes that are gone. Before every
+      // release-of-a-write, the previous behaviour rewrote unconditionally and so
+      // repaired that silently; skipping without checking would turn a repaired
+      // snapshot into a lost one at the next restart.
+      //
+      // Verify against the file itself, keyed to the resolved target so a retargeted
+      // symlink is also a miss. Reading back a matching-size file costs far less
+      // than the atomic replace it avoids, and only happens when the digest already
+      // matched — the amplification this fixes is the repeated WRITE, not the read.
+      const unchanged = lastSnapshotDigest !== null
+        && payloadDigest === lastSnapshotDigest
+        && payloadBytes === lastSnapshotBytes
+        && lastSnapshotTarget === resolveWriteTarget(path)
+        && existsSync(path)
+        && await snapshotOnDiskMatches(path, payload, payloadBytes);
+      if (!unchanged) {
+        mkdirSync(dirname(path), { recursive: true, mode: 0o700 });
+        try { chmodSync(dirname(path), 0o700); } catch { /* best-effort (e.g. Windows) */ }
+        await atomicWriteFileAsync(path, payload);
+        lastSnapshotDigest = payloadDigest;
+        lastSnapshotBytes = payloadBytes;
+        lastSnapshotTarget = resolveWriteTarget(path);
+      }
       persistAttemptHookForTests?.();
       if (revision === stateRevision) return "stable";
     }
@@ -809,11 +870,26 @@ function drainPendingSpillUnlinks(): void {
   }
 }
 
+/**
+ * Debounce scaled by the size of the last snapshot written.
+ *
+ * The whole snapshot is re-serialized and atomically replaced on every flush, so at
+ * the 24 MiB bound a fixed 2 s debounce is up to ~12 MB/s of write amplification for
+ * state nothing reads until the next start (#2460). Small snapshots keep the base
+ * cadence; the stretch is linear in size and clamped, so the write rate is roughly
+ * flat instead of growing with the file.
+ */
+function snapshotDebounceMs(): number {
+  if (lastSnapshotBytes <= SNAPSHOT_DEBOUNCE_SCALE_FROM_BYTES) return SNAPSHOT_DEBOUNCE_MS;
+  const scaled = Math.round(SNAPSHOT_DEBOUNCE_MS * (lastSnapshotBytes / SNAPSHOT_DEBOUNCE_SCALE_FROM_BYTES));
+  return Math.min(scaled, SNAPSHOT_DEBOUNCE_MAX_MS);
+}
+
 function schedulePersistAt(path: string, replace = false): void {
   if (persistTimer && !replace) return;
   if (persistTimer) clearTimeout(persistTimer);
   pendingPersistPath = path;
-  persistTimer = setTimeout(() => { void persistNow(path); }, SNAPSHOT_DEBOUNCE_MS);
+  persistTimer = setTimeout(() => { void persistNow(path); }, snapshotDebounceMs());
   (persistTimer as { unref?: () => void }).unref?.();
 }
 
@@ -1418,6 +1494,9 @@ export function clearResponseStateMemoryForTests(): void {
   replayScopeMismatchDrops = 0;
   replayOverlapSkips = 0;
   persistAttemptHookForTests = null;
+  lastSnapshotBytes = 0;
+  lastSnapshotDigest = null;
+  lastSnapshotTarget = null;
   loaded = false;
 }
 

tests/responses-state-write-amplification.test.ts

@@ -0,0 +1,218 @@
+/**
+ * #2460 — `responses-state.json` was rewritten in full on a fixed 2 s debounce.
+ *
+ * The snapshot is bounded at 24 MiB, so under sustained traffic every debounce
+ * paid a complete re-serialization plus an atomic replacement of a file nothing
+ * reads until the next start. Two narrow measures are covered here: a
+ * byte-identical payload is not rewritten, and the debounce scales with the size
+ * of the snapshot actually being written.
+ */
+import { afterEach, beforeEach, describe, expect, test } from "bun:test";
+import { existsSync, mkdtempSync, readFileSync, rmSync, statSync, utimesSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import {
+  clearResponseStateForTests,
+  clearResponseStateMemoryForTests,
+  flushResponseState,
+  rememberResponseState,
+  setResponseStateByteCapForTests,
+} from "../src/responses/state";
+
+function remember(id: string, text: string): void {
+  rememberResponseState(
+    { model: "test/model", input: text, store: false },
+    { id, output: [{ type: "message", role: "assistant", content: text }], status: "completed" },
+    undefined,
+    { force: true },
+  );
+}
+
+describe("responses-state snapshot write amplification (#2460)", () => {
+  let home: string;
+  const priorHome = process.env["OPENCODEX_HOME"];
+  let snapshot: string;
+
+  beforeEach(() => {
+    home = mkdtempSync(join(tmpdir(), "ocx-state-amp-"));
+    process.env["OPENCODEX_HOME"] = home;
+    snapshot = join(home, "responses-state.json");
+    clearResponseStateMemoryForTests();
+  });
+
+  afterEach(() => {
+    setResponseStateByteCapForTests(null);
+    clearResponseStateForTests();
+    rmSync(home, { recursive: true, force: true });
+    if (priorHome === undefined) delete process.env["OPENCODEX_HOME"];
+    else process.env["OPENCODEX_HOME"] = priorHome;
+  });
+
+  /** Record the delay the store hands to setTimeout when it schedules its next write. */
+  function scheduledDelay(schedule: () => void): number {
+    const realSetTimeout = globalThis.setTimeout;
+    const delays: number[] = [];
+    globalThis.setTimeout = ((handler: TimerHandler, ms?: number, ...rest: unknown[]) => {
+      delays.push(ms ?? 0);
+      return (realSetTimeout as (...a: unknown[]) => unknown)(handler, ms, ...rest);
+    }) as unknown as typeof setTimeout;
+    try {
+      schedule();
+    } finally {
+      globalThis.setTimeout = realSetTimeout;
+    }
+    expect(delays).toHaveLength(1);
+    return delays[0]!;
+  }
+
+  /** Backdate the snapshot so "was it rewritten?" is a mtime comparison, not a clock race. */
+  function backdate(): number {
+    const past = new Date(Date.now() - 60_000);
+    utimesSync(snapshot, past, past);
+    return statSync(snapshot).mtimeMs;
+  }
+
+  test("a flush that would reproduce the same bytes does not rewrite the file", async () => {
+    remember("resp_amp_small", "kept");
+    await flushResponseState();
+    expect(existsSync(snapshot)).toBe(true);
+    const before = backdate();
+
+    // An entry past the 2 MiB per-entry bound is dropped from the selection, so
+    // recording it mutates state without changing a single persisted byte.
+    remember("resp_amp_oversized", "x".repeat(3 * 1024 * 1024));
+    await flushResponseState();
+
+    expect(statSync(snapshot).mtimeMs).toBe(before);
+  });
+
+  test("a flush that changes the payload still rewrites the file", async () => {
+    remember("resp_amp_first", "first");
+    await flushResponseState();
+    const before = backdate();
+
+    remember("resp_amp_second", "second");
+    await flushResponseState();
+
+    expect(statSync(snapshot).mtimeMs).toBeGreaterThan(before);
+  });
+
+  test("a snapshot deleted underneath us is rewritten even when the payload matches", async () => {
+    remember("resp_amp_restore", "kept");
+    await flushResponseState();
+    rmSync(snapshot, { force: true });
+
+    remember("resp_amp_oversized_2", "y".repeat(3 * 1024 * 1024));
+    await flushResponseState();
+
+    expect(existsSync(snapshot)).toBe(true);
+  });
+
+  // Deletion is the easy half. The cached digest describes what THIS process last
+  // wrote, which is not the same claim as "that is what is on disk now" — a second
+  // proxy sharing the home, or anything rewriting the file in place, leaves the
+  // digest describing bytes that are gone. Before the skip existed, every flush
+  // rewrote and so repaired that silently; skipping on the digest alone would turn
+  // a self-healing snapshot into a permanently corrupt one, discovered only at the
+  // next restart when the continuation state fails to load.
+  //
+  // Same-length replacement is the case that defeats a size-only check, so that is
+  // what this drives.
+  test("a snapshot replaced with different bytes of the same length is rewritten", async () => {
+    remember("resp_amp_replaced", "kept");
+    await flushResponseState();
+
+    const original = readFileSync(snapshot, "utf-8");
+    writeFileSync(snapshot, "X".repeat(Buffer.byteLength(original, "utf8")));
+
+    // A mutation whose bounded payload is byte-identical to the last write: the
+    // oversized entry is dropped by the per-entry bound, so the digest still matches.
+    remember("resp_amp_replaced_oversized", "y".repeat(3 * 1024 * 1024));
+    await flushResponseState();
+
+    expect(readFileSync(snapshot, "utf-8")).toBe(original);
+  });
+
+  test("the scheduled debounce stays at its base value for a small snapshot", async () => {
+    remember("resp_amp_tiny", "tiny");
+    await flushResponseState();
+
+    const delay = scheduledDelay(() => remember("resp_amp_tiny_2", "tiny"));
+    await flushResponseState();
+
+    expect(delay).toBe(2_000);
+  });
+
+  test("the scheduled debounce stretches once the snapshot is large", async () => {
+    // Four ~800 KiB entries: each under the 2 MiB per-entry bound, so all four are
+    // persisted and the payload lands well past the 1 MiB scaling threshold.
+    for (let i = 0; i < 4; i += 1) remember(`resp_amp_big_${i}`, "z".repeat(800 * 1024));
+    await flushResponseState();
+
+    const delay = scheduledDelay(() => remember("resp_amp_big_next", "next"));
+    await flushResponseState();
+
+    expect(delay).toBeGreaterThan(2_000);
+    expect(delay).toBeLessThanOrEqual(30_000);
+    // Roughly proportional to size: ~3.2 MiB of payload is ~6 s, not ~2 s.
+    expect(delay).toBeGreaterThanOrEqual(5_000);
+  });
+
+  // The ceiling is the half of the scaling rule that a proportional formula gets
+  // wrong silently: `<= 30_000` passes for any well-behaved input, so it proves
+  // the clamp exists only when something actually reaches it. A snapshot at the
+  // byte cap is the case that does.
+  test("the scheduled debounce clamps to exactly 30s at the snapshot cap", async () => {
+    // The scaling rule is 2s per MiB, so a payload at or past 15 MiB computes above
+    // the ceiling. 900 KB of text per entry matters: the entry is serialized with the
+    // input echoed alongside the output, so ~1.5 MB of text lands past the 2 MiB
+    // per-entry bound and is skipped entirely, leaving an empty snapshot.
+    for (let i = 0; i < 20; i += 1) remember(`resp_amp_cap_${i}`, "c".repeat(900 * 1024));
+    await flushResponseState();
+    // Guard the premise: if the snapshot did not actually get large, a passing
+    // clamp assertion below would be proving nothing.
+    expect(statSync(snapshot).size).toBeGreaterThan(15 * 1024 * 1024);
+
+    const delay = scheduledDelay(() => remember("resp_amp_cap_next", "next"));
+    await flushResponseState();
+
+    expect(delay).toBe(30_000);
+  });
+
+  // Shutdown is the one moment the debounce must not apply: a pending write that
+  // waits out a 30s timer during a graceful stop is a lost turn, and the longer
+  // the debounce grows the more there is to lose. The existing tests all flush
+  // immediately after each change, so none of them proves the pending case.
+  test("a graceful flush writes changes still sitting behind the debounce", async () => {
+    remember("resp_amp_drain_seed", "seed");
+    await flushResponseState();
+
+    // Two changes with NO flush between them: the debounce timer is pending and
+    // neither has reached disk when the graceful flush arrives.
+    remember("resp_amp_drain_late", "late");
+    remember("resp_amp_drain_late2", "late2");
+
+    await flushResponseState();
+
+    const parsed = JSON.parse(await Bun.file(snapshot).text()) as {
+      states: [string, Record<string, unknown>][];
+    };
+    const ids = parsed.states.map(([id]) => id);
+    expect(ids).toContain("resp_amp_drain_late");
+    expect(ids).toContain("resp_amp_drain_late2");
+  });
+
+  test("the snapshot still round-trips after a skipped write", async () => {
+    remember("resp_amp_roundtrip", "payload");
+    await flushResponseState();
+    remember("resp_amp_roundtrip_oversized", "w".repeat(3 * 1024 * 1024));
+    await flushResponseState();
+
+    const parsed = JSON.parse(await Bun.file(snapshot).text()) as {
+      version: number;
+      states: [string, Record<string, unknown>][];
+    };
+    expect(parsed.version).toBe(2);
+    expect(parsed.states.map(([id]) => id)).toContain("resp_amp_roundtrip");
+  });
+});