From e4b9fabfe2c1efd9e3f1d7fd2c55eaee59609dc4 Mon Sep 17 00:00:00 2001 From: Neil Date: Sat, 12 Sep 2026 18:08:27 -0700 Subject: [PATCH] fix: page journal replay reads so no SQLite snapshot outlives its statement - iterateJournalEpochRows fetches one completed LIMIT statement per page instead of a lazily consumed .iterate() cursor, so reduction never runs inside an open read snapshot and a WAL checkpoint can pass mid-replay. Regression test: a checkpoint issued from inside the reducer is not busy. - The retention test now asserts the applyJournalRow spy observed every row, so the 8 MiB bound cannot pass vacuously if the spy stops intercepting. - Reliability gate manifest records the new assertion and the paged read design. --- config/reliability-gates.jsonc | 5 ++-- .../journal-row-table.ts | 30 ++++++++++++------- .../journal-streaming-replay.test.ts | 22 ++++++++++++++ 3 files changed, 45 insertions(+), 12 deletions(-) diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index f1218ccdd07..85c5eb08b0f 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -27,7 +27,7 @@ "https://github.com/stablyai/orca/blob/main/src/main/native-chat/agent-session-journal/journal-open.ts" ], "invariant": "Replay preserves latest revisions, original item order, fences, aliases, submissions, repair precedence, read-only schema latching and cursor cleanup while retaining live items rather than all historical bodies.", - "oracle": "Replay 2,048 16 KiB revisions into one latest item with less than 8 MiB sampled live heap growth; preserve prefix and future-schema latching after a gap, malformed suffix repair precedence, and release SQLite read cursors on early exit. Existing journal and subscriber tests cover replayed content and recovery.", + "oracle": "Replay 2,048 16 KiB revisions into one latest item with less than 8 MiB sampled live heap growth; preserve prefix and future-schema latching after a gap, malformed suffix repair precedence, and hold no SQLite read snapshot across reduction (a mid-replay checkpoint is not busy). Existing journal and subscriber tests cover replayed content and recovery.", "commands": [ "ORCA_BACKGROUND_LAUNCH=1 pnpm test src/main/native-chat/agent-session-journal src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts src/main/native-chat/agent-session-wire/agent-session-history-forward-read-budget.test.ts src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts src/main/native-chat/agent-session-wire/agent-session-journal-recovery.test.ts", "ORCA_BACKGROUND_LAUNCH=1 pnpm test src/main/native-chat/agent-session-journal/journal-streaming-replay.test.ts src/main/native-chat/agent-session-journal/journal-corruption-repair.test.ts" @@ -41,6 +41,7 @@ "file": "src/main/native-chat/agent-session-journal/journal-streaming-replay.test.ts", "assertions": [ "releases superseded revision bodies while reducing a long journal", + "holds no read snapshot while reducing, so a checkpoint can pass mid-replay", "keeps the prefix but latches read-only for a future row beyond a gap", "keeps gap repair precedence when a later row is malformed", "rejects an unanchored prefix before a later gap" @@ -72,7 +73,7 @@ }, "performanceBudget": { "required": true, - "evidence": "Synchronous SQLite iteration and immediate reduction retain reduced state plus the current row. No cooperative await pins the cursor; no new polling, subprocess, provider call or wire change." + "evidence": "Paged SQLite reads (one completed statement per page) and immediate reduction retain reduced state plus one page of rows. No cursor or read snapshot outlives its statement, so a WAL checkpoint can pass mid-replay; no new polling, subprocess, provider call or wire change." }, "knownGaps": [ "No Windows/Linux runtime execution or real remote-host validation.", diff --git a/src/main/native-chat/agent-session-journal/journal-row-table.ts b/src/main/native-chat/agent-session-journal/journal-row-table.ts index fa566b2060d..f07ad19d4d5 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-table.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-table.ts @@ -52,17 +52,29 @@ export function readJournalEpochRows( sessionId: string, epoch: string ): JournalStoredRow[] { - return [...iterateJournalEpochRows(db, sessionId, epoch)] + return toStoredRows(db.prepare(SELECT_EPOCH_ROWS).all(sessionId, epoch)) } -/** Consume synchronously: closing or breaking the loop releases SQLite's read snapshot. */ +// Why pages, not `.iterate()`: a lazily consumed cursor pins a read snapshot for as long as the +// consumer reduces, and a WAL checkpoint cannot pass an open snapshot. Each page is one completed +// statement, so the consumer's memory is bounded by a page while no snapshot outlives a fetch. +const EPOCH_ROW_PAGE_SIZE = 128 + +/** Epoch rows in sequence order, fetched one completed statement at a time. */ export function* iterateJournalEpochRows( db: Database.Database, sessionId: string, epoch: string ): Generator { - for (const row of db.prepare(SELECT_EPOCH_ROWS).iterate(sessionId, epoch)) { - yield toStoredRow(row) + let afterSeq = Number.MIN_SAFE_INTEGER + for (;;) { + const page = readJournalRowsAfter(db, sessionId, epoch, afterSeq, EPOCH_ROW_PAGE_SIZE) + yield* page + const last = page.at(-1) + if (page.length < EPOCH_ROW_PAGE_SIZE || last === undefined) { + return + } + afterSeq = last.seq } } @@ -103,10 +115,8 @@ export function deleteJournalRowSuffix( } function toStoredRows(rows: readonly unknown[]): JournalStoredRow[] { - return rows.map(toStoredRow) -} - -function toStoredRow(entry: unknown): JournalStoredRow { - const record = entry as { epoch: string; seq: number; ts: number; row_json: string } - return { epoch: record.epoch, seq: record.seq, ts: record.ts, rowJson: record.row_json } + return rows.map((entry) => { + const record = entry as { epoch: string; seq: number; ts: number; row_json: string } + return { epoch: record.epoch, seq: record.seq, ts: record.ts, rowJson: record.row_json } + }) } diff --git a/src/main/native-chat/agent-session-journal/journal-streaming-replay.test.ts b/src/main/native-chat/agent-session-journal/journal-streaming-replay.test.ts index 39eb70ee172..0f1490c3ca8 100644 --- a/src/main/native-chat/agent-session-journal/journal-streaming-replay.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-streaming-replay.test.ts @@ -73,11 +73,13 @@ describe('streaming journal replay', () => { gc() const initial = process.memoryUsage().heapUsed let peak = initial + let applied = 0 const apply = reducer.applyJournalRow const spy = vi.spyOn(reducer, 'applyJournalRow') spy.mockImplementation((state, row) => { // The probe must not retain old row bodies in Vitest's call history. spy.mockClear() + applied += 1 if (row.seq % 256 === 0) { gc() peak = Math.max(peak, process.memoryUsage().heapUsed) @@ -88,9 +90,29 @@ describe('streaming journal replay', () => { expect(loaded.state.items.size).toBe(1) expect(loaded.state.items.get('message-1')?.revision).toBe(2049) expect(loaded.state.lastSequence).toBe(2049) + // The probe must have measured every row, or the heap bound above is vacuous. + expect(applied).toBe(2049) expect(peak - initial).toBeLessThan(8 * 1024 * 1024) }) + it('holds no read snapshot while reducing, so a checkpoint can pass mid-replay', () => { + put(anchor()) + for (let seq = 2; seq <= 300; seq++) { + put(revision(seq)) + } + const apply = reducer.applyJournalRow + const checkpoints: { busy: number }[] = [] + vi.spyOn(reducer, 'applyJournalRow').mockImplementation((state, row) => { + if (row.seq === 2 || row.seq === 200) { + checkpoints.push(...(opened.db.pragma('wal_checkpoint(PASSIVE)') as { busy: number }[])) + } + apply(state, row) + }) + const loaded = replayJournal(opened.db, false, sessionId)! + expect(loaded.state.lastSequence).toBe(300) + expect(checkpoints.map((entry) => entry.busy)).toEqual([0, 0]) + }) + it('keeps the prefix but latches read-only for a future row beyond a gap', () => { put(anchor()) put(revision(2))