diff --git a/src/relay/relay.ts b/src/relay/relay.ts index 94a0637c5..1b22d7c3a 100644 --- a/src/relay/relay.ts +++ b/src/relay/relay.ts @@ -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 { 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 { } await new Promise((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 { 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 { 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 diff --git a/src/relay/subprocess.test.ts b/src/relay/subprocess.test.ts index 7e6841384..ced23197a 100644 --- a/src/relay/subprocess.test.ts +++ b/src/relay/subprocess.test.ts @@ -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 () => {