fix(native-chat): preserve journal durability across compaction

This commit is contained in:
Brennan Benson
2026-08-07 14:06:55 -07:00
parent a0c960097c
commit d4bcc028a3
8 changed files with 242 additions and 82 deletions
@@ -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)
}
}
@@ -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))
@@ -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 }
}
@@ -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<void> {
await mkdir(journalDir, { recursive: true })
}
export async function readJournalSnapshotFile(
journalDir: string
): Promise<JournalSnapshotFile | null> {
): Promise<JournalSnapshotReadResult> {
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<JournalSnapshotFile>
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<JournalReadResult> {
let raw: string
try {
@@ -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<JournalLoad | null> {
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
}
@@ -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
}
@@ -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<string, unknown>
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 })
@@ -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<void> {
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<void> {
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<void> {
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<AgentJournalCursor> {
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<void> {
@@ -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<JournalRow> {
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<T>(operation: () => Promise<T>): Promise<T> {
const run = this.writes.then(operation)
this.writes = run.catch(() => undefined)
return run
}