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..075d9736b64e --- /dev/null +++ b/test/regression/issue/39846.test.ts @@ -0,0 +1,44 @@ +import { expect, test } from "bun:test"; +import { isWindows } from "harness"; +import net from "node:net"; + +// https://github.com/oven-sh/bun/issues/39846 +// 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; + 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; + + const res = await fetch(`http://127.0.0.1:${port}/`, { + method: "POST", + body: new ReadableStream({ start: () => {} }), + duplex: "half", + }); + + // "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!.resetAndDestroy(); + + expect(await res.text()).toBe("firsttail"); + } finally { + conn?.destroy(); + server.close(); + } + } +});