Repository: denoland/deno (language: rust)

Issue:
Uncatchable exception when using `TransformStream` with requests.
The code below simulates a proxy server that makes a fetch() call, transforms the body, and then returns the transformed body.
When the client cancels a request, the server crashes with an uncaught exception that cannot be caught in any way.

```js
// Response transforming server that crashes with an uncaught AbortError.
function startServer() {
  Deno.serve({ port: 8000 }, async (req) => {
    const upstreamResponse = await fetch("http://localhost:8001", req);

    // Use a TransformStream to convert the response body to uppercase.
    const transformStream = new TransformStream({
      transform(chunk, controller) {
        const decoder = new TextDecoder();
        const encoder = new TextEncoder();
        const chunk2 = encoder.encode(decoder.decode(chunk).toUpperCase());
        controller.enqueue(chunk2);
      },

      cancel(reason) {
        console.log("server: TransformStream cancelled:", reason);
      },
    });

    upstreamResponse.body?.pipeTo(transformStream.writable).catch((err) => {
      console.error("server: pipeTo() failed", err);
    });

    return new Response(transformStream.readable);
  });
}

// ==== THE ISSUE IS NOT IN THE CODE BELOW ====

// Upstream server that sends a response with a body that never ends.
// This is not where the error happens (it handlers the cancellation correctly).
function startUpstreamServer() {
  Deno.serve({ port: 8001 }, (req) => {
    // Create an infinite readable stream that emits 'a'
    let pushTimeout: number | null = null;
    const readableStream = new ReadableStream({
      start(controller) {
        const encoder = new TextEncoder();
        const chunk = encoder.encode("a");

        function push() {
          controller.enqueue(chunk);
          pushTimeout = setTimeout(push, 100);
        }

        push();
      },

      cancel(reason) {
        console.log("upstream_server: ReadableStream cancelled:", reason);
        clearTimeout(pushTimeout!);
      },
    });

    return new Response(readableStream, {
      headers: { "Content-Type": "text/plain" },
    });
  });
}

// The client is just there to simulate a client that cancels a request.
async function startClient() {
  const controller = new AbortController();
  const signal = controller.signal;

  try {
    const response = await fetch("http://localhost:8000", { signal });
    const reader = response.body?.getReader();
    if (!reader) throw new Error("client: failed to get reader from response");

    let received = 0;
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      received += value.length;
      console.log(`client: received ${received} bytes`);

      if (received >= 5) {
        console.log("client: aborting request after receiving 5 bytes");
        controller.abort();
        break;
      }
    }
  } catch (err) {
    if (err.name === "AbortError") {
      console.log("client: Request aborted as expected.");
    } else {
      console.error("client: an error occurred:", err);
    }
  }
}

startUpstreamServer();
startServer();
startClient();
```

Output:
```
Listening on http://0.0.0.0:8001/
Listening on http://0.0.0.0:8000/
client: received 1 bytes
client: received 2 bytes
client: received 3 bytes
client: received 4 bytes
client: received 5 bytes
client: aborting request after receiving 5 bytes
server: TransformStream cancelled: DOMException {
  message: "The request has been cancelled.",
  name: "AbortError",
  code: 20
}
server: pipeTo() failed DOMException {
  message: "The request has been cancelled.",
  name: "AbortError",
  code: 20
}
error: Uncaught (in promise) AbortError: The request has been cancelled.
```


Discussion / hints:
It seems that the issue can be worked around by passing the `{ preventAbort: true }` option to `pipeTo()`/`pipeThrough()`. However, this shouldn't be necessary.

Test code:
```js
function startServer() {
  Deno.serve({ port: 8000 }, async (req) => {
    const upstreamResponse = await fetch("http://localhost:8001/", req);

    // Use a TransformStream to convert the response body to uppercase.
    const transformStream = new TransformStream({
      transform(chunk, controller) {
        const decoder = new TextDecoder();
        const encoder = new TextEncoder();
        const chunk2 = encoder.encode(decoder.decode(chunk).toUpperCase());
        controller.enqueue(chunk2);
      },

      cancel(reason) {
        console.log("server: TransformStream cancelled:", reason);
      },
    });

    const readable = upstreamResponse.body?.pipeThrough(transformStream, {
      preventAbort: true,
    });
    return new Response(readable);
  });
}

// Code for upstream server and client omitted for brevity.
// They're the same as the code in the issue description.
```
The uncaught rejection seems happening at this line https://github.com/denoland/deno/blob/b5c3f4f782d3cdc1a7c51b03f597799e1c9c773b/ext/web/06_streams.js#L2546

This rejects `reader.closed` promise, but in this construction of streams, `reader` object is implicitly created in `pipeTo` (or `pipeThrough`) call. I'm not sure there's a way to get access to that promise when we use `pipeTo` function.

Fix patch:
diff --git a/ext/web/06_streams.js b/ext/web/06_streams.js
index 2ad350c5531e90..c9fab6364a9dea 100644
--- a/ext/web/06_streams.js
+++ b/ext/web/06_streams.js
@@ -779,37 +779,35 @@ class ResourceStreamResourceSink {
  * @param {any} sink
  * @param {Uint8Array} chunk
  */
-function readableStreamWriteChunkFn(reader, sink, chunk) {
+async function readableStreamWriteChunkFn(reader, sink, chunk) {
   // Empty chunk. Re-read.
   if (chunk.length == 0) {
-    readableStreamReadFn(reader, sink);
+    await readableStreamReadFn(reader, sink);
     return;
   }
 
   const res = op_readable_stream_resource_write_sync(sink.external, chunk);
   if (res == 0) {
     // Closed
-    reader.cancel("resource closed");
+    await reader.cancel("resource closed");
     sink.close();
   } else if (res == 1) {
     // Successfully written (synchronous). Re-read.
-    readableStreamReadFn(reader, sink);
+    await readableStreamReadFn(reader, sink);
   } else if (res == 2) {
     // Full. If the channel is full, we perform an async await until we can write, and then return
     // to a synchronous loop.
-    (async () => {
-      if (
-        await op_readable_stream_resource_write_buf(
-          sink.external,
-          chunk,
-        )
-      ) {
-        readableStreamReadFn(reader, sink);
-      } else {
-        reader.cancel("resource closed");
-        sink.close();
-      }
-    })();
+    if (
+      await op_readable_stream_resource_write_buf(
+        sink.external,
+        chunk,
+      )
+    ) {
+      await readableStreamReadFn(reader, sink);
+    } else {
+      await reader.cancel("resource closed");
+      sink.close();
+    }
   }
 }
 
@@ -822,17 +820,23 @@ function readableStreamReadFn(reader, sink) {
   // real resource.
   let reentrant = true;
   let gotChunk = undefined;
+  const promise = new Deferred();
   readableStreamDefaultReaderRead(reader, {
     chunkSteps(chunk) {
       // If the chunk has non-zero length, write it
       if (reentrant) {
         gotChunk = chunk;
       } else {
-        readableStreamWriteChunkFn(reader, sink, chunk);
+        PromisePrototypeThen(
+          readableStreamWriteChunkFn(reader, sink, chunk),
+          () => promise.resolve(),
+          (e) => promise.reject(e),
+        );
       }
     },
     closeSteps() {
       sink.close();
+      promise.resolve();
     },
     errorSteps(error) {
       const success = op_readable_stream_resource_write_error(
@@ -842,15 +846,29 @@ function readableStreamReadFn(reader, sink) {
       // We don't cancel the reader if there was an error reading. We'll let the downstream
       // consumer close the resource after it receives the error.
       if (!success) {
-        reader.cancel("resource closed");
+        PromisePrototypeThen(
+          reader.cancel("resource closed"),
+          () => {
+            sink.close();
+            promise.resolve();
+          },
+          (e) => promise.reject(e),
+        );
+      } else {
+        sink.close();
+        promise.resolve();
       }
-      sink.close();
     },
   });
   reentrant = false;
   if (gotChunk) {
-    readableStreamWriteChunkFn(reader, sink, gotChunk);
+    PromisePrototypeThen(
+      readableStreamWriteChunkFn(reader, sink, gotChunk),
+      () => promise.resolve(),
+      (e) => promise.reject(e),
+    );
   }
+  return promise.promise;
 }
 
 /**
@@ -873,7 +891,9 @@ function resourceForReadableStream(stream, length) {
   PromisePrototypeCatch(
     PromisePrototypeThen(
       op_readable_stream_resource_await_close(rid),
-      () => reader.cancel("resource closed"),
+      () => {
+        PromisePrototypeCatch(reader.cancel("resource closed"), () => {});
+      },
     ),
     () => {},
   );
@@ -884,7 +904,9 @@ function resourceForReadableStream(stream, length) {
   );
 
   // Trigger the first read
-  readableStreamReadFn(reader, sink);
+  PromisePrototypeCatch(readableStreamReadFn(reader, sink), (err) => {
+    PromisePrototypeCatch(reader.cancel(err), () => {});
+  });
 
   return rid;
 }


Test patch:
diff --git a/tests/unit/streams_test.ts b/tests/unit/streams_test.ts
index 84a87d166d68ca..d0a457ddc4fea5 100644
--- a/tests/unit/streams_test.ts
+++ b/tests/unit/streams_test.ts
@@ -547,3 +547,107 @@ Deno.test(function readableStreamFromWithStringThrows() {
     "Failed to execute 'ReadableStream.from': Argument 1 can not be converted to async iterable.",
   );
 });
+
+Deno.test(async function readableStreamFromWithStringThrows() {
+  const serverPort = 4592;
+  const upstreamServerPort = 4593;
+
+  const stopSignal = new AbortController();
+  const promise = Promise.withResolvers();
+  // Response transforming server that crashes with an uncaught AbortError.
+  function startServer() {
+    Deno.serve({ port: serverPort, signal: stopSignal.signal }, async (req) => {
+      const upstreamResponse = await fetch(
+        `http://localhost:${upstreamServerPort}`,
+        req,
+      );
+
+      // Use a TransformStream to convert the response body to uppercase.
+      const transformStream = new TransformStream({
+        transform(chunk, controller) {
+          const decoder = new TextDecoder();
+          const encoder = new TextEncoder();
+          const chunk2 = encoder.encode(decoder.decode(chunk).toUpperCase());
+          controller.enqueue(chunk2);
+        },
+      });
+
+      upstreamResponse.body?.pipeTo(transformStream.writable).catch(() => {});
+
+      return new Response(transformStream.readable);
+    });
+  }
+
+  // ==== THE ISSUE IS NOT IN THE CODE BELOW ====
+
+  // Upstream server that sends a response with a body that never ends.
+  // This is not where the error happens (it handlers the cancellation correctly).
+  function startUpstreamServer() {
+    Deno.serve({ port: upstreamServerPort, signal: stopSignal.signal }, (_) => {
+      // Create an infinite readable stream that emits 'a'
+      let pushTimeout: number | null = null;
+      const readableStream = new ReadableStream({
+        start(controller) {
+          const encoder = new TextEncoder();
+          const chunk = encoder.encode("a");
+
+          function push() {
+            controller.enqueue(chunk);
+            pushTimeout = setTimeout(push, 100);
+          }
+
+          push();
+        },
+
+        cancel(reason) {
+          assertEquals(reason, "resource closed");
+          promise.resolve(undefined);
+          clearTimeout(pushTimeout!);
+        },
+      });
+
+      return new Response(readableStream, {
+        headers: { "Content-Type": "text/plain" },
+      });
+    });
+  }
+
+  // The client is just there to simulate a client that cancels a request.
+  async function startClient() {
+    const controller = new AbortController();
+    const signal = controller.signal;
+
+    try {
+      const response = await fetch(`http://localhost:${serverPort}`, {
+        signal,
+      });
+      const reader = response.body?.getReader();
+      if (!reader) {
+        throw new Error("client: failed to get reader from response");
+      }
+
+      let received = 0;
+      while (true) {
+        const { done, value } = await reader.read();
+        if (done) break;
+
+        received += value.length;
+
+        if (received >= 5) {
+          controller.abort();
+          break;
+        }
+      }
+    } catch (_) {
+      //
+    }
+  }
+
+  startUpstreamServer();
+  startServer();
+  const p = startClient();
+
+  await promise.promise;
+  stopSignal.abort();
+  await p;
+});