From 8cb2017a3a0f101bd8a055da90f678756ceb7563 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Mon, 26 Oct 2026 10:09:27 +0900 Subject: [PATCH 1/3] fix(coding-agent): build forks atomically and keep a header write's prefix (#4295) forkFrom writes through replaceFileAtomically so a full disk leaves no partial fork; a failed catch-up append in _writeHeaderAsync keeps the header and whole entries, cuts the rest, and writes the remaining entries before the next append. --- changes/unreleased/4305.md | 23 +++ .../coding-agent/src/core/session-manager.ts | 61 ++++++-- .../test/session-manager-header-write.test.ts | 38 +++++ .../session-manager/fork-full-disk.test.ts | 131 ++++++++++++++++++ 4 files changed, 240 insertions(+), 13 deletions(-) create mode 100644 changes/unreleased/4305.md create mode 100644 packages/coding-agent/test/session-manager/fork-full-disk.test.ts diff --git a/changes/unreleased/4305.md b/changes/unreleased/4305.md new file mode 100644 index 0000000000..7f474519ce --- /dev/null +++ b/changes/unreleased/4305.md @@ -0,0 +1,23 @@ + +## 2026-10-26 - Fork files are built atomically and a header catch-up keeps its prefix (senpi#4295) + +### What changed + +- `packages/coding-agent/src/core/session-manager.ts`: `forkFrom` builds the fork through `replaceFileAtomically` (temp file, fsync, rename) instead of an exclusive header write followed by one `appendFileSync` per entry; an existing target still fails with `EEXIST`. When the post-close catch-up append of `_writeHeaderAsync` fails, the file is kept and cut back to the bytes known whole (`_keepCaughtUpPrefix`), the transcript counts as flushed, and the entries that append carried are written, one line each, before the next append. + +### Why + +- A full disk part-way through `forkFrom` left a half-copied fork in the session list. A failed catch-up deleted the whole new file, including the header and entries already on disk. + +### Why an extension could not handle it + +- Both writes run inside `SessionManager`. + +### Expected merge conflict zones + +- MEDIUM: `packages/coding-agent/src/core/session-manager.ts`, `_writeHeaderAsync` and the flushed append in `_persist`. +- LOW: `forkFrom` file creation. + + +### Fixed +- A fork interrupted by a full disk leaves no partial session file behind, and a session whose first save was interrupted after its header was written keeps what was written and saves the rest next ([#4295](https://github.com/code-yeongyu/senpi/issues/4295)) diff --git a/packages/coding-agent/src/core/session-manager.ts b/packages/coding-agent/src/core/session-manager.ts index e377b8238b..9c49fa35e7 100644 --- a/packages/coding-agent/src/core/session-manager.ts +++ b/packages/coding-agent/src/core/session-manager.ts @@ -24,7 +24,7 @@ import { statSync, writeFileSync, } from "fs"; -import { appendFile, type FileHandle, open, readdir, rm } from "fs/promises"; +import { appendFile, type FileHandle, open, readdir, rm, truncate } from "fs/promises"; import { basename, join, resolve } from "path"; import { StringDecoder } from "string_decoder"; import { APP_NAME, getAgentDir as getDefaultAgentDir, getSessionsDir } from "../config.ts"; @@ -1172,6 +1172,8 @@ export class SessionManager { private ownedHeaderFile: string | undefined; // Set when an append to the flushed file failed and may have left a partial last line. private tailMayBeTorn = false; + /** The file entries a failed header catch-up left off disk (from `from` on); written before the next append. */ + private unwritten: { readonly entries: FileEntry[]; from: number } | undefined; private fileEntries: FileEntry[] = []; private byId: Map = new Map(); // Runtime-only identity tracking lets AgentSession compare messages to a @@ -1676,12 +1678,15 @@ export class SessionManager { const serialize = (batch: readonly FileEntry[]): string => batch.map((e) => `${JSON.stringify(this.residentStore.materialize(e))}\n`).join(""); let written = 0; + let durable = { entries: 0, bytes: 0 }; let handle: FileHandle | undefined = await open(sessionFile, "wx"); try { while (written < entries.length) { const batch = entries.slice(written); written += batch.length; - await handle.writeFile(serialize(batch)); + const text = serialize(batch); + await handle.writeFile(text); + durable = { entries: written, bytes: durable.bytes + Buffer.byteLength(text) }; } const closing = handle; handle = undefined; @@ -1690,12 +1695,18 @@ export class SessionManager { while (written < entries.length && this.sessionFile === sessionFile && this.fileEntries === entries) { const batch = entries.slice(written); written += batch.length; - await appendFile(sessionFile, serialize(batch)); + const text = serialize(batch); + await appendFile(sessionFile, text); + durable = { entries: written, bytes: durable.bytes + Buffer.byteLength(text) }; } } catch (error) { - // This write created the file: a part-written one would fail every later first flush with - // EEXIST while the entries it was carrying stayed in memory only. Nothing counts as flushed yet. - return discardFailedFirstFlushAsync(sessionFile, handle, error); + // While the exclusive handle is open nothing is whole yet: a part-written file would fail every later first + // flush with EEXIST, so it is removed and nothing counts as flushed. + if (handle !== undefined) return discardFailedFirstFlushAsync(sessionFile, handle, error); + // The header and the first entries are on disk and closed: keep them (senpi#4295). Cut what the failed + // append landed, and write the entries it carried before this session's next append. + await this._keepCaughtUpPrefix(sessionFile, entries, durable); + throw error; } // Synchronous with the last check above: no entry can land between it and the flag. if (this.sessionFile === sessionFile && this.fileEntries === entries && written === entries.length) { @@ -1704,6 +1715,23 @@ export class SessionManager { } } + private async _keepCaughtUpPrefix( + sessionFile: string, + entries: FileEntry[], + durable: { readonly entries: number; readonly bytes: number }, + ): Promise { + try { + await truncate(sessionFile, durable.bytes); + } catch (cutError) { + if (!(cutError instanceof Error)) throw cutError; + this.tailMayBeTorn = true; + } + if (this.sessionFile !== sessionFile || this.fileEntries !== entries) return; + this.unwritten = { entries, from: durable.entries }; + this.flushed = true; + this.ownedHeaderFile = sessionFile; + } + /** * A new session file is created only once the session contains a user or assistant message. * Setup entries alone (model, thinking level, system prompt) stay in memory so opening and @@ -1733,6 +1761,13 @@ export class SessionManager { this.tailMayBeTorn = false; } try { + // Entries a failed header catch-up left off disk go first, one line each, so a later failure resumes there. + const unwritten = this.unwritten?.entries === this.fileEntries ? this.unwritten : undefined; + for (const pending of unwritten?.entries.slice(unwritten.from) ?? []) { + appendFileSync(this.sessionFile, `${JSON.stringify(this.residentStore.materialize(pending))}\n`); + if (unwritten) unwritten.from++; + } + this.unwritten = undefined; appendFileSync(this.sessionFile, `${JSON.stringify(persistedEntry)}\n`); } catch (error) { this.tailMayBeTorn = true; @@ -3088,14 +3123,14 @@ export class SessionManager { parentSession: resolvedSourcePath, }; reserveSessionWrite(newSessionFile); - writeFileSync(newSessionFile, `${JSON.stringify(newHeader)}\n`, { flag: "wx" }); - - // Copy all non-header entries from source - for (const entry of sourceEntries) { - if (entry.type !== "session") { - appendFileSync(newSessionFile, `${JSON.stringify(entry)}\n`); - } + if (existsSync(newSessionFile)) { + throw Object.assign(new Error(`EEXIST: file already exists, open '${newSessionFile}'`), { code: "EEXIST" }); } + // Built in a temp file and renamed into place: a full disk part-way leaves no partial fork in the session list. + const forkLines = [newHeader, ...sourceEntries.filter((entry) => entry.type !== "session")].map( + (entry) => `${JSON.stringify(entry)}\n`, + ); + replaceFileAtomically(newSessionFile, forkLines); return new SessionManager(resolvedTargetCwd, dir, newSessionFile, true); } diff --git a/packages/coding-agent/test/session-manager-header-write.test.ts b/packages/coding-agent/test/session-manager-header-write.test.ts index 9491587515..8a88f374d3 100644 --- a/packages/coding-agent/test/session-manager-header-write.test.ts +++ b/packages/coding-agent/test/session-manager-header-write.test.ts @@ -8,12 +8,20 @@ import { SessionManager } from "../src/core/session-manager.ts"; const hooks = vi.hoisted(() => ({ beforeClose: undefined as (() => void) | undefined, writeFails: undefined as Error | undefined, + appendFails: undefined as Error | undefined, })); vi.mock("fs/promises", async (importOriginal) => { const real = await importOriginal(); return { ...real, + appendFile: async (...args: Parameters) => { + const failure = hooks.appendFails; + if (failure === undefined || typeof args[1] !== "string") return real.appendFile(...args); + hooks.appendFails = undefined; + await real.appendFile(args[0], args[1].slice(0, Math.floor(args[1].length / 2))); + throw failure; + }, open: async (...args: Parameters) => { const handle = await real.open(...args); const writeFile = handle.writeFile.bind(handle); @@ -39,6 +47,7 @@ const roots: string[] = []; afterEach(() => { hooks.beforeClose = undefined; hooks.writeFails = undefined; + hooks.appendFails = undefined; for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); }); @@ -140,4 +149,33 @@ describe("persistHeaderNow's asynchronous header write", () => { expect(idsOnDisk(file)).toEqual(manager.getEntries().map((entry) => entry.id)); expect(manager.getEntries()).toHaveLength(4); }); + + it("keeps the header and entries already on disk when only the catch-up append fails, and writes the rest next", async () => { + // Given a header write whose catch-up append (entries persisted while its handle closed) runs out of space + const { manager, file } = freshSession(); + manager.appendCustomEntry("before", { n: 0 }); + hooks.beforeClose = () => { + hooks.beforeClose = undefined; + manager.appendCustomEntry("during-close", { n: 1 }); + hooks.appendFails = Object.assign(new Error("ENOSPC: no space left on device, write"), { code: "ENOSPC" }); + }; + + // When the header write settles and, once there is space, the next entry is appended + await expect(manager.persistHeaderNow()).rejects.toMatchObject({ code: "ENOSPC" }); + const kept = existsSync(file) + ? linesOnDisk(file).map((line) => (line.type === "session" ? "session" : line.customType)) + : []; + manager.appendCustomEntry("after", { n: 2 }); + + // Then the written prefix stayed on disk, and the file then holds every entry memory holds, once, in order + expect(kept).toEqual(["session", "before"]); + expect(manager.isTranscriptFlushed()).toBe(true); + expect(linesOnDisk(file).map((line) => (line.type === "session" ? "session" : line.customType))).toEqual([ + "session", + "before", + "during-close", + "after", + ]); + expect(idsOnDisk(file)).toEqual(manager.getEntries().map((entry) => entry.id)); + }); }); diff --git a/packages/coding-agent/test/session-manager/fork-full-disk.test.ts b/packages/coding-agent/test/session-manager/fork-full-disk.test.ts new file mode 100644 index 0000000000..a7dc8d6f14 --- /dev/null +++ b/packages/coding-agent/test/session-manager/fork-full-disk.test.ts @@ -0,0 +1,131 @@ +import type { AssistantMessage } from "@earendil-works/pi-ai"; +import type * as FsModule from "fs"; +import { mkdtempSync, readdirSync, rmSync } from "fs"; +import { tmpdir } from "os"; +import { join } from "path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { SessionManager } from "../../src/core/session-manager.ts"; + +/** A full disk part-way through creating a fork or a branch must leave no fork file in the session list (senpi#4295). */ + +const fault = vi.hoisted(() => ({ budget: undefined as number | undefined })); + +function noSpace(): Error { + return Object.assign(new Error("ENOSPC: no space left on device, write"), { code: "ENOSPC" }); +} + +vi.mock("fs", async (importOriginal) => { + const actual = await importOriginal(); + /** Lands what fits in the remaining budget, then fails the write the way a full volume does. */ + function limited(write: (target: T[0], data: string, ...rest: unknown[]) => void) { + return (target: T[0], data: unknown, ...rest: unknown[]) => { + if (fault.budget === undefined || typeof data !== "string") return write(target, data as string, ...rest); + if (data.length <= fault.budget) { + fault.budget -= data.length; + return write(target, data, ...rest); + } + const landed = data.slice(0, fault.budget); + fault.budget = undefined; + if (landed.length > 0) write(target, landed, ...rest); + throw noSpace(); + }; + } + return { + ...actual, + appendFileSync: limited((target, data, ...rest) => + actual.appendFileSync(target as FsModule.PathOrFileDescriptor, data, ...(rest as [])), + ), + writeFileSync: limited((target, data, ...rest) => + actual.writeFileSync(target as FsModule.PathOrFileDescriptor, data, ...(rest as [])), + ), + }; +}); + +const assistant = (text: string): AssistantMessage => ({ + role: "assistant", + content: [{ type: "text", text }], + api: "anthropic-messages", + provider: "anthropic", + model: "test", + usage: { + input: 1, + output: 1, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 2, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: 2, +}); + +const roots: string[] = []; + +afterEach(() => { + fault.budget = undefined; + for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); +}); + +function savedSession(): { readonly root: string; readonly session: SessionManager; readonly file: string } { + const root = mkdtempSync(join(tmpdir(), "senpi-fork-full-")); + roots.push(root); + const session = SessionManager.create(root, join(root, "sessions")); + for (let turn = 0; turn < 3; turn++) { + session.appendMessage({ role: "user", content: `question ${turn} ${"x".repeat(200)}`, timestamp: 1 }); + session.appendMessage(assistant(`answer ${turn} ${"y".repeat(200)}`)); + } + const file = session.getSessionFile(); + if (!file) throw new Error("test setup: persisted session has no file"); + return { root, session, file }; +} + +function captureError(run: () => unknown): unknown { + try { + run(); + } catch (error) { + return error; + } + return undefined; +} + +describe("creating a session copy on a full disk", () => { + it("leaves no fork file behind when forkFrom runs out of space part-way", () => { + // Given a saved session and a target session directory + const { root, file } = savedSession(); + const forkDir = join(root, "forks"); + const failures: unknown[] = []; + + // When the disk fills up after each possible amount of the fork has been written + for (const budget of [0, 150, 600, 1200]) { + fault.budget = budget; + failures.push(captureError(() => SessionManager.forkFrom(file, root, forkDir))); + fault.budget = undefined; + } + + // Then every attempt reported the full disk and no partial fork (or temp file) is left in the directory + expect(failures.map((failure) => (failure as { code?: string } | undefined)?.code)).toEqual([ + "ENOSPC", + "ENOSPC", + "ENOSPC", + "ENOSPC", + ]); + expect(readdirSync(forkDir)).toEqual([]); + }); + + it("leaves no branch file behind when createBranchedSession runs out of space part-way", () => { + // Given a saved session + const { root, session } = savedSession(); + const leaf = session.getLeafId(); + if (!leaf) throw new Error("test setup: no leaf"); + const before = readdirSync(join(root, "sessions")).sort(); + + // When the branch copy runs out of space part-way + fault.budget = 300; + const failure = captureError(() => session.createBranchedSession(leaf)); + fault.budget = undefined; + + // Then the failure surfaced and the session directory holds exactly what it held before + expect(failure).toMatchObject({ code: "ENOSPC" }); + expect(readdirSync(join(root, "sessions")).sort()).toEqual(before); + }); +}); From cc1991a8e3c061e843fa4efb85ad734ebeb16fbc Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Mon, 26 Oct 2026 10:51:15 +0900 Subject: [PATCH 2/3] fix(coding-agent): no extra existence probe in forkFrom; drop its retired ledger entries (#4295) --- packages/coding-agent/src/core/session-manager.ts | 3 --- .../test/suite/no-sync-in-session-path.ledger.json | 14 -------------- 2 files changed, 17 deletions(-) diff --git a/packages/coding-agent/src/core/session-manager.ts b/packages/coding-agent/src/core/session-manager.ts index 9c49fa35e7..b9f5b00734 100644 --- a/packages/coding-agent/src/core/session-manager.ts +++ b/packages/coding-agent/src/core/session-manager.ts @@ -3123,9 +3123,6 @@ export class SessionManager { parentSession: resolvedSourcePath, }; reserveSessionWrite(newSessionFile); - if (existsSync(newSessionFile)) { - throw Object.assign(new Error(`EEXIST: file already exists, open '${newSessionFile}'`), { code: "EEXIST" }); - } // Built in a temp file and renamed into place: a full disk part-way leaves no partial fork in the session list. const forkLines = [newHeader, ...sourceEntries.filter((entry) => entry.type !== "session")].map( (entry) => `${JSON.stringify(entry)}\n`, diff --git a/packages/coding-agent/test/suite/no-sync-in-session-path.ledger.json b/packages/coding-agent/test/suite/no-sync-in-session-path.ledger.json index c68d8643b0..8542d06808 100644 --- a/packages/coding-agent/test/suite/no-sync-in-session-path.ledger.json +++ b/packages/coding-agent/test/suite/no-sync-in-session-path.ledger.json @@ -206,20 +206,6 @@ "count": 1, "note": "DELIBERATE transcript path: one JSONL entry appended per persisted entry (KB-scale), or a full flush of the pending entries on the first assistant message (session-scale)" }, - { - "file": "src/core/session-manager.ts", - "api": "appendFileSync", - "symbol": "forkFrom", - "count": 1, - "note": "DELIBERATE transcript path: forked transcript written once (session-scale)" - }, - { - "file": "src/core/session-manager.ts", - "api": "writeFileSync", - "symbol": "forkFrom", - "count": 1, - "note": "DELIBERATE transcript path: forked transcript written once (session-scale)" - }, { "file": "src/core/session-resident-store.ts", "api": "readFileSync", From 83966e2247642d353313b63ad09ec9d59c080f61 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Mon, 26 Oct 2026 20:34:01 +0900 Subject: [PATCH 3/3] fix(coding-agent): unify transcript write tracking (#4295) Co-authored-by: sisyphus-dev-ai --- changes/unreleased/4305.md | 9 +- .../coding-agent/src/core/session-manager.ts | 119 +++++++------- .../src/core/session-write-recovery.ts | 14 +- .../src/core/session-write-state.ts | 19 +++ .../test/session-manager-header-write.test.ts | 67 +++++++- .../torn-append-every-offset.test.ts | 123 ++++++++++++++ .../write-state-interleavings.test.ts | 153 ++++++++++++++++++ 7 files changed, 437 insertions(+), 67 deletions(-) create mode 100644 packages/coding-agent/src/core/session-write-state.ts create mode 100644 packages/coding-agent/test/session-manager/torn-append-every-offset.test.ts create mode 100644 packages/coding-agent/test/session-manager/write-state-interleavings.test.ts diff --git a/changes/unreleased/4305.md b/changes/unreleased/4305.md index 7f474519ce..7f80856388 100644 --- a/changes/unreleased/4305.md +++ b/changes/unreleased/4305.md @@ -3,11 +3,14 @@ ### What changed -- `packages/coding-agent/src/core/session-manager.ts`: `forkFrom` builds the fork through `replaceFileAtomically` (temp file, fsync, rename) instead of an exclusive header write followed by one `appendFileSync` per entry; an existing target still fails with `EEXIST`. When the post-close catch-up append of `_writeHeaderAsync` fails, the file is kept and cut back to the bytes known whole (`_keepCaughtUpPrefix`), the transcript counts as flushed, and the entries that append carried are written, one line each, before the next append. +- `packages/coding-agent/src/core/session-manager.ts`: `forkFrom` builds the fork through `replaceFileAtomically` (temp file, fsync, rename) instead of an exclusive header write followed by one `appendFileSync` per entry. The fresh timestamp and session id make collisions unlikely; the atomic rename replaces an existing target rather than rejecting it with `EEXIST`. When the post-close catch-up append of `_writeHeaderAsync` fails, the file is kept and cut back to the bytes known whole (`_keepCaughtUpPrefix`), and the missing entries are written, one line each, before the next append. +- `session-write-state.ts` makes buffering, an in-flight header writer, and a flushed file mutually exclusive states. Discard, reload, new-session, branch and full-rewrite transitions replace the entire state, so no stale append cursor or torn-tail cause survives a new file. The header writer owns its queue across compaction instead of relying on the resident array's identity. +- `session-write-recovery.ts` accepts the exact durable byte boundary when a header catch-up's cut-back failed. Retrying cuts the whole failed batch, including any complete lines it landed, before replaying it. The unified state retains the original write error for #4300's recovery notice integration. +- The atomic replacement helper removes its temporary file on a caught failure. A process crash before rename can still leave a hidden `.tmp` file; it is not a session and is not swept automatically, since another live writer may own it. ### Why -- A full disk part-way through `forkFrom` left a half-copied fork in the session list. A failed catch-up deleted the whole new file, including the header and entries already on disk. +- A full disk part-way through `forkFrom` left a half-copied fork in the session list. A failed catch-up deleted the whole new file, including the header and entries already on disk. Independently tracked flush flags and pending entries could replay entries after discard/recreation, while compaction could strand a header-owned file behind repeated `EEXIST`. ### Why an extension could not handle it @@ -15,7 +18,7 @@ ### Expected merge conflict zones -- MEDIUM: `packages/coding-agent/src/core/session-manager.ts`, `_writeHeaderAsync` and the flushed append in `_persist`. +- MEDIUM: `packages/coding-agent/src/core/session-manager.ts`, `_writeHeaderAsync` and the flushed append in `_persist`; #4300 must use the unified state's `tailFailure.cause`, not a separate cause field, and preserve the exact `completeBytes` cut-back boundary while setting the failed batch aside before replay. - LOW: `forkFrom` file creation. diff --git a/packages/coding-agent/src/core/session-manager.ts b/packages/coding-agent/src/core/session-manager.ts index b9f5b00734..9fa89b090a 100644 --- a/packages/coding-agent/src/core/session-manager.ts +++ b/packages/coding-agent/src/core/session-manager.ts @@ -69,6 +69,7 @@ import { reserveSessionWrite, unregisterSessionWriter, } from "./session-write-reservation.ts"; +import type { HeaderWriteState, SessionWriteState } from "./session-write-state.ts"; import { isVirtualModel } from "./virtual-models.ts"; export type { SessionListProgress } from "./session-discovery.ts"; @@ -1166,14 +1167,8 @@ export class SessionManager { private persist: boolean; private readonly writeGuard = new SessionWriteGuard(); private readonly writeFence = new SessionWriteFence(); - private flushed: boolean = false; + private writeState: SessionWriteState = { kind: "buffered" }; private headerWrite: Promise | undefined; - // The file persistHeaderNow() created in this process; only it may be discarded on close (senpi#3263). - private ownedHeaderFile: string | undefined; - // Set when an append to the flushed file failed and may have left a partial last line. - private tailMayBeTorn = false; - /** The file entries a failed header catch-up left off disk (from `from` on); written before the next append. */ - private unwritten: { readonly entries: FileEntry[]; from: number } | undefined; private fileEntries: FileEntry[] = []; private byId: Map = new Map(); // Runtime-only identity tracking lets AgentSession compare messages to a @@ -1361,6 +1356,7 @@ export class SessionManager { this.writeFence.opened(resolvedFile); } this.sessionFile = resolvedFile; + this.writeState = { kind: "buffered" }; this.mirrorTrimmed = false; this.trimmedIds.clear(); this.residentStore.clear(); @@ -1377,7 +1373,6 @@ export class SessionManager { // different identity and the caller's record would point at nothing (#2010). if (created?.id !== undefined) { this._rewriteFile(); - this.flushed = true; } return undefined; } @@ -1399,7 +1394,6 @@ export class SessionManager { // An empty file carries no identity yet, so a caller-supplied id still applies. this._resetToNewSession(withFileNameId(sessionFile, newSessionOptions)); this._rewriteFile(); - this.flushed = true; return false; } @@ -1431,13 +1425,11 @@ export class SessionManager { this.historyView = { source: this.fileEntries, length: this.fileEntries.length, entries: loadedHistory }; } this.mutationCount++; - this.flushed = true; - this.ownedHeaderFile = undefined; + this.writeState = { kind: "flushed" }; } newSession(options?: NewSessionOptions): string | undefined { const timestamp = this._resetToNewSession(options); - this.ownedHeaderFile = undefined; if (this.persist) { const fileTimestamp = timestamp.replace(/[:.]/g, "-"); const path = join(this.getSessionDir(), `${fileTimestamp}_${this.sessionId}.jsonl`); @@ -1493,7 +1485,7 @@ export class SessionManager { latestCacheHitRate: undefined, }; this.mutationCount++; - this.flushed = false; + this.writeState = { kind: "buffered" }; return timestamp; } @@ -1586,6 +1578,7 @@ export class SessionManager { this.writeFence.assertOwned(this.sessionFile); reserveSessionWrite(this.sessionFile); replaceFileAtomically(this.sessionFile, this._serializedFileEntries()); + this.writeState = { kind: "flushed" }; } private *_serializedFileEntries(): Generator { @@ -1638,16 +1631,18 @@ export class SessionManager { * persisted while it runs are appended by it before the transcript counts as flushed. */ persistHeaderNow(): Promise { - if (!this.persist || !this.sessionFile || this.flushed) return this.headerWrite ?? Promise.resolve(); - const sessionFile = this.sessionFile; - this.headerWrite ??= this._writeHeaderAsync(sessionFile).finally(() => { + if (!this.persist || !this.sessionFile || this.writeState.kind !== "buffered") + return this.headerWrite ?? Promise.resolve(); + const state: HeaderWriteState = { kind: "writing", file: this.sessionFile, entries: [...this.fileEntries] }; + this.writeState = state; + this.headerWrite = this._writeHeaderAsync(state).finally(() => { this.headerWrite = undefined; }); return this.headerWrite; } isTranscriptFlushed(): boolean { - return this.flushed; + return this.writeState.kind === "flushed"; } /** @@ -1662,25 +1657,26 @@ export class SessionManager { */ async discardHeaderOnlyFile(): Promise { await this.headerWrite; - if (!this.persist || !this.sessionFile || !this.flushed) return false; - if (this.ownedHeaderFile !== this.sessionFile) return false; + if (!this.persist || !this.sessionFile || this.writeState.kind !== "flushed") return false; + if (this.writeState.ownedHeaderFile !== this.sessionFile) return false; if (!this.fileEntries.every(isDisposableEntry)) return false; + const state = this.writeState; await rm(this.sessionFile, { force: true }); - this.ownedHeaderFile = undefined; - this.flushed = false; + if (this.writeState === state) this.writeState = { kind: "buffered" }; return true; } - private async _writeHeaderAsync(sessionFile: string): Promise { + private async _writeHeaderAsync(state: HeaderWriteState): Promise { + const { file: sessionFile, entries } = state; this.writeFence.assertOwned(sessionFile); reserveSessionWrite(sessionFile); - const entries = this.fileEntries; const serialize = (batch: readonly FileEntry[]): string => batch.map((e) => `${JSON.stringify(this.residentStore.materialize(e))}\n`).join(""); let written = 0; let durable = { entries: 0, bytes: 0 }; - let handle: FileHandle | undefined = await open(sessionFile, "wx"); + let handle: FileHandle | undefined; try { + handle = await open(sessionFile, "wx"); while (written < entries.length) { const batch = entries.slice(written); written += batch.length; @@ -1692,7 +1688,7 @@ export class SessionManager { handle = undefined; await closing.close(); // Entries persisted while the handle closed: append until a pass finds nothing new. - while (written < entries.length && this.sessionFile === sessionFile && this.fileEntries === entries) { + while (written < entries.length && this.writeState === state) { const batch = entries.slice(written); written += batch.length; const text = serialize(batch); @@ -1702,34 +1698,42 @@ export class SessionManager { } catch (error) { // While the exclusive handle is open nothing is whole yet: a part-written file would fail every later first // flush with EEXIST, so it is removed and nothing counts as flushed. - if (handle !== undefined) return discardFailedFirstFlushAsync(sessionFile, handle, error); + if (handle !== undefined || durable.entries === 0) { + if (this.writeState === state) this.writeState = { kind: "buffered" }; + if (handle !== undefined) return discardFailedFirstFlushAsync(sessionFile, handle, error); + throw error; + } // The header and the first entries are on disk and closed: keep them (senpi#4295). Cut what the failed // append landed, and write the entries it carried before this session's next append. - await this._keepCaughtUpPrefix(sessionFile, entries, durable); + await this._keepCaughtUpPrefix(state, durable, error); throw error; } // Synchronous with the last check above: no entry can land between it and the flag. - if (this.sessionFile === sessionFile && this.fileEntries === entries && written === entries.length) { - this.flushed = true; - this.ownedHeaderFile = sessionFile; + if (this.writeState === state) { + this.writeState = { kind: "flushed", ownedHeaderFile: sessionFile }; } } private async _keepCaughtUpPrefix( - sessionFile: string, - entries: FileEntry[], + state: HeaderWriteState, durable: { readonly entries: number; readonly bytes: number }, + cause: unknown, ): Promise { + if (this.writeState !== state) return; + const { file: sessionFile, entries } = state; + let tailFailure: { cause: unknown; completeBytes: number } | undefined; try { await truncate(sessionFile, durable.bytes); - } catch (cutError) { - if (!(cutError instanceof Error)) throw cutError; - this.tailMayBeTorn = true; - } - if (this.sessionFile !== sessionFile || this.fileEntries !== entries) return; - this.unwritten = { entries, from: durable.entries }; - this.flushed = true; - this.ownedHeaderFile = sessionFile; + } catch { + tailFailure = { cause, completeBytes: durable.bytes }; + } + if (this.writeState !== state) return; + this.writeState = { + kind: "flushed", + ownedHeaderFile: sessionFile, + pending: { entries, from: durable.entries }, + tailFailure, + }; } /** @@ -1755,37 +1759,38 @@ export class SessionManager { reserveSessionWrite(this.sessionFile); const persistedEntry = this.residentStore.materialize(entry); - if (this.flushed) { - if (this.tailMayBeTorn) { - truncateToLastCompleteLine(this.sessionFile); - this.tailMayBeTorn = false; + const state = this.writeState; + if (state.kind === "flushed") { + if (state.tailFailure) { + truncateToLastCompleteLine(this.sessionFile, state.tailFailure.completeBytes); + state.tailFailure = undefined; } try { // Entries a failed header catch-up left off disk go first, one line each, so a later failure resumes there. - const unwritten = this.unwritten?.entries === this.fileEntries ? this.unwritten : undefined; + const unwritten = state.pending; for (const pending of unwritten?.entries.slice(unwritten.from) ?? []) { appendFileSync(this.sessionFile, `${JSON.stringify(this.residentStore.materialize(pending))}\n`); if (unwritten) unwritten.from++; } - this.unwritten = undefined; + state.pending = undefined; appendFileSync(this.sessionFile, `${JSON.stringify(persistedEntry)}\n`); } catch (error) { - this.tailMayBeTorn = true; + state.tailFailure = { cause: error }; throw error; } return; } - // Entries stay in memory only until the branch holds a user or assistant message; then all are - // written (#10000: the prompt reaches disk even if the first turn never completes). const durableLifecycle = entry.type === "custom" && LIFECYCLE_ENTRY_TYPES.has(entry.customType); - if (!isConversationEntry(entry) && !durableLifecycle && !this._hasConversation()) return; - // An asynchronous header write owns the file until it finishes, and appends this entry. - if (this.headerWrite) { + if (state.kind === "writing") { if (durableLifecycle) throw new Error("Lifecycle append requires the pending session header write to finish"); + state.entries.push(entry); return; } + // Entries stay in memory only until the branch holds a user or assistant message; then all are + // written (#10000: the prompt reaches disk even if the first turn never completes). + if (!isConversationEntry(entry) && !durableLifecycle && !this._hasConversation()) return; const fd = openSync(this.sessionFile, "wx"); try { for (const e of [...this.fileEntries, entry]) { @@ -1795,9 +1800,11 @@ export class SessionManager { discardFailedFirstFlush(this.sessionFile, fd, error); } closeSync(fd); - this.flushed = true; // Lifecycle bookkeeping created this file before any conversation, so an empty session can still discard it. - if (!isConversationEntry(entry) && !this._hasConversation()) this.ownedHeaderFile = this.sessionFile; + this.writeState = { + kind: "flushed", + ownedHeaderFile: !isConversationEntry(entry) && !this._hasConversation() ? this.sessionFile : undefined, + }; } private _appendEntry(entry: SessionEntry, shareable?: object): void { @@ -2947,6 +2954,7 @@ export class SessionManager { this.trimmedIds.clear(); this.sessionId = newSessionId; this.sessionFile = newSessionFile; + this.writeState = { kind: "buffered" }; this.fileEntries = [header, ...branchedEntries.map((entry) => this.residentStore.externalize(entry))]; this._buildIndex(); this.historyGeneration++; @@ -2956,9 +2964,6 @@ export class SessionManager { // has a conversation, otherwise let _persist() create the file later. if (this._hasConversation()) { this._rewriteFile(); - this.flushed = true; - } else { - this.flushed = false; } return newSessionFile; diff --git a/packages/coding-agent/src/core/session-write-recovery.ts b/packages/coding-agent/src/core/session-write-recovery.ts index 96671f82ad..e93281e312 100644 --- a/packages/coding-agent/src/core/session-write-recovery.ts +++ b/packages/coding-agent/src/core/session-write-recovery.ts @@ -7,25 +7,27 @@ const SCAN_CHUNK_BYTES = 64 * 1024; /** * Cuts a JSONL session file back to its last complete line. An append that failed part-way * (ENOSPC) can leave a partial line, and the next append would be glued onto it, losing both - * entries on reload. A file that already ends in a newline is left untouched. + * entries on reload. A file that already ends in a newline is left untouched unless a failed + * header catch-up supplied its exact durable boundary: that batch may contain whole lines too, + * and all of its bytes must be cut before replaying the batch. */ -export function truncateToLastCompleteLine(path: string): void { +export function truncateToLastCompleteLine(path: string, knownCompleteBytes?: number): void { const fd = openSync(path, "r+"); try { const size = fstatSync(fd).size; const chunk = Buffer.alloc(Math.min(SCAN_CHUNK_BYTES, size)); let end = size; - while (end > 0) { + let completeLength = knownCompleteBytes; + while (completeLength === undefined && end > 0) { const start = Math.max(0, end - chunk.length); const read = readSync(fd, chunk, 0, end - start, start); const newline = chunk.subarray(0, read).lastIndexOf(NEWLINE); if (newline >= 0) { - const completeLength = start + newline + 1; - if (completeLength < size) ftruncateSync(fd, completeLength); - return; + completeLength = start + newline + 1; } end = start; } + if (completeLength !== undefined && completeLength < size) ftruncateSync(fd, completeLength); } finally { closeSync(fd); } diff --git a/packages/coding-agent/src/core/session-write-state.ts b/packages/coding-agent/src/core/session-write-state.ts new file mode 100644 index 0000000000..7a16248198 --- /dev/null +++ b/packages/coding-agent/src/core/session-write-state.ts @@ -0,0 +1,19 @@ +import type { FileEntry } from "./session-manager.ts"; + +/** A header writer owns its queue even when compaction replaces the resident mirror. */ +export interface HeaderWriteState { + readonly kind: "writing"; + readonly file: string; + readonly entries: FileEntry[]; +} + +/** Buffering cannot retain a previous file's append cursor, ownership, or torn-tail cause. */ +export type SessionWriteState = + | { readonly kind: "buffered" } + | HeaderWriteState + | { + readonly kind: "flushed"; + readonly ownedHeaderFile?: string; + pending?: { readonly entries: FileEntry[]; from: number }; + tailFailure?: { readonly cause: unknown; readonly completeBytes?: number }; + }; diff --git a/packages/coding-agent/test/session-manager-header-write.test.ts b/packages/coding-agent/test/session-manager-header-write.test.ts index 8a88f374d3..76017027b2 100644 --- a/packages/coding-agent/test/session-manager-header-write.test.ts +++ b/packages/coding-agent/test/session-manager-header-write.test.ts @@ -9,6 +9,8 @@ const hooks = vi.hoisted(() => ({ beforeClose: undefined as (() => void) | undefined, writeFails: undefined as Error | undefined, appendFails: undefined as Error | undefined, + cutFails: false, + keepWholeLine: false, })); vi.mock("fs/promises", async (importOriginal) => { @@ -19,9 +21,15 @@ vi.mock("fs/promises", async (importOriginal) => { const failure = hooks.appendFails; if (failure === undefined || typeof args[1] !== "string") return real.appendFile(...args); hooks.appendFails = undefined; - await real.appendFile(args[0], args[1].slice(0, Math.floor(args[1].length / 2))); + const keep = hooks.keepWholeLine ? args[1].indexOf("\n") + 18 : Math.floor(args[1].length / 2); + await real.appendFile(args[0], args[1].slice(0, keep)); throw failure; }, + truncate: async (...args: Parameters) => { + if (!hooks.cutFails) return real.truncate(...args); + hooks.cutFails = false; + throw Object.assign(new Error("EIO"), { code: "EIO" }); + }, open: async (...args: Parameters) => { const handle = await real.open(...args); const writeFile = handle.writeFile.bind(handle); @@ -48,6 +56,8 @@ afterEach(() => { hooks.beforeClose = undefined; hooks.writeFails = undefined; hooks.appendFails = undefined; + hooks.cutFails = false; + hooks.keepWholeLine = false; for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); }); @@ -94,6 +104,61 @@ function idsOnDisk(file: string): string[] { } describe("persistHeaderNow's asynchronous header write", () => { + it("retries an exact failed cut-back before replaying a batch that landed whole lines too", async () => { + const { manager, file } = freshSession(); + manager.appendCustomEntry("before"); + hooks.beforeClose = () => { + hooks.beforeClose = undefined; + manager.appendCustomEntry("during-one"); + manager.appendCustomEntry("during-two"); + hooks.appendFails = Object.assign(new Error("ENOSPC"), { code: "ENOSPC" }); + hooks.keepWholeLine = true; + hooks.cutFails = true; + }; + await expect(manager.persistHeaderNow()).rejects.toMatchObject({ code: "ENOSPC" }); + expect(readFileSync(file, "utf8")).toContain('"customType":"during-one"'); + manager.appendCustomEntry("after"); + expect(idsOnDisk(file)).toEqual(manager.getEntries().map((entry) => entry.id)); + }); + + it.each([false, true])("keeps header ownership across compaction, catch-up failure=%s", async (fails) => { + const { manager, file } = freshSession(); + const first = manager.appendCustomEntry("before"); + let compaction = ""; + hooks.beforeClose = () => { + hooks.beforeClose = undefined; + compaction = manager.appendCompaction("summary", first, 1); + if (fails) hooks.appendFails = Object.assign(new Error("ENOSPC"), { code: "ENOSPC" }); + }; + const writing = manager.persistHeaderNow(); + if (fails) await expect(writing).rejects.toMatchObject({ code: "ENOSPC" }); + else await writing; + expect(manager.isTranscriptFlushed()).toBe(true); + const next = manager.appendCustomEntry("after"); + expect(idsOnDisk(file)).toEqual([first, compaction, next]); + expect(idsOnDisk(file)).toEqual( + SessionManager.open(file) + .getEntries() + .map((entry) => entry.id), + ); + }); + + it("does not replay a failed catch-up after discarding and recreating its startup-only file", async () => { + const { manager, file } = freshSession(); + manager.appendCustomEntry("before", {}, { startupBookkeeping: true }); + hooks.beforeClose = () => { + hooks.beforeClose = undefined; + manager.appendCustomEntry("during-close", {}, { startupBookkeeping: true }); + hooks.appendFails = Object.assign(new Error("ENOSPC"), { code: "ENOSPC" }); + }; + await expect(manager.persistHeaderNow()).rejects.toMatchObject({ code: "ENOSPC" }); + expect(await manager.discardHeaderOnlyFile()).toBe(true); + expect(existsSync(file)).toBe(false); + manager.appendMessage({ role: "user", content: "first flush", timestamp: 1 }); + manager.appendMessage(assistant("next append")); + expect(idsOnDisk(file)).toEqual(manager.getEntries().map((entry) => entry.id)); + }); + it("writes the entries persisted while its file handle closes before the transcript counts as flushed", async () => { const { manager, file } = freshSession(); manager.appendCustomEntry("before", { n: 0 }); diff --git a/packages/coding-agent/test/session-manager/torn-append-every-offset.test.ts b/packages/coding-agent/test/session-manager/torn-append-every-offset.test.ts new file mode 100644 index 0000000000..978611c398 --- /dev/null +++ b/packages/coding-agent/test/session-manager/torn-append-every-offset.test.ts @@ -0,0 +1,123 @@ +import type { AssistantMessage } from "@earendil-works/pi-ai"; +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { SessionManager } from "../../src/core/session-manager.ts"; + +const fault = vi.hoisted(() => ({ keepBytes: undefined as number | undefined })); +vi.mock("fs", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + appendFileSync: (...args: Parameters) => { + const keep = fault.keepBytes; + if (keep === undefined || typeof args[1] !== "string") return actual.appendFileSync(...args); + fault.keepBytes = undefined; + actual.appendFileSync(args[0], Buffer.from(args[1]).subarray(0, keep)); + throw Object.assign(new Error("ENOSPC"), { code: "ENOSPC" }); + }, + }; +}); + +const roots: string[] = []; +const managers: SessionManager[] = []; +afterEach(() => { + fault.keepBytes = undefined; + for (const manager of managers.splice(0)) manager.dispose(); + for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); +}); + +function assistant(text: string): AssistantMessage { + return { + role: "assistant", + content: [{ type: "text", text }], + api: "anthropic-messages", + provider: "anthropic", + model: "test", + usage: { + input: 1, + output: 1, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 2, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: 2, + }; +} + +function transcript(): { root: string; file: string; manager: SessionManager } { + const root = mkdtempSync(join(tmpdir(), "senpi-every-offset-")); + roots.push(root); + const manager = SessionManager.create(root, join(root, "sessions")); + managers.push(manager); + manager.appendMessage({ role: "user", content: "u1", timestamp: 1 }); + manager.appendMessage(assistant("a1")); + manager.appendMessage({ role: "user", content: "u2", timestamp: 3 }); + manager.appendMessage(assistant("a2")); + const file = manager.getSessionFile(); + if (!file) throw new Error("missing transcript"); + return { root, file, manager }; +} + +function reopen(file: string): SessionManager { + const manager = SessionManager.open(file); + managers.push(manager); + return manager; +} + +describe("torn transcript byte boundaries (senpi#4295)", () => { + it("opens every byte-offset prefix of a four-message transcript and appends without losing whole entries", () => { + const { root, file } = transcript(); + const bytes = readFileSync(file); + let end = 0; + const entryEnds = bytes + .toString("utf8") + .trimEnd() + .split("\n") + .map((line) => { + end += Buffer.byteLength(line) + 1; + return { end: end - 1, entry: JSON.parse(line) as { type: string; id: string } }; + }) + .filter(({ entry }) => entry.type !== "session"); + for (let cut = 0; cut <= bytes.length; cut++) { + const path = join(root, `cut-${cut}.jsonl`); + writeFileSync(path, bytes.subarray(0, cut)); + const expected = entryEnds.filter(({ end: entryEnd }) => entryEnd <= cut).map(({ entry }) => entry.id); + const manager = reopen(path); + expect( + manager.getEntries().map((entry) => entry.id), + `cut ${cut}`, + ).toEqual(expected); + const next = manager.appendMessage({ role: "user", content: "retry", timestamp: 4 }); + expect( + reopen(path) + .getEntries() + .map((entry) => entry.id), + `cut ${cut} after append`, + ).toEqual([...expected, next]); + } + }); + + it.each(Array.from({ length: 159 }, (_, offset) => offset))( + "retries in the same process after an ENOSPC append at byte %i", + (offset) => { + const { file, manager } = transcript(); + const prefix = manager.getEntries().map((entry) => entry.id); + fault.keepBytes = offset; + expect(() => manager.appendCustomEntry("torn", { pad: "x".repeat(160) })).toThrow("ENOSPC"); + expect(manager.getEntries().map((entry) => entry.id)).toEqual(prefix); + const next = manager.appendCustomEntry("retry"); + const following = manager.appendMessage({ role: "user", content: "after retry", timestamp: 5 }); + expect( + reopen(file) + .getEntries() + .map((entry) => entry.id), + ).toEqual([...prefix, next, following]); + const disk = readFileSync(file, "utf8").trimEnd().split("\n"); + expect(disk.map((line) => JSON.parse(line)).filter((entry) => entry.type !== "session")).toHaveLength(6); + }, + ); +}); diff --git a/packages/coding-agent/test/session-manager/write-state-interleavings.test.ts b/packages/coding-agent/test/session-manager/write-state-interleavings.test.ts new file mode 100644 index 0000000000..a4faee793e --- /dev/null +++ b/packages/coding-agent/test/session-manager/write-state-interleavings.test.ts @@ -0,0 +1,153 @@ +import { existsSync, mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { SessionManager } from "../../src/core/session-manager.ts"; + +const fault = vi.hoisted(() => ({ + beforeClose: undefined as (() => void) | undefined, + asyncAppend: false, + syncAppend: false, +})); + +function noSpace(): Error { + return Object.assign(new Error("ENOSPC"), { code: "ENOSPC" }); +} + +vi.mock("fs/promises", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + appendFile: async (...args: Parameters) => { + if (!fault.asyncAppend || typeof args[1] !== "string") return actual.appendFile(...args); + fault.asyncAppend = false; + await actual.appendFile(args[0], args[1].slice(0, 17)); + throw noSpace(); + }, + open: async (...args: Parameters) => { + const handle = await actual.open(...args); + const close = handle.close.bind(handle); + handle.close = async () => { + const callback = fault.beforeClose; + fault.beforeClose = undefined; + callback?.(); + await close(); + }; + return handle; + }, + }; +}); + +vi.mock("fs", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + appendFileSync: (...args: Parameters) => { + if (!fault.syncAppend || typeof args[1] !== "string") return actual.appendFileSync(...args); + fault.syncAppend = false; + actual.appendFileSync(args[0], args[1].slice(0, 17)); + throw noSpace(); + }, + }; +}); + +const roots: string[] = []; +const managers: SessionManager[] = []; +afterEach(() => { + fault.beforeClose = undefined; + fault.asyncAppend = false; + fault.syncAppend = false; + for (const manager of managers.splice(0)) manager.dispose(); + for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); +}); + +/** Read only whole lines: an unsuccessful append may leave one uncommitted partial line. */ +function durableIds(file: string): string[] { + if (!existsSync(file)) return []; + const text = readFileSync(file, "utf8"); + return text + .slice(0, text.lastIndexOf("\n") + 1) + .trim() + .split("\n") + .map((line) => JSON.parse(line) as { type: string; id: string }) + .filter((entry) => entry.type !== "session") + .map((entry) => entry.id); +} + +describe("transcript write-state interleavings", () => { + it.each(Array.from({ length: 32 }, (_, index) => index + 1))( + "preserves exactly one ordered durable prefix under seed %i", + async (seed) => { + const root = mkdtempSync(join(tmpdir(), "senpi-write-state-")); + roots.push(root); + let manager = SessionManager.create(root, join(root, "sessions")); + managers.push(manager); + const file = manager.getSessionFile(); + if (!file) throw new Error("missing session path"); + const accepted: string[] = []; + let previous: string[] = []; + let random = seed; + const next = (): number => (random = (Math.imul(random, 1664525) + 1013904223) >>> 0); + const appendSetup = (): void => { + accepted.push(manager.appendCustomEntry("startup", {}, { startupBookkeeping: true })); + }; + const append = (conversation: boolean): void => { + const id = conversation + ? manager.appendMessage({ role: "user", content: "message", timestamp: 1 }) + : manager.appendCustomEntry("startup", {}, { startupBookkeeping: true }); + accepted.push(id); + }; + const flush = async (fails: boolean): Promise => { + if (!manager.isTranscriptFlushed()) { + fault.beforeClose = () => { + appendSetup(); + fault.asyncAppend = fails; + }; + } + try { + await manager.persistHeaderNow(); + } catch (error) { + expect(error).toMatchObject({ code: "ENOSPC" }); + } + }; + // Every seed reaches the reported catch-up/discard boundary, then varies what comes next. + appendSetup(); + await flush(true); + for (let step = 0; step < 64; step++) { + const operation = (next() >>> 16) % 6; + if (operation === 0) append((next() >>> 16) % 2 === 0); + else if (operation === 1) { + fault.syncAppend = manager.isTranscriptFlushed(); + try { + append(false); + } catch (error) { + expect(error).toMatchObject({ code: "ENOSPC" }); + } + } else if (operation === 2) { + if (await manager.discardHeaderOnlyFile()) previous = []; + } else if (operation === 3 || operation === 4) await flush(operation === 4); + else if (existsSync(file)) { + // A reopened reader can observe only the durable prefix, not the failed writer's pending queue. + const reader = SessionManager.open(file); + managers.push(reader); + expect(reader.getEntries().map((entry) => entry.id)).toEqual(durableIds(file)); + if (durableIds(file).length === accepted.length) manager = reader; + } + const disk = durableIds(file); + expect(new Set(disk).size).toBe(disk.length); + expect(disk).toEqual(accepted.slice(0, disk.length)); + expect(disk.slice(0, previous.length)).toEqual(previous); + expect(manager.getEntries().map((entry) => entry.id)).toEqual(accepted); + previous = disk; + } + fault.syncAppend = false; + fault.asyncAppend = false; + append(true); + append(false); + expect(durableIds(file)).toEqual(accepted); + const reader = SessionManager.open(file); + managers.push(reader); + expect(reader.getEntries().map((entry) => entry.id)).toEqual(accepted); + }, + ); +});