From bc36c91ae8ecf5d271cd2713637751248d40dddc Mon Sep 17 00:00:00 2001 From: OrcaWin <293788423+OrcaWin@users.noreply.github.com> Date: Tue, 28 Jul 2026 18:03:48 -0700 Subject: [PATCH] fix(ssh): prioritize mux control writes after drain --- config/reliability-gates.jsonc | 16 +++- ...h-channel-multiplexer-backpressure.test.ts | 20 +++-- .../ssh-multiplexer-transport-writer.test.ts | 46 +++++++++++- .../ssh/ssh-multiplexer-transport-writer.ts | 9 ++- .../ssh-multiplexer-writer-lane-scheduler.ts | 73 +++++++++++++++++++ 5 files changed, 151 insertions(+), 13 deletions(-) create mode 100644 src/main/ssh/ssh-multiplexer-writer-lane-scheduler.ts diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index cd98ec1d7e4..3594eb2a22e 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -3104,7 +3104,7 @@ "https://github.com/stablyai/orca/pull/7054" ], "invariant": "High-volume terminal output must stay bounded across daemon socket writes, SSH relay writer/decoder/source retention/reconnect, main model and projection admission, runtime metadata, detectors, and tail checks, main-to-renderer in-flight bytes, renderer scheduler queues, and hidden-output restore without starving focused input. Negotiated SSH spans remain exact and contiguous through activation, exit, cancellation, recovery, desktop projection, and required remote replacement; additional subscribers cannot stall the owner, while the required legacy primary retains backpressure.", - "oracle": "The current executable slice injects main-process renderer backlog pressure, then asserts unchanged path-provenance history reuse for pathless output, ordinary terminal-wait detection without a joined-tail allocation and with full prompt-family continuity, per-PTY and total pending-output caps, preserved sequenced-tail metadata, active-pending protection ahead of background trimming, and ACK-gated in-flight bounds. The joined source-intake/multiplex oracle blocks renderer or headless serialization, admits one snapshot-covered span and one trailing span, requires replacement reservation only after the authoritative sequence is known, advances upstream ACK eligibility exactly to the covered boundary after SnapshotEnd, and delivers then ACKs the trailing span through the ordinary live path without cancellation. Source-range ledger and multiplex seams additionally require cumulative byte ACK 40 then 100 for one 100-byte frame to release exact byte credit without early source settlement, admit a contiguous higher-generation recovered token while the prior token remains unsettled, and reject stale stream or source generations. Direct-SSH contracts additionally prove activation-response settlement before the first source frame, exact early-ACK reservation through send settlement or same-token retry, recovery-completion settlement before live admission, isolated saturated-subscriber eviction with healthy owner liveness, preserved gate-off primary backpressure, one-write ownership across write(false)/drain, liveness rebasing, failed-exit retention through late ACK or exact recovery, response-settlement cancellation authority, token-scoped exit cancellation, exact checkpoint-to-recoveryEnd continuity, stale-owner one-shot retry, no physical PTY teardown on recovery failure, exact provider-generation closure and once-only cleanup when recovery cancellation publication or proof rejects, 16 MiB-plus-1 MiB decoder caps, charged per-PTY/session retention budgets, and eight-wide isolated reattach. The live Electron perf oracle adds hidden-output floods, renderer scheduler queue depth, dropped-output-zero normal scenarios, and active key latency budgets before promotion.", + "oracle": "The current executable slice injects main-process renderer backlog pressure, then asserts unchanged path-provenance history reuse for pathless output, ordinary terminal-wait detection without a joined-tail allocation and with full prompt-family continuity, per-PTY and total pending-output caps, preserved sequenced-tail metadata, active-pending protection ahead of background trimming, and ACK-gated in-flight bounds. The joined source-intake/multiplex oracle blocks renderer or headless serialization, admits one snapshot-covered span and one trailing span, requires replacement reservation only after the authoritative sequence is known, advances upstream ACK eligibility exactly to the covered boundary after SnapshotEnd, and delivers then ACKs the trailing span through the ordinary live path without cancellation. Source-range ledger and multiplex seams additionally require cumulative byte ACK 40 then 100 for one 100-byte frame to release exact byte credit without early source settlement, admit a contiguous higher-generation recovered token while the prior token remains unsettled, and reject stale stream or source generations. Direct-SSH contracts additionally prove activation-response settlement before the first source frame, exact early-ACK reservation through send settlement or same-token retry, recovery-completion settlement before live admission, isolated saturated-subscriber eviction with healthy owner liveness, preserved gate-off primary backpressure, one-write ownership across write(false)/drain, control selection ahead of queued ordinary PTY input with FIFO lanes and bounded fairness, liveness rebasing, failed-exit retention through late ACK or exact recovery, response-settlement cancellation authority, token-scoped exit cancellation, exact checkpoint-to-recoveryEnd continuity, stale-owner one-shot retry, no physical PTY teardown on recovery failure, exact provider-generation closure and once-only cleanup when recovery cancellation publication or proof rejects, 16 MiB-plus-1 MiB decoder caps, charged per-PTY/session retention budgets, and eight-wide isolated reattach. The live Electron perf oracle adds hidden-output floods, renderer scheduler queue depth, dropped-output-zero normal scenarios, and active key latency budgets before promotion.", "commands": [ "pnpm exec vitest run --config config/vitest.config.ts src/main/ipc/pty.test.ts", "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orca-runtime-path-candidate-history.test.ts", @@ -3114,7 +3114,7 @@ "pnpm exec vitest run --config config/vitest.config.ts src/relay/dispatcher.test.ts src/relay/pty-handler-source-publication.test.ts --reporter=dot", "pnpm exec vitest run --config config/vitest.config.ts src/main/ssh/ssh-relay-session-recovery-races.test.ts --reporter=dot", "pnpm exec vitest run --config config/vitest.config.ts src/main/ipc/ssh-pty-remote-source-range-consumers.test.ts src/main/runtime/rpc/terminal-source-range-ledger.test.ts src/main/runtime/rpc/terminal-multiplex.test.ts --reporter=dot", - "pnpm exec vitest run --config config/vitest.config.ts src/shared/pty-consumer-session.test.ts src/main/ipc/ssh-pty-output-intake.test.ts src/main/ipc/ssh-pty-remote-source-range-consumers.test.ts src/main/runtime/rpc/terminal-multiplex.test.ts src/main/ssh/ssh-multiplexer-transport-writer.test.ts src/main/ssh/ssh-channel-multiplexer.test.ts src/main/ssh/ssh-pty-source-credit-rollout.test.ts src/main/ssh/ssh-relay-deploy.test.ts src/relay/dispatcher-client-writer.test.ts --reporter=dot", + "pnpm exec vitest run --config config/vitest.config.ts src/shared/pty-consumer-session.test.ts src/main/ipc/ssh-pty-output-intake.test.ts src/main/ipc/ssh-pty-remote-source-range-consumers.test.ts src/main/runtime/rpc/terminal-multiplex.test.ts src/main/ssh/ssh-multiplexer-transport-writer.test.ts src/main/ssh/ssh-channel-multiplexer.test.ts src/main/ssh/ssh-channel-multiplexer-backpressure.test.ts src/main/ssh/ssh-pty-source-credit-rollout.test.ts src/main/ssh/ssh-relay-deploy.test.ts src/relay/dispatcher-client-writer.test.ts --reporter=dot", "pnpm exec vitest run --config config/vitest.config.ts src/main/agent-hooks/wsl-hook-relay-sentinel.test.ts --reporter=dot", "ORCA_E2E_SSH_DOCKER=1 ORCA_SSH_PTY_SOURCE_CREDIT_V1=1 pnpm exec playwright test tests/e2e/ssh-docker-relay-perf.spec.ts --config tests/playwright.config.ts --project electron-headless --workers=1" ], @@ -3148,6 +3148,7 @@ "src/main/runtime/rpc/terminal-multiplex.test.ts", "src/main/ssh/ssh-multiplexer-transport-writer.test.ts", "src/main/ssh/ssh-channel-multiplexer.test.ts", + "src/main/ssh/ssh-channel-multiplexer-backpressure.test.ts", "src/main/ssh/ssh-pty-source-credit-rollout.test.ts", "src/main/ssh/ssh-relay-deploy.test.ts", "src/relay/dispatcher-client-writer.test.ts", @@ -3309,10 +3310,19 @@ "file": "src/main/ssh/ssh-multiplexer-transport-writer.test.ts", "assertions": [ "write(false) owns one frame and later ordinary traffic waits for drain", + "queued control is selected before ordinary PTY backlog at drain while each lane remains FIFO", + "four-control fairness guarantees ordinary progress without starving control", "one coalesced liveness bypass is allowed per saturated epoch", "saturation and drain transitions are reported exactly once" ] }, + { + "file": "src/main/ssh/ssh-channel-multiplexer-backpressure.test.ts", + "assertions": [ + "source ACK, cancellation, exit, request, and response control frames precede queued pasted PTY input after drain", + "control and ordinary frames preserve lane FIFO while ordinary input progresses after four control writes" + ] + }, { "file": "src/main/ssh/ssh-channel-multiplexer.test.ts", "assertions": [ @@ -3408,7 +3418,7 @@ }, "redGreenEvidence": { "status": "partial", - "evidence": "Tests assert pathless runtime output reuses unchanged path-provenance history with zero old-candidate byte scans, ordinary saturated terminal tails retain no rebuilt wait text or repeated phrase scans while every blocked/ready prompt family remains live, main pending renderer output is capped per PTY and in total, direct-SSH source/recovery memory and both frame decoders stay within exact caps, and 50-PTY reconnect isolates slow/failing siblings. Before the snapshot-admission fix, the focused consumer oracle failed because the trailing span entered transferring instead of remaining open; after retaining model-sequence ends and reserving at the serialized fence, the joined main/runtime oracle passed for renderer and headless sources with upstream boundaries 4 then 8. Before cumulative partial ACK and source-rotation support, the ledger rejected ACK 40 of 100 and the multiplex seam detached instead of publishing recovered token B; both exact oracles pass while stale stream and source generations remain rejected. Other controls previously failed with serial reattach, UTF-8-only accounting, and roughly 32 MiB decoder retention. Needs daemon-stream contract coverage and broader hidden-output/input-latency perf artifacts before promotion." + "evidence": "Tests assert pathless runtime output reuses unchanged path-provenance history with zero old-candidate byte scans, ordinary saturated terminal tails retain no rebuilt wait text or repeated phrase scans while every blocked/ready prompt family remains live, main pending renderer output is capped per PTY and in total, direct-SSH source/recovery memory and both frame decoders stay within exact caps, and 50-PTY reconnect isolates slow/failing siblings. The deterministic main-to-relay drain oracle failed with ordinary-2 and ordinary-3 ahead of a later control frame under the single FIFO, then passed with control-first lane selection and four-write ordinary fairness. Before the snapshot-admission fix, the focused consumer oracle failed because the trailing span entered transferring instead of remaining open; after retaining model-sequence ends and reserving at the serialized fence, the joined main/runtime oracle passed for renderer and headless sources with upstream boundaries 4 then 8. Before cumulative partial ACK and source-rotation support, the ledger rejected ACK 40 of 100 and the multiplex seam detached instead of publishing recovered token B; both exact oracles pass while stale stream and source generations remain rejected. Other controls previously failed with serial reattach, UTF-8-only accounting, and roughly 32 MiB decoder retention. Needs daemon-stream contract coverage and broader hidden-output/input-latency perf artifacts before promotion." }, "performanceBudget": { "required": true, diff --git a/src/main/ssh/ssh-channel-multiplexer-backpressure.test.ts b/src/main/ssh/ssh-channel-multiplexer-backpressure.test.ts index 582a3d66c35..7f6100f6d75 100644 --- a/src/main/ssh/ssh-channel-multiplexer-backpressure.test.ts +++ b/src/main/ssh/ssh-channel-multiplexer-backpressure.test.ts @@ -136,7 +136,7 @@ describe('SshChannelMultiplexer backpressure hardening', () => { expect(seen).toEqual(Array.from({ length: 65 }, (_, index) => index)) }) - it('keeps acknowledgements, cancellation, exit, requests, and responses ordered after drain', async () => { + it('prioritizes source ACK and control while preserving FIFO and ordinary progress', async () => { mux.dispose() const written: Buffer[] = [] let drain = (): void => {} @@ -163,7 +163,9 @@ describe('SshChannelMultiplexer backpressure hardening', () => { mux.onRequest('client.control', () => ({ accepted: true })) const controller = new AbortController() - mux.notify('pty.data', { id: 'pty-1', data: 'ordinary' }) + mux.notify('pty.data', { id: 'pty-1', data: 'ordinary-1' }) + mux.notify('pty.data', { id: 'pty-1', data: 'ordinary-2' }) + mux.notify('pty.data', { id: 'pty-1', data: 'ordinary-3' }) mux.notify('pty.ackData', { acknowledgements: [] }) const request = mux.request('fs.scan', {}, { signal: controller.signal }) controller.abort() @@ -180,13 +182,21 @@ describe('SshChannelMultiplexer backpressure hardening', () => { const payloads = written.map((frame) => JSON.parse(frame.subarray(HEADER_LENGTH, HEADER_LENGTH + frame.readUInt32BE(9)).toString()) ) - expect(payloads.map((payload) => payload.method ?? `response:${payload.id}`)).toEqual([ - 'pty.data', + expect( + payloads.map((payload) => + payload.method === 'pty.data' + ? `pty.data:${payload.params.data}` + : (payload.method ?? `response:${payload.id}`) + ) + ).toEqual([ + 'pty.data:ordinary-1', 'pty.ackData', 'fs.scan', 'rpc.cancel', 'pty.exit', - 'response:91' + 'pty.data:ordinary-2', + 'response:91', + 'pty.data:ordinary-3' ]) await expect(request).rejects.toMatchObject({ name: 'AbortError' }) }) diff --git a/src/main/ssh/ssh-multiplexer-transport-writer.test.ts b/src/main/ssh/ssh-multiplexer-transport-writer.test.ts index 649af6d04d8..ffae9948226 100644 --- a/src/main/ssh/ssh-multiplexer-transport-writer.test.ts +++ b/src/main/ssh/ssh-multiplexer-transport-writer.test.ts @@ -47,7 +47,51 @@ function transportHarness(writeResults: (boolean | void)[]): WriterHarness { } describe('SshMultiplexerTransportWriter', () => { - it('preserves FIFO order and waits for drain after write(false)', () => { + it('selects queued control before ordinary backlog at the drain boundary', () => { + const harness = transportHarness([false, true, true, true]) + const writer = new SshMultiplexerTransportWriter(harness.transport, vi.fn()) + + writer.enqueue(Buffer.from('ordinary-1'), 'ordinary') + writer.enqueue(Buffer.from('ordinary-2'), 'ordinary') + writer.enqueue(Buffer.from('ordinary-3'), 'ordinary') + writer.enqueue(Buffer.from('control'), 'control') + + expect(harness.writes.map(String)).toEqual(['ordinary-1']) + harness.drain() + expect(harness.writes.map(String)).toEqual([ + 'ordinary-1', + 'control', + 'ordinary-2', + 'ordinary-3' + ]) + writer.dispose() + }) + + it('preserves FIFO within each lane and prevents ordinary starvation', () => { + const harness = transportHarness([false, ...Array(8).fill(true)]) + const writer = new SshMultiplexerTransportWriter(harness.transport, vi.fn()) + + writer.enqueue(Buffer.from('ordinary-1'), 'ordinary') + writer.enqueue(Buffer.from('ordinary-2'), 'ordinary') + for (let index = 1; index <= 6; index++) { + writer.enqueue(Buffer.from(`control-${index}`), 'control') + } + + harness.drain() + expect(harness.writes.map(String)).toEqual([ + 'ordinary-1', + 'control-1', + 'control-2', + 'control-3', + 'control-4', + 'ordinary-2', + 'control-5', + 'control-6' + ]) + writer.dispose() + }) + + it('waits for drain and settles each write once', () => { const harness = transportHarness([false, true, true]) const failed = vi.fn() const writer = new SshMultiplexerTransportWriter(harness.transport, failed) diff --git a/src/main/ssh/ssh-multiplexer-transport-writer.ts b/src/main/ssh/ssh-multiplexer-transport-writer.ts index 287ad4e0fcb..d428e85889a 100644 --- a/src/main/ssh/ssh-multiplexer-transport-writer.ts +++ b/src/main/ssh/ssh-multiplexer-transport-writer.ts @@ -1,4 +1,5 @@ import { HEADER_LENGTH, MAX_MESSAGE_SIZE } from './relay-protocol' +import { SshMultiplexerWriterLaneScheduler } from './ssh-multiplexer-writer-lane-scheduler' export type MultiplexerWriteSettlement = { ok: true } | { ok: false; error: Error } @@ -41,7 +42,7 @@ function onceSettlement( } export class SshMultiplexerTransportWriter { - private readonly queue: WriterEntry[] = [] + private readonly scheduler = new SshMultiplexerWriterLaneScheduler() private readonly inFlight = new Set() private readonly settleOnDrain = new Set() private ordinaryBytes = 0 @@ -91,7 +92,7 @@ export class SshMultiplexerTransportWriter { if (lane === 'liveness' && this.saturated) { this.writeEntry(entry) } else { - this.queue.push(entry) + this.scheduler.enqueue(entry, lane) this.pump() } return true @@ -105,7 +106,7 @@ export class SshMultiplexerTransportWriter { this.saturated = false this.removeDrainListener?.() this.removeDrainListener = null - for (const entry of this.queue.splice(0)) { + for (const entry of this.scheduler.clear()) { this.release(entry, { ok: false, error }) } for (const entry of Array.from(this.inFlight)) { @@ -133,7 +134,7 @@ export class SshMultiplexerTransportWriter { this.pumping = true try { while (!this.closed && !this.saturated) { - const entry = this.queue.shift() + const entry = this.scheduler.select() if (!entry) { return } diff --git a/src/main/ssh/ssh-multiplexer-writer-lane-scheduler.ts b/src/main/ssh/ssh-multiplexer-writer-lane-scheduler.ts new file mode 100644 index 00000000000..74ad5d593b6 --- /dev/null +++ b/src/main/ssh/ssh-multiplexer-writer-lane-scheduler.ts @@ -0,0 +1,73 @@ +import type { MultiplexerWriterLane } from './ssh-multiplexer-transport-writer' + +const CONTROL_WRITES_BEFORE_ORDINARY = 4 + +type LaneQueue = { + entries: T[] + head: number +} + +function createLaneQueue(): LaneQueue { + return { entries: [], head: 0 } +} + +function hasEntries(queue: LaneQueue): boolean { + return queue.head < queue.entries.length +} + +function shift(queue: LaneQueue): T | undefined { + const entry = queue.entries[queue.head] + if (entry === undefined) { + return undefined + } + queue.head += 1 + if (queue.head === queue.entries.length) { + queue.entries.length = 0 + queue.head = 0 + } + return entry +} + +function clear(queue: LaneQueue): T[] { + const entries = queue.entries.slice(queue.head) + queue.entries.length = 0 + queue.head = 0 + return entries +} + +export class SshMultiplexerWriterLaneScheduler { + private readonly ordinary = createLaneQueue() + private readonly control = createLaneQueue() + private readonly liveness = createLaneQueue() + private controlWritesSinceOrdinary = 0 + + enqueue(entry: T, lane: MultiplexerWriterLane): void { + this[lane].entries.push(entry) + } + + select(): T | undefined { + const liveness = shift(this.liveness) + if (liveness) { + return liveness + } + if ( + hasEntries(this.control) && + (!hasEntries(this.ordinary) || + this.controlWritesSinceOrdinary < CONTROL_WRITES_BEFORE_ORDINARY) + ) { + this.controlWritesSinceOrdinary += 1 + return shift(this.control) + } + const ordinary = shift(this.ordinary) + if (ordinary) { + this.controlWritesSinceOrdinary = 0 + return ordinary + } + return shift(this.control) + } + + clear(): T[] { + this.controlWritesSinceOrdinary = 0 + return [...clear(this.liveness), ...clear(this.control), ...clear(this.ordinary)] + } +}