diff --git a/src/persist/index.ts b/src/persist/index.ts index 942e76c..5289c6b 100644 --- a/src/persist/index.ts +++ b/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"; diff --git a/src/persist/store.ts b/src/persist/store.ts index 2bc3e62..0b36600 100644 --- a/src/persist/store.ts +++ b/src/persist/store.ts @@ -88,6 +88,29 @@ export interface StateStoreOptions { 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 { /** 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>> { + * 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>> { const out = new Map>(); 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 { } 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 { + 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(); + 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(envelope: PersistedEnvelope): boolean { diff --git a/tests/persist.test.ts b/tests/persist.test.ts index 70b323b..02b7575 100644 --- a/tests/persist.test.ts +++ b/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 = { 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 }); + } +});