diff --git a/packages/bun-usockets/src/eventing/epoll_kqueue.c b/packages/bun-usockets/src/eventing/epoll_kqueue.c index 4dce3538d154..cc25ec6842b5 100644 --- a/packages/bun-usockets/src/eventing/epoll_kqueue.c +++ b/packages/bun-usockets/src/eventing/epoll_kqueue.c @@ -280,8 +280,10 @@ static void us_internal_dispatch_ready_polls(struct us_loop_t *loop) { uint8_t writable : 1; uint8_t error : 1; uint8_t eof : 1; + uint8_t send_eof : 1; + uint8_t send_eof_err : 1; uint8_t skip : 1; - uint8_t _pad : 3; + uint8_t _pad : 1; }; _Static_assert(sizeof(struct kevent_flags) == 1, "kevent_flags must be 1 byte"); @@ -305,7 +307,15 @@ static void us_internal_dispatch_ready_polls(struct us_loop_t *loop) { #endif .writable = (filter == EVFILT_WRITE), .error = !!(flags & EV_ERROR), - .eof = !!(flags & EV_EOF), + /* eof tracks the read side only; the write filter's EV_EOF + * (SS_CANTSENDMORE) is kept separate so the zero-event detector + * below can route it to the error path without fabricating a + * second read EOF. Dispatch still ORs them, preserving behavior + * for polls with events armed. */ + .eof = (flags & EV_EOF) && filter != EVFILT_WRITE, + .send_eof = (filter == EVFILT_WRITE) && (flags & EV_EOF), + /* kevent(2): with EV_EOF set, fflags carries the socket error. */ + .send_eof_err = (filter == EVFILT_WRITE) && (flags & EV_EOF) && loop->ready_polls[i].fflags != 0, }; /* Look backward for a prior entry with the same poll to coalesce into. @@ -317,6 +327,8 @@ static void us_internal_dispatch_ready_polls(struct us_loop_t *loop) { coalesced[j].writable |= bits.writable; coalesced[j].error |= bits.error; coalesced[j].eof |= bits.eof; + coalesced[j].send_eof |= bits.send_eof; + coalesced[j].send_eof_err |= bits.send_eof_err; coalesced[i] = (struct kevent_flags){ .skip = 1 }; merged = 1; break; @@ -344,9 +356,56 @@ static void us_internal_dispatch_ready_polls(struct us_loop_t *loop) { int events = (bits.readable ? LIBUS_SOCKET_READABLE : 0) | (bits.writable ? LIBUS_SOCKET_WRITABLE : 0); - events &= us_poll_events(poll); - if (events || bits.error || bits.eof) { - us_internal_dispatch_ready_poll(poll, bits.error, bits.eof, events); + int error = bits.error; + int eof = bits.eof | bits.send_eof; + + const int wanted = us_poll_events(poll); + events &= wanted; + + /* The EV_CLEAR write knote kqueue_change keeps as a connection-death + * detector exists whenever WRITABLE is not armed: it is the only + * kernel presence of a poll at zero events (a half-open socket past + * on_end whose writes drained, or a paused socket), and it survives a + * later 0 -> READABLE transition, whose diff has no reason to touch + * the write filter. EV_EOF from it is SS_CANTSENDMORE, and what that + * means depends on what we know. */ + if (bits.send_eof && !(wanted & LIBUS_SOCKET_WRITABLE)) { + const int kind = us_internal_poll_type(poll); + if (kind == POLL_TYPE_SOCKET || kind == POLL_TYPE_SOCKET_SHUT_DOWN) { + if (bits.send_eof_err) { + /* A pending socket error (a TCP reset) is the poll error, + * routed exactly like epoll's implicit EPOLLERR. */ + error = 1; + eof = bits.eof; + } else if (kind == POLL_TYPE_SOCKET_SHUT_DOWN && wanted == 0) { + /* A shut-down socket with no read knote has no other way + * to learn the connection is over: eof stays set + * (bits.eof | bits.send_eof) and the shut-down branch in + * loop.c clean-closes. For the half-open path that is + * exact epoll parity (the peer's FIN was consumed before + * the read knote was deleted, and EPOLLHUP would fire for + * the same both-directions-shut state). For a paused + * socket it means closing on our own shutdown's echo, + * which node:http's half-close-with-unread-body depends + * on (nothing ever resumes that socket), where epoll + * waits for the peer's FIN to complete SHUTDOWN_MASK. */ + } else { + /* Stay silent. A resumed reader (wanted == READABLE) + * hears the peer through the read filter, so a stale + * shutdown echo must not clean-close it first. And on a + * not-shut-down socket, a unix peer's graceful close + * (SS_CANTSENDMORE with so_error 0) must not fabricate + * the ECONNRESET the error path would clamp to, nor + * re-run on_end via eof, for a disconnect epoll reports + * as a mere EPOLLHUP; a paused one keeps its queued data + * readable for resume(). */ + eof = bits.eof; + } + } + } + + if (events || error || eof) { + us_internal_dispatch_ready_poll(poll, error, eof, events); } } #endif @@ -543,10 +602,18 @@ int kqueue_change(int kqfd, int fd, int old_events, int new_events, void *user_d } 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); - } + /* Polling neither direction must still see the connection die, the way a + * zero-event epoll registration still reports EPOLLHUP/EPOLLERR (see + * us_poll_start_rc). Keep an EV_CLEAR write knote armed: it fires once on + * arming (the socket is trivially writable; the dispatcher masks it out) + * and then only on write-side state changes, so an idle half-open or + * paused socket does not wake the loop, while a peer reset reports + * EV_EOF/fflags through it. EV_ONESHOT here got consumed by that first + * masked wakeup, leaving the fd with no knote at all: the reset was never + * seen and the socket leaked until process exit. Unconditional because + * EV_ADD must also convert a still-armed oneshot from prior WRITABLE + * interest. */ + EV_SET64(&change_list[change_length++], fd, EVFILT_WRITE, EV_ADD | EV_CLEAR, 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? */ 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); @@ -660,6 +727,19 @@ void us_poll_change(struct us_poll_t *p, struct us_loop_t *loop, int events) { /* 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); } +#ifdef LIBUS_USE_KQUEUE + else if (events == 0) { + /* 0 -> 0 is not a no-op on kqueue for a socket: a writable dispatch + * consumed the EV_ONESHOT write knote (loop.c clears + * POLL_TYPE_POLLING_OUT to mirror that), so the fd may hold no knote + * at all. Re-register the EV_CLEAR detector (see kqueue_change) so a + * peer reset can still wake the socket. */ + const int kind = us_internal_poll_type(p); + if (kind == POLL_TYPE_SOCKET || kind == POLL_TYPE_SOCKET_SHUT_DOWN) { + kqueue_change(loop->fd, p->state.fd, 0, 0, p); + } + } +#endif } void us_poll_stop(struct us_poll_t *p, struct us_loop_t *loop) { @@ -672,9 +752,20 @@ 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); - } + /* Delete both filters explicitly, whatever the tracked events say: a poll + * at 0 events still holds the EV_CLEAR detector knote (see kqueue_change), + * and kqueue_change's diff cannot express "delete everything" (its 0 -> 0 + * transition arms the detector instead). This matters for detach paths + * (us_socket_detach) that stop the poll but keep the fd open: a leftover + * knote would keep the freed poll as udata. Deleting an absent filter just + * reports ENOENT through KEVENT_FLAG_ERROR_EVENTS, which we ignore. */ + struct kevent64_s change_list[2]; + EV_SET64(&change_list[0], p->state.fd, EVFILT_READ, EV_DELETE, 0, 0, 0, 0, 0); + EV_SET64(&change_list[1], p->state.fd, EVFILT_WRITE, EV_DELETE, 0, 0, 0, 0, 0); + int rc; + do { + rc = kevent64(loop->fd, change_list, 2, change_list, 2, KEVENT_FLAG_ERROR_EVENTS, NULL); + } while (IS_EINTR(rc)); #endif /* Disable any instance of us in the pending ready poll list */ diff --git a/test/js/bun/net/socket.test.ts b/test/js/bun/net/socket.test.ts index f9e7ae1f55c1..17f6e9b056ef 100644 --- a/test/js/bun/net/socket.test.ts +++ b/test/js/bun/net/socket.test.ts @@ -655,6 +655,292 @@ describe.concurrent("socket", () => { } }); + // An allowHalfOpen socket that consumed the peer's FIN and drained its writes + // polls for no events at all. On Linux the fd stays registered in epoll, which + // reports EPOLLERR/EPOLLHUP even at zero interest, so a later RST still closes + // the socket. On macOS the kqueue write oneshot used to be consumed by the + // first writable wakeup, leaving the fd with no filter: the peer's reset was + // never delivered and the socket leaked until process exit (this test timed + // out). Windows (libuv) tracks this state separately and is skipped here. + it.skipIf(isWindows)("allowHalfOpen socket sees the peer reset after end + drain", async () => { + const { promise: ended, resolve: onEnd } = Promise.withResolvers(); + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + + using server = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + allowHalfOpen: true, + socket: { + open() {}, + data() {}, + end() { + onEnd(); + }, + close() { + onClosed("close"); + }, + error() { + onClosed("error"); + }, + }, + }); + + const client = net.connect({ port: server.port, host: "127.0.0.1", allowHalfOpen: true }); + // Post-connect teardown errors must not become uncaught exceptions. + client.on("error", () => {}); + try { + await new Promise((resolve, reject) => { + client.once("connect", resolve); + client.once("error", reject); + }); + client.end(); // FIN; the server side stays half-open + + await ended; + // Let the server's post-end writable dispatch run so its poll drops to zero + // requested events (the state that used to lose the last kqueue filter) + // before the reset arrives. + await new Promise(resolve => setImmediate(resolve)); + await new Promise(resolve => setImmediate(resolve)); + + client.resetAndDestroy(); + + // Hangs forever on a deaf socket; the test timeout is the failure signal. + expect(await closed).toBeOneOf(["close", "error"]); + } finally { + client.destroy(); + } + }); + + // The zero-event state must distinguish a dead connection from a graceful + // unix-domain disconnect: a unix peer's close() makes our write side + // unusable with no socket error, which epoll reports as a mere EPOLLHUP + // (and not at all at zero interest before both directions are shut). A + // paused unix socket with data still queued in the kernel must stay open + // through that disconnect so resume() can deliver the data and the real + // end-of-stream, instead of being torn down with a fabricated ECONNRESET. + it.skipIf(isWindows)("paused unix socket keeps data from a peer that closed gracefully", async () => { + const { promise: opened, resolve: onOpen } = Promise.withResolvers>(); + const { promise: gotData, resolve: onData, reject: onDataLost } = Promise.withResolvers(); + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + let teardown: string | null = null; + + using dir = tempDir("unix-paused-close", {}); + const sock = join(String(dir), "s.sock"); + + using server = Bun.listen({ + unix: sock, + socket: { + open(socket) { + socket.pause(); + onOpen(socket); + }, + data(_socket, buffer) { + onData(buffer.toString()); + }, + end() {}, + close() { + teardown ??= "close"; + // No-op on the expected path (data already resolved); an early + // teardown fails the data await with the cause instead of hanging. + onDataLost(new Error("server socket closed before the queued data was delivered")); + onClosed(teardown); + }, + error() { + teardown ??= "error"; + onDataLost(new Error("server socket errored before the queued data was delivered")); + onClosed(teardown); + }, + }, + }); + + const client = net.connect({ path: sock }); + client.on("error", () => {}); + try { + await new Promise((resolve, reject) => { + client.once("connect", resolve); + client.once("error", reject); + }); + const socket = await opened; + // Let the pause's writable dispatch drop the poll to zero events. + await new Promise(resolve => setImmediate(resolve)); + await new Promise(resolve => setImmediate(resolve)); + + await new Promise((resolve, reject) => client.write("hello", err => (err ? reject(err) : resolve()))); + client.destroy(); // graceful full close; unix sockets have no RST + + // Give a misrouted detector event plenty of turns to close the socket. + for (let i = 0; i < 10; i++) { + await new Promise(resolve => setImmediate(resolve)); + } + expect(teardown).toBeNull(); + + // resume() re-arms reading: the buffered bytes arrive, then the + // end-of-stream closes the (not half-open) socket cleanly. + socket.resume(); + expect(await gotData).toBe("hello"); + expect(await closed).toBe("close"); + } finally { + client.destroy(); + } + }); + + // A paused socket reaches the same zero-event state (pause arms WRITABLE, + // whose dispatch drops the poll to no events). A peer reset must still + // close it: Linux delivers EPOLLERR at zero interest; kqueue needs the + // detector knote and its fflags error routing. + it.skipIf(isWindows)("paused socket sees the peer reset", async () => { + const { promise: opened, resolve: onOpen } = Promise.withResolvers>(); + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + + using server = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + socket: { + open(socket) { + socket.pause(); + onOpen(socket); + }, + data() {}, + end() {}, + close() { + onClosed("close"); + }, + error() { + onClosed("error"); + }, + }, + }); + + const client = net.connect({ port: server.port, host: "127.0.0.1" }); + client.on("error", () => {}); + try { + await new Promise((resolve, reject) => { + client.once("connect", resolve); + client.once("error", reject); + }); + await opened; + // Let the pause's writable dispatch drop the poll to zero events. + await new Promise(resolve => setImmediate(resolve)); + await new Promise(resolve => setImmediate(resolve)); + + client.resetAndDestroy(); + + // Hangs forever on a deaf socket; the test timeout is the failure signal. + expect(await closed).toBeOneOf(["close", "error"]); + } finally { + client.destroy(); + } + }); + + // Replying to the peer's FIN with our own shutdown (what node:net's end() + // does) leaves the socket shut down at zero poll events, and the write-side + // shutdown echo is then its only close signal (the read knote went away + // with the consumed FIN). It must be delivered as the clean close, the way + // epoll's EPOLLHUP closes the same both-directions-shut state. + it.skipIf(isWindows)("allowHalfOpen socket closes cleanly after both sides shut down", async () => { + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + + using server = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + allowHalfOpen: true, + socket: { + open() {}, + data() {}, + end(socket) { + // The canonical reply to the peer's FIN: finish our side too. + // shutdown() half-closes natively; end() would close outright at + // the JS layer and never depend on the kernel's close signal. + socket.shutdown(); + }, + close() { + onClosed("close"); + }, + error() { + onClosed("error"); + }, + }, + }); + + const client = net.connect({ port: server.port, host: "127.0.0.1", allowHalfOpen: true }); + client.on("error", () => {}); + try { + await new Promise((resolve, reject) => { + client.once("connect", resolve); + client.once("error", reject); + }); + client.end(); + + // Hangs forever if the shutdown echo is suppressed at zero events. + expect(await closed).toBe("close"); + } finally { + client.destroy(); + } + }); + + // The reverse ordering: half-close while paused, then resume reading. The + // detector knote survives the 0 -> READABLE transition, so the stale + // shutdown echo is delivered while the socket legitimately reads; it must + // not clean-close the socket before the peer's reply and FIN arrive. + it.skipIf(isWindows)("half-closed paused socket still hears the peer after resume", async () => { + const { promise: opened, resolve: onOpen } = Promise.withResolvers>(); + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + let received = ""; + + using server = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + allowHalfOpen: true, + socket: { + open(socket) { + socket.pause(); + onOpen(socket); + }, + data(_socket, buffer) { + received += buffer.toString(); + }, + end() {}, + close() { + onClosed("close"); + }, + error() { + onClosed("error"); + }, + }, + }); + + // A raw half-open client: it must keep writing after the server's FIN + // arrives (node:net's client tears its stream down there). + const client = await Bun.connect({ + hostname: "127.0.0.1", + port: server.port, + allowHalfOpen: true, + socket: { open() {}, data() {}, end() {}, close() {}, error() {} }, + }); + try { + const socket = await opened; + // Let the pause's writable dispatch drop the poll to zero events and + // arm the detector. + await new Promise(resolve => setImmediate(resolve)); + await new Promise(resolve => setImmediate(resolve)); + + socket.shutdown(); // half-close while paused; the shutdown echo is now queued + socket.resume(); // re-arms reading; the detector knote survives + + // The peer answers in two chunks. A stale echo closing the socket + // early loses part or all of the reply. + client.write("part1"); + await new Promise(resolve => setImmediate(resolve)); + client.write("part2"); + client.end(); + + expect(await closed).toBe("close"); + expect(received).toBe("part1part2"); + } finally { + client.terminate(); + } + }); + it("upgradeTLS handles errors", async () => { using server = Bun.serve({ port: 0,