fix(ext/http): ensure request body resource lives as long as response is alive (#20206)

Deno.serve's fast streaming implementation was not keeping the request
body resource ID alive. We were taking the `Rc<Resource>` from the
resource table during the response, so a hairpin duplex response that
fed back the request body would work.

However, if any JS code attempted to read from the request body (which
requires the resource ID to be valid), the response would fail with a
difficult-to-diagnose "EOF" error.

This was affecting more complex duplex uses of `Deno.fetch` (though as
far as I can tell was unreported).

Simple test:

```ts
        const reader = request.body.getReader();
        return new Response(
          new ReadableStream({
            async pull(controller) {
              const { done, value } = await reader.read();
              if (done) {
                controller.close();
              } else {
                controller.enqueue(value);
              }
            },
          }),
```

And then attempt to use the stream in duplex mode:

```ts

async function testDuplex(
  reader: ReadableStreamDefaultReader<Uint8Array>,
  writable: WritableStreamDefaultWriter<Uint8Array>,
) {
  await writable.write(new Uint8Array([1]));
  const chunk1 = await reader.read();
  assert(!chunk1.done);
  assertEquals(chunk1.value, new Uint8Array([1]));
  await writable.write(new Uint8Array([2]));
  const chunk2 = await reader.read();
  assert(!chunk2.done);
  assertEquals(chunk2.value, new Uint8Array([2]));
  await writable.close();
  const chunk3 = await reader.read();
  assert(chunk3.done);
}
```

In older versions of Deno, this would just lock up. I believe after
23ff0e722e3c4b0827940853c53c5ee2ede5ec9f, it started throwing a more
explicit error:

```
httpServerStreamDuplexJavascript => ./cli/tests/unit/serve_test.ts:1339:6
error: TypeError: request or response body error: error reading a body from connection: Connection reset by peer (os error 54)
    at async Object.pull (ext:deno_web/06_streams.js:810:27)
```

1 files changed, 59 insertions, 3 deletions

diff --git a/ext/http/slab.rs b/ext/http/slab.rs
index dbe1a6635..8c285c860 100644
--- a/ext/http/slab.rs
+++ b/ext/http/slab.rs
@@ -3,6 +3,8 @@ use crate::request_properties::HttpConnectionProperties;
 use crate::response_body::CompletionHandle;
 use crate::response_body::ResponseBytes;
 use deno_core::error::AnyError;
+use deno_core::OpState;
+use deno_core::ResourceId;
 use http::request::Parts;
 use http::HeaderMap;
 use hyper1::body::Incoming;
@@ -18,10 +20,36 @@ pub type Request = hyper1::Request<Incoming>;
 pub type Response = hyper1::Response<ResponseBytes>;
 pub type SlabId = u32;
 
+enum RequestBodyState {
+  Incoming(Incoming),
+  Resource(HttpRequestBodyAutocloser),
+}
+
+impl From<Incoming> for RequestBodyState {
+  fn from(value: Incoming) -> Self {
+    RequestBodyState::Incoming(value)
+  }
+}
+
+/// Ensures that the request body closes itself when no longer needed.
+pub struct HttpRequestBodyAutocloser(ResourceId, Rc<RefCell<OpState>>);
+
+impl HttpRequestBodyAutocloser {
+  pub fn new(res: ResourceId, op_state: Rc<RefCell<OpState>>) -> Self {
+    Self(res, op_state)
+  }
+}
+
+impl Drop for HttpRequestBodyAutocloser {
+  fn drop(&mut self) {
+    _ = self.1.borrow_mut().resource_table.close(self.0);
+  }
+}
+
 pub struct HttpSlabRecord {
   request_info: HttpConnectionProperties,
   request_parts: Parts,
-  request_body: Option<Incoming>,
+  request_body: Option<RequestBodyState>,
   // The response may get taken before we tear this down
   response: Option<Response>,
   promise: CompletionHandle,
@@ -98,6 +126,7 @@ fn slab_insert_raw(
     let mut slab = slab.borrow_mut();
     let body = ResponseBytes::default();
     let trailers = body.trailers();
+    let request_body = request_body.map(|r| r.into());
     slab.insert(HttpSlabRecord {
       request_info,
       request_parts,
@@ -174,8 +203,35 @@ impl SlabEntry {
   }
 
   /// Take the Hyper body from this entry.
-  pub fn take_body(&mut self) -> Incoming {
-    self.self_mut().request_body.take().unwrap()
+  pub fn take_body(&mut self) -> Option<Incoming> {
+    let body_holder = &mut self.self_mut().request_body;
+    let body = body_holder.take();
+    match body {
+      Some(RequestBodyState::Incoming(body)) => Some(body),
+      x => {
+        *body_holder = x;
+        None
+      }
+    }
+  }
+
+  pub fn take_resource(&mut self) -> Option<HttpRequestBodyAutocloser> {
+    let body_holder = &mut self.self_mut().request_body;
+    let body = body_holder.take();
+    match body {
+      Some(RequestBodyState::Resource(res)) => Some(res),
+      x => {
+        *body_holder = x;
+        None
+      }
+    }
+  }
+
+  /// Replace the request body with a resource ID and the OpState we'll need to shut it down.
+  /// We cannot keep just the resource itself, as JS code might be reading from the resource ID
+  /// to generate the response data (requiring us to keep it in the resource table).
+  pub fn put_resource(&mut self, res: HttpRequestBodyAutocloser) {
+    self.self_mut().request_body = Some(RequestBodyState::Resource(res));
   }
 
   /// Complete this entry, potentially expunging it if it is complete.