From b313731b80ff0b8616e327fdeb114f25addb91fc Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Wed, 23 Sep 2026 09:54:05 +0200 Subject: [PATCH] feat: add JSONL sidecar reclamation --- packages/durable/docs/pico-v5-handoff.md | 15 +- packages/durable/docs/pico-v5.md | 22 +- packages/durable/src/storage/jsonl/storage.ts | 280 ++++++++-- packages/durable/test/jsonl-storage.test.ts | 498 +++++++++++++++++- 4 files changed, 742 insertions(+), 73 deletions(-) diff --git a/packages/durable/docs/pico-v5-handoff.md b/packages/durable/docs/pico-v5-handoff.md index 15fef74b3..825017d63 100644 --- a/packages/durable/docs/pico-v5-handoff.md +++ b/packages/durable/docs/pico-v5-handoff.md @@ -79,9 +79,10 @@ arguments, and reads never expose backend-owned cached objects. JSONL creation has `fsync?: boolean`, defaulting to `false`. With `false`, append sidecars and then the marker without an explicit flush. With `true`, append all affected sidecars, flush each affected sidecar, and then append the main marker. -Do not explicitly flush `main.jsonl`. A main-only commit has no sidecars to -flush. Any uncertain append or flush failure poisons the open backend and -publishes no prepared in-memory mutation. +Do not explicitly flush `main.jsonl` for ordinary publication. A main-only commit +has no sidecars to flush. Any uncertain publication append or flush failure +poisons the open backend and publishes no prepared in-memory mutation. Package 5 +adds the separate post-publication flush required to authorize reclamation. Fault-test torn/short sidecar writes, failures between sidecars, every marker boundary, unconfirmed tails, missing confirmed data, poisoned writes, exact-byte @@ -92,7 +93,13 @@ conformance suite directly and after reopen. ## 5. JSONL reclamation Implement task-document retirement and current-only base reclamation using -committed markers, temporary replacement, rename, and descriptor invalidation. +committed markers and descriptor invalidation. Remove a sidecar directly when no +records remain; otherwise use temporary replacement and rename. With `fsync: +true`, flush `main.jsonl` once before destructive reclamation so the authorizing +marker cannot disappear while cleanup survives; if that flush fails, skip +reclamation without failing the already-published commit. Flush a non-empty +temporary replacement before rename. This is not publication flushing or main +compaction. Crash-test every rewrite/rename boundary. Verify that rewindable history is never reclaimed and default no-fsync behavior matches the specification. diff --git a/packages/durable/docs/pico-v5.md b/packages/durable/docs/pico-v5.md index 58d2a3df7..3e96ba963 100644 --- a/packages/durable/docs/pico-v5.md +++ b/packages/durable/docs/pico-v5.md @@ -2120,10 +2120,14 @@ JSONL creation accepts an `fsync` option that defaults to `false`. Without `fsync`, it guarantees ordinary process-crash consistency, not survival of power, host, kernel, or filesystem failure. With `fsync: true`, the backend appends all affected sidecar records, flushes each affected sidecar, and only -then appends the main marker. It does not explicitly flush `main.jsonl`; an -acknowledged tail commit may therefore still disappear, but a marker that -survives should not overtake its sidecar data. A main-only commit has no -sidecars to flush. +then appends the main marker. Ordinary publication does not explicitly flush +`main.jsonl`; an acknowledged tail commit may therefore still disappear, but a +marker that survives should not overtake its sidecar data. A main-only commit has +no sidecars to flush. Before destructive reclamation with `fsync: true`, the +backend flushes `main.jsonl` once so the authorizing marker cannot disappear +while its replacement or removal survives. If that flush fails, the committed +state remains published and reclamation is deferred. A non-empty temporary +replacement is also flushed before rename. Recovery: @@ -2135,10 +2139,12 @@ Recovery: record unnecessary. - Any uncertain append failure poisons the open backend. -Reclamation starts only after the authorizing base/retirement commits. It writes -a temporary replacement, renames it, and invalidates cached file descriptors so -future appends cannot target an unlinked inode. `main.jsonl` is not compacted in -the initial implementation. +Reclamation starts only after the authorizing base/retirement commits. When no +sidecar records remain, it removes the sidecar directly. Otherwise, it writes a +temporary replacement, renames it, and invalidates cached file descriptors so +future appends cannot target an unlinked inode. Flushing `main.jsonl` to +authorize reclamation does not compact it. `main.jsonl` is not compacted in the +initial implementation. ## 12. API footguns diff --git a/packages/durable/src/storage/jsonl/storage.ts b/packages/durable/src/storage/jsonl/storage.ts index 0ec5e59f8..36b62851b 100644 --- a/packages/durable/src/storage/jsonl/storage.ts +++ b/packages/durable/src/storage/jsonl/storage.ts @@ -19,6 +19,7 @@ import { MemoryStorage } from "../memory.ts"; const FORMAT_VERSION = 1; const MAIN_FILE = "main.jsonl"; +const RECLAIM_SUFFIX = ".reclaim"; const textDecoder = new TextDecoder("utf-8", { fatal: true }); type StoredTask = TaskRecord; @@ -92,6 +93,11 @@ const sidecarFileName = (kind: "doc" | "task", id: Id): string => `${kind}-${id} const isSidecarFileName = (name: string): boolean => /^(?:doc|task)-(?:0|[1-9]\d*)\.jsonl$/.test(name); +const isReclaimFileName = (name: string): boolean => /^(?:doc|task)-(?:0|[1-9]\d*)\.jsonl\.reclaim$/.test(name); + +const isCurrentOnly = (record: DocumentCreate): boolean => + record.scope.kind !== "conversation" || record.history === "latest"; + const sidecarKey = (file: string, seq: Seq, ordinal: number): string => JSON.stringify([file, seq, ordinal]); const jsonLine = (value: unknown): string => { @@ -233,21 +239,16 @@ export class JsonlStorage implements Storage { private readonly directory: string; private readonly mainPath: string; private readonly fsync: boolean; - private readonly memory: MemoryStorage; + private readonly memory = new MemoryStorage(); + private readonly currentOnlyDocuments = new Set(); + private readonly liveTaskSidecars = new Set(); private closed = false; private poisonError: JsonlStoragePoisonedError | undefined; - private constructor( - fs: FileSystem, - directory: string, - mainPath: string, - memory: MemoryStorage, - options: JsonlStorageOptions, - ) { + private constructor(fs: FileSystem, directory: string, mainPath: string, options: JsonlStorageOptions) { this.fs = fs; this.directory = directory; this.mainPath = mainPath; - this.memory = memory; this.fsync = options.fsync ?? false; } @@ -264,15 +265,16 @@ export class JsonlStorage implements Storage { if (!created.ok) throw errorFromFile("directory creation", created.error); const mainPathResult = await fs.joinPath([absolute.value, MAIN_FILE], context); if (!mainPathResult.ok) throw errorFromFile("path join", mainPathResult.error); - const memory = new MemoryStorage(); - await JsonlStorage.recover(fs, absolute.value, mainPathResult.value, memory, context); - return new JsonlStorage(fs, absolute.value, mainPathResult.value, memory, options); + const storage = new JsonlStorage(fs, absolute.value, mainPathResult.value, options); + await storage.recover(context); + return storage; } async commit(writes: readonly StorageWrite[], context: Context): Promise { this.assertUsable(); const prepared = this.memory.prepareCommit(writes); const encoded = this.encodeCommit(prepared.seq, prepared.writes); + const reclamations = this.planReclamations(prepared.writes, encoded); const sidecars = await Promise.all( [...encoded.sidecars].map(async ([file, content]) => ({ file, @@ -293,7 +295,10 @@ export class JsonlStorage implements Storage { } const marker = await this.fs.appendFile(this.mainPath, encoded.marker, context); if (!marker.ok) throw this.poison(errorFromFile(`append to ${MAIN_FILE}`, marker.error)); - return prepared.apply(); + const seq = prepared.apply(); + this.adoptSidecarState(prepared.writes); + await this.reclaimSidecars(reclamations, context); + return seq; } async mintId() { @@ -421,14 +426,95 @@ export class JsonlStorage implements Storage { return { marker: jsonLine(marker), sidecars }; } - private static async recover( - fs: FileSystem, - directory: string, - mainPath: string, - memory: MemoryStorage, - context: Context, - ): Promise { - const main = await JsonlStorage.readLines(fs, mainPath, MAIN_FILE, context, (text, line) => + private planReclamations(writes: readonly StorageWrite[], encoded: EncodedCommit): ReadonlyMap { + const createdCurrentOnlyDocuments = new Set(); + const retiredDocuments = new Set(); + const baseDocuments = new Set(); + const finalTasks = new Map(); + for (const write of writes) { + switch (write.type) { + case "document.create": + if (isCurrentOnly(write.record)) createdCurrentOnlyDocuments.add(write.record.id); + break; + case "document.change": + if (write.content.kind === "base") baseDocuments.add(write.id); + break; + case "document.retire": + retiredDocuments.add(write.id); + break; + case "task": + finalTasks.set(write.value.id, write.value); + break; + } + } + + const replacements = new Map(); + const isCurrentOnlyDocument = (id: Id): boolean => + this.currentOnlyDocuments.has(id) || createdCurrentOnlyDocuments.has(id); + for (const id of retiredDocuments) { + if (isCurrentOnlyDocument(id)) replacements.set(sidecarFileName("doc", id), ""); + } + for (const id of baseDocuments) { + if (!isCurrentOnlyDocument(id) || retiredDocuments.has(id)) continue; + const file = sidecarFileName("doc", id); + const content = encoded.sidecars.get(file); + if (content !== undefined) replacements.set(file, content); + } + for (const [id, task] of finalTasks) { + if ( + task.state.status === "terminal" && + (this.liveTaskSidecars.has(id) || encoded.sidecars.has(sidecarFileName("task", id))) + ) { + replacements.set(sidecarFileName("task", id), ""); + } + } + return replacements; + } + + private adoptSidecarState(writes: readonly StorageWrite[]): void { + for (const write of writes) { + if (write.type === "document.create") { + if (isCurrentOnly(write.record)) this.currentOnlyDocuments.add(write.record.id); + } else if (write.type === "task") { + if (write.value.state.status === "terminal") this.liveTaskSidecars.delete(write.value.id); + else this.liveTaskSidecars.add(write.value.id); + } + } + } + + /** The marker already published this state, so reclamation is retryable best-effort maintenance. */ + private async reclaimSidecars(replacements: ReadonlyMap, context: Context): Promise { + if (replacements.size === 0) return; + if (this.fsync) { + const flushed = await this.fs.flushFile(this.mainPath, context); + if (!flushed.ok) return; + } + for (const [file, content] of replacements) await this.replaceSidecar(file, content, context); + } + + private async replaceSidecar(file: string, content: string, context: Context): Promise { + const path = await this.fs.joinPath([this.directory, file], context); + if (!path.ok) return; + if (content === "") { + await this.fs.remove(path.value, { force: true }, context); + return; + } + const temporaryPath = await this.fs.joinPath([this.directory, `${file}${RECLAIM_SUFFIX}`], context); + if (!temporaryPath.ok) return; + const written = await this.fs.writeFile(temporaryPath.value, content, context); + if (!written.ok) return; + if (this.fsync) { + const flushed = await this.fs.flushFile(temporaryPath.value, context); + if (!flushed.ok) return; + } + await this.fs.renameFile(temporaryPath.value, path.value, context); + } + + private async recover(context: Context): Promise { + const fs = this.fs; + const directory = this.directory; + const memory = this.memory; + const main = await JsonlStorage.readLines(fs, this.mainPath, MAIN_FILE, context, (text, line) => parseMainMarker(text, line), ); let previousSeq = 0; @@ -441,6 +527,11 @@ export class JsonlStorage implements Storage { const listed = await fs.listDir(directory, context); if (!listed.ok) throw errorFromFile("directory listing", listed.error); + for (const info of listed.value) { + if (info.kind === "file" && isReclaimFileName(info.name)) { + await fs.remove(info.path, { force: true }, context); + } + } const sidecarFiles = listed.value .filter((info) => info.kind === "file" && isSidecarFileName(info.name)) .map((info) => info.name) @@ -468,6 +559,56 @@ export class JsonlStorage implements Storage { } } + const currentOnlyDocuments = new Set(); + const retiredDocuments = new Set(); + const finalTaskIsLive = new Map(); + for (const { value: marker } of main.lines) { + for (const operation of marker.writes) { + if (operation.type === "document.create") { + if (isCurrentOnly(operation.record)) currentOnlyDocuments.add(operation.record.id); + } else if (operation.type === "document.retire") { + retiredDocuments.add(operation.id); + } else if (operation.type === "task") { + finalTaskIsLive.set(operation.value.id, false); + } else if (operation.type === "task.sidecar") { + finalTaskIsLive.set(operation.id, true); + } + } + } + const retiredCurrentOnlyDocuments = new Set([...retiredDocuments].filter((id) => currentOnlyDocuments.has(id))); + + const latestBases = new Map(); + for (const { value: marker } of main.lines) { + for (const operation of marker.writes) { + if (operation.type !== "document.create" && operation.type !== "document.change") continue; + const id = operation.type === "document.create" ? operation.record.id : operation.id; + if (!currentOnlyDocuments.has(id)) continue; + const record = recordByKey.get( + sidecarKey(sidecarFileName("doc", id), marker.seq, operation.ordinal), + )?.value; + if ( + record?.payload.type !== "document" || + record.payload.id !== id || + record.payload.content.kind !== "base" + ) { + continue; + } + const previous = latestBases.get(id); + if ( + previous === undefined || + record.seq > previous.seq || + (record.seq === previous.seq && record.ordinal > previous.ordinal) + ) { + latestBases.set(id, record); + } + } + } + + const isBeforeLatestBase = (id: Id, seq: Seq, ordinal: number): boolean => { + const base = latestBases.get(id); + return base !== undefined && (seq < base.seq || (seq === base.seq && ordinal < base.ordinal)); + }; + const terminalTasks = new Set([...finalTaskIsLive].filter(([, live]) => !live).map(([id]) => id)); const confirmed = new Set(); for (const line of main.lines) { const marker = line.value; @@ -482,52 +623,61 @@ export class JsonlStorage implements Storage { writes.push(operation); break; case "task.sidecar": { + const optional = terminalTasks.has(operation.id); const record = JsonlStorage.confirmRecord( marker, operation.ordinal, sidecarFileName("task", operation.id), recordByKey, confirmed, + optional, ); - if (record.payload.type !== "task" || record.payload.value.id !== operation.id) { - throw new JsonlCorruptionError(`Confirmed task sidecar data does not match commit ${marker.seq}`); + if (record !== undefined) { + if (record.payload.type !== "task" || record.payload.value.id !== operation.id) { + throw new JsonlCorruptionError( + `Confirmed task sidecar data does not match commit ${marker.seq}`, + ); + } + if (!optional) writes.push({ type: "task", value: record.payload.value }); } - writes.push({ type: "task", value: record.payload.value }); - break; - } - case "document.create": { - const record = JsonlStorage.confirmRecord( - marker, - operation.ordinal, - sidecarFileName("doc", operation.record.id), - recordByKey, - confirmed, - ); - if (record.payload.type !== "document" || record.payload.id !== operation.record.id) { - throw new JsonlCorruptionError( - `Confirmed document sidecar data does not match commit ${marker.seq}`, - ); - } - if (record.payload.content.kind !== "base") { - throw new JsonlCorruptionError(`Document creation lacks a confirmed base in commit ${marker.seq}`); - } - writes.push({ type: "document.create", record: operation.record, content: record.payload.content }); break; } + case "document.create": case "document.change": { + const id = operation.type === "document.create" ? operation.record.id : operation.id; + const reclaimed = + retiredCurrentOnlyDocuments.has(id) || isBeforeLatestBase(id, marker.seq, operation.ordinal); const record = JsonlStorage.confirmRecord( marker, operation.ordinal, - sidecarFileName("doc", operation.id), + sidecarFileName("doc", id), recordByKey, confirmed, + reclaimed, ); - if (record.payload.type !== "document" || record.payload.id !== operation.id) { - throw new JsonlCorruptionError( - `Confirmed document sidecar data does not match commit ${marker.seq}`, - ); + let content: DocumentContent | undefined; + if (record !== undefined) { + if (record.payload.type !== "document" || record.payload.id !== id) { + throw new JsonlCorruptionError( + `Confirmed document sidecar data does not match commit ${marker.seq}`, + ); + } + content = record.payload.content; + } + if (operation.type === "document.create") { + if (content !== undefined && content.kind !== "base") { + throw new JsonlCorruptionError( + `Document creation lacks a confirmed base in commit ${marker.seq}`, + ); + } + writes.push({ + type: "document.create", + record: operation.record, + content: reclaimed || content === undefined ? { kind: "base", version: 1, value: {} } : content, + }); + } else if (!reclaimed && content !== undefined) { + writes.push({ type: "document.change", id, content }); } - writes.push({ type: "document.change", id: operation.id, content: record.payload.content }); break; } } @@ -542,6 +692,7 @@ export class JsonlStorage implements Storage { } } + const reclamations = new Map(); for (const [file, parsed] of parsedFiles) { let unconfirmedAt: number | undefined; for (const line of parsed.lines) { @@ -558,6 +709,33 @@ export class JsonlStorage implements Storage { const truncated = await fs.truncateFile(parsed.path, unconfirmedAt, context); if (!truncated.ok) throw errorFromFile(`tail truncation of ${file}`, truncated.error); } + + const id = Number(file.slice(file.indexOf("-") + 1, -".jsonl".length)); + const confirmedLines = parsed.lines.filter((line) => + confirmed.has(sidecarKey(file, line.value.seq, line.value.ordinal)), + ); + let retainedLines: readonly ParsedLine[] | undefined; + if (file.startsWith("task-") && terminalTasks.has(id)) { + retainedLines = []; + } else if (file.startsWith("doc-") && retiredCurrentOnlyDocuments.has(id)) { + retainedLines = []; + } else if (file.startsWith("doc-") && latestBases.has(id)) { + retainedLines = confirmedLines.filter( + (line) => !isBeforeLatestBase(id, line.value.seq, line.value.ordinal), + ); + } + if ( + retainedLines !== undefined && + (retainedLines.length < confirmedLines.length || retainedLines.length === 0) + ) { + reclamations.set(file, retainedLines.map((line) => jsonLine(line.value)).join("")); + } + } + await this.reclaimSidecars(reclamations, context); + + for (const id of currentOnlyDocuments) this.currentOnlyDocuments.add(id); + for (const [id, live] of finalTaskIsLive) { + if (live) this.liveTaskSidecars.add(id); } } @@ -567,11 +745,13 @@ export class JsonlStorage implements Storage { file: string, recordByKey: ReadonlyMap>, confirmed: Set, - ): SidecarRecord { + optional: boolean, + ): SidecarRecord | undefined { const key = sidecarKey(file, marker.seq, ordinal); if (confirmed.has(key)) throw new JsonlCorruptionError(`Sidecar record is confirmed more than once`); const line = recordByKey.get(key); if (line === undefined) { + if (optional) return undefined; throw new JsonlCorruptionError(`Missing confirmed sidecar record ${file} at sequence ${marker.seq}`); } confirmed.add(key); diff --git a/packages/durable/test/jsonl-storage.test.ts b/packages/durable/test/jsonl-storage.test.ts index 484c8e572..0de533b4d 100644 --- a/packages/durable/test/jsonl-storage.test.ts +++ b/packages/durable/test/jsonl-storage.test.ts @@ -1,4 +1,4 @@ -import { mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises"; +import { mkdtemp, readdir, readFile, rm, stat, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { basename, join } from "node:path"; import type { Context, JsonValue } from "@earendil-works/chord"; @@ -139,7 +139,7 @@ function sessionDocument(id: number, kind = "test.document"): DocumentCreate { } type Failure = { - readonly operation: "append" | "flush"; + readonly operation: "append" | "flush" | "write" | "rename" | "remove"; readonly call: number; readonly mode: "before" | "after" | "short"; }; @@ -149,18 +149,26 @@ class InstrumentedEnv extends NodeExecutionEnv { private failure: Failure | undefined; private appendCalls = 0; private flushCalls = 0; + private writeCalls = 0; + private renameCalls = 0; + private removeCalls = 0; fail(failure: Failure): void { this.failure = failure; - this.appendCalls = 0; - this.flushCalls = 0; - this.operations.length = 0; + this.resetObservations(); } clear(): void { this.failure = undefined; + this.resetObservations(); + } + + private resetObservations(): void { this.appendCalls = 0; this.flushCalls = 0; + this.writeCalls = 0; + this.renameCalls = 0; + this.removeCalls = 0; this.operations.length = 0; } @@ -201,6 +209,66 @@ class InstrumentedEnv extends NodeExecutionEnv { } return err(new FileError("unknown", "injected flush failure", path)); } + + override async writeFile( + path: string, + content: string | Uint8Array, + writeContext: Context, + ): Promise> { + this.writeCalls++; + this.operations.push(`write:${basename(path)}`); + const failure = this.failure; + if (failure?.operation !== "write" || failure.call !== this.writeCalls) { + return super.writeFile(path, content, writeContext); + } + if (failure.mode === "before") return err(new FileError("unknown", "injected write failure", path)); + if (failure.mode === "short") { + const bytes = typeof content === "string" ? new TextEncoder().encode(content) : content; + const partial = bytes.subarray(0, Math.max(1, Math.floor(bytes.length / 2))); + const written = await super.writeFile(path, partial, writeContext); + if (!written.ok) return written; + return err(new FileError("unknown", "injected short write", path)); + } + const written = await super.writeFile(path, content, writeContext); + if (!written.ok) return written; + return err(new FileError("unknown", "injected post-write failure", path)); + } + + override async renameFile( + sourcePath: string, + destinationPath: string, + renameContext: Context, + ): Promise> { + this.renameCalls++; + this.operations.push(`rename:${basename(sourcePath)}->${basename(destinationPath)}`); + const failure = this.failure; + if (failure?.operation !== "rename" || failure.call !== this.renameCalls) { + return super.renameFile(sourcePath, destinationPath, renameContext); + } + if (failure.mode === "after") { + const renamed = await super.renameFile(sourcePath, destinationPath, renameContext); + if (!renamed.ok) return renamed; + } + return err(new FileError("unknown", "injected rename failure", sourcePath)); + } + + override async remove( + path: string, + options: { recursive?: boolean; force?: boolean } | undefined, + removeContext: Context, + ): Promise> { + this.removeCalls++; + this.operations.push(`remove:${basename(path)}`); + const failure = this.failure; + if (failure?.operation !== "remove" || failure.call !== this.removeCalls) { + return super.remove(path, options, removeContext); + } + if (failure.mode === "after") { + const removed = await super.remove(path, options, removeContext); + if (!removed.ok) return removed; + } + return err(new FileError("unknown", "injected remove failure", path)); + } } async function readLines(path: string): Promise { @@ -208,6 +276,16 @@ async function readLines(path: string): Promise { return text === "" ? [] : text.trimEnd().split("\n"); } +async function fileExists(path: string): Promise { + try { + await stat(path); + return true; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return false; + throw error; + } +} + describe("Pico JsonlStorage publication and recovery", () => { it("opens through the Node adapter", async () => { const storage = await openNodeJsonlStorage(await tempDirectory(), context); @@ -216,7 +294,7 @@ describe("Pico JsonlStorage publication and recovery", () => { expect(await storage.conversation(ROOT_CONVERSATION_ID, context)).toEqual({ id: ROOT_CONVERSATION_ID }); }); - it("persists live tasks and document revisions in sidecars with one marker for every commit", async () => { + it("publishes every commit before reclaiming current-only document and terminal-task sidecars", async () => { const directory = await tempDirectory(); const storage = await openStorage(directory); await createRoot(storage); @@ -247,12 +325,15 @@ describe("Pico JsonlStorage publication and recovery", () => { ], context, ); + expect(await readLines(join(directory, `task-${taskId}.jsonl`))).toHaveLength(1); + expect(await readLines(join(directory, `doc-${documentId}.jsonl`))).toHaveLength(1); + await storage.commit([{ type: "document.retire", id: documentId }], context); await storage.commit([{ type: "task", value: terminalTask(taskId) }], context); expect(await readLines(join(directory, "main.jsonl"))).toHaveLength(7); - expect(await readLines(join(directory, `task-${taskId}.jsonl`))).toHaveLength(1); - expect(await readLines(join(directory, `doc-${documentId}.jsonl`))).toHaveLength(3); + expect(await fileExists(join(directory, `task-${taskId}.jsonl`))).toBe(false); + expect(await fileExists(join(directory, `doc-${documentId}.jsonl`))).toBe(false); const markerTypes = (await readLines(join(directory, "main.jsonl"))).map( (line) => (JSON.parse(line) as { readonly type: string }).type, ); @@ -420,7 +501,7 @@ describe("Pico JsonlStorage publication and recovery", () => { }); } - it("orders appends and optional flushes exactly and never flushes main.jsonl", async () => { + it("orders publication flushes exactly and flushes main only to authorize reclamation", async () => { for (const fsync of [false, true]) { const directory = await tempDirectory(); const env = new InstrumentedEnv({ cwd: directory }); @@ -452,6 +533,27 @@ describe("Pico JsonlStorage publication and recovery", () => { ]; expect(env.operations).toEqual(expected); + env.clear(); + await storage.commit( + [ + { + type: "document.change", + id: firstId, + content: { kind: "base", version: 1, value: { checkpoint: true } }, + }, + ], + context, + ); + expect(env.operations).toEqual([ + `append:doc-${firstId}.jsonl`, + ...(fsync ? [`flush:doc-${firstId}.jsonl`] : []), + "append:main.jsonl", + ...(fsync ? ["flush:main.jsonl"] : []), + `write:doc-${firstId}.jsonl.reclaim`, + ...(fsync ? [`flush:doc-${firstId}.jsonl.reclaim`] : []), + `rename:doc-${firstId}.jsonl.reclaim->doc-${firstId}.jsonl`, + ]); + env.clear(); await storage.commit( [ @@ -467,9 +569,383 @@ describe("Pico JsonlStorage publication and recovery", () => { context, ); expect(env.operations).toEqual(["append:main.jsonl"]); + + const taskId = await storage.mintId(); + await storage.commit([{ type: "task", value: pendingTask(taskId) }], context); + env.clear(); + await storage.commit([{ type: "task", value: terminalTask(taskId) }], context); + expect(env.operations).toEqual([ + "append:main.jsonl", + ...(fsync ? ["flush:main.jsonl"] : []), + `remove:task-${taskId}.jsonl`, + ]); } }); + for (const failure of [ + { operation: "write", call: 1, mode: "before" }, + { operation: "write", call: 1, mode: "short" }, + { operation: "write", call: 1, mode: "after" }, + { operation: "rename", call: 1, mode: "before" }, + { operation: "rename", call: 1, mode: "after" }, + ] as const satisfies readonly Failure[]) { + it(`recovers a committed base across reclaim ${failure.operation} ${failure.mode}`, async () => { + const directory = await tempDirectory(); + const env = new InstrumentedEnv({ cwd: directory }); + const storage = await openStorage(directory, env); + await createRoot(storage); + const id = await storage.mintId(); + await storage.commit( + [ + { + type: "document.create", + record: sessionDocument(id), + content: { kind: "base", version: 1, value: { count: 0 } }, + }, + ], + context, + ); + await storage.commit( + [ + { + type: "document.change", + id, + content: { kind: "delta", version: 1, ops: [["s", ["count"], 1]] }, + }, + ], + context, + ); + + env.fail(failure); + await expect( + storage.commit( + [ + { + type: "document.change", + id, + content: { kind: "base", version: 1, value: { count: 2 } }, + }, + ], + context, + ), + ).resolves.toBe(4); + expect((await storage.document(id, "current", context))?.value).toEqual({ count: 2 }); + await storage.close(context); + + const reopened = await openStorage(directory); + expect((await reopened.document(id, "current", context))?.value).toEqual({ count: 2 }); + expect(await readLines(join(directory, `doc-${id}.jsonl`))).toHaveLength(1); + expect((await readdir(directory)).filter((name) => name.endsWith(".reclaim"))).toEqual([]); + }); + } + + for (const mode of ["before", "after"] as const) { + it(`recovers document-retirement reclamation across remove ${mode}`, async () => { + const directory = await tempDirectory(); + const env = new InstrumentedEnv({ cwd: directory }); + const storage = await openStorage(directory, env); + await createRoot(storage); + const taskId = await storage.mintId(); + const id = await storage.mintId(); + await storage.commit( + [ + { type: "task", value: pendingTask(taskId) }, + { + type: "document.create", + record: { id, kind: "task.document", scope: { kind: "task", taskId } }, + content: { kind: "base", version: 1, value: { count: 1 } }, + }, + ], + context, + ); + env.fail({ operation: "remove", call: 1, mode }); + await expect(storage.commit([{ type: "document.retire", id }], context)).resolves.toBe(3); + expect(await storage.document(id, "current", context)).toBeUndefined(); + await storage.close(context); + + const reopened = await openStorage(directory); + expect(await reopened.document(id, "current", context)).toBeUndefined(); + expect(await reopened.task(taskId, context)).toEqual(pendingTask(taskId)); + expect(await fileExists(join(directory, `doc-${id}.jsonl`))).toBe(false); + expect((await readdir(directory)).filter((name) => name.endsWith(".reclaim"))).toEqual([]); + }); + } + + for (const mode of ["before", "after"] as const) { + it(`defers reclamation after authorizing-main flush ${mode} failure`, async () => { + const directory = await tempDirectory(); + const env = new InstrumentedEnv({ cwd: directory }); + const storage = await openStorage(directory, env, { fsync: true }); + await createRoot(storage); + const id = await storage.mintId(); + await storage.commit( + [ + { + type: "document.create", + record: sessionDocument(id), + content: { kind: "base", version: 1, value: { count: 0 } }, + }, + ], + context, + ); + env.fail({ operation: "flush", call: 2, mode }); + await expect( + storage.commit( + [ + { + type: "document.change", + id, + content: { kind: "base", version: 1, value: { count: 2 } }, + }, + ], + context, + ), + ).resolves.toBe(3); + expect(env.operations).toEqual([ + `append:doc-${id}.jsonl`, + `flush:doc-${id}.jsonl`, + "append:main.jsonl", + "flush:main.jsonl", + ]); + expect((await storage.document(id, "current", context))?.value).toEqual({ count: 2 }); + expect(await readLines(join(directory, `doc-${id}.jsonl`))).toHaveLength(2); + await storage.close(context); + + const recoveryEnv = new InstrumentedEnv({ cwd: directory }); + recoveryEnv.fail({ operation: "flush", call: 1, mode }); + const deferred = await openStorage(directory, recoveryEnv, { fsync: true }); + expect((await deferred.document(id, "current", context))?.value).toEqual({ count: 2 }); + expect(recoveryEnv.operations).toEqual(["flush:main.jsonl"]); + expect(await readLines(join(directory, `doc-${id}.jsonl`))).toHaveLength(2); + await deferred.close(context); + + const reclaimed = await openStorage(directory, new NodeExecutionEnv({ cwd: directory }), { fsync: true }); + expect((await reclaimed.document(id, "current", context))?.value).toEqual({ count: 2 }); + expect(await readLines(join(directory, `doc-${id}.jsonl`))).toHaveLength(1); + }); + } + + for (const mode of ["before", "after"] as const) { + it(`keeps a committed base usable after reclaim-temp flush ${mode} failure`, async () => { + const directory = await tempDirectory(); + const env = new InstrumentedEnv({ cwd: directory }); + const storage = await openStorage(directory, env, { fsync: true }); + await createRoot(storage); + const id = await storage.mintId(); + await storage.commit( + [ + { + type: "document.create", + record: sessionDocument(id), + content: { kind: "base", version: 1, value: { count: 0 } }, + }, + ], + context, + ); + env.fail({ operation: "flush", call: 3, mode }); + await expect( + storage.commit( + [ + { + type: "document.change", + id, + content: { kind: "base", version: 1, value: { count: 2 } }, + }, + ], + context, + ), + ).resolves.toBe(3); + env.clear(); + await storage.commit( + [ + { + type: "document.change", + id, + content: { kind: "delta", version: 1, ops: [["s", ["count"], 3]] }, + }, + ], + context, + ); + await storage.close(context); + + const reopened = await openStorage(directory, new NodeExecutionEnv({ cwd: directory }), { fsync: true }); + expect((await reopened.document(id, "current", context))?.value).toEqual({ count: 3 }); + expect(await readLines(join(directory, `doc-${id}.jsonl`))).toHaveLength(2); + }); + } + + for (const mode of ["before", "after"] as const) { + it(`recovers terminal-task reclamation across remove ${mode}`, async () => { + const directory = await tempDirectory(); + const env = new InstrumentedEnv({ cwd: directory }); + const storage = await openStorage(directory, env); + await createRoot(storage); + const id = await storage.mintId(); + await storage.commit([{ type: "task", value: pendingTask(id) }], context); + env.fail({ operation: "remove", call: 1, mode }); + await expect(storage.commit([{ type: "task", value: terminalTask(id) }], context)).resolves.toBe(3); + expect(await storage.task(id, context)).toEqual(terminalTask(id)); + await storage.close(context); + + const reopened = await openStorage(directory); + expect(await reopened.task(id, context)).toEqual(terminalTask(id)); + expect(await fileExists(join(directory, `task-${id}.jsonl`))).toBe(false); + expect((await readdir(directory)).filter((name) => name.endsWith(".reclaim"))).toEqual([]); + }); + } + + it("appends later deltas to the replacement sidecar after a current-only base", async () => { + const directory = await tempDirectory(); + const storage = await openStorage(directory); + await createRoot(storage); + const id = await storage.mintId(); + await storage.commit( + [ + { + type: "document.create", + record: sessionDocument(id), + content: { kind: "base", version: 1, value: { count: 0 } }, + }, + ], + context, + ); + await storage.commit( + [ + { + type: "document.change", + id, + content: { kind: "base", version: 1, value: { count: 10 } }, + }, + ], + context, + ); + await storage.commit( + [ + { + type: "document.change", + id, + content: { kind: "delta", version: 1, ops: [["s", ["count"], 11]] }, + }, + ], + context, + ); + expect(await readLines(join(directory, `doc-${id}.jsonl`))).toHaveLength(2); + const reopened = await openStorage(directory); + expect((await reopened.document(id, "current", context))?.value).toEqual({ count: 11 }); + }); + + it("never reclaims rewindable document history, including after a base and retirement", async () => { + const directory = await tempDirectory(); + const storage = await openStorage(directory); + await createRoot(storage); + const id = await storage.mintId(); + const record = { + id, + kind: "rewindable", + scope: { kind: "conversation" as const, conversationId: ROOT_CONVERSATION_ID }, + history: "rewindable" as const, + fork: "asOf" as const, + }; + const createdAt = await storage.commit( + [{ type: "document.create", record, content: { kind: "base", version: 1, value: { count: 0 } } }], + context, + ); + const changedAt = await storage.commit( + [ + { + type: "document.change", + id, + content: { kind: "delta", version: 1, ops: [["s", ["count"], 1]] }, + }, + ], + context, + ); + await storage.commit( + [ + { + type: "document.change", + id, + content: { kind: "base", version: 1, value: { count: 2 } }, + }, + ], + context, + ); + await storage.commit([{ type: "document.retire", id }], context); + + expect(await readLines(join(directory, `doc-${id}.jsonl`))).toHaveLength(3); + const reopened = await openStorage(directory); + expect((await reopened.document(id, createdAt, context))?.value).toEqual({ count: 0 }); + expect((await reopened.document(id, changedAt, context))?.value).toEqual({ count: 1 }); + expect(await reopened.document(id, "current", context)).toBeUndefined(); + expect(await readLines(join(directory, `doc-${id}.jsonl`))).toHaveLength(3); + }); + + it("reclaims retired task-, session-, and latest-conversation document sidecars", async () => { + const directory = await tempDirectory(); + const storage = await openStorage(directory); + await createRoot(storage); + const taskId = await storage.mintId(); + const sessionId = await storage.mintId(); + const latestId = await storage.mintId(); + const taskDocumentId = await storage.mintId(); + const createdAt = await storage.commit( + [ + { type: "task", value: pendingTask(taskId) }, + { + type: "document.create", + record: sessionDocument(sessionId, "session"), + content: { kind: "base", version: 1, value: {} }, + }, + { + type: "document.create", + record: { + id: latestId, + kind: "latest", + scope: { kind: "conversation", conversationId: ROOT_CONVERSATION_ID }, + history: "latest", + fork: "current", + }, + content: { kind: "base", version: 1, value: {} }, + }, + { + type: "document.create", + record: { id: taskDocumentId, kind: "task", scope: { kind: "task", taskId } }, + content: { kind: "base", version: 1, value: {} }, + }, + ], + context, + ); + const retiredAt = await storage.commit( + [ + { type: "document.retire", id: sessionId }, + { type: "document.retire", id: latestId }, + { type: "document.retire", id: taskDocumentId }, + { type: "task", value: terminalTask(taskId) }, + ], + context, + ); + for (const file of [ + `doc-${sessionId}.jsonl`, + `doc-${latestId}.jsonl`, + `doc-${taskDocumentId}.jsonl`, + `task-${taskId}.jsonl`, + ]) { + expect(await fileExists(join(directory, file))).toBe(false); + } + + const reopened = await openStorage(directory); + expect(await reopened.task(taskId, context)).toEqual(terminalTask(taskId)); + expect(await reopened.document(sessionId, "current", context)).toBeUndefined(); + expect(await reopened.document(latestId, "current", context)).toBeUndefined(); + expect(await reopened.document(taskDocumentId, "current", context)).toBeUndefined(); + expect( + await reopened.findDocument({ kind: "session", scope: { kind: "session" } }, createdAt, context), + ).toMatchObject({ id: sessionId, createdAt, retiredAt }); + expect( + await reopened.findDocument({ kind: "session", scope: { kind: "session" } }, retiredAt, context), + ).toBeUndefined(); + }); + it("truncates torn UTF-8 tails at exact byte offsets and reuses the unconfirmed sequence", async () => { const directory = await tempDirectory(); const storage = await openStorage(directory); @@ -512,7 +988,7 @@ describe("Pico JsonlStorage publication and recovery", () => { await storage.commit([{ type: "task", value: pendingTask(taskId) }], context); await storage.commit([{ type: "task", value: terminalTask(taskId) }], context); const sidecarPath = join(directory, `task-${taskId}.jsonl`); - const confirmedSize = (await stat(sidecarPath)).size; + expect(await fileExists(sidecarPath)).toBe(false); await writeFile( sidecarPath, `${JSON.stringify({ @@ -526,7 +1002,7 @@ describe("Pico JsonlStorage publication and recovery", () => { ); const reopened = await openStorage(directory); - expect((await stat(sidecarPath)).size).toBe(confirmedSize); + expect(await fileExists(sidecarPath)).toBe(false); expect(await reopened.task(taskId, context)).toEqual(terminalTask(taskId)); expect(await reopened.commit([], context)).toBe(4); });