From 2eff180e99dec29e2c87a17be129b795b13badcf Mon Sep 17 00:00:00 2001 From: Jinjing <6427696+AmethystLiang@users.noreply.github.com> Date: Sat, 1 Aug 2026 14:26:54 -0700 Subject: [PATCH] fix(relay): prevent infinite redeploy on flapping SSH transports Charge reconnect attempts when connection restores mid-backoff, preventing infinite loop on transports that flap between states. Refactor control overflow handling to use entry property instead of WeakSet marker for clarity. --- src/main/ipc/ssh.ts | 13 +++++++---- src/relay/dispatcher-client-writer.ts | 29 ++++++++---------------- src/relay/dispatcher-writer-admission.ts | 11 +++------ src/relay/dispatcher.ts | 8 +++---- src/relay/fs-handler-stream.test.ts | 2 +- 5 files changed, 26 insertions(+), 37 deletions(-) diff --git a/src/main/ipc/ssh.ts b/src/main/ipc/ssh.ts index e1f74bff7b0..0227b84f36a 100644 --- a/src/main/ipc/ssh.ts +++ b/src/main/ipc/ssh.ts @@ -745,7 +745,7 @@ function configureRelaySessionCallbacks(session: SshRelaySession): void { return } - const scheduleRelayRedeploy = (delay: number): void => { + const scheduleRelayRedeploy = (delay: number, attemptCharged: boolean): void => { state.reconnectTimer = setTimeout(() => { state.reconnectTimer = null relayLostBackoff.set(tid, state) @@ -756,6 +756,11 @@ function configureRelaySessionCallbacks(session: SshRelaySession): void { } const status = connectionManager?.getState(tid)?.status if (status === 'connected') { + if (!attemptCharged) { + // Why: waiting is free, but the deploy it defers is real — charge it here so a transport that + // flaps back to 'connected' can't redeploy forever on an uncharged budget. + state.attempts += 1 + } void s.reconnect(liveConn, relayGracePeriodForTarget(t)) return } @@ -766,7 +771,7 @@ function configureRelaySessionCallbacks(session: SshRelaySession): void { } // Why: still mid-transition — re-arm at the max delay without consuming an attempt. It ends once // the transport settles: 'connected' redeploys, a terminal status or a dropped session clears above. - scheduleRelayRedeploy(RELAY_LOST_MAX_DELAY_MS) + scheduleRelayRedeploy(RELAY_LOST_MAX_DELAY_MS, false) }, delay) relayLostBackoff.set(tid, state) } @@ -779,7 +784,7 @@ function configureRelaySessionCallbacks(session: SshRelaySession): void { 'Relay channel lost. Reconnecting...', state.attempts ) - scheduleRelayRedeploy(RELAY_LOST_MAX_DELAY_MS) + scheduleRelayRedeploy(RELAY_LOST_MAX_DELAY_MS, false) console.warn( `[ssh] Relay channel for ${tid} lost while the SSH transport is ${transportStatus ?? 'unknown'}; waiting ${RELAY_LOST_MAX_DELAY_MS}ms without consuming an attempt` ) @@ -795,7 +800,7 @@ function configureRelaySessionCallbacks(session: SshRelaySession): void { 'Relay channel lost. Reconnecting...', state.attempts ) - scheduleRelayRedeploy(delay) + scheduleRelayRedeploy(delay, true) console.warn( `[ssh] Relay channel for ${tid} lost; reconnect attempt ${state.attempts}/${RELAY_LOST_MAX_ATTEMPTS} in ${delay}ms` ) diff --git a/src/relay/dispatcher-client-writer.ts b/src/relay/dispatcher-client-writer.ts index 1966ce31dfc..58be59173cf 100644 --- a/src/relay/dispatcher-client-writer.ts +++ b/src/relay/dispatcher-client-writer.ts @@ -83,7 +83,8 @@ export class DispatcherClientWriter { lane: DispatcherWriterLane, encode: () => Buffer, estimatedBytes: number, - onSettled: (result: SinkWriteSettlement) => void = () => {} + onSettled: (result: SinkWriteSettlement) => void = () => {}, + overflowIsNonFatal = false ): boolean { if (this.closed) { onSettled({ ok: false, error: new Error('Relay writer is closed') }) @@ -94,7 +95,8 @@ export class DispatcherClientWriter { encode, estimatedBytes, onSettled: onceDispatcherWriterSettlement(onSettled), - settled: false + settled: false, + overflowIsNonFatal } const admission = this.admission.admit(entry, this.sink.frameCapacity(lane, this.saturated)) if (!admission.accepted) { @@ -176,29 +178,18 @@ export class DispatcherClientWriter { } return this.laneScheduler.select( this.admission, - (entry) => this.canWriteControl(entry), - (entry) => this.canWriteProducer(entry) + (entry) => this.canWrite(entry, this.sink.highWaterMark), + (entry) => this.canWrite(entry, this.sink.producerFrameCapacity) ) } - private canWriteControl(entry: DispatcherWriterEntry): boolean { - const highWaterMark = this.sink.highWaterMark - if (!Number.isFinite(highWaterMark)) { + // Why: a zero-length writable always takes one frame, else a single oversized frame could never drain. + private canWrite(entry: DispatcherWriterEntry, capacity: number): boolean { + if (!Number.isFinite(this.sink.highWaterMark)) { return true } const writableLength = this.sink.writableLength - return writableLength + entry.estimatedBytes <= highWaterMark || writableLength === 0 - } - - private canWriteProducer(entry: DispatcherWriterEntry): boolean { - const highWaterMark = this.sink.highWaterMark - if (!Number.isFinite(highWaterMark)) { - return true - } - return ( - this.sink.writableLength + entry.estimatedBytes <= this.sink.producerFrameCapacity || - this.sink.writableLength === 0 - ) + return writableLength + entry.estimatedBytes <= capacity || writableLength === 0 } private writeEntry(entry: DispatcherWriterEntry): void { diff --git a/src/relay/dispatcher-writer-admission.ts b/src/relay/dispatcher-writer-admission.ts index 677bab60210..07db61c12e8 100644 --- a/src/relay/dispatcher-writer-admission.ts +++ b/src/relay/dispatcher-writer-admission.ts @@ -15,6 +15,8 @@ export type DispatcherWriterEntry = { estimatedBytes: number onSettled: (result: DispatcherWriterSettlement) => void settled: boolean + // Why: control overflow closes the client by default; best-effort frames opt out of that instead. + overflowIsNonFatal?: boolean } type AdmissionResult = @@ -27,13 +29,6 @@ const LIVENESS_QUEUE_MAX_FRAMES = 2 export const DEFAULT_PRODUCER_QUEUE_MAX_BYTES = 2 * 1024 * 1024 -const nonFatalControlEncodes = new WeakSet<() => Buffer>() - -export function markControlOverflowNonFatal(encode: () => Buffer): () => Buffer { - nonFatalControlEncodes.add(encode) - return encode -} - export function onceDispatcherWriterSettlement( callback: (result: DispatcherWriterSettlement) => void ): (result: DispatcherWriterSettlement) => void { @@ -151,7 +146,7 @@ export class DispatcherWriterAdmission { private admitControl(entry: DispatcherWriterEntry): AdmissionResult { // Why: best-effort controls may fail soft without weakening protocol-critical admission. - const overflowIsNonFatal = nonFatalControlEncodes.delete(entry.encode) + const overflowIsNonFatal = entry.overflowIsNonFatal === true if ( this.controlFrames >= DISPATCHER_CONTROL_QUEUE_MAX_FRAMES || this.controlBytes + entry.estimatedBytes > DISPATCHER_CONTROL_QUEUE_MAX_BYTES diff --git a/src/relay/dispatcher.ts b/src/relay/dispatcher.ts index e7be71adf8f..edf55403d15 100644 --- a/src/relay/dispatcher.ts +++ b/src/relay/dispatcher.ts @@ -23,7 +23,6 @@ import { type RelayClientWrite, type SinkWriteSettlement } from './dispatcher-client-writer' -import { markControlOverflowNonFatal } from './dispatcher-writer-admission' import { LegacyRelayPublicationLedger, type LegacyPublicationLease @@ -991,11 +990,10 @@ export class RelayDispatcher { } return client.writer.enqueue( lane, - lane === 'control' && controlOverflow === 'reject' - ? markControlOverflowNonFatal(encode) - : encode, + encode, frameBytes, - onSettled + onSettled, + lane === 'control' && controlOverflow === 'reject' ) } diff --git a/src/relay/fs-handler-stream.test.ts b/src/relay/fs-handler-stream.test.ts index c770fe72f23..ba6cfbf1877 100644 --- a/src/relay/fs-handler-stream.test.ts +++ b/src/relay/fs-handler-stream.test.ts @@ -453,7 +453,7 @@ describe('FsHandler readFileStream', () => { it('caps undelivered terminal frames so they cannot overflow the link-killing control lane', async () => { const filePath = path.join(tmpDir, 'terminal-budget.png') - writeFileSync(filePath, Buffer.alloc(64 * 1024, 0x42)) // one chunk per stream + writeFileSync(filePath, Buffer.alloc(STREAM_CHUNK_SIZE, 0x42)) // one chunk per stream // A socket that accepts nothing: control frames queue, and the real writer destroys the // client once 256 of them pile up. dispatcher.holdControlSettlements()