diff --git a/packages/bun-usockets/src/context.c b/packages/bun-usockets/src/context.c index 9489f5468fb7..c3156070364d 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->fin_deferred = 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->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 c3a00130d8f2..5c3933a54761 100644 --- a/packages/bun-usockets/src/eventing/libuv.c +++ b/packages/bun-usockets/src/eventing/libuv.c @@ -37,10 +37,26 @@ int us_internal_libuv_peer_reset_probe(LIBUS_SOCKET_DESCRIPTOR fd) { if (send(fd, "", 0, 0) != SOCKET_ERROR) { return 0; } - 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. */ - return err != WSAEWOULDBLOCK && err != WSAESHUTDOWN; + /* Allowlist of definitely-peer-gone codes, the WSA mirror of + * us_internal_send_errno_is_peer_gone (socket.c). A transient failure must + * not count - WSAENOBUFS is documented transient-on-healthy + * (bsd_send_is_transient_error) - because the sweep re-probes every 4s, so + * a missed detection self-heals, while a false positive reset-closes a + * healthy paused connection. WSAESHUTDOWN is our own shutdown(SD_SEND), + * not the peer; the sweep probes sockets after local shutdown. */ + switch (WSAGetLastError()) { + case WSAECONNRESET: + case WSAECONNABORTED: + case WSAENETRESET: + case WSAENOTCONN: + case WSAETIMEDOUT: + case WSAENETDOWN: + case WSAENETUNREACH: + case WSAEHOSTUNREACH: + return 1; + default: + return 0; + } } static struct us_socket_t *us_internal_poll_cb_adopted_socket(struct us_poll_t *wp) { @@ -103,14 +119,23 @@ static void poll_cb(uv_poll_t *p, int status, int events) { * until resume; pending data keeps the pause honored untouched. */ char probe; ssize_t peeked = bsd_recv(us_poll_fd(wp), &probe, 1, MSG_PEEK); + struct us_socket_t *sock = us_internal_poll_cb_adopted_socket(wp); if (peeked == 0) { + /* Graceful FIN with nothing buffered: the shared dispatch defers the + * eof until resume (paused-EOF contract), which leaves this socket in + * the same consumed-DISCONNECT state as the data-deferred branch + * below - a LATER reset has no event left to ride. Hand it to the + * sweep as well. */ + if (!sock->fin_deferred) { + sock->fin_deferred = 1; + sock->group->loop->data.fin_deferred_count++; + } 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 diff --git a/packages/bun-usockets/src/loop.c b/packages/bun-usockets/src/loop.c index 18faa4f94478..8c10db258909 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->fin_deferred = 0; /* We always use nodelay */ bsd_socket_nodelay(client_fd, 1); diff --git a/packages/bun-usockets/src/socket.c b/packages/bun-usockets/src/socket.c index 25f798b1a017..2783292dddc7 100644 --- a/packages/bun-usockets/src/socket.c +++ b/packages/bun-usockets/src/socket.c @@ -360,6 +360,15 @@ __attribute__((always_inline)) struct us_socket_t *us_socket_close(struct us_soc // - does not emit on_close event // - does not close struct us_socket_t *us_socket_detach(struct us_socket_t *s) { +#ifdef LIBUS_USE_LIBUV + /* The fd leaves usockets' management, so the sweep must forget it + * (mirrors us_internal_socket_close_raw; a stale flag would leak + * fin_deferred_count and keep the sweep walking forever). */ + if (s->fin_deferred) { + s->fin_deferred = 0; + s->group->loop->data.fin_deferred_count--; + } +#endif if (!us_socket_is_closed(s)) { struct us_loop_t *loop = s->group->loop; @@ -457,6 +466,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->fin_deferred = 0; s->connect_state = NULL; /* We always use nodelay */ diff --git a/test/js/bun/net/socket.test.ts b/test/js/bun/net/socket.test.ts index f9e7ae1f55c1..14cffae4cd3f 100644 --- a/test/js/bun/net/socket.test.ts +++ b/test/js/bun/net/socket.test.ts @@ -3667,3 +3667,179 @@ describe.concurrent("connect() failure promise settlement", () => { ).rejects.toBe(boom); }); }); + +// A paused socket polls without READABLE, so a peer FIN is only reported +// through AFD's one-shot DISCONNECT. The eof is deferred until resume (pause +// contract), which consumes the only event the poll had; when the connection +// later dies, nothing is subscribed that could report it, so the 4s sweep has +// to probe the socket. Writing one byte into the dead connection resets the +// victim's TCB the same way a remote peer's RST segment would (Windows emits +// no RST from FIN_WAIT_2 on loopback, so the peer's abort alone is invisible +// to an idle victim here). The sleeps in these tests are structural: the +// sweep runs on a fixed 4s cadence, and the deferred-FIN window asserts the +// absence of a close across one full sweep. +function pausedVictimPair() { + const state = { + victimClosed: Promise.withResolvers(), + closedHow: null as string | null, + closeError: undefined as unknown, + endFired: false, + dataReceived: "", + }; + const victimOpen = Promise.withResolvers(); + const server = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + socket: { + open(s) { + s.pause(); // receive backpressure before any bytes flow + victimOpen.resolve(s); + }, + data(_s, buf) { + state.dataReceived += buf.toString(); + }, + end() { + state.endFired = true; + }, + error(_s, e) { + state.closedHow ??= "error"; + state.closeError = e; + state.victimClosed.resolve("error"); + }, + close(_s, e) { + state.closedHow ??= "close"; + state.closeError ??= e; + state.victimClosed.resolve("close"); + }, + }, + }); + const peer = (async () => { + const peerOpened = Promise.withResolvers(); + await Bun.connect({ + hostname: "127.0.0.1", + port: server.port, + socket: { + open(s) { + peerOpened.resolve(s); + }, + data() {}, + end() {}, + error() {}, + close() {}, + }, + }); + return peerOpened.promise; + })(); + return { state, server, victimOpen: victimOpen.promise, peer }; +} + +// Exercises the peeked == 0 path in poll_cb: the FIN lands on an empty +// receive buffer. +it.concurrent.skipIf(!isWindows)( + "paused socket with a deferred empty-buffer FIN still closes when the connection later dies", + async () => { + const { state, server, victimOpen, peer: peerP } = pausedVictimPair(); + using _server = server; + const peer = await peerP; + const victim = await victimOpen; + + // Let the pause settle: the writable dispatch drops WRITABLE, leaving the + // victim's poll subscribed to DISCONNECT only. + await Bun.sleep(200); + + // Clean FIN with the victim's receive buffer empty. + peer.shutdown(); + + // A full sweep period passes: the deferred FIN alone must not close the + // paused victim (its peer is alive and half-closed; the sweep's probe has + // to keep it deferred). + await Bun.sleep(4600); + expect(state.closedHow).toBeNull(); + + // Peer dies; the victim streams on, and the byte's RST reply resets its + // TCB. Without the sweep escalation nothing is ever delivered and the + // socket strands, so a bounded race is the condition check. + peer.terminate(); + await Bun.sleep(100); + victim.write("x"); + + const result = await Promise.race([state.victimClosed.promise, Bun.sleep(12_000).then(() => "stranded")]); + expect(result).not.toBe("stranded"); + // The deferred eof must not have been delivered as end; the socket died. + expect(state.endFired).toBe(false); + }, + 40_000, +); + +// Exercises the pre-existing peeked > 0 latch in poll_cb (FIN deferred behind +// buffered data), which the fin_deferred init fix first makes reachable: the +// garbage-broken count kept the sweep gate closed, so this path had never +// actually run. +it.concurrent.skipIf(!isWindows)( + "paused socket with a FIN deferred behind buffered data still closes when the connection later dies", + async () => { + const { state, server, victimOpen, peer: peerP } = pausedVictimPair(); + using _server = server; + const peer = await peerP; + const victim = await victimOpen; + + await Bun.sleep(200); + + // Data, then a clean FIN: the victim is paused, so the bytes sit in its + // kernel receive buffer and MSG_PEEK sees them at the DISCONNECT. + peer.write("data-behind-fin"); + peer.flush(); + await Bun.sleep(50); + peer.shutdown(); + + await Bun.sleep(4600); + expect(state.closedHow).toBeNull(); + + peer.terminate(); + await Bun.sleep(100); + victim.write("x"); + + const result = await Promise.race([state.victimClosed.promise, Bun.sleep(12_000).then(() => "stranded")]); + expect(result).not.toBe("stranded"); + // Node parity for a paused socket whose peer died: the reset wins; the + // buffered bytes and the end are never delivered. + expect(state.dataReceived).toBe(""); + expect(state.endFired).toBe(false); + }, + 40_000, +); + +// The resume side of the contract: a deferred FIN (and the data in front of +// it) is delivered once the socket resumes, and the close is clean. Resume +// also hands the socket back from the sweep (the fin_deferred count returns +// to zero), so a wrong count here would silently re-break the sweep gate. +it.concurrent.skipIf(!isWindows)( + "resuming a paused socket delivers the data and FIN that were deferred while paused", + async () => { + const { state, server, victimOpen, peer: peerP } = pausedVictimPair(); + using _server = server; + const peer = await peerP; + const victim = await victimOpen; + + await Bun.sleep(200); + + peer.write("deferred-data"); + peer.flush(); + await Bun.sleep(50); + peer.shutdown(); + + // Let the FIN's DISCONNECT report arrive and defer while paused. + await Bun.sleep(300); + expect(state.closedHow).toBeNull(); + + victim.resume(); + + const result = await Promise.race([state.victimClosed.promise, Bun.sleep(12_000).then(() => "stranded")]); + expect(result).not.toBe("stranded"); + expect(state.dataReceived).toBe("deferred-data"); + expect(state.endFired).toBe(true); + // Clean shutdown, not a reset. + expect(state.closeError).toBeFalsy(); + }, + 30_000, +);