diff --git a/src/main/runtime/rpc/methods/terminal.ts b/src/main/runtime/rpc/methods/terminal.ts index f30e1b2f591..942d4664f1b 100644 --- a/src/main/runtime/rpc/methods/terminal.ts +++ b/src/main/runtime/rpc/methods/terminal.ts @@ -75,6 +75,9 @@ type TerminalMultiplexStream = { ackInFlightBytes: number buffering: boolean ackPendingOutput: TerminalOutputFrameChunk[] + ackPendingOutputBytes: number + ackPendingOutputOverflowed: boolean + ackRecoverySnapshotInFlight: boolean pendingOutput: TerminalOutputChunk[] pendingOutputBytes: number pendingOutputOverflowed: boolean @@ -249,6 +252,26 @@ function appendPendingMultiplexOutput( stream.pendingOutputOverflowed ||= trimmed.overflowed } +function appendAckPendingOutput( + stream: TerminalMultiplexStream, + chunk: TerminalOutputFrameChunk +): void { + stream.ackPendingOutput.push(chunk) + stream.ackPendingOutputBytes += chunk.bytes.byteLength + let omittedChunkCount = 0 + while ( + stream.ackPendingOutputBytes > TERMINAL_MULTIPLEX_PENDING_MAX_BYTES && + omittedChunkCount < stream.ackPendingOutput.length + ) { + stream.ackPendingOutputBytes -= stream.ackPendingOutput[omittedChunkCount]!.bytes.byteLength + omittedChunkCount += 1 + } + if (omittedChunkCount > 0) { + stream.ackPendingOutput.splice(0, omittedChunkCount) + stream.ackPendingOutputOverflowed = true + } +} + function trimPendingOutputToBudget( pendingOutput: (string | TerminalOutputChunk)[], pendingOutputBytes: number @@ -949,15 +972,63 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ return } if ( + stream.ackPendingOutputOverflowed || stream.ackPendingOutput.length > 0 || !canSendAckGatedOutput(stream, chunk.bytes.byteLength) ) { - stream.ackPendingOutput.push(chunk) + appendAckPendingOutput(stream, chunk) return } sendAckGatedOutput(stream, chunk) } + const sendAckRecoverySnapshot = async (stream: TerminalMultiplexStream): Promise => { + if ( + closed || + streams.get(stream.streamId) !== stream || + stream.ackRecoverySnapshotInFlight + ) { + return + } + stream.ackRecoverySnapshotInFlight = true + try { + const serialized = await serializeBudgetedRequestedSnapshot(runtime, stream.ptyId, 0) + if (closed || streams.get(stream.streamId) !== stream) { + return + } + const size = runtime.getTerminalSize(stream.ptyId) + const displayMode = runtime.getMobileDisplayMode(stream.ptyId) + // Why: dropped ACK-pending output means live frames are no longer a + // complete replay. Send a fresh model snapshot before resuming output. + sendSnapshotFrames((opcode, payload) => sendFrame(stream.streamId, opcode, payload), { + kind: 'scrollback', + cols: serialized?.cols ?? size?.cols ?? 80, + rows: serialized?.rows ?? size?.rows ?? 24, + displayMode, + reason: 'ack-pending-overflow', + seq: serialized?.seq, + source: serialized?.source, + truncated: true, + truncatedByByteBudget: serialized?.truncatedByByteBudget, + data: serialized?.data ?? '' + }) + stream.ackPendingOutputOverflowed = false + } catch (error) { + sendStreamError( + stream.streamId, + error instanceof Error ? error.message : 'Remote terminal recovery snapshot failed.' + ) + } finally { + if (streams.get(stream.streamId) === stream) { + stream.ackRecoverySnapshotInFlight = false + flushAckPendingOutput(stream) + } + } + } const flushAckPendingOutput = (stream: TerminalMultiplexStream): void => { + if (stream.ackPendingOutputOverflowed) { + void sendAckRecoverySnapshot(stream) + return + } let flushed = 0 while ( flushed < stream.ackPendingOutput.length && @@ -968,6 +1039,10 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ } if (flushed > 0) { stream.ackPendingOutput.splice(0, flushed) + stream.ackPendingOutputBytes = stream.ackPendingOutput.reduce( + (total, pending) => total + pending.bytes.byteLength, + 0 + ) } } const acknowledgeOutput = (stream: TerminalMultiplexStream, bytes: number): void => { @@ -989,6 +1064,9 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ ackTotalInFlightBytes = Math.max(0, ackTotalInFlightBytes - stream.ackInFlightBytes) stream.ackInFlightBytes = 0 stream.ackPendingOutput = [] + stream.ackPendingOutputBytes = 0 + stream.ackPendingOutputOverflowed = false + stream.ackRecoverySnapshotInFlight = false stream.unsubscribeData() stream.unsubscribeResize() stream.unsubscribeFit() @@ -1199,6 +1277,9 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ ackInFlightBytes: 0, buffering: true, ackPendingOutput: [], + ackPendingOutputBytes: 0, + ackPendingOutputOverflowed: false, + ackRecoverySnapshotInFlight: false, pendingOutput: [], pendingOutputBytes: 0, pendingOutputOverflowed: false, diff --git a/src/main/runtime/rpc/terminal-multiplex.test.ts b/src/main/runtime/rpc/terminal-multiplex.test.ts index 169ab9a3ed7..608d5ae1d11 100644 --- a/src/main/runtime/rpc/terminal-multiplex.test.ts +++ b/src/main/runtime/rpc/terminal-multiplex.test.ts @@ -489,6 +489,176 @@ describe('terminal multiplex RPC', () => { await dispatchPromise }) + it('caps stalled ACK output and snapshots before resuming retained tail frames', async () => { + const messages: string[] = [] + const binaryFrames: Uint8Array[] = [] + const handlers = new Map< + number, + (frame: NonNullable>) => void + >() + const cleanups = new Map void>() + const dataListenerRef: { + current?: (data: string, meta?: { seq?: number; rawLength?: number }) => void + } = {} + const runtime = stubRuntime({ + resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), + readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }), + serializeTerminalBuffer: vi + .fn() + .mockResolvedValueOnce({ data: 'initial snapshot', cols: 120, rows: 40 }) + .mockResolvedValue({ data: 'recovered snapshot', cols: 120, rows: 40, seq: 99 }), + 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, meta?: { seq?: number; rawLength?: number }) => 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) + }), + cleanupSubscription: vi.fn((id: string) => { + const cleanup = cleanups.get(id) + cleanups.delete(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-ack-overflow', + 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: 17, + terminal: 'terminal-1', + client: { id: 'desktop-1', type: 'desktop' }, + viewport: { cols: 120, rows: 40 }, + capabilities: { ackOutput: 1 } + }) + }) + )! + ) + await vi.waitFor(() => + expect(messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe(true) + ) + binaryFrames.splice(0) + + const output = 'x'.repeat(3 * 1024 * 1024) + dataListenerRef.current?.(output, { seq: output.length, rawLength: output.length }) + + const initialOutputFrames = binaryFrames + .map((frame) => decodeTerminalStreamFrame(frame)) + .filter((frame) => frame?.opcode === TerminalStreamOpcode.Output) + const initialBytes = initialOutputFrames.reduce( + (total, frame) => total + (frame?.payload.byteLength ?? 0), + 0 + ) + expect(initialBytes).toBeLessThanOrEqual(512 * 1024) + + handlers.get(17)?.( + decodeTerminalStreamFrame( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Input, + streamId: 17, + seq: 2, + payload: encodeTerminalStreamText('still interactive\r') + }) + )! + ) + await vi.waitFor(() => + expect(runtime.sendTerminal).toHaveBeenCalledWith('terminal-1', { + text: 'still interactive\r', + enter: false, + interrupt: false + }) + ) + + binaryFrames.splice(0) + handlers.get(17)?.( + decodeTerminalStreamFrame( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Ack, + streamId: 17, + seq: 3, + payload: encodeTerminalStreamJson({ bytes: initialBytes }) + }) + )! + ) + + await vi.waitFor(() => + expect( + binaryFrames + .map((frame) => decodeTerminalStreamFrame(frame)) + .some((frame) => { + if (frame?.opcode !== TerminalStreamOpcode.SnapshotStart) { + return false + } + const payload = decodeTerminalStreamJson<{ reason?: string }>(frame.payload) + return payload?.reason === 'ack-pending-overflow' + }) + ).toBe(true) + ) + const drainFrames = binaryFrames.map((frame) => decodeTerminalStreamFrame(frame)) + const recoveryStartIndex = drainFrames.findIndex((frame) => { + if (frame?.opcode !== TerminalStreamOpcode.SnapshotStart) { + return false + } + const payload = decodeTerminalStreamJson<{ reason?: string }>(frame.payload) + return payload?.reason === 'ack-pending-overflow' + }) + const firstOutputAfterAckIndex = drainFrames.findIndex( + (frame) => frame?.opcode === TerminalStreamOpcode.Output + ) + expect(recoveryStartIndex).toBeGreaterThanOrEqual(0) + expect(firstOutputAfterAckIndex).toBeGreaterThan(recoveryStartIndex) + expect( + drainFrames + .filter((frame) => frame?.opcode === TerminalStreamOpcode.SnapshotChunk) + .map((frame) => (frame ? decodeTerminalStreamText(frame.payload) : '')) + .join('') + ).toBe('recovered snapshot') + + const outputBytesAfterRecovery = drainFrames + .filter((frame) => frame?.opcode === TerminalStreamOpcode.Output) + .reduce((total, frame) => total + (frame?.payload.byteLength ?? 0), 0) + expect(outputBytesAfterRecovery).toBeLessThanOrEqual(256 * 1024) + + runtime.cleanupSubscription('terminal-multiplex:conn-ack-overflow') + await dispatchPromise + }) + it('marks multiplex fallback snapshots truncated when the uncursored read is limited', async () => { const messages: string[] = [] const binaryFrames: Uint8Array[] = []