node:stream: Readable.fromWeb emits an AbortError when the web stream fails with a falsy value

src/js/internal/streams/native-readable.ts

@@ -135,6 +135,11 @@ function getRemainingChunk(stream: NativeReadable, maxToRead?: number) {
   return chunk;
 }
 
+// destroy(falsy) emits no 'error'. Node's adapter makes it an AbortError (destroyer()).
+function failPull(stream: NativeReadable, reason: unknown) {
+  errorOrDestroy(stream, reason || $makeAbortError());
+}
+
 function read(this: NativeReadable, maxToRead: number) {
   $debug(`[${this.debugId}] read${this[kPendingRead] ? ", is already pending" : ""}`);
   var ptr = this.$bunNativePtr;
@@ -168,7 +173,12 @@ function read(this: NativeReadable, maxToRead: number) {
     }
   }
   const chunk = getRemainingChunk(this, maxToRead);
-  var result = ptr.pull(chunk, this[kCloseState]);
+  var result;
+  try {
+    result = ptr.pull(chunk, this[kCloseState]);
+  } catch (error) {
+    return failPull(this, error);
+  }
   $assert(result !== undefined);
   $debug(
     `[${this.debugId}] pull ${chunk?.byteLength} bytes, result: ${$isPromise(result) ? "<pending>" : $isTypedArrayView(result) ? `<${result.byteLength} bytes>` : result}, closeState: ${this[kCloseState][0]}`,
@@ -183,9 +193,7 @@ function read(this: NativeReadable, maxToRead: number) {
         this[kPendingRead] = false;
         this[kRemainingChunk] = handleResult(this, result, chunk, this[kCloseState][0]);
       },
-      reason => {
-        errorOrDestroy(this, reason);
-      },
+      reason => failPull(this, reason),
     );
   } else {
     this[kRemainingChunk] = handleResult(this, result, chunk, this[kCloseState][0]);

src/js/internal/webstreams_adapters.ts

@@ -82,7 +82,7 @@ class ReadableFromWeb extends Readable {
       } catch {}
     }
     this.#closed = true;
-    this.destroy(error);
+    destroyer(this, error);
   }
 
   // One reader.read() per _read(). readMany() would drain a start()-enqueued

test/js/node/stream/node-stream.test.js

@@ -476,6 +476,56 @@ it("Readable.fromWeb on an already-errored web stream emits 'error' and destroys
   expect(r.errored?.message).toBe("start-boom");
 });
 
+// destroy(null) emits 'close' alone, so a node stream cannot carry a falsy error. Node's adapter
+// destroys through destroyer(), which makes an AbortError of it while the stream is unfinished.
+it.each([[null], [undefined], [0], [""]])(
+  "Readable.fromWeb on a web stream errored with %p emits an AbortError",
+  async reason => {
+    const web = new ReadableStream({
+      start(c) {
+        c.error(reason);
+      },
+    });
+    const r = Readable.fromWeb(web);
+    const events = [];
+    const { promise, resolve } = Promise.withResolvers();
+    r.on("error", error => events.push(["error", error.name, error.code]));
+    r.on("end", () => events.push(["end"]));
+    r.on("close", resolve);
+    r.resume();
+    await promise;
+    expect(events).toEqual([["error", "AbortError", "ABORT_ERR"]]);
+    expect(r.destroyed).toBe(true);
+  },
+);
+
+// An HTMLRewriter output is a native stream, which Readable.fromWeb pulls from directly. If the
+// input failed before the first pull, pull() throws the stored value. If it fails later, the pull rejects.
+it.each([[null], [undefined], [0], [""]])(
+  "Readable.fromWeb on a native stream that failed with %p emits an AbortError",
+  async reason => {
+    let controller;
+    const input = new ReadableStream({
+      start(c) {
+        controller = c;
+      },
+    });
+    const body = new HTMLRewriter().transform(new Response(input)).body;
+    controller.error(reason);
+    await new Promise(resolve => setImmediate(resolve));
+    const r = Readable.fromWeb(body);
+    const events = [];
+    const { promise, resolve } = Promise.withResolvers();
+    r.on("error", error => events.push(["error", error.name, error.code]));
+    r.on("end", () => events.push(["end"]));
+    r.on("close", resolve);
+    r.resume();
+    await promise;
+    expect(events).toEqual([["error", "AbortError", "ABORT_ERR"]]);
+    expect(r.destroyed).toBe(true);
+  },
+);
+
 // Delivering a 64 KiB file chunk re-enters the native reader: push() over the
 // highWaterMark pauses it, and the next _read unpauses it mid-delivery. On
 // Windows that used to free the buffer an in-flight libuv file read was still

test/js/web/fetch/fetch-abort-stream-body.test.ts

@@ -389,6 +389,32 @@ describe("aborting mid-body fails every reader of res.body with signal.reason",
     },
   );
 
+  // A node stream cannot carry a falsy error, so a falsy reason arrives as an AbortError, as in Node.
+  describe.each(["while it waits for more", "before Readable.fromWeb()"])("Readable.fromWeb(res.body), %s", timing => {
+    test.concurrent.each([[null], [0], [""]])("abort(%p) emits an AbortError", async reason => {
+      using server = await stalledBodyServer();
+      const controller = new AbortController();
+      const res = await fetch(server.url, { signal: controller.signal });
+      expect(res.body).toBeInstanceOf(ReadableStream);
+      if (timing === "before Readable.fromWeb()") controller.abort(reason);
+      const readable = Readable.fromWeb(res.body as any);
+      const settled = new Promise(resolve =>
+        readable
+          .on("error", (error: any) => resolve({ name: error.name, code: error.code }))
+          .on("end", () => resolve("ended"))
+          .on("close", () => resolve("closed")),
+      );
+      if (timing === "while it waits for more") {
+        await once(readable, "data");
+        await new Promise(resolve => setImmediate(resolve));
+        controller.abort(reason);
+      } else {
+        readable.resume();
+      }
+      expect(await settled).toEqual({ name: "AbortError", code: "ABORT_ERR" });
+    });
+  });
+
   // Nothing waits, so nothing can take the reason, and to keep it natively would root it. The
   // source still has to fail: main ended such a stream as if the body were complete.
   test.concurrent("Readable.fromWeb(res.body) that has not read yet fails with an AbortError", async () => {