Fix stale relay socket startup recovery (#2563)
* fix: address pr-bug-scan validated finding from #2447 Restore self-recovery from a stale socket file (left behind when a prior relay was killed by SIGKILL/OOM/host crash) without unlinking a live duplicate's socket. * fix: address review findings --------- Co-authored-by: orca-bug-scan-bot <orca-bug-scan-bot@stably.ai>
This commit is contained in:
Jinjing
GitHub
parent
ba0bae0a0d
commit
be305c8e0a
|
|
@ -49,6 +49,7 @@ import { resolveOpenCodeSourceConfigDir, resolvePiSourceAgentDir } from './plugi
|
|||
const DEFAULT_GRACE_MS = DEFAULT_SSH_RELAY_GRACE_PERIOD_SECONDS * 1000
|
||||
const SOCK_NAME = 'relay.sock'
|
||||
const CONNECT_TIMEOUT_MS = 5_000
|
||||
const STALE_SOCKET_PROBE_TIMEOUT_MS = 500
|
||||
const EMPTY_DETACHED_STARTUP_GRACE_MS = parseNonNegativeIntEnv(
|
||||
'ORCA_RELAY_EMPTY_STARTUP_GRACE_MS',
|
||||
60_000
|
||||
|
|
@ -60,6 +61,10 @@ type SocketIdentity = {
|
|||
ctimeNs: bigint
|
||||
}
|
||||
|
||||
function sameSocketIdentity(a: SocketIdentity, b: SocketIdentity): boolean {
|
||||
return a.dev === b.dev && a.ino === b.ino && a.ctimeNs === b.ctimeNs
|
||||
}
|
||||
|
||||
function parseNonNegativeIntEnv(name: string, fallback: number): number {
|
||||
const raw = process.env[name]
|
||||
if (raw === undefined) {
|
||||
|
|
@ -196,9 +201,7 @@ async function main(): Promise<void> {
|
|||
ownsSocketPath &&
|
||||
ownedSocketIdentity !== null &&
|
||||
currentIdentity !== null &&
|
||||
currentIdentity.dev === ownedSocketIdentity.dev &&
|
||||
currentIdentity.ino === ownedSocketIdentity.ino &&
|
||||
currentIdentity.ctimeNs === ownedSocketIdentity.ctimeNs
|
||||
sameSocketIdentity(currentIdentity, ownedSocketIdentity)
|
||||
)
|
||||
}
|
||||
const cleanupOwnedSocket = (): void => {
|
||||
|
|
@ -561,11 +564,25 @@ async function main(): Promise<void> {
|
|||
}
|
||||
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const onListening = (): void => {
|
||||
let staleRetryAttempted = false
|
||||
|
||||
function removeStartupListeners(): void {
|
||||
server.off('listening', onListening)
|
||||
server.off('error', onInitialError)
|
||||
server.off('error', failInitial)
|
||||
}
|
||||
|
||||
function listenForStartupError(onError: (err: NodeJS.ErrnoException) => void): void {
|
||||
server.once('listening', onListening)
|
||||
server.once('error', onError)
|
||||
server.listen(sockPath)
|
||||
}
|
||||
|
||||
function onListening(): void {
|
||||
removeStartupListeners()
|
||||
restoreUmask()
|
||||
ownsSocketPath = true
|
||||
ownedSocketIdentity = readSocketIdentity(sockPath)
|
||||
server.off('error', onInitialError)
|
||||
server.on('error', (err) => {
|
||||
process.stderr.write(`[relay] Socket server error: ${err.message}\n`)
|
||||
})
|
||||
|
|
@ -573,9 +590,9 @@ async function main(): Promise<void> {
|
|||
resolve()
|
||||
}
|
||||
|
||||
const onInitialError = (err: NodeJS.ErrnoException): void => {
|
||||
function failInitial(err: NodeJS.ErrnoException): void {
|
||||
removeStartupListeners()
|
||||
restoreUmask()
|
||||
server.off('listening', onListening)
|
||||
if (err.code === 'EADDRINUSE') {
|
||||
process.stderr.write(
|
||||
`[relay] Socket path already in use: ${sockPath}; another relay is likely active. Use --connect instead of starting a new daemon.\n`
|
||||
|
|
@ -586,9 +603,81 @@ async function main(): Promise<void> {
|
|||
reject(err)
|
||||
}
|
||||
|
||||
server.once('listening', onListening)
|
||||
server.once('error', onInitialError)
|
||||
server.listen(sockPath)
|
||||
function unlinkIfStillStale(blockedIdentity: SocketIdentity | null): boolean {
|
||||
const currentIdentity = readSocketIdentity(sockPath)
|
||||
if (currentIdentity === null) {
|
||||
return true
|
||||
}
|
||||
if (blockedIdentity === null || !sameSocketIdentity(currentIdentity, blockedIdentity)) {
|
||||
return false
|
||||
}
|
||||
try {
|
||||
unlinkSync(sockPath)
|
||||
return true
|
||||
} catch (unlinkErr) {
|
||||
const e = unlinkErr as NodeJS.ErrnoException
|
||||
return e.code === 'ENOENT'
|
||||
}
|
||||
}
|
||||
|
||||
// Why: a previous relay killed by SIGKILL/OOM/host-crash leaves the
|
||||
// socket file on disk with no listener. EADDRINUSE on bind in that
|
||||
// case is not "duplicate active" — it is a stale inode. Probe with a
|
||||
// short connect; if it refuses, the socket is dead and we may unlink
|
||||
// and retry once. If it connects, a live relay owns it and we keep
|
||||
// the existing "duplicate detected" rejection.
|
||||
function onInitialError(err: NodeJS.ErrnoException): void {
|
||||
if (err.code !== 'EADDRINUSE' || staleRetryAttempted) {
|
||||
failInitial(err)
|
||||
return
|
||||
}
|
||||
staleRetryAttempted = true
|
||||
const blockedIdentity = readSocketIdentity(sockPath)
|
||||
const probe = createConnection({ path: sockPath })
|
||||
let probeSettled = false
|
||||
let probeTimeout: NodeJS.Timeout | null = null
|
||||
const finishProbe = (callback: () => void): void => {
|
||||
if (probeSettled) {
|
||||
return
|
||||
}
|
||||
probeSettled = true
|
||||
if (probeTimeout) {
|
||||
clearTimeout(probeTimeout)
|
||||
}
|
||||
callback()
|
||||
}
|
||||
probe.once('connect', () => {
|
||||
finishProbe(() => {
|
||||
probe.destroy()
|
||||
failInitial(err)
|
||||
})
|
||||
})
|
||||
probe.once('error', (probeErr: NodeJS.ErrnoException) => {
|
||||
finishProbe(() => {
|
||||
if (probeErr.code !== 'ECONNREFUSED' && probeErr.code !== 'ENOENT') {
|
||||
failInitial(err)
|
||||
return
|
||||
}
|
||||
if (!unlinkIfStillStale(blockedIdentity)) {
|
||||
failInitial(err)
|
||||
return
|
||||
}
|
||||
process.stderr.write(
|
||||
`[relay] Removed stale socket at ${sockPath} and retrying listen\n`
|
||||
)
|
||||
removeStartupListeners()
|
||||
listenForStartupError(failInitial)
|
||||
})
|
||||
})
|
||||
probeTimeout = setTimeout(() => {
|
||||
finishProbe(() => {
|
||||
probe.destroy()
|
||||
failInitial(err)
|
||||
})
|
||||
}, STALE_SOCKET_PROBE_TIMEOUT_MS)
|
||||
}
|
||||
|
||||
listenForStartupError(onInitialError)
|
||||
})
|
||||
|
||||
return server
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
/* oxlint-disable max-lines -- Why: subprocess coverage shares one bundled relay artifact; splitting this file would rebuild the same daemon bundle across suites and make these lifecycle tests slower/flakier. */
|
||||
import { afterAll, beforeAll, describe, expect, it, afterEach } from 'vitest'
|
||||
import { mkdtempSync, readFileSync, unlinkSync, writeFileSync } from 'fs'
|
||||
import { existsSync, mkdtempSync, readFileSync, unlinkSync, writeFileSync } from 'fs'
|
||||
import { rm } from 'fs/promises'
|
||||
import * as path from 'path'
|
||||
import { tmpdir } from 'os'
|
||||
|
|
@ -267,6 +267,49 @@ describe('Subprocess: Relay entry point', () => {
|
|||
10_000
|
||||
)
|
||||
|
||||
it.skipIf(process.platform === 'win32')(
|
||||
'reclaims a socket path left behind by a killed detached relay',
|
||||
async () => {
|
||||
tmpDir = mkdtempSync(path.join(tmpdir(), 'relay-stale-'))
|
||||
const sockPath = path.join(tmpDir, 'relay.sock')
|
||||
const first = spawn(['--detached', '--grace-time', '10', '--sock-path', sockPath])
|
||||
let bridge: RelayProcess | null = null
|
||||
try {
|
||||
await first.sentinelReceived
|
||||
|
||||
first.kill('SIGKILL')
|
||||
await first.waitForExit(2000)
|
||||
expect(existsSync(sockPath)).toBe(true)
|
||||
|
||||
relay = spawn(['--detached', '--grace-time', '10', '--sock-path', sockPath])
|
||||
await relay.sentinelReceived
|
||||
|
||||
bridge = spawn(['--connect', '--sock-path', sockPath])
|
||||
await bridge.sentinelReceived
|
||||
const id = bridge.send('relay.status')
|
||||
const resp = await bridge.waitForResponse(id)
|
||||
expect(resp.error).toBeUndefined()
|
||||
expect(
|
||||
resp.result as {
|
||||
pid: number | undefined
|
||||
socket: { path: string; owned: boolean; listening: boolean }
|
||||
}
|
||||
).toMatchObject({
|
||||
pid: relay.proc.pid,
|
||||
socket: { path: sockPath, owned: true, listening: true }
|
||||
})
|
||||
} finally {
|
||||
bridge?.kill('SIGTERM')
|
||||
await bridge?.waitForExit().catch(() => {})
|
||||
if (first.proc.exitCode === null && first.proc.signalCode === null) {
|
||||
first.kill('SIGKILL')
|
||||
await first.waitForExit().catch(() => {})
|
||||
}
|
||||
}
|
||||
},
|
||||
10_000
|
||||
)
|
||||
|
||||
it.skipIf(process.platform === 'win32')(
|
||||
'does not unlink a newer relay socket when an older relay exits',
|
||||
async () => {
|
||||
|
|
|
|||
Loading…
Reference in New Issue