mirror of
https://github.com/stablyai/orca.git
synced 2026-11-01 01:27:35 +00:00
fix(remote): a missed wake probe marks a socket suspect instead of closing it (#33071)
On macOS wake and network online events every remote-runtime socket was probed with an 8 s deadline and closed on a miss. Relayed or lossy links still rerouting after a wake lost connections that would have survived. A miss now re-pings and leaves the socket until the normal 25 s keepalive window ends; the probe is skipped when data arrived in the last 3 s.
This commit is contained in:
@@ -19,12 +19,15 @@ export type RemoteRuntimeSocketLivenessOptions = {
|
||||
}
|
||||
|
||||
// Why: after sleep or a network change a socket can read OPEN while its path is gone; the normal
|
||||
// cadence needs ~30-40 s to prove that. A resume probe settles it in one short, explicit window.
|
||||
// cadence needs ~30-40 s to prove that. A resume probe starts the liveness window at once, and a
|
||||
// miss at this deadline only re-pings: a relayed or lossy path can take longer to come back.
|
||||
export const REMOTE_RUNTIME_SOCKET_RESUME_PROBE_DEADLINE_MS = 8_000
|
||||
// Why: inbound data this recent suggests the path survived; the regular keepalive still covers a miss.
|
||||
export const REMOTE_RUNTIME_SOCKET_RESUME_PROBE_RECENT_ACTIVITY_MS = 3_000
|
||||
|
||||
export type RemoteRuntimeSocketLivenessMonitor = {
|
||||
noteActivity: () => void
|
||||
/** Ping now; declare the socket dead unless anything arrives within `deadlineMs`. */
|
||||
/** Ping now; declare the socket dead unless anything arrives within the liveness window. */
|
||||
probeNow: (deadlineMs?: number) => void
|
||||
stop: () => void
|
||||
}
|
||||
@@ -56,6 +59,7 @@ export function startRemoteRuntimeSocketLiveness(args: {
|
||||
let lastTickAt = now()
|
||||
let probeSentAt: number | null = null
|
||||
let activitySinceResumeProbe = false
|
||||
let lastActivityAt: number | null = null
|
||||
let resumeProbeTimer: ReturnType<typeof setTimeout> | null = null
|
||||
let stopped = false
|
||||
|
||||
@@ -105,26 +109,52 @@ export function startRemoteRuntimeSocketLiveness(args: {
|
||||
if (stopped || resumeProbeTimer !== null) {
|
||||
return
|
||||
}
|
||||
const probedAt = now()
|
||||
const sinceActivityMs = lastActivityAt === null ? null : probedAt - lastActivityAt
|
||||
if (
|
||||
sinceActivityMs !== null &&
|
||||
sinceActivityMs >= 0 &&
|
||||
sinceActivityMs < REMOTE_RUNTIME_SOCKET_RESUME_PROBE_RECENT_ACTIVITY_MS
|
||||
) {
|
||||
return
|
||||
}
|
||||
activitySinceResumeProbe = false
|
||||
// Why: a socket still connecting has no path to probe yet; a deadline with no ping could kill it mid-handshake.
|
||||
if (!tryPing()) {
|
||||
return
|
||||
}
|
||||
const probedAt = now()
|
||||
// Why: the probe restarts the keepalive window, so an older unanswered ping cannot cut it short.
|
||||
probeSentAt = probedAt
|
||||
const deathDelayMs = Math.max(livenessTimeoutMs - deadlineMs, deadlineMs)
|
||||
armResumeProbeTimer(probedAt, deadlineMs, deadlineMs, () => {
|
||||
// Why: a missed resume probe is suspect, not dead; a second ping covers a first one lost mid-reroute.
|
||||
tryPing()
|
||||
armResumeProbeTimer(now(), deathDelayMs, deadlineMs, () => {
|
||||
stop()
|
||||
args.onDead()
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
function armResumeProbeTimer(
|
||||
armedAt: number,
|
||||
delayMs: number,
|
||||
probeDeadlineMs: number,
|
||||
onSilent: () => void
|
||||
): void {
|
||||
resumeProbeTimer = setTimeout(() => {
|
||||
resumeProbeTimer = null
|
||||
if (stopped || activitySinceResumeProbe) {
|
||||
return
|
||||
}
|
||||
// Why: a late timer means this process slept or stalled, so replies may still be queued; probe afresh.
|
||||
const elapsedMs = now() - probedAt
|
||||
if (elapsedMs < 0 || elapsedMs > deadlineMs * 1.5) {
|
||||
probeNow(deadlineMs)
|
||||
const elapsedMs = now() - armedAt
|
||||
if (elapsedMs < 0 || elapsedMs > delayMs * 1.5) {
|
||||
probeNow(probeDeadlineMs)
|
||||
return
|
||||
}
|
||||
stop()
|
||||
args.onDead()
|
||||
}, deadlineMs)
|
||||
onSilent()
|
||||
}, delayMs)
|
||||
// Why: mobile typechecks shared code with DOM timer types where unref is absent.
|
||||
const timer: unknown = resumeProbeTimer
|
||||
if (
|
||||
@@ -150,6 +180,7 @@ export function startRemoteRuntimeSocketLiveness(args: {
|
||||
noteActivity: () => {
|
||||
probeSentAt = null
|
||||
activitySinceResumeProbe = true
|
||||
lastActivityAt = now()
|
||||
},
|
||||
probeNow,
|
||||
stop
|
||||
|
||||
@@ -5,8 +5,9 @@ import {
|
||||
} from './remote-runtime-socket-liveness'
|
||||
import { SharedControlReconnectScheduler } from './remote-runtime/control/remote-runtime-shared-control-reconnect'
|
||||
|
||||
// Regression for #9092: after sleep a socket can still read OPEN while its path is gone, and the
|
||||
// normal cadence took ~30-40 s to notice; resume now probes every socket with a short deadline.
|
||||
// Regression for #9092: after sleep a socket can still read OPEN while its path is gone, so resume
|
||||
// probes every socket at once. A miss at the probe deadline marks it suspect and re-pings; only the
|
||||
// normal 25 s keepalive window kills it, because relayed links can take longer than 8 s to reroute.
|
||||
describe('remote runtime resume probe', () => {
|
||||
const monitors: { stop: () => void }[] = []
|
||||
|
||||
@@ -21,26 +22,59 @@ describe('remote runtime resume probe', () => {
|
||||
})
|
||||
|
||||
function start(onDead = vi.fn(), ping = vi.fn(() => true)) {
|
||||
const monitor = startRemoteRuntimeSocketLiveness({
|
||||
ping,
|
||||
onDead,
|
||||
options: { pingIntervalMs: 10_000, livenessTimeoutMs: 25_000 }
|
||||
})
|
||||
const monitor = startRemoteRuntimeSocketLiveness({ ping, onDead })
|
||||
monitors.push(monitor)
|
||||
return { monitor, onDead, ping }
|
||||
}
|
||||
|
||||
it('declares a silent socket dead within the resume deadline', async () => {
|
||||
const { onDead, ping } = start()
|
||||
it('keeps a socket that answers the wake probe 12 s late', async () => {
|
||||
const { monitor, onDead, ping } = start()
|
||||
|
||||
expect(probeAllRemoteRuntimeSocketsNow(2_000)).toBe(1)
|
||||
expect(ping).toHaveBeenCalledTimes(1)
|
||||
await vi.advanceTimersByTimeAsync(1_999)
|
||||
probeAllRemoteRuntimeSocketsNow()
|
||||
await vi.advanceTimersByTimeAsync(8_000)
|
||||
expect(ping).toHaveBeenCalledTimes(2)
|
||||
await vi.advanceTimersByTimeAsync(4_000)
|
||||
monitor.noteActivity()
|
||||
await vi.advanceTimersByTimeAsync(20_000)
|
||||
expect(onDead).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('declares a silent socket dead when the normal keepalive window ends', async () => {
|
||||
const { onDead } = start()
|
||||
|
||||
expect(probeAllRemoteRuntimeSocketsNow()).toBe(1)
|
||||
await vi.advanceTimersByTimeAsync(24_999)
|
||||
expect(onDead).not.toHaveBeenCalled()
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
expect(onDead).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('gives a network-change probe a full window even with an older keepalive unanswered', async () => {
|
||||
const { monitor, onDead } = start()
|
||||
|
||||
await vi.advanceTimersByTimeAsync(29_000)
|
||||
probeAllRemoteRuntimeSocketsNow()
|
||||
await vi.advanceTimersByTimeAsync(12_000)
|
||||
expect(onDead).not.toHaveBeenCalled()
|
||||
monitor.noteActivity()
|
||||
await vi.advanceTimersByTimeAsync(20_000)
|
||||
expect(onDead).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('skips the probe when data arrived in the last few seconds', async () => {
|
||||
const { monitor, onDead, ping } = start()
|
||||
|
||||
monitor.noteActivity()
|
||||
await vi.advanceTimersByTimeAsync(2_000)
|
||||
probeAllRemoteRuntimeSocketsNow()
|
||||
expect(ping).not.toHaveBeenCalled()
|
||||
await vi.advanceTimersByTimeAsync(7_000)
|
||||
expect(onDead).not.toHaveBeenCalled()
|
||||
|
||||
probeAllRemoteRuntimeSocketsNow()
|
||||
expect(ping).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('probes afresh instead of failing a socket when the process stalled through the deadline', async () => {
|
||||
let clock = 0
|
||||
const onDead = vi.fn()
|
||||
@@ -60,21 +94,14 @@ describe('remote runtime resume probe', () => {
|
||||
expect(onDead).not.toHaveBeenCalled()
|
||||
expect(ping).toHaveBeenCalledTimes(2)
|
||||
|
||||
clock += 2_000
|
||||
await vi.advanceTimersByTimeAsync(2_000)
|
||||
// The fresh probe still ends a socket that stays silent.
|
||||
for (let elapsed = 0; elapsed < 120_000; elapsed += 1_000) {
|
||||
clock += 1_000
|
||||
await vi.advanceTimersByTimeAsync(1_000)
|
||||
}
|
||||
expect(onDead).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('keeps a socket that answers the probe', async () => {
|
||||
const { monitor, onDead } = start()
|
||||
|
||||
probeAllRemoteRuntimeSocketsNow(2_000)
|
||||
await vi.advanceTimersByTimeAsync(500)
|
||||
monitor.noteActivity()
|
||||
await vi.advanceTimersByTimeAsync(5_000)
|
||||
expect(onDead).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('skips stopped monitors and coalesces a repeated resume into one probe', async () => {
|
||||
const stopped = start()
|
||||
stopped.monitor.stop()
|
||||
@@ -84,7 +111,7 @@ describe('remote runtime resume probe', () => {
|
||||
probeAllRemoteRuntimeSocketsNow(2_000)
|
||||
expect(stopped.ping).not.toHaveBeenCalled()
|
||||
expect(live.ping).toHaveBeenCalledTimes(1)
|
||||
await vi.advanceTimersByTimeAsync(2_000)
|
||||
await vi.advanceTimersByTimeAsync(25_000)
|
||||
expect(live.onDead).toHaveBeenCalledTimes(1)
|
||||
expect(stopped.onDead).not.toHaveBeenCalled()
|
||||
})
|
||||
@@ -95,13 +122,13 @@ describe('remote runtime resume probe', () => {
|
||||
const { onDead } = start(vi.fn(), ping)
|
||||
|
||||
probeAllRemoteRuntimeSocketsNow(2_000)
|
||||
await vi.advanceTimersByTimeAsync(2_000)
|
||||
await vi.advanceTimersByTimeAsync(9_000)
|
||||
expect(onDead).not.toHaveBeenCalled()
|
||||
|
||||
// Once open, the next resume probes it normally.
|
||||
open = true
|
||||
probeAllRemoteRuntimeSocketsNow(2_000)
|
||||
await vi.advanceTimersByTimeAsync(2_000)
|
||||
await vi.advanceTimersByTimeAsync(25_000)
|
||||
expect(onDead).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user