fix(artifacts): retain reader work through cleanup

This commit is contained in:
morluto committed 2026-10-27 03:12:22 +08:00
1 parent d4c511ae6e
commit 5da50ab3f4
4 files changed
+500 -48

No files matched your search

+16 -3
View File
@@ -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;
},
);
+216 -1
View File
@@ -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<number> => {
return files;
};
/** Hold real zip.js finalization after its streaming work settles. */
const holdZipProducerSettlement = () => {
const reached = createDeferred<void>();
const release = createDeferred<void>();
const original = ZipReader.prototype.getEntriesGenerator;
const enumeration = vi
.spyOn(ZipReader.prototype, "getEntriesGenerator")
.mockImplementation(async function* (this: ZipReader<unknown>, 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<void>();
const release = createDeferred<void>();
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<unknown>,
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();
}
});
+109 -44
View File
@@ -176,6 +176,10 @@ export class ZipArtifactReader implements ArtifactReader {
readonly format: ZipPackageFormat;
readonly #source: NodeFileReader;
readonly #reader: ZipReader<string>;
readonly #streams = new Map<Readable, Promise<void>>();
readonly #enumerations = new Set<AsyncGenerator<Entry, void>>();
#closing = false;
#closePromise: Promise<void> | 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<ArtifactEntry> {
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<Readable> {
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<void> {
close(): Promise<void> {
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<void> {
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<void> } => {
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<T>(
source: AsyncIterable<T>,
): AsyncIterable<T> {
): AsyncGenerator<T, void> {
try {
yield* source;
} catch (cause: unknown) {
@@ -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<ReturnType<typeof open>> | 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<typeof file.stat>) => {
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) => {