diff --git a/src/artifacts/ArtifactFileSource.ts b/src/artifacts/ArtifactFileSource.ts new file mode 100644 index 0000000000..4eccd7f0dc --- /dev/null +++ b/src/artifacts/ArtifactFileSource.ts @@ -0,0 +1,146 @@ +import { artifactIntegrityDetails } from "../domain/artifactIntegrity.js"; +import { OwnedFileHandle } from "../filesystem/OwnedFileHandle.js"; +import { readFileHandleChunks } from "../filesystem/readFileHandleChunks.js"; +import { + NonRegularFileReadError, + RegularFileAdmissionFailure, + RegularFileOperationCancelled, + openRegularFile, + sameRegularFileState, + type StableRegularFileDescriptor, +} from "../filesystem/RegularFile.js"; +import { assertArtifactReadActive } from "./ArtifactFileOperation.js"; +import { hashReadable } from "./ArtifactHash.js"; +import { ArtifactReaderFailure } from "./ArtifactReader.js"; +import { ArtifactResourceScope } from "./ArtifactResourceScope.js"; + +/** One persistent descriptor and hash binding, borrowing caller-admitted sources. */ +export class ArtifactFileSource { + readonly #scope = new ArtifactResourceScope(); + #owner: OwnedFileHandle | undefined; + #opening: Promise | undefined; + #boundSha256: string | undefined; + + constructor( + private readonly path: string, + private readonly diagnosticLabel: string, + private readonly admitted?: StableRegularFileDescriptor, + ) { + this.#scope.addCloseHook(async () => { + if (this.#owner !== undefined) { + await this.#owner.close(); + this.#owner = undefined; + } + this.#opening = undefined; + this.#boundSha256 = undefined; + }); + } + + /** Verify the admitted descriptor; the selected pathname is not reinterpreted. */ + read(signal?: AbortSignal): Promise { + return this.#scope.run(() => this.#verify(signal)); + } + + /** Bind verified bytes to the inventory's expected whole-carrier identity. */ + bindSha256(expectedSha256: string, signal?: AbortSignal): Promise { + return this.#scope.run(async () => { + const { handle } = await this.#verify(signal); + const observed = + this.#boundSha256 ?? + ( + await hashReadable( + readFileHandleChunks(handle, { + signal, + cancellationFailure: (reason) => + new ArtifactReaderFailure( + "cancelled", + `${this.diagnosticLabel} verification cancelled`, + { cause: reason }, + ), + }), + signal, + ) + ).sha256; + await this.#verify(signal); + if (observed !== expectedSha256) + throw new ArtifactReaderFailure( + "integrity", + `${this.diagnosticLabel} carrier changed after inventory`, + undefined, + artifactIntegrityDetails( + ".", + { algorithm: "sha256", digest: expectedSha256 }, + observed, + false, + ), + ); + this.#boundSha256 = observed; + }); + } + + /** Join admitted operations before closing; failed descriptor ownership stays retained. */ + close(): Promise { + return this.#scope.close(); + } + + async #verify(signal?: AbortSignal): Promise { + assertArtifactReadActive(signal); + this.#opening ??= this.#admit(signal); + const source = await this.#opening; + const observed = await source.handle.stat(); + if (!sameRegularFileState(source.initial, observed)) + throw new ArtifactReaderFailure( + "integrity", + `${this.diagnosticLabel} source identity changed: ${this.path}`, + ); + assertArtifactReadActive(signal); + return source; + } + + async #admit(signal?: AbortSignal): Promise { + try { + const handle = + this.admitted?.handle ?? + (await openRegularFile(this.path, { symlinks: "reject", signal })); + if (this.admitted === undefined) + this.#owner = new OwnedFileHandle(handle); + const initial = this.admitted?.initial ?? (await handle.stat()); + return { handle, initial }; + } catch (cause: unknown) { + if (cause instanceof RegularFileAdmissionFailure) { + this.#owner = cause.owner; + throw ArtifactReaderFailure.withCleanup( + this.#admissionFailure(cause.cause), + ArtifactReaderFailure.cleanupObservation( + cause.cleanupCause, + this.path, + ), + ); + } + throw this.#admissionFailure(cause); + } + } + + #admissionFailure(cause: unknown): ArtifactReaderFailure { + if (cause instanceof ArtifactReaderFailure) return cause; + if (cause instanceof RegularFileOperationCancelled) + return new ArtifactReaderFailure( + "cancelled", + `${this.diagnosticLabel} admission cancelled`, + { cause: cause.reason }, + ); + if (cause instanceof NonRegularFileReadError) + return new ArtifactReaderFailure( + "format", + `${this.diagnosticLabel} source is not a regular file: ${this.path}`, + { cause }, + ); + return new ArtifactReaderFailure( + "io", + cause instanceof Error + ? cause.message + : `Cannot admit ${this.diagnosticLabel} source: ${this.path}`, + { cause }, + ); + } +} diff --git a/src/artifacts/GodotPckArtifactReader.ts b/src/artifacts/GodotPckArtifactReader.ts index 962241e0eb..7233553864 100644 --- a/src/artifacts/GodotPckArtifactReader.ts +++ b/src/artifacts/GodotPckArtifactReader.ts @@ -1,7 +1,4 @@ import { artifactExpansionBudgetBytes } from "../domain/artifactExpansionBudget.js"; -import { hashReadable } from "./ArtifactHash.js"; -import { artifactIntegrityDetails } from "../domain/artifactIntegrity.js"; -import type { Stats } from "node:fs"; import type { FileHandle } from "node:fs/promises"; import { setImmediate } from "node:timers/promises"; import { Readable, Transform, pipeline } from "node:stream"; @@ -18,16 +15,9 @@ import { assertArtifactReadActive } from "./ArtifactFileOperation.js"; import { locateGodotPck, readPckBytes } from "./GodotPckLocation.js"; import type { GodotPckMetadata } from "../domain/artifactFormatMetadata.js"; import { jsonValueByteLength } from "../domain/jsonValueByteLength.js"; -import { OwnedFileHandle } from "../filesystem/OwnedFileHandle.js"; import { readFileHandleChunks } from "../filesystem/readFileHandleChunks.js"; -import { - NonRegularFileReadError, - RegularFileAdmissionFailure, - RegularFileOperationCancelled, - openRegularFile, - sameRegularFileState, - type StableRegularFileDescriptor, -} from "../filesystem/RegularFile.js"; +import type { StableRegularFileDescriptor } from "../filesystem/RegularFile.js"; +import { ArtifactFileSource } from "./ArtifactFileSource.js"; const PACK_DIR_ENCRYPTED = 1; const PACK_REL_FILEBASE = 2; @@ -45,18 +35,14 @@ export class GodotPckArtifactReader implements ArtifactReader { readonly #scope = new ArtifactResourceScope(); readonly #entries = new Map(); readonly #streams = new Set(); - #owner: OwnedFileHandle | undefined; - #initial: Stats | undefined; - #opening: Promise | undefined; + readonly #source: ArtifactFileSource; #prepared: Promise | undefined; #header: GodotPckMetadata | undefined; #closing = false; - #boundSha256: string | undefined; - constructor( - private readonly path: string, - private readonly admitted?: StableRegularFileDescriptor, - ) {} + constructor(path: string, admitted?: StableRegularFileDescriptor) { + this.#source = new ArtifactFileSource(path, "Godot PCK", admitted); + } formatMetadata(): GodotPckMetadata | undefined { return this.#header === undefined ? undefined : { ...this.#header }; @@ -68,37 +54,7 @@ export class GodotPckArtifactReader implements ArtifactReader { ): Promise { await this.#scope.run(async () => { this.#assertOpen(signal); - const handle = await this.#assertStable(signal); - const observed = - this.#boundSha256 ?? - ( - await hashReadable( - readFileHandleChunks(handle, { - signal, - cancellationFailure: (reason) => - new ArtifactReaderFailure( - "cancelled", - "Godot PCK verification cancelled", - { cause: reason }, - ), - }), - signal, - ) - ).sha256; - await this.#assertStable(signal); - if (observed !== expectedSha256) - throw new ArtifactReaderFailure( - "integrity", - "Godot PCK carrier changed after inventory", - undefined, - artifactIntegrityDetails( - ".", - { algorithm: "sha256", digest: expectedSha256 }, - observed, - false, - ), - ); - this.#boundSha256 = observed; + await this.#source.bindSha256(expectedSha256, signal); }); } @@ -110,7 +66,7 @@ export class GodotPckArtifactReader implements ArtifactReader { this.#assertOpen(signal); this.#prepared ??= this.#prepare(signal); await this.#prepared; - await this.#assertStable(signal); + await this.#source.read(signal); }); if (this.#header?.directory_status !== "available") { options?.onCoverage?.({ @@ -145,7 +101,7 @@ export class GodotPckArtifactReader implements ArtifactReader { "unavailable", `Encrypted Godot PCK member is unsupported: ${stored.path}`, ); - const handle = await this.#assertStable(signal); + const { handle } = await this.#source.read(signal); this.#assertOpen(signal); if (stored.byteOffset === null || stored.declaredSize === null) throw formatFailure("member has no physical byte range"); @@ -196,10 +152,7 @@ export class GodotPckArtifactReader implements ArtifactReader { new ArtifactReaderFailure("unavailable", "Godot PCK reader closed"), ); await this.#scope.close(); - if (this.#owner !== undefined) { - await this.#owner.close(); - this.#owner = undefined; - } + await this.#source.close(); this.#entries.clear(); } @@ -216,56 +169,8 @@ export class GodotPckArtifactReader implements ArtifactReader { ); } - #open(signal?: AbortSignal): Promise { - this.#opening ??= this.#admit(signal); - return this.#opening; - } - - async #admit(signal?: AbortSignal): Promise { - try { - const handle = - this.admitted?.handle ?? - (await openRegularFile(this.path, { symlinks: "reject", signal })); - if (this.admitted === undefined) - this.#owner = new OwnedFileHandle(handle); - this.#initial = this.admitted?.initial ?? (await handle.stat()); - return handle; - } catch (cause: unknown) { - if (cause instanceof RegularFileAdmissionFailure) { - this.#owner = cause.owner; - throw ArtifactReaderFailure.withCleanup( - admissionFailure(cause.cause, this.path), - ArtifactReaderFailure.cleanupObservation( - cause.cleanupCause, - this.path, - ), - ); - } - throw admissionFailure(cause, this.path); - } - } - - async #assertStable(signal?: AbortSignal): Promise { - assertArtifactReadActive(signal); - const handle = await this.#open(signal); - const observed = await handle.stat(); - if ( - this.#initial === undefined || - !sameRegularFileState(this.#initial, observed) - ) - throw new ArtifactReaderFailure( - "integrity", - `Godot PCK source identity changed: ${this.path}`, - ); - assertArtifactReadActive(signal); - return handle; - } - async #prepare(signal?: AbortSignal): Promise { - const handle = await this.#assertStable(signal); - const initial = this.#initial; - if (initial === undefined) - throw formatFailure("source identity was not admitted"); + const { handle, initial } = await this.#source.read(signal); const header = await readGodotPckHeader(handle, initial.size, signal); this.#header = header; if (header.directory_status !== "available") return; @@ -353,7 +258,7 @@ export class GodotPckArtifactReader implements ArtifactReader { if (retainedBytes > MAX_DIRECTORY_METADATA_BYTES) throw metadataLimit(); this.#entries.set(metadata.adapterKey, metadata); } - await this.#assertStable(signal); + await this.#source.read(signal); } #memberStream(entry: ArtifactEntry, signal?: AbortSignal): Transform { @@ -361,7 +266,7 @@ export class GodotPckArtifactReader implements ArtifactReader { const check = async (): Promise => { if (bytes !== entry.declaredSize) throw formatFailure(`member ended early: ${entry.path}`); - await this.#assertStable(signal); + await this.#source.read(signal); }; return new Transform({ transform(chunk: Buffer, _encoding, callback) { @@ -536,29 +441,3 @@ const metadataLimit = (): ArtifactReaderFailure => "Godot PCK directory exceeds the 64 MiB encoded metadata budget", { boundary: "memory" }, ); - -const admissionFailure = ( - cause: unknown, - path: string, -): ArtifactReaderFailure => { - if (cause instanceof ArtifactReaderFailure) return cause; - if (cause instanceof RegularFileOperationCancelled) - return new ArtifactReaderFailure( - "cancelled", - "Godot PCK admission cancelled", - { cause: cause.reason }, - ); - if (cause instanceof NonRegularFileReadError) - return new ArtifactReaderFailure( - "format", - `Godot PCK source is not a regular file: ${path}`, - { cause }, - ); - return new ArtifactReaderFailure( - "io", - cause instanceof Error - ? cause.message - : `Cannot admit Godot PCK source: ${path}`, - { cause }, - ); -}; diff --git a/src/artifacts/NodeSeaArtifactReader.ts b/src/artifacts/NodeSeaArtifactReader.ts index fc566ae9a8..c8971a15d6 100644 --- a/src/artifacts/NodeSeaArtifactReader.ts +++ b/src/artifacts/NodeSeaArtifactReader.ts @@ -1,11 +1,8 @@ -import type { Stats } from "node:fs"; -import type { FileHandle } from "node:fs/promises"; import { Readable, Transform, pipeline } from "node:stream"; import { setImmediate } from "node:timers/promises"; import { artifactExpansionBudgetBytes } from "../domain/artifactExpansionBudget.js"; import type { NodeSeaMetadata } from "../domain/artifactFormatMetadata.js"; -import { artifactIntegrityDetails } from "../domain/artifactIntegrity.js"; import { ArtifactPathRegistry, normalizeArtifactPath, @@ -19,7 +16,6 @@ import { type ArtifactTraversalOptions, } from "./ArtifactReader.js"; import { ArtifactResourceScope } from "./ArtifactResourceScope.js"; -import { hashReadable } from "./ArtifactHash.js"; import { assertArtifactReadActive } from "./ArtifactFileOperation.js"; import { NODE_SEA_BLOB_TOO_LARGE, @@ -33,16 +29,9 @@ import { readNodeSeaBlob, scanNodeSeaFuse, } from "./nodeSea/NodeSeaLocation.js"; -import { OwnedFileHandle } from "../filesystem/OwnedFileHandle.js"; import { readFileHandleChunks } from "../filesystem/readFileHandleChunks.js"; -import { - NonRegularFileReadError, - RegularFileAdmissionFailure, - RegularFileOperationCancelled, - openRegularFile, - sameRegularFileState, - type StableRegularFileDescriptor, -} from "../filesystem/RegularFile.js"; +import type { StableRegularFileDescriptor } from "../filesystem/RegularFile.js"; +import { ArtifactFileSource } from "./ArtifactFileSource.js"; const SNAPSHOT_PATH = "__node_sea__/snapshot"; const CODE_CACHE_PATH = "__node_sea__/code-cache"; @@ -57,18 +46,14 @@ export class NodeSeaArtifactReader implements ArtifactReader { readonly #scope = new ArtifactResourceScope(); readonly #entries = new Map(); readonly #streams = new Set(); - #owner: OwnedFileHandle | undefined; - #initial: Stats | undefined; - #opening: Promise | undefined; + readonly #source: ArtifactFileSource; #prepared: Promise | undefined; #header: NodeSeaMetadata | undefined; #closing = false; - #boundSha256: string | undefined; - constructor( - private readonly path: string, - private readonly admitted?: StableRegularFileDescriptor, - ) {} + constructor(path: string, admitted?: StableRegularFileDescriptor) { + this.#source = new ArtifactFileSource(path, "Node SEA", admitted); + } formatMetadata(): NodeSeaMetadata | undefined { if (this.#header === undefined) return undefined; @@ -81,37 +66,7 @@ export class NodeSeaArtifactReader implements ArtifactReader { ): Promise { await this.#scope.run(async () => { this.#assertOpen(signal); - const handle = await this.#assertStable(signal); - const observed = - this.#boundSha256 ?? - ( - await hashReadable( - readFileHandleChunks(handle, { - signal, - cancellationFailure: (reason) => - new ArtifactReaderFailure( - "cancelled", - "Node SEA verification cancelled", - { cause: reason }, - ), - }), - signal, - ) - ).sha256; - await this.#assertStable(signal); - if (observed !== expectedSha256) - throw new ArtifactReaderFailure( - "integrity", - "Node SEA carrier changed after inventory", - undefined, - artifactIntegrityDetails( - ".", - { algorithm: "sha256", digest: expectedSha256 }, - observed, - false, - ), - ); - this.#boundSha256 = observed; + await this.#source.bindSha256(expectedSha256, signal); }); } @@ -123,7 +78,7 @@ export class NodeSeaArtifactReader implements ArtifactReader { this.#assertOpen(signal); this.#prepared ??= this.#prepare(signal); await this.#prepared; - await this.#assertStable(signal); + await this.#source.read(signal); }); const contradiction = this.#header?.fuse_contradiction; if (contradiction !== null && contradiction !== undefined) @@ -149,7 +104,7 @@ export class NodeSeaArtifactReader implements ArtifactReader { "integrity", "Node SEA entry does not match its owning inventory", ); - const handle = await this.#assertStable(signal); + const { handle } = await this.#source.read(signal); this.#assertOpen(signal); if (stored.byteOffset === null || stored.declaredSize === null) throw formatFailure("member has no physical byte range"); @@ -200,10 +155,7 @@ export class NodeSeaArtifactReader implements ArtifactReader { new ArtifactReaderFailure("unavailable", "Node SEA reader closed"), ); await this.#scope.close(); - if (this.#owner !== undefined) { - await this.#owner.close(); - this.#owner = undefined; - } + await this.#source.close(); this.#entries.clear(); } @@ -220,56 +172,8 @@ export class NodeSeaArtifactReader implements ArtifactReader { ); } - #open(signal?: AbortSignal): Promise { - this.#opening ??= this.#admit(signal); - return this.#opening; - } - - async #admit(signal?: AbortSignal): Promise { - try { - const handle = - this.admitted?.handle ?? - (await openRegularFile(this.path, { symlinks: "reject", signal })); - if (this.admitted === undefined) - this.#owner = new OwnedFileHandle(handle); - this.#initial = this.admitted?.initial ?? (await handle.stat()); - return handle; - } catch (cause: unknown) { - if (cause instanceof RegularFileAdmissionFailure) { - this.#owner = cause.owner; - throw ArtifactReaderFailure.withCleanup( - admissionFailure(cause.cause, this.path), - ArtifactReaderFailure.cleanupObservation( - cause.cleanupCause, - this.path, - ), - ); - } - throw admissionFailure(cause, this.path); - } - } - - async #assertStable(signal?: AbortSignal): Promise { - assertArtifactReadActive(signal); - const handle = await this.#open(signal); - const observed = await handle.stat(); - if ( - this.#initial === undefined || - !sameRegularFileState(this.#initial, observed) - ) - throw new ArtifactReaderFailure( - "integrity", - `Node SEA source identity changed: ${this.path}`, - ); - assertArtifactReadActive(signal); - return handle; - } - async #prepare(signal?: AbortSignal): Promise { - const handle = await this.#assertStable(signal); - const initial = this.#initial; - if (initial === undefined) - throw formatFailure("source identity was not admitted"); + const { handle, initial } = await this.#source.read(signal); const location = await locateNodeSea(handle, initial.size, signal); if (location === undefined) throw formatFailure("blob was not found"); if (location.blobLength > NODE_SEA_MAX_BLOB_BYTES) @@ -341,7 +245,7 @@ export class NodeSeaArtifactReader implements ArtifactReader { fuse_sentinel: fuse.fuseSentinel, fuse_contradiction: fuse.fuseContradiction, }; - await this.#assertStable(signal); + await this.#source.read(signal); } #memberStream(entry: ArtifactEntry, signal?: AbortSignal): Transform { @@ -349,7 +253,7 @@ export class NodeSeaArtifactReader implements ArtifactReader { const check = async (): Promise => { if (bytes !== entry.declaredSize) throw formatFailure(`member ended early: ${entry.path}`); - await this.#assertStable(signal); + await this.#source.read(signal); }; return new Transform({ transform(chunk: Buffer, _encoding, callback) { @@ -438,31 +342,3 @@ function* seaMembers(decoded: DecodedNodeSeaBlob): Generator<{ const formatFailure = (detail: string): ArtifactReaderFailure => new ArtifactReaderFailure("format", `Node SEA blob is malformed: ${detail}`); - -const admissionFailure = ( - cause: unknown, - path: string, -): ArtifactReaderFailure => { - if (cause instanceof ArtifactReaderFailure) return cause; - if (cause instanceof RegularFileOperationCancelled) - return new ArtifactReaderFailure( - "cancelled", - "Node SEA admission cancelled", - { - cause: cause.reason, - }, - ); - if (cause instanceof NonRegularFileReadError) - return new ArtifactReaderFailure( - "format", - `Node SEA source is not a regular file: ${path}`, - { cause }, - ); - return new ArtifactReaderFailure( - "io", - cause instanceof Error - ? cause.message - : `Cannot admit Node SEA source: ${path}`, - { cause }, - ); -}; diff --git a/tests/process-global/artifactFileSourceLifecycle.test.ts b/tests/process-global/artifactFileSourceLifecycle.test.ts new file mode 100644 index 0000000000..7bf58a85a4 --- /dev/null +++ b/tests/process-global/artifactFileSourceLifecycle.test.ts @@ -0,0 +1,464 @@ +import { createHash } from "node:crypto"; +import { + open, + readFile, + rename, + writeFile, + type FileHandle, +} from "node:fs/promises"; +import { dirname, join } from "node:path"; +import { text } from "node:stream/consumers"; +import { setImmediate } from "node:timers/promises"; +import { afterEach, expect, it, vi } from "vitest"; + +import { + ArtifactReaderFailure, + type ArtifactEntry, + type ArtifactReader, +} from "../../src/artifacts/ArtifactReader.js"; +import { ArtifactResourceScope } from "../../src/artifacts/ArtifactResourceScope.js"; +import { GodotPckArtifactReader } from "../../src/artifacts/GodotPckArtifactReader.js"; +import { NodeSeaArtifactReader } from "../../src/artifacts/NodeSeaArtifactReader.js"; +import { nodeSeaElf } from "../../src/artifacts/nodeSea/NodeSea.fixture.js"; +import type { StableRegularFileDescriptor } from "../../src/filesystem/RegularFile.js"; +import { createDeferred } from "../fixtures/binarySession.js"; +import { buildGodotPck } from "../fixtures/godotPck.js"; +import { createTestTempDirectory } from "../fixtures/temporaryDirectory.js"; + +interface OpenControl { + readonly paths: Set; + readonly handles: Map Promise>; + onOpen: ((handle: FileHandle, path: string) => Promise) | undefined; +} + +const control = vi.hoisted((): OpenControl => ({ + paths: new Set(), + handles: new Map(), + onOpen: undefined, +})); + +vi.mock("node:fs/promises", async (importOriginal) => { + const original = await importOriginal(); + return { + ...original, + open: async (...args: Parameters) => { + const handle = await original.open(...args); + const path = String(args[0]); + if (control.paths.has(path)) { + control.handles.set(handle, handle.close.bind(handle)); + await control.onOpen?.(handle, path); + } + return handle; + }, + }; +}); + +afterEach(async () => { + control.onOpen = undefined; + for (const [handle, close] of control.handles) + if (handle.fd >= 0) await close(); + control.handles.clear(); + control.paths.clear(); + vi.restoreAllMocks(); +}); + +const BODY = "export const artifact = 'retained';\n"; +const carriers = [ + { + name: "Godot PCK", + bytes: (body = BODY) => buildGodotPck(3, [{ path: "main.js", data: body }]), + reader: (path: string, admitted?: StableRegularFileDescriptor) => + new GodotPckArtifactReader(path, admitted), + }, + { + name: "Node SEA", + bytes: (body = BODY) => nodeSeaElf({ codePath: "main.js", main: body }), + reader: (path: string, admitted?: StableRegularFileDescriptor) => + new NodeSeaArtifactReader(path, admitted), + }, +] as const; + +const carrierFile = async (carrier: (typeof carriers)[number]) => { + const root = await createTestTempDirectory("rea-artifact-source-lifecycle-"); + const path = join(root, "carrier.bin"); + const bytes = carrier.bytes(); + await writeFile(path, bytes); + control.paths.add(path); + return { path, sha256: createHash("sha256").update(bytes).digest("hex") }; +}; + +const entries = async (reader: ArtifactReader): Promise => { + const values: ArtifactEntry[] = []; + for await (const entry of reader.entries()) values.push(entry); + return values; +}; + +const openedHandle = (): FileHandle => { + const handles = [...control.handles.keys()]; + expect(handles).toHaveLength(1); + const handle = handles[0]; + if (handle === undefined) throw new Error("Missing admitted descriptor"); + return handle; +}; + +const awaitAcknowledgement = async ( + acknowledgement: Promise, + operation: Promise, + boundary: string, +): Promise => { + let deadline: ReturnType | undefined; + try { + const observed = await Promise.race([ + acknowledgement.then(() => ({ kind: "acknowledged" }) as const), + operation.then( + (value) => ({ kind: "settled", status: "fulfilled", value }) as const, + (cause: unknown) => + ({ kind: "settled", status: "rejected", cause }) as const, + ), + new Promise<{ readonly kind: "deadline" }>((resolve) => { + deadline = setTimeout(() => resolve({ kind: "deadline" }), 2_000); + }), + ]); + if (observed.kind !== "acknowledged") + throw new Error( + `${boundary} was not acknowledged before ${observed.kind === "deadline" ? "the gate deadline" : "the triggering operation settled"}`, + { cause: observed }, + ); + } finally { + if (deadline !== undefined) clearTimeout(deadline); + } +}; + +it.each(carriers)( + "$name keeps the caller's expected digest and rechecks identity after binding", + async (carrier) => { + const { path, sha256 } = await carrierFile(carrier); + const reader = carrier.reader(path); + const expected = "0".repeat(64); + try { + await reader.prepareContainer(sha256); + await expect(reader.prepareContainer(expected)).rejects.toMatchObject({ + reason: "integrity", + details: { + logicalPath: ".", + declaredSha256: expected, + calculatedSha256: sha256, + unpacked: false, + }, + }); + await reader.prepareContainer(sha256); + await writeFile(path, carrier.bytes(BODY.repeat(3))); + await expect(reader.prepareContainer(sha256)).rejects.toMatchObject({ + reason: "integrity", + message: `${carrier.name} source identity changed: ${path}`, + }); + } finally { + await reader.close(); + } + }, +); + +it.each(carriers)( + "$name joins late admission before close and refuses replacement work", + async (carrier) => { + const { path, sha256 } = await carrierFile(carrier); + const admitted = createDeferred(); + const release = createDeferred(); + control.onOpen = async (handle) => { + vi.spyOn(handle, "close"); + admitted.resolve(); + await release.promise; + }; + const reader = carrier.reader(path); + const operations: Promise[] = []; + let closed = false; + try { + const preparing = reader.prepareContainer(sha256); + operations.push(preparing); + await awaitAcknowledgement( + admitted.promise, + preparing, + "carrier admission", + ); + const handle = openedHandle(); + const closeAttempts = [reader.close(), reader.close()]; + operations.push(...closeAttempts); + const closing = Promise.all(closeAttempts).then(() => { + closed = true; + }); + operations.push(closing); + await setImmediate(); + expect(closed).toBe(false); + expect(handle.fd).toBeGreaterThanOrEqual(0); + expect(handle.close).not.toHaveBeenCalled(); + release.resolve(); + await preparing; + await closing; + expect(handle.fd).toBe(-1); + expect(handle.close).toHaveBeenCalledOnce(); + await expect(reader.prepareContainer(sha256)).rejects.toMatchObject({ + reason: "unavailable", + }); + await expect( + reader.entries()[Symbol.asyncIterator]().next(), + ).rejects.toMatchObject({ + reason: "unavailable", + }); + expect(control.handles.size).toBe(1); + await reader.close(); + expect(handle.close).toHaveBeenCalledOnce(); + } finally { + release.resolve(); + await Promise.allSettled(operations); + await reader.close(); + } + }, +); + +it.each(carriers)( + "$name retains a failed physical close and retries only that exact descriptor", + async (carrier) => { + const { path, sha256 } = await carrierFile(carrier); + const entered = createDeferred(); + const release = createDeferred(); + const failure = new Error("controlled carrier close failure"); + let blocked = true; + control.onOpen = async (handle) => { + const close = handle.close.bind(handle); + vi.spyOn(handle, "close").mockImplementation(async () => { + entered.resolve(); + await release.promise; + if (blocked) throw failure; + await close(); + }); + }; + const reader = carrier.reader(path); + let closing: Promise[]> | undefined; + try { + await reader.prepareContainer(sha256); + const [entry] = await entries(reader); + if (entry === undefined) throw new Error("Missing carrier member"); + expect(await text(await reader.open(entry))).toBe(BODY); + const handle = openedHandle(); + closing = Promise.allSettled([reader.close(), reader.close()]); + await awaitAcknowledgement(entered.promise, closing, "carrier close"); + expect(handle.fd).toBeGreaterThanOrEqual(0); + expect(handle.close).toHaveBeenCalledOnce(); + release.resolve(); + expect(await closing).toEqual([ + { status: "rejected", reason: failure }, + { status: "rejected", reason: failure }, + ]); + expect(handle.fd).toBeGreaterThanOrEqual(0); + blocked = false; + await Promise.all([reader.close(), reader.close()]); + expect(handle.fd).toBe(-1); + expect(handle.close).toHaveBeenCalledTimes(2); + await reader.close(); + expect(handle.close).toHaveBeenCalledTimes(2); + expect(control.handles.size).toBe(1); + } finally { + blocked = false; + release.resolve(); + await closing; + await reader.close(); + } + }, +); + +it.each(carriers)( + "$name borrows the admitted descriptor after path replacement and leaves its close to the caller", + async (carrier) => { + const { path, sha256 } = await carrierFile(carrier); + const handle = await open(path, "r"); + const close = vi.spyOn(handle, "close"); + let reader: ReturnType | undefined; + try { + const replacement = join(dirname(path), "replacement.bin"); + const replacementBytes = carrier.bytes("replacement source\n"); + await writeFile(replacement, replacementBytes); + await rename(replacement, path); + reader = carrier.reader(path, { + handle, + initial: await handle.stat(), + }); + await reader.prepareContainer(sha256); + const [entry] = await entries(reader); + if (entry === undefined) + throw new Error("Missing original carrier member"); + expect(await text(await reader.open(entry))).toBe(BODY); + await reader.close(); + await reader.close(); + expect(close).not.toHaveBeenCalled(); + expect(handle.fd).toBeGreaterThanOrEqual(0); + expect(await readFile(path)).toEqual(replacementBytes); + expect(control.handles.size).toBe(1); + await handle.close(); + expect(close).toHaveBeenCalledOnce(); + expect(handle.fd).toBe(-1); + await reader.close(); + expect(close).toHaveBeenCalledOnce(); + } finally { + try { + await reader?.close(); + } finally { + if (handle.fd >= 0) await handle.close(); + } + } + }, +); + +const admissionCases = carriers.flatMap((carrier) => + ( + [ + { primary: "io", closeFails: false }, + { primary: "io", closeFails: true }, + { primary: "cancelled", closeFails: false }, + { primary: "cancelled", closeFails: true }, + ] as const + ).map((scenario) => ({ name: carrier.name, carrier, ...scenario })), +); + +it.each(admissionCases)( + "$name awaits physical admission cleanup and preserves $primary (closeFails=$closeFails)", + async ({ carrier, primary, closeFails }) => { + const { path } = await carrierFile(carrier); + const entered = createDeferred(); + const release = createDeferred(); + const controller = new AbortController(); + const admissionFailure = new Error("controlled descriptor stat failure"); + const cleanupFailure = new Error("controlled admission close failure"); + const cancellation = { request: "cancelled carrier admission" }; + let blocked = closeFails; + control.onOpen = async (handle) => { + if (primary === "io") + vi.spyOn(handle, "stat").mockRejectedValueOnce(admissionFailure); + else controller.abort(cancellation); + const close = handle.close.bind(handle); + vi.spyOn(handle, "close").mockImplementation(async () => { + entered.resolve(); + await release.promise; + if (blocked) throw cleanupFailure; + await close(); + }); + }; + const reader = carrier.reader(path); + let settled = false; + let reading: Promise | undefined; + try { + const operation = reader + .entries(controller.signal) + [Symbol.asyncIterator]() + .next(); + reading = operation + .catch((cause: unknown) => cause) + .finally(() => { + settled = true; + }); + await awaitAcknowledgement( + entered.promise, + operation, + "admission cleanup", + ); + const handle = openedHandle(); + // An earlier metadata failure stays primary when cancellation arrives in close. + controller.abort(cancellation); + await setImmediate(); + expect(settled).toBe(false); + expect(handle.fd).toBeGreaterThanOrEqual(0); + release.resolve(); + const failure = await reading; + if (!(failure instanceof ArtifactReaderFailure)) + throw new Error("Expected a typed carrier admission failure"); + expect(failure.reason).toBe(primary); + const admission = closeFails ? failure.cause : failure; + if (!(admission instanceof ArtifactReaderFailure)) + throw new Error("Missing the primary admission failure"); + expect(admission.cause).toBe( + primary === "io" ? admissionFailure : cancellation, + ); + expect(failure.cancellation).toEqual( + primary === "cancelled" ? { reason: cancellation } : undefined, + ); + if (closeFails) { + expect(failure.cleanup).toEqual({ + reason: cleanupFailure.message, + resources: [path], + }); + expect(handle.fd).toBeGreaterThanOrEqual(0); + await expect(reader.close()).rejects.toBe(cleanupFailure); + expect(handle.close).toHaveBeenCalledTimes(2); + blocked = false; + await reader.close(); + expect(handle.close).toHaveBeenCalledTimes(3); + } else { + expect(failure.cleanup).toBeUndefined(); + expect(handle.fd).toBe(-1); + await reader.close(); + expect(handle.close).toHaveBeenCalledOnce(); + } + expect(handle.fd).toBe(-1); + expect(control.handles.size).toBe(1); + } finally { + controller.abort(cancellation); + blocked = false; + release.resolve(); + await reading; + await reader.close(); + } + }, +); + +it("retries a failed reader without closing a successfully released peer again", async () => { + const good = await carrierFile(carriers[0]); + const failed = await carrierFile(carriers[1]); + const cleanupFailure = new Error("controlled Node SEA peer close failure"); + let blocked = true; + control.onOpen = async (handle, path) => { + const close = handle.close.bind(handle); + vi.spyOn(handle, "close").mockImplementation(async () => { + if (path === failed.path && blocked) throw cleanupFailure; + await close(); + }); + }; + const goodReader = carriers[0].reader(good.path); + const failedReader = carriers[1].reader(failed.path); + const scope = new ArtifactResourceScope(); + try { + await goodReader.prepareContainer(good.sha256); + await failedReader.prepareContainer(failed.sha256); + const handles = [...control.handles.keys()]; + const [goodHandle, failedHandle] = handles; + if (goodHandle === undefined || failedHandle === undefined) + throw new Error("Missing the two exact admitted descriptors"); + expect( + await scope.release({ + kind: "reader", + resource: good.path, + reader: goodReader, + }), + ).toEqual({ kind: "released" }); + const release = await scope.release({ + kind: "reader", + resource: failed.path, + reader: failedReader, + }); + expect(release).toMatchObject({ kind: "failed", cause: cleanupFailure }); + expect(goodHandle.fd).toBe(-1); + expect(failedHandle.fd).toBeGreaterThanOrEqual(0); + expect(scope.cleanupResources).toEqual([failed.path]); + blocked = false; + await Promise.all([scope.close(), scope.close()]); + await scope.close(); + expect(goodHandle.close).toHaveBeenCalledOnce(); + expect(failedHandle.close).toHaveBeenCalledTimes(2); + expect(failedHandle.fd).toBe(-1); + expect(scope.cleanupResources).toEqual([]); + expect(control.handles.size).toBe(2); + } finally { + blocked = false; + await scope.close(); + await goodReader.close(); + await failedReader.close(); + } +});