fix(ssh): park one PTY's exhausted delivery recovery instead of dropping the channel

A per-PTY recovery budget running out disposed the whole relay channel, so one PTY that could not re-prove its delivery aborted every in-flight filesystem and git request on that host and stalled every sibling pane. A retry count is not proof of anything, and it certainly is not proof about the other sessions sharing the channel. Exhaustion now parks that PTY's delivery. The remote shell keeps running, its lease stands, and the next relay open reattaches it with a fresh delivery generation — the parked state is cleared on teardown and the generation changes on reconnect, so a reconnect recovers it. The consecutive-attempt ceiling goes away entirely; the per-generation one is what bounds the retry cost, and the second ceiling only existed to reach the channel drop sooner. Tradeoff worth stating: the failing pane used to self-heal within seconds because the forced reconnect wiped all rejection state, and it now stays frozen until the next relay open. That is a worse outcome for that one pane and a much better one for every other session on the host, and reconnecting is user-reachable. Co-authored-by: Orca <help@stably.ai>
2026-09-23 00:02:29 +00:00 · 2026-08-08 19:11:53 -07:00
parent 8480d8f26d
commit 07e382ec19
2 changed files with 85 additions and 48 deletions
@@ -54,6 +54,7 @@ type RejectedDeliverySession = {
  >
  rejectedPtyRecoveryAttempts: Map<string, unknown>
  sourceIdentityByRelayPtyId: Map<string, unknown>
+  watchMuxForRelayLoss: (mux: SshChannelMultiplexer) => void
  retireExitedPty: (payload: {
    id: string
    code: number
@@ -89,9 +90,18 @@ function rejectedPayload(overrides: Partial<SshPtyDataPayload> = {}): SshPtyData

 function prepareSession() {
  const deps = createMockDeps()
+  const disposeHandlers: ((reason: 'shutdown' | 'connection_lost') => void)[] = []
  const mux = {
    isDisposed: vi.fn(() => false),
-    dispose: vi.fn()
+    onDispose: vi.fn((handler: (reason: 'shutdown' | 'connection_lost') => void) => {
+      disposeHandlers.push(handler)
+      return () => {}
+    }),
+    dispose: vi.fn((reason: 'shutdown' | 'connection_lost' = 'shutdown') => {
+      for (const handler of disposeHandlers) {
+        handler(reason)
+      }
+    })
  } as unknown as SshChannelMultiplexer
  const session = new SshRelaySession(
    'target-1',
@@ -101,6 +111,11 @@ function prepareSession() {
  )
  const internals = session as unknown as RejectedDeliverySession
  internals.mux = mux
+  // Why wired: disposing this mux is what fires the host-wide relay-lost reconnect, so a test can
+  // observe the blast radius of a per-PTY escalation instead of only the local dispose call.
+  const relayLost = vi.fn()
+  session.setOnRelayLost(relayLost)
+  internals.watchMuxForRelayLoss(mux)
  internals.activePtyProviderGeneration = 23
  internals.ptyConsumerSessionState = {
    mode: 'negotiated',
@@ -110,7 +125,7 @@ function prepareSession() {
    ownerLease: 'owner-lease',
    outputFlowControl: { version: 1, windowSu: 64 }
  }
-  return { deps, internals, mux, session }
+  return { deps, internals, mux, relayLost, session }
 }

 describe('SshRelaySession rejected PTY delivery recovery', () => {
@@ -243,8 +258,10 @@ describe('SshRelaySession rejected PTY delivery recovery', () => {
    expect(mux.dispose).not.toHaveBeenCalled()
  })

+  // Why this one still drops the channel: the relay refused to confirm the delivery was canceled, so
+  // no per-PTY fence is available — it is an unprovable state, not an exhausted retry count.
  it('reconnects instead of canceling an unprovable malformed delivery', async () => {
-    const { internals, mux } = prepareSession()
+    const { internals, mux, relayLost } = prepareSession()
    const reattach = vi.fn().mockResolvedValue(true)
    internals.reattachRejectedPty = reattach
    vi.spyOn(console, 'warn').mockImplementation(() => {})
@@ -259,6 +276,7 @@ describe('SshRelaySession rejected PTY delivery recovery', () => {
    )

    expect(mux.dispose).toHaveBeenCalledWith('connection_lost')
+    expect(relayLost).toHaveBeenCalledWith('target-1')
    expect(reattach).not.toHaveBeenCalled()
  })

@@ -357,11 +375,11 @@ describe('SshRelaySession rejected PTY delivery recovery', () => {
    expect(internals.rejectedPtyRecoveryAttempts).toHaveLength(0)
  })

-  // Why a channel drop rather than a terminal relay error: a terminal error clears the reconnect
-  // backoff and rotates provider authority, aborting every fs and git request on the target, so one
-  // PTY's undeliverable output would strand the whole connection in manual recovery.
-  it('bounds failed targeted recovery and escalates to a recoverable relay reconnect', async () => {
-    const { internals, mux, session } = prepareSession()
+  // Why this is the oracle: an exhausted retry budget is not proof of anything, and the relay channel
+  // is shared — dropping it rotates provider authority, aborts every in-flight fs and git request on
+  // the target and stalls every sibling PTY over one PTY's undeliverable output.
+  it('parks one PTY after its recovery budget runs out and leaves the shared channel and a sibling alone', async () => {
+    const { internals, mux, relayLost, session } = prepareSession()
    const reattach = vi.fn().mockResolvedValue(false)
    internals.reattachRejectedPty = reattach
    getSshPtyProviderMock.mockReturnValue({ hasPty: () => true } as unknown as SshPtyProvider)
@@ -370,19 +388,41 @@ describe('SshRelaySession rejected PTY delivery recovery', () => {
    const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})

    await internals.acceptPtyData(rejectedPayload())
-    await vi.waitFor(() => expect(mux.dispose).toHaveBeenCalledOnce(), { timeout: 2000 })
-
-    expect(reattach).toHaveBeenCalledTimes(2)
-    expect(reattach.mock.calls).toEqual([
-      ['pty-bad', mux, 23, 'confirm-existing'],
-      ['pty-bad', mux, 23, 'confirm-existing']
-    ])
-    expect(mux.dispose).toHaveBeenCalledWith('connection_lost')
-    expect(warn).toHaveBeenCalledWith(
-      expect.stringContaining('PTY pty-bad delivery recovery exhausted')
+    await vi.waitFor(
+      () =>
+        expect(warn).toHaveBeenCalledWith(
+          expect.stringContaining('PTY pty-bad delivery recovery exhausted')
+        ),
+      { timeout: 10_000 }
    )
+    // Why the extra wait: a dispose would be scheduled off the same recovery pass as the warning.
+    await new Promise((resolve) => setTimeout(resolve, 400))
+
+    expect(mux.dispose).not.toHaveBeenCalled()
+    expect(relayLost).not.toHaveBeenCalled()
    expect(onTerminalError).not.toHaveBeenCalled()
+    expect(reattach).toHaveBeenCalledTimes(12)
    expect(acceptOutputDataMock).not.toHaveBeenCalled()
+
+    await internals.acceptPtyData(
+      rejectedPayload({
+        id: 'ssh:target-1@@pty-healthy',
+        data: 'healthy',
+        ptyIncarnation: 'incarnation-healthy',
+        source: source({
+          relayPtyId: 'pty-healthy',
+          spanId: 'token-healthy:0:7',
+          deliveryToken: 'token-healthy',
+          sourceEndSu: 7
+        }),
+        sourceRejected: undefined
+      })
+    )
+
+    expect(acceptOutputDataMock).toHaveBeenCalledOnce()
+    expect(acceptOutputDataMock).toHaveBeenCalledWith(
+      expect.objectContaining({ id: 'ssh:target-1@@pty-healthy', data: 'healthy' })
+    )
  })

  it('stops without an error when the rejected PTY exited during recovery', async () => {
@@ -452,18 +492,18 @@ describe('SshRelaySession rejected PTY delivery recovery', () => {
  })

  it('does not let accepted frames refill the recovery budget indefinitely', async () => {
-    const { internals, mux, session } = prepareSession()
+    const { internals, mux, relayLost, session } = prepareSession()
    const reattach = vi.fn().mockResolvedValue(true)
    internals.reattachRejectedPty = reattach
    getSshPtyProviderMock.mockReturnValue({ hasPty: () => true } as unknown as SshPtyProvider)
    const onTerminalError = vi.fn()
    session.setOnTerminalRelayError(onTerminalError)
+    const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})

    // Why alternating, with a fresh bad token each round: every rejection retires its own delivery,
    // so a flapping PTY only keeps asking for recovery by moving onto new ones, and the accepted
-    // frame in between is what used to clear the budget outright. Each reattach here succeeds, so
-    // the consecutive budget is cleared legitimately too — only the per-generation ceiling can stop
-    // it.
+    // frame in between is what used to clear the budget outright — only the per-generation ceiling
+    // can stop it.
    for (let round = 0; round < 40; round++) {
      await internals.acceptPtyData(
        rejectedPayload({
@@ -487,9 +527,17 @@ describe('SshRelaySession rejected PTY delivery recovery', () => {
      )
      await Promise.resolve()
    }
-    await vi.waitFor(() => expect(mux.dispose).toHaveBeenCalledOnce(), { timeout: 2000 })
+    await vi.waitFor(
+      () =>
+        expect(warn).toHaveBeenCalledWith(
+          expect.stringContaining('PTY pty-bad delivery recovery exhausted')
+        ),
+      { timeout: 2000 }
+    )

    expect(onTerminalError).not.toHaveBeenCalled()
+    expect(mux.dispose).not.toHaveBeenCalled()
+    expect(relayLost).not.toHaveBeenCalled()
    expect(acceptOutputDataMock).toHaveBeenCalledTimes(40)
    expect(reattach.mock.calls.length).toBeLessThanOrEqual(12)
  })
@@ -135,10 +135,9 @@ const SSH_PTY_REATTACH_MAX_CONCURRENCY = 8
 const SSH_PTY_REATTACH_ATTEMPT_TIMEOUT_MS = 10_000
 const SSH_PTY_REATTACH_RETRY_MIN_DELAY_MS = 50
 const SSH_PTY_REATTACH_RETRY_JITTER_MS = 200
-const SSH_REJECTED_PTY_RECOVERY_MAX_ATTEMPTS = 2
-// Why a second ceiling: the consecutive budget resets whenever a reattach succeeds, so a PTY that
-// alternates recovered and rejected frames would otherwise reattach forever — each one costs a
-// store read, an attach round trip and a store write.
+// Why a ceiling at all: a PTY that alternates recovered and rejected frames would otherwise reattach
+// forever — each one costs a store read, an attach round trip and a store write. Exhausting it parks
+// this PTY's delivery; it never authorizes tearing anything down.
 const SSH_REJECTED_PTY_RECOVERY_MAX_GENERATION_ATTEMPTS = 12
 const SSH_REJECTED_PTY_RECOVERY_RETRY_DELAY_MS = 150
 const SSH_SOURCE_RECOVERY_CANCELLATION_FAILED = 'ssh_source_recovery_cancellation_failed'
@@ -339,7 +338,6 @@ export class SshRelaySession {
    string,
    {
      providerGeneration: number
-      attempts: number
      generationAttempts: number
      reported: boolean
    }
@@ -1831,26 +1829,21 @@ export class SshRelaySession {
    const attempt =
      previous?.providerGeneration === providerGeneration
        ? previous
-        : { providerGeneration, attempts: 0, generationAttempts: 0, reported: false }
-    if (
-      attempt.attempts >= SSH_REJECTED_PTY_RECOVERY_MAX_ATTEMPTS ||
-      attempt.generationAttempts >= SSH_REJECTED_PTY_RECOVERY_MAX_GENERATION_ATTEMPTS
-    ) {
+        : { providerGeneration, generationAttempts: 0, reported: false }
+    if (attempt.generationAttempts >= SSH_REJECTED_PTY_RECOVERY_MAX_GENERATION_ATTEMPTS) {
      if (!attempt.reported) {
        attempt.reported = true
+        // Why park instead of dropping the relay channel: a retry count is not proof of anything, and
+        // the channel is shared — dropping it rotates provider authority, aborts every in-flight fs
+        // and git request on the target and stalls every sibling PTY over one PTY's delivery. The
+        // remote shell keeps running, its lease stands, and the next relay open reattaches it with a
+        // fresh delivery generation.
        console.warn(
-          `[ssh-relay-session] PTY ${relayPtyId} delivery recovery exhausted for ${this.targetId}; dropping the relay channel to reconnect`
+          `[ssh-relay-session] PTY ${relayPtyId} delivery recovery exhausted for ${this.targetId}; parking its delivery until the next relay open`
        )
-        // Why a channel drop and not a terminal relay error: a terminal error clears the reconnect
-        // backoff, rotates provider authority (aborting every in-flight fs and git request on the
-        // target) and parks the target in a manual-recovery state — over one PTY's delivery. Losing
-        // the channel is the recoverable escalation, and it is what this path did before targeted
-        // recovery existed.
-        mux.dispose('connection_lost')
      }
      return
    }
-    attempt.attempts++
    attempt.generationAttempts++
    this.rejectedPtyRecoveryAttempts.set(appPtyId, attempt)
    void this.rejectedPtyReattaches
@@ -1866,13 +1859,9 @@ export class SshRelaySession {
      )
      .then(
        (recovered) => {
-          if (recovered) {
-            // Why only a completed reattach clears this: an accepted frame proves nothing about the
-            // delivery that was rejected, and resetting on one lets a flapping PTY reattach forever.
-            attempt.attempts = 0
-            return
+          if (!recovered) {
+            this.retryRejectedPtyDelivery(payload, source, appPtyId)
          }
-          this.retryRejectedPtyDelivery(payload, source, appPtyId)
        },
        (error: unknown) => {
          console.warn(`[ssh-relay-session] PTY ${relayPtyId} targeted delivery recovery failed`, {
@@ -1886,7 +1875,7 @@ export class SshRelaySession {

  // Why liveness is checked before retrying: reattachKnownPty resolves without claiming the lease
  // when the PTY exited mid-attach, which is indistinguishable from a failed reattach at the call
-  // site. Retrying that race twice would drop the relay channel over an ordinary PTY exit.
+  // site, so an ordinary exit would otherwise burn the whole recovery budget on attach round trips.
  private retryRejectedPtyDelivery(
    payload: SshPtyDataPayload,
    source: SshPtyDataPayload['source'],