fix(relay): bound how long a stalled subscriber can hold every PTY paused

Review follow-up on the back-pressure change:

- A refused projection pauses the PTY dispatcher-wide, and nothing else on the
  daemon reaps a subscriber that never drains: the writer only closes on a write
  *error*, and a peer that is alive but not reading never raises one. Arm a
  bounded reap when a projection is refused — a non-primary sink that settles
  zero bytes for a full 30s window is detached (it reconnects and replays) so
  the other viewers' terminals resume. The primary link, the one #12041 is
  about, still back-pressures forever.
- `projectToClients` encoded the frame before filtering, so the common
  single-owner case (no legacy projection target at all) paid a full
  JSON.stringify plus two Buffer copies per span and threw them away.

Co-authored-by: Orca <help@stably.ai>

src/relay/dispatcher-client-writer.ts

@@ -33,6 +33,7 @@ export class DispatcherClientWriter {
   private readonly settleOnDrain = new Set<DispatcherWriterEntry>()
   private readonly capacityListeners = new Set<() => void>()
   private readonly idleWaiters = new Set<() => void>()
+  private settledBytes = 0
   private saturated = false
   private drainArmed = false
   private removeDrainListener: (() => void) | null = null
@@ -55,6 +56,14 @@ export class DispatcherClientWriter {
     return this.admission.retainedProducerBytes
   }
 
+  /**
+   * Monotonic bytes the sink has settled. Only real writes move it — a liveness entry replaced in the
+   * queue never reaches releaseEntry — so a flat counter across a window means the peer moved nothing.
+   */
+  get drainProgressBytes(): number {
+    return this.settledBytes
+  }
+
   get producerFrameCapacity(): number {
     return this.sink.producerFrameCapacity
   }
@@ -286,6 +295,7 @@ export class DispatcherClientWriter {
       return
     }
     entry.settled = true
+    this.settledBytes += entry.estimatedBytes
     this.inFlight.delete(entry)
     this.settleOnDrain.delete(entry)
     this.admission.release(entry)

src/relay/dispatcher-projection-stall.ts

@@ -0,0 +1,86 @@
+/**
+ * Bounds the cost of back-pressuring a refused PTY projection.
+ *
+ * A projection is all-or-nothing, so one subscriber that never drains holds every PTY paused for every
+ * viewer. Nothing else reaps such a sink: the writer only closes on a write *error*, and a peer that is
+ * alive but not reading never raises one. This detaches a subscriber that made zero drain progress for a
+ * full window — it reconnects and replays, while the sinks that are merely slow keep their back-pressure.
+ */
+export const PROJECTION_STALL_EVICTION_MS = 30_000
+
+export type ProjectionStallProbe = {
+  // Retaining producer bytes: the sink has accepted nothing back since they were queued.
+  stalled: boolean
+  // Monotonic settled bytes; unchanged across a window means the sink moved nothing at all.
+  drainProgressBytes: number
+}
+
+export class ProjectionStallEvictor<T> {
+  private readonly armed = new Map<T, number>()
+  private timer: ReturnType<typeof setTimeout> | null = null
+
+  constructor(
+    private readonly probe: (target: T) => ProjectionStallProbe,
+    private readonly evict: (target: T) => void,
+    private readonly evictionMs: number = PROJECTION_STALL_EVICTION_MS
+  ) {}
+
+  noteRefusedProjection(targets: readonly T[]): void {
+    for (const target of targets) {
+      const probe = this.probe(target)
+      if (!probe.stalled) {
+        this.armed.delete(target)
+        continue
+      }
+      if (!this.armed.has(target)) {
+        this.armed.set(target, probe.drainProgressBytes)
+      }
+    }
+    this.schedule()
+  }
+
+  forget(target: T): void {
+    this.armed.delete(target)
+  }
+
+  dispose(): void {
+    this.armed.clear()
+    this.clearTimer()
+  }
+
+  private schedule(): void {
+    if (this.timer || this.armed.size === 0) {
+      return
+    }
+    this.timer = setTimeout(() => {
+      this.timer = null
+      this.sweep()
+    }, this.evictionMs)
+    this.timer.unref?.()
+  }
+
+  private clearTimer(): void {
+    if (!this.timer) {
+      return
+    }
+    clearTimeout(this.timer)
+    this.timer = null
+  }
+
+  private sweep(): void {
+    for (const [target, progressBytes] of Array.from(this.armed)) {
+      const probe = this.probe(target)
+      if (!probe.stalled) {
+        this.armed.delete(target)
+        continue
+      }
+      if (probe.drainProgressBytes !== progressBytes) {
+        this.armed.set(target, probe.drainProgressBytes)
+        continue
+      }
+      this.armed.delete(target)
+      this.evict(target)
+    }
+    this.schedule()
+  }
+}

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

@@ -1,5 +1,6 @@
-import { describe, expect, it, vi } from 'vitest'
+import { afterEach, describe, expect, it, vi } from 'vitest'
 import { RelayDispatcher, type RelayClientSinkOptions } from './dispatcher'
+import { PROJECTION_STALL_EVICTION_MS } from './dispatcher-projection-stall'
 import { DISPATCHER_CONTROL_QUEUE_MAX_FRAMES } from './dispatcher-writer-admission'
 
 type SlowClient = {
@@ -186,6 +187,31 @@ describe('RelayDispatcher slow-client PTY projection', () => {
     }
   })
 
+  it('never encodes a projection with no subscribers', () => {
+    const primary = makeStallingClient(65536)
+    primary.drain()
+    const dispatcher = new RelayDispatcher(primary.write, primary.options)
+    try {
+      // A flow-controlled owner is excluded from the legacy projection, so "no target" is the hot path.
+      let reads = 0
+      const params = {
+        id: 'pty-1',
+        get data(): string {
+          reads++
+          return 'y'.repeat(40_000)
+        }
+      }
+
+      expect(dispatcher.projectPtyDataToMatchingClients(() => false, params)).toBe(true)
+      expect(reads).toBe(0)
+
+      expect(dispatcher.projectPtyDataToMatchingClients(() => true, params)).toBe(true)
+      expect(reads).toBeGreaterThan(0)
+    } finally {
+      dispatcher.dispose()
+    }
+  })
+
   it('a >1MiB response while the producer queue is full does not close the client', async () => {
     const primary = makeStallingClient(65536)
     const dispatcher = new RelayDispatcher(primary.write, primary.options)
@@ -203,6 +229,91 @@ describe('RelayDispatcher slow-client PTY projection', () => {
   })
 })
 
+describe('RelayDispatcher stalled-subscriber eviction', () => {
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  it('detaches a subscriber that never drains so the paused PTY can resume', () => {
+    vi.useFakeTimers()
+    const primary = makeStallingClient(65536)
+    primary.drain()
+    const stalled = makeStallingClient(65536)
+    const dispatcher = new RelayDispatcher(primary.write, primary.options)
+    const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
+    try {
+      const stalledId = dispatcher.attachClient(stalled.write, stalled.options)
+      fillProducerQueue(dispatcher, stalledId)
+      let capacityNotifications = 0
+      dispatcher.onLegacyPtyCapacity(() => {
+        capacityNotifications++
+      })
+
+      const params = { id: 'pty-1', data: 'y'.repeat(40_000) }
+      expect(dispatcher.projectPtyDataToMatchingClients(() => true, params)).toBe(false)
+      expect(stalled.closes).toBe(0)
+
+      vi.advanceTimersByTime(PROJECTION_STALL_EVICTION_MS + 1)
+
+      expect(stalled.closes).toBe(1)
+      // The eviction is only worth anything if it un-wedges the producer it was blocking.
+      expect(capacityNotifications).toBeGreaterThan(0)
+      expect(dispatcher.projectPtyDataToMatchingClients(() => true, params)).toBe(true)
+      expect(primary.closes).toBe(0)
+    } finally {
+      stderr.mockRestore()
+      dispatcher.dispose()
+    }
+  })
+
+  it('never evicts the primary link, however long it back-pressures', () => {
+    vi.useFakeTimers()
+    const primary = makeStallingClient(65536)
+    const dispatcher = new RelayDispatcher(primary.write, primary.options)
+    const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
+    try {
+      fillProducerQueue(dispatcher, dispatcher.activeClientIds()[0])
+
+      const params = { id: 'pty-1', data: 'y'.repeat(40_000) }
+      expect(dispatcher.projectPtyDataToMatchingClients(() => true, params)).toBe(false)
+
+      vi.advanceTimersByTime(PROJECTION_STALL_EVICTION_MS * 10)
+
+      expect(primary.closes).toBe(0)
+      expect(dispatcher.activeClientIds()).toHaveLength(1)
+    } finally {
+      stderr.mockRestore()
+      dispatcher.dispose()
+    }
+  })
+
+  it('spares a subscriber that drains inside the stall window', () => {
+    vi.useFakeTimers()
+    const primary = makeStallingClient(65536)
+    primary.drain()
+    const slow = makeStallingClient(65536, true)
+    const dispatcher = new RelayDispatcher(primary.write, primary.options)
+    const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
+    try {
+      const slowId = dispatcher.attachClient(slow.write, slow.options)
+      fillProducerQueue(dispatcher, slowId)
+
+      const params = { id: 'pty-1', data: 'y'.repeat(40_000) }
+      expect(dispatcher.projectPtyDataToMatchingClients(() => true, params)).toBe(false)
+
+      vi.advanceTimersByTime(PROJECTION_STALL_EVICTION_MS - 1)
+      slow.drain()
+      vi.advanceTimersByTime(PROJECTION_STALL_EVICTION_MS * 4)
+
+      expect(slow.closes).toBe(0)
+      expect(dispatcher.projectPtyDataToMatchingClients(() => true, params)).toBe(true)
+    } finally {
+      stderr.mockRestore()
+      dispatcher.dispose()
+    }
+  })
+})
+
 describe('RelayDispatcher pty.replay under a full control lane', () => {
   it('drops replay instead of killing a link that would regenerate the same replay', () => {
     const primary = makeStallingClient(65536)

src/relay/dispatcher.ts

@@ -27,6 +27,7 @@ import {
   LegacyRelayPublicationLedger,
   type LegacyPublicationLease
 } from './legacy-relay-publication-ledger'
+import { ProjectionStallEvictor } from './dispatcher-projection-stall'
 
 export type {
   RelayClientSinkOptions,
@@ -100,6 +101,15 @@ export class RelayDispatcher {
   private notificationHandlers = new Map<string, NotificationHandler>()
   private readonly requestAborts = new ClientRequestAborts()
   private readonly publicationLedger = new LegacyRelayPublicationLedger()
+  private readonly projectionStalls = new ProjectionStallEvictor<RelayClient>(
+    (client) => ({
+      stalled: !client.closed && client.writer.retainedProducerBytes > 0,
+      drainProgressBytes: client.writer.drainProgressBytes
+    }),
+    (client) => {
+      this.closeClient(client, new Error('Relay PTY subscriber stalled without draining'), true)
+    }
+  )
   private pendingRelayRequests = new Map<number, PendingRelayRequest>()
   private clientDetachListeners = new Set<(clientId: number) => void>()
   private disposeListeners = new Set<() => void>()
@@ -758,6 +768,7 @@ export class RelayDispatcher {
       return
     }
     this.disposed = true
+    this.projectionStalls.dispose()
     if (this.keepaliveTimer) {
       clearInterval(this.keepaliveTimer)
       this.keepaliveTimer = null
@@ -1128,6 +1139,11 @@ export class RelayDispatcher {
     msg: JsonRpcNotification,
     lane: 'interactive' | 'ordinary'
   ): boolean {
+    // Why: having no projection target is the norm (a flow-controlled owner is excluded from the legacy
+    // mirror), and that is the PTY hot path — never encode the frame just to discard it.
+    if (clients.length === 0) {
+      return true
+    }
     const bytes = this.estimateFrameBytes(msg)
     // Why: closing over a transiently full queue just makes the producer regenerate the same span for a
     // fresh sink of identical capacity, so back-pressure instead; only a frame that can never be admitted
@@ -1142,7 +1158,15 @@ export class RelayDispatcher {
       this.logDroppedProducerNotification(client, msg.method, bytes)
       return false
     })
-    return this.tryPublishToClients(deliverable, msg, lane, bytes)
+    if (this.tryPublishToClients(deliverable, msg, lane, bytes)) {
+      return true
+    }
+    // Why: the pause this rejection triggers is dispatcher-wide, so a sink that never drains would freeze
+    // every PTY for every viewer; arm the bounded reap that detaches it (it reconnects and replays).
+    this.projectionStalls.noteRefusedProjection(
+      deliverable.filter((client) => client !== this.primaryClient)
+    )
+    return false
   }
 
   private publishToClient(
@@ -1270,6 +1294,7 @@ export class RelayDispatcher {
       return
     }
     client.closed = true
+    this.projectionStalls.forget(client)
     this.requestAborts.abortClient(client.id)
     client.writer.close(error)
     client.generation++