From 79e08dff79d696b304e73b0cca7762d7994d7dac Mon Sep 17 00:00:00 2001 From: Neil <4138956+nwparker@users.noreply.github.com> Date: Mon, 3 Aug 2026 01:31:50 -0700 Subject: [PATCH] fix(relay): close a saturated PTY mirror instead of wedging every terminal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 68 ++++++++++++------- src/relay/dispatcher.ts | 45 ++++++++---- .../pty-handler-source-publication.test.ts | 9 +-- .../relay-pty-source-publication.test.ts | 6 +- 4 files changed, 81 insertions(+), 47 deletions(-) diff --git a/src/relay/dispatcher-slow-client.test.ts b/src/relay/dispatcher-slow-client.test.ts index d5f6072899a..0b07e4ecd60 100644 --- a/src/relay/dispatcher-slow-client.test.ts +++ b/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) diff --git a/src/relay/dispatcher.ts b/src/relay/dispatcher.ts index 5a8c509796e..04c06547a68 100644 --- a/src/relay/dispatcher.ts +++ b/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 }) diff --git a/src/relay/pty-handler-source-publication.test.ts b/src/relay/pty-handler-source-publication.test.ts index 2348c8316f9..3978f6eb5a8 100644 --- a/src/relay/pty-handler-source-publication.test.ts +++ b/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() }) }) diff --git a/src/relay/relay-pty-source-publication.test.ts b/src/relay/relay-pty-source-publication.test.ts index 62854a38663..67fefd92039 100644 --- a/src/relay/relay-pty-source-publication.test.ts +++ b/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)