fix(ext/http): surface errors thrown while streaming a Deno.serve body (#35104)

## Summary

A `Deno.serve` response whose body is a `ReadableStream` /
`TransformStream`
that throws **while being drained** swallowed the error completely. The
error
was flattened to a bare message that was only used to abort the
underlying
transport, so:

- the client saw a truncated response, and
- **nothing on the server implicated the faulty callback**.

The motivating example from #19867 — a typo'd `value.toUppperCase()`
inside a
transformer — failed silently with no stack trace anywhere.

## What happens now

```
TypeError: value.toUppperCase is not a function
    at Object.transform (file:///.../main.ts:27:31)
    at Object.invokeCallbackFunction (ext:deno_webidl/00_webidl.js:1154:16)
    at transformAlgorithm (ext:deno_web/06_streams.js:3998:14)
    ...
```

The stack trace pointing at the throwing transformer is printed, and the
server stays alive for subsequent requests.

## How

- `resourceForReadableStream` (`ext/web/06_streams.js`) gains an
optional
`onError` callback. When the stream errors while being drained into the
resource, the callback is invoked with the **original error object**
(stack
included) *before* it is flattened to a string for the Rust transport.
Other
  callers (`fetch`, `cache`) pass no callback and are unaffected.
- `Deno.serve` (`ext/http/00_serve.ts`) passes a callback that routes
the error
through the server's error handler (`onError`). The default handler logs
a
stack trace; a user-provided `onError` can observe it for
logging/metrics.

### Note on returning 500

The issue also asks whether a 500 should be returned. By the time the
body
stream starts producing chunks the response status and headers have
already
been committed to the wire, so the response can no longer be changed and
the
value returned from `onError` cannot be used. The error is therefore
surfaced
for observability only; the connection is still aborted (so the client
sees a
truncated body) and the server keeps serving. Returning a 500 would only
be
possible by buffering the entire streamed body before sending headers,
which
would defeat streaming.

## Test

`tests/specs/serve/stream_body_error` — a serve handler whose streaming
body
throws in a `TransformStream` transformer must print a stack trace
implicating
the transformer and keep the server alive for a follow-up request.

Fixes #19867

Closes denoland/divybot#552

Co-authored-by: divybot <divybot@users.noreply.github.com>
Co-authored-by: Divy Srivastava <me@littledivy.com>

Author: divybot

ext/http/00_serve.ts

@@ -52,6 +52,7 @@ const {
   ArrayPrototypePush,
   ObjectHasOwn,
   ObjectPrototypeIsPrototypeOf,
+  PromisePrototype,
   PromisePrototypeCatch,
   PromiseResolve,
   SafeArrayIterator,
@@ -562,12 +563,37 @@ function trySetServeFastStaticResponse(
   }
 }
 
+// Report an error that was thrown while a streaming response body was being
+// drained. The response status/headers are already on the wire at this point,
+// so the value returned from `onError` can no longer be used; we route the
+// error through the handler purely so it is observed (the default handler logs
+// a stack trace) instead of being silently swallowed.
+function reportResponseStreamError(onError, error) {
+  let result;
+  try {
+    result = onError(error);
+  } catch (e) {
+    internals.log("error", "Exception in onError while handling exception", e);
+    return;
+  }
+  if (ObjectPrototypeIsPrototypeOf(PromisePrototype, result)) {
+    PromisePrototypeThen(result, undefined, (e) => {
+      internals.log(
+        "error",
+        "Exception in onError while handling exception",
+        e,
+      );
+    });
+  }
+}
+
 function fastSyncResponseOrStream(
   req,
   respBody,
   status,
   innerRequest: InnerRequest,
   headers,
+  onError,
 ) {
   if (respBody === null || respBody === undefined) {
     // Don't set the body
@@ -661,7 +687,13 @@ function fastSyncResponseOrStream(
     rid = resourceBacking.rid;
     autoClose = resourceBacking.autoClose;
   } else {
-    rid = resourceForReadableStream(stream);
+    // The response headers/status have already been committed by the time the
+    // body stream starts producing chunks, so an error thrown while draining
+    // the stream (e.g. inside a `TransformStream` transformer) can no longer
+    // change the response. Report it through the server's error handler so it
+    // is not silently swallowed and a stack trace implicating the faulty
+    // callback is surfaced. See https://github.com/denoland/deno/issues/19867.
+    rid = resourceForReadableStream(stream, undefined, onError);
     autoClose = true;
   }
   PromisePrototypeThen(
@@ -830,6 +862,7 @@ function mapToCallback(context, callback, onError) {
       status,
       innerRequest,
       headers,
+      (error) => reportResponseStreamError(onError, error),
     );
   };
 
@@ -995,6 +1028,7 @@ function mapToNativeResponseCallback(context, callback, onError) {
       inner.status,
       innerRequest,
       inner.headerList,
+      (error) => reportResponseStreamError(onError, error),
     );
     return undefined;
   }

ext/web/06_streams.js

@@ -900,8 +900,9 @@ async function readableStreamWriteChunkFn(reader, sink, chunk) {
 /**
  * @param {ReadableStreamDefaultReader<Uint8Array>} reader
  * @param {any} sink
+ * @param {((error: unknown) => void) | undefined} onError
  */
-async function readableStreamReadFn(reader, sink) {
+async function readableStreamReadFn(reader, sink, onError) {
   let loop = true;
 
   while (loop) {
@@ -929,6 +930,16 @@ async function readableStreamReadFn(reader, sink) {
         promise.resolve(false);
       },
       errorSteps(error) {
+        // Surface the original error (with its stack) to the owner of the
+        // resource before we flatten it to a string for the Rust side. Without
+        // this hook the error is otherwise swallowed: it is converted to a
+        // plain message, pushed into the channel and used only to abort the
+        // underlying transport, so e.g. a throwing `TransformStream` feeding a
+        // `Deno.serve` response body would fail with nothing implicating the
+        // faulty callback. See https://github.com/denoland/deno/issues/19867.
+        if (onError !== undefined) {
+          onError(error);
+        }
         const success = op_readable_stream_resource_write_error(
           sink.external,
           extractStringErrorFromError(error),
@@ -969,9 +980,13 @@ async function readableStreamReadFn(reader, sink) {
  * ReadableStream source.
  * @param {ReadableStream<Uint8Array>} stream
  * @param {number | undefined} length
+ * @param {((error: unknown) => void) | undefined} onError Invoked with the
+ *   original error (including its stack) if the stream errors while being
+ *   drained into the resource. Lets the owner report an otherwise swallowed
+ *   error before it is flattened to a string for the transport.
  * @returns {number}
  */
-function resourceForReadableStream(stream, length) {
+function resourceForReadableStream(stream, length, onError) {
   const reader = acquireReadableStreamDefaultReader(stream);
 
   // Allocate the resource
@@ -996,7 +1011,7 @@ function resourceForReadableStream(stream, length) {
   );
 
   // Trigger the first read
-  PromisePrototypeCatch(readableStreamReadFn(reader, sink), (err) => {
+  PromisePrototypeCatch(readableStreamReadFn(reader, sink, onError), (err) => {
     PromisePrototypeCatch(reader.cancel(err), () => {});
   });
 

tests/specs/serve/stream_body_error/__test__.jsonc

@@ -0,0 +1,4 @@
+{
+  "args": "run -A main.ts",
+  "output": "main.out"
+}

tests/specs/serve/stream_body_error/main.out

@@ -0,0 +1,4 @@
+[WILDCARD]TypeError: value.toUppperCase is not a function
+    at [WILDCARD]transform[WILDLINE]
+[WILDCARD]second status: 200 - second request ok
+server shut down cleanly

tests/specs/serve/stream_body_error/main.ts

@@ -0,0 +1,59 @@
+// Regression test for https://github.com/denoland/deno/issues/19867
+//
+// A `Deno.serve` response whose body is a `ReadableStream`/`TransformStream`
+// that throws while being drained used to swallow the error completely: the
+// client saw a truncated response and nothing on the server implicated the
+// faulty callback. The error must now be surfaced through the server's error
+// handler so a stack trace pointing at the throwing transformer is printed,
+// and the server must stay alive for subsequent requests.
+
+const ac = new AbortController();
+const server = Deno.serve(
+  { port: 0, signal: ac.signal, onListen() {} },
+  (req) => {
+    if (new URL(req.url).pathname === "/ok") {
+      return new Response("second request ok");
+    }
+    return new Response(
+      new ReadableStream({
+        start(c) {
+          c.enqueue(new TextEncoder().encode("test"));
+          c.close();
+        },
+      })
+        .pipeThrough(new TextDecoderStream())
+        .pipeThrough(
+          new TransformStream({
+            transform(value, c) {
+              // Intentional typo (`toUppperCase`) so the transformer throws.
+              // @ts-ignore intentional typo
+              c.enqueue(value.toUppperCase());
+            },
+          }),
+        )
+        .pipeThrough(new TextEncoderStream()),
+    );
+  },
+);
+
+const port = (server.addr as Deno.NetAddr).port;
+
+const res = await fetch(`http://localhost:${port}/`);
+console.log("status:", res.status);
+try {
+  await res.text();
+  console.log("body read: ok");
+} catch {
+  console.log("body read: errored");
+}
+
+// Give the streaming error a moment to be reported on the server side.
+await new Promise((r) => setTimeout(r, 100));
+
+// The server must still be alive to serve a second request.
+const res2 = await fetch(`http://localhost:${port}/ok`);
+console.log("second status:", res2.status, "-", await res2.text());
+
+ac.abort();
+await server.finished;
+console.log("server shut down cleanly");