Skip to content

usockets: drain the receive queue before closing on a peer reset #39860

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Jarred-Sumner merged 2 commits into from
Aug 21, 2026
Merged

usockets: drain the receive queue before closing on a peer reset #39860

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6d1ac96
usockets: drain the receive queue before closing on a peer reset
Jarred-Sumner Aug 21, 2026
edfdcfb
Sequence the paused-reset regression test without sleeps
Jarred-Sumner Aug 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 0 additions & 2 deletions packages/bun-usockets/src/context.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
}
Expand Down
88 changes: 24 additions & 64 deletions packages/bun-usockets/src/eventing/libuv.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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) {
Expand Down Expand Up @@ -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
Expand Down
6 changes: 1 addition & 5 deletions packages/bun-usockets/src/internal/internal.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
7 changes: 0 additions & 7 deletions packages/bun-usockets/src/internal/loop_data.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
59 changes: 22 additions & 37 deletions packages/bun-usockets/src/loop.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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 */
Expand All @@ -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) {
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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) {
Expand Down
15 changes: 0 additions & 15 deletions packages/bun-usockets/src/socket.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 */
Expand Down Expand Up @@ -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
Expand Down
Loading