From 6d1ac9685ae58b33cf4d818e6584cdd9621101fb Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Fri, 21 Aug 2026 06:01:12 +0000 Subject: [PATCH 1/2] usockets: drain the receive queue before closing on a peer reset A poll error event (EPOLLERR / EV_EOF+fflags / AFD abort) closed the socket without reading when it carried no READABLE bit for us - which is always the case for a paused socket. The kernel keeps the receive queue on a reset, so the tail of the peer's stream queued ahead of it was discarded with the fd: a streamed fetch() response under receive backpressure lost its end with ECONNRESET although every byte had arrived (#39846), and node:net / Bun.Socket lost it the same way. An error is the end of the connection, so a pause no longer protects anything: run the read loop for an error event even without READABLE interest and through a pause, so recv() delivers the queued data and then the error, and the same dispatch closes with it. No per-socket opt-in and no deferred-error state. Windows: the paused-socket MSG_PEEK probe, the fin_deferred bit, its loop counter and the sweep that escalated it are removed; a paused socket takes the existing not-reading branch (reset -> error dispatch, FIN -> ABORT-only re-arm until resume), and a shut-down paused socket no longer has READABLE forced on. --- packages/bun-usockets/src/context.c | 2 - packages/bun-usockets/src/eventing/libuv.c | 88 +++++-------------- packages/bun-usockets/src/internal/internal.h | 6 +- .../bun-usockets/src/internal/loop_data.h | 7 -- packages/bun-usockets/src/loop.c | 59 +++++-------- packages/bun-usockets/src/socket.c | 15 ---- test/js/bun/net/socket.test.ts | 70 ++++++++++++--- test/js/node/net/node-net.test.ts | 73 +++++++++++++++ test/js/node/tls/node-tls-server.test.ts | 4 + test/regression/issue/39846.test.ts | 71 +++++++++++++++ 10 files changed, 252 insertions(+), 143 deletions(-) create mode 100644 test/regression/issue/39846.test.ts diff --git a/packages/bun-usockets/src/context.c b/packages/bun-usockets/src/context.c index 6005f46f7e26..0761aef4b317 100644 --- a/packages/bun-usockets/src/context.c +++ b/packages/bun-usockets/src/context.c @@ -365,7 +365,6 @@ static void us_internal_init_listen_socket(struct us_listen_socket_t *ls, s->flags.allow_half_open = (options & LIBUS_SOCKET_ALLOW_HALF_OPEN); s->unclassified_send_failures = 0; s->read_eof = 0; - s->fin_deferred = 0; s->next = 0; s->prev = 0; s->connect_state = NULL; @@ -545,7 +544,6 @@ static inline void us_internal_init_connect_socket(struct us_socket_t *s, s->flags.last_write_failed = 0; s->unclassified_send_failures = 0; s->read_eof = 0; - s->fin_deferred = 0; s->connect_state = NULL; s->connect_next = NULL; } diff --git a/packages/bun-usockets/src/eventing/libuv.c b/packages/bun-usockets/src/eventing/libuv.c index ea5e72904745..3ea446c7bdc0 100644 --- a/packages/bun-usockets/src/eventing/libuv.c +++ b/packages/bun-usockets/src/eventing/libuv.c @@ -22,11 +22,6 @@ #ifdef LIBUS_USE_LIBUV -/* The shared dispatch follows socket adoption (a tunneled/upgraded socket - * moves; the old allocation stays readable with flags.adopted set and prev - * pointing at the live one) and skips closed sockets. The paused-probe below - * must honor the same contract - dereferencing the raw poll cast crashed the - * CONNECT-tunnel tests on the aarch64 agent. */ /* Windows does not reliably latch a received RST in SO_ERROR (POSIX does); * the reset surfaces on the next I/O. A zero-byte send observes it without * touching the stream: 0 on a healthy socket, SOCKET_ERROR with a fatal @@ -39,19 +34,19 @@ int us_internal_libuv_peer_reset_probe(LIBUS_SOCKET_DESCRIPTOR fd) { } int err = WSAGetLastError(); /* WSAESHUTDOWN means our own shutdown(SD_SEND) ran; that is not a peer - * reset. The fin_deferred sweep probes sockets after local shutdown. */ + * reset (us_socket_stalled_write_means_peer_gone can ask after one). */ return err != WSAEWOULDBLOCK && err != WSAESHUTDOWN; } +/* The shared dispatch follows socket adoption (a tunneled/upgraded socket + * moves; the old allocation stays readable with flags.adopted set and prev + * pointing at the live one) and skips closed sockets. poll_cb's probes must + * honor the same contract - dereferencing the raw poll cast crashed the + * CONNECT-tunnel tests on the aarch64 agent. */ static struct us_socket_t *us_internal_poll_cb_adopted_socket(struct us_poll_t *wp) { return us_internal_socket_follow_adopted((struct us_socket_t *)wp); } -static int us_internal_poll_cb_socket_is_probeable(struct us_poll_t *wp) { - struct us_socket_t *s = us_internal_poll_cb_adopted_socket(wp); - return !s->flags.is_closed && s->flags.is_paused; -} - /* uv_poll_t->data always (except for most times after calling us_poll_stop) * points to the us_poll_t */ static void poll_cb(uv_poll_t *p, int status, int events) { @@ -87,61 +82,26 @@ static void poll_cb(uv_poll_t *p, int status, int events) { * never cut at an EAGAIN. */ if (kind == POLL_TYPE_SOCKET_SHUT_DOWN) { eof = 1; - events |= UV_READABLE; - } else if (kind == POLL_TYPE_SOCKET && us_internal_poll_cb_socket_is_probeable(wp)) { - /* A paused socket polls without READABLE, so the read loop cannot - * discover terminal states for it - and the pause contract forbids - * consuming deferred bytes. MSG_PEEK discriminates without consuming: - * an error is an abortive reset (our libuv patch reports AFD_POLL_ABORT - * as DISCONNECT so it reaches write-only polls at all) and must close - * now like epoll's unmaskable EPOLLERR; 0 is a graceful FIN with no - * data, deferred by the shared dispatch's existing paused-EOF contract - * until resume; pending data keeps the pause honored untouched. */ - char probe; - ssize_t peeked = bsd_recv(us_poll_fd(wp), &probe, 1, MSG_PEEK); - if (peeked == 0) { - eof = 1; - events |= UV_READABLE; - } else if (peeked < 0 && !bsd_would_block()) { - error = 1; - events |= UV_READABLE; - } else if (peeked > 0) { - struct us_socket_t *sock = us_internal_poll_cb_adopted_socket(wp); - if (us_socket_get_error(sock) != 0 || us_internal_libuv_peer_reset_probe(us_poll_fd(wp))) { - /* Data is buffered ahead of whatever ended the connection. If the - * peer ABORTED, the kernel already discarded the stream's tail and - * a paused socket that never resumes would otherwise never learn - - * node's paused sockets error immediately on a reset, buffered - * data included. SO_ERROR separates that from a graceful FIN - * behind data, which stays deferred until resume. */ - error = 1; - events |= UV_READABLE; - } else if (!sock->fin_deferred) { - /* Graceful FIN deferred behind data. This one-shot DISCONNECT - * report is now consumed, so a LATER reset (an error-path peer - * ends, flushes, then destroys - FIN, then RST) has no event left - * to ride. Mark the socket; the sweep timer escalates via - * SO_ERROR. */ - sock->fin_deferred = 1; - sock->group->loop->data.fin_deferred_count++; - } - } + /* A paused socket keeps the hint only; the dispatcher leaves it for + * resume(), whose poll change re-arms DISCONNECT and lands here again. */ + events |= us_poll_events(wp) & LIBUS_SOCKET_READABLE; } else if (kind == POLL_TYPE_SOCKET && !(us_poll_events(wp) & LIBUS_SOCKET_READABLE)) { - /* A half-open data socket whose end was already delivered: the EOF path - * moved its poll to WRITABLE-only (loop.c), and us_poll_change re-adds - * UV_DISCONNECT unconditionally, so AFD keeps reporting the FIN's - * level-triggered DISCONNECT. Re-adding READABLE here made recv() - * rediscover the same EOF and busy-loop on_end; keeping DISCONNECT - * armed would complete instantly forever. But the peer's later RST - * must still close the socket (epoll parity: EPOLLERR is unmaskable), - * so ask the kernel which of the two this wakeup is: a dead peer - * surfaces via SO_ERROR or the zero-byte send probe and closes through - * the shared error path; a FIN re-report quiesces with only the - * ABORT-only subscription (UV_PRIORITIZED) kept armed so the RST still - * has an event to ride. Non-SOCKET kinds keep the unconditional - * READABLE below: SEMI_SOCKET checks error/eof (set from status) and - * listen polls READABLE only. */ + /* A data socket that is not reading: paused, or half-open with its end + * already delivered (the EOF path moved its poll to WRITABLE-only, and + * us_poll_change re-adds UV_DISCONNECT unconditionally, so AFD keeps + * reporting the FIN's level-triggered DISCONNECT). Re-adding READABLE + * here would pull bytes a paused caller asked to defer, or rediscover + * the same EOF and busy-loop on_end; keeping DISCONNECT armed would + * complete instantly forever. A dead peer surfaces via SO_ERROR or the + * zero-byte send probe and goes through the shared error path (which + * reads off whatever is still queued and closes); a FIN, fresh on a + * paused socket or re-reported on a half-open one, quiesces with only + * the ABORT-only subscription (UV_PRIORITIZED) kept armed so a later + * RST still has an event to ride, and a paused socket meets the FIN + * again through recv() once resume() re-arms READABLE. Non-SOCKET kinds + * keep the unconditional READABLE below: SEMI_SOCKET checks error/eof + * (set from status) and listen polls READABLE only. */ struct us_socket_t *sock = us_internal_poll_cb_adopted_socket(wp); /* A reported UV_PRIORITIZED is AFD's own ABORT signal and needs no * probe; the probe covers a reset that arrives while PRIORITIZED was diff --git a/packages/bun-usockets/src/internal/internal.h b/packages/bun-usockets/src/internal/internal.h index d8289364c3ed..f8fd5d534f45 100644 --- a/packages/bun-usockets/src/internal/internal.h +++ b/packages/bun-usockets/src/internal/internal.h @@ -326,11 +326,7 @@ struct us_socket_t { * would-block/transient nor a known peer-gone error (see * us_socket_write_check_error). Reset by any send that makes progress. * Lives in the pad-to-pointer gap before `group`, so it costs nothing. */ - /* 7 bits fit the 32-cap retry counter; the spare bit marks a paused - * socket whose peer FIN was deferred behind buffered data (libuv path - - * the sweep escalates via SO_ERROR when the peer later resets). */ - unsigned char unclassified_send_failures : 7; - unsigned char fin_deferred : 1; + unsigned char unclassified_send_failures; struct us_socket_group_t *group; /* NULL for plain TCP. Direct BoringSSL `SSL*`; set by us_internal_ssl_attach diff --git a/packages/bun-usockets/src/internal/loop_data.h b/packages/bun-usockets/src/internal/loop_data.h index 3937ea5d70d5..959a9110204c 100644 --- a/packages/bun-usockets/src/internal/loop_data.h +++ b/packages/bun-usockets/src/internal/loop_data.h @@ -45,13 +45,6 @@ struct us_internal_loop_data_t { long long sweep_next_tick_ns; #endif int sweep_timer_count; -#ifdef LIBUS_USE_LIBUV - /* Sockets whose peer FIN was deferred behind buffered data while paused - * (poll_cb's MSG_PEEK probe): the sweep escalates them via SO_ERROR when - * the peer later resets, since the one-shot DISCONNECT report was already - * consumed by the FIN. Zero cost while no socket is in that state. */ - int fin_deferred_count; -#endif struct us_internal_async *wakeup_async; struct us_socket_group_t *head; /* QUIC engines on this loop. us_quic_loop_process walks the list from diff --git a/packages/bun-usockets/src/loop.c b/packages/bun-usockets/src/loop.c index ce1335294657..e5ef50cc9927 100644 --- a/packages/bun-usockets/src/loop.c +++ b/packages/bun-usockets/src/loop.c @@ -389,30 +389,6 @@ void us_internal_free_closed_sockets(struct us_loop_t *loop) { #ifdef LIBUS_USE_LIBUV void sweep_timer_cb(struct us_internal_callback_t *cb) { us_internal_timer_sweep(cb->loop); - /* Escalate paused sockets whose peer FIN was deferred behind buffered - * data and whose peer has since reset (poll_cb consumed the only - * DISCONNECT report on the FIN; AFD has no event left to deliver the - * abort to a read-less poll). Zero cost unless such sockets exist; - * closing unlinks the socket, so restart the walk after each close. */ - while (cb->loop->data.fin_deferred_count > 0) { - struct us_socket_t *victim = 0; - for (struct us_socket_group_t *g = cb->loop->data.head; g && !victim; g = g->next) { - for (struct us_socket_t *s = g->head_sockets; s; s = s->next) { - if (s->fin_deferred && !s->flags.is_closed - && (us_socket_get_error(s) != 0 - || us_internal_libuv_peer_reset_probe(us_poll_fd(&s->p)))) { - victim = s; - break; - } - } - } - if (!victim) { - break; - } - victim->fin_deferred = 0; - cb->loop->data.fin_deferred_count--; - us_internal_socket_close_raw(victim, LIBUS_SOCKET_CLOSE_CODE_CONNECTION_RESET, 0); - } } #endif @@ -535,7 +511,6 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in s->flags.last_write_failed = 0; s->unclassified_send_failures = 0; s->read_eof = 0; - s->fin_deferred = 0; /* We always use nodelay */ bsd_socket_nodelay(client_fd, 1); @@ -613,7 +588,19 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in } } - if (events & LIBUS_SOCKET_READABLE) { + /* An error event (EPOLLERR, EV_EOF with the socket error in fflags, an AFD + * abort) is the connection's death and this dispatch closes the socket with + * it below. The kernel keeps the receive queue on a reset, so the tail of the + * peer's stream may still be queued ahead of the error, and closing without + * reading would discard it (a streamed response cut short although every + * byte arrived, #39846). So the read loop runs for an error even when this + * event carried no READABLE bit or the socket is paused: a pause is flow + * control, and there is no later for a dead connection to flow into. recv() + * then returns the data and after it the error, which is what libuv reports + * to node as well. A socket parked in the low-priority queue is not linked + * where on_data expects it and takes the plain error close. */ + const int drain_for_error = error && !s->read_eof && s->flags.low_prio_state != 1; + if ((events & LIBUS_SOCKET_READABLE) || drain_for_error) { /* Contexts may prioritize down sockets that are currently readable, e.g. when SSL handshake has to be done. * SSL handshakes are CPU intensive, so we limit the number of handshakes per loop iteration, and move the rest * to the low-priority queue */ @@ -622,7 +609,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in * non-SSL arm dispatched a full vtable lookup just to read * NULL — no Zig handler defines isLowPrio and every C++ vtable * sets is_low_prio = nullptr — so it's been dropped. */ - if (s->ssl && us_internal_ssl_is_low_prio(s)) { + if (!error && s->ssl && us_internal_ssl_is_low_prio(s)) { if (flags->low_prio_state == 2) { flags->low_prio_state = 0; /* Socket has been delayed and now it's time to process incoming data for one iteration */ } else if (loop->data.low_prio_budget > 0) { @@ -756,7 +743,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in * buffer. This is what the comment above always described; it * was keyed on the error flag, which kqueue does not set for * a peer FIN. */ - if (s && !us_socket_is_closed(s) && !s->flags.is_paused && (eof || error)) { + if (s && !us_socket_is_closed(s) && (error || (!s->flags.is_paused && eof))) { continue; } /* Stop if on_data paused us (us_socket_pause from the data @@ -788,7 +775,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in * a large response on Windows only). recv() returning * 0 or WSAEWOULDBLOCK ends the loop, so this is * bounded by the kernel receive buffer. */ - if (s && !us_socket_is_closed(s) && !s->flags.is_paused && (eof || error)) { + if (s && !us_socket_is_closed(s) && (error || (!s->flags.is_paused && eof))) { continue; } /* Windows AFD_POLL_ABORT is not level-triggered the way @@ -870,14 +857,12 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in eof = 0; } if (eof && error && !read_fin) { - /* An error event whose read loop did not reach a FIN (the socket is - * paused, or on_data paused it mid-drain): the eof hint next to the - * error flag is the reset taking both directions down (EPOLLHUP beside - * EPOLLERR; EV_EOF with the error in fflags), not an end of stream, so - * it must not take the end path below. That path dispatched on_end for - * a reset, and a TLS socket's on_end closes with a clean code itself, - * so the error close never ran. A FIN this dispatch did read still - * delivers its end first; the error close follows either way. */ + /* The eof hint next to an error flag is the reset taking both directions + * down (EPOLLHUP beside EPOLLERR; EV_EOF with the error in fflags), not an + * end of stream, so it must not take the end path below (a TLS socket's + * on_end closes with a clean code itself, and the error would be lost). A + * FIN this dispatch did read still delivers its end first; the error + * close follows either way. */ eof = 0; } if(eof && s) { diff --git a/packages/bun-usockets/src/socket.c b/packages/bun-usockets/src/socket.c index 8d76a1116e6e..b84ec3a542ab 100644 --- a/packages/bun-usockets/src/socket.c +++ b/packages/bun-usockets/src/socket.c @@ -264,12 +264,6 @@ void us_connecting_socket_close(struct us_connecting_socket_t *c) { * handshake/secureConnection event. openssl.c re-enters here once that * graceful path is done. */ struct us_socket_t *us_internal_socket_close_raw(struct us_socket_t *s, int code, void *reason) { -#ifdef LIBUS_USE_LIBUV - if (s->fin_deferred) { - s->fin_deferred = 0; - s->group->loop->data.fin_deferred_count--; - } -#endif if (s->ssl && s->ssl_in_use) { /* A JS callback running from inside SSL_do_handshake/SSL_read (ALPN, SNI, * keylog, ...) destroyed this socket. Closing now frees the SSL and @@ -465,7 +459,6 @@ struct us_socket_t *us_socket_from_fd(struct us_socket_group_t *group, unsigned s->flags.last_write_failed = 0; s->unclassified_send_failures = 0; s->read_eof = 0; - s->fin_deferred = 0; s->connect_state = NULL; /* We always use nodelay */ @@ -859,14 +852,6 @@ void us_socket_pause(struct us_socket_t *s) { } void us_socket_resume(struct us_socket_t *s) { -#ifdef LIBUS_USE_LIBUV - /* Reads flow again: normal delivery discovers the deferred FIN (and any - * reset behind it), so the sweep no longer owns this socket. */ - if (s->fin_deferred) { - s->fin_deferred = 0; - s->group->loop->data.fin_deferred_count--; - } -#endif if (!s->flags.is_paused) return; s->flags.is_paused = 0; // closed cannot be resumed diff --git a/test/js/bun/net/socket.test.ts b/test/js/bun/net/socket.test.ts index 0b5f1358f746..a0aaf47ad318 100644 --- a/test/js/bun/net/socket.test.ts +++ b/test/js/bun/net/socket.test.ts @@ -4087,17 +4087,17 @@ describe("allowHalfOpen socket whose peer resets behind pending writes", () => { }); }); -// A paused socket polls for nothing. epoll reports the reset anyway (EPOLLERR cannot be -// masked); kqueue only reports it through the read knote that epoll_kqueue.c keeps -// registered while reads are off. Before that, the pause left a one-shot writable event -// behind and nothing else: a reset that landed after it was consumed was never reported, -// and the socket stayed paused for good. The greeting round trip below guarantees the -// one-shot has been consumed before the reset is sent. The node:net and node:tls shapes -// of this scenario are in test/js/node/tls/node-tls-server.test.ts. +// A paused socket polls for nothing, but a peer reset still reaches it (epoll reports EPOLLERR +// regardless of interest; kqueue keeps a read knote registered while reads are off, see +// epoll_kqueue.c). The reset is the end of the connection, so the pause no longer protects +// anything: the data queued ahead of the reset is delivered, then the socket closes with read +// ECONNRESET. Closing without reading discarded that data (a streamed body cut short although +// every byte arrived, #39846). Windows discards the receive queue on a reset itself. The +// node:net and node:tls shapes are in test/js/node/tls/node-tls-server.test.ts. describe.concurrent.each(["tcp", "tls"] as const)("%s socket paused when its peer resets the connection", transport => { - it("closes with read ECONNRESET while still paused and delivers none of the unread data", async () => { + it("delivers the data queued ahead of the reset, then closes with read ECONNRESET, while still paused", async () => { const closedWith = Promise.withResolvers(); - let dataCalls = 0; + let received = ""; const pauseAndGreet = (socket: Socket) => { socket.pause(); socket.write("greeting"); @@ -4115,8 +4115,8 @@ describe.concurrent.each(["tcp", "tls"] as const)("%s socket paused when its pee if (success) pauseAndGreet(socket); else closedWith.reject(authorizationError ?? new Error("server handshake failed")); }, - data() { - dataCalls++; + data(_socket, chunk) { + received += chunk.toString(); }, close(_socket, error) { closedWith.resolve(error); @@ -4138,23 +4138,67 @@ describe.concurrent.each(["tcp", "tls"] as const)("%s socket paused when its pee }); const peer = await greeted.promise; peer.write("queued behind the pause"); + peer.flush(); peer.terminate(); const error = (await closedWith.promise) as NodeJS.ErrnoException | undefined; expect({ + received: isWindows ? "" : received, reported: error instanceof Error, syscall: error?.syscall, - dataCalls, code: error?.code, }).toEqual({ + received: isWindows ? "" : "queued behind the pause", reported: true, syscall: "read", - dataCalls: 0, code: "ECONNRESET", }); }); }); +// A paused socket with a backpressured write of its own must also close on the reset: an owner +// that resumes only after 'drain' (node:http's flood guard) would otherwise wait forever. +it("a paused socket with a backpressured write still closes when its peer resets", async () => { + const closedWith = Promise.withResolvers(); + let backpressured!: () => void; + const isBackpressured = new Promise(resolve => (backpressured = resolve)); + const big = Buffer.alloc(4 * 1024 * 1024, "x"); + using server = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + socket: { + open(socket) { + socket.pause(); + // The peer never reads, so this fills both kernel buffers and is refused part-way. + while (socket.write(big) === big.length) {} + backpressured(); + }, + drain(socket) { + while (socket.write(big) === big.length) {} + }, + data() {}, + close(_socket, error) { + closedWith.resolve(error); + }, + }, + }); + const peer = await Bun.connect({ + hostname: "127.0.0.1", + port: server.port, + socket: { + open(socket) { + socket.pause(); + }, + data() {}, + close() {}, + }, + }); + await isBackpressured; + peer.terminate(); + const error = (await closedWith.promise) as NodeJS.ErrnoException | undefined; + expect(error?.code).toBe("ECONNRESET"); +}); + // A close that the event loop initiated passes the read error to close(). usockets // reports that error in the platform's own numbering (an errno on POSIX, a WSA code // such as WSAECONNRESET = 10054 on Windows) and on_close has to map it: unmapped, a diff --git a/test/js/node/net/node-net.test.ts b/test/js/node/net/node-net.test.ts index eadca761d75b..d44ffb1cf657 100644 --- a/test/js/node/net/node-net.test.ts +++ b/test/js/node/net/node-net.test.ts @@ -1191,6 +1191,79 @@ describe.concurrent("socket that already sent FIN and is paused with unread data }); }); +// A reset that reaches a read-stopped handle ends the connection, and the bytes the kernel +// still holds ahead of it are read off the socket before it is closed rather than discarded +// with the fd (#39846: a streamed fetch() body was cut short under receive backpressure this +// way). They land in the paused stream's buffer, so bytesRead accounts for every byte the peer +// sent. Windows discards the receive queue on a reset itself. +describe.concurrent("read-stopped socket whose peer resets behind unread data", () => { + it("reads the queued bytes off the socket before reporting ECONNRESET", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const net = require("net"); + const server = net.createServer({ allowHalfOpen: true, highWaterMark: 64 * 1024 }, s => { + // 64 KiB reaches the highWaterMark so the handle is read-stopped; the next + // 32 KiB and the reset stay in the kernel. + s.pause(); + const events = []; + s.on("end", () => events.push("end")); + s.on("error", e => events.push("error " + e.code)); + s.on("close", () => { + events.push("close"); + console.log(JSON.stringify({ events, bytesRead: s.bytesRead, buffered: s.readableLength })); + process.exit(0); + }); + (function waitReadStopped() { + if (s.readableLength >= 64 * 1024) console.log("read-stopped"); + else setImmediate(waitReadStopped); + })(); + }); + server.listen(0, "127.0.0.1", () => console.log("port " + server.address().port)); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "inherit", + }); + const reader = proc.stdout.getReader(); + let buffered = ""; + async function line() { + while (!buffered.includes("\n")) { + const { value, done } = await reader.read(); + if (done) break; + buffered += new TextDecoder().decode(value); + } + const i = buffered.indexOf("\n"); + const out = i === -1 ? buffered : buffered.slice(0, i); + buffered = i === -1 ? "" : buffered.slice(i + 1); + return out; + } + const port = Number((await line()).split(" ")[1]); + const peer = await Bun.connect({ + hostname: "127.0.0.1", + port, + socket: { data() {}, drain() {}, close() {}, error() {} }, + }); + expect(peer.write(Buffer.alloc(64 * 1024, "a"))).toBe(64 * 1024); + peer.flush(); + expect(await line()).toBe("read-stopped"); + // In the kernel ahead of the reset: same connection, so TCP orders them. + expect(peer.write(Buffer.alloc(32 * 1024, "b"))).toBe(32 * 1024); + peer.flush(); + peer.terminate(); + const result = JSON.parse(await line()); + expect(result).toEqual({ + events: ["error ECONNRESET", "close"], + bytesRead: isWindows ? result.bytesRead : 96 * 1024, + buffered: isWindows ? result.buffered : 96 * 1024, + }); + expect(await proc.exited).toBe(0); + }); +}); + // A socket whose reads are stopped for backpressure must not hold the process // open: in node a handle that is not reading is inactive, so a program that // never consumes a reply (or a request) still exits. Each fixture leaves such a diff --git a/test/js/node/tls/node-tls-server.test.ts b/test/js/node/tls/node-tls-server.test.ts index 32f369386c1c..b2fbbdb5cedd 100644 --- a/test/js/node/tls/node-tls-server.test.ts +++ b/test/js/node/tls/node-tls-server.test.ts @@ -2547,6 +2547,10 @@ describe.each(["tls", "net"])("%s server socket whose peer resets the connection t.peer.terminate(); await t.settled; expect(t.events).toEqual(["error ECONNRESET", "close hadError=true"]); + // The data queued ahead of the reset was read off the socket before it closed + // (kept in the paused stream's buffer), not discarded with the fd. Windows + // discards the receive queue on a reset. + if (!isWindows) expect(t.socket.bytesRead).toBe(64 * 1024); }); it("delivers the data queued ahead of the reset and then reports ECONNRESET, not 'end'", async () => { diff --git a/test/regression/issue/39846.test.ts b/test/regression/issue/39846.test.ts new file mode 100644 index 000000000000..0b1ff5b50109 --- /dev/null +++ b/test/regression/issue/39846.test.ts @@ -0,0 +1,71 @@ +import { expect, test } from "bun:test"; +import { isWindows } from "harness"; +import net from "node:net"; + +// https://github.com/oven-sh/bun/issues/39846 +// A fetch() response tail that arrived while receive backpressure had the +// socket paused was discarded when a TCP reset arrived behind it: the event +// loop closed the paused socket without draining the bytes the kernel still +// held. Windows cannot recover the tail (AFD discards the receive queue on a +// reset, and node loses it there too), so the test is skipped on Windows. +test.skipIf(isWindows)("fetch: response tail received before a reset is delivered, not discarded", async () => { + for (let i = 0; i < 2; i++) { + let conn: net.Socket | undefined; + const server = net.createServer(c => { + conn = c; + c.on("error", () => {}); + let buf = ""; + const onData = (d: Buffer) => { + buf += d.toString("latin1"); + if (buf.includes("\r\n\r\n")) { + c.off("data", onData); + c.write("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n5\r\nfirst\r\n"); + } + }; + c.on("data", onData); + }); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + try { + const port = (server.address() as net.AddressInfo).port; + + let push!: (s: string) => void; + const requestBody = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("hello")); + push = s => controller.enqueue(new TextEncoder().encode(s)); + }, + }); + + const res = await fetch(`http://127.0.0.1:${port}/`, { + method: "POST", + body: requestBody, + duplex: "half", + }); + + // The first body chunk pauses the transport (nothing consumes the body + // yet). The pause is the precondition under test and is unobservable + // from JS (a paused socket produces no events), so the sleeps below are + // the only way to sequence the race deterministically. + await Bun.sleep(100); + + // Finish the response and close the server socket cleanly. The tail and + // the FIN sit unread in the paused client's kernel buffer. + await new Promise(resolve => conn!.write("4\r\ntail\r\n0\r\n\r\n", () => resolve())); + conn!.destroy(); + await Bun.sleep(50); + + // A request-body chunk sent to the closed server socket makes its kernel + // answer with a reset. The reset reaches the paused client while the + // response tail is still queued behind it. + push("x"); + await Bun.sleep(100); + + // Consuming the body resumes the transport, which must drain the tail + // before it surfaces the reset. + expect(await res.text()).toBe("firsttail"); + } finally { + conn?.destroy(); + server.close(); + } + } +}); From edfdcfbe8097bcf3e9fc069b80c5553aafbc532f Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Fri, 21 Aug 2026 14:53:54 -0700 Subject: [PATCH 2/2] Sequence the paused-reset regression test without sleeps The first body chunk arrives with the headers, so the transport is already paused when fetch() resolves. The server resets with resetAndDestroy after the tail is flushed, so the tail and the reset are queued before the body is consumed. No-Verification-Needed: test-only change --- test/regression/issue/39846.test.ts | 37 ++++------------------------- 1 file changed, 5 insertions(+), 32 deletions(-) diff --git a/test/regression/issue/39846.test.ts b/test/regression/issue/39846.test.ts index 0b1ff5b50109..075d9736b64e 100644 --- a/test/regression/issue/39846.test.ts +++ b/test/regression/issue/39846.test.ts @@ -3,11 +3,7 @@ import { isWindows } from "harness"; import net from "node:net"; // https://github.com/oven-sh/bun/issues/39846 -// A fetch() response tail that arrived while receive backpressure had the -// socket paused was discarded when a TCP reset arrived behind it: the event -// loop closed the paused socket without draining the bytes the kernel still -// held. Windows cannot recover the tail (AFD discards the receive queue on a -// reset, and node loses it there too), so the test is skipped on Windows. +// Windows cannot recover the tail: AFD discards the receive queue on a reset, and node loses it there too. test.skipIf(isWindows)("fetch: response tail received before a reset is delivered, not discarded", async () => { for (let i = 0; i < 2; i++) { let conn: net.Socket | undefined; @@ -28,40 +24,17 @@ test.skipIf(isWindows)("fetch: response tail received before a reset is delivere try { const port = (server.address() as net.AddressInfo).port; - let push!: (s: string) => void; - const requestBody = new ReadableStream({ - start(controller) { - controller.enqueue(new TextEncoder().encode("hello")); - push = s => controller.enqueue(new TextEncoder().encode(s)); - }, - }); - const res = await fetch(`http://127.0.0.1:${port}/`, { method: "POST", - body: requestBody, + body: new ReadableStream({ start: () => {} }), duplex: "half", }); - // The first body chunk pauses the transport (nothing consumes the body - // yet). The pause is the precondition under test and is unobservable - // from JS (a paused socket produces no events), so the sleeps below are - // the only way to sequence the race deterministically. - await Bun.sleep(100); - - // Finish the response and close the server socket cleanly. The tail and - // the FIN sit unread in the paused client's kernel buffer. + // "first" arrived with the headers and nothing pulls the body, so the transport is paused. + // The tail and the reset now queue behind each other in the paused client's kernel buffer. await new Promise(resolve => conn!.write("4\r\ntail\r\n0\r\n\r\n", () => resolve())); - conn!.destroy(); - await Bun.sleep(50); - - // A request-body chunk sent to the closed server socket makes its kernel - // answer with a reset. The reset reaches the paused client while the - // response tail is still queued behind it. - push("x"); - await Bun.sleep(100); + conn!.resetAndDestroy(); - // Consuming the body resumes the transport, which must drain the tail - // before it surfaces the reset. expect(await res.text()).toBe("firsttail"); } finally { conn?.destroy();