diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index 6c2d4a42f0f..2afb1f89482 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -3096,17 +3096,18 @@ "providers": ["local", "daemon", "ssh", "remote-runtime"], "coveredPlatforms": ["macos"], "coveredProviders": ["ssh"], - "coverageNotes": "Local macOS evidence covers main-process pending-output caps, runtime path-provenance history reuse, saturated-tail wait detection, direct-SSH source retention/accounting, decoder input bounds, bounded reconnect scheduling, and WSL stdio transport settlement through deterministic unit contracts. A joined main/runtime oracle exercises renderer-sourced headed semantics and headless-model snapshot semantics without launching a live paired server. Separately, a macOS-hosted Docker OpenSSH run exercises only the deployed Linux relay and direct SSH provider, with an exact 256 KiB source-credit plateau, concurrent PTY typing, fixed-size filesystem frames, Git churn, and owner-lease reconnect. Neither deterministic topology labels nor Docker SSH constitute live headed desktop, headless orca serve, or physical WSL evidence; local/daemon, prior-version daemons, Windows named pipes/ConPTY, folder workspaces, and mixed-version clients also remain uncovered.", + "coverageNotes": "Local macOS evidence covers main-process pending-output and projection-admission caps, runtime path-provenance history reuse, saturated-tail wait detection, direct-SSH source retention/accounting, exact provider-generation pause ownership, renderer exit/data ordering, decoder input bounds, bounded reconnect scheduling, and WSL stdio transport settlement through deterministic unit contracts. A joined main/runtime oracle exercises renderer-sourced headed semantics and headless-model snapshot semantics without launching a live paired server. Separately, a macOS-hosted Docker OpenSSH run exercises only the deployed Linux relay and direct SSH provider, with an exact 256 KiB source-credit plateau, concurrent PTY typing, fixed-size filesystem frames, Git churn, and owner-lease reconnect. Neither deterministic topology labels nor Docker SSH constitute live headed desktop, headless orca serve, or physical WSL evidence; local/daemon, prior-version daemons, Windows named pipes/ConPTY, folder workspaces, and mixed-version clients also remain uncovered.", "motivatingLinks": [ "https://github.com/stablyai/orca/pull/6836", "https://github.com/stablyai/orca/pull/6858", "https://github.com/stablyai/orca/pull/7002", "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; provisional activation data cannot project before authority commits, exit-time cancellation transfers published projections before proof commit, and generation close fences late proof from final exit while draining exact projection waiters. 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 without releasing in-flight byte credit. Replacement reservation rolls back every earlier span on a later failure, commit requires the exact frozen transfer states, and authoritatively reclaimed covered spans complete idempotently. The exit-deadline oracle publishes a source-backed desktop projection, advances a fake clock to cancellation, and requires one projection transfer, proof commit, preparation, and final exit with no provider close or retained obligation; a second controlled proof remains pending across generation close and must never publish final exit. Direct-SSH contracts additionally prove activation-response settlement before the first source frame, claim-gated projection under same-decoder-turn response/data delivery, exact early-ACK reservation through send settlement or same-token retry, recovery-completion settlement before live admission, invalid-checkpoint restore retention through response settlement followed by fresh-token retry and one live source frame, 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.", + "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; provisional activation data cannot project before authority commits, pending renderer projection IDs are capped and compacted across split remainders, exit preparation owns its renderer fence through finalization, exit-time cancellation transfers published projections before proof commit, generation close fences late proof while draining exact projection waiters, and pause/resume targets only the exact provider generation. 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, a 1,024-ID projection cap with split-remainder compaction, preserved sequenced-tail metadata, active-pending protection ahead of background trimming, exact provider-generation pause ownership, renderer exit/data ordering, 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 without releasing in-flight byte credit. Replacement reservation rolls back every earlier span on a later failure, commit requires the exact frozen transfer states, and authoritatively reclaimed covered spans complete idempotently. The exit-deadline oracle publishes a source-backed desktop projection, advances a fake clock to cancellation, and requires one projection transfer, proof commit, preparation, and final exit with no provider close or retained obligation; a second controlled proof remains pending across generation close and must never publish final exit, while both paths release the renderer preparation lease exactly once. Direct-SSH contracts additionally prove activation-response settlement before the first source frame, claim-gated projection under same-decoder-turn response/data delivery, exact early-ACK reservation through send settlement or same-token retry, recovery-completion settlement before live admission, invalid-checkpoint restore retention through response settlement followed by fresh-token retry and one live source frame, 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/ipc/pty-pending-projection-admissions.test.ts src/main/ipc/ssh-pty-legacy-projection.test.ts src/main/ipc/ssh-pty-model-admission.test.ts --reporter=dot", "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orca-runtime-path-candidate-history.test.ts", "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orca-runtime.test.ts src/main/runtime/orca-runtime-tail-wait-memo.test.ts", "pnpm exec vitest run --config config/vitest.config.ts src/main/providers/ssh-pty-provider.test.ts src/main/ssh/ssh-relay-session.test.ts src/main/ssh/ssh-relay-session-data-delivery.test.ts src/main/ssh/ssh-relay-session-incarnation.test.ts src/main/ssh/ssh-relay-session-reconnect-incarnation.test.ts src/main/ssh/ssh-relay-session-terminal-error.test.ts src/main/ssh/ssh-pty-recovery-retention-budget.test.ts src/main/ssh/relay-protocol-backpressure.test.ts src/relay/protocol-backpressure.test.ts src/relay/pty-source-credit-ledger.test.ts src/relay/pty-source-credit-scheduler.test.ts src/relay/relay-pty-source-publication.test.ts src/relay/ssh-pty-source-credit-adapter.test.ts --reporter=dot", @@ -3125,6 +3126,8 @@ ], "testFiles": [ "src/main/ipc/pty.test.ts", + "src/main/ipc/pty-pending-projection-admissions.test.ts", + "src/main/ipc/ssh-pty-legacy-projection.test.ts", "src/main/runtime/orca-runtime-path-candidate-history.test.ts", "src/main/runtime/orca-runtime.test.ts", "src/main/runtime/orca-runtime-tail-wait-memo.test.ts", @@ -3153,7 +3156,7 @@ "src/relay/pty-handler-source-publication.test.ts", "src/relay/ssh-pty-source-credit-adapter.test.ts", "src/shared/pty-consumer-session.test.ts", - "src/main/ipc/ssh-pty-legacy-projection.test.ts", + "src/main/ipc/ssh-pty-model-admission.test.ts", "src/main/ipc/ssh-pty-output-intake.test.ts", "src/main/ipc/ssh-pty-output-exit-deadline.test.ts", "src/main/ipc/ssh-pty-remote-source-range-consumers.test.ts", @@ -3174,7 +3177,22 @@ "assertions": [ "total renderer in-flight output is capped across many PTYs", "active PTY pending output is prioritized during renderer backpressure", - "combined pending output exceeding the interactive size limit is batched" + "combined pending output exceeding the interactive size limit is batched", + "reconnect replacement cannot redirect pause or resume away from the exact provider generation", + "data arriving while SSH exit waits for renderer projection settlement cannot follow exit preparation" + ] + }, + { + "file": "src/main/ipc/pty-pending-projection-admissions.test.ts", + "assertions": [ + "pending projection IDs retain exactly 1,024 entries then transfer the full ordered run", + "published prefixes compact before append while transfer state survives every split remainder" + ] + }, + { + "file": "src/main/ipc/ssh-pty-legacy-projection.test.ts", + "assertions": [ + "split projection publication remains pending until its exact display and accounting ranges are fully published" ] }, { @@ -3317,12 +3335,19 @@ "file": "src/main/ipc/ssh-pty-legacy-projection.test.ts", "assertions": ["provider-generation close drains exact projection terminality waiters"] }, + { + "file": "src/main/ipc/ssh-pty-model-admission.test.ts", + "assertions": [ + "disposal resumes every paused provider generation exactly once, including pause-only rejected entries" + ] + }, { "file": "src/main/ipc/ssh-pty-output-intake.test.ts", "assertions": [ "exit remains behind accepted model, desktop projection, and required remote obligations", "exit timeout cancels only the matching delivery and keeps unrelated provider work usable", - "desktop source identity and scanner facts commit atomically or roll back without credit" + "desktop source identity and scanner facts commit atomically or roll back without credit", + "renderer exit preparation remains owned through finalization and releases on duplicate, failure, or generation close" ] }, { @@ -3522,7 +3547,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. 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. On exact pre-fix baseline db167ea3d with test-only oracles, cancellation proof reclaimed the published projection span before transfer and generation close allowed a late proof to publish final exit; both deterministic oracles pass after proof acquisition and commit are separated under the exact deadline barrier. 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. Before restore-record retirement, invalid-checkpoint recovery followed by restore/retry returned restoreRequired a second time instead of opening a fresh activation; the same deterministic oracle passes after exact response-settlement cleanup. With test-only oracles on pre-fix 4bf54da9b, an existing-owner mismatch lacked the stale-owner code and both shared authority and main fallback rows failed; the candidate types that refusal while stale fresh admission remains a subscriber and the retained principal and lease can still recover. 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 plus projection IDs stay within exact caps, direct-SSH source/recovery memory and both frame decoders stay within exact caps, and 50-PTY reconnect isolates slow/failing siblings. On pre-fix 4bf54da9b, reconnect pause targeted the replacement provider, disposal omitted a pause-only generation, normal and generation-close exit paths never released the renderer preparation lease, and the projection-cap seam did not exist; the same oracles pass with exact generation indexing, paused-generation enumeration, barrier-owned exit leases, and the 1,024-ID transfer latch. 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. On exact pre-fix baseline db167ea3d with test-only oracles, cancellation proof reclaimed the published projection span before transfer and generation close allowed a late proof to publish final exit; both deterministic oracles pass after proof acquisition and commit are separated under the exact deadline barrier. Before snapshot admission, 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, 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. Before restore-record retirement, invalid-checkpoint recovery followed by restore/retry returned restoreRequired a second time instead of opening a fresh activation; the same deterministic oracle passes after exact response-settlement cleanup. With test-only oracles on pre-fix 4bf54da9b, an existing-owner mismatch lacked the stale-owner code and both shared authority and main fallback rows failed; the candidate types that refusal while stale fresh admission remains a subscriber and the retained principal and lease can still recover. 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/ipc/pty-pending-data-drain-contract.ts b/src/main/ipc/pty-pending-data-drain-contract.ts index 9ddfa755ffb..d1d4c5be1a8 100644 --- a/src/main/ipc/pty-pending-data-drain-contract.ts +++ b/src/main/ipc/pty-pending-data-drain-contract.ts @@ -10,6 +10,7 @@ export type PendingPtyData = { droppedMode2031Data?: string droppedMode2031ScanState?: Mode2031ReplyScanState projectionAdmissionIds?: readonly string[] + projectionAdmissionsTransferred?: true } export type PtyPendingDataDrainDisposition = 'active' | 'background' | 'blocked' diff --git a/src/main/ipc/pty-pending-projection-admissions.test.ts b/src/main/ipc/pty-pending-projection-admissions.test.ts new file mode 100644 index 00000000000..62863e27fa6 --- /dev/null +++ b/src/main/ipc/pty-pending-projection-admissions.test.ts @@ -0,0 +1,64 @@ +import { describe, expect, it, vi } from 'vitest' +import { + appendPendingProjectionAdmission, + compactPendingProjectionAdmissions, + PTY_PENDING_PROJECTION_ADMISSION_MAX_IDS, + propagatePendingProjectionRemainder +} from './pty-pending-projection-admissions' + +describe('pending PTY projection admissions', () => { + it('retains the exact cap then transfers the whole ordered run on overflow', () => { + const transfer = vi.fn() + let state = compactPendingProjectionAdmissions() + + for (let index = 0; index < PTY_PENDING_PROJECTION_ADMISSION_MAX_IDS; index++) { + state = appendPendingProjectionAdmission(state, `projection-${index}`, { + isPending: () => true, + transfer + }) + } + + expect(state.projectionAdmissionIds).toHaveLength(PTY_PENDING_PROJECTION_ADMISSION_MAX_IDS) + expect(transfer).not.toHaveBeenCalled() + + state = appendPendingProjectionAdmission(state, 'projection-overflow', { + isPending: () => true, + transfer + }) + + expect(state).toEqual({ projectionAdmissionsTransferred: true }) + expect(transfer).toHaveBeenCalledOnce() + expect(transfer.mock.calls[0]?.[0]).toHaveLength(PTY_PENDING_PROJECTION_ADMISSION_MAX_IDS + 1) + }) + + it('compacts terminal prefixes and transfers later admissions until the remainder drains', () => { + const transfer = vi.fn() + const compacted = propagatePendingProjectionRemainder( + { + projectionAdmissionIds: ['projection-published', 'projection-partial', 'projection-tail'] + }, + { sent: true, projectionsTransferred: false }, + { isPending: (id) => id !== 'projection-published', transfer } + ) + + expect(compacted).toEqual({ + projectionAdmissionIds: ['projection-partial', 'projection-tail'] + }) + + const transferred = appendPendingProjectionAdmission( + { projectionAdmissionsTransferred: true }, + 'projection-after-transfer', + { isPending: () => true, transfer } + ) + expect(transferred).toEqual({ projectionAdmissionsTransferred: true }) + expect(transfer).toHaveBeenCalledWith(['projection-after-transfer'], 'pending-projection-cap') + + expect( + propagatePendingProjectionRemainder( + compacted, + { sent: true, projectionsTransferred: true }, + { isPending: () => true, transfer } + ) + ).toEqual({ projectionAdmissionsTransferred: true }) + }) +}) diff --git a/src/main/ipc/pty-pending-projection-admissions.ts b/src/main/ipc/pty-pending-projection-admissions.ts new file mode 100644 index 00000000000..9fa0d423bc9 --- /dev/null +++ b/src/main/ipc/pty-pending-projection-admissions.ts @@ -0,0 +1,57 @@ +export const PTY_PENDING_PROJECTION_ADMISSION_MAX_IDS = 1024 + +export type PendingProjectionAdmissions = Readonly<{ + projectionAdmissionIds?: readonly string[] + projectionAdmissionsTransferred?: true +}> + +type PendingProjectionAdmissionOptions = Readonly<{ + isPending: (id: string) => boolean + transfer: (ids: readonly string[], reason: string) => void +}> + +const DEFAULT_OPTIONS: PendingProjectionAdmissionOptions = { + isPending: () => true, + transfer: () => {} +} + +export function compactPendingProjectionAdmissions( + state: PendingProjectionAdmissions = {}, + options: PendingProjectionAdmissionOptions = DEFAULT_OPTIONS +): PendingProjectionAdmissions { + if (state.projectionAdmissionsTransferred) { + return { projectionAdmissionsTransferred: true } + } + const ids = Array.from( + new Set((state.projectionAdmissionIds ?? []).filter((id) => options.isPending(id))) + ) + return ids.length > 0 ? { projectionAdmissionIds: ids } : {} +} + +export function appendPendingProjectionAdmission( + state: PendingProjectionAdmissions, + id: string, + options: PendingProjectionAdmissionOptions +): PendingProjectionAdmissions { + if (state.projectionAdmissionsTransferred) { + options.transfer([id], 'pending-projection-cap') + return { projectionAdmissionsTransferred: true } + } + const compacted = compactPendingProjectionAdmissions(state, options) + const ids = [...(compacted.projectionAdmissionIds ?? []), id] + if (ids.length <= PTY_PENDING_PROJECTION_ADMISSION_MAX_IDS) { + return { projectionAdmissionIds: ids } + } + options.transfer(ids, 'pending-projection-cap') + return { projectionAdmissionsTransferred: true } +} + +export function propagatePendingProjectionRemainder( + state: PendingProjectionAdmissions, + delivery: Readonly<{ sent: boolean; projectionsTransferred: boolean }>, + options: PendingProjectionAdmissionOptions +): PendingProjectionAdmissions { + return delivery.sent && !delivery.projectionsTransferred + ? compactPendingProjectionAdmissions(state, options) + : { projectionAdmissionsTransferred: true } +} diff --git a/src/main/ipc/pty.test.ts b/src/main/ipc/pty.test.ts index 4ff670577e9..7a29d953527 100644 --- a/src/main/ipc/pty.test.ts +++ b/src/main/ipc/pty.test.ts @@ -247,7 +247,11 @@ import { resolveWindowsShellLaunchArgs } from '../providers/windows-shell-args' import { _resetWslCachesForTests, _setWslCachesForTests } from '../wsl' import { wslHookRelayManager } from '../agent-hooks/wsl-hook-relay-manager' import { acquireWatcherRemovalGate } from './watcher-removal-gate' -import { acceptSshPtyOutputData } from './ssh-pty-output-intake-registry' +import { + acceptSshPtyOutputData, + acceptSshPtyOutputExit, + closeSshPtyOutputGeneration +} from './ssh-pty-output-intake-registry' // Why: Windows resolves a bare PowerShell name to an absolute exe before ConPTY, else CreateProcessW fails with error 5 (PR #6537 / #5161). const RESOLVED_WINDOWS_POWERSHELL = 'C:\\Windows\\System32\\WindowsPowerShell\\v1.0\\powershell.exe' @@ -450,7 +454,8 @@ describe('registerPtyHandlers', () => { 'ssh-reattach-1', 'ssh-reattach-fail', 'ssh-reattach-ok', - 'ssh-runtime-env' + 'ssh-runtime-env', + 'ssh-generation-replacement' ]) { unregisterSshPtyProvider(leakedConnectionId) } @@ -11535,6 +11540,130 @@ describe('registerPtyHandlers', () => { } }) + it('pauses and resumes the exact SSH provider generation across reconnect replacement', async () => { + vi.useFakeTimers() + const completion = makeDeferred() + let sequence = 0 + let captures = 0 + const runtime = { + setPtyController: vi.fn(), + setRemoteTerminalSourceRangeConsumerHooks: vi.fn(), + getPtyOutputSequence: vi.fn(() => sequence), + acceptPtyDataBounded: vi.fn((_id: string, _data: string, _at: number, rawLength: number) => { + sequence += rawLength + captures++ + return { + sequence, + completion: captures === 1 ? completion.promise : Promise.resolve() + } + }) + } + const original = { + providerGeneration: 41, + hasPtyDeliveryPauseAdapter: () => true, + pauseProducer: vi.fn(), + resumeProducer: vi.fn() + } + const replacement = { + providerGeneration: 42, + hasPtyDeliveryPauseAdapter: () => true, + pauseProducer: vi.fn(), + resumeProducer: vi.fn() + } + const id = 'ssh:ssh-generation-replacement@@relay-pty' + const receipts: Promise[] = [] + + try { + registerPtyHandlers(mainWindow as never, runtime as never) + registerSshPtyProvider('ssh-generation-replacement', original as never) + const running = acceptSshPtyOutputData({ + id, + data: 'a'.repeat(256 * 1024), + providerGeneration: 41, + ptyIncarnation: 'incarnation-41', + rawLength: 256 * 1024, + transformed: false + }) + receipts.push(running) + registerSshPtyProvider('ssh-generation-replacement', replacement as never) + const pressured = acceptSshPtyOutputData({ + id, + data: 'b', + providerGeneration: 41, + ptyIncarnation: 'incarnation-41', + rawLength: 1, + transformed: false + }) + receipts.push(pressured) + + expect(original.pauseProducer).toHaveBeenCalledWith(id) + expect(replacement.pauseProducer).not.toHaveBeenCalled() + + completion.resolve() + await Promise.all([running, pressured]) + expect(original.resumeProducer).toHaveBeenCalledWith(id) + expect(replacement.resumeProducer).not.toHaveBeenCalled() + } finally { + completion.resolve() + await Promise.allSettled(receipts) + closeSshPtyOutputGeneration(41, 'test-cleanup') + unregisterSshPtyProvider('ssh-generation-replacement') + } + }) + + it('rejects local data while an SSH renderer exit waits for projection settlement', async () => { + const provider = installObservableDaemonTestProvider() + let sequence = 0 + const runtime = { + setPtyController: vi.fn(), + setRemoteTerminalSourceRangeConsumerHooks: vi.fn(), + getPtyOutputSequence: vi.fn(() => sequence), + acceptPtyDataBounded: vi.fn((_id: string, _data: string, _at: number, rawLength: number) => { + sequence += rawLength + return { sequence, completion: Promise.resolve() } + }), + onPtyData: vi.fn(), + onPtyExit: vi.fn() + } + const id = 'ssh:exit-data-race@@relay-pty' + + registerPtyHandlers(mainWindow as never, runtime as never) + mainWindow.webContents.send.mockClear() + await acceptSshPtyOutputData({ + id, + data: 'before-exit', + providerGeneration: 51, + ptyIncarnation: 'incarnation-51', + rawLength: 'before-exit'.length, + transformed: false + }) + const exit = acceptSshPtyOutputExit({ + id, + code: 0, + providerGeneration: 51, + ptyIncarnation: 'incarnation-51' + }) + await Promise.resolve() + + provider.emitData(id, 'must-not-follow-exit') + expect(mainWindow.webContents.send).not.toHaveBeenCalledWith('pty:data', { + id, + data: 'must-not-follow-exit' + }) + + getPtyAckDataListener()(null, { id, processedChars: 'before-exit'.length }) + await exit + expect(mainWindow.webContents.send.mock.calls.at(-1)).toEqual([ + 'pty:exit', + { + id, + code: 0, + providerGeneration: 51, + ptyIncarnation: 'incarnation-51' + } + ]) + }) + it('resumes a paused producer when the PTY exits before draining', async () => { vi.useFakeTimers() try { diff --git a/src/main/ipc/pty.ts b/src/main/ipc/pty.ts index ec25c7df6a5..8503569c301 100644 --- a/src/main/ipc/pty.ts +++ b/src/main/ipc/pty.ts @@ -158,6 +158,12 @@ import { unmarkHiddenRendererPty } from './pty-hidden-delivery-gate' import { PtyPendingDataDrainQueue, type PendingPtyData } from './pty-pending-data-drain-queue' +import { + appendPendingProjectionAdmission, + compactPendingProjectionAdmissions, + propagatePendingProjectionRemainder, + type PendingProjectionAdmissions +} from './pty-pending-projection-admissions' import { SshPtyOutputIntake } from './ssh-pty-output-intake' import { cancelSshPtySourceDelivery, @@ -209,6 +215,7 @@ type FreshLocalFallbackProvider = IPtyProvider & { routesFreshSpawnsToLocalProvider?: true } const sshProviders = new Map() +const sshProvidersByGeneration = new Map() type RegisteredPtyProvider = { provider: IPtyProvider @@ -1303,10 +1310,19 @@ function beginPtySpawnForWorktree( /** Register an SSH PTY provider for a connection. */ export function registerSshPtyProvider(connectionId: string, provider: IPtyProvider): void { sshProviders.set(connectionId, provider) + const generation = (provider as { providerGeneration?: number }).providerGeneration + if (Number.isSafeInteger(generation) && generation! > 0) { + sshProvidersByGeneration.set(generation!, provider) + } } /** Remove an SSH PTY provider when a connection is closed. */ export function unregisterSshPtyProvider(connectionId: string): void { + const provider = sshProviders.get(connectionId) + const generation = (provider as { providerGeneration?: number } | undefined)?.providerGeneration + if (generation !== undefined && sshProvidersByGeneration.get(generation) === provider) { + sshProvidersByGeneration.delete(generation) + } sshProviders.delete(connectionId) } @@ -2402,7 +2418,7 @@ export function registerPtyHandlers( id: string, payload: PtyDataPayload, projectionAdmissionIds?: readonly string[] - ): boolean { + ): { sent: boolean; projectionsTransferred: boolean } { const charCount = getPtyPayloadCharCount(payload) const accounting = rendererDeliveryAccountingByPty.get(id) const hadAccounting = accounting !== undefined @@ -2445,8 +2461,9 @@ export function registerPtyHandlers( chars: charCount }) console.error('[pty] renderer data send failed; payload will not be retried', error) - return false + return { sent: false, projectionsTransferred: projectionAdmissionIds !== undefined } } + let projectionsTransferred = false if (projectionAdmissionIds) { try { sshOutputIntake?.publishProjectionPrefix( @@ -2456,6 +2473,7 @@ export function registerPtyHandlers( ) } catch { sshOutputIntake?.transferProjections(projectionAdmissionIds, 'projection-publish-failed') + projectionsTransferred = true } } if (rendererDeliveryRestoreNeededPtys.has(id)) { @@ -2469,7 +2487,7 @@ export function registerPtyHandlers( ) } } - return true + return { sent: true, projectionsTransferred } } function rendererPtyIsKnownHidden(id: string): boolean { @@ -2585,6 +2603,39 @@ export function registerPtyHandlers( } } + function updatePendingProjectionAdmissions( + pending: PendingPtyData, + state: PendingProjectionAdmissions + ): void { + delete pending.projectionAdmissionIds + delete pending.projectionAdmissionsTransferred + if (state.projectionAdmissionIds) { + pending.projectionAdmissionIds = state.projectionAdmissionIds + } + if (state.projectionAdmissionsTransferred) { + pending.projectionAdmissionsTransferred = true + } + } + + function compactPendingProjectionState( + pending: PendingProjectionAdmissions, + projectionSemanticsId?: string + ): PendingProjectionAdmissions { + const options = pendingProjectionAdmissionOptions() + const compacted = compactPendingProjectionAdmissions(pending, options) + return projectionSemanticsId + ? appendPendingProjectionAdmission(compacted, projectionSemanticsId, options) + : compacted + } + + function pendingProjectionAdmissionOptions() { + return { + isPending: (id: string) => sshOutputIntake?.hasUnpublishedProjection(id) ?? false, + transfer: (ids: readonly string[], reason: string) => + sshOutputIntake?.transferProjections(ids, reason) + } + } + function appendPendingPtyData( id: string, existing: PendingPtyData | undefined, @@ -2617,17 +2668,19 @@ export function registerPtyHandlers( droppedMode2031ScanState: mode2031.state } } + const projectionState = compactPendingProjectionState(existing ?? {}, projectionSemanticsId) const nextContainsBackgroundOutput = existing?.containsBackgroundOutput === true || containsBackgroundOutput if (!existing) { - return dropOversizedPendingPtyData(id, { + const pending: PendingPtyData = { data, ...(typeof startSeq === 'number' ? { startSeq } : {}), ...(rawLength !== data.length ? { rawLength } : {}), ...(transformed ? { transformed: true } : {}), - ...(nextContainsBackgroundOutput ? { containsBackgroundOutput: true } : {}), - ...(projectionSemanticsId ? { projectionAdmissionIds: [projectionSemanticsId] } : {}) - }) + ...(nextContainsBackgroundOutput ? { containsBackgroundOutput: true } : {}) + } + updatePendingProjectionAdmissions(pending, projectionState) + return dropOversizedPendingPtyData(id, pending) } const existingRawLength = existing.rawLength ?? existing.data.length const next: PendingPtyData = { @@ -2635,16 +2688,9 @@ export function registerPtyHandlers( ...(!preservesSeq || existing.transformed || transformed ? { rawLength: existingRawLength + rawLength, transformed: true as const } : {}), - ...(nextContainsBackgroundOutput ? { containsBackgroundOutput: true } : {}), - ...((existing.projectionAdmissionIds?.length ?? 0) > 0 || projectionSemanticsId - ? { - projectionAdmissionIds: [ - ...(existing.projectionAdmissionIds ?? []), - ...(projectionSemanticsId ? [projectionSemanticsId] : []) - ] - } - : {}) + ...(nextContainsBackgroundOutput ? { containsBackgroundOutput: true } : {}) } + updatePendingProjectionAdmissions(next, projectionState) if (typeof existing.startSeq === 'number') { next.startSeq = existing.startSeq } @@ -2754,7 +2800,7 @@ export function registerPtyHandlers( droppedOutput: true }, pending.projectionAdmissionIds - ) + ).sent ) { sendFailed = true break @@ -2766,8 +2812,9 @@ export function registerPtyHandlers( const indivisible = pending.transformed === true const chunk = indivisible ? data : data.slice(0, PTY_BATCH_FLUSH_CHUNK_CHARS) const remaining = indivisible ? '' : data.slice(PTY_BATCH_FLUSH_CHUNK_CHARS) + let nextPending: PendingPtyData | undefined if (remaining) { - const nextPending: PendingPtyData = { data: remaining } + nextPending = { data: remaining } if (typeof pending.startSeq === 'number') { nextPending.startSeq = pending.startSeq + chunk.length } @@ -2777,26 +2824,38 @@ export function registerPtyHandlers( if (pending.projectionAdmissionIds) { nextPending.projectionAdmissionIds = pending.projectionAdmissionIds } + if (pending.projectionAdmissionsTransferred) { + nextPending.projectionAdmissionsTransferred = true + } pendingData.replaceWithRemainder(selection, nextPending) } else { pendingData.remove(selection) pendingOverflowMarkedPtys.delete(id) } updateProducerFlowControl(id) - if ( - !sendPtyDataToRenderer( + const delivery = sendPtyDataToRenderer( + id, + makePtyDataPayload( id, - makePtyDataPayload( - id, - chunk, - pending.startSeq, - pending.containsBackgroundOutput, - pending.rawLength, - pending.transformed - ), - pending.projectionAdmissionIds + chunk, + pending.startSeq, + pending.containsBackgroundOutput, + pending.rawLength, + pending.transformed + ), + pending.projectionAdmissionIds + ) + if (nextPending) { + updatePendingProjectionAdmissions( + nextPending, + propagatePendingProjectionRemainder( + nextPending, + delivery, + pendingProjectionAdmissionOptions() + ) ) - ) { + } + if (!delivery.sent) { sendFailed = true break } @@ -2859,15 +2918,23 @@ export function registerPtyHandlers( return true } - function preparePtyExitForRenderer(payload: { id: string; code: number }): void { + function preparePtyExitForRenderer(payload: { id: string; code: number }): (() => void) | null { if (mainWindow.isDestroyed()) { sshOutputIntake?.transferPtyProjections(payload.id, 'renderer-destroyed') - return + return () => {} } if (rendererExitingPtyIds.has(payload.id)) { - return + return null } rendererExitingPtyIds.add(payload.id) + let released = false + const release = (): void => { + if (released) { + return + } + released = true + rendererExitingPtyIds.delete(payload.id) + } try { if (!rendererCreditBeforeExitByPty.has(payload.id)) { rendererCreditBeforeExitByPty.set( @@ -2905,8 +2972,10 @@ export function registerPtyHandlers( ) } } - } finally { - rendererExitingPtyIds.delete(payload.id) + return release + } catch (error) { + release() + throw error } } @@ -2915,48 +2984,47 @@ export function registerPtyHandlers( rendererCreditBeforeExitByPty.delete(payload.id) return } - if (rendererExitingPtyIds.has(payload.id)) { - return - } - rendererExitingPtyIds.add(payload.id) - try { - const hadReleasableRendererCredit = - rendererCreditBeforeExitByPty.get(payload.id) ?? - getRendererInFlightCharsForPty(payload.id) > 0 - rendererCreditBeforeExitByPty.delete(payload.id) - // Why resume a dead PTY (no-op): avoid leaving a stale paused mark behind for a reused id. - producerFlowControl.release(payload.id) - sourceCreditPendingPtys.delete(payload.id) - pendingOverflowMarkedPtys.delete(payload.id) - rendererDeliveryRestoreNeededPtys.delete(payload.id) - lastInputAtByPty.delete(payload.id) - interactiveOutputCharsByPty.delete(payload.id) - const releasedRendererCredit = getRendererInFlightCharsForPty(payload.id) - rendererInFlightTotalChars = Math.max(0, rendererInFlightTotalChars - releasedRendererCredit) - // Why: the renderer also drops its cumulative total on pty:exit, so a reused id restarts aligned at zero on both sides. - rendererDeliveryAccountingByPty.delete(payload.id) - if (hadReleasableRendererCredit) { - if (pendingDataFlushActive) { - // Why: let the open round coalesce this wake into its one post-round continuation. - const reactivatedBlocked = pendingData.reactivateBlocked() - pendingDataCreditReleasedDuringFlush ||= reactivatedBlocked - } else { - schedulePendingDataAfterCreditReport(true) - } + const hadReleasableRendererCredit = + rendererCreditBeforeExitByPty.get(payload.id) ?? + getRendererInFlightCharsForPty(payload.id) > 0 + rendererCreditBeforeExitByPty.delete(payload.id) + // Why resume a dead PTY (no-op): avoid leaving a stale paused mark behind for a reused id. + producerFlowControl.release(payload.id) + sourceCreditPendingPtys.delete(payload.id) + pendingOverflowMarkedPtys.delete(payload.id) + rendererDeliveryRestoreNeededPtys.delete(payload.id) + lastInputAtByPty.delete(payload.id) + interactiveOutputCharsByPty.delete(payload.id) + const releasedRendererCredit = getRendererInFlightCharsForPty(payload.id) + rendererInFlightTotalChars = Math.max(0, rendererInFlightTotalChars - releasedRendererCredit) + // Why: the renderer also drops its cumulative total on pty:exit, so a reused id restarts aligned at zero on both sides. + rendererDeliveryAccountingByPty.delete(payload.id) + if (hadReleasableRendererCredit) { + if (pendingDataFlushActive) { + // Why: let the open round coalesce this wake into its one post-round continuation. + const reactivatedBlocked = pendingData.reactivateBlocked() + pendingDataCreditReleasedDuringFlush ||= reactivatedBlocked + } else { + schedulePendingDataAfterCreditReport(true) } - mainWindow.webContents.send('pty:exit', { - ...payload, - ...(reversibleStopOwnersByPtyId.has(payload.id) ? { preserveRendererBinding: true } : {}) - }) - } finally { - rendererExitingPtyIds.delete(payload.id) } + mainWindow.webContents.send('pty:exit', { + ...payload, + ...(reversibleStopOwnersByPtyId.has(payload.id) ? { preserveRendererBinding: true } : {}) + }) } function sendPtyExitToRenderer(payload: { id: string; code: number }): void { - preparePtyExitForRenderer(payload) - sshOutputIntake?.transferPtyProjections(payload.id, 'legacy-pty-exit') - finalizePtyExitForRenderer(payload) + const release = preparePtyExitForRenderer(payload) + if (!release) { + return + } + try { + sshOutputIntake?.transferPtyProjections(payload.id, 'legacy-pty-exit') + finalizePtyExitForRenderer(payload) + } finally { + release() + } } function sendPtySpawnedToRenderer(id: string): void { @@ -3131,13 +3199,19 @@ export function registerPtyHandlers( projection.identity.sequenceEnd, projection ), - prepareExit: (event) => preparePtyExitForRenderer(event), + prepareExit: (event) => { + const release = preparePtyExitForRenderer(event) + if (!release) { + throw new Error('pty_renderer_exit_in_progress') + } + return release + }, finalizeExit: (event) => { runtime?.onPtyExit(event.id, event.code, event.ptyIncarnation) finalizePtyExitForRenderer(event) }, - pauseProvider: (_generation, id) => { - const provider = tryGetProviderForPty(id) as + pauseProvider: (generation, id) => { + const provider = sshProvidersByGeneration.get(generation) as | (IPtyProvider & { hasPtyDeliveryPauseAdapter?: () => boolean }) | undefined if (!provider?.hasPtyDeliveryPauseAdapter?.()) { @@ -3146,16 +3220,13 @@ export function registerPtyHandlers( provider.pauseProducer?.(id) return true }, - resumeProvider: (_generation, id) => tryGetProviderForPty(id)?.resumeProducer?.(id), + resumeProvider: (generation, id) => + sshProvidersByGeneration.get(generation)?.resumeProducer?.(id), closeProvider: (generation, reason) => { - for (const provider of sshProviders.values()) { - if ((provider as { providerGeneration?: number }).providerGeneration === generation) { - ;( - provider as IPtyProvider & { closeOutputIntake?: (reason: string) => void } - ).closeOutputIntake?.(reason) - return - } - } + const provider = sshProvidersByGeneration.get(generation) + ;( + provider as (IPtyProvider & { closeOutputIntake?: (reason: string) => void }) | undefined + )?.closeOutputIntake?.(reason) }, onGenerationClosed: (providerGeneration) => { for (const id of pendingData.keys()) { @@ -3172,6 +3243,7 @@ export function registerPtyHandlers( pendingOverflowMarkedPtys.delete(id) } } + sshProvidersByGeneration.delete(providerGeneration) }, publishSourceAck: publishSshPtySourceAck, cancelSourceDelivery: cancelSshPtySourceDelivery diff --git a/src/main/ipc/ssh-pty-legacy-projection-publication.ts b/src/main/ipc/ssh-pty-legacy-projection-publication.ts index 730eb0067ae..395b9233d60 100644 --- a/src/main/ipc/ssh-pty-legacy-projection-publication.ts +++ b/src/main/ipc/ssh-pty-legacy-projection-publication.ts @@ -57,6 +57,22 @@ export function publishLegacyProjectionPrefix( } } +export function hasUnpublishedLegacyProjection( + records: ReadonlyMap, + id: string +): boolean { + const record = records.get(id) + if (!record || record.state === 'reserved') { + return false + } + const displayLength = + record.semantics.identity.displayEnd - record.semantics.identity.displayStart + return ( + record.publishedDisplay < displayLength || + record.publishedAccounting < record.semantics.identity.rawLength + ) +} + export function settlePublishedLegacyProjectionPrefix( records: Map, idsByPty: Map, diff --git a/src/main/ipc/ssh-pty-legacy-projection.test.ts b/src/main/ipc/ssh-pty-legacy-projection.test.ts index 3147f973190..e926456619a 100644 --- a/src/main/ipc/ssh-pty-legacy-projection.test.ts +++ b/src/main/ipc/ssh-pty-legacy-projection.test.ts @@ -170,11 +170,14 @@ describe('SshPtyLegacyProjectionLedger', () => { }) ) const id = projection.identity.projectionSemanticsId + expect(ledger.hasUnpublished(id)).toBe(true) ledger.publishPrefix([id], 2, 2) + expect(ledger.hasUnpublished(id)).toBe(true) ledger.settlePublishedPrefix('pty-1', 2) expect(settled).toEqual([]) ledger.publishPrefix([id], 2, 2) + expect(ledger.hasUnpublished(id)).toBe(false) ledger.settlePublishedPrefix('pty-1', 2) expect(settled).toEqual([ diff --git a/src/main/ipc/ssh-pty-legacy-projection.ts b/src/main/ipc/ssh-pty-legacy-projection.ts index c9576d718a5..0e779a82b4f 100644 --- a/src/main/ipc/ssh-pty-legacy-projection.ts +++ b/src/main/ipc/ssh-pty-legacy-projection.ts @@ -23,6 +23,7 @@ import { unpublishedProjectionIds } from './ssh-pty-projection-terminality' import { + hasUnpublishedLegacyProjection, publishLegacyProjectionPrefix, settlePublishedLegacyProjectionPrefix } from './ssh-pty-legacy-projection-publication' @@ -303,6 +304,10 @@ export class SshPtyLegacyProjectionLedger { return this.records.get(id)?.semantics } + hasUnpublished(id: string): boolean { + return hasUnpublishedLegacyProjection(this.records, id) + } + getDebugSnapshot(): LegacySshProjectionDebugSnapshot { return projectionDebugSnapshot(this.records, this.cursorByPty, { settled: this.settledCount, diff --git a/src/main/ipc/ssh-pty-model-admission-pressure.ts b/src/main/ipc/ssh-pty-model-admission-pressure.ts index 18837e6c51c..5b34db79e5b 100644 --- a/src/main/ipc/ssh-pty-model-admission-pressure.ts +++ b/src/main/ipc/ssh-pty-model-admission-pressure.ts @@ -45,6 +45,10 @@ export class SshPtyModelAdmissionPressure { return this.pausedKeys.size } + get pausedProviderGenerations(): ReadonlySet { + return new Set(Array.from(this.pausedKeys.values(), (key) => key.providerGeneration)) + } + has(key: SshPtyModelAdmissionKey): boolean { return pressureHasAdmissionKey(this.entries, key) } diff --git a/src/main/ipc/ssh-pty-model-admission.test.ts b/src/main/ipc/ssh-pty-model-admission.test.ts index 4d267ef83b0..e03a6676155 100644 --- a/src/main/ipc/ssh-pty-model-admission.test.ts +++ b/src/main/ipc/ssh-pty-model-admission.test.ts @@ -43,6 +43,37 @@ describe('SshPtyModelAdmission', () => { expect(admission.getDebugSnapshot()).toMatchObject({ sourceUnits: 0, bytes: 0 }) }) + it('resumes every paused provider generation exactly once on disposal', async () => { + const resumeProvider = vi.fn() + const admission = new SshPtyModelAdmission({ + perPtyHighSourceUnits: 4, + perPtyHighBytes: 1024, + globalHighSourceUnits: 4, + globalHighBytes: 1024, + pressureMaxFrames: 1, + pressureMaxBytes: 1024, + pauseProvider: () => true, + resumeProvider + }) + const running = accept(admission, new Promise(() => {})) + const pressured = accept(admission, Promise.resolve()) + const rejected = admission.accept({ ptyId: 'pty-2', providerGeneration: 8 }, 'data', 4, () => ({ + sequence: 4, + completion: Promise.resolve() + })) + + await expect(rejected).rejects.toThrow('ssh_model_admission_pressure_exhausted') + admission.dispose() + admission.dispose() + + await expect(running).rejects.toThrow('ssh_model_admission_disposed') + await expect(pressured).rejects.toThrow('ssh_model_admission_disposed') + expect(resumeProvider.mock.calls).toEqual([ + [{ ptyId: 'pty-1', providerGeneration: 7 }], + [{ ptyId: 'pty-2', providerGeneration: 8 }] + ]) + }) + it.each(['resolve', 'reject'] as const)( 'ignores a late completion %s after cancellation', async (settle) => { diff --git a/src/main/ipc/ssh-pty-model-admission.ts b/src/main/ipc/ssh-pty-model-admission.ts index 1f387ddd516..cc8c57acc7d 100644 --- a/src/main/ipc/ssh-pty-model-admission.ts +++ b/src/main/ipc/ssh-pty-model-admission.ts @@ -156,6 +156,9 @@ export class SshPtyModelAdmission { for (const entry of this.pressure.values) { generations.add(entry.key.providerGeneration) } + for (const generation of this.pressure.pausedProviderGenerations) { + generations.add(generation) + } for (const generation of generations) { this.closeGeneration(generation, 'ssh_model_admission_disposed') } diff --git a/src/main/ipc/ssh-pty-output-exit-deadline.test.ts b/src/main/ipc/ssh-pty-output-exit-deadline.test.ts index 8a14aada9d3..7c49df2ec91 100644 --- a/src/main/ipc/ssh-pty-output-exit-deadline.test.ts +++ b/src/main/ipc/ssh-pty-output-exit-deadline.test.ts @@ -11,6 +11,7 @@ type ExitDeadlineHarness = Readonly<{ intake: SshPtyOutputIntake dependencies: SshPtyOutputIntakeDependencies cancelSourceDelivery: ReturnType + releaseExit: ReturnType exits: string[] }> @@ -24,12 +25,13 @@ function createHarness( cancelSourceDelivery: NonNullable ): ExitDeadlineHarness { const exits: string[] = [] + const releaseExit = vi.fn() const cancellation = vi.fn(cancelSourceDelivery) const dependencies: SshPtyOutputIntakeDependencies = { getModelSequence: () => 0, acceptModel: (event) => ({ sequence: event.rawLength, completion: Promise.resolve() }), project: vi.fn(), - prepareExit: vi.fn(), + prepareExit: vi.fn(() => releaseExit), finalizeExit: () => exits.push('exit'), pauseProvider: vi.fn(() => true), resumeProvider: vi.fn(), @@ -43,6 +45,7 @@ function createHarness( }), dependencies, cancelSourceDelivery: cancellation, + releaseExit, exits } } @@ -96,6 +99,7 @@ describe('SshPtyOutputExitDeadline', () => { expect(harness.cancelSourceDelivery).toHaveBeenCalledOnce() expect(harness.dependencies.closeProvider).not.toHaveBeenCalled() expect(harness.dependencies.prepareExit).toHaveBeenCalledOnce() + expect(harness.releaseExit).toHaveBeenCalledOnce() expect(harness.exits).toEqual(['exit']) expect(harness.intake.getDebugSnapshot()).toMatchObject({ projection: { records: 0 }, diff --git a/src/main/ipc/ssh-pty-output-exit-deadline.ts b/src/main/ipc/ssh-pty-output-exit-deadline.ts index 30f1ee5bf96..4dfa71c09ef 100644 --- a/src/main/ipc/ssh-pty-output-exit-deadline.ts +++ b/src/main/ipc/ssh-pty-output-exit-deadline.ts @@ -23,6 +23,7 @@ export class SshPtyOutputExitDeadline { private readonly barriersByGeneration = new Map>() private readonly cancellationOwnedExits = new Set() private readonly preparedExits = new Set() + private readonly preparedExitReleases = new Map void>() private readonly barrierMs: number private readonly cancellationProofMs: number @@ -41,6 +42,7 @@ export class SshPtyOutputExitDeadline { } settled = true clearTimeout(barrier.timer) + this.releasePreparedExit(event) this.remove(event.providerGeneration, barrier) if (result.ok) { resolve() @@ -107,6 +109,7 @@ export class SshPtyOutputExitDeadline { } for (const key of this.preparedExits) { if (key.startsWith(prefix)) { + this.releasePreparedExitKey(key) this.preparedExits.delete(key) } } @@ -123,8 +126,11 @@ export class SshPtyOutputExitDeadline { if (this.preparedExits.has(key)) { return } + const release = this.dependencies.intake.prepareExit(event) this.preparedExits.add(key) - this.dependencies.intake.prepareExit(event) + if (release) { + this.preparedExitReleases.set(key, release) + } } get activeBarriers(): number { @@ -192,6 +198,21 @@ export class SshPtyOutputExitDeadline { return `${event.providerGeneration}\0${event.id}\0${event.ptyIncarnation}` } + private releasePreparedExit(event: SshPtyOutputExitEvent): void { + const key = this.exitKey(event) + this.preparedExits.delete(key) + this.releasePreparedExitKey(key) + } + + private releasePreparedExitKey(key: string): void { + const release = this.preparedExitReleases.get(key) + if (!release) { + return + } + this.preparedExitReleases.delete(key) + release() + } + private withCancellationProofDeadline(promise: Promise): Promise { return new Promise((resolve, reject) => { const timer = setTimeout( diff --git a/src/main/ipc/ssh-pty-output-intake-contract.ts b/src/main/ipc/ssh-pty-output-intake-contract.ts index 50776c67778..c6e4cc8238a 100644 --- a/src/main/ipc/ssh-pty-output-intake-contract.ts +++ b/src/main/ipc/ssh-pty-output-intake-contract.ts @@ -53,7 +53,7 @@ export type SshPtyOutputIntakeDependencies = { projection: LegacySshProjectionSemantics ) => { sequence: number; completion: Promise } project: (event: SshPtyOutputDataEvent, projection: LegacySshProjectionSemantics) => void - prepareExit: (event: SshPtyOutputExitEvent) => void + prepareExit: (event: SshPtyOutputExitEvent) => void | (() => void) finalizeExit: (event: SshPtyOutputExitEvent) => void pauseProvider?: (providerGeneration: number, id: string) => boolean resumeProvider?: (providerGeneration: number, id: string) => void diff --git a/src/main/ipc/ssh-pty-output-intake.test.ts b/src/main/ipc/ssh-pty-output-intake.test.ts index 332da53ab58..8b1672f4730 100644 --- a/src/main/ipc/ssh-pty-output-intake.test.ts +++ b/src/main/ipc/ssh-pty-output-intake.test.ts @@ -467,6 +467,65 @@ describe('SshPtyOutputIntake', () => { expect(harness.intake.getDebugSnapshot().projection.records).toBe(0) }) + it('owns renderer exit preparation through finalization and duplicate rejection', async () => { + const releaseRendererExit = vi.fn() + const harness = createHarness({ + prepareExit: vi.fn(() => releaseRendererExit) + }) + const dataReceipt = harness.intake.acceptData(event()) + harness.completions[0]!.resolve() + const receipt = await dataReceipt + harness.intake.publishProjectionPrefix( + [receipt.projection.identity.projectionSemanticsId], + 4, + 4 + ) + const exitEvent = { + id: 'pty-1', + code: 0, + providerGeneration: 1, + ptyIncarnation: 'incarnation-1' + } + const exit = harness.intake.acceptExit(exitEvent) + await Promise.resolve() + + expect(releaseRendererExit).not.toHaveBeenCalled() + await expect(harness.intake.acceptExit(exitEvent)).rejects.toThrow('ssh_output_duplicate_exit') + expect(releaseRendererExit).not.toHaveBeenCalled() + + harness.intake.settleProjectionPrefix('pty-1', 4) + await exit + expect(releaseRendererExit).toHaveBeenCalledOnce() + }) + + it('releases renderer exit preparation when generation close aborts finalization', async () => { + const releaseRendererExit = vi.fn() + const harness = createHarness({ + prepareExit: vi.fn(() => releaseRendererExit) + }) + const dataReceipt = harness.intake.acceptData(event()) + harness.completions[0]!.resolve() + const receipt = await dataReceipt + harness.intake.publishProjectionPrefix( + [receipt.projection.identity.projectionSemanticsId], + 4, + 4 + ) + const exit = harness.intake.acceptExit({ + id: 'pty-1', + code: 0, + providerGeneration: 1, + ptyIncarnation: 'incarnation-1' + }) + await Promise.resolve() + + harness.intake.closeGeneration(1, 'provider-replaced') + + await expect(exit).rejects.toThrow('provider-replaced') + expect(releaseRendererExit).toHaveBeenCalledOnce() + expect(harness.order).not.toContain('exit') + }) + it('retains exit until a required remote source consumer settles', async () => { const harness = createHarness({}, { exitBarrierMs: 1000 }) const remote = harness.intake.getRemoteSourceRangeConsumerHooks() @@ -627,7 +686,9 @@ describe('SshPtyOutputIntake', () => { }) it('closes the provider when exit finalization fails', async () => { + const releaseRendererExit = vi.fn() const harness = createHarness({ + prepareExit: () => releaseRendererExit, finalizeExit: () => { throw new Error('renderer exit send failed') } @@ -642,6 +703,7 @@ describe('SshPtyOutputIntake', () => { }) ).rejects.toThrow('renderer exit send failed') expect(harness.dependencies.closeProvider).toHaveBeenCalledWith(1, 'pty-exit-finalize-failed') + expect(releaseRendererExit).toHaveBeenCalledOnce() }) it('cancels queued work and exit waiters on generation cleanup', async () => { diff --git a/src/main/ipc/ssh-pty-output-intake.ts b/src/main/ipc/ssh-pty-output-intake.ts index 1aab7d82874..67a5b74d28b 100644 --- a/src/main/ipc/ssh-pty-output-intake.ts +++ b/src/main/ipc/ssh-pty-output-intake.ts @@ -208,6 +208,10 @@ export class SshPtyOutputIntake { ) } + hasUnpublishedProjection(id: string): boolean { + return this.projections.hasUnpublished(id) + } + closeGeneration(providerGeneration: number, reason: string): void { this.generationGuard.closeGeneration(providerGeneration) this.admission.closeGeneration(providerGeneration, reason)