diff --git a/src/main/runtime/rpc/methods/terminal.ts b/src/main/runtime/rpc/methods/terminal.ts index 61ad7c7bafa..04d4d993b88 100644 --- a/src/main/runtime/rpc/methods/terminal.ts +++ b/src/main/runtime/rpc/methods/terminal.ts @@ -1986,6 +1986,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ stream.pendingOutputOverflowed = false stream.buffering = true const requestId = request.requestId + let sentSnapshotOutputSeq: number | undefined try { const scrollbackRows = normalizeMultiplexSnapshotScrollbackRows(request.scrollbackRows) let serialized = await serializeBudgetedRequestedSnapshot( @@ -2027,6 +2028,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ return } } + sentSnapshotOutputSeq = serialized?.seq sendSnapshotFrames((opcode, payload) => sendFrame(stream.streamId, opcode, payload), { kind: 'scrollback', cols: serialized?.cols ?? size?.cols ?? 80, @@ -2054,7 +2056,17 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ const pendingOutput = stream.pendingOutput.splice(0) if (shouldFlushPendingOutput) { for (const chunk of pendingOutput) { - stream.outputBatcher.push(chunk.data, chunk.meta) + // Why: an untagged reply resets the client to the snapshot's + // high-water, so covered bytes would render twice; tagged + // snapshots feed a side consumer and the live view still + // needs every buffered chunk. + const uncoveredData = + typeof requestId === 'number' + ? chunk.data + : getOutputAfterSnapshotSeq(chunk, sentSnapshotOutputSeq) + if (uncoveredData) { + stream.outputBatcher.push(uncoveredData, chunk.meta) + } } } stream.pendingOutputBytes = 0 diff --git a/src/main/runtime/rpc/terminal-multiplex-resync-replay-trim.test.ts b/src/main/runtime/rpc/terminal-multiplex-resync-replay-trim.test.ts new file mode 100644 index 00000000000..99c386c8492 --- /dev/null +++ b/src/main/runtime/rpc/terminal-multiplex-resync-replay-trim.test.ts @@ -0,0 +1,226 @@ +import { describe, expect, it, vi } from 'vitest' +import { RpcDispatcher } from './dispatcher' +import type { RpcRequest } from './core' +import type { OrcaRuntimeService } from '../orca-runtime' +import { TERMINAL_METHODS } from './methods/terminal' +import type { RuntimeTerminalWait } from '../../../shared/runtime-types' +import { + TerminalStreamOpcode, + decodeTerminalStreamFrame, + decodeTerminalStreamJson, + decodeTerminalStreamText, + encodeTerminalStreamFrame, + encodeTerminalStreamJson +} from '../../../shared/terminal-stream-protocol' + +// An untagged SnapshotRequest is the client's frame-drop resync: its reply +// resets the client terminal to the snapshot's output high-water. Output +// buffered while the snapshot serialized is already inside that snapshot up to +// its seq, so replaying the covered bytes afterward renders the recovered tail +// twice. A tagged (requestId) snapshot feeds a side consumer instead — the +// live view still needs every buffered chunk, so that replay must stay whole. + +type OutputMeta = { seq?: number; rawLength?: number } + +async function setupMultiplexStream(): Promise<{ + binaryFrames: Uint8Array[] + sendClientFrame: (opcode: TerminalStreamOpcode, payload: Uint8Array) => void + emitOutput: (data: string, meta?: OutputMeta) => void + setSnapshot: (snapshot: { data: string; seq?: number }) => void + deferNextSerialize: () => void + releaseSerialize: () => Promise + finish: () => Promise +}> { + const messages: string[] = [] + const binaryFrames: Uint8Array[] = [] + const handlers = new Map< + number, + (frame: NonNullable>) => void + >() + const cleanups = new Map void>() + let emitOutput: ((data: string, meta?: OutputMeta) => void) | null = null + let snapshot: { data: string; seq?: number } = { data: 'INITIAL', seq: 0 } + let deferSerialize = false + let releaseDeferredSerialize: (() => void) | null = null + + const runtime = { + getRuntimeId: () => 'test-runtime', + resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), + readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }), + serializeTerminalBuffer: vi.fn(async () => { + if (deferSerialize) { + deferSerialize = false + await new Promise((resolve) => { + releaseDeferredSerialize = resolve + }) + } + return { data: snapshot.data, cols: 80, rows: 24, seq: snapshot.seq } + }), + getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }), + getMobileDisplayMode: vi.fn().mockReturnValue('auto'), + getLayout: vi.fn().mockReturnValue({ seq: 1 }), + registerRemoteTerminalViewSubscriber: vi.fn(() => () => {}), + subscribeToTerminalData: vi.fn( + (_ptyId: string, cb: (data: string, meta?: OutputMeta) => void) => { + emitOutput = cb + return vi.fn() + } + ), + subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()), + subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()), + subscribeToDriverChanges: vi.fn().mockReturnValue(vi.fn()), + getTerminalFitOverride: vi.fn().mockReturnValue(null), + getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), + registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => { + cleanups.set(id, cleanup) + }), + cleanupSubscription: vi.fn((id: string) => { + const cleanup = cleanups.get(id) + cleanups.delete(id) + cleanup?.() + }), + waitForTerminal: vi.fn(() => new Promise(() => {})), + updateDesktopViewport: vi.fn().mockResolvedValue(true) + } as unknown as OrcaRuntimeService + const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + + const request: RpcRequest = { + id: 'req-1', + authToken: 'tok', + method: 'terminal.multiplex', + params: {} + } + const dispatchPromise = dispatcher.dispatchStreaming(request, (msg) => messages.push(msg), { + connectionId: 'conn-1', + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, + registerBinaryStreamHandler: (streamId, handler) => { + handlers.set(streamId, handler) + return () => handlers.delete(streamId) + } + }) + + await vi.runOnlyPendingTimersAsync() + expect(messages.some((msg) => JSON.parse(msg).result?.type === 'ready')).toBe(true) + + handlers.get(0)?.( + decodeTerminalStreamFrame( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Subscribe, + streamId: 0, + seq: 1, + payload: encodeTerminalStreamJson({ + streamId: 5, + terminal: 'terminal-1', + client: { id: 'desktop-1', type: 'desktop' } + }) + }) + )! + ) + for (let i = 0; i < 5; i += 1) { + await vi.runOnlyPendingTimersAsync() + } + expect(emitOutput).not.toBeNull() + + return { + binaryFrames, + sendClientFrame: (opcode, payload) => { + handlers.get(5)?.( + decodeTerminalStreamFrame( + encodeTerminalStreamFrame({ opcode, streamId: 5, seq: 1, payload }) + )! + ) + }, + emitOutput: (data, meta) => emitOutput!(data, meta), + setSnapshot: (next) => { + snapshot = next + }, + deferNextSerialize: () => { + deferSerialize = true + }, + releaseSerialize: async () => { + releaseDeferredSerialize?.() + releaseDeferredSerialize = null + for (let i = 0; i < 5; i += 1) { + await vi.runOnlyPendingTimersAsync() + } + }, + finish: async () => { + runtime.cleanupSubscription('terminal-multiplex:conn-1') + await dispatchPromise + } + } +} + +function outputTextsAfterLastSnapshotEnd(frames: Uint8Array[]): string[] { + const decoded = frames.map((frame) => decodeTerminalStreamFrame(frame)) + const lastEnd = decoded.reduce( + (last, frame, index) => (frame?.opcode === TerminalStreamOpcode.SnapshotEnd ? index : last), + -1 + ) + return decoded.slice(lastEnd + 1).flatMap((frame) => { + if (frame?.opcode === TerminalStreamOpcode.Output) { + return [decodeTerminalStreamText(frame.payload)] + } + if (frame?.opcode === TerminalStreamOpcode.OutputSpan) { + return [decodeTerminalStreamJson<{ data?: string }>(frame.payload)?.data ?? ''] + } + return [] + }) +} + +describe('terminal.multiplex requested-snapshot replay trim', () => { + it('drops snapshot-covered buffered output after an untagged resync reply', async () => { + vi.useFakeTimers() + try { + const harness = await setupMultiplexStream() + + harness.setSnapshot({ data: 'RECOVERED', seq: 12 }) + harness.deferNextSerialize() + harness.sendClientFrame(TerminalStreamOpcode.SnapshotRequest, encodeTerminalStreamJson({})) + // Buffered while the snapshot serialized: fully covered by seq 12, and a + // partial chunk straddling the boundary whose tail the client still needs. + harness.emitOutput('xxx', { seq: 9, rawLength: 3 }) + harness.emitOutput('bbbccc', { seq: 15, rawLength: 6 }) + await harness.releaseSerialize() + + const snapshotStart = harness.binaryFrames + .map((frame) => decodeTerminalStreamFrame(frame)) + .findLast((frame) => frame?.opcode === TerminalStreamOpcode.SnapshotStart)! + expect(decodeTerminalStreamJson(snapshotStart.payload)).toMatchObject({ seq: 12 }) + expect(outputTextsAfterLastSnapshotEnd(harness.binaryFrames).join('')).toBe('ccc') + + await harness.finish() + } finally { + vi.useRealTimers() + } + }) + + it('replays all buffered output untouched after a tagged snapshot reply', async () => { + vi.useFakeTimers() + try { + const harness = await setupMultiplexStream() + + harness.setSnapshot({ data: 'MANUAL', seq: 12 }) + harness.deferNextSerialize() + harness.sendClientFrame( + TerminalStreamOpcode.SnapshotRequest, + encodeTerminalStreamJson({ requestId: 7 }) + ) + harness.emitOutput('xxx', { seq: 9, rawLength: 3 }) + harness.emitOutput('bbbccc', { seq: 15, rawLength: 6 }) + await harness.releaseSerialize() + + const snapshotStart = harness.binaryFrames + .map((frame) => decodeTerminalStreamFrame(frame)) + .findLast((frame) => frame?.opcode === TerminalStreamOpcode.SnapshotStart)! + expect(decodeTerminalStreamJson(snapshotStart.payload)).toMatchObject({ requestId: 7 }) + expect(outputTextsAfterLastSnapshotEnd(harness.binaryFrames).join('')).toBe('xxxbbbccc') + + await harness.finish() + } finally { + vi.useRealTimers() + } + }) +}) 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 1d851a0e6ef..03a6a944618 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 @@ -39,6 +39,9 @@ class FakeMultiplexServer { dropNextOutput = false droppedFrames = 0 holdNextManualSnapshot = false + truncateNextRecoverySnapshot = false + dropNextRecoverySnapshotEnd = false + holdNextRecoverySnapshot = false snapshotRequests: (number | undefined)[] = [] private heldManualRequestId: number | null = null private snapshotData = 'INITIAL' @@ -71,6 +74,21 @@ class FakeMultiplexServer { // Resync request: the server serializes the *current* buffer, so recovery // includes everything the client missed. this.snapshotData = 'RECOVERED' + if (typeof payload?.requestId !== 'number' && this.holdNextRecoverySnapshot) { + // The reply's binary frames were all dropped under backpressure. + this.holdNextRecoverySnapshot = false + return + } + if (typeof payload?.requestId !== 'number' && this.truncateNextRecoverySnapshot) { + this.truncateNextRecoverySnapshot = false + this.sendSnapshot(undefined, { truncated: true }) + return + } + if (typeof payload?.requestId !== 'number' && this.dropNextRecoverySnapshotEnd) { + this.dropNextRecoverySnapshotEnd = false + this.sendSnapshot(undefined, { omitEnd: true }) + return + } this.sendSnapshot(payload?.requestId) } } @@ -79,14 +97,27 @@ class FakeMultiplexServer { this.toClient(encodeTerminalStreamFrame({ opcode, streamId: this.streamId, seq, payload })) } - private sendSnapshot(requestId?: number): void { + private sendSnapshot( + requestId?: number, + options?: { truncated?: boolean; omitEnd?: boolean } + ): void { this.send( TerminalStreamOpcode.SnapshotStart, - encodeTerminalStreamJson({ cols: 80, rows: 24, seq: this.cursorUnits, requestId }), + encodeTerminalStreamJson({ + cols: 80, + rows: 24, + seq: options?.truncated ? undefined : this.cursorUnits, + requestId, + truncated: options?.truncated + }), 0 ) - this.send(TerminalStreamOpcode.SnapshotChunk, encodeTerminalStreamText(this.snapshotData), 0) - this.send(TerminalStreamOpcode.SnapshotEnd, new Uint8Array(), 0) + if (!options?.truncated) { + this.send(TerminalStreamOpcode.SnapshotChunk, encodeTerminalStreamText(this.snapshotData), 0) + } + if (!options?.omitEnd) { + this.send(TerminalStreamOpcode.SnapshotEnd, new Uint8Array(), 0) + } } /** Emit an Output chunk, honoring simulated websocket backpressure. */ @@ -123,6 +154,10 @@ class FakeMultiplexServer { ) } + replaySnapshotCoveredOutput(text: string): void { + this.send(TerminalStreamOpcode.Output, encodeTerminalStreamText(text), this.cursorUnits) + } + flushHeldManualSnapshot(): void { if (this.heldManualRequestId === null) { throw new Error('No manual snapshot is held') @@ -209,6 +244,109 @@ describe('remote terminal frame-drop resync', () => { expect(server.droppedFrames).toBe(1) // Instead, a fresh authoritative snapshot recovers the terminal. expect(snapshots).toEqual(['INITIAL', '\x1b[2J\x1b[3J\x1b[HRECOVERED']) + + server.replaySnapshotCoveredOutput('ccc') + server.output('ddd') + expect(data).toEqual(['aaa', 'ddd']) + }) + + it('retries a truncated recovery on a backoff without accepting output across the gap', async () => { + vi.useFakeTimers() + try { + const { data, snapshots } = await subscribeClient() + server.truncateNextRecoverySnapshot = true + + server.output('aaa') + 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. + server.output('ddd') + expect(server.snapshotRequests).toEqual([undefined]) + + // The retry fires from the backoff timer alone — no further output needed. + await vi.advanceTimersByTimeAsync(500) + expect(server.snapshotRequests).toEqual([undefined, undefined]) + + server.output('eee') + expect(snapshots).toEqual(['INITIAL', '\x1b[2J\x1b[3J\x1b[HRECOVERED']) + expect(data).toEqual(['aaa', 'eee']) + } finally { + vi.useRealTimers() + } + }) + + 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 + + server.output('aaa') + server.dropNextOutput = true + server.output('bbb') + server.output('ccc') + expect(server.snapshotRequests).toEqual([undefined]) + + // The paired binary Error frame was dropped under backpressure; only the + // reliable JSON error event arrives. It must release the resync gate. + subscriptionCallbacks.onResponse({ + ok: true, + result: { type: 'error', streamId: stream.streamId, message: 'snapshot failed' } + }) + + server.output('ddd') + server.output('eee') + + expect(server.snapshotRequests).toEqual([undefined, undefined]) + expect(snapshots).toEqual(['INITIAL', '\x1b[2J\x1b[3J\x1b[HRECOVERED']) + expect(data).toEqual(['aaa', 'eee']) + }) + + it('dispatches the deferred resync when a JSON error consumes the manual snapshot', async () => { + const { data, snapshots, stream } = await subscribeClient() + server.holdNextManualSnapshot = true + const manualSnapshot = stream.serializeBuffer({ scrollbackRows: 100 }) + await Promise.resolve() + + server.output('aaa') + server.dropNextOutput = true + server.output('bbb') + server.output('ccc') + expect(server.snapshotRequests).toHaveLength(1) + + subscriptionCallbacks.onResponse({ + ok: true, + result: { type: 'error', streamId: stream.streamId, message: 'stream failed' } + }) + await expect(manualSnapshot).rejects.toThrow('stream failed') + + expect(server.snapshotRequests).toEqual([expect.any(Number), undefined]) + expect(snapshots).toEqual(['INITIAL', '\x1b[2J\x1b[3J\x1b[HRECOVERED']) + + server.output('ddd') + expect(data).toEqual(['aaa', 'ddd']) + }) + + it('times out a dropped recovery end and retries on the next sequence gap', async () => { + vi.useFakeTimers() + try { + const { data, snapshots } = await subscribeClient() + server.dropNextRecoverySnapshotEnd = true + + server.output('aaa') + server.dropNextOutput = true + server.output('bbb') + server.output('ccc') + await vi.advanceTimersByTimeAsync(10_000) + server.output('ddd') + server.output('eee') + + expect(server.snapshotRequests).toEqual([undefined, undefined]) + expect(snapshots).toEqual(['INITIAL', '\x1b[2J\x1b[3J\x1b[HRECOVERED']) + expect(data).toEqual(['aaa', 'eee']) + } finally { + vi.useRealTimers() + } }) it('passes contiguous output straight through without resyncing', async () => { @@ -352,5 +490,8 @@ describe('remote terminal frame-drop resync', () => { await expect(manualSnapshot).resolves.toMatchObject({ data: 'MANUAL' }) expect(server.snapshotRequests).toHaveLength(2) expect(snapshots).toEqual(['INITIAL', '\x1b[2J\x1b[3J\x1b[HRECOVERED']) + + server.output('ddd') + expect(data).toEqual(['aaa', 'ddd']) }) }) diff --git a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts index 455261b0f9c..a88e6f8283f 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts @@ -90,8 +90,11 @@ type RemoteRuntimeMultiplexedTerminalState = { // Track it so a gap triggers a self-healing snapshot resync instead of // silently rendering corrupt/missing output (frame-drop resync). expectedSeq: number | undefined + recoverySnapshotSeq: number | undefined resyncInFlight: boolean resyncPendingSend: boolean + resyncTimer: ReturnType | null + resyncAttempts: number } type RemoteRuntimeSnapshotInfo = { @@ -126,6 +129,11 @@ type RemoteRuntimeSnapshotRequest = { const CONTROL_STREAM_ID = 0 const MAX_REMOTE_TERMINAL_SNAPSHOT_BYTES = 2 * 1024 * 1024 const REMOTE_TERMINAL_SNAPSHOT_REQUEST_TIMEOUT_MS = 10_000 +const REMOTE_TERMINAL_RESYNC_TIMEOUT_MS = 10_000 +// Why: a truncated recovery means the server is too flooded to serialize; +// retrying once per incoming chunk would stampede it, so back off instead. +const REMOTE_TERMINAL_RESYNC_RETRY_BASE_MS = 500 +const REMOTE_TERMINAL_RESYNC_RETRY_MAX_MS = 5_000 // Why: exported so the transport can classify it as benign — the snapshot was // skipped but live output continues, so it must not surface a fatal red banner. export const REMOTE_TERMINAL_SNAPSHOT_TOO_LARGE = @@ -247,8 +255,11 @@ class RemoteRuntimeTerminalMultiplexer { initialSnapshotReceived: false, pendingSnapshotRequest: null, expectedSeq: undefined, + recoverySnapshotSeq: undefined, resyncInFlight: false, - resyncPendingSend: false + resyncPendingSend: false, + resyncTimer: null, + resyncAttempts: 0 } this.streams.set(streamId, state) @@ -282,6 +293,7 @@ class RemoteRuntimeTerminalMultiplexer { close: () => { if (this.streams.get(streamId) === state) { this.sendFrame(streamId, TerminalStreamOpcode.Unsubscribe) + clearResyncTimer(state) rejectPendingSnapshotRequest(state, 'Remote terminal stream closed.') this.streams.delete(streamId) this.closeIfIdle() @@ -418,6 +430,7 @@ class RemoteRuntimeTerminalMultiplexer { } if (event.type === 'end') { clearSnapshot(stream) + clearResyncTimer(stream) rejectPendingSnapshotRequest(stream, 'Remote terminal stream ended.') this.streams.delete(event.streamId) stream.callbacks.onEnd?.() @@ -428,6 +441,15 @@ class RemoteRuntimeTerminalMultiplexer { stream, typeof event.message === 'string' ? event.message : 'Remote terminal stream failed.' ) + // Why: the paired binary Error frame can be dropped under backpressure; + // this reliable event must also dispatch or release the resync gate, and + // must never disarm the watchdog while leaving the gate shut. + if (stream.resyncPendingSend) { + this.sendDeferredResyncSnapshot(stream) + } else { + clearResyncTimer(stream) + stream.resyncInFlight = false + } stream.callbacks.onError?.( typeof event.message === 'string' ? event.message : 'Remote terminal stream failed.' ) @@ -509,6 +531,15 @@ class RemoteRuntimeTerminalMultiplexer { 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. + if ( + typeof seq === 'number' && + typeof stream.recoverySnapshotSeq === 'number' && + seq <= stream.recoverySnapshotSeq + ) { + return + } if (this.detectOutputGap(stream, seq, rawLength)) { this.requestResyncSnapshot(stream) return @@ -570,13 +601,14 @@ class RemoteRuntimeTerminalMultiplexer { const target = stream.snapshotTarget const info = stream.snapshotInfo const pendingRequest = stream.pendingSnapshotRequest + const snapshotApplied = !stream.snapshotOverflowed && info?.truncated !== true const matchesPendingRequest = target === 'request' && pendingRequest && (typeof info?.requestId === 'number' ? info.requestId === pendingRequest.requestId : stream.initialSnapshotReceived) - if (!stream.snapshotOverflowed && info?.truncated !== true) { + if (snapshotApplied) { if (matchesPendingRequest) { pendingRequest.resolve({ data: data ?? '', @@ -605,15 +637,31 @@ class RemoteRuntimeTerminalMultiplexer { clearPendingSnapshotRequest(stream) } clearSnapshot(stream) - // Why: the snapshot is the new authoritative output high-water; align the - // gap detector to it and re-open the live path (used by both the initial - // snapshot and a frame-drop resync, which reuses the 'initial' target). if (target === 'initial') { + clearResyncTimer(stream) stream.expectedSeq = typeof info?.seq === 'number' ? info.seq : undefined stream.resyncInFlight = false stream.resyncPendingSend = false stream.initialSnapshotReceived = true stream.callbacks.onSubscribed?.() + } else if (target === 'recovery') { + // Why: only an applied recovery is authoritative; retaining the prior + // high-water after a discarded snapshot keeps the gap detectable. + if (snapshotApplied) { + clearResyncTimer(stream) + stream.expectedSeq = typeof info?.seq === 'number' ? info.seq : undefined + stream.recoverySnapshotSeq = typeof info?.seq === 'number' ? info.seq : undefined + stream.resyncAttempts = 0 + stream.resyncInFlight = false + stream.resyncPendingSend = false + } else if (stream.resyncInFlight) { + this.scheduleResyncRetry(stream) + } else { + // Why: a discarded server-pushed recovery leaves dropped output + // unrepresented; pull a fresh snapshot now instead of waiting for + // the next chunk to expose the gap. + this.requestResyncSnapshot(stream) + } } else { this.sendDeferredResyncSnapshot(stream) } @@ -629,6 +677,7 @@ class RemoteRuntimeTerminalMultiplexer { return } // Why: a failed resync must re-open the live path or output stalls forever. + clearResyncTimer(stream) stream.resyncInFlight = false stream.resyncPendingSend = false stream.callbacks.onError?.(decodeTerminalStreamText(frame.payload)) @@ -660,11 +709,13 @@ class RemoteRuntimeTerminalMultiplexer { return } stream.resyncInFlight = true - stream.expectedSeq = undefined if (stream.pendingSnapshotRequest) { // 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. stream.resyncPendingSend = true + this.startResyncTimer(stream) return } this.sendResyncSnapshot(stream) @@ -679,6 +730,7 @@ class RemoteRuntimeTerminalMultiplexer { private sendResyncSnapshot(stream: RemoteRuntimeMultiplexedTerminalState): void { stream.resyncPendingSend = false + this.startResyncTimer(stream) const sent = this.sendFrame( stream.streamId, TerminalStreamOpcode.SnapshotRequest, @@ -686,10 +738,62 @@ class RemoteRuntimeTerminalMultiplexer { ) if (!sent) { // Transport is down; the reconnect path re-subscribes from scratch. + clearResyncTimer(stream) stream.resyncInFlight = false } } + // 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. + private scheduleResyncRetry(stream: RemoteRuntimeMultiplexedTerminalState): void { + stream.resyncAttempts += 1 + const delay = Math.min( + REMOTE_TERMINAL_RESYNC_RETRY_MAX_MS, + REMOTE_TERMINAL_RESYNC_RETRY_BASE_MS * 2 ** Math.min(stream.resyncAttempts - 1, 4) + ) + clearResyncTimer(stream) + const timer = setTimeout(() => { + if ( + stream.resyncTimer !== timer || + this.streams.get(stream.streamId) !== stream || + !stream.resyncInFlight + ) { + return + } + stream.resyncTimer = null + if (stream.pendingSnapshotRequest) { + stream.resyncPendingSend = true + this.startResyncTimer(stream) + return + } + this.sendResyncSnapshot(stream) + }, delay) + if (typeof timer.unref === 'function') { + timer.unref() + } + stream.resyncTimer = timer + } + + private startResyncTimer(stream: RemoteRuntimeMultiplexedTerminalState): void { + clearResyncTimer(stream) + const timer = setTimeout(() => { + if ( + stream.resyncTimer !== timer || + this.streams.get(stream.streamId) !== stream || + !stream.resyncInFlight + ) { + return + } + stream.resyncTimer = null + stream.resyncInFlight = false + stream.resyncPendingSend = false + }, REMOTE_TERMINAL_RESYNC_TIMEOUT_MS) + if (typeof timer.unref === 'function') { + timer.unref() + } + stream.resyncTimer = timer + } + private requestSnapshot( stream: RemoteRuntimeMultiplexedTerminalState, opts?: { scrollbackRows?: number } @@ -820,6 +924,7 @@ class RemoteRuntimeTerminalMultiplexer { this.releaseIfCurrent(this.environmentId, this) for (const stream of streams) { clearSnapshot(stream) + clearResyncTimer(stream) rejectPendingSnapshotRequest(stream, message ?? 'Remote runtime connection closed.') const canHandleClose = Boolean(stream.callbacks.onTransportClose) stream.callbacks.onTransportClose?.({ recoverable }) @@ -909,6 +1014,14 @@ function clearPendingSnapshotRequest(stream: RemoteRuntimeMultiplexedTerminalSta } } +function clearResyncTimer(stream: RemoteRuntimeMultiplexedTerminalState): void { + const timer = stream.resyncTimer + stream.resyncTimer = null + if (timer) { + clearTimeout(timer) + } +} + function rejectPendingSnapshotRequest( stream: RemoteRuntimeMultiplexedTerminalState, message: string