From 1f768d3dbef34a45e7f85767c6697a39bbfe88df Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Fri, 10 Jul 2026 04:24:29 -0400 Subject: [PATCH] fix(remote): resync remote terminal on dropped output frames The multiplex output path silently drops Output frames when the websocket buffer exceeds MAX_BINARY_BUFFERED_AMOUNT (encryptedBinaryReply returns false). The client never noticed the gap, so a congested link (fast PTY over slow/SSH transport) permanently corrupted the remote terminal with missing/garbled output and no error. Output frame seq is a byte high-water, so a drop is self-describing on the wire. The client now tracks the expected next offset and, on a gap, discards the corrupt tail and pulls a fresh authoritative snapshot (self-healing, no user-facing error). Server-side, seq-less Output chunks now carry the sentinel 0 instead of the control-frame cursor so they don't poison the client's gap detector, and sendFrame propagates the send/drop result end-to-end. Adds an end-to-end repro test driving the real client multiplexer through a fake backpressured transport. Co-authored-by: Orca --- src/main/runtime/rpc/dispatcher.ts | 2 +- src/main/runtime/rpc/methods/terminal.ts | 20 +- .../terminal-multiplex-escape-tail.test.ts | 9 +- .../runtime/rpc/terminal-multiplex.test.ts | 217 ++++++++++++++---- .../rpc/terminal-output-batching.test.ts | 33 ++- .../rpc/terminal-subscribe-buffer.test.ts | 125 +++++++--- ...runtime-terminal-frame-drop-resync.test.ts | 174 ++++++++++++++ .../remote-runtime-terminal-multiplexer.ts | 72 +++++- 8 files changed, 554 insertions(+), 98 deletions(-) create mode 100644 src/renderer/src/runtime/remote-runtime-terminal-frame-drop-resync.test.ts diff --git a/src/main/runtime/rpc/dispatcher.ts b/src/main/runtime/rpc/dispatcher.ts index 818ddb92fca..24ff44597b0 100644 --- a/src/main/runtime/rpc/dispatcher.ts +++ b/src/main/runtime/rpc/dispatcher.ts @@ -102,7 +102,7 @@ export class RpcDispatcher { signal?: AbortSignal clientId?: string clientKind?: 'mobile' | 'runtime' - sendBinary?: (bytes: Uint8Array) => void + sendBinary?: (bytes: Uint8Array) => boolean | void registerBinaryStreamHandler?: ( streamId: number, handler: (frame: TerminalStreamFrame) => void diff --git a/src/main/runtime/rpc/methods/terminal.ts b/src/main/runtime/rpc/methods/terminal.ts index a750438e03e..d35642a436e 100644 --- a/src/main/runtime/rpc/methods/terminal.ts +++ b/src/main/runtime/rpc/methods/terminal.ts @@ -1221,18 +1221,20 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ opcode: TerminalStreamOpcode, payload: Uint8Array = new Uint8Array(), seq?: number - ): void => { + ): boolean => { if (closed) { - return + return false } - sendBinary( - encodeTerminalStreamFrame({ - opcode, - streamId, - seq: typeof seq === 'number' ? seq : cursor++, - payload - }) + // Why: Output `seq` is a byte high-water the client uses for frame-drop + // gap detection, so a seq-less Output chunk must carry the sentinel 0 + // (== "no seq") rather than the cursor value that orders control frames; + // a cursor value would poison the client's expected-seq tracker. + const resolvedSeq = + typeof seq === 'number' ? seq : opcode === TerminalStreamOpcode.Output ? 0 : cursor++ + const sent = sendBinary( + encodeTerminalStreamFrame({ opcode, streamId, seq: resolvedSeq, payload }) ) + return sent !== false } const sendStreamError = (streamId: number, message: string): void => { sendFrame(streamId, TerminalStreamOpcode.Error, encodeTerminalStreamText(message)) diff --git a/src/main/runtime/rpc/terminal-multiplex-escape-tail.test.ts b/src/main/runtime/rpc/terminal-multiplex-escape-tail.test.ts index 59e4d2d1b29..24367102745 100644 --- a/src/main/runtime/rpc/terminal-multiplex-escape-tail.test.ts +++ b/src/main/runtime/rpc/terminal-multiplex-escape-tail.test.ts @@ -70,14 +70,19 @@ describe('terminal.multiplex pending-escape-tail threading (#7329)', () => { waitForTerminal: vi.fn(() => new Promise(() => {})), updateDesktopViewport: vi.fn().mockResolvedValue(true) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.multiplex', {}), (msg) => messages.push(msg), { connectionId: 'conn-1', - sendBinary: (bytes) => binaryFrames.push(bytes), + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, registerBinaryStreamHandler: (streamId, handler) => { handlers.set(streamId, handler) return () => handlers.delete(streamId) diff --git a/src/main/runtime/rpc/terminal-multiplex.test.ts b/src/main/runtime/rpc/terminal-multiplex.test.ts index 69e956aa155..edb5afc030a 100644 --- a/src/main/runtime/rpc/terminal-multiplex.test.ts +++ b/src/main/runtime/rpc/terminal-multiplex.test.ts @@ -77,14 +77,19 @@ describe('terminal multiplex RPC', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.multiplex', {}), (msg) => messages.push(msg), { connectionId: 'conn-1', - sendBinary: (bytes) => binaryFrames.push(bytes), + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, registerBinaryStreamHandler: (streamId, handler) => { handlers.set(streamId, handler) return () => handlers.delete(streamId) @@ -199,7 +204,10 @@ describe('terminal multiplex RPC', () => { opcode: TerminalStreamOpcode.SnapshotRequest, streamId: 5, seq: 4, - payload: encodeTerminalStreamJson({ requestId: 7, scrollbackRows: 5000 }) + payload: encodeTerminalStreamJson({ + requestId: 7, + scrollbackRows: 5000 + }) }) )! ) @@ -291,14 +299,19 @@ describe('terminal multiplex RPC', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateMobileViewport: vi.fn().mockResolvedValue({ updated: true, applied: true }) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.multiplex', {}), (msg) => messages.push(msg), { connectionId: 'conn-stale-multiplex-resize', - sendBinary: (bytes) => binaryFrames.push(bytes), + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, registerBinaryStreamHandler: (streamId, handler) => { handlers.set(streamId, handler) return () => handlers.delete(streamId) @@ -327,8 +340,20 @@ describe('terminal multiplex RPC', () => { await vi.waitFor(() => expect(resizeListener).toBeDefined()) binaryFrames.splice(0) - resizeListener?.({ cols: 90, rows: 24, displayMode: 'auto', reason: 'apply-layout', seq: 2 }) - resizeListener?.({ cols: 100, rows: 24, displayMode: 'auto', reason: 'apply-layout', seq: 3 }) + resizeListener?.({ + cols: 90, + rows: 24, + displayMode: 'auto', + reason: 'apply-layout', + seq: 2 + }) + resizeListener?.({ + cols: 100, + rows: 24, + displayMode: 'auto', + reason: 'apply-layout', + seq: 3 + }) await vi.waitFor(() => expect(restreamResolves).toHaveLength(2)) restreamResolves[1]?.({ data: 'newer', cols: 100, rows: 24 }) @@ -402,14 +427,19 @@ describe('terminal multiplex RPC', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.multiplex', {}), (msg) => messages.push(msg), { connectionId: 'conn-multibyte-output-batch', - sendBinary: (bytes) => binaryFrames.push(bytes), + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, registerBinaryStreamHandler: (streamId, handler) => { handlers.set(streamId, handler) return () => handlers.delete(streamId) @@ -478,9 +508,11 @@ describe('terminal multiplex RPC', () => { const cleanups = new Map void>() const runtime = stubRuntime({ resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), - readTerminal: vi - .fn() - .mockResolvedValue({ tail: ['line 120'], truncated: false, limited: true }), + readTerminal: vi.fn().mockResolvedValue({ + tail: ['line 120'], + truncated: false, + limited: true + }), serializeTerminalBuffer: vi.fn().mockResolvedValue(null), getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }), getMobileDisplayMode: vi.fn().mockReturnValue('auto'), @@ -502,14 +534,19 @@ describe('terminal multiplex RPC', () => { waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.multiplex', {}), (msg) => messages.push(msg), { connectionId: 'conn-multiplex-limited', - sendBinary: (bytes) => binaryFrames.push(bytes), + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, registerBinaryStreamHandler: (streamId, handler) => { handlers.set(streamId, handler) return () => handlers.delete(streamId) @@ -578,8 +615,16 @@ describe('terminal multiplex RPC', () => { serializeTerminalBuffer: vi .fn() .mockResolvedValueOnce({ data: 'initial', cols: 120, rows: 40 }) - .mockResolvedValueOnce({ data: 'x'.repeat(2 * 1024 * 1024 + 1), cols: 120, rows: 40 }) - .mockResolvedValueOnce({ data: 'budgeted snapshot', cols: 120, rows: 40 }), + .mockResolvedValueOnce({ + data: 'x'.repeat(2 * 1024 * 1024 + 1), + cols: 120, + rows: 40 + }) + .mockResolvedValueOnce({ + data: 'budgeted snapshot', + cols: 120, + rows: 40 + }), getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }), getMobileDisplayMode: vi.fn().mockReturnValue('auto'), getLayout: vi.fn().mockReturnValue({ seq: 1 }), @@ -596,14 +641,19 @@ describe('terminal multiplex RPC', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.multiplex', {}), (msg) => messages.push(msg), { connectionId: 'conn-budgeted-request', - sendBinary: (bytes) => binaryFrames.push(bytes), + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, registerBinaryStreamHandler: (streamId, handler) => { handlers.set(streamId, handler) return () => handlers.delete(streamId) @@ -640,7 +690,10 @@ describe('terminal multiplex RPC', () => { opcode: TerminalStreamOpcode.SnapshotRequest, streamId: 14, seq: 2, - payload: encodeTerminalStreamJson({ requestId: 55, scrollbackRows: 5000 }) + payload: encodeTerminalStreamJson({ + requestId: 55, + scrollbackRows: 5000 + }) }) )! ) @@ -704,7 +757,10 @@ describe('terminal multiplex RPC', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.multiplex', {}), @@ -805,7 +861,10 @@ describe('terminal multiplex RPC', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.multiplex', {}), @@ -892,7 +951,10 @@ describe('terminal multiplex RPC', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.subscribe', { @@ -981,7 +1043,10 @@ describe('terminal multiplex RPC', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.subscribe', { @@ -992,7 +1057,9 @@ describe('terminal multiplex RPC', () => { (msg) => messages.push(msg), { connectionId: 'conn-subscribe-output-chunking', - sendBinary: (bytes) => binaryFrames.push(bytes), + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, registerBinaryStreamHandler: (streamId, handler) => { handlers.set(streamId, handler) return () => handlers.delete(streamId) @@ -1067,7 +1134,10 @@ describe('terminal multiplex RPC', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.subscribe', { @@ -1078,13 +1148,18 @@ describe('terminal multiplex RPC', () => { (msg) => messages.push(msg), { connectionId: 'conn-buffered-output-on-subscribe', - sendBinary: (bytes) => binaryFrames.push(bytes), + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, registerBinaryStreamHandler: vi.fn(() => vi.fn()) } ) await vi.waitFor(() => expect(dataListenerRef.current).toBeDefined()) - dataListenerRef.current?.('starting shell\r\n', { seq: 16, rawLength: 16 }) + dataListenerRef.current?.('starting shell\r\n', { + seq: 16, + rawLength: 16 + }) resolveSnapshot({ data: '', cols: 120, rows: 40 }) await vi.waitFor(() => expect(messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe(true) @@ -1125,7 +1200,12 @@ describe('terminal multiplex RPC', () => { readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }), serializeTerminalBuffer: vi.fn( () => - new Promise<{ data: string; cols: number; rows: number; seq: number }>((resolve) => { + new Promise<{ + data: string + cols: number + rows: number + seq: number + }>((resolve) => { resolveSnapshot = resolve }) ), @@ -1151,7 +1231,10 @@ describe('terminal multiplex RPC', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.subscribe', { @@ -1162,7 +1245,9 @@ describe('terminal multiplex RPC', () => { (msg) => messages.push(msg), { connectionId: 'conn-buffered-output-covered-by-snapshot', - sendBinary: (bytes) => binaryFrames.push(bytes), + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, registerBinaryStreamHandler: vi.fn(() => vi.fn()) } ) @@ -1222,7 +1307,12 @@ describe('terminal multiplex RPC', () => { readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }), serializeTerminalBuffer: vi.fn( () => - new Promise<{ data: string; cols: number; rows: number; seq: number }>((resolve) => { + new Promise<{ + data: string + cols: number + rows: number + seq: number + }>((resolve) => { resolveSnapshot = resolve }) ), @@ -1248,7 +1338,10 @@ describe('terminal multiplex RPC', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.subscribe', { @@ -1259,7 +1352,9 @@ describe('terminal multiplex RPC', () => { (msg) => messages.push(msg), { connectionId: 'conn-buffered-output-partially-covered-by-snapshot', - sendBinary: (bytes) => binaryFrames.push(bytes), + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, registerBinaryStreamHandler: vi.fn(() => vi.fn()) } ) @@ -1319,7 +1414,10 @@ describe('terminal multiplex RPC', () => { cleanups.set(id, cleanup) }) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.multiplex', {}), @@ -1327,7 +1425,9 @@ describe('terminal multiplex RPC', () => { { signal: controller.signal, connectionId: 'conn-phone-multiplex', - sendBinary: (bytes) => binaryFrames.push(bytes), + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, registerBinaryStreamHandler: (streamId, handler) => { handlers.set(streamId, handler) return () => handlers.delete(streamId) @@ -1393,14 +1493,19 @@ describe('terminal multiplex RPC', () => { cleanups.set(id, cleanup) }) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.multiplex', {}), (msg) => messages.push(msg), { connectionId: 'conn-stale-handle', - sendBinary: (bytes) => binaryFrames.push(bytes), + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, registerBinaryStreamHandler: (streamId, handler) => { handlers.set(streamId, handler) return () => handlers.delete(streamId) @@ -1484,14 +1589,19 @@ describe('terminal multiplex RPC', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.multiplex', {}), (msg) => messages.push(msg), { connectionId: 'conn-buffered', - sendBinary: (bytes) => binaryFrames.push(bytes), + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, registerBinaryStreamHandler: (streamId, handler) => { handlers.set(streamId, handler) return () => handlers.delete(streamId) @@ -1593,14 +1703,19 @@ describe('terminal multiplex RPC', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.multiplex', {}), (msg) => messages.push(msg), { connectionId: 'conn-buffered-multibyte', - sendBinary: (bytes) => binaryFrames.push(bytes), + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, registerBinaryStreamHandler: (streamId, handler) => { handlers.set(streamId, handler) return () => handlers.delete(streamId) @@ -1690,7 +1805,11 @@ describe('terminal multiplex RPC', () => { resolveRequestedSnapshot = resolve }) ) - .mockResolvedValueOnce({ data: 'retry snapshot', cols: 120, rows: 40 }), + .mockResolvedValueOnce({ + data: 'retry snapshot', + cols: 120, + rows: 40 + }), getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }), getMobileDisplayMode: vi.fn().mockReturnValue('auto'), getLayout: vi.fn().mockReturnValue({ seq: 1 }), @@ -1710,14 +1829,19 @@ describe('terminal multiplex RPC', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.multiplex', {}), (msg) => messages.push(msg), { connectionId: 'conn-request-overflow', - sendBinary: (bytes) => binaryFrames.push(bytes), + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, registerBinaryStreamHandler: (streamId, handler) => { handlers.set(streamId, handler) return () => handlers.delete(streamId) @@ -1754,7 +1878,10 @@ describe('terminal multiplex RPC', () => { opcode: TerminalStreamOpcode.SnapshotRequest, streamId: 12, seq: 2, - payload: encodeTerminalStreamJson({ requestId: 44, scrollbackRows: 5000 }) + payload: encodeTerminalStreamJson({ + requestId: 44, + scrollbackRows: 5000 + }) }) )! ) diff --git a/src/main/runtime/rpc/terminal-output-batching.test.ts b/src/main/runtime/rpc/terminal-output-batching.test.ts index 2df3dbc7173..df81e05e057 100644 --- a/src/main/runtime/rpc/terminal-output-batching.test.ts +++ b/src/main/runtime/rpc/terminal-output-batching.test.ts @@ -52,7 +52,10 @@ describe('terminal output batching', () => { }), waitForTerminal: vi.fn(() => new Promise(() => {})) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.subscribe', { @@ -78,7 +81,9 @@ describe('terminal output batching', () => { .map((msg) => JSON.parse(msg)) .filter((message) => message.result?.type === 'data') expect(dataMessages).toHaveLength(1) - expect(dataMessages[0]).toMatchObject({ result: { type: 'data', chunk: 'ab' } }) + expect(dataMessages[0]).toMatchObject({ + result: { type: 'data', chunk: 'ab' } + }) runtime.cleanupSubscription('terminal-1:desktop-1') await dispatchPromise @@ -123,7 +128,10 @@ describe('terminal output batching', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateMobileViewport: vi.fn().mockResolvedValue(false) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.subscribe', { @@ -134,7 +142,9 @@ describe('terminal output batching', () => { (msg) => messages.push(msg), { connectionId: 'conn-1', - sendBinary: (bytes) => binaryFrames.push(bytes) + sendBinary: (bytes) => { + binaryFrames.push(bytes) + } } ) @@ -144,7 +154,10 @@ describe('terminal output batching', () => { const subscribed = messages .map((msg) => JSON.parse(msg)) .find((msg) => msg.result?.type === 'subscribed') - expect(subscribed?.result).toMatchObject({ type: 'subscribed', streamId: expect.any(Number) }) + expect(subscribed?.result).toMatchObject({ + type: 'subscribed', + streamId: expect.any(Number) + }) await vi.waitFor(() => expect(dataListenerRef.current).toBeDefined()) const emitData = dataListenerRef.current @@ -206,7 +219,10 @@ describe('terminal output batching', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateMobileViewport: vi.fn().mockResolvedValue(false) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.subscribe', { @@ -286,7 +302,10 @@ describe('terminal output batching', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateMobileViewport: vi.fn().mockResolvedValue(false) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const messages: string[] = [] const dispatchPromise = dispatcher.dispatchStreaming( diff --git a/src/main/runtime/rpc/terminal-subscribe-buffer.test.ts b/src/main/runtime/rpc/terminal-subscribe-buffer.test.ts index adad88dded0..f4a70cbeebd 100644 --- a/src/main/runtime/rpc/terminal-subscribe-buffer.test.ts +++ b/src/main/runtime/rpc/terminal-subscribe-buffer.test.ts @@ -40,7 +40,10 @@ describe('terminal subscribe buffering', () => { ), readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.subscribe', { @@ -77,11 +80,16 @@ describe('terminal subscribe buffering', () => { const messages: string[] = [] const runtime = stubRuntime({ resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: null }), - readTerminal: vi - .fn() - .mockResolvedValue({ tail: ['line 120'], truncated: false, limited: true }) + readTerminal: vi.fn().mockResolvedValue({ + tail: ['line 120'], + truncated: false, + limited: true + }) + }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) await dispatcher.dispatchStreaming( makeRequest('terminal.subscribe', { terminal: 'terminal-1' }), @@ -104,9 +112,11 @@ describe('terminal subscribe buffering', () => { const cleanups = new Map void>() const runtime = stubRuntime({ resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), - readTerminal: vi - .fn() - .mockResolvedValue({ tail: ['line 120'], truncated: false, limited: true }), + readTerminal: vi.fn().mockResolvedValue({ + tail: ['line 120'], + truncated: false, + limited: true + }), serializeTerminalBuffer: vi.fn().mockResolvedValue(null), getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }), getMobileDisplayMode: vi.fn().mockReturnValue('auto'), @@ -122,7 +132,10 @@ describe('terminal subscribe buffering', () => { }), waitForTerminal: vi.fn(() => new Promise(() => {})) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.subscribe', { @@ -172,7 +185,10 @@ describe('terminal subscribe buffering', () => { cleanupSubscription: vi.fn(), waitForTerminal: vi.fn(() => new Promise(() => {})) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.subscribe', { @@ -209,9 +225,11 @@ describe('terminal subscribe buffering', () => { const cleanups = new Map void>() const runtime = stubRuntime({ resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), - readTerminal: vi - .fn() - .mockResolvedValue({ tail: ['line 120'], truncated: false, limited: true }), + readTerminal: vi.fn().mockResolvedValue({ + tail: ['line 120'], + truncated: false, + limited: true + }), serializeTerminalBuffer: vi.fn().mockResolvedValue(null), getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }), getMobileDisplayMode: vi.fn().mockReturnValue('auto'), @@ -230,7 +248,10 @@ describe('terminal subscribe buffering', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateMobileViewport: vi.fn().mockResolvedValue(false) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.subscribe', { @@ -241,7 +262,9 @@ describe('terminal subscribe buffering', () => { (msg) => messages.push(msg), { connectionId: 'conn-binary-limited', - sendBinary: (bytes) => binaryFrames.push(bytes) + sendBinary: (bytes) => { + binaryFrames.push(bytes) + } } ) @@ -273,9 +296,11 @@ describe('terminal subscribe buffering', () => { const cleanups = new Map void>() const runtime = stubRuntime({ resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), - readTerminal: vi - .fn() - .mockResolvedValue({ tail: ['line 120'], truncated: false, limited: true }), + readTerminal: vi.fn().mockResolvedValue({ + tail: ['line 120'], + truncated: false, + limited: true + }), serializeTerminalBuffer: vi.fn().mockResolvedValue({ data: 'serialized snapshot\r\n', cols: 100, @@ -298,7 +323,10 @@ describe('terminal subscribe buffering', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateMobileViewport: vi.fn().mockResolvedValue(false) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.subscribe', { @@ -309,7 +337,9 @@ describe('terminal subscribe buffering', () => { (msg) => messages.push(msg), { connectionId: 'conn-binary-serialized-limited', - sendBinary: (bytes) => binaryFrames.push(bytes) + sendBinary: (bytes) => { + binaryFrames.push(bytes) + } } ) @@ -375,7 +405,10 @@ describe('terminal subscribe buffering', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateMobileViewport: vi.fn().mockResolvedValue(false) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.subscribe', { @@ -386,7 +419,9 @@ describe('terminal subscribe buffering', () => { (msg) => messages.push(msg), { connectionId: 'conn-buffered', - sendBinary: (bytes) => binaryFrames.push(bytes) + sendBinary: (bytes) => { + binaryFrames.push(bytes) + } } ) @@ -446,7 +481,12 @@ describe('terminal subscribe buffering', () => { data: string cols: number rows: number - oscLinks?: { row: number; startCol: number; endCol: number; uri: string }[] + oscLinks?: { + row: number + startCol: number + endCol: number + uri: string + }[] }) => void)[] = [] const serializeTerminalBuffer = vi .fn() @@ -457,7 +497,12 @@ describe('terminal subscribe buffering', () => { data: string cols: number rows: number - oscLinks?: { row: number; startCol: number; endCol: number; uri: string }[] + oscLinks?: { + row: number + startCol: number + endCol: number + uri: string + }[] }>((resolve) => { restreamResolves.push(resolve) }) @@ -490,7 +535,10 @@ describe('terminal subscribe buffering', () => { sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateMobileViewport: vi.fn().mockResolvedValue({ updated: true, applied: true }) }) - const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const dispatcher = new RpcDispatcher({ + runtime, + methods: TERMINAL_METHODS + }) const dispatchPromise = dispatcher.dispatchStreaming( makeRequest('terminal.subscribe', { @@ -501,19 +549,38 @@ describe('terminal subscribe buffering', () => { (msg) => messages.push(msg), { connectionId: 'conn-stale-resize', - sendBinary: (bytes) => binaryFrames.push(bytes) + sendBinary: (bytes) => { + binaryFrames.push(bytes) + } } ) await vi.waitFor(() => expect(resizeListener).toBeDefined()) binaryFrames.splice(0) - resizeListener?.({ cols: 90, rows: 24, displayMode: 'auto', reason: 'apply-layout', seq: 2 }) - resizeListener?.({ cols: 100, rows: 24, displayMode: 'auto', reason: 'apply-layout', seq: 3 }) + resizeListener?.({ + cols: 90, + rows: 24, + displayMode: 'auto', + reason: 'apply-layout', + seq: 2 + }) + resizeListener?.({ + cols: 100, + rows: 24, + displayMode: 'auto', + reason: 'apply-layout', + seq: 3 + }) await vi.waitFor(() => expect(restreamResolves).toHaveLength(2)) const newerOscLinks = [{ row: 0, startCol: 4, endCol: 9, uri: 'https://example.com' }] - restreamResolves[1]?.({ data: 'newer', cols: 100, rows: 24, oscLinks: newerOscLinks }) + restreamResolves[1]?.({ + data: 'newer', + cols: 100, + rows: 24, + oscLinks: newerOscLinks + }) await vi.waitFor(() => expect( binaryFrames.some((frame) => { 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 new file mode 100644 index 00000000000..a7b1d883cf2 --- /dev/null +++ b/src/renderer/src/runtime/remote-runtime-terminal-frame-drop-resync.test.ts @@ -0,0 +1,174 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { + TerminalStreamOpcode, + decodeTerminalStreamFrame, + decodeTerminalStreamJson, + encodeTerminalStreamFrame, + encodeTerminalStreamJson, + encodeTerminalStreamText +} from '../../../shared/terminal-stream-protocol' +import { + getRemoteRuntimeTerminalMultiplexer, + resetRemoteRuntimeTerminalMultiplexersForTests +} from './remote-runtime-terminal-multiplexer' + +// Why: reproduces the silent frame-drop corruption. The server multiplex path +// drops Output frames when the websocket buffer is over its cap +// (encryptedBinaryReply returns false); the wire `seq` is a byte high-water, so +// a drop leaves a detectable gap. This harness drives the real client +// multiplexer through the same subscribe transport the app uses and forces a +// drop, asserting the client resyncs instead of rendering a corrupt tail. + +type SubscribeCallbacks = { + onResponse: (response: unknown) => void + onBinary?: (bytes: Uint8Array) => void + onError?: (error: { message: string }) => void + onClose?: () => void +} + +/** + * Minimal server that mimics src/main/runtime/rpc/methods/terminal.ts's multiplex + * output path: Output frames carry a monotonic byte high-water `seq`, and a + * SnapshotRequest is answered with an initial-style snapshot (no requestId). + */ +class FakeMultiplexServer { + private cursorBytes = 0 + private streamId = 0 + dropNextOutput = false + droppedFrames = 0 + private snapshotData = 'INITIAL' + + constructor( + private readonly toClient: (bytes: Uint8Array) => void, + private readonly onServerSideDrop?: () => void + ) {} + + /** Client -> server frames arrive here (Subscribe / SnapshotRequest / Input). */ + receive(bytes: Uint8Array): void { + const frame = decodeTerminalStreamFrame(bytes) + if (!frame) { + return + } + if (frame.opcode === TerminalStreamOpcode.Subscribe) { + const payload = decodeTerminalStreamJson<{ streamId: number }>(frame.payload) + this.streamId = payload?.streamId ?? 0 + this.sendSnapshot() + return + } + if (frame.opcode === TerminalStreamOpcode.SnapshotRequest) { + // Resync request: the server serializes the *current* buffer, so recovery + // includes everything the client missed. + this.snapshotData = 'RECOVERED' + this.sendSnapshot() + } + } + + private send(opcode: TerminalStreamOpcode, payload: Uint8Array, seq: number): void { + this.toClient(encodeTerminalStreamFrame({ opcode, streamId: this.streamId, seq, payload })) + } + + private sendSnapshot(): void { + this.send( + TerminalStreamOpcode.SnapshotStart, + encodeTerminalStreamJson({ cols: 80, rows: 24, seq: this.cursorBytes }), + 0 + ) + this.send(TerminalStreamOpcode.SnapshotChunk, encodeTerminalStreamText(this.snapshotData), 0) + this.send(TerminalStreamOpcode.SnapshotEnd, new Uint8Array(), 0) + } + + /** Emit an Output chunk, honoring simulated websocket backpressure. */ + output(text: string): void { + const startSeq = this.cursorBytes + this.cursorBytes += text.length + if (this.dropNextOutput) { + // encryptedBinaryReply returned false: frame is NOT sent. The byte + // high-water still advances (server keeps producing), so the next frame's + // seq jumps past what the client last saw. + this.dropNextOutput = false + this.droppedFrames += 1 + this.onServerSideDrop?.() + return + } + void startSeq + this.send(TerminalStreamOpcode.Output, encodeTerminalStreamText(text), this.cursorBytes) + } +} + +describe('remote terminal frame-drop resync', () => { + const unsubscribe = vi.fn() + let server: FakeMultiplexServer + + beforeEach(() => { + vi.clearAllMocks() + resetRemoteRuntimeTerminalMultiplexersForTests() + + const subscribe = vi.fn(async (_args: unknown, callbacks: SubscribeCallbacks) => { + server = new FakeMultiplexServer((bytes) => callbacks.onBinary?.(bytes)) + queueMicrotask(() => callbacks.onResponse({ ok: true, result: { type: 'ready' } })) + return { + unsubscribe, + sendBinary: (bytes: Uint8Array) => server.receive(bytes) + } + }) + + vi.stubGlobal('window', { + api: { runtimeEnvironments: { subscribe } } + }) + }) + + afterEach(() => { + vi.unstubAllGlobals() + }) + + async function subscribeClient(): Promise<{ data: string[]; snapshots: string[] }> { + const data: string[] = [] + const snapshots: string[] = [] + const multiplexer = getRemoteRuntimeTerminalMultiplexer('env-1') + await multiplexer.subscribeTerminal({ + terminal: 'terminal-1', + client: { id: 'desktop-1', type: 'desktop' }, + callbacks: { + onData: (chunk) => data.push(chunk), + onSnapshot: (chunk) => snapshots.push(chunk) + } + }) + // Let the initial snapshot round-trip settle. + await Promise.resolve() + await Promise.resolve() + return { data, snapshots } + } + + it('detects a dropped Output frame via the seq gap and resyncs', async () => { + const { data, snapshots } = await subscribeClient() + expect(snapshots).toEqual(['INITIAL']) + + server.output('aaa') + server.dropNextOutput = true + server.output('bbb') // dropped under backpressure — never reaches the client + server.output('ccc') // seq jumps past 'bbb', exposing the gap + + // Flush the client's resync SnapshotRequest -> server snapshot round-trip. + await Promise.resolve() + await Promise.resolve() + + // The corrupt tail ('ccc', which followed a gap) is NOT rendered as live data. + expect(data).toEqual(['aaa']) + expect(server.droppedFrames).toBe(1) + // Instead, a fresh authoritative snapshot recovers the terminal. + expect(snapshots).toEqual(['INITIAL', 'RECOVERED']) + }) + + it('passes contiguous output straight through without resyncing', async () => { + const { data, snapshots } = await subscribeClient() + + server.output('one') + server.output('two') + server.output('three') + await Promise.resolve() + await Promise.resolve() + + expect(data).toEqual(['one', 'two', 'three']) + expect(snapshots).toEqual(['INITIAL']) + }) +}) diff --git a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts index 70fb729b9db..4ea70856a23 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts @@ -77,6 +77,12 @@ type RemoteRuntimeMultiplexedTerminalState = { snapshotInfo: RemoteRuntimeSnapshotInfo | null initialSnapshotReceived: boolean pendingSnapshotRequest: RemoteRuntimeSnapshotRequest | null + // Why: Output frames carry a byte high-water `seq`; a jump past the expected + // next offset means the server dropped frames under websocket backpressure. + // Track it so a gap triggers a self-healing snapshot resync instead of + // silently rendering corrupt/missing output (frame-drop resync). + expectedSeq: number | undefined + resyncInFlight: boolean } type RemoteRuntimeSnapshotInfo = { @@ -149,7 +155,9 @@ class RemoteRuntimeTerminalMultiplexer { snapshotTarget: 'initial', snapshotInfo: null, initialSnapshotReceived: false, - pendingSnapshotRequest: null + pendingSnapshotRequest: null, + expectedSeq: undefined, + resyncInFlight: false } this.streams.set(streamId, state) @@ -334,10 +342,20 @@ class RemoteRuntimeTerminalMultiplexer { } if (frame.opcode === TerminalStreamOpcode.Output) { const data = decodeTerminalStreamText(frame.payload) - stream.callbacks.onData(data, { - seq: typeof frame.seq === 'number' && frame.seq > 0 ? frame.seq : undefined, - rawLength: data.length - }) + // Why: a resync snapshot is authoritative; drop live output that arrives + // while it is in flight so the corrupt post-gap tail is never rendered. + if (stream.resyncInFlight) { + return + } + const seq = typeof frame.seq === 'number' && frame.seq > 0 ? frame.seq : undefined + if (this.detectOutputGap(stream, seq, data.length)) { + this.requestResyncSnapshot(stream) + return + } + if (typeof seq === 'number') { + stream.expectedSeq = seq + } + stream.callbacks.onData(data, { seq, rawLength: data.length }) return } if (frame.opcode === TerminalStreamOpcode.SnapshotStart) { @@ -400,7 +418,12 @@ class RemoteRuntimeTerminalMultiplexer { clearPendingSnapshotRequest(stream) } clearSnapshot(stream) + // Why: the snapshot is the new authoritative byte 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') { + stream.expectedSeq = typeof info?.seq === 'number' ? info.seq : undefined + stream.resyncInFlight = false stream.initialSnapshotReceived = true stream.callbacks.onSubscribed?.() } @@ -408,6 +431,8 @@ class RemoteRuntimeTerminalMultiplexer { } if (frame.opcode === TerminalStreamOpcode.Error) { clearSnapshot(stream) + // Why: a failed resync must re-open the live path or output stalls forever. + stream.resyncInFlight = false const pendingSnapshotRequest = stream.pendingSnapshotRequest if (pendingSnapshotRequest) { clearPendingSnapshotRequest(stream) @@ -418,6 +443,43 @@ class RemoteRuntimeTerminalMultiplexer { } } + // Why: Output `seq` is the byte high-water at the end of a chunk, so a chunk + // that begins after the last high-water (startSeq > expectedSeq) means the + // server dropped intervening frames under backpressure. Only flag a gap when + // both offsets are known, and never on the first seq (nothing to compare to). + private detectOutputGap( + stream: RemoteRuntimeMultiplexedTerminalState, + seq: number | undefined, + rawLength: number + ): boolean { + if (typeof seq !== 'number' || typeof stream.expectedSeq !== 'number') { + return false + } + const startSeq = seq - rawLength + 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. + private requestResyncSnapshot(stream: RemoteRuntimeMultiplexedTerminalState): void { + if (stream.resyncInFlight) { + return + } + stream.resyncInFlight = true + stream.expectedSeq = undefined + const sent = this.sendFrame( + stream.streamId, + TerminalStreamOpcode.SnapshotRequest, + encodeTerminalStreamJson({ scrollbackRows: undefined }) + ) + if (!sent) { + // Transport is down; the reconnect path re-subscribes from scratch. + stream.resyncInFlight = false + } + } + private requestSnapshot( stream: RemoteRuntimeMultiplexedTerminalState, opts?: { scrollbackRows?: number }