From b606117eedfa654030bac3bdc6f5f2a814ca794a Mon Sep 17 00:00:00 2001 From: OrcaWin Date: Wed, 22 Jul 2026 19:03:42 -0400 Subject: [PATCH] Fix paired remote terminal parse backpressure (#9683) * fix(terminal): defer remote output ACKs until parse * test(terminal): document synchronous credit claims --------- Co-authored-by: OrcaWin <293788423+OrcaWin@users.noreply.github.com> --- .../terminal-pane/pty-connection.test.ts | 18 ++- .../terminal-pane/pty-connection.ts | 50 ++++++-- .../terminal-pane/terminal-pty-ack-gate.ts | 41 +----- .../terminal-delivery-credit.test.ts | 44 +++++++ .../pane-manager/terminal-delivery-credit.ts | 49 ++++++++ .../remote-runtime-terminal-multiplexer.ts | 22 ++-- ...untime-terminal-parse-backpressure.test.ts | 118 ++++++++++++++++++ 7 files changed, 289 insertions(+), 53 deletions(-) create mode 100644 src/renderer/src/lib/pane-manager/terminal-delivery-credit.test.ts create mode 100644 src/renderer/src/lib/pane-manager/terminal-delivery-credit.ts create mode 100644 src/renderer/src/runtime/remote-runtime-terminal-parse-backpressure.test.ts diff --git a/src/renderer/src/components/terminal-pane/pty-connection.test.ts b/src/renderer/src/components/terminal-pane/pty-connection.test.ts index 64704880bc3..7e70b5474b4 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection.test.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection.test.ts @@ -13463,6 +13463,8 @@ describe('connectPanePty', () => { it('holds newer live bytes until a later replay frame has fully parsed', async () => { const { connectPanePty } = await import('./pty-connection') + const { deliverTerminalDataWithDeferredCredit } = + await import('@/lib/pane-manager/terminal-delivery-credit') enableActiveRuntimeEnvironment() const transport = createMockTransport('remote:env-1@@terminal-live-order') const callbacksRef: { @@ -13484,8 +13486,12 @@ describe('connectPanePty', () => { await flushAsyncTicks(8) expect(writes).toEqual(['\x1b[2J\x1b[3J\x1b[H']) - callbacksRef.data?.('NEWER-LIVE\r\n') + const acknowledgeLiveFrame = vi.fn() + deliverTerminalDataWithDeferredCredit(acknowledgeLiveFrame, () => { + callbacksRef.data?.('NEWER-LIVE\r\n') + }) expect(writes).not.toContain('NEWER-LIVE\r\n') + expect(acknowledgeLiveFrame).not.toHaveBeenCalled() for (let index = 0; index < 12 && parseCallbacks.length > 0; index += 1) { parseCallbacks.shift()?.() await flushAsyncTicks(4) @@ -13498,7 +13504,17 @@ describe('connectPanePty', () => { expect(replayIndex).toBeGreaterThan(0) expect(resetIndex).toBeGreaterThan(replayIndex) expect(liveIndex).toBeGreaterThan(resetIndex) + expect(acknowledgeLiveFrame).toHaveBeenCalledOnce() + + callbacksRef.replay?.('stalled replay') + await flushAsyncTicks(8) + const acknowledgeDisposedFrame = vi.fn() + deliverTerminalDataWithDeferredCredit(acknowledgeDisposedFrame, () => { + callbacksRef.data?.('LIVE-BEHIND-STALLED-REPLAY') + }) + expect(acknowledgeDisposedFrame).not.toHaveBeenCalled() binding.dispose() + expect(acknowledgeDisposedFrame).toHaveBeenCalledOnce() }) it('drops a queued relay replay instead of retagging it for a replacement PTY', async () => { diff --git a/src/renderer/src/components/terminal-pane/pty-connection.ts b/src/renderer/src/components/terminal-pane/pty-connection.ts index 821a9688d17..0a8505cde68 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection.ts @@ -23,7 +23,10 @@ import { isStatefulRendererReplyCsiQuery, isStatelessRendererReplyCsiQuery } from '../../../../shared/terminal-reply-query-extraction' -import { takeCurrentPtyDeliveryAckCredit } from './terminal-pty-ack-gate' +import { + deliverTerminalDataWithDeferredCredit, + takeCurrentTerminalDeliveryCredit +} from '@/lib/pane-manager/terminal-delivery-credit' import { serializeWithAbsoluteCursor } from '../../../../shared/terminal-serialize-absolute-cursor' import { isTerminalQueryReply } from '../../../../shared/terminal-query-reply' import type { PtyBufferSnapshot, PtyConnectResult } from './pty-transport' @@ -3312,7 +3315,13 @@ export function connectPanePty( let lastTerminalInputAt = Number.NEGATIVE_INFINITY let hasReceivedPtyOutput = false let deferredReattachLiveData: - | { data: string; ptyId: string | null; streamGeneration: number; meta?: PtyDataMeta }[] + | { + data: string + ptyId: string | null + streamGeneration: number + meta?: PtyDataMeta + ackCredit?: () => void + }[] | null = null let deferredReattachLiveDataChars = 0 let reattachLiveDataDeferralDepth = 0 @@ -5787,7 +5796,7 @@ export function connectPanePty( 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. - ackCredit: takeCurrentPtyDeliveryAckCredit() ?? undefined, + ackCredit: takeCurrentTerminalDeliveryCredit() ?? undefined, onBackgroundBacklogDropped: markHiddenOutputRestoreNeeded, latencySensitive: !foreground || parseHiddenStartupOutput @@ -6881,20 +6890,26 @@ export function connectPanePty( } if (deferredReattachLiveData !== null) { // Why: a replacement stream must not inherit bytes or a gap marker from the replay owner it superseded. - deferredReattachLiveData = deferredReattachLiveData.filter( - (chunk) => chunk.streamGeneration === streamGeneration - ) + deferredReattachLiveData = deferredReattachLiveData.filter((chunk) => { + const keep = chunk.streamGeneration === streamGeneration + if (!keep) { + chunk.ackCredit?.() + } + return keep + }) deferredReattachLiveDataChars = deferredReattachLiveData.reduce( (total, chunk) => total + chunk.data.length, 0 ) const oversized = data.length > MAX_DEFERRED_REATTACH_LIVE_CHARS const deferredData = oversized ? data.slice(-MAX_DEFERRED_REATTACH_LIVE_CHARS) : data + const ackCredit = takeCurrentTerminalDeliveryCredit() deferredReattachLiveData.push({ data: deferredData, ptyId: transport.getPtyId(), streamGeneration, - ...(meta ? { meta } : {}) + ...(meta ? { meta } : {}), + ...(ackCredit ? { ackCredit } : {}) }) deferredReattachLiveDataChars += deferredData.length // Why: one huge IPC frame would bypass the queue's memory bound; mark a stream gap so snapshot recovery replaces it, not a partial ANSI frame. @@ -6906,6 +6921,7 @@ export function connectPanePty( ) { const removed = deferredReattachLiveData.shift() deferredReattachLiveDataChars -= removed?.data.length ?? 0 + removed?.ackCredit?.() dropped = true } if (dropped && deferredReattachLiveData[0]) { @@ -7111,6 +7127,9 @@ export function connectPanePty( const currentOwner = deferredReattachLiveDataOwners.get(currentGeneration) deferredReattachLiveDataOwners = new Map() if (disposed || !chunks) { + for (const chunk of chunks ?? []) { + chunk.ackCredit?.() + } return } // Why: paint the authoritative replay first, then admit deferred live chunks so the replay clear can't erase newer output. @@ -7121,9 +7140,16 @@ export function connectPanePty( chunk.streamGeneration !== currentGeneration || currentOwner?.failed === true ) { + chunk.ackCredit?.() continue } - dataCallback(chunk.data, chunk.meta, chunk.streamGeneration) + if (chunk.ackCredit) { + deliverTerminalDataWithDeferredCredit(chunk.ackCredit, () => { + dataCallback(chunk.data, chunk.meta, chunk.streamGeneration) + }) + } else { + dataCallback(chunk.data, chunk.meta, chunk.streamGeneration) + } deliveredDeferredChunks += 1 } if (deliveredDeferredChunks > 0) { @@ -8157,6 +8183,14 @@ export function connectPanePty( reconcileIfSessionMissing, dispose() { disposed = true + // Why: a stalled xterm replay may never reach its finally; release live-frame credit when this renderer no longer owns the stream. + for (const chunk of deferredReattachLiveData ?? []) { + chunk.ackCredit?.() + } + deferredReattachLiveData = null + deferredReattachLiveDataChars = 0 + reattachLiveDataDeferralDepth = 0 + deferredReattachLiveDataOwners = new Map() cancelPendingSafeFitContinuations(pane) pendingHiddenSnapshotFit = null pendingReattachFit = null diff --git a/src/renderer/src/components/terminal-pane/terminal-pty-ack-gate.ts b/src/renderer/src/components/terminal-pane/terminal-pty-ack-gate.ts index 31733aebc47..8340dfa1b48 100644 --- a/src/renderer/src/components/terminal-pane/terminal-pty-ack-gate.ts +++ b/src/renderer/src/components/terminal-pane/terminal-pty-ack-gate.ts @@ -1,4 +1,8 @@ import { e2eConfig } from '@/lib/e2e-config' +import { + deliverTerminalDataWithDeferredCredit, + takeCurrentTerminalDeliveryCredit +} from '@/lib/pane-manager/terminal-delivery-credit' type E2eTerminalPtyAckGateSnapshot = { gatedPtyCount: number @@ -89,25 +93,6 @@ export function ackPtyData(ptyId: string, chars: number): void { // true parse backpressure and main's producer flow control pauses the shell // instead of dropping. -type DeferredPtyAckCredit = { - ptyId: string - chars: number - claimed: boolean - credited: boolean -} - -let currentDeliveryCredit: DeferredPtyAckCredit | null = null - -function creditDeferredPtyAck(credit: DeferredPtyAckCredit): void { - // Why fire-once: split queue chunks and discard paths may both touch the - // same delivery; the invariant is exactly one credit per delivered chunk. - if (credit.credited) { - return - } - credit.credited = true - ackPtyData(credit.ptyId, credit.chars) -} - /** Runs one pty:data delivery with a parse-deferred ACK credit. If the * handler hands bytes to the output scheduler, the claimed credit fires when * the scheduler consumes (writes or discards) them; any credit left @@ -118,28 +103,14 @@ export function deliverPtyDataWithDeferredAck( chars: number, deliver: () => void ): void { - const credit: DeferredPtyAckCredit = { ptyId, chars, claimed: false, credited: false } - currentDeliveryCredit = credit - try { - deliver() - } finally { - currentDeliveryCredit = null - if (!credit.claimed) { - creditDeferredPtyAck(credit) - } - } + 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 { - const credit = currentDeliveryCredit - if (!credit || credit.claimed) { - return null - } - credit.claimed = true - return () => creditDeferredPtyAck(credit) + return takeCurrentTerminalDeliveryCredit() } export function getProcessedPtyCharTotals(): Record { diff --git a/src/renderer/src/lib/pane-manager/terminal-delivery-credit.test.ts b/src/renderer/src/lib/pane-manager/terminal-delivery-credit.test.ts new file mode 100644 index 00000000000..01e20fb3fbd --- /dev/null +++ b/src/renderer/src/lib/pane-manager/terminal-delivery-credit.test.ts @@ -0,0 +1,44 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +describe('terminal delivery credit', () => { + beforeEach(() => { + vi.resetModules() + }) + + it('restores an outer delivery after a nested delivery returns', async () => { + const { deliverTerminalDataWithDeferredCredit, takeCurrentTerminalDeliveryCredit } = + await import('./terminal-delivery-credit') + const completeOuter = vi.fn() + const completeInner = vi.fn() + let outerCredit: (() => void) | null = null + let innerCredit: (() => void) | null = null + + deliverTerminalDataWithDeferredCredit(completeOuter, () => { + deliverTerminalDataWithDeferredCredit(completeInner, () => { + innerCredit = takeCurrentTerminalDeliveryCredit() + }) + outerCredit = takeCurrentTerminalDeliveryCredit() + }) + + expect(completeOuter).not.toHaveBeenCalled() + expect(completeInner).not.toHaveBeenCalled() + innerCredit!() + outerCredit!() + expect(completeInner).toHaveBeenCalledOnce() + expect(completeOuter).toHaveBeenCalledOnce() + }) + + it('auto-settles before a deferred consumer can claim the delivery', async () => { + const { deliverTerminalDataWithDeferredCredit, takeCurrentTerminalDeliveryCredit } = + await import('./terminal-delivery-credit') + const complete = vi.fn() + let claimLater: (() => (() => void) | null) | null = null + + deliverTerminalDataWithDeferredCredit(complete, () => { + claimLater = takeCurrentTerminalDeliveryCredit + }) + + expect(complete).toHaveBeenCalledOnce() + expect(claimLater!()).toBeNull() + }) +}) diff --git a/src/renderer/src/lib/pane-manager/terminal-delivery-credit.ts b/src/renderer/src/lib/pane-manager/terminal-delivery-credit.ts new file mode 100644 index 00000000000..d467cedda23 --- /dev/null +++ b/src/renderer/src/lib/pane-manager/terminal-delivery-credit.ts @@ -0,0 +1,49 @@ +type TerminalDeliveryCredit = { + complete: () => void + claimed: boolean + credited: boolean +} + +// Why: consumers must claim during deliver(); after it returns this synchronous slot is restored and unclaimed credit settles. +let currentDeliveryCredit: TerminalDeliveryCredit | null = null + +function completeTerminalDeliveryCredit(credit: TerminalDeliveryCredit): void { + // Why: queue splitting and discard paths can both settle one delivery. + if (credit.credited) { + return + } + credit.credited = true + credit.complete() +} + +/** Defers producer credit until the output scheduler consumes or discards the delivery. */ +export function deliverTerminalDataWithDeferredCredit( + complete: () => void, + deliver: () => void +): void { + const credit: TerminalDeliveryCredit = { + complete, + claimed: false, + credited: false + } + const previousCredit = currentDeliveryCredit + currentDeliveryCredit = credit + try { + deliver() + } finally { + currentDeliveryCredit = previousCredit + if (!credit.claimed) { + 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) { + return null + } + credit.claimed = true + return () => completeTerminalDeliveryCredit(credit) +} diff --git a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts index 4823c119430..12e4ea7f8b3 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts @@ -11,6 +11,7 @@ import { encodeTerminalStreamText } from '../../../shared/terminal-stream-protocol' import { e2eConfig } from '@/lib/e2e-config' +import { deliverTerminalDataWithDeferredCredit } from '@/lib/pane-manager/terminal-delivery-credit' import { unwrapRuntimeRpcResult } from './runtime-rpc-client' type RuntimeEnvironmentSubscriptionHandle = { @@ -472,7 +473,7 @@ class RemoteRuntimeTerminalMultiplexer { ? (span!.data as string) : '' : decodeTerminalStreamText(frame.payload) - try { + const deliverOutput = (): void => { if (!validSpan) { // Why: rendering malformed span JSON would expose protocol framing // as terminal text and lose its raw sequence accounting. @@ -501,15 +502,18 @@ class RemoteRuntimeTerminalMultiplexer { rawLength, ...(frame.opcode === TerminalStreamOpcode.OutputSpan ? { transformed: true } : {}) }) - } finally { - if (stream.acknowledgeOutput) { - if (shouldHoldE2eRemoteTerminalAck(stream.terminal)) { - stream.heldAckBytes += frame.payload.byteLength - } else { - this.acknowledgeOutput(stream, frame.payload.byteLength) - } - } } + if (!stream.acknowledgeOutput) { + deliverOutput() + return + } + deliverTerminalDataWithDeferredCredit(() => { + if (shouldHoldE2eRemoteTerminalAck(stream.terminal)) { + stream.heldAckBytes += frame.payload.byteLength + } else { + this.acknowledgeOutput(stream, frame.payload.byteLength) + } + }, deliverOutput) return } if (frame.opcode === TerminalStreamOpcode.SnapshotStart) { diff --git a/src/renderer/src/runtime/remote-runtime-terminal-parse-backpressure.test.ts b/src/renderer/src/runtime/remote-runtime-terminal-parse-backpressure.test.ts new file mode 100644 index 00000000000..645e99f4d7f --- /dev/null +++ b/src/renderer/src/runtime/remote-runtime-terminal-parse-backpressure.test.ts @@ -0,0 +1,118 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { + TerminalStreamOpcode, + decodeTerminalStreamFrame, + decodeTerminalStreamJson, + encodeTerminalStreamFrame, + encodeTerminalStreamJson, + encodeTerminalStreamText +} from '../../../shared/terminal-stream-protocol' + +describe('remote terminal renderer backpressure', () => { + const sendBinary = vi.fn() + let callbacks: { + onResponse: (response: unknown) => void + onBinary: (bytes: Uint8Array) => void + } | null = null + + beforeEach(() => { + vi.resetModules() + sendBinary.mockReset() + callbacks = null + vi.stubGlobal('window', { + api: { + runtimeEnvironments: { + subscribe: vi.fn(async (_args, nextCallbacks) => { + callbacks = nextCallbacks + queueMicrotask(() => { + callbacks?.onResponse({ ok: true, result: { type: 'ready' } }) + }) + return { unsubscribe: vi.fn(), sendBinary } + }) + } + } + }) + }) + + afterEach(() => { + vi.unstubAllGlobals() + }) + + it('withholds server credit until xterm consumes the output frame', 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-codex', + client: { id: 'mac-viewer', type: 'desktop' }, + callbacks: { + onData: (data) => { + writeTerminalOutput(terminal, data, { + foreground: true, + ackCredit: takeCurrentTerminalDeliveryCredit() ?? undefined + }) + }, + onSnapshot: vi.fn() + } + }) + + callbacks?.onBinary( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.SnapshotStart, + streamId: stream.streamId, + seq: 1, + payload: encodeTerminalStreamJson({ kind: 'scrollback', seq: 0 }) + }) + ) + callbacks?.onBinary( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.SnapshotEnd, + streamId: stream.streamId, + seq: 2, + payload: new Uint8Array() + }) + ) + sendBinary.mockClear() + + const text = '\x1b[?1049h\x1b[?2026h\x1b[2J\x1b[H\x1b[31m-red 🙂 界\x1b[0m\x1b[?2026l' + const output = encodeTerminalStreamText(text) + callbacks?.onBinary( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Output, + streamId: stream.streamId, + seq: text.length, + payload: output + }) + ) + + expect(terminal.write).toHaveBeenCalledWith(text, expect.any(Function)) + expect(parsedCallbacks).toHaveLength(1) + expect(sentAckBytes()).toEqual([]) + + parsedCallbacks.shift()?.() + expect(sentAckBytes()).toEqual([output.byteLength]) + stream.close() + }) + + function sentAckBytes(): number[] { + return sendBinary.mock.calls.flatMap(([bytes]) => { + const frame = decodeTerminalStreamFrame(bytes) + if (frame?.opcode !== TerminalStreamOpcode.Ack) { + return [] + } + const payload = decodeTerminalStreamJson<{ bytes?: number }>(frame.payload) + return typeof payload?.bytes === 'number' ? [payload.bytes] : [] + }) + } +})