From 0cbcec01eec360bdca567912e2c4ca7fd598e2ea Mon Sep 17 00:00:00 2001 From: Neil <4138956+nwparker@users.noreply.github.com> Date: Sun, 7 Jun 2026 05:50:51 -0700 Subject: [PATCH] Cap terminal multiplex backlog by bytes (#4785) --- src/main/runtime/rpc/methods/terminal.ts | 47 ++++---- .../runtime/rpc/terminal-multiplex.test.ts | 101 ++++++++++++++++++ 2 files changed, 128 insertions(+), 20 deletions(-) diff --git a/src/main/runtime/rpc/methods/terminal.ts b/src/main/runtime/rpc/methods/terminal.ts index 104c0f3b466..10f775bf024 100644 --- a/src/main/runtime/rpc/methods/terminal.ts +++ b/src/main/runtime/rpc/methods/terminal.ts @@ -26,7 +26,10 @@ const REQUESTED_SNAPSHOT_BYTE_BUDGET = 2 * 1024 * 1024 const TERMINAL_STREAM_CHUNK_BYTES = 48 * 1024 const TERMINAL_OUTPUT_FLUSH_MS = 5 const TERMINAL_OUTPUT_BATCH_MAX_CHARS = 64 * 1024 -const TERMINAL_MULTIPLEX_PENDING_MAX_CHARS = 256 * 1024 +// Why: pending output is held for later binary frames, so cap the encoded +// payload bytes rather than UTF-16 code units. +const TERMINAL_MULTIPLEX_PENDING_MAX_BYTES = 256 * 1024 +const terminalStreamTextEncoder = new TextEncoder() let nextTerminalStreamId = 1 type SnapshotFrameOptions = { @@ -66,7 +69,7 @@ type TerminalMultiplexStream = { isMobile: boolean buffering: boolean pendingOutput: TerminalOutputChunk[] - pendingOutputChars: number + pendingOutputBytes: number pendingOutputOverflowed: boolean outputBatcher: ReturnType unsubscribeData: () => void @@ -182,29 +185,33 @@ function appendPendingMultiplexOutput( meta?: { seq?: number; rawLength?: number } ): void { stream.pendingOutput.push({ data, meta }) - stream.pendingOutputChars += data.length - const trimmed = trimPendingOutputToBudget(stream.pendingOutput, stream.pendingOutputChars) - stream.pendingOutputChars = trimmed.chars + stream.pendingOutputBytes += terminalStreamByteLength(data) + const trimmed = trimPendingOutputToBudget(stream.pendingOutput, stream.pendingOutputBytes) + stream.pendingOutputBytes = trimmed.bytes stream.pendingOutputOverflowed ||= trimmed.overflowed } function trimPendingOutputToBudget( pendingOutput: (string | TerminalOutputChunk)[], - pendingOutputChars: number -): { chars: number; overflowed: boolean } { + pendingOutputBytes: number +): { bytes: number; overflowed: boolean } { let omittedChunkCount = 0 while ( - pendingOutputChars > TERMINAL_MULTIPLEX_PENDING_MAX_CHARS && + pendingOutputBytes > TERMINAL_MULTIPLEX_PENDING_MAX_BYTES && omittedChunkCount < pendingOutput.length ) { const chunk = pendingOutput[omittedChunkCount] - pendingOutputChars -= typeof chunk === 'string' ? chunk.length : chunk.data.length + pendingOutputBytes -= terminalStreamByteLength(typeof chunk === 'string' ? chunk : chunk.data) omittedChunkCount += 1 } if (omittedChunkCount > 0) { pendingOutput.splice(0, omittedChunkCount) } - return { chars: pendingOutputChars, overflowed: omittedChunkCount > 0 } + return { bytes: pendingOutputBytes, overflowed: omittedChunkCount > 0 } +} + +function terminalStreamByteLength(data: string): number { + return terminalStreamTextEncoder.encode(data).byteLength } function isTerminalReadPayloadIncomplete(read: { truncated: boolean; limited?: boolean }): boolean { @@ -238,7 +245,7 @@ async function serializeBudgetedRequestedSnapshot( if (!serialized) { return null } - const bytes = new TextEncoder().encode(serialized.data).byteLength + const bytes = terminalStreamByteLength(serialized.data) if (bytes <= REQUESTED_SNAPSHOT_BYTE_BUDGET || rows === 0) { return { ...serialized, @@ -297,7 +304,7 @@ async function serializeBudgetedMobileSnapshot( if (!serialized) { return null } - const bytes = new TextEncoder().encode(serialized.data).byteLength + const bytes = terminalStreamByteLength(serialized.data) if (bytes <= MOBILE_SNAPSHOT_BYTE_BUDGET || rows === 0) { return { ...serialized, @@ -931,7 +938,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ // Why: the overflowed tail is newer than the first snapshot. Retry // so hidden restore receives a current terminal image instead of null. stream.pendingOutput.splice(0) - stream.pendingOutputChars = 0 + stream.pendingOutputBytes = 0 stream.pendingOutputOverflowed = false serialized = await serializeBudgetedRequestedSnapshot( runtime, @@ -984,7 +991,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ stream.outputBatcher.push(chunk.data, chunk.meta) } } - stream.pendingOutputChars = 0 + stream.pendingOutputBytes = 0 stream.pendingOutputOverflowed = false stream.outputBatcher.flush() } @@ -1027,7 +1034,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ isMobile, buffering: true, pendingOutput: [], - pendingOutputChars: 0, + pendingOutputBytes: 0, pendingOutputOverflowed: false, outputBatcher: createTerminalOutputBatcher((data, meta) => { sendFrame( @@ -1143,7 +1150,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ for (const chunk of stream.pendingOutput.splice(0)) { stream.outputBatcher.push(chunk.data, chunk.meta) } - stream.pendingOutputChars = 0 + stream.pendingOutputBytes = 0 stream.pendingOutputOverflowed = false stream.outputBatcher.flush() @@ -1310,7 +1317,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ let closed = false let buffering = true const pendingOutput: string[] = [] - let pendingOutputChars = 0 + let pendingOutputBytes = 0 let unsubscribeData = (): void => {} let unsubscribeResize = (): void => {} let unsubscribeFit = (): void => {} @@ -1416,8 +1423,8 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ } if (buffering) { pendingOutput.push(data) - pendingOutputChars += data.length - pendingOutputChars = trimPendingOutputToBudget(pendingOutput, pendingOutputChars).chars + pendingOutputBytes += terminalStreamByteLength(data) + pendingOutputBytes = trimPendingOutputToBudget(pendingOutput, pendingOutputBytes).bytes return } outputBatcher?.push(data) @@ -1468,7 +1475,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ for (const item of pendingOutput.splice(0)) { outputBatcher.push(item) } - pendingOutputChars = 0 + pendingOutputBytes = 0 outputBatcher.flush() unsubscribeResize = runtime.subscribeToTerminalResize(ptyId, (event) => { diff --git a/src/main/runtime/rpc/terminal-multiplex.test.ts b/src/main/runtime/rpc/terminal-multiplex.test.ts index 0578b651661..36930b4c5f8 100644 --- a/src/main/runtime/rpc/terminal-multiplex.test.ts +++ b/src/main/runtime/rpc/terminal-multiplex.test.ts @@ -882,6 +882,107 @@ describe('terminal multiplex RPC', () => { } }) + it('bounds multibyte live output by encoded bytes while a multiplex snapshot is loading', async () => { + vi.useFakeTimers() + try { + const messages: string[] = [] + const binaryFrames: Uint8Array[] = [] + const handlers = new Map< + number, + (frame: NonNullable>) => void + >() + const cleanups = new Map void>() + const dataListenerRef: { current?: (data: string) => void } = {} + let resolveSnapshot: (value: { data: string; cols: number; rows: number }) => void = () => {} + const runtime = stubRuntime({ + resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), + readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }), + serializeTerminalBuffer: vi.fn( + () => + new Promise<{ data: string; cols: number; rows: number }>((resolve) => { + resolveSnapshot = resolve + }) + ), + getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }), + getMobileDisplayMode: vi.fn().mockReturnValue('auto'), + getLayout: vi.fn().mockReturnValue({ seq: 1 }), + subscribeToTerminalData: vi.fn((_: string, listener: (data: string) => void) => { + dataListenerRef.current = listener + 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) + }), + waitForTerminal: vi.fn(() => new Promise(() => {})), + sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), + updateDesktopViewport: vi.fn().mockResolvedValue(true) + }) + 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), + registerBinaryStreamHandler: (streamId, handler) => { + handlers.set(streamId, handler) + return () => handlers.delete(streamId) + } + } + ) + + await vi.waitFor(() => + 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: 10, + terminal: 'terminal-1', + client: { id: 'desktop-1', type: 'desktop' }, + viewport: { cols: 120, rows: 40 } + }) + }) + )! + ) + await vi.waitFor(() => expect(dataListenerRef.current).toBeDefined()) + + for (let index = 0; index < 400; index += 1) { + dataListenerRef.current?.(`${String(index).padStart(3, '0')}${'界'.repeat(341)}`) + } + await vi.waitFor(() => expect(runtime.serializeTerminalBuffer).toHaveBeenCalled()) + resolveSnapshot({ data: '', cols: 120, rows: 40 }) + await vi.waitFor(() => + expect(messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe(true) + ) + await vi.runOnlyPendingTimersAsync() + + const output = binaryFrames + .map((frame) => decodeTerminalStreamFrame(frame)) + .filter((frame) => frame?.opcode === TerminalStreamOpcode.Output) + .map((frame) => (frame ? decodeTerminalStreamText(frame.payload) : '')) + .join('') + expect(new TextEncoder().encode(output).byteLength).toBeLessThanOrEqual(256 * 1024) + expect(output).not.toContain('000') + expect(output).toContain('399') + + cleanups.get('terminal-multiplex:conn-buffered-multibyte')?.() + await dispatchPromise + } finally { + vi.useRealTimers() + } + }) + it('retries requested snapshots after live output overflows during serialization', async () => { vi.useFakeTimers() try {