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.
This commit is contained in:
OrcaWin
GitHub
parent
88b7e69ba1
commit
72a2d7bc7d
|
|
@ -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."
|
||||
},
|
||||
|
|
|
|||
|
|
@ -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<boolean>
|
||||
// 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<number, TerminalMultiplexStream>()
|
||||
const pendingPtyWaitControllers = new Map<number, Set<AbortController>>()
|
||||
let ackTotalInFlightBytes = 0
|
||||
let ackTotalWindowBytes = TERMINAL_MULTIPLEX_ACK_TOTAL_INITIAL_WINDOW_BYTES
|
||||
let ackFlushCursorStreamId: number | null = null
|
||||
let resolveMultiplex = (): void => {}
|
||||
const multiplexClosed = new Promise<void>((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,
|
||||
|
|
|
|||
|
|
@ -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<ParserMeasurement> {
|
||||
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<void>((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<number, { streamIndex: number; bytes: number }[]>()
|
||||
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<number, ParserMeasurement>()
|
||||
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)
|
||||
})
|
||||
})
|
||||
|
|
@ -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])
|
||||
})
|
||||
})
|
||||
|
|
@ -0,0 +1,41 @@
|
|||
type TerminalMultiplexDrainStream = { streamId: number }
|
||||
|
||||
export function drainTerminalMultiplexRoundRobin<T extends TerminalMultiplexDrainStream>(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<T extends TerminalMultiplexDrainStream>(
|
||||
streams: readonly T[],
|
||||
cursorStreamId: number | null
|
||||
): number {
|
||||
if (cursorStreamId === null) {
|
||||
return 0
|
||||
}
|
||||
return (streams.findIndex((stream) => stream.streamId === cursorStreamId) + 1) % streams.length
|
||||
}
|
||||
|
|
@ -43,7 +43,8 @@ function makeRequest(method: string, params?: unknown): RpcRequest {
|
|||
|
||||
function startDesktopMultiplexSubscribe(
|
||||
overrides: Partial<OrcaRuntimeService> = {},
|
||||
trace?: string[]
|
||||
trace?: string[],
|
||||
sendBinaryOverride?: (bytes: Uint8Array<ArrayBufferLike>) => boolean | void
|
||||
) {
|
||||
const messages: string[] = []
|
||||
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
|
||||
|
|
@ -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<ArrayBufferLike>[] = []
|
||||
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<ArrayBufferLike>[] = []
|
||||
|
|
@ -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<string>((_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 }),
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
})
|
||||
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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 () => {
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
})
|
||||
})
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<typeof setTimeout> | null
|
||||
snapshotChunks: Uint8Array<ArrayBufferLike>[]
|
||||
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<RemoteRuntimeMultiplexedTerminalState> {
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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<ArrayBufferLike>) => 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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
Loading…
Reference in New Issue