fix(artifacts): preserve physical failures during cancellation (#9641)

* fix(ines): preserve physical read failures on cancellation

* fix(artifacts): preserve root identity failures on cancellation

* refactor(artifacts): reuse cancellation projection

src/artifacts/ArtifactFileOperation.ts

@@ -338,7 +338,8 @@ export const withArtifactFileHandle = async <Value>(
   }
 };
 
-const artifactOperationCause = (cause: unknown): unknown =>
+/** Convert only cancellation produced by the regular-file owner. */
+export const artifactOperationCause = (cause: unknown): unknown =>
   cause instanceof RegularFileOperationCancelled
     ? new ArtifactReaderFailure(
         "cancelled",

src/artifacts/InesArtifactReader.ts

@@ -1,3 +1,4 @@
+import { artifactOperationCause } from "./ArtifactFileOperation.js";
 import { readFileHandlePrefix } from "../filesystem/readFileHandlePrefix.js";
 import { Readable } from "node:stream";
 import {
@@ -92,6 +93,10 @@ export class InesArtifactReader implements ArtifactReader {
       start: offset,
       end: offset + size - 1,
       signal,
+      cancellationFailure: (reason) =>
+        new ArtifactReaderFailure("cancelled", "Artifact inventory cancelled", {
+          cause: reason,
+        }),
     });
     const verify = () => this.#verifySource(signal);
     return Readable.from(
@@ -111,7 +116,6 @@ export class InesArtifactReader implements ArtifactReader {
             );
           await verify();
         } catch (cause: unknown) {
-          abortIfNeeded(signal);
           if (cause instanceof ArtifactReaderFailure) throw cause;
           throw new ArtifactReaderFailure("io", "Could not read iNES region", {
             cause,
@@ -177,14 +181,13 @@ export class InesArtifactReader implements ArtifactReader {
         { symlinks: "reject", ...(signal === undefined ? {} : { signal }) },
       );
     } catch (cause: unknown) {
-      abortIfNeeded(signal);
       if (cause instanceof RegularFileChangedError)
         throw new ArtifactReaderFailure(
           "integrity",
           "iNES source changed during reading",
           { cause: cause.cause ?? cause },
         );
-      throw cause;
+      throw artifactOperationCause(cause);
     }
     // State validation is asynchronous even after the final byte was consumed.
     abortIfNeeded(signal);

src/artifacts/inventory/classify.ts

@@ -1,3 +1,4 @@
+import { artifactOperationCause } from "../ArtifactFileOperation.js";
 import {
   nesContainerFormat,
   parseInesLayout,
@@ -244,14 +245,13 @@ const classifyRootFormat = async (
       ...(signal === undefined ? {} : { signal }),
     });
   } catch (cause: unknown) {
-    abortIfNeeded(signal);
     if (cause instanceof RegularFileChangedError)
       throw new ArtifactReaderFailure(
         "integrity",
         `Root artifact changed during classification: ${path}`,
         { cause: cause.cause ?? cause },
       );
-    throw cause;
+    throw artifactOperationCause(cause);
   }
   // Fill each range: a short result is end of file, not a partial host read.
   const read = async (position: number, length: number): Promise<Buffer> => {

src/artifacts/inventory/hashStableRootArtifact.ts

@@ -1,3 +1,4 @@
+import { artifactOperationCause } from "../ArtifactFileOperation.js";
 import { regularFileCancellation } from "../../filesystem/OwnedRegularFile.js";
 import type { Stats } from "node:fs";
 import type { FileHandle } from "node:fs/promises";
@@ -161,14 +162,13 @@ export const hashStableRootArtifactHandle = async (
       ...(signal === undefined ? {} : { signal }),
     });
   } catch (cause: unknown) {
-    abortIfNeeded(signal);
     if (cause instanceof RegularFileChangedError)
       throw new ArtifactReaderFailure(
         "integrity",
         `Root artifact changed during inventory: ${path}`,
         { cause: cause.cause ?? cause },
       );
-    throw cause;
+    throw artifactOperationCause(cause);
   }
   if (digest.bytes !== initial.size)
     throw new ArtifactReaderFailure(

tests/boundary/filesystem/inesCancellation.test.ts

@@ -318,3 +318,167 @@ it.each([
     }
   },
 );
+
+it.each(["read", "stat"] as const)(
+  "preserves an iNES descriptor %s failure concurrent with cancellation",
+  async (phase) => {
+    const root = await createTestTempDirectory("rea-ines-physical-failure-");
+    const path = join(root, "source.nes");
+    await writeFile(path, inesFixture());
+    const handle = await open(path, "r");
+    const reader = new InesArtifactReader(path, {
+      handle,
+      initial: await handle.stat(),
+    });
+    const controller = new AbortController();
+    const physical = Object.assign(new Error("descriptor I/O failed"), {
+      code: "EIO",
+    });
+    let restore = () => {};
+    try {
+      const entries = [];
+      for await (const entry of reader.entries()) entries.push(entry);
+      const entry = entries[0];
+      if (entry === undefined) throw new Error("Missing iNES region");
+      if (phase === "stat") {
+        const spy = vi
+          .spyOn(handle, "stat")
+          .mockImplementationOnce(async () => {
+            controller.abort("caller stopped");
+            throw physical;
+          });
+        restore = () => spy.mockRestore();
+        await expect(reader.open(entry, controller.signal)).rejects.toBe(
+          physical,
+        );
+      } else {
+        const stream = await reader.open(entry, controller.signal);
+        const spy = vi
+          .spyOn(handle, "read")
+          .mockImplementationOnce(async () => {
+            controller.abort("caller stopped");
+            throw physical;
+          });
+        restore = () => spy.mockRestore();
+        await expect(
+          (async () => {
+            for await (const _ of stream) {
+              /* consume */
+            }
+          })(),
+        ).rejects.toMatchObject({ reason: "io", cause: physical });
+      }
+    } finally {
+      restore();
+      await reader.close();
+      await handle.close();
+    }
+  },
+);
+
+it("preserves the caller reason when a completed iNES read observes cancellation", async () => {
+  const root = await createTestTempDirectory("rea-ines-stream-cancel-");
+  const path = join(root, "source.nes");
+  await writeFile(path, inesFixture());
+  const handle = await open(path, "r");
+  const reader = new InesArtifactReader(path, {
+    handle,
+    initial: await handle.stat(),
+  });
+  const controller = new AbortController();
+  const reason = { source: "caller" };
+  let restore = () => {};
+  try {
+    const entries = [];
+    for await (const entry of reader.entries()) entries.push(entry);
+    const entry = entries[0];
+    if (entry === undefined) throw new Error("Missing iNES region");
+    const stream = await reader.open(entry, controller.signal);
+    const original = handle.read.bind(handle);
+    const spy = vi
+      .spyOn(handle, "read")
+      .mockImplementationOnce(async (...args) => {
+        const result = await original(...args);
+        controller.abort(reason);
+        return result;
+      });
+    restore = () => spy.mockRestore();
+    await expect(
+      (async () => {
+        for await (const _ of stream) {
+          /* consume */
+        }
+      })(),
+    ).rejects.toMatchObject({ reason: "cancelled", cause: reason });
+  } finally {
+    restore();
+    await reader.close();
+    await handle.close();
+  }
+});
+
+it.each(["classification", "hash", "region"] as const)(
+  "preserves physical failure and concurrent cancellation through %s inventory",
+  async (phase) => {
+    const root = await createTestTempDirectory("rea-ines-inventory-io-");
+    const path = join(root, "source.nes");
+    await writeFile(path, inesFixture());
+    const sample = await open(path, "r");
+    const prototype = Object.getPrototypeOf(sample) as FileHandle;
+    await sample.close();
+    const originalRead = prototype.read;
+    const originalStat = prototype.stat;
+    const controller = new AbortController();
+    const physical = Object.assign(new Error("inventory descriptor failure"), {
+      code: "EIO",
+    });
+    let afterRead = false;
+    let statCount = 0;
+    const read = vi.spyOn(prototype, "read").mockImplementation(async function (
+      this: FileHandle,
+      ...args
+    ) {
+      const arguments_: readonly unknown[] = args;
+      const position = arguments_[3];
+      if (
+        phase === "region" &&
+        afterRead &&
+        (typeof position === "number" || typeof position === "bigint") &&
+        Number(position) === 16
+      ) {
+        controller.abort("caller stopped");
+        throw physical;
+      }
+      const value = await originalRead.call(this, ...args);
+      afterRead = true;
+      return value;
+    });
+    const stat = vi.spyOn(prototype, "stat").mockImplementation(async function (
+      this: FileHandle,
+      ...args
+    ) {
+      if (
+        afterRead &&
+        ++statCount ===
+          (phase === "classification" ? 2 : phase === "hash" ? 3 : -1)
+      ) {
+        controller.abort("caller stopped");
+        throw physical;
+      }
+      return originalStat.call(this, ...args);
+    });
+    try {
+      const failure = await inventoryArtifact(path, {
+        signal: controller.signal,
+      }).catch((cause: unknown) => cause);
+      expect(controller.signal.aborted).toBe(true);
+      expect(failure).not.toMatchObject({ reason: "cancelled" });
+      if (phase === "region")
+        expect(failure).toMatchObject({ reason: "io", cause: physical });
+      else expect(failure).toBe(physical);
+    } finally {
+      read.mockRestore();
+      stat.mockRestore();
+    }
+  },
+);