diff --git a/src/relay/dispatcher-client-writer.ts b/src/relay/dispatcher-client-writer.ts index 04d14547992..dd9a2e3b245 100644 --- a/src/relay/dispatcher-client-writer.ts +++ b/src/relay/dispatcher-client-writer.ts @@ -33,7 +33,6 @@ 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 @@ -56,14 +55,6 @@ 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 } @@ -295,7 +286,6 @@ 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 deleted file mode 100644 index b0c56ab9d9b..00000000000 --- a/src/relay/dispatcher-projection-stall.ts +++ /dev/null @@ -1,86 +0,0 @@ -/** - * 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 f0565d393bf..d5f6072899a 100644 --- a/src/relay/dispatcher-slow-client.test.ts +++ b/src/relay/dispatcher-slow-client.test.ts @@ -1,6 +1,10 @@ import { afterEach, describe, expect, it, vi } from 'vitest' -import { RelayDispatcher, type RelayClientSinkOptions } from './dispatcher' -import { PROJECTION_STALL_EVICTION_MS } from './dispatcher-projection-stall' +import { + RelayDispatcher, + type LegacyProjectionRole, + type RelayClientSinkOptions, + type SinkWriteSettlement +} from './dispatcher' import { DISPATCHER_CONTROL_QUEUE_MAX_FRAMES } from './dispatcher-writer-admission' type SlowClient = { @@ -72,6 +76,18 @@ function fillProducerQueue(dispatcher: RelayDispatcher, clientId: number): void } } +function roles( + byId: Record +): (clientId: number) => LegacyProjectionRole { + 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) @@ -94,14 +110,15 @@ describe('RelayDispatcher slow-client PTY projection', () => { } }) - it('rejects a projected pty.data frame instead of closing a subscriber that cannot drain', () => { + it('rejects a projected pty.data frame instead of closing an owner that cannot drain', () => { 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 primaryId = dispatcher.activeClientIds()[0] + fillProducerQueue(dispatcher, primaryId) - const projected = dispatcher.projectPtyDataToMatchingClients(() => true, { + const projected = dispatcher.projectPtyDataToLegacyClients(roles({ [primaryId]: 'owner' }), { id: 'pty-1', data: 'y'.repeat(40_000) }) @@ -115,13 +132,48 @@ describe('RelayDispatcher slow-client PTY projection', () => { } }) - it('sheds a projected frame no sink can ever admit rather than wedging the PTY', () => { + it('sheds a saturated mirror instead of pausing the PTY for every other viewer', () => { + 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) + const mirrorBefore = ptyDataFrameCount(mirror.frames) + stderr.mockClear() + + const projected = dispatcher.projectPtyDataToLegacyClients( + roles({ [ownerId]: 'owner', [mirrorId]: 'mirror' }), + { id: 'pty-1', data: 'y'.repeat(40_000) } + ) + + // A bystander mirror (agent hook, orca CLI, second window) must not gate the native PTY. + expect(projected).toBe(true) + expect(ptyDataFrameCount(owner.frames)).toBe(1) + expect(ptyDataFrameCount(mirror.frames)).toBe(mirrorBefore) + expect(mirror.closes).toBe(0) + 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() + } + }) + + it('sheds a projected frame no owner sink can ever admit rather than wedging the PTY', () => { const primary = makeStallingClient(65536) const dispatcher = new RelayDispatcher(primary.write, primary.options) const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true) try { + const primaryId = dispatcher.activeClientIds()[0] // Larger than producerFrameCapacity (49152B), so retrying it forever would stall the PTY. - const projected = dispatcher.projectPtyDataToMatchingClients(() => true, { + const projected = dispatcher.projectPtyDataToLegacyClients(roles({ [primaryId]: 'owner' }), { id: 'pty-1', data: 'y'.repeat(60_000) }) @@ -141,53 +193,57 @@ describe('RelayDispatcher slow-client PTY projection', () => { } }) - it('delivers a rejected projection exactly once to every subscriber after the stalled sink drains', () => { - const stalled = makeStallingClient(65536, true) - const healthy = makeStallingClient(65536, true) - healthy.drain() - const dispatcher = new RelayDispatcher(stalled.write, stalled.options) + it('delivers a rejected projection exactly once to every mirror after the owner drains', () => { + const owner = makeStallingClient(65536, true) + const mirror = makeStallingClient(65536, true) + mirror.drain() + const dispatcher = new RelayDispatcher(owner.write, owner.options) const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true) try { - dispatcher.attachClient(healthy.write, healthy.options) - fillProducerQueue(dispatcher, dispatcher.activeClientIds()[0]) - const healthyBefore = ptyDataFrameCount(healthy.frames) + const ownerId = dispatcher.activeClientIds()[0] + const mirrorId = dispatcher.attachClient(mirror.write, mirror.options) + fillProducerQueue(dispatcher, ownerId) + const mirrorBefore = ptyDataFrameCount(mirror.frames) + const roleOf = roles({ [ownerId]: 'owner', [mirrorId]: 'mirror' }) const params = { id: 'pty-1', data: 'y'.repeat(40_000) } - expect(dispatcher.projectPtyDataToMatchingClients(() => true, params)).toBe(false) - // All-or-nothing: the healthy subscriber must not get a copy the retry would duplicate. - expect(ptyDataFrameCount(healthy.frames)).toBe(healthyBefore) + expect(dispatcher.projectPtyDataToLegacyClients(roleOf, params)).toBe(false) + // The mirror must not get a copy the owner's retry would duplicate. + expect(ptyDataFrameCount(mirror.frames)).toBe(mirrorBefore) - stalled.drain() - expect(dispatcher.projectPtyDataToMatchingClients(() => true, params)).toBe(true) - expect(ptyDataFrameCount(healthy.frames)).toBe(healthyBefore + 1) - expect(stalled.closes).toBe(0) - expect(healthy.closes).toBe(0) + owner.drain() + expect(dispatcher.projectPtyDataToLegacyClients(roleOf, params)).toBe(true) + expect(ptyDataFrameCount(mirror.frames)).toBe(mirrorBefore + 1) + expect(owner.closes).toBe(0) + expect(mirror.closes).toBe(0) } finally { stderr.mockRestore() dispatcher.dispose() } }) - it('holds a projected pty.exit for retry instead of closing a saturated subscriber', () => { + it('holds a projected pty.exit for retry instead of closing a saturated owner', () => { const primary = makeStallingClient(65536, true) const dispatcher = new RelayDispatcher(primary.write, primary.options) const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true) try { - fillProducerQueue(dispatcher, dispatcher.activeClientIds()[0]) + const primaryId = dispatcher.activeClientIds()[0] + fillProducerQueue(dispatcher, primaryId) + const roleOf = roles({ [primaryId]: 'owner' }) const params = { id: 'pty-1', exitCode: 0 } - expect(dispatcher.projectPtyExitToMatchingClients(() => true, params)).toBe(false) + expect(dispatcher.projectPtyExitToLegacyClients(roleOf, params)).toBe(false) expect(primary.closes).toBe(0) primary.drain() - expect(dispatcher.projectPtyExitToMatchingClients(() => true, params)).toBe(true) + expect(dispatcher.projectPtyExitToLegacyClients(roleOf, params)).toBe(true) } finally { stderr.mockRestore() dispatcher.dispose() } }) - it('never encodes a projection with no subscribers', () => { + it('never encodes a projection with no targets', () => { const primary = makeStallingClient(65536) primary.drain() const dispatcher = new RelayDispatcher(primary.write, primary.options) @@ -201,11 +257,14 @@ describe('RelayDispatcher slow-client PTY projection', () => { return 'y'.repeat(40_000) } } + const primaryId = dispatcher.activeClientIds()[0] - expect(dispatcher.projectPtyDataToMatchingClients(() => false, params)).toBe(true) + expect(dispatcher.projectPtyDataToLegacyClients(roles({}), params)).toBe(true) expect(reads).toBe(0) - expect(dispatcher.projectPtyDataToMatchingClients(() => true, params)).toBe(true) + expect( + dispatcher.projectPtyDataToLegacyClients(roles({ [primaryId]: 'owner' }), params) + ).toBe(true) expect(reads).toBeGreaterThan(0) } finally { dispatcher.dispose() @@ -229,84 +288,58 @@ describe('RelayDispatcher slow-client PTY projection', () => { }) }) -describe('RelayDispatcher stalled-subscriber eviction', () => { +describe('RelayDispatcher projection never reaps a client on a timer', () => { afterEach(() => { vi.useRealTimers() }) - it('detaches a subscriber that never drains so the paused PTY can resume', () => { + it('never closes a peer that is merely mid-write while another target back-pressures', () => { vi.useFakeTimers() - const primary = makeStallingClient(65536) - primary.drain() - const stalled = makeStallingClient(65536) - const dispatcher = new RelayDispatcher(primary.write, primary.options) + const owner = makeStallingClient(65536) + const neverDrains = makeStallingClient(65536) + const dispatcher = new RelayDispatcher(owner.write, owner.options) const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true) + let midWriteCloses = 0 try { - const stalledId = dispatcher.attachClient(stalled.write, stalled.options) - fillProducerQueue(dispatcher, stalledId) - let capacityNotifications = 0 - dispatcher.onLegacyPtyCapacity(() => { - capacityNotifications++ + 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. + void settle + return true + }, + { + supportsWriteCallback: true, + writableHighWaterMark: () => 65536, + writableLength: () => 0, + close: () => { + midWriteCloses++ + } + } + ) + fillProducerQueue(dispatcher, ownerId) + fillProducerQueue(dispatcher, neverDrainsId) + const roleOf = roles({ + [ownerId]: 'owner', + [neverDrainsId]: 'mirror', + [midWriteId]: 'mirror' }) - 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(dispatcher.projectPtyDataToLegacyClients(roleOf, params)).toBe(false) + // Just inside the 30s window an earlier revision watched, so the peer armed here would be + // judged over ~1ms of observation. + vi.advanceTimersByTime(29_999) + dispatcher.tryNotifyPtyDataToClient(midWriteId, { id: 'pty-1', data: 'z' }, () => {}) + expect(dispatcher.projectPtyDataToLegacyClients(roleOf, params)).toBe(false) + vi.advanceTimersByTime(10 * 60_000) - 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) + // Reaping a peer 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) } finally { stderr.mockRestore() dispatcher.dispose() diff --git a/src/relay/dispatcher.ts b/src/relay/dispatcher.ts index 00dfa399cbc..5a8c509796e 100644 --- a/src/relay/dispatcher.ts +++ b/src/relay/dispatcher.ts @@ -27,7 +27,6 @@ import { LegacyRelayPublicationLedger, type LegacyPublicationLease } from './legacy-relay-publication-ledger' -import { ProjectionStallEvictor } from './dispatcher-projection-stall' export type { RelayClientSinkOptions, @@ -55,6 +54,15 @@ export type RelayClientSourceOptions = { resumeReads?: () => void } +/** + * 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. + */ +export type LegacyProjectionRole = 'skip' | 'owner' | 'mirror' + export type MethodHandler = ( params: Record, context: RequestContext @@ -101,15 +109,6 @@ 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>() @@ -344,8 +343,8 @@ export class RelayDispatcher { ) } - projectPtyDataToMatchingClients( - matchesClient: (clientId: number) => boolean, + projectPtyDataToLegacyClients( + roleOf: (clientId: number) => LegacyProjectionRole, params: Record, options: { interactive?: boolean } = {} ): boolean { @@ -353,7 +352,7 @@ export class RelayDispatcher { return false } return this.projectToClients( - this.activeClients().filter((client) => matchesClient(client.id)), + roleOf, { jsonrpc: '2.0', method: 'pty.data', params }, options.interactive ? 'interactive' : 'ordinary' ) @@ -410,18 +409,14 @@ export class RelayDispatcher { ) } - projectPtyExitToMatchingClients( - matchesClient: (clientId: number) => boolean, + projectPtyExitToLegacyClients( + roleOf: (clientId: number) => LegacyProjectionRole, params: Record ): boolean { if (this.disposed) { return false } - return this.projectToClients( - this.activeClients().filter((client) => matchesClient(client.id)), - { jsonrpc: '2.0', method: 'pty.exit', params }, - 'ordinary' - ) + return this.projectToClients(roleOf, { jsonrpc: '2.0', method: 'pty.exit', params }, 'ordinary') } tryNotifyPtyExitToClient( @@ -768,7 +763,6 @@ export class RelayDispatcher { return } this.disposed = true - this.projectionStalls.dispose() if (this.keepaliveTimer) { clearInterval(this.keepaliveTimer) this.keepaliveTimer = null @@ -1135,38 +1129,54 @@ export class RelayDispatcher { } private projectToClients( - clients: readonly RelayClient[], + roleOf: (clientId: number) => LegacyProjectionRole, msg: JsonRpcNotification, lane: 'interactive' | 'ordinary' ): boolean { + const owners: RelayClient[] = [] + const mirrors: RelayClient[] = [] + for (const client of this.clients.values()) { + if (client.closed) { + continue + } + const role = roleOf(client.id) + if (role === 'owner') { + owners.push(client) + } else if (role === 'mirror') { + mirrors.push(client) + } + } // 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) { + if (owners.length === 0 && mirrors.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 - // is shed, because retrying it forever would wedge the PTY. - const deliverable = clients.filter((client) => { - if (client.closed) { - return false - } + // Why: closing over a transiently full queue only makes the producer regenerate the same span for a + // fresh sink of identical capacity, so an owner — which has no second delivery lane — back-pressures + // instead. A frame past its frame capacity can never be admitted, so shed that one or the retry wedges. + const blocking = owners.filter((client) => { if (bytes <= client.writer.producerFrameCapacity) { return true } this.logDroppedProducerNotification(client, msg.method, bytes) return false }) - 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 + return this.runPublicationTransaction(() => { + 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. + for (const client of mirrors) { + if (client.closed || this.publishToClient(client, msg, lane, undefined, bytes)) { + continue + } + this.logDroppedProducerNotification(client, msg.method, bytes) + } + return !this.disposed + }) } private publishToClient( @@ -1294,7 +1304,6 @@ export class RelayDispatcher { return } client.closed = true - this.projectionStalls.forget(client) this.requestAborts.abortClient(client.id) client.writer.close(error) client.generation++ diff --git a/src/relay/pty-handler-source-publication.test.ts b/src/relay/pty-handler-source-publication.test.ts index ffe7f3c45a6..2348c8316f9 100644 --- a/src/relay/pty-handler-source-publication.test.ts +++ b/src/relay/pty-handler-source-publication.test.ts @@ -652,7 +652,7 @@ describe('PtyHandler negotiated source publication', () => { await spawn({}) const appendSpy = vi.spyOn(adapter, 'appendSource') const projectSpy = vi - .spyOn(dispatcher, 'projectPtyDataToMatchingClients') + .spyOn(dispatcher, 'projectPtyDataToLegacyClients') .mockReturnValueOnce(false) // Why: a larger capacity result on the retry must not move the memoized span boundary. const maxCharsSpy = vi.spyOn(dispatcher, 'maxLegacyPtyDataChars') @@ -679,7 +679,7 @@ describe('PtyHandler negotiated source publication', () => { it('does not emit pty.exit ahead of output a failed projection re-queued', async () => { await spawn({}) const projectSpy = vi - .spyOn(dispatcher, 'projectPtyDataToMatchingClients') + .spyOn(dispatcher, 'projectPtyDataToLegacyClients') .mockImplementationOnce(() => { // Why: capacity fans out from a bare socket callback mid-drain, when the captured queue has // already been removed from pendingOutputByPty and the exit guard would read it as empty. @@ -709,28 +709,17 @@ describe('PtyHandler negotiated source publication', () => { expect(ordered.indexOf('pty.data')).toBeLessThan(ordered.lastIndexOf('pty.exit')) }) - it('pauses the PTY instead of detaching a subscriber that saturates, and resumes on drain', async () => { + it('keeps the native PTY, the V1 owner and a saturated subscriber all live', async () => { await spawn({}) const detached: number[] = [] const healthyWrites: Buffer[] = [] - const saturatedSettlements: ((result: SinkWriteSettlement) => void)[] = [] - const saturatedDrainWaiters: (() => void)[] = [] - let saturatedStalled = true + const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true) dispatcher.onClientDetached((clientId) => detached.push(clientId)) - const saturatedId = dispatcher.attachClient( - (_data, settle) => { - saturatedSettlements.push(settle) - return !saturatedStalled - }, - { - supportsWriteCallback: true, - writableLength: () => (saturatedStalled ? 16 * 1024 : 0), - writableHighWaterMark: () => 4 * 1024 * 1024, - waitWriteDrain: (callback: () => void) => { - saturatedDrainWaiters.push(callback) - } - } - ) + const saturatedId = dispatcher.attachClient(() => false, { + supportsWriteCallback: true, + writableLength: () => 16 * 1024, + writableHighWaterMark: () => 4 * 1024 * 1024 + }) const healthyId = dispatcher.attachClient( (data, settle) => { healthyWrites.push(Buffer.from(data)) @@ -739,8 +728,6 @@ describe('PtyHandler negotiated source publication', () => { }, { supportsWriteCallback: true } ) - const healthyPtyData = (): unknown[] => - healthyWrites.map(notification).filter((frame) => frame?.method === 'pty.data') const payload = 's'.repeat(16 * 1024) let admitted = 0 while ( @@ -754,26 +741,18 @@ 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([]) expect(detached).not.toContain(healthyId) - // All-or-nothing: the healthy mirror waits rather than taking a copy the retry would duplicate. - expect(healthyPtyData()).toHaveLength(0) - expect(pausePty).toHaveBeenCalled() - // The span is already in the source ledger, so the credited owner keeps draining it. + expect( + healthyWrites.map(notification).filter((frame) => frame?.method === 'pty.data') + ).toHaveLength(1) expect(sourceDataFrames()).toHaveLength(1) expect(publication.getDebugSnapshot()).toMatchObject({ sendCommitted: 1 }) - - saturatedStalled = false - for (const waiter of saturatedDrainWaiters.splice(0)) { - waiter() - } - while (saturatedSettlements.length > 0) { - saturatedSettlements.shift()!({ ok: true }) - } - await vi.advanceTimersByTimeAsync(8) - - expect(healthyPtyData()).toHaveLength(1) - expect(sourceDataFrames()).toHaveLength(1) - expect(detached).toEqual([]) + expect(pausePty).not.toHaveBeenCalled() + expect( + stderr.mock.calls.filter((call) => /Dropped pty\.data/.test(String(call[0]))) + ).toHaveLength(1) + stderr.mockRestore() }) }) diff --git a/src/relay/relay-pty-source-exit-publication.test.ts b/src/relay/relay-pty-source-exit-publication.test.ts index f5a567a33fc..46488cce737 100644 --- a/src/relay/relay-pty-source-exit-publication.test.ts +++ b/src/relay/relay-pty-source-exit-publication.test.ts @@ -106,7 +106,7 @@ function createScenario( tryNotifyPtyExitToMatchingClients: vi.fn( (_matches: (clientId: number) => boolean, _params: unknown) => notifyAccepted ), - projectPtyExitToMatchingClients: vi.fn(() => true), + projectPtyExitToLegacyClients: vi.fn(() => true), tryNotifyPtyExitToClient: vi.fn(() => true) } const sender = { pump: vi.fn(), wakeSendWaiters: vi.fn() } @@ -315,7 +315,7 @@ describe('RelayPtySourceLegacyExitIndex', () => { it('keeps the exit pending when the subscriber projection is refused', () => { const record = deliveryRecord({ sealed: true }) const scenario = createScenario(closedSnapshot({ state: 'sealed-unsettled' }), record) - scenario.dispatcher.projectPtyExitToMatchingClients.mockReturnValueOnce(false) + scenario.dispatcher.projectPtyExitToLegacyClients.mockReturnValueOnce(false) expect(scenario.run()).toBe(false) expect(record.legacyExitAccepted).toBe(false) @@ -325,7 +325,7 @@ describe('RelayPtySourceLegacyExitIndex', () => { // The handler's retry after drain lands the exit once, with no duplicate legacy projection. expect(scenario.run()).toBe(true) expect(record.legacyExitAccepted).toBe(true) - expect(scenario.dispatcher.projectPtyExitToMatchingClients).toHaveBeenCalledTimes(2) + expect(scenario.dispatcher.projectPtyExitToLegacyClients).toHaveBeenCalledTimes(2) expect(scenario.dispatcher.tryNotifyPtyExitToClient).toHaveBeenCalledOnce() }) @@ -338,7 +338,7 @@ describe('RelayPtySourceLegacyExitIndex', () => { }) expect(() => scenario.runTracked(index)).toThrow('owner write failed') - expect(scenario.dispatcher.projectPtyExitToMatchingClients).toHaveBeenCalledOnce() + expect(scenario.dispatcher.projectPtyExitToLegacyClients).toHaveBeenCalledOnce() expect(scenario.session.cancelDelivery).toHaveBeenCalledWith( record.identity, 'exit-publication-failed' diff --git a/src/relay/relay-pty-source-exit-publication.ts b/src/relay/relay-pty-source-exit-publication.ts index 6257ae25442..4588c82b823 100644 --- a/src/relay/relay-pty-source-exit-publication.ts +++ b/src/relay/relay-pty-source-exit-publication.ts @@ -1,4 +1,5 @@ import type { RelayDispatcher } from './dispatcher' +import { legacyProjectionRole } from './relay-pty-source-output' import { onceSinkSettlement, type RelayPtySourceDeliveryRecord, @@ -177,8 +178,8 @@ export function sealAndPublishPtySourceExit(options: PtySourceExitOptions): bool return false } if (!record.legacyExitAccepted) { - record.legacyExitAccepted = dispatcher.projectPtyExitToMatchingClients( - (clientId) => session.deliveryMode(clientId) !== 'source-owner', + record.legacyExitAccepted = dispatcher.projectPtyExitToLegacyClients( + (clientId) => legacyProjectionRole(session, clientId), params ) options.legacyExits?.remember(params, record.legacyExitAccepted) diff --git a/src/relay/relay-pty-source-output.ts b/src/relay/relay-pty-source-output.ts index bbc57d1c643..1cda5e0893c 100644 --- a/src/relay/relay-pty-source-output.ts +++ b/src/relay/relay-pty-source-output.ts @@ -1,6 +1,6 @@ import { randomUUID } from 'node:crypto' import type { PtySourceDeliveryIdentity } from '../shared/pty-source-credit-contract' -import type { RelayDispatcher } from './dispatcher' +import type { LegacyProjectionRole, RelayDispatcher } from './dispatcher' import type { RelayPtySourceDeliveryRecord } from './relay-pty-source-send-scheduler' import type { SshPtyConsumerSessionAdapter } from './ssh-pty-consumer-session-adapter' @@ -50,6 +50,22 @@ export function appendPtySourceOutput( return true } +/** + * Only a session owner is served solely by the legacy projection, so only it may back-pressure the native + * PTY. Every other target — agent hooks, `orca` CLI processes, a second window, a client still completing + * pty.openClient — is a bystander mirror that would otherwise wedge the terminal for everyone. + */ +export function legacyProjectionRole( + session: SshPtyConsumerSessionAdapter, + clientId: number +): LegacyProjectionRole { + const mode = session.deliveryMode(clientId) + if (mode === 'source-owner') { + return 'skip' + } + return mode === 'legacy-owner' ? 'owner' : 'mirror' +} + export function projectPtySourceOutputToLegacy( dispatcher: RelayDispatcher, session: SshPtyConsumerSessionAdapter, @@ -57,8 +73,8 @@ export function projectPtySourceOutputToLegacy( output: RelayPtySourceOutput, interactive: boolean ): boolean { - return dispatcher.projectPtyDataToMatchingClients( - (clientId) => session.deliveryMode(clientId) !== 'source-owner', + return dispatcher.projectPtyDataToLegacyClients( + (clientId) => legacyProjectionRole(session, clientId), { id, data: output.data, diff --git a/src/relay/relay-pty-source-publication.test.ts b/src/relay/relay-pty-source-publication.test.ts index b79895f2713..62854a38663 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('backpressures instead of detaching a saturated subscriber while the V1 owner stays live', async () => { + it('sheds a saturated subscriber without detaching it or stalling the V1 owner', async () => { const harness = await createHarness(8) const detached: number[] = [] const saturatedWrites: Buffer[] = [] @@ -324,8 +324,8 @@ describe('RelayPtySourcePublication', () => { expect(admitted).toBeGreaterThan(0) expect(admitted).toBeLessThan(20) - // False tells pty-handler to pause the PTY and republish the span once the sink drains. - expect(harness.publication.publish('pty-1', { data: saturatedPayload }, false)).toBe(false) + // A bystander subscriber is neither closed (#12041) nor allowed to pause the PTY for everyone. + expect(harness.publication.publish('pty-1', { data: saturatedPayload }, false)).toBe(true) expect(detached).toEqual([]) expect(detached).not.toContain(healthyId) @@ -333,8 +333,7 @@ describe('RelayPtySourcePublication', () => { expect(heldSettlements).toHaveLength(1) expect( healthyWrites.map(notification).filter((frame) => frame?.method === 'pty.data') - ).toHaveLength(0) - // The span is already appended to the source ledger, so the credited owner still drains it. + ).toHaveLength(1) expect( harness.writes.map(notification).filter((frame) => frame?.method === 'pty.data') ).toHaveLength(1)