fix(ext/node): retry named-pipe connect on ERROR_PIPE_BUSY (Windows) (#33974)

On Windows, when all of a named-pipe server's instances are in use (e.g.
Docker Desktop under load from testcontainers spinning up containers),
`CreateFileW` returns `ERROR_PIPE_BUSY` (231). libuv handles this
transparently by queueing the connect to a worker thread that calls
`WaitNamedPipeW` and retries — node code like `docker-modem` relies on
this. In Deno today the same condition surfaces to JavaScript as
`Error: connect EINVAL //./pipe/docker_engine`, which is the
intermittent failure reported in the linked issue.

The fix mirrors libuv's `pipe_connect_thread_proc`: when the synchronous
`ClientOptions::open()` in `uv_pipe_connect` returns 231, spawn a
blocking task that loops on `WaitNamedPipeW(30s)` + retry until an
instance becomes available, and fire the deferred connect callback with
the resolved status once the worker completes. Also cleans up a dead
branch in `io_error_to_uv` that claimed `ERROR_PIPE_BUSY` mapped to
`PermissionDenied` in Rust std (it actually maps to `Uncategorized`, so
the check was unreachable), and adds explicit raw-os-error mappings for
the common Win32 pipe codes so they no longer fall through to
`UV_EINVAL`.

Fixes #33923

Author: bartlomieju

libs/core/uv_compat.rs

@@ -81,6 +81,24 @@ pub const UV_EOF: i32 = -4095;
 /// Map a `std::io::Error` to the closest libuv error code.
 pub(crate) fn io_error_to_uv(err: &std::io::Error) -> c_int {
   use std::io::ErrorKind;
+  // On Windows, several Win32 error codes don't get a stable ErrorKind
+  // mapping from std (they all end up as `Uncategorized`). Handle the
+  // pipe-related ones explicitly first so they don't fall through to
+  // the catch-all UV_EINVAL.
+  #[cfg(windows)]
+  if let Some(code) = err.raw_os_error() {
+    match code {
+      231 => return UV_EBUSY,  // ERROR_PIPE_BUSY
+      536 => return UV_EAGAIN, // ERROR_PIPE_LISTENING
+      230 => return UV_EPIPE,  // ERROR_BAD_PIPE
+      // `ERROR_PIPE_NOT_CONNECTED` maps to `UV_EPIPE` to match libuv's
+      // `uv_translate_sys_error` — node code pattern-matches against
+      // libuv's actual values, so semantic accuracy (`UV_ENOTCONN`)
+      // would break those callers.
+      233 => return UV_EPIPE, // ERROR_PIPE_NOT_CONNECTED
+      _ => {}
+    }
+  }
   match err.kind() {
     ErrorKind::AddrInUse => UV_EADDRINUSE,
     ErrorKind::AddrNotAvailable => UV_EINVAL,
@@ -93,15 +111,7 @@ pub(crate) fn io_error_to_uv(err: &std::io::Error) -> c_int {
     ErrorKind::InvalidInput => UV_EINVAL,
     ErrorKind::WouldBlock => UV_EAGAIN,
     ErrorKind::TimedOut => UV_ETIMEDOUT,
-    ErrorKind::PermissionDenied => {
-      // On Windows, ERROR_PIPE_BUSY (231) is mapped to PermissionDenied
-      // by Rust std, but it means the named pipe is already in use.
-      #[cfg(windows)]
-      if let Some(231) = err.raw_os_error() {
-        return UV_EADDRINUSE;
-      }
-      UV_EACCES
-    }
+    ErrorKind::PermissionDenied => UV_EACCES,
     _ => {
       // On Unix, try to use the raw OS error for a more accurate mapping.
       #[cfg(unix)]

libs/core/uv_compat/pipe.rs

@@ -80,6 +80,20 @@ pub struct uv_pipe_t {
       Box<dyn std::future::Future<Output = std::io::Result<()>> + Send>,
     >,
   >,
+  /// In-flight `WaitNamedPipeW` retry task spawned when a synchronous
+  /// `ClientOptions::open()` returned `ERROR_PIPE_BUSY`. The deferred
+  /// connect callback fires once this task completes (success or final
+  /// failure). Mirrors libuv's `pipe_connect_thread_proc`.
+  #[cfg(windows)]
+  #[allow(
+    clippy::type_complexity,
+    reason = "JoinHandle<io::Result<NamedPipeClient>> is inherently complex"
+  )]
+  pub(crate) internal_win_connect_retry: Option<
+    tokio::task::JoinHandle<
+      std::io::Result<tokio::net::windows::named_pipe::NamedPipeClient>,
+    >,
+  >,
 
   // Connected stream (from connect or accept)
   #[cfg(unix)]
@@ -219,6 +233,8 @@ pub fn new_pipe(ipc: bool) -> uv_pipe_t {
     internal_win_client: None,
     #[cfg(windows)]
     internal_win_connect_fut: None,
+    #[cfg(windows)]
+    internal_win_connect_retry: None,
     #[cfg(unix)]
     internal_stream: None,
     #[cfg(unix)]
@@ -813,6 +829,32 @@ pub unsafe fn uv_pipe_connect(
         }
         0
       }
+      Err(e) if e.raw_os_error() == Some(231) => {
+        // ERROR_PIPE_BUSY: all of the server's pipe instances are in use.
+        // Match libuv (`pipe_connect_thread_proc`) by retrying on a worker
+        // thread using `WaitNamedPipeW`. The connect callback fires
+        // asynchronously once the retry resolves.
+        if !req.is_null() {
+          (*req).handle = pipe as *mut super::stream::uv_stream_t;
+        }
+        let path_owned = path.to_owned();
+        let join = tokio::task::spawn_blocking(move || {
+          wait_named_pipe_and_open(&path_owned)
+        });
+        (*pipe).internal_win_connect_retry = Some(join);
+        (*pipe).internal_connect = Some(PipeConnectPending { req, cb });
+        (*pipe).flags |= UV_HANDLE_ACTIVE;
+
+        let inner = super::get_inner((*pipe).loop_);
+        let mut handles = inner.pipe_handles.borrow_mut();
+        if !handles.iter().any(|&h| std::ptr::eq(h, pipe)) {
+          handles.push(pipe);
+        }
+        if let Some(w) = (*pipe).internal_waker.as_ref() {
+          w.mark_ready();
+        }
+        0
+      }
       Err(e) => {
         // If the path exists but isn't a named pipe, return ENOTSOCK
         // (matching libuv behavior). ClientOptions::open returns NotFound
@@ -828,6 +870,49 @@ pub unsafe fn uv_pipe_connect(
   }
 }
 
+/// Wait for a Windows named pipe instance to become free and open a client.
+///
+/// Mirrors libuv's `pipe_connect_thread_proc`: loop on `WaitNamedPipeW`
+/// (30s per attempt) until an instance is available, then try to open it
+/// via `CreateFileW`. If another client races us, the wait/open loop
+/// continues until success or `WaitNamedPipeW` itself fails (e.g. the
+/// server is gone).
+#[cfg(windows)]
+pub(crate) fn wait_named_pipe_and_open(
+  path: &str,
+) -> std::io::Result<tokio::net::windows::named_pipe::NamedPipeClient> {
+  use std::os::windows::ffi::OsStrExt;
+  let wide: Vec<u16> = std::ffi::OsStr::new(path)
+    .encode_wide()
+    .chain(std::iter::once(0))
+    .collect();
+  loop {
+    // SAFETY: `wide` is a null-terminated wide string for the lifetime
+    // of this call. WaitNamedPipeW has no other safety requirements.
+    let ok = unsafe {
+      windows_sys::Win32::System::Pipes::WaitNamedPipeW(wide.as_ptr(), 30000)
+    };
+    if ok == 0 {
+      // Any `WaitNamedPipeW` failure — including `ERROR_SEM_TIMEOUT` —
+      // exits the loop. Matches libuv's `pipe_connect_thread_proc`:
+      // the 30s timeout signals the server is unresponsive and the
+      // connect should be reported as failed rather than retried
+      // indefinitely.
+      return Err(std::io::Error::last_os_error());
+    }
+    match tokio::net::windows::named_pipe::ClientOptions::new().open(path) {
+      Ok(client) => return Ok(client),
+      Err(e) if e.raw_os_error() == Some(231) => {
+        // Another client raced us to the freed instance. Loop and wait
+        // for another one. Yield first to avoid a hot spin.
+        std::thread::yield_now();
+        continue;
+      }
+      Err(e) => return Err(e),
+    }
+  }
+}
+
 /// Attempt a non-blocking write to a pipe. Returns the number of bytes
 /// written (>= 0) or a negative `UV_*` error code. `UV_EAGAIN` when the
 /// pipe would block. Mirrors `uv__try_write()` in libuv.
@@ -963,6 +1048,9 @@ pub(crate) unsafe fn close_pipe(pipe: *mut uv_pipe_t) {
       if let Some(handle) = (*pipe).internal_pending_read.take() {
         handle.abort();
       }
+      if let Some(retry) = (*pipe).internal_win_connect_retry.take() {
+        retry.abort();
+      }
       if let Some(handle) = (*pipe).internal_handle.take() {
         // SAFETY: handle is a valid OS handle from get_osfhandle.
         windows_sys::Win32::Foundation::CloseHandle(handle as _);
@@ -1115,7 +1203,50 @@ pub(crate) unsafe fn poll_pipe_handle(
 
   unsafe {
     // 1. Poll deferred connect callback (from uv_pipe_connect on Windows).
-    if let Some(pending) = (*pipe_ptr).internal_connect.take() {
+    //
+    // Two sub-cases:
+    //   a) Sync-success: ClientOptions::open() returned a connected client
+    //      immediately; the client is already stored in `internal_win_client`
+    //      and we just need to fire cb(req, 0).
+    //   b) Retry-in-flight: ClientOptions::open() returned ERROR_PIPE_BUSY
+    //      and a worker task is calling WaitNamedPipeW + retry. Wait for the
+    //      task to finish before firing cb with the resolved status.
+    if let Some(ref mut join) = (*pipe_ptr).internal_win_connect_retry {
+      match std::pin::Pin::new(join).poll(cx) {
+        Poll::Pending => { /* keep waiting */ }
+        Poll::Ready(join_res) => {
+          (*pipe_ptr).internal_win_connect_retry = None;
+          let status = match join_res {
+            Ok(Ok(client)) => {
+              // Same FILE_TYPE_PIPE sanity check as the sync path.
+              use std::os::windows::io::AsRawHandle;
+              let handle = client.as_raw_handle();
+              let file_type =
+                windows_sys::Win32::Storage::FileSystem::GetFileType(
+                  handle as _,
+                );
+              if file_type
+                != windows_sys::Win32::Storage::FileSystem::FILE_TYPE_PIPE
+              {
+                drop(client);
+                super::UV_ENOTSOCK
+              } else {
+                (*pipe_ptr).internal_win_client = Some(client);
+                0
+              }
+            }
+            Ok(Err(ref e)) => io_error_to_uv(e),
+            Err(_) => super::UV_ECANCELED,
+          };
+          if let Some(pending) = (*pipe_ptr).internal_connect.take()
+            && let Some(cb) = pending.cb
+          {
+            cb(pending.req, status);
+          }
+          any_work = true;
+        }
+      }
+    } else if let Some(pending) = (*pipe_ptr).internal_connect.take() {
       if let Some(cb) = pending.cb {
         cb(pending.req, 0);
       }

libs/core/uv_compat/tests.rs

@@ -2989,3 +2989,175 @@ async fn pipe_open_not_active_until_read_start() {
   })
   .await;
 }
+
+/// Regression test for https://github.com/denoland/deno/issues/33923.
+///
+/// On Windows, when all named-pipe instances are in use (e.g., Docker
+/// Desktop under load), `CreateFileW` returns `ERROR_PIPE_BUSY` (231).
+/// libuv handles this by transparently retrying via `WaitNamedPipeW` on
+/// a worker thread; node code (like `docker-modem`) relies on this. We
+/// must do the same — without it, the connection surfaces as
+/// `connect EINVAL` to JavaScript.
+#[cfg(windows)]
+#[tokio::test(flavor = "current_thread")]
+async fn pipe_connect_retries_on_error_pipe_busy() {
+  use tokio::net::windows::named_pipe::ClientOptions;
+  use tokio::net::windows::named_pipe::ServerOptions;
+  // Unique pipe name per test run.
+  let pipe_name =
+    format!(r"\\.\pipe\deno-uvcompat-busy-{}", std::process::id());
+
+  // Server with max_instances=1 -- only one client can be connected at
+  // a time. The first server instance must use first_pipe_instance(true).
+  let server = ServerOptions::new()
+    .first_pipe_instance(true)
+    .max_instances(1)
+    .create(&pipe_name)
+    .unwrap();
+
+  // Client 1 connects, consuming the only instance.
+  let _client1 = ClientOptions::new().open(&pipe_name).unwrap();
+  // Drive the server's ConnectNamedPipe so the pair is fully established.
+  server.connect().await.unwrap();
+
+  // Sanity: a second synchronous open MUST now return ERROR_PIPE_BUSY.
+  // (If this assertion fails, the test setup no longer triggers the
+  // condition we're trying to exercise.)
+  let busy_err = ClientOptions::new().open(&pipe_name).unwrap_err();
+  assert_eq!(
+    busy_err.raw_os_error(),
+    Some(231),
+    "expected ERROR_PIPE_BUSY, got {:?}",
+    busy_err
+  );
+
+  // Now exercise the retry helper. It should block in WaitNamedPipeW
+  // until we drop the existing server and create a fresh instance.
+  let pipe_name_clone = pipe_name.clone();
+  let join = tokio::task::spawn_blocking(move || {
+    pipe::wait_named_pipe_and_open(&pipe_name_clone)
+  });
+
+  // Give the worker a moment to actually enter WaitNamedPipeW. Without
+  // this the new server instance below could appear before the wait
+  // starts, masking whether the retry actually waits.
+  tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+  assert!(!join.is_finished(), "wait must block while pipe is busy");
+
+  // Free the instance: drop the original server, then create a new one.
+  drop(server);
+  let new_server = ServerOptions::new()
+    .first_pipe_instance(false)
+    .create(&pipe_name)
+    .unwrap();
+  // Drive the server side so the client's CreateFileW can complete.
+  let connect_fut = tokio::spawn(async move {
+    new_server.connect().await.unwrap();
+    new_server
+  });
+
+  // The retry helper must observe the new instance and succeed.
+  let result = tokio::time::timeout(std::time::Duration::from_secs(5), join)
+    .await
+    .expect("wait_named_pipe_and_open did not complete in time")
+    .expect("spawn_blocking join failed")
+    .expect("wait_named_pipe_and_open returned an error");
+
+  // Hand-off succeeded: result is a connected client.
+  drop(result);
+  let _ = connect_fut.await;
+}
+
+/// End-to-end check that `uv_pipe_connect` returns successfully (0 from
+/// the function, status=0 to the connect callback) even when the initial
+/// `ClientOptions::open()` fails with `ERROR_PIPE_BUSY`. Prior to the
+/// retry fix, this scenario surfaced as `UV_EINVAL` (the
+/// `connect EINVAL //./pipe/...` error from #33923).
+#[cfg(windows)]
+#[tokio::test(flavor = "current_thread")]
+async fn uv_pipe_connect_busy_then_succeeds() {
+  use std::sync::atomic::AtomicI32;
+  use std::sync::atomic::Ordering;
+
+  use tokio::net::windows::named_pipe::ClientOptions;
+  use tokio::net::windows::named_pipe::ServerOptions;
+
+  let pipe_name =
+    format!(r"\\.\pipe\deno-uvcompat-busy-cb-{}", std::process::id());
+
+  // Saturate the pipe so the synchronous open in uv_pipe_connect hits
+  // ERROR_PIPE_BUSY.
+  let server = ServerOptions::new()
+    .first_pipe_instance(true)
+    .max_instances(1)
+    .create(&pipe_name)
+    .unwrap();
+  let _client1 = ClientOptions::new().open(&pipe_name).unwrap();
+  server.connect().await.unwrap();
+
+  run_test(async |runtime, uv_loop| {
+    // Static result slot for the connect callback. The pointer is
+    // stable for the test's duration.
+    static RESULT: AtomicI32 = AtomicI32::new(i32::MIN);
+    RESULT.store(i32::MIN, Ordering::SeqCst);
+
+    unsafe extern "C" fn connect_cb(_req: *mut uv_connect_t, status: i32) {
+      RESULT.store(status, Ordering::SeqCst);
+    }
+
+    let mut pipe = pipe::new_pipe(false);
+    unsafe {
+      assert_ok(pipe::uv_pipe_init(uv_loop, &mut pipe, 0));
+    }
+
+    let mut req = new_connect();
+    let rc = unsafe {
+      pipe::uv_pipe_connect(&mut req, &mut pipe, &pipe_name, Some(connect_cb))
+    };
+    // The retry path returns 0 synchronously; the callback fires later.
+    assert_eq!(rc, 0, "uv_pipe_connect should defer, not fail with {rc}");
+    assert_eq!(
+      RESULT.load(Ordering::SeqCst),
+      i32::MIN,
+      "callback must not fire before the pipe is free"
+    );
+
+    // Free up the pipe and create a new server instance so the worker's
+    // WaitNamedPipeW returns.
+    drop(server);
+    let new_server = ServerOptions::new()
+      .first_pipe_instance(false)
+      .create(&pipe_name)
+      .unwrap();
+    // Drive the server side concurrently so the client's CreateFileW
+    // (issued from the retry worker) can complete.
+    let server_task = tokio::spawn(async move {
+      new_server.connect().await.unwrap();
+      new_server
+    });
+
+    // Tick the loop until the callback fires (with timeout).
+    let deadline =
+      std::time::Instant::now() + std::time::Duration::from_secs(5);
+    while RESULT.load(Ordering::SeqCst) == i32::MIN {
+      if std::time::Instant::now() > deadline {
+        panic!("connect callback did not fire within timeout");
+      }
+      tick(runtime).await;
+      tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+    }
+    assert_eq!(
+      RESULT.load(Ordering::SeqCst),
+      0,
+      "expected status=0, got {}",
+      RESULT.load(Ordering::SeqCst)
+    );
+
+    let _ = server_task.await;
+    unsafe {
+      uv_close(&mut pipe as *mut uv_pipe_t as *mut uv_handle_t, None);
+    }
+    tick(runtime).await;
+  })
+  .await;
+}