diff --git a/src/artifacts/ArtifactFileOperation.ts b/src/artifacts/ArtifactFileOperation.ts new file mode 100644 index 0000000000..ece86f6979 --- /dev/null +++ b/src/artifacts/ArtifactFileOperation.ts @@ -0,0 +1,248 @@ +import type { FileHandle } from "node:fs/promises"; +import { OwnedFileHandle } from "../filesystem/OwnedFileHandle.js"; +import { + withOwnedFileHandle, + withOwnedRegularFile, + RegularFileCancelled, + RegularFileCleanupFailure, +} from "../filesystem/OwnedRegularFile.js"; +import { ArtifactReaderFailure } from "./ArtifactReader.js"; +import { + ArtifactResourceRetryFailure, + type ArtifactResourceScope, +} from "./ArtifactResourceScope.js"; +import { + AnalysisCleanupLease, + AnalysisCleanupLeaseGroup, +} from "./AnalysisCleanupLease.js"; +import type { AnalysisError } from "../domain/analysisErrorBase.js"; + +/** Artifact projection plus the exact descriptor owner and physical operation outcome. */ +export class ArtifactFileCleanupFailure< + Value = unknown, +> extends ArtifactReaderFailure { + readonly cleanupLeases: readonly AnalysisCleanupLease[] | undefined; + constructor( + readonly fileCleanupFailure: RegularFileCleanupFailure, + readonly cleanupLease: AnalysisCleanupLease, + readonly completionCancellation?: { readonly reason: unknown }, + ) { + const primary = + fileCleanupFailure.outcome.kind === "failed" + ? fileCleanupFailure.outcome.cause + : completionCancellation === undefined + ? fileCleanupFailure.cleanupCause + : new ArtifactReaderFailure( + "cancelled", + "Artifact read cancelled during close", + { + cause: completionCancellation.reason, + }, + ); + const projected = ArtifactReaderFailure.withCleanup( + primary, + ArtifactReaderFailure.cleanupObservation( + fileCleanupFailure.cleanupCause, + fileCleanupFailure.path, + ), + ); + super( + projected.reason, + projected.message, + { + cause: fileCleanupFailure, + cleanup: projected.cleanup, + partialObservation: projected.partialObservation, + cancellation: projected.cancellation, + ...(projected.boundary === undefined + ? {} + : { boundary: projected.boundary }), + }, + projected.details, + ); + const inherited = + primary instanceof ArtifactFileCleanupFailure || + primary instanceof ArtifactResourceRetryFailure + ? primary.cleanupLeases + : primary instanceof ArtifactReaderFailure && + primary.cleanup !== undefined + ? undefined + : []; + this.cleanupLeases = + inherited === undefined + ? undefined + : [...new Set([cleanupLease, ...inherited])]; + } +} + +/** Normalize the physical primary failure without replacing its exact cleanup owner. */ +export const mapArtifactFileFailure = ( + cause: unknown, + classify: (cause: unknown) => unknown, +): unknown => { + if (!(cause instanceof ArtifactFileCleanupFailure)) return classify(cause); + const physical = cause.fileCleanupFailure; + if (physical.outcome.kind !== "failed") return cause; + const primary = + physical.outcome.cause instanceof ArtifactFileCleanupFailure + ? mapArtifactFileFailure(physical.outcome.cause, classify) + : classify(physical.outcome.cause); + if (primary === physical.outcome.cause) return cause; + return new ArtifactFileCleanupFailure( + new RegularFileCleanupFailure( + physical.path, + physical.owner, + { kind: "failed", cause: primary }, + physical.cleanupCause, + physical.retention, + ), + cause.cleanupLease, + cause.completionCancellation, + ); +}; + +/** Read only the physical outcomes carried by known nested file owners. */ +export const artifactFilePrimaryFailure = (cause: unknown): unknown => { + while ( + cause instanceof ArtifactFileCleanupFailure && + cause.fileCleanupFailure.outcome.kind === "failed" + ) + cause = cause.fileCleanupFailure.outcome.cause; + return cause; +}; + +/** Cancellation after confirmed close retains the completed physical operation. */ +export class ArtifactFileCancelled< + Value = unknown, +> extends ArtifactReaderFailure { + constructor( + readonly fileCancellation: Pick< + RegularFileCancelled, + "reason" | "outcome" + >, + ) { + super("cancelled", "Artifact read cancelled", { + cause: fileCancellation.reason, + }); + } +} + +/** Bind a translated failure to its exact descriptor leases, including failed retries. */ +export const bindArtifactFileCleanup = ( + cause: unknown, + error: AnalysisError, +): void => { + const leases = + cause instanceof ArtifactFileCleanupFailure + ? cause.cleanupLeases + : cause instanceof ArtifactResourceRetryFailure + ? cause.cleanupLeases + : undefined; + if (leases !== undefined) new AnalysisCleanupLeaseGroup(error, leases); +}; + +/** Interpret this owner's typed cancellation, preserving an undefined reason. */ +export const artifactFileCancellationCause = ( + cause: ArtifactReaderFailure, +): unknown => { + return cause.cancellation === undefined + ? cause.cause + : cause.cancellation.reason; +}; + +/** Report cancellation at the read stage that actually observed the caller signal. */ +export const assertArtifactReadActive = ( + signal: AbortSignal | undefined, +): void => { + if (signal?.aborted) + throw new ArtifactReaderFailure("cancelled", "Artifact read cancelled", { + cause: signal.reason, + }); +}; + +/** Admit a regular file without blocking on pipes, retaining failed admission closes. */ +export const withArtifactRegularFile = async ( + path: string, + resources: ArtifactResourceScope, + signal: AbortSignal | undefined, + operation: (file: FileHandle) => Promise, + symlinks: "follow" | "reject" = "reject", +): Promise => { + let cleanupLease: AnalysisCleanupLease | undefined; + assertArtifactReadActive(signal); + try { + const value = await withOwnedRegularFile( + path, + { + symlinks, + retainFailedClose: (owner) => { + cleanupLease = new AnalysisCleanupLease(() => owner.close()); + resources.retain({ + kind: "lease", + resource: path, + lease: cleanupLease, + }); + }, + }, + async (file) => { + assertArtifactReadActive(signal); + return await operation(file); + }, + ); + if (signal?.aborted) + throw new ArtifactFileCancelled({ + reason: signal.reason, + outcome: { kind: "completed", value }, + }); + return value; + } catch (cause: unknown) { + if ( + cause instanceof RegularFileCleanupFailure && + cleanupLease !== undefined + ) + throw new ArtifactFileCleanupFailure( + cause, + cleanupLease, + signal?.aborted ? { reason: signal.reason } : undefined, + ); + throw cause; + } +}; + +/** Keep admission, operation outcome, final close and failed-owner transfer together. */ +export const withArtifactFileHandle = async ( + path: string, + file: FileHandle, + resources: ArtifactResourceScope | undefined, + signal: AbortSignal | undefined, + operation: (file: FileHandle) => Promise, +): Promise => { + const owner = new OwnedFileHandle(file); + const cleanupLease = new AnalysisCleanupLease(() => owner.close()); + try { + return await withOwnedFileHandle( + path, + owner, + { + ...(signal === undefined ? {} : { signal }), + retainFailedClose: () => + resources?.retain({ + kind: "lease", + resource: path, + lease: cleanupLease, + }), + }, + operation, + ); + } catch (cause: unknown) { + if (cause instanceof RegularFileCleanupFailure) + throw new ArtifactFileCleanupFailure( + cause, + cleanupLease, + signal?.aborted ? { reason: signal.reason } : undefined, + ); + if (cause instanceof RegularFileCancelled) + throw new ArtifactFileCancelled(cause); + throw cause; + } +}; diff --git a/src/artifacts/ArtifactHash.ts b/src/artifacts/ArtifactHash.ts index c1234f847e..2c06ae2c26 100644 --- a/src/artifacts/ArtifactHash.ts +++ b/src/artifacts/ArtifactHash.ts @@ -24,6 +24,7 @@ export const abortIfNeeded = (signal?: AbortSignal): void => { throw new ArtifactReaderFailure( "cancelled", "Artifact inventory cancelled", + { cause: signal.reason }, ); }; @@ -63,5 +64,6 @@ export const hashReadable = async ( xmlProbe.write(selected); } } + abortIfNeeded(signal); return { sha256: hash.digest("hex"), bytes, prefix: Buffer.concat(prefixes) }; }; diff --git a/src/artifacts/ArtifactReader.test.ts b/src/artifacts/ArtifactReader.test.ts new file mode 100644 index 0000000000..ef410cc0c8 --- /dev/null +++ b/src/artifacts/ArtifactReader.test.ts @@ -0,0 +1,28 @@ +import { expect, it } from "vitest"; +import { ArtifactReaderFailure } from "./ArtifactReader.js"; + +it.each([ + undefined, + null, + "caller reason", + new ArtifactReaderFailure("cancelled", "caller-provided error"), +])( + "preserves explicit cancellation through repeated cleanup composition: %s", + (reason) => { + const primary = new ArtifactReaderFailure("cancelled", "Read cancelled", { + cause: reason, + }); + const first = ArtifactReaderFailure.withCleanup(primary, { + reason: "first close failed", + resources: ["first"], + }); + const second = ArtifactReaderFailure.withCleanup(first, { + reason: "second close failed", + resources: ["second"], + }); + expect(second.cancellation).toBe(primary.cancellation); + expect(second.cancellation?.reason).toBe(reason); + expect(second.cleanup?.resources).toEqual(["first", "second"]); + expect(second.cause).toBe(first); + }, +); diff --git a/src/artifacts/ArtifactReader.ts b/src/artifacts/ArtifactReader.ts index 417fa54748..7b2edcbb92 100644 --- a/src/artifacts/ArtifactReader.ts +++ b/src/artifacts/ArtifactReader.ts @@ -13,12 +13,10 @@ import type { export type ArtifactReaderBoundary = "file-size" | "capacity" | "memory"; export interface ArtifactReaderFailureOptions extends ErrorOptions { - readonly cleanup?: AnalysisCleanupObservation; - readonly partialObservation?: AnalysisPartialObservation; - /** - * The constrained bound for a `limit` reason. Callers report the resource - * from this instead of assuming every budget is memory. - */ + readonly cancellation?: { readonly reason: unknown } | undefined; + readonly cleanup?: AnalysisCleanupObservation | undefined; + readonly partialObservation?: AnalysisPartialObservation | undefined; + /** The constrained resource bound for a limit failure. */ readonly boundary?: ArtifactReaderBoundary; } @@ -138,6 +136,7 @@ export interface ArtifactReader { /** Typed adapter failure translated at provider boundary. */ export class ArtifactReaderFailure extends Error { + readonly cancellation: { readonly reason: unknown } | undefined; readonly cleanup: AnalysisCleanupObservation | undefined; readonly partialObservation: AnalysisPartialObservation | undefined; /** The constrained bound for a `limit` reason, when the reader knows it. */ @@ -163,6 +162,10 @@ export class ArtifactReaderFailure extends Error { ) { super(message, options); this.name = "ArtifactReaderFailure"; + this.cancellation = + reason === "cancelled" + ? (options?.cancellation ?? { reason: options?.cause }) + : undefined; this.cleanup = options?.cleanup; this.partialObservation = options?.partialObservation; this.boundary = options?.boundary; @@ -184,6 +187,7 @@ export class ArtifactReaderFailure extends Error { primary.message, { cause: primary, + cancellation: primary.cancellation, cleanup: combineCleanup(primary.cleanup, cleanup), // The constrained bound survives cleanup wrapping, or a file-size // refusal would be reported as a memory constraint once a close fails. diff --git a/src/artifacts/ArtifactResourceScope.ts b/src/artifacts/ArtifactResourceScope.ts index 648521bf98..c183d1f6b9 100644 --- a/src/artifacts/ArtifactResourceScope.ts +++ b/src/artifacts/ArtifactResourceScope.ts @@ -7,6 +7,7 @@ import type { ArtifactReader } from "./ArtifactReader.js"; import { ArtifactReaderFailure } from "./ArtifactReader.js"; import type { SafeOutputTree } from "./SafeOutputTree.js"; import type { PrivateRuntimeRoot } from "../process/PrivateRuntimeRoot.js"; +import { AnalysisCleanupLease } from "./AnalysisCleanupLease.js"; /** One artifact resource and the capability that owns its cleanup. */ export type ArtifactResourceOwner = @@ -57,12 +58,22 @@ export type ArtifactCleanupAttempt = /** Retry failure carrying the exact retained owners that did not release. */ export class ArtifactResourceRetryFailure extends ArtifactReaderFailure { + readonly cleanupLeases: readonly AnalysisCleanupLease[] | undefined; constructor( readonly failedOwners: readonly ArtifactResourceOwner[], message: string, options: ConstructorParameters[2], ) { super("unavailable", message, options); + const leases = failedOwners.flatMap((owner) => + owner.kind === "lease" && owner.lease instanceof AnalysisCleanupLease + ? [owner.lease] + : [], + ); + this.cleanupLeases = + leases.length > 0 && leases.length === failedOwners.length + ? [...new Set(leases)] + : undefined; } } diff --git a/src/artifacts/AsarArtifactReader.ts b/src/artifacts/AsarArtifactReader.ts index c988ba056e..cb54d80b86 100644 --- a/src/artifacts/AsarArtifactReader.ts +++ b/src/artifacts/AsarArtifactReader.ts @@ -25,7 +25,8 @@ import { import { closeAsarHandle, readValidatedAsarEntry } from "./AsarEntryStream.js"; import { admitAsarHeader } from "./AsarHeader.js"; import { OwnedFileHandle } from "../filesystem/OwnedFileHandle.js"; -import { readFileHandleChunks } from "../filesystem/readFileHandleChunks.js"; +import type { FileHandleChunkStream } from "../filesystem/readFileHandleChunks.js"; +import { readArtifactFileChunks } from "./readArtifactFileChunks.js"; import { readAsarListing, type AsarListedEntry, @@ -336,19 +337,14 @@ export class AsarArtifactReader implements ArtifactReader { source, digesting, createWriteStream(snapshotPath, { flags: "wx", mode: 0o600 }), - signal === undefined ? {} : { signal }, ); const completedInput = inputHandle; inputHandle = undefined; await this.#closeOwnedHandle(completedInput); + abortIfNeeded(signal); this.#snapshot = { path: snapshotPath, sha256: hash.digest("hex") }; } catch (cause: unknown) { - let failure = - signal?.aborted === true - ? new ArtifactReaderFailure("cancelled", "ASAR operation cancelled", { - cause, - }) - : asarFailure(this.path, "snapshot", cause); + let failure = asarFailure(this.path, "snapshot", cause); if (inputHandle !== undefined) { const failedInput = inputHandle; inputHandle = undefined; @@ -491,7 +487,7 @@ export class AsarArtifactReader implements ArtifactReader { ); } abortIfNeeded(signal); - const source = this.#ownedReadStream(openedHandle, { start: 0 }); + const source = this.#ownedReadStream(openedHandle, { start: 0, signal }); handle = undefined; const output = readValidatedAsarEntry( source, @@ -573,9 +569,9 @@ export class AsarArtifactReader implements ArtifactReader { readonly signal?: AbortSignal | undefined; readonly releaseOnClose?: boolean; }, - ): Readable { + ): FileHandleChunkStream { const { releaseOnClose, ...readOptions } = options; - const source = readFileHandleChunks(handle, readOptions); + const source = readArtifactFileChunks(handle, readOptions); const owned = this.#ownedHandles.get(handle); if (owned === undefined) throw new ArtifactReaderFailure("integrity", "ASAR handle has no owner"); @@ -662,7 +658,9 @@ const parseArchiveOffset = (value: string): number | undefined => { const abortIfNeeded = (signal?: AbortSignal): void => { if (signal?.aborted === true) - throw new ArtifactReaderFailure("cancelled", "ASAR operation cancelled"); + throw new ArtifactReaderFailure("cancelled", "ASAR operation cancelled", { + cause: signal.reason, + }); }; const waitForOperation = async ( @@ -677,7 +675,9 @@ const waitForOperation = async ( const onAbort = (): void => { removeAbortListener(); reject( - new ArtifactReaderFailure("cancelled", "ASAR operation cancelled"), + new ArtifactReaderFailure("cancelled", "ASAR operation cancelled", { + cause: signal.reason, + }), ); }; signal.addEventListener("abort", onAbort, { once: true }); diff --git a/src/artifacts/AsarEntryStream.test.ts b/src/artifacts/AsarEntryStream.test.ts new file mode 100644 index 0000000000..d5a29bc899 --- /dev/null +++ b/src/artifacts/AsarEntryStream.test.ts @@ -0,0 +1,43 @@ +import type { FileHandle } from "node:fs/promises"; +import { expect, it } from "vitest"; +import { readArtifactFileChunks } from "./readArtifactFileChunks.js"; +import { readValidatedAsarEntry } from "./AsarEntryStream.js"; + +it.each(["complete", "truncated"])( + "preserves %s source facts when output is cancelled after physical EOF", + async (kind) => { + const controller = new AbortController(); + const bytes = Buffer.from("ASAR member"); + let delivered = false; + const read = (async (target: Buffer) => { + if (delivered) return { bytesRead: 0, buffer: target }; + delivered = true; + bytes.copy(target); + return { bytesRead: bytes.length, buffer: target }; + }) as FileHandle["read"]; + const source = readArtifactFileChunks( + { read }, + { signal: controller.signal }, + ); + const output = readValidatedAsarEntry( + source, + bytes.length + (kind === "truncated" ? 1 : 0), + "member", + "/fixture/archive.asar", + controller.signal, + ); + const failed = new Promise((resolve) => + output.once("error", resolve), + ); + output.read(0); + await expect.poll(() => source.readCoverage().kind).toBe("complete"); + expect(output.readableLength).toBeGreaterThan(0); + expect(output.readableEnded).toBe(false); + controller.abort(null); + await expect(failed).resolves.toMatchObject( + kind === "complete" + ? { reason: "cancelled", cancellation: { reason: null } } + : { reason: "integrity" }, + ); + }, +); diff --git a/src/artifacts/AsarEntryStream.ts b/src/artifacts/AsarEntryStream.ts index b11b233d6a..12b524af9b 100644 --- a/src/artifacts/AsarEntryStream.ts +++ b/src/artifacts/AsarEntryStream.ts @@ -1,3 +1,4 @@ +import type { FileHandleChunkStream } from "../filesystem/readFileHandleChunks.js"; import { Readable } from "node:stream"; import { ArtifactReaderFailure } from "./ArtifactReader.js"; @@ -13,19 +14,39 @@ export const closeAsarHandle = async ( /** Validate and contextualize bytes read from one ASAR member source. */ export const readValidatedAsarEntry = ( - source: Readable, + source: FileHandleChunkStream, expectedBytes: number | undefined, entryPath: string, archivePath: string, signal?: AbortSignal, ): Readable => { + const expectedExtent = + expectedBytes === undefined ? undefined : BigInt(expectedBytes); + const completionFailure = (): ArtifactReaderFailure | undefined => { + const coverage = source.readCoverage(); + if (coverage.kind !== "complete") + return new ArtifactReaderFailure( + "integrity", + `ASAR entry read coverage is incomplete: ${entryPath}`, + ); + if (expectedExtent !== undefined && coverage.bytes !== expectedExtent) + return new ArtifactReaderFailure( + "integrity", + `ASAR entry size disagrees with its header: ${entryPath}`, + ); + if (signal?.aborted) + return new ArtifactReaderFailure( + "cancelled", + "ASAR operation cancelled", + { cause: signal.reason }, + ); + return undefined; + }; const output = Readable.from( (async function* () { - let observedBytes = 0; const iterator = source[Symbol.asyncIterator](); try { while (true) { - abortIfNeeded(signal); const next = await iterator.next(); if (next.done) break; const chunk: unknown = next.value; @@ -34,21 +55,20 @@ export const readValidatedAsarEntry = ( "format", `ASAR entry did not return bytes: ${entryPath}`, ); - observedBytes += Buffer.isBuffer(chunk) - ? chunk.length - : chunk.byteLength; - if (expectedBytes !== undefined && observedBytes > expectedBytes) + const coverage = source.readCoverage(); + if (expectedExtent !== undefined && coverage.bytes > expectedExtent) throw new ArtifactReaderFailure( "integrity", `ASAR entry exceeded its declared size: ${entryPath}`, ); + if (coverage.kind === "complete" && signal?.aborted) { + const failure = completionFailure(); + if (failure !== undefined) throw failure; + } yield chunk; } - if (expectedBytes !== undefined && observedBytes !== expectedBytes) - throw new ArtifactReaderFailure( - "integrity", - `ASAR entry size disagrees with its header: ${entryPath}`, - ); + const failure = completionFailure(); + if (failure !== undefined) throw failure; } catch (cause: unknown) { throw asarFailure(archivePath, `read ${entryPath}`, cause); } finally { @@ -61,25 +81,46 @@ export const readValidatedAsarEntry = ( } })(), ); - const closeSource = (): void => { - signal?.removeEventListener("abort", onAbort); - if (!source.destroyed) source.destroy(); - }; - const onAbort = (): void => { - closeSource(); + const onSourceError = (cause: unknown): void => { + const completedCancellation = + cause instanceof ArtifactReaderFailure && + cause.reason === "cancelled" && + source.readCoverage().kind === "complete" + ? completionFailure() + : undefined; output.destroy( - new ArtifactReaderFailure("cancelled", "ASAR operation cancelled"), + completedCancellation ?? + asarFailure(archivePath, `read ${entryPath}`, cause), ); }; - output.once("close", closeSource); + source.on("error", onSourceError); + source.once("close", () => source.removeListener("error", onSourceError)); + const onAbort = (): void => { + if ( + source.readCoverage().kind === "complete" && + !output.destroyed && + (source.errored === null || + (source.errored instanceof ArtifactReaderFailure && + source.errored.reason === "cancelled")) + ) { + const failure = completionFailure(); + if (failure !== undefined) output.destroy(failure); + } + }; signal?.addEventListener("abort", onAbort, { once: true }); - if (signal?.aborted === true) onAbort(); + output.once("close", () => { + signal?.removeEventListener("abort", onAbort); + if (!source.destroyed) source.destroy(); + }); + if (signal?.aborted) onAbort(); return output; }; const abortIfNeeded = (signal?: AbortSignal): void => { if (signal?.aborted === true) - throw new ArtifactReaderFailure("cancelled", "ASAR operation cancelled"); + throw new ArtifactReaderFailure("cancelled", "ASAR operation cancelled", { + cause: signal.reason, + }); }; const asarFailure = ( diff --git a/src/artifacts/DirectoryArtifactReader.ts b/src/artifacts/DirectoryArtifactReader.ts index 56ef4ccc76..dfab54f223 100644 --- a/src/artifacts/DirectoryArtifactReader.ts +++ b/src/artifacts/DirectoryArtifactReader.ts @@ -20,7 +20,7 @@ import { import { hostExecutableMarking } from "./hostExecutableMarking.js"; import { OwnedDirectoryHandle } from "../filesystem/OwnedDirectoryHandle.js"; import { OwnedFileHandle } from "../filesystem/OwnedFileHandle.js"; -import { readFileHandleChunks } from "../filesystem/readFileHandleChunks.js"; +import { readArtifactFileChunks } from "./readArtifactFileChunks.js"; interface DirectoryTraversalOptions extends ArtifactTraversalOptions { readonly descendInto?: (path: string) => boolean; @@ -377,7 +377,7 @@ const mapStableFileRead = async function* ( "integrity", `Directory entry has no captured file state: ${entry.path}`, ); - for await (const chunk of readFileHandleChunks(handle, { + for await (const chunk of readArtifactFileChunks(handle, { // Include one extra byte so a growing file cannot make this read unbounded. end: expectedBytes, signal, @@ -404,12 +404,6 @@ const mapStableFileRead = async function* ( `Directory entry changed during read: ${entry.path}`, ); } catch (cause: unknown) { - if (signal?.aborted === true) - throw new ArtifactReaderFailure( - "cancelled", - "Directory artifact read cancelled", - { cause }, - ); if ( cause instanceof Error && "code" in cause && @@ -443,6 +437,7 @@ const abortIfNeeded = (signal?: AbortSignal): void => { throw new ArtifactReaderFailure( "cancelled", "Artifact traversal cancelled", + { cause: signal.reason }, ); }; diff --git a/src/artifacts/MachOSliceArtifactReader.test.ts b/src/artifacts/MachOSliceArtifactReader.test.ts index ba282e3b63..6193040109 100644 --- a/src/artifacts/MachOSliceArtifactReader.test.ts +++ b/src/artifacts/MachOSliceArtifactReader.test.ts @@ -1,9 +1,9 @@ import { EventEmitter } from "node:events"; import { PassThrough } from "node:stream"; -import { writeFile } from "node:fs/promises"; +import { truncate, writeFile } from "node:fs/promises"; import { join } from "node:path"; -import { describe, expect, it, onTestFinished } from "vitest"; +import { describe, expect, it, onTestFinished, vi } from "vitest"; import { createTestTempDirectory } from "../../tests/fixtures/temporaryDirectory.js"; import { CPU, fatImage, machoImage } from "./apple/MachoImage.fixture.js"; @@ -155,7 +155,7 @@ describe("Mach-O command cleanup lifecycle", () => { }); describe("MachOSliceArtifactReader", () => { - it("keeps a borrowed source usable after a cancelled slice read", async () => { + it("keeps a borrowed source usable after idle and active cancelled slice reads", async () => { const root = await createTestTempDirectory("rea-slices-cancelled-"); const binary = join(root, "fat"); const x86 = machoImage({ cpu: CPU.x86_64 }); @@ -202,6 +202,18 @@ describe("MachOSliceArtifactReader", () => { const armEntry = entries.find((entry) => entry.path === "slices/arm64"); if (armEntry === undefined) throw new Error("Expected an arm64 slice"); + const idleController = new AbortController(); + const idle = await reader.open(armEntry, idleController.signal); + const idleFailure = new Promise((resolve) => + idle.once("error", resolve), + ); + idleController.abort(null); + await expect(idleFailure).resolves.toMatchObject({ + reason: "cancelled", + cancellation: { reason: null }, + }); + expect((await handle.stat()).isFile()).toBe(true); + const controller = new AbortController(); const cancelledRead = await reader.open(armEntry, controller.signal); const iterator = cancelledRead[Symbol.asyncIterator](); @@ -224,6 +236,32 @@ describe("MachOSliceArtifactReader", () => { for await (const chunk of verifiedRead) verifiedChunks.push(Buffer.from(chunk)); expect(Buffer.concat(verifiedChunks)).toEqual(arm); + + const truncatedController = new AbortController(); + const truncated = await reader.open( + verifiedArmEntry, + truncatedController.signal, + ); + await truncate(binary, 8193); + let readEOF = false; + const read = handle.read.bind(handle); + vi.spyOn(handle, "read").mockImplementation((async (...args: unknown[]) => { + const result = (await Reflect.apply(read, handle, args)) as { + bytesRead: number; + buffer: Buffer; + }; + if (result.bytesRead === 0) readEOF = true; + return result; + }) as typeof handle.read); + const failed = new Promise((resolve) => + truncated.once("error", resolve), + ); + truncated.read(0); + await expect.poll(() => readEOF).toBe(true); + await new Promise((resolve) => setImmediate(resolve)); + expect(truncated.readableLength).toBeGreaterThan(0); + truncatedController.abort(null); + await expect(failed).resolves.toMatchObject({ reason: "integrity" }); }); }); diff --git a/src/artifacts/MachOSliceArtifactReader.ts b/src/artifacts/MachOSliceArtifactReader.ts index 4665015688..d44dbfab22 100644 --- a/src/artifacts/MachOSliceArtifactReader.ts +++ b/src/artifacts/MachOSliceArtifactReader.ts @@ -1,3 +1,4 @@ +import { assertArtifactReadActive } from "./ArtifactFileOperation.js"; import { lstat } from "node:fs/promises"; import type { FileHandle } from "node:fs/promises"; import { Readable } from "node:stream"; @@ -30,7 +31,7 @@ import { sameRegularFileState, type StableRegularFileDescriptor, } from "../filesystem/RegularFile.js"; -import { readFileHandleChunks } from "../filesystem/readFileHandleChunks.js"; +import { readArtifactFileChunks } from "./readArtifactFileChunks.js"; /** Read-only universal Mach-O slice reader backed by native lipo metadata. */ export class MachOSliceArtifactReader implements ArtifactReader { @@ -225,7 +226,9 @@ export class MachOSliceArtifactReader implements ArtifactReader { async open(entry: ArtifactEntry, signal?: AbortSignal): Promise { if (signal?.aborted === true) return Promise.reject( - new ArtifactReaderFailure("cancelled", "Mach-O slice read cancelled"), + new ArtifactReaderFailure("cancelled", "Mach-O slice read cancelled", { + cause: signal.reason, + }), ); const produced = this.#entries.get(entry.adapterKey); if (produced === undefined || !sameArtifactEntry(entry, produced)) @@ -257,48 +260,109 @@ export class MachOSliceArtifactReader implements ArtifactReader { "integrity", `Mach-O slice range is outside the artifact: ${entry.path}`, ); - const sourceStream = readFileHandleChunks(source.handle, { + const sourceStream = readArtifactFileChunks(source.handle, { start: offset, end: offset + size - 1, ...(signal === undefined ? {} : { signal }), }); const verifySource = () => this.#verifySource(source); - return Readable.from( + let validatingSource = false; + const completionFailure = (): ArtifactReaderFailure | undefined => { + const coverage = sourceStream.readCoverage(); + if (coverage.kind !== "complete") + return new ArtifactReaderFailure( + "integrity", + `Mach-O slice read coverage is incomplete: ${entry.path}`, + ); + if (coverage.bytes !== BigInt(size)) + return new ArtifactReaderFailure( + "integrity", + `Mach-O slice size disagrees with lipo metadata: ${entry.path}`, + ); + if (signal?.aborted) + return new ArtifactReaderFailure( + "cancelled", + "Mach-O slice read cancelled", + { cause: signal.reason }, + ); + return undefined; + }; + const output = Readable.from( (async function* () { - let observedBytes = 0; try { for await (const raw of sourceStream) { const chunk = streamChunkToBuffer(raw); - observedBytes += chunk.byteLength; - if (observedBytes > size) - throw new ArtifactReaderFailure( - "integrity", - `Mach-O slice exceeded its reported size: ${entry.path}`, - ); + if ( + sourceStream.readCoverage().kind === "complete" && + signal?.aborted + ) { + const failure = completionFailure(); + if (failure !== undefined) throw failure; + } yield chunk; } } catch (cause: unknown) { if (cause instanceof ArtifactReaderFailure) throw cause; - if (signal?.aborted === true) - throw new ArtifactReaderFailure( - "cancelled", - "Mach-O slice read cancelled", - { cause }, - ); throw new ArtifactReaderFailure( "integrity", `Could not read Mach-O slice range: ${entry.path}`, { cause }, ); } - if (observedBytes !== size) - throw new ArtifactReaderFailure( - "integrity", - `Mach-O slice size disagrees with lipo metadata: ${entry.path}`, - ); - await verifySource(); + validatingSource = true; + const failure = completionFailure(); + if (failure !== undefined) throw failure; + try { + await verifySource(); + assertArtifactReadActive(signal); + } finally { + validatingSource = false; + } })(), ); + const onSourceError = (cause: unknown): void => { + const completedCancellation = + cause instanceof ArtifactReaderFailure && + cause.reason === "cancelled" && + sourceStream.readCoverage().kind === "complete" && + !validatingSource + ? completionFailure() + : undefined; + output.destroy( + completedCancellation ?? + (cause instanceof ArtifactReaderFailure + ? cause + : new ArtifactReaderFailure( + "integrity", + `Could not read Mach-O slice range: ${entry.path}`, + { cause }, + )), + ); + }; + sourceStream.on("error", onSourceError); + sourceStream.once("close", () => + sourceStream.removeListener("error", onSourceError), + ); + const onAbort = (): void => { + if ( + sourceStream.readCoverage().kind === "complete" && + (sourceStream.errored === null || + (sourceStream.errored instanceof ArtifactReaderFailure && + sourceStream.errored.reason === "cancelled")) && + !validatingSource && + !output.destroyed + ) { + const failure = completionFailure(); + if (failure !== undefined) output.destroy(failure); + } + }; + signal?.addEventListener("abort", onAbort, { once: true }); + output.once("close", () => { + signal?.removeEventListener("abort", onAbort); + if (!sourceStream.destroyed) sourceStream.destroy(); + }); + if (signal?.aborted) onAbort(); + return output; } provenance(): readonly ArtifactCommand[] { diff --git a/src/artifacts/SafeOutputTree.ts b/src/artifacts/SafeOutputTree.ts index 4c661f1980..6435100271 100644 --- a/src/artifacts/SafeOutputTree.ts +++ b/src/artifacts/SafeOutputTree.ts @@ -605,6 +605,7 @@ const abortIfNeeded = (signal?: AbortSignal): void => { throw new ArtifactReaderFailure( "cancelled", "Artifact extraction cancelled", + { cause: signal.reason }, ); }; diff --git a/src/artifacts/SafeOutputTreeCreationFailure.ts b/src/artifacts/SafeOutputTreeCreationFailure.ts index d8f132e352..49a4abd642 100644 --- a/src/artifacts/SafeOutputTreeCreationFailure.ts +++ b/src/artifacts/SafeOutputTreeCreationFailure.ts @@ -15,6 +15,7 @@ export class SafeOutputTreeCreationFailure extends ArtifactReaderFailure { primary.message, { cause: primary, + cancellation: primary.cancellation, ...(primary.cleanup === undefined ? {} : { cleanup: primary.cleanup }), ...(primary.partialObservation === undefined ? {} diff --git a/src/artifacts/SafeOutputTreeReadback.ts b/src/artifacts/SafeOutputTreeReadback.ts index 3cee16f488..d524726399 100644 --- a/src/artifacts/SafeOutputTreeReadback.ts +++ b/src/artifacts/SafeOutputTreeReadback.ts @@ -1,6 +1,7 @@ import { createHash } from "node:crypto"; import type { OwnedFileHandle } from "../filesystem/OwnedFileHandle.js"; -import { readFileHandleChunks } from "../filesystem/readFileHandleChunks.js"; +import { readArtifactFileChunks } from "./readArtifactFileChunks.js"; +import { assertArtifactReadActive } from "./ArtifactFileOperation.js"; import { streamChunkToBuffer } from "./StreamBytes.js"; import { ArtifactReaderFailure } from "./ArtifactReader.js"; @@ -20,6 +21,7 @@ export const hashOutputFile = async ( readonly signal?: AbortSignal | undefined; }, ): Promise<{ readonly sha256: string; readonly bytes: number }> => { + assertArtifactReadActive(options.signal); const admittedIdentity = await readFileIdentity(owner.handle, path); await assertFilePathIdentity(path, identity); if ( @@ -32,31 +34,17 @@ export const hashOutputFile = async ( ); const hash = createHash("sha256"); let bytes = 0; - try { - for await (const raw of readFileHandleChunks(owner.handle, options)) { - if (options.signal?.aborted === true) - throw new ArtifactReaderFailure( - "cancelled", - "Artifact extraction cancelled", - ); - const chunk = streamChunkToBuffer(raw); - bytes += chunk.length; - if (bytes > options.maximum) - throw new ArtifactReaderFailure( - "integrity", - "Readback exceeded the bytes written", - ); - hash.update(chunk); - } - } catch (cause: unknown) { - if (options.signal?.aborted === true) + for await (const raw of readArtifactFileChunks(owner.handle, options)) { + const chunk = streamChunkToBuffer(raw); + bytes += chunk.length; + if (bytes > options.maximum) throw new ArtifactReaderFailure( - "cancelled", - "Artifact extraction cancelled", - { cause }, + "integrity", + "Readback exceeded the bytes written", ); - throw cause; + hash.update(chunk); } await assertFilePathIdentity(path, identity); + assertArtifactReadActive(options.signal); return { sha256: hash.digest("hex"), bytes }; }; diff --git a/src/artifacts/inventory/hashStableRootArtifact.ts b/src/artifacts/inventory/hashStableRootArtifact.ts index 40a8a5fd6d..0956771b00 100644 --- a/src/artifacts/inventory/hashStableRootArtifact.ts +++ b/src/artifacts/inventory/hashStableRootArtifact.ts @@ -1,7 +1,7 @@ import { lstat } from "node:fs/promises"; import type { Stats } from "node:fs"; import type { FileHandle } from "node:fs/promises"; -import { readFileHandleChunks } from "../../filesystem/readFileHandleChunks.js"; +import { readArtifactFileChunks } from "../readArtifactFileChunks.js"; import { OwnedFileHandle } from "../../filesystem/OwnedFileHandle.js"; import { ArtifactReaderFailure } from "../ArtifactReader.js"; @@ -101,7 +101,7 @@ export const hashStableRootArtifactHandle = async ( const digest = await hashReadable( // Read at most the admitted extent plus one byte to detect growth without // following an actively appended file indefinitely. - readFileHandleChunks(handle, { start: 0, end: initial.size }), + readArtifactFileChunks(handle, { start: 0, end: initial.size, signal }), signal, ); abortIfNeeded(signal); @@ -115,5 +115,6 @@ export const hashStableRootArtifactHandle = async ( "integrity", `Root artifact changed during inventory: ${path}`, ); + abortIfNeeded(signal); return digest; }; diff --git a/src/artifacts/readArtifactFileChunks.ts b/src/artifacts/readArtifactFileChunks.ts new file mode 100644 index 0000000000..a9d6679ec5 --- /dev/null +++ b/src/artifacts/readArtifactFileChunks.ts @@ -0,0 +1,19 @@ +import type { FileHandle } from "node:fs/promises"; +import { + readFileHandleChunks, + type FileHandleChunkOptions, +} from "../filesystem/readFileHandleChunks.js"; +import { ArtifactReaderFailure } from "./ArtifactReader.js"; + +/** Produce artifact cancellation facts at the actual descriptor read boundary. */ +export const readArtifactFileChunks = ( + handle: Pick, + options: Pick = {}, +) => + readFileHandleChunks(handle, { + ...options, + cancellationFailure: (reason) => + new ArtifactReaderFailure("cancelled", "Artifact file read cancelled", { + cause: reason, + }), + }); diff --git a/src/artifacts/readStableArtifact.ts b/src/artifacts/readStableArtifact.ts index d42af37c2f..0a9a5a2c73 100644 --- a/src/artifacts/readStableArtifact.ts +++ b/src/artifacts/readStableArtifact.ts @@ -3,7 +3,16 @@ import { constants, type BigIntStats } from "node:fs"; import { lstat, open, type FileHandle } from "node:fs/promises"; import { ArtifactReaderFailure } from "./ArtifactReader.js"; import type { ArtifactResourceScope } from "./ArtifactResourceScope.js"; -import { OwnedFileHandle } from "../filesystem/OwnedFileHandle.js"; +import { + withArtifactFileHandle, + assertArtifactReadActive, +} from "./ArtifactFileOperation.js"; + +/** Exact physical read completed before descriptor finalization. */ +export interface StableArtifactSnapshot { + readonly bytes: Buffer; + readonly sha256: string; +} /** Filesystem boundary for stable reads, including deterministic replacement probes. */ export interface StableArtifactFileSystem { @@ -38,13 +47,6 @@ const afterValidation = async ( } }; -const assertReadActive = (signal: AbortSignal | undefined): void => { - if (signal?.aborted) - throw new ArtifactReaderFailure("cancelled", "Artifact read cancelled", { - cause: signal.reason, - }); -}; - /** Read one bounded regular file and reject replacement or in-place changes. */ export const readStableArtifact = async ( path: string, @@ -52,8 +54,8 @@ export const readStableArtifact = async ( signal?: AbortSignal, fileSystem: StableArtifactFileSystem = FILE_SYSTEM, resourceScope?: ArtifactResourceScope, -): Promise<{ readonly bytes: Buffer; readonly sha256: string }> => { - assertReadActive(signal); +): Promise => { + assertArtifactReadActive(signal); const before = await fileSystem.lstat(path); if (!before.isFile() || before.isSymbolicLink()) throw new ArtifactReaderFailure( @@ -72,82 +74,71 @@ export const readStableArtifact = async ( constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK, ), ); - const owner = - resourceScope === undefined ? undefined : new OwnedFileHandle(file); - let failure: unknown; - try { - const initial = await file.stat({ bigint: true }); - if ( - !initial.isFile() || - initial.dev !== before.dev || - initial.ino !== before.ino - ) - throw new ArtifactReaderFailure( - "integrity", - `Artifact changed before open: ${path}`, - ); - const chunks: Buffer[] = []; - let count = 0; - for await (const chunk of file.createReadStream({ - autoClose: false, - highWaterMark: 65536, - })) { - assertReadActive(signal); - const value: unknown = chunk; - if (!Buffer.isBuffer(value)) + return withArtifactFileHandle( + path, + file, + resourceScope, + signal, + async (file) => { + const initial = await file.stat({ bigint: true }); + if ( + !initial.isFile() || + initial.dev !== before.dev || + initial.ino !== before.ino + ) throw new ArtifactReaderFailure( - "format", - `Artifact stream did not return bytes: ${path}`, + "integrity", + `Artifact changed before open: ${path}`, ); - const bytes = value; - count += bytes.length; - if (count > maximumBytes) - throw new ArtifactReaderFailure( - "limit", - `Selected artifact grew beyond its byte limit: ${path}`, - { boundary: "file-size" }, - ); - chunks.push(bytes); - } - const after = await file.stat({ bigint: true }); - const current = await afterValidation(path, "during read", () => - fileSystem.lstat(path), - ); - if ( - [after, current].some( - (stat) => - !stat.isFile() || - stat.dev !== initial.dev || - stat.ino !== initial.ino || - stat.size !== initial.size || - stat.mtimeNs !== initial.mtimeNs || - stat.ctimeNs !== initial.ctimeNs, - ) || - BigInt(count) !== initial.size - ) - throw new ArtifactReaderFailure( - "integrity", - `Artifact changed during read: ${path}`, + const chunks: Buffer[] = []; + let count = 0; + for await (const chunk of file.createReadStream({ + autoClose: false, + highWaterMark: 65536, + })) { + assertArtifactReadActive(signal); + const value: unknown = chunk; + if (!Buffer.isBuffer(value)) + throw new ArtifactReaderFailure( + "format", + `Artifact stream did not return bytes: ${path}`, + ); + const bytes = value; + count += bytes.length; + if (count > maximumBytes) + throw new ArtifactReaderFailure( + "limit", + `Selected artifact grew beyond its byte limit: ${path}`, + { boundary: "file-size" }, + ); + chunks.push(bytes); + } + const after = await file.stat({ bigint: true }); + const current = await afterValidation(path, "during read", () => + fileSystem.lstat(path), ); - assertReadActive(signal); - const bytes = Buffer.concat(chunks, count); - return { bytes, sha256: createHash("sha256").update(bytes).digest("hex") }; - } catch (cause: unknown) { - failure = cause; - throw cause; - } finally { - if (owner === undefined || resourceScope === undefined) await file.close(); - else { - const cleanup = await resourceScope.release({ - kind: "file-handle", - resource: path, - handle: owner, - }); - if (cleanup.kind === "failed") - throw ArtifactReaderFailure.withCleanup( - failure ?? cleanup.cause, - ArtifactReaderFailure.cleanupObservation(cleanup.cause, path), + if ( + [after, current].some( + (stat) => + !stat.isFile() || + stat.dev !== initial.dev || + stat.ino !== initial.ino || + stat.size !== initial.size || + stat.mtimeNs !== initial.mtimeNs || + stat.ctimeNs !== initial.ctimeNs, + ) || + BigInt(count) !== initial.size + ) + throw new ArtifactReaderFailure( + "integrity", + `Artifact changed during read: ${path}`, ); - } - } + assertArtifactReadActive(signal); + const bytes = Buffer.concat(chunks, count); + return { + bytes, + sha256: createHash("sha256").update(bytes).digest("hex"), + }; + }, + ); }; diff --git a/src/domain/analysisErrorBase.ts b/src/domain/analysisErrorBase.ts index a2a52c1d74..f02df6b604 100644 --- a/src/domain/analysisErrorBase.ts +++ b/src/domain/analysisErrorBase.ts @@ -62,6 +62,8 @@ export interface AnalysisErrorOptions extends ErrorOptions { /** Base class for expected analysis, provider, and session failures. */ export abstract class AnalysisError extends Error { + readonly #cleanupDependencies = new Set(); + #cleanupDependents: WeakRef[] = []; readonly diagnostics: Readonly> | undefined; readonly capturedOutput: AnalysisCapturedOutput | undefined; readonly cleanup: AnalysisCleanupObservation | undefined; @@ -107,9 +109,39 @@ export abstract class AnalysisError extends Error { return this; } + /** Keep a composed error incomplete until all explicitly owned source errors resolve. */ + bindCleanupDependencies(dependencies: readonly AnalysisError[]): this { + for (const dependency of dependencies) { + if ( + !dependency.cleanupIncomplete || + this.#cleanupDependencies.has(dependency) + ) + continue; + this.#cleanupDependencies.add(dependency); + dependency.#cleanupDependents = [ + ...dependency.#cleanupDependents.filter( + (reference) => reference.deref() !== undefined, + ), + new WeakRef(this), + ]; + } + if (this.#cleanupDependencies.size === 0) this.resolveCleanup(); + return this; + } + /** Clear live cleanup status only after its owning resource confirms retry success. */ resolveCleanup(): this { + if ( + [...this.#cleanupDependencies].some( + (dependency) => dependency.cleanupIncomplete, + ) + ) + return this; + this.#cleanupDependencies.clear(); this.cleanupIncomplete = false; + for (const reference of this.#cleanupDependents) + reference.deref()?.resolveCleanup(); + this.#cleanupDependents = []; return this; } diff --git a/src/filesystem/OwnedRegularFile.ts b/src/filesystem/OwnedRegularFile.ts index d5e1893ffd..f79978e021 100644 --- a/src/filesystem/OwnedRegularFile.ts +++ b/src/filesystem/OwnedRegularFile.ts @@ -63,12 +63,15 @@ export class RegularFileCleanupFailure extends Error { } } -export interface OwnedRegularFileOptions { +export interface OwnedFileOperationOptions { readonly signal?: AbortSignal; - readonly symlinks?: "follow" | "reject"; readonly retainFailedClose?: (owner: OwnedFileHandle) => void; } +export interface OwnedRegularFileOptions extends OwnedFileOperationOptions { + readonly symlinks?: "follow" | "reject"; +} + /** Run an operation while one admitted descriptor and its cleanup owner stay together. */ export const withOwnedRegularFile = async ( path: string, @@ -89,7 +92,22 @@ export const withOwnedRegularFile = async ( retention, ); }); - const owner = new OwnedFileHandle(handle); + return withOwnedFileHandle( + path, + new OwnedFileHandle(handle), + options, + operation, + ); +}; + +/** Own an already admitted descriptor through its operation and final close. */ +export const withOwnedFileHandle = async ( + path: string, + owner: OwnedFileHandle, + options: OwnedFileOperationOptions, + operation: (handle: FileHandle) => Promise, +): Promise => { + const handle = owner.handle; let outcome: RegularFileOperationOutcome; try { outcome = { kind: "completed", value: await operation(handle) }; diff --git a/src/filesystem/readFileHandleChunks.ts b/src/filesystem/readFileHandleChunks.ts index e9b9acc2f7..a7459db5e5 100644 --- a/src/filesystem/readFileHandleChunks.ts +++ b/src/filesystem/readFileHandleChunks.ts @@ -1,15 +1,39 @@ import type { FileHandle } from "node:fs/promises"; import { Readable } from "node:stream"; +/** Cancellation observed by the descriptor read, retaining the caller's exact reason. */ +export class FileReadCancelled extends Error { + constructor(readonly reason: unknown) { + super("File read cancelled", { cause: reason }); + this.name = "FileReadCancelled"; + } +} + +export interface FileHandleChunkOptions { + readonly start?: number | bigint; + readonly end?: number | bigint; + readonly signal?: AbortSignal | undefined; + readonly cancellationFailure?: (reason: unknown) => Error; +} + +/** Physical descriptor coverage is separate from delivery of buffered stream bytes. */ +export type FileHandleReadCoverage = + | { readonly kind: "incomplete"; readonly bytes: bigint } + | { + readonly kind: "complete"; + readonly bytes: bigint; + readonly boundary: "eof" | "range"; + }; + +export type FileHandleChunkStream = Readable & { + readCoverage(): FileHandleReadCoverage; +}; + /** Stream positional reads while leaving descriptor cleanup to its owner. */ export const readFileHandleChunks = ( handle: Pick, - options: { - readonly start?: number | bigint; - readonly end?: number | bigint; - readonly signal?: AbortSignal | undefined; - } = {}, -): Readable => { + options: FileHandleChunkOptions = {}, +): FileHandleChunkStream => { const start = fileOffset(options.start ?? 0); const unlimited = options.end === undefined || options.end === Infinity; const end = unlimited ? undefined : fileOffset(options.end); @@ -19,24 +43,49 @@ export const readFileHandleChunks = ( (end !== undefined && end < start) ) throw new RangeError("Invalid file read range"); - return Readable.from( + let readPending = false; + let totalBytes = 0n; + let coverage: + | Extract + | undefined; + const cancellationFailure = (): Error => + options.cancellationFailure?.(options.signal?.reason) ?? + new FileReadCancelled(options.signal?.reason); + const assertActive = (): void => { + if (options.signal?.aborted) throw cancellationFailure(); + }; + const source = Readable.from( (async function* () { let position = start; while (end === undefined || position <= end) { - options.signal?.throwIfAborted(); + assertActive(); const remaining = end === undefined ? undefined : end - position + 1n; const chunk = Buffer.allocUnsafe( remaining === undefined || remaining > 64n * 1024n ? 64 * 1024 : Number(remaining), ); - const { bytesRead } = await handle.read( - chunk, - 0, - chunk.length, - position, - ); - options.signal?.throwIfAborted(); + readPending = true; + let bytesRead: number; + try { + ({ bytesRead } = await handle.read(chunk, 0, chunk.length, position)); + } finally { + readPending = false; + } + totalBytes += BigInt(bytesRead); + if (bytesRead === 0) + coverage = Object.freeze({ + kind: "complete", + bytes: totalBytes, + boundary: "eof", + }); + else if (end !== undefined && position + BigInt(bytesRead) > end) + coverage = Object.freeze({ + kind: "complete", + bytes: totalBytes, + boundary: "range", + }); + assertActive(); if (bytesRead === 0) return; position += BigInt(bytesRead); yield chunk.subarray(0, bytesRead); @@ -44,9 +93,23 @@ export const readFileHandleChunks = ( })(), { objectMode: false, - ...(options.signal === undefined ? {} : { signal: options.signal }), }, ); + const onAbort = (): void => { + // A native descriptor read cannot be interrupted. Its rejection remains + // primary; a fulfilled read observes cancellation before yielding bytes. + if (!readPending && !source.destroyed) + source.destroy(cancellationFailure()); + }; + options.signal?.addEventListener("abort", onAbort, { once: true }); + source.once("close", () => + options.signal?.removeEventListener("abort", onAbort), + ); + if (options.signal?.aborted) onAbort(); + return Object.assign(source, { + readCoverage: (): FileHandleReadCoverage => + coverage ?? Object.freeze({ kind: "incomplete", bytes: totalBytes }), + }); }; const fileOffset = (value: number | bigint): bigint | undefined => { diff --git a/tests/boundary/filesystem/asarReaderCleanup.test.ts b/tests/boundary/filesystem/asarReaderCleanup.test.ts index 073ee1c7d4..170c4fde9b 100644 --- a/tests/boundary/filesystem/asarReaderCleanup.test.ts +++ b/tests/boundary/filesystem/asarReaderCleanup.test.ts @@ -264,3 +264,72 @@ describe("ASAR preparation cleanup", () => { } }); }); + +it("forwards cancellation of an unconsumed packed member and closes its exact descriptor", async () => { + const { archive } = await createArchive(); + const { reader, opened } = createTrackedReader(archive); + try { + const entry = await packedMember(reader); + const controller = new AbortController(); + const output = await reader.open(entry, controller.signal); + const descriptor = latestHandle(opened); + const failed = new Promise((resolve) => + output.once("error", resolve), + ); + controller.abort(null); + await expect(failed).resolves.toMatchObject({ + reason: "cancelled", + cancellation: { reason: null }, + }); + await reader.close(); + expectDescriptorClosed(descriptor.fd); + } finally { + await reader.close(); + } +}); + +it.each(["snapshot", "member"])( + "preserves %s read failure when caller cancellation arrives during that read", + async (stage) => { + const { archive } = await createArchive(); + const controller = new AbortController(); + const failure = Object.assign( + new Error("controlled ASAR native read failure"), + { code: "EIO", errno: -5, syscall: "read" }, + ); + let active = stage === "snapshot"; + const { reader, opened } = createTrackedReader(archive, { + beforeAdoption: async (_path, handle) => { + if (active) + handle.read = (async () => { + controller.abort(null); + throw failure; + }) as FileHandle["read"]; + }, + }); + try { + const consume = async () => { + if (stage === "snapshot") { + for await (const _entry of reader.entries(controller.signal)) { + /* failure precedes admission */ + } + } else { + const entry = await packedMember(reader); + active = true; + const source = await reader.open(entry, controller.signal); + for await (const _chunk of source) { + /* failure precedes bytes */ + } + } + }; + await expect(consume()).rejects.toMatchObject({ + reason: "io", + cause: failure, + }); + await reader.close(); + for (const handle of opened) expectDescriptorClosed(handle.fd); + } finally { + await reader.close(); + } + }, +); diff --git a/tests/boundary/filesystem/asarStreamingAllocation.test.ts b/tests/boundary/filesystem/asarStreamingAllocation.test.ts index e8c46bd8cf..dcde4272ff 100644 --- a/tests/boundary/filesystem/asarStreamingAllocation.test.ts +++ b/tests/boundary/filesystem/asarStreamingAllocation.test.ts @@ -9,6 +9,7 @@ import { writeFile, truncate, } from "node:fs/promises"; +import type { FileHandle } from "node:fs/promises"; import { dirname, join } from "node:path"; import { Readable } from "node:stream"; import { buffer } from "node:stream/consumers"; @@ -21,6 +22,7 @@ import { readValidatedAsarEntry, } from "../../../src/artifacts/AsarEntryStream.js"; import { hashReadable } from "../../../src/artifacts/ArtifactHash.js"; +import { readArtifactFileChunks } from "../../../src/artifacts/readArtifactFileChunks.js"; import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js"; const memberBytes = 64 * 1024 * 1024; @@ -714,12 +716,15 @@ it("preserves filesystem errors raised while consuming a member stream", async ( errno: -5, syscall: "read", }); - const source = Readable.from( - (async function* () { - yield Buffer.from("partial"); - throw ioError; - })(), - ); + let delivered = false; + const read = (async (target: Buffer) => { + if (delivered) throw ioError; + delivered = true; + const partial = Buffer.from("partial"); + partial.copy(target); + return { bytesRead: partial.length, buffer: target }; + }) as FileHandle["read"]; + const source = readArtifactFileChunks({ read }); const output = readValidatedAsarEntry( source, undefined, diff --git a/tests/boundary/filesystem/fileHandleChunks.test.ts b/tests/boundary/filesystem/fileHandleChunks.test.ts new file mode 100644 index 0000000000..67c26e290d --- /dev/null +++ b/tests/boundary/filesystem/fileHandleChunks.test.ts @@ -0,0 +1,126 @@ +import type { FileHandle } from "node:fs/promises"; +import { buffer } from "node:stream/consumers"; +import { expect, it, vi } from "vitest"; +import { + FileReadCancelled, + readFileHandleChunks, +} from "../../../src/filesystem/readFileHandleChunks.js"; + +it.each(["complete", "fail"])( + "settles the pending descriptor read before cancellation: %s", + async (outcome) => { + let enter!: () => void; + let release!: (value: { bytesRead: number; buffer: Buffer }) => void; + let reject!: (cause: unknown) => void; + const entered = new Promise((resolve) => { + enter = resolve; + }); + const pending = new Promise<{ bytesRead: number; buffer: Buffer }>( + (resolve, fail) => { + release = resolve; + reject = fail; + }, + ); + const read = vi.fn(async () => { + enter(); + return pending; + }) as unknown as FileHandle["read"]; + const controller = new AbortController(); + const source = readFileHandleChunks( + { read }, + { signal: controller.signal }, + ); + let settled = false; + const result = buffer(source) + .catch((cause: unknown) => cause) + .finally(() => { + settled = true; + }); + await entered; + controller.abort(null); + await new Promise((resolve) => setImmediate(resolve)); + expect(settled).toBe(false); + expect(source.destroyed).toBe(false); + const failure = new Error("controlled native read failure"); + if (outcome === "fail") reject(failure); + else release({ bytesRead: 0, buffer: Buffer.alloc(0) }); + const error = await result; + if (outcome === "fail") expect(error).toBe(failure); + else { + expect(error).toBeInstanceOf(FileReadCancelled); + if (error instanceof FileReadCancelled) expect(error.reason).toBeNull(); + } + expect(read).toHaveBeenCalledTimes(1); + }, +); + +it.each([false, true])( + "cancels an unconsumed stream without admitting a read (already aborted: %s)", + async (alreadyAborted) => { + const controller = new AbortController(); + const reason = new Error("caller cancellation"); + const read = vi.fn() as unknown as FileHandle["read"]; + if (alreadyAborted) controller.abort(reason); + const source = readFileHandleChunks( + { read }, + { signal: controller.signal }, + ); + const failed = new Promise((resolve) => + source.once("error", resolve), + ); + if (!alreadyAborted) controller.abort(reason); + const error = await failed; + expect(error).toBeInstanceOf(FileReadCancelled); + if (error instanceof FileReadCancelled) expect(error.reason).toBe(reason); + expect(read).not.toHaveBeenCalled(); + }, +); + +it("removes cancellation observation after normal stream completion", async () => { + const controller = new AbortController(); + const remove = vi.spyOn(controller.signal, "removeEventListener"); + const read = vi.fn(async () => ({ + bytesRead: 0, + buffer: Buffer.alloc(0), + })) as unknown as FileHandle["read"]; + const source = readFileHandleChunks({ read }, { signal: controller.signal }); + expect(await buffer(source)).toEqual(Buffer.alloc(0)); + expect(remove).toHaveBeenCalledWith("abort", expect.any(Function)); + controller.abort(null); + expect(source.errored).toBeNull(); +}); + +it.each(["range", "eof"])( + "retains physical %s coverage independently of stream delivery", + async (boundary) => { + let calls = 0; + const read = (async ( + target: Buffer, + _offset: number, + length: number, + position: bigint, + ) => { + calls++; + expect(position).toBe(calls === 1 ? 5n : 6n); + if (boundary === "eof" && calls === 2) + return { bytesRead: 0, buffer: target }; + const bytes = boundary === "range" ? 3 : 1; + target.fill(0x61, 0, bytes); + expect(length).toBe(3); + return { bytesRead: bytes, buffer: target }; + }) as FileHandle["read"]; + const source = readFileHandleChunks({ read }, { start: 5, end: 7 }); + expect(source.readCoverage()).toEqual({ kind: "incomplete", bytes: 0n }); + expect(await buffer(source)).toEqual( + Buffer.alloc(boundary === "range" ? 3 : 1, 0x61), + ); + const coverage = source.readCoverage(); + expect(coverage).toEqual({ + kind: "complete", + bytes: boundary === "range" ? 3n : 1n, + boundary, + }); + expect(Object.isFrozen(coverage)).toBe(true); + expect(calls).toBe(boundary === "range" ? 1 : 2); + }, +); diff --git a/tests/boundary/filesystem/stableArtifact.test.ts b/tests/boundary/filesystem/stableArtifact.test.ts index 3bd587ad0d..38c9f7d6ce 100644 --- a/tests/boundary/filesystem/stableArtifact.test.ts +++ b/tests/boundary/filesystem/stableArtifact.test.ts @@ -7,8 +7,14 @@ import { writeFile, } from "node:fs/promises"; import { join } from "node:path"; -import { expect, it } from "vitest"; +import { createHash } from "node:crypto"; +import { expect, it, onTestFinished } from "vitest"; import { readStableArtifact } from "../../../src/artifacts/readStableArtifact.js"; +import { + ArtifactFileCleanupFailure, + ArtifactFileCancelled, +} from "../../../src/artifacts/ArtifactFileOperation.js"; +import { ArtifactResourceScope } from "../../../src/artifacts/ArtifactResourceScope.js"; import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js"; it("reads exact binary and empty bytes and rejects symlink/size/cancellation boundaries", async () => { @@ -78,3 +84,173 @@ it("keeps initially missing inputs distinguishable from later changes", async () readStableArtifact(join(root, "missing"), 1024), ).rejects.toMatchObject({ code: "ENOENT" }); }); + +const heldDescriptor = async () => { + const root = await createTestTempDirectory("rea-stable-close-owner-"); + const path = join(root, "selected.bin"); + const bytes = Buffer.from([0, 255, 1, 254]); + await writeFile(path, bytes); + const handle = await open(path, "r"); + const nativeClose = handle.close.bind(handle); + const closeFailure = new Error("controlled descriptor close failure"); + let allowClose = false; + handle.close = async () => { + if (!allowClose) throw closeFailure; + await nativeClose(); + }; + onTestFinished(async () => { + if (handle.fd >= 0) await nativeClose(); + }); + return { + path, + bytes, + handle, + nativeClose, + closeFailure, + release: () => { + allowClose = true; + }, + fileSystem: { + lstat: (selected: string) => lstat(selected, { bigint: true }), + open: async () => handle, + }, + }; +}; + +it.each(["scoped", "transferred"])( + "retains exact completed bytes and descriptor cleanup under %s ownership", + async (ownership) => { + const file = await heldDescriptor(); + const scope = + ownership === "scoped" ? new ArtifactResourceScope() : undefined; + const failure = await readStableArtifact( + file.path, + 1024, + undefined, + file.fileSystem, + scope, + ).catch((cause: unknown) => cause); + expect(failure).toBeInstanceOf(ArtifactFileCleanupFailure); + if (!(failure instanceof ArtifactFileCleanupFailure)) + throw new Error("Missing descriptor owner"); + expect(failure.fileCleanupFailure.owner.handle).toBe(file.handle); + expect(failure.fileCleanupFailure.cleanupCause).toBe(file.closeFailure); + expect(failure.fileCleanupFailure.outcome).toEqual({ + kind: "completed", + value: { + bytes: file.bytes, + sha256: createHash("sha256").update(file.bytes).digest("hex"), + }, + }); + expect(file.handle.fd).toBeGreaterThanOrEqual(0); + expect(failure.cleanup?.resources).toEqual([file.path]); + expect(scope?.hasPending ?? false).toBe(ownership === "scoped"); + file.release(); + if (scope !== undefined) await scope.close(); + else await failure.cleanupLease.release(); + expect(file.handle.fd).toBe(-1); + }, +); + +it.each([new Error("primary read failure"), undefined, null])( + "preserves the explicit failed read outcome %j through close rejection", + async (primary) => { + const file = await heldDescriptor(); + file.handle.stat = async () => { + throw primary; + }; + const scope = new ArtifactResourceScope(); + const failure = await readStableArtifact( + file.path, + 1024, + undefined, + file.fileSystem, + scope, + ).catch((cause: unknown) => cause); + if (!(failure instanceof ArtifactFileCleanupFailure)) + throw new Error("Missing typed read outcome"); + expect(failure.fileCleanupFailure.outcome).toEqual({ + kind: "failed", + cause: primary, + }); + expect(failure.fileCleanupFailure.cleanupCause).toBe(file.closeFailure); + file.release(); + await scope.close(); + expect(file.handle.fd).toBe(-1); + }, +); + +it("retains physical read facts and the caller's reason when cancellation arrives during successful close", async () => { + const file = await heldDescriptor(); + const controller = new AbortController(); + const reason = new Error("cancelled during descriptor finalization"); + file.handle.close = async () => { + controller.abort(reason); + await file.nativeClose(); + }; + const failure = await readStableArtifact( + file.path, + 1024, + controller.signal, + file.fileSystem, + ).catch((cause: unknown) => cause); + if (!(failure instanceof ArtifactFileCancelled)) + throw new Error("Missing late cancellation"); + expect(failure.cause).toBe(reason); + expect(failure.fileCancellation.outcome.value.bytes).toEqual(file.bytes); + expect(file.handle.fd).toBe(-1); +}); + +it("retains a completed read and failed cleanup when cancellation arrives during close rejection", async () => { + const file = await heldDescriptor(); + const controller = new AbortController(); + const reason = new Error("cancelled during failed close"); + file.handle.close = async () => { + controller.abort(reason); + throw file.closeFailure; + }; + const failure = await readStableArtifact( + file.path, + 1024, + controller.signal, + file.fileSystem, + ).catch((cause: unknown) => cause); + if (!(failure instanceof ArtifactFileCleanupFailure)) + throw new Error("Missing cleanup ownership"); + expect(failure.fileCleanupFailure.outcome).toMatchObject({ + kind: "completed", + value: { bytes: file.bytes }, + }); + expect(controller.signal.reason).toBe(reason); + file.handle.close = file.nativeClose; + await failure.cleanupLease.release(); + expect(file.handle.fd).toBe(-1); +}); + +it("transfers the exact owner even when the caller's retention boundary rejects it", async () => { + const file = await heldDescriptor(); + const retentionFailure = new Error("controlled owner-retention failure"); + class RejectingScope extends ArtifactResourceScope { + override retain(): void { + throw retentionFailure; + } + } + const failure = await readStableArtifact( + file.path, + 1024, + undefined, + file.fileSystem, + new RejectingScope(), + ).catch((cause: unknown) => cause); + if (!(failure instanceof ArtifactFileCleanupFailure)) + throw new Error("Lost cleanup capability after retention failure"); + expect(failure.fileCleanupFailure.retention).toEqual({ + kind: "callback_threw", + cause: retentionFailure, + }); + expect(failure.fileCleanupFailure.cleanupCause).toBe(file.closeFailure); + expect(failure.fileCleanupFailure.owner.handle).toBe(file.handle); + file.release(); + await failure.cleanupLease.release(); + expect(file.handle.fd).toBe(-1); +}); diff --git a/tests/process-global/artifactReadFailurePrecedence.test.ts b/tests/process-global/artifactReadFailurePrecedence.test.ts new file mode 100644 index 0000000000..83270a4a10 --- /dev/null +++ b/tests/process-global/artifactReadFailurePrecedence.test.ts @@ -0,0 +1,207 @@ +import { createHash } from "node:crypto"; +import { constants } from "node:fs"; +import { access, mkdir, writeFile, type FileHandle } from "node:fs/promises"; +import { join } from "node:path"; +import { buffer } from "node:stream/consumers"; +import { createPackageWithOptions } from "@electron/asar"; +import { Readable } from "node:stream"; +import { afterEach, beforeEach, expect, it, vi } from "vitest"; + +const fault = vi.hoisted(() => ({ + path: "", + controller: undefined as AbortController | undefined, + reason: null as unknown, + failure: undefined as Error | undefined, + abortAfterRead: false, + abortAtFinalIdentity: false, + readFinished: false, + opened: [] as FileHandle[], +})); +vi.mock("node:fs/promises", async (importOriginal) => { + const original = await importOriginal(); + return { + ...original, + lstat: async (...args: Parameters) => { + const value = await original.lstat(...args); + if ( + args[0] === fault.path && + fault.readFinished && + fault.abortAtFinalIdentity + ) { + fault.abortAtFinalIdentity = false; + fault.controller?.abort(fault.reason); + } + return value; + }, + open: async (...args: Parameters) => { + const handle = await original.open(...args); + const flags = args[1]; + if ( + args[0] !== fault.path || + typeof flags !== "number" || + (flags & (constants.O_WRONLY | constants.O_RDWR)) !== 0 + ) + return handle; + fault.opened.push(handle); + const read = handle.read.bind(handle); + handle.read = (async (...readArgs: unknown[]) => { + if (fault.failure !== undefined) { + fault.controller?.abort(fault.reason); + throw fault.failure; + } + const result = (await Reflect.apply(read, handle, readArgs)) as { + bytesRead: number; + buffer: Buffer; + }; + if (result.bytesRead === 0) { + fault.readFinished = true; + if (fault.abortAfterRead) fault.controller?.abort(fault.reason); + } + return result; + }) as FileHandle["read"]; + return handle; + }, + }; +}); + +import { AsarArtifactReader } from "../../src/artifacts/AsarArtifactReader.js"; +import { SafeOutputTree } from "../../src/artifacts/SafeOutputTree.js"; +import { ArtifactReaderFailure } from "../../src/artifacts/ArtifactReader.js"; +import { createTestTempDirectory } from "../fixtures/temporaryDirectory.js"; + +const trees: SafeOutputTree[] = []; +beforeEach(() => { + fault.path = ""; + fault.controller = undefined; + fault.reason = null; + fault.failure = undefined; + fault.abortAfterRead = false; + fault.abortAtFinalIdentity = false; + fault.readFinished = false; + fault.opened = []; +}); +afterEach(async () => { + fault.path = ""; + await Promise.all(trees.splice(0).map((tree) => tree.rollback())); +}); + +const fixture = async () => { + const root = await createTestTempDirectory("rea-artifact-read-precedence-"); + const tree = await SafeOutputTree.create(join(root, "output")); + trees.push(tree); + const bytes = Buffer.from("durable readback bytes"); + fault.path = join(tree.outputRoot, "file.bin"); + const controller = new AbortController(); + fault.controller = controller; + return { + tree, + controller, + write: () => + tree.write( + "file.bin", + Readable.from(bytes), + { + sha256: createHash("sha256").update(bytes).digest("hex"), + bytes: bytes.length, + }, + controller.signal, + ), + }; +}; + +it("preserves a readback IO failure when cancellation arrives in the same read completion", async () => { + const { tree, write } = await fixture(); + const failure = Object.assign(new Error("controlled readback IO failure"), { + code: "EIO", + }); + fault.failure = failure; + await expect(write()).rejects.toBe(failure); + fault.failure = undefined; + await tree.rollback(); + await expect(access(tree.outputRoot)).rejects.toMatchObject({ + code: "ENOENT", + }); +}); + +it.each([ + null, + "caller reason", + new ArtifactReaderFailure("io", "caller reason"), +])( + "preserves the exact reason when cancellation is observed at read EOF: %s", + async (reason) => { + const { write } = await fixture(); + fault.reason = reason; + fault.abortAfterRead = true; + const error = await write().catch((cause: unknown) => cause); + expect(error).toBeInstanceOf(ArtifactReaderFailure); + if (!(error instanceof ArtifactReaderFailure)) + throw new Error("Expected typed cancellation"); + expect(error.reason).toBe("cancelled"); + expect(error.cancellation?.reason).toBe(reason); + }, +); + +it("observes cancellation during final readback identity validation", async () => { + const { write } = await fixture(); + fault.abortAtFinalIdentity = true; + const error = await write().catch((cause: unknown) => cause); + expect(error).toBeInstanceOf(ArtifactReaderFailure); + if (error instanceof ArtifactReaderFailure) { + expect(error.reason).toBe("cancelled"); + expect(error.cancellation?.reason).toBeNull(); + } +}); + +it.each(["idle", "active", "read-failure"])( + "preserves unpacked ASAR cancellation and exact descriptor ownership: %s", + async (stage) => { + const root = await createTestTempDirectory( + "rea-asar-unpacked-read-cancel-", + ); + const input = join(root, "input"); + const archive = join(root, "fixture.asar"); + await mkdir(input); + await writeFile(join(input, "member.bin"), "unpacked member bytes"); + await createPackageWithOptions(input, archive, { unpack: "*.bin" }); + const reader = new AsarArtifactReader(archive); + try { + const entries = []; + for await (const entry of reader.entries()) entries.push(entry); + const entry = entries.find((value) => value.path === "member.bin"); + if (entry === undefined || !entry.unpacked) + throw new Error("Expected unpacked member"); + fault.path = join(`${archive}.unpacked`, "member.bin"); + fault.controller = new AbortController(); + fault.reason = "unpacked cancellation"; + const failure = Object.assign( + new Error("controlled unpacked read failure"), + { code: "EIO", errno: -5, syscall: "read" }, + ); + if (stage === "read-failure") fault.failure = failure; + const source = await reader.open(entry, fault.controller.signal); + expect(fault.opened).toHaveLength(1); + const result = + stage === "idle" + ? new Promise((resolve) => source.once("error", resolve)) + : buffer(source).catch((cause: unknown) => cause); + if (stage !== "read-failure") fault.controller.abort(fault.reason); + if (stage === "read-failure") + await expect(result).resolves.toMatchObject({ + reason: "io", + cause: failure, + }); + else + await expect(result).resolves.toMatchObject({ + reason: "cancelled", + cancellation: { reason: fault.reason }, + }); + fault.failure = undefined; + await reader.close(); + expect(fault.opened[0]?.fd).toBe(-1); + } finally { + fault.failure = undefined; + await reader.close(); + } + }, +);