refactor(ext/http): Use HttpRecord as response body to track until body completion (#20822)

Use HttpRecord as response body so requests can be tracked all the way
to response body completion.

This allows Request properties to be accessed while the response body is
streaming.

Graceful shutdown now awaits a future instead of async spinning waiting
for requests to finish.

On the minimal benchmark this refactor improves performance an
additional 2% over pooling alone for a net 3% increase over the previous
deno main branch.

Builds upon https://github.com/denoland/deno/pull/20809 and
https://github.com/denoland/deno/pull/20770.

---------

Co-authored-by: Matt Mastracci <matthew@mastracci.com>

1 files changed, 22 insertions, 13 deletions

diff --git a/ext/http/00_serve.js b/ext/http/00_serve.js
index 05e0bb5c3..adb03a22c 100644
--- a/ext/http/00_serve.js
+++ b/ext/http/00_serve.js
@@ -43,6 +43,7 @@ const {
   ObjectHasOwn,
   ObjectPrototypeIsPrototypeOf,
   PromisePrototypeCatch,
+  PromisePrototypeThen,
   Symbol,
   TypeError,
   Uint8Array,
@@ -50,6 +51,7 @@ const {
 } = primordials;
 
 const {
+  op_http_close_after_finish,
   op_http_get_request_headers,
   op_http_get_request_method_and_url,
   op_http_read_request_body,
@@ -386,9 +388,10 @@ class ServeHandlerInfo {
   }
 }
 
-function fastSyncResponseOrStream(req, respBody, status) {
+function fastSyncResponseOrStream(req, respBody, status, innerRequest) {
   if (respBody === null || respBody === undefined) {
     // Don't set the body
+    innerRequest?.close();
     op_http_set_promise_complete(req, status);
     return;
   }
@@ -397,36 +400,43 @@ function fastSyncResponseOrStream(req, respBody, status) {
   const body = stream.body;
 
   if (ObjectPrototypeIsPrototypeOf(Uint8ArrayPrototype, body)) {
+    innerRequest?.close();
     op_http_set_response_body_bytes(req, body, status);
     return;
   }
 
   if (typeof body === "string") {
+    innerRequest?.close();
     op_http_set_response_body_text(req, body, status);
     return;
   }
 
   // At this point in the response it needs to be a stream
   if (!ObjectPrototypeIsPrototypeOf(ReadableStreamPrototype, stream)) {
+    innerRequest?.close();
     throw TypeError("invalid response");
   }
   const resourceBacking = getReadableStreamResourceBacking(stream);
+  let rid, autoClose;
   if (resourceBacking) {
-    op_http_set_response_body_resource(
-      req,
-      resourceBacking.rid,
-      resourceBacking.autoClose,
-      status,
-    );
+    rid = resourceBacking.rid;
+    autoClose = resourceBacking.autoClose;
   } else {
-    const rid = resourceForReadableStream(stream);
+    rid = resourceForReadableStream(stream);
+    autoClose = true;
+  }
+  PromisePrototypeThen(
     op_http_set_response_body_resource(
       req,
       rid,
-      true,
+      autoClose,
       status,
-    );
-  }
+    ),
+    () => {
+      innerRequest?.close();
+      op_http_close_after_finish(req);
+    },
+  );
 }
 
 /**
@@ -499,8 +509,7 @@ function mapToCallback(context, callback, onError) {
       }
     }
 
-    innerRequest?.close();
-    fastSyncResponseOrStream(req, inner.body, status);
+    fastSyncResponseOrStream(req, inner.body, status, innerRequest);
   };
 }