diff --git a/src/erlang/BeamProvider.ts b/src/erlang/BeamProvider.ts index 637f7c9944..28ed4648c2 100644 --- a/src/erlang/BeamProvider.ts +++ b/src/erlang/BeamProvider.ts @@ -1,5 +1,8 @@ import { ArtifactReaderFailure } from "../artifacts/ArtifactReader.js"; -import { readStableArtifact } from "../artifacts/readStableArtifact.js"; +import { + completedStableArtifact, + readStableArtifact, +} from "../artifacts/readStableArtifact.js"; import { artifactFilePrimaryFacts, type ArtifactFilePrimaryFacts, @@ -9,6 +12,7 @@ import { } from "../artifacts/ArtifactFileOperation.js"; import { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js"; import { analysisErrorWithCleanupFailure } from "../domain/analysisErrorCleanup.js"; +import { fileSnapshotPartialObservationSchema } from "../domain/fileSnapshotObservation.js"; import type { BeamPort } from "../application/erlang/BeamPort.js"; import type { ExecutionOptions } from "../application/AnalysisProvider.js"; import { ArtifactOperationError } from "../domain/artifactOperationError.js"; @@ -203,6 +207,16 @@ const beamFailurePrimary = ( const beamFailure = (cause: unknown, input: InspectErlangBeamInput) => { const primary = beamFailurePrimary(artifactFilePrimaryFacts(cause), input); + const snapshot = completedStableArtifact(cause); + if (snapshot !== undefined) + primary.retainPartialObservation( + fileSnapshotPartialObservationSchema.parse({ + kind: "file-snapshot-hash", + path: input.path, + sha256: snapshot.sha256, + bytes: snapshot.bytes.length, + }), + ); if (!(cause instanceof ArtifactReaderFailure) || cause.cleanup === undefined) return primary; const cleanup = new ProviderCleanupError( diff --git a/src/python/PythonPycProvider.ts b/src/python/PythonPycProvider.ts index b463bb8186..461fc9e12a 100644 --- a/src/python/PythonPycProvider.ts +++ b/src/python/PythonPycProvider.ts @@ -6,9 +6,13 @@ import { bindArtifactFileCleanup, } from "../artifacts/ArtifactFileOperation.js"; import { ArtifactReaderFailure } from "../artifacts/ArtifactReader.js"; -import { readStableArtifact } from "../artifacts/readStableArtifact.js"; +import { + completedStableArtifact, + readStableArtifact, +} from "../artifacts/readStableArtifact.js"; import { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js"; import { analysisErrorWithCleanupFailure } from "../domain/analysisErrorCleanup.js"; +import { fileSnapshotPartialObservationSchema } from "../domain/fileSnapshotObservation.js"; import type { PythonPycPort } from "../application/python/PythonPycPort.js"; import type { ExecutionOptions } from "../application/AnalysisProvider.js"; import { AnalysisError } from "../domain/analysisErrorBase.js"; @@ -164,6 +168,16 @@ const failureForPrimary = ( const failureFor = (cause: unknown, input: InspectPythonPycInput) => { const primary = failureForPrimary(artifactFilePrimaryFacts(cause), input); + const snapshot = completedStableArtifact(cause); + if (snapshot !== undefined) + primary.retainPartialObservation( + fileSnapshotPartialObservationSchema.parse({ + kind: "file-snapshot-hash", + path: input.path, + sha256: snapshot.sha256, + bytes: snapshot.bytes.length, + }), + ); if (!(cause instanceof ArtifactReaderFailure) || cause.cleanup === undefined) return primary; const cleanup = new ProviderCleanupError( @@ -182,8 +196,24 @@ export class PythonPycProvider implements PythonPycPort { readonly identity = PYTHON_PYC_PROVIDER_IDENTITY; readonly #resources = new ArtifactResourceScope(); - close(): Promise { - return this.#resources.close(); + async close(): Promise { + try { + await this.#resources.close(); + } catch (cause: unknown) { + if ( + !(cause instanceof ArtifactReaderFailure) || + cause.cleanup === undefined + ) + throw cause; + const error = new ProviderCleanupError( + PYTHON_PYC_PROVIDER_IDENTITY.id, + cause.cleanup.resources, + { reason: cause.cleanup.reason }, + { cause, operation: OPERATION }, + ); + bindArtifactFileCleanup(cause, error); + throw error; + } } async inspect(input: InspectPythonPycInput, options?: ExecutionOptions) { diff --git a/tests/boundary/filesystem/pythonBeamCleanup.test.ts b/tests/boundary/filesystem/pythonBeamCleanup.test.ts new file mode 100644 index 0000000000..4d8bec7057 --- /dev/null +++ b/tests/boundary/filesystem/pythonBeamCleanup.test.ts @@ -0,0 +1,154 @@ +import { createHash } from "node:crypto"; +import { fstatSync } from "node:fs"; +import type { FileHandle } from "node:fs/promises"; +import * as fileSystem from "node:fs/promises"; +import { join } from "node:path"; +import { afterEach, expect, it, vi } from "vitest"; +import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js"; +import { ProviderCleanupError } from "../../../src/domain/providerCleanupError.js"; +import { pythonPycFixture } from "../../../src/python/PythonPyc.fixture.js"; +import { PythonPycProvider } from "../../../src/python/PythonPycProvider.js"; +import { PythonPycService } from "../../../src/application/python/PythonPycService.js"; +import { BeamService } from "../../../src/application/erlang/BeamService.js"; +import { BeamProvider } from "../../../src/erlang/BeamProvider.js"; +import { erlangBeamFixture } from "../../../src/erlang/Beam.fixture.js"; + +const injected = vi.hoisted(() => ({ + opened: undefined as + | ((path: unknown, handle: FileHandle) => void) + | undefined, +})); + +vi.mock("node:fs/promises", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + open: async (...args: Parameters) => { + const handle = await actual.open(...args); + injected.opened?.(args[0], handle); + return handle; + }, + }; +}); + +afterEach(() => { + injected.opened = undefined; + vi.restoreAllMocks(); +}); + +const providers = [ + { + name: "Python pyc", + operation: "inspect_python_pyc", + bytes: pythonPycFixture(), + create: () => new PythonPycService(new PythonPycProvider()), + }, + { + name: "Erlang BEAM", + operation: "inspect_erlang_beam", + bytes: erlangBeamFixture(), + create: () => new BeamService(new BeamProvider()), + }, +]; + +const outcomes = [ + { stage: "completed read", cancelDuringClose: false, failRead: false }, + { + stage: "cancellation during close", + cancelDuringClose: true, + failRead: false, + }, + { + stage: "read failure then cancellation", + cancelDuringClose: true, + failRead: true, + }, +]; + +it.each( + providers.flatMap((provider) => + outcomes.map((outcome) => ({ ...provider, ...outcome })), + ), +)( + "$name preserves acquired facts and resolves cleanup after retries: $stage", + async ({ operation, bytes, create, cancelDuringClose, failRead }) => { + const root = await createTestTempDirectory("rea-python-beam-close-owner-"); + const path = join(root, "selected-artifact"); + await fileSystem.writeFile(path, bytes); + const releases: (() => Promise)[] = []; + const descriptors: number[] = []; + const controller = new AbortController(); + const reason = { source: "cancelled during descriptor finalization" }; + const readFailure = new Error("injected physical artifact read failure"); + let denyClose = true; + injected.opened = (openedPath, file) => { + if (String(openedPath) !== path) return; + descriptors.push(file.fd); + if (failRead) vi.spyOn(file, "read").mockRejectedValueOnce(readFailure); + const close = file.close.bind(file); + releases.push(close); + vi.spyOn(file, "close").mockImplementation(async () => { + if (cancelDuringClose) controller.abort(reason); + if (denyClose) + throw new Error("injected artifact descriptor close failure"); + await close(); + }); + }; + const service = create(); + try { + const result = await service.inspect( + { path }, + { signal: controller.signal }, + ); + if (result.ok) throw new Error("Expected incomplete descriptor cleanup"); + expect(result.error.cleanupIncomplete).toBe(true); + expect(result.error.cleanupResources).toEqual([path]); + const expectedSnapshot = failRead + ? undefined + : { + kind: "file-snapshot-hash", + path, + sha256: createHash("sha256").update(bytes).digest("hex"), + bytes: bytes.length, + }; + expect(result.error.partialObservation).toEqual(expectedSnapshot); + if (failRead) + expect(result.error.cleanupPrimary?.cause).toBe(readFailure); + else if (cancelDuringClose) { + expect(result.error.cleanupPrimary?._tag).toBe( + "AnalysisCancelledError", + ); + expect(result.error.cleanupPrimary?.cause).toBe(reason); + } + expect(fstatSync(descriptors[0]!)).toBeDefined(); + const closeErrors: ProviderCleanupError[] = []; + for (let attempt = 0; attempt < 2; attempt += 1) { + const failure: unknown = await service.close().then( + () => undefined, + (cause: unknown) => cause, + ); + expect(failure).toBeInstanceOf(ProviderCleanupError); + if (!(failure instanceof ProviderCleanupError)) + throw new Error("Expected a translated provider cleanup failure"); + expect(failure.operation).toBe(operation); + expect(failure.cleanupIncomplete).toBe(true); + expect(failure.cleanupResources).toEqual([path]); + closeErrors.push(failure); + } + expect(descriptors).toHaveLength(1); + expect(result.error.cleanupIncomplete).toBe(true); + denyClose = false; + await service.close(); + expect(() => fstatSync(descriptors[0]!)).toThrow(); + for (const failure of [result.error, ...closeErrors]) { + expect(failure.cleanupIncomplete).toBe(false); + expect(failure.cleanupResources).toEqual([]); + } + expect(result.error.partialObservation).toEqual(expectedSnapshot); + } finally { + denyClose = false; + await service.close(); + for (const close of releases) await close(); + } + }, +);