diff --git a/packages/bun-usockets/src/eventing/epoll_kqueue.c b/packages/bun-usockets/src/eventing/epoll_kqueue.c index 4dce3538d154..30231d4dbf1a 100644 --- a/packages/bun-usockets/src/eventing/epoll_kqueue.c +++ b/packages/bun-usockets/src/eventing/epoll_kqueue.c @@ -538,13 +538,34 @@ 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); + /* Unconditionally drop any write filter: a still-armed one-shot would + * report our own SS_CANTSENDMORE, and old_events cannot be trusted to + * know about it - pause() before shutdown() arms the 0-event fallback + * one-shot without recording it anywhere (ENOENT from the receipt is + * harmless when none exists). 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)) { diff --git a/packages/bun-usockets/src/internal/internal.h b/packages/bun-usockets/src/internal/internal.h index 38106f69c95a..ddaa9e0699e7 100644 --- a/packages/bun-usockets/src/internal/internal.h +++ b/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" diff --git a/packages/bun-usockets/src/libusockets.h b/packages/bun-usockets/src/libusockets.h index 1040e1377585..158ef203b84a 100644 --- a/packages/bun-usockets/src/libusockets.h +++ b/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 } diff --git a/packages/bun-usockets/src/socket.c b/packages/bun-usockets/src/socket.c index 25f798b1a017..2e3d6ca53928 100644 --- a/packages/bun-usockets/src/socket.c +++ b/packages/bun-usockets/src/socket.c @@ -418,6 +418,17 @@ static void us_internal_rearm_writable(struct us_socket_t *s) { 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; @@ -710,6 +721,28 @@ 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 + /* 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). + * AFTER the shutdown(2): macOS 26 only delivers the peer's close on + * a filter registered after SHUT_WR (node-http-halfclose-midupload + * timed out with the watch armed before it), and EV_EOF is level + * state, so a FIN landing in the gap is still reported by the fresh + * registration. */ + if (us_poll_events(&s->p) == 0) { + kqueue_change(s->group->loop->fd, us_poll_fd(&s->p), 0, 0, &s->p); + } else { + /* Still reading: scrub any phantom write one-shot the 0-event + * fallback armed during an earlier pause (poll_events never + * records it, so the diff above cannot see it; ENOENT is + * harmless when none exists). */ + kqueue_change(s->group->loop->fd, us_poll_fd(&s->p), + LIBUS_SOCKET_READABLE | LIBUS_SOCKET_WRITABLE, + LIBUS_SOCKET_READABLE, &s->p); + } +#endif } } @@ -840,8 +873,14 @@ void us_socket_pause(struct us_socket_t *s) { 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; } @@ -864,6 +903,11 @@ void us_socket_resume(struct us_socket_t *s) { 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); + /* Re-add readable, but like pause() only KEEP writable interest: any + * backpressure during the pause already armed it via + * us_internal_rearm_writable, and manufacturing it here fired a bogus + * drain on every pause/resume round trip (libuv's uv_read_start only + * adds POLLIN). */ + us_poll_change(&s->p, s->group->loop, + LIBUS_SOCKET_READABLE | (us_poll_events(&s->p) & LIBUS_SOCKET_WRITABLE)); } diff --git a/packages/bun-uws/src/HttpContext.h b/packages/bun-uws/src/HttpContext.h index 2a88ad687ad8..c3eb2d432d5f 100644 --- a/packages/bun-uws/src/HttpContext.h +++ b/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 *) 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::HTTP_NODE_READS_PAUSED; httpResponseData->nodeHttpParkAtNextBoundary = true; ((HttpResponse *) 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); } } } diff --git a/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp b/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp index e7d895f5e3b8..4f60efc5e64f 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp @@ -500,6 +500,11 @@ static void onNodeHttpReadsPaused(us_socket_t* socket) auto* d = reinterpret_cast*>(us_socket_ext(socket)); d->nodeHttpParkAtNextBoundary = true; d->state |= uWS::HttpResponseData::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) diff --git a/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp b/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp index 59e3be6835e9..fb5cd79f3f94 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp @@ -223,16 +223,16 @@ JSC_DEFINE_HOST_FUNCTION(jsFunctionNodeHTTPServerSocketEnd, (JSC::JSGlobalObject } auto bufferedSize = thisObject->streamBuffer.bufferedSize(); if (bufferedSize == 0) { - // onNodeHTTPRequest no longer pauses at dispatch; pause here so the - // shutdown+resume below still cycles kqueue's EVFILT_READ (delete then - // re-add), without which macOS 26 does not deliver the peer's close. + // Pause so the shutdown cycles kqueue's EVFILT_READ (pause drops it; + // us_internal_socket_raw_shutdown's teardown transition re-adds it, + // ordering documented there), without which macOS 26 does not + // deliver the peer's close. if (thisObject->socket && !thisObject->upgraded) { us_socket_pause(thisObject->socket); } auto result = us_socket_buffered_js_write(thisObject->socket, thisObject->is_ssl, thisObject->ended, &thisObject->streamBuffer, globalObject, JSValue::encode(JSC::jsUndefined()), JSValue::encode(JSC::jsUndefined())); - // Undo the pause above after the shutdown so the unread body drains - // and kqueue's one-shot EVFILT_WRITE (which delivers EV_EOF on - // SHUT_WR) is not deleted by a W -> R|W -> R step. + // Undo the pause after the shutdown so the unread body drains; the + // teardown transition already scrubbed the stale write filter. if (thisObject->socket && !thisObject->upgraded) { us_socket_resume(thisObject->socket); } diff --git a/src/uws_sys/libuwsockets.cpp b/src/uws_sys/libuwsockets.cpp index 989380a317b1..8419c35fb0fe 100644 --- a/src/uws_sys/libuwsockets.cpp +++ b/src/uws_sys/libuwsockets.cpp @@ -1834,11 +1834,7 @@ size_t uws_req_get_header(uws_req_t *res, const char *lower_case_header, void us_socket_mark_needs_more_not_ssl(uws_res_r res) { - us_socket_r s = (us_socket_t *)res; - if(us_socket_is_closed(s)) return; - s->flags.last_write_failed = 1; - us_poll_change(&s->p, s->group->loop, - LIBUS_SOCKET_READABLE | LIBUS_SOCKET_WRITABLE); + us_socket_mark_writable_pending((us_socket_t *)res); } void uws_res_override_write_offset(int ssl, uws_res_r res, uint64_t offset) @@ -2007,9 +2003,9 @@ __attribute__((callback (corker, ctx))) } void us_socket_sendfile_needs_more(us_socket_r s) { - if(us_socket_is_closed(s)) return; - s->flags.last_write_failed = 1; - us_poll_change(&s->p, s->group->loop, LIBUS_SOCKET_READABLE | LIBUS_SOCKET_WRITABLE); + /* The pending file tail lives outside the socket's write path; see + * us_socket_mark_writable_pending in libusockets.h. */ + us_socket_mark_writable_pending(s); } LIBUS_SOCKET_DESCRIPTOR us_socket_get_fd(us_socket_r s) { diff --git a/test/js/bun/net/socket.test.ts b/test/js/bun/net/socket.test.ts index f9e7ae1f55c1..d55145891818 100644 --- a/test/js/bun/net/socket.test.ts +++ b/test/js/bun/net/socket.test.ts @@ -362,6 +362,240 @@ describe.concurrent("socket", () => { expect(await bunRun(fileURLToPath(new URL("./kqueue-filter-coalesce-fixture.ts", import.meta.url)))).toSpawn(); }); + // Deterministic checkpoint for the negative assertions below: an echo round + // trip through an independent pair on the SAME event loop cannot complete + // before events that were already ready for other sockets have dispatched, + // so N round trips prove N full poll cycles ran. + async function loopCycles(n: number) { + using echo = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + socket: { + open() {}, + data(s, d) { + s.write(d); + }, + end() {}, + error() {}, + close() {}, + }, + }); + const done = Promise.withResolvers(); + let count = 0; + const opened = Promise.withResolvers(); + await Bun.connect({ + hostname: "127.0.0.1", + port: echo.port, + socket: { + open: s => opened.resolve(s), + data(s) { + if (++count >= n) done.resolve(); + else s.write("p"); + }, + end() {}, + error() {}, + close() {}, + }, + }); + const probe = await opened.promise; + probe.write("p"); + await done.promise; + probe.terminate(); + } + + // 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(); + 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 loopCycles(2); // any connect-time writable has dispatched + const before = drains; + s.pause(); + await loopCycles(3); // the buggy pause-armed writable would have fired + s.resume(); + await loopCycles(3); // resume() manufacturing WRITABLE would have too + 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(); + 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; + // The buggy kqueue EV_EOF close dispatched in the first poll cycle after + // the pause; several full cycles prove it is not coming. + await loopCycles(3); + const closed = closedEarly; + peer.terminate(); // release before asserting so a failure does not leak the socket + expect(closed).toBe(false); + }); + + // The sibling ordering: pause() first arms the 0-event fallback write + // one-shot before the socket is shut down, and the teardown transition must + // still scrub it or it echoes our own SS_CANTSENDMORE as EV_EOF. + it("pause() then shutdown() keeps a half-closed socket open while the peer is silent", async () => { + let closedEarly = false; + const opened = Promise.withResolvers(); + using server = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + socket: { + open(s) { + s.pause(); + s.shutdown(); + 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; + // Same checkpoint rationale as the shutdown-then-pause test. + await loopCycles(3); + const closed = closedEarly; + peer.terminate(); + expect(closed).toBe(false); + }); + + // Third ordering: pause() then resume() leaves the 0-event fallback's + // write one-shot armed but unrecorded; shutdown() must scrub it or it + // echoes our own SS_CANTSENDMORE. + it("pause() then resume() then shutdown() keeps a half-closed socket open while the peer is silent", async () => { + let closedEarly = false; + const opened = Promise.withResolvers(); + using server = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + socket: { + open(s) { + s.pause(); + s.resume(); + s.shutdown(); + 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; + // Same checkpoint rationale as the shutdown-then-pause test. + await loopCycles(3); + const closed = closedEarly; + peer.terminate(); + 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). Windows is + // excluded: the libuv backend has no event for a reset against a paused + // 0-event poll (AFD only reports subscribed events), a pre-existing gap + // tracked separately from this change. + it.skipIf(isWindows)("shutdown() then pause() still closes when the peer terminates", async () => { + const closed = Promise.withResolvers(); + const opened = Promise.withResolvers(); + 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(); + 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))],