usockets: keep the kernel poll registration in sync across us_poll_resize

packages/bun-usockets/src/context.c

@@ -16,8 +16,10 @@
  */
 
 #include "internal/internal.h"
+#include "internal/fault_inject.h"
 #include "libusockets.h"
 #include <errno.h>
+#include <limits.h>
 #include <stdlib.h>
 #include <string.h>
 #ifndef _WIN32
@@ -302,6 +304,22 @@ struct us_socket_t *us_socket_adopt(struct us_socket_t *s, struct us_socket_grou
     struct us_connecting_socket_t *c = s->connect_state;
     struct us_socket_t *new_s = s;
     if (ext_size != -1) {
+#if defined(LIBUS_SOCKET_FAULT_INJECTION) && LIBUS_SOCKET_FAULT_INJECTION
+        /* US_FAULT_ADOPT_GROW (action "short", bytes = N) inflates the new ext
+         * size by N so us_poll_resize below takes its grow path, which no
+         * in-tree adopter reaches (all pass equal or smaller ext sizes). The
+         * SHORT action writes the rule's byte count through the clamp
+         * out-param; INT_MAX is the did-not-fire sentinel, unreachable as a
+         * real count because the JS setter requires bytes > 0 and the rule
+         * field is a plain int. Over-allocating is safe: us_calloc zeroes the
+         * tail and the memcpy in us_poll_resize copies old_size bytes. */
+        ssize_t fault_out_unused = 0;
+        int fault_grow_bytes = INT_MAX;
+        (void) US_FAULT_CHECK(US_FAULT_ADOPT_GROW, us_poll_fd(&s->p), fault_out_unused, fault_grow_bytes);
+        if (fault_grow_bytes != INT_MAX) {
+            ext_size += fault_grow_bytes;
+        }
+#endif
         struct us_poll_t *poll_ref = &s->p;
         new_s = (struct us_socket_t *) us_poll_resize(poll_ref, loop,
             sizeof(struct us_socket_t) - sizeof(struct us_poll_t) + old_ext_size,

packages/bun-usockets/src/eventing/epoll_kqueue.c

@@ -584,12 +584,81 @@ struct us_poll_t *us_poll_resize(struct us_poll_t *p, struct us_loop_t *loop, un
 
     int events = us_poll_events(p);
 #ifdef LIBUS_USE_EPOLL
-    /* Hack: forcefully update poll by stripping away already set events */
-    new_p->state.poll_type = us_internal_poll_type(new_p);
-    us_poll_change(new_p, loop, events);
+    /* Re-point the kernel's epitem at new_p directly instead of through
+     * us_poll_change, whose old==new diff would skip the epoll_ctl at
+     * events == 0 (a real steady state: half-open socket after on_end, see
+     * us_poll_start_rc) - and the MOD is what moves data.ptr off the old
+     * poll, which the caller frees. */
+    struct epoll_event event;
+    event.events = events;
+    if (!(events & LIBUS_SOCKET_READABLE) && !(events & LIBUS_SOCKET_WRITABLE)) {
+        /* See us_poll_start_rc: 0-event polls rely on implicit EPOLLHUP/EPOLLERR. */
+        event.events |= EPOLLHUP | EPOLLERR;
+    }
+    event.data.ptr = new_p;
+    int rc;
+    do {
+        rc = epoll_ctl(loop->fd, EPOLL_CTL_MOD, new_p->state.fd, &event);
+    } while (IS_EINTR(rc));
+    if (rc != 0) {
+        /* A failed MOD leaves the kernel's data.ptr on the old poll, which the
+         * caller frees: crash now rather than use-after-free later (the
+         * eventfd failure in us_internal_create_async sets the precedent). */
+        BUN_PANIC("us_poll_resize: epoll_ctl failed to re-register the poll");
+    }
 #else
-    /* Forcefully update poll by resetting them with new_p as user data */
-    kqueue_change(loop->fd, new_p->state.fd, 0, LIBUS_SOCKET_WRITABLE | LIBUS_SOCKET_READABLE, new_p);
+    /* Re-register each filter with new_p as udata (EV_ADD on an existing knote
+     * updates udata in place), arming exactly the poll's current interest: the
+     * registration must match us_poll_events, because the dispatcher masks
+     * delivered events with it but never deletes an over-armed filter, and
+     * us_poll_change cannot diff away a filter the poll state never armed.
+     * The deletes drop filters the poll does not want, so a stale FIN-detector
+     * oneshot (kqueue_change arms EVFILT_WRITE at 0 events, and that knote
+     * survives a later 0 -> READABLE transition) cannot keep the old poll as
+     * udata past its free. EV_DELETE of an absent filter reports ENOENT, and
+     * on FreeBSD the first error aborts the rest of the changelist (the
+     * kevent64 shim passes no eventlist), so the EV_ADDs - the udata move -
+     * go first and the one possible EV_DELETE comes last. */
+    struct kevent64_s change_list[2];
+    int change_length = 0;
+    if (events & LIBUS_SOCKET_READABLE) {
+        EV_SET64(&change_list[change_length++], new_p->state.fd, EVFILT_READ,
+            EV_ADD, 0, 0, (uint64_t)(void *)new_p, 0, 0);
+    }
+    /* A 0-event poll may still have a pending FIN-detector oneshot in the
+     * kernel (kqueue_change arms one on transitions to 0 events; delivery
+     * consumes it). EV_ADD moves a pending knote's udata off the soon-freed
+     * old poll, and re-arms the detector when it was already consumed. */
+    if ((events & LIBUS_SOCKET_WRITABLE) || events == 0) {
+        EV_SET64(&change_list[change_length++], new_p->state.fd, EVFILT_WRITE,
+            EV_ADD | EV_ONESHOT, 0, 0, (uint64_t)(void *)new_p, 0, 0);
+    }
+    if (!(events & LIBUS_SOCKET_READABLE)) {
+        EV_SET64(&change_list[change_length++], new_p->state.fd, EVFILT_READ,
+            EV_DELETE, 0, 0, (uint64_t)(void *)new_p, 0, 0);
+    }
+    if ((events & LIBUS_SOCKET_READABLE) && !(events & LIBUS_SOCKET_WRITABLE)) {
+        EV_SET64(&change_list[change_length++], new_p->state.fd, EVFILT_WRITE,
+            EV_DELETE, 0, 0, (uint64_t)(void *)new_p, 0, 0);
+    }
+    int ret;
+    do {
+        ret = kevent64(loop->fd, change_list, change_length, change_list, change_length, KEVENT_FLAG_ERROR_EVENTS, NULL);
+    } while (IS_EINTR(ret));
+    /* A change that failed to apply leaves the kernel referencing the old
+     * poll, which the caller frees: crash now rather than use-after-free
+     * later. ENOENT is the trailing EV_DELETE of an absent filter, the one
+     * benign failure; on FreeBSD it surfaces as ret < 0 (the shim passes no
+     * eventlist) after every EV_ADD before it has already applied. */
+    if (ret < 0 && errno != ENOENT) {
+        BUN_PANIC("us_poll_resize: kevent failed to re-register the poll");
+    }
+    for (int i = 0; i < ret; i++) {
+        if ((change_list[i].flags & EV_ERROR) && change_list[i].data != 0 &&
+            change_list[i].data != ENOENT) {
+            BUN_PANIC("us_poll_resize: kevent failed to re-register the poll");
+        }
+    }
 #endif
     /* This is needed for epoll also (us_change_poll doesn't update the old poll) */
     us_internal_loop_update_pending_ready_polls(loop, p, new_p, events, events);

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

@@ -54,6 +54,13 @@ enum us_fault_syscall {
      * US_FAULT_ERRNO applies, and the errno value is ignored — the simulated
      * failure is a thrown JS out-of-memory error, not an errno. */
     US_FAULT_SESSION_BUFFER,
+    /* Not a fault: inflates the adopted ext size in us_socket_adopt by the
+     * rule's clamp_bytes so us_poll_resize takes its grow path (reallocate,
+     * re-register the kernel poll under the new pointer, retire the old
+     * socket). Every in-tree adopter passes an equal or smaller ext size, so
+     * that path is unreachable without injection. Only US_FAULT_SHORT
+     * applies, reusing clamp_bytes as the number of bytes to add. */
+    US_FAULT_ADOPT_GROW,
     US_FAULT_COUNT
 };
 

src/js/internal-for-testing.ts

@@ -345,8 +345,8 @@ export const setSocketOptions: setSocketOptionsFn = $newRustFunction(
 /**
  * The syscalls instrumented in bsd.c, plus non-syscall hooks whose failure
  * paths are otherwise unreachable without injection ("ssl_loop_buffer",
- * "poll_start", "session_buffer"; see fault_inject.h for the per-hook
- * description). Arming anything else is rejected.
+ * "poll_start", "session_buffer", "adopt_grow"; see fault_inject.h for the
+ * per-hook description). Arming anything else is rejected.
  */
 export type SocketFaultSyscall =
   | "recv"
@@ -358,7 +358,8 @@ export type SocketFaultSyscall =
   | "accept"
   | "ssl_loop_buffer"
   | "poll_start"
-  | "session_buffer";
+  | "session_buffer"
+  | "adopt_grow";
 
 export type SocketFaultRule = {
   syscall: SocketFaultSyscall;
@@ -379,7 +380,7 @@ export type SocketFaultRule = {
     | "ENETUNREACH"
     | "EHOSTUNREACH"
     | number;
-  /** clamp recv/send length to this many bytes; required and > 0 when action === "short" */
+  /** clamp recv/send length to this many bytes, or the ext growth for "adopt_grow"; required and > 0 when action === "short" */
   bytes?: number;
   /** skip the first N matching calls before triggering. Default 0. */
   after?: number;

src/runtime/socket/socket_body.rs

@@ -5098,11 +5098,13 @@ pub mod testing_apis {
                 fi::POLL_START
             } else if syscall_str.eql_comptime(b"session_buffer") {
                 fi::SESSION_BUFFER
+            } else if syscall_str.eql_comptime(b"adopt_grow") {
+                fi::ADOPT_GROW
             } else {
                 // socket/close/shutdown have enum slots but no bsd.c hooks;
                 // accepting them would arm rules that can never fire.
                 return Err(global.throw(format_args!(
-                    "rule.syscall must be one of: recv, send, writev, sendmsg, recvmsg, connect, accept, ssl_loop_buffer, poll_start, session_buffer"
+                    "rule.syscall must be one of: recv, send, writev, sendmsg, recvmsg, connect, accept, ssl_loop_buffer, poll_start, session_buffer, adopt_grow"
                 )));
             };
 
@@ -5131,11 +5133,24 @@ pub mod testing_apis {
                 )));
             };
 
-            // "short" clamps a byte count, which only recv/send have; arming it
-            // on any other syscall would silently never fire.
-            if action == fi::ACTION_SHORT && syscall != fi::RECV && syscall != fi::SEND {
+            // "short" carries a byte count, which only recv/send (clamp) and
+            // adopt_grow (growth) have; arming it on any other syscall would
+            // silently never fire.
+            if action == fi::ACTION_SHORT
+                && !matches!(syscall, fi::RECV | fi::SEND | fi::ADOPT_GROW)
+            {
+                return Err(global.throw(format_args!(
+                    "rule.action \"short\" is only supported for syscall \"recv\", \"send\" or \"adopt_grow\""
+                )));
+            }
+
+            // adopt_grow has exactly one meaningful action: "short" with the
+            // byte count to add. The other actions would arm a rule whose
+            // effect (errno/zero short-circuit) the adopt hook never reads.
+            if syscall == fi::ADOPT_GROW && action != fi::ACTION_SHORT && action != fi::ACTION_NONE
+            {
                 return Err(global.throw(format_args!(
-                    "rule.action \"short\" is only supported for syscall \"recv\" or \"send\""
+                    "rule.action must be \"short\" for syscall \"adopt_grow\""
                 )));
             }
 

src/uws_sys/lib.rs

@@ -418,6 +418,10 @@ pub mod fault_inject {
     /// Not a syscall: the JS `Buffer` allocated for a TLS session/keylog
     /// payload in the `on_session`/`on_keylog` dispatch.
     pub const SESSION_BUFFER: c_int = 12;
+    /// Not a fault: inflates the adopted ext size in `us_socket_adopt` by the
+    /// rule's byte count (`bytes`, action "short") so `us_poll_resize` takes
+    /// its otherwise-unreachable grow path.
+    pub const ADOPT_GROW: c_int = 13;
 
     pub const ACTION_NONE: c_int = 0;
     pub const ACTION_ERRNO: c_int = 1;

test/js/bun/net/socket.test.ts

@@ -655,6 +655,152 @@ describe.concurrent("socket", () => {
     }
   });
 
+  // Adoption (us_socket_adopt -> us_poll_resize) can run after the peer's FIN
+  // was already consumed by end() and the write buffer drained, i.e. with the
+  // poll watching neither direction and relying on implicit EPOLLHUP/EPOLLERR.
+  // The kernel registration must follow the adopted socket so the peer's later
+  // reset dispatches to the live poll, exactly once, instead of touching a
+  // freed one. Linux-only: the zero-event steady state is epoll's (kqueue
+  // keeps no kernel filter on such a socket, so the reset goes unseen there).
+  it.skipIf(!isLinux)("upgradeTLS after the peer half-closed survives a subsequent reset", async () => {
+    const { promise: ended, resolve: onEnd, reject: onEndFail } = Promise.withResolvers<Socket<undefined>>();
+    const { promise: torndown, resolve: onTeardown } = Promise.withResolvers<string>();
+    let endCount = 0;
+
+    using server = Bun.listen({
+      hostname: "127.0.0.1",
+      port: 0,
+      allowHalfOpen: true,
+      socket: {
+        open() {},
+        data() {},
+        end(socket) {
+          endCount++;
+          onEnd(socket);
+        },
+        // Latched by the end() resolve on the expected path; these only fire
+        // it when the socket dies early, turning a would-be hang into a
+        // diagnosable failure.
+        close() {
+          onEndFail(new Error("server socket closed before end fired"));
+        },
+        error(_socket, error) {
+          onEndFail(error);
+        },
+      },
+    });
+
+    const client = net.connect({ port: server.port, host: "127.0.0.1", allowHalfOpen: true });
+    client.on("error", () => {}); // post-connect errors (the reset) are expected
+    await new Promise<void>((resolve, reject) => {
+      client.once("connect", resolve);
+      client.once("error", reject);
+    });
+    client.end(); // FIN; the server side stays half-open
+
+    const socket = await ended;
+    // end() leaves the poll armed for WRITABLE only; the next writable
+    // dispatch finds nothing buffered and drops it to zero events. One
+    // event-loop turn suffices, and the state is stable once reached (nothing
+    // re-arms the poll without a JS-side write), so the extra turns are slack
+    // for scheduling differences, not a timing condition.
+    for (let i = 0; i < 10; i++) {
+      await new Promise(resolve => setImmediate(resolve));
+    }
+
+    // Grow the adopted ext so us_poll_resize actually reallocates and must
+    // re-register the kernel poll under the new pointer (no in-tree adopter
+    // grows on its own). The rule is armed and consumed inside this
+    // synchronous block, so concurrent tests cannot hit it.
+    if (socketFaultInjection.available()) {
+      socketFaultInjection.set({ syscall: "adopt_grow", action: "short", bytes: 512, repeat: 1 });
+    }
+    try {
+      socket.upgradeTLS({
+        tls: { cert: tls.cert, key: tls.key },
+        isServer: true,
+        data: {},
+        socket: {
+          data() {},
+          end() {},
+          close() {
+            onTeardown("close");
+          },
+          error() {
+            onTeardown("error");
+          },
+        },
+      });
+    } finally {
+      if (socketFaultInjection.available()) {
+        socketFaultInjection.clear();
+      }
+    }
+
+    // Let the loop retire the replaced socket (freed at the outermost
+    // loop_post) before the peer resets, so the reset must find the live
+    // registration rather than racing the retirement.
+    for (let i = 0; i < 10; i++) {
+      await new Promise(resolve => setImmediate(resolve));
+    }
+
+    // The reset must reach the adopted socket (EPOLLHUP/EPOLLERR have no mask).
+    client.resetAndDestroy();
+    expect(["close", "error"]).toContain(await torndown);
+    expect(endCount).toBe(1);
+  });
+
+  // Same forced reallocation on a socket that is actively polling readable: a
+  // full TLS handshake and an echo must flow through the relocated socket
+  // (covers the grow path's normal-interest re-registration on every backend).
+  it.skipIf(!socketFaultInjection.available())(
+    "upgradeTLS survives a forced poll reallocation (handshake + echo)",
+    async () => {
+      const { promise: echoed, resolve: onEcho, reject: onEchoFail } = Promise.withResolvers<string>();
+
+      using server = Bun.listen({
+        hostname: "127.0.0.1",
+        port: 0,
+        socket: {
+          open(socket) {
+            socketFaultInjection.set({ syscall: "adopt_grow", action: "short", bytes: 512, repeat: 1 });
+            try {
+              socket.upgradeTLS({
+                tls: { cert: tls.cert, key: tls.key },
+                isServer: true,
+                data: {},
+                socket: {
+                  data(tlsSocket, chunk) {
+                    tlsSocket.write(chunk);
+                  },
+                  close() {},
+                  error(_socket, error) {
+                    onEchoFail(error);
+                  },
+                },
+              });
+            } finally {
+              socketFaultInjection.clear();
+            }
+          },
+          data() {},
+          close() {},
+          error() {},
+        },
+      });
+
+      const client = tlsConnect({ port: server.port, host: "127.0.0.1", rejectUnauthorized: false }, () => {
+        client.write("ping");
+      });
+      client.on("error", onEchoFail);
+      client.on("data", data => {
+        onEcho(String(data));
+        client.end();
+      });
+      expect(await echoed).toBe("ping");
+    },
+  );
+
   it("upgradeTLS handles errors", async () => {
     using server = Bun.serve({
       port: 0,