From d4bcc028a3d1bdc2a825d877f39a9745a5da881c Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Fri, 7 Aug 2026 14:06:55 -0700 Subject: [PATCH] fix(native-chat): preserve journal durability across compaction --- .../journal-compaction.ts | 14 +- .../journal-crash-boundary.test.ts | 2 +- .../journal-epoch-rollover.ts | 6 +- .../agent-session-journal/journal-log-file.ts | 29 ++++- .../agent-session-journal/journal-open.ts | 16 ++- .../journal-store-contracts.ts | 38 ++++++ .../journal-store.test.ts | 96 +++++++++++++- .../agent-session-journal/journal-store.ts | 123 +++++++++--------- 8 files changed, 242 insertions(+), 82 deletions(-) create mode 100644 src/main/native-chat/agent-session-journal/journal-store-contracts.ts diff --git a/src/main/native-chat/agent-session-journal/journal-compaction.ts b/src/main/native-chat/agent-session-journal/journal-compaction.ts index dd8f4f38066..c2c70476fd4 100644 --- a/src/main/native-chat/agent-session-journal/journal-compaction.ts +++ b/src/main/native-chat/agent-session-journal/journal-compaction.ts @@ -15,12 +15,13 @@ import { } from './journal-reducer' import { pruneJournalBlobs } from './journal-blob-store' import { + journalSnapshotByteLength, rewriteJournalLog, writeJournalSnapshotFile, type JournalSnapshotFile } from './journal-log-file' import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types' -import type { JournalRow } from './journal-row-schema' +import { journalRowByteLength, type JournalRow } from './journal-row-schema' export type JournalCompactionPolicy = { /** Always keep at least this many rows, however old they are. */ @@ -40,6 +41,7 @@ export type JournalCompactionResult = { tailRows: JournalRow[] compactedThrough: number oldestSequence: number + sizeBytes: number } export async function compactJournal(input: { @@ -58,7 +60,12 @@ export async function compactJournal(input: { v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, epoch: input.state.epoch, compactedThrough, + highestFence: input.state.highestFence, items: rendered.items, + tombstones: [...input.state.tombstones.entries()].map(([itemId, revision]) => ({ + itemId, + revision + })), submissions: rendered.submissions, receipts: [...input.state.receipts.values()].map((receipt) => ({ clientMessageId: receipt.clientMessageId, @@ -83,7 +90,10 @@ export async function compactJournal(input: { return { tailRows: retained, compactedThrough, - oldestSequence: retained[0]?.seq ?? compactedThrough + 1 + oldestSequence: retained[0]?.seq ?? compactedThrough + 1, + sizeBytes: + journalSnapshotByteLength(snapshot) + + retained.reduce((total, row) => total + journalRowByteLength(row), 0) } } diff --git a/src/main/native-chat/agent-session-journal/journal-crash-boundary.test.ts b/src/main/native-chat/agent-session-journal/journal-crash-boundary.test.ts index 0f8d85d34fc..f34f2ce171c 100644 --- a/src/main/native-chat/agent-session-journal/journal-crash-boundary.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-crash-boundary.test.ts @@ -215,7 +215,7 @@ describe('crash between provider accept and journal commit', () => { providerIdentity: ACCEPTED_IDENTITY, fence: 1 }) - await journal.compact() + await journal.compact(1) const reopened = await open() expect(reopened.receiptFor('cm_1')?.providerItemId).toBe(agentJournalItemKey(ACCEPTED_IDENTITY)) diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-rollover.ts b/src/main/native-chat/agent-session-journal/journal-epoch-rollover.ts index b6b9a5ea9be..6f35a71da39 100644 --- a/src/main/native-chat/agent-session-journal/journal-epoch-rollover.ts +++ b/src/main/native-chat/agent-session-journal/journal-epoch-rollover.ts @@ -23,7 +23,7 @@ export async function publishNewEpoch(input: { reason: AgentJournalEpochReason fence: number now: number -}): Promise<{ state: JournalReducerState; row: JournalRow }> { +}): Promise<{ state: JournalReducerState; row: JournalRow; sizeBytes: number }> { const row: JournalRow = { kind: 'epoch', reason: input.reason, @@ -35,7 +35,7 @@ export async function publishNewEpoch(input: { ts: input.now } const state = createJournalReducerState(input.sessionId, input.epoch) - await compactJournal({ + const compacted = await compactJournal({ journalDir: input.journalDir, state, tailRows: [row], @@ -44,5 +44,5 @@ export async function publishNewEpoch(input: { }) applyJournalRow(state, row) state.oldestSequence = 1 - return { state, row } + return { state, row, sizeBytes: compacted.sizeBytes } } diff --git a/src/main/native-chat/agent-session-journal/journal-log-file.ts b/src/main/native-chat/agent-session-journal/journal-log-file.ts index 795b7436c64..a951954bb0b 100644 --- a/src/main/native-chat/agent-session-journal/journal-log-file.ts +++ b/src/main/native-chat/agent-session-journal/journal-log-file.ts @@ -15,6 +15,7 @@ import type { AgentJournalRenderItem, AgentJournalSubmission } from '../../../shared/agent-session-journal-types' +import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types' import { parseJournalRow, serializeJournalRow, type JournalRow } from './journal-row-schema' export const JOURNAL_LOG_FILE = 'log.jsonl' @@ -25,7 +26,9 @@ export type JournalSnapshotFile = { epoch: string /** Highest sequence folded into `items`; the tail starts after it. */ compactedThrough: number + highestFence: number items: AgentJournalRenderItem[] + tombstones: { itemId: string; revision: number }[] submissions: AgentJournalSubmission[] /** Receipts outlive the rows that minted them: a client reconnecting after * compaction must still get the same answer instead of re-sending. */ @@ -52,19 +55,33 @@ export type JournalReadResult = { malformed: number } +export type JournalSnapshotReadResult = { + snapshot: JournalSnapshotFile | null + unreadable: boolean +} + export async function ensureJournalDir(journalDir: string): Promise { await mkdir(journalDir, { recursive: true }) } export async function readJournalSnapshotFile( journalDir: string -): Promise { +): Promise { try { const raw = await readFile(join(journalDir, JOURNAL_SNAPSHOT_FILE), 'utf-8') - const parsed = JSON.parse(raw) as JournalSnapshotFile - return typeof parsed?.epoch === 'string' ? parsed : null + const parsed = JSON.parse(raw) as Partial + if (typeof parsed?.v !== 'number' || !Number.isInteger(parsed.v) || parsed.v < 1) { + return { snapshot: null, unreadable: false } + } + if (parsed.v > AGENT_SESSION_JOURNAL_SCHEMA_VERSION) { + return { snapshot: null, unreadable: true } + } + return { + snapshot: typeof parsed?.epoch === 'string' ? (parsed as JournalSnapshotFile) : null, + unreadable: false + } } catch { - return null + return { snapshot: null, unreadable: false } } } @@ -76,6 +93,10 @@ export async function writeJournalSnapshotFile( await writeFileDurable(durableWriteTempPath(target), target, JSON.stringify(snapshot)) } +export function journalSnapshotByteLength(snapshot: JournalSnapshotFile): number { + return Buffer.byteLength(JSON.stringify(snapshot), 'utf8') +} + export async function readJournalLog(journalDir: string): Promise { let raw: string try { diff --git a/src/main/native-chat/agent-session-journal/journal-open.ts b/src/main/native-chat/agent-session-journal/journal-open.ts index 9fbd730fb01..e51918f3138 100644 --- a/src/main/native-chat/agent-session-journal/journal-open.ts +++ b/src/main/native-chat/agent-session-journal/journal-open.ts @@ -9,6 +9,7 @@ import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' import { findSequenceGap } from './journal-cursor' import { + journalSnapshotByteLength, readJournalLog, readJournalSnapshotFile, type JournalSnapshotFile @@ -38,11 +39,12 @@ export async function loadJournal( journalDir: string, sessionId: string ): Promise { - const snapshot = await readJournalSnapshotFile(journalDir) + const snapshotRead = await readJournalSnapshotFile(journalDir) + const snapshot = snapshotRead.snapshot const log = await readJournalLog(journalDir) const epoch = resolveEpoch(snapshot, log.rows) if (!epoch) { - return log.unreadable ? emptyReadOnlyLoad(sessionId) : null + return log.unreadable || snapshotRead.unreadable ? emptyReadOnlyLoad(sessionId) : null } const compactedThrough = snapshot?.epoch === epoch ? snapshot.compactedThrough : 0 @@ -71,9 +73,11 @@ export async function loadJournal( state, tailRows, compactedThrough, - readOnly: log.unreadable, + readOnly: log.unreadable || snapshotRead.unreadable, corrupt, - sizeBytes: tailRows.reduce((total, row) => total + journalRowByteLength(row), 0) + sizeBytes: + (snapshot?.epoch === epoch ? journalSnapshotByteLength(snapshot) : 0) + + tailRows.reduce((total, row) => total + journalRowByteLength(row), 0) } } @@ -115,6 +119,9 @@ function seedState( for (const item of snapshot.items) { state.items.set(item.itemId, item) } + for (const tombstone of snapshot.tombstones ?? []) { + state.tombstones.set(tombstone.itemId, tombstone.revision) + } for (const submission of snapshot.submissions) { state.submissions.set(submission.clientMessageId, { ...submission } as AgentJournalSubmission) } @@ -131,6 +138,7 @@ function seedState( } state.lastSequence = snapshot.compactedThrough state.oldestSequence = snapshot.compactedThrough + 1 + state.highestFence = snapshot.highestFence ?? 0 return state } diff --git a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts b/src/main/native-chat/agent-session-journal/journal-store-contracts.ts new file mode 100644 index 00000000000..ac4ba7d1e11 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-store-contracts.ts @@ -0,0 +1,38 @@ +import type { + AgentJournalCursor, + AgentJournalItemIdentity, + AgentJournalResetReason, + AgentSessionJournalIdentity +} from '../../../shared/agent-session-journal-types' +import type { JournalCompactionPolicy } from './journal-compaction' +import type { JournalPayloadLimits } from './journal-payload-bounds' +import type { JournalRow } from './journal-row-schema' + +export type AgentSessionJournalOptions = { + identity: AgentSessionJournalIdentity + journalDir: string + limits?: JournalPayloadLimits + compaction?: JournalCompactionPolicy + now?: () => number + mintEpoch?: () => string +} + +export type JournalReadSince = + | { ok: true; rows: JournalRow[]; cursor: AgentJournalCursor } + | { ok: false; reset: AgentJournalResetReason } + +export type ResolveDispatchInput = { + clientMessageId: string + fence: number + /** Crash reconciliation, rather than live dispatch, settled this. */ + recovered?: true +} & ( + | { state: 'accepted'; providerIdentity: AgentJournalItemIdentity } + | { state: 'rejected' | 'unknown'; reason?: string | null } +) + +export type JournalAppendResult = { + cursor: AgentJournalCursor + itemId: string + revision: number +} diff --git a/src/main/native-chat/agent-session-journal/journal-store.test.ts b/src/main/native-chat/agent-session-journal/journal-store.test.ts index f6e85e63adf..8475e4c4349 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.test.ts @@ -108,6 +108,17 @@ describe('fences', () => { await journal.appendItem(item(2), body('c'), { fence: 7 }) expect(journal.snapshot().items.map((entry) => entry.body)).toEqual([body('a'), body('c')]) }) + + it('rejects rollover and compaction from a stale writer', async () => { + const journal = await open() + await journal.appendItem(item(0), body('a'), { fence: 7 }) + + await expect(journal.rollEpoch('handle_forked', 6)).rejects.toMatchObject({ + code: 'journal_stale_fence' + }) + await expect(journal.compact(6)).rejects.toMatchObject({ code: 'journal_stale_fence' }) + expect(journal.snapshot().items.map((entry) => entry.body)).toEqual([body('a')]) + }) }) describe('replay', () => { @@ -147,6 +158,18 @@ describe('replay', () => { expect(reopened.snapshot().items).toHaveLength(0) }) + it('serializes an append requested during rollover into the new epoch', async () => { + const journal = await open() + await journal.appendItem(item(0), body('old epoch'), { fence: 1 }) + + const rolled = journal.rollEpoch('handle_forked', 2) + const appended = journal.appendItem(item(1), body('new epoch'), { fence: 2 }) + await Promise.all([rolled, appended]) + + expect(journal.snapshot().items.map((entry) => entry.body)).toEqual([body('new epoch')]) + expect((await open()).snapshot()).toEqual(journal.snapshot()) + }) + it('rolls the epoch when the log lost a row', async () => { const journal = await open() for (let index = 0; index < 4; index += 1) { @@ -171,7 +194,7 @@ describe('compaction and retention', () => { } const rendered = journal.snapshot() const tip = journal.cursor() - await journal.compact() + await journal.compact(1) expect(journal.snapshot()).toEqual(rendered) expect(journal.readSince({ epoch: tip.epoch, sequence: 1 })).toEqual({ @@ -193,7 +216,7 @@ describe('compaction and retention', () => { } const rendered = journal.snapshot() const logBefore = await readFile(join(root, JOURNAL_LOG_FILE), 'utf-8') - await journal.compact() + await journal.compact(1) // Simulate the crash: the snapshot landed, the truncation did not. await writeFile(join(root, JOURNAL_LOG_FILE), logBefore, 'utf-8') @@ -202,6 +225,35 @@ describe('compaction and retention', () => { expect(reopened.snapshot().items).toHaveLength(5) }) + it('serializes compaction behind an append so the log rewrite cannot discard it', async () => { + const journal = await open({ compaction: { minTailRows: 1, retainTailMs: 0 } }) + await journal.appendItem(item(0), body('before'), { fence: 1 }) + + const appended = journal.appendItem(item(1), body('during'), { fence: 1 }) + await Promise.all([appended, journal.compact(1)]) + + const reopened = await open() + expect(reopened.snapshot()).toEqual(journal.snapshot()) + expect(reopened.snapshot().items.map((entry) => entry.body)).toEqual([ + body('before'), + body('during') + ]) + }) + + it('preserves tombstones and the highest writer fence across compaction', async () => { + const journal = await open({ compaction: { minTailRows: 1, retainTailMs: 0 } }) + await journal.appendItem(item(0), body('removed'), { fence: 7 }) + await journal.appendTombstone(item(0), { fence: 7 }) + await journal.compact(7) + + const reopened = await open() + await expect( + reopened.appendItem(item(1), body('stale writer'), { fence: 6 }) + ).rejects.toMatchObject({ code: 'journal_stale_fence' }) + await reopened.appendItem(item(0), body('late stale revision'), { fence: 7 }) + expect(reopened.snapshot().items).toHaveLength(0) + }) + it('prunes blobs no live row references and keeps the ones that survive', async () => { const journal = await open({ compaction: { minTailRows: 1, retainTailMs: 0 } }) const kept = boundPayload('k'.repeat(64), { @@ -220,7 +272,7 @@ describe('compaction and retention', () => { { kind: 'tool-call', name: 'bash', input: {}, state: 'completed', output: kept }, { fence: 1 } ) - await journal.compact() + await journal.compact(1) expect(await readJournalBlob(root, kept.digest)).toBe('k'.repeat(64)) expect(await readJournalBlob(root, dropped.digest)).toBeNull() @@ -274,6 +326,24 @@ describe('bounds', () => { ).rejects.toMatchObject({ code: 'journal_bound_exceeded' }) }) + it('keeps the size bound across repeated compaction cycles', async () => { + const journal = await open({ + limits: { ...DEFAULT_JOURNAL_PAYLOAD_LIMITS, maxSessionBytes: 6_000 }, + compaction: { minTailRows: 1, retainTailMs: 0 } + }) + let rejection: unknown + for (let index = 0; index < 30; index += 1) { + try { + await journal.appendItem(item(index), body('x'.repeat(512)), { fence: 1 }) + await journal.compact(1) + } catch (error) { + rejection = error + break + } + } + expect(rejection).toMatchObject({ code: 'journal_bound_exceeded' }) + }) + it('refuses an append past the per-window rate bound', async () => { const journal = await open({ limits: { ...DEFAULT_JOURNAL_PAYLOAD_LIMITS, maxAppendsPerWindow: 3, appendWindowMs: 60_000 } @@ -312,7 +382,7 @@ describe('schema', () => { await expect(reopened.appendItem(item(1), body('b'), { fence: 1 })).rejects.toMatchObject({ code: 'journal_read_only' }) - await expect(reopened.compact()).rejects.toMatchObject({ code: 'journal_read_only' }) + await expect(reopened.compact(1)).rejects.toMatchObject({ code: 'journal_read_only' }) expect(reopened.readSince({ epoch: reopened.epoch, sequence: 0 })).toEqual({ ok: false, reset: 'schema_unreadable' @@ -321,6 +391,24 @@ describe('schema', () => { expect(await readFile(logPath, 'utf-8')).toContain('"v":99') }) + it('degrades to read-only on a snapshot from a newer build without overwriting it', async () => { + const journal = await open() + await journal.appendItem(item(0), body('a'), { fence: 1 }) + await journal.compact(1) + const snapshotPath = join(root, JOURNAL_SNAPSHOT_FILE) + const snapshot = JSON.parse(await readFile(snapshotPath, 'utf-8')) as Record + const future = JSON.stringify({ ...snapshot, v: 99, futureState: 'keep me' }) + await writeFile(snapshotPath, future, 'utf-8') + + const reopened = await open() + expect(reopened.isReadOnly).toBe(true) + await expect(reopened.appendItem(item(1), body('b'), { fence: 1 })).rejects.toMatchObject({ + code: 'journal_read_only' + }) + await expect(reopened.compact(1)).rejects.toMatchObject({ code: 'journal_read_only' }) + expect(await readFile(snapshotPath, 'utf-8')).toBe(future) + }) + it('skips a malformed line without giving up the journal', async () => { const journal = await open() await journal.appendItem(item(0), body('a'), { fence: 1 }) diff --git a/src/main/native-chat/agent-session-journal/journal-store.ts b/src/main/native-chat/agent-session-journal/journal-store.ts index d4cb6358454..6ea1220b423 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -12,7 +12,6 @@ import type { AgentJournalItemBody, AgentJournalItemIdentity, AgentJournalMessageItem, - AgentJournalResetReason, AgentJournalSnapshot, AgentJournalSubmission, AgentSessionJournalIdentity @@ -24,7 +23,7 @@ import { resolveJournalResume, type JournalResume } from './journal-cursor' import { publishNewEpoch } from './journal-epoch-rollover' import { appendJournalRows, ensureJournalDir } from './journal-log-file' import { loadJournal } from './journal-open' -import { DEFAULT_JOURNAL_PAYLOAD_LIMITS, type JournalPayloadLimits } from './journal-payload-bounds' +import { DEFAULT_JOURNAL_PAYLOAD_LIMITS } from './journal-payload-bounds' import { applyJournalRow, createJournalReducerState, @@ -36,6 +35,12 @@ import { type AgentJournalEpochReason, type JournalRow } from './journal-row-schema' +import type { + AgentSessionJournalOptions, + JournalAppendResult, + JournalReadSince, + ResolveDispatchInput +} from './journal-store-contracts' import { assertJournalFence, assertJournalWritable, @@ -43,35 +48,12 @@ import { } from './journal-write-guards' export { AgentSessionJournalError } from './journal-write-guards' - -export type AgentSessionJournalOptions = { - identity: AgentSessionJournalIdentity - journalDir: string - limits?: JournalPayloadLimits - compaction?: JournalCompactionPolicy - now?: () => number - mintEpoch?: () => string -} - -export type JournalReadSince = - | { ok: true; rows: JournalRow[]; cursor: AgentJournalCursor } - | { ok: false; reset: AgentJournalResetReason } - -export type ResolveDispatchInput = { - clientMessageId: string - fence: number - /** Set when crash reconciliation, not the live dispatch, settled this. */ - recovered?: true -} & ( - | { state: 'accepted'; providerIdentity: AgentJournalItemIdentity } - | { state: 'rejected' | 'unknown'; reason?: string | null } -) - -export type JournalAppendResult = { - cursor: AgentJournalCursor - itemId: string - revision: number -} +export type { + AgentSessionJournalOptions, + JournalAppendResult, + JournalReadSince, + ResolveDispatchInput +} from './journal-store-contracts' export async function openAgentSessionJournal( options: AgentSessionJournalOptions @@ -129,22 +111,24 @@ export class AgentSessionJournal { } async open(): Promise { - await ensureJournalDir(this.journalDir) - const loaded = await loadJournal(this.journalDir, this.identity.sessionId) - if (!loaded) { - await this.startEpoch('session_created', 0) - return - } - this.state = loaded.state - this.tailRows = loaded.tailRows - this.compactedThrough = loaded.compactedThrough - this.sizeBytes = loaded.sizeBytes - this.readOnly = loaded.readOnly - if (loaded.corrupt && !loaded.readOnly) { - // A reader that observes a gap rolls the epoch rather than rendering a - // partial timeline; clients take the snapshot reload they already handle. - await this.rollEpoch('corruption', this.state.highestFence) - } + await this.serializeWrite(async () => { + await ensureJournalDir(this.journalDir) + const loaded = await loadJournal(this.journalDir, this.identity.sessionId) + if (!loaded) { + await this.startEpoch('session_created', 0) + return + } + this.state = loaded.state + this.tailRows = loaded.tailRows + this.compactedThrough = loaded.compactedThrough + this.sizeBytes = loaded.sizeBytes + this.readOnly = loaded.readOnly + if (loaded.corrupt && !loaded.readOnly) { + // A reader that observes a gap rolls the epoch rather than rendering a + // partial timeline; clients take the snapshot reload they already handle. + await this.startEpoch('corruption', this.state.highestFence) + } + }) } cursor(): AgentJournalCursor { @@ -291,27 +275,34 @@ export class AgentSessionJournal { return pending } - async compact(): Promise { - assertJournalWritable(this.readOnly, this.identity.sessionId) - const result = await compactJournal({ - journalDir: this.journalDir, - state: this.state, - tailRows: this.tailRows, - policy: this.compaction, - now: this.now() + async compact(fence: number): Promise { + await this.serializeWrite(async () => { + assertJournalWritable(this.readOnly, this.identity.sessionId) + assertJournalFence(fence, this.state.highestFence) + this.state.highestFence = Math.max(this.state.highestFence, fence) + const result = await compactJournal({ + journalDir: this.journalDir, + state: this.state, + tailRows: this.tailRows, + policy: this.compaction, + now: this.now() + }) + this.tailRows = result.tailRows + this.compactedThrough = result.compactedThrough + this.state.oldestSequence = result.oldestSequence + this.sizeBytes = result.sizeBytes }) - this.tailRows = result.tailRows - this.compactedThrough = result.compactedThrough - this.state.oldestSequence = result.oldestSequence - this.sizeBytes = this.tailRows.reduce((total, row) => total + journalRowByteLength(row), 0) } /** The escape hatch for corruption, an unreconcilable prefix, a forked handle, * and an unreadable schema. It invalidates every cursor; clients reload. */ async rollEpoch(reason: AgentJournalEpochReason, fence: number): Promise { - assertJournalWritable(this.readOnly, this.identity.sessionId) - await this.startEpoch(reason, fence) - return this.cursor() + return this.serializeWrite(async () => { + assertJournalWritable(this.readOnly, this.identity.sessionId) + assertJournalFence(fence, this.state.highestFence) + await this.startEpoch(reason, fence) + return this.cursor() + }) } private async startEpoch(reason: AgentJournalEpochReason, fence: number): Promise { @@ -327,7 +318,7 @@ export class AgentSessionJournal { this.state = published.state this.tailRows = [published.row] this.compactedThrough = 0 - this.sizeBytes = journalRowByteLength(published.row) + this.sizeBytes = published.sizeBytes } private rowBase( @@ -350,7 +341,7 @@ export class AgentSessionJournal { * callers cannot interleave and mint the same sequence. */ private enqueue(build: (seq: number, ts: number) => JournalRow): Promise { - const run = this.writes.then(async () => { + return this.serializeWrite(async () => { assertJournalWritable(this.readOnly, this.identity.sessionId) const ts = this.now() const row = build(this.state.lastSequence + 1, ts) @@ -362,6 +353,10 @@ export class AgentSessionJournal { this.sizeBytes += journalRowByteLength(row) return row }) + } + + private serializeWrite(operation: () => Promise): Promise { + const run = this.writes.then(operation) this.writes = run.catch(() => undefined) return run }