diff --git a/src/artifacts/DirectoryArtifactReader.ts b/src/artifacts/DirectoryArtifactReader.ts index b96066adcb..b05795752a 100644 --- a/src/artifacts/DirectoryArtifactReader.ts +++ b/src/artifacts/DirectoryArtifactReader.ts @@ -404,8 +404,19 @@ export class DirectoryArtifactReader implements ArtifactReader { const [opened, named] = await Promise.all([ file.stat({ bigint: true }), lstat(entry.adapterKey, { bigint: true }), - ]); - abortIfNeeded(signal); + ]).catch((validationCause: unknown) => { + if (outcome.kind !== "failed") throw validationCause; + throw new ArtifactReaderFailure( + "io", + `Directory inspection failed: ${failureMessage(outcome.cause)}; final source identity could not be validated: ${failureMessage(validationCause)}`, + { + cause: new AggregateError( + [outcome.cause, validationCause], + "Directory inspection and final source validation failed", + ), + }, + ); + }); if ( !opened.isFile() || !named.isFile() || @@ -414,9 +425,11 @@ export class DirectoryArtifactReader implements ArtifactReader { ) throw new ArtifactReaderFailure( "integrity", - `Directory entry changed during inspection: ${entry.path}`, + `Directory entry changed during inspection: ${entry.path}${outcome.kind === "failed" ? `; inspection failed: ${failureMessage(outcome.cause)}` : ""}`, + outcome.kind === "failed" ? { cause: outcome.cause } : {}, ); if (outcome.kind === "failed") throw outcome.cause; + abortIfNeeded(signal); return outcome.value; }, ); diff --git a/src/artifacts/ZipArtifactReader.test.ts b/src/artifacts/ZipArtifactReader.test.ts index 79da39efd1..fcf5600c34 100644 --- a/src/artifacts/ZipArtifactReader.test.ts +++ b/src/artifacts/ZipArtifactReader.test.ts @@ -1,14 +1,17 @@ import { writeOrderedZip } from "../../tests/fixtures/artifactEntryOrder.js"; import { zipSymlinkMetadataFixture } from "../../tests/fixtures/zipSymlinkMetadata.js"; -import { ArtifactReaderFailure } from "./ArtifactReader.js"; +import { ArtifactReaderFailure, type ArtifactEntry } from "./ArtifactReader.js"; +import { createDeferred } from "../../tests/fixtures/binarySession.js"; import { readFile, writeFile } from "node:fs/promises"; import { join } from "node:path"; +import { once } from "node:events"; import { TextReader, Uint8ArrayReader, Uint8ArrayWriter, ZipWriter, + ZipReader, } from "@zip.js/zip.js"; import { describe, expect, it, vi } from "vitest"; @@ -17,6 +20,8 @@ import { createTestTempDirectory } from "../../tests/fixtures/temporaryDirectory import { ZipArtifactReader } from "./ZipArtifactReader.js"; import { openRegularFile } from "../filesystem/RegularFile.js"; import { MINIMUM_ARTIFACT_EXPANDED_BYTES } from "../domain/artifactExpansionBudget.js"; +import { OwnedFileHandle } from "../filesystem/OwnedFileHandle.js"; +import { hashReadable } from "./ArtifactHash.js"; /** Small real ZIP carriers with forged sizes; metadata-only controls, never decompressed. */ const writeZipWithDeclaredSize = async ( @@ -55,6 +60,33 @@ const enumerate = async (reader: ZipArtifactReader): Promise => { return files; }; +/** Hold real zip.js finalization after its streaming work settles. */ +const holdZipProducerSettlement = () => { + const reached = createDeferred(); + const release = createDeferred(); + const original = ZipReader.prototype.getEntriesGenerator; + const enumeration = vi + .spyOn(ZipReader.prototype, "getEntriesGenerator") + .mockImplementation(async function* (this: ZipReader, options) { + for await (const entry of original.call(this, options)) { + if (!entry.directory) { + const getData = entry.getData.bind(entry); + entry.getData = async (writer, dataOptions) => { + try { + return await getData(writer, dataOptions); + } finally { + reached.resolve(); + await release.promise; + } + }; + } + yield entry; + } + return true; + }); + return { reached: reached.promise, release: release.resolve, enumeration }; +}; + const admittedZip = async () => { const directory = await createTestTempDirectory("rea-zip-short-read-"); const path = join(directory, "valid.zip"); @@ -392,3 +424,186 @@ it.each([null, { caller: "opaque ZIP interruption" }])( } }, ); + +describe("ZipArtifactReader producer ownership", () => { + it.each(["unread", "abandoned", "cancelled"] as const)( + "joins late extraction settlement for an %s stream before closing its source", + async (mode) => { + const directory = await createTestTempDirectory("rea-zip-producer-"); + const path = join(directory, "member.zip"); + const writer = new ZipWriter(new Uint8ArrayWriter()); + await writer.add( + "member.bin", + new Uint8ArrayReader(new Uint8Array(256 * 1024)), + { + level: 0, + }, + ); + await writeFile(path, await writer.close()); + const held = holdZipProducerSettlement(); + const fileClose = vi.spyOn(OwnedFileHandle.prototype, "close"); + const reader = new ZipArtifactReader(path, "zip"); + const caller = new AbortController(); + try { + const entries: ArtifactEntry[] = []; + for await (const entry of reader.entries()) entries.push(entry); + const stream = await reader.open(entries[0]!, caller.signal); + await once(stream, "readable"); + const reason = { caller: "ZIP owner cancellation" }; + if (mode === "cancelled") { + const digest = hashReadable(stream, caller.signal); + caller.abort(reason); + await expect(digest).rejects.toMatchObject({ + reason: "cancelled", + cause: reason, + }); + } else if (mode === "abandoned") { + for await (const _chunk of stream) break; + } + const closing = reader.close(); + expect(reader.close()).toBe(closing); + await held.reached; + expect(stream.destroyed).toBe(true); + expect(fileClose).not.toHaveBeenCalled(); + let closed = false; + void closing.then(() => { + closed = true; + }); + await Promise.resolve(); + expect(closed).toBe(false); + held.release(); + await closing; + expect(fileClose).toHaveBeenCalledTimes(1); + await reader.close(); + expect(fileClose).toHaveBeenCalledTimes(1); + expect(() => reader.open(entries[0]!)).toThrow("ZIP reader is closed"); + } finally { + held.release(); + await reader.close(); + held.enumeration.mockRestore(); + fileClose.mockRestore(); + } + }, + ); + + it("retires completed buffered output and retries a failed descriptor close", async () => { + const directory = await createTestTempDirectory("rea-zip-buffered-close-"); + const path = join(directory, "small.zip"); + await writeOrderedZip(path, ["member.txt"]); + const held = holdZipProducerSettlement(); + const reader = new ZipArtifactReader(path, "zip"); + try { + const entries: ArtifactEntry[] = []; + for await (const entry of reader.entries()) entries.push(entry); + const stream = await reader.open(entries[0]!); + await held.reached; + held.release(); + // getData has finished, but this tiny member remains buffered and unread. + await Promise.resolve(); + const closeFailure = new Error("ZIP descriptor close denied"); + const close = vi + .spyOn(OwnedFileHandle.prototype, "close") + .mockRejectedValueOnce(closeFailure); + try { + await expect(reader.close()).rejects.toBe(closeFailure); + expect(stream.destroyed).toBe(true); + await reader.close(); + expect(close).toHaveBeenCalledTimes(2); + } finally { + close.mockRestore(); + } + } finally { + held.release(); + await reader.close(); + held.enumeration.mockRestore(); + } + }); +}); + +it("joins an admitted ZIP enumeration before reader close completes", async () => { + const fixture = await admittedZip(); + const reached = createDeferred(); + const release = createDeferred(); + const nativeRead = fixture.positional.read.bind(fixture.positional); + const reads = vi + .spyOn(fixture.positional, "read") + .mockImplementation(async (...args) => { + reached.resolve(); + await release.promise; + return nativeRead(...args); + }); + const reader = new ZipArtifactReader(fixture.path, "zip", undefined, fixture); + const traversal = reader.entries()[Symbol.asyncIterator](); + const next = traversal.next(); + try { + await reached.promise; + let closed = false; + const closing = reader.close().then(() => { + closed = true; + }); + await Promise.resolve(); + expect(closed).toBe(false); + release.resolve(); + await expect(next).rejects.toMatchObject({ + reason: "unavailable", + message: "ZIP reader is closed", + }); + await closing; + const completedReads = reads.mock.calls.length; + await expect( + reader.entries()[Symbol.asyncIterator]().next(), + ).rejects.toThrow("ZIP reader is closed"); + expect(reads).toHaveBeenCalledTimes(completedReads); + } finally { + release.resolve(); + await next.catch(() => undefined); + await reader.close(); + reads.mockRestore(); + await fixture.handle.close(); + } +}); + +it("joins a late ZIP producer and closes the source despite an enumeration cleanup failure", async () => { + const directory = await createTestTempDirectory("rea-zip-mixed-cleanup-"); + const path = join(directory, "mixed.zip"); + await writeOrderedZip(path, ["member.txt"]); + const held = holdZipProducerSettlement(); + const enumerate = held.enumeration.getMockImplementation()!; + const failure = new Error("ZIP enumeration finalization denied"); + held.enumeration.mockImplementation(function ( + this: ZipReader, + options, + ) { + const source = enumerate.call(this, options); + const finish = source.return.bind(source); + vi.spyOn(source, "return").mockImplementationOnce(async (value) => { + await finish(value); + throw failure; + }); + return source; + }); + const fileClose = vi.spyOn(OwnedFileHandle.prototype, "close"); + const reader = new ZipArtifactReader(path, "zip"); + const traversal = reader.entries()[Symbol.asyncIterator](); + try { + const first = await traversal.next(); + if (first.done) throw new Error("Expected ZIP member"); + await reader.open(first.value); + await held.reached; + const closing = reader.close(); + const failureCheck = expect(closing).rejects.toBe(failure); + await Promise.resolve(); + expect(fileClose).not.toHaveBeenCalled(); + held.release(); + await failureCheck; + expect(fileClose).toHaveBeenCalledTimes(1); + await reader.close(); + expect(fileClose).toHaveBeenCalledTimes(1); + } finally { + held.release(); + await reader.close(); + await traversal.return?.(); + held.enumeration.mockRestore(); + fileClose.mockRestore(); + } +}); diff --git a/src/artifacts/ZipArtifactReader.ts b/src/artifacts/ZipArtifactReader.ts index dc03966bcf..9a7fc2fcd3 100644 --- a/src/artifacts/ZipArtifactReader.ts +++ b/src/artifacts/ZipArtifactReader.ts @@ -176,6 +176,10 @@ export class ZipArtifactReader implements ArtifactReader { readonly format: ZipPackageFormat; readonly #source: NodeFileReader; readonly #reader: ZipReader; + readonly #streams = new Map>(); + readonly #enumerations = new Set>(); + #closing = false; + #closePromise: Promise | undefined; readonly #entries = new Map< string, { readonly source: Entry; readonly metadata: ArtifactEntry } @@ -202,50 +206,60 @@ export class ZipArtifactReader implements ArtifactReader { } async *entries(signal?: AbortSignal): AsyncIterable { + this.#assertOpen(signal); let index = 0; // A traversal owns its accounting, including overlapping or repeated passes. let declaredFileBytes = 0; - for await (const entry of typedZipFailures( + const enumeration = typedZipFailures( this.#reader.getEntriesGenerator({ decodeText: decodeZipEntryText }), - )) { - assertArtifactReadActive(signal); - // ZIP member identity is its directory occurrence, not its filename. - // Logical path uniqueness belongs to the artifact inventory boundary. - const metadata = projectZipEntry(entry, index++); - if (metadata === undefined) continue; - // zip.js enforces each member's declared output size, so this sum bounds streamed work. - if ( - metadata.kind === "file" && - !metadata.encrypted && - metadata.declaredSize !== null && - Number.isSafeInteger(metadata.declaredSize) && - metadata.declaredSize >= 0 - ) { - declaredFileBytes += metadata.declaredSize; - const budget = artifactExpansionBudgetBytes(this.#source.size); - if (declaredFileBytes > budget) + ); + this.#enumerations.add(enumeration); + try { + for await (const entry of enumeration) { + this.#assertOpen(signal); + // ZIP member identity is its directory occurrence, not its filename. + // Logical path uniqueness belongs to the artifact inventory boundary. + const metadata = projectZipEntry(entry, index++); + if (metadata === undefined) continue; + // zip.js enforces each member's declared output size, so this sum bounds streamed work. + if ( + metadata.kind === "file" && + !metadata.encrypted && + metadata.declaredSize !== null && + Number.isSafeInteger(metadata.declaredSize) && + metadata.declaredSize >= 0 + ) { + declaredFileBytes += metadata.declaredSize; + const budget = artifactExpansionBudgetBytes(this.#source.size); + if (declaredFileBytes > budget) + throw new ArtifactReaderFailure( + "limit", + `ZIP entries declare ${String(declaredFileBytes)} uncompressed bytes, exceeding this archive's ${String(budget)}-byte decompression budget (${String(MINIMUM_ARTIFACT_EXPANDED_BYTES)}-byte minimum, ${String(ARTIFACT_EXPANSION_RATIO)}x archive size); a small archive cannot expand this far`, + ); + } + const stored = this.#entries.get(metadata.adapterKey); + if ( + stored !== undefined && + !sameArtifactEntry(metadata, stored.metadata) + ) throw new ArtifactReaderFailure( - "limit", - `ZIP entries declare ${String(declaredFileBytes)} uncompressed bytes, exceeding this archive's ${String(budget)}-byte decompression budget (${String(MINIMUM_ARTIFACT_EXPANDED_BYTES)}-byte minimum, ${String(ARTIFACT_EXPANSION_RATIO)}x archive size); a small archive cannot expand this far`, + "integrity", + `ZIP member identity changed during enumeration: ${entry.filename}`, ); + if (stored === undefined) + this.#entries.set(metadata.adapterKey, { + source: entry, + metadata: copyArtifactEntry(metadata), + }); + yield metadata; } - const stored = this.#entries.get(metadata.adapterKey); - if (stored !== undefined && !sameArtifactEntry(metadata, stored.metadata)) - throw new ArtifactReaderFailure( - "integrity", - `ZIP member identity changed during enumeration: ${entry.filename}`, - ); - if (stored === undefined) - this.#entries.set(metadata.adapterKey, { - source: entry, - metadata: copyArtifactEntry(metadata), - }); - yield metadata; + } finally { + this.#enumerations.delete(enumeration); } } open(entry: ArtifactEntry, signal?: AbortSignal): Promise { - assertArtifactReadActive(signal); + this.#assertOpen(signal); const stored = this.#entries.get(entry.adapterKey); if (stored === undefined || !sameArtifactEntry(entry, stored.metadata)) return Promise.reject( @@ -266,13 +280,55 @@ export class ZipArtifactReader implements ArtifactReader { "Encrypted ZIP entry is unsupported", ), ); - return Promise.resolve(extractStream(source, signal)); + const { output, completion } = extractStream(source, signal); + this.#streams.set(output, completion); + // Producer completion can precede consumption of a small buffered member. + // Retain the output until both it and its producer have finished. + output.once("close", () => { + void completion.then(() => this.#streams.delete(output)); + }); + return Promise.resolve(output); } - async close(): Promise { + close(): Promise { + this.#closing = true; + this.#closePromise ??= this.#close().finally(() => { + // A failed descriptor close retains its owner for a later retry. + this.#closePromise = undefined; + }); + return this.#closePromise; + } + + async #close(): Promise { + for (const output of this.#streams.keys()) + output.destroy( + new ArtifactReaderFailure("unavailable", "ZIP reader closed"), + ); + const outcomes = await Promise.allSettled([ + ...this.#streams.values(), + ...Array.from(this.#enumerations, (enumeration) => enumeration.return()), + ]); + this.#enumerations.clear(); // best-effort cleanup: zip-reader close must not mask prior extraction state. await this.#reader.close().catch(() => undefined); - await this.#source.closeHandle(); + const failures = outcomes.flatMap((outcome) => + outcome.status === "rejected" ? [outcome.reason as unknown] : [], + ); + try { + await this.#source.closeHandle(); + this.#entries.clear(); + } catch (cause: unknown) { + failures.push(cause); + } + if (failures.length === 1) throw failures[0]; + if (failures.length > 1) + throw new AggregateError(failures, "ZIP reader cleanup failed"); + } + + #assertOpen(signal?: AbortSignal): void { + assertArtifactReadActive(signal); + if (this.#closing) + throw new ArtifactReaderFailure("unavailable", "ZIP reader is closed"); } provenance(): readonly [] { @@ -377,8 +433,13 @@ const decodeStrictZipUtf8 = (bytes: Uint8Array, message: string): string => { } }; -const extractStream = (entry: FileEntry, signal?: AbortSignal): Readable => { +const extractStream = ( + entry: FileEntry, + signal?: AbortSignal, +): { readonly output: Readable; readonly completion: Promise } => { const output = new PassThrough(); + // Reader retirement can precede a caller attaching its own error listener. + output.on("error", () => undefined); const controller = new AbortController(); const onAbort = (): void => { controller.abort(signal?.reason); @@ -392,9 +453,8 @@ const extractStream = (entry: FileEntry, signal?: AbortSignal): Readable => { ); }; signal?.addEventListener("abort", onAbort, { once: true }); - // `close` covers both a consumer walking away mid-entry and normal completion - // after `end()`; aborting then is harmless because `getData` has already - // resolved, and it releases the producer in the abandoned-consumer case. + // A closed output cancels its producer; the reader separately awaits getData + // settlement before releasing the source descriptor. output.once("close", () => { signal?.removeEventListener("abort", onAbort); controller.abort(); @@ -412,15 +472,20 @@ const extractStream = (entry: FileEntry, signal?: AbortSignal): Readable => { output.destroy(toError(typedZipFailure(reason))); }, }); - void entry + const completion = entry .getData(writable, { signal: controller.signal, checkSignature: true, checkOverlappingEntry: true, onprogress: () => assertArtifactReadActive(signal), }) - .catch((cause: unknown) => output.destroy(toError(typedZipFailure(cause)))); - return output; + .then( + () => undefined, + (cause: unknown) => { + output.destroy(toError(typedZipFailure(cause))); + }, + ); + return { output, completion }; }; /** zip.js reports archive faults as plain errors identified by message. */ @@ -466,7 +531,7 @@ const typedZipFailure = (cause: unknown): unknown => { async function* typedZipFailures( source: AsyncIterable, -): AsyncIterable { +): AsyncGenerator { try { yield* source; } catch (cause: unknown) { diff --git a/tests/boundary/filesystem/desktopApplication.test.ts b/tests/boundary/filesystem/desktopApplication.test.ts index 3021aaacc8..906003e778 100644 --- a/tests/boundary/filesystem/desktopApplication.test.ts +++ b/tests/boundary/filesystem/desktopApplication.test.ts @@ -33,6 +33,7 @@ import { import { locateGodotPck } from "../../../src/artifacts/GodotPckLocation.js"; import { DirectoryArtifactReader } from "../../../src/artifacts/DirectoryArtifactReader.js"; import { ArtifactReaderFailure } from "../../../src/artifacts/ArtifactReader.js"; +import { artifactFilePrimaryFacts } from "../../../src/artifacts/ArtifactFileOperation.js"; import { DesktopApplicationReader } from "../../../src/artifacts/DesktopApplicationReader.js"; import { DesktopApplicationService } from "../../../src/application/artifacts/DesktopApplicationService.js"; @@ -749,6 +750,164 @@ it("preserves the selected root alias and directs follow-ups to the admitted phy } }); +it.each( + ["physical read", "format"].flatMap((variant) => + ["unavailable", "mismatch"].map((validationState) => ({ + variant, + validationState, + })), + ), +)( + "keeps $variant failure through $validationState final validation while discarding unverified desktop byte facts", + async ({ variant, validationState }) => { + const root = await createTestTempDirectory( + "rea-desktop-failed-validation-", + ); + const header = Buffer.alloc(20); + header.write("GDPC"); + header.writeUInt32LE(2, 4); + header.writeUInt32LE(4, 8); + await put(root, "game.pck", header); + const primary = + variant === "physical read" + ? Object.assign(new Error("physical read denied"), { code: "EACCES" }) + : new ArtifactReaderFailure( + "format", + "invalid later format declaration", + ); + const validation = Object.assign( + new Error("final descriptor stat failed"), + { code: "EIO" }, + ); + let observed: unknown; + const service = new DesktopApplicationService( + new DesktopApplicationReader((path) => { + const reader = new DirectoryArtifactReader(path); + const inspectFile = reader.inspectFile.bind(reader); + reader.inspectFile = (entry, inspect, signal) => + inspectFile( + entry, + async (file) => { + await inspect(file); + if (validationState === "unavailable") + vi.spyOn(file, "stat").mockRejectedValueOnce(validation); + else { + const changed = await file.stat({ bigint: true }); + changed.ino += 1n; + vi.spyOn(file, "stat").mockResolvedValueOnce(changed); + } + if (variant === "physical read") { + vi.spyOn(file, "read").mockRejectedValueOnce(primary); + await file.read(Buffer.alloc(4), 0, 4, 0); + } + throw primary; + }, + signal, + ).catch((cause: unknown) => { + observed = cause; + throw cause; + }); + return reader; + }), + ); + try { + const result = await service.inspect({ path: root }); + expect(observed).toMatchObject({ + reason: validationState === "unavailable" ? "io" : "integrity", + }); + if (validationState === "unavailable") + expect((observed as ArtifactReaderFailure).cause).toMatchObject({ + errors: [primary, validation], + }); + else expect((observed as ArtifactReaderFailure).cause).toBe(primary); + if (!result.ok) throw result.error; + const report = desktopApplicationSchema.parse( + result.value.normalized_result, + ); + expect(report.coverage.status).toBe("partial"); + expect(report.files[0]).toMatchObject({ + read_status: "unavailable", + ranges: [], + }); + expect( + report.files[0]?.facts.filter((fact) => fact.offset !== null), + ).toEqual([]); + expect(report.files[0]?.issues.join(" ")).toContain(primary.message); + expect(report.files[0]?.issues.join(" ")).toContain( + validationState === "unavailable" + ? validation.message + : "Directory entry changed during inspection", + ); + expect(report.candidates).toEqual([]); + } finally { + await service.close(); + vi.restoreAllMocks(); + } + }, +); + +it.each([false, true])( + "preserves a physical desktop read failure when cancellation arrives during final validation (failed close: %s)", + async (failedClose) => { + const root = await createTestTempDirectory( + "rea-desktop-read-final-validation-", + ); + await put(root, "game.pck", Buffer.from("completed source prefix")); + const reader = new DirectoryArtifactReader(root); + const entries = []; + for await (const entry of reader.entries()) entries.push(entry); + const controller = new AbortController(); + const reason = { phase: "after physical read failure" }; + const physical = Object.assign(new Error("physical read denied"), { + code: "EACCES", + }); + let fileHandle: Awaited> | undefined; + let denyClose = failedClose; + try { + const failure: unknown = await reader + .inspectFile( + entries[0]!, + async (file) => { + fileHandle = file; + const stat = file.stat.bind(file); + file.stat = ((...args: Parameters) => { + controller.abort(reason); + return stat(...args); + }) as typeof file.stat; + const close = file.close.bind(file); + file.close = async () => { + if (denyClose) throw new Error("desktop descriptor close denied"); + await close(); + }; + vi.spyOn(file, "read").mockRejectedValueOnce(physical); + await file.read(Buffer.alloc(4), 0, 4, 0); + }, + controller.signal, + ) + .catch((cause: unknown) => cause); + expect(controller.signal.reason).toBe(reason); + expect(failure).toMatchObject({ reason: "io" }); + const primary = artifactFilePrimaryFacts(failure).cause; + if (failedClose) expect(primary).toBe(physical); + else expect(primary).toMatchObject({ cause: physical }); + if (failedClose) { + expect(failure).toMatchObject({ + cleanup: { resources: [entries[0]!.adapterKey] }, + }); + await expect(reader.close()).rejects.toBeDefined(); + expect(fileHandle!.fd).toBeGreaterThanOrEqual(0); + } + denyClose = false; + await reader.close(); + expect(fileHandle!.fd).toBe(-1); + } finally { + denyClose = false; + await reader.close(); + vi.restoreAllMocks(); + } + }, +); + it.each([false, true])( "retains verified header facts and exact release authority when file close fails (late cancellation: %s)", async (cancelDuringClose) => {