From 5e6756b1ef22ff79f3d1d58d3b190fe5ec649e70 Mon Sep 17 00:00:00 2001 From: Neil <4138956+nwparker@users.noreply.github.com> Date: Mon, 3 Aug 2026 00:09:08 -0700 Subject: [PATCH] fix(relay): bound how long a stalled subscriber can hold every PTY paused MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- src/relay/dispatcher-client-writer.ts | 10 ++ src/relay/dispatcher-projection-stall.ts | 86 +++++++++++++++++ src/relay/dispatcher-slow-client.test.ts | 113 ++++++++++++++++++++++- src/relay/dispatcher.ts | 27 +++++- 4 files changed, 234 insertions(+), 2 deletions(-) create mode 100644 src/relay/dispatcher-projection-stall.ts diff --git a/src/relay/dispatcher-client-writer.ts b/src/relay/dispatcher-client-writer.ts index dd9a2e3b245..04d14547992 100644 --- a/src/relay/dispatcher-client-writer.ts +++ b/src/relay/dispatcher-client-writer.ts @@ -33,6 +33,7 @@ export class DispatcherClientWriter { private readonly settleOnDrain = new Set() 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) diff --git a/src/relay/dispatcher-projection-stall.ts b/src/relay/dispatcher-projection-stall.ts new file mode 100644 index 00000000000..b0c56ab9d9b --- /dev/null +++ b/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 { + private readonly armed = new Map() + private timer: ReturnType | 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() + } +} diff --git a/src/relay/dispatcher-slow-client.test.ts b/src/relay/dispatcher-slow-client.test.ts index f89d22a9d29..f0565d393bf 100644 --- a/src/relay/dispatcher-slow-client.test.ts +++ b/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) diff --git a/src/relay/dispatcher.ts b/src/relay/dispatcher.ts index c5b531db005..00dfa399cbc 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 { ProjectionStallEvictor } from './dispatcher-projection-stall' export type { RelayClientSinkOptions, @@ -100,6 +101,15 @@ export class RelayDispatcher { private notificationHandlers = new Map() private readonly requestAborts = new ClientRequestAborts() private readonly publicationLedger = new LegacyRelayPublicationLedger() + private readonly projectionStalls = new ProjectionStallEvictor( + (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() 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++