Windows: dispatch libuv events from the event loop, not from inside libuv callbacks

packages/bun-usockets/src/internal/eventing/libuv.h

@@ -27,24 +27,70 @@
 /* Defined in eventing/libuv.c; used by the sweep escalation in loop.c. */
 int us_internal_libuv_peer_reset_probe(LIBUS_SOCKET_DESCRIPTOR fd);
 
+struct us_poll_t;
+
 struct us_loop_t {
   alignas(LIBUS_EXT_ALIGNMENT) struct us_internal_loop_data_t data;
 
   uv_loop_t *uv_loop;
   int is_default;
 
-  uv_prepare_t *uv_pre;
-  uv_check_t *uv_check;
+  /* libuv is only the readiness source here, the way epoll/kqueue are on the
+   * other backend: the callbacks it runs inside uv_run (poll_cb, timer_cb,
+   * async_cb) do nothing but link the poll into this list. us_loop_run and
+   * us_loop_pump dispatch it once uv_run has returned, so no libuv frame is
+   * ever on the stack below a socket, timer or wakeup handler. That is what
+   * makes it sound for a handler to drive the loop again (waitForPromise) or
+   * to close any handle: uv_run is not re-entrant, and closing a handle whose
+   * libuv dispatch frame is still live corrupts the loop once a nested run
+   * completes the close. The list is intrusive (a poll is in it at most once,
+   * later reports for the same poll merge into its entry) and loop-wide, so a
+   * nested dispatch keeps draining what an outer uv_run collected. */
+  struct us_poll_t *ready_head;
+  struct us_poll_t *ready_tail;
+
+  /* Unref'd timer that bounds how long uv_run may park, so a tick can take a
+   * timeout the way epoll_wait/kevent do (us_loop_run_with_timeout). Its
+   * callback does nothing; expiring is enough to end the poll phase. */
+  uv_timer_t *deadline_timer;
+
+  /* Non-zero while this loop's uv_run is on the stack, i.e. while whatever
+   * runs is running inside a libuv callback. Ticking the loop from there is a
+   * nested uv_run, which libuv does not support; libuv callbacks record and
+   * defer (see the top of libuv.c) so that never happens. */
+  int in_uv_run;
+
+  /* For a uv loop this us_loop created itself: head and tail of what Bun's
+   * own libuv callbacks (pipes, processes, files, dns, ...) deferred during
+   * uv_run - the Rust-side counterpart of the ready list
+   * (src/libuv_sys/deferred.rs), reached through uv_loop->data. */
+  void *deferred[2];
 };
 
-// it is no longer valid to cast a pointer to us_poll_t to a pointer of
-// uv_poll_t
+/* Not castable to uv_poll_t: the libuv handle is a separate allocation so the
+ * poll block can be resized (us_poll_resize) while the handle stays put. */
 struct us_poll_t {
-  /* We need to hold a pointer to this uv_poll_t since we need to be able to
-   * resize our block */
+  /* NULL once the poll no longer owns a handle: us_poll_stop handed it to
+   * uv_close (libuv frees it in the close callback), or us_poll_resize moved
+   * it to the replacement block. */
   uv_poll_t *uv_p;
+  struct us_loop_t *loop;
   LIBUS_SOCKET_DESCRIPTOR fd;
   unsigned char poll_type;
+  /* Linked into loop->ready_head. ready_status/ready_events accumulate what
+   * libuv reported since the poll was last dispatched: the first non-zero
+   * status, and the union of the event bits. */
+  unsigned char ready;
+  int ready_status;
+  int ready_events;
+  struct us_poll_t *ready_prev, *ready_next;
 };
 
-#endif // LIBUV_H
+/* One non-blocking tick regardless of whether libuv considers the loop alive,
+ * and one tick parked for at most timeout_ms (< 0 unbounded, 0 = pump); see
+ * libuv.c. us_loop_run (libusockets.h) is the unbounded form. */
+void us_loop_pump(struct us_loop_t *loop);
+void us_loop_run_with_timeout(struct us_loop_t *loop, long long timeout_ms);
+int us_loop_in_uv_run(struct us_loop_t *loop);
+
+#endif // LIBUV_H

packages/bun-usockets/src/internal/loop_data.h

@@ -88,10 +88,11 @@ struct us_internal_loop_data_t {
     /* We do not care if this flips or not, it doesn't matter */
     size_t iteration_nr;
     void* jsc_vm;
-    /* Reentrancy depth of us_loop_run_bun_tick. When >1, we are inside a
-     * nested tick (e.g. waitForPromise from a poll callback). Freeing closed
-     * sockets must be deferred to the outermost tick so the outer dispatch
-     * doesn't read a freed poll. */
+    /* Reentrancy depth of the tick (us_loop_run_bun_tick; us_loop_run /
+     * us_loop_pump on libuv). When >1, we are inside a nested tick (e.g.
+     * waitForPromise from a socket handler). Freeing closed sockets must be
+     * deferred to the outermost tick so the outer dispatch doesn't read a
+     * freed poll. */
     int tick_depth;
 };
 

packages/bun-usockets/src/loop.c

@@ -415,10 +415,10 @@ void us_internal_loop_post(struct us_loop_t *loop) {
 #endif
     if (loop->data.nq_head) us_nq_loop_flush_if_pending(loop);
     /* A poll callback may re-enter the loop (e.g. expect().toThrow() →
-     * waitForPromise → us_loop_run_bun_tick). The inner tick must not free
-     * closed sockets: the outer tick's dispatch is mid-iteration and may still
-     * hold a pointer to one (it reads s->flags right after on_data returns).
-     * Defer to the outermost tick's loop_post. */
+     * waitForPromise → us_loop_run_bun_tick / us_loop_run). The inner tick
+     * must not free closed sockets: the outer tick's dispatch is
+     * mid-iteration and may still hold a pointer to one (it reads s->flags
+     * right after on_data returns). Defer to the outermost tick's loop_post. */
     if (loop->data.tick_depth <= 1) {
         us_internal_free_closed_sockets(loop);
     }

packages/bun-usockets/src/socket.c

@@ -792,7 +792,8 @@ unsigned int us_get_local_address_info(char *buf, struct us_socket_t *s, const c
 
 void us_socket_ref(struct us_socket_t *s) {
 #ifdef LIBUS_USE_LIBUV
-    uv_ref((uv_handle_t *) s->p.uv_p);
+    /* A closed (or relocated) socket no longer owns a libuv handle. */
+    if (s->p.uv_p) uv_ref((uv_handle_t *) s->p.uv_p);
 #endif
     // do nothing if not using libuv
 }
@@ -837,7 +838,7 @@ int us_socket_keepalive(us_socket_r s, int enabled, unsigned int delay) {
 
 void us_socket_unref(struct us_socket_t *s) {
 #ifdef LIBUS_USE_LIBUV
-    uv_unref((uv_handle_t *) s->p.uv_p);
+    if (s->p.uv_p) uv_unref((uv_handle_t *) s->p.uv_p);
 #endif
     // do nothing if not using libuv
 }

src/io/MaxBuf.rs

@@ -145,13 +145,27 @@ impl MaxBuf {
     /// `owned_by_reader` is set, which every caller has just checked via
     /// `Some(maxbuf)`).
     pub(crate) fn on_read_bytes(this: NonNull<MaxBuf>, bytes: u64) -> bool {
+        if !Self::charge(this, bytes) {
+            return false;
+        }
+        Self::overflowed(this)
+    }
+
+    /// Charges `bytes` against the budget; `true` once it is overdrawn. Pure
+    /// bookkeeping (no callback), for read completions that are recorded
+    /// inside the poll backend and dispatched later; pair with [`overflowed`].
+    pub(crate) fn charge(this: NonNull<MaxBuf>, bytes: u64) -> bool {
         let mb = Self::live(&this);
         let delta = i64::try_from(bytes).unwrap_or(0);
         let remaining = mb.remaining_bytes.get().checked_sub(delta).unwrap_or(-1);
         mb.remaining_bytes.set(remaining);
-        if remaining >= 0 {
-            return false;
-        }
+        remaining < 0
+    }
+
+    /// Tells the owning subprocess its `maxBuffer` was overdrawn (it kills the
+    /// child). `true` if there was an owner to tell.
+    pub(crate) fn overflowed(this: NonNull<MaxBuf>) -> bool {
+        let mb = Self::live(&this);
         let Some(owner) = mb.owned_by_subprocess.get() else {
             return false;
         };