diff --git a/.changeset/event-endpoint-orphan-lock-reclaim.md b/.changeset/event-endpoint-orphan-lock-reclaim.md new file mode 100644 index 000000000..8a6991c16 --- /dev/null +++ b/.changeset/event-endpoint-orphan-lock-reclaim.md @@ -0,0 +1,5 @@ +--- +"agent-bundle": patch +--- + +Reclaim provably dead orphaned event-runtime endpoint claims using owner pid and process start-time identity, while keeping live and ambiguous claims fail-closed. diff --git a/packages/agent-bundle/src/events/ipc.ts b/packages/agent-bundle/src/events/ipc.ts index 981e1968a..7fc75e365 100644 --- a/packages/agent-bundle/src/events/ipc.ts +++ b/packages/agent-bundle/src/events/ipc.ts @@ -1,5 +1,5 @@ import { createHash } from 'node:crypto'; -import { chmod, mkdir, open, rm, stat, type FileHandle } from 'node:fs/promises'; +import { chmod, mkdir, open, readFile, rm, stat, type FileHandle } from 'node:fs/promises'; import { createConnection, createServer, type Server, type Socket } from 'node:net'; import { dirname, join } from 'node:path'; import { StringDecoder } from 'node:string_decoder'; @@ -239,6 +239,21 @@ interface EndpointClaim { readonly path: string; } +interface EndpointClaimOwner { + readonly linuxStartTime?: string; + readonly pid: number; +} + +interface EndpointClaimSnapshot { + readonly identity: Readonly<{ readonly device: number; readonly inode: number }>; + readonly owner: EndpointClaimOwner; +} + +const endpointClaimOwnerSchema = z.object({ + linuxStartTime: z.string().regex(/^\d+$/u).optional(), + pid: z.number().int().positive(), +}).strict(); + const probeEndpoint = (endpoint: string): Effect.Effect => Effect.callback((resume) => { const socket = createConnection(endpoint); @@ -277,14 +292,47 @@ const probeEndpoint = (endpoint: string): Effect.Effect => { + const processStat = await readFile(`/proc/${pid}/stat`, 'utf8'); + const commEnd = processStat.lastIndexOf(')'); + if (commEnd === -1) throw new Error(`Unable to parse process stat for pid ${pid}.`); + const fieldsAfterComm = processStat.slice(commEnd + 1).trim().split(/\s+/u); + const startTime = fieldsAfterComm[19]; + if (startTime === undefined) throw new Error(`Process stat for pid ${pid} has no start time.`); + return startTime; +}; + +const currentEndpointClaimOwner = async (): Promise => ({ + ...(process.platform === 'linux' ? { linuxStartTime: await linuxProcessStartTime(process.pid) } : {}), + pid: process.pid, +}); + +const removeFileIfIdentityMatches = async ( + path: string, + identity: Readonly<{ readonly device: number; readonly inode: number }>, +): Promise => { + let current; + try { + current = await stat(path); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return true; + throw error; + } + if (current.dev !== identity.device || current.ino !== identity.inode) return false; + await rm(path, { force: true }); + return true; +}; + const tryClaimEndpoint = ( endpoint: string, ): Effect.Effect => liftPromise(async () => { const path = `${endpoint}.lock`; + const owner = await currentEndpointClaimOwner(); let handle: FileHandle | undefined; try { handle = await open(path, 'wx', 0o600); + await handle.writeFile(JSON.stringify(owner), 'utf8'); const lockStat = await handle.stat(); return { handle, @@ -292,7 +340,13 @@ const tryClaimEndpoint = ( path, }; } catch (error) { - await handle?.close(); + if (handle !== undefined) { + const lockStat = await handle.stat().catch(() => undefined); + await handle.close(); + if (lockStat !== undefined) { + await removeFileIfIdentityMatches(path, { device: lockStat.dev, inode: lockStat.ino }); + } + } if ((error as NodeJS.ErrnoException).code === 'EEXIST') return undefined; throw error; } @@ -300,22 +354,59 @@ const tryClaimEndpoint = ( Effect.mapError((error) => transportError('runtime-failed', 'Unable to claim the event runtime endpoint.', error)), ); -const releaseEndpointClaim = (claim: EndpointClaim): Effect.Effect => - liftPromise(async () => { - await claim.handle.close(); - let current; +const readEndpointClaimSnapshot = async (path: string): Promise => { + try { + const handle = await open(path, 'r'); try { - current = await stat(claim.path); - } catch (error) { - if ((error as NodeJS.ErrnoException).code === 'ENOENT') return; - throw error; + const rawOwner = await handle.readFile('utf8'); + const lockStat = await handle.stat(); + const owner = endpointClaimOwnerSchema.safeParse(JSON.parse(rawOwner)); + if (!owner.success) return undefined; + return { + identity: { device: lockStat.dev, inode: lockStat.ino }, + owner: owner.data, + }; + } finally { + await handle.close(); } - if ( - current.dev === claim.identity.device - && current.ino === claim.identity.inode - ) { - await rm(claim.path, { force: true }); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return 'missing'; + return undefined; + } +}; + +const isEndpointClaimOwnerProvablyDead = async (owner: EndpointClaimOwner): Promise => { + try { + process.kill(owner.pid, 0); + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code === 'ESRCH') return true; + if (code !== 'EPERM') return false; + } + if (process.platform !== 'linux' || owner.linuxStartTime === undefined) return false; + try { + return await linuxProcessStartTime(owner.pid) !== owner.linuxStartTime; + } catch { + return false; + } +}; + +const reclaimOrphanedEndpointClaim = (path: string): Effect.Effect => + liftPromise(async () => { + const snapshot = await readEndpointClaimSnapshot(path); + if (snapshot === 'missing') return true; + if (snapshot === undefined || !await isEndpointClaimOwnerProvablyDead(snapshot.owner)) return false; + try { + return await removeFileIfIdentityMatches(path, snapshot.identity); + } catch { + return false; } + }).pipe(Effect.catch(() => Effect.succeed(false))); + +const releaseEndpointClaim = (claim: EndpointClaim): Effect.Effect => + liftPromise(async () => { + await claim.handle.close(); + await removeFileIfIdentityMatches(claim.path, claim.identity); }).pipe(Effect.ignore); const claimEndpoint = Effect.fnUntraced(function*( @@ -332,14 +423,13 @@ const claimEndpoint = Effect.fnUntraced(function*( 'Event runtime endpoint already has a live server.', )); } + if (yield* reclaimOrphanedEndpointClaim(`${endpoint}.lock`)) continue; if (attempt + 1 < ENDPOINT_CLAIM_RETRY_COUNT) { yield* Effect.sleep(ENDPOINT_CLAIM_RETRY_DELAY); } } - // A crashed process can leave the claim file behind. Retrying the endpoint - // probe is bounded rather than stealing an unverifiable claim and recreating - // the same unlink race that this lock prevents. + // Unverifiable claims remain fail-closed rather than being stolen. return yield* Effect.fail(transportError( 'runtime-failed', 'Event runtime endpoint claim did not clear after bounded retries.', diff --git a/packages/agent-bundle/tests/event-ipc.test.ts b/packages/agent-bundle/tests/event-ipc.test.ts index f641991be..c3319910d 100644 --- a/packages/agent-bundle/tests/event-ipc.test.ts +++ b/packages/agent-bundle/tests/event-ipc.test.ts @@ -1,5 +1,8 @@ -import { stat, writeFile } from 'node:fs/promises'; +import { spawn, type ChildProcess } from 'node:child_process'; +import { once } from 'node:events'; +import { mkdir, readFile, rm, stat, writeFile } from 'node:fs/promises'; import { createConnection, type Socket } from 'node:net'; +import { dirname } from 'node:path'; import { Effect } from 'effect'; import { expect, it } from 'effect-rstest'; @@ -7,10 +10,83 @@ import { expect, it } from 'effect-rstest'; import { createEventRuntimeServer, createEventRuntimeServerForTest, + eventRuntimeEndpoint, EventRuntimeTransportError, requestEventRuntime, } from '../src/events/ipc.ts'; +interface EndpointClaimOwner { + readonly linuxStartTime?: string; + readonly pid: number; +} + +const linuxProcessStartTime = async (pid: number): Promise => { + const processStat = await readFile(`/proc/${pid}/stat`, 'utf8'); + const commEnd = processStat.lastIndexOf(')'); + if (commEnd === -1) throw new Error(`Unable to parse process stat for pid ${pid}.`); + const fieldsAfterComm = processStat.slice(commEnd + 1).trim().split(/\s+/u); + const startTime = fieldsAfterComm[19]; + if (startTime === undefined) throw new Error(`Process stat for pid ${pid} has no start time.`); + return startTime; +}; + +const currentProcessOwner = async (): Promise => ({ + ...(process.platform === 'linux' ? { linuxStartTime: await linuxProcessStartTime(process.pid) } : {}), + pid: process.pid, +}); + +const spawnChildOwner = async (): Promise> => { + const child = spawn(process.execPath, ['-e', 'setInterval(() => {}, 1000)'], { + stdio: 'ignore', + }); + await once(child, 'spawn'); + const pid = child.pid; + if (pid === undefined) { + child.kill('SIGKILL'); + throw new Error('Spawned child has no pid.'); + } + const owner: EndpointClaimOwner = { + ...(process.platform === 'linux' ? { linuxStartTime: await linuxProcessStartTime(pid) } : {}), + pid, + }; + return { child, owner }; +}; + +const killChild = async (child: ChildProcess): Promise => { + if (child.exitCode !== null || child.signalCode !== null) return; + child.kill('SIGKILL'); + await once(child, 'exit'); +}; + +const deadChildOwner = async (): Promise => { + const { child, owner } = await spawnChildOwner(); + await killChild(child); + return owner; +}; + +const writeEndpointClaim = async (endpointId: string, contents: string): Promise => { + const endpoint = eventRuntimeEndpoint(endpointId); + await mkdir(dirname(endpoint), { mode: 0o700, recursive: true }); + const claimPath = `${endpoint}.lock`; + await writeFile(claimPath, contents, { mode: 0o600 }); + return claimPath; +}; + +const expectBoundedClaimFailure = async (endpointId: string): Promise => { + await expect(createEventRuntimeServer({ + artifactEpoch: 'epoch-1', + endpointId, + handle: async () => undefined, + })).rejects.toMatchObject({ + code: 'runtime-failed', + message: expect.stringMatching(/claim did not clear after bounded retries/u), + name: EventRuntimeTransportError.name, + }); +}; + it.live('round-trips a bounded event envelope through the epoch-bound runtime socket', () => Effect.gen(function*() { const endpointId = `event-ipc-${crypto.randomUUID()}`; const server = yield* Effect.acquireRelease( @@ -207,6 +283,67 @@ it.live('does not unlink a concurrent winner after both servers probe a stale en expect(response).toEqual({ owner: 'first' }); })); +it.live('reclaims an endpoint claim whose owner process was killed', () => Effect.gen(function*() { + if (process.platform === 'win32') return; + const endpointId = `event-ipc-dead-claim-${crypto.randomUUID()}`; + const owner = yield* Effect.promise(deadChildOwner); + const claimPath = yield* Effect.promise(() => writeEndpointClaim(endpointId, JSON.stringify(owner))); + const server = yield* Effect.acquireRelease( + Effect.promise(() => createEventRuntimeServer({ + artifactEpoch: 'epoch-1', + endpointId, + handle: async () => undefined, + })), + (server) => Effect.promise(() => server.close()), + ); + expect(server.endpoint).toBe(eventRuntimeEndpoint(endpointId)); + yield* Effect.promise(() => rm(claimPath, { force: true })); +})); + +it.live('fails closed when an endpoint claim owner is still alive', () => Effect.gen(function*() { + if (process.platform === 'win32') return; + const endpointId = `event-ipc-live-claim-${crypto.randomUUID()}`; + const owner = yield* Effect.promise(currentProcessOwner); + const claimPath = yield* Effect.promise(() => writeEndpointClaim(endpointId, JSON.stringify(owner))); + yield* Effect.promise(() => expectBoundedClaimFailure(endpointId)).pipe( + Effect.ensuring(Effect.promise(() => rm(claimPath, { force: true }))), + ); +})); + +it.live('fails closed when an endpoint claim has unparseable contents', () => Effect.gen(function*() { + if (process.platform === 'win32') return; + const endpointId = `event-ipc-garbage-claim-${crypto.randomUUID()}`; + const claimPath = yield* Effect.promise(() => writeEndpointClaim(endpointId, 'not-json')); + yield* Effect.promise(() => expectBoundedClaimFailure(endpointId)).pipe( + Effect.ensuring(Effect.promise(() => rm(claimPath, { force: true }))), + ); +})); + +it.live('reclaims an endpoint claim whose pid has been recycled', () => Effect.gen(function*() { + if (process.platform !== 'linux') return; + const endpointId = `event-ipc-recycled-claim-${crypto.randomUUID()}`; + const { child, owner } = yield* Effect.acquireRelease( + Effect.promise(spawnChildOwner), + ({ child }) => Effect.promise(() => killChild(child)), + ); + const recycledOwner = { + ...owner, + linuxStartTime: String(BigInt(owner.linuxStartTime ?? '0') + 1n), + }; + const claimPath = yield* Effect.promise(() => writeEndpointClaim(endpointId, JSON.stringify(recycledOwner))); + const server = yield* Effect.acquireRelease( + Effect.promise(() => createEventRuntimeServer({ + artifactEpoch: 'epoch-1', + endpointId, + handle: async () => undefined, + })), + (server) => Effect.promise(() => server.close()), + ); + expect(server.endpoint).toBe(eventRuntimeEndpoint(endpointId)); + yield* Effect.promise(() => rm(claimPath, { force: true })); + expect(child.exitCode).toBeNull(); +})); + it.live('interrupts an in-flight event handler when the client disconnects', () => Effect.gen(function*() { const endpointId = `event-ipc-disconnect-${crypto.randomUUID()}`; let markStarted!: () => void;