fix(ext/flash): Correctly handle errors for chunked responses (#17303)

The leading cause of the problem was that `handleResponse` has
`tryRespondChunked` passed as an argument, which in turn is implemented
as a call to `core.ops.op_try_flash_respond_chuncked`, that throws in
the repro code.

`handleResponse` was not handled correctly, as it not returned any
value, and had no `catch` attached to it.
It also effectively was never correctly handled inside two other blocks
with `resp.then` and `PromisePrototypeCatch(PromisePrototypeThen(resp,
"..."))` as well, as it just short-circuited the promise with an empty
resolve, instead of relying on the last `(async () => {})` block.

This change makes `handleResponse` return a correct value and attach
`onError` handler to the "non-thenable" variant of response handling
code.

cli/tests/unit/flash_test.ts

@@ -12,6 +12,7 @@ import {
   assert,
   assertEquals,
   assertRejects,
+  assertStringIncludes,
   assertThrows,
   Deferred,
   deferred,
@@ -495,6 +496,51 @@ Deno.test(
   },
 );
 
+// https://github.com/denoland/deno/issues/17291
+Deno.test(
+  { permissions: { net: true } },
+  async function httpServerIncorrectChunkedResponse() {
+    const ac = new AbortController();
+    const listeningPromise = deferred();
+    const errorPromise = deferred();
+    const server = Deno.serve({
+      handler: () => {
+        const body = new ReadableStream({
+          start(controller) {
+            // Non-encoded string is not a valid readable chunk.
+            controller.enqueue("wat");
+          },
+        });
+        return new Response(body);
+      },
+      port: 4501,
+      signal: ac.signal,
+      onListen: onListen(listeningPromise),
+      onError: (err) => {
+        const errResp = new Response(
+          `Internal server error: ${(err as Error).message}`,
+          { status: 500 },
+        );
+        ac.abort();
+        errorPromise.resolve(errResp);
+        return errResp;
+      },
+    });
+
+    await listeningPromise;
+
+    const resp = await fetch("http://127.0.0.1:4501/");
+    // Incorrectly implemented reader ReadableStream should reject.
+    await assertRejects(() => resp.body!.getReader().read());
+
+    const err = await errorPromise as Response;
+    assertStringIncludes(await err.text(), "Expected ArrayBufferView");
+
+    ac.abort();
+    await server;
+  },
+);
+
 Deno.test(
   { permissions: { net: true } },
   async function httpServerCorrectLengthForUnicodeString() {

ext/flash/01_http.js

@@ -327,7 +327,7 @@
       );
     }
 
-    (async () => {
+    return (async () => {
       if (!ws) {
         if (hasBody && body[_state] !== "closed") {
           // TODO(@littledivy): Optimize by draining in a single op.
@@ -590,7 +590,6 @@
                     ),
                     onError,
                   );
-                  continue;
                 } else if (typeof resp?.then === "function") {
                   resp.then((resp) =>
                     handleResponse(
@@ -606,24 +605,23 @@
                       tryRespondChunked,
                     )
                   ).catch(onError);
-                  continue;
+                } else {
+                  handleResponse(
+                    req,
+                    resp,
+                    body,
+                    hasBody,
+                    method,
+                    serverId,
+                    i,
+                    respondFast,
+                    respondChunked,
+                    tryRespondChunked,
+                  ).catch(onError);
                 }
               } catch (e) {
                 resp = await onError(e);
               }
-
-              handleResponse(
-                req,
-                resp,
-                body,
-                hasBody,
-                method,
-                serverId,
-                i,
-                respondFast,
-                respondChunked,
-                tryRespondChunked,
-              );
             }
 
             offset += tokens;