Repository: denoland/deno
Pull request: fix(ext/node): drain all ready connections in poll_tcp_handle to prevent accept starvation

## Summary

- Fix `node:net` accept starvation under high concurrency when mixed with Deno-native APIs (`Deno.listenTls()`, `Deno.connect()`)
- `poll_tcp_handle` now drains all ready connections from `poll_accept` in a loop before firing callbacks, instead of accepting only one at a time
- Fires the connection callback once per accepted stream (matching libuv semantics), with a safety check that stops if the callback doesn't call `uv_accept`

## Root cause

Under high concurrent load, `poll_tcp_handle` previously accepted one connection, fired the JS callback, and moved on. The callback's async work (e.g., `Deno.listenTls`) created pending tokio ops processed in Phase 2 of the event loop, delaying the next Phase 4 (I/O) iteration. Since tokio's reactor only re-polls epoll when the event loop returns `Pending`, connections that arrived in the same epoll batch were invisible until the next reactor turn — creating a feedback loop where accept throughput degraded to zero.

Closes #32941

## Test plan

- [x] New test `tcp_batch_accept_concurrent` — opens 8 concurrent connections before ticking the event loop, verifies all 8 are accepted in a single tick
- [x] All 59 `uv_compat` tests pass (58 existing + 1 new)
- [x] `./x fmt` and `./x lint` pass

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Changed files:
filename: libs/core/uv_compat/tcp.rs
patch: @@ -606,34 +606,41 @@ pub(crate) unsafe fn poll_tcp_handle(
   }
 
   // 2. Poll listener for new connections.
-  //    Match libuv: accept one connection at a time. Only poll when the
-  //    backlog is empty. If the user doesn't call uv_accept in the
-  //    callback, we stop polling to avoid spinning.
-  //    Poll with a short-lived borrow on the listener, drop it before the
-  //    connection callback.
+  //    Drain all ready connections from poll_accept before firing callbacks.
+  //    This prevents starvation under high concurrency: tokio's reactor
+  //    only re-polls epoll when poll_event_loop_inner returns Pending, so
+  //    connections that arrived during the same epoll batch must be accepted
+  //    now or they'll be invisible until the next reactor turn.
+  //    After draining, fire the connection callback once per accepted stream
+  //    (libuv fires the callback once per connection). If the user doesn't
+  //    call uv_accept in the callback, stop firing to avoid spinning.
   unsafe {
-    let accepted = if (*tcp_ptr).internal_listener.is_some()
+    if (*tcp_ptr).internal_listener.is_some()
       && (*tcp_ptr).internal_connection_cb.is_some()
       && (*tcp_ptr).internal_backlog.is_empty()
     {
       let listener = (*tcp_ptr).internal_listener.as_ref().unwrap();
-      if let Poll::Ready(Ok((stream, _))) = listener.poll_accept(cx) {
-        Some(stream)
-      } else {
-        None
+      while let Poll::Ready(Ok((stream, _))) = listener.poll_accept(cx) {
+        (*tcp_ptr).internal_backlog.push_back(stream);
+        any_work = true;
       }
-    } else {
-      None
-    };
-    // Listener borrow dropped.
-    if let Some(stream) = accepted {
-      (*tcp_ptr).internal_backlog.push_back(stream);
-      any_work = true;
     }
-    if !(*tcp_ptr).internal_backlog.is_empty()
+    // Fire the connection callback once per backlog entry.
+    // After each callback, re-check that the user called uv_accept
+    // (which drains one entry). If backlog didn't shrink, the user
+    // isn't consuming connections — stop to avoid infinite loop.
+    while !(*tcp_ptr).internal_backlog.is_empty()
       && let Some(cb) = (*tcp_ptr).internal_connection_cb
     {
+      let before = (*tcp_ptr).internal_backlog.len();
       cb(tcp_ptr as *mut uv_stream_t, 0);
+      let after = (*tcp_ptr).internal_backlog.len();
+      if after < before {
+        any_work = true;
+      } else {
+        // If uv_accept wasn't called, backlog didn't shrink — stop.
+        break;
+      }
     }
   }
 

filename: libs/core/uv_compat/tests.rs
patch: @@ -2327,3 +2327,128 @@ async fn uv_write_callback_is_deferred() {
   })
   .await;
 }
+
+// ========== TCP batch accept ==========
+
+/// Verify that multiple connections queued before a tick are all accepted
+/// in a single event loop iteration. This prevents starvation when the
+/// connection handler does async work (e.g., Deno.listenTls) that delays
+/// returning to the I/O phase.
+#[tokio::test(flavor = "current_thread")]
+async fn tcp_batch_accept() {
+  run_test(async |runtime, uv_loop| {
+    // Shared state passed through server.data: a counter and a Vec of
+    // heap-allocated accepted client handles (so uv_close's deferred
+    // processing doesn't hit a dangling stack pointer).
+    struct AcceptState {
+      count: Cell<u32>,
+      clients: RefCell<Vec<*mut uv_tcp_t>>,
+    }
+    let state = Rc::new(AcceptState {
+      count: Cell::new(0),
+      clients: RefCell::new(Vec::new()),
+    });
+    let state_ptr = Rc::into_raw(state.clone());
+
+    let mut server = std::mem::MaybeUninit::<uv_tcp_t>::uninit();
+    let server_ptr = server.as_mut_ptr();
+
+    unsafe extern "C" fn on_connection(
+      server: *mut uv_stream_t,
+      status: i32,
+    ) {
+      unsafe {
+        assert_eq!(status, 0);
+        let state_ptr = (*server).data as *const AcceptState;
+        let state = &*state_ptr;
+
+        // Heap-allocate the client handle so it stays valid until
+        // uv_close processes it in the close phase.
+        let client_ptr = Box::into_raw(Box::new(
+          std::mem::MaybeUninit::<uv_tcp_t>::uninit(),
+        )) as *mut uv_tcp_t;
+        let loop_ = (*(server as *const uv_tcp_t)).loop_;
+        uv_tcp_init(loop_, client_ptr);
+        let rc = uv_accept(server, client_ptr as *mut uv_stream_t);
+        assert_eq!(rc, 0);
+        state.count.set(state.count.get() + 1);
+
+        // Track the heap pointer so the outer scope can close + free them.
+        state.clients.borrow_mut().push(client_ptr);
+      }
+    }
+
+    let server_port: u16;
+    unsafe {
+      uv_tcp_init(uv_loop, server_ptr);
+      (*server_ptr).data = state_ptr as *mut c_void;
+
+      let mut addr = std::mem::MaybeUninit::<sockaddr_in>::uninit();
+      let ip = std::ffi::CString::new("127.0.0.1").unwrap();
+      uv_ip4_addr(ip.as_ptr(), 0, addr.as_mut_ptr());
+      assert_ok(uv_tcp_bind(
+        server_ptr,
+        addr.as_ptr() as *const c_void,
+        0,
+        0,
+      ));
+      assert_ok(uv_listen(
+        server_ptr as *mut uv_stream_t,
+        128,
+        Some(on_connection),
+      ));
+
+      let mut name = std::mem::MaybeUninit::<sockaddr_in>::zeroed();
+      let mut namelen = std::mem::size_of::<sockaddr_in>() as i32;
+      uv_tcp_getsockname(
+        server_ptr,
+        name.as_mut_ptr() as *mut c_void,
+        &mut namelen,
+      );
+      server_port = u16::from_be(name.assume_init_ref().sin_port);
+    }
+
+    // Open multiple connections BEFORE ticking the event loop so they all
+    // land in the same poll_accept batch.
+    const NUM_CONNECTIONS: u32 = 8;
+    let mut client_streams = Vec::new();
+    for _ in 0..NUM_CONNECTIONS {
+      let stream =
+        tokio::net::TcpStream::connect(format!("127.0.0.1:{}", server_port))
+          .await
+          .unwrap();
+      client_streams.push(stream);
+    }
+
+    // A single tick should accept all connections thanks to batch-accept.
+    tick(runtime).await;
+
+    let count = state.count.get();
+    assert_eq!(
+      count, NUM_CONNECTIONS,
+      "Expected all {NUM_CONNECTIONS} connections to be accepted in one tick, got {count}"
+    );
+
+    // Cleanup: close all accepted client handles, then the server.
+    drop(client_streams);
+    unsafe {
+      for client_ptr in state.clients.borrow().iter() {
+        uv_close(*client_ptr as *mut uv_handle_t, None);
+      }
+      uv_close(server_ptr as *mut uv_handle_t, None);
+    }
+    // Tick to process deferred closes.
+    tick(runtime).await;
+
+    // Free heap-allocated client handles now that close phase is done.
+    unsafe {
+      for client_ptr in state.clients.borrow().iter() {
+        drop(Box::from_raw(
+          *client_ptr as *mut std::mem::MaybeUninit<uv_tcp_t>,
+        ));
+      }
+      Rc::from_raw(state_ptr);
+    }
+  })
+  .await;
+}