fix(ext/node): support HTTP over Windows named pipes in node:http (#32414)

Add Windows Named Pipe support to Deno's Node.js HTTP compatibility
layer, enabling libraries like `dockerode`/`docker-modem` that use
`http.request({socketPath: "//./pipe/docker_engine"})` to work on
Windows.

Add `WindowsPipe` variant to `NetworkStream` with proper handling across
HTTP server, request properties, and Node HTTP client ops.

Implement EBUSY (ERROR_PIPE_BUSY / os error 231) retry logic in
`Pipe.connect()` for Windows, emulating libuv's `WaitNamedPipeW`
behavior with polling retries.

Author: fraidev

ext/http/http_next.rs

@@ -1007,6 +1007,10 @@ where
     NetworkStream::Tunnel(conn) => {
       serve_http(conn, connection_properties, lifetime, tx, options)
     }
+    #[cfg(windows)]
+    NetworkStream::WindowsPipe(conn) => {
+      serve_http(conn, connection_properties, lifetime, tx, options)
+    }
   }
 }
 

ext/http/request_properties.rs

@@ -172,6 +172,8 @@ impl HttpPropertyExtractor for DefaultHttpPropertyExtractor {
       ))]
       NetworkStreamAddress::Vsock(vsock) => Some(vsock.port()),
       NetworkStreamAddress::Tunnel(ref addr) => Some(addr.port() as _),
+      #[cfg(windows)]
+      NetworkStreamAddress::WindowsPipe(_) => None,
     };
     let peer_address = match peer_address {
       NetworkStreamAddress::Ip(addr) => Rc::from(addr.ip().to_string()),
@@ -186,6 +188,8 @@ impl HttpPropertyExtractor for DefaultHttpPropertyExtractor {
         Rc::from(format!("vsock:{}", addr.cid()))
       }
       NetworkStreamAddress::Tunnel(ref addr) => Rc::from(addr.hostname()),
+      #[cfg(windows)]
+      NetworkStreamAddress::WindowsPipe(_) => Rc::from("pipe"),
     };
     let local_port = listen_properties.local_port;
     let stream_type = listen_properties.stream_type;
@@ -231,6 +235,8 @@ fn listener_properties(
     ))]
     NetworkStreamAddress::Vsock(vsock) => Some(vsock.port()),
     NetworkStreamAddress::Tunnel(addr) => Some(addr.port() as _),
+    #[cfg(windows)]
+    NetworkStreamAddress::WindowsPipe(_) => None,
   };
   Ok(HttpListenProperties {
     scheme,
@@ -290,6 +296,8 @@ fn req_host_from_addr(
         format!("{}:{}", addr.hostname(), addr.port())
       }
     }
+    #[cfg(windows)]
+    NetworkStreamAddress::WindowsPipe(_) => "localhost".to_owned(),
   }
 }
 
@@ -305,6 +313,8 @@ fn req_scheme_from_stream_type(stream_type: NetworkStreamType) -> &'static str {
       target_os = "macos"
     ))]
     NetworkStreamType::Vsock => "http+vsock://",
+    #[cfg(windows)]
+    NetworkStreamType::WindowsPipe => "http://",
   }
 }
 
@@ -335,6 +345,8 @@ fn req_host<'a>(
         target_os = "macos"
       ))]
       NetworkStreamType::Vsock => {}
+      #[cfg(windows)]
+      NetworkStreamType::WindowsPipe => {}
     }
     return Some(Cow::Borrowed(auth.as_str()));
   }

ext/net/lib.rs

@@ -13,7 +13,7 @@ pub mod resolve_addr;
 pub mod tcp;
 pub mod tunnel;
 #[cfg(windows)]
-mod win_pipe;
+pub mod win_pipe;
 
 use std::sync::Arc;
 

ext/net/raw.rs

@@ -468,6 +468,17 @@ network_stream!(
     crate::tunnel::TunnelStreamResource,
     crate::tunnel::OwnedReadHalf,
     crate::tunnel::OwnedWriteHalf,
+  ],
+  #[cfg(windows)]
+  [
+    WindowsPipe,
+    windowsPipe,
+    crate::win_pipe::WindowsPipeStream,
+    crate::win_pipe::WindowsPipeListener,
+    crate::win_pipe::WindowsPipeAddr,
+    crate::win_pipe::NamedPipe,
+    tokio::io::ReadHalf<crate::win_pipe::WindowsPipeStream>,
+    tokio::io::WriteHalf<crate::win_pipe::WindowsPipeStream>,
   ]
 );
 
@@ -478,6 +489,8 @@ pub enum NetworkStreamAddress {
   #[cfg(any(target_os = "android", target_os = "linux", target_os = "macos"))]
   Vsock(tokio_vsock::VsockAddr),
   Tunnel(crate::tunnel::TunnelAddr),
+  #[cfg(windows)]
+  WindowsPipe(crate::win_pipe::WindowsPipeAddr),
 }
 
 impl From<std::net::SocketAddr> for NetworkStreamAddress {
@@ -506,6 +519,13 @@ impl From<crate::tunnel::TunnelAddr> for NetworkStreamAddress {
   }
 }
 
+#[cfg(windows)]
+impl From<crate::win_pipe::WindowsPipeAddr> for NetworkStreamAddress {
+  fn from(value: crate::win_pipe::WindowsPipeAddr) -> Self {
+    NetworkStreamAddress::WindowsPipe(value)
+  }
+}
+
 #[derive(Debug, thiserror::Error, deno_error::JsError)]
 pub enum TakeNetworkStreamError {
   #[class("Busy")]
@@ -525,6 +545,10 @@ pub enum TakeNetworkStreamError {
   #[class("Busy")]
   #[error("Tunnel socket is currently in use")]
   TunnelBusy,
+  #[cfg(windows)]
+  #[class("Busy")]
+  #[error("Windows named pipe is currently in use")]
+  WindowsPipeBusy,
   #[class(generic)]
   #[error(transparent)]
   ReuniteTcp(#[from] tokio::net::tcp::ReuniteError),
@@ -606,6 +630,20 @@ pub fn take_network_stream_resource(
     return Ok(NetworkStream::Tunnel(stream));
   }
 
+  #[cfg(windows)]
+  if let Ok(resource_rc) =
+    resource_table.take::<crate::win_pipe::NamedPipe>(stream_rid)
+  {
+    let resource = Rc::try_unwrap(resource_rc)
+      .map_err(|_| TakeNetworkStreamError::WindowsPipeBusy)?;
+    let client = resource.into_client().map_err(|_| {
+      TakeNetworkStreamError::Resource(ResourceError::BadResourceId)
+    })?;
+    return Ok(NetworkStream::WindowsPipe(
+      crate::win_pipe::WindowsPipeStream::new(client),
+    ));
+  }
+
   Err(TakeNetworkStreamError::Resource(
     ResourceError::BadResourceId,
   ))

ext/net/win_pipe.rs

@@ -3,16 +3,22 @@
 use std::borrow::Cow;
 use std::ffi::OsStr;
 use std::io;
+use std::pin::Pin;
 use std::rc::Rc;
+use std::task::Context;
+use std::task::Poll;
 
 use deno_core::AsyncRefCell;
 use deno_core::AsyncResult;
 use deno_core::CancelHandle;
 use deno_core::CancelTryFuture;
 use deno_core::RcRef;
 use deno_core::Resource;
+use tokio::io::AsyncRead;
 use tokio::io::AsyncReadExt;
+use tokio::io::AsyncWrite;
 use tokio::io::AsyncWriteExt;
+use tokio::io::ReadBuf;
 use tokio::io::ReadHalf;
 use tokio::io::WriteHalf;
 use tokio::net::windows::named_pipe;
@@ -120,6 +126,28 @@ impl NamedPipe {
       )),
     }
   }
+
+  /// Cancel all pending read/write operations on this pipe.
+  /// This triggers the `CancelHandle`, causing any in-flight async ops
+  /// (e.g., reads started by `readStart()`) to complete with a cancellation
+  /// error, releasing their `Rc` references to this resource.
+  pub fn cancel_pending_ops(&self) {
+    self.cancel.cancel();
+  }
+
+  /// Consume this `NamedPipe` and reunite the split read/write halves back
+  /// into a `NamedPipeClient`. Only works for client pipes.
+  pub fn into_client(self) -> io::Result<named_pipe::NamedPipeClient> {
+    let read_half = self.read_half.into_inner();
+    let write_half = self.write_half.into_inner();
+    match (read_half, write_half) {
+      (NamedPipeRead::Client(r), NamedPipeWrite::Client(w)) => Ok(r.unsplit(w)),
+      _ => Err(io::Error::new(
+        io::ErrorKind::InvalidInput,
+        "cannot extract client from non-client pipe",
+      )),
+    }
+  }
 }
 
 impl Resource for NamedPipe {
@@ -134,3 +162,90 @@ impl Resource for NamedPipe {
     self.cancel.cancel();
   }
 }
+
+/// A stub address type for Windows named pipes.
+/// Pipes don't have traditional network addresses.
+#[derive(Copy, Clone, PartialEq, Eq)]
+pub struct WindowsPipeAddr;
+
+/// A wrapper around `NamedPipeClient` that implements the stream traits
+/// required by the network stream abstraction.
+pub struct WindowsPipeStream(named_pipe::NamedPipeClient);
+
+impl WindowsPipeStream {
+  pub fn new(client: named_pipe::NamedPipeClient) -> Self {
+    Self(client)
+  }
+
+  pub fn local_addr(&self) -> io::Result<WindowsPipeAddr> {
+    Ok(WindowsPipeAddr)
+  }
+
+  pub fn peer_addr(&self) -> io::Result<WindowsPipeAddr> {
+    Ok(WindowsPipeAddr)
+  }
+
+  pub fn into_split(
+    self,
+  ) -> (tokio::io::ReadHalf<Self>, tokio::io::WriteHalf<Self>) {
+    tokio::io::split(self)
+  }
+}
+
+impl AsyncRead for WindowsPipeStream {
+  fn poll_read(
+    self: Pin<&mut Self>,
+    cx: &mut Context<'_>,
+    buf: &mut ReadBuf<'_>,
+  ) -> Poll<io::Result<()>> {
+    Pin::new(&mut self.get_mut().0).poll_read(cx, buf)
+  }
+}
+
+impl AsyncWrite for WindowsPipeStream {
+  fn poll_write(
+    self: Pin<&mut Self>,
+    cx: &mut Context<'_>,
+    buf: &[u8],
+  ) -> Poll<io::Result<usize>> {
+    Pin::new(&mut self.get_mut().0).poll_write(cx, buf)
+  }
+
+  fn poll_flush(
+    self: Pin<&mut Self>,
+    cx: &mut Context<'_>,
+  ) -> Poll<io::Result<()>> {
+    Pin::new(&mut self.get_mut().0).poll_flush(cx)
+  }
+
+  fn poll_shutdown(
+    self: Pin<&mut Self>,
+    cx: &mut Context<'_>,
+  ) -> Poll<io::Result<()>> {
+    Pin::new(&mut self.get_mut().0).poll_shutdown(cx)
+  }
+}
+
+/// A stub listener type for Windows named pipes.
+/// Not used for HTTP client connections, but required by the
+/// network stream abstraction.
+pub struct WindowsPipeListener;
+
+impl WindowsPipeListener {
+  #[allow(clippy::unused_async)]
+  pub async fn accept(
+    &self,
+  ) -> io::Result<(WindowsPipeStream, WindowsPipeAddr)> {
+    Err(io::Error::new(
+      io::ErrorKind::Unsupported,
+      "WindowsPipeListener::accept is not supported",
+    ))
+  }
+
+  pub fn local_addr(&self) -> io::Result<WindowsPipeAddr> {
+    Err(io::Error::new(
+      io::ErrorKind::Unsupported,
+      "WindowsPipeListener::local_addr is not supported",
+    ))
+  }
+}

ext/node/ops/http.rs

@@ -176,6 +176,62 @@ pub async fn op_node_http_request_with_conn(
         .any(|part| part.trim_ascii() == b"upgrade")
   });
 
+  // Take the network stream resource for HTTP communication.
+  // On Windows, NamedPipe resources may have pending read operations
+  // (from readStart() in stream_wrap.ts) that hold extra Rc references,
+  // preventing Rc::try_unwrap(). We handle this by cancelling pending ops
+  // and yielding to let them complete before extracting the pipe.
+  #[cfg(windows)]
+  let stream = {
+    let is_pipe = state
+      .borrow()
+      .resource_table
+      .get::<deno_net::win_pipe::NamedPipe>(conn_rid)
+      .is_ok();
+    if is_pipe {
+      // Take the NamedPipe from the resource table
+      let pipe_rc = state
+        .borrow_mut()
+        .resource_table
+        .take::<deno_net::win_pipe::NamedPipe>(conn_rid)
+        .map_err(ConnError::Resource)?;
+
+      // Cancel pending read/write operations. This triggers the CancelHandle,
+      // causing in-flight ops to complete with a cancellation error and
+      // release their Rc references.
+      pipe_rc.cancel_pending_ops();
+
+      // Yield to the event loop so cancelled ops can be polled, see the
+      // cancellation, and drop their Rc references to the NamedPipe.
+      //
+      // Invariant: a single yield is sufficient because:
+      // 1. cancel_pending_ops() triggers the CancelHandle, which causes
+      //    all in-flight read/write futures to resolve on their next poll.
+      // 2. The `if (!this.#reading) return;` guard in stream_wrap.ts's
+      //    #read() (after its own PromiseResolve yield) ensures the JS
+      //    read loop bails out before starting a new op_read.
+      // 3. yield_now() gives the executor one turn to poll those cancelled
+      //    futures and drop their Rc references.
+      tokio::task::yield_now().await;
+
+      // Now we should be the sole Rc owner
+      let resource = Rc::try_unwrap(pipe_rc)
+        .map_err(|_| ConnError::Resource(ResourceError::BadResourceId))?;
+      let client = resource
+        .into_client()
+        .map_err(|_| ConnError::Resource(ResourceError::BadResourceId))?;
+      NetworkStream::WindowsPipe(deno_net::win_pipe::WindowsPipeStream::new(
+        client,
+      ))
+    } else {
+      take_network_stream_resource(
+        &mut state.borrow_mut().resource_table,
+        conn_rid,
+      )
+      .map_err(|_| ConnError::Resource(ResourceError::BadResourceId))?
+    }
+  };
+  #[cfg(not(windows))]
   let stream = take_network_stream_resource(
     &mut state.borrow_mut().resource_table,
     conn_rid,
@@ -466,6 +522,10 @@ pub async fn op_node_http_response_reclaim_conn(
     NetworkStream::Tunnel(_) => {
       return Ok(None);
     }
+    #[cfg(windows)]
+    NetworkStream::WindowsPipe(_) => {
+      return Ok(None);
+    }
   };
 
   Ok(Some(rid))