refactor(process): own selected artifact read failures

Classify selected-input read failures at the shared artifact boundary and remove duplicate phase-based translations from TShark and sigrok. Keep other read stages raw for their existing owners and retain cleanup resources until retry succeeds.

Extend the descriptor outcome matrix with post-run private-copy integrity failures, late cancellation, retained command output, failed cleanup, and successful retry.
This commit is contained in:
morluto
2026-10-24 06:53:48 +08:00
parent 0134e15ab1
commit edc805eca2
4 changed files with 157 additions and 118 deletions
+11 -55
View File
@@ -37,19 +37,14 @@ import {
type AnalysisCapturedOutput,
} from "../domain/analysisErrorBase.js";
import {
AnalysisAccessDeniedError,
AnalysisArtifactChangedError,
AnalysisCancelledError,
AnalysisCapabilityUnavailableError,
AnalysisInputError,
AnalysisOutputError,
AnalysisResourceConstraintError,
} from "../domain/analysisErrorCore.js";
import { createEvidence } from "../domain/evidence.js";
import {
fixedAllowanceOptions,
fixedProducerInputAllowance,
} from "../domain/fixedAllowanceRemediation.js";
import { fixedAllowanceOptions } from "../domain/fixedAllowanceRemediation.js";
import { ProviderAdapterError } from "../domain/providerAdapterError.js";
import {
type InspectPacketCaptureInput,
@@ -137,12 +132,8 @@ export class TsharkCaptureProvider implements PacketCapturePort {
// TShark 4.6.9 emits a 22.3 MB glossary; keep capture output capped at 4 MiB.
const limits =
"path" in input ? LIMITS : { ...LIMITS, output_bytes: 32 * 1024 * 1024 };
let phase:
| "configuration"
| "artifact"
| "workspace"
| "producer"
| "integrity" = "configuration";
let phase: "configuration" | "workspace" | "producer" | "integrity" =
"configuration";
let capturedOutput: AnalysisCapturedOutput | undefined;
let resourcePath = "path" in input ? input.path : "TShark field glossary";
let effectiveLimits: LinuxResourceLimits | undefined;
@@ -298,12 +289,13 @@ export class TsharkCaptureProvider implements PacketCapturePort {
}),
);
} else {
phase = "artifact";
const selected = await this.#resources.read(
input.path,
LIMITS.input_bytes,
options?.signal,
);
const selected = await this.#resources.readSelectedArtifact({
path: input.path,
maximumBytes: LIMITS.input_bytes,
allowanceName: "the fixed packet-capture input allowance",
observedOutput: capturedOutput,
signal: options?.signal,
});
phase = "workspace";
await writeFile(snapshotPath, selected.bytes, {
flag: "wx",
@@ -428,12 +420,7 @@ export class TsharkCaptureProvider implements PacketCapturePort {
}
#failure(
cause: unknown,
phase:
| "configuration"
| "artifact"
| "workspace"
| "producer"
| "integrity",
phase: "configuration" | "workspace" | "producer" | "integrity",
path: string,
workspacePath: string | null,
effectiveLimits: LinuxResourceLimits | undefined,
@@ -521,41 +508,10 @@ export class TsharkCaptureProvider implements PacketCapturePort {
userMessage: `${reason} On Linux, supply TShark 4.x on PATH or through absolute REA_TSHARK_COMMAND, and util-linux prlimit through REA_TSHARK_PRLIMIT_COMMAND (default /usr/bin/prlimit).`,
},
);
if (
phase === "artifact" &&
cause instanceof ArtifactReaderFailure &&
cause.reason === "limit"
)
return new AnalysisResourceConstraintError(
operation,
"file-size",
reason,
{ input_bytes: LIMITS.input_bytes },
{
cause,
...fixedProducerInputAllowance(
LIMITS.input_bytes,
"the fixed packet-capture input allowance",
),
},
);
if (cause instanceof ArtifactReaderFailure && cause.reason === "integrity")
return new AnalysisArtifactChangedError(operation, path, reason, {
cause,
});
if (
phase === "artifact" &&
cause instanceof Error &&
"code" in cause &&
(cause.code === "EACCES" || cause.code === "EPERM")
)
return new AnalysisAccessDeniedError(operation, path, cause.code, {
cause,
});
if (phase === "artifact")
return new AnalysisInputError(operation, { cause }, [
{ path: ["path"], reason: "invalid_value", message: reason },
]);
return new AnalysisOutputError(operation, reason, {
cause,
...(observedOutput === undefined
+74 -1
View File
@@ -15,15 +15,20 @@ import {
type AnalysisPartialObservation,
} from "../domain/analysisErrorBase.js";
import {
AnalysisAccessDeniedError,
AnalysisCancelledError,
AnalysisArtifactChangedError,
AnalysisInputError,
AnalysisResourceConstraintError,
AnalysisOutputError,
AnalysisTimeoutError,
} from "../domain/analysisErrorCore.js";
import { analysisErrorWithCleanupFailure } from "../domain/analysisErrorCleanup.js";
import { projectAnalysisError } from "../domain/analysisErrorProjection.js";
import { fixedAllowanceOptions } from "../domain/fixedAllowanceRemediation.js";
import {
fixedAllowanceOptions,
fixedProducerInputAllowance,
} from "../domain/fixedAllowanceRemediation.js";
import { ProviderAdapterError } from "../domain/providerAdapterError.js";
import { ProviderCleanupError } from "../domain/providerCleanupError.js";
import type { JsonValue } from "../domain/jsonValue.js";
@@ -44,6 +49,14 @@ export type ArtifactAnalysisRuntime = Pick<
"path" | "close"
>;
interface SelectedArtifactRead {
readonly path: string;
readonly maximumBytes: number;
readonly allowanceName: string;
readonly observedOutput: AnalysisCapturedOutput | undefined;
readonly signal?: AbortSignal | undefined;
}
export const artifactCommandOutput = (
snapshot: ProviderProcessSnapshot,
): AnalysisCapturedOutput => ({
@@ -146,6 +159,66 @@ export class ArtifactAnalysisResources {
);
}
/** Admit a selected input and classify its physical failure before later stages. */
async readSelectedArtifact(input: SelectedArtifactRead) {
const operation = this.#artifacts.operation;
try {
return await this.read(input.path, input.maximumBytes, input.signal);
} catch (cause: unknown) {
throw this.#selectedArtifactFailure(cause, input, operation);
}
}
#selectedArtifactFailure(
cause: unknown,
input: SelectedArtifactRead,
operation: string,
): AnalysisError {
const { path, maximumBytes, allowanceName, observedOutput } = input;
if (cause instanceof AnalysisError) return cause;
if (cause instanceof ArtifactReaderFailure && cause.cleanup !== undefined)
return this.cleanupFailure(
cause,
(primary) => this.#selectedArtifactFailure(primary, input, operation),
observedOutput,
operation,
);
if (cause instanceof ArtifactReaderFailure && cause.reason === "cancelled")
return new AnalysisCancelledError(operation, {
cause: artifactFileCancellationCause(cause),
...(observedOutput === undefined
? {}
: { capturedOutput: observedOutput }),
});
const reason = cause instanceof Error ? cause.message : String(cause);
if (cause instanceof ArtifactReaderFailure && cause.reason === "limit")
return new AnalysisResourceConstraintError(
operation,
"file-size",
reason,
{ input_bytes: maximumBytes },
{
cause,
...fixedProducerInputAllowance(maximumBytes, allowanceName),
},
);
if (cause instanceof ArtifactReaderFailure && cause.reason === "integrity")
return new AnalysisArtifactChangedError(operation, path, reason, {
cause,
});
if (
cause instanceof Error &&
"code" in cause &&
(cause.code === "EACCES" || cause.code === "EPERM")
)
return new AnalysisAccessDeniedError(operation, path, cause.code, {
cause,
});
return new AnalysisInputError(operation, { cause }, [
{ path: ["path"], reason: "invalid_value", message: reason },
]);
}
readLimits(limits: LinuxResourceLimits): Promise<LinuxResourceLimits> {
return this.readKernelLimits(limits, this.#artifacts.scope);
}
+11 -54
View File
@@ -39,7 +39,6 @@ import {
type AnalysisCapturedOutput,
} from "../domain/analysisErrorBase.js";
import {
AnalysisAccessDeniedError,
AnalysisArtifactChangedError,
AnalysisCancelledError,
AnalysisCapabilityUnavailableError,
@@ -48,10 +47,7 @@ import {
AnalysisResourceConstraintError,
} from "../domain/analysisErrorCore.js";
import { createEvidence } from "../domain/evidence.js";
import {
fixedAllowanceOptions,
fixedProducerInputAllowance,
} from "../domain/fixedAllowanceRemediation.js";
import { fixedAllowanceOptions } from "../domain/fixedAllowanceRemediation.js";
import { ProviderAdapterError } from "../domain/providerAdapterError.js";
import { windowAnnotationText } from "../domain/signals/annotationTextWindow.js";
import {
@@ -129,12 +125,8 @@ export class SigrokRecordingProvider implements SignalRecordingPort {
options?: ExecutionOptions,
): Promise<Result<AnalysisExecution, AnalysisError>> {
const operation = "path" in input ? OPERATION : "inspect_signal_decoders";
let phase:
| "configuration"
| "artifact"
| "workspace"
| "producer"
| "integrity" = "configuration";
let phase: "configuration" | "workspace" | "producer" | "integrity" =
"configuration";
let capturedOutput: AnalysisCapturedOutput | undefined;
let resourcePath = "path" in input ? input.path : "sigrok decoder catalog";
let effectiveLimits: LinuxResourceLimits | undefined;
@@ -321,12 +313,13 @@ export class SigrokRecordingProvider implements SignalRecordingPort {
}),
);
} else {
phase = "artifact";
const selected = await this.#resources.read(
input.path,
LIMITS.input_bytes,
options?.signal,
);
const selected = await this.#resources.readSelectedArtifact({
path: input.path,
maximumBytes: LIMITS.input_bytes,
allowanceName: "the fixed sigrok input allowance",
observedOutput: capturedOutput,
signal: options?.signal,
});
phase = "workspace";
await writeFile(snapshotPath, selected.bytes, {
flag: "wx",
@@ -429,12 +422,7 @@ export class SigrokRecordingProvider implements SignalRecordingPort {
}
#failure(
cause: unknown,
phase:
| "configuration"
| "artifact"
| "workspace"
| "producer"
| "integrity",
phase: "configuration" | "workspace" | "producer" | "integrity",
path: string,
workspacePath: string | null,
effectiveLimits: LinuxResourceLimits | undefined,
@@ -517,41 +505,10 @@ export class SigrokRecordingProvider implements SignalRecordingPort {
userMessage: `${reason} On Linux, configure sigrok-cli 0.7.2 through REA_SIGROK_COMMAND and util-linux prlimit through REA_SIGROK_PRLIMIT_COMMAND (default /usr/bin/prlimit).`,
},
);
if (
phase === "artifact" &&
cause instanceof ArtifactReaderFailure &&
cause.reason === "limit"
)
return new AnalysisResourceConstraintError(
operation,
"file-size",
reason,
{ input_bytes: LIMITS.input_bytes },
{
cause,
...fixedProducerInputAllowance(
LIMITS.input_bytes,
"the fixed sigrok input allowance",
),
},
);
if (cause instanceof ArtifactReaderFailure && cause.reason === "integrity")
return new AnalysisArtifactChangedError(operation, path, reason, {
cause,
});
if (
phase === "artifact" &&
cause instanceof Error &&
"code" in cause &&
(cause.code === "EACCES" || cause.code === "EPERM")
)
return new AnalysisAccessDeniedError(operation, path, cause.code, {
cause,
});
if (phase === "artifact")
return new AnalysisInputError(operation, { cause }, [
{ path: ["path"], reason: "invalid_value", message: reason },
]);
return new AnalysisOutputError(operation, reason, {
cause,
...(observedOutput === undefined
@@ -1,6 +1,6 @@
import { lstat, writeFile } from "node:fs/promises";
import type { FileHandle } from "node:fs/promises";
import { join } from "node:path";
import { basename, join } from "node:path";
import { expect, it } from "vitest";
import { SigrokRecordingProvider } from "../../../src/signals/SigrokRecordingProvider.js";
import { TsharkCaptureProvider } from "../../../src/captures/TsharkCaptureProvider.js";
@@ -50,6 +50,12 @@ const fileOutcomeControls = (["signal", "capture"] as const).flatMap((kind) =>
resource: "limiter",
discovery: true,
},
{
stage: "integrity",
failure: "integrity",
tag: "AnalysisArtifactChangedError",
resource: "private-copy",
},
].map((control) => ({ kind, ...control })),
);
@@ -98,6 +104,7 @@ const createFileOutcomeFixture = async ({
let launches = 0;
let handle: FileHandle | undefined;
let runtimeRoot: PrivateRuntimeRoot | undefined;
let privateCopyPath: string | undefined;
const caller = new AbortController();
const cancellationReason = {
request: "late cancellation during failed close",
@@ -123,6 +130,10 @@ const createFileOutcomeFixture = async ({
},
async () => {
runtimeRoot = await PrivateRuntimeRoot.create({ parent: root });
privateCopyPath = join(
runtimeRoot.path,
kind === "signal" ? basename(inputPath) : "capture",
);
return runtimeRoot;
},
"linux",
@@ -130,11 +141,16 @@ const createFileOutcomeFixture = async ({
if (args[0] === command) commandReads++;
if (args[0] === limiter) limiterReads++;
const selected =
args[0] === selectedPath &&
(stage !== "integrity" ||
(resource === "limiter" ? limiterReads === 2 : commandReads === 3));
resource === "private-copy"
? args[0] === privateCopyPath
: args[0] === selectedPath &&
(stage !== "integrity" ||
(resource === "limiter"
? limiterReads === 2
: commandReads === 3));
if (!selected) return readStableArtifact(...args);
if (resource === "limiter") expect(args[2]).toBeUndefined();
if (resource === "limiter" || resource === "private-copy")
expect(args[2]).toBeUndefined();
const scope = args[4];
if (scope === undefined)
throw new Error("Expected retained artifact scope");
@@ -174,6 +190,9 @@ const createFileOutcomeFixture = async ({
get runtimeRoot() {
return runtimeRoot;
},
get privateCopyPath() {
return privateCopyPath;
},
allowClose() {
allowed = true;
},
@@ -221,6 +240,38 @@ it.each(fileOutcomeControls)(
expect(primary).toMatchObject({ _tag: tag, cause: fixture.physical });
expect(result.error.cleanupIncomplete).toBe(true);
expect(fixture.handle?.fd).toBeGreaterThanOrEqual(0);
if (resource === "private-copy") {
if (
fixture.runtimeRoot === undefined ||
fixture.privateCopyPath === undefined
)
throw new Error("Expected retained private recording copy");
const operation =
kind === "signal"
? "decode_signal_recording"
: "inspect_packet_capture";
const capturedOutput = {
stdout: kind === "signal" ? "annotation\n" : "[]\n",
stderr: "",
truncated: false,
exit_code: 0,
signal: null,
};
expect(primary).toMatchObject({
operation,
path: fixture.privateCopyPath,
capturedOutput,
});
expect(result.error).toMatchObject({ operation, capturedOutput });
expect(fixture.caller.signal.reason).toBe(fixture.cancellationReason);
expect(fixture.launches).toBe(2);
expect(result.error.cleanupResources).toEqual([
fixture.privateCopyPath,
fixture.runtimeRoot.path,
]);
await expect(lstat(fixture.privateCopyPath)).resolves.toBeDefined();
await expect(lstat(fixture.inputPath)).resolves.toBeDefined();
}
if (resource === "limiter") {
const operation =
kind === "signal"
@@ -258,7 +309,7 @@ it.each(fileOutcomeControls)(
.close()
.catch((cause: unknown) => cause);
expect(failed).toMatchObject({ cleanupIncomplete: true });
if (resource === "limiter") {
if (resource === "limiter" || resource === "private-copy") {
expect(fixture.handle?.fd).toBeGreaterThanOrEqual(0);
await expect(lstat(fixture.runtimeRoot!.path)).resolves.toBeDefined();
}
@@ -280,13 +331,15 @@ it.each(fileOutcomeControls)(
);
expect(failed).toMatchObject({ cleanupIncomplete: false });
expect(fixture.launches).toBe(launchCount);
if (resource === "limiter") {
if (resource === "limiter" || resource === "private-copy") {
expect(result.error.capturedOutput).toEqual(primary?.capturedOutput);
expect(result.error.cleanupResources).toEqual([]);
await expect(lstat(fixture.runtimeRoot!.path)).rejects.toMatchObject({
code: "ENOENT",
});
await expect(lstat(fixture.limiter)).resolves.toBeDefined();
await expect(
lstat(resource === "limiter" ? fixture.limiter : fixture.inputPath),
).resolves.toBeDefined();
}
} finally {
await fixture.dispose();