fix daemon history reanchor after checkpoint failure

This commit is contained in:
Brennan Benson
2026-08-13 02:13:38 -07:00
parent 77bd4f8302
commit 295bbb24a6
4 changed files with 176 additions and 19 deletions
@@ -32,7 +32,8 @@ export function terminalSnapshotFromColdRestore(
modes: info.modes,
cols: info.cols,
rows: info.rows,
scrollbackLines: countAnsiRows(info.scrollbackAnsi),
scrollbackLines:
info.scrollbackLines ?? Math.max(0, countAnsiRows(info.scrollbackAnsi) - info.rows),
...(info.lastTitle ? { lastTitle: info.lastTitle } : {}),
...(opts?.outputSequence !== undefined ? { outputSequence: opts.outputSequence } : {})
}
@@ -103,7 +104,8 @@ export async function buildDurableCheckpointSnapshot(opts: {
? { frameRestoreAnsi: opts.liveSnapshot.frameRestoreAnsi }
: {})
}
} catch {
} catch (error) {
console.warn('[history] durable snapshot rebuild failed:', error)
return opts.liveSnapshot
} finally {
emulator.dispose()
+58 -17
View File
@@ -70,6 +70,7 @@ import { resolveSafePtyDefaultCwd } from '../providers/pty-default-cwd'
import { PtyWriteUnavailableError } from '../providers/pty-write-unavailable-error'
import { ColdRestorePayloadCache, type ColdRestorePayload } from './cold-restore-payload-cache'
import { buildDurableCheckpointSnapshot } from './daemon-durable-history-snapshot'
import { DAEMON_RESTORE_SCROLLBACK_ROWS } from './daemon-restore-scrollback-depth'
import { DAEMON_SESSION_SCROLLBACK_ROWS } from './daemon-session-scrollback-window'
import { PtyProcessListAdmission } from '../providers/pty-process-list-admission'
import {
@@ -102,6 +103,11 @@ type HistoryRecoveryContext = {
identityChanged: boolean
}
type SnapshotCheckpointResult = {
checkpoint: HistoryCheckpointResult
snapshot: NonNullable<TakePendingOutputResult['snapshot']> | null
}
// Why take-and-clear together: every consuming branch must reset the field, so pairing them stops one from forgetting.
function takeRecoveryFreeze(
historyRecovery: HistoryRecoveryContext,
@@ -226,6 +232,8 @@ export class DaemonPtyAdapter implements IPtyProvider {
private dirtySessionVersions = new Map<string, number>()
// Why: a cold-restored session is a fresh shell atop a pre-crash log; incremental appends would be rejected on restore, so the first tick re-anchors with a full snapshot.
private sessionsNeedingFullCheckpoint = new Set<string>()
// Why: overflow or an unpersisted drain breaks disk-to-queue continuity, so only the daemon emulator can safely re-anchor the next checkpoint.
private sessionsNeedingLiveCheckpoint = new Set<string>()
private checkpointTimer: ReturnType<typeof setTimeout> | null = null
private checkpointInFlight: Promise<void> | null = null
private keepHistoryShutdowns = new Set<Promise<void>>()
@@ -755,6 +763,7 @@ export class DaemonPtyAdapter implements IPtyProvider {
})
if (this.historyManager.hasWriter(sessionId)) {
this.sessionsNeedingFullCheckpoint.add(sessionId)
this.sessionsNeedingLiveCheckpoint.add(sessionId)
this.lastFullCheckpointAt.delete(sessionId)
}
} else if (canReanchorHistory) {
@@ -816,6 +825,7 @@ export class DaemonPtyAdapter implements IPtyProvider {
if (!wasAlreadyManaged) {
// Why: a previous adapter may have drained records it never persisted, so appending would leave a seq gap the reader rejects; force a full snapshot to re-anchor.
this.sessionsNeedingFullCheckpoint.add(sessionId)
this.sessionsNeedingLiveCheckpoint.add(sessionId)
this.lastFullCheckpointAt.delete(sessionId)
}
}
@@ -1130,6 +1140,7 @@ export class DaemonPtyAdapter implements IPtyProvider {
}
// Why: the !keepHistory path takes no final checkpoint, so clear sessionsNeedingFullCheckpoint here or it stays stranded (no-op under keepHistory).
this.sessionsNeedingFullCheckpoint.delete(id)
this.sessionsNeedingLiveCheckpoint.delete(id)
this.lastFullCheckpointAt.delete(id)
this.stopCheckpointTimerIfIdle()
this.initialCwds.delete(id)
@@ -1282,12 +1293,23 @@ export class DaemonPtyAdapter implements IPtyProvider {
try {
// Why exclusive compact: an independent take/append races the 5s tick, can
// seq-gap the log, and would remount a stale checkpoint over the live window.
const checkpointResult: { value?: SnapshotCheckpointResult } = {}
await this.runExclusiveCheckpoint(async () => {
const result = await this.takeSnapshotAndCheckpoint(sessionId, { teardown: false })
if (result === 'committed') {
checkpointResult.value = await this.takeSnapshotAndCheckpoint(sessionId, {
teardown: false,
forceLiveSnapshot: this.sessionsNeedingLiveCheckpoint.has(sessionId)
})
if (checkpointResult.value.checkpoint === 'committed') {
this.sessionsNeedingFullCheckpoint.delete(sessionId)
}
})
const checkpoint = checkpointResult.value
if (checkpoint?.checkpoint !== 'committed' || !checkpoint.snapshot) {
return checkpoint?.snapshot ?? liveSnapshot
}
if (scrollbackRows === undefined || scrollbackRows >= DAEMON_RESTORE_SCROLLBACK_ROWS) {
return checkpoint.snapshot
}
const restoreInfo = await this.historyReader.detectColdRestore(sessionId, {
ignoreCleanEnd: true,
wslDistro: this.wslDistrosBySessionId.get(sessionId)
@@ -1296,11 +1318,12 @@ export class DaemonPtyAdapter implements IPtyProvider {
return liveSnapshot
}
return await buildDurableCheckpointSnapshot({
liveSnapshot,
liveSnapshot: checkpoint.snapshot,
restoreInfo,
scrollbackRows
})
} catch {
} catch (error) {
console.warn('[history] durable snapshot overlay failed:', sessionId, error)
return liveSnapshot
}
}
@@ -1472,6 +1495,7 @@ export class DaemonPtyAdapter implements IPtyProvider {
}
if (historyManager.hasWriter(session.sessionId)) {
this.sessionsNeedingFullCheckpoint.add(session.sessionId)
this.sessionsNeedingLiveCheckpoint.add(session.sessionId)
this.lastFullCheckpointAt.delete(session.sessionId)
this.markSessionDirty(session.sessionId)
}
@@ -1591,6 +1615,7 @@ export class DaemonPtyAdapter implements IPtyProvider {
this.dirtySessionVersions.clear()
this.lastFullCheckpointAt.clear()
this.sessionsNeedingFullCheckpoint.clear()
this.sessionsNeedingLiveCheckpoint.clear()
this.pausedProducerSessionIds.clear()
this.producerResumesOwedOnReconnect.clear()
this.stopCheckpointTimer()
@@ -1698,6 +1723,8 @@ export class DaemonPtyAdapter implements IPtyProvider {
this.stopCheckpointTimer()
this.dirtySessionVersions.clear()
this.lastFullCheckpointAt.clear()
this.sessionsNeedingFullCheckpoint.clear()
this.sessionsNeedingLiveCheckpoint.clear()
this.coldRestoreCache.clear()
this.wslDistrosBySessionId.clear()
this.pausedProducerSessionIds.clear()
@@ -2092,13 +2119,15 @@ export class DaemonPtyAdapter implements IPtyProvider {
// Why take-with-snapshot not plain getSnapshot: it clears pending records in the same turn as the serialize,
// so a warm reattach won't re-append records the checkpoint already contains (double-replay on cold restore).
const checkpoint = await this.takeSnapshotAndCheckpoint(sessionId, {
teardown: opts.teardown
teardown: opts.teardown,
forceLiveSnapshot: this.sessionsNeedingLiveCheckpoint.has(sessionId)
})
if (checkpoint === 'retryable') {
if (checkpoint.checkpoint !== 'committed') {
this.sessionsNeedingFullCheckpoint.add(sessionId)
return 'deferred'
}
this.sessionsNeedingFullCheckpoint.delete(sessionId)
this.sessionsNeedingLiveCheckpoint.delete(sessionId)
return 'done'
}
const take = await this.client.request<TakePendingOutputResult | null>('takePendingOutput', {
@@ -2113,8 +2142,11 @@ export class DaemonPtyAdapter implements IPtyProvider {
this.sessionsNeedingFullCheckpoint.add(sessionId)
return 'deferred'
}
const checkpoint = await this.takeSnapshotAndCheckpoint(sessionId, { teardown: false })
if (checkpoint === 'retryable') {
const checkpoint = await this.takeSnapshotAndCheckpoint(sessionId, {
teardown: false,
forceLiveSnapshot: true
})
if (checkpoint.checkpoint !== 'committed') {
this.sessionsNeedingFullCheckpoint.add(sessionId)
return 'deferred'
}
@@ -2137,8 +2169,11 @@ export class DaemonPtyAdapter implements IPtyProvider {
this.sessionsNeedingFullCheckpoint.add(sessionId)
return 'deferred'
}
const checkpoint = await this.takeSnapshotAndCheckpoint(sessionId, { teardown: false })
if (checkpoint === 'retryable') {
const checkpoint = await this.takeSnapshotAndCheckpoint(sessionId, {
teardown: false,
forceLiveSnapshot: true
})
if (checkpoint.checkpoint !== 'committed') {
this.sessionsNeedingFullCheckpoint.add(sessionId)
return 'deferred'
}
@@ -2148,8 +2183,8 @@ export class DaemonPtyAdapter implements IPtyProvider {
private async takeSnapshotAndCheckpoint(
sessionId: string,
opts: { teardown: boolean }
): Promise<HistoryCheckpointResult> {
opts: { teardown: boolean; forceLiveSnapshot?: boolean }
): Promise<SnapshotCheckpointResult> {
const take = await this.client.request<TakePendingOutputResult | null>('takePendingOutput', {
sessionId,
includeSnapshot: true,
@@ -2160,7 +2195,7 @@ export class DaemonPtyAdapter implements IPtyProvider {
// queue on includeSnapshot but omits the field. Treating absence as []
// would compact stale disk history and reset the log.
const snapshot =
take.drainedRecords === undefined
take.drainedRecords === undefined || opts.forceLiveSnapshot === true || take.overflowed
? take.snapshot
: await this.buildDurableHistorySnapshot(sessionId, take.snapshot, [
...take.drainedRecords,
@@ -2171,16 +2206,20 @@ export class DaemonPtyAdapter implements IPtyProvider {
// Why take.records is dropped, not appended: the pending output this take drained went into the snapshot that
// failed to land, so appending the held tail at the next contiguous seq would splice it over that hole and
// defeat the log's seq-gap detection. A stale prefix beats an undetectable hole.
return checkpoint
this.sessionsNeedingFullCheckpoint.add(sessionId)
this.sessionsNeedingLiveCheckpoint.add(sessionId)
this.markSessionDirty(sessionId)
return { checkpoint, snapshot: take.snapshot }
}
this.lastFullCheckpointAt.set(sessionId, Date.now())
if (take.records.length > 0 && snapshot === take.snapshot) {
// Why: live-window fallback still lacks held parser-state bytes; keep them as a post-checkpoint log tail.
await this.historyManager.appendIncrements(sessionId, take.seq, take.records)
}
return 'committed'
this.sessionsNeedingLiveCheckpoint.delete(sessionId)
return { checkpoint: 'committed', snapshot }
}
return 'unavailable'
return { checkpoint: 'unavailable', snapshot: take?.snapshot ?? null }
}
private async buildDurableHistorySnapshot(
@@ -2201,7 +2240,8 @@ export class DaemonPtyAdapter implements IPtyProvider {
restoreInfo,
pendingRecords
})
} catch {
} catch (error) {
console.warn('[history] durable history rebuild failed:', sessionId, error)
return liveSnapshot
}
}
@@ -2491,6 +2531,7 @@ export class DaemonPtyAdapter implements IPtyProvider {
}
// Why: an exited session can't be checkpointed again; clearing its pending-full flag prevents a permanent leak.
this.sessionsNeedingFullCheckpoint.delete(event.sessionId)
this.sessionsNeedingLiveCheckpoint.delete(event.sessionId)
// Why: a reused sessionId (renderer respawns a persisted ptyId) must not inherit the dead session's snapshot cooldown.
this.lastFullCheckpointAt.delete(event.sessionId)
this.stopCheckpointTimerIfIdle()
@@ -20,6 +20,7 @@ import type { SubprocessHandle } from './session'
const PREVIOUSLY_RECOVERABLE_LINE = 'LINE_01000'
const OLDEST_WRITTEN_LINE = 'LINE_00001'
const NEWEST_WRITTEN_LINE = `LINE_${String(DESKTOP_TERMINAL_SCROLLBACK_ROWS_DEFAULT).padStart(5, '0')}`
const FRESH_AFTER_CHECKPOINT = 'FRESH_AFTER_CHECKPOINT'
function numberedOutput(lineCount: number): string {
let output = ''
@@ -192,6 +193,7 @@ describe('STA-4091 previously recoverable restore depth', () => {
})
expect(durable.outputSequence).toBe(9)
expect(durable.scrollbackAnsi).toBe('')
expect(durable.scrollbackLines).toBe(restoreInfo?.scrollbackLines)
expect(snapshotText(durable)).toContain(PREVIOUSLY_RECOVERABLE_LINE)
} finally {
live.dispose()
@@ -351,5 +353,114 @@ describe('STA-4091 previously recoverable restore depth', () => {
expect(text).toContain(NEWEST_WRITTEN_LINE)
expect(text).not.toContain(OLDEST_WRITTEN_LINE)
})
it('reanchors from the live window after pending output overflows', async () => {
const { id } = await adapter.spawn({
cols: 80,
rows: 24,
sessionId: 'overflow-reanchor',
cwd: '/tmp'
})
lastSubprocess.emitData(numberedOutput(DESKTOP_TERMINAL_SCROLLBACK_ROWS_DEFAULT))
await adapter.getBufferSnapshot(id)
const overflowLine = `BULK_${'x'.repeat(100)}\r\n`
lastSubprocess.emitData(overflowLine.repeat(20_000))
lastSubprocess.emitData(`${FRESH_AFTER_CHECKPOINT}\r\n`)
const snapshot = await adapter.getBufferSnapshot(id)
const text = `${snapshot?.scrollbackAnsi ?? ''}${snapshot?.data ?? ''}`
expect(text).toContain(FRESH_AFTER_CHECKPOINT)
const restore = await new HistoryReader(historyDir).detectColdRestore(id, {
ignoreCleanEnd: true
})
expect(snapshotText(restore ?? {})).toContain(FRESH_AFTER_CHECKPOINT)
})
it.each(['retryable', 'unavailable'] as const)(
'returns the current live snapshot when a durable checkpoint is %s',
async (checkpointResult) => {
const { id } = await adapter.spawn({
cols: 80,
rows: 24,
sessionId: `checkpoint-${checkpointResult}`,
cwd: '/tmp'
})
lastSubprocess.emitData(numberedOutput(DESKTOP_TERMINAL_SCROLLBACK_ROWS_DEFAULT))
await adapter.getBufferSnapshot(id)
lastSubprocess.emitData(`${FRESH_AFTER_CHECKPOINT}\r\n`)
const internals = adapter as unknown as {
historyManager: HistoryManager
sessionsNeedingLiveCheckpoint: Set<string>
}
const checkpoint = vi
.spyOn(internals.historyManager, 'checkpoint')
.mockResolvedValueOnce(checkpointResult)
const snapshot = await adapter.getBufferSnapshot(id)
const text = `${snapshot?.scrollbackAnsi ?? ''}${snapshot?.data ?? ''}`
expect(text).toContain(FRESH_AFTER_CHECKPOINT)
expect(internals.sessionsNeedingLiveCheckpoint.has(id)).toBe(true)
const recovered = await adapter.getBufferSnapshot(id)
expect(`${recovered?.scrollbackAnsi ?? ''}${recovered?.data ?? ''}`).toContain(
FRESH_AFTER_CHECKPOINT
)
const restore = await new HistoryReader(historyDir).detectColdRestore(id, {
ignoreCleanEnd: true
})
expect(snapshotText(restore ?? {})).toContain(FRESH_AFTER_CHECKPOINT)
expect(checkpoint.mock.calls.at(-1)?.[1].scrollbackLines).toBe(
DAEMON_SESSION_SCROLLBACK_ROWS
)
expect(internals.sessionsNeedingLiveCheckpoint.has(id)).toBe(false)
checkpoint.mockResolvedValueOnce(checkpointResult)
const reattach = await adapter.spawn({
cols: 80,
rows: 24,
sessionId: id,
cwd: '/tmp'
})
expect(reattach.isReattach).toBe(true)
expect(reattach.snapshot).toContain(FRESH_AFTER_CHECKPOINT)
}
)
it('uses the post-drain sequence for output produced during an overlay', async () => {
const { id } = await adapter.spawn({
cols: 80,
rows: 24,
sessionId: 'overlay-sequence',
cwd: '/tmp'
})
lastSubprocess.emitData(numberedOutput(DESKTOP_TERMINAL_SCROLLBACK_ROWS_DEFAULT))
await adapter.getBufferSnapshot(id)
const internals = adapter as unknown as {
client: { request: (method: string, params?: unknown) => Promise<unknown> }
}
const request = internals.client.request.bind(internals.client)
let injected = false
vi.spyOn(internals.client, 'request').mockImplementation(async (method, params) => {
if (
method === 'takePendingOutput' &&
(params as { includeSnapshot?: boolean } | undefined)?.includeSnapshot &&
!injected
) {
injected = true
lastSubprocess.emitData(`${FRESH_AFTER_CHECKPOINT}\r\n`)
}
return request(method, params)
})
const overlaid = await adapter.getBufferSnapshot(id)
const current = await adapter.getBufferSnapshot(id, { scrollbackRows: 24 })
expect(`${overlaid?.scrollbackAnsi ?? ''}${overlaid?.data ?? ''}`).toContain(
FRESH_AFTER_CHECKPOINT
)
expect(overlaid?.seq).toBe(current?.seq)
})
})
})
@@ -10,6 +10,7 @@ export type ColdRestoreInfo = {
cwd: string
cols: number
rows: number
scrollbackLines?: number
modes: TerminalModes
pendingEscapeTailAnsi?: string
lastTitle?: string
@@ -22,6 +23,7 @@ type RestoredSnapshot = {
rehydrateSequences: string
cols: number
rows: number
scrollbackLines?: number
modes: TerminalModes
pendingEscapeTailAnsi?: string
lastTitle?: string
@@ -43,6 +45,7 @@ export function coldRestoreInfoFromSnapshot(
cwd: cwd ?? meta.cwd,
cols: snapshot.cols,
rows: snapshot.rows,
scrollbackLines: snapshot.scrollbackLines,
modes: snapshot.modes,
...(snapshot.pendingEscapeTailAnsi
? { pendingEscapeTailAnsi: snapshot.pendingEscapeTailAnsi }