usockets: keep the kernel poll registration in sync across us_poll_resize

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

@@ -584,12 +584,63 @@ 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));
 #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.
+     * Arming READABLE|WRITABLE unconditionally here desynced kernel vs poll
+     * state for a poll not watching both directions: the dispatcher masks
+     * delivered events with us_poll_events() but never deletes the filter, and
+     * us_poll_change cannot diff away a filter the poll state says was never
+     * armed, so a level-triggered EVFILT_READ with data or a FIN pending would
+     * re-fire on every kevent call forever.
+     * 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);
+    }
+    /* At 0 events the FIN-detector oneshot is the poll's only kernel presence;
+     * re-add it so its udata moves to new_p, matching what kqueue_change
+     * maintains for that state. */
+    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));
 #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);

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

@@ -655,6 +655,66 @@ 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 } = 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);
+        },
+        close() {},
+        error() {},
+      },
+    });
+
+    const client = net.connect({ port: server.port, host: "127.0.0.1", allowHalfOpen: true });
+    client.on("error", () => {});
+    await new Promise<void>(resolve => client.once("connect", resolve));
+    client.end(); // FIN; the server side stays half-open
+
+    const socket = await ended;
+    // Let the post-end writable dispatch drop the poll to zero events before adopting.
+    await new Promise(resolve => setImmediate(resolve));
+    await new Promise(resolve => setImmediate(resolve));
+
+    socket.upgradeTLS({
+      tls: { cert: tls.cert, key: tls.key },
+      isServer: true,
+      data: {},
+      socket: {
+        data() {},
+        end() {},
+        close() {
+          onTeardown("close");
+        },
+        error() {
+          onTeardown("error");
+        },
+      },
+    });
+
+    // The reset must reach the adopted socket (EPOLLHUP/EPOLLERR have no mask).
+    client.resetAndDestroy();
+    expect(["close", "error"]).toContain(await torndown);
+    expect(endCount).toBe(1);
+  });
+
   it("upgradeTLS handles errors", async () => {
     using server = Bun.serve({
       port: 0,