diff --git a/docs/process-capture.md b/docs/process-capture.md index 931135baf0..0c4871171a 100644 --- a/docs/process-capture.md +++ b/docs/process-capture.md @@ -428,6 +428,17 @@ Every capture requires `truncation_details`, with separate accounting for: imply incomplete path enumeration. - `process`: sampling limit and whether sampling ended partially. Coverage remains `sampled`, including when that limit was not exhausted. + `sampling_failures` counts failed read observations plus at most one + unfinished pass interrupted by sampler shutdown. `first_sampling_failure` + retains the first attributed read diagnostic, with shutdown context appended + when applicable; interruption alone has an explicit shutdown diagnostic. + Denied or failed procfs reads mark + sampling partial while preserving readable process rows and thread children; + a process with unreadable identity is not attributed. Expected process-exit + races are separate from failed observations. Shutdown counts only when it + interrupts a read or leaves traversal unfinished; a completed pass does not + become partial merely because the sampler stops. Completed rows + remain available and issued reads settle before final coverage is returned. The aggregate `truncated` flag summarizes these observations. Comparisons retain results for unaffected dimensions and mark affected dimensions unknown; an diff --git a/src/domain/process/processCapture.validation.test.ts b/src/domain/process/processCapture.validation.test.ts index 4fb6333be3..ac2e9a7ce7 100644 --- a/src/domain/process/processCapture.validation.test.ts +++ b/src/domain/process/processCapture.validation.test.ts @@ -45,6 +45,71 @@ const captureWithEnumerationOmission = ( }; }; +const captureWithSamplingInterruption = () => { + const capture = emptyCapture(); + return { + ...capture, + truncated: true, + truncation_details: { + ...capture.truncation_details, + process: { + ...capture.truncation_details.process, + sampling_partial: true, + sampling_failures: 1, + first_sampling_failure: "Process sampling interrupted by shutdown", + }, + }, + }; +}; + +it("requires interrupted sampling diagnostics to agree with partial and aggregate truncation", () => { + const interrupted = captureWithSamplingInterruption(); + expect(parseProcessCapture(interrupted)).toEqual(interrupted); + expect(() => + parseProcessCapture({ + ...interrupted, + truncated: false, + }), + ).toThrow("truncation_details"); + expect(() => + parseProcessCapture({ + ...interrupted, + truncation_details: { + ...interrupted.truncation_details, + process: { + ...interrupted.truncation_details.process, + sampling_partial: false, + }, + }, + }), + ).toThrow("truncation_details.process"); + expect(() => + parseProcessCapture({ + ...interrupted, + truncation_details: { + ...interrupted.truncation_details, + process: { + ...interrupted.truncation_details.process, + first_sampling_failure: null, + }, + }, + }), + ).toThrow("truncation_details.process"); +}); + +it("limits interruption impact to process comparison", () => { + const baseline = emptyCapture(); + const interrupted = captureWithSamplingInterruption(); + expect(compareProcessCaptures(interrupted, baseline)).toMatchObject({ + status: "unknown", + terminal: "unchanged", + interaction: "unchanged", + exit: "unchanged", + filesystem: "unchanged", + process: "unknown", + }); +}); + it.each([ { path: "root_0:.", reason: "entry_unavailable", system_code: "ENOENT" }, { path: "root_0:.", reason: "identity_changed", system_code: null }, diff --git a/src/domain/process/processCaptureCoverage.ts b/src/domain/process/processCaptureCoverage.ts index cba5b47660..cf61ed0992 100644 --- a/src/domain/process/processCaptureCoverage.ts +++ b/src/domain/process/processCaptureCoverage.ts @@ -115,7 +115,7 @@ export const hasCaptureTruncation = ( /** Whether producer accounting reports a gap in the selected observation source. */ export const processSourceTruncated = ( - capture: UnverifiedProcessCapture, + capture: Pick, source: ProcessObservationSource, ): boolean => { const details = capture.truncation_details; diff --git a/src/process/capture/LinuxProcessSampling.ts b/src/process/capture/LinuxProcessSampling.ts new file mode 100644 index 0000000000..67ba0fcad9 --- /dev/null +++ b/src/process/capture/LinuxProcessSampling.ts @@ -0,0 +1,280 @@ +import { readFile, readdir } from "node:fs/promises"; +import { + mergeSamplingDiagnostics, + readProcSamplingSource, + samplingFailure, + samplingObservation, + samplingInterrupted, + type SamplingObservation, +} from "./ProcessSamplingObservation.js"; + +export const PROC_READ_CONCURRENCY = 64; + +export interface SampledProcessRow { + readonly pid: number; + readonly parent_pid: number; + readonly process_group_id: number | null; + readonly session_id: number | null; + readonly command: string; + readonly startTime: string | undefined; +} + +export interface SampledProcessNode extends SampledProcessRow { + readonly children: readonly number[] | undefined; +} + +const parseProcStat = ( + pid: number, + stat: string, +): SampledProcessRow | undefined => { + const commandStart = stat.indexOf("("); + const commandEnd = stat.lastIndexOf(")"); + if ( + commandStart < 0 || + commandEnd <= commandStart || + !/^\d+$/u.test(stat.slice(0, commandStart).trim()) || + Number(stat.slice(0, commandStart).trim()) !== pid + ) + return undefined; + const fields = stat + .slice(commandEnd + 2) + .trim() + .split(/\s+/u); + const parentPid = Number(fields[1]); + const processGroupId = Number(fields[2]); + const sessionId = Number(fields[3]); + const startTime = fields[19]; + if ( + !Number.isSafeInteger(pid) || + pid <= 0 || + !Number.isSafeInteger(parentPid) || + parentPid < 0 || + !Number.isSafeInteger(processGroupId) || + processGroupId <= 0 || + !Number.isSafeInteger(sessionId) || + sessionId <= 0 || + startTime === undefined || + !/^\d+$/u.test(startTime) + ) + return undefined; + return { + pid, + parent_pid: parentPid, + process_group_id: processGroupId, + session_id: sessionId, + command: "", + startTime, + }; +}; + +export const readLinuxProcessStat = async ( + pid: number, + signal: AbortSignal, +): Promise> => { + const path = `/proc/${String(pid)}/stat`; + const read = await readProcSamplingSource(path, signal, () => + readFile(path, { encoding: "utf8", signal }), + ); + if (read.value === undefined) + return samplingObservation(undefined, read.diagnostics); + const row = parseProcStat(pid, read.value); + return samplingObservation( + row, + read.diagnostics, + ...(row === undefined + ? [samplingFailure(`${path}: malformed process identity`)] + : []), + ); +}; + +/** Procfs operations remain injectable below error classification and composition. */ +export interface LinuxChildrenHost { + taskIds(pid: number, signal: AbortSignal): Promise; + children(pid: number, taskId: number, signal: AbortSignal): Promise; +} + +const systemLinuxChildrenHost: LinuxChildrenHost = { + async taskIds(pid) { + const entries = await readdir(`/proc/${String(pid)}/task`); + return entries + .filter((entry) => /^\d+$/u.test(entry)) + .map(Number) + .filter((taskId) => Number.isSafeInteger(taskId) && taskId > 0) + .sort((left, right) => left - right); + }, + children(pid, taskId, signal) { + return readFile(`/proc/${String(pid)}/task/${String(taskId)}/children`, { + encoding: "utf8", + signal, + }); + }, +}; + +/** Keep the child union and diagnostics from independently readable process threads. */ +export const readLinuxChildren = async ( + pid: number, + signal: AbortSignal, + host: LinuxChildrenHost = systemLinuxChildrenHost, +): Promise> => { + const tasks = await readProcSamplingSource( + `/proc/${String(pid)}/task`, + signal, + () => host.taskIds(pid, signal), + ); + if (tasks.value === undefined) + return samplingObservation(undefined, tasks.diagnostics); + if (tasks.value.length === 0) + return samplingObservation([], tasks.diagnostics); + const children = new Set(); + let diagnostics = tasks.diagnostics; + let successfulReads = 0; + let offset = 0; + for ( + ; + offset < tasks.value.length && !signal.aborted; + offset += PROC_READ_CONCURRENCY + ) { + const texts = await Promise.all( + tasks.value + .slice(offset, offset + PROC_READ_CONCURRENCY) + .map((taskId) => + readProcSamplingSource( + `/proc/${String(pid)}/task/${String(taskId)}/children`, + signal, + () => host.children(pid, taskId, signal), + ), + ), + ); + for (const text of texts) { + diagnostics = mergeSamplingDiagnostics(diagnostics, text.diagnostics); + if (text.value === undefined) continue; + successfulReads += 1; + for (const value of text.value.trim().split(/\s+/u)) { + if (!/^\d+$/u.test(value)) continue; + const child = Number(value); + if (Number.isSafeInteger(child) && child > 0) children.add(child); + } + } + } + if (offset < tasks.value.length && signal.aborted) + diagnostics = mergeSamplingDiagnostics(diagnostics, samplingInterrupted()); + return samplingObservation( + successfulReads === 0 ? undefined : [...children].sort((a, b) => a - b), + diagnostics, + ); +}; + +/** Identity brackets own attribution for both procfs and ps-discovered candidates. */ +const inspectStableLinuxProcess = async ( + pid: number, + expectedParent: number | undefined, + signal: AbortSignal, + identities: ReadonlyMap, + readDetails: () => Promise>, +): Promise< + SamplingObservation< + { readonly row: SampledProcessRow; readonly details: T } | undefined + > +> => { + const before = await readLinuxProcessStat(pid, signal); + if (before.value === undefined || before.value.startTime === undefined) + return samplingObservation(undefined, before.diagnostics); + const existing = identities.get(pid); + if ( + (existing !== undefined && existing !== before.value.startTime) || + (expectedParent !== undefined && before.value.parent_pid !== expectedParent) + ) + return samplingObservation(undefined, before.diagnostics); + const details = await readDetails(); + const after = await readLinuxProcessStat(pid, signal); + const diagnostics = mergeSamplingDiagnostics( + before.diagnostics, + details.diagnostics, + after.diagnostics, + ); + if ( + after.value === undefined || + before.value.startTime !== after.value.startTime || + before.value.parent_pid !== after.value.parent_pid || + (expectedParent !== undefined && after.value.parent_pid !== expectedParent) + ) + return samplingObservation(undefined, diagnostics); + return samplingObservation( + { row: before.value, details: details.value }, + diagnostics, + ); +}; + +const readLinuxCommand = async ( + pid: number, + signal: AbortSignal, +): Promise> => { + const path = `/proc/${String(pid)}/cmdline`; + const command = await readProcSamplingSource(path, signal, () => + readFile(path, { signal }), + ); + return samplingObservation( + command.value?.toString().replaceAll("\0", " ").trim() ?? "", + command.diagnostics, + ); +}; + +/** Inspect a ps-discovered PID without trusting ps for Linux identity or command. */ +export const inspectLinuxProcessMetadata = async ( + pid: number, + expectedParent: number, + signal: AbortSignal, + identities: ReadonlyMap, +): Promise> => { + const observed = await inspectStableLinuxProcess( + pid, + expectedParent, + signal, + identities, + () => readLinuxCommand(pid, signal), + ); + return samplingObservation( + observed.value === undefined + ? undefined + : { + ...observed.value.row, + command: observed.value.details, + }, + observed.diagnostics, + ); +}; + +/** Failed optional reads qualify a stable row instead of erasing it or its peers. */ +export const inspectLinuxProcess = async ( + pid: number, + expectedParent: number | undefined, + signal: AbortSignal, + identities: ReadonlyMap, +): Promise> => { + const observed = await inspectStableLinuxProcess( + pid, + expectedParent, + signal, + identities, + async () => { + const [command, children] = await Promise.all([ + readLinuxCommand(pid, signal), + readLinuxChildren(pid, signal), + ]); + return samplingObservation( + { command: command.value, children: children.value }, + command.diagnostics, + children.diagnostics, + ); + }, + ); + return samplingObservation( + observed.value === undefined + ? undefined + : { + ...observed.value.row, + ...observed.value.details, + }, + observed.diagnostics, + ); +}; diff --git a/src/process/capture/ProcessCaptureLifecycle.ts b/src/process/capture/ProcessCaptureLifecycle.ts index a3538a91af..155a8dfd8d 100644 --- a/src/process/capture/ProcessCaptureLifecycle.ts +++ b/src/process/capture/ProcessCaptureLifecycle.ts @@ -459,7 +459,9 @@ const captureCoverageUnknowns = ( unknowns.push({ scope: "process", reason: - "Process sampling stopped with incomplete observations; sample_limit is reported in truncation_details.process.", + details.process.first_sampling_failure !== null + ? `Process sampling has incomplete observations: ${details.process.first_sampling_failure}. See truncation_details.process.` + : "The process sample limit was reached; see truncation_details.process.", }); return unknowns; }; diff --git a/src/process/capture/ProcessCaptureObservation.test.ts b/src/process/capture/ProcessCaptureObservation.test.ts index 51ebe20fa9..ace89ca938 100644 --- a/src/process/capture/ProcessCaptureObservation.test.ts +++ b/src/process/capture/ProcessCaptureObservation.test.ts @@ -2,7 +2,8 @@ import { expect, it } from "vitest"; import { snapshotRoots } from "./FilesystemSnapshot.js"; import { buildCaptureResult } from "./ProcessCaptureLifecycle.js"; -import { isInitializedPtyRoot, readLinuxChildren } from "./ProcessSampling.js"; +import { isInitializedPtyRoot } from "./ProcessSampling.js"; +import { readLinuxChildren } from "./LinuxProcessSampling.js"; import { TerminalRenderer } from "./TerminalRenderer.js"; import { parseProcessScenario } from "../../domain/process/processScenario.js"; import type { ProcessCapture } from "../../domain/process/processCaptureParsing.js"; @@ -215,7 +216,10 @@ it("collects and deduplicates children from every Linux thread", async () => { taskId === 100 ? "201 202" : taskId === 101 ? "202 203" : "", ), }), - ).toEqual([201, 202, 203]); + ).toEqual({ + value: [201, 202, 203], + diagnostics: { count: 0, first: null, interrupted: false }, + }); }); it("admits PTY samples only after stable session and token setup", () => { @@ -275,5 +279,9 @@ it("cancels filesystem snapshots before traversing declared roots", async () => }), controller.signal, ), - ).rejects.toMatchObject({ name: "AbortError" }); + ).rejects.toMatchObject({ + reason: "cancelled", + cause: controller.signal.reason, + snapshot: { files: [], truncated: true }, + }); }); diff --git a/src/process/capture/ProcessSampling.procReadCoverage.test.ts b/src/process/capture/ProcessSampling.procReadCoverage.test.ts new file mode 100644 index 0000000000..9d6ed6849a --- /dev/null +++ b/src/process/capture/ProcessSampling.procReadCoverage.test.ts @@ -0,0 +1,568 @@ +import { afterAll, beforeAll, expect, it, onTestFinished, vi } from "vitest"; + +const fixture = vi.hoisted(() => ({ + failures: new Map(), + abortErrors: new Set(), + reads: [] as string[], + tasks: new Map(), + children: new Map(), + holdPath: undefined as string | undefined, + holdStarted: false, + holdStartedResolve: undefined as (() => void) | undefined, + abortObserved: false, + abortObservedResolve: undefined as (() => void) | undefined, + holdReaddirPath: undefined as string | undefined, + readdirGate: undefined as Promise | undefined, + readdirStarted: false, + readdirStartedResolve: undefined as (() => void) | undefined, + readdirCount: 0, + psFailure: undefined as string | undefined, + psOutput: undefined as string | undefined, + psReads: 0, + statWaitPid: undefined as number | undefined, + statWaitCount: 0, + statWaitResolve: undefined as (() => void) | undefined, + statReads: new Map(), + statMutations: new Map< + number, + { + readonly read: number; + readonly pid?: number; + readonly parent?: number; + readonly start?: number; + } + >(), + runId: "fixture-run-token", + runIdFailure: undefined as string | undefined, + runIdReads: 0, +})); + +vi.mock("node:fs/promises", async (importOriginal) => { + const filesystem = await importOriginal(); + return { + ...filesystem, + readFile: async (...args: Parameters) => { + const [path, options] = args; + const filename = String(path); + if (!filename.startsWith("/proc/")) return filesystem.readFile(...args); + fixture.reads.push(filename); + const code = fixture.failures.get(filename); + if (code !== undefined) throw Object.assign(new Error(code), { code }); + if (fixture.abortErrors.has(filename)) { + const error = new Error("unexpected abort-shaped filesystem error"); + error.name = "AbortError"; + throw error; + } + if (filename === fixture.holdPath) { + fixture.holdStarted = true; + fixture.holdStartedResolve?.(); + const signal = + typeof options === "object" && options !== null && "signal" in options + ? options.signal + : undefined; + await new Promise((_resolve, reject) => { + if (signal?.aborted) { + const error = Object.assign(new Error("sampling cancelled"), { + cause: signal.reason, + }); + error.name = "AbortError"; + reject(error); + return; + } + signal?.addEventListener( + "abort", + () => { + fixture.abortObserved = true; + fixture.abortObservedResolve?.(); + const error = Object.assign(new Error("sampling cancelled"), { + cause: signal?.reason, + }); + error.name = "AbortError"; + reject(error); + }, + { once: true }, + ); + }); + } + const statMatch = /^\/proc\/(\d+)\/stat$/u.exec(filename); + let value: string | Buffer; + if (statMatch !== null) { + const pid = Number(statMatch[1]); + const reads = (fixture.statReads.get(pid) ?? 0) + 1; + fixture.statReads.set(pid, reads); + if (pid === fixture.statWaitPid && reads >= fixture.statWaitCount) + fixture.statWaitResolve?.(); + value = statLine(pid, reads); + } else { + const childMatch = /^\/proc\/(\d+)\/task\/(\d+)\/children$/u.exec( + filename, + ); + if (childMatch !== null) + value = + fixture.children.get(`${childMatch[1]}:${childMatch[2]}`) ?? ""; + else if (/^\/proc\/\d+\/cmdline$/u.test(filename)) + value = Buffer.from(`fixture-command-${filename.split("/")[2]}\0`); + else throw new Error(`Unexpected proc fixture read: ${filename}`); + } + if ( + typeof options === "object" && + options !== null && + "encoding" in options && + options.encoding === "utf8" + ) + return typeof value === "string" ? value : value.toString("utf8"); + return Buffer.isBuffer(value) ? value : Buffer.from(value); + }, + readdir: async (...args: Parameters) => { + const [path] = args; + const filename = String(path); + if (!filename.startsWith("/proc/")) return filesystem.readdir(...args); + fixture.reads.push(filename); + if (filename === fixture.holdReaddirPath) { + fixture.readdirStarted = true; + fixture.readdirCount += 1; + fixture.readdirStartedResolve?.(); + await fixture.readdirGate; + } + const code = fixture.failures.get(filename); + if (code !== undefined) throw Object.assign(new Error(code), { code }); + if (fixture.abortErrors.has(filename)) { + const error = new Error("unexpected abort-shaped filesystem error"); + error.name = "AbortError"; + throw error; + } + const taskMatch = /^\/proc\/(\d+)\/task$/u.exec(filename); + if (taskMatch === null) + throw new Error(`Unexpected proc fixture directory: ${filename}`); + const pid = taskMatch[1]; + if (pid === undefined) + throw new Error(`Missing task pid in fixture path: ${filename}`); + return fixture.tasks.get(Number(pid)) ?? [pid]; + }, + }; +}); + +vi.mock("../ProcessOwnership.js", () => ({ + readProcessRunId: async () => { + fixture.runIdReads += 1; + if (fixture.runIdFailure !== undefined) + throw Object.assign(new Error(fixture.runIdFailure), { + code: fixture.runIdFailure, + }); + return fixture.runId; + }, +})); + +vi.mock("../ExecFileOutput.js", () => ({ + execFileOutput: async () => { + fixture.psReads += 1; + if (fixture.psFailure !== undefined) + throw Object.assign(new Error(fixture.psFailure), { + code: fixture.psFailure, + }); + return { stdout: fixture.psOutput ?? psFixture() }; + }, +})); + +import { startProcessSampler } from "./ProcessSampling.js"; +import type { ProcessSample } from "../../domain/process/processCapture.js"; + +const platformDescriptor = Object.getOwnPropertyDescriptor(process, "platform"); +beforeAll(() => { + Object.defineProperty(process, "platform", { + configurable: true, + value: "linux", + }); +}); +afterAll(() => { + if (platformDescriptor !== undefined) + Object.defineProperty(process, "platform", platformDescriptor); +}); + +const processRows = [ + { pid: 100, parent: 1, start: 1100 }, + { pid: 200, parent: 100, start: 1200 }, + { pid: 300, parent: 200, start: 1300 }, + { pid: 400, parent: 100, start: 1400 }, +] as const; + +const statLine = (pid: number, read: number): string => { + const row = processRows.find((candidate) => candidate.pid === pid); + if (row === undefined) throw new Error(`Unknown fixture pid ${String(pid)}`); + const mutation = fixture.statMutations.get(pid); + const fields = [ + "S", + String( + mutation?.read === read ? (mutation.parent ?? row.parent) : row.parent, + ), + "100", + "100", + ]; + while (fields.length < 19) fields.push("0"); + fields.push( + String(mutation?.read === read ? (mutation.start ?? row.start) : row.start), + ); + const reportedPid = mutation?.read === read ? (mutation.pid ?? pid) : pid; + return `${String(reportedPid)} (fixture-${String(pid)}) ${fields.join(" ")}`; +}; + +const psFixture = (): string => + processRows + .map( + ({ pid, parent }) => + `${String(pid)} ${String(parent)} 100 100 fixture-command-${String(pid)}`, + ) + .join("\n"); + +const setUpFixture = (): void => { + fixture.failures.clear(); + fixture.abortErrors.clear(); + fixture.reads.length = 0; + fixture.tasks.clear(); + fixture.children.clear(); + fixture.holdPath = undefined; + fixture.holdStarted = false; + fixture.holdStartedResolve = undefined; + fixture.abortObserved = false; + fixture.abortObservedResolve = undefined; + fixture.holdReaddirPath = undefined; + fixture.readdirGate = undefined; + fixture.readdirStarted = false; + fixture.readdirStartedResolve = undefined; + fixture.readdirCount = 0; + fixture.psFailure = undefined; + fixture.psOutput = undefined; + fixture.psReads = 0; + fixture.statWaitPid = undefined; + fixture.statWaitCount = 0; + fixture.statWaitResolve = undefined; + fixture.statReads.clear(); + fixture.statMutations.clear(); + fixture.runIdFailure = undefined; + fixture.runIdReads = 0; + for (const pid of [100, 200, 300, 400]) fixture.tasks.set(pid, [String(pid)]); + fixture.children.set("100:100", "200 400"); + fixture.children.set("200:200", "300"); +}; + +const start = () => { + const samples: ProcessSample[] = []; + const stop = startProcessSampler({ + rootPid: 100, + runId: fixture.runId, + started: 0, + limit: 20, + samples, + }); + onTestFinished(async () => { + await stop(); + }); + return { samples, stop }; +}; + +const waitFor = async (condition: () => boolean): Promise => { + const deadline = Date.now() + 3_000; + while (!condition()) { + if (Date.now() >= deadline) + throw new Error("proc fixture condition timed out"); + await new Promise((resolve) => setTimeout(resolve, 1)); + } +}; + +const flushImmediateProcPass = async (): Promise => { + await new Promise((resolve) => setTimeout(resolve, 0)); + await Promise.resolve(); +}; + +const deferred = (): { readonly promise: Promise; resolve(): void } => { + let resolve!: () => void; + const promise = new Promise((done) => { + resolve = done; + }); + return { promise, resolve }; +}; + +const expectPartialFailure = async ( + path: string, + code: string, + expectedPids: readonly number[], +): Promise => { + setUpFixture(); + fixture.failures.set(path, code); + const { samples, stop } = start(); + await waitFor(() => fixture.reads.includes(path)); + await flushImmediateProcPass(); + const result = await stop(); + expect(samples.map(({ pid }) => pid)).toEqual( + expect.arrayContaining([...expectedPids]), + ); + expect(result.partial).toBe(true); + expect(result.coverage.sampling_failures).toBeGreaterThan(0); + expect(result.coverage.first_sampling_failure).toMatch(code); + expect(result.coverage.first_sampling_failure).toContain(path); +}; + +it.each([ + ["/proc/200/stat", "EACCES", [100, 400]], + ["/proc/200/stat", "EIO", [100, 400]], + ["/proc/200/cmdline", "EPERM", [100, 200, 300, 400]], + ["/proc/100/task", "EIO", [100]], + ["/proc/100/task/100/children", "EACCES", [100, 200, 300, 400]], + ["/proc/200/task/200/children", "EPERM", [100, 200, 400]], +] as const)( + "retains readable process rows and reports %s %s as partial", + async (path, code, expectedPids) => { + await expectPartialFailure(path, code, expectedPids); + }, +); + +it("retains only fully bracketed rows when stop interrupts a held child read", async () => { + const childReadStarted = deferred(); + const siblingBracketed = deferred(); + setUpFixture(); + fixture.holdPath = "/proc/200/cmdline"; + fixture.holdStartedResolve = childReadStarted.resolve; + fixture.statWaitPid = 400; + fixture.statWaitCount = 2; + fixture.statWaitResolve = siblingBracketed.resolve; + const { samples, stop } = start(); + await Promise.all([childReadStarted.promise, siblingBracketed.promise]); + await Promise.resolve(); + await Promise.resolve(); + expect(fixture.statReads.get(400)).toBe(2); + + const result = await stop(); + expect(samples.map(({ pid }) => pid)).toEqual( + expect.arrayContaining([100, 400]), + ); + expect(samples.some(({ pid }) => pid === 200 || pid === 300)).toBe(false); + expect(result.partial).toBe(true); + expect(result.coverage.sampling_failures).toBe(1); + expect(result.coverage.first_sampling_failure).toMatch(/interrupted/iu); +}); + +it("retains only the stable procfs root if it changes during ps fallback discovery", async () => { + setUpFixture(); + fixture.failures.set("/proc/100/task/100/children", "EACCES"); + fixture.statMutations.set(100, { read: 5, start: 9999 }); + fixture.psOutput = [ + "100 1 100 100 ps-root-command", + "200 100 100 100 discovery-only-child-command", + "300 200 100 100 discovery-only-grandchild-command", + "400 100 100 100 discovery-only-sibling-command", + ].join("\n"); + const { samples, stop } = start(); + await waitFor(() => fixture.psReads > 0); + await waitFor( + () => + fixture.statReads.get(100) === 5 && + samples.some((sample) => sample.pid === 100), + ); + const result = await stop(); + expect(samples.map(({ pid }) => pid)).toEqual([100]); + expect(samples[0]?.command).toBe("fixture-command-100"); + expect(result.partial).toBe(true); + expect(result.coverage.sampling_failures).toBeGreaterThan(0); +}); + +it("rejects a ps child edge when procfs reports a different parent", async () => { + setUpFixture(); + fixture.failures.set("/proc/100/task/100/children", "EACCES"); + fixture.statMutations.set(200, { read: 1, parent: 201 }); + fixture.psOutput = [ + "100 1 100 100 ps-root-command", + "200 100 100 100 discovery-only-child-command", + "300 200 100 100 discovery-only-grandchild-command", + "400 100 100 100 discovery-only-sibling-command", + ].join("\n"); + const { samples, stop } = start(); + await waitFor( + () => + samples.some((sample) => sample.pid === 100) && + samples.some((sample) => sample.pid === 400), + ); + const result = await stop(); + expect(samples.map(({ pid }) => pid)).toEqual([100, 400]); + expect(samples.map(({ command }) => command)).toEqual([ + "fixture-command-100", + "fixture-command-400", + ]); + expect(result.partial).toBe(true); +}); + +it("retains children from readable threads while reporting another thread's EIO", async () => { + setUpFixture(); + fixture.tasks.set(100, ["100", "101"]); + fixture.children.set("100:101", ""); + fixture.failures.set("/proc/100/task/101/children", "EIO"); + const { samples, stop } = start(); + await waitFor(() => fixture.reads.includes("/proc/100/task/101/children")); + await flushImmediateProcPass(); + const result = await stop(); + expect(samples.map(({ pid }) => pid)).toEqual( + expect.arrayContaining([100, 200, 300, 400]), + ); + expect(result.partial).toBe(true); + expect(result.coverage.first_sampling_failure).toContain("EIO"); + expect(result.coverage.first_sampling_failure).toContain( + "/proc/100/task/101/children", + ); +}); + +it.each(["ENOENT", "ESRCH"] as const)( + "treats %s as an expected process disappearance rather than a sampling failure", + async (code) => { + setUpFixture(); + fixture.failures.set("/proc/200/stat", code); + const { samples, stop } = start(); + await waitFor(() => samples.some((row) => row.pid === 400)); + const result = await stop(); + expect(samples.map(({ pid }) => pid)).toEqual( + expect.arrayContaining([100, 400]), + ); + expect(result.partial).toBe(false); + expect(result.coverage.sampling_failures).toBe(0); + expect(result.coverage.first_sampling_failure).toBeNull(); + }, +); + +it.each(["EACCES", "EIO"] as const)( + "records root run-token inspection failure %s as incomplete sampling", + async (code) => { + setUpFixture(); + fixture.runIdFailure = code; + const { samples, stop } = start(); + await waitFor(() => fixture.runIdReads > 0); + await flushImmediateProcPass(); + const result = await stop(); + expect(samples).toEqual([]); + expect(result.partial).toBe(true); + expect(result.coverage.sampling_failures).toBeGreaterThan(0); + expect(result.coverage.first_sampling_failure).toContain(code); + expect(result.coverage.first_sampling_failure).toContain("100"); + }, +); + +it("does not swallow an unrelated filesystem AbortError as sampler cancellation", async () => { + setUpFixture(); + fixture.abortErrors.add("/proc/200/stat"); + const { samples, stop } = start(); + await waitFor(() => fixture.reads.includes("/proc/200/stat")); + await flushImmediateProcPass(); + const result = await stop(); + expect(samples.map(({ pid }) => pid)).toEqual( + expect.arrayContaining([100, 400]), + ); + expect(result.partial).toBe(true); + expect(result.coverage.sampling_failures).toBeGreaterThan(0); + expect(result.coverage.first_sampling_failure).toContain("AbortError"); + expect(result.coverage.first_sampling_failure).toContain("/proc/200/stat"); +}); + +it("does not report sampler shutdown cancellation as a proc-read failure", async () => { + setUpFixture(); + fixture.holdPath = "/proc/100/task/100/children"; + const { stop } = start(); + await waitFor(() => fixture.holdStarted); + const result = await stop(); + expect(result.partial).toBe(true); + expect(result.coverage.sampling_failures).toBe(1); + expect(result.coverage.first_sampling_failure).toMatch(/interrupted/iu); +}); + +it("awaits issued proc reads through stop, retaining a late I/O failure but excluding the matching abort", async () => { + vi.useFakeTimers(); + const readdirGate = deferred(); + const readdirStarted = deferred(); + const commandStarted = deferred(); + const commandAborted = deferred(); + const releaseReaddir = readdirGate.resolve; + let stopSampling: ReturnType["stop"] | undefined; + try { + setUpFixture(); + fixture.holdReaddirPath = "/proc/100/task"; + fixture.readdirGate = readdirGate.promise; + fixture.readdirStartedResolve = readdirStarted.resolve; + fixture.holdPath = "/proc/100/cmdline"; + fixture.holdStartedResolve = commandStarted.resolve; + fixture.abortObservedResolve = commandAborted.resolve; + const { stop } = start(); + stopSampling = stop; + await Promise.all([readdirStarted.promise, commandStarted.promise]); + + await vi.advanceTimersByTimeAsync(150); + expect(fixture.readdirCount).toBe(1); + expect(fixture.runIdReads).toBe(1); + + let stopSettled = false; + const stopping = stop().then((result) => { + stopSettled = true; + return result; + }); + await commandAborted.promise; + expect(stopSettled).toBe(false); + fixture.failures.set("/proc/100/task", "EIO"); + releaseReaddir(); + const result = await stopping; + expect(fixture.readdirCount).toBe(1); + expect(result.partial).toBe(true); + expect(result.coverage.sampling_failures).toBe(2); + expect(result.coverage.first_sampling_failure).toContain("EIO"); + expect(result.coverage.first_sampling_failure).toContain("/proc/100/task"); + expect(result.coverage.first_sampling_failure).toMatch(/interrupted/iu); + expect(result.coverage.first_sampling_failure).not.toContain("AbortError"); + } finally { + releaseReaddir(); + await stopSampling?.(); + vi.useRealTimers(); + } +}); + +it("retains a stable root row when child enumeration and its ps fallback both fail", async () => { + setUpFixture(); + fixture.tasks.set(100, ["100", "101"]); + fixture.failures.set("/proc/100/task/100/children", "EACCES"); + fixture.failures.set("/proc/100/task/101/children", "EIO"); + fixture.psFailure = "EIO"; + const { samples, stop } = start(); + await waitFor(() => fixture.reads.includes("/proc/100/task/101/children")); + await flushImmediateProcPass(); + const result = await stop(); + expect(samples.map(({ pid }) => pid)).toContain(100); + expect(result.partial).toBe(true); + expect(result.coverage.sampling_failures).toBeGreaterThanOrEqual(3); + expect(result.coverage.first_sampling_failure).toContain( + "/proc/100/task/100/children", + ); +}); + +it.each([ + { description: "PID", mutation: { read: 2, pid: 101 } }, + { description: "start time", mutation: { read: 2, start: 9999 } }, + { description: "parent", mutation: { read: 2, parent: 2 } }, +] as const)( + "does not attribute a root whose initialized identity changes by $description", + async ({ mutation }) => { + setUpFixture(); + fixture.statMutations.set(100, mutation); + const { samples, stop } = start(); + await waitFor(() => fixture.runIdReads > 0); + await flushImmediateProcPass(); + await stop(); + expect(samples).toEqual([]); + }, +); + +it("marks a completed clean pass as neither partial nor interrupted", async () => { + setUpFixture(); + const { samples, stop } = start(); + await waitFor(() => + [100, 200, 300, 400].every((pid) => + samples.some((sample) => sample.pid === pid), + ), + ); + const result = await stop(); + expect(result.partial).toBe(false); + expect(result.coverage.sampling_failures).toBe(0); + expect(result.coverage.first_sampling_failure).toBeNull(); +}); diff --git a/src/process/capture/ProcessSampling.ts b/src/process/capture/ProcessSampling.ts index 59b67f577b..eaceec0b43 100644 --- a/src/process/capture/ProcessSampling.ts +++ b/src/process/capture/ProcessSampling.ts @@ -1,4 +1,3 @@ -import { readFile, readdir } from "node:fs/promises"; import type { ProcessSamplingCoverage } from "../../domain/process/processCaptureCoverage.js"; import type { ProcessSample, @@ -7,20 +6,23 @@ import type { import { execFileOutput } from "../ExecFileOutput.js"; import { readProcessRunId } from "../ProcessOwnership.js"; import { monotonicProcessCaptureClock } from "./ProcessCaptureClock.js"; -const PROC_READ_CONCURRENCY = 64; - -interface ProcessRow { - readonly pid: number; - readonly parent_pid: number; - readonly process_group_id: number | null; - readonly session_id: number | null; - readonly command: string; - readonly startTime: string | undefined; -} - -interface ProcessNode extends ProcessRow { - readonly children: readonly number[]; -} +import { + inspectLinuxProcess, + inspectLinuxProcessMetadata, + PROC_READ_CONCURRENCY, + readLinuxProcessStat, + type SampledProcessRow as ProcessRow, +} from "./LinuxProcessSampling.js"; +import { + isSamplingAbort, + mergeSamplingDiagnostics, + readProcSamplingSource, + readSamplingSource, + samplingObservation, + samplingInterrupted, + samplingReadObservation, + type SamplingObservation, +} from "./ProcessSamplingObservation.js"; interface SampleProcessContext { readonly rootPid: number; @@ -59,255 +61,70 @@ export const isInitializedPtyRoot = (options: { ); }; -const isExpectedProcReadFailure = (cause: unknown): boolean => - cause instanceof Error && - "code" in cause && - (cause.code === "ENOENT" || - cause.code === "ESRCH" || - cause.code === "EACCES" || - cause.code === "EPERM"); - -const handleProcReadFailure = (cause: unknown): undefined => { - if (isExpectedProcReadFailure(cause)) return undefined; - throw cause; -}; - -const parseProcStat = ( - identifier: string, - stat: string, -): ProcessRow | undefined => { - const commandEnd = stat.lastIndexOf(")"); - if (commandEnd < 0) return undefined; - const fields = stat - .slice(commandEnd + 2) - .trim() - .split(/\s+/); - const pid = Number(identifier); - const parentPid = Number(fields[1]); - const processGroupId = Number(fields[2]); - const sessionId = Number(fields[3]); - const startTime = fields[19]; - if ( - !Number.isSafeInteger(pid) || - pid <= 0 || - !Number.isSafeInteger(parentPid) || - parentPid < 0 || - !Number.isSafeInteger(processGroupId) || - processGroupId <= 0 || - !Number.isSafeInteger(sessionId) || - sessionId <= 0 || - startTime === undefined || - !/^\d+$/.test(startTime) - ) - return undefined; - return { - pid, - parent_pid: parentPid, - process_group_id: processGroupId, - session_id: sessionId, - command: "", - startTime, - }; -}; - -const readProcessStat = async ( - pid: number, - signal: AbortSignal, -): Promise => { - try { - const stat = await readFile(`/proc/${String(pid)}/stat`, { - encoding: "utf8", - signal, - }); - return parseProcStat(String(pid), stat); - } catch (cause: unknown) { - return handleProcReadFailure(cause); - } -}; - -const readProcessCommand = async ( - pid: number, - signal: AbortSignal, -): Promise => { - try { - return await readFile(`/proc/${String(pid)}/cmdline`, { signal }); - } catch (cause: unknown) { - return handleProcReadFailure(cause); - } -}; - -/** Procfs reads used to enumerate children created by every process thread. */ -export interface LinuxChildrenHost { - taskIds( - pid: number, - signal: AbortSignal, - ): Promise; - children( - pid: number, - taskId: number, - signal: AbortSignal, - ): Promise; -} - -const systemLinuxChildrenHost: LinuxChildrenHost = { - async taskIds(pid, signal) { - try { - signal.throwIfAborted(); - const entries = await readdir(`/proc/${String(pid)}/task`); - signal.throwIfAborted(); - return entries - .filter((entry) => /^\d+$/.test(entry)) - .map(Number) - .filter((taskId) => Number.isSafeInteger(taskId) && taskId > 0) - .sort((left, right) => left - right); - } catch (cause: unknown) { - return handleProcReadFailure(cause); - } - }, - async children(pid, taskId, signal) { - try { - return await readFile( - `/proc/${String(pid)}/task/${String(taskId)}/children`, - { encoding: "utf8", signal }, - ); - } catch (cause: unknown) { - return handleProcReadFailure(cause); - } - }, -}; - -/** Read the deduplicated children created by every thread in a Linux process. */ -export const readLinuxChildren = async ( - pid: number, - signal: AbortSignal, - host: LinuxChildrenHost = systemLinuxChildrenHost, -): Promise => { - const taskIds = await host.taskIds(pid, signal); - if (taskIds === undefined) return undefined; - const children = new Set(); - let successfulReads = 0; - for ( - let offset = 0; - offset < taskIds.length; - offset += PROC_READ_CONCURRENCY - ) { - const texts = await Promise.all( - taskIds - .slice(offset, offset + PROC_READ_CONCURRENCY) - .map((taskId) => host.children(pid, taskId, signal)), - ); - for (const text of texts) { - if (text === undefined) continue; - successfulReads += 1; - for (const value of text.trim().split(/\s+/)) { - if (!/^\d+$/.test(value)) continue; - const child = Number(value); - if (Number.isSafeInteger(child) && child > 0) children.add(child); - } - } - } - return successfulReads === 0 - ? undefined - : [...children].sort((left, right) => left - right); -}; - -const inspectProcess = async ( - pid: number, - expectedParent: number | undefined, - signal: AbortSignal, - identities: Map, -): Promise => { - const before = await readProcessStat(pid, signal); - if (before === undefined || before.startTime === undefined) return undefined; - - const existing = identities.get(pid); - if (existing !== undefined && existing !== before.startTime) return undefined; - if (expectedParent !== undefined && before.parent_pid !== expectedParent) - return undefined; - - const [commandBuffer, children] = await Promise.all([ - readProcessCommand(pid, signal), - readLinuxChildren(pid, signal), - ]); - - const after = await readProcessStat(pid, signal); - if (after === undefined || after.startTime === undefined) return undefined; - if ( - before.startTime !== after.startTime || - before.parent_pid !== after.parent_pid - ) - return undefined; - if (expectedParent !== undefined && after.parent_pid !== expectedParent) - return undefined; - - const command = - commandBuffer === undefined - ? "" - : commandBuffer.toString().replaceAll("\0", " ").trim(); - - return { - pid: before.pid, - parent_pid: before.parent_pid, - process_group_id: before.process_group_id, - session_id: before.session_id, - startTime: before.startTime, - command, - children: children ?? [], - }; -}; - const sampleLinux = async ( context: SampleProcessContext, -): Promise => { +): Promise> => { const { rootPid, limit, signal, identities } = context; - if (limit <= 0) return []; - - const rootStat = await readProcessStat(rootPid, signal); - if (rootStat === undefined) return []; - const rootChildren = await readLinuxChildren(rootPid, signal); - if (rootChildren === undefined) return samplePs(context); - - const rows: ProcessRow[] = []; + if (limit <= 0) return samplingObservation([]); + const root = await inspectLinuxProcess( + rootPid, + undefined, + signal, + identities, + ); + if (root.value === undefined) + return samplingObservation([], root.diagnostics); + // Preserve the existing ps fallback when procfs cannot enumerate the root. + if (root.value.children === undefined && !signal.aborted) { + const fallback = await sampleLinuxFallback(context, root.value); + return samplingObservation( + fallback.value, + root.diagnostics, + fallback.diagnostics, + ); + } + const rows: ProcessRow[] = [root.value]; + let diagnostics = root.diagnostics; const visited = new Set([rootPid]); - let level: Array<{ - readonly pid: number; - readonly expectedParent: number | undefined; - }> = [{ pid: rootPid, expectedParent: undefined }]; - - while (level.length > 0 && rows.length < limit) { - const batchSize = Math.min(level.length, PROC_READ_CONCURRENCY); + let level = (root.value.children ?? []).map((pid) => ({ + pid, + expectedParent: rootPid, + })); + for (const { pid } of level) visited.add(pid); + while (level.length > 0 && rows.length < limit && !signal.aborted) { + const batchSize = Math.min( + level.length, + PROC_READ_CONCURRENCY, + limit - rows.length, + ); const batch = level.slice(0, batchSize); level = level.slice(batchSize); - - const batchResults = await Promise.all( - batch.map((item) => - inspectProcess(item.pid, item.expectedParent, signal, identities), + const observations = await Promise.all( + batch.map(({ pid, expectedParent }) => + inspectLinuxProcess(pid, expectedParent, signal, identities), ), ); - - const nextLevel: typeof batch = []; - for (const node of batchResults) { + for (const observation of observations) { + diagnostics = mergeSamplingDiagnostics( + diagnostics, + observation.diagnostics, + ); + const node = observation.value; if (node === undefined) continue; - - if (rows.length >= limit) break; rows.push(node); - if (rows.length >= limit) break; - - for (const child of node.children) { + for (const child of node.children ?? []) { if (visited.has(child)) continue; visited.add(child); - nextLevel.push({ pid: child, expectedParent: node.pid }); + level.push({ pid: child, expectedParent: node.pid }); } - if (rows.length >= limit) break; } - - level.push(...nextLevel); } - - return rows; + if (level.length > 0 && signal.aborted) + diagnostics = mergeSamplingDiagnostics(diagnostics, samplingInterrupted()); + return samplingObservation(rows, diagnostics); }; -const readPsRows = async ( +const readPsRowsRaw = async ( signal: AbortSignal, ): Promise => { const { stdout } = await execFileOutput( @@ -337,56 +154,151 @@ const readPsRows = async ( .sort((left, right) => left.pid - right.pid); }; +const readPsRows = async ( + signal: AbortSignal, +): Promise> => + samplingReadObservation( + await readSamplingSource("process table (ps)", signal, () => + readPsRowsRaw(signal), + ), + ); + +/** ps supplies candidate edges only; Linux identity and command remain procfs-owned. */ +const sampleLinuxFallback = async ( + context: SampleProcessContext, + root: ProcessRow, +): Promise> => { + const observed = await readPsRows(context.signal); + if (observed.value === undefined) + return samplingObservation([root], observed.diagnostics); + const childrenByParent = new Map(); + for (const row of observed.value) { + const children = childrenByParent.get(row.parent_pid) ?? []; + children.push(row.pid); + childrenByParent.set(row.parent_pid, children); + } + const rows: ProcessRow[] = [root]; + let diagnostics = observed.diagnostics; + const visited = new Set([root.pid]); + let level = (childrenByParent.get(root.pid) ?? []).map((pid) => ({ + pid, + expectedParent: root.pid, + })); + while ( + level.length > 0 && + rows.length < context.limit && + !context.signal.aborted + ) { + const batch = level.splice( + 0, + Math.min(PROC_READ_CONCURRENCY, context.limit - rows.length), + ); + const candidates = batch.filter(({ pid }) => !visited.has(pid)); + for (const { pid } of candidates) visited.add(pid); + const observations = await Promise.all( + candidates.map(({ pid, expectedParent }) => + inspectLinuxProcessMetadata( + pid, + expectedParent, + context.signal, + context.identities, + ), + ), + ); + for (const observation of observations) { + diagnostics = mergeSamplingDiagnostics( + diagnostics, + observation.diagnostics, + ); + if (observation.value === undefined) continue; + rows.push(observation.value); + for (const child of childrenByParent.get(observation.value.pid) ?? []) + level.push({ pid: child, expectedParent: observation.value.pid }); + } + } + if (level.length > 0 && context.signal.aborted) + diagnostics = mergeSamplingDiagnostics(diagnostics, samplingInterrupted()); + const after = await readLinuxProcessStat(root.pid, context.signal); + diagnostics = mergeSamplingDiagnostics(diagnostics, after.diagnostics); + return samplingObservation( + after.value !== undefined && + after.value.startTime === root.startTime && + after.value.parent_pid === root.parent_pid + ? rows + : [root], + diagnostics, + ); +}; + const readRootIdentity = async ( rootPid: number, signal: AbortSignal, -): Promise => - process.platform === "linux" - ? readProcessStat(rootPid, signal) - : (await readPsRows(signal)).find(({ pid }) => pid === rootPid); +): Promise> => { + if (process.platform === "linux") + return readLinuxProcessStat(rootPid, signal); + const rows = await readPsRows(signal); + return samplingObservation( + rows.value?.find(({ pid }) => pid === rootPid), + rows.diagnostics, + ); +}; const inspectInitializedPtyRoot = async ( rootPid: number, runId: string, signal: AbortSignal, -): Promise<{ readonly startTime: string | undefined } | undefined> => { +): Promise< + SamplingObservation<{ readonly startTime: string | undefined } | undefined> +> => { const before = await readRootIdentity(rootPid, signal); if ( - before === undefined || - before.process_group_id !== rootPid || - (before.session_id !== rootPid && before.session_id !== null) + before.value === undefined || + before.value.process_group_id !== rootPid || + (before.value.session_id !== rootPid && before.value.session_id !== null) ) - return undefined; - let observedRunId: string | undefined; - try { - observedRunId = await readProcessRunId(rootPid, undefined, signal); - } catch (cause: unknown) { - // Unreadable run identity means the root cannot be attributed. - void cause; - return undefined; - } + return samplingObservation(undefined, before.diagnostics); + const observedRunId = + process.platform === "linux" + ? await readProcSamplingSource( + `/proc/${String(rootPid)}/environ`, + signal, + () => readProcessRunId(rootPid, undefined, signal), + ) + : samplingReadObservation( + await readSamplingSource( + `run token for PID ${String(rootPid)}`, + signal, + () => readProcessRunId(rootPid, undefined, signal), + ), + ); const after = await readRootIdentity(rootPid, signal); + const diagnostics = mergeSamplingDiagnostics( + before.diagnostics, + observedRunId.diagnostics, + after.diagnostics, + ); if ( - after === undefined || + after.value === undefined || !isInitializedPtyRoot({ rootPid, expectedRunId: runId, - before, - observedRunId, - after, + before: before.value, + observedRunId: observedRunId.value, + after: after.value, }) ) - return undefined; - return { startTime: after.startTime }; + return samplingObservation(undefined, diagnostics); + return samplingObservation({ startTime: after.value.startTime }, diagnostics); }; const samplePs = async ( context: SampleProcessContext, -): Promise => { +): Promise> => { const { rootPid, limit, signal } = context; - if (limit <= 0) return []; + if (limit <= 0) return samplingObservation([]); - const rows = await readPsRows(signal); + const observed = await readPsRows(signal); + const rows = observed.value ?? []; const rowByPid = new Map(rows.map((row) => [row.pid, row])); const childrenByParent = new Map(); for (const row of rows) { @@ -416,13 +328,14 @@ const samplePs = async ( } } - return result; + return samplingObservation(result, observed.diagnostics); }; const sampleProcesses = async ( context: SampleProcessContext, -): Promise => { - if (process.platform === "win32" || context.limit <= 0) return []; +): Promise> => { + if (process.platform === "win32" || context.limit <= 0) + return samplingObservation([]); const rows = process.platform === "linux" @@ -431,7 +344,7 @@ const sampleProcesses = async ( const { elapsedMs, identities } = context; const samples: ProcessSample[] = []; - for (const row of rows) { + for (const row of rows.value) { if (row.startTime !== undefined) { const existing = identities.get(row.pid); if (existing !== undefined && existing !== row.startTime) continue; @@ -446,7 +359,7 @@ const sampleProcesses = async ( session_id: row.session_id, }); } - return samples; + return samplingObservation(samples, rows.diagnostics); }; /** Start bounded process-tree sampling and expose an awaited stop. */ @@ -467,9 +380,9 @@ export const startProcessSampler = (options: { let pending: Promise | undefined; let stopped = false; let abortCurrent: (() => void) | undefined; - let partial = false; let limitReached = false; let samplingFailures = 0; + let samplingWasInterrupted = false; let firstSamplingFailure: string | null = null; let rootInitialized = false; @@ -479,27 +392,38 @@ export const startProcessSampler = (options: { const controller = new AbortController(); abortCurrent = () => controller.abort(); pending = (async () => { + let diagnostics = mergeSamplingDiagnostics(); if (!rootInitialized) { const identity = await inspectInitializedPtyRoot( rootPid, runId, controller.signal, ); - if (identity === undefined) return []; - if (identity.startTime !== undefined) - identities.set(rootPid, identity.startTime); + diagnostics = identity.diagnostics; + if (identity.value === undefined) + return samplingObservation([], diagnostics); + if (identity.value.startTime !== undefined) + identities.set(rootPid, identity.value.startTime); rootInitialized = true; } - return sampleProcesses({ + const observation = await sampleProcesses({ rootPid, elapsedMs: monotonicProcessCaptureClock() - started, limit: limit + 1, identities, signal: controller.signal, }); + return samplingObservation( + observation.value, + diagnostics, + observation.diagnostics, + ); })() - .then((values) => { - for (const value of values) { + .then((observation) => { + samplingFailures += observation.diagnostics.count; + samplingWasInterrupted ||= observation.diagnostics.interrupted; + firstSamplingFailure ??= observation.diagnostics.first; + for (const value of observation.value) { const observation = JSON.stringify({ parent_pid: value.parent_pid, command: value.command, @@ -508,7 +432,6 @@ export const startProcessSampler = (options: { }); if (lastObservations.get(value.pid) === observation) continue; if (samples.length >= limit) { - partial = true; limitReached = true; continue; } @@ -519,8 +442,9 @@ export const startProcessSampler = (options: { } }) .catch((cause: unknown) => { - if (!(cause instanceof Error && cause.name === "AbortError")) { - partial = true; + if (isSamplingAbort(cause, controller.signal)) { + samplingWasInterrupted = true; + } else { samplingFailures += 1; firstSamplingFailure ??= cause instanceof Error @@ -544,6 +468,15 @@ export const startProcessSampler = (options: { clearInterval(timer); abortCurrent?.(); await pending; + const failureCount = samplingFailures + (samplingWasInterrupted ? 1 : 0); + const interrupted = + "Sampler shutdown interrupted an unfinished process observation; unexamined processes remain unknown"; + const firstFailure = samplingWasInterrupted + ? firstSamplingFailure === null + ? interrupted + : `${firstSamplingFailure}; ${interrupted}` + : firstSamplingFailure; + const partial = limitReached || failureCount > 0; return { partial, coverage: { @@ -551,8 +484,8 @@ export const startProcessSampler = (options: { sampling_partial: partial, retained_samples: samples.length, sample_limit_reached: limitReached, - sampling_failures: samplingFailures, - first_sampling_failure: firstSamplingFailure, + sampling_failures: failureCount, + first_sampling_failure: firstFailure, coverage: "sampled", }, }; diff --git a/src/process/capture/ProcessSamplingObservation.ts b/src/process/capture/ProcessSamplingObservation.ts new file mode 100644 index 0000000000..3f6a0f2207 --- /dev/null +++ b/src/process/capture/ProcessSamplingObservation.ts @@ -0,0 +1,117 @@ +import { err, ok, type Result } from "../../domain/result.js"; + +/** Retain a count, first failure and interruption fact with their observations. */ +export interface SamplingDiagnostics { + readonly count: number; + readonly first: string | null; + readonly interrupted: boolean; +} + +export interface SamplingObservation { + readonly value: T; + readonly diagnostics: SamplingDiagnostics; +} + +export type SamplingReadFailure = + | { readonly kind: "cancelled"; readonly reason: unknown } + | { + readonly kind: "unavailable"; + readonly systemCode: string | undefined; + readonly diagnostic: string; + }; + +export const mergeSamplingDiagnostics = ( + ...parts: readonly SamplingDiagnostics[] +): SamplingDiagnostics => ({ + count: parts.reduce((count, part) => count + part.count, 0), + first: parts.find((part) => part.first !== null)?.first ?? null, + interrupted: parts.some((part) => part.interrupted), +}); + +export const samplingObservation = ( + value: T, + ...diagnostics: readonly SamplingDiagnostics[] +): SamplingObservation => ({ + value, + diagnostics: mergeSamplingDiagnostics(...diagnostics), +}); + +export const samplingFailure = (diagnostic: string): SamplingDiagnostics => ({ + count: 1, + first: diagnostic, + interrupted: false, +}); + +/** An interrupted attempt left an observation or declared traversal unfinished. */ +export const samplingInterrupted = (): SamplingDiagnostics => ({ + count: 0, + first: null, + interrupted: true, +}); + +/** Only this attempt's abort can end a read without a failure observation. */ +export const isSamplingAbort = (cause: unknown, signal: AbortSignal): boolean => + signal.aborted && + (cause === signal.reason || + (cause instanceof Error && + cause.name === "AbortError" && + cause.cause === signal.reason)); + +/** Settle each issued read, including peers still completing after shutdown. */ +export const readSamplingSource = async ( + source: string, + signal: AbortSignal, + read: () => Promise, +): Promise> => { + if (signal.aborted) return err({ kind: "cancelled", reason: signal.reason }); + try { + return ok(await read()); + } catch (cause: unknown) { + if (isSamplingAbort(cause, signal)) + return err({ kind: "cancelled", reason: signal.reason }); + const systemCode = + cause instanceof Error && + "code" in cause && + typeof cause.code === "string" + ? cause.code + : undefined; + const reason = + cause instanceof Error + ? `${cause.name}: ${cause.message}` + : String(cause); + return err({ + kind: "unavailable", + systemCode, + diagnostic: `${source}: ${systemCode === undefined ? "" : `${systemCode}: `}${reason}`, + }); + } +}; + +export const samplingReadObservation = ( + result: Result, +): SamplingObservation => + result.ok + ? samplingObservation(result.value) + : samplingObservation( + undefined, + result.error.kind === "cancelled" + ? samplingInterrupted() + : samplingFailure(result.error.diagnostic), + ); + +/** Missing procfs entries are expected exit races; denied reads stay unknown. */ +export const readProcSamplingSource = async ( + path: string, + signal: AbortSignal, + read: () => Promise, +): Promise> => { + const result = await readSamplingSource(path, signal, read); + if ( + !result.ok && + result.error.kind === "unavailable" && + (result.error.systemCode === "ENOENT" || + result.error.systemCode === "ESRCH") + ) + return samplingObservation(undefined); + return samplingReadObservation(result); +}; diff --git a/tests/boundary/process/processCaptureLifecycle.test.ts b/tests/boundary/process/processCaptureLifecycle.test.ts index bf566ec029..29f39e7468 100644 --- a/tests/boundary/process/processCaptureLifecycle.test.ts +++ b/tests/boundary/process/processCaptureLifecycle.test.ts @@ -30,6 +30,7 @@ import { } from "../../../src/process/capture/ProcessCaptureLifecycle.js"; import { parseProcessScenario } from "../../../src/domain/process/processScenario.js"; import { emptyProcessCapture } from "../../../src/domain/process/processCapture.fixture.js"; +import { processSourceTruncated } from "../../../src/domain/process/processCaptureCoverage.js"; import type { ProcessCapture } from "../../../src/domain/process/processCaptureParsing.js"; const processFixture = fileURLToPath( @@ -542,7 +543,7 @@ itWithCaptureCapability( file.path.endsWith(":escape"), ); expect(escaped?.symlink_target).toBe("/etc/passwd"); - expect(capture.truncated).toBe(false); + expect(processSourceTruncated(capture, "filesystem")).toBe(false); expect(JSON.stringify(capture.files_after)).not.toContain(root); expect( capture.files_after.some((file) => file.path.includes("passwd")),