From b49031bf098e4d217d146d41712323f88a4e2425 Mon Sep 17 00:00:00 2001 From: Ciro Spaciari MacBook Date: Thu, 6 Aug 2026 16:06:38 -0700 Subject: [PATCH 1/9] usockets(kqueue): close half-open sockets whose write side hit EV_EOF instead of spinning On kqueue EVFILT_WRITE is one-shot and the half-open EOF branch re-adds it unconditionally. Once the peer is fully gone the filter fires again immediately with EV_EOF, the writable dispatch lets the owner send() -> EPIPE which us_socket_write masked as backpressure and re-armed, and the loop spun a core forever with no error surfaced. Treat EV_EOF on EVFILT_WRITE for a socket we did not shut down and are no longer reading as an error, taking the same SO_ERROR close path epoll's EPOLLERR|EPOLLHUP already takes, and stop re-arming writable from the legacy write entry points on peer-gone errnos (as libuv's uv__write does). --- packages/bun-usockets/src/loop.c | 15 +++++ packages/bun-usockets/src/socket.c | 65 +++++++++++-------- .../node-http-server-socket-end-drain.test.ts | 13 +++- ...ode-http-upgrade-halfopen-reset-fixture.ts | 46 +++++++++++++ 4 files changed, 110 insertions(+), 29 deletions(-) create mode 100644 test/js/node/http/node-http-upgrade-halfopen-reset-fixture.ts diff --git a/packages/bun-usockets/src/loop.c b/packages/bun-usockets/src/loop.c index 18faa4f94478..de5bcaf1b0f2 100644 --- a/packages/bun-usockets/src/loop.c +++ b/packages/bun-usockets/src/loop.c @@ -586,6 +586,21 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in } /* The group can change after calling a callback but the loop is always the same */ struct us_loop_t* loop = s->group->loop; + #ifdef LIBUS_USE_KQUEUE + /* EV_EOF on EVFILT_WRITE we didn't shutdown() = SS_CANTSENDMORE (xnu bsd/kern/uipc_socket.c filt_sowrite); + * with the read side already ended nothing else closes us and the one-shot filter re-fires forever, so take + * the POLLERR close below like epoll's EPOLLERR|EPOLLHUP (libuv: deps/uv/src/unix/stream.c uv__write error path). */ + if ((events & LIBUS_SOCKET_WRITABLE) && eof && !error + && !(s->p.state.poll_type & POLL_TYPE_POLLING_IN) + && !s->flags.is_paused + && !us_socket_is_shut_down(s)) { + /* Clear POLLING_OUT for the consumed one-shot filter (POLLING_IN is already clear per the guard) */ + s->p.state.poll_type = us_internal_poll_type(&s->p); + error = 1; + eof = 0; /* on_end already ran when POLLING_IN was dropped (or the read side never + * started, e.g. low-prio parked mid-handshake) - error-close fits both */ + } + #endif if (events & LIBUS_SOCKET_WRITABLE && !error) { s->flags.last_write_failed = 0; #ifdef LIBUS_USE_KQUEUE diff --git a/packages/bun-usockets/src/socket.c b/packages/bun-usockets/src/socket.c index 25f798b1a017..1607429bf1ff 100644 --- a/packages/bun-usockets/src/socket.c +++ b/packages/bun-usockets/src/socket.c @@ -418,13 +418,46 @@ static void us_internal_rearm_writable(struct us_socket_t *s) { LIBUS_SOCKET_WRITABLE | (s->flags.is_paused ? 0 : LIBUS_SOCKET_READABLE)); } +#ifndef _WIN32 +/* send() errnos that mean the peer or the path to it is gone: no retry can ever + * succeed, so report them to the caller now; only the EAGAIN/ENOBUFS class waits for + * another writable event (https://github.com/libuv/libuv/blob/v1.51.0/src/unix/stream.c#L820-L837). */ +static int us_internal_send_errno_is_peer_gone(int e) { + switch (e) { + case EPIPE: + case ECONNRESET: + case ECONNABORTED: + case ENOTCONN: + case ETIMEDOUT: + case ENETDOWN: + case ENETUNREACH: + case EHOSTUNREACH: + return 1; + default: + return 0; + } +} +#endif + +/* Peer-gone errnos are final: don't wait for writable on them (libuv fails the req and stops POLLOUT, + * deps/uv/src/unix/stream.c uv__try_write/uv__write). Re-arming would spin kqueue's one-shot EVFILT_WRITE, + * which re-fires with EV_EOF; loop.c's error/EOF paths close the socket instead. */ +static int us_internal_send_should_rearm(ssize_t written) { +#ifndef _WIN32 + if (written < 0 && us_internal_send_errno_is_peer_gone(errno)) { + return 0; + } +#endif + return 1; +} + 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; } int written = bsd_write2(us_poll_fd(&s->p), header, header_length, payload, payload_length); - if (written != header_length + payload_length) { + if (written != header_length + payload_length && us_internal_send_should_rearm(written)) { us_internal_rearm_writable(s); } @@ -497,7 +530,7 @@ int us_socket_write(struct us_socket_t *s, const char *data, int length) { } int written = bsd_send(us_poll_fd(&s->p), data, length); - if (written != length) { + if (written != length && us_internal_send_should_rearm(written)) { s->flags.last_write_failed = 1; us_internal_rearm_writable(s); } @@ -506,28 +539,6 @@ int us_socket_write(struct us_socket_t *s, const char *data, int length) { } #ifndef _WIN32 -/* send() errnos that mean the peer or the path to it is gone: no retry can - * ever succeed, so they are reported to the caller immediately (libuv fails - * the write request for these in uv__try_write, unix/stream.c; only the - * EAGAIN/ENOBUFS class waits for another writable event). Mirrored by the - * blanket `result < -1` fatal handling in h2_frame_parser's - * is_transport_fatal_write_result - classification lives here only. */ -static int us_internal_send_errno_is_peer_gone(int e) { - switch (e) { - case EPIPE: - case ECONNRESET: - case ECONNABORTED: - case ENOTCONN: - case ETIMEDOUT: - case ENETDOWN: - case ENETUNREACH: - case EHOSTUNREACH: - return 1; - default: - return 0; - } -} - /* One retry runs per writable dispatch (one event-loop iteration), so 32 * consecutive failures is far beyond any observed transient race window * (the macOS EPROTOTYPE race resolves within a dispatch or two) while still @@ -618,7 +629,7 @@ int us_socket_raw_writev(struct us_socket_t *s, const struct us_iovec_t *iov, in for (int i = 0; i < count; i++) total += iov[i].iov_len; ssize_t written = bsd_writev(us_poll_fd(&s->p), iov, count); - if (written != (ssize_t)total) { + if (written != (ssize_t)total && us_internal_send_should_rearm(written)) { s->flags.last_write_failed = 1; us_internal_rearm_writable(s); } @@ -637,7 +648,7 @@ int us_socket_raw_write(struct us_socket_t *s, const char *data, int length) { } int written = bsd_send(us_poll_fd(&s->p), data, length); - if (written != length) { + if (written != length && us_internal_send_should_rearm(written)) { s->flags.last_write_failed = 1; us_internal_rearm_writable(s); } @@ -674,7 +685,7 @@ int us_socket_ipc_write_fd(struct us_socket_t *s, const char *data, int length, int sent = bsd_sendmsg(us_poll_fd(&s->p), &msg, 0); - if (sent != length) { + if (sent != length && us_internal_send_should_rearm(sent)) { s->flags.last_write_failed = 1; us_internal_rearm_writable(s); } diff --git a/test/js/node/http/node-http-server-socket-end-drain.test.ts b/test/js/node/http/node-http-server-socket-end-drain.test.ts index 05d1a0e74443..891bb4480497 100644 --- a/test/js/node/http/node-http-server-socket-end-drain.test.ts +++ b/test/js/node/http/node-http-server-socket-end-drain.test.ts @@ -1,12 +1,13 @@ import { expect, test } from "bun:test"; -import { bunEnv, bunExe } from "harness"; +import { bunEnv, bunExe, bunRun } from "harness"; +import path from "node:path"; // res.socket.end() half-closes the connection; the server must still release the // socket (drain the unconsumed body on epoll, or take kqueue's EVFILT_WRITE // EV_EOF from its own SHUT_WR) so server.close() resolves. On macOS that early // close can RST the still-writing client, so the client's EPIPE is expected and // the close wait must not be once(c, "close"), which would reject on it. -test("server.close() completes after res.socket.end() with a 2 MB upload in flight", async () => { +test.concurrent("server.close() completes after res.socket.end() with a 2 MB upload in flight", async () => { await using proc = Bun.spawn({ cmd: [ bunExe(), @@ -57,3 +58,11 @@ test("server.close() completes after res.socket.end() with a 2 MB upload in flig const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); expect({ stdout, stderr, exitCode }).toEqual({ stdout: "closed destroyed=true\n", stderr: "", exitCode: 0 }); }); + +// Half-open Upgrade socket with queued writes; peer FINs then RSTs. kqueue reports the RST as EV_EOF on the +// one-shot EVFILT_WRITE: the loop must close (like epoll's EPOLLERR|EPOLLHUP / libuv uv__write's error path), +// not re-arm forever. Before the fix this spun a core on macOS and 'close' never fired. +test.concurrent("upgrade socket with queued writes is closed, not spun, when the half-closed peer resets", async () => { + const { stdout, exitCode } = await bunRun(path.join(import.meta.dir, "node-http-upgrade-halfopen-reset-fixture.ts")); + expect({ stdout, exitCode }).toEqual({ stdout: "closed", exitCode: 0 }); +}); diff --git a/test/js/node/http/node-http-upgrade-halfopen-reset-fixture.ts b/test/js/node/http/node-http-upgrade-halfopen-reset-fixture.ts new file mode 100644 index 000000000000..13ef6ac9180c --- /dev/null +++ b/test/js/node/http/node-http-upgrade-halfopen-reset-fixture.ts @@ -0,0 +1,46 @@ +// Half-open Upgrade socket with queued writes; the peer never reads, FINs, then RSTs. +// The server socket must be closed (and the process exit), not left spinning on the +// one-shot EVFILT_WRITE that kqueue keeps re-delivering with EV_EOF. +import { once } from "node:events"; +import http from "node:http"; +import net from "node:net"; + +const server = http.createServer((req, res) => res.end()); +const closed = Promise.withResolvers(); +const halfClosed = Promise.withResolvers(); +const backedUp = Promise.withResolvers(); +server.on("upgrade", (req, socket) => { + socket.on("error", () => {}); + socket.on("close", () => closed.resolve()); + // Like ws: answer the peer's FIN with end(), which queues behind the pending writes. + // 'end' firing also means read interest was already dropped, the state the fix keys on. + socket.on("end", () => { + socket.end(); + halfClosed.resolve(); + }); + // Flowing mode so the peer's FIN is consumed and 'end' actually fires (node semantics). + socket.resume(); + socket.write("HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: x\r\n\r\n"); + const chunk = Buffer.alloc(4 * 1024 * 1024, 0x61); + for (let i = 0; i < 8; i++) socket.write(chunk); + backedUp.resolve(); +}); +await once(server.listen(0, "127.0.0.1"), "listening"); + +const c = net.connect((server.address() as net.AddressInfo).port, "127.0.0.1"); +c.on("error", () => {}); +// Never read: the server's 32 MB stays queued behind a full pipe. +c.pause(); +await once(c, "connect"); +c.write("GET / HTTP/1.1\r\nHost: x\r\nConnection: Upgrade\r\nUpgrade: x\r\n\r\n"); +await backedUp.promise; +// Half-close once the server has its writes queued. +c.end(); +// The server has read our FIN; now vanish. +await halfClosed.promise; +c.resetAndDestroy(); + +await closed.promise; +console.log("closed"); +server.close(); +// Exiting naturally (no process.exit) proves the loop went idle instead of spinning. From 90907efa90d46dcd60a155c2564de160a63aefba Mon Sep 17 00:00:00 2001 From: Ciro Spaciari MacBook Date: Thu, 6 Aug 2026 16:26:33 -0700 Subject: [PATCH 2/9] test: assert fixture stderr is empty --- test/js/node/http/node-http-server-socket-end-drain.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/js/node/http/node-http-server-socket-end-drain.test.ts b/test/js/node/http/node-http-server-socket-end-drain.test.ts index 891bb4480497..adce53d5b787 100644 --- a/test/js/node/http/node-http-server-socket-end-drain.test.ts +++ b/test/js/node/http/node-http-server-socket-end-drain.test.ts @@ -63,6 +63,6 @@ test.concurrent("server.close() completes after res.socket.end() with a 2 MB upl // one-shot EVFILT_WRITE: the loop must close (like epoll's EPOLLERR|EPOLLHUP / libuv uv__write's error path), // not re-arm forever. Before the fix this spun a core on macOS and 'close' never fired. test.concurrent("upgrade socket with queued writes is closed, not spun, when the half-closed peer resets", async () => { - const { stdout, exitCode } = await bunRun(path.join(import.meta.dir, "node-http-upgrade-halfopen-reset-fixture.ts")); - expect({ stdout, exitCode }).toEqual({ stdout: "closed", exitCode: 0 }); + const { stdout, stderr, exitCode } = await bunRun(path.join(import.meta.dir, "node-http-upgrade-halfopen-reset-fixture.ts")); + expect({ stdout, stderr, exitCode }).toEqual({ stdout: "closed", stderr: "", exitCode: 0 }); }); From 71548fe9eed7cd3894b2051e4cc25f621792a2e7 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Thu, 6 Aug 2026 23:30:56 +0000 Subject: [PATCH 3/9] [autofix.ci] apply automated fixes --- src/bun_core/tty.rs | 4 +++- src/md/ansi_renderer.rs | 6 +++++- src/runtime/bake/bake_body.rs | 2 +- src/runtime/server/server_body.rs | 5 ++++- test/js/node/http/node-http-server-socket-end-drain.test.ts | 4 +++- 5 files changed, 16 insertions(+), 5 deletions(-) diff --git a/src/bun_core/tty.rs b/src/bun_core/tty.rs index 29926732e65a..4bf156f91360 100644 --- a/src/bun_core/tty.rs +++ b/src/bun_core/tty.rs @@ -89,7 +89,9 @@ impl RawModeGuard { impl Drop for RawModeGuard { #[inline] fn drop(&mut self) { - let _ = self.state.set_mode(self.fd, Mode::Normal, SetAttrWhen::Drain); + let _ = self + .state + .set_mode(self.fd, Mode::Normal, SetAttrWhen::Drain); } } diff --git a/src/md/ansi_renderer.rs b/src/md/ansi_renderer.rs index b35a5419d7c0..0deed8793f5d 100644 --- a/src/md/ansi_renderer.rs +++ b/src/md/ansi_renderer.rs @@ -2535,7 +2535,11 @@ fn probe_kitty_graphics() -> bool { Err(_) => return false, }; let mut tty_state = bun_core::tty::State::new(); - let _ = tty_state.set_mode(0, bun_core::tty::Mode::Raw, bun_core::tty::SetAttrWhen::Drain); + let _ = tty_state.set_mode( + 0, + bun_core::tty::Mode::Raw, + bun_core::tty::SetAttrWhen::Drain, + ); let _restore = scopeguard::guard((saved_termios, tty_state), |(saved, mut state)| { if bun_sys::posix::tcsetattr(0, bun_sys::posix::TCSA::Now, &saved).is_err() { let _ = state.set_mode( diff --git a/src/runtime/bake/bake_body.rs b/src/runtime/bake/bake_body.rs index 3e4cc72b47bb..cdd8ab2dadcd 100644 --- a/src/runtime/bake/bake_body.rs +++ b/src/runtime/bake/bake_body.rs @@ -10,8 +10,8 @@ use core::ptr::NonNull; use bun_alloc::Arena; // = bumpalo::Bump use bun_collections::ArrayHashMap; use bun_core::Output; -use bun_jsc::{JSGlobalObject, JSValue, JsError, JsResult, ZigStringSlice}; use bun_core::{ZStr, strings}; +use bun_jsc::{JSGlobalObject, JSValue, JsError, JsResult, ZigStringSlice}; use bun_options_types::schema as bun_schema; use bun_paths::{self as paths, PathBuffer}; diff --git a/src/runtime/server/server_body.rs b/src/runtime/server/server_body.rs index 309f72865893..2867551551ae 100644 --- a/src/runtime/server/server_body.rs +++ b/src/runtime/server/server_body.rs @@ -492,7 +492,10 @@ pub mod BunInfo { // `JSON.toAST(allocator, BunInfo, info)` — hand-expanded: let platform_props = bun_alloc::AstAlloc::vec_from_iter([ prop(b"os", str_expr(os_tag_name(info.platform.os))), - prop(b"arch", str_expr(arch_tag_name(bun_core::Environment::ARCH))), + prop( + b"arch", + str_expr(arch_tag_name(bun_core::Environment::ARCH)), + ), prop(b"version", str_expr(info.platform.version)), ]); let platform_expr = Expr::init( diff --git a/test/js/node/http/node-http-server-socket-end-drain.test.ts b/test/js/node/http/node-http-server-socket-end-drain.test.ts index adce53d5b787..b962b3b72562 100644 --- a/test/js/node/http/node-http-server-socket-end-drain.test.ts +++ b/test/js/node/http/node-http-server-socket-end-drain.test.ts @@ -63,6 +63,8 @@ test.concurrent("server.close() completes after res.socket.end() with a 2 MB upl // one-shot EVFILT_WRITE: the loop must close (like epoll's EPOLLERR|EPOLLHUP / libuv uv__write's error path), // not re-arm forever. Before the fix this spun a core on macOS and 'close' never fired. test.concurrent("upgrade socket with queued writes is closed, not spun, when the half-closed peer resets", async () => { - const { stdout, stderr, exitCode } = await bunRun(path.join(import.meta.dir, "node-http-upgrade-halfopen-reset-fixture.ts")); + const { stdout, stderr, exitCode } = await bunRun( + path.join(import.meta.dir, "node-http-upgrade-halfopen-reset-fixture.ts"), + ); expect({ stdout, stderr, exitCode }).toEqual({ stdout: "closed", stderr: "", exitCode: 0 }); }); From 8379d507e801f215eddef1e4d566262f9a5e1031 Mon Sep 17 00:00:00 2001 From: Ciro Spaciari MacBook Date: Thu, 6 Aug 2026 16:35:01 -0700 Subject: [PATCH 4/9] yaml: allow needless_pass_by_value on bind_anchor PendingAnchor is a #[must_use] token deliberately consumed by binding, so by-value is the design; clippy (which only runs on PRs, not main pushes) flags it on every PR since #37055 landed. No-Verification-Needed: lint attribute only, no runtime behavior change --- src/parsers/yaml.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/parsers/yaml.rs b/src/parsers/yaml.rs index af318a3d03d1..f28607eb5e9e 100644 --- a/src/parsers/yaml.rs +++ b/src/parsers/yaml.rs @@ -3311,6 +3311,7 @@ impl CollectionData for E::Object { } impl<'i, Enc: Encoding> Parser<'i, Enc> { + #[allow(clippy::needless_pass_by_value)] // must_use token: binding consumes the anchor fn bind_anchor(&mut self, anchor: PendingAnchor, node: Expr) -> Result<(), AllocError> { self.anchors .put(Enc::key_bytes(anchor.name.slice(self.input)), node) From 17b5cb082a09c232d9b92eee46013014d877dbc4 Mon Sep 17 00:00:00 2001 From: Ciro Spaciari MacBook Date: Thu, 6 Aug 2026 16:41:22 -0700 Subject: [PATCH 5/9] socket: allow clippy::unit_arg in uws_handlers swallow() is a deliberate polymorphic sink over () and Result<(), E> consumer methods; routing the unit-returning ones through it is the module's design. This lint was masked in CI until bun_parsers compiled again (clippy only runs on PRs, and a failing dep crate hides its dependents' lints). No-Verification-Needed: lint attribute only, no runtime behavior change --- src/runtime/socket/uws_handlers.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/runtime/socket/uws_handlers.rs b/src/runtime/socket/uws_handlers.rs index 5bc312e75db5..8a0a1d8bdd27 100644 --- a/src/runtime/socket/uws_handlers.rs +++ b/src/runtime/socket/uws_handlers.rs @@ -8,6 +8,10 @@ //! old `NewSocketHandler.configure`/`unsafeConfigure` machinery, which built //! the same trampolines at runtime per `us_socket_context_t`. +// `swallow()` uniformly sinks `()` and `Result<(), E>` consumers; passing the +// unit-returning ones through it is the point, not an accident. +#![allow(clippy::unit_arg)] + use bun_ptr::ThisPtr; use core::ffi::{c_int, c_void}; use core::ptr::NonNull; From 07b01096a50a6c96e7143409f59059c797060a30 Mon Sep 17 00:00:00 2001 From: Ciro Spaciari MacBook Date: Thu, 6 Aug 2026 16:57:58 -0700 Subject: [PATCH 6/9] usockets: preserve read interest in us_internal_rearm_writable us_poll_change sets absolute flags, and rearm_writable forced READABLE back on for any non-paused socket. For a half-open socket whose read side already hit EOF that re-arms the dead read filter, which re-fires and delivers on_end a second time. Read interest is only ever off because something dropped it deliberately (pause, half-open EOF, low-prio parking), so preserve it instead of forcing it on. --- packages/bun-usockets/src/socket.c | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/packages/bun-usockets/src/socket.c b/packages/bun-usockets/src/socket.c index 1607429bf1ff..b53840906ca6 100644 --- a/packages/bun-usockets/src/socket.c +++ b/packages/bun-usockets/src/socket.c @@ -409,13 +409,14 @@ struct us_socket_t *us_socket_pair(struct us_socket_group_t *group, unsigned cha #endif } -/* 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. */ +/* Re-arm writable for a backpressured write, preserving the current read + * interest: us_poll_change sets absolute flags, and read interest is only ever + * off because something dropped it deliberately - us_socket_pause, the + * half-open EOF branch (re-adding it would re-deliver EOF and fire on_end + * again), or low-priority parking. Forcing READABLE back on undid all three. */ 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)); + LIBUS_SOCKET_WRITABLE | (us_poll_events(&s->p) & LIBUS_SOCKET_READABLE)); } #ifndef _WIN32 From 6bc53ffdbaf05d1cd2092514a1f595bcebd55862 Mon Sep 17 00:00:00 2001 From: Ciro Spaciari MacBook Date: Thu, 6 Aug 2026 16:58:24 -0700 Subject: [PATCH 7/9] test: add half-open peer-reset test matrix Nine parallel tests (Bun.listen/Bun.serve plain+TLS, fetch upload, node:http/https response+upgrade) driving a backpressured socket whose peer stops reading, FINs, then RSTs: the victim must close, with 'end' delivered at most once. From #37077. --- test/js/bun/net/half-open-peer-reset.mjs | 195 ++++++++++++++++++ ...et-half-open-peer-reset-bun-listen-tls.mjs | 3 + ...st-net-half-open-peer-reset-bun-listen.mjs | 3 + ...net-half-open-peer-reset-bun-serve-tls.mjs | 3 + ...est-net-half-open-peer-reset-bun-serve.mjs | 3 + ...-net-half-open-peer-reset-fetch-upload.mjs | 3 + ...alf-open-peer-reset-node-http-response.mjs | 3 + ...half-open-peer-reset-node-http-upgrade.mjs | 3 + ...lf-open-peer-reset-node-https-response.mjs | 3 + ...alf-open-peer-reset-node-https-upgrade.mjs | 3 + 10 files changed, 222 insertions(+) create mode 100644 test/js/bun/net/half-open-peer-reset.mjs create mode 100644 test/js/bun/test/parallel/test-net-half-open-peer-reset-bun-listen-tls.mjs create mode 100644 test/js/bun/test/parallel/test-net-half-open-peer-reset-bun-listen.mjs create mode 100644 test/js/bun/test/parallel/test-net-half-open-peer-reset-bun-serve-tls.mjs create mode 100644 test/js/bun/test/parallel/test-net-half-open-peer-reset-bun-serve.mjs create mode 100644 test/js/bun/test/parallel/test-net-half-open-peer-reset-fetch-upload.mjs create mode 100644 test/js/bun/test/parallel/test-net-half-open-peer-reset-node-http-response.mjs create mode 100644 test/js/bun/test/parallel/test-net-half-open-peer-reset-node-http-upgrade.mjs create mode 100644 test/js/bun/test/parallel/test-net-half-open-peer-reset-node-https-response.mjs create mode 100644 test/js/bun/test/parallel/test-net-half-open-peer-reset-node-https-upgrade.mjs diff --git a/test/js/bun/net/half-open-peer-reset.mjs b/test/js/bun/net/half-open-peer-reset.mjs new file mode 100644 index 000000000000..9c2e9a9eac31 --- /dev/null +++ b/test/js/bun/net/half-open-peer-reset.mjs @@ -0,0 +1,195 @@ +// Body of test/js/bun/test/parallel/test-net-half-open-peer-reset-*: a backpressured socket whose peer stops reading, FINs, then RSTs must close (with `end` at most once), not spin. +import { tls as tlsCert } from "../../../harness"; +import { once } from "node:events"; +import http from "node:http"; +import https from "node:https"; + +const big = Buffer.alloc(4 * 1024 * 1024, 0x78); +const { key, cert } = tlsCert; +const upgradeReq = "GET / HTTP/1.1\r\nHost: x\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n"; +const getReq = "GET / HTTP/1.1\r\nHost: x\r\n\r\n"; + +export async function run(mode) { + const issued = Promise.withResolvers(); // victim has queued more than the kernel will buffer + const ended = Promise.withResolvers(); // victim saw the peer's FIN (half-open modes) + const closed = Promise.withResolvers(); // victim socket/request was torn down + let endCount = 0; + const onEnd = () => { + if (++endCount > 1) fail("end delivered " + endCount + " times"); + ended.resolve(); + }; + const fail = why => { + console.error("FAIL", why); + process.exit(1); + }; + + // The peer is a raw Bun socket so FIN (shutdown) and RST (terminate) are exactly what hits the wire, in that order. + async function peer(port, { preface, useTls, waitForEnd }) { + const opened = Promise.withResolvers(); + const s = await Bun.connect({ + hostname: "127.0.0.1", + port, + allowHalfOpen: true, + tls: useTls ? { rejectUnauthorized: false } : false, + socket: { + open: s => (useTls ? undefined : opened.resolve(s)), + handshake: s => opened.resolve(s), + data() {}, + end() {}, + error() {}, + close() {}, + }, + }); + await opened.promise; + if (preface) s.write(preface); + s.flush(); + s.pause(); + await issued.promise; + s.shutdown(); + if (waitForEnd) await Promise.race([ended.promise, closed.promise]); + // The FIN→RST gap is where the bug lived; an immediate RST can overtake the FIN and just read as ECONNRESET. + await Bun.sleep(50); + s.terminate(); + const deadline = setTimeout(() => fail("still open 5s after peer reset"), 5000); + closed.promise.then(() => clearTimeout(deadline)); + } + + switch (mode) { + case "bun-listen": + case "bun-listen-tls": { + const useTls = mode === "bun-listen-tls"; + using server = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + allowHalfOpen: true, + tls: useTls ? { key, cert } : undefined, + socket: { + open(s) { + s.write(big); + issued.resolve(); + }, + data() {}, + drain(s) { + s.write(big); + }, + end: onEnd, + error() {}, + close: () => closed.resolve(""), + }, + }); + await peer(server.port, { useTls, waitForEnd: true }); + await closed.promise; + break; + } + case "node-http-upgrade": + case "node-https-upgrade": { + const useTls = mode === "node-https-upgrade"; + await using server = (useTls ? https : http).createServer(useTls ? { key, cert } : {}, (_q, res) => res.end()); + server.on("upgrade", (_req, socket) => { + socket.on("error", () => {}); + socket.on("close", () => closed.resolve("")); + socket.on("end", () => { + onEnd(); + socket.end(); + }); + socket.write("HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n"); + for (let i = 0; i < 8; i++) socket.write(big); + issued.resolve(); + }); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + await peer(server.address().port, { preface: upgradeReq, useTls, waitForEnd: true }); + await closed.promise; + break; + } + case "node-http-response": + case "node-https-response": { + const useTls = mode === "node-https-response"; + await using server = (useTls ? https : http).createServer(useTls ? { key, cert } : {}, (_req, res) => { + res.on("close", () => closed.resolve("res")); + res.writeHead(200); + for (let i = 0; i < 8; i++) res.write(big); + issued.resolve(); + }); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + await peer(server.address().port, { preface: getReq, useTls, waitForEnd: false }); + await closed.promise; + break; + } + case "bun-serve": + case "bun-serve-tls": { + const useTls = mode === "bun-serve-tls"; + using server = Bun.serve({ + port: 0, + hostname: "127.0.0.1", + idleTimeout: 0, + tls: useTls ? { key, cert } : undefined, + fetch(req) { + req.signal.addEventListener("abort", () => closed.resolve("abort")); + let i = 0; + return new Response( + new ReadableStream({ + pull(ctrl) { + if (i++ < 64) ctrl.enqueue(big); + else ctrl.close(); + issued.resolve(); + }, + cancel: () => closed.resolve("cancel"), + }), + ); + }, + }); + await peer(server.port, { preface: getReq, useTls, waitForEnd: false }); + await closed.promise; + break; + } + case "fetch-upload": { + // Roles flipped: the server is the peer that stops reading / FINs / RSTs; fetch() is the victim with a pending body. + using server = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + allowHalfOpen: true, + socket: { + async open(s) { + s.pause(); + await issued.promise; + s.shutdown(); + await Bun.sleep(50); + s.terminate(); + const deadline = setTimeout(() => fail("fetch still pending 5s after peer reset"), 5000); + closed.promise.then(() => clearTimeout(deadline)); + }, + data() {}, + end() {}, + error() {}, + close() {}, + }, + }); + let i = 0; + fetch(`http://127.0.0.1:${server.port}/`, { + method: "POST", + duplex: "half", + body: new ReadableStream({ + pull(c) { + if (i++ < 64) c.enqueue(big); + else c.close(); + issued.resolve(); + }, + }), + }).then( + r => + r.text().then( + () => closed.resolve("resolved"), + e => closed.resolve("body rejected " + e?.code), + ), + e => closed.resolve("rejected " + e?.code), + ); + await closed.promise; + break; + } + default: + fail("unknown mode " + mode); + } + console.log("closed", await closed.promise); +} diff --git a/test/js/bun/test/parallel/test-net-half-open-peer-reset-bun-listen-tls.mjs b/test/js/bun/test/parallel/test-net-half-open-peer-reset-bun-listen-tls.mjs new file mode 100644 index 000000000000..629bc54ffeab --- /dev/null +++ b/test/js/bun/test/parallel/test-net-half-open-peer-reset-bun-listen-tls.mjs @@ -0,0 +1,3 @@ +// https://github.com/oven-sh/bun/pull/37077 +import { run } from "../../net/half-open-peer-reset.mjs"; +await run("bun-listen-tls"); diff --git a/test/js/bun/test/parallel/test-net-half-open-peer-reset-bun-listen.mjs b/test/js/bun/test/parallel/test-net-half-open-peer-reset-bun-listen.mjs new file mode 100644 index 000000000000..ccb2eda53d82 --- /dev/null +++ b/test/js/bun/test/parallel/test-net-half-open-peer-reset-bun-listen.mjs @@ -0,0 +1,3 @@ +// https://github.com/oven-sh/bun/pull/37077 +import { run } from "../../net/half-open-peer-reset.mjs"; +await run("bun-listen"); diff --git a/test/js/bun/test/parallel/test-net-half-open-peer-reset-bun-serve-tls.mjs b/test/js/bun/test/parallel/test-net-half-open-peer-reset-bun-serve-tls.mjs new file mode 100644 index 000000000000..e1f4502ebe53 --- /dev/null +++ b/test/js/bun/test/parallel/test-net-half-open-peer-reset-bun-serve-tls.mjs @@ -0,0 +1,3 @@ +// https://github.com/oven-sh/bun/pull/37077 +import { run } from "../../net/half-open-peer-reset.mjs"; +await run("bun-serve-tls"); diff --git a/test/js/bun/test/parallel/test-net-half-open-peer-reset-bun-serve.mjs b/test/js/bun/test/parallel/test-net-half-open-peer-reset-bun-serve.mjs new file mode 100644 index 000000000000..719c10bc7520 --- /dev/null +++ b/test/js/bun/test/parallel/test-net-half-open-peer-reset-bun-serve.mjs @@ -0,0 +1,3 @@ +// https://github.com/oven-sh/bun/pull/37077 +import { run } from "../../net/half-open-peer-reset.mjs"; +await run("bun-serve"); diff --git a/test/js/bun/test/parallel/test-net-half-open-peer-reset-fetch-upload.mjs b/test/js/bun/test/parallel/test-net-half-open-peer-reset-fetch-upload.mjs new file mode 100644 index 000000000000..3fa0a19c0a9d --- /dev/null +++ b/test/js/bun/test/parallel/test-net-half-open-peer-reset-fetch-upload.mjs @@ -0,0 +1,3 @@ +// https://github.com/oven-sh/bun/pull/37077 +import { run } from "../../net/half-open-peer-reset.mjs"; +await run("fetch-upload"); diff --git a/test/js/bun/test/parallel/test-net-half-open-peer-reset-node-http-response.mjs b/test/js/bun/test/parallel/test-net-half-open-peer-reset-node-http-response.mjs new file mode 100644 index 000000000000..e5004e3f8aa1 --- /dev/null +++ b/test/js/bun/test/parallel/test-net-half-open-peer-reset-node-http-response.mjs @@ -0,0 +1,3 @@ +// https://github.com/oven-sh/bun/pull/37077 +import { run } from "../../net/half-open-peer-reset.mjs"; +await run("node-http-response"); diff --git a/test/js/bun/test/parallel/test-net-half-open-peer-reset-node-http-upgrade.mjs b/test/js/bun/test/parallel/test-net-half-open-peer-reset-node-http-upgrade.mjs new file mode 100644 index 000000000000..6461019c9138 --- /dev/null +++ b/test/js/bun/test/parallel/test-net-half-open-peer-reset-node-http-upgrade.mjs @@ -0,0 +1,3 @@ +// https://github.com/oven-sh/bun/pull/37077 +import { run } from "../../net/half-open-peer-reset.mjs"; +await run("node-http-upgrade"); diff --git a/test/js/bun/test/parallel/test-net-half-open-peer-reset-node-https-response.mjs b/test/js/bun/test/parallel/test-net-half-open-peer-reset-node-https-response.mjs new file mode 100644 index 000000000000..a65a4813f923 --- /dev/null +++ b/test/js/bun/test/parallel/test-net-half-open-peer-reset-node-https-response.mjs @@ -0,0 +1,3 @@ +// https://github.com/oven-sh/bun/pull/37077 +import { run } from "../../net/half-open-peer-reset.mjs"; +await run("node-https-response"); diff --git a/test/js/bun/test/parallel/test-net-half-open-peer-reset-node-https-upgrade.mjs b/test/js/bun/test/parallel/test-net-half-open-peer-reset-node-https-upgrade.mjs new file mode 100644 index 000000000000..82f7821ffadd --- /dev/null +++ b/test/js/bun/test/parallel/test-net-half-open-peer-reset-node-https-upgrade.mjs @@ -0,0 +1,3 @@ +// https://github.com/oven-sh/bun/pull/37077 +import { run } from "../../net/half-open-peer-reset.mjs"; +await run("node-https-upgrade"); From 7c0b0e48bcb4b172e3bc300761be0c14a3f090c9 Mon Sep 17 00:00:00 2001 From: Ciro Spaciari MacBook Date: Thu, 6 Aug 2026 17:02:00 -0700 Subject: [PATCH 8/9] usockets: only skip the peer-gone rearm for paused sockets Skipping the rearm for every peer-gone failure left a race window where a half-open socket ended up with no kqueue filter registered at all: a writable event without EV_EOF, an RST landing before the flush send(), and the epilogue no-ops with zero events - nothing ever dispatches again. Re-arming the one-shot EVFILT_WRITE instead makes the next dispatch carry EV_EOF and take the loop.c error-close. Paused sockets keep the skip: the close branch deliberately excludes them, so their re-arm would spin. --- packages/bun-usockets/src/socket.c | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/packages/bun-usockets/src/socket.c b/packages/bun-usockets/src/socket.c index b53840906ca6..a6668a8333ee 100644 --- a/packages/bun-usockets/src/socket.c +++ b/packages/bun-usockets/src/socket.c @@ -440,12 +440,13 @@ static int us_internal_send_errno_is_peer_gone(int e) { } #endif -/* Peer-gone errnos are final: don't wait for writable on them (libuv fails the req and stops POLLOUT, - * deps/uv/src/unix/stream.c uv__try_write/uv__write). Re-arming would spin kqueue's one-shot EVFILT_WRITE, - * which re-fires with EV_EOF; loop.c's error/EOF paths close the socket instead. */ -static int us_internal_send_should_rearm(ssize_t written) { +/* On a peer-gone errno a PAUSED socket must not re-arm: kqueue's one-shot EVFILT_WRITE would + * re-fire with EV_EOF forever, and the loop.c close branch deliberately excludes paused sockets + * (deferred-EOF keeps their unread data). Everything else still re-arms so a live socket always + * has a filter registered - the re-fired EV_EOF then takes the loop.c error-close next tick. */ +static int us_internal_send_should_rearm(struct us_socket_t *s, ssize_t written) { #ifndef _WIN32 - if (written < 0 && us_internal_send_errno_is_peer_gone(errno)) { + if (written < 0 && s->flags.is_paused && us_internal_send_errno_is_peer_gone(errno)) { return 0; } #endif @@ -458,7 +459,7 @@ int us_socket_write2(struct us_socket_t *s, const char *header, int header_lengt } int written = bsd_write2(us_poll_fd(&s->p), header, header_length, payload, payload_length); - if (written != header_length + payload_length && us_internal_send_should_rearm(written)) { + if (written != header_length + payload_length && us_internal_send_should_rearm(s, written)) { us_internal_rearm_writable(s); } @@ -531,7 +532,7 @@ int us_socket_write(struct us_socket_t *s, const char *data, int length) { } int written = bsd_send(us_poll_fd(&s->p), data, length); - if (written != length && us_internal_send_should_rearm(written)) { + if (written != length && us_internal_send_should_rearm(s, written)) { s->flags.last_write_failed = 1; us_internal_rearm_writable(s); } @@ -630,7 +631,7 @@ int us_socket_raw_writev(struct us_socket_t *s, const struct us_iovec_t *iov, in for (int i = 0; i < count; i++) total += iov[i].iov_len; ssize_t written = bsd_writev(us_poll_fd(&s->p), iov, count); - if (written != (ssize_t)total && us_internal_send_should_rearm(written)) { + if (written != (ssize_t)total && us_internal_send_should_rearm(s, written)) { s->flags.last_write_failed = 1; us_internal_rearm_writable(s); } @@ -649,7 +650,7 @@ int us_socket_raw_write(struct us_socket_t *s, const char *data, int length) { } int written = bsd_send(us_poll_fd(&s->p), data, length); - if (written != length && us_internal_send_should_rearm(written)) { + if (written != length && us_internal_send_should_rearm(s, written)) { s->flags.last_write_failed = 1; us_internal_rearm_writable(s); } @@ -686,7 +687,7 @@ int us_socket_ipc_write_fd(struct us_socket_t *s, const char *data, int length, int sent = bsd_sendmsg(us_poll_fd(&s->p), &msg, 0); - if (sent != length && us_internal_send_should_rearm(sent)) { + if (sent != length && us_internal_send_should_rearm(s, sent)) { s->flags.last_write_failed = 1; us_internal_rearm_writable(s); } From 3319b55a85517f51045328812a0e1db5093d39c5 Mon Sep 17 00:00:00 2001 From: Ciro Spaciari MacBook Date: Thu, 6 Aug 2026 17:10:44 -0700 Subject: [PATCH 9/9] usockets: adopt the read_eof latch and write-side EV_EOF error translation from #37077 Replaces the dispatch-layer EV_EOF branch with #37077's design: kqueue's EV_EOF on EVFILT_WRITE (SS_CANTSENDMORE without our own shutdown) is translated to the poll error at the event layer, matching epoll's EPOLLERR, and a read_eof latch guarantees on_end fires at most once - readable interest is never re-added by rearm or resume after the peer's FIN was delivered. Covers two cases the previous mechanism missed: resume() after 'end' re-firing on_end, and a paused socket never observing a peer reset behind pending writes. --- packages/bun-usockets/src/context.c | 2 + .../bun-usockets/src/eventing/epoll_kqueue.c | 19 ++++- packages/bun-usockets/src/internal/internal.h | 2 + packages/bun-usockets/src/loop.c | 22 ++--- packages/bun-usockets/src/socket.c | 83 ++++++++----------- 5 files changed, 61 insertions(+), 67 deletions(-) diff --git a/packages/bun-usockets/src/context.c b/packages/bun-usockets/src/context.c index 9489f5468fb7..8503c6196e84 100644 --- a/packages/bun-usockets/src/context.c +++ b/packages/bun-usockets/src/context.c @@ -364,6 +364,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->read_eof = 0; s->next = 0; s->prev = 0; s->connect_state = NULL; @@ -508,6 +509,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->read_eof = 0; s->connect_state = NULL; s->connect_next = NULL; } diff --git a/packages/bun-usockets/src/eventing/epoll_kqueue.c b/packages/bun-usockets/src/eventing/epoll_kqueue.c index 4dce3538d154..7504da40d774 100644 --- a/packages/bun-usockets/src/eventing/epoll_kqueue.c +++ b/packages/bun-usockets/src/eventing/epoll_kqueue.c @@ -280,8 +280,9 @@ 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 skip : 1; - uint8_t _pad : 3; + uint8_t _pad : 2; }; _Static_assert(sizeof(struct kevent_flags) == 1, "kevent_flags must be 1 byte"); @@ -305,7 +306,9 @@ 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), + /* EV_EOF on EVFILT_READ is the peer's FIN; on EVFILT_WRITE it is SS_CANTSENDMORE (peer gone, or our own shutdown) - not a read EOF (libuv kqueue.c ignores it there too). */ + .eof = (flags & EV_EOF) && filter == EVFILT_READ, + .send_eof = (flags & EV_EOF) && filter == EVFILT_WRITE, }; /* Look backward for a prior entry with the same poll to coalesce into. @@ -317,6 +320,7 @@ 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[i] = (struct kevent_flags){ .skip = 1 }; merged = 1; break; @@ -344,9 +348,16 @@ 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); + int error = bits.error; + /* Write side dead without our own shutdown(): peer reset (or connect refused). Surface it as the poll error epoll would report as EPOLLERR. */ + if (bits.send_eof) { + int type = us_internal_poll_type(poll); + if (type == POLL_TYPE_SOCKET || type == POLL_TYPE_SEMI_SOCKET) error = 1; + } + events &= us_poll_events(poll); - if (events || bits.error || bits.eof) { - us_internal_dispatch_ready_poll(poll, bits.error, bits.eof, events); + if (events || error || bits.eof) { + us_internal_dispatch_ready_poll(poll, error, bits.eof, events); } } #endif diff --git a/packages/bun-usockets/src/internal/internal.h b/packages/bun-usockets/src/internal/internal.h index 38106f69c95a..4661e0bc8f37 100644 --- a/packages/bun-usockets/src/internal/internal.h +++ b/packages/bun-usockets/src/internal/internal.h @@ -307,6 +307,8 @@ struct us_socket_t { * the driver's epilogue via ssl_pending_detach. */ unsigned char ssl_in_use : 1; unsigned char ssl_pending_detach : 1; + /* Peer FIN was dispatched as on_end on a half-open socket; readable interest is never re-added and on_end never re-fires. */ + unsigned char read_eof : 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 de5bcaf1b0f2..392ed173dcab 100644 --- a/packages/bun-usockets/src/loop.c +++ b/packages/bun-usockets/src/loop.c @@ -540,6 +540,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->read_eof = 0; /* We always use nodelay */ bsd_socket_nodelay(client_fd, 1); @@ -586,21 +587,6 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in } /* The group can change after calling a callback but the loop is always the same */ struct us_loop_t* loop = s->group->loop; - #ifdef LIBUS_USE_KQUEUE - /* EV_EOF on EVFILT_WRITE we didn't shutdown() = SS_CANTSENDMORE (xnu bsd/kern/uipc_socket.c filt_sowrite); - * with the read side already ended nothing else closes us and the one-shot filter re-fires forever, so take - * the POLLERR close below like epoll's EPOLLERR|EPOLLHUP (libuv: deps/uv/src/unix/stream.c uv__write error path). */ - if ((events & LIBUS_SOCKET_WRITABLE) && eof && !error - && !(s->p.state.poll_type & POLL_TYPE_POLLING_IN) - && !s->flags.is_paused - && !us_socket_is_shut_down(s)) { - /* Clear POLLING_OUT for the consumed one-shot filter (POLLING_IN is already clear per the guard) */ - s->p.state.poll_type = us_internal_poll_type(&s->p); - error = 1; - eof = 0; /* on_end already ran when POLLING_IN was dropped (or the read side never - * started, e.g. low-prio parked mid-handshake) - error-close fits both */ - } - #endif if (events & LIBUS_SOCKET_WRITABLE && !error) { s->flags.last_write_failed = 0; #ifdef LIBUS_USE_KQUEUE @@ -866,7 +852,11 @@ 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->flags.allow_half_open && s->read_eof) { + /* on_end already delivered (libuv UV_HANDLE_READ_EOF): just drop the readable interest that re-surfaced it. */ + us_poll_change(&s->p, loop, us_poll_events(&s->p) & LIBUS_SOCKET_WRITABLE); + } else if(s->flags.allow_half_open) { + s->read_eof = 1; /* 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 diff --git a/packages/bun-usockets/src/socket.c b/packages/bun-usockets/src/socket.c index a6668a8333ee..6077d2e4a4d1 100644 --- a/packages/bun-usockets/src/socket.c +++ b/packages/bun-usockets/src/socket.c @@ -409,48 +409,13 @@ struct us_socket_t *us_socket_pair(struct us_socket_group_t *group, unsigned cha #endif } -/* Re-arm writable for a backpressured write, preserving the current read - * interest: us_poll_change sets absolute flags, and read interest is only ever - * off because something dropped it deliberately - us_socket_pause, the - * half-open EOF branch (re-adding it would re-deliver EOF and fire on_end - * again), or low-priority parking. Forcing READABLE back on undid all three. */ +/* 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. */ static void us_internal_rearm_writable(struct us_socket_t *s) { us_poll_change(&s->p, s->group->loop, - LIBUS_SOCKET_WRITABLE | (us_poll_events(&s->p) & LIBUS_SOCKET_READABLE)); -} - -#ifndef _WIN32 -/* send() errnos that mean the peer or the path to it is gone: no retry can ever - * succeed, so report them to the caller now; only the EAGAIN/ENOBUFS class waits for - * another writable event (https://github.com/libuv/libuv/blob/v1.51.0/src/unix/stream.c#L820-L837). */ -static int us_internal_send_errno_is_peer_gone(int e) { - switch (e) { - case EPIPE: - case ECONNRESET: - case ECONNABORTED: - case ENOTCONN: - case ETIMEDOUT: - case ENETDOWN: - case ENETUNREACH: - case EHOSTUNREACH: - return 1; - default: - return 0; - } -} -#endif - -/* On a peer-gone errno a PAUSED socket must not re-arm: kqueue's one-shot EVFILT_WRITE would - * re-fire with EV_EOF forever, and the loop.c close branch deliberately excludes paused sockets - * (deferred-EOF keeps their unread data). Everything else still re-arms so a live socket always - * has a filter registered - the re-fired EV_EOF then takes the loop.c error-close next tick. */ -static int us_internal_send_should_rearm(struct us_socket_t *s, ssize_t written) { -#ifndef _WIN32 - if (written < 0 && s->flags.is_paused && us_internal_send_errno_is_peer_gone(errno)) { - return 0; - } -#endif - return 1; + LIBUS_SOCKET_WRITABLE | ((s->flags.is_paused || s->read_eof) ? 0 : LIBUS_SOCKET_READABLE)); } int us_socket_write2(struct us_socket_t *s, const char *header, int header_length, const char *payload, int payload_length) { @@ -459,7 +424,7 @@ int us_socket_write2(struct us_socket_t *s, const char *header, int header_lengt } int written = bsd_write2(us_poll_fd(&s->p), header, header_length, payload, payload_length); - if (written != header_length + payload_length && us_internal_send_should_rearm(s, written)) { + if (written != header_length + payload_length) { us_internal_rearm_writable(s); } @@ -492,6 +457,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->read_eof = 0; s->connect_state = NULL; /* We always use nodelay */ @@ -532,7 +498,7 @@ int us_socket_write(struct us_socket_t *s, const char *data, int length) { } int written = bsd_send(us_poll_fd(&s->p), data, length); - if (written != length && us_internal_send_should_rearm(s, written)) { + if (written != length) { s->flags.last_write_failed = 1; us_internal_rearm_writable(s); } @@ -541,6 +507,28 @@ int us_socket_write(struct us_socket_t *s, const char *data, int length) { } #ifndef _WIN32 +/* send() errnos that mean the peer or the path to it is gone: no retry can + * ever succeed, so they are reported to the caller immediately (libuv fails + * the write request for these in uv__try_write, unix/stream.c; only the + * EAGAIN/ENOBUFS class waits for another writable event). Mirrored by the + * blanket `result < -1` fatal handling in h2_frame_parser's + * is_transport_fatal_write_result - classification lives here only. */ +static int us_internal_send_errno_is_peer_gone(int e) { + switch (e) { + case EPIPE: + case ECONNRESET: + case ECONNABORTED: + case ENOTCONN: + case ETIMEDOUT: + case ENETDOWN: + case ENETUNREACH: + case EHOSTUNREACH: + return 1; + default: + return 0; + } +} + /* One retry runs per writable dispatch (one event-loop iteration), so 32 * consecutive failures is far beyond any observed transient race window * (the macOS EPROTOTYPE race resolves within a dispatch or two) while still @@ -631,7 +619,7 @@ int us_socket_raw_writev(struct us_socket_t *s, const struct us_iovec_t *iov, in for (int i = 0; i < count; i++) total += iov[i].iov_len; ssize_t written = bsd_writev(us_poll_fd(&s->p), iov, count); - if (written != (ssize_t)total && us_internal_send_should_rearm(s, written)) { + if (written != (ssize_t)total) { s->flags.last_write_failed = 1; us_internal_rearm_writable(s); } @@ -650,7 +638,7 @@ int us_socket_raw_write(struct us_socket_t *s, const char *data, int length) { } int written = bsd_send(us_poll_fd(&s->p), data, length); - if (written != length && us_internal_send_should_rearm(s, written)) { + if (written != length) { s->flags.last_write_failed = 1; us_internal_rearm_writable(s); } @@ -687,7 +675,7 @@ int us_socket_ipc_write_fd(struct us_socket_t *s, const char *data, int length, int sent = bsd_sendmsg(us_poll_fd(&s->p), &msg, 0); - if (sent != length && us_internal_send_should_rearm(s, sent)) { + if (sent != length) { s->flags.last_write_failed = 1; us_internal_rearm_writable(s); } @@ -872,11 +860,12 @@ void us_socket_resume(struct us_socket_t *s) { // closed cannot be resumed if (us_socket_is_closed(s)) return; + int readable = s->read_eof ? 0 : LIBUS_SOCKET_READABLE; if (us_socket_is_shut_down(s)) { // we already sent FIN so we resume only readable side we are read-only - us_poll_change(&s->p, s->group->loop, LIBUS_SOCKET_READABLE); + us_poll_change(&s->p, s->group->loop, 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); + us_poll_change(&s->p, s->group->loop, readable | LIBUS_SOCKET_WRITABLE); }