node:cluster: one handle for each descriptor that workers name

src/js/internal/cluster/RoundRobinHandle.ts

@@ -110,6 +110,12 @@ export default class RoundRobinHandle {
     return this.all.has(worker.id);
   }
 
+  // The descriptor that the listener of this handle holds, -1 before it listens and after remove() closed it.
+  get fd() {
+    const fd = this.server?._handle?.fd;
+    return typeof fd === "number" ? fd : -1;
+  }
+
   // With the channel still up the unacked newconn is settled by its ack; once it is gone, a crashed worker's goes to another worker and a disconnected worker's (already settled by it) is dropped.
   remove(worker, channelGone = false) {
     if (channelGone) {

src/js/internal/cluster/SharedHandle.ts

@@ -1,6 +1,6 @@
 const clusterRawBind = $newRustFunction("node_cluster_binding.rs", "clusterRawBind", 4);
 const closeRawHandle = $newRustFunction("node_cluster_binding.rs", "clusterCloseHandle", 1);
-const validateFd = $newRustFunction("node_cluster_binding.rs", "clusterValidateFd", 1);
+const validateFd = $newRustFunction("node_cluster_binding.rs", "clusterValidateFd", 2);
 
 export default class SharedHandle {
   key;
@@ -19,7 +19,7 @@ export default class SharedHandle {
     this.sharedOnly = sharedOnly === true;
 
     if (typeof fd === "number" && fd >= 0) {
-      const err = validateFd(fd);
+      const err = validateFd(fd, addressType === "udp4" || addressType === "udp6");
       if (err !== 0) {
         this.errno = err;
       } else {
@@ -47,6 +47,12 @@ export default class SharedHandle {
     return this.workers.has(worker.id);
   }
 
+  // The descriptor that remove() closes: the number a worker named, or the socket that clusterRawBind made.
+  get fd() {
+    const handle = this.handle;
+    return handle ? handle.fd : -1;
+  }
+
   remove(worker) {
     const workers = this.workers;
     if (!workers.has(worker.id)) return false;

src/js/internal/cluster/primary.ts

@@ -7,11 +7,13 @@ const { kHandle } = require("internal/shared");
 
 const sendHelper = $newRustFunction("node_cluster_binding.rs", "sendHelperPrimary", 4);
 const onInternalMessage = $newRustFunction("node_cluster_binding.rs", "onInternalMessagePrimary", 3);
-const { UV_EINVAL, UV_ENOBUFS } = process.binding("uv");
+const { UV_EEXIST, UV_EINVAL, UV_ENOBUFS } = process.binding("uv");
 
 let child_process;
 let RoundRobinHandle;
 let SharedHandle;
+let validateFd;
+let isFdOfDgramSocket;
 
 const ArrayPrototypeSlice = Array.prototype.slice;
 const ObjectValues = Object.values;
@@ -45,7 +47,7 @@ const SCHED_RR = 2;
 
 export default cluster;
 
-const handles = new Map();
+const handles = new Map<string, any>();
 cluster.isWorker = false;
 cluster.isMaster = true; // Deprecated alias. Must be same as isPrimary.
 cluster.isPrimary = true;
@@ -240,6 +242,21 @@ function exitedAfterDisconnect(worker, message) {
   send(worker, { ack: message.seq });
 }
 
+// A handle closes its descriptor when its last worker leaves, so a second handle for a held descriptor closes it under the holder. Looked up by number: `udp4` and `udp6` are two keys for one descriptor.
+function errnoOfFdQuery(message) {
+  const fd = message.fd;
+  const isUdp = message.addressType === "udp4" || message.addressType === "udp6";
+  let held = isUdp && (isFdOfDgramSocket ??= $newRustFunction("udp_socket.rs", "jsDgramIsFdAdopted", 1))(fd);
+  if (!held) {
+    handles.$forEach(handle => {
+      if (handle.fd === fd) held = true;
+    });
+  }
+  if (!held) return 0;
+  // The kind comes before the holder, as in node. Not as in node v26.3.0: EEXIST also under SCHED_NONE and for udp4 then udp6, where its worker dies on an assertion (nodejs/node#64869) or node serves and closes the number two times.
+  return (validateFd ??= $newRustFunction("node_cluster_binding.rs", "clusterValidateFd", 2))(fd, isUdp) || UV_EEXIST;
+}
+
 function queryServer(worker, message) {
   // Stop processing if worker already disconnecting
   if (worker.exitedAfterDisconnect) return;
@@ -294,6 +311,13 @@ function queryServer(worker, message) {
       worker.emit("error", error);
       return;
     }
+    if (typeof message.fd === "number" && message.fd >= 0) {
+      const errno = errnoOfFdQuery(message);
+      if (errno !== 0) {
+        send(worker, { errno, key, ack: message.seq, data: cachedHandle ? cachedHandle.data : message.data }, null);
+        return;
+      }
+    }
     if (
       schedulingPolicy !== SCHED_RR ||
       message.sharedOnly === true ||

src/runtime/node/node_cluster_binding.rs

@@ -731,6 +731,7 @@ pub(crate) fn cluster_raw_bind(global: &JSGlobalObject, frame: &CallFrame) -> Js
     }
 }
 
+/// `(fd, wantDgram)` → 0 when `fd` can serve a query of that kind (datagram, else stream), or a negative errno.
 #[bun_jsc::host_fn]
 pub(crate) fn cluster_validate_fd(global: &JSGlobalObject, frame: &CallFrame) -> JsResult<JSValue> {
     let _ = global;
@@ -741,12 +742,21 @@ pub(crate) fn cluster_validate_fd(global: &JSGlobalObject, frame: &CallFrame) ->
     #[cfg(not(windows))]
     {
         let fd = value.to_int32();
+        // `fd: 3.5` is not descriptor 3: https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L1904-L1911
+        if f64::from(fd) != value.as_number() {
+            return Ok(JSValue::js_number_from_int32(-bun_sys::UV_E::INVAL));
+        }
         if fd < 0 {
             return Ok(JSValue::js_number_from_int32(-bun_sys::UV_E::BADF));
         }
+        let wanted = if frame.argument(1).to_boolean() {
+            libc::SOCK_DGRAM
+        } else {
+            libc::SOCK_STREAM
+        };
         let mut ty: libc::c_int = 0;
         let mut len = core::mem::size_of::<libc::c_int>() as libc::socklen_t;
-        // SAFETY: plain getsockopt on a caller-supplied fd; out-params are
+        // SAFETY: plain getsockopt on a caller-supplied fd; out-params are live locals, and `len` is the size of `ty`.
         let rc = unsafe {
             libc::getsockopt(
                 fd,
@@ -756,8 +766,8 @@ pub(crate) fn cluster_validate_fd(global: &JSGlobalObject, frame: &CallFrame) ->
                 &raw mut len,
             )
         };
-        // node's createServerHandle: EINVAL for anything that cannot listen (e.g. a connected stdio socketpair), fd left untouched.
-        if rc != 0 || (ty != libc::SOCK_STREAM && ty != libc::SOCK_DGRAM) {
+        // A descriptor of another kind is EINVAL before node opens it: https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L1904-L1911 and https://github.com/nodejs/node/blob/v26.3.0/lib/internal/dgram.js#L67-L70
+        if rc != 0 || ty != wanted {
             return Ok(JSValue::js_number_from_int32(-bun_sys::UV_E::INVAL));
         }
         if ty == libc::SOCK_STREAM {
@@ -771,6 +781,21 @@ pub(crate) fn cluster_validate_fd(global: &JSGlobalObject, frame: &CallFrame) ->
             if connected {
                 return Ok(JSValue::js_number_from_int32(-bun_sys::UV_E::INVAL));
             }
+        } else {
+            // SAFETY: sockaddr_storage is plain data; getsockname only writes within `name_len`.
+            let family = unsafe {
+                let mut name: libc::sockaddr_storage = bun_core::ffi::zeroed_unchecked();
+                let mut name_len =
+                    core::mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
+                if libc::getsockname(fd, (&raw mut name).cast(), &raw mut name_len) != 0 {
+                    return Ok(JSValue::js_number_from_int32(-bun_sys::UV_E::INVAL));
+                }
+                libc::c_int::from(name.ss_family)
+            };
+            // https://github.com/nodejs/node/blob/v26.3.0/deps/uv/src/unix/tty.c#L458-L460
+            if family != libc::AF_INET && family != libc::AF_INET6 {
+                return Ok(JSValue::js_number_from_int32(-bun_sys::UV_E::INVAL));
+            }
         }
         Ok(JSValue::js_number_from_int32(0))
     }