feat: StateStore.loadAll({ maxParseBytes }) — stat-only parse budget for sub-linear startup

src/persist/index.ts

@@ -14,5 +14,5 @@
  * - the store never deletes files; cleanup is a downstream decision
  */
 export { StateStore, flatFileNameFor } from "./store.js";
-export type { PersistedEnvelope, PersistLogger, StateStoreOptions } from "./store.js";
+export type { LoadAllOptions, PersistedEnvelope, PersistLogger, StateStoreOptions } from "./store.js";
 export { mergeCompressionState } from "./state-merge.js";

src/persist/store.ts

@@ -88,6 +88,29 @@ export interface StateStoreOptions<T> {
     retryMaxMs?: number;
 }
 
+/** Options for {@link StateStore.loadAll}. */
+export interface LoadAllOptions {
+    /**
+     * Cap on the total bytes of record files loadAll will read and parse.
+     * When set, loadAll does a stat-only pass first (readdir + stat, no
+     * content reads), groups each canonical file with its `.fb.json` spill
+     * variant (same id ⇒ all-or-nothing, so the pair's freshest-wins
+     * reconciliation always sees both sides), then includes groups
+     * newest-mtime-first while the running total stays within the budget.
+     * A group that does not fit is skipped individually and selection
+     * continues with older groups — the byte cap is hard even when the
+     * newest group alone exceeds it; only if NO group fits is the result
+     * empty. Excluded groups are left on disk untouched (the store never deletes)
+     * and stay individually reachable via loadSync(id, hint). If no group
+     * fits, the result is an empty map (a warning is logged) — the budget
+     * is a hard cap, so one oversized record can never blow past it. Group
+     * freshness uses file mtime as a stat-level proxy for the in-envelope
+     * savedAt; ordering among parsed records still comes from savedAt
+     * reconciliation. Omitted ⇒ parse every record (default, unchanged).
+     */
+    maxParseBytes?: number;
+}
+
 /**
  * Crash-safe, debounce-coalescing JSON state store. Mechanism only — lifted
  * from billion-context's proxy SessionStore and generalized:
@@ -288,11 +311,20 @@ export class StateStore<T> {
     /** Load every record under dir. Populates the discovery map (enables
      *  loadSync for namespaced relPaths). Skips corrupt files, `.tmp-*`
      *  orphans, and records whose filename does not match their id — one
-     *  bad file never blocks boot. Never throws. */
-    async loadAll(): Promise<Map<string, PersistedEnvelope<T>>> {
+     *  bad file never blocks boot; filesystem and record problems never
+     *  throw. An invalid `maxParseBytes` (non-finite or negative) throws
+     *  TypeError fail-fast instead.
+     *
+     * With `options.maxParseBytes`, parsing is budget-limited via a stat-only
+     *  preselection pass (see {@link LoadAllOptions.maxParseBytes}); skipped
+     *  records are not deleted and remain loadable via loadSync(id, hint). */
+    async loadAll(options?: LoadAllOptions): Promise<Map<string, PersistedEnvelope<T>>> {
         const out = new Map<string, PersistedEnvelope<T>>();
         if (!this.enabled) return out;
-        const files = await this.walkJsonFiles(this.dir);
+        const files =
+            options?.maxParseBytes == null
+                ? await this.walkJsonFiles(this.dir)
+                : await this.selectWithinBudget(options.maxParseBytes);
         for (const file of files) {
             const envelope = this.readEnvelope(file);
             if (!envelope) continue;
@@ -573,6 +605,63 @@ export class StateStore<T> {
         }
         return out;
     }
+
+    /** Stat-only preselection for budgeted loadAll: pairs each canonical
+     *  file with its `.fb.json` spill under a dir+stem key (identical stems
+     *  in different directories are different records), sums group sizes,
+     *  and fills the budget newest-mtime-first. Never reads file content. */
+    private async selectWithinBudget(budget: number): Promise<string[]> {
+        if (!Number.isFinite(budget) || budget < 0) {
+            throw new TypeError(
+                `maxParseBytes must be a finite non-negative number, got ${String(budget)}`,
+            );
+        }
+        const entries = await Promise.all(
+            (await this.walkJsonFiles(this.dir)).map(async (file) => {
+                try {
+                    const st = await fsp.stat(file);
+                    return { file, size: st.size, mtimeMs: st.mtimeMs };
+                } catch {
+                    return null; // vanished or unreadable between walk and stat
+                }
+            }),
+        );
+        const groups = new Map<string, { key: string; files: string[]; size: number; mtimeMs: number }>();
+        for (const e of entries) {
+            if (!e) continue;
+            const base = path.basename(e.file);
+            const stem = base.endsWith(".fb.json")
+                ? base.slice(0, -".fb.json".length)
+                : base.slice(0, -".json".length);
+            const key = `${path.dirname(e.file)}\u0000${stem}`;
+            const g = groups.get(key) ?? { key, files: [], size: 0, mtimeMs: 0 };
+            g.files.push(e.file);
+            g.size += e.size;
+            g.mtimeMs = Math.max(g.mtimeMs, e.mtimeMs);
+            groups.set(key, g);
+        }
+        const ordered = [...groups.values()].sort(
+            (a, b) => b.mtimeMs - a.mtimeMs || b.size - a.size || (a.key < b.key ? -1 : a.key > b.key ? 1 : 0),
+        );
+        const selected: string[] = [];
+        let used = 0;
+        let skipped = 0;
+        for (const g of ordered) {
+            if (used + g.size > budget) {
+                skipped++;
+                continue;
+            }
+            selected.push(...g.files);
+            used += g.size;
+        }
+        if (skipped > 0) {
+            this.log(
+                "warn",
+                `[persist] loadAll budget ${budget} bytes: parsed ${ordered.length - skipped}/${ordered.length} record groups (skipped ${skipped})`,
+            );
+        }
+        return selected;
+    }
 }
 
 function defaultValidate<T>(envelope: PersistedEnvelope<T>): boolean {

tests/persist.test.ts

@@ -1,7 +1,7 @@
 import { test } from "node:test";
 import assert from "node:assert/strict";
 import fs from "node:fs";
-import { mkdtempSync, rmSync, writeFileSync, mkdirSync, readdirSync, readFileSync, existsSync } from "node:fs";
+import { mkdtempSync, rmSync, writeFileSync, mkdirSync, readdirSync, readFileSync, existsSync, statSync } from "node:fs";
 import fsp from "node:fs/promises";
 import { tmpdir } from "node:os";
 import path from "node:path";
@@ -656,3 +656,114 @@ test("flushSync retries use a fresh temp name per attempt (no tombstoned reuse)"
         rmSync(dir, { recursive: true, force: true });
     }
 });
+
+function writeRawRecord(dir: string, id: string, opts: { name?: string; savedAt?: number; mtimeSec?: number; pad?: number } = {}): string {
+    const file = path.join(dir, opts.name ?? flatFileNameFor(id));
+    mkdirSync(path.dirname(file), { recursive: true });
+    const payload: Record<string, unknown> = { label: id, count: 0 };
+    if (opts.pad !== undefined) payload.pad = "x".repeat(opts.pad);
+    writeFileSync(file, JSON.stringify({ version: 1, savedAt: opts.savedAt ?? 1_000_000, id, payload }), "utf8");
+    if (opts.mtimeSec !== undefined) fs.utimesSync(file, opts.mtimeSec, opts.mtimeSec);
+    return file;
+}
+
+test("loadAll maxParseBytes parses only the newest records within the budget", async () => {
+    const dir = tmpDir();
+    try {
+        const s = store(dir);
+        ["old-1", "mid-1", "mid-2", "new-1"].forEach((id, i) =>
+            writeRawRecord(dir, id, { pad: 1000, mtimeSec: 1000 + i * 100 }),
+        );
+        const size = fs.statSync(path.join(dir, flatFileNameFor("old-1"))).size;
+        const out = await s.loadAll({ maxParseBytes: 2 * size + 1 });
+        assert.deepEqual([...out.keys()].sort(), ["mid-2", "new-1"]);
+        // Without a budget the behavior is unchanged: everything parses.
+        assert.equal((await s.loadAll()).size, 4);
+        // Skipped records are not deleted and stay directly loadable.
+        assert.ok(existsSync(path.join(dir, flatFileNameFor("old-1"))), "skipped file stays on disk");
+        assert.equal(s.loadSync("old-1")?.payload.label, "old-1");
+    } finally {
+        rmSync(dir, { recursive: true, force: true });
+    }
+});
+
+test("loadAll budget skips an oversized newest group and fills from older groups", async () => {
+    const dir = tmpDir();
+    try {
+        const s = store(dir);
+        writeRawRecord(dir, "huge-new", { pad: 50_000, mtimeSec: 4000 });
+        writeRawRecord(dir, "old-a", { pad: 100, mtimeSec: 1000 });
+        writeRawRecord(dir, "old-b", { pad: 100, mtimeSec: 2000 });
+        const small = fs.statSync(path.join(dir, flatFileNameFor("old-a"))).size;
+        const out = await s.loadAll({ maxParseBytes: 2 * small + 1 });
+        assert.deepEqual([...out.keys()].sort(), ["old-a", "old-b"]);
+    } finally {
+        rmSync(dir, { recursive: true, force: true });
+    }
+});
+
+test("loadAll budget treats a canonical file and its spill as one all-or-nothing unit", async () => {
+    const dir = tmpDir();
+    try {
+        const s = store(dir);
+        writeRawRecord(dir, "dup-id", { savedAt: 100, mtimeSec: 3000, pad: 500 });
+        writeRawRecord(dir, "dup-id", { name: spillNameFor("dup-id"), savedAt: 200, mtimeSec: 3001, pad: 500 });
+        writeRawRecord(dir, "fresh-id", { mtimeSec: 4000, pad: 100 });
+        const dupSize = fs.statSync(path.join(dir, flatFileNameFor("dup-id"))).size;
+        const freshSize = fs.statSync(path.join(dir, flatFileNameFor("fresh-id"))).size;
+        // Room for the fresh record plus only one half of the pair → the pair is skipped whole.
+        const tight = await s.loadAll({ maxParseBytes: freshSize + dupSize });
+        assert.deepEqual([...tight.keys()], ["fresh-id"]);
+        // Room for the whole pair → both halves parse and reconcile by savedAt.
+        const roomy = await s.loadAll({ maxParseBytes: freshSize + 2 * dupSize });
+        assert.equal(roomy.get("dup-id")?.savedAt, 200);
+        assert.equal(roomy.get("dup-id")?.payload.label, "dup-id");
+    } finally {
+        rmSync(dir, { recursive: true, force: true });
+    }
+});
+
+test("loadAll budget smaller than every record parses nothing and warns", async () => {
+    const dir = tmpDir();
+    try {
+        const logs: { level: string; msg: string }[] = [];
+        const s = store(dir, { log: (level, msg) => logs.push({ level, msg }) });
+        writeRawRecord(dir, "a-1", { mtimeSec: 1000 });
+        writeRawRecord(dir, "b-1", { mtimeSec: 2000 });
+        const out = await s.loadAll({ maxParseBytes: 1 });
+        assert.equal(out.size, 0);
+        assert.ok(existsSync(path.join(dir, flatFileNameFor("a-1"))), "files remain on disk");
+        assert.ok(existsSync(path.join(dir, flatFileNameFor("b-1"))), "files remain on disk");
+        assert.ok(logs.some((l) => l.level === "warn" && l.msg.includes("budget")), "warn logged");
+    } finally {
+        rmSync(dir, { recursive: true, force: true });
+    }
+});
+
+test("loadAll rejects invalid budgets fail-fast", async () => {
+    const dir = tmpDir();
+    try {
+        const s = store(dir);
+        await assert.rejects(s.loadAll({ maxParseBytes: -1 }), TypeError);
+        await assert.rejects(s.loadAll({ maxParseBytes: Number.NaN }), TypeError);
+    } finally {
+        rmSync(dir, { recursive: true, force: true });
+    }
+});
+
+test("loadAll budget pairs canonical and spill within a directory only", async () => {
+    const dir = tmpDir();
+    try {
+        const relPath = (id: string): string =>
+            id === "p" ? "n1/same.json" : id === "q" ? "n2/same.json" : flatFileNameFor(id);
+        const s = store(dir, { relPath });
+        writeRawRecord(dir, "p", { name: "n1/same.json", mtimeSec: 3000, pad: 500 });
+        writeRawRecord(dir, "q", { name: "n2/same.json", mtimeSec: 4000, pad: 500 });
+        const size = fs.statSync(path.join(dir, "n1/same.json")).size;
+        // Equal-sized same-stem files in different dirs must NOT merge into one oversized group.
+        const out = await s.loadAll({ maxParseBytes: size + 1 });
+        assert.deepEqual([...out.keys()], ["q"]);
+    } finally {
+        rmSync(dir, { recursive: true, force: true });
+    }
+});