diff --git a/src/main/ipc/ssh.test.ts b/src/main/ipc/ssh.test.ts index 4e2996f4c..538cf4bf8 100644 --- a/src/main/ipc/ssh.test.ts +++ b/src/main/ipc/ssh.test.ts @@ -424,6 +424,7 @@ describe('SSH IPC handlers', () => { role: 'session-owner', ownerGeneration: 1, ownerLease: 'ipc-test-owner', + resumed: false, capabilities: { outputFlowControl: { version: 1, windowSu: DEFAULT_PTY_SOURCE_WINDOW_SU } } diff --git a/src/main/ssh/ssh-owner-admission-blocked-error.ts b/src/main/ssh/ssh-owner-admission-blocked-error.ts new file mode 100644 index 000000000..35055cb48 --- /dev/null +++ b/src/main/ssh/ssh-owner-admission-blocked-error.ts @@ -0,0 +1,21 @@ +// Why a typed error instead of raw transport classification: another authenticated connection holds +// the PTY session owner claim, and no relay redeploy or backoff attempt can reconcile that. The +// reconnect ladder has to stop and say what actually happened rather than blame the link. + +export class SshOwnerAdmissionBlockedError extends Error { + readonly name = 'SshOwnerAdmissionBlockedError' + + constructor(targetId: string, options?: { cause?: unknown }) { + super( + `Another connection currently owns the remote terminals for ${targetId}. ` + + `Disconnect that client, or wait for it to release ownership, before reconnecting.`, + options + ) + } +} + +export function isSshOwnerAdmissionBlockedError( + err: unknown +): err is SshOwnerAdmissionBlockedError { + return err instanceof SshOwnerAdmissionBlockedError +} diff --git a/src/main/ssh/ssh-owner-recovery-retry.test.ts b/src/main/ssh/ssh-owner-recovery-retry.test.ts index 9c8f4bd6a..68e835abc 100644 --- a/src/main/ssh/ssh-owner-recovery-retry.test.ts +++ b/src/main/ssh/ssh-owner-recovery-retry.test.ts @@ -1,9 +1,17 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import { + PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR, + PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR, + PTY_CONSUMER_OWNER_HELD_SELF_ERROR, PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR, PTY_CONSUMER_OWNER_RECOVERY_SUPERSEDED_ERROR } from '../../shared/pty-consumer-session' -import { retrySshOwnerRecoveryWhileBlocked } from './ssh-owner-recovery-retry' +import { + isSshOwnerAdmissionBlocked, + retrySshOwnerRecoveryWhileBlocked, + SSH_OWNER_HELD_DISCONNECTED_WAIT_MS, + SSH_OWNER_HELD_SELF_WAIT_MS +} from './ssh-owner-recovery-retry' function publicationPendingError(): Error & { code: number } { return Object.assign(new Error('Owner grant publication is still pending'), { @@ -17,6 +25,10 @@ function supersededError(): Error & { code: number } { }) } +function heldError(code: number): Error & { code: number } { + return Object.assign(new Error('PTY session owner is held'), { code }) +} + function openGate() { return { isCurrent: () => true, @@ -79,6 +91,137 @@ describe('SSH owner recovery retry', () => { expect(attempt).toHaveBeenCalledTimes(2) }) + it('retries a disconnected holder until it releases admission', async () => { + vi.useFakeTimers() + const attempt = vi + .fn<() => Promise>() + .mockRejectedValueOnce(heldError(PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR)) + .mockRejectedValueOnce(heldError(PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR)) + .mockResolvedValue('recovered') + + const recovery = retrySshOwnerRecoveryWhileBlocked(attempt, openGate()) + await vi.advanceTimersByTimeAsync(75) + + await expect(recovery).resolves.toBe('recovered') + expect(attempt).toHaveBeenCalledTimes(3) + }) + + it('never retries an attached holder', async () => { + const error = heldError(PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR) + const attempt = vi.fn<() => Promise>().mockRejectedValue(error) + + // Why blocked, not transient: another connection is live on the claim, so no amount of waiting + // inside this admission changes the answer. + await expect(retrySshOwnerRecoveryWhileBlocked(attempt, openGate())).rejects.toBe(error) + expect(attempt).toHaveBeenCalledOnce() + expect(isSshOwnerAdmissionBlocked(error)).toBe(true) + expect(isSshOwnerAdmissionBlocked(publicationPendingError())).toBe(false) + }) + + it('gives a disconnected holder its own budget rather than the publication one', async () => { + vi.useFakeTimers() + let failures = 0 + const attempt = vi.fn<() => Promise>().mockImplementation(async () => { + if (failures++ < 6) { + throw heldError(PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR) + } + return 'recovered' + }) + + // Why a 60ms publication budget: six backoff waits run well past it, so a single shared deadline + // would give up before the incumbent's clamped grace floor could ever elapse. + const recovery = retrySshOwnerRecoveryWhileBlocked(attempt, openGate(), 60) + await vi.advanceTimersByTimeAsync(SSH_OWNER_HELD_DISCONNECTED_WAIT_MS) + + await expect(recovery).resolves.toBe('recovered') + expect(attempt).toHaveBeenCalledTimes(7) + }) + + it('reports each exhausted retry budget under its own reason', async () => { + vi.useFakeTimers() + const exhausted: string[] = [] + const gate = { ...openGate(), onRetryExhausted: (reason: string) => exhausted.push(reason) } + + const pending = retrySshOwnerRecoveryWhileBlocked( + vi.fn<() => Promise>().mockRejectedValue(publicationPendingError()), + gate, + 60 + ) + const pendingRejection = expect(pending).rejects.toThrow('publication') + await vi.advanceTimersByTimeAsync(60) + await pendingRejection + + const held = retrySshOwnerRecoveryWhileBlocked( + vi + .fn<() => Promise>() + .mockRejectedValue(heldError(PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR)), + gate, + 60 + ) + const heldRejection = expect(held).rejects.toThrow('held') + // Why the disconnected budget and not the 60ms one: each reason carries its own deadline so a + // settling publication cannot spend the budget that waits out a grace floor. + await vi.advanceTimersByTimeAsync(SSH_OWNER_HELD_DISCONNECTED_WAIT_MS) + await heldRejection + + expect(exhausted).toEqual(['publication-pending', 'disconnected-holder']) + }) + + it('starts each budget when its own phase begins', async () => { + vi.useFakeTimers() + const start = Date.now() + let disconnectedAttempts = 0 + const attempt = vi.fn<() => Promise>().mockImplementation(async () => { + // A publication that takes longer to settle than the whole disconnected budget. + if (Date.now() - start < SSH_OWNER_HELD_DISCONNECTED_WAIT_MS + 100) { + throw publicationPendingError() + } + return disconnectedAttempts++ < 3 + ? Promise.reject(heldError(PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR)) + : 'recovered' + }) + + const recovery = retrySshOwnerRecoveryWhileBlocked(attempt, openGate(), 30_000) + await vi.advanceTimersByTimeAsync(30_000) + + // Why this fails on eagerly computed deadlines: the disconnected budget would have started at + // entry and be long gone by the time the first -32045 arrives, giving that phase zero attempts. + await expect(recovery).resolves.toBe('recovered') + expect(disconnectedAttempts).toBeGreaterThan(1) + }) + + it("treats the client's own attached connection as transient, not blocked", async () => { + vi.useFakeTimers() + const selfError = heldError(PTY_CONSUMER_OWNER_HELD_SELF_ERROR) + const attempt = vi + .fn<() => Promise>() + .mockRejectedValueOnce(selfError) + .mockResolvedValue('recovered') + + const recovery = retrySshOwnerRecoveryWhileBlocked(attempt, openGate()) + await vi.advanceTimersByTimeAsync(25) + + await expect(recovery).resolves.toBe('recovered') + // Why not blocked: in a one-app deployment the incumbent is this client's own zombie, so parking + // the target in 'error' with no retry strands the user until they restart the app. + expect(isSshOwnerAdmissionBlocked(selfError)).toBe(false) + }) + + it('lets an exhausted self-holder fall through to ordinary reconnect backoff', async () => { + vi.useFakeTimers() + const selfError = heldError(PTY_CONSUMER_OWNER_HELD_SELF_ERROR) + const attempt = vi.fn<() => Promise>().mockRejectedValue(selfError) + + const recovery = retrySshOwnerRecoveryWhileBlocked(attempt, openGate()) + const rejection = expect(recovery).rejects.toBe(selfError) + await vi.advanceTimersByTimeAsync(SSH_OWNER_HELD_SELF_WAIT_MS) + await rejection + + // The error still surfaces, but unblocked — the relay-lost ladder retries it on backoff, which is + // how this recovered before owner admission became explicit. + expect(isSshOwnerAdmissionBlocked(selfError)).toBe(false) + }) + it('stops waiting when the relay channel closes', async () => { vi.useFakeTimers() let current = true diff --git a/src/main/ssh/ssh-owner-recovery-retry.ts b/src/main/ssh/ssh-owner-recovery-retry.ts index e5e78d9a4..5a958b1dd 100644 --- a/src/main/ssh/ssh-owner-recovery-retry.ts +++ b/src/main/ssh/ssh-owner-recovery-retry.ts @@ -1,16 +1,57 @@ import { + PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR, + PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR, + PTY_CONSUMER_OWNER_HELD_SELF_ERROR, PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR, PTY_CONSUMER_OWNER_RECOVERY_SUPERSEDED_ERROR } from '../../shared/pty-consumer-session' // Why: bound polling when publication is settling or a superseded attempt is closing its transport. export const SSH_OWNER_RECOVERY_WAIT_MS = 3_000 +// Why separate and longer than the relay's grace floor: a disconnected incumbent releases admission +// only after that floor elapses, so this budget must outlast it without borrowing the pending budget. +export const SSH_OWNER_HELD_DISCONNECTED_WAIT_MS = 2_000 +// Why short: the incumbent is this client's own half-open connection, and the relay frees it when its +// keepalive notices — well outside any budget worth blocking a connect on. Poll briefly in case the +// close is already in flight, then let the ordinary relay-lost backoff carry the retry. +export const SSH_OWNER_HELD_SELF_WAIT_MS = 1_000 + +export type SshOwnerRecoveryRetryReason = + | 'publication-pending' + | 'disconnected-holder' + | 'self-holder' + +// Why exported: an attached holder is a decision, not a transport fault, so callers must be able to +// report it as blocked instead of feeding it to reconnect classification as an unexplained failure. +export function isSshOwnerAdmissionBlocked(error: unknown): boolean { + return ( + (error as { code?: unknown } | null | undefined)?.code === + PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR + ) +} + +function retryReasonFor(error: unknown): SshOwnerRecoveryRetryReason | null { + const code = (error as { code?: unknown } | null | undefined)?.code + if ( + code === PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR || + code === PTY_CONSUMER_OWNER_RECOVERY_SUPERSEDED_ERROR + ) { + return 'publication-pending' + } + if (code === PTY_CONSUMER_OWNER_HELD_SELF_ERROR) { + return 'self-holder' + } + // Why an attached holder is deliberately absent: it is blocked, not transient — retrying cannot + // change the answer while another connection is live on the claim. + return code === PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR ? 'disconnected-holder' : null +} const SSH_OWNER_RECOVERY_INITIAL_DELAY_MS = 25 const SSH_OWNER_RECOVERY_MAX_DELAY_MS = 250 type SshOwnerRecoveryRetryGate = { isCurrent: () => boolean onClosed: (listener: () => void) => () => void + onRetryExhausted?: (reason: SshOwnerRecoveryRetryReason) => void } function waitForRetry(delayMs: number, gate: SshOwnerRecoveryRetryGate): Promise { @@ -41,22 +82,32 @@ export async function retrySshOwnerRecoveryWhileBlocked( gate: SshOwnerRecoveryRetryGate, waitMs: number = SSH_OWNER_RECOVERY_WAIT_MS ): Promise { - const deadline = Date.now() + waitMs + const budgetByReason: Record = { + 'publication-pending': waitMs, + 'disconnected-holder': SSH_OWNER_HELD_DISCONNECTED_WAIT_MS, + 'self-holder': SSH_OWNER_HELD_SELF_WAIT_MS + } + // Why each deadline starts when its own phase does: the phases are sequential and independent, so + // charging one reason's budget for time another reason spent leaves it with nothing. A connect that + // waits out a publication first would otherwise get zero attempts at the refusal that follows it. + const deadlineByReason = new Map() let delayMs = SSH_OWNER_RECOVERY_INITIAL_DELAY_MS while (true) { try { return await attempt() } catch (error) { - const code = (error as { code?: unknown } | null | undefined)?.code - if ( - (code !== PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR && - code !== PTY_CONSUMER_OWNER_RECOVERY_SUPERSEDED_ERROR) || - !gate.isCurrent() - ) { + const reason = retryReasonFor(error) + if (reason === null || !gate.isCurrent()) { throw error } + let deadline = deadlineByReason.get(reason) + if (deadline === undefined) { + deadline = Date.now() + budgetByReason[reason] + deadlineByReason.set(reason, deadline) + } const remainingMs = deadline - Date.now() if (remainingMs <= 0) { + gate.onRetryExhausted?.(reason) throw error } await waitForRetry(Math.min(delayMs, remainingMs), gate) diff --git a/src/main/ssh/ssh-pty-consumer-session.test.ts b/src/main/ssh/ssh-pty-consumer-session.test.ts index 76dc25ffa..a1fc612f1 100644 --- a/src/main/ssh/ssh-pty-consumer-session.test.ts +++ b/src/main/ssh/ssh-pty-consumer-session.test.ts @@ -18,6 +18,7 @@ function legacyOwnerGrant(overrides: Record = {}): Record { expectedServerBuildId: 'build-a' }) ).resolves.toEqual({ - mode: 'negotiated', - clientInstanceId: 'client-a', - clientGeneration: 3, - ownerGeneration: 7, - ownerLease: 'lease-a' + state: { + mode: 'negotiated', + clientInstanceId: 'client-a', + clientGeneration: 3, + ownerGeneration: 7, + ownerLease: 'lease-a' + }, + resumed: false }) expect(request).toHaveBeenCalledWith( 'pty.openClient', @@ -51,9 +55,9 @@ describe('openSshPtyConsumerSession', () => { it('carries recovery generation and lease on reconnect', async () => { const { mux, request } = muxReturning( - legacyOwnerGrant({ ownerGeneration: 8, ownerLease: 'lease-b' }) + legacyOwnerGrant({ ownerGeneration: 8, ownerLease: 'lease-a', resumed: true }) ) - await openSshPtyConsumerSession(mux, { + const admission = await openSshPtyConsumerSession(mux, { clientInstanceId: 'client-a', expectedServerBuildId: 'build-a', resume: { ownerGeneration: 7, ownerLease: 'lease-a' } @@ -62,8 +66,30 @@ describe('openSshPtyConsumerSession', () => { expect(request.mock.calls[0][1]).toMatchObject({ resume: { ownerGeneration: 7, ownerLease: 'lease-a' } }) + expect(admission.resumed).toBe(true) }) + it.each([undefined, 'yes', 1, null])( + 'rejects an owner grant that does not state whether the claim was resumed', + async (resumed) => { + const grant = legacyOwnerGrant() + // Why not a legacy peer: the build id already matched, and client and relay ship together. + if (resumed === undefined) { + delete grant.resumed + } else { + grant.resumed = resumed + } + const { mux } = muxReturning(grant) + + await expect( + openSshPtyConsumerSession(mux, { + clientInstanceId: 'client-a', + expectedServerBuildId: 'build-a' + }) + ).rejects.toThrow('whether the claim was resumed') + } + ) + it('rejects a prior or mismatched relay build', async () => { const { mux } = muxReturning(legacyOwnerGrant({ serverBuildId: 'old-build' })) @@ -126,9 +152,12 @@ describe('openSshPtyConsumerSession', () => { outputFlowControl: { requestedWindowSu: 64 } }) ).resolves.toEqual({ - mode: 'legacy-fallback', - clientInstanceId: 'client-a', - serverBuildId: 'build-a' + state: { + mode: 'legacy-fallback', + clientInstanceId: 'client-a', + serverBuildId: 'build-a' + }, + resumed: false }) }) diff --git a/src/main/ssh/ssh-pty-consumer-session.ts b/src/main/ssh/ssh-pty-consumer-session.ts index 8bb272d72..aa5853490 100644 --- a/src/main/ssh/ssh-pty-consumer-session.ts +++ b/src/main/ssh/ssh-pty-consumer-session.ts @@ -27,6 +27,13 @@ export type SshPtyLegacyFallbackState = { export type SshPtyConsumerSessionState = SshPtyConsumerOwnerState | SshPtyLegacyFallbackState +export type SshPtyConsumerAdmission = { + state: SshPtyConsumerSessionState + // Why not on the owner state itself: this describes one admission's outcome, not the persisted + // claim, and it must never round-trip through the recovery record. + resumed: boolean +} + export type OpenSshPtyConsumerSessionOptions = { clientInstanceId: string expectedServerBuildId: string | undefined @@ -68,6 +75,11 @@ function validateGrant( ) { throw new Error('Remote relay did not grant an authenticated PTY session owner') } + // Why not treated as a legacy relay: client and relay ship in one build, and the build id was already + // matched above — a missing `resumed` here is corruption, not an older peer. + if (typeof grant.resumed !== 'boolean') { + throw new Error('Remote relay owner grant did not state whether the claim was resumed') + } const requestedFlow = options.outputFlowControl const grantedFlow = grant.capabilities?.outputFlowControl if (requestedFlow) { @@ -88,7 +100,7 @@ function validateGrant( export async function openSshPtyConsumerSession( mux: SshChannelMultiplexer, options: OpenSshPtyConsumerSessionOptions -): Promise { +): Promise { let result: unknown try { result = await mux.request( @@ -119,23 +131,29 @@ export async function openSshPtyConsumerSession( typeof options.expectedServerBuildId === 'string' && options.expectedServerBuildId.length > 0 ) { - return Object.freeze({ - mode: 'legacy-fallback', - clientInstanceId: options.clientInstanceId, - serverBuildId: options.expectedServerBuildId - }) + return { + state: Object.freeze({ + mode: 'legacy-fallback', + clientInstanceId: options.clientInstanceId, + serverBuildId: options.expectedServerBuildId + }), + resumed: false + } } throw error } const grant = validateGrant(result, options) return { - mode: 'negotiated', - clientInstanceId: options.clientInstanceId, - clientGeneration: grant.clientGeneration, - ownerGeneration: grant.ownerGeneration!, - ownerLease: grant.ownerLease!, - ...(grant.capabilities?.outputFlowControl - ? { outputFlowControl: grant.capabilities.outputFlowControl } - : {}) + state: { + mode: 'negotiated', + clientInstanceId: options.clientInstanceId, + clientGeneration: grant.clientGeneration, + ownerGeneration: grant.ownerGeneration!, + ownerLease: grant.ownerLease!, + ...(grant.capabilities?.outputFlowControl + ? { outputFlowControl: grant.capabilities.outputFlowControl } + : {}) + }, + resumed: grant.resumed! } } diff --git a/src/main/ssh/ssh-relay-session-agent-hooks.integration.test.ts b/src/main/ssh/ssh-relay-session-agent-hooks.integration.test.ts index 1a384a32f..f54afef3f 100644 --- a/src/main/ssh/ssh-relay-session-agent-hooks.integration.test.ts +++ b/src/main/ssh/ssh-relay-session-agent-hooks.integration.test.ts @@ -111,6 +111,7 @@ function createFakeRelay(): FakeRelay { ? (params.resume as { ownerGeneration: number }).ownerGeneration + 1 : 1, ownerLease: 'test-owner-lease', + resumed: params.resume !== undefined, capabilities: { outputFlowControl: { version: 1, windowSu: DEFAULT_PTY_SOURCE_WINDOW_SU } } diff --git a/src/main/ssh/ssh-relay-session-data-delivery.test.ts b/src/main/ssh/ssh-relay-session-data-delivery.test.ts index 56f6c6eef..0966b07fd 100644 --- a/src/main/ssh/ssh-relay-session-data-delivery.test.ts +++ b/src/main/ssh/ssh-relay-session-data-delivery.test.ts @@ -1,10 +1,6 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { SshRelaySession } from './ssh-relay-session' -import { - createMismatchedOwnerRecoveryError, - createMockDeps, - mockDeploySuccess -} from './ssh-relay-session-test-fixtures' +import { createMockDeps, mockDeploySuccess } from './ssh-relay-session-test-fixtures' const { acceptOutputDataMock, @@ -145,16 +141,22 @@ describe('SshRelaySession data delivery', () => { vi.mocked(applySshPtySourceCancellationProof).mockReturnValue(true) vi.mocked(applySshPtySourceRecoveryCancellationProof).mockReturnValue(true) openConsumerSessionMock.mockImplementation(async (_mux, options) => ({ - mode: 'negotiated', - clientInstanceId: options.clientInstanceId, - clientGeneration: 1, - ownerGeneration: 1, - ownerLease: 'test-owner-lease', - ...(options.outputFlowControl - ? { - outputFlowControl: { version: 1, windowSu: options.outputFlowControl.requestedWindowSu } - } - : {}) + state: { + mode: 'negotiated', + clientInstanceId: options.clientInstanceId, + clientGeneration: 1, + ownerGeneration: 1, + ownerLease: 'test-owner-lease', + ...(options.outputFlowControl + ? { + outputFlowControl: { + version: 1, + windowSu: options.outputFlowControl.requestedWindowSu + } + } + : {}) + }, + resumed: options.resume !== undefined })) muxRequestMock.mockResolvedValue([]) mockDeploySuccess() @@ -173,12 +175,15 @@ describe('SshRelaySession data delivery', () => { ) let generation = 0 openConsumerSessionMock.mockImplementation(async (_mux, options) => ({ - mode: 'negotiated', - clientInstanceId: options.clientInstanceId, - clientGeneration: ++generation, - ownerGeneration: generation, - ownerLease: `owner-lease-${generation}`, - outputFlowControl: { version: 1, windowSu: 256 * 1024 } + state: { + mode: 'negotiated', + clientInstanceId: options.clientInstanceId, + clientGeneration: ++generation, + ownerGeneration: generation, + ownerLease: `owner-lease-${generation}`, + outputFlowControl: { version: 1, windowSu: 256 * 1024 } + }, + resumed: options.resume !== undefined })) vi.mocked(getSshPtyAcceptedSourceCheckpoints).mockReturnValue([ { @@ -342,7 +347,7 @@ describe('SshRelaySession data delivery', () => { expect(mockStore.removeSshPtyConsumerRecovery).toHaveBeenCalledWith(targetId) }) - it('clears stale owner recovery and retries a fresh same-build relay once without resume', async () => { + it('voids checkpoints for a fresh claim without a second owner request', async () => { const targetId = 'fresh-relay-retry' const { mockConn, mockStore, mockPortForward, getMainWindow } = createMockDeps() vi.mocked(deployAndLaunchRelay).mockResolvedValue({ @@ -365,16 +370,17 @@ describe('SshRelaySession data delivery', () => { await first.establish(mockConn) first.detach() - openConsumerSessionMock - .mockRejectedValueOnce(createMismatchedOwnerRecoveryError()) - .mockImplementationOnce(async (_mux, options) => ({ + openConsumerSessionMock.mockImplementationOnce(async (_mux, options) => ({ + state: { mode: 'negotiated', clientInstanceId: options.clientInstanceId, clientGeneration: 1, ownerGeneration: 1, ownerLease: 'fresh-owner-lease', outputFlowControl: { version: 1, windowSu: 256 * 1024 } - })) + }, + resumed: false + })) vi.mocked(getPtyIdsForConnection).mockReturnValue([`ssh:${targetId}@@pty-1`]) vi.mocked(getSshPtyProvider).mockImplementation( () => vi.mocked(registerSshPtyProvider).mock.calls.at(-1)?.[1] @@ -391,9 +397,10 @@ describe('SshRelaySession data delivery', () => { const retryCalls = openConsumerSessionMock.mock.calls .slice(openCallCountBeforeRetry) .map(([, options]) => options) - expect(retryCalls).toHaveLength(2) + // Why one call: the relay answers a proof it cannot match with a fresh claim, so the client never + // needs a second, resume-less request to get owner authority back. + expect(retryCalls).toHaveLength(1) expect(retryCalls[0]).toHaveProperty('resume') - expect(retryCalls[1]).not.toHaveProperty('resume') expect(attachForReconnectMock).toHaveBeenCalledWith( 'pty-1', undefined, @@ -644,12 +651,15 @@ describe('SshRelaySession data delivery', () => { openConsumerSessionMock.mockImplementation(async (_mux, options) => { generation++ return { - mode: 'negotiated', - clientInstanceId: options.clientInstanceId, - clientGeneration: generation, - ownerGeneration: generation, - ownerLease: `owner-lease-${generation}`, - outputFlowControl: { version: 1, windowSu: 256 * 1024 } + state: { + mode: 'negotiated', + clientInstanceId: options.clientInstanceId, + clientGeneration: generation, + ownerGeneration: generation, + ownerLease: `owner-lease-${generation}`, + outputFlowControl: { version: 1, windowSu: 256 * 1024 } + }, + resumed: options.resume !== undefined } }) vi.mocked(getSshPtyAcceptedSourceCheckpoints).mockReturnValue([ diff --git a/src/main/ssh/ssh-relay-session-model-migration.test.ts b/src/main/ssh/ssh-relay-session-model-migration.test.ts index 9bd221635..ae5db55e0 100644 --- a/src/main/ssh/ssh-relay-session-model-migration.test.ts +++ b/src/main/ssh/ssh-relay-session-model-migration.test.ts @@ -1,10 +1,6 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { SshRelaySession } from './ssh-relay-session' -import { - createMismatchedOwnerRecoveryError, - createMockDeps, - mockDeploySuccess -} from './ssh-relay-session-test-fixtures' +import { createMockDeps, mockDeploySuccess } from './ssh-relay-session-test-fixtures' type SettledMigration = { status: 'settled' @@ -135,16 +131,22 @@ describe('SshRelaySession model migration', () => { vi.mocked(getPtyIdsForConnection).mockReturnValue([]) vi.mocked(getSshPtyAcceptedSourceCheckpoints).mockReturnValue([]) openConsumerSessionMock.mockImplementation(async (_mux, options) => ({ - mode: 'negotiated', - clientInstanceId: options.clientInstanceId, - clientGeneration: 1, - ownerGeneration: 1, - ownerLease: 'owner-lease-1', - ...(options.outputFlowControl - ? { - outputFlowControl: { version: 1, windowSu: options.outputFlowControl.requestedWindowSu } - } - : {}) + state: { + mode: 'negotiated', + clientInstanceId: options.clientInstanceId, + clientGeneration: 1, + ownerGeneration: 1, + ownerLease: 'owner-lease-1', + ...(options.outputFlowControl + ? { + outputFlowControl: { + version: 1, + windowSu: options.outputFlowControl.requestedWindowSu + } + } + : {}) + }, + resumed: options.resume !== undefined })) muxRequestMock.mockResolvedValue([]) mockDeploySuccess() @@ -209,7 +211,7 @@ describe('SshRelaySession model migration', () => { ) }) - it('waits for a stale-owner migration before requesting restore', async () => { + it('waits for a voided-checkpoint migration before requesting restore', async () => { const targetId = 'migration-stale-owner' const appPtyId = `ssh:${targetId}@@pty-1` const migration = pendingMigration() @@ -232,23 +234,24 @@ describe('SshRelaySession model migration', () => { vi.mocked(getSshPtyProvider).mockImplementation( () => vi.mocked(registerSshPtyProvider).mock.calls.at(-1)?.[1] ) - openConsumerSessionMock - .mockRejectedValueOnce(createMismatchedOwnerRecoveryError()) - .mockImplementationOnce(async (_mux, options) => ({ + openConsumerSessionMock.mockImplementationOnce(async (_mux, options) => ({ + state: { mode: 'negotiated', clientInstanceId: options.clientInstanceId, clientGeneration: 2, ownerGeneration: 2, ownerLease: 'fresh-owner-lease', outputFlowControl: { version: 1, windowSu: 256 * 1024 } - })) + }, + resumed: false + })) attachForReconnectMock.mockResolvedValue({ incarnationId: 'incarnation-1', sourceRecovery: { status: 'restoreRequired', reason: 'checkpointUnavailable' } }) const reconnect = session.reconnect(deps.mockConn) - await vi.waitFor(() => expect(openConsumerSessionMock).toHaveBeenCalledTimes(3)) + await vi.waitFor(() => expect(openConsumerSessionMock).toHaveBeenCalledTimes(2)) expect(attachForReconnectMock).not.toHaveBeenCalled() migration.resolve(settledMigration(appPtyId, 8)) diff --git a/src/main/ssh/ssh-relay-session-recovery-durability.test.ts b/src/main/ssh/ssh-relay-session-recovery-durability.test.ts index e2b3ba434..ed0136f91 100644 --- a/src/main/ssh/ssh-relay-session-recovery-durability.test.ts +++ b/src/main/ssh/ssh-relay-session-recovery-durability.test.ts @@ -1,13 +1,12 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { + PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR, + PTY_CONSUMER_OWNER_HELD_SELF_ERROR, PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR, PTY_CONSUMER_OWNER_RECOVERY_SUPERSEDED_ERROR } from '../../shared/pty-consumer-session' import { SshRelaySession } from './ssh-relay-session' -import { - createMismatchedOwnerRecoveryError, - createMockDeps -} from './ssh-relay-session-test-fixtures' +import { createMockDeps } from './ssh-relay-session-test-fixtures' import { getSshPtyConsumerRecovery } from './ssh-pty-consumer-recovery' const { muxRequestMock, openConsumerSessionMock } = vi.hoisted(() => ({ @@ -108,11 +107,14 @@ describe('SshRelaySession consumer recovery durability', () => { vi.clearAllMocks() muxRequestMock.mockResolvedValue([]) openConsumerSessionMock.mockImplementation(async (_mux, options) => ({ - mode: 'negotiated', - clientInstanceId: options.clientInstanceId, - clientGeneration: 1, - ownerGeneration: 1, - ownerLease: 'test-owner-lease' + state: { + mode: 'negotiated', + clientInstanceId: options.clientInstanceId, + clientGeneration: 1, + ownerGeneration: 1, + ownerLease: 'test-owner-lease' + }, + resumed: false })) vi.mocked(deployAndLaunchRelay).mockResolvedValue({ transport: { write: vi.fn(), onData: vi.fn(), onClose: vi.fn() }, @@ -277,8 +279,68 @@ describe('SshRelaySession consumer recovery durability', () => { expect(completed).toBe(true) }) - it('does not retry a stale owner after disposal wins the recovery-removal race', async () => { - const targetId = 'target-stale-owner-disposal' + it('reports an attached owner as terminal instead of a lost relay', async () => { + const targetId = 'target-owner-held-attached' + const { mockConn, mockStore, mockPortForward, getMainWindow } = createMockDeps() + const session = new SshRelaySession(targetId, getMainWindow, mockStore, mockPortForward) + const onTerminal = vi.fn() + const onRelayLost = vi.fn() + session.setOnTerminalRelayError(onTerminal) + session.setOnRelayLost(onRelayLost) + await session.establish(mockConn) + + openConsumerSessionMock.mockRejectedValueOnce( + Object.assign(new Error('PTY session owner is held by an attached connection'), { + code: PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR + }) + ) + await session.reconnect(mockConn) + + // Why not relay-lost: the relay answered and refused on purpose, so redeploy backoff would spend + // its whole budget on a link that is working. + expect(onRelayLost).not.toHaveBeenCalled() + expect(onTerminal).toHaveBeenCalledWith( + targetId, + expect.objectContaining({ name: 'SshOwnerAdmissionBlockedError' }) + ) + expect(onTerminal.mock.calls[0]?.[1]?.message).toContain('owns the remote terminals') + expect(mockStore.removeSshPtyConsumerRecovery).not.toHaveBeenCalled() + expect(openConsumerSessionMock).toHaveBeenCalledTimes(2) + session.dispose() + }) + + it("routes this client's own attached connection to relay-lost recovery, not to a parked error", async () => { + const targetId = 'target-owner-held-self' + const { mockConn, mockStore, mockPortForward, getMainWindow } = createMockDeps() + const session = new SshRelaySession(targetId, getMainWindow, mockStore, mockPortForward) + const onTerminal = vi.fn() + const onRelayLost = vi.fn() + session.setOnTerminalRelayError(onTerminal) + session.setOnRelayLost(onRelayLost) + await session.establish(mockConn) + + // The incumbent is this app's own half-open connection, which the relay has not yet observed + // closing — the ordinary single-app case, since only SshRelaySession ever requests session-owner. + openConsumerSessionMock.mockRejectedValue( + Object.assign( + new Error("PTY session owner is held by this client's own earlier connection"), + { + code: PTY_CONSUMER_OWNER_HELD_SELF_ERROR + } + ) + ) + await session.reconnect(mockConn) + + // Why not terminal: parking here clears backoff and never retries, so the user stays locked out + // of their own session until they restart the app. Backoff is what lets keepalive reap the zombie. + expect(onTerminal).not.toHaveBeenCalled() + expect(onRelayLost).toHaveBeenCalled() + expect(mockStore.removeSshPtyConsumerRecovery).not.toHaveBeenCalled() + session.dispose() + }) + + it('recovers a forgotten owner record in one request without deleting recovery identity', async () => { + const targetId = 'target-forgotten-owner-record' const { mockConn, mockStore, mockPortForward, getMainWindow } = createMockDeps() vi.mocked(mockStore.getSshPtyConsumerRecovery).mockReturnValue({ targetId, @@ -286,30 +348,75 @@ describe('SshRelaySession consumer recovery durability', () => { serverBuildId: 'test-relay-build', clientGeneration: 1, ownerGeneration: 1, - ownerLease: 'stale-owner' + ownerLease: 'forgotten-owner' }) - let signalRemovalStarted!: () => void - const removalStarted = new Promise((resolve) => { - signalRemovalStarted = resolve - }) - let settleRemoval!: () => void - vi.mocked(mockStore.removeSshPtyConsumerRecovery).mockImplementationOnce(() => { - signalRemovalStarted() - return new Promise((resolve) => { - settleRemoval = resolve - }) - }) - openConsumerSessionMock.mockRejectedValueOnce(createMismatchedOwnerRecoveryError()) + openConsumerSessionMock.mockImplementationOnce(async () => ({ + state: { + mode: 'negotiated', + clientInstanceId: 'persisted-client', + clientGeneration: 4, + ownerGeneration: 9, + ownerLease: 'fresh-owner' + }, + resumed: false + })) const session = new SshRelaySession(targetId, getMainWindow, mockStore, mockPortForward) + getSshPtyConsumerRecovery(targetId)!.checkpointsByAppPtyId.set('pty-1', { + id: 'pty-1' + } as unknown as never) - const establishing = session.establish(mockConn) - const failed = expect(establishing).rejects.toThrow('Session disposed during establish') - await removalStarted - const disposal = session.disposeAndPersist() - settleRemoval() - await Promise.all([failed, disposal]) + await session.establish(mockConn) expect(openConsumerSessionMock).toHaveBeenCalledTimes(1) + expect(openConsumerSessionMock.mock.calls[0]?.[1]).toMatchObject({ + resume: { ownerGeneration: 1, ownerLease: 'forgotten-owner' } + }) + // Why both: the fresh claim voids checkpoints taken under the old lease, but the recovery identity + // is what lets this client keep resuming the target at all — a refusal must never cost it. + expect(getSshPtyConsumerRecovery(targetId)!.checkpointsByAppPtyId.size).toBe(0) + expect(mockStore.removeSshPtyConsumerRecovery).not.toHaveBeenCalled() + expect(mockStore.upsertSshPtyConsumerRecovery).toHaveBeenCalledWith( + expect.objectContaining({ + targetId, + clientInstanceId: 'persisted-client', + ownerGeneration: 9, + ownerLease: 'fresh-owner' + }) + ) + session.dispose() + }) + + it('keeps checkpoints when a resumed claim comes back', async () => { + const targetId = 'target-resumed-owner-record' + const { mockConn, mockStore, mockPortForward, getMainWindow } = createMockDeps() + vi.mocked(mockStore.getSshPtyConsumerRecovery).mockReturnValue({ + targetId, + clientInstanceId: 'persisted-client', + serverBuildId: 'test-relay-build', + clientGeneration: 1, + ownerGeneration: 1, + ownerLease: 'persisted-owner' + }) + openConsumerSessionMock.mockImplementationOnce(async () => ({ + state: { + mode: 'negotiated', + clientInstanceId: 'persisted-client', + clientGeneration: 4, + ownerGeneration: 2, + ownerLease: 'persisted-owner' + }, + resumed: true + })) + const session = new SshRelaySession(targetId, getMainWindow, mockStore, mockPortForward) + getSshPtyConsumerRecovery(targetId)!.checkpointsByAppPtyId.set('pty-1', { + id: 'pty-1' + } as unknown as never) + + await session.establish(mockConn) + + expect(getSshPtyConsumerRecovery(targetId)!.checkpointsByAppPtyId.has('pty-1')).toBe(true) + expect(mockStore.removeSshPtyConsumerRecovery).not.toHaveBeenCalled() + session.dispose() }) it('leaves recovery state alone when a newer owner already claimed the target record', async () => { @@ -330,28 +437,84 @@ describe('SshRelaySession consumer recovery durability', () => { ownerGeneration: 5, ownerLease: 'winner-owner' } - openConsumerSessionMock.mockImplementationOnce(() => { - // Why inside the rejection: the record is target-scoped, so the winner can land while this - // attempt is still unwinding its own resume. + openConsumerSessionMock.mockImplementationOnce(async () => { + // Why inside the open: the record is target-scoped, so the winner can land while this attempt is + // still waiting on its own resume, leaving this attempt's `previousOwner` snapshot stale. const record = getSshPtyConsumerRecovery(targetId)! record.owner = winner - record.checkpointsByAppPtyId.set('pty-1', { - id: 'pty-1' - } as unknown as never) - return Promise.reject(createMismatchedOwnerRecoveryError()) + record.checkpointsByAppPtyId.set('pty-1', { id: 'pty-1' } as unknown as never) + return { + state: { + mode: 'negotiated', + clientInstanceId: 'persisted-client', + clientGeneration: 3, + ownerGeneration: 1, + ownerLease: 'loser-owner' + }, + resumed: false + } }) - openConsumerSessionMock.mockRejectedValueOnce(new Error('fresh open failed')) const session = new SshRelaySession(targetId, getMainWindow, mockStore, mockPortForward) - await expect(session.establish(mockConn)).rejects.toThrow('fresh open failed') + await session.establish(mockConn) - const record = getSshPtyConsumerRecovery(targetId)! - expect(record.owner).toBe(winner) - expect(record.checkpointsByAppPtyId.has('pty-1')).toBe(true) + expect(getSshPtyConsumerRecovery(targetId)!.checkpointsByAppPtyId.has('pty-1')).toBe(true) expect(mockStore.removeSshPtyConsumerRecovery).not.toHaveBeenCalled() session.dispose() }) + it('mutates no recovery state after a local attempt loses its authority', async () => { + const targetId = 'target-owner-attempt-superseded' + const { mockConn, mockStore, mockPortForward, getMainWindow } = createMockDeps() + vi.mocked(mockStore.getSshPtyConsumerRecovery).mockReturnValue({ + targetId, + clientInstanceId: 'persisted-client', + serverBuildId: 'test-relay-build', + clientGeneration: 1, + ownerGeneration: 1, + ownerLease: 'persisted-owner' + }) + let signalOpenStarted!: () => void + const openStarted = new Promise((resolve) => { + signalOpenStarted = resolve + }) + let finishOpen!: (value: unknown) => void + openConsumerSessionMock.mockImplementationOnce(() => { + signalOpenStarted() + return new Promise((resolve) => { + finishOpen = resolve + }) + }) + const session = new SshRelaySession(targetId, getMainWindow, mockStore, mockPortForward) + getSshPtyConsumerRecovery(targetId)!.checkpointsByAppPtyId.set('pty-1', { + id: 'pty-1' + } as unknown as never) + + const superseded = session.establish(mockConn) + const failed = expect(superseded).rejects.toThrow('Session disposed during establish') + await openStarted + // Why detach and not disposal: disposal legitimately removes the record, which would hide whether + // the attempt that lost its authority mutated anything on its way out. + await session.detachAndPersist() + finishOpen({ + state: { + mode: 'negotiated', + clientInstanceId: 'persisted-client', + clientGeneration: 3, + ownerGeneration: 3, + ownerLease: 'superseded-owner' + }, + resumed: false + }) + await failed + + expect(getSshPtyConsumerRecovery(targetId)!.checkpointsByAppPtyId.has('pty-1')).toBe(true) + expect(mockStore.upsertSshPtyConsumerRecovery).not.toHaveBeenCalledWith( + expect.objectContaining({ ownerLease: 'superseded-owner' }) + ) + expect(mockStore.removeSshPtyConsumerRecovery).not.toHaveBeenCalled() + }) + it('does not remember a consumer opened after establish was disposed', async () => { const { mockConn, mockStore, mockPortForward, getMainWindow } = createMockDeps() let signalOpenStarted!: () => void @@ -377,11 +540,14 @@ describe('SshRelaySession consumer recovery durability', () => { await openStarted await session.disposeAndPersist() finishOpen({ - mode: 'negotiated', - clientInstanceId: 'late-client', - clientGeneration: 1, - ownerGeneration: 1, - ownerLease: 'late-owner' + state: { + mode: 'negotiated', + clientInstanceId: 'late-client', + clientGeneration: 1, + ownerGeneration: 1, + ownerLease: 'late-owner' + }, + resumed: false }) await failed diff --git a/src/main/ssh/ssh-relay-session-recovery-races.test.ts b/src/main/ssh/ssh-relay-session-recovery-races.test.ts index 81039d901..e44359ba5 100644 --- a/src/main/ssh/ssh-relay-session-recovery-races.test.ts +++ b/src/main/ssh/ssh-relay-session-recovery-races.test.ts @@ -191,12 +191,15 @@ describe('SshRelaySession recovery race fencing', () => { }> { let generation = 0 openConsumerSessionMock.mockImplementation(async (_mux, options) => ({ - mode: 'negotiated', - clientInstanceId: options.clientInstanceId, - clientGeneration: ++generation, - ownerGeneration: generation, - ownerLease: `owner-lease-${generation}`, - outputFlowControl: { version: 1, windowSu: 256 * 1024 } + state: { + mode: 'negotiated', + clientInstanceId: options.clientInstanceId, + clientGeneration: ++generation, + ownerGeneration: generation, + ownerLease: `owner-lease-${generation}`, + outputFlowControl: { version: 1, windowSu: 256 * 1024 } + }, + resumed: options.resume !== undefined })) vi.mocked(getSshPtyAcceptedSourceCheckpoints).mockReturnValue([ { diff --git a/src/main/ssh/ssh-relay-session-test-fixtures.ts b/src/main/ssh/ssh-relay-session-test-fixtures.ts index 5da4f90c3..c5f084a87 100644 --- a/src/main/ssh/ssh-relay-session-test-fixtures.ts +++ b/src/main/ssh/ssh-relay-session-test-fixtures.ts @@ -1,6 +1,5 @@ import { vi, type Mock } from 'vitest' import type { BrowserWindow } from 'electron' -import { PtyConsumerSession } from '../../shared/pty-consumer-session' import type { SshConnection } from './ssh-connection' import type { Store } from '../persistence' import type { SshPortForwardManager } from './ssh-port-forward' @@ -54,39 +53,3 @@ export function mockDeploySuccess(): void { platform: 'linux-x64' }) } - -export function createMismatchedOwnerRecoveryError(): unknown { - const stateMachine = new PtyConsumerSession({ - serverBuildId: 'test-relay-build', - createLease: () => 'retained-owner-lease' - }) - const owner = stateMachine.admit( - { clientInstanceId: 'retained-client', requestedRole: 'session-owner' }, - { - connectionId: 'retained-connection', - principal: 'retained-principal', - authenticated: true, - allowSessionOwner: true - } - ) - owner.commitPublication() - stateMachine.close('retained-connection') - try { - stateMachine.admit( - { - clientInstanceId: 'retained-client', - requestedRole: 'session-owner', - resume: { ownerGeneration: 1, ownerLease: 'retained-owner-lease' } - }, - { - connectionId: 'stale-connection', - principal: 'stale-principal', - authenticated: true, - allowSessionOwner: true - } - ) - } catch (error) { - return error - } - throw new Error('Expected mismatched owner recovery to fail') -} diff --git a/src/main/ssh/ssh-relay-session.ts b/src/main/ssh/ssh-relay-session.ts index 8126043c8..2e8816dd8 100644 --- a/src/main/ssh/ssh-relay-session.ts +++ b/src/main/ssh/ssh-relay-session.ts @@ -6,7 +6,6 @@ import type { BrowserWindow } from 'electron' import { deployAndLaunchRelay } from './ssh-relay-deploy' import { execCommand } from './ssh-relay-deploy-helpers' import { isRelayVersionMismatchError } from './ssh-relay-version-mismatch-error' -import type { RelayVersionMismatchError } from './ssh-relay-version-mismatch-error' import { SshChannelMultiplexer } from './ssh-channel-multiplexer' import { SshPtyProvider } from '../providers/ssh-pty-provider' import type { SshPtyAttachResult } from '../providers/ssh-pty-session-reattach' @@ -80,8 +79,14 @@ import { import type { Store } from '../persistence' import type { OrcaRuntimeService } from '../runtime/orca-runtime' import { DEFAULT_PTY_SOURCE_WINDOW_SU } from '../../shared/pty-source-credit-contract' -import { PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR } from '../../shared/pty-consumer-session' -import { retrySshOwnerRecoveryWhileBlocked } from './ssh-owner-recovery-retry' +import { + isSshOwnerAdmissionBlocked, + retrySshOwnerRecoveryWhileBlocked +} from './ssh-owner-recovery-retry' +import { + isSshOwnerAdmissionBlockedError, + SshOwnerAdmissionBlockedError +} from './ssh-owner-admission-blocked-error' import { runRemoteOrcaCli } from './ssh-remote-orca-cli' import { acknowledgeRemoteOrcaCliPostOutput, @@ -97,6 +102,8 @@ import { isTerminalLeafId, makePaneKey } from '../../shared/stable-pane-id' import { isValidTerminalTabId } from '../../shared/terminal-tab-id' import { openSshPtyConsumerSession, + type OpenSshPtyConsumerSessionOptions, + type SshPtyConsumerAdmission, type SshPtyConsumerOwnerState, type SshPtyConsumerSessionState } from './ssh-pty-consumer-session' @@ -112,8 +119,7 @@ import { detachSshPtyConsumerRecovery, forgetSshPtyConsumerRecovery, getSshPtyConsumerRecovery, - rememberSshPtyConsumerRecovery, - removeSshPtyConsumerOwnerRecovery + rememberSshPtyConsumerRecovery } from './ssh-pty-consumer-recovery' export type RelaySessionState = 'idle' | 'deploying' | 'ready' | 'reconnecting' | 'disposed' @@ -286,10 +292,9 @@ export class SshRelaySession { private muxNotificationCleanup: (() => void) | null = null // Why: onStateChange never fires when the relay channel closes but SSH stays up; this callback lets ssh.ts drive relay-level reconnect. private _onRelayLost: ((targetId: string) => void) | null = null - // Why: version mismatch is terminal, so it needs a separate callback from _onRelayLost (which expects a recoverable transport drop). - private _onTerminalRelayError: - | ((targetId: string, err: RelayVersionMismatchError) => void) - | null = null + // Why: a version mismatch or a blocked owner admission is terminal, so it needs a separate callback + // from _onRelayLost (which expects a recoverable transport drop). + private _onTerminalRelayError: ((targetId: string, err: Error) => void) | null = null private _onReady: ((targetId: string) => void) | null = null private portScanner: PortScanner | null = null private currentConnection: SshConnection | null = null @@ -356,7 +361,7 @@ export class SshRelaySession { this._onRelayLost = cb } - setOnTerminalRelayError(cb: (targetId: string, err: RelayVersionMismatchError) => void): void { + setOnTerminalRelayError(cb: (targetId: string, err: Error) => void): void { this._onTerminalRelayError = cb } @@ -559,10 +564,11 @@ export class SshRelaySession { ) this._state = 'idle' } - // Why: a version mismatch on first connect is terminal (deployed binary vs. a still-running legacy daemon); notify the callback but still rethrow. - if (isRelayVersionMismatchError(err)) { + // Why: terminal on first connect — a deployed binary against a still-running legacy daemon, or a + // claim another connection holds. Notify the callback but still rethrow. + if (isRelayVersionMismatchError(err) || isSshOwnerAdmissionBlockedError(err)) { console.warn( - `[ssh-relay-session] Terminal relay version mismatch on initial connect for ${this.targetId}: ${err.message}` + `[ssh-relay-session] Terminal relay error on initial connect for ${this.targetId}: ${err.message}` ) this._onTerminalRelayError?.(this.targetId, err) } @@ -710,10 +716,11 @@ export class SshRelaySession { : 'connection_lost' ) } - // Why: version-mismatch is terminal — fire the typed callback and drop out of 'reconnecting' since backoff retry can't reconcile it. - if (isRelayVersionMismatchError(err)) { + // Why terminal: neither a version mismatch nor a blocked owner claim is reconcilable by backoff + // retry, so fire the typed callback and drop out of 'reconnecting'. + if (isRelayVersionMismatchError(err) || isSshOwnerAdmissionBlockedError(err)) { console.warn( - `[ssh-relay-session] Terminal relay version mismatch for ${this.targetId}: ${err.message}` + `[ssh-relay-session] Terminal relay error for ${this.targetId}: ${err.message}` ) if (this.abortController === abortController && !this.isDisposed()) { this._state = 'idle' @@ -1071,6 +1078,19 @@ export class SshRelaySession { allowSameBuildLegacyFallback: true, outputFlowControl: { requestedWindowSu: DEFAULT_PTY_SOURCE_WINDOW_SU } } + const admission = await this.admitPtyConsumerOwner(mux, previousOwner, options, ownsAttempt) + if (previousOwner && !admission.resumed) { + this.voidPtyConsumerCheckpoints(previousOwner, ownsAttempt) + } + return admission.state + } + + private async admitPtyConsumerOwner( + mux: SshChannelMultiplexer, + previousOwner: SshPtyConsumerOwnerState | null, + options: OpenSshPtyConsumerSessionOptions, + ownsAttempt: () => boolean + ): Promise { try { return await retrySshOwnerRecoveryWhileBlocked( () => @@ -1091,48 +1111,47 @@ export class SshRelaySession { } ) } catch (error) { - if ( - !previousOwner || - (error as { code?: unknown }).code !== PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR - ) { - throw error + // Why converted here: past this point the failure travels the same path as a dropped transport, + // where backoff would keep redeploying a relay that is working fine and refusing on purpose. + if (isSshOwnerAdmissionBlocked(error)) { + throw new SshOwnerAdmissionBlockedError(this.targetId, { cause: error }) } - const recovery = getSshPtyConsumerRecovery(this.targetId) - // Why identity-guarded: the record is target-scoped and its clientInstanceId is shared by every - // session for that target, so only a record still describing the owner this attempt tried to - // resume is ours to drop — otherwise a loser wipes the winner's checkpoints. - const ownsRecoveryRecord = - ownsAttempt() && - (!recovery?.owner || - (recovery.owner.ownerGeneration === previousOwner.ownerGeneration && - recovery.owner.ownerLease === previousOwner.ownerLease)) - if (ownsRecoveryRecord) { - if (recovery) { - delete recovery.owner - recovery.checkpointsByAppPtyId.clear() - for (const [ptyId, migration] of recovery.modelMigrationsByAppPtyId) { - recovery.modelMigrationsByAppPtyId.set( - ptyId, - migration.then(() => - Object.freeze({ - status: 'checkpoint-unavailable' as const, - reason: 'completion-failed' as const - }) - ) - ) - } - } - await removeSshPtyConsumerOwnerRecovery( - this.targetId, - this.ptyConsumerClientInstanceId, - this.store + throw error + } + } + + // Why the recovery row and clientInstanceId survive: the relay minted a fresh claim, which voids the + // checkpoints taken under the old one but says nothing about our identity for this target. The caller + // durably records the new lease before ready, so removal here would only lose the identity. + private voidPtyConsumerCheckpoints( + previousOwner: SshPtyConsumerOwnerState, + ownsAttempt: () => boolean + ): void { + const recovery = getSshPtyConsumerRecovery(this.targetId) + // Why identity-guarded: the record is target-scoped and its clientInstanceId is shared by every + // session for that target, so only a record still describing the owner this attempt tried to + // resume is ours to void — otherwise a loser wipes the winner's checkpoints. + if ( + !recovery || + !ownsAttempt() || + (recovery.owner && + (recovery.owner.ownerGeneration !== previousOwner.ownerGeneration || + recovery.owner.ownerLease !== previousOwner.ownerLease)) + ) { + return + } + delete recovery.owner + recovery.checkpointsByAppPtyId.clear() + for (const [ptyId, migration] of recovery.modelMigrationsByAppPtyId) { + recovery.modelMigrationsByAppPtyId.set( + ptyId, + migration.then(() => + Object.freeze({ + status: 'checkpoint-unavailable' as const, + reason: 'completion-failed' as const + }) ) - } - if (!ownsAttempt()) { - throw new Error('Session disposed during establish') - } - this.ptyConsumerSessionState = null - return openSshPtyConsumerSession(mux, options) + ) } } diff --git a/src/relay/dispatcher-client-close-cause.test.ts b/src/relay/dispatcher-client-close-cause.test.ts new file mode 100644 index 000000000..2eb926900 --- /dev/null +++ b/src/relay/dispatcher-client-close-cause.test.ts @@ -0,0 +1,67 @@ +import { describe, expect, it, vi } from 'vitest' +import { RelayDispatcher, type SinkWriteSettlement } from './dispatcher' +import { DISPATCHER_CONTROL_QUEUE_MAX_FRAMES } from './dispatcher-writer-admission' + +// A detach carries the reason the client went away, because the PTY owner grace is only safe to +// shorten against an owner the relay watched leave. Everything here is about keeping those two +// answers apart. +describe('RelayDispatcher client close cause', () => { + it('reports a backpressure teardown as a relay-local close, not a peer close', () => { + // Why this matters beyond the argument shape: "capacity exceeded" is what a client that is alive + // but slow to drain looks like, and a consumer that read this as a peer close would shorten that + // owner's grace and hand its session to someone else while it is still there. + const detachListener = vi.fn() + const settlements: ((result: SinkWriteSettlement) => void)[] = [] + const dispatcher = new RelayDispatcher( + (_data, onSettled) => { + settlements.push(onSettled) + return true + }, + { supportsWriteCallback: true } + ) + try { + dispatcher.onClientDetached(detachListener) + + // Every write stays in flight, so the control lane fills and the relay destroys its own client. + for (let frame = 0; frame <= DISPATCHER_CONTROL_QUEUE_MAX_FRAMES; frame += 1) { + dispatcher.notifyControl('control.fill', { frame }) + } + + expect(detachListener).toHaveBeenCalledWith(1, 'local') + } finally { + dispatcher.dispose() + } + }) + + it('reports a detach the transport observed as a peer close', () => { + const detachListener = vi.fn() + const dispatcher = new RelayDispatcher(() => true) + try { + dispatcher.onClientDetached(detachListener) + const clientId = dispatcher.attachClient(() => true) + + // Why the caller states it: only the socket layer sees the peer's transport end, so the + // evidence has to travel from there rather than be inferred here. + dispatcher.detachClient(clientId, 'peer-closed') + + expect(detachListener).toHaveBeenCalledWith(clientId, 'peer-closed') + } finally { + dispatcher.dispose() + } + }) + + it('defaults an unqualified detach to the cautious answer', () => { + const detachListener = vi.fn() + const dispatcher = new RelayDispatcher(() => true) + try { + dispatcher.onClientDetached(detachListener) + const clientId = dispatcher.attachClient(() => true) + + dispatcher.detachClient(clientId) + + expect(detachListener).toHaveBeenCalledWith(clientId, 'local') + } finally { + dispatcher.dispose() + } + }) +}) diff --git a/src/relay/dispatcher.test.ts b/src/relay/dispatcher.test.ts index 3ec57dcf8..f38d3833d 100644 --- a/src/relay/dispatcher.test.ts +++ b/src/relay/dispatcher.test.ts @@ -411,7 +411,9 @@ describe('RelayDispatcher', () => { dispatcher.invalidateClient() - expect(listener).toHaveBeenCalledWith(1) + // Why the cause is asserted: an unqualified invalidate is the relay's own decision, and only a + // caller that watched the peer's transport end may report 'peer-closed'. + expect(listener).toHaveBeenCalledWith(1, 'local') }) it('detaches the primary client when its write throws (frame lost, trigger reconnect)', () => { @@ -436,7 +438,8 @@ describe('RelayDispatcher', () => { // Fix: the write failure detaches the primary so the reconnect/reattach // machinery runs promptly instead of waiting for keepalive timeout. - expect(detachListener).toHaveBeenCalledWith(1) + // Why 'local': a throwing sink is this relay's write failing, not observed proof the peer left. + expect(detachListener).toHaveBeenCalledWith(1, 'local') // Recovery: a reconnecting socket swaps the write via setWrite; the client // is usable again and later frames flow to the new sink. diff --git a/src/relay/dispatcher.ts b/src/relay/dispatcher.ts index d89d0795e..c3fc87609 100644 --- a/src/relay/dispatcher.ts +++ b/src/relay/dispatcher.ts @@ -27,6 +27,7 @@ import { LegacyRelayPublicationLedger, type LegacyPublicationLease } from './legacy-relay-publication-ledger' +import type { PtyConsumerCloseCause } from '../shared/pty-consumer-session-contract' export type { RelayClientSinkOptions, @@ -101,7 +102,9 @@ export class RelayDispatcher { private readonly requestAborts = new ClientRequestAborts() private readonly publicationLedger = new LegacyRelayPublicationLedger() private pendingRelayRequests = new Map() - private clientDetachListeners = new Set<(clientId: number) => void>() + private clientDetachListeners = new Set< + (clientId: number, cause: PtyConsumerCloseCause) => void + >() private disposeListeners = new Set<() => void>() private legacyCapacityListeners = new Set<() => void>() private clientCapacityListeners = new Map void>>() @@ -137,8 +140,13 @@ export class RelayDispatcher { } // Why: mark in-flight requests stale on disconnect so a late pty.spawn/fs.watch can't create unowned remote state. - invalidateClient(): void { - this.closeClient(this.primaryClient, new Error('Relay primary client invalidated'), false) + invalidateClient(cause: PtyConsumerCloseCause = 'local'): void { + this.closeClient( + this.primaryClient, + new Error('Relay primary client invalidated'), + false, + cause + ) } // Why: seq numbers and request ids are per SSH channel, so each attached client needs independent protocol state. @@ -153,12 +161,12 @@ export class RelayDispatcher { return client.id } - detachClient(clientId: number): void { + detachClient(clientId: number, cause: PtyConsumerCloseCause = 'local'): void { const client = this.clients.get(clientId) if (!client || client === this.primaryClient) { return } - this.closeClient(client, new Error('Relay client detached'), true) + this.closeClient(client, new Error('Relay client detached'), true, cause) } // Why: a displaced owner must lose its transport whichever client holds it, and the launch channel is @@ -187,7 +195,7 @@ export class RelayDispatcher { this.notificationHandlers.set(method, handler) } - onClientDetached(listener: (clientId: number) => void): () => void { + onClientDetached(listener: (clientId: number, cause: PtyConsumerCloseCause) => void): () => void { this.clientDetachListeners.add(listener) return () => this.clientDetachListeners.delete(listener) } @@ -1269,7 +1277,14 @@ export class RelayDispatcher { } } - private closeClient(client: RelayClient, error: Error, remove: boolean): void { + private closeClient( + client: RelayClient, + error: Error, + remove: boolean, + // Why the default is the cautious one: most closes here are the relay's own doing, and only the + // callers holding real evidence of a peer-side transport end may say so. + cause: PtyConsumerCloseCause = 'local' + ): void { if (client.closed) { return } @@ -1280,7 +1295,7 @@ export class RelayDispatcher { if (remove) { this.clients.delete(client.id) } - this.notifyClientDetached(client.id) + this.notifyClientDetached(client.id, cause) if (remove) { // Only for a client that is gone for good: an invalidated primary is revived by setWrite, and a // frame stranded by its retired sink must stay armed to retry. After the detach fan-out, so a @@ -1329,10 +1344,10 @@ export class RelayDispatcher { } } - private notifyClientDetached(clientId: number): void { + private notifyClientDetached(clientId: number, cause: PtyConsumerCloseCause): void { for (const listener of this.clientDetachListeners) { try { - listener(clientId) + listener(clientId, cause) } catch (err) { process.stderr.write( `[relay] Client detach listener failed: ${err instanceof Error ? err.message : String(err)}\n` diff --git a/src/relay/relay.ts b/src/relay/relay.ts index 4169c4fc4..1da1dd70b 100644 --- a/src/relay/relay.ts +++ b/src/relay/relay.ts @@ -1008,7 +1008,9 @@ async function main(): Promise { const clientId = socketClients.get(sock) socketClients.delete(sock) if (clientId !== undefined) { - dispatcher.detachClient(clientId) + // Why 'peer-closed' only here: the socket itself ended, which is the one signal that + // actually says the client is gone rather than merely slow. + dispatcher.detachClient(clientId, 'peer-closed') } relayLogLine(`[relay] Socket client closed (clients=${socketClients.size})`) if (!stdoutAlive && socketClients.size === 0) { @@ -1159,7 +1161,7 @@ async function main(): Promise { process.stdout.on('error', () => { stdoutAlive = false flushStdoutDrainWaiters() - dispatcher.invalidateClient() + dispatcher.invalidateClient('peer-closed') }) function startGrace(reason: string, options?: { retryDeferredShutdown?: boolean }): void { @@ -1214,7 +1216,7 @@ async function main(): Promise { // Why: stdin close means the SSH channel is gone; mark stdout dead so its write callback no-ops instead of hitting a dead pipe. stdoutAlive = false flushStdoutDrainWaiters() - dispatcher.invalidateClient() + dispatcher.invalidateClient('peer-closed') if (socketClients.size === 0) { startGrace('stdin ended') } @@ -1223,7 +1225,7 @@ async function main(): Promise { process.stdin.on('error', () => { stdoutAlive = false flushStdoutDrainWaiters() - dispatcher.invalidateClient() + dispatcher.invalidateClient('peer-closed') if (socketClients.size === 0) { startGrace('stdin error') } diff --git a/src/relay/ssh-pty-consumer-session-adapter.test.ts b/src/relay/ssh-pty-consumer-session-adapter.test.ts index a0f040814..7c38799b8 100644 --- a/src/relay/ssh-pty-consumer-session-adapter.test.ts +++ b/src/relay/ssh-pty-consumer-session-adapter.test.ts @@ -1,5 +1,10 @@ import { afterEach, describe, expect, it, vi } from 'vitest' -import { PTY_CONSUMER_OWNER_GRACE_MS } from '../shared/pty-consumer-session' +import { + PTY_CONSUMER_OWNER_GRACE_MS, + PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR, + PTY_CONSUMER_OWNER_HELD_GRACE_FLOOR_MS, + PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR +} from '../shared/pty-consumer-session' import { RelayDispatcher, type RelayClientSessionIdentity } from './dispatcher' import { encodeJsonRpcFrame, MessageType } from './protocol' import { SshPtyConsumerSessionAdapter } from './ssh-pty-consumer-session-adapter' @@ -29,10 +34,18 @@ function openFrame(id: number, overrides: Record = {}): Buffer ) } -function responseResult(buffer: Buffer): Record { +function responsePayload(buffer: Buffer): Record { expect(buffer[0]).toBe(MessageType.Regular) const length = buffer.readUInt32BE(9) - return JSON.parse(buffer.subarray(13, 13 + length).toString('utf8')).result + return JSON.parse(buffer.subarray(13, 13 + length).toString('utf8')) +} + +function responseResult(buffer: Buffer): Record { + return responsePayload(buffer).result as Record +} + +function responseError(buffer: Buffer): Record { + return responsePayload(buffer).error as Record } async function flushRequests(): Promise { @@ -80,9 +93,14 @@ describe('SshPtyConsumerSessionAdapter', () => { expect(responseResult(firstWrites[0])).toMatchObject({ role: 'session-owner', - ownerGeneration: 1 + ownerGeneration: 1, + resumed: false + }) + // Why a coded refusal reaches the wire: the dispatcher transports code and message only, so the + // competitor has to be able to tell "still settling" from "held" without parsing prose. + expect(responseError(secondWrites[0])).toMatchObject({ + code: PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR }) - expect(responseResult(secondWrites[0])).toMatchObject({ role: 'subscriber' }) firstSettlements[0]({ ok: true }) }) @@ -118,6 +136,79 @@ describe('SshPtyConsumerSessionAdapter', () => { }) }) + it.each([['peer-closed'], ['local']] as const)( + 'shortens a refused owner grace only for a %s detach', + async (cause) => { + // Why only Date: flushRequests rides setImmediate, which fake timers would otherwise capture. + vi.useFakeTimers({ toFake: ['Date'] }) + dispatcher = new RelayDispatcher( + (_data, onSettled) => { + onSettled({ ok: true }) + return true + }, + { supportsWriteCallback: true }, + endpointIdentity + ) + new SshPtyConsumerSessionAdapter(dispatcher, 'build-a') + + const ownerWrites: Buffer[] = [] + const ownerId = dispatcher.attachClient( + (data, onSettled) => { + ownerWrites.push(Buffer.from(data)) + onSettled({ ok: true }) + return true + }, + { supportsWriteCallback: true }, + endpointIdentity + ) + dispatcher.feedClient(ownerId, openFrame(1)) + await flushRequests() + expect(responseResult(ownerWrites[0])).toMatchObject({ role: 'session-owner' }) + + dispatcher.detachClient(ownerId, cause) + + const rivalWrites: Buffer[] = [] + const rivalId = dispatcher.attachClient( + (data, onSettled) => { + rivalWrites.push(Buffer.from(data)) + onSettled({ ok: true }) + return true + }, + { supportsWriteCallback: true }, + { ...endpointIdentity, principal: 'competitor' } + ) + // The first refusal is what applies the floor, so it has to happen before the clock moves. + dispatcher.feedClient(rivalId, openFrame(2)) + await flushRequests() + expect(responseError(rivalWrites[0])).toMatchObject({ + code: PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR + }) + + vi.setSystemTime(Date.now() + PTY_CONSUMER_OWNER_HELD_GRACE_FLOOR_MS + 1) + const retryId = dispatcher.attachClient( + (data, onSettled) => { + rivalWrites.push(Buffer.from(data)) + onSettled({ ok: true }) + return true + }, + { supportsWriteCallback: true }, + { ...endpointIdentity, principal: 'competitor' } + ) + dispatcher.feedClient(retryId, openFrame(3)) + await flushRequests() + + // Why this pair is the whole point of the cause: a socket that ended is evidence the owner is + // gone, and a queue the relay overran is not. Only the first may cost the incumbent its claim. + if (cause === 'peer-closed') { + expect(responseResult(rivalWrites[1])).toMatchObject({ role: 'session-owner' }) + } else { + expect(responseError(rivalWrites[1])).toMatchObject({ + code: PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR + }) + } + } + ) + it('rejects an unproved constructor stream as an owner principal', async () => { const writes: Buffer[] = [] dispatcher = new RelayDispatcher((data, onSettled) => { diff --git a/src/relay/ssh-pty-consumer-session-adapter.ts b/src/relay/ssh-pty-consumer-session-adapter.ts index b28082aa7..1b3bf7bad 100644 --- a/src/relay/ssh-pty-consumer-session-adapter.ts +++ b/src/relay/ssh-pty-consumer-session-adapter.ts @@ -95,12 +95,12 @@ export class SshPtyConsumerSessionAdapter { dispatcher.onRequest(SSH_PTY_OPEN_CLIENT_METHOD, (params, context) => this.openClient(params, context) ) - dispatcher.onClientDetached((clientId) => { + dispatcher.onClientDetached((clientId, cause) => { const grant = this.session.activeGrant(String(clientId)) if (grant) { this.clearPausedForGrant(grant) } - this.session.close(String(clientId)) + this.session.close(String(clientId), cause) if (grant) { this.sourceCredit.retainOrCloseOnDetach(grant) } diff --git a/src/shared/pty-consumer-owner-recovery.ts b/src/shared/pty-consumer-owner-recovery.ts index dfac94d7e..a71d73636 100644 --- a/src/shared/pty-consumer-owner-recovery.ts +++ b/src/shared/pty-consumer-owner-recovery.ts @@ -19,6 +19,37 @@ function throwRecoveryError(message: string, code: number): never { throw Object.assign(new Error(message), { code }) } +// Why identity excludes ownerGeneration: the generation fences the data path, but a reconnecting +// owner legitimately arrives holding whatever generation it last persisted. Matching on the logical +// triple is what lets one claim survive reconnects. +export function matchesPtyConsumerOwnerClaim( + hello: PtyConsumerSessionHello, + authentication: PtyConsumerAuthentication, + current: IncumbentOwner +): boolean { + const resume = hello.resume + return ( + resume !== undefined && + resume.ownerLease === current.lease && + hello.clientInstanceId === current.clientInstanceId && + authentication.principal === current.principal + ) +} + +// Why identity without the lease, next to the claim match above: a client that lost its recovery +// record still knows its own instance id, and against an incumbent carrying that id the incumbent is +// its own earlier connection. Enough to call a refusal transient; never enough to hand over a claim. +export function isPtyConsumerOwnerSameClient( + hello: PtyConsumerSessionHello, + authentication: PtyConsumerAuthentication, + current: IncumbentOwner +): boolean { + return ( + hello.clientInstanceId === current.clientInstanceId && + authentication.principal === current.principal + ) +} + export function assertPtyConsumerOwnerRecovery( hello: PtyConsumerSessionHello, authentication: PtyConsumerAuthentication, @@ -28,11 +59,7 @@ export function assertPtyConsumerOwnerRecovery( if (!resume) { throw new Error('Owner recovery proof is required') } - if ( - resume.ownerLease !== current.lease || - hello.clientInstanceId !== current.clientInstanceId || - authentication.principal !== current.principal - ) { + if (!matchesPtyConsumerOwnerClaim(hello, authentication, current)) { throwRecoveryError( 'Owner recovery lease is stale or belongs to another principal', PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR diff --git a/src/shared/pty-consumer-session-capabilities.ts b/src/shared/pty-consumer-session-capabilities.ts new file mode 100644 index 000000000..89962f6b3 --- /dev/null +++ b/src/shared/pty-consumer-session-capabilities.ts @@ -0,0 +1,45 @@ +import type { + PtyConsumerSessionGrant, + PtyConsumerSessionHello, + PtyConsumerSessionOptions +} from './pty-consumer-session-contract' +import { assertNonEmptyString, MAX_CAPABILITY_VERSIONS } from './pty-consumer-session-hello' + +export function assertPtyConsumerSessionOptions(options: PtyConsumerSessionOptions): void { + assertNonEmptyString(options.serverBuildId, 'serverBuildId') + if ( + options.outputFlowControl && + (!Number.isSafeInteger(options.outputFlowControl.maxWindowSu) || + options.outputFlowControl.maxWindowSu <= 0 || + options.outputFlowControl.versions.length > MAX_CAPABILITY_VERSIONS || + options.outputFlowControl.versions.some( + (version) => !Number.isSafeInteger(version) || version <= 0 + )) + ) { + throw new Error('outputFlowControl support is invalid') + } + if ( + options.ownerGraceMs !== undefined && + (!Number.isSafeInteger(options.ownerGraceMs) || options.ownerGraceMs < 0) + ) { + throw new Error('ownerGraceMs must be a non-negative safe integer') + } +} + +export function intersectPtyConsumerCapabilities( + hello: PtyConsumerSessionHello, + support: PtyConsumerSessionOptions['outputFlowControl'] +): Pick { + const offer = hello.capabilities?.outputFlowControl + if (!offer || !support || !offer.versions.includes(1) || !support.versions.includes(1)) { + return {} + } + return { + capabilities: { + outputFlowControl: { + version: 1, + windowSu: Math.min(offer.requestedWindowSu, support.maxWindowSu) + } + } + } +} diff --git a/src/shared/pty-consumer-session-contract.ts b/src/shared/pty-consumer-session-contract.ts index 1e359a11f..49527f839 100644 --- a/src/shared/pty-consumer-session-contract.ts +++ b/src/shared/pty-consumer-session-contract.ts @@ -5,6 +5,23 @@ export const PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR = -32041 // window bounded by one response write, so the client may retry within a short budget. export const PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR = -32042 export const PTY_CONSUMER_OWNER_RECOVERY_SUPERSEDED_ERROR = -32043 +// Why two codes, not one message: the dispatcher transports only code and message, and the two +// holders need opposite client behavior — an attached incumbent blocks, a disconnected one is transient. +export const PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR = -32044 +export const PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR = -32045 +// Why a third code: only `SshRelaySession` requests session-owner and every endpoint-credential socket +// shares one principal, so an attached incumbent carrying the requester's own clientInstanceId is that +// client's own half-open connection the relay never saw close — transient, not another client's claim. +export const PTY_CONSUMER_OWNER_HELD_SELF_ERROR = -32046 +// Why: a disconnected incumbent keeps at most this much of its remaining grace once a different +// owner-capable client asks, so admission converges inside one bounded retry instead of the full grace. +export const PTY_CONSUMER_OWNER_HELD_GRACE_FLOOR_MS = 250 + +// Why the grace floor needs this: shortening a grace is only safe against an owner the relay has +// evidence is gone, and that evidence exists only where the transport ended on the peer's side. A +// teardown the relay itself initiated — backpressure, a decode fault — proves nothing about liveness, +// so 'local' is the default and never shortens anything. +export type PtyConsumerCloseCause = 'peer-closed' | 'local' export type PtyConsumerRole = 'session-owner' | 'subscriber' @@ -30,6 +47,9 @@ export type PtyConsumerSessionGrant = { role: PtyConsumerRole ownerGeneration?: number ownerLease?: string + // Why: always present on a 'session-owner' grant, absent on a subscriber grant. `false` means the + // relay minted a fresh claim, so the client's checkpoints for the previous claim no longer apply. + resumed?: boolean capabilities?: { outputFlowControl?: { version: 1 diff --git a/src/shared/pty-consumer-session-hello.ts b/src/shared/pty-consumer-session-hello.ts index c8119da88..9dc54adf3 100644 --- a/src/shared/pty-consumer-session-hello.ts +++ b/src/shared/pty-consumer-session-hello.ts @@ -34,19 +34,3 @@ export function validateHello(hello: PtyConsumerSessionHello): void { throw new Error('outputFlowControl.requestedWindowSu must be a positive safe integer') } } - -// Why: a duplicate open on one connection is only idempotent if it asks for exactly the same thing. -export function helloFingerprint(hello: PtyConsumerSessionHello): string { - const flow = hello.capabilities?.outputFlowControl - return JSON.stringify({ - clientInstanceId: hello.clientInstanceId, - requestedRole: hello.requestedRole, - resume: hello.resume, - outputFlowControl: flow - ? { - versions: [...flow.versions].sort((a, b) => a - b), - requestedWindowSu: flow.requestedWindowSu - } - : undefined - }) -} diff --git a/src/shared/pty-consumer-session.test.ts b/src/shared/pty-consumer-session.test.ts index ce6282e86..c53991a0f 100644 --- a/src/shared/pty-consumer-session.test.ts +++ b/src/shared/pty-consumer-session.test.ts @@ -1,8 +1,11 @@ import { describe, expect, it } from 'vitest' import { + PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR, + PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR, + PTY_CONSUMER_OWNER_HELD_GRACE_FLOOR_MS, + PTY_CONSUMER_OWNER_HELD_SELF_ERROR, PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR, PTY_CONSUMER_OWNER_RECOVERY_SUPERSEDED_ERROR, - PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR, PtyConsumerSession, type PtyConsumerAuthentication, type PtyConsumerSessionHello @@ -40,38 +43,51 @@ function createSession(options: { now?: () => number } = {}): PtyConsumerSession } describe('PtyConsumerSession', () => { - it('types a resume against a fresh relay without weakening other owner refusals', () => { + it('grants a fresh claim when the relay no longer holds the resumed record', () => { const session = createSession() - expect(() => - session.admit( - ownerHello({ resume: { ownerGeneration: 1, ownerLease: 'stale' } }), - auth('connection-1') - ) - ).toThrow( - expect.objectContaining({ - code: PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR, - message: expect.stringContaining('stale') - }) + const admission = session.admit( + ownerHello({ resume: { ownerGeneration: 1, ownerLease: 'forgotten' } }), + auth('connection-1') ) + + // Why one round trip: the client named a record this relay does not have, which is a fresh claim, + // not a refusal — `resumed: false` is what tells it the old checkpoints no longer apply. + expect(admission.grant).toMatchObject({ + role: 'session-owner', + ownerGeneration: 1, + ownerLease: 'lease-1', + resumed: false + }) }) it('activates an authenticated owner only after its publication fence', () => { const session = createSession() const first = session.admit(ownerHello(), auth('connection-1')) - const competitor = session.admit( - ownerHello({ clientInstanceId: 'client-b' }), - auth('connection-2', { principal: 'other' }) - ) expect(first.grant).toMatchObject({ clientGeneration: 1, role: 'session-owner', ownerGeneration: 1, - ownerLease: 'lease-1' + ownerLease: 'lease-1', + resumed: false }) - expect(competitor.grant).toMatchObject({ clientGeneration: 2, role: 'subscriber' }) + // Why a coded refusal and not a subscriber grant: a subscriber grant is unusable to a client that + // asked to own the PTY, and it arrives shaped like success. + expect(() => + session.admit( + ownerHello({ clientInstanceId: 'client-b' }), + auth('connection-2', { principal: 'other' }) + ) + ).toThrow(expect.objectContaining({ code: PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR })) first.commitPublication() + + expect(() => + session.admit( + ownerHello({ clientInstanceId: 'client-b' }), + auth('connection-3', { principal: 'other' }) + ) + ).toThrow(expect.objectContaining({ code: PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR })) }) it('rolls back an unpublished owner without consuming authority', () => { @@ -86,14 +102,15 @@ describe('PtyConsumerSession', () => { }) }) - it('returns the same generation and lease for duplicate opens on one connection', () => { + it('rejects an identical duplicate open before it registers a second publication', () => { const session = createSession() const first = session.admit(ownerHello(), auth('connection-1')) - const duplicate = session.admit(ownerHello(), auth('connection-1')) - expect(duplicate.grant).toBe(first.grant) + // Why even an identical repeat: two responses settle independently, so one admission cannot make + // one response's rollback and the other's commit atomic. + expect(() => session.admit(ownerHello(), auth('connection-1'))).toThrow('only once') first.commitPublication() - duplicate.commitPublication() + expect(session.activeGrant('connection-1')).toBe(first.grant) }) it('rejects a second, different open on one connection', () => { @@ -339,48 +356,145 @@ describe('PtyConsumerSession', () => { expect(session.activeGrant('connection-2')).toBeNull() }) - it('types mismatched recovery without disturbing principal or lease ownership', () => { - const session = createSession() + it('separates a disconnected holder from the owner it belongs to', () => { + let now = 10 + const session = createSession({ now: () => now }) const first = session.admit(ownerHello(), auth('connection-1')) first.commitPublication() session.close('connection-1') - expect(() => - session.admit( - ownerHello({ resume: { ownerGeneration: 1, ownerLease: 'lease-1' } }), - auth('connection-2', { principal: 'stale-desktop' }) - ) - ).toThrow( - expect.objectContaining({ - code: PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR, - message: expect.stringContaining('principal') - }) - ) - expect(() => - session.admit( - ownerHello({ resume: { ownerGeneration: 1, ownerLease: 'wrong' } }), - auth('connection-3') - ) - ).toThrow(expect.objectContaining({ code: PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR })) - - const staleFresh = session.admit( - ownerHello(), - auth('connection-2', { principal: 'stale-desktop' }) - ) - expect(staleFresh.grant).toMatchObject({ role: 'subscriber' }) - expect(staleFresh.grant.ownerLease).toBeUndefined() + for (const [connectionId, hello] of [ + ['connection-2', ownerHello({ resume: { ownerGeneration: 1, ownerLease: 'lease-1' } })], + ['connection-3', ownerHello({ resume: { ownerGeneration: 1, ownerLease: 'wrong' } })], + ['connection-4', ownerHello()] + ] as const) { + expect(() => + session.admit(hello, auth(connectionId, { principal: 'other-desktop' })) + ).toThrow(expect.objectContaining({ code: PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR })) + } + // Why the incumbent still wins: a matching proof is routed as a replacement and never reaches the + // held-owner branch, so shortening the grace cannot cost the real owner its lease. const recovered = session.admit( ownerHello({ resume: { ownerGeneration: 1, ownerLease: 'lease-1' } }), - auth('connection-4') + auth('connection-5') ) expect(recovered.grant).toMatchObject({ role: 'session-owner', ownerGeneration: 2, - ownerLease: 'lease-1' + ownerLease: 'lease-1', + resumed: true }) }) + it('clamps a refused disconnected holder to the shared grace floor', () => { + let now = 10 + const session = createSession({ now: () => now }) + const first = session.admit(ownerHello(), auth('connection-1')) + first.commitPublication() + // Why 'peer-closed': the floor is only for an owner the relay watched leave. + session.close('connection-1', 'peer-closed') + + const rival = ownerHello({ clientInstanceId: 'client-b' }) + expect(() => session.admit(rival, auth('connection-2', { principal: 'other' }))).toThrow( + expect.objectContaining({ code: PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR }) + ) + now += PTY_CONSUMER_OWNER_HELD_GRACE_FLOOR_MS - 1 + expect(() => session.admit(rival, auth('connection-3', { principal: 'other' }))).toThrow( + expect.objectContaining({ code: PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR }) + ) + + now += 1 + const promoted = session.admit(rival, auth('connection-4', { principal: 'other' })) + + expect(promoted.grant).toMatchObject({ + role: 'session-owner', + ownerGeneration: 2, + ownerLease: 'lease-2', + resumed: false + }) + }) + + it('keeps the whole grace for an owner the relay tore down for backpressure', () => { + let now = 1_000 + const session = createSession({ now: () => now }) + const first = session.admit(ownerHello(), auth('connection-1')) + first.commitPublication() + // The relay destroyed this socket because its lane queue was full. That is the signature of an + // owner that is alive and slow, so the default 'local' cause must leave the grace untouched. + session.close('connection-1') + + const rival = ownerHello({ clientInstanceId: 'client-b' }) + now += 10 + expect(() => session.admit(rival, auth('connection-2'))).toThrow( + expect.objectContaining({ code: PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR }) + ) + // Why past the floor and still refused: no owner finishes notice-close, connect, handshake and + // openClient inside 250 ms, so a floor that applied here would hand the claim away every time. + now += PTY_CONSUMER_OWNER_HELD_GRACE_FLOOR_MS + 20 + expect(() => session.admit(rival, auth('connection-3'))).toThrow( + expect.objectContaining({ code: PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR }) + ) + + now += 5_000 + const recovered = session.admit( + ownerHello({ resume: { ownerGeneration: 1, ownerLease: 'lease-1' } }), + auth('connection-4') + ) + + // The point of the whole sequence: the live owner still gets back in. Losing here is permanent — + // the refusal it would have received routes as blocked and parks the target with no retry. + expect(recovered.grant).toMatchObject({ + role: 'session-owner', + ownerLease: 'lease-1', + resumed: true + }) + }) + + it("refuses a client's own attached connection as transient, not as another client's claim", () => { + const session = createSession() + const first = session.admit(ownerHello(), auth('connection-1')) + first.commitPublication() + + // The app's previous connection is a half-open zombie the relay never observed closing. Its + // re-open carries the same instance id and no proof, because the recovery record went with it. + expect(() => session.admit(ownerHello(), auth('connection-2'))).toThrow( + expect.objectContaining({ code: PTY_CONSUMER_OWNER_HELD_SELF_ERROR }) + ) + + // A genuinely different client is still blocked — this narrows the terminal case, it does not + // remove it. + expect(() => + session.admit( + ownerHello({ clientInstanceId: 'client-b' }), + auth('connection-3', { principal: 'other-desktop' }) + ) + ).toThrow(expect.objectContaining({ code: PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR })) + // Same instance id under a different principal is a different client too. + expect(() => + session.admit(ownerHello(), auth('connection-4', { principal: 'other-desktop' })) + ).toThrow(expect.objectContaining({ code: PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR })) + }) + + it('never converts an owner-capable request into a subscriber grant', () => { + const session = createSession() + const first = session.admit(ownerHello(), auth('connection-1')) + first.commitPublication() + + const ineligible = session.admit( + ownerHello(), + auth('connection-2', { allowSessionOwner: false }) + ) + + // Why this is the only subscriber outcome left: the request was not owner-capable in the first + // place, so no refusal code applies and the grant carries no `resumed`. + expect(ineligible.grant.role).toBe('subscriber') + expect(ineligible.grant).not.toHaveProperty('resumed') + expect(() => + session.admit(ownerHello({ clientInstanceId: 'client-b' }), auth('connection-3')) + ).toThrow(expect.objectContaining({ code: PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR })) + }) + it('elects a new owner after disconnected-owner grace expires', () => { let now = 10 const session = createSession({ now: () => now }) diff --git a/src/shared/pty-consumer-session.ts b/src/shared/pty-consumer-session.ts index 3baf502ac..31e6d4dce 100644 --- a/src/shared/pty-consumer-session.ts +++ b/src/shared/pty-consumer-session.ts @@ -1,27 +1,34 @@ import { randomUUID } from 'node:crypto' import { PTY_CONSUMER_OWNER_GRACE_MS, + PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR, + PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR, + PTY_CONSUMER_OWNER_HELD_GRACE_FLOOR_MS, + PTY_CONSUMER_OWNER_HELD_SELF_ERROR, + PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR, PTY_CONSUMER_SESSION_PROTOCOL_VERSION, - PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR, type PtyConsumerAuthentication, + type PtyConsumerCloseCause, type PtyConsumerDisplacedOwner, type PtyConsumerSessionAdmission, type PtyConsumerSessionGrant, type PtyConsumerSessionHello, type PtyConsumerSessionOptions } from './pty-consumer-session-contract' +import { assertNonEmptyString, validateHello } from './pty-consumer-session-hello' import { - assertNonEmptyString, - helloFingerprint, - MAX_CAPABILITY_VERSIONS, - validateHello -} from './pty-consumer-session-hello' -import { assertPtyConsumerOwnerRecovery } from './pty-consumer-owner-recovery' + assertPtyConsumerSessionOptions, + intersectPtyConsumerCapabilities +} from './pty-consumer-session-capabilities' +import { + assertPtyConsumerOwnerRecovery, + isPtyConsumerOwnerSameClient, + matchesPtyConsumerOwnerClaim +} from './pty-consumer-owner-recovery' export * from './pty-consumer-session-contract' type ClientRecord = { - fingerprint: string principal: string clientInstanceId: string grant: Readonly @@ -35,11 +42,17 @@ type OwnerRecord = { clientInstanceId: string generation: number lease: string + resumed: boolean state: 'pending' | 'active' | 'disconnected' disconnectedAt?: number + disconnectCause?: PtyConsumerCloseCause replaces?: OwnerRecord } +function throwOwnerError(message: string, code: number): never { + throw Object.assign(new Error(message), { code }) +} + export class PtyConsumerSession { private readonly clients = new Map() private readonly now: () => number @@ -50,24 +63,7 @@ export class PtyConsumerSession { private owner: OwnerRecord | null = null constructor(private readonly options: PtyConsumerSessionOptions) { - assertNonEmptyString(options.serverBuildId, 'serverBuildId') - if ( - options.outputFlowControl && - (!Number.isSafeInteger(options.outputFlowControl.maxWindowSu) || - options.outputFlowControl.maxWindowSu <= 0 || - options.outputFlowControl.versions.length > MAX_CAPABILITY_VERSIONS || - options.outputFlowControl.versions.some( - (version) => !Number.isSafeInteger(version) || version <= 0 - )) - ) { - throw new Error('outputFlowControl support is invalid') - } - if ( - options.ownerGraceMs !== undefined && - (!Number.isSafeInteger(options.ownerGraceMs) || options.ownerGraceMs < 0) - ) { - throw new Error('ownerGraceMs must be a non-negative safe integer') - } + assertPtyConsumerSessionOptions(options) this.now = options.now ?? Date.now this.createLease = options.createLease ?? randomUUID this.ownerGraceMs = options.ownerGraceMs ?? PTY_CONSUMER_OWNER_GRACE_MS @@ -85,16 +81,11 @@ export class PtyConsumerSession { } this.expireOwner() - const fingerprint = helloFingerprint(hello) - const duplicate = this.clients.get(authentication.connectionId) - if (duplicate) { - if ( - duplicate.fingerprint !== fingerprint || - duplicate.principal !== authentication.principal - ) { - throw new Error('pty.openClient may be used only once per transport connection') - } - return this.admissionFor(duplicate) + // Why even an identical repeat is rejected: the two responses settle their publications + // independently, so one shared admission cannot make one response's rollback and the other's + // commit atomic. A client recovering from an RPC timeout opens a new connection instead. + if (this.clients.has(authentication.connectionId)) { + throw new Error('pty.openClient may be used only once per transport connection') } const owner = this.selectOwner(hello, authentication) @@ -103,11 +94,12 @@ export class PtyConsumerSession { serverBuildId: this.options.serverBuildId, clientGeneration: this.nextClientGeneration++, role: owner ? ('session-owner' as const) : ('subscriber' as const), - ...(owner ? { ownerGeneration: owner.generation, ownerLease: owner.lease } : {}), - ...this.intersectCapabilities(hello) + ...(owner + ? { ownerGeneration: owner.generation, ownerLease: owner.lease, resumed: owner.resumed } + : {}), + ...intersectPtyConsumerCapabilities(hello, this.options.outputFlowControl) }) const client: ClientRecord = { - fingerprint, principal: authentication.principal, clientInstanceId: hello.clientInstanceId, grant, @@ -121,7 +113,9 @@ export class PtyConsumerSession { return this.admissionFor(client, this.displacedOwnerFor(owner)) } - close(connectionId: string): void { + // Why the cause defaults to 'local': it only ever widens the grace this record keeps, so a caller + // that cannot prove the peer's transport ended gets the answer that costs a live owner nothing. + close(connectionId: string, cause: PtyConsumerCloseCause = 'local'): void { const client = this.clients.get(connectionId) if (!client) { return @@ -136,7 +130,12 @@ export class PtyConsumerSession { ) { this.owner = { ...this.owner, - replaces: { ...this.owner.replaces, state: 'disconnected', disconnectedAt: this.now() } + replaces: { + ...this.owner.replaces, + state: 'disconnected', + disconnectedAt: this.now(), + disconnectCause: cause + } } } return @@ -148,7 +147,8 @@ export class PtyConsumerSession { this.owner = { ...this.owner, state: 'disconnected', - disconnectedAt: this.now() + disconnectedAt: this.now(), + disconnectCause: cause } } @@ -217,16 +217,14 @@ export class PtyConsumerSession { return null } const current = this.owner + // Why resume proof for a vacant record is not an error: the relay simply no longer has the record + // the client is naming. Minting a fresh claim here resolves it in one round trip, and `resumed: + // false` tells the client its checkpoints are void without making it delete its identity first. if (!current) { - if (hello.resume) { - throw Object.assign(new Error('Owner recovery lease is stale'), { - code: PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR - }) - } return this.newOwner(hello, authentication, null) } - if (!hello.resume) { - return null + if (!matchesPtyConsumerOwnerClaim(hello, authentication, current)) { + this.refuseHeldOwner(hello, authentication, current) } assertPtyConsumerOwnerRecovery(hello, authentication, current) // Why an active owner is displaced rather than refused: the resume proof matched this owner's @@ -236,6 +234,61 @@ export class PtyConsumerSession { return this.newOwner(hello, authentication, current) } + // Why an owner-capable request is refused rather than demoted: a subscriber grant is unusable to a + // client that needs to drive the PTY, and it arrives shaped like success. A coded refusal lets the + // caller retry the transient case and stop on the blocked one. + private refuseHeldOwner( + hello: PtyConsumerSessionHello, + authentication: PtyConsumerAuthentication, + current: OwnerRecord + ): never { + if (current.state === 'pending') { + throwOwnerError( + 'Owner grant publication is still pending', + PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR + ) + } + if (current.state === 'active') { + // Why identity without the lease: a client that lost its proof — a fresh process, a dropped + // recovery record — still knows who it is. Against an incumbent carrying its own instance id + // the honest answer is "your other connection is still registered", which resolves itself once + // the relay notices that socket. Blocking here strands the single-app case forever. + if (isPtyConsumerOwnerSameClient(hello, authentication, current)) { + throwOwnerError( + "PTY session owner is held by this client's own earlier connection", + PTY_CONSUMER_OWNER_HELD_SELF_ERROR + ) + } + throwOwnerError( + 'PTY session owner is held by an attached connection', + PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR + ) + } + this.clampDisconnectedOwnerGrace(current) + throwOwnerError( + 'PTY session owner is held by a disconnected connection within its grace period', + PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR + ) + } + + // Why only a peer-closed disconnect may shorten this: the floor is a bet that the incumbent is gone, + // and the relay tears a client's socket down for its own reasons too — a full lane queue is the + // signature of an owner that is alive but not draining fast enough. No owner completes a reconnect + // ladder in 250 ms, so clamping on a teardown we initiated hands a live owner's admission away and + // it can never get it back. Expiring a record never stops the remote PTY, but it does cost the user + // every route back to it. + private clampDisconnectedOwnerGrace(current: OwnerRecord): void { + if (current.disconnectCause !== 'peer-closed') { + return + } + const floorStart = + this.now() - Math.max(this.ownerGraceMs - PTY_CONSUMER_OWNER_HELD_GRACE_FLOOR_MS, 0) + if ((current.disconnectedAt ?? 0) <= floorStart) { + return + } + this.owner = { ...current, disconnectedAt: floorStart } + } + private displacedOwnerFor( owner: OwnerRecord | null ): Readonly | undefined { @@ -275,29 +328,12 @@ export class PtyConsumerSession { clientInstanceId: hello.clientInstanceId, generation: this.nextOwnerGeneration++, lease, + resumed: replaces !== null, state: 'pending', ...(replaces ? { replaces } : {}) } } - private intersectCapabilities( - hello: PtyConsumerSessionHello - ): Pick { - const offer = hello.capabilities?.outputFlowControl - const support = this.options.outputFlowControl - if (!offer || !support || !offer.versions.includes(1) || !support.versions.includes(1)) { - return {} - } - return { - capabilities: { - outputFlowControl: { - version: 1, - windowSu: Math.min(offer.requestedWindowSu, support.maxWindowSu) - } - } - } - } - private expireOwner(): void { if ( this.owner?.state === 'disconnected' &&