usockets: stop pause() from arming writable interest it never had

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

@@ -538,13 +538,33 @@ int kqueue_change(int kqfd, int fd, int old_events, int new_events, void *user_d
     /* 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)) {
+    /* 0-event polls need a filter for peer teardown to ride (epoll gets this
+     * for free via the implicit EPOLLHUP|EPOLLERR). For a socket whose write
+     * side WE shut down, a write filter is useless: our own SS_CANTSENDMORE
+     * makes it report EV_EOF instantly, which read as the connection being
+     * over and closed a paused half-closed socket whose peer was alive. The
+     * read filter has no such echo - its EV_EOF is the PEER's FIN/RST - so
+     * keep one armed (EV_CLEAR: buffered data fires once and is masked by the
+     * dispatcher, instead of level-triggering every tick). */
+    int own_shutdown = user_data &&
+        us_internal_poll_type((struct us_poll_t *) user_data) == POLL_TYPE_SOCKET_SHUT_DOWN;
+    int teardown_watch = !is_readable && !is_writable && own_shutdown;
+    if (teardown_watch) {
+        EV_SET64(&change_list[change_length++], fd, EVFILT_READ, EV_ADD | EV_CLEAR, 0, 0, (uint64_t)(void*)user_data, 0, 0);
+        if (old_events & LIBUS_SOCKET_WRITABLE) {
+            /* A still-armed write one-shot would report our own
+             * SS_CANTSENDMORE; nothing can ever be sent again anyway. */
+            EV_SET64(&change_list[change_length++], fd, EVFILT_WRITE, EV_DELETE, 0, 0, (uint64_t)(void*)user_data, 0, 0);
+        }
+    } else 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);
     }
 
     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
+        /* One-shot write filter as the teardown ride for 0-event polls that
+         * did not shut down themselves (see above; their read side may be
+         * gone entirely, e.g. half-open after on_end). */
+        if(!(old_events & LIBUS_SOCKET_WRITABLE) && !own_shutdown) {
             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)) {

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

@@ -51,6 +51,13 @@ void us_internal_loop_update_pending_ready_polls(struct us_loop_t *loop,
                                                  int new_events);
 #endif
 
+#ifdef LIBUS_USE_KQUEUE
+/* Defined in eventing/epoll_kqueue.c. Applies an interest-set transition as
+ * kevent changes; a 0->0 transition on a shut-down socket's poll arms the
+ * read-side teardown watch (see us_internal_socket_raw_shutdown). */
+int kqueue_change(int kqfd, int fd, int old_events, int new_events, void *user_data);
+#endif
+
 /* We only have one networking implementation so far */
 #include "internal/networking/bsd.h"
 

packages/bun-usockets/src/libusockets.h

@@ -725,6 +725,12 @@ int us_socket_set_tos(us_socket_r s, int tos);
 int us_socket_get_tos(us_socket_r s);
 void us_socket_resume(us_socket_r s);
 void us_socket_pause(us_socket_r s);
+/* Arm writable interest for bytes the caller holds OUTSIDE the socket's own
+ * write path (uws's queued pipelined responses sit in the AsyncSocket buffer
+ * without any send having been attempted, so no write failure ever armed the
+ * poll). The next writable event flushes them. Respects pause: readable
+ * interest is not re-added. */
+void us_socket_mark_writable_pending(us_socket_r s);
 
 #ifdef __cplusplus
 }

packages/bun-usockets/src/socket.c

@@ -418,6 +418,17 @@
                    LIBUS_SOCKET_WRITABLE | (s->flags.is_paused ? 0 : LIBUS_SOCKET_READABLE));
 }
 
+/* See libusockets.h. last_write_failed makes the writable dispatch keep the
+ * interest armed until a flush succeeds (it is cleared at writable-event
+ * entry and re-set by any failing write). */
+void us_socket_mark_writable_pending(struct us_socket_t *s) {
+    if (us_socket_is_closed(s) || us_socket_is_shut_down(s)) {
+        return;
+    }
+    s->flags.last_write_failed = 1;
+    us_internal_rearm_writable(s);
+}
+
 int us_socket_write2(struct us_socket_t *s, const char *header, int header_length, const char *payload, int payload_length) {
     if (us_socket_is_closed(s) || us_socket_is_shut_down(s)) {
         return 0;
@@ -709,6 +720,15 @@
     if (!us_socket_is_closed(s) && us_internal_poll_type(&s->p) != POLL_TYPE_SOCKET_SHUT_DOWN) {
         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);
+#ifdef LIBUS_USE_KQUEUE
+        /* A socket already at 0 events (paused with nothing buffered) diffs
+         * to a no-op above, leaving no kqueue filter at all; arm the
+         * read-side teardown watch directly so the peer's FIN/RST still
+         * closes us (epoll's implicit EPOLLHUP|EPOLLERR needs no filter). */
+        if (us_poll_events(&s->p) == 0) {
+            kqueue_change(s->group->loop->fd, us_poll_fd(&s->p), 0, 0, &s->p);
+        }
+#endif
         bsd_shutdown_socket(us_poll_fd((struct us_poll_t *) s));
     }
 }
@@ -840,8 +860,14 @@
     if (s->flags.is_paused) return;
     // closed cannot be paused because it is already closed
     if (us_socket_is_closed(s)) return;
-    // we are readable and writable so we can just pause readable side
-    us_poll_change(&s->p, s->group->loop, LIBUS_SOCKET_WRITABLE);
+    /* Drop readable interest but only KEEP writable interest, never add it:
+     * forcing WRITABLE here dispatched a bogus writable (a JS drain event
+     * with nothing buffered) on every pause, and on a shut-down socket the
+     * fresh kqueue EVFILT_WRITE one-shot reported our own SS_CANTSENDMORE
+     * as EV_EOF immediately, closing a half-closed socket whose peer was
+     * still alive (libuv's uv_read_stop only removes read interest). A
+     * backpressured write keeps its interest; none means nothing to drain. */
+    us_poll_change(&s->p, s->group->loop, us_poll_events(&s->p) & LIBUS_SOCKET_WRITABLE);
     s->flags.is_paused = 1;
 }
 

packages/bun-uws/src/HttpContext.h

@@ -421,6 +421,11 @@ struct HttpContext {
                      * right now — pausing the socket alone cannot bound it. */
                     httpResponseData->nodeHttpParkAtNextBoundary = true;
                     ((HttpResponse<SSL> *) s)->pause();
+                    /* The buffered bytes were queued without a kernel write
+                     * (response ordering), so no write failure armed the poll;
+                     * the onWritable that flushes them and replays the parked
+                     * requests needs explicit writable interest. */
+                    us_socket_mark_writable_pending((us_socket_t *) s);
                 }
                 }
             } else {
@@ -454,6 +459,10 @@ struct HttpContext {
                         httpResponseData->state |= HttpResponseData<SSL>::HTTP_NODE_READS_PAUSED;
                         httpResponseData->nodeHttpParkAtNextBoundary = true;
                         ((HttpResponse<SSL> *) s)->pause();
+                        /* Same as the pipelined branch above: the queued bytes
+                         * never hit the kernel, so arm the writable that will
+                         * flush them and replay the parked requests. */
+                        us_socket_mark_writable_pending((us_socket_t *) s);
                     }
                 }
             }

src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp

@@ -500,6 +500,11 @@ static void onNodeHttpReadsPaused(us_socket_t* socket)
     auto* d = reinterpret_cast<uWS::NodeHttpResponseData<SSL>*>(us_socket_ext(socket));
     d->nodeHttpParkAtNextBoundary = true;
     d->state |= uWS::HttpResponseData<SSL>::HTTP_NODE_READS_PAUSED;
+    // The replay of parked requests (and the eventual read resume) runs from
+    // onWritable, but the queued pipelined responses live in the JS pipeline
+    // queue: no socket write happens here to arm the poll, so ask for the
+    // writable event explicitly.
+    us_socket_mark_writable_pending(socket);
 }
 
 extern "C" void Bun__NodeHTTP__onReadsPaused(int ssl, us_socket_t* socket)

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

@@ -362,6 +362,117 @@ describe.concurrent("socket", () => {
     expect(await bunRun(fileURLToPath(new URL("./kqueue-filter-coalesce-fixture.ts", import.meta.url)))).toSpawn();
   });
 
+  // us_socket_pause armed WRITABLE unconditionally; the always-writable socket
+  // then dispatched a bogus drain with nothing buffered.
+  it("pause() with nothing buffered must not fire a drain event", async () => {
+    using server = Bun.listen({
+      hostname: "127.0.0.1",
+      port: 0,
+      socket: { open() {}, data() {}, end() {}, error() {}, close() {} },
+    });
+    let drains = 0;
+    const opened = Promise.withResolvers<any>();
+    await Bun.connect({
+      hostname: "127.0.0.1",
+      port: server.port,
+      socket: {
+        open: s => opened.resolve(s),
+        drain() {
+          drains++;
+        },
+        data() {},
+        end() {},
+        error() {},
+        close() {},
+      },
+    });
+    const s = await opened.promise;
+    await Bun.sleep(50); // let any connect-time writable settle
+    const before = drains;
+    s.pause();
+    await Bun.sleep(100); // window in which the buggy pause-armed writable fired
+    const delta = drains - before;
+    s.terminate(); // release before asserting so a failure does not leak the socket
+    expect(delta).toBe(0);
+  });
+
+  // On kqueue, pause() after shutdown() armed an EVFILT_WRITE one-shot that
+  // reported our own SS_CANTSENDMORE as EV_EOF immediately: the half-closed
+  // socket closed within milliseconds even though the peer was alive and
+  // silent. Linux always kept it open; this pins the behavior on both.
+  it("shutdown() then pause() keeps a half-closed socket open while the peer is silent", async () => {
+    let closedEarly = false;
+    const opened = Promise.withResolvers<void>();
+    using server = Bun.listen({
+      hostname: "127.0.0.1",
+      port: 0,
+      socket: {
+        open(s) {
+          s.shutdown();
+          s.pause();
+          opened.resolve();
+        },
+        data() {},
+        end() {},
+        error() {},
+        close() {
+          closedEarly = true;
+        },
+      },
+    });
+    // allowHalfOpen peer ignores our FIN and stays silently connected.
+    const peer = await Bun.connect({
+      hostname: "127.0.0.1",
+      port: server.port,
+      allowHalfOpen: true,
+      socket: { open() {}, data() {}, end() {}, error() {}, close() {} },
+    });
+    await opened.promise;
+    // Negative-assertion window: the buggy kqueue EV_EOF closed within ~2ms,
+    // so 250ms is >100x margin without spending the whole per-test budget.
+    await Bun.sleep(250);
+    const closed = closedEarly;
+    peer.terminate(); // release before asserting so a failure does not leak the socket
+    expect(closed).toBe(false);
+  });
+
+  // The flip side: with the write filter unusable after our own shutdown()
+  // (its EV_EOF echoes SS_CANTSENDMORE), the read-side teardown watch must
+  // still deliver the peer's actual termination to a paused half-closed
+  // socket (epoll gets this via the implicit EPOLLHUP|EPOLLERR).
+  it("shutdown() then pause() still closes when the peer terminates", async () => {
+    const closed = Promise.withResolvers<void>();
+    const opened = Promise.withResolvers<void>();
+    using server = Bun.listen({
+      hostname: "127.0.0.1",
+      port: 0,
+      socket: {
+        open(s) {
+          s.shutdown();
+          s.pause();
+          opened.resolve();
+        },
+        data() {},
+        end() {},
+        error() {},
+        close() {
+          closed.resolve();
+        },
+      },
+    });
+    const peerOpened = Promise.withResolvers<any>();
+    await Bun.connect({
+      hostname: "127.0.0.1",
+      port: server.port,
+      allowHalfOpen: true,
+      socket: { open: s => peerOpened.resolve(s), data() {}, end() {}, error() {}, close() {} },
+    });
+    const peer = await peerOpened.promise;
+    await opened.promise;
+    peer.terminate(); // RST; the victim's close must still fire while paused
+    await closed.promise;
+  });
+
   it("reload() should preserve active_connections (no UAF / counter underflow)", async () => {
     await using proc = Bun.spawn({
       cmd: [bunExe(), fileURLToPath(new URL("./socket-reload-fixture.ts", import.meta.url))],