From 295bbb24a68f68cbeee54b249cd58f4435adcfcd Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 13 Aug 2026 02:13:38 -0700 Subject: [PATCH] fix daemon history reanchor after checkpoint failure --- .../daemon/daemon-durable-history-snapshot.ts | 6 +- src/main/daemon/daemon-pty-adapter.ts | 75 +++++++++--- .../daemon-restore-scrollback-depth.test.ts | 111 ++++++++++++++++++ .../terminal-history-cold-restore-info.ts | 3 + 4 files changed, 176 insertions(+), 19 deletions(-) diff --git a/src/main/daemon/daemon-durable-history-snapshot.ts b/src/main/daemon/daemon-durable-history-snapshot.ts index cad0fcc7e2d..c8d71736607 100644 --- a/src/main/daemon/daemon-durable-history-snapshot.ts +++ b/src/main/daemon/daemon-durable-history-snapshot.ts @@ -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() diff --git a/src/main/daemon/daemon-pty-adapter.ts b/src/main/daemon/daemon-pty-adapter.ts index 7af91fbf370..9569af1249e 100644 --- a/src/main/daemon/daemon-pty-adapter.ts +++ b/src/main/daemon/daemon-pty-adapter.ts @@ -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 | 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() // 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() + // 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() private checkpointTimer: ReturnType | null = null private checkpointInFlight: Promise | null = null private keepHistoryShutdowns = new Set>() @@ -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('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 { + opts: { teardown: boolean; forceLiveSnapshot?: boolean } + ): Promise { const take = await this.client.request('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() diff --git a/src/main/daemon/daemon-restore-scrollback-depth.test.ts b/src/main/daemon/daemon-restore-scrollback-depth.test.ts index c5aa21fefa4..ee79bbcf035 100644 --- a/src/main/daemon/daemon-restore-scrollback-depth.test.ts +++ b/src/main/daemon/daemon-restore-scrollback-depth.test.ts @@ -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 + } + 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 } + } + 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) + }) }) }) diff --git a/src/main/daemon/terminal-history-cold-restore-info.ts b/src/main/daemon/terminal-history-cold-restore-info.ts index fe71909f3b4..a3a334f8045 100644 --- a/src/main/daemon/terminal-history-cold-restore-info.ts +++ b/src/main/daemon/terminal-history-cold-restore-info.ts @@ -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 }