From 8a247fc70f8294d4e8cfe920007eb135a965bb77 Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Sun, 18 Oct 2026 02:38:30 +0800 Subject: [PATCH] fix(providers): preserve artifact producer cancellation facts Wire sigrok and tshark finalization to the effective signal. Services and file-failure adapters retain exact opaque reasons and completed Evidence instead of guessing from error names. Preserve typed primary outcomes through failed cleanup and recovery. Validation: latest per-file coverage is 114 passing tests across nine selected files with 14 explicit Linux-only skips; core foundation passes independently. check:fast, docs and discovery pass. Four finalization controls failed before the signal ownership fix. Bounded completed authored-producer service workflows have identical Evidence with no performance improvement claim. Real sigrok/tshark and native Linux/Windows were not verified. Closes #4858 --- .../captures/PacketCaptureService.ts | 9 +- .../signals/SignalRecordingService.ts | 9 +- src/captures/TsharkCaptureProvider.ts | 19 +- src/signals/SigrokRecordingProvider.ts | 19 +- .../boundary/captures/captureProvider.test.ts | 40 +++- .../artifactProducerCancellation.test.ts | 221 ++++++++++++++++++ .../signals/recordingProvider.test.ts | 42 +++- 7 files changed, 341 insertions(+), 18 deletions(-) create mode 100644 tests/boundary/process/artifactProducerCancellation.test.ts diff --git a/src/application/captures/PacketCaptureService.ts b/src/application/captures/PacketCaptureService.ts index e2877fa3fa..741270e936 100644 --- a/src/application/captures/PacketCaptureService.ts +++ b/src/application/captures/PacketCaptureService.ts @@ -49,7 +49,9 @@ export class PacketCaptureService { ): Promise> { const operation = "inspect_packet_capture"; if (options?.signal?.aborted) - return err(new AnalysisCancelledError(operation)); + return err( + new AnalysisCancelledError(operation, { cause: options.signal.reason }), + ); const input = inspectPacketCaptureInputSchema.safeParse(raw); if (!input.success) return err( @@ -92,7 +94,9 @@ export class PacketCaptureService { ): Promise> { const operation = "inspect_packet_fields"; if (options?.signal?.aborted) - return err(new AnalysisCancelledError(operation)); + return err( + new AnalysisCancelledError(operation, { cause: options.signal.reason }), + ); const input = inspectPacketFieldsInputSchema.safeParse(raw); if (!input.success) return err( @@ -139,6 +143,7 @@ export class PacketCaptureService { return options?.signal?.aborted ? err( new AnalysisCancelledError(operation, { + cause: options.signal.reason, partialObservation: evidence, }), ) diff --git a/src/application/signals/SignalRecordingService.ts b/src/application/signals/SignalRecordingService.ts index e5b99f1b7f..e97b8bdc33 100644 --- a/src/application/signals/SignalRecordingService.ts +++ b/src/application/signals/SignalRecordingService.ts @@ -49,7 +49,9 @@ export class SignalRecordingService { ): Promise> { const operation = "decode_signal_recording"; if (options?.signal?.aborted) - return err(new AnalysisCancelledError(operation)); + return err( + new AnalysisCancelledError(operation, { cause: options.signal.reason }), + ); const input = decodeSignalRecordingInputSchema.safeParse(raw); if (!input.success) return err( @@ -92,7 +94,9 @@ export class SignalRecordingService { ): Promise> { const operation = "inspect_signal_decoders"; if (options?.signal?.aborted) - return err(new AnalysisCancelledError(operation)); + return err( + new AnalysisCancelledError(operation, { cause: options.signal.reason }), + ); const input = inspectSignalDecodersInputSchema.safeParse(raw); if (!input.success) return err( @@ -139,6 +143,7 @@ export class SignalRecordingService { return options?.signal?.aborted ? err( new AnalysisCancelledError(operation, { + cause: options.signal.reason, partialObservation: evidence, }), ) diff --git a/src/captures/TsharkCaptureProvider.ts b/src/captures/TsharkCaptureProvider.ts index 5103ac40b7..a42413b2cd 100644 --- a/src/captures/TsharkCaptureProvider.ts +++ b/src/captures/TsharkCaptureProvider.ts @@ -1,3 +1,4 @@ +import { artifactFileCancellationCause } from "../artifacts/ArtifactFileOperation.js"; import { type InspectPacketFieldsInput, packetFieldsCatalogSchema, @@ -130,7 +131,10 @@ export class TsharkCaptureProvider implements PacketCapturePort { let root: ArtifactAnalysisRuntime | undefined; let result: Result; try { - if (options?.signal?.aborted) throw new AnalysisCancelledError(operation); + if (options?.signal?.aborted) + throw new AnalysisCancelledError(operation, { + cause: options.signal.reason, + }); await this.#resources.release(); if (this.platform !== "linux") throw new Error( @@ -419,6 +423,7 @@ export class TsharkCaptureProvider implements PacketCapturePort { limitations: execution.limitations, }), operation, + options?.signal, ); } #failure( @@ -447,11 +452,13 @@ export class TsharkCaptureProvider implements PacketCapturePort { observedOutput, operation, ); - if ( - (cause instanceof Error && cause.name === "AbortError") || - (cause instanceof ArtifactReaderFailure && cause.reason === "cancelled") - ) - return new AnalysisCancelledError(operation, { cause }); + if (cause instanceof ArtifactReaderFailure && cause.reason === "cancelled") + return new AnalysisCancelledError(operation, { + cause: artifactFileCancellationCause(cause), + ...(observedOutput === undefined + ? {} + : { capturedOutput: observedOutput }), + }); if (cause instanceof FieldDiscoveryBudgetError) return new AnalysisResourceConstraintError( operation, diff --git a/src/signals/SigrokRecordingProvider.ts b/src/signals/SigrokRecordingProvider.ts index 816556c62a..c14b20d3d7 100644 --- a/src/signals/SigrokRecordingProvider.ts +++ b/src/signals/SigrokRecordingProvider.ts @@ -1,3 +1,4 @@ +import { artifactFileCancellationCause } from "../artifacts/ArtifactFileOperation.js"; import { type InspectSignalDecodersInput, signalDecodersSchema, @@ -126,7 +127,10 @@ export class SigrokRecordingProvider implements SignalRecordingPort { let root: ArtifactAnalysisRuntime | undefined; let result: Result; try { - if (options?.signal?.aborted) throw new AnalysisCancelledError(operation); + if (options?.signal?.aborted) + throw new AnalysisCancelledError(operation, { + cause: options.signal.reason, + }); await this.#resources.release(); if (this.platform !== "linux") throw new Error( @@ -383,6 +387,7 @@ export class SigrokRecordingProvider implements SignalRecordingPort { limitations: execution.limitations, }), operation, + options?.signal, ); } #failure( @@ -407,11 +412,13 @@ export class SigrokRecordingProvider implements SignalRecordingPort { observedOutput, operation, ); - if ( - (cause instanceof Error && cause.name === "AbortError") || - (cause instanceof ArtifactReaderFailure && cause.reason === "cancelled") - ) - return new AnalysisCancelledError(operation, { cause }); + if (cause instanceof ArtifactReaderFailure && cause.reason === "cancelled") + return new AnalysisCancelledError(operation, { + cause: artifactFileCancellationCause(cause), + ...(observedOutput === undefined + ? {} + : { capturedOutput: observedOutput }), + }); if (cause instanceof DecoderDiscoveryBudgetError) return new AnalysisResourceConstraintError( operation, diff --git a/tests/boundary/captures/captureProvider.test.ts b/tests/boundary/captures/captureProvider.test.ts index 30281bed3e..a7ec58b29f 100644 --- a/tests/boundary/captures/captureProvider.test.ts +++ b/tests/boundary/captures/captureProvider.test.ts @@ -97,6 +97,8 @@ else { if(readFileSync('capture', 'utf8') !== 'recording bytes') throw Error('wr return { path: root.path, close: async () => { + if (mode === "close-abort") + controller.abort({ request: "cancel during workspace close" }); if (!mayClean) throw new Error("cleanup fixture failure"); await root.close(); }, @@ -248,6 +250,8 @@ it.each([ ); if (result.ok) throw new Error("Expected capture analysis failure"); expect(result.error._tag).toBe(tag); + if (mode === "wait") + expect(result.error.cause).toBe(f.controller.signal.reason); expect(result.error.cleanupIncomplete).toBe(false); expect( (await readdir(f.directory)).filter((name) => @@ -327,7 +331,7 @@ it("retains a completed observation when cancellation arrives at the service bou throw new Error("Unexpected field discovery call"); }, inspect: async () => { - f.controller.abort(); + f.controller.abort({ request: "cancel at Evidence boundary" }); return execution; }, close: async () => {}, @@ -335,6 +339,7 @@ it("retains a completed observation when cancellation arrives at the service bou const result = await service.inspect(input, { signal: f.controller.signal }); if (result.ok) throw new Error("Expected late cancellation"); expect(result.error._tag).toBe("AnalysisCancelledError"); + expect(result.error.cause).toBe(f.controller.signal.reason); expect(result.error.partialObservation).toMatchObject({ operation: "inspect_packet_capture", normalized_result: { @@ -676,3 +681,36 @@ it("preserves discovery resource ownership through CLI shutdown retries", async await f.service.close(); } }); + +it.each( + [false, true].flatMap((failsClose) => + [false, true].map((discovery) => ({ failsClose, discovery })), + ), +)( + "retains completed Evidence and exact cancellation during workspace close: %j", + async ({ failsClose, discovery }) => { + const f = await fixture("close-abort", failsClose); + const result = discovery + ? await f.service.inspectFields({}, { signal: f.controller.signal }) + : await f.service.inspect( + { path: f.path, fields: ["ip.addr"] }, + { signal: f.controller.signal }, + ); + if (result.ok) throw new Error("Expected finalization cancellation"); + const primary = failsClose ? result.error.cleanupPrimary : result.error; + expect(primary).toMatchObject({ _tag: "AnalysisCancelledError" }); + expect(primary?.cause).toBe(f.controller.signal.reason); + expect(result.error.partialObservation).toMatchObject({ + operation: discovery ? "inspect_packet_fields" : "inspect_packet_capture", + }); + expect(result.error.cleanupIncomplete).toBe(failsClose); + if (failsClose) + await expect(f.service.close()).rejects.toMatchObject({ + cleanupIncomplete: true, + }); + f.allowCleanup(); + await f.service.close(); + expect(result.error.cleanupIncomplete).toBe(false); + expect(primary?.cause).toBe(f.controller.signal.reason); + }, +); diff --git a/tests/boundary/process/artifactProducerCancellation.test.ts b/tests/boundary/process/artifactProducerCancellation.test.ts new file mode 100644 index 0000000000..55c808550c --- /dev/null +++ b/tests/boundary/process/artifactProducerCancellation.test.ts @@ -0,0 +1,221 @@ +import { stat, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { expect, it, vi } from "vitest"; +import { SigrokRecordingProvider } from "../../../src/signals/SigrokRecordingProvider.js"; +import { TsharkCaptureProvider } from "../../../src/captures/TsharkCaptureProvider.js"; +import { SignalRecordingService } from "../../../src/application/signals/SignalRecordingService.js"; +import { PacketCaptureService } from "../../../src/application/captures/PacketCaptureService.js"; +import { decodeSignalRecordingInputSchema } from "../../../src/domain/signals/signalRecording.js"; +import { inspectSignalDecodersInputSchema } from "../../../src/domain/signals/signalDecoders.js"; +import { inspectPacketCaptureInputSchema } from "../../../src/domain/captures/packetCapture.js"; +import { inspectPacketFieldsInputSchema } from "../../../src/domain/captures/packetFields.js"; +import { OwnedFileHandle } from "../../../src/filesystem/OwnedFileHandle.js"; +import { readStableArtifact } from "../../../src/artifacts/readStableArtifact.js"; +import { ArtifactReaderFailure } from "../../../src/artifacts/ArtifactReader.js"; +import { ProviderCleanupError } from "../../../src/domain/providerCleanupError.js"; +import { AnalysisError } from "../../../src/domain/analysisErrorBase.js"; +import type { ExecutionOptions } from "../../../src/application/AnalysisProvider.js"; +import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js"; + +// Simulated Linux admission, real descriptors; no producer, limiter or target runs. +const fixture = async ( + kind: "signal" | "capture", + readArtifact = readStableArtifact, +) => { + const root = await createTestTempDirectory("rea-producer-cancellation-"); + const command = join(root, "selected-tool"); + const path = join(root, "recording"); + await writeFile(command, "not an executable producer", { mode: 0o700 }); + await writeFile(path, "recording bytes"); + const Provider = + kind === "signal" ? SigrokRecordingProvider : TsharkCaptureProvider; + const provider = new Provider( + { + REA_SIGROK_COMMAND: command, + REA_SIGROK_PRLIMIT_COMMAND: command, + REA_TSHARK_COMMAND: command, + REA_TSHARK_PRLIMIT_COMMAND: command, + }, + async () => { + throw new Error("Must not launch a producer"); + }, + async () => { + throw new Error("Must not allocate a workspace"); + }, + "linux", + readArtifact, + ); + const requests = + provider instanceof SigrokRecordingProvider + ? [ + (options: ExecutionOptions) => + provider.decode( + decodeSignalRecordingInputSchema.parse({ + path, + decoders: "uart", + }), + options, + ), + (options: ExecutionOptions) => + provider.inspectDecoders( + inspectSignalDecodersInputSchema.parse({}), + options, + ), + (options: ExecutionOptions) => + new SignalRecordingService(provider).decode( + { path, decoders: "uart" }, + options, + ), + (options: ExecutionOptions) => + new SignalRecordingService(provider).inspectDecoders({}, options), + ] + : [ + (options: ExecutionOptions) => + provider.inspect( + inspectPacketCaptureInputSchema.parse({ + path, + fields: ["frame.number"], + }), + options, + ), + (options: ExecutionOptions) => + provider.inspectFields( + inspectPacketFieldsInputSchema.parse({}), + options, + ), + (options: ExecutionOptions) => + new PacketCaptureService(provider).inspect( + { path, fields: ["frame.number"] }, + options, + ), + (options: ExecutionOptions) => + new PacketCaptureService(provider).inspectFields({}, options), + ]; + return { command, provider, requests }; +}; +const reasons = [ + null, + "opaque caller reason", + new ProviderCleanupError("foreign", ["foreign-resource"], { + reason: "caller-owned cleanup", + }), +]; + +it.each(["signal", "capture"] as const)( + "preserves opaque reasons at every %s service/provider admission", + async (kind) => { + const f = await fixture(kind); + try { + for (const reason of reasons) + for (const request of f.requests) { + const result = await request({ signal: AbortSignal.abort(reason) }); + if (result.ok) throw new Error("Expected cancelled admission"); + expect(result.error).toMatchObject({ + _tag: "AnalysisCancelledError", + cleanupIncomplete: false, + }); + expect(result.error.cause).toBe(reason); + expect(result.error.cleanupResources).toEqual([]); + if (reason instanceof ProviderCleanupError) + expect(reason.cleanupIncomplete).toBe(true); + } + } finally { + await f.provider.close(); + } + }, +); + +it.each(["signal", "capture"] as const)( + "uses artifact cancellation facts for the %s producer", + async (kind) => { + for (const reason of reasons) { + const f = await fixture(kind, async () => { + throw new ArtifactReaderFailure("cancelled", "read interrupted", { + cause: reason, + }); + }); + try { + const result = await f.requests[0]!({}); + if (result.ok) throw new Error("Expected artifact cancellation"); + expect(result.error).toMatchObject({ _tag: "AnalysisCancelledError" }); + expect(result.error.cause).toBe(reason); + expect(result.error.cleanupResources).toEqual([]); + } finally { + await f.provider.close(); + } + } + }, +); + +it.each(["signal", "capture"] as const)( + "does not infer %s cancellation from an arbitrary error name", + async (kind) => { + const physical = new Error("physical read failed"); + physical.name = "AbortError"; + const f = await fixture(kind, async () => { + throw physical; + }); + try { + const result = await f.requests[0]!({}); + if (result.ok) throw new Error("Expected read failure"); + expect(result.error).toMatchObject({ + _tag: "AnalysisCapabilityUnavailableError", + cause: physical, + }); + } finally { + await f.provider.close(); + } + }, +); + +it.each( + (["signal", "capture"] as const).flatMap((kind) => + [false, true].map((failsClose) => ({ kind, failsClose })), + ), +)( + "preserves the selected-tool file outcome through final-close cancellation: %j", + async ({ kind, failsClose }) => { + const f = await fixture(kind); + const identity = await stat(f.command); + const reason = { request: "cancel during selected-tool descriptor close" }; + const caller = new AbortController(); + const original = OwnedFileHandle.prototype.close; + let owner: OwnedFileHandle | undefined; + let denied = failsClose; + const close = vi + .spyOn(OwnedFileHandle.prototype, "close") + .mockImplementation(async function (this: OwnedFileHandle) { + const observed = await this.handle.stat(); + if (observed.dev === identity.dev && observed.ino === identity.ino) { + owner = this; + caller.abort(reason); + if (denied) throw new Error("injected selected-tool close failure"); + } + await original.call(this); + }); + try { + const result = await f.requests[0]!({ signal: caller.signal }); + if (result.ok) throw new Error("Expected final-close cancellation"); + const primary = failsClose ? result.error.cleanupPrimary : result.error; + expect(primary).toMatchObject({ _tag: "AnalysisCancelledError" }); + expect(primary?.cause).toBe(reason); + expect(result.error.cleanupIncomplete).toBe(failsClose); + if (failsClose) { + expect(owner?.handle.fd).toBeGreaterThanOrEqual(0); + await expect(f.provider.close()).rejects.toMatchObject({ + cleanupIncomplete: true, + }); + } + denied = false; + await f.provider.close(); + expect(owner?.handle.fd).toBe(-1); + expect(result.error.cleanupIncomplete).toBe(false); + expect(primary?.cause).toBe(reason); + } finally { + denied = false; + close.mockRestore(); + await f.provider.close(); + if (owner !== undefined) await original.call(owner); + } + }, +); diff --git a/tests/boundary/signals/recordingProvider.test.ts b/tests/boundary/signals/recordingProvider.test.ts index f99b252dbf..8b6c608e99 100644 --- a/tests/boundary/signals/recordingProvider.test.ts +++ b/tests/boundary/signals/recordingProvider.test.ts @@ -97,6 +97,8 @@ else { if(readFileSync('recording', 'utf8') !== 'recording bytes') throw Error(' return { path: root.path, close: async () => { + if (mode === "close-abort") + controller.abort({ request: "cancel during workspace close" }); if (!mayClean) throw new Error("cleanup fixture failure"); await root.close(); }, @@ -236,6 +238,8 @@ it.each([ ); if (result.ok) throw new Error("Expected decoder failure"); expect(result.error._tag).toBe(tag); + if (mode === "wait") + expect(result.error.cause).toBe(f.controller.signal.reason); expect(result.error.cleanupIncomplete).toBe(false); expect( (await readdir(f.directory)).filter((name) => @@ -311,7 +315,7 @@ it("retains a completed observation when cancellation arrives at the service bou throw new Error("Unexpected discovery call"); }, decode: async () => { - f.controller.abort(); + f.controller.abort({ request: "cancel at Evidence boundary" }); return execution; }, close: async () => {}, @@ -319,6 +323,7 @@ it("retains a completed observation when cancellation arrives at the service bou const result = await service.decode(input, { signal: f.controller.signal }); if (result.ok) throw new Error("Expected late cancellation"); expect(result.error._tag).toBe("AnalysisCancelledError"); + expect(result.error.cause).toBe(f.controller.signal.reason); expect(result.error.partialObservation).toMatchObject({ operation: "decode_signal_recording", normalized_result: { annotations_text: "10-20 uart-1: 42\n" }, @@ -626,3 +631,38 @@ it("retains completed discovery on cancellation during identity verification", a }, }); }); + +it.each( + [false, true].flatMap((failsClose) => + [false, true].map((discovery) => ({ failsClose, discovery })), + ), +)( + "retains completed Evidence and exact cancellation during workspace close: %j", + async ({ failsClose, discovery }) => { + const f = await fixture("close-abort", failsClose); + const result = discovery + ? await f.service.inspectDecoders({}, { signal: f.controller.signal }) + : await f.service.decode( + { path: f.path, decoders: "uart" }, + { signal: f.controller.signal }, + ); + if (result.ok) throw new Error("Expected finalization cancellation"); + const primary = failsClose ? result.error.cleanupPrimary : result.error; + expect(primary).toMatchObject({ _tag: "AnalysisCancelledError" }); + expect(primary?.cause).toBe(f.controller.signal.reason); + expect(result.error.partialObservation).toMatchObject({ + operation: discovery + ? "inspect_signal_decoders" + : "decode_signal_recording", + }); + expect(result.error.cleanupIncomplete).toBe(failsClose); + if (failsClose) + await expect(f.service.close()).rejects.toMatchObject({ + cleanupIncomplete: true, + }); + f.allowCleanup(); + await f.service.close(); + expect(result.error.cleanupIncomplete).toBe(false); + expect(primary?.cause).toBe(f.controller.signal.reason); + }, +);