mirror of
https://github.com/stablyai/orca.git
synced 2026-10-17 08:03:14 +00:00
fix(ssh): guard mux dead-link detector against sleep/App Nap timer pauses
After system sleep or App Nap timer throttling, the first post-wake
timeout-check tick saw pre-pause keepalives as >20s stale and killed a
healthy link (false 'Connection timed out (no ack received)' ->
dispose('connection_lost') -> reconnect overlay churn). Track the last
tick time; when a tick gap far exceeds the interval, reset staleness
tracking, probe with a fresh keepalive, and let the next full window
make an honest liveness determination. A genuinely dead link is still
detected within ~25s after wake.
Also adds probeLiveness(timeoutMs): a keepalive round-trip primitive
that resolves true on the first frame of any kind, used by the resume
path to distinguish surviving links from dead ones.
Refs #7773
Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
@@ -307,6 +307,81 @@ describe('SshChannelMultiplexer', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('wake guard (timer pause across system sleep, #7773)', () => {
|
||||
it('does not kill a healthy link on the first tick after a long timer pause', () => {
|
||||
// Reach steady state with pending unacked keepalives (<5s old at pause).
|
||||
vi.advanceTimersByTime(5_000)
|
||||
expect(mux.isDisposed()).toBe(false)
|
||||
|
||||
// Simulate sleep/App Nap: wall clock jumps far ahead with no ticks.
|
||||
// Without the guard, the first post-wake tick sees lastReceivedAt and
|
||||
// the pre-pause keepalive both >20s stale and disposes the mux.
|
||||
vi.setSystemTime(Date.now() + 60 * 60 * 1000)
|
||||
const writesBefore = transport.written.length
|
||||
vi.advanceTimersByTime(5_000)
|
||||
|
||||
expect(mux.isDisposed()).toBe(false)
|
||||
// The guard probes immediately with a fresh keepalive.
|
||||
expect(transport.written.length).toBeGreaterThan(writesBefore)
|
||||
expect(transport.written.at(-1)![0]).toBe(MessageType.KeepAlive)
|
||||
})
|
||||
|
||||
it('keeps the link alive after wake when frames resume', () => {
|
||||
vi.advanceTimersByTime(5_000)
|
||||
vi.setSystemTime(Date.now() + 60 * 60 * 1000)
|
||||
vi.advanceTimersByTime(5_000) // guard tick
|
||||
|
||||
// The relay answers the post-wake probe; the link must stay up.
|
||||
let seq = 1
|
||||
for (let i = 0; i < 8; i++) {
|
||||
vi.advanceTimersByTime(5_000)
|
||||
transport.dataCallbacks[0](encodeKeepAliveFrame(seq++, 0))
|
||||
}
|
||||
expect(mux.isDisposed()).toBe(false)
|
||||
})
|
||||
|
||||
it('still detects a genuinely dead link within the next window after wake', () => {
|
||||
vi.advanceTimersByTime(5_000)
|
||||
vi.setSystemTime(Date.now() + 60 * 60 * 1000)
|
||||
vi.advanceTimersByTime(5_000) // guard tick: reset + probe, no kill
|
||||
|
||||
expect(mux.isDisposed()).toBe(false)
|
||||
// No frames arrive after the guard reset; the honest window expires.
|
||||
vi.advanceTimersByTime(25_000)
|
||||
expect(mux.isDisposed()).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('probeLiveness', () => {
|
||||
it('sends a keepalive and resolves true when any frame arrives', async () => {
|
||||
const writesBefore = transport.written.length
|
||||
const probe = mux.probeLiveness(5_000)
|
||||
|
||||
expect(transport.written.length).toBe(writesBefore + 1)
|
||||
expect(transport.written.at(-1)![0]).toBe(MessageType.KeepAlive)
|
||||
|
||||
transport.dataCallbacks[0](encodeKeepAliveFrame(1, 0))
|
||||
await expect(probe).resolves.toBe(true)
|
||||
})
|
||||
|
||||
it('resolves false when no frame arrives before the timeout', async () => {
|
||||
const probe = mux.probeLiveness(5_000)
|
||||
vi.advanceTimersByTime(5_000)
|
||||
await expect(probe).resolves.toBe(false)
|
||||
})
|
||||
|
||||
it('resolves false when the mux is disposed while probing', async () => {
|
||||
const probe = mux.probeLiveness(5_000)
|
||||
mux.dispose()
|
||||
await expect(probe).resolves.toBe(false)
|
||||
})
|
||||
|
||||
it('resolves false immediately on a disposed mux', async () => {
|
||||
mux.dispose()
|
||||
await expect(mux.probeLiveness(5_000)).resolves.toBe(false)
|
||||
})
|
||||
})
|
||||
|
||||
describe('dispose', () => {
|
||||
it('rejects all pending requests on dispose', async () => {
|
||||
const promise = mux.request('pty.spawn')
|
||||
|
||||
@@ -34,6 +34,9 @@ export type MethodNotificationHandler = (params: Record<string, unknown>) => voi
|
||||
export type RequestHandler = (params: Record<string, unknown>) => Promise<unknown> | unknown
|
||||
|
||||
const REQUEST_TIMEOUT_MS = 30_000
|
||||
// Why: a tick gap far beyond the interval means the process was paused
|
||||
// (system sleep, App Nap timer throttling) — not that the link is dead (#7773).
|
||||
const WAKE_GAP_MS = KEEPALIVE_SEND_MS * 3
|
||||
|
||||
export class SshChannelMultiplexer {
|
||||
private decoder: FrameDecoder
|
||||
@@ -58,6 +61,10 @@ export class SshChannelMultiplexer {
|
||||
// Track the oldest unacked outgoing message timestamp
|
||||
private unackedTimestamps = new Map<number, number>()
|
||||
|
||||
// Why: liveness probes (#7773) resolve on the first frame of any kind —
|
||||
// a keepalive ack proves the relay round-trip without a full RPC.
|
||||
private livenessProbeWaiters: { succeed: () => void; fail: () => void }[] = []
|
||||
|
||||
constructor(transport: MultiplexerTransport) {
|
||||
this.transport = transport
|
||||
|
||||
@@ -232,6 +239,31 @@ export class SshChannelMultiplexer {
|
||||
this.sendMessage(msg)
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a fresh keepalive and resolve true when any frame arrives before the
|
||||
* timeout. Used on system resume to distinguish a link that survived sleep
|
||||
* from a dead one before tearing the session down (#7773).
|
||||
*/
|
||||
probeLiveness(timeoutMs: number): Promise<boolean> {
|
||||
if (this.disposed) {
|
||||
return Promise.resolve(false)
|
||||
}
|
||||
return new Promise<boolean>((resolve) => {
|
||||
const settle = (alive: boolean): void => {
|
||||
clearTimeout(timer)
|
||||
const idx = this.livenessProbeWaiters.indexOf(waiter)
|
||||
if (idx !== -1) {
|
||||
this.livenessProbeWaiters.splice(idx, 1)
|
||||
}
|
||||
resolve(alive)
|
||||
}
|
||||
const waiter = { succeed: () => settle(true), fail: () => settle(false) }
|
||||
const timer = setTimeout(() => settle(false), timeoutMs)
|
||||
this.livenessProbeWaiters.push(waiter)
|
||||
this.sendKeepAlive()
|
||||
})
|
||||
}
|
||||
|
||||
dispose(reason: 'shutdown' | 'connection_lost' = 'shutdown'): void {
|
||||
if (this.disposed) {
|
||||
return
|
||||
@@ -259,6 +291,10 @@ export class SshChannelMultiplexer {
|
||||
reason === 'connection_lost' ? 'SSH connection lost, reconnecting...' : 'Multiplexer disposed'
|
||||
const errorCode = reason === 'connection_lost' ? 'CONNECTION_LOST' : 'DISPOSED'
|
||||
|
||||
for (const waiter of this.livenessProbeWaiters.splice(0)) {
|
||||
waiter.fail()
|
||||
}
|
||||
|
||||
for (const [id, pending] of this.pendingRequests) {
|
||||
pending.cleanup()
|
||||
const err = new Error(errorMessage) as Error & { code: string }
|
||||
@@ -322,6 +358,12 @@ export class SshChannelMultiplexer {
|
||||
}
|
||||
|
||||
private handleFrame(frame: DecodedFrame): void {
|
||||
// Why: any decoded frame proves the relay round-trip is alive; resolve
|
||||
// pending resume probes before ordinary dispatch (#7773).
|
||||
for (const waiter of this.livenessProbeWaiters.splice(0)) {
|
||||
waiter.succeed()
|
||||
}
|
||||
|
||||
// Update ack tracking
|
||||
if (frame.id > this.highestReceivedSeq) {
|
||||
this.highestReceivedSeq = frame.id
|
||||
@@ -453,12 +495,28 @@ export class SshChannelMultiplexer {
|
||||
}
|
||||
|
||||
private startTimeoutCheck(): void {
|
||||
let lastTickAt = Date.now()
|
||||
this.timeoutTimer = setInterval(() => {
|
||||
if (this.disposed) {
|
||||
return
|
||||
}
|
||||
|
||||
const now = Date.now()
|
||||
const sinceLastTick = now - lastTickAt
|
||||
lastTickAt = now
|
||||
// Why: after sleep/App Nap the pre-pause keepalive looks stale on the
|
||||
// first post-wake tick, killing a healthy link (#7773). Reset staleness
|
||||
// tracking, probe with a fresh keepalive, and let the NEXT full window
|
||||
// make an honest liveness determination.
|
||||
if (sinceLastTick > WAKE_GAP_MS) {
|
||||
this.lastReceivedAt = now
|
||||
for (const seq of this.unackedTimestamps.keys()) {
|
||||
this.unackedTimestamps.set(seq, now)
|
||||
}
|
||||
this.sendKeepAlive()
|
||||
return
|
||||
}
|
||||
|
||||
const noDataReceived = now - this.lastReceivedAt > TIMEOUT_MS
|
||||
|
||||
// Check oldest unacked message
|
||||
|
||||
Reference in New Issue
Block a user