fix(ext/http): fix crash in dropped Deno.serve requests (#21252)

Fixes #21250

We were attempting to recycle dropped resource responses too early.

cli/tests/unit/serve_test.ts

@@ -2727,73 +2727,81 @@ Deno.test(
   },
 );
 
-for (const url of ["text", "file", "stream"]) {
+for (const delay of ["delay", "nodelay"]) {
-  // Ensure that we don't panic when the incoming TCP request was dropped
+  for (const url of ["text", "file", "stream"]) {
-  // https://github.com/denoland/deno/issues/20315 and that we correctly
+    // Ensure that we don't panic when the incoming TCP request was dropped
-  // close/cancel the response
+    // https://github.com/denoland/deno/issues/20315 and that we correctly
-  Deno.test({
+    // close/cancel the response
-    permissions: { read: true, write: true, net: true },
+    Deno.test({
-    name: `httpServerTcpCancellation_${url}`,
+      permissions: { read: true, write: true, net: true },
-    fn: async function () {
+      name: `httpServerTcpCancellation_${url}_${delay}`,
-      const ac = new AbortController();
+      fn: async function () {
-      const streamCancelled = url == "stream" ? deferred() : undefined;
+        const ac = new AbortController();
-      const listeningPromise = deferred();
+        const streamCancelled = url == "stream" ? deferred() : undefined;
-      const waitForAbort = deferred();
+        const listeningPromise = deferred();
-      const waitForRequest = deferred();
+        const waitForAbort = deferred();
-      const server = Deno.serve({
+        const waitForRequest = deferred();
-        port: servePort,
+        const server = Deno.serve({
-        signal: ac.signal,
+          port: servePort,
-        onListen: onListen(listeningPromise),
+          signal: ac.signal,
-        handler: async (req: Request) => {
+          onListen: onListen(listeningPromise),
-          let respBody = null;
+          handler: async (req: Request) => {
-          if (req.url.includes("/text")) {
+            let respBody = null;
-            respBody = "text";
+            if (req.url.includes("/text")) {
-          } else if (req.url.includes("/file")) {
+              respBody = "text";
-            respBody = (await makeTempFile(1024)).readable;
+            } else if (req.url.includes("/file")) {
-          } else if (req.url.includes("/stream")) {
+              respBody = (await makeTempFile(1024)).readable;
-            respBody = new ReadableStream({
+            } else if (req.url.includes("/stream")) {
-              start(controller) {
+              respBody = new ReadableStream({
-                controller.enqueue(new Uint8Array([1]));
+                start(controller) {
-              },
+                  controller.enqueue(new Uint8Array([1]));
-              cancel(reason) {
+                },
-                streamCancelled!.resolve(reason);
+                cancel(reason) {
-              },
+                  streamCancelled!.resolve(reason);
-            });
+                },
-          } else {
+              });
-            fail();
+            } else {
-          }
+              fail();
-          waitForRequest.resolve();
+            }
-          await waitForAbort;
+            waitForRequest.resolve();
-          // Allocate the request body
+            await waitForAbort;
-          req.body;
-          return new Response(respBody);
-        },
-      });
 
-      await listeningPromise;
+            if (delay == "delay") {
+              await new Promise((r) => setTimeout(r, 1000));
+            }
+            // Allocate the request body
+            req.body;
+            return new Response(respBody);
+          },
+        });
 
-      // Create a POST request and drop it once the server has received it
+        await listeningPromise;
-      const conn = await Deno.connect({ port: servePort });
-      const writer = conn.writable.getWriter();
-      await writer.write(new TextEncoder().encode(`POST /${url} HTTP/1.0\n\n`));
-      await waitForRequest;
-      await writer.close();
 
-      waitForAbort.resolve();
+        // Create a POST request and drop it once the server has received it
+        const conn = await Deno.connect({ port: servePort });
+        const writer = conn.writable.getWriter();
+        await writer.write(
+          new TextEncoder().encode(`POST /${url} HTTP/1.0\n\n`),
+        );
+        await waitForRequest;
+        await writer.close();
 
-      // Wait for cancellation before we shut the server down
+        waitForAbort.resolve();
-      if (streamCancelled !== undefined) {
-        await streamCancelled;
-      }
 
-      // Since the handler has a chance of creating resources or running async
+        // Wait for cancellation before we shut the server down
-      // ops, we need to use a graceful shutdown here to ensure they have fully
+        if (streamCancelled !== undefined) {
-      // drained.
+          await streamCancelled;
-      await server.shutdown();
+        }
 
-      await server.finished;
+        // Since the handler has a chance of creating resources or running async
-    },
+        // ops, we need to use a graceful shutdown here to ensure they have fully
-  });
+        // drained.
+        await server.shutdown();
+
+        await server.finished;
+      },
+    });
+  }
 }
 
 Deno.test(

ext/http/http_next.rs

@@ -716,6 +716,8 @@ pub async fn op_http_set_response_body_resource(
     }
   };
 
+  *http.needs_close_after_finish() = true;
+
   set_response(
     http.clone(),
     resource.size_hint().1.map(|s| s as usize),
@@ -726,7 +728,6 @@ pub async fn op_http_set_response_body_resource(
     },
   );
 
-  *http.needs_close_after_finish() = true;
   http.response_body_finished().await;
   Ok(())
 }