diff --git a/src/process/ArtifactAnalysisResources.test.ts b/src/process/ArtifactAnalysisResources.test.ts index 58aff744d..ede5260ab 100644 --- a/src/process/ArtifactAnalysisResources.test.ts +++ b/src/process/ArtifactAnalysisResources.test.ts @@ -1,9 +1,13 @@ +import { OwnedCommandFailure } from "./OwnedCommand.js"; import { EventEmitter } from "node:events"; import { PassThrough } from "node:stream"; import { expect, it } from "vitest"; import { readStableArtifact } from "../artifacts/readStableArtifact.js"; import { ArtifactReaderFailure } from "../artifacts/ArtifactReader.js"; -import { AnalysisOutputError } from "../domain/analysisErrorCore.js"; +import { + AnalysisCancelledError, + AnalysisOutputError, +} from "../domain/analysisErrorCore.js"; import { createEvidence } from "../domain/evidence.js"; import { projectAnalysisError } from "../domain/analysisErrorProjection.js"; import { ProviderAdapterError } from "../domain/providerAdapterError.js"; @@ -332,3 +336,334 @@ it("keeps a runtime root until its descriptor cleanup owner is released", async expect(outer.error.cleanupIncomplete).toBe(false); await resources.close(); }); + +it.each([ + null, + "opaque reason", + new ProviderCleanupError("foreign", ["foreign-resource"], { + reason: "caller cleanup", + }), +])( + "preserves caller cancellation at resource admission: %s", + async (reason) => { + const resources = new ArtifactAnalysisResources( + "fixture", + "inspect_signal_decoders", + readStableArtifact, + readLinuxResourceLimits, + { message: "busy", details: {} }, + ); + let calls = 0; + try { + const result = await resources.run(async () => { + calls++; + return ok("unexpected"); + }, AbortSignal.abort(reason)); + expect(result.ok).toBe(false); + if (result.ok) throw new Error("Expected cancelled admission"); + expect(result.error.cause).toBe(reason); + expect(result.error.cleanupResources).toEqual([]); + expect(calls).toBe(0); + if (reason instanceof ProviderCleanupError) + expect(reason.cleanupIncomplete).toBe(true); + } finally { + await resources.close(); + } + }, +); + +it.each([ + null, + "opaque reason", + new ProviderCleanupError("foreign", ["foreign-resource"], { + reason: "caller cleanup", + }), +])( + "projects the command owner's exact cancellation independently of physical causes: %s", + async (reason) => { + const resources = new ArtifactAnalysisResources( + "fixture", + "inspect_signal_decoders", + readStableArtifact, + readLinuxResourceLimits, + { message: "busy", details: {} }, + ); + const physical = new Error("physical collection interruption"); + const failure = new OwnedCommandFailure( + "cancelled", + "cancelled command", + null, + null, + { cause: physical, cancellation: { reason } }, + ); + try { + const error = resources.commandFailure( + failure, + { command_timeout_ms: 100, output_bytes: 1024 }, + undefined, + ); + expect(error).toMatchObject({ _tag: "AnalysisCancelledError" }); + expect(error.cause).toBe(reason); + expect(error.cleanupResources).toEqual([]); + expect(failure.cause).toBe(physical); + if (reason instanceof ProviderCleanupError) + expect(reason.cleanupIncomplete).toBe(true); + } finally { + await resources.close(); + } + }, +); + +it.each([false, true])( + "preserves artifact cancellation and independent failed cleanup (failed close: %s)", + async (failedClose) => { + const reason = { request: "artifact cancellation" }; + let mayClose = false; + const readArtifact: typeof readStableArtifact = async ( + _path, + _limit, + _signal, + _fs, + scope, + ) => { + if (failedClose) { + if (scope === undefined) + throw new Error("Expected retained file scope"); + const retained = { + kind: "lease", + resource: "descriptor", + lease: { + release: async () => { + if (!mayClose) + throw new Error("injected descriptor close failure"); + }, + }, + } as const; + expect(await scope.release(retained)).toMatchObject({ kind: "failed" }); + } + throw new ArtifactReaderFailure("cancelled", "read cancelled", { + cause: reason, + ...(failedClose + ? { + cleanup: { + reason: "injected descriptor close failure", + resources: ["descriptor"], + }, + } + : {}), + }); + }; + const resources = new ArtifactAnalysisResources( + "fixture", + "inspect_signal_decoders", + readArtifact, + readLinuxResourceLimits, + { message: "busy", details: {} }, + ); + try { + const result = await resources.run(async () => + ok(await resources.read("fixture", 1)), + ); + if (result.ok) throw new Error("Expected cancelled artifact read"); + const primary = failedClose ? result.error.cleanupPrimary : result.error; + expect(primary).toMatchObject({ _tag: "AnalysisCancelledError" }); + expect(primary?.cause).toBe(reason); + expect(result.error.cleanupIncomplete).toBe(failedClose); + mayClose = true; + await resources.close(); + expect(result.error.cleanupIncomplete).toBe(false); + expect(primary?.cause).toBe(reason); + } finally { + mayClose = true; + await resources.close(); + } + }, +); + +it("retains a physical command failure when the owner also observed late cancellation", async () => { + const resources = new ArtifactAnalysisResources( + "fixture", + "inspect_signal_decoders", + readStableArtifact, + readLinuxResourceLimits, + { message: "busy", details: {} }, + ); + const reason = { request: "late cancellation" }; + const failure = new OwnedCommandFailure( + "process", + "physical failure", + null, + null, + { cancellation: { reason } }, + ); + try { + const error = resources.commandFailure( + failure, + { command_timeout_ms: 100, output_bytes: 1024 }, + undefined, + ); + expect(error).toMatchObject({ _tag: "AnalysisOutputError" }); + expect(error.cause).toBe(failure); + } finally { + await resources.close(); + } +}); + +it("preserves typed caller cancellation through failed workspace cleanup and retry", async () => { + let denied = true; + const resources = new ArtifactAnalysisResources( + "fixture", + "inspect_signal_decoders", + readStableArtifact, + readLinuxResourceLimits, + { message: "busy", details: {} }, + ); + const reason = { request: "cancelled before workspace cleanup" }; + const primary = new AnalysisCancelledError("inspect_signal_decoders", { + cause: reason, + }); + resources.retain({ + path: "workspace", + close: async () => { + if (denied) throw new Error("workspace close denied"); + }, + }); + try { + const result = await resources.finish(err(primary), () => { + throw new Error("Must not produce success Evidence"); + }); + if (result.ok) throw new Error("Expected retained workspace"); + expect(result.error.cleanupPrimary).toBe(primary); + expect(result.error.cleanupPrimary?.cause).toBe(reason); + await expect(resources.close()).rejects.toMatchObject({ + cleanupIncomplete: true, + }); + denied = false; + await resources.close(); + expect(result.error.cleanupIncomplete).toBe(false); + expect(result.error.cleanupPrimary?.cause).toBe(reason); + } finally { + denied = false; + await resources.close(); + } +}); + +it("observes cancellation after pending descriptor cleanup before admitting replacement work", async () => { + let closes = 0; + let cleanupStarted: () => void = () => {}; + const started = new Promise((resolve) => { + cleanupStarted = resolve; + }); + let releaseCleanup: () => void = () => {}; + const held = new Promise((resolve) => { + releaseCleanup = resolve; + }); + const readArtifact: typeof readStableArtifact = async ( + _path, + _limit, + _signal, + _fs, + scope, + ) => { + if (scope === undefined) throw new Error("Expected file scope"); + const retained = { + kind: "lease", + resource: "descriptor", + lease: { + release: async () => { + closes++; + if (closes === 1) throw new Error("descriptor close denied"); + cleanupStarted(); + await held; + }, + }, + } as const; + await scope.release(retained); + throw new ArtifactReaderFailure("io", "descriptor cleanup failed", { + cleanup: { reason: "descriptor close denied", resources: ["descriptor"] }, + }); + }; + const resources = new ArtifactAnalysisResources( + "fixture", + "inspect_signal_decoders", + readArtifact, + readLinuxResourceLimits, + { message: "busy", details: {} }, + ); + let calls = 0; + try { + const first = await resources.run(async () => + ok(await resources.read("fixture", 1)), + ); + expect(first).toMatchObject({ + ok: false, + error: { cleanupIncomplete: true }, + }); + const caller = new AbortController(); + const reason = { request: "cancel while old descriptor closes" }; + const pending = resources.run(async () => { + calls++; + return ok("Must not run"); + }, caller.signal); + await started; + caller.abort(reason); + releaseCleanup(); + const result = await pending; + if (result.ok) throw new Error("Expected cancelled replacement"); + expect(result.error.cause).toBe(reason); + expect(calls).toBe(0); + if (!first.ok) expect(first.error.cleanupIncomplete).toBe(false); + } finally { + releaseCleanup(); + await resources.close(); + } +}); + +it.each([false, true])( + "keeps an existing physical primary across cleanup-time cancellation (failed close: %s)", + async (failsClose) => { + const resources = new ArtifactAnalysisResources( + "fixture", + "inspect_signal_decoders", + readStableArtifact, + readLinuxResourceLimits, + { message: "busy", details: {} }, + ); + const caller = new AbortController(); + const primary = new AnalysisOutputError( + "inspect_signal_decoders", + "physical producer report failure", + ); + let denied = failsClose; + resources.retain({ + path: "workspace", + close: async () => { + caller.abort({ request: "late cancellation" }); + if (denied) throw new Error("workspace close denied"); + }, + }); + try { + const result = await resources.finish( + err(primary), + () => { + throw new Error("Failed work must not produce success Evidence"); + }, + "inspect_signal_decoders", + caller.signal, + ); + if (result.ok) throw new Error("Expected original physical failure"); + expect(failsClose ? result.error.cleanupPrimary : result.error).toBe( + primary, + ); + denied = false; + await resources.close(); + expect(result.error.cleanupIncomplete).toBe(false); + expect(failsClose ? result.error.cleanupPrimary : result.error).toBe( + primary, + ); + } finally { + denied = false; + await resources.close(); + } + }, +); diff --git a/src/process/ArtifactAnalysisResources.ts b/src/process/ArtifactAnalysisResources.ts index 8afad44e9..ad1277fc5 100644 --- a/src/process/ArtifactAnalysisResources.ts +++ b/src/process/ArtifactAnalysisResources.ts @@ -1,3 +1,4 @@ +import { artifactFileCancellationCause } from "../artifacts/ArtifactFileOperation.js"; import { ArtifactReaderFailure } from "../artifacts/ArtifactReader.js"; import { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js"; import { readStableArtifact } from "../artifacts/readStableArtifact.js"; @@ -75,7 +76,13 @@ export class ArtifactAnalysisResources { operation = this.operation, ): Promise> { if (this.#closed || signal?.aborted) - return Promise.resolve(err(new AnalysisCancelledError(operation))); + return Promise.resolve( + err( + new AnalysisCancelledError(operation, { + cause: signal?.aborted ? signal.reason : undefined, + }), + ), + ); if (this.#busy) return Promise.resolve( err( @@ -95,6 +102,12 @@ export class ArtifactAnalysisResources { // this scope has drained it and admits acquisition for the new request. this.#artifacts.operation = operation; this.#resolveCleanupFailures(); + if (signal?.aborted) + return Promise.resolve( + err( + new AnalysisCancelledError(operation, { cause: signal.reason }), + ), + ); return work(); }), ) @@ -221,7 +234,7 @@ export class ArtifactAnalysisResources { cause: unknown, capturedOutput?: AnalysisCapturedOutput, operation = this.operation, - ): ProviderCleanupError { + ): AnalysisError { const failure = new ProviderCleanupError( this.providerId, observation.resources, @@ -234,6 +247,21 @@ export class ArtifactAnalysisResources { { operation, cause }, ); this.#trackCleanupFailure(failure, [], failure, true); + if ( + cause instanceof ArtifactReaderFailure && + cause.reason === "cancelled" + ) { + const primary = new AnalysisCancelledError(operation, { + cause: artifactFileCancellationCause(cause), + ...(capturedOutput === undefined ? {} : { capturedOutput }), + partialObservation: cause.partialObservation, + }); + return analysisErrorWithCleanupFailure( + primary, + failure, + operation, + ).bindCleanupDependencies([primary, failure]); + } return failure; } @@ -261,7 +289,11 @@ export class ArtifactAnalysisResources { ...(capturedOutput === undefined ? {} : { capturedOutput }), }; if (cause.reason === "cancelled") - return new AnalysisCancelledError(operation, detail); + return new AnalysisCancelledError(operation, { + ...detail, + cause: + cause.cancellation === undefined ? cause : cause.cancellation.reason, + }); if (cause.reason === "timeout") return new AnalysisTimeoutError( operation, @@ -295,14 +327,30 @@ export class ArtifactAnalysisResources { result: Result, observation: (value: T) => AnalysisPartialObservation, operation = this.operation, + signal?: AbortSignal, ): Promise> { try { await this.release(); + if (result.ok && signal?.aborted) + return err( + new AnalysisCancelledError(operation, { + cause: signal.reason, + partialObservation: observation(result.value), + }), + ); return result; } catch (cause: unknown) { const partialObservation = result.ok ? observation(result.value) : result.error.partialObservation; + const primaryFailure = !result.ok + ? result.error + : signal?.aborted + ? new AnalysisCancelledError(operation, { + cause: signal.reason, + partialObservation, + }) + : undefined; const cleanupDiagnostics = cause instanceof ProviderCleanupError ? cause.diagnostics : undefined; const cleanupFailure = new ProviderCleanupError( @@ -321,21 +369,27 @@ export class ArtifactAnalysisResources { cause instanceof ProviderCleanupError ? String(cleanupDiagnostics?.reason ?? cause.message) : String(cause), - previous_error: result.ok ? null : projectAnalysisError(result.error), + previous_error: + primaryFailure === undefined + ? null + : projectAnalysisError(primaryFailure), }, { operation, - cause: result.ok - ? cause - : new AggregateError( - [result.error, cause], - "Analysis failed and artifact cleanup remains incomplete.", - { cause: result.error }, - ), + ...(primaryFailure === undefined + ? {} + : { cleanupPrimary: primaryFailure }), + cause: + primaryFailure === undefined + ? cause + : new AggregateError( + [primaryFailure, cause], + "Analysis failed and artifact cleanup remains incomplete.", + { cause: primaryFailure }, + ), ...(partialObservation === undefined ? {} : { partialObservation }), }, ); - const primaryFailure = result.ok ? undefined : result.error; this.#trackCleanupFailure( cleanupFailure, [...this.#pending.keys()], @@ -408,6 +462,10 @@ export class ArtifactAnalysisResources { if (cause instanceof AnalysisError) return cause; if (cause instanceof ArtifactReaderFailure && cause.cleanup !== undefined) return this.cleanupFailure(cause.cleanup, cause, undefined, operation); + if (cause instanceof ArtifactReaderFailure && cause.reason === "cancelled") + return new AnalysisCancelledError(operation, { + cause: artifactFileCancellationCause(cause), + }); return new ProviderAdapterError(this.providerId, operation, { cause, diagnostics: { stage: "artifact-resources", reason: String(cause) },