fix(ext/node): support http over unix sockets (#29182)

Closes #20255

---------

Co-authored-by: Divy Srivastava <[email protected]>

Author: satyarohith

ext/node/ops/http.rs

@@ -139,6 +139,10 @@ pub enum ConnError {
   #[class(generic)]
   #[error(transparent)]
   ReuniteTcp(#[from] tokio::net::tcp::ReuniteError),
+  #[cfg(unix)]
+  #[class(generic)]
+  #[error(transparent)]
+  ReuniteUnix(#[from] tokio::net::unix::ReuniteError),
   #[class(inherit)]
   #[error(transparent)]
   Canceled(#[from] deno_core::Canceled),
@@ -149,6 +153,9 @@ pub enum ConnError {
 
 #[op2(async, stack_trace)]
 #[serde]
+// This is triggering a known false positive for explicit drop(state) calls.
+// See https://rust-lang.github.io/rust-clippy/master/index.html#await_holding_refcell_ref
+#[allow(clippy::await_holding_refcell_ref)]
 pub async fn op_node_http_request_with_conn<P>(
   state: Rc<RefCell<OpState>>,
   #[serde] method: ByteString,
@@ -162,43 +169,73 @@ pub async fn op_node_http_request_with_conn<P>(
 where
   P: crate::NodePermissions + 'static,
 {
-  let (_handle, mut sender) = if encrypted {
-    let resource_rc = state
-      .borrow_mut()
-      .resource_table
-      .take::<TlsStreamResource>(conn_rid)
-      .map_err(ConnError::Resource)?;
-    let resource =
-      Rc::try_unwrap(resource_rc).map_err(|_e| ConnError::TlsStreamBusy)?;
-    let (read_half, write_half) = resource.into_inner();
-    let tcp_stream = read_half.unsplit(write_half);
-    let io = TokioIo::new(tcp_stream);
-    let (sender, conn) = hyper::client::conn::http1::handshake(io).await?;
-    (
-      tokio::task::spawn(async move { conn.with_upgrades().await }),
-      sender,
-    )
-  } else {
-    let resource_rc = state
-      .borrow_mut()
+  let (_handle, mut sender) = {
+    let mut state = state.borrow_mut();
+    if encrypted {
+      let resource_rc = state
+        .resource_table
+        .take::<TlsStreamResource>(conn_rid)
+        .map_err(ConnError::Resource)?;
+      let resource =
+        Rc::try_unwrap(resource_rc).map_err(|_e| ConnError::TlsStreamBusy)?;
+      let (read_half, write_half) = resource.into_inner();
+      let tcp_stream = read_half.unsplit(write_half);
+      let io = TokioIo::new(tcp_stream);
+      drop(state);
+      let (sender, conn) = hyper::client::conn::http1::handshake(io).await?;
+      (
+        tokio::task::spawn(async move { conn.with_upgrades().await }),
+        sender,
+      )
+    } else if let Ok(resource_rc) = state
       .resource_table
       .take::<TcpStreamResource>(conn_rid)
-      .map_err(ConnError::Resource)?;
-    let resource =
-      Rc::try_unwrap(resource_rc).map_err(|_| ConnError::TcpStreamBusy)?;
-    let (read_half, write_half) = resource.into_inner();
-    let tcp_stream = read_half.reunite(write_half)?;
-    let io = TokioIo::new(tcp_stream);
-    let (sender, conn) = hyper::client::conn::http1::handshake(io).await?;
-
-    // Spawn a task to poll the connection, driving the HTTP state
-    (
-      tokio::task::spawn(async move {
-        conn.with_upgrades().await?;
-        Ok::<_, _>(())
-      }),
-      sender,
-    )
+      .map_err(ConnError::Resource)
+    {
+      let resource =
+        Rc::try_unwrap(resource_rc).map_err(|_| ConnError::TcpStreamBusy)?;
+      let (read_half, write_half) = resource.into_inner();
+      let tcp_stream = read_half.reunite(write_half)?;
+      let io = TokioIo::new(tcp_stream);
+      drop(state);
+      let (sender, conn) = hyper::client::conn::http1::handshake(io).await?;
+
+      // Spawn a task to poll the connection, driving the HTTP state
+      (
+        tokio::task::spawn(async move {
+          conn.with_upgrades().await?;
+          Ok::<_, _>(())
+        }),
+        sender,
+      )
+    } else {
+      #[cfg(unix)]
+      {
+        let resource_rc = state
+          .resource_table
+          .take::<deno_net::io::UnixStreamResource>(conn_rid)
+          .map_err(ConnError::Resource)?;
+        let resource =
+          Rc::try_unwrap(resource_rc).map_err(|_| ConnError::TcpStreamBusy)?;
+        let (read_half, write_half) = resource.into_inner();
+        let tcp_stream = read_half.reunite(write_half)?;
+        let io = TokioIo::new(tcp_stream);
+        drop(state);
+        let (sender, conn) = hyper::client::conn::http1::handshake(io).await?;
+
+        // Spawn a task to poll the connection, driving the HTTP state
+        (
+          tokio::task::spawn(async move {
+            conn.with_upgrades().await?;
+            Ok::<_, _>(())
+          }),
+          sender,
+        )
+      }
+
+      #[cfg(not(unix))]
+      return Err(ConnError::Resource(ResourceError::BadResourceId));
+    }
   };
 
   // Create the request.

ext/node/polyfills/http.ts

@@ -515,7 +515,12 @@ class ClientRequest extends OutgoingMessage {
           span.setAttribute("url.query", parsedUrl.search.slice(1));
         }
 
-        let baseConnRid = handle[kStreamBaseField][internalRidSymbol];
+        let baseConnRid;
+        try {
+          baseConnRid = handle[kStreamBaseField][internalRidSymbol];
+        } catch (err) {
+          throw (this.socket.errored || err);
+        }
         if (this._encrypted) {
           const hasCaCerts = this.agent?.options?.ca !== undefined;
           const caCerts = hasCaCerts

tests/unit_node/http_test.ts

@@ -2080,3 +2080,31 @@ Deno.test("[node/http] rawHeaders are in flattened format", async () => {
   await promise;
   await new Promise((resolve) => server.close(resolve));
 });
+
+Deno.test("[node/http] client http over unix socket works", {
+  ignore: Deno.build.os == "windows",
+}, async () => {
+  const { promise, resolve } = Promise.withResolvers<void>();
+  const socketPath = Deno.makeTempDirSync() + "/server.sock";
+  const server = Deno.serve({
+    transport: "unix",
+    path: socketPath,
+    onListen,
+  }, (_req) => new Response("ok"));
+
+  function onListen() {
+    const options = {
+      socketPath,
+      path: "/",
+      method: "GET",
+    };
+    http.request(options, async (res) => {
+      assertEquals(res.statusCode, 200);
+      assertEquals(await text(res), "ok");
+      resolve();
+      server.shutdown();
+    }).end();
+  }
+  await promise;
+  await server.finished;
+});