fix(relay): close a saturated PTY mirror instead of wedging every terminal

Shedding the frame left the mirror's retained leases above the relay-wide
low-water mark with nothing left to release them, so
`legacyRetentionBelowLowWater` never cleared, `PtyHandler.maybeResumePtyOutput`
early-returned and every paused PTY on the host stayed paused forever. Trading
#12041's disconnect for an unbounded global stall is strictly worse, so a mirror
that cannot admit a projection is closed again (main's behaviour) and recovers by
reconnecting. Owners still back-pressure — that is the actual #12041 fix.

Also name the control lane in the dropped-`pty.replay` diagnostic; it was
classified against the unrelated producer frame bound.

src/relay/dispatcher-slow-client.test.ts

@@ -82,12 +82,6 @@ function roles(
   return (clientId) => byId[clientId] ?? 'skip'
 }
 
-function queueFullDrops(stderr: { mock: { calls: unknown[][] } }): string[] {
-  return stderr.mock.calls
-    .map((call) => String(call[0]))
-    .filter((line) => /producer queue full/.test(line))
-}
-
 describe('RelayDispatcher slow-client PTY projection', () => {
   it('tryNotifyPtyData applies backpressure without closing the client', () => {
     const primary = makeStallingClient(65536)
@@ -132,7 +126,7 @@ describe('RelayDispatcher slow-client PTY projection', () => {
     }
   })
 
-  it('sheds a saturated mirror instead of pausing the PTY for every other viewer', () => {
+  it('closes a saturated mirror instead of pausing the PTY for every other viewer', () => {
     const owner = makeStallingClient(65536)
     owner.drain()
     const mirror = makeStallingClient(65536)
@@ -143,7 +137,6 @@ describe('RelayDispatcher slow-client PTY projection', () => {
       const mirrorId = dispatcher.attachClient(mirror.write, mirror.options)
       fillProducerQueue(dispatcher, mirrorId)
       const mirrorBefore = ptyDataFrameCount(mirror.frames)
-      stderr.mockClear()
 
       const projected = dispatcher.projectPtyDataToLegacyClients(
         roles({ [ownerId]: 'owner', [mirrorId]: 'mirror' }),
@@ -154,12 +147,35 @@ describe('RelayDispatcher slow-client PTY projection', () => {
       expect(projected).toBe(true)
       expect(ptyDataFrameCount(owner.frames)).toBe(1)
       expect(ptyDataFrameCount(mirror.frames)).toBe(mirrorBefore)
-      expect(mirror.closes).toBe(0)
+      expect(mirror.closes).toBe(1)
+      expect(owner.closes).toBe(0)
+    } finally {
+      stderr.mockRestore()
+      dispatcher.dispose()
+    }
+  })
+
+  it('releases the relay-wide legacy gate when a mirror can never drain', () => {
+    const owner = makeStallingClient(65536)
+    owner.drain()
+    const mirror = makeStallingClient(65536)
+    const dispatcher = new RelayDispatcher(owner.write, owner.options)
+    const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
+    try {
+      const ownerId = dispatcher.activeClientIds()[0]
+      const mirrorId = dispatcher.attachClient(mirror.write, mirror.options)
+      fillProducerQueue(dispatcher, mirrorId)
+      // Its retained leases sit above the relay-wide low-water mark, so every paused PTY is gated.
+      expect(dispatcher.legacyRetentionBelowLowWater).toBe(false)
+
+      const roleOf = roles({ [ownerId]: 'owner', [mirrorId]: 'mirror' })
+      dispatcher.projectPtyDataToLegacyClients(roleOf, { id: 'pty-1', data: 'y'.repeat(40_000) })
+
+      // PtyHandler.maybeResumePtyOutput reads this gate; if a shed mirror keeps it false with nothing
+      // left to evict it, every paused PTY on the host stays paused forever.
+      expect(mirror.closes).toBe(1)
+      expect(dispatcher.legacyRetentionBelowLowWater).toBe(true)
       expect(owner.closes).toBe(0)
-      expect(queueFullDrops(stderr)).toHaveLength(1)
-      expect(queueFullDrops(stderr)[0]).toMatch(
-        /^\[relay\] Dropped pty\.data \(\d+B, producer queue full/
-      )
     } finally {
       stderr.mockRestore()
       dispatcher.dispose()
@@ -295,14 +311,12 @@ describe('RelayDispatcher projection never reaps a client on a timer', () => {
 
   it('never closes a peer that is merely mid-write while another target back-pressures', () => {
     vi.useFakeTimers()
-    const owner = makeStallingClient(65536)
-    const neverDrains = makeStallingClient(65536)
+    const owner = makeStallingClient(65536, true)
     const dispatcher = new RelayDispatcher(owner.write, owner.options)
     const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
     let midWriteCloses = 0
     try {
       const ownerId = dispatcher.activeClientIds()[0]
-      const neverDrainsId = dispatcher.attachClient(neverDrains.write, neverDrains.options)
       const midWriteId = dispatcher.attachClient(
         (_data, settle: (result: SinkWriteSettlement) => void) => {
           // Accepted by the sink but not acknowledged yet: normal for any real socket mid-write.
@@ -319,12 +333,7 @@ describe('RelayDispatcher projection never reaps a client on a timer', () => {
         }
       )
       fillProducerQueue(dispatcher, ownerId)
-      fillProducerQueue(dispatcher, neverDrainsId)
-      const roleOf = roles({
-        [ownerId]: 'owner',
-        [neverDrainsId]: 'mirror',
-        [midWriteId]: 'mirror'
-      })
+      const roleOf = roles({ [ownerId]: 'owner', [midWriteId]: 'mirror' })
       const params = { id: 'pty-1', data: 'y'.repeat(40_000) }
 
       expect(dispatcher.projectPtyDataToLegacyClients(roleOf, params)).toBe(false)
@@ -335,11 +344,15 @@ describe('RelayDispatcher projection never reaps a client on a timer', () => {
       expect(dispatcher.projectPtyDataToLegacyClients(roleOf, params)).toBe(false)
       vi.advanceTimersByTime(10 * 60_000)
 
-      // Reaping a peer on a timer is exactly the disconnect #12041 reported.
+      // A peer with a frame merely in flight still has queue room; reaping it on a timer is exactly
+      // the disconnect #12041 reported.
       expect(midWriteCloses).toBe(0)
-      expect(neverDrains.closes).toBe(0)
       expect(owner.closes).toBe(0)
-      expect(dispatcher.activeClientIds()).toHaveLength(3)
+      expect(dispatcher.activeClientIds()).toHaveLength(2)
+
+      owner.drain()
+      expect(dispatcher.projectPtyDataToLegacyClients(roleOf, params)).toBe(true)
+      expect(midWriteCloses).toBe(0)
     } finally {
       stderr.mockRestore()
       dispatcher.dispose()
@@ -378,7 +391,10 @@ describe('RelayDispatcher pty.replay under a full control lane', () => {
       dispatcher.notify('pty.replay', { id: 'pty-1', data: 'x' })
       expect(primary.closes).toBe(0)
       expect(stderr).toHaveBeenCalledTimes(1)
-      expect(String(stderr.mock.calls[0][0])).toMatch(/^\[relay\] Dropped pty\.replay \(/)
+      // The control lane rejected it; naming the producer bound would point at the wrong limit.
+      expect(String(stderr.mock.calls[0][0])).toMatch(
+        /^\[relay\] Dropped pty\.replay \(\d+B, control queue full\)\n$/
+      )
 
       dispatcher.notify('pty.replay', { id: 'pty-2', data: 'x' })
       expect(primary.closes).toBe(0)

src/relay/dispatcher.ts

@@ -57,9 +57,11 @@ export type RelayClientSourceOptions = {
 /**
  * How the legacy PTY projection serves a client whose producer queue is full.
  * - `skip`: not a projection target (a flow-controlled source owner has its own credited lane).
- * - `owner`: the projection is this client's only delivery lane, so a full queue back-pressures the PTY.
- * - `mirror`: a bystander copy that recovers by reopening the pty, so a full queue sheds this one frame for
- *   that client alone — a mirror must never hold the native PTY paused for every other viewer.
+ * - `owner`: the projection is this client's only delivery lane, so a full queue back-pressures the PTY
+ *   until it drains — the #12041 disconnect was closing this client instead.
+ * - `mirror`: a bystander copy that recovers by reconnecting, so a full queue closes just that client.
+ *   Shedding the frame instead would leave its retained bytes gating `legacyRetentionBelowLowWater`
+ *   with nothing left to evict it, wedging every paused PTY on the host forever.
  */
 export type LegacyProjectionRole = 'skip' | 'owner' | 'mirror'
 
@@ -538,7 +540,7 @@ export class RelayDispatcher {
           // Why: reattach regenerates replay, so killing the link over it just re-kills the reconnect;
           // drop the pane's scrollback instead and let live output repaint it.
           if (!this.enqueueFrame(client, msg, 'control', undefined, frameBytes, 'reject')) {
-            this.logDroppedProducerNotification(client, method, frameBytes)
+            this.logDroppedProducerNotification(client, method, frameBytes, 'control')
           }
           continue
         }
@@ -586,10 +588,17 @@ export class RelayDispatcher {
 
   // Why: one line per generation, method and drop reason — a flooding producer retries every batch and would
   // spam stderr, but a transient queue-full drop must not consume the slot a real over-capacity drop needs.
-  private logDroppedProducerNotification(client: RelayClient, method: string, bytes: number): void {
+  private logDroppedProducerNotification(
+    client: RelayClient,
+    method: string,
+    bytes: number,
+    // Why: a control-lane drop has nothing to do with the producer bound; naming that lane sends the
+    // reader to the wrong limit.
+    lane: 'producer' | 'control' = 'producer'
+  ): void {
     const capacity = client.writer.producerFrameCapacity
-    const overCapacity = bytes > capacity
-    const key = `${method}:${overCapacity ? 'over-capacity' : 'queue-full'}`
+    const overCapacity = lane === 'producer' && bytes > capacity
+    const key = `${method}:${lane === 'control' ? 'control-full' : overCapacity ? 'over-capacity' : 'queue-full'}`
     let log = client.droppedNotificationLog
     if (!log || log.generation !== client.generation) {
       log = { generation: client.generation, loggedKeys: new Set() }
@@ -600,9 +609,11 @@ export class RelayDispatcher {
     }
     log.loggedKeys.add(key)
     process.stderr.write(
-      overCapacity
-        ? `[relay] Dropped ${method} (${bytes}B > producer frame capacity ${capacity}B)\n`
-        : `[relay] Dropped ${method} (${bytes}B, producer queue full; frame capacity ${capacity}B)\n`
+      lane === 'control'
+        ? `[relay] Dropped ${method} (${bytes}B, control queue full)\n`
+        : overCapacity
+          ? `[relay] Dropped ${method} (${bytes}B > producer frame capacity ${capacity}B)\n`
+          : `[relay] Dropped ${method} (${bytes}B, producer queue full; frame capacity ${capacity}B)\n`
     )
   }
 
@@ -1166,14 +1177,20 @@ export class RelayDispatcher {
       if (!this.tryPublishToClients(blocking, msg, lane, bytes)) {
         return false
       }
-      // Why: a mirror is a bystander view that recovers on its own (reopening the pty replays it), so it
-      // must never hold the native PTY paused for every other viewer. Publishing only after the owners
-      // committed also keeps a back-pressured span from reaching a mirror twice when it is retried.
+      // Why: a mirror must not hold the native PTY paused for every other viewer, but shedding its frame
+      // would strand its retained bytes above the relay-wide low-water mark with nothing left to release
+      // them, so every paused PTY would stay paused. Closing bounds that, and the client reconnects.
+      // Publishing only after the owners committed also keeps a back-pressured span from reaching a
+      // mirror twice when it is retried.
       for (const client of mirrors) {
         if (client.closed || this.publishToClient(client, msg, lane, undefined, bytes)) {
           continue
         }
-        this.logDroppedProducerNotification(client, msg.method, bytes)
+        this.closeClient(
+          client,
+          new Error('Relay PTY subscriber projection capacity exceeded'),
+          client !== this.primaryClient
+        )
       }
       return !this.disposed
     })

src/relay/pty-handler-source-publication.test.ts

@@ -709,7 +709,7 @@ describe('PtyHandler negotiated source publication', () => {
     expect(ordered.indexOf('pty.data')).toBeLessThan(ordered.lastIndexOf('pty.exit'))
   })
 
-  it('keeps the native PTY, the V1 owner and a saturated subscriber all live', async () => {
+  it('keeps the native PTY and V1 owner live when one subscriber saturates', async () => {
     await spawn({})
     const detached: number[] = []
     const healthyWrites: Buffer[] = []
@@ -741,8 +741,9 @@ describe('PtyHandler negotiated source publication', () => {
 
     expect(admitted).toBeGreaterThan(100)
     expect(admitted).toBeLessThan(140)
-    // Neither closed (that is #12041) nor gating: a bystander subscriber sheds its own copy only.
-    expect(detached).toEqual([])
+    // A bystander subscriber is dropped rather than allowed to pause the PTY for every viewer or to
+    // pin the relay-wide legacy gate open.
+    expect(detached).toEqual([saturatedId])
     expect(detached).not.toContain(healthyId)
     expect(
       healthyWrites.map(notification).filter((frame) => frame?.method === 'pty.data')
@@ -752,7 +753,7 @@ describe('PtyHandler negotiated source publication', () => {
     expect(pausePty).not.toHaveBeenCalled()
     expect(
       stderr.mock.calls.filter((call) => /Dropped pty\.data/.test(String(call[0])))
-    ).toHaveLength(1)
+    ).toHaveLength(0)
     stderr.mockRestore()
   })
 })

src/relay/relay-pty-source-publication.test.ts

@@ -283,7 +283,7 @@ describe('RelayPtySourcePublication', () => {
     ).toHaveLength(2)
   })
 
-  it('sheds a saturated subscriber without detaching it or stalling the V1 owner', async () => {
+  it('detaches only a saturated subscriber while the V1 owner stays live', async () => {
     const harness = await createHarness(8)
     const detached: number[] = []
     const saturatedWrites: Buffer[] = []
@@ -324,10 +324,10 @@ describe('RelayPtySourcePublication', () => {
 
     expect(admitted).toBeGreaterThan(0)
     expect(admitted).toBeLessThan(20)
-    // A bystander subscriber is neither closed (#12041) nor allowed to pause the PTY for everyone.
+    // A bystander subscriber is dropped rather than allowed to pause the PTY for every viewer.
     expect(harness.publication.publish('pty-1', { data: saturatedPayload }, false)).toBe(true)
 
-    expect(detached).toEqual([])
+    expect(detached).toEqual([saturatedId])
     expect(detached).not.toContain(healthyId)
     expect(saturatedWrites).toHaveLength(1)
     expect(heldSettlements).toHaveLength(1)