perf(process): share batched token observation across ownership checks

src/process/DarwinProcessRunTokenReader.ts

@@ -262,11 +262,17 @@ export const createDarwinProcessRunTokenReader = (
 
   const read = async (
     processes: readonly ProcessTableEntry[],
+    signal?: AbortSignal,
   ): Promise<ReadonlyMap<number, ProcessRunTokenObservation>> => {
+    signal?.throwIfAborted();
     if (processes.length === 0) return new Map();
-    const binary = await compile();
+    const binary = await compile(signal);
     const pids = processes.map(({ pid }) => String(pid));
-    const { stdout } = await execFileOutput(binary, pids, { timeout: 15_000 });
+    const { stdout } = await execFileOutput(binary, pids, {
+      timeout: 15_000,
+      ...(signal === undefined ? {} : { signal }),
+    });
+    signal?.throwIfAborted();
     const decoded = safeParseJson(stdout);
     if (!decoded.ok)
       throw new Error(

src/process/ProcessOwnership.observation.test.ts

@@ -0,0 +1,169 @@
+import { describe, expect, it, vi } from "vitest";
+import {
+  cleanupOwnedProcessGroup,
+  type ProcessOwnershipHost,
+} from "./ProcessOwnership.js";
+import {
+  observeOwnedProcessGroup,
+  observeOwnedProcessLineage,
+} from "./ProcessOwnershipObservation.js";
+import { host, ownership } from "./ProcessOwnership.fixture.js";
+
+const processes = [100, 101].map((pid) => ({
+  pid,
+  parentPid: pid === 100 ? 1 : 100,
+  processGroupId: 100,
+  state: "S",
+  command: "fixture",
+}));
+
+describe("batched ownership observation", () => {
+  it("observes group and lineage with one batch each and falls back only for missing rows", async () => {
+    const { adapter } = host({
+      100: { REA_PROCESS_RUN_ID: ownership.runId },
+      101: { REA_PROCESS_RUN_ID: ownership.runId },
+    });
+    const environment = vi.fn(adapter.environment);
+    const runTokens = vi.fn<NonNullable<ProcessOwnershipHost["runTokens"]>>(
+      () =>
+        Promise.resolve(
+          new Map([[100, { state: "readable", runId: ownership.runId }]]),
+        ),
+    );
+    const observedHost = { ...adapter, environment, runTokens };
+    await expect(
+      observeOwnedProcessGroup(ownership, observedHost),
+    ).resolves.toEqual({ state: "alive" });
+    await expect(
+      observeOwnedProcessLineage(ownership, observedHost),
+    ).resolves.toMatchObject({
+      status: "verified",
+      lineage: {
+        descendants: [{ pid: 101, parentPid: 100, processGroupId: 100 }],
+      },
+    });
+    expect(runTokens.mock.calls).toEqual([[processes], [processes]]);
+    expect(environment.mock.calls).toEqual([[101], [101]]);
+  });
+
+  it.each(["batch", "environment"] as const)(
+    "preserves cancellation during %s reads, including successful completion races",
+    async (boundary) => {
+      const controller = new AbortController();
+      const reason = new Error("cancelled ownership read");
+      const listProcesses = vi.fn(() => Promise.resolve(processes));
+      const environment = vi.fn<ProcessOwnershipHost["environment"]>(
+        async (_pid, signal) => {
+          expect(signal).toBe(controller.signal);
+          controller.abort(reason);
+          return { REA_PROCESS_RUN_ID: ownership.runId };
+        },
+      );
+      const runTokens = vi.fn<NonNullable<ProcessOwnershipHost["runTokens"]>>(
+        async (_processes, signal) => {
+          expect(signal).toBe(controller.signal);
+          if (boundary === "batch") controller.abort(reason);
+          return new Map();
+        },
+      );
+      await expect(
+        observeOwnedProcessGroup(
+          ownership,
+          { listProcesses, environment, runTokens, signalGroup: vi.fn() },
+          controller.signal,
+        ),
+      ).rejects.toBe(reason);
+      expect(listProcesses).toHaveBeenCalledTimes(1);
+      expect(environment).toHaveBeenCalledTimes(boundary === "batch" ? 0 : 1);
+    },
+  );
+
+  it.each([100, 101])(
+    "rechecks unavailable batch rows for exited PID %i without overriding them through fallback",
+    async (exitedPid) => {
+      const listProcesses = vi
+        .fn<ProcessOwnershipHost["listProcesses"]>()
+        .mockResolvedValueOnce(processes)
+        .mockResolvedValue(processes.filter(({ pid }) => pid !== exitedPid));
+      const environment = vi.fn(() =>
+        Promise.resolve({ REA_PROCESS_RUN_ID: ownership.runId }),
+      );
+      const observedHost: ProcessOwnershipHost = {
+        listProcesses,
+        environment,
+        signalGroup: vi.fn(),
+        runTokens: () =>
+          Promise.resolve(
+            new Map(
+              processes.map(({ pid }) => [
+                pid,
+                pid === exitedPid
+                  ? { state: "unavailable" as const, reason: "process exited" }
+                  : { state: "readable" as const, runId: ownership.runId },
+              ]),
+            ),
+          ),
+      };
+      const result = await observeOwnedProcessLineage(ownership, observedHost);
+      if (exitedPid === 100)
+        expect(result).toMatchObject({
+          status: "unavailable",
+          reason: "owned launcher exited during lineage validation",
+        });
+      else
+        expect(result).toMatchObject({
+          status: "verified",
+          lineage: { descendants: [] },
+        });
+      expect(environment).not.toHaveBeenCalled();
+    },
+  );
+});
+
+describe("token sweep batch failures", () => {
+  it("cleans token-owned groups when a failed batch has readable fallback observations", async () => {
+    const { adapter, signalGroup } = host({
+      100: { REA_PROCESS_RUN_ID: ownership.runId },
+      101: { REA_PROCESS_RUN_ID: ownership.runId },
+    });
+    await expect(
+      cleanupOwnedProcessGroup(
+        { ...ownership, sweepTokenOwnedProcesses: true, captureBaseline: [] },
+        {
+          ...adapter,
+          runTokens: () =>
+            Promise.reject(new Error("batch reader unavailable")),
+        },
+      ),
+    ).resolves.toEqual({ cleaned: true, signaled: true });
+    expect(signalGroup.mock.calls).toEqual([[100, "SIGKILL"]]);
+  });
+
+  it("uses fallback ownership when the initial batch fails and preserves both failures through retries", async () => {
+    const signalGroup = vi.fn();
+    const adapter: ProcessOwnershipHost = {
+      listProcesses: () => Promise.resolve(processes),
+      environment: (pid) =>
+        pid === 100
+          ? Promise.resolve({ REA_PROCESS_RUN_ID: ownership.runId })
+          : Promise.reject(new Error("individual unreadable")),
+      runTokens: () => Promise.reject(new Error("batch reader unavailable")),
+      signalGroup,
+    };
+    const result = await cleanupOwnedProcessGroup(
+      { ...ownership, sweepTokenOwnedProcesses: true, captureBaseline: [] },
+      adapter,
+    );
+    expect(result).toMatchObject({
+      cleaned: false,
+      failures: expect.arrayContaining([
+        expect.objectContaining({
+          pid: 101,
+          diagnostic:
+            "individual unreadable; run-token batch failed: batch reader unavailable",
+        }),
+      ]),
+    });
+    expect(signalGroup).not.toHaveBeenCalled();
+  });
+});

src/process/ProcessOwnership.ts

@@ -7,6 +7,7 @@ import {
 } from "./ProcessOwnershipObservation.js";
 import { launcherIdentityFailure } from "./ProcessOwnershipIdentity.js";
 import { descendantsOf, liveProcesses } from "./ProcessOwnershipProcessTree.js";
+import { readProcessRunTokens } from "./ProcessRunTokenObservations.js";
 
 const execFileAsync = promisify(execFile);
 
@@ -90,10 +91,14 @@ export interface ProcessOwnershipHost {
   close?(): Promise<void>;
   /** List current processes; startup inspection may be cancelled by the caller. */
   listProcesses(signal?: AbortSignal): Promise<readonly ProcessTableEntry[]>;
-  environment(pid: number): Promise<Readonly<Record<string, string>>>;
+  environment(
+    pid: number,
+    signal?: AbortSignal,
+  ): Promise<Readonly<Record<string, string>>>;
   /** Read only run-token values in one host operation when the OS supports it. */
   runTokens?(
     processes: readonly ProcessTableEntry[],
+    signal?: AbortSignal,
   ): Promise<ReadonlyMap<number, ProcessRunTokenObservation>>;
   /** Read stable identities; omit the signal during cleanup so it can finish. */
   processIdentities?(
@@ -678,29 +683,10 @@ const processRunTokenFailures = async (
   host: ProcessOwnershipHost,
 ): Promise<readonly ProcessOwnershipValidationFailure[]> => {
   const failures: ProcessOwnershipValidationFailure[] = [];
-  let runTokenObservations:
-    | ReadonlyMap<number, ProcessRunTokenObservation>
-    | undefined;
-  let runTokenReadFailure: string | undefined;
-  if (host.runTokens !== undefined) {
-    try {
-      runTokenObservations = await host.runTokens(members);
-    } catch (cause: unknown) {
-      runTokenReadFailure = errorMessage(cause);
-    }
-  }
-  for (const member of members) {
-    const batchObservation = runTokenObservations?.get(member.pid);
-    let observation =
-      batchObservation ?? (await readEnvironmentRunToken(host, member.pid));
-    if (
-      runTokenReadFailure !== undefined &&
-      observation.state === "unavailable"
-    )
-      observation = {
-        state: "unavailable",
-        reason: `${observation.reason}; run-token batch failed: ${runTokenReadFailure}`,
-      };
+  for await (const { process: member, observation } of readProcessRunTokens(
+    host,
+    members,
+  )) {
     if (observation.state === "readable") {
       if (observation.runId !== runId)
         failures.push({ pid: member.pid, reason: "run-token-mismatch" });
@@ -823,26 +809,14 @@ const scanTokenOwnedProcesses = async (
       identity?.state === "readable" ? identity.identity : null,
     );
   };
-  let bulkTokens: ReadonlyMap<number, ProcessRunTokenObservation> | undefined;
-  if (host.runTokens !== undefined)
-    bulkTokens = await host.runTokens(candidates);
   const unreadable: UnreadableTokenCandidate[] = [];
-  for (const process of candidates) {
-    const bulkToken = bulkTokens?.get(process.pid);
-    if (bulkToken !== undefined) {
-      if (bulkToken.state === "readable") {
-        if (bulkToken.runId === runId) recordOwned(process);
-        continue;
-      }
-      unreadable.push({ process, diagnostic: bulkToken.reason });
-      continue;
-    }
-    try {
-      if ((await host.environment(process.pid)).REA_PROCESS_RUN_ID === runId)
-        recordOwned(process);
-    } catch (cause: unknown) {
-      unreadable.push({ process, diagnostic: errorMessage(cause) });
-    }
+  for await (const { process, observation } of readProcessRunTokens(
+    host,
+    candidates,
+  )) {
+    if (observation.state === "readable") {
+      if (observation.runId === runId) recordOwned(process);
+    } else unreadable.push({ process, diagnostic: observation.reason });
   }
   const remaining = await settleUnreadableTokenCandidates(
     unreadable,
@@ -968,31 +942,20 @@ const settleUnreadableTokenCandidates = async (
       void cause;
       live = undefined;
     }
-    let observations:
-      | ReadonlyMap<number, ProcessRunTokenObservation>
-      | undefined;
-    if (host.runTokens !== undefined) {
-      try {
-        observations = await host.runTokens(
-          remaining.map(({ process }) => process),
-        );
-      } catch (cause: unknown) {
-        void cause;
-        observations = undefined;
-      }
-    }
     const stillUnreadable: UnreadableTokenCandidate[] = [];
-    for (const candidate of remaining) {
-      const pid = candidate.process.pid;
-      if (live !== undefined && !live.has(pid)) continue;
-      const observation =
-        observations?.get(pid) ?? (await readEnvironmentRunToken(host, pid));
+    const candidates = remaining
+      .map(({ process }) => process)
+      .filter(({ pid }) => live === undefined || live.has(pid));
+    for await (const { process, observation } of readProcessRunTokens(
+      host,
+      candidates,
+    )) {
       if (observation.state === "readable") {
-        if (observation.runId === runId) recordOwned(candidate.process);
+        if (observation.runId === runId) recordOwned(process);
         continue;
       }
       stillUnreadable.push({
-        process: candidate.process,
+        process,
         diagnostic: observation.reason,
       });
     }
@@ -1001,20 +964,6 @@ const settleUnreadableTokenCandidates = async (
   return remaining;
 };
 
-const readEnvironmentRunToken = async (
-  host: ProcessOwnershipHost,
-  pid: number,
-): Promise<ProcessRunTokenObservation> => {
-  try {
-    return {
-      state: "readable",
-      runId: (await host.environment(pid)).REA_PROCESS_RUN_ID,
-    };
-  } catch (cause: unknown) {
-    return { state: "unavailable", reason: errorMessage(cause) };
-  }
-};
-
 const processIsGone = async (
   host: ProcessOwnershipHost,
   pid: number,

src/process/ProcessOwnershipObservation.ts

@@ -5,6 +5,7 @@ import { launcherIdentityFailure } from "./ProcessOwnershipIdentity.js";
 import { descendantsOf, liveProcesses } from "./ProcessOwnershipProcessTree.js";
 import { execFileOutput } from "./ExecFileOutput.js";
 import { createDarwinProcessRunTokenReader } from "./DarwinProcessRunTokenReader.js";
+import { readProcessRunTokens } from "./ProcessRunTokenObservations.js";
 import type {
   OwnedProcessGroup,
   ProcessGroupObservation,
@@ -37,26 +38,28 @@ export const observeOwnedProcessGroup = async (
   signal?.throwIfAborted();
   const liveMembers = liveProcesses(members);
   if (liveMembers.length === 0) return { state: "empty" };
-  for (const member of liveMembers) {
-    signal?.throwIfAborted();
-    try {
-      const runId = (await host.environment(member.pid)).REA_PROCESS_RUN_ID;
-      signal?.throwIfAborted();
-      if (runId !== ownership.runId)
+  for await (const { process: member, observation } of readProcessRunTokens(
+    host,
+    liveMembers,
+    signal,
+  )) {
+    if (observation.state === "readable") {
+      if (observation.runId !== ownership.runId)
         return {
           state: "unverifiable",
           reason: "process ownership did not match",
         };
-    } catch (cause: unknown) {
-      signal?.throwIfAborted();
+    } else {
       if (!(await processIsGone(host, member.pid))) {
+        signal?.throwIfAborted();
         return {
           state: "unverifiable",
-          reason: `process ownership could not be revalidated for PID ${member.pid}: ${errorMessage(cause)}`,
+          reason: `process ownership could not be revalidated for PID ${member.pid}: ${observation.reason}`,
         };
       }
     }
   }
+  signal?.throwIfAborted();
   return { state: "alive" };
 };
 
@@ -88,18 +91,18 @@ export const observeOwnedProcessLineage = async (
     return unavailableLineage(ownership, identityFailure);
   const descendants = descendantsOf(launcher.pid, processes);
   const verifiedDescendants: ProcessTableEntry[] = [];
-  for (const member of [launcher, ...descendants]) {
-    try {
-      if (
-        (await host.environment(member.pid)).REA_PROCESS_RUN_ID !==
-        ownership.runId
-      )
+  for await (const { process: member, observation } of readProcessRunTokens(
+    host,
+    [launcher, ...descendants],
+  )) {
+    if (observation.state === "readable") {
+      if (observation.runId !== ownership.runId)
         return unavailableLineage(
           ownership,
           "process lineage contains an unowned or PID-reused process",
         );
       if (member.pid !== launcher.pid) verifiedDescendants.push(member);
-    } catch (cause: unknown) {
+    } else {
       if (await processIsGone(host, member.pid)) {
         if (member.pid === launcher.pid)
           return unavailableLineage(
@@ -110,7 +113,7 @@ export const observeOwnedProcessLineage = async (
       }
       return unavailableLineage(
         ownership,
-        `process ownership could not be revalidated for PID ${member.pid}: ${errorMessage(cause)}`,
+        `process ownership could not be revalidated for PID ${member.pid}: ${observation.reason}`,
       );
     }
   }
@@ -330,18 +333,21 @@ export const createSystemProcessOwnershipHost = (
       }
     },
     listProcesses,
-    async environment(pid) {
+    async environment(pid, signal) {
+      signal?.throwIfAborted();
       if (platform === "linux")
         return parseProcessEnvironment(
-          await readFile(`/proc/${pid}/environ`, "utf8"),
+          await readFile(`/proc/${pid}/environ`, { encoding: "utf8", signal }),
         );
       if (platform === "darwin") {
-        const process = (await listProcesses()).find(
+        const process = (await listProcesses(signal)).find(
           (entry) => entry.pid === pid,
         );
         if (process === undefined)
           throw new Error(`process ${String(pid)} is not live`);
-        const observation = (await darwinTokens?.read([process]))?.get(pid);
+        const observation = (await darwinTokens?.read([process], signal))?.get(
+          pid,
+        );
         if (observation === undefined || observation.state === "unavailable")
           throw new Error(
             observation?.reason ?? "process run token could not be read",
@@ -355,6 +361,7 @@ export const createSystemProcessOwnershipHost = (
         ["eww", "-p", String(pid)],
         {
           env: hostEnvironment,
+          ...(signal === undefined ? {} : { signal }),
         },
       );
       const observedEnvironment: Record<string, string> = {};
@@ -366,11 +373,13 @@ export const createSystemProcessOwnershipHost = (
       }
       return observedEnvironment;
     },
-    async runTokens(processes) {
+    async runTokens(processes, signal) {
+      signal?.throwIfAborted();
       if (platform !== "darwin" || processes.length === 0) return new Map();
       try {
-        return (await darwinTokens?.read(processes)) ?? new Map();
+        return (await darwinTokens?.read(processes, signal)) ?? new Map();
       } catch (cause: unknown) {
+        if (isExpectedAbort(cause, signal)) throw cause;
         void cause;
         return new Map(
           processes.map(({ pid }) => [

src/process/ProcessRunTokenObservations.ts

@@ -0,0 +1,62 @@
+import type {
+  ProcessOwnershipHost,
+  ProcessRunTokenObservation,
+  ProcessTableEntry,
+} from "./ProcessOwnership.js";
+
+const errorMessage = (cause: unknown): string =>
+  cause instanceof Error ? cause.message : String(cause);
+
+/** Read only ownership tokens, using per-process fallback for missing batch rows. */
+export async function* readProcessRunTokens(
+  host: ProcessOwnershipHost,
+  processes: readonly ProcessTableEntry[],
+  signal?: AbortSignal,
+): AsyncGenerator<{
+  readonly process: ProcessTableEntry;
+  readonly observation: ProcessRunTokenObservation;
+}> {
+  signal?.throwIfAborted();
+  if (processes.length === 0) return;
+  let batch: ReadonlyMap<number, ProcessRunTokenObservation> | undefined;
+  let batchFailure: string | undefined;
+  if (host.runTokens !== undefined) {
+    try {
+      batch =
+        signal === undefined
+          ? await host.runTokens(processes)
+          : await host.runTokens(processes, signal);
+    } catch (cause: unknown) {
+      signal?.throwIfAborted();
+      batchFailure = errorMessage(cause);
+    }
+  }
+  for (const process of processes) {
+    signal?.throwIfAborted();
+    let observation = batch?.get(process.pid);
+    if (observation === undefined) {
+      try {
+        const environment =
+          signal === undefined
+            ? await host.environment(process.pid)
+            : await host.environment(process.pid, signal);
+        observation = {
+          state: "readable",
+          runId: environment.REA_PROCESS_RUN_ID,
+        };
+      } catch (cause: unknown) {
+        signal?.throwIfAborted();
+        const reason = errorMessage(cause);
+        observation = {
+          state: "unavailable",
+          reason:
+            batchFailure === undefined
+              ? reason
+              : `${reason}; run-token batch failed: ${batchFailure}`,
+        };
+      }
+    }
+    signal?.throwIfAborted();
+    yield { process, observation };
+  }
+}

tests/boundary/process/darwinProcessRunTokenReader.test.ts

@@ -567,9 +567,9 @@ it.skipIf(process.platform === "win32")(
   },
 );
 
-it.skipIf(process.platform === "win32")(
-  "preserves cancellation through system-host baseline identity inspection",
-  async () => {
+it.skipIf(process.platform === "win32").each(["baseline", "tokens"] as const)(
+  "preserves cancellation through system-host %s inspection",
+  async (kind) => {
     const directory = await mkdtemp(
       join(tmpdir(), "rea-process-baseline-abort-test-"),
     );
@@ -579,8 +579,23 @@ it.skipIf(process.platform === "win32")(
     });
     const controller = new AbortController();
     let helperRoot: string | undefined;
+    const inspection =
+      kind === "baseline"
+        ? host.captureBaseline?.(controller.signal)
+        : host.runTokens?.(
+            [
+              {
+                pid: process.pid,
+                parentPid: process.ppid,
+                processGroupId: process.pid,
+                state: "S",
+                command: process.execPath,
+              },
+            ],
+            controller.signal,
+          );
     const baseline =
-      host.captureBaseline?.(controller.signal).then(
+      inspection?.then(
         () => ({ state: "fulfilled" as const }),
         (cause: unknown) => ({ state: "rejected" as const, cause }),
       ) ?? Promise.resolve({ state: "missing" as const });