diff --git a/src/process/ProcessOwnership.observation.test.ts b/src/process/ProcessOwnership.observation.test.ts index e29aa5965e..58085eb21a 100644 --- a/src/process/ProcessOwnership.observation.test.ts +++ b/src/process/ProcessOwnership.observation.test.ts @@ -4,7 +4,7 @@ import { type ProcessOwnershipHost, } from "./ProcessOwnership.js"; import { - observeOwnedProcessGroup, + observeOwnedProcessGroups, observeOwnedProcessLineage, } from "./ProcessOwnershipObservation.js"; import { host, ownership } from "./ProcessOwnership.fixture.js"; @@ -17,6 +17,158 @@ const processes = [100, 101].map((pid) => ({ command: "fixture", })); +describe("multi-group ownership observation", () => { + it("observes multiple groups from one process snapshot and token batch", async () => { + const rows = [ + ...processes, + { + pid: 200, + parentPid: 1, + processGroupId: 200, + state: "S", + command: "other-owned-group", + }, + { + pid: 300, + parentPid: 1, + processGroupId: 300, + state: "Z", + command: "[exited]", + }, + ]; + const listProcesses = vi.fn(() => Promise.resolve(rows)); + const environment = vi.fn(() => Promise.resolve({})); + const runTokens = vi.fn>( + (members) => + Promise.resolve( + new Map( + members.map(({ pid }) => [ + pid, + { + state: "readable" as const, + runId: pid === 200 ? "other-run" : ownership.runId, + }, + ]), + ), + ), + ); + + await expect( + observeOwnedProcessGroups(ownership.runId, [100, 200, 300], { + listProcesses, + environment, + runTokens, + signalGroup: vi.fn(), + }), + ).resolves.toEqual( + new Map([ + [100, { state: "alive" }], + [ + 200, + { state: "unverifiable", reason: "process ownership did not match" }, + ], + [300, { state: "empty" }], + ]), + ); + expect(listProcesses).toHaveBeenCalledTimes(1); + expect(runTokens).toHaveBeenCalledTimes(1); + expect(runTokens).toHaveBeenCalledWith( + rows.filter(({ state }) => state !== "Z"), + ); + expect(environment).not.toHaveBeenCalled(); + }); + + it("rechecks unavailable members individually before classifying their groups", async () => { + const rows = [ + ...processes, + { + pid: 200, + parentPid: 1, + processGroupId: 200, + state: "S", + command: "unreadable-group", + }, + ]; + const listProcesses = vi + .fn() + .mockResolvedValueOnce(rows) + .mockResolvedValueOnce(rows.filter(({ pid }) => pid !== 101)) + .mockResolvedValueOnce(rows); + const runTokens = vi.fn>( + (members) => + Promise.resolve( + new Map( + members.map(({ pid }) => [ + pid, + pid === 101 || pid === 200 + ? { + state: "unavailable" as const, + reason: `unreadable-${String(pid)}`, + } + : { state: "readable" as const, runId: ownership.runId }, + ]), + ), + ), + ); + + await expect( + observeOwnedProcessGroups(ownership.runId, [100, 200], { + listProcesses, + environment: vi.fn(() => Promise.resolve({})), + runTokens, + signalGroup: vi.fn(), + }), + ).resolves.toEqual( + new Map([ + [100, { state: "alive" }], + [ + 200, + { + state: "unverifiable", + reason: + "process ownership could not be revalidated for PID 200: unreadable-200", + }, + ], + ]), + ); + expect(listProcesses).toHaveBeenCalledTimes(3); + expect(runTokens).toHaveBeenCalledTimes(1); + }); +}); + +describe("multi-group observation short-circuit", () => { + it("stops missing-row fallbacks once every live group is unverifiable", async () => { + const listProcesses = vi.fn(() => Promise.resolve(processes)); + const environment = vi.fn((pid: number) => + Promise.resolve({ + REA_PROCESS_RUN_ID: pid === 100 ? "other-run" : ownership.runId, + }), + ); + const runTokens = vi.fn>( + () => Promise.resolve(new Map()), + ); + + await expect( + observeOwnedProcessGroups(ownership.runId, [100], { + listProcesses, + environment, + runTokens, + signalGroup: vi.fn(), + }), + ).resolves.toEqual( + new Map([ + [ + 100, + { state: "unverifiable", reason: "process ownership did not match" }, + ], + ]), + ); + expect(runTokens).toHaveBeenCalledTimes(1); + expect(environment.mock.calls).toEqual([[100]]); + expect(listProcesses).toHaveBeenCalledTimes(1); + }); +}); + describe("batched ownership observation", () => { it("observes group and lineage with one batch each and falls back only for missing rows", async () => { const { adapter } = host({ @@ -32,8 +184,14 @@ describe("batched ownership observation", () => { ); const observedHost = { ...adapter, environment, runTokens }; await expect( - observeOwnedProcessGroup(ownership, observedHost), - ).resolves.toEqual({ state: "alive" }); + observeOwnedProcessGroups( + ownership.runId, + [ownership.processGroupId], + observedHost, + ), + ).resolves.toEqual( + new Map([[ownership.processGroupId, { state: "alive" }]]), + ); await expect( observeOwnedProcessLineage(ownership, observedHost), ).resolves.toMatchObject({ @@ -67,8 +225,9 @@ describe("batched ownership observation", () => { }, ); await expect( - observeOwnedProcessGroup( - ownership, + observeOwnedProcessGroups( + ownership.runId, + [ownership.processGroupId], { listProcesses, environment, runTokens, signalGroup: vi.fn() }, controller.signal, ), diff --git a/src/process/ProcessOwnership.validation.test.ts b/src/process/ProcessOwnership.validation.test.ts index 2d6dd7275f..a516141736 100644 --- a/src/process/ProcessOwnership.validation.test.ts +++ b/src/process/ProcessOwnership.validation.test.ts @@ -6,7 +6,7 @@ import { type WindowsProcessTreeHost, } from "./ProcessOwnership.js"; import { - observeOwnedProcessGroup, + observeOwnedProcessGroups, observeOwnedProcessLineage, } from "./ProcessOwnershipObservation.js"; import { host, ownership } from "./ProcessOwnership.fixture.js"; @@ -160,7 +160,12 @@ describe("owned process-group cleanup validation: exited members", () => { }, }; await expect( - observeOwnedProcessGroup(ownership, adapter, controller.signal), + observeOwnedProcessGroups( + ownership.runId, + [ownership.processGroupId], + adapter, + controller.signal, + ), ).rejects.toBe(reason); }); it("does not inspect a process group after observation was cancelled", async () => { @@ -177,7 +182,12 @@ describe("owned process-group cleanup validation: exited members", () => { }, }; await expect( - observeOwnedProcessGroup(ownership, adapter, controller.signal), + observeOwnedProcessGroups( + ownership.runId, + [ownership.processGroupId], + adapter, + controller.signal, + ), ).rejects.toBe(reason); }); it("ignores exited zombie members during live ownership checks", async () => { @@ -229,9 +239,13 @@ describe("owned process-group cleanup validation: exited members", () => { environment, signalGroup: vi.fn(), }; - expect(await observeOwnedProcessGroup(ownership, adapter)).toEqual({ - state: "empty", - }); + expect( + await observeOwnedProcessGroups( + ownership.runId, + [ownership.processGroupId], + adapter, + ), + ).toEqual(new Map([[ownership.processGroupId, { state: "empty" }]])); expect(environment).not.toHaveBeenCalled(); }); it("is idempotent when the owned group has already exited", async () => { diff --git a/src/process/ProcessOwnershipObservation.ts b/src/process/ProcessOwnershipObservation.ts index 43fc65e52b..db5c7cb3a9 100644 --- a/src/process/ProcessOwnershipObservation.ts +++ b/src/process/ProcessOwnershipObservation.ts @@ -16,51 +16,78 @@ import type { ProcessTableEntry, } from "./ProcessOwnership.js"; -/** Observe one group without signaling it, failing closed on identity doubt. */ -export const observeOwnedProcessGroup = async ( - ownership: OwnedProcessGroup, +/** Observe multiple groups from one process snapshot and one token batch. */ +export const observeOwnedProcessGroups = async ( + runId: string, + processGroupIds: readonly number[], host: ProcessOwnershipHost = systemProcessOwnershipHost, signal?: AbortSignal, -): Promise => { +): Promise> => { signal?.throwIfAborted(); - let members: readonly ProcessTableEntry[]; + const groupIds = [...new Set(processGroupIds)]; + if (groupIds.length === 0) return new Map(); + let processes: readonly ProcessTableEntry[]; try { - members = (await host.listProcesses(signal)).filter( - ({ processGroupId }) => processGroupId === ownership.processGroupId, - ); + processes = liveProcesses(await host.listProcesses(signal)); } catch (cause: unknown) { signal?.throwIfAborted(); - return { - state: "unverifiable", + const failure = { + state: "unverifiable" as const, reason: `process group could not be inspected: ${errorMessage(cause)}`, }; + return new Map(groupIds.map((groupId) => [groupId, failure])); } signal?.throwIfAborted(); - const liveMembers = liveProcesses(members); - if (liveMembers.length === 0) return { state: "empty" }; + const membersByGroup = new Map( + groupIds.map((groupId) => [groupId, [] as ProcessTableEntry[]]), + ); + for (const process of processes) + membersByGroup.get(process.processGroupId)?.push(process); + const observations = new Map( + groupIds.map((groupId) => [ + groupId, + (membersByGroup.get(groupId)?.length ?? 0) > 0 + ? { state: "alive" } + : { state: "empty" }, + ]), + ); + const liveMembers = groupIds.flatMap( + (groupId) => membersByGroup.get(groupId) ?? [], + ); + const pendingGroups = new Set( + groupIds.filter( + (groupId) => (membersByGroup.get(groupId)?.length ?? 0) > 0, + ), + ); for await (const { process: member, observation } of readProcessRunTokens( host, liveMembers, signal, )) { + if (!pendingGroups.has(member.processGroupId)) continue; if (observation.state === "readable") { - if (observation.runId !== ownership.runId) - return { + if (observation.runId !== runId) { + observations.set(member.processGroupId, { state: "unverifiable", reason: "process ownership did not match", - }; + }); + pendingGroups.delete(member.processGroupId); + } } else { - if (!(await processIsGone(host, member.pid))) { - signal?.throwIfAborted(); - return { + const gone = await processIsGone(host, member.pid); + signal?.throwIfAborted(); + if (!gone) { + observations.set(member.processGroupId, { state: "unverifiable", reason: `process ownership could not be revalidated for PID ${member.pid}: ${observation.reason}`, - }; + }); + pendingGroups.delete(member.processGroupId); } } + if (pendingGroups.size === 0) break; } signal?.throwIfAborted(); - return { state: "alive" }; + return observations; }; /** diff --git a/src/process/capture/ProcessCaptureLifecycle.ts b/src/process/capture/ProcessCaptureLifecycle.ts index 19ac0173a9..e4e59d1716 100644 --- a/src/process/capture/ProcessCaptureLifecycle.ts +++ b/src/process/capture/ProcessCaptureLifecycle.ts @@ -25,7 +25,7 @@ import { type ProcessOwnershipBaseline, } from "../ProcessOwnership.js"; import { - observeOwnedProcessGroup, + observeOwnedProcessGroups, prepareProcessOwnershipInspection, systemProcessOwnershipHost, } from "../ProcessOwnershipObservation.js"; @@ -616,21 +616,20 @@ export const observeSettlement = async ( while (!deadlineReached) { onLiveProgress?.(); assertNotCancelled(signal); - const observations = await Promise.all( - [...new Set(processGroupIds)].map((processGroupId) => - observeOwnedProcessGroup( - { runId, leaderPid: processGroupId, processGroupId }, - undefined, - signal, - ), - ), + const observations = await observeOwnedProcessGroups( + runId, + processGroupIds, + undefined, + signal, ); assertNotCancelled(signal); - if (observations.some(({ state }) => state === "unverifiable")) { + if ( + [...observations.values()].some(({ state }) => state === "unverifiable") + ) { recordEvent("lifecycle", 1); return { state: "unverifiable", elapsed_ms: Date.now() - started }; } - if (observations.every(({ state }) => state === "empty")) { + if ([...observations.values()].every(({ state }) => state === "empty")) { consecutiveEmpty += 1; if (consecutiveEmpty >= 2) { recordEvent("lifecycle", 1);