From 1508a308e5827fdd7ac6a73ba2a47a4d4221027a Mon Sep 17 00:00:00 2001 From: Neil <4138956+nwparker@users.noreply.github.com> Date: Thu, 15 Oct 2026 17:21:20 -0700 Subject: [PATCH] fix(remote-terminal): keep live output flowing during resync and retry an unanswered one (#29236) * fix(remote-terminal): keep live output flowing during resync and retry an unanswered one Co-authored-by: Claude * fix(remote-terminal): escalate to stall recovery when a resync is never answered Co-authored-by: Claude --------- Co-authored-by: Claude --- ...mote-runtime-terminal-binary-controller.ts | 13 +- ...emote-runtime-terminal-binary-snapshots.ts | 6 + ...runtime-terminal-frame-drop-resync.test.ts | 175 +++++++++++++++++- ...ime-terminal-multiplexer-implementation.ts | 1 + ...mote-runtime-terminal-multiplexer-types.ts | 2 + ...te-runtime-terminal-snapshot-controller.ts | 34 +++- .../remote-runtime-terminal-snapshot-state.ts | 2 + 7 files changed, 209 insertions(+), 24 deletions(-) diff --git a/src/renderer/src/runtime/remote-runtime-terminal-binary-controller.ts b/src/renderer/src/runtime/remote-runtime-terminal-binary-controller.ts index 906c2d82451..14d5b7d3d91 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-binary-controller.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-binary-controller.ts @@ -97,11 +97,6 @@ export abstract class RemoteRuntimeTerminalBinaryController extends RemoteRuntim frame.opcode === TerminalStreamOpcode.OutputSpan && typeof span?.rawLength === 'number' ? span.rawLength : data.length - // Why: a resync snapshot is authoritative; discard live output while - // it is in flight, but still return transport credit in finally. - if (stream.resyncInFlight) { - return - } const seq = typeof frame.seq === 'number' && frame.seq > 0 ? frame.seq : undefined // Why: older servers replay snapshot-covered buffered chunks after a // requested recovery; rendering them would duplicate the recovered tail. @@ -112,11 +107,15 @@ export abstract class RemoteRuntimeTerminalBinaryController extends RemoteRuntim ) { return } - if (this.detectOutputGap(stream, seq, rawLength)) { + const resyncInFlight = stream.resyncInFlight + // Why: keep painting while recovery is in flight — gating froze the pane for + // the whole round trip (#18488); the snapshot replaces the screen when it lands. + // expectedSeq stays at the gap so a recovery that never lands can still be retried. + if (!resyncInFlight && this.detectOutputGap(stream, seq, rawLength)) { this.requestResyncSnapshot(stream) return } - if (typeof seq === 'number') { + if (typeof seq === 'number' && !resyncInFlight) { stream.expectedSeq = seq stream.commandProbeBaselineSeq = undefined } diff --git a/src/renderer/src/runtime/remote-runtime-terminal-binary-snapshots.ts b/src/renderer/src/runtime/remote-runtime-terminal-binary-snapshots.ts index 58f9559ed04..adece51d459 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-binary-snapshots.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-binary-snapshots.ts @@ -38,6 +38,9 @@ export abstract class RemoteRuntimeTerminalBinarySnapshots extends RemoteRuntime : stream.initialSnapshotReceived ? 'recovery' : 'initial' + if (stream.snapshotTarget === 'recovery') { + this.noteResyncSnapshotProgress(stream) + } return } if (frame.opcode === TerminalStreamOpcode.SnapshotChunk) { @@ -53,6 +56,9 @@ export abstract class RemoteRuntimeTerminalBinarySnapshots extends RemoteRuntime return } stream.snapshotChunks.push(frame.payload) + if (stream.snapshotTarget === 'recovery') { + this.noteResyncSnapshotProgress(stream) + } return } if (frame.opcode === TerminalStreamOpcode.SnapshotEnd) { diff --git a/src/renderer/src/runtime/remote-runtime-terminal-frame-drop-resync.test.ts b/src/renderer/src/runtime/remote-runtime-terminal-frame-drop-resync.test.ts index 3d71ef1b5fe..7ada7a3b283 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-frame-drop-resync.test.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-frame-drop-resync.test.ts @@ -43,9 +43,11 @@ class FakeMultiplexServer { truncateNextRecoverySnapshot = false dropNextRecoverySnapshotEnd = false holdNextRecoverySnapshot = false + streamNextRecoverySnapshot = false snapshotRequests: (number | undefined)[] = [] snapshotRequestRows: (number | undefined)[] = [] private heldManualRequestId: number | null = null + private heldRecoverySnapshot = false private snapshotData = 'INITIAL' constructor( @@ -82,6 +84,17 @@ class FakeMultiplexServer { if (typeof payload?.requestId !== 'number' && this.holdNextRecoverySnapshot) { // The reply's binary frames were all dropped under backpressure. this.holdNextRecoverySnapshot = false + this.heldRecoverySnapshot = true + return + } + if (typeof payload?.requestId !== 'number' && this.streamNextRecoverySnapshot) { + // A slow link: the reply starts now and its chunks trickle in later. + this.streamNextRecoverySnapshot = false + this.send( + TerminalStreamOpcode.SnapshotStart, + encodeTerminalStreamJson({ cols: 80, rows: 24, seq: this.cursorUnits }), + 0 + ) return } if (typeof payload?.requestId !== 'number' && this.truncateNextRecoverySnapshot) { @@ -172,6 +185,22 @@ class FakeMultiplexServer { this.snapshotData = 'MANUAL' this.sendSnapshot(requestId) } + + streamRecoveryChunk(text: string): void { + this.send(TerminalStreamOpcode.SnapshotChunk, encodeTerminalStreamText(text), 0) + } + + endStreamedRecovery(): void { + this.send(TerminalStreamOpcode.SnapshotEnd, new Uint8Array(), 0) + } + + flushHeldRecoverySnapshot(): void { + if (!this.heldRecoverySnapshot) { + throw new Error('No recovery snapshot is held') + } + this.heldRecoverySnapshot = false + this.sendSnapshot() + } } describe('remote terminal frame-drop resync', () => { @@ -242,7 +271,8 @@ describe('remote terminal frame-drop resync', () => { await Promise.resolve() await Promise.resolve() - // The corrupt tail ('ccc', which followed a gap) is NOT rendered as live data. + // The immediate snapshot covers the corrupt tail ('ccc'), so it is not + // rendered separately or duplicated after the recovery reset. expect(data).toEqual(['aaa']) expect(server.droppedFrames).toBe(1) // Instead, a fresh authoritative snapshot recovers the terminal. @@ -255,7 +285,7 @@ describe('remote terminal frame-drop resync', () => { expect(data).toEqual(['aaa', 'ddd']) }) - it('retries a truncated recovery on a backoff without accepting output across the gap', async () => { + it('retries a truncated recovery on a backoff while keeping live output flowing', async () => { vi.useFakeTimers() try { const { data, snapshots } = await subscribeClient() @@ -265,8 +295,8 @@ describe('remote terminal frame-drop resync', () => { server.dropNextOutput = true server.output('bbb') server.output('ccc') - // The gate stays shut across the backoff: the post-gap tail is corrupt, - // and retrying once per chunk would stampede a flooded server. + // Recovery retries are backoff-limited, but the live path stays open; + // output must not freeze while a jittery host serializes the snapshot. server.output('ddd') expect(server.snapshotRequests).toEqual([undefined]) @@ -276,12 +306,59 @@ describe('remote terminal frame-drop resync', () => { server.output('eee') expect(snapshots).toEqual(['INITIAL', '\x1b[?2026l\x1b[2J\x1b[3J\x1b[HRECOVERED']) - expect(data).toEqual(['aaa', 'eee']) + expect(data).toEqual(['aaa', 'ddd', 'eee']) } finally { vi.useRealTimers() } }) + it('keeps painting live output while a recovery snapshot is delayed', async () => { + const { data, snapshots, stream } = await subscribeClient() + try { + server.holdNextRecoverySnapshot = true + + server.output('aaa') + server.dropNextOutput = true + server.output('bbb') + server.output('ccc') // exposes the gap and starts the delayed recovery + server.output('ddd') + server.output('eee') + + // A slow snapshot must not become a renderer backpressure gate. + expect(data).toEqual(['aaa', 'ddd', 'eee']) + expect(snapshots).toEqual(['INITIAL']) + } finally { + stream.close() + } + }) + + it('reconciles painted output when a delayed recovery snapshot arrives', async () => { + const { data, snapshots, stream } = await subscribeClient() + try { + server.holdNextRecoverySnapshot = true + + server.output('aaa') + server.dropNextOutput = true + server.output('bbb') + server.output('ccc') + server.output('ddd') + server.output('eee') + + expect(data).toEqual(['aaa', 'ddd', 'eee']) + expect(snapshots).toEqual(['INITIAL']) + + server.flushHeldRecoverySnapshot() + expect(snapshots).toEqual(['INITIAL', '\x1b[?2026l\x1b[2J\x1b[3J\x1b[HRECOVERED']) + + // A replay covered by the authoritative snapshot must not duplicate. + server.replaySnapshotCoveredOutput('eee') + server.output('fff') + expect(data).toEqual(['aaa', 'ddd', 'eee', 'fff']) + } finally { + stream.close() + } + }) + it('re-opens the live path when only the JSON error event for a resync survives', async () => { const { data, snapshots, stream } = await subscribeClient() server.holdNextRecoverySnapshot = true @@ -332,7 +409,7 @@ describe('remote terminal frame-drop resync', () => { expect(data).toEqual(['aaa', 'ddd']) }) - it('times out a dropped recovery end and retries on the next sequence gap', async () => { + it('retries a recovery whose end was dropped without waiting for more output', async () => { vi.useFakeTimers() try { const { data, snapshots } = await subscribeClient() @@ -342,13 +419,95 @@ describe('remote terminal frame-drop resync', () => { server.dropNextOutput = true server.output('bbb') server.output('ccc') + server.output('ddd') await vi.advanceTimersByTimeAsync(10_000) + expect(server.snapshotRequests).toEqual([undefined]) + + // #18488: the timeout itself retries; an idle pane still heals. + await vi.advanceTimersByTimeAsync(500) + expect(server.snapshotRequests).toEqual([undefined, undefined]) + expect(snapshots).toEqual(['INITIAL', '\x1b[?2026l\x1b[2J\x1b[3J\x1b[HRECOVERED']) + + server.output('eee') + expect(data).toEqual(['aaa', 'ddd', 'eee']) + } finally { + vi.useRealTimers() + } + }) + + it('does not re-gate on every timeout while recovery cannot complete', async () => { + vi.useFakeTimers() + try { + const { data } = await subscribeClient() + server.holdNextRecoverySnapshot = true + + server.output('aaa') + server.dropNextOutput = true + server.output('bbb') + server.output('ccc') + await vi.advanceTimersByTimeAsync(10_000) + server.holdNextRecoverySnapshot = true + // Before #18488 the frame after each timeout re-tripped the stale gap and was dropped. server.output('ddd') server.output('eee') + expect(data).toEqual(['aaa', 'ddd', 'eee']) + } finally { + vi.useRealTimers() + } + }) + + it('escalates to stall recovery when the host never answers a recovery', async () => { + vi.useFakeTimers() + try { + const onTransportClose = vi.fn() + await getRemoteRuntimeTerminalMultiplexer('env-1').subscribeTerminal({ + terminal: 'terminal-1', + client: { id: 'desktop-1', type: 'desktop' }, + callbacks: { onData: vi.fn(), onSnapshot: vi.fn(), onTransportClose } + }) + await Promise.resolve() + await Promise.resolve() + server.holdNextRecoverySnapshot = true + server.output('aaa') + server.dropNextOutput = true + server.output('bbb') + server.output('ccc') + await vi.advanceTimersByTimeAsync(10_000) + server.holdNextRecoverySnapshot = true + await vi.advanceTimersByTimeAsync(500) expect(server.snapshotRequests).toEqual([undefined, undefined]) - expect(snapshots).toEqual(['INITIAL', '\x1b[?2026l\x1b[2J\x1b[3J\x1b[HRECOVERED']) - expect(data).toEqual(['aaa', 'eee']) + expect(onTransportClose).not.toHaveBeenCalled() + + await vi.advanceTimersByTimeAsync(10_000) + expect(onTransportClose).toHaveBeenCalledWith({ recoverable: true }) + } finally { + vi.useRealTimers() + } + }) + + it('does not re-request a recovery that is still streaming on a slow link', async () => { + vi.useFakeTimers() + try { + const { data, snapshots } = await subscribeClient() + server.streamNextRecoverySnapshot = true + + server.output('aaa') + server.dropNextOutput = true + server.output('bbb') + server.output('ccc') + await vi.advanceTimersByTimeAsync(6_000) + server.streamRecoveryChunk('SLOW-') + await vi.advanceTimersByTimeAsync(6_000) + server.streamRecoveryChunk('RECOVERY') + await vi.advanceTimersByTimeAsync(6_000) + server.endStreamedRecovery() + await vi.advanceTimersByTimeAsync(10_000) + + expect(server.snapshotRequests).toEqual([undefined]) + expect(snapshots).toEqual(['INITIAL', '\x1b[?2026l\x1b[2J\x1b[3J\x1b[HSLOW-RECOVERY']) + server.output('ddd') + expect(data).toEqual(['aaa', 'ddd']) } finally { vi.useRealTimers() } diff --git a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-implementation.ts b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-implementation.ts index 6e37cf12f8c..8ad2f2759d9 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-implementation.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-implementation.ts @@ -56,6 +56,7 @@ export class RemoteRuntimeTerminalMultiplexer extends RemoteRuntimeTerminalBinar resyncPendingSend: false, resyncTimer: null, resyncAttempts: 0, + resyncUnansweredTimeouts: 0, capacityRejected: false, watchdog: createRemoteTerminalStreamWatchdog((stall) => { if (e2eDisableRemoteTerminalStallRecovery) { diff --git a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-types.ts b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-types.ts index eb366169401..9961fafc844 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-types.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-types.ts @@ -193,6 +193,8 @@ export type RemoteRuntimeMultiplexedTerminalState = { resyncPendingSend: boolean resyncTimer: ReturnType | null resyncAttempts: number + /** Consecutive resync timeouts with no recovery frame at all. */ + resyncUnansweredTimeouts: number capacityRejected: boolean watchdog: RemoteTerminalStreamWatchdog } diff --git a/src/renderer/src/runtime/remote-runtime-terminal-snapshot-controller.ts b/src/renderer/src/runtime/remote-runtime-terminal-snapshot-controller.ts index 7ec782cfaaa..3fe96d3b98b 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-snapshot-controller.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-snapshot-controller.ts @@ -6,6 +6,7 @@ import { TERMINAL_MULTIPLEX_RECOVERY_SCROLLBACK_ROWS } from '../../../shared/ter import { recordRendererCrashBreadcrumb } from '@/lib/crash-breadcrumb-recorder' import { RemoteRuntimeTerminalMultiplexerBase } from './remote-runtime-terminal-multiplexer-base' import { + REMOTE_TERMINAL_RESYNC_MAX_UNANSWERED_TIMEOUTS, REMOTE_TERMINAL_RESYNC_RETRY_BASE_MS, REMOTE_TERMINAL_RESYNC_RETRY_MAX_MS, REMOTE_TERMINAL_RESYNC_TIMEOUT_MS, @@ -38,10 +39,9 @@ export abstract class RemoteRuntimeTerminalSnapshotController extends RemoteRunt return startSeq > stream.expectedSeq } - // Why: on a detected gap, discard the corrupt tail and pull a fresh - // authoritative snapshot. The request carries no requestId so the server - // reply renders through the initial-snapshot path (full reset), self-healing - // without surfacing an error to the user. + // Why: on a detected gap, pull a fresh authoritative snapshot. The request + // carries no requestId so the server reply uses the recovery path (full + // reset), self-healing without surfacing an error to the user. protected requestResyncSnapshot(stream: RemoteRuntimeMultiplexedTerminalState): void { if (stream.resyncInFlight) { return @@ -51,7 +51,7 @@ export abstract class RemoteRuntimeTerminalSnapshotController extends RemoteRunt // Why: snapshot frame groups are not multiplexed; wait for the manual // snapshot to finish so its response cannot be mistaken for recovery. // Arm the watchdog now so a dispatch path that consumes the pending - // request without re-dispatching cannot hold the gate shut forever. + // request without re-dispatching cannot leave recovery pending forever. stream.resyncPendingSend = true this.startResyncTimer(stream) return @@ -82,8 +82,8 @@ export abstract class RemoteRuntimeTerminalSnapshotController extends RemoteRunt } } - // Why: keep the gate shut across the backoff — the post-gap tail is corrupt - // either way — and heal even if the flood ends with no further output. + // Why: keep recovery pending across the backoff and heal even if the flood + // ends with no further output. protected scheduleResyncRetry(stream: RemoteRuntimeMultiplexedTerminalState): void { stream.resyncAttempts += 1 const delay = Math.min( @@ -113,6 +113,15 @@ export abstract class RemoteRuntimeTerminalSnapshotController extends RemoteRunt stream.resyncTimer = timer } + // Why: a large recovery on a slow link can take longer than the timeout to + // stream; expiring mid-transfer would request a second one and repaint twice. + protected noteResyncSnapshotProgress(stream: RemoteRuntimeMultiplexedTerminalState): void { + stream.resyncUnansweredTimeouts = 0 + if (stream.resyncInFlight && !stream.resyncPendingSend) { + this.startResyncTimer(stream) + } + } + private startResyncTimer(stream: RemoteRuntimeMultiplexedTerminalState): void { clearResyncTimer(stream) const timer = setTimeout(() => { @@ -124,8 +133,15 @@ export abstract class RemoteRuntimeTerminalSnapshotController extends RemoteRunt return } stream.resyncTimer = null - stream.resyncInFlight = false - stream.resyncPendingSend = false + stream.resyncUnansweredTimeouts += 1 + if (stream.resyncUnansweredTimeouts >= REMOTE_TERMINAL_RESYNC_MAX_UNANSWERED_TIMEOUTS) { + // Why: pending recovery refuses the watchdog's probe, so escalate here instead. + this.recoverStalledStream(stream) + return + } + // Why retry, not reopen: reopening left expectedSeq before the gap, so the + // next frame re-tripped it and recovery ran on the timeout's cadence (#18488). + this.scheduleResyncRetry(stream) }, REMOTE_TERMINAL_RESYNC_TIMEOUT_MS) if (typeof timer.unref === 'function') { timer.unref() diff --git a/src/renderer/src/runtime/remote-runtime-terminal-snapshot-state.ts b/src/renderer/src/runtime/remote-runtime-terminal-snapshot-state.ts index d47bbff92e4..244b14f1be6 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-snapshot-state.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-snapshot-state.ts @@ -13,6 +13,8 @@ export const CONTROL_STREAM_ID = 0 export const MAX_REMOTE_TERMINAL_SNAPSHOT_BYTES = 2 * 1024 * 1024 export const REMOTE_TERMINAL_SNAPSHOT_REQUEST_TIMEOUT_MS = 10_000 export const REMOTE_TERMINAL_RESYNC_TIMEOUT_MS = 10_000 +// Why: a host that never answers a recovery must reach stall recovery, not retry forever. +export const REMOTE_TERMINAL_RESYNC_MAX_UNANSWERED_TIMEOUTS = 2 // Why: a truncated recovery means the server is too flooded to serialize; // retrying once per incoming chunk would stampede it, so back off instead. export const REMOTE_TERMINAL_RESYNC_RETRY_BASE_MS = 500