usockets: keep READABLE off a half-open socket whose on_end already fired

packages/bun-usockets/src/context.c

@@ -347,6 +347,7 @@ static void us_internal_init_listen_socket(struct us_listen_socket_t *ls,
     s->flags.adopted = 0;
     s->flags.allow_half_open = (options & LIBUS_SOCKET_ALLOW_HALF_OPEN);
     s->unclassified_send_failures = 0;
+    s->readable_ended = 0;
     s->next = 0;
     s->prev = 0;
     s->connect_state = NULL;
@@ -491,6 +492,7 @@ static inline void us_internal_init_connect_socket(struct us_socket_t *s,
     s->flags.adopted = 0;
     s->flags.last_write_failed = 0;
     s->unclassified_send_failures = 0;
+    s->readable_ended = 0;
     s->connect_state = NULL;
     s->connect_next = NULL;
 }

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

@@ -305,6 +305,10 @@ struct us_socket_t {
    * the driver's epilogue via ssl_pending_detach. */
   unsigned char ssl_in_use : 1;
   unsigned char ssl_pending_detach : 1;
+  /* on_end has been dispatched for the half-open path; recv() can only return
+   * 0 now. Guards the callers that would otherwise re-arm READABLE (partial
+   * write, resume) so on_end is not re-derived and re-fired every tick. */
+  unsigned char readable_ended : 1;
   /* The close code passed to the deferred close (e.g. a reset requested from
    * inside a handshake callback must still RST, not FIN, when it is finally
    * performed). */

packages/bun-usockets/src/loop.c

@@ -536,6 +536,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                         s->flags.adopted = 0;
                         s->flags.last_write_failed = 0;
                         s->unclassified_send_failures = 0;
+                        s->readable_ended = 0;
 
                         /* We always use nodelay */
                         bsd_socket_nodelay(client_fd, 1);
@@ -847,7 +848,14 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                     s = us_internal_socket_close_raw(s, LIBUS_SOCKET_CLOSE_CODE_CLEAN_SHUTDOWN, NULL);
                     return;
                 }
-                if(s->flags.allow_half_open) {
+                if (s->readable_ended) {
+                    /* on_end already fired (half-open). A backpressured write's
+                     * rearm_writable (or resume/a level-triggered eof hint) put
+                     * READABLE back; absorb the 0-byte read here instead of
+                     * re-dispatching on_end. Drop READABLE so the next tick is
+                     * driven by WRITABLE alone. */
+                    us_poll_change(&s->p, loop, us_poll_events(&s->p) & LIBUS_SOCKET_WRITABLE);
+                } else if(s->flags.allow_half_open) {
                     /* EOF with half-open allowed: stop polling readable but KEEP
                      * polling writable. Masking with the current events dropped
                      * writable when the EOF landed before the poll had been
@@ -858,6 +866,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                      * http response tests hung on every Linux target. The
                      * writable dispatch disables writable polling again once
                      * the buffer is drained, so this does not busy-poll. */
+                    s->readable_ended = 1;
                     us_poll_change(&s->p, loop, LIBUS_SOCKET_WRITABLE);
                     s = s->ssl ? us_internal_ssl_on_end(s) : us_dispatch_end(s);
                 } else {

packages/bun-usockets/src/socket.c

@@ -412,7 +412,11 @@ struct us_socket_t *us_socket_pair(struct us_socket_group_t *group, unsigned cha
 /* Re-arm writable for a backpressured write without resuming the read side of
  * a paused socket: us_poll_change sets absolute flags, so including READABLE
  * unconditionally would silently undo us_socket_pause mid-backpressure and
- * deliver data the caller asked to defer. */
+ * deliver data the caller asked to defer. READABLE stays on even when
+ * readable_ended is set - loop.c's readable_ended guard absorbs the 0-byte
+ * read without re-dispatching, and the Windows/TLS half-close drain (libuv.c
+ * poll_cb's UV_DISCONNECT handling) relies on READABLE being present to keep
+ * the writable dispatch flowing after the eof branch drops it. */
 static void us_internal_rearm_writable(struct us_socket_t *s) {
     us_poll_change(&s->p, s->group->loop,
                    LIBUS_SOCKET_WRITABLE | (s->flags.is_paused ? 0 : LIBUS_SOCKET_READABLE));
@@ -457,6 +461,7 @@ struct us_socket_t *us_socket_from_fd(struct us_socket_group_t *group, unsigned
     s->flags.adopted = 0;
     s->flags.last_write_failed = 0;
     s->unclassified_send_failures = 0;
+    s->readable_ended = 0;
     s->connect_state = NULL;
 
     /* We always use nodelay */
@@ -703,12 +708,18 @@ int us_connecting_socket_is_shut_down(struct us_connecting_socket_t *c) {
 }
 
 void us_internal_socket_raw_shutdown(struct us_socket_t *s) {
-    /* Todo: should we emit on_close if calling shutdown on an already half-closed socket?
-     * We need more states in that case, we need to track RECEIVED_FIN
-     * so far, the app has to track this and call close as needed */
     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);
+        /* Peer FIN already delivered: the half-open poll sits at WRITABLE-only
+         * (or 0 after drain), so `events & READABLE` would be 0 and on
+         * kqueue/libuv nothing would wake the SHUT_DOWN close path (epoll gets
+         * it via unmaskable EPOLLHUP). Arm READABLE so the next poll reports
+         * the 0-byte read / DISCONNECT and closes via the existing SHUT_DOWN
+         * branch - next iteration, not synchronously, so callers still see a
+         * live socket after shutdown() returns. */
+        us_poll_change(&s->p, s->group->loop,
+                       s->readable_ended ? LIBUS_SOCKET_READABLE
+                                         : (us_poll_events(&s->p) & LIBUS_SOCKET_READABLE));
         bsd_shutdown_socket(us_poll_fd((struct us_poll_t *) s));
     }
 }
@@ -860,10 +871,15 @@ void us_socket_resume(struct us_socket_t *s) {
     if (us_socket_is_closed(s)) return;
 
     if (us_socket_is_shut_down(s)) {
-        // we already sent FIN so we resume only readable side we are read-only
+        /* We already sent FIN. Re-deriving eof here is what closes us (loop.c
+         * checks is_shut_down before readable_ended), so READABLE stays on
+         * even if readable_ended - same as raw_shutdown. */
         us_poll_change(&s->p, s->group->loop, LIBUS_SOCKET_READABLE);
         return;
     }
-    // we are readable and writable so we resume everything
-    us_poll_change(&s->p, s->group->loop, LIBUS_SOCKET_READABLE | LIBUS_SOCKET_WRITABLE);
+    /* Peer FIN already delivered: recv() can only return 0 now, so skip
+     * READABLE and leave the half-open poll at WRITABLE-only. */
+    us_poll_change(&s->p, s->group->loop,
+                   LIBUS_SOCKET_WRITABLE |
+                       (s->readable_ended ? 0 : LIBUS_SOCKET_READABLE));
 }

test/js/bun/net/tcp-server.test.ts

@@ -1,4 +1,5 @@
 import { connect, listen, SocketHandler, TCPSocketListener } from "bun";
+import { setSocketOptions } from "bun:internal-for-testing";
 import { describe, expect, it } from "bun:test";
 import { expectMaxObjectTypeCount, isWindows } from "harness";
 
@@ -295,6 +296,80 @@ describe("tcp socket binaryType", () => {
   }
 });
 
+// With allowHalfOpen, a server's end() handler that writes more than the kernel
+// send buffer accepts (a partial write) triggered us_internal_rearm_writable,
+// which re-added READABLE to the poll mask. The half-open eof branch had just
+// set it to WRITABLE-only, so the next epoll tick re-derived recv()==0 -> eof
+// and re-dispatched end(), forever. Drain fired at most once between re-entries.
+it("allowHalfOpen: end() fires once when the handler's write is partially accepted", async () => {
+  const PAYLOAD = Buffer.alloc(256 * 1024, 0x61);
+  let endCount = 0;
+  let drainCount = 0;
+  const serverClosed = Promise.withResolvers<void>();
+  const clientClosed = Promise.withResolvers<void>();
+
+  using server = listen<{ sent: number }>({
+    hostname: "127.0.0.1",
+    port: 0,
+    allowHalfOpen: true,
+    socket: {
+      open(s) {
+        // Clamp SO_SNDBUF so the write from end() is a partial write on every
+        // POSIX kernel. No-op on Windows; the drainCount assertion is gated.
+        setSocketOptions(s, 1, 4096);
+        s.data = { sent: 0 };
+      },
+      data() {},
+      end(s) {
+        if (++endCount > 1) {
+          // The bug re-enters end() every tick; terminate so the test fails on
+          // the assertion below instead of spinning.
+          s.terminate();
+          return;
+        }
+        s.data.sent = s.write(PAYLOAD);
+        if (s.data.sent >= PAYLOAD.length) s.shutdown();
+      },
+      drain(s) {
+        drainCount++;
+        if (s.data.sent === 0) return;
+        s.data.sent += s.write(PAYLOAD.subarray(s.data.sent));
+        if (s.data.sent >= PAYLOAD.length) s.shutdown();
+      },
+      close() {
+        serverClosed.resolve();
+      },
+    },
+  });
+
+  let received = 0;
+  await connect({
+    hostname: "127.0.0.1",
+    port: server.port,
+    socket: {
+      open(s) {
+        s.write("hi");
+        s.shutdown();
+      },
+      data(_s, chunk) {
+        received += chunk.byteLength;
+      },
+      end() {},
+      close() {
+        clientClosed.resolve();
+      },
+    },
+  });
+
+  await Promise.all([serverClosed.promise, clientClosed.promise]);
+
+  expect({ endCount, received }).toEqual({ endCount: 1, received: PAYLOAD.length });
+  // setSocketOptions is a POSIX-only no-op on Windows, where loopback
+  // auto-tuning can accept 256 KiB in one send(). The assertion above already
+  // proves the fix there (endCount == 1 with the whole payload delivered).
+  if (!isWindows) expect(drainCount).toBeGreaterThanOrEqual(1);
+});
+
 it("should not leak memory", async () => {
   // assert we don't leak the sockets
   // we expect 1 or 2 because that's the prototype / structure