mirror of
https://github.com/stablyai/orca.git
synced 2026-10-20 00:03:45 +00:00
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.
This commit is contained in:
+9
-4
@@ -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`
|
||||
)
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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'
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user