Uncatchable exception when using TransformStream with requests.

[piscisaureus]

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.

// 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.

[piscisaureus]

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:

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.

[kt3k]

The uncaught rejection seems happening at this line

deno/ext/web/06_streams.js

Line 2546 in b5c3f4f

closedPromise.reject(e);

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.