fix(events): reclaim provably-dead orphaned endpoint claims instead of failing permanently

.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.

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<EndpointProbe, EventRuntimeTransportError> =>
   Effect.callback<EndpointProbe, EventRuntimeTransportError>((resume) => {
     const socket = createConnection(endpoint);
@@ -277,45 +292,121 @@ const probeEndpoint = (endpoint: string): Effect.Effect<EndpointProbe, EventRunt
     });
   });
 
+const linuxProcessStartTime = async (pid: number): Promise<string> => {
+  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<EndpointClaimOwner> => ({
+  ...(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<boolean> => {
+  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<EndpointClaim | undefined, EventRuntimeTransportError> =>
   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,
         identity: { device: lockStat.dev, inode: lockStat.ino },
         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;
     }
   }).pipe(
     Effect.mapError((error) => transportError('runtime-failed', 'Unable to claim the event runtime endpoint.', error)),
   );
 
-const releaseEndpointClaim = (claim: EndpointClaim): Effect.Effect<void> =>
-  liftPromise(async () => {
-    await claim.handle.close();
-    let current;
+const readEndpointClaimSnapshot = async (path: string): Promise<EndpointClaimSnapshot | 'missing' | undefined> => {
+  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<boolean> => {
+  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<boolean> =>
+  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<void> =>
+  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.',

packages/agent-bundle/tests/event-ipc.test.ts

@@ -1,16 +1,92 @@
-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';
 
 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<string> => {
+  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<EndpointClaimOwner> => ({
+  ...(process.platform === 'linux' ? { linuxStartTime: await linuxProcessStartTime(process.pid) } : {}),
+  pid: process.pid,
+});
+
+const spawnChildOwner = async (): Promise<Readonly<{
+  child: ChildProcess;
+  owner: EndpointClaimOwner;
+}>> => {
+  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<void> => {
+  if (child.exitCode !== null || child.signalCode !== null) return;
+  child.kill('SIGKILL');
+  await once(child, 'exit');
+};
+
+const deadChildOwner = async (): Promise<EndpointClaimOwner> => {
+  const { child, owner } = await spawnChildOwner();
+  await killChild(child);
+  return owner;
+};
+
+const writeEndpointClaim = async (endpointId: string, contents: string): Promise<string> => {
+  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<void> => {
+  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;