fix(#10182): hang during http server response (#10197)

cli/tests/unit/http_test.ts

@@ -200,3 +200,30 @@ unitTest(
     client.close();
   },
 );
+
+unitTest(
+  { perms: { net: true } },
+  async function httpServerRegressionHang() {
+    const promise = (async () => {
+      const listener = Deno.listen({ port: 4501 });
+      const conn = await listener.accept();
+      const httpConn = Deno.serveHttp(conn);
+      const event = await httpConn.nextRequest();
+      assert(event);
+      const { request, respondWith } = event;
+      const reqBody = await request.text();
+      assertEquals("request", reqBody);
+      await respondWith(new Response("response"));
+      httpConn.close();
+      listener.close();
+    })();
+
+    const resp = await fetch("http://127.0.0.1:4501/", {
+      method: "POST",
+      body: "request",
+    });
+    const respBody = await resp.text();
+    assertEquals("response", respBody);
+    await promise;
+  },
+);

runtime/js/40_http.js

@@ -130,7 +130,7 @@
         zeroCopyBuf = null;
       }
 
-      const responseBodyRid = Deno.core.opSync("op_http_response", [
+      const responseBodyRid = await Deno.core.opAsync("op_http_response", [
         responseSenderRid,
         resp.status ?? 200,
         flattenHeaders(resp.headers),

runtime/ops/http.rs

@@ -48,7 +48,7 @@ pub fn init(rt: &mut deno_core::JsRuntime) {
   super::reg_async(rt, "op_http_request_next", op_http_request_next);
   super::reg_async(rt, "op_http_request_read", op_http_request_read);
 
-  super::reg_sync(rt, "op_http_response", op_http_response);
+  super::reg_async(rt, "op_http_response", op_http_response);
   super::reg_async(rt, "op_http_response_write", op_http_response_write);
   super::reg_async(rt, "op_http_response_close", op_http_response_close);
 }
@@ -312,16 +312,15 @@ struct RespondArgs(
   Vec<String>,
 );
 
-fn op_http_response(
-  state: &mut OpState,
+async fn op_http_response(
+  state: Rc<RefCell<OpState>>,
   args: RespondArgs,
   data: Option<ZeroCopyBuf>,
 ) -> Result<Option<ResourceId>, AnyError> {
-  let rid = args.0;
-  let status = args.1;
-  let headers = args.2;
+  let RespondArgs(rid, status, headers) = args;
 
   let response_sender = state
+    .borrow_mut()
     .resource_table
     .take::<ResponseSenderResource>(rid)
     .ok_or_else(bad_resource_id)?;
@@ -329,6 +328,12 @@ fn op_http_response(
     .ok()
     .expect("multiple op_http_respond ongoing");
 
+  let conn_resource = state
+    .borrow()
+    .resource_table
+    .get::<ConnResource>(response_sender.conn_rid)
+    .ok_or_else(bad_resource_id)?;
+
   let mut builder = Response::builder().status(status);
 
   debug_assert_eq!(headers.len() % 2, 0);
@@ -348,11 +353,12 @@ fn op_http_response(
     let (sender, body) = Body::channel();
     res = builder.body(body)?;
 
-    let response_body_rid = state.resource_table.add(ResponseBodyResource {
-      body: AsyncRefCell::new(sender),
-      cancel: CancelHandle::default(),
-      conn_rid: response_sender.conn_rid,
-    });
+    let response_body_rid =
+      state.borrow_mut().resource_table.add(ResponseBodyResource {
+        body: AsyncRefCell::new(sender),
+        cancel: CancelHandle::default(),
+        conn_rid: response_sender.conn_rid,
+      });
 
     Some(response_body_rid)
   };
@@ -364,6 +370,12 @@ fn op_http_response(
     return Err(type_error("internal communication error"));
   }
 
+  poll_fn(|cx| match conn_resource.poll(cx) {
+    Poll::Ready(x) => Poll::Ready(x),
+    Poll::Pending => Poll::Ready(Ok(())),
+  })
+  .await?;
+
   Ok(maybe_response_body_rid)
 }