From 5d656ae6d47ee0677636838e5634f9c9ab81eefa Mon Sep 17 00:00:00 2001 From: Jinjing <6427696+AmethystLiang@users.noreply.github.com> Date: Wed, 22 Jul 2026 21:39:12 -0700 Subject: [PATCH] fix(terminal): prevent double rendering on frame-drop resync After an untagged resync snapshot, the client resets to the snapshot's output high-water. Buffered output generated while serializing is already included in the snapshot, so replaying the entire buffer afterward renders those bytes twice. Trim to output after the snapshot seq. Tagged snapshots feed side consumers and need all buffered output unchanged. Add exponential backoff for truncated recovery retries to avoid stampeding flooded servers. --- src/main/runtime/rpc/methods/terminal.ts | 14 +- ...minal-multiplex-resync-replay-trim.test.ts | 226 ++++++++++++++++++ ...runtime-terminal-frame-drop-resync.test.ts | 73 +++++- .../remote-runtime-terminal-multiplexer.ts | 68 +++++- 4 files changed, 371 insertions(+), 10 deletions(-) create mode 100644 src/main/runtime/rpc/terminal-multiplex-resync-replay-trim.test.ts diff --git a/src/main/runtime/rpc/methods/terminal.ts b/src/main/runtime/rpc/methods/terminal.ts index 8f952fcc15f..81c582c8f30 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 1bb0597cda0..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 @@ -41,6 +41,7 @@ class FakeMultiplexServer { holdNextManualSnapshot = false truncateNextRecoverySnapshot = false dropNextRecoverySnapshotEnd = false + holdNextRecoverySnapshot = false snapshotRequests: (number | undefined)[] = [] private heldManualRequestId: number | null = null private snapshotData = 'INITIAL' @@ -73,6 +74,11 @@ 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 }) @@ -244,14 +250,50 @@ describe('remote terminal frame-drop resync', () => { expect(data).toEqual(['aaa', 'ddd']) }) - it('retries after a truncated recovery without accepting output across the gap', async () => { - const { data, snapshots } = await subscribeClient() - server.truncateNextRecoverySnapshot = true + 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') @@ -260,6 +302,31 @@ describe('remote terminal frame-drop resync', () => { 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 { diff --git a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts index 2bbeb4f7043..a88e6f8283f 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts @@ -94,6 +94,7 @@ type RemoteRuntimeMultiplexedTerminalState = { resyncInFlight: boolean resyncPendingSend: boolean resyncTimer: ReturnType | null + resyncAttempts: number } type RemoteRuntimeSnapshotInfo = { @@ -129,6 +130,10 @@ 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 = @@ -253,7 +258,8 @@ class RemoteRuntimeTerminalMultiplexer { recoverySnapshotSeq: undefined, resyncInFlight: false, resyncPendingSend: false, - resyncTimer: null + resyncTimer: null, + resyncAttempts: 0 } this.streams.set(streamId, state) @@ -431,11 +437,19 @@ class RemoteRuntimeTerminalMultiplexer { this.closeIfIdle() } else if (event.type === 'error') { clearSnapshot(stream) - clearResyncTimer(stream) rejectPendingSnapshotRequest( 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.' ) @@ -631,15 +645,23 @@ class RemoteRuntimeTerminalMultiplexer { stream.initialSnapshotReceived = true stream.callbacks.onSubscribed?.() } else if (target === 'recovery') { - clearResyncTimer(stream) // Why: only an applied recovery is authoritative; retaining the prior - // high-water after a discarded snapshot makes live output retry the gap. + // 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) } - stream.resyncInFlight = false - stream.resyncPendingSend = false } else { this.sendDeferredResyncSnapshot(stream) } @@ -690,7 +712,10 @@ class RemoteRuntimeTerminalMultiplexer { 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) @@ -718,6 +743,37 @@ class RemoteRuntimeTerminalMultiplexer { } } + // 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(() => {