usockets: report a peer reset on a paused socket on kqueue

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

@@ -574,55 +574,62 @@ void us_internal_loop_update_pending_ready_polls(struct us_loop_t *loop, struct
 /* Poll */
 
 #ifdef LIBUS_USE_KQUEUE
-/* Helper function for setting or updating EVFILT_READ and EVFILT_WRITE */
-int kqueue_change(int kqfd, int fd, int old_events, int new_events, void *user_data) {
-    struct kevent64_s change_list[2];
+static int kqueue_is_socket_poll(struct us_poll_t *p) {
+    int type = us_internal_poll_type(p);
+    return type == POLL_TYPE_SOCKET || type == POLL_TYPE_SOCKET_SHUT_DOWN;
+}
+
+/* Registers the difference between old_events and new_events for fd.
+ *
+ * With keep_read_knote (us_poll_change passes it for socket polls) the EVFILT_READ knote
+ * stays registered while the socket does not poll for reads (paused, half-open after the
+ * peer's FIN, shut down with reads off, parked as low priority), re-added with EV_CLEAR.
+ * It is kqueue's stand-in for epoll's implicit EPOLLHUP/EPOLLERR: the peer's FIN or RST
+ * still reaches the dispatcher as eof/error (us_poll_events masks the readable bit out),
+ * and EV_CLEAR keeps unread data or a consumed EOF from re-firing every tick. Nothing else
+ * reports them: the one-shot write filter is consumed by the first, immediate, writable
+ * event, so a reset of a paused socket went unreported until resume(), unlike on epoll
+ * and libuv. EV_ADD on an existing knote updates its udata but keeps its flags, so each
+ * switch between the two modes deletes the knote and adds a new one; us_poll_resize relies
+ * on the same rule to move the udata without changing the mode. */
+int kqueue_change(int kqfd, int fd, int old_events, int new_events, void *user_data, int keep_read_knote) {
+    struct kevent64_s change_list[3];
     int change_length = 0;
 
-    /* Do they differ in readable? */
-    int is_readable =  (new_events & LIBUS_SOCKET_READABLE);
-    int is_writable =  (new_events & LIBUS_SOCKET_WRITABLE);
-    if ((new_events & LIBUS_SOCKET_READABLE) != (old_events & LIBUS_SOCKET_READABLE)) {
-        EV_SET64(&change_list[change_length++], fd, EVFILT_READ, is_readable ? EV_ADD : EV_DELETE, 0, 0, (uint64_t)(void*)user_data, 0, 0);
+    int is_readable = (new_events & LIBUS_SOCKET_READABLE);
+    if (is_readable != (old_events & LIBUS_SOCKET_READABLE)) {
+        if (keep_read_knote) {
+            EV_SET64(&change_list[change_length++], fd, EVFILT_READ, EV_DELETE, 0, 0, 0, 0, 0);
+            EV_SET64(&change_list[change_length++], fd, EVFILT_READ, EV_ADD | (is_readable ? 0 : EV_CLEAR), 0, 0, (uint64_t)(void*)user_data, 0, 0);
+        } else {
+            EV_SET64(&change_list[change_length++], fd, EVFILT_READ, is_readable ? EV_ADD : EV_DELETE, 0, 0, (uint64_t)(void*)user_data, 0, 0);
+        }
     }
 
-    if(!is_readable && !is_writable) {
-        if(!(old_events & LIBUS_SOCKET_WRITABLE)) {
-            // if we are not reading or writing, we need to add writable to receive FIN
-            EV_SET64(&change_list[change_length++], fd, EVFILT_WRITE, EV_ADD | EV_ONESHOT, 0, 0, (uint64_t)(void*)user_data, 0, 0);
-        }
-    } else if ((new_events & LIBUS_SOCKET_WRITABLE) != (old_events & LIBUS_SOCKET_WRITABLE)) {
-        /* Do they differ in writable? */
+    if ((new_events & LIBUS_SOCKET_WRITABLE) != (old_events & LIBUS_SOCKET_WRITABLE)) {
         EV_SET64(&change_list[change_length++], fd, EVFILT_WRITE, (new_events & LIBUS_SOCKET_WRITABLE) ? EV_ADD | EV_ONESHOT : EV_DELETE, 0, 0, (uint64_t)(void*)user_data, 0, 0);
     }
+
     int ret;
     do {
         ret = kevent64(kqfd, change_list, change_length, change_list, change_length, KEVENT_FLAG_ERROR_EVENTS, NULL);
     } while (IS_EINTR(ret));
 
-    // ret should be 0 in most cases (not guaranteed when removing async)
-
-    /* KEVENT_FLAG_ERROR_EVENTS reports per-filter failures as EV_ERROR entries
-     * with the errno in .data; kevent64 itself returns the count and does not
-     * set errno. Mirror epoll's contract so us_poll_start_rc callers can read it. */
-    if (ret > 0) {
-        errno = (int) change_list[0].data;
+    /* KEVENT_FLAG_ERROR_EVENTS reports each failed change as an EV_ERROR entry with the
+     * errno in .data while the other changes still apply; kevent64 itself returns their
+     * count and does not set errno. A delete of a filter that is not registered (a write
+     * one-shot the kernel already consumed) leaves the fd in the state asked for and is not
+     * a failure. Anything else mirrors epoll's contract so us_poll_start_rc callers can
+     * read errno. */
+    for (int i = 0; i < ret; i++) {
+        if ((change_list[i].flags & EV_DELETE) && change_list[i].data == ENOENT) {
+            continue;
+        }
+        errno = (int) change_list[i].data;
+        return 1;
     }
 
-    return ret;
-}
-
-/* Kqueue's stand-in for epoll's implicit EPOLLHUP/EPOLLERR: an EV_CLEAR read
- * filter re-fires on peer FIN/RST but not forever on a consumed EOF; the
- * dispatcher masks its readable bit out via us_poll_events (no reads). */
-void us_internal_kqueue_socket_arm_read_sentinel(struct us_socket_t *s) {
-    struct us_loop_t *loop = s->group->loop;
-    struct kevent64_s event;
-    EV_SET64(&event, us_poll_fd(&s->p), EVFILT_READ, EV_ADD | EV_CLEAR, 0, 0, (uint64_t)(void *)&s->p, 0, 0);
-    int ret;
-    do {
-        ret = kevent64(loop->fd, &event, 1, &event, 1, KEVENT_FLAG_ERROR_EVENTS, NULL);
-    } while (IS_EINTR(ret));
+    return ret < 0 ? ret : 0;
 }
 
 #endif
@@ -646,8 +653,10 @@ struct us_poll_t *us_poll_resize(struct us_poll_t *p, struct us_loop_t *loop, un
     new_p->state.poll_type = us_internal_poll_type(new_p);
     us_poll_change(new_p, loop, events);
 #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-add both filters to move their udata to new_p, whether or not they are polled: the
+     * EV_CLEAR read knote of a socket that is not reading (see kqueue_change) has to follow
+     * the relocation too, and EV_ADD keeps the mode of a knote that already exists. */
+    kqueue_change(loop->fd, new_p->state.fd, 0, LIBUS_SOCKET_WRITABLE | LIBUS_SOCKET_READABLE, new_p, 0);
 #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);
@@ -685,7 +694,7 @@ int us_poll_start_rc(struct us_poll_t *p, struct us_loop_t *loop, int events) {
     } while (IS_EINTR(ret));
     return ret;
 #else
-    return kqueue_change(loop->fd, p->state.fd, 0, events, p);
+    return kqueue_change(loop->fd, p->state.fd, 0, events, p, 0);
 #endif
 }
 
@@ -720,7 +729,7 @@ int us_poll_change(struct us_poll_t *p, struct us_loop_t *loop, int events) {
             rc = us_poll_start_rc(p, loop, events);
         }
 #else
-        kqueue_change(loop->fd, p->state.fd, old_events, events, p);
+        kqueue_change(loop->fd, p->state.fd, old_events, events, p, kqueue_is_socket_poll(p));
 #endif
         /* Set all removed events to null-polls in pending ready poll list */
         us_internal_loop_update_pending_ready_polls(loop, p, p, old_events, events);
@@ -738,8 +747,11 @@ void us_poll_stop(struct us_poll_t *p, struct us_loop_t *loop) {
          rc = epoll_ctl(loop->fd, EPOLL_CTL_DEL, p->state.fd, &event);
     } while (IS_EINTR(rc));
 #else
-    if (old_events) {
-        kqueue_change(loop->fd, p->state.fd, old_events, new_events, NULL);
+    /* A socket poll has a read knote in both of its modes (see kqueue_change), so there is
+     * one to delete even when it was not polling for reads. */
+    int registered = kqueue_is_socket_poll(p) ? old_events | LIBUS_SOCKET_READABLE : old_events;
+    if (registered) {
+        kqueue_change(loop->fd, p->state.fd, registered, new_events, NULL, 0);
     }
 #endif
 

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

@@ -249,13 +249,6 @@ void us_internal_ssl_ctx_unref(struct ssl_ctx_st *ssl_ctx);
 /* TCP-level FIN, bypassing the SSL layer (used by ssl_on_end). */
 void us_internal_socket_raw_shutdown(us_socket_r s);
 
-#ifdef LIBUS_USE_KQUEUE
-/* Arm an EV_CLEAR read filter on a socket with no readable interest so the
- * peer's FIN/RST still reaches the dispatcher (kqueue's stand-in for epoll's
- * implicit EPOLLHUP/EPOLLERR). See the definition in epoll_kqueue.c. */
-void us_internal_kqueue_socket_arm_read_sentinel(us_socket_r s);
-#endif
-
 int us_internal_handle_dns_results(us_loop_r loop);
 
 /* Sockets are polls */

packages/bun-usockets/src/loop.c

@@ -914,12 +914,6 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                      * writable dispatch disables writable polling again once
                      * the buffer is drained, so this does not busy-poll. */
                     us_poll_change(&s->p, loop, LIBUS_SOCKET_WRITABLE);
-#ifdef LIBUS_USE_KQUEUE
-                    /* The change above deleted the read filter; without a sentinel
-                     * the peer's later RST is never reported (the one-shot write
-                     * filter may already be consumed) and the socket strands. */
-                    us_internal_kqueue_socket_arm_read_sentinel(s);
-#endif
                     s = s->ssl ? us_internal_ssl_on_end(s) : us_dispatch_end(s);
                 } else {
                     /* Half-open not allowed, or a hangup (both directions down, level-triggered):

packages/bun-usockets/src/socket.c

@@ -731,13 +731,6 @@ void us_internal_socket_raw_shutdown(struct us_socket_t *s) {
         us_internal_poll_set_type(&s->p, POLL_TYPE_SOCKET_SHUT_DOWN);
         us_poll_change(&s->p, s->group->loop, us_poll_events(&s->p) & LIBUS_SOCKET_READABLE);
         bsd_shutdown_socket(us_poll_fd((struct us_poll_t *) s));
-#ifdef LIBUS_USE_KQUEUE
-        if (!(us_poll_events(&s->p) & LIBUS_SOCKET_READABLE)) {
-            /* Shut down with reads off: no filter remains, so a peer FIN/RST
-             * would never be delivered (epoll still reports HUP/ERR). */
-            us_internal_kqueue_socket_arm_read_sentinel(s);
-        }
-#endif
     }
 }
 
@@ -871,13 +864,6 @@ void us_socket_pause(struct us_socket_t *s) {
     // we are readable and writable so we can just pause readable side
     us_poll_change(&s->p, s->group->loop, LIBUS_SOCKET_WRITABLE);
     s->flags.is_paused = 1;
-#ifdef LIBUS_USE_KQUEUE
-    if (us_socket_is_shut_down(s)) {
-        /* Pausing dropped a shut-down socket's read filter; same as
-         * us_internal_socket_raw_shutdown. */
-        us_internal_kqueue_socket_arm_read_sentinel(s);
-    }
-#endif
 }
 
 void us_socket_resume(struct us_socket_t *s) {

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

@@ -4086,3 +4086,74 @@ describe("allowHalfOpen socket whose peer resets behind pending writes", () => {
     expect(endCount).toBe(1);
   });
 });
+
+// A paused socket polls for nothing. epoll reports the reset anyway (EPOLLERR cannot be
+// masked); kqueue only reports it through the read knote that epoll_kqueue.c keeps
+// registered while reads are off. Before that, the pause left a one-shot writable event
+// behind and nothing else: a reset that landed after it was consumed was never reported,
+// and the socket stayed paused for good. The greeting round trip below guarantees the
+// one-shot has been consumed before the reset is sent. The node:net and node:tls shapes
+// of this scenario are in test/js/node/tls/node-tls-server.test.ts.
+describe.concurrent.each(["tcp", "tls"] as const)("%s socket paused when its peer resets the connection", transport => {
+  it("closes with read ECONNRESET while still paused and delivers none of the unread data", async () => {
+    const closedWith = Promise.withResolvers<Error | undefined>();
+    let dataCalls = 0;
+    const pauseAndGreet = (socket: Socket) => {
+      socket.pause();
+      socket.write("greeting");
+    };
+    using server = Bun.listen({
+      hostname: "127.0.0.1",
+      port: 0,
+      tls: transport === "tls" ? tls : undefined,
+      socket: {
+        open(socket) {
+          // A TLS socket cannot be paused before its handshake has been read.
+          if (transport === "tcp") pauseAndGreet(socket);
+        },
+        handshake(socket, success, authorizationError) {
+          if (success) pauseAndGreet(socket);
+          else closedWith.reject(authorizationError ?? new Error("server handshake failed"));
+        },
+        data() {
+          dataCalls++;
+        },
+        close(_socket, error) {
+          closedWith.resolve(error);
+        },
+      },
+    });
+
+    const greeted = Promise.withResolvers<Socket>();
+    await Bun.connect({
+      hostname: "127.0.0.1",
+      port: server.port,
+      tls: transport === "tls" ? { ca: tls.cert } : undefined,
+      socket: {
+        data: socket => greeted.resolve(socket),
+        error: (_socket, error) => greeted.reject(error),
+        connectError: (_socket, error) => greeted.reject(error),
+        close: () => greeted.reject(new Error("peer closed before the greeting arrived")),
+      },
+    });
+    const peer = await greeted.promise;
+    peer.write("queued behind the pause");
+    peer.terminate();
+
+    const error = (await closedWith.promise) as NodeJS.ErrnoException | undefined;
+    // On Windows the close reports the reset too, but the error it carries has no code: the
+    // raw WSA code reaches on_close unmapped (node:net papers over it, see SocketEmitEndNT).
+    // That is a separate bug. Until it is fixed only POSIX can check the code.
+    expect({
+      reported: error instanceof Error,
+      syscall: error?.syscall,
+      dataCalls,
+      code: isWindows ? null : error?.code,
+    }).toEqual({
+      reported: true,
+      syscall: "read",
+      dataCalls: 0,
+      code: isWindows ? null : "ECONNRESET",
+    });
+  });
+});