From 72a2d7bc7d560c27fe2dd649cc45a9cd3573561b Mon Sep 17 00:00:00 2001 From: OrcaWin Date: Wed, 22 Jul 2026 22:58:38 -0700 Subject: [PATCH] fix(terminal): bound parse-deferred remote flow (#10012) Credits remote terminal output only after parse or intentional discard, with bounded adaptive windows, fair draining, recovery cleanup, and RTT/parser benchmarks. --- config/reliability-gates.jsonc | 56 +++- src/main/runtime/rpc/methods/terminal.ts | 126 +++++-- ...minal-multiplex-flow-control.bench.test.ts | 208 ++++++++++++ .../terminal-multiplex-round-robin.test.ts | 67 ++++ .../rpc/terminal-multiplex-round-robin.ts | 41 +++ .../runtime/rpc/terminal-multiplex.test.ts | 307 +++++++++++++++--- .../terminal-pane/pty-connection.test.ts | 9 +- .../terminal-pane/pty-connection.ts | 2 +- .../terminal-pty-ack-gate.test.ts | 8 +- .../terminal-pane/terminal-pty-ack-gate.ts | 3 - .../terminal-delivery-credit.test.ts | 22 ++ .../pane-manager/terminal-delivery-credit.ts | 39 ++- .../remote-runtime-terminal-multiplexer.ts | 93 +++++- ...untime-terminal-parse-backpressure.test.ts | 261 ++++++++++++++- .../runtime/runtime-terminal-stream.test.ts | 22 ++ src/shared/terminal-multiplex-flow-control.ts | 10 + 16 files changed, 1157 insertions(+), 117 deletions(-) create mode 100644 src/main/runtime/rpc/terminal-multiplex-flow-control.bench.test.ts create mode 100644 src/main/runtime/rpc/terminal-multiplex-round-robin.test.ts create mode 100644 src/main/runtime/rpc/terminal-multiplex-round-robin.ts create mode 100644 src/shared/terminal-multiplex-flow-control.ts diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index fa180e87b..fad227b81 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -4778,16 +4778,18 @@ "macos" ], "coveredProviders": [], - "coverageNotes": "Local macOS evidence over the runtime-RPC stream budgets on main@1282f5c2d. PR #5824 adds a platform-neutral mobile decision gate proving chat-covered terminal streams pause and resume only after the mounted WebView is ready; live Android restore evidence remains required. The pending stack adds byte-exact 512KB/2MB/256KB/48KB budget assertions; legacy JSON subscribe parity remains undecided.", + "coverageNotes": "Local macOS evidence covers runtime-RPC stream budgets plus paired-renderer parse/discard credit. Deferred credit is shared by local and remote transports, batches ACKs at 192 KiB or 4 ms, grows per-stream windows from 512 KiB to 2 MiB and aggregate windows from 2 MiB to 8 MiB, bounds queued output to 256 KiB per stream, and caps each multiplex connection at 32 active or pending streams for an 8 MiB aggregate pending-output ceiling. Deterministic tests cover replay ordering, stale generations, malformed frames, hidden panes, queue eviction, disposal, send/recovery failure, repeated pending-slot replacement, reconnect, and round-robin fairness. The opt-in benchmark covers 1/20/100 ms RTT and 1/4/8 viewers, exact protocol-frame allocations, scheduler CPU, and measured @xterm/headless parser CPU/retained heap. Live Android restore evidence, browser/WebGL parser measurements, and legacy JSON subscribe parity remain required.", "motivatingLinks": [ "https://github.com/stablyai/orca/pull/6951", "https://github.com/stablyai/orca/pull/6955", "https://github.com/stablyai/orca/pull/7009" ], - "invariant": "Runtime and mobile terminal subscriptions must cap initial snapshots, live output buffered while snapshots load, chunk sizes, and batch sizes; a terminal covered by native chat must have no live output subscription and must restore from fresh scrollback when revealed, while preserving output order, input locks, resize/driver events, and fallback parity or explicit fallback deprecation.", - "oracle": "The current executable slice asserts mobile initial snapshots downgrade until they fit <=512KB, requested binary snapshots downgrade until they fit <=2MB, binary live output queued while the initial snapshot loads stays <=256KB while preserving the newest tail, large binary output is split into <=48KB frames, output bursts are coalesced before emit, aborts do not register stale listeners, and stale mobile resize re-stream completions are dropped. The mobile native-chat decision test asserts an active stream pauses while covered and resumes only for a ready active terminal. JSON fallback parity and live Android scrollback restoration remain explicit gaps.", + "invariant": "Runtime and mobile terminal subscriptions must cap initial snapshots, live output buffered while snapshots load, chunk sizes, batches, and aggregate in-flight credit. ACK means the renderer parsed the bytes or intentionally discarded them; receipt-time ACK is forbidden. Every replay, stale-generation, malformed-frame, hidden-pane, eviction, disposal, error, and reconnect path must settle credit exactly once so streams neither leak memory nor stall. A terminal covered by native chat must restore from fresh scrollback when revealed, while preserving output order, input locks, resize/driver events, fairness, and fallback parity or explicit fallback deprecation.", + "oracle": "Assert mobile initial snapshots downgrade until they fit <=512KB, requested binary snapshots downgrade until they fit <=2MB, live output queued while snapshots load stays <=256KB per stream, large output splits into <=48KB frames, and output bursts coalesce. Feed paired output through the xterm parse callback and prove ACK is deferred until parse or intentional discard, then inject stale generation, malformed/transformed frames, replay failure, queue eviction, hidden panes, pane disposal, ACK send failure, recovery serialization failure, and reconnect races; assert ordered replay and exactly-once credit settlement. Fill the aggregate window across bulk and interactive streams, ACK once, and prove round-robin progress. Run the opt-in 64 MiB/viewer RTT matrix and enforce bounded 8 MiB aggregate in-flight memory, >7 MiB/s/viewer at 100 ms RTT, and <200 ms completion spread. JSON fallback parity and live Android scrollback restoration remain explicit gaps.", "commands": [ "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/terminal-subscribe-buffer.test.ts src/main/runtime/rpc/terminal-output-batching.test.ts src/main/runtime/rpc/terminal-multiplex.test.ts", + "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/terminal-subscribe-buffer.test.ts src/main/runtime/rpc/terminal-output-batching.test.ts src/main/runtime/rpc/terminal-multiplex.test.ts src/renderer/src/components/terminal-pane/pty-connection.test.ts src/renderer/src/components/terminal-pane/terminal-pty-ack-gate.test.ts src/renderer/src/lib/pane-manager/terminal-delivery-credit.test.ts src/renderer/src/runtime/remote-runtime-terminal-parse-backpressure.test.ts src/renderer/src/runtime/runtime-terminal-stream.test.ts --maxWorkers=1", + "ORCA_TERMINAL_PERF_BENCH=1 pnpm exec vitest run --config config/vitest.config.ts --disableConsoleIntercept src/main/runtime/rpc/terminal-multiplex-flow-control.bench.test.ts", "pnpm --dir mobile exec vitest run --root .. mobile/src/session/mobile-native-chat-terminal-stream.test.ts", "pnpm --dir mobile exec vitest run --root .. mobile/src/session/use-mobile-native-chat-terminal-stream.test.ts" ], @@ -4795,6 +4797,12 @@ "src/main/runtime/rpc/terminal-subscribe-buffer.test.ts", "src/main/runtime/rpc/terminal-output-batching.test.ts", "src/main/runtime/rpc/terminal-multiplex.test.ts", + "src/main/runtime/rpc/terminal-multiplex-flow-control.bench.test.ts", + "src/renderer/src/components/terminal-pane/pty-connection.test.ts", + "src/renderer/src/components/terminal-pane/terminal-pty-ack-gate.test.ts", + "src/renderer/src/lib/pane-manager/terminal-delivery-credit.test.ts", + "src/renderer/src/runtime/remote-runtime-terminal-parse-backpressure.test.ts", + "src/renderer/src/runtime/runtime-terminal-stream.test.ts", "mobile/src/session/mobile-native-chat-terminal-stream.test.ts", "mobile/src/session/use-mobile-native-chat-terminal-stream.test.ts" ], @@ -4819,7 +4827,33 @@ "assertions": [ "requested snapshots fall back smaller when serialized data exceeds the send budget", "oversized live output frames are bounded for subscribed binary streams", - "multibyte live output flushes when encoded bytes reach the batch budget" + "multibyte live output flushes when encoded bytes reach the batch budget", + "adaptive credit grows only after ACK, stays globally bounded, and drains pending streams round-robin", + "send and recovery serialization failures detach once instead of leaking credit or retrying forever", + "32 active or pending slots cap aggregate queued output and repeated pending-slot subscribe cancels its older waiter" + ] + }, + { + "file": "src/renderer/src/lib/pane-manager/terminal-delivery-credit.test.ts", + "assertions": [ + "nested synchronous deliveries restore the outer credit owner", + "unclaimed intentional discards settle automatically while every claimed scheduler child must settle before the parent credits" + ] + }, + { + "file": "src/renderer/src/runtime/remote-runtime-terminal-parse-backpressure.test.ts", + "assertions": [ + "paired renderer ACK waits for xterm parse completion or explicit discard", + "192 KiB parsed output batches into one ACK while the 4 ms timer releases interactive output", + "malformed frames, malformed transformed output, disposal, late parse, renderer delivery failure, and ACK transport failure release credit or close the owning stream without reordering output" + ] + }, + { + "file": "src/main/runtime/rpc/terminal-multiplex-flow-control.bench.test.ts", + "assertions": [ + "one through eight viewers stay within the 8 MiB aggregate adaptive window", + "the 100 ms RTT model sustains more than 7 MiB/s per viewer with less than 200 ms fairness spread", + "the opt-in benchmark reports RTT throughput, scheduler CPU time, exact protocol frame allocations, completion spread, and measured @xterm/headless parser CPU and retained heap" ] }, { @@ -4856,6 +4890,15 @@ "result": "passed", "durationSeconds": 0.2, "summary": "The focused mobile native-chat suite passed with 3 terminal-stream lifecycle assertions in the staged PR #5824 worktree." + }, + { + "date": "2026-07-22", + "runner": "local", + "platform": "macos", + "command": "ORCA_TERMINAL_PERF_BENCH=1 pnpm exec vitest run --config config/vitest.config.ts --disableConsoleIntercept src/main/runtime/rpc/terminal-multiplex-flow-control.bench.test.ts", + "result": "passed", + "durationSeconds": 0.91, + "summary": "The 1/20/100 ms RTT x 1/4/8 viewer matrix stayed at or below 8 MiB in flight with zero completion spread. At 100 ms it modeled 18.8 MiB/s per viewer for 1-4 viewers and 9.7 MiB/s for 8 viewers. Measured @xterm/headless parsing was 26.7/63.6/95.3 aggregate MiB/s for 1/4/8 viewers, with 84.4/236.5/336.0 ms CPU and 2893/13409/28991 KiB retained heap for 4 MiB per viewer." } ], "runtimeBudget": { @@ -4872,7 +4915,7 @@ }, "performanceBudget": { "required": true, - "evidence": "This gate is the byte and batching budget for runtime/mobile terminal streaming." + "evidence": "Parsed/discarded credit uses 192 KiB/4 ms ACK batching, 512 KiB-to-2 MiB adaptive per-stream windows, a 2 MiB-to-8 MiB aggregate window, <=48 KiB output frames, <=256 KiB queued output per stream, and <=32 streams per connection (8 MiB aggregate pending output). The 64 MiB/viewer model gate requires >7 MiB/s/viewer at 100 ms RTT, <200 ms completion spread, and aggregate in-flight bytes <=8 MiB. The 2026-07-22 run modeled 9.7 MiB/s/viewer at 100 ms with eight viewers and measured @xterm/headless at 95.3 aggregate MiB/s, 336.0 ms parser CPU, and 28991 KiB retained heap for eight 4 MiB viewers." }, "promotionCriteria": [ "Gate binary multiplex first.", @@ -4881,7 +4924,8 @@ ], "knownGaps": [ "The pure mobile decision gate does not yet prove live Android WebView scrollback restore after a chat toggle.", - "Legacy JSON subscribe parity is undecided." + "Legacy JSON subscribe parity is undecided.", + "The parser measurement uses @xterm/headless; browser renderer/WebGL CPU, GPU, and allocation behavior still need packaged-app performance evidence." ], "demotionRule": "Cannot promote while a supported stream path has uncapped snapshot or live-output buffering." }, diff --git a/src/main/runtime/rpc/methods/terminal.ts b/src/main/runtime/rpc/methods/terminal.ts index 91f2444cd..487f79f26 100644 --- a/src/main/runtime/rpc/methods/terminal.ts +++ b/src/main/runtime/rpc/methods/terminal.ts @@ -42,17 +42,20 @@ import { navigationTargetsHost, resolveRuntimeNavigationTarget } from '../../../../shared/runtime-navigation' +import { + TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES, + TERMINAL_MULTIPLEX_ACK_STREAM_MAX_WINDOW_BYTES, + TERMINAL_MULTIPLEX_ACK_TOTAL_INITIAL_WINDOW_BYTES, + TERMINAL_MULTIPLEX_ACK_TOTAL_MAX_WINDOW_BYTES, + TERMINAL_MULTIPLEX_MAX_STREAMS_PER_CONNECTION, + TERMINAL_MULTIPLEX_PENDING_MAX_BYTES, + TERMINAL_OUTPUT_BATCH_MAX_BYTES, + TERMINAL_STREAM_CHUNK_BYTES +} from '../../../../shared/terminal-multiplex-flow-control' +import { drainTerminalMultiplexRoundRobin } from '../terminal-multiplex-round-robin' const REQUESTED_SNAPSHOT_BYTE_BUDGET = 2 * 1024 * 1024 -const TERMINAL_STREAM_CHUNK_BYTES = 48 * 1024 const TERMINAL_OUTPUT_FLUSH_MS = 5 -// Why: output batches become binary stream payloads; byte size is the transport cost. -const TERMINAL_OUTPUT_BATCH_MAX_BYTES = 64 * 1024 -// Why: remote clients can apply output pressure without pausing runtime PTY ingestion. -const TERMINAL_MULTIPLEX_ACK_STREAM_HIGH_WATER_BYTES = 512 * 1024 -const TERMINAL_MULTIPLEX_ACK_TOTAL_HIGH_WATER_BYTES = 2 * 1024 * 1024 -// Why: pending output becomes binary frames, so cap encoded payload bytes, not UTF-16 code units. -const TERMINAL_MULTIPLEX_PENDING_MAX_BYTES = 256 * 1024 const TERMINAL_QUERY_REPLAY_MAX_CHARS = 16 * 1024 // Why: bound initial subscribe latency; readiness after this deadline triggers an in-stream recovery snapshot. const MOBILE_RENDERER_MOUNT_READY_TIMEOUT_MS = 3_000 @@ -102,6 +105,7 @@ type TerminalMultiplexStream = { isMobile: boolean ackOutput: boolean ackInFlightBytes: number + ackWindowBytes: number supportsDesktopViewportClaims: boolean desktopClaimTail: Promise // Whether THIS stream registered the width driver, so detach won't release a peer stream's floor. @@ -1603,6 +1607,8 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ const streams = new Map() const pendingPtyWaitControllers = new Map>() let ackTotalInFlightBytes = 0 + let ackTotalWindowBytes = TERMINAL_MULTIPLEX_ACK_TOTAL_INITIAL_WINDOW_BYTES + let ackFlushCursorStreamId: number | null = null let resolveMultiplex = (): void => {} const multiplexClosed = new Promise((resolve) => { resolveMultiplex = resolve @@ -1619,10 +1625,21 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ // Why: a seq-less Output chunk must carry sentinel 0, not the control-frame cursor, or it poisons the client's frame-drop tracker. const resolvedSeq = typeof seq === 'number' ? seq : opcode === TerminalStreamOpcode.Output ? 0 : cursor++ - const sent = sendBinary( - encodeTerminalStreamFrame({ opcode, streamId, seq: resolvedSeq, payload }) - ) - return sent !== false + let sent: boolean | void + try { + sent = sendBinary( + encodeTerminalStreamFrame({ opcode, streamId, seq: resolvedSeq, payload }) + ) + } catch { + closeMultiplex() + return false + } + if (sent === false) { + // Why: false means the transport discarded this frame; reconnect is the only available retry boundary with an authoritative snapshot. + closeMultiplex() + return false + } + return true } const sendStreamError = (streamId: number, message: string): void => { sendFrame(streamId, TerminalStreamOpcode.Error, encodeTerminalStreamText(message)) @@ -1650,24 +1667,28 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ return true } return ( - stream.ackInFlightBytes + bytes <= TERMINAL_MULTIPLEX_ACK_STREAM_HIGH_WATER_BYTES && - ackTotalInFlightBytes + bytes <= TERMINAL_MULTIPLEX_ACK_TOTAL_HIGH_WATER_BYTES + stream.ackInFlightBytes + bytes <= stream.ackWindowBytes && + ackTotalInFlightBytes + bytes <= ackTotalWindowBytes ) } const sendAckGatedOutput = ( stream: TerminalMultiplexStream, chunk: TerminalOutputFrameChunk - ): void => { - sendFrame( + ): boolean => { + const sent = sendFrame( stream.streamId, chunk.opcode ?? TerminalStreamOpcode.Output, chunk.bytes, chunk.seq ) + if (!sent) { + return false + } if (stream.ackOutput) { stream.ackInFlightBytes += chunk.bytes.byteLength ackTotalInFlightBytes += chunk.bytes.byteLength } + return true } const queueOrSendOutput = ( stream: TerminalMultiplexStream, @@ -1700,23 +1721,23 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ if (closed || streams.get(stream.streamId) !== stream) { return } - const size = runtime.getTerminalSize(stream.ptyId) + if (!serialized) { + throw new Error('Remote terminal recovery snapshot unavailable.') + } const displayMode = runtime.getMobileDisplayMode(stream.ptyId) // Why: dropped ACK-pending output breaks live replay; send a fresh snapshot before resuming output. - // Why: clients discard truncated snapshots, so mark truncated only when serialization actually failed. sendSnapshotFrames((opcode, payload) => sendFrame(stream.streamId, opcode, payload), { kind: 'scrollback', - cols: serialized?.cols ?? size?.cols ?? 80, - rows: serialized?.rows ?? size?.rows ?? 24, + cols: serialized.cols, + rows: serialized.rows, displayMode, reason: 'ack-pending-overflow', - seq: serialized?.seq, - source: serialized?.source, - truncated: !serialized, - truncatedByByteBudget: serialized?.truncatedByByteBudget, - data: serialized?.data ?? '' + seq: serialized.seq, + source: serialized.source, + truncatedByByteBudget: serialized.truncatedByByteBudget, + data: serialized.data }) - if (serialized && typeof serialized.seq === 'number') { + if (typeof serialized.seq === 'number') { // Why: chunks queued before the snapshot serialized are already in it; replaying them would duplicate output. const snapshotSeq = serialized.seq const retained = stream.ackPendingOutput.filter( @@ -1734,24 +1755,32 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ stream.streamId, error instanceof Error ? error.message : 'Remote terminal recovery snapshot failed.' ) + // Why: retrying the same failed recovery from finally creates an unbounded error loop. + detachStream(stream.streamId, true) } finally { if (streams.get(stream.streamId) === stream) { stream.ackRecoverySnapshotInFlight = false - flushAckPendingOutput(stream) + flushAllAckPendingOutput() } } } - const flushAckPendingOutput = (stream: TerminalMultiplexStream): void => { + const flushAckPendingOutput = ( + stream: TerminalMultiplexStream, + maxChunks = Number.POSITIVE_INFINITY + ): number => { if (stream.ackPendingOutputOverflowed) { void sendAckRecoverySnapshot(stream) - return + return 0 } let flushed = 0 while ( flushed < stream.ackPendingOutput.length && + flushed < maxChunks && canSendAckGatedOutput(stream, stream.ackPendingOutput[flushed]!.bytes.byteLength) ) { - sendAckGatedOutput(stream, stream.ackPendingOutput[flushed]!) + if (!sendAckGatedOutput(stream, stream.ackPendingOutput[flushed]!)) { + return flushed + } flushed += 1 } if (flushed > 0) { @@ -1761,17 +1790,38 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ 0 ) } + return flushed } const flushAllAckPendingOutput = (): void => { - for (const stream of streams.values()) { - flushAckPendingOutput(stream) - } + const ordered = Array.from(streams.values()) + ackFlushCursorStreamId = drainTerminalMultiplexRoundRobin({ + streams: ordered, + cursorStreamId: ackFlushCursorStreamId, + canContinue: () => !closed, + drainOne: (stream) => { + if (streams.get(stream.streamId) !== stream) { + return false + } + if (flushAckPendingOutput(stream, 1) > 0) { + return true + } + return false + } + }) } const acknowledgeOutput = (stream: TerminalMultiplexStream, bytes: number): void => { if (!stream.ackOutput || bytes <= 0) { return } const acknowledged = Math.min(stream.ackInFlightBytes, bytes) + stream.ackWindowBytes = Math.min( + TERMINAL_MULTIPLEX_ACK_STREAM_MAX_WINDOW_BYTES, + stream.ackWindowBytes + acknowledged + ) + ackTotalWindowBytes = Math.min( + TERMINAL_MULTIPLEX_ACK_TOTAL_MAX_WINDOW_BYTES, + ackTotalWindowBytes + acknowledged + ) stream.ackInFlightBytes -= acknowledged ackTotalInFlightBytes = Math.max(0, ackTotalInFlightBytes - acknowledged) flushAllAckPendingOutput() @@ -2108,6 +2158,15 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ } const request = parsed.data detachStream(request.streamId, false) + cancelPendingPtyWaits(request.streamId) + if ( + streams.size + pendingPtyWaitControllers.size >= + TERMINAL_MULTIPLEX_MAX_STREAMS_PER_CONNECTION + ) { + sendStreamError(request.streamId, 'terminal_stream_limit_exceeded') + emit({ type: 'end', streamId: request.streamId }) + return + } const isMobile = request.client?.type === 'mobile' let leaf: { ptyId: string | null } | null @@ -2180,6 +2239,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ isMobile, ackOutput: request.capabilities?.ackOutput === 1, ackInFlightBytes: 0, + ackWindowBytes: TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES, supportsDesktopViewportClaims: request.capabilities?.desktopViewportClaims === 1, desktopClaimTail: Promise.resolve(true), registeredRemoteDesktopDriver: false, diff --git a/src/main/runtime/rpc/terminal-multiplex-flow-control.bench.test.ts b/src/main/runtime/rpc/terminal-multiplex-flow-control.bench.test.ts new file mode 100644 index 000000000..cfa660ce7 --- /dev/null +++ b/src/main/runtime/rpc/terminal-multiplex-flow-control.bench.test.ts @@ -0,0 +1,208 @@ +import { performance } from 'node:perf_hooks' +import { describe, expect, it } from 'vitest' +import { + TERMINAL_MULTIPLEX_ACK_BATCH_BYTES, + TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES, + TERMINAL_MULTIPLEX_ACK_STREAM_MAX_WINDOW_BYTES, + TERMINAL_MULTIPLEX_ACK_TOTAL_INITIAL_WINDOW_BYTES, + TERMINAL_MULTIPLEX_ACK_TOTAL_MAX_WINDOW_BYTES, + TERMINAL_STREAM_CHUNK_BYTES +} from '../../../shared/terminal-multiplex-flow-control' +import { drainTerminalMultiplexRoundRobin } from './terminal-multiplex-round-robin' + +const MIB = 1024 * 1024 +const PAYLOAD_BYTES_PER_STREAM = 64 * MIB +const PARSER_PAYLOAD_BYTES_PER_VIEWER = 4 * MIB +const benchEnabled = process.env.ORCA_TERMINAL_PERF_BENCH === '1' + +type SimulationResult = { + throughputMiBPerSecond: number + perStreamCompletionMs: number[] + maxInFlightBytes: number + outputFrames: number + ackFrames: number + loopIterations: number +} + +type ParserMeasurement = { + aggregateMiBPerSecond: number + cpuMs: number + retainedHeapKiB: number + xtermWrites: number +} + +async function measureHeadlessXtermParsing(viewers: number): Promise { + const { Terminal } = await import('@xterm/headless') + const sample = '\x1b[?25l\x1b[38;5;45mremote output | build | status | 0123456789\x1b[0m\r\n' + const chunk = sample + .repeat(Math.ceil(TERMINAL_STREAM_CHUNK_BYTES / sample.length)) + .slice(0, TERMINAL_STREAM_CHUNK_BYTES) + const terminals = Array.from( + { length: viewers }, + () => new Terminal({ cols: 120, rows: 40, scrollback: 5_000 }) + ) + const heapBefore = process.memoryUsage().heapUsed + const cpuBefore = process.cpuUsage() + const startedAt = performance.now() + let xtermWrites = 0 + await Promise.all( + terminals.map(async (terminal) => { + let remaining = PARSER_PAYLOAD_BYTES_PER_VIEWER + while (remaining > 0) { + const data = remaining >= chunk.length ? chunk : chunk.slice(0, remaining) + xtermWrites += 1 + await new Promise((resolve) => terminal.write(data, resolve)) + remaining -= data.length + } + }) + ) + const elapsedMs = performance.now() - startedAt + const cpu = process.cpuUsage(cpuBefore) + const heapAfter = process.memoryUsage().heapUsed + for (const terminal of terminals) { + terminal.dispose() + } + return { + aggregateMiBPerSecond: (PARSER_PAYLOAD_BYTES_PER_VIEWER * viewers) / MIB / (elapsedMs / 1_000), + cpuMs: (cpu.user + cpu.system) / 1_000, + retainedHeapKiB: Math.max(0, heapAfter - heapBefore) / 1_024, + xtermWrites + } +} + +function simulateParsedCredit(streamCount: number, rttMs: number): SimulationResult { + const remaining = Array.from({ length: streamCount }, () => PAYLOAD_BYTES_PER_STREAM) + const inFlight = Array.from({ length: streamCount }, () => 0) + const windows = Array.from( + { length: streamCount }, + () => TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES + ) + const streams = Array.from({ length: streamCount }, (_, streamIndex) => ({ + streamId: streamIndex + 1, + streamIndex + })) + const perStreamCompletionMs = Array.from({ length: streamCount }, () => 0) + const acknowledgements = new Map() + let totalInFlight = 0 + let totalWindow = TERMINAL_MULTIPLEX_ACK_TOTAL_INITIAL_WINDOW_BYTES + let maxInFlightBytes = 0 + let outputFrames = 0 + let ackFrames = 0 + let nowMs = 0 + let loopIterations = 0 + let sendCursorStreamId: number | null = null + while (remaining.some((bytes) => bytes > 0) || totalInFlight > 0) { + for (const acknowledgement of acknowledgements.get(nowMs) ?? []) { + inFlight[acknowledgement.streamIndex] -= acknowledgement.bytes + totalInFlight -= acknowledgement.bytes + if ( + remaining[acknowledgement.streamIndex] === 0 && + inFlight[acknowledgement.streamIndex] === 0 + ) { + perStreamCompletionMs[acknowledgement.streamIndex] = nowMs + } + windows[acknowledgement.streamIndex] = Math.min( + TERMINAL_MULTIPLEX_ACK_STREAM_MAX_WINDOW_BYTES, + windows[acknowledgement.streamIndex]! + acknowledgement.bytes + ) + totalWindow = Math.min( + TERMINAL_MULTIPLEX_ACK_TOTAL_MAX_WINDOW_BYTES, + totalWindow + acknowledgement.bytes + ) + ackFrames += Math.ceil(acknowledgement.bytes / TERMINAL_MULTIPLEX_ACK_BATCH_BYTES) + } + acknowledgements.delete(nowMs) + sendCursorStreamId = drainTerminalMultiplexRoundRobin({ + streams, + cursorStreamId: sendCursorStreamId, + drainOne: ({ streamIndex }) => { + if ( + remaining[streamIndex]! <= 0 || + inFlight[streamIndex]! >= windows[streamIndex]! || + totalInFlight >= totalWindow + ) { + return false + } + const bytes = Math.min( + TERMINAL_STREAM_CHUNK_BYTES, + remaining[streamIndex]!, + windows[streamIndex]! - inFlight[streamIndex]!, + totalWindow - totalInFlight + ) + remaining[streamIndex] -= bytes + inFlight[streamIndex] += bytes + totalInFlight += bytes + outputFrames += 1 + const due = nowMs + rttMs + const dueAcks = acknowledgements.get(due) ?? [] + const existingAck = dueAcks.find((ack) => ack.streamIndex === streamIndex) + if (existingAck) { + existingAck.bytes += bytes + } else { + dueAcks.push({ streamIndex, bytes }) + } + acknowledgements.set(due, dueAcks) + return true + } + }) + maxInFlightBytes = Math.max(maxInFlightBytes, totalInFlight) + nowMs += 1 + loopIterations += 1 + } + return { + throughputMiBPerSecond: (PAYLOAD_BYTES_PER_STREAM * streamCount) / MIB / (nowMs / 1000), + perStreamCompletionMs, + maxInFlightBytes, + outputFrames, + ackFrames, + loopIterations + } +} + +describe('terminal multiplex parsed-credit bounds', () => { + it('keeps aggregate memory bounded and streams fair at 100 ms RTT', () => { + const result = simulateParsedCredit(8, 100) + expect(result.maxInFlightBytes).toBeLessThanOrEqual( + TERMINAL_MULTIPLEX_ACK_TOTAL_MAX_WINDOW_BYTES + ) + expect(result.throughputMiBPerSecond / 8).toBeGreaterThan(7) + expect(result.ackFrames).toBeLessThan(result.outputFrames / 3) + expect( + Math.max(...result.perStreamCompletionMs) - Math.min(...result.perStreamCompletionMs) + ).toBeLessThan(200) + }) +}) + +describe.skipIf(!benchEnabled)('terminal multiplex parsed-credit benchmark', () => { + it('reports RTT, fairness, protocol allocations, and measured xterm parser cost', async () => { + const parserMeasurements = new Map() + for (const viewers of [1, 4, 8]) { + parserMeasurements.set(viewers, await measureHeadlessXtermParsing(viewers)) + } + const rows = [1, 20, 100].flatMap((rttMs) => + [1, 4, 8].map((viewers) => { + const startedAt = performance.now() + const result = simulateParsedCredit(viewers, rttMs) + const parser = parserMeasurements.get(viewers)! + return { + rttMs, + viewers, + aggregateMiBps: Number(result.throughputMiBPerSecond.toFixed(1)), + perViewerMiBps: Number((result.throughputMiBPerSecond / viewers).toFixed(1)), + schedulerCpuMs: Number((performance.now() - startedAt).toFixed(2)), + protocolFrameAllocations: result.outputFrames + result.ackFrames, + loopIterations: result.loopIterations, + maxInFlightKiB: result.maxInFlightBytes / 1024, + completionSpreadMs: + Math.max(...result.perStreamCompletionMs) - Math.min(...result.perStreamCompletionMs), + measuredParserMiBps: Number(parser.aggregateMiBPerSecond.toFixed(1)), + parserCpuMs: Number(parser.cpuMs.toFixed(1)), + parserRetainedHeapKiB: Number(parser.retainedHeapKiB.toFixed(0)), + xtermWriteAllocations: parser.xtermWrites + } + }) + ) + // eslint-disable-next-line no-console -- opt-in benchmark evidence + console.table(rows) + }) +}) diff --git a/src/main/runtime/rpc/terminal-multiplex-round-robin.test.ts b/src/main/runtime/rpc/terminal-multiplex-round-robin.test.ts new file mode 100644 index 000000000..96e04718c --- /dev/null +++ b/src/main/runtime/rpc/terminal-multiplex-round-robin.test.ts @@ -0,0 +1,67 @@ +import { describe, expect, it } from 'vitest' +import { drainTerminalMultiplexRoundRobin } from './terminal-multiplex-round-robin' + +describe('terminal multiplex round-robin drain', () => { + it('admits a later interactive stream before older bulk queues refill the window', () => { + const streams = Array.from({ length: 8 }, (_, index) => ({ + streamId: index + 1, + pendingChunks: index === 7 ? 1 : 8 + })) + const order: number[] = [] + let remainingSlots = 8 + + const cursor = drainTerminalMultiplexRoundRobin({ + streams, + cursorStreamId: null, + canContinue: () => remainingSlots > 0, + drainOne: (stream) => { + if (stream.pendingChunks === 0) { + return false + } + stream.pendingChunks -= 1 + remainingSlots -= 1 + order.push(stream.streamId) + return true + } + }) + + expect(order).toEqual([1, 2, 3, 4, 5, 6, 7, 8]) + expect(cursor).toBe(8) + }) + + it('resumes after the previous sender on the next release', () => { + const streams = [1, 2, 3].map((streamId) => ({ streamId, pendingChunks: 2 })) + let slots = 2 + const firstOrder: number[] = [] + const cursor = drainTerminalMultiplexRoundRobin({ + streams, + cursorStreamId: null, + canContinue: () => slots > 0, + drainOne: (stream) => { + stream.pendingChunks -= 1 + slots -= 1 + firstOrder.push(stream.streamId) + return true + } + }) + slots = 2 + const secondOrder: number[] = [] + drainTerminalMultiplexRoundRobin({ + streams, + cursorStreamId: cursor, + canContinue: () => slots > 0, + drainOne: (stream) => { + if (stream.pendingChunks === 0) { + return false + } + stream.pendingChunks -= 1 + slots -= 1 + secondOrder.push(stream.streamId) + return true + } + }) + + expect(firstOrder).toEqual([1, 2]) + expect(secondOrder).toEqual([3, 1]) + }) +}) diff --git a/src/main/runtime/rpc/terminal-multiplex-round-robin.ts b/src/main/runtime/rpc/terminal-multiplex-round-robin.ts new file mode 100644 index 000000000..d5c8f57d8 --- /dev/null +++ b/src/main/runtime/rpc/terminal-multiplex-round-robin.ts @@ -0,0 +1,41 @@ +type TerminalMultiplexDrainStream = { streamId: number } + +export function drainTerminalMultiplexRoundRobin(args: { + streams: readonly T[] + cursorStreamId: number | null + drainOne: (stream: T) => boolean + canContinue?: () => boolean +}): number | null { + const { streams, drainOne } = args + if (streams.length === 0) { + return null + } + const canContinue = args.canContinue ?? (() => true) + let cursorStreamId = args.cursorStreamId + let startIndex = getStartIndex(streams, cursorStreamId) + while (canContinue()) { + let progressed = false + for (let offset = 0; offset < streams.length && canContinue(); offset += 1) { + const stream = streams[(startIndex + offset) % streams.length]! + if (drainOne(stream)) { + cursorStreamId = stream.streamId + progressed = true + } + } + if (!progressed) { + break + } + startIndex = getStartIndex(streams, cursorStreamId) + } + return cursorStreamId +} + +function getStartIndex( + streams: readonly T[], + cursorStreamId: number | null +): number { + if (cursorStreamId === null) { + return 0 + } + return (streams.findIndex((stream) => stream.streamId === cursorStreamId) + 1) % streams.length +} diff --git a/src/main/runtime/rpc/terminal-multiplex.test.ts b/src/main/runtime/rpc/terminal-multiplex.test.ts index f281d72e4..d86161d03 100644 --- a/src/main/runtime/rpc/terminal-multiplex.test.ts +++ b/src/main/runtime/rpc/terminal-multiplex.test.ts @@ -43,7 +43,8 @@ function makeRequest(method: string, params?: unknown): RpcRequest { function startDesktopMultiplexSubscribe( overrides: Partial = {}, - trace?: string[] + trace?: string[], + sendBinaryOverride?: (bytes: Uint8Array) => boolean | void ) { const messages: string[] = [] const binaryFrames: Uint8Array[] = [] @@ -87,6 +88,10 @@ function startDesktopMultiplexSubscribe( { connectionId: 'conn-desktop-first-paint', sendBinary: (bytes) => { + const sent = sendBinaryOverride?.(bytes) + if (sent === false) { + return false + } binaryFrames.push(bytes) const opcode = decodeTerminalStreamFrame(bytes)?.opcode if ( @@ -96,10 +101,15 @@ function startDesktopMultiplexSubscribe( ) { trace?.push('snapshot') } + return sent }, registerBinaryStreamHandler: (streamId, handler) => { handlers.set(streamId, handler) - return () => handlers.delete(streamId) + return () => { + if (handlers.get(streamId) === handler) { + handlers.delete(streamId) + } + } } } ) @@ -128,6 +138,65 @@ function sendDesktopMultiplexSubscribe( } describe('terminal multiplex RPC', () => { + it.each(['refuses', 'throws'] as const)( + 'closes without reserving ACK debt when the transport %s an output frame', + async (failureMode) => { + let dataListener: + | ((data: string, meta?: { seq?: number; rawLength?: number }) => void) + | null = null + let rejectOutput = false + const unsubscribeData = vi.fn() + const harness = startDesktopMultiplexSubscribe( + { + subscribeToTerminalData: vi.fn((_ptyId, listener) => { + dataListener = listener + return unsubscribeData + }) + }, + undefined, + (bytes) => { + const frame = decodeTerminalStreamFrame(bytes) + if (!rejectOutput || frame?.opcode !== TerminalStreamOpcode.Output) { + return true + } + if (failureMode === 'throws') { + throw new Error('socket closed') + } + return false + } + ) + + await vi.waitFor(() => + expect(harness.messages.some((msg) => JSON.parse(msg).result?.type === 'ready')).toBe(true) + ) + sendDesktopMultiplexSubscribe(harness.handlers) + await vi.waitFor(() => + expect(harness.messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe( + true + ) + ) + await vi.waitFor(() => expect(dataListener).not.toBeNull()) + harness.binaryFrames.splice(0) + rejectOutput = true + + const output = 'x'.repeat(64 * 1024) + const deliverData = dataListener as unknown as ( + data: string, + meta?: { seq?: number; rawLength?: number } + ) => void + deliverData(output, { seq: output.length, rawLength: output.length }) + + await vi.waitFor(() => expect(unsubscribeData).toHaveBeenCalledOnce()) + await harness.dispatchPromise + expect( + harness.binaryFrames + .map((bytes) => decodeTerminalStreamFrame(bytes)) + .filter((frame) => frame?.opcode === TerminalStreamOpcode.Output) + ).toEqual([]) + expect(harness.handlers.size).toBe(0) + } + ) + it('multiplexes terminal streams and routes desktop resize to the source PTY', async () => { vi.useFakeTimers() try { @@ -1197,7 +1266,7 @@ describe('terminal multiplex RPC', () => { await dispatchPromise }) - it('releases shared ACK budget to other stalled multiplex streams', async () => { + it('round-robins released ACK budget to a later interactive stream', async () => { const messages: string[] = [] const binaryFrames: Uint8Array[] = [] const handlers = new Map< @@ -1272,7 +1341,7 @@ describe('terminal multiplex RPC', () => { expect(messages.some((msg) => JSON.parse(msg).result?.type === 'ready')).toBe(true) ) - const streamIds = [21, 22, 23, 24, 25, 26] + const streamIds = [21, 22, 23, 24, 25, 26, 27, 28] for (const streamId of streamIds) { handlers.get(0)?.( decodeTerminalStreamFrame( @@ -1302,22 +1371,33 @@ describe('terminal multiplex RPC', () => { await vi.waitFor(() => expect(dataListeners.size).toBe(streamIds.length)) binaryFrames.splice(0) - const fillerOutput = 'f'.repeat(480 * 1024) + const fillerOutput = 'f'.repeat(512 * 1024) for (let index = 1; index <= 4; index += 1) { dataListeners.get(`pty-${index}`)?.(fillerOutput, { seq: fillerOutput.length, rawLength: fillerOutput.length }) } - const stalledOutput = 's'.repeat(700 * 1024) - dataListeners.get('pty-5')?.(stalledOutput, { - seq: stalledOutput.length, - rawLength: stalledOutput.length - }) - dataListeners.get('pty-6')?.(stalledOutput, { - seq: stalledOutput.length, - rawLength: stalledOutput.length + const queuedFillerOutput = 'q'.repeat(256 * 1024) + for (let index = 1; index <= 4; index += 1) { + dataListeners.get(`pty-${index}`)?.(queuedFillerOutput, { + seq: fillerOutput.length + queuedFillerOutput.length, + rawLength: queuedFillerOutput.length + }) + } + const stalledOutput = 's'.repeat(256 * 1024) + for (let index = 5; index <= 7; index += 1) { + dataListeners.get(`pty-${index}`)?.(stalledOutput, { + seq: stalledOutput.length, + rawLength: stalledOutput.length + }) + } + const interactiveOutput = 'interactive-output\r\n' + dataListeners.get('pty-8')?.(interactiveOutput, { + seq: interactiveOutput.length, + rawLength: interactiveOutput.length }) + await new Promise((resolve) => setTimeout(resolve, 10)) const initialOutputFrames = binaryFrames .map((frame) => decodeTerminalStreamFrame(frame)) @@ -1337,25 +1417,27 @@ describe('terminal multiplex RPC', () => { 0 ) expect(initialBytes).toBeLessThanOrEqual(2 * 1024 * 1024) - expect(initialBytesByStream.get(21)).toBe(480 * 1024) - expect(initialBytesByStream.get(22)).toBe(480 * 1024) - expect(initialBytesByStream.get(23)).toBe(480 * 1024) - expect(initialBytesByStream.get(24)).toBe(480 * 1024) - expect(initialBytesByStream.get(25)).toBeGreaterThan(0) + expect(initialBytesByStream.get(21)).toBe(512 * 1024) + expect(initialBytesByStream.get(22)).toBe(512 * 1024) + expect(initialBytesByStream.get(23)).toBe(512 * 1024) + expect(initialBytesByStream.get(24)).toBe(512 * 1024) + expect(initialBytesByStream.get(25) ?? 0).toBe(0) expect(initialBytesByStream.get(26) ?? 0).toBe(0) + expect(initialBytesByStream.get(27) ?? 0).toBe(0) + expect(initialBytesByStream.get(28) ?? 0).toBe(0) - handlers.get(26)?.( + handlers.get(28)?.( decodeTerminalStreamFrame( encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Input, - streamId: 26, + streamId: 28, seq: 200, payload: encodeTerminalStreamText('remote-still-interactive\r') }) )! ) await vi.waitFor(() => - expect(runtime.sendTerminal).toHaveBeenCalledWith('terminal-6', { + expect(runtime.sendTerminal).toHaveBeenCalledWith('terminal-8', { text: 'remote-still-interactive\r', enter: false, interrupt: false @@ -1379,25 +1461,17 @@ describe('terminal multiplex RPC', () => { binaryFrames .slice(frameCountBeforeAck) .map((frame) => decodeTerminalStreamFrame(frame)) - .some((frame) => { - if (frame?.streamId !== 25 || frame.opcode !== TerminalStreamOpcode.SnapshotStart) { - return false - } - const payload = decodeTerminalStreamJson<{ reason?: string }>(frame.payload) - return payload?.reason === 'ack-pending-overflow' - }) + .some( + (frame) => + frame?.streamId === 28 && + frame.opcode === TerminalStreamOpcode.Output && + decodeTerminalStreamText(frame.payload) === interactiveOutput + ) ).toBe(true) ) const framesAfterAck = binaryFrames .slice(frameCountBeforeAck) .map((frame) => decodeTerminalStreamFrame(frame)) - const snapshotStartIndex = framesAfterAck.findIndex((frame) => { - if (frame?.streamId !== 25 || frame.opcode !== TerminalStreamOpcode.SnapshotStart) { - return false - } - const payload = decodeTerminalStreamJson<{ reason?: string }>(frame.payload) - return payload?.reason === 'ack-pending-overflow' - }) const outputFramesAfterAck = framesAfterAck.filter( (frame) => frame?.opcode === TerminalStreamOpcode.Output ) @@ -1411,17 +1485,14 @@ describe('terminal multiplex RPC', () => { (bytesAfterAckByStream.get(frame.streamId) ?? 0) + frame.payload.byteLength ) } - expect(snapshotStartIndex).toBeGreaterThanOrEqual(0) - expect( - framesAfterAck - .filter((frame) => frame?.streamId === 25 && frame.opcode === TerminalStreamOpcode.Output) - .every((frame) => framesAfterAck.indexOf(frame) > snapshotStartIndex) - ).toBe(true) expect(bytesAfterAckByStream.get(25) ?? 0).toBeGreaterThan(0) - expect(bytesAfterAckByStream.get(21) ?? 0).toBe(0) + expect(bytesAfterAckByStream.get(26) ?? 0).toBeGreaterThan(0) + expect(bytesAfterAckByStream.get(27) ?? 0).toBeGreaterThan(0) + expect(bytesAfterAckByStream.get(28) ?? 0).toBe(interactiveOutput.length) + expect(bytesAfterAckByStream.get(21) ?? 0).toBeGreaterThan(0) expect( outputFramesAfterAck.reduce((total, frame) => total + (frame?.payload.byteLength ?? 0), 0) - ).toBeLessThanOrEqual(initialBytesByStream.get(21) ?? 0) + ).toBeLessThanOrEqual((initialBytesByStream.get(21) ?? 0) * 2) runtime.cleanupSubscription('terminal-multiplex:conn-ack-shared-budget') await dispatchPromise @@ -1605,6 +1676,81 @@ describe('terminal multiplex RPC', () => { await dispatchPromise }) + it.each([ + { + failure: 'throws', + recover: () => Promise.reject(new Error('snapshot unavailable')) + }, + { + failure: 'returns no snapshot', + recover: () => Promise.resolve(null) + } + ])('ends a stream when ACK overflow recovery serialization $failure', async ({ recover }) => { + const dataListenerRef: { + current?: (data: string, meta?: { seq?: number; rawLength?: number }) => void + } = {} + const serializeTerminalBuffer = vi + .fn() + .mockResolvedValueOnce({ data: 'initial snapshot', cols: 120, rows: 40 }) + .mockImplementation(recover) + const harness = startDesktopMultiplexSubscribe({ + serializeTerminalBuffer, + subscribeToTerminalData: vi.fn( + ( + _: string, + listener: (data: string, meta?: { seq?: number; rawLength?: number }) => void + ) => { + dataListenerRef.current = listener + return vi.fn() + } + ) + }) + + await vi.waitFor(() => + expect(harness.messages.some((message) => JSON.parse(message).result?.type === 'ready')).toBe( + true + ) + ) + sendDesktopMultiplexSubscribe(harness.handlers) + await vi.waitFor(() => + expect( + harness.messages.some((message) => JSON.parse(message).result?.type === 'subscribed') + ).toBe(true) + ) + harness.binaryFrames.splice(0) + + const output = 'x'.repeat(3 * 1024 * 1024) + dataListenerRef.current?.(output, { seq: output.length, rawLength: output.length }) + const inFlightBytes = harness.binaryFrames + .map((bytes) => decodeTerminalStreamFrame(bytes)) + .filter((frame) => frame?.opcode === TerminalStreamOpcode.Output) + .reduce((total, frame) => total + (frame?.payload.byteLength ?? 0), 0) + harness.handlers.get(7)?.( + decodeTerminalStreamFrame( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Ack, + streamId: 7, + seq: 2, + payload: encodeTerminalStreamJson({ bytes: inFlightBytes }) + }) + )! + ) + + await vi.waitFor(() => { + const eventTypes = harness.messages.map((message) => JSON.parse(message).result?.type) + expect(eventTypes).toContain('error') + expect(eventTypes).toContain('end') + }) + expect(harness.handlers.has(7)).toBe(false) + expect(serializeTerminalBuffer).toHaveBeenCalledTimes(2) + await Promise.resolve() + await Promise.resolve() + expect(serializeTerminalBuffer).toHaveBeenCalledTimes(2) + + harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + await harness.dispatchPromise + }) + it('trims recovery-covered ACK pending output instead of replaying it', async () => { const messages: string[] = [] const binaryFrames: Uint8Array[] = [] @@ -2881,6 +3027,81 @@ describe('terminal multiplex RPC', () => { await harness.dispatchPromise }) + it('cancels an older pending PTY wait when the same multiplex slot resubscribes', async () => { + const waitSignals: AbortSignal[] = [] + const waitForLeafPtyId = vi.fn( + (_handle: string, _timeoutMs?: number, signal?: AbortSignal) => + new Promise((_resolve, reject) => { + if (signal) { + waitSignals.push(signal) + } + signal?.addEventListener('abort', () => reject(new Error('request_aborted')), { + once: true + }) + }) + ) + const harness = startDesktopMultiplexSubscribe({ + resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: null }), + waitForLeafPtyId + }) + await vi.waitFor(() => + expect(harness.messages.some((message) => JSON.parse(message).result?.type === 'ready')).toBe( + true + ) + ) + + sendDesktopMultiplexSubscribe(harness.handlers) + await vi.waitFor(() => expect(waitSignals).toHaveLength(1)) + sendDesktopMultiplexSubscribe(harness.handlers) + await vi.waitFor(() => expect(waitSignals).toHaveLength(2)) + + expect(waitSignals[0]?.aborted).toBe(true) + expect(waitSignals[1]?.aborted).toBe(false) + harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + await vi.waitFor(() => expect(waitSignals[1]?.aborted).toBe(true)) + await harness.dispatchPromise + }) + + it('caps multiplex stream slots so aggregate pending output stays bounded', async () => { + const harness = startDesktopMultiplexSubscribe() + await vi.waitFor(() => + expect(harness.messages.some((message) => JSON.parse(message).result?.type === 'ready')).toBe( + true + ) + ) + for (let streamId = 1; streamId <= 33; streamId += 1) { + harness.handlers.get(0)?.( + decodeTerminalStreamFrame( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Subscribe, + streamId: 0, + seq: streamId, + payload: encodeTerminalStreamJson({ + streamId, + terminal: 'terminal-1', + client: { id: 'desktop-1', type: 'desktop' }, + capabilities: { ackOutput: 1 } + }) + }) + )! + ) + } + + await vi.waitFor(() => { + const results = harness.messages.map((message) => JSON.parse(message).result) + expect(results.filter((result) => result?.type === 'subscribed')).toHaveLength(32) + expect(results).toContainEqual({ + type: 'error', + streamId: 33, + message: 'terminal_stream_limit_exceeded' + }) + expect(results).toContainEqual({ type: 'end', streamId: 33 }) + }) + + harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + await harness.dispatchPromise + }) + it("still reports no_connected_pty when a desktop multiplex subscriber's PTY never appears", async () => { const runtime = stubRuntime({ resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: null }), diff --git a/src/renderer/src/components/terminal-pane/pty-connection.test.ts b/src/renderer/src/components/terminal-pane/pty-connection.test.ts index 5a59e70d9..3020dab37 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection.test.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection.test.ts @@ -13560,6 +13560,8 @@ describe('connectPanePty', () => { it('requests snapshot recovery for one oversized live frame deferred by replay', async () => { const { connectPanePty } = await import('./pty-connection') + const { deliverTerminalDataWithDeferredCredit } = + await import('@/lib/pane-manager/terminal-delivery-credit') const transport = createMockTransport('pty-large-live') const callbacksRef: { replay: ((data: string) => void) | null @@ -13584,7 +13586,11 @@ describe('connectPanePty', () => { callbacksRef.replay?.('authoritative replay') await flushAsyncTicks(8) const oversizedLiveFrame = 'L'.repeat(512 * 1024 + 1) - callbacksRef.data?.(oversizedLiveFrame) + const acknowledgeDroppedFrame = vi.fn() + deliverTerminalDataWithDeferredCredit(acknowledgeDroppedFrame, () => { + callbacksRef.data?.(oversizedLiveFrame) + }) + expect(acknowledgeDroppedFrame).not.toHaveBeenCalled() while (parseCallbacks.length > 0) { parseCallbacks.shift()?.() await flushAsyncTicks(4) @@ -13595,6 +13601,7 @@ describe('connectPanePty', () => { scrollbackRows: 5000 }) expect(writes.some((write) => write.startsWith('L'))).toBe(false) + expect(acknowledgeDroppedFrame).toHaveBeenCalledOnce() binding.dispose() }) diff --git a/src/renderer/src/components/terminal-pane/pty-connection.ts b/src/renderer/src/components/terminal-pane/pty-connection.ts index c0598d912..8a410c4e0 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection.ts @@ -5823,7 +5823,7 @@ export function connectPanePty( writeTerminalOutput(pane.terminal, data, { foreground: foregroundOutput, beforeWrite: beforeTerminalOutputWrite, - // Why: claim the delivery's parse-deferred ACK credit (null outside a delivery); the FIRST scheduler write carries it all and fires when bytes are consumed. + // Why: every scheduler write claims one child so a split delivery is credited only after all children parse or discard. ackCredit: takeCurrentTerminalDeliveryCredit() ?? undefined, onBackgroundBacklogDropped: markHiddenOutputRestoreNeeded, latencySensitive: diff --git a/src/renderer/src/components/terminal-pane/terminal-pty-ack-gate.test.ts b/src/renderer/src/components/terminal-pane/terminal-pty-ack-gate.test.ts index 3280ca36d..ba001989c 100644 --- a/src/renderer/src/components/terminal-pane/terminal-pty-ack-gate.test.ts +++ b/src/renderer/src/components/terminal-pane/terminal-pty-ack-gate.test.ts @@ -91,7 +91,7 @@ describe('terminal-pty-ack-gate parse-deferred crediting', () => { expect(ackDataMock).toHaveBeenCalledTimes(1) }) - it('hands out the credit only once per delivery', async () => { + it('waits for every scheduler write produced by one delivery', async () => { const { deliverPtyDataWithDeferredAck, takeCurrentPtyDeliveryAckCredit } = await loadAckGate() let first: (() => void) | null = null let second: (() => void) | null = null @@ -102,7 +102,11 @@ describe('terminal-pty-ack-gate parse-deferred crediting', () => { }) expect(first).not.toBeNull() - expect(second).toBeNull() + expect(second).not.toBeNull() + first!() + expect(ackDataMock).not.toHaveBeenCalled() + second!() + expect(ackDataMock).toHaveBeenCalledWith('pty-a', 5, 5) }) it('returns null outside a delivery', async () => { diff --git a/src/renderer/src/components/terminal-pane/terminal-pty-ack-gate.ts b/src/renderer/src/components/terminal-pane/terminal-pty-ack-gate.ts index 8340dfa1b..12f065377 100644 --- a/src/renderer/src/components/terminal-pane/terminal-pty-ack-gate.ts +++ b/src/renderer/src/components/terminal-pane/terminal-pty-ack-gate.ts @@ -106,9 +106,6 @@ export function deliverPtyDataWithDeferredAck( deliverTerminalDataWithDeferredCredit(() => ackPtyData(ptyId, chars), deliver) } -/** Claims the in-progress delivery's credit for the output scheduler. Returns - * a fire-once callback, or null when outside a delivery or already claimed - * (only the FIRST scheduler write of a delivery carries the credit). */ export function takeCurrentPtyDeliveryAckCredit(): (() => void) | null { return takeCurrentTerminalDeliveryCredit() } diff --git a/src/renderer/src/lib/pane-manager/terminal-delivery-credit.test.ts b/src/renderer/src/lib/pane-manager/terminal-delivery-credit.test.ts index 01e20fb3f..3b7fcc2fd 100644 --- a/src/renderer/src/lib/pane-manager/terminal-delivery-credit.test.ts +++ b/src/renderer/src/lib/pane-manager/terminal-delivery-credit.test.ts @@ -41,4 +41,26 @@ describe('terminal delivery credit', () => { expect(complete).toHaveBeenCalledOnce() expect(claimLater!()).toBeNull() }) + it('settles only after every scheduler write claimed by one delivery completes', async () => { + const { deliverTerminalDataWithDeferredCredit, takeCurrentTerminalDeliveryCredit } = + await import('./terminal-delivery-credit') + const complete = vi.fn() + let first: (() => void) | null = null + let second: (() => void) | null = null + + deliverTerminalDataWithDeferredCredit(complete, () => { + first = takeCurrentTerminalDeliveryCredit() + second = takeCurrentTerminalDeliveryCredit() + }) + + expect(first).not.toBeNull() + expect(second).not.toBeNull() + first!() + expect(complete).not.toHaveBeenCalled() + second!() + expect(complete).toHaveBeenCalledOnce() + first!() + second!() + expect(complete).toHaveBeenCalledOnce() + }) }) diff --git a/src/renderer/src/lib/pane-manager/terminal-delivery-credit.ts b/src/renderer/src/lib/pane-manager/terminal-delivery-credit.ts index d467cedda..da15be636 100644 --- a/src/renderer/src/lib/pane-manager/terminal-delivery-credit.ts +++ b/src/renderer/src/lib/pane-manager/terminal-delivery-credit.ts @@ -1,30 +1,31 @@ type TerminalDeliveryCredit = { complete: () => void - claimed: boolean - credited: boolean + open: boolean + pendingClaims: number + completed: boolean } -// Why: consumers must claim during deliver(); after it returns this synchronous slot is restored and unclaimed credit settles. +// Why: claims are synchronous; nesting restores an outer delivery after an inner callback returns. let currentDeliveryCredit: TerminalDeliveryCredit | null = null function completeTerminalDeliveryCredit(credit: TerminalDeliveryCredit): void { - // Why: queue splitting and discard paths can both settle one delivery. - if (credit.credited) { + if (credit.completed || credit.open || credit.pendingClaims > 0) { return } - credit.credited = true + credit.completed = true credit.complete() } -/** Defers producer credit until the output scheduler consumes or discards the delivery. */ +/** Defers producer credit until every output scheduler consumer parses or discards it. */ export function deliverTerminalDataWithDeferredCredit( complete: () => void, deliver: () => void ): void { const credit: TerminalDeliveryCredit = { complete, - claimed: false, - credited: false + open: true, + pendingClaims: 0, + completed: false } const previousCredit = currentDeliveryCredit currentDeliveryCredit = credit @@ -32,18 +33,24 @@ export function deliverTerminalDataWithDeferredCredit( deliver() } finally { currentDeliveryCredit = previousCredit - if (!credit.claimed) { - completeTerminalDeliveryCredit(credit) - } + credit.open = false + completeTerminalDeliveryCredit(credit) } } -/** Claims the current delivery for parse-deferred settlement by the output scheduler. */ export function takeCurrentTerminalDeliveryCredit(): (() => void) | null { const credit = currentDeliveryCredit - if (!credit || credit.claimed) { + if (!credit || !credit.open) { return null } - credit.claimed = true - return () => completeTerminalDeliveryCredit(credit) + credit.pendingClaims += 1 + let settled = false + return () => { + if (settled) { + return + } + settled = true + credit.pendingClaims -= 1 + completeTerminalDeliveryCredit(credit) + } } diff --git a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts index a88e6f828..0c13d2d20 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts @@ -14,6 +14,10 @@ import { e2eConfig } from '@/lib/e2e-config' import { deliverTerminalDataWithDeferredCredit } from '@/lib/pane-manager/terminal-delivery-credit' import { unwrapRuntimeRpcResult } from './runtime-rpc-client' import { getRuntimeEnvironmentRevision } from './runtime-environment-revision' +import { + TERMINAL_MULTIPLEX_ACK_BATCH_BYTES, + TERMINAL_MULTIPLEX_ACK_FLUSH_MS +} from '../../../shared/terminal-multiplex-flow-control' type RuntimeEnvironmentSubscriptionHandle = { unsubscribe: () => void @@ -78,6 +82,8 @@ type RemoteRuntimeMultiplexedTerminalState = { subscriptionRequested: boolean acknowledgeOutput: boolean heldAckBytes: number + pendingAckBytes: number + ackFlushTimer: ReturnType | null snapshotChunks: Uint8Array[] snapshotBytes: number snapshotOverflowed: boolean @@ -247,6 +253,8 @@ class RemoteRuntimeTerminalMultiplexer { subscriptionRequested: false, acknowledgeOutput: args.client.type === 'desktop', heldAckBytes: 0, + pendingAckBytes: 0, + ackFlushTimer: null, snapshotChunks: [], snapshotBytes: 0, snapshotOverflowed: false, @@ -292,6 +300,7 @@ class RemoteRuntimeTerminalMultiplexer { serializeBuffer: (opts) => this.requestSnapshot(state, opts), close: () => { if (this.streams.get(streamId) === state) { + discardOutputAcknowledgements(state) this.sendFrame(streamId, TerminalStreamOpcode.Unsubscribe) clearResyncTimer(state) rejectPendingSnapshotRequest(state, 'Remote terminal stream closed.') @@ -429,6 +438,7 @@ class RemoteRuntimeTerminalMultiplexer { return } if (event.type === 'end') { + discardOutputAcknowledgements(stream) clearSnapshot(stream) clearResyncTimer(stream) rejectPendingSnapshotRequest(stream, 'Remote terminal stream ended.') @@ -483,10 +493,19 @@ class RemoteRuntimeTerminalMultiplexer { } const frame = decodeTerminalStreamFrame(bytes) if (!frame) { + // Why: malformed framing cannot be credited safely; closing makes the server release every stream window. + this.failConnection(new Error('Remote terminal stream received a malformed frame.')) return } const stream = this.streams.get(frame.streamId) if (!stream) { + if ( + frame.opcode === TerminalStreamOpcode.Output || + frame.opcode === TerminalStreamOpcode.OutputSpan + ) { + // Why: the renderer already disposed this stream; unsubscribe releases server credit that cannot reach a parser. + this.sendFrame(frame.streamId, TerminalStreamOpcode.Unsubscribe) + } return } if ( @@ -557,13 +576,19 @@ class RemoteRuntimeTerminalMultiplexer { deliverOutput() return } - deliverTerminalDataWithDeferredCredit(() => { - if (shouldHoldE2eRemoteTerminalAck(stream.terminal)) { - stream.heldAckBytes += frame.payload.byteLength - } else { - this.acknowledgeOutput(stream, frame.payload.byteLength) - } - }, deliverOutput) + try { + deliverTerminalDataWithDeferredCredit(() => { + if (shouldHoldE2eRemoteTerminalAck(stream.terminal)) { + stream.heldAckBytes += frame.payload.byteLength + } else { + this.queueOutputAcknowledgement(stream, frame.payload.byteLength) + } + }, deliverOutput) + } catch (error) { + this.failConnection( + error instanceof Error ? error : new Error('Remote terminal output delivery failed.') + ) + } return } if (frame.opcode === TerminalStreamOpcode.SnapshotStart) { @@ -856,6 +881,33 @@ class RemoteRuntimeTerminalMultiplexer { ) } + private queueOutputAcknowledgement( + stream: RemoteRuntimeMultiplexedTerminalState, + bytes: number + ): boolean { + if (this.streams.get(stream.streamId) !== stream) { + return true + } + stream.pendingAckBytes += bytes + if (stream.pendingAckBytes >= TERMINAL_MULTIPLEX_ACK_BATCH_BYTES) { + return this.flushOutputAcknowledgement(stream) + } + if (stream.ackFlushTimer === null) { + stream.ackFlushTimer = setTimeout(() => { + stream.ackFlushTimer = null + this.flushOutputAcknowledgement(stream) + }, TERMINAL_MULTIPLEX_ACK_FLUSH_MS) + } + return true + } + + private flushOutputAcknowledgement(stream: RemoteRuntimeMultiplexedTerminalState): boolean { + clearAckFlushTimer(stream) + const bytes = stream.pendingAckBytes + stream.pendingAckBytes = 0 + return bytes <= 0 || this.acknowledgeOutput(stream, bytes) + } + getStreamsForE2e(): Iterable { return this.streams.values() } @@ -868,7 +920,7 @@ class RemoteRuntimeTerminalMultiplexer { } const bytes = stream.heldAckBytes stream.heldAckBytes = 0 - if (this.acknowledgeOutput(stream, bytes)) { + if (this.queueOutputAcknowledgement(stream, bytes)) { released += bytes } } @@ -883,8 +935,15 @@ class RemoteRuntimeTerminalMultiplexer { if (!this.matchesCurrentEnvironmentRevision() || !this.ready || !this.subscription) { return false } - this.subscription.sendBinary(encodeTerminalStreamFrame({ opcode, streamId, seq: 0, payload })) - return true + try { + this.subscription.sendBinary(encodeTerminalStreamFrame({ opcode, streamId, seq: 0, payload })) + return true + } catch (error) { + this.handleClose( + error instanceof Error ? error.message : 'Remote terminal transport write failed.' + ) + return false + } } private resolveReadyIfConnected(): void { @@ -923,6 +982,7 @@ class RemoteRuntimeTerminalMultiplexer { // Why: close callbacks may resubscribe synchronously; release first so every replacement shares the new environment multiplexer. this.releaseIfCurrent(this.environmentId, this) for (const stream of streams) { + discardOutputAcknowledgements(stream) clearSnapshot(stream) clearResyncTimer(stream) rejectPendingSnapshotRequest(stream, message ?? 'Remote runtime connection closed.') @@ -1006,6 +1066,19 @@ function clearSnapshot(stream: RemoteRuntimeMultiplexedTerminalState): void { stream.snapshotInfo = null } +function clearAckFlushTimer(stream: RemoteRuntimeMultiplexedTerminalState): void { + if (stream.ackFlushTimer !== null) { + clearTimeout(stream.ackFlushTimer) + stream.ackFlushTimer = null + } +} + +function discardOutputAcknowledgements(stream: RemoteRuntimeMultiplexedTerminalState): void { + clearAckFlushTimer(stream) + stream.pendingAckBytes = 0 + stream.heldAckBytes = 0 +} + function clearPendingSnapshotRequest(stream: RemoteRuntimeMultiplexedTerminalState): void { const request = stream.pendingSnapshotRequest stream.pendingSnapshotRequest = null diff --git a/src/renderer/src/runtime/remote-runtime-terminal-parse-backpressure.test.ts b/src/renderer/src/runtime/remote-runtime-terminal-parse-backpressure.test.ts index 645e99f4d..9d4c13bcf 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-parse-backpressure.test.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-parse-backpressure.test.ts @@ -10,6 +10,7 @@ import { describe('remote terminal renderer backpressure', () => { const sendBinary = vi.fn() + const unsubscribe = vi.fn() let callbacks: { onResponse: (response: unknown) => void onBinary: (bytes: Uint8Array) => void @@ -18,6 +19,7 @@ describe('remote terminal renderer backpressure', () => { beforeEach(() => { vi.resetModules() sendBinary.mockReset() + unsubscribe.mockReset() callbacks = null vi.stubGlobal('window', { api: { @@ -27,7 +29,7 @@ describe('remote terminal renderer backpressure', () => { queueMicrotask(() => { callbacks?.onResponse({ ok: true, result: { type: 'ready' } }) }) - return { unsubscribe: vi.fn(), sendBinary } + return { unsubscribe, sendBinary } }) } } @@ -101,10 +103,265 @@ describe('remote terminal renderer backpressure', () => { expect(sentAckBytes()).toEqual([]) parsedCallbacks.shift()?.() - expect(sentAckBytes()).toEqual([output.byteLength]) + await vi.waitFor(() => expect(sentAckBytes()).toEqual([output.byteLength])) stream.close() }) + it('batches parsed bulk output credit up to the byte threshold', async () => { + const { getRemoteRuntimeTerminalMultiplexer } = + await import('./remote-runtime-terminal-multiplexer') + const { takeCurrentTerminalDeliveryCredit } = + await import('../lib/pane-manager/terminal-delivery-credit') + const { writeTerminalOutput } = + await import('../lib/pane-manager/pane-terminal-output-scheduler') + const parsedCallbacks: (() => void)[] = [] + const terminal = { + write: vi.fn((_data: string, parsed?: () => void) => { + if (parsed) { + parsedCallbacks.push(parsed) + } + }) + } + const stream = await getRemoteRuntimeTerminalMultiplexer('windows-test').subscribeTerminal({ + terminal: 'term-bulk', + client: { id: 'mac-viewer', type: 'desktop' }, + callbacks: { + onData: (data) => { + writeTerminalOutput(terminal, data, { + foreground: true, + ackCredit: takeCurrentTerminalDeliveryCredit() ?? undefined + }) + }, + onSnapshot: vi.fn() + } + }) + sendBinary.mockClear() + const output = encodeTerminalStreamText('x'.repeat(64 * 1024)) + + for (let index = 0; index < 3; index += 1) { + callbacks?.onBinary( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Output, + streamId: stream.streamId, + seq: index + 1, + payload: output + }) + ) + } + + expect(sentAckBytes()).toEqual([]) + for (const parsed of parsedCallbacks) { + parsed() + } + expect(sentAckBytes()).toEqual([output.byteLength * 3]) + stream.close() + }) + + it('releases unknown streams and closes malformed connections instead of leaking credit', async () => { + const { getRemoteRuntimeTerminalMultiplexer } = + await import('./remote-runtime-terminal-multiplexer') + const multiplexer = getRemoteRuntimeTerminalMultiplexer('windows-test') + const stream = await multiplexer.subscribeTerminal({ + terminal: 'term-codex', + client: { id: 'mac-viewer', type: 'desktop' }, + callbacks: { onData: vi.fn(), onSnapshot: vi.fn() } + }) + sendBinary.mockClear() + + callbacks?.onBinary( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Output, + streamId: stream.streamId + 100, + seq: 1, + payload: encodeTerminalStreamText('x') + }) + ) + expect( + sendBinary.mock.calls.some(([bytes]) => { + const frame = decodeTerminalStreamFrame(bytes) + return ( + frame?.opcode === TerminalStreamOpcode.Unsubscribe && + frame.streamId === stream.streamId + 100 + ) + }) + ).toBe(true) + + callbacks?.onBinary(new Uint8Array([1, 2, 3])) + expect(unsubscribe).toHaveBeenCalledOnce() + }) + + it('credits malformed transformed output only after intentionally discarding it', async () => { + const { getRemoteRuntimeTerminalMultiplexer } = + await import('./remote-runtime-terminal-multiplexer') + const onData = vi.fn() + const stream = await getRemoteRuntimeTerminalMultiplexer('windows-test').subscribeTerminal({ + terminal: 'term-codex', + client: { id: 'mac-viewer', type: 'desktop' }, + callbacks: { onData, onSnapshot: vi.fn() } + }) + sendBinary.mockClear() + const malformed = encodeTerminalStreamJson({ data: 42, rawLength: 'wrong' }) + + callbacks?.onBinary( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.OutputSpan, + streamId: stream.streamId, + seq: 4, + payload: malformed + }) + ) + + expect(onData).not.toHaveBeenCalled() + await vi.waitFor(() => expect(sentAckBytes()).toEqual([malformed.byteLength])) + expect( + sendBinary.mock.calls.some(([bytes]) => { + const frame = decodeTerminalStreamFrame(bytes) + return frame?.opcode === TerminalStreamOpcode.SnapshotRequest + }) + ).toBe(true) + stream.close() + }) + + it('passes transformed sequence metadata and cancels pending credit on disposal', async () => { + const { getRemoteRuntimeTerminalMultiplexer } = + await import('./remote-runtime-terminal-multiplexer') + const onData = vi.fn() + const stream = await getRemoteRuntimeTerminalMultiplexer('windows-test').subscribeTerminal({ + terminal: 'term-codex', + client: { id: 'mac-viewer', type: 'desktop' }, + callbacks: { onData, onSnapshot: vi.fn() } + }) + sendBinary.mockClear() + const transformed = encodeTerminalStreamJson({ + data: 'visible', + rawLength: 11, + transformed: true + }) + callbacks?.onBinary( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.OutputSpan, + streamId: stream.streamId, + seq: 21, + payload: transformed + }) + ) + + expect(onData).toHaveBeenCalledWith('visible', { + seq: 21, + rawLength: 11, + transformed: true + }) + stream.close() + await new Promise((resolve) => setTimeout(resolve, 10)) + expect(sentAckBytes()).toEqual([]) + }) + + it('settles a late parser callback locally after the server ends the stream', async () => { + const { getRemoteRuntimeTerminalMultiplexer } = + await import('./remote-runtime-terminal-multiplexer') + const { takeCurrentTerminalDeliveryCredit } = + await import('../lib/pane-manager/terminal-delivery-credit') + const parsedCredits: (() => void)[] = [] + const stream = await getRemoteRuntimeTerminalMultiplexer('windows-test').subscribeTerminal({ + terminal: 'term-codex', + client: { id: 'mac-viewer', type: 'desktop' }, + callbacks: { + onData: () => { + const credit = takeCurrentTerminalDeliveryCredit() + if (credit) { + parsedCredits.push(credit) + } + }, + onSnapshot: vi.fn() + } + }) + sendBinary.mockClear() + callbacks?.onBinary( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Output, + streamId: stream.streamId, + seq: 1, + payload: encodeTerminalStreamText('x') + }) + ) + callbacks?.onResponse({ ok: true, result: { type: 'end', streamId: stream.streamId } }) + + parsedCredits[0]?.() + await new Promise((resolve) => setTimeout(resolve, 10)) + expect(sentAckBytes()).toEqual([]) + }) + + it('closes without ACKing when the renderer delivery callback throws', async () => { + const { getRemoteRuntimeTerminalMultiplexer } = + await import('./remote-runtime-terminal-multiplexer') + const stream = await getRemoteRuntimeTerminalMultiplexer('windows-test').subscribeTerminal({ + terminal: 'term-codex', + client: { id: 'mac-viewer', type: 'desktop' }, + callbacks: { + onData: () => { + throw new Error('renderer delivery failed') + }, + onSnapshot: vi.fn() + } + }) + sendBinary.mockClear() + + callbacks?.onBinary( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Output, + streamId: stream.streamId, + seq: 1, + payload: encodeTerminalStreamText('x') + }) + ) + + expect(unsubscribe).toHaveBeenCalledOnce() + await new Promise((resolve) => setTimeout(resolve, 10)) + expect(sentAckBytes()).toEqual([]) + }) + + it('closes and releases server debt when an ACK transport write throws', async () => { + const { getRemoteRuntimeTerminalMultiplexer } = + await import('./remote-runtime-terminal-multiplexer') + const { takeCurrentTerminalDeliveryCredit } = + await import('../lib/pane-manager/terminal-delivery-credit') + const parseCredits: (() => void)[] = [] + const stream = await getRemoteRuntimeTerminalMultiplexer('windows-test').subscribeTerminal({ + terminal: 'term-codex', + client: { id: 'mac-viewer', type: 'desktop' }, + callbacks: { + onData: () => { + const credit = takeCurrentTerminalDeliveryCredit() + if (credit) { + parseCredits.push(credit) + } + }, + onSnapshot: vi.fn() + } + }) + sendBinary.mockClear() + sendBinary.mockImplementation((bytes) => { + const frame = decodeTerminalStreamFrame(bytes) + if (frame?.opcode === TerminalStreamOpcode.Ack) { + throw new Error('socket closed') + } + }) + callbacks?.onBinary( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Output, + streamId: stream.streamId, + seq: 1, + payload: encodeTerminalStreamText('x') + }) + ) + + expect(parseCredits).toHaveLength(1) + parseCredits[0]?.() + + await vi.waitFor(() => expect(unsubscribe).toHaveBeenCalledOnce()) + expect(sentAckBytes()).toEqual([1]) + }) + function sentAckBytes(): number[] { return sendBinary.mock.calls.flatMap(([bytes]) => { const frame = decodeTerminalStreamFrame(bytes) diff --git a/src/renderer/src/runtime/runtime-terminal-stream.test.ts b/src/renderer/src/runtime/runtime-terminal-stream.test.ts index 385c7d9fe..a990b146e 100644 --- a/src/renderer/src/runtime/runtime-terminal-stream.test.ts +++ b/src/renderer/src/runtime/runtime-terminal-stream.test.ts @@ -120,6 +120,14 @@ describe('remote runtime terminal data subscriptions', () => { ) expect(watcher).toHaveBeenCalledWith('live') + await vi.waitFor(() => + expect( + sendBinary.mock.calls + .slice(1) + .map((call) => decodeTerminalStreamFrame(call[0])) + .some((frame) => frame?.opcode === TerminalStreamOpcode.Ack) + ).toBe(true) + ) const ackFrame = sendBinary.mock.calls .slice(1) .map((call) => decodeTerminalStreamFrame(call[0])) @@ -362,6 +370,13 @@ describe('remote runtime terminal multiplex ACK gate', () => { }) ) + await vi.waitFor(() => + expect( + sendBinary.mock.calls + .map((call) => decodeTerminalStreamFrame(call[0])) + .filter((frame) => frame?.opcode === TerminalStreamOpcode.Ack) + ).toHaveLength(1) + ) const immediateAckFrames = sendBinary.mock.calls .map((call) => decodeTerminalStreamFrame(call[0])) .filter((frame) => frame?.opcode === TerminalStreamOpcode.Ack) @@ -374,6 +389,13 @@ describe('remote runtime terminal multiplex ACK gate', () => { }) gate?.release() + await vi.waitFor(() => + expect( + sendBinary.mock.calls + .map((call) => decodeTerminalStreamFrame(call[0])) + .filter((frame) => frame?.opcode === TerminalStreamOpcode.Ack) + ).toHaveLength(2) + ) const allAckFrames = sendBinary.mock.calls .map((call) => decodeTerminalStreamFrame(call[0])) .filter((frame) => frame?.opcode === TerminalStreamOpcode.Ack) diff --git a/src/shared/terminal-multiplex-flow-control.ts b/src/shared/terminal-multiplex-flow-control.ts new file mode 100644 index 000000000..a14f45028 --- /dev/null +++ b/src/shared/terminal-multiplex-flow-control.ts @@ -0,0 +1,10 @@ +export const TERMINAL_STREAM_CHUNK_BYTES = 48 * 1024 +export const TERMINAL_OUTPUT_BATCH_MAX_BYTES = 64 * 1024 +export const TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES = 512 * 1024 +export const TERMINAL_MULTIPLEX_ACK_STREAM_MAX_WINDOW_BYTES = 2 * 1024 * 1024 +export const TERMINAL_MULTIPLEX_ACK_TOTAL_INITIAL_WINDOW_BYTES = 2 * 1024 * 1024 +export const TERMINAL_MULTIPLEX_ACK_TOTAL_MAX_WINDOW_BYTES = 8 * 1024 * 1024 +export const TERMINAL_MULTIPLEX_PENDING_MAX_BYTES = 256 * 1024 +export const TERMINAL_MULTIPLEX_ACK_BATCH_BYTES = 192 * 1024 +export const TERMINAL_MULTIPLEX_ACK_FLUSH_MS = 4 +export const TERMINAL_MULTIPLEX_MAX_STREAMS_PER_CONNECTION = 32