diff --git a/packages/bun-usockets/src/context.c b/packages/bun-usockets/src/context.c index 66eba3028bb6..856fd6c2cbcd 100644 --- a/packages/bun-usockets/src/context.c +++ b/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; } diff --git a/packages/bun-usockets/src/internal/internal.h b/packages/bun-usockets/src/internal/internal.h index 644b4f71a529..7b38fa4fb4b6 100644 --- a/packages/bun-usockets/src/internal/internal.h +++ b/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). */ diff --git a/packages/bun-usockets/src/loop.c b/packages/bun-usockets/src/loop.c index baba410af040..d72ad08ff81b 100644 --- a/packages/bun-usockets/src/loop.c +++ b/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 { diff --git a/packages/bun-usockets/src/socket.c b/packages/bun-usockets/src/socket.c index 25f798b1a017..7af529045b24 100644 --- a/packages/bun-usockets/src/socket.c +++ b/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)); } diff --git a/test/js/bun/net/tcp-server.test.ts b/test/js/bun/net/tcp-server.test.ts index 9ffcd056439c..68de6fb4fae3 100644 --- a/test/js/bun/net/tcp-server.test.ts +++ b/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(); + const clientClosed = Promise.withResolvers(); + + 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