Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
36 changes: 21 additions & 15 deletions packages/bun-usockets/src/context.c
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,6 @@
#include <arpa/inet.h>
#include <netinet/in.h>
#endif
#define CONCURRENT_CONNECTIONS 4

#if defined(BUN_DEBUG) || (defined(__has_feature) && __has_feature(address_sanitizer)) || defined(__SANITIZE_ADDRESS__)
#include <assert.h>
Expand Down Expand Up @@ -692,11 +691,17 @@ struct us_socket_t *us_socket_group_connect_unix(struct us_socket_group_t *group
return connect_socket;
}

int start_connections(struct us_connecting_socket_t *c, int count) {
/* One address at a time, the way RFC 8305 §5 dials: the next address starts
* LIBUS_CONNECT_ATTEMPT_DELAY_NS after the previous one, or at once when an
* attempt fails. The first to connect wins and the rest are reset. Opening
* every address at once instead would park the dial on the kernel's SYN
* retries (about two minutes) whenever the first few are black holes, and
* would cost the server one accepted connection per address. */
int us_internal_socket_start_next_attempt(struct us_connecting_socket_t *c) {
int opened = 0;
struct us_socket_group_t *group = c->group;
struct us_loop_t *loop = group->loop;
for (; c->addrinfo_head != NULL && opened < count; c->addrinfo_head = c->addrinfo_head->ai_next) {
for (; c->addrinfo_head != NULL && opened < 1; c->addrinfo_head = c->addrinfo_head->ai_next) {
struct sockaddr_storage addr;
init_addr_with_port(c->addrinfo_head, c->port, &addr);
/* The deferred-DNS path does not carry a local binding. */
Expand Down Expand Up @@ -765,13 +770,14 @@ void us_internal_socket_after_resolve(struct us_connecting_socket_t *c) {

c->addrinfo_head = &result->entries->info;

int opened = start_connections(c, CONCURRENT_CONNECTIONS);
if (opened == 0) {
if (!us_internal_socket_start_next_attempt(c)) {
/* Same as the exhausted path in us_internal_socket_after_open: a
* real connect failure must not be reported as a caller abort. */
c->error = ECONNREFUSED;
us_connecting_socket_close(c);
return;
}
us_internal_connect_attempt_schedule(c);
}

void us_internal_socket_after_open(struct us_socket_t *s, int error) {
Expand Down Expand Up @@ -809,16 +815,16 @@ void us_internal_socket_after_open(struct us_socket_t *s, int error) {
}
us_socket_close(s, LIBUS_SOCKET_CLOSE_CODE_CONNECTION_RESET, 0);

if (c->connecting_head == NULL || c->connecting_head->connect_next == NULL) {
int opened = start_connections(c, c->connecting_head == NULL ? CONCURRENT_CONNECTIONS : 1);
if (opened == 0 && c->connecting_head == NULL) {
/* Every resolved address failed to connect. Without this,
* us_connecting_socket_close defaults c->error to
* ECANCELED (caller abort) and never invalidates the
* DNS cache entry for the dead host. */
c->error = ECONNREFUSED;
us_connecting_socket_close(c);
}
/* A failure advances to the next address at once. */
if (!us_internal_socket_start_next_attempt(c) && c->connecting_head == NULL) {
/* Every resolved address failed to connect. Without this,
* us_connecting_socket_close defaults c->error to
* ECANCELED (caller abort) and never invalidates the
* DNS cache entry for the dead host. */
c->error = ECONNREFUSED;
us_connecting_socket_close(c);
} else {
us_internal_connect_attempt_schedule(c);
}
} else {
us_internal_socket_connect_failed(s, error);
Expand Down
9 changes: 8 additions & 1 deletion packages/bun-usockets/src/eventing/epoll_kqueue.c
Original file line number Diff line number Diff line change
Expand Up @@ -428,9 +428,14 @@ static void us_internal_drain_ready_polls(struct us_loop_t *loop) {
}
}

/* Bound `timeout` by the socket-timeout sweep deadline (NULL == forever). */
/* Bound `timeout` by the socket-timeout sweep deadline and the next scheduled
* connect attempt (NULL == forever). */
static const struct timespec *us_internal_clamp_to_sweep(struct us_loop_t *loop, const struct timespec *timeout, struct timespec *storage) {
long long ns = us_internal_sweep_timeout_ns(loop);
long long connect_ns = us_internal_connect_attempt_timeout_ns(loop);
if (connect_ns >= 0 && (ns < 0 || connect_ns < ns)) {
ns = connect_ns;
}
if (ns < 0) {
return timeout;
}
Expand Down Expand Up @@ -465,6 +470,7 @@ void us_loop_run(struct us_loop_t *loop) {
us_internal_dispatch_ready_polls(loop);
us_internal_drain_ready_polls(loop);
us_internal_sweep_if_due(loop);
us_internal_connect_attempts_if_due(loop);

/* Emit post callback */
us_internal_loop_post(loop);
Expand Down Expand Up @@ -558,6 +564,7 @@ void us_loop_run_bun_tick(struct us_loop_t *loop, const struct timespec* timeout
us_internal_dispatch_ready_polls(loop);
us_internal_drain_ready_polls(loop);
us_internal_sweep_if_due(loop);
us_internal_connect_attempts_if_due(loop);

/* Emit post callback */
us_internal_loop_post(loop);
Expand Down
19 changes: 18 additions & 1 deletion packages/bun-usockets/src/internal/internal.h
Original file line number Diff line number Diff line change
Expand Up @@ -182,13 +182,26 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
void us_internal_timer_sweep(us_loop_r loop);
void us_internal_enable_sweep_timer(struct us_loop_t *loop);
void us_internal_disable_sweep_timer(struct us_loop_t *loop);
#ifndef LIBUS_USE_LIBUV
/* CLOCK_MONOTONIC in ns. The clock every deadline on the loop is measured
* against, so anything comparing against one must read it and not another. */
uint64_t us_internal_monotonic_ns(void);
#ifndef LIBUS_USE_LIBUV
long long us_internal_sweep_timeout_ns(struct us_loop_t *loop);
void us_internal_sweep_if_due(struct us_loop_t *loop);
/* ns until the earliest scheduled connect attempt on this loop, or -1. */
long long us_internal_connect_attempt_timeout_ns(struct us_loop_t *loop);
#endif
/* Delay between two connect attempts to the addresses of one name while the
* earlier attempts are still pending (RFC 8305 §5 "Connection Attempt Delay"). */
#define LIBUS_CONNECT_ATTEMPT_DELAY_NS (250LL * 1000000LL)
/* Schedule the next address of `c` for now + LIBUS_CONNECT_ATTEMPT_DELAY_NS.
* A no-op when `c` has no address left. */
void us_internal_connect_attempt_schedule(struct us_connecting_socket_t *c);
/* Start the scheduled attempt of every connecting socket whose delay elapsed. */
void us_internal_connect_attempts_if_due(struct us_loop_t *loop);
/* Open a connect() to the next address of `c`, at most one. Returns 1 when a
* socket was opened, 0 when every remaining address failed to open. */
int us_internal_socket_start_next_attempt(struct us_connecting_socket_t *c);
void us_internal_free_closed_sockets(us_loop_r loop);
/* Defers closing `s` with `code` to us_internal_loop_post. */
void us_internal_loop_close_unresumable_socket(struct us_loop_t *loop, struct us_socket_t *s, int code);
Expand Down Expand Up @@ -452,6 +465,10 @@ struct us_connecting_socket_t {
uint16_t port;
int error;
struct addrinfo *addrinfo_head;
/* Absolute monotonic ns at which the next address in addrinfo_head is
* tried while the current attempts are still pending (RFC 8305 §5), or 0
* when nothing is scheduled. */
long long next_attempt_ns;
// this is used to track pending connecting sockets in the context
struct us_connecting_socket_t* next_pending;
struct us_connecting_socket_t* prev_pending;
Expand Down
7 changes: 7 additions & 0 deletions packages/bun-usockets/src/internal/loop_data.h
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,13 @@ struct us_internal_loop_data_t {
int low_prio_budget;
struct us_connecting_socket_t *dns_ready_head;
struct us_connecting_socket_t *closed_connecting_head;
/* Absolute monotonic ns of the earliest us_connecting_socket_t::next_attempt_ns
* on this loop, or -1. POSIX folds it into the poll timeout next to the
* sweep deadline; libuv arms connect_timer to it. */
long long connect_next_attempt_ns;
#ifdef LIBUS_USE_LIBUV
struct us_timer_t *connect_timer;
#endif
zig_mutex_t mutex;
void *parent_ptr;
char parent_tag;
Expand Down
93 changes: 93 additions & 0 deletions packages/bun-usockets/src/loop.c
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,10 @@ void sweep_timer_cb(struct us_internal_callback_t *cb);
// when the sweep timer is disabled, we don't need to do anything
void sweep_timer_noop(struct us_timer_t *timer) {}

uint64_t us_internal_monotonic_ns(void) {
return uv_hrtime();
}

void us_internal_enable_sweep_timer(struct us_loop_t *loop) {
loop->data.sweep_timer_count++;
if (loop->data.sweep_timer_count == 1) {
Expand Down Expand Up @@ -118,7 +122,92 @@ void us_internal_sweep_if_due(struct us_loop_t *loop) {
us_internal_timer_sweep(loop);
}

long long us_internal_connect_attempt_timeout_ns(struct us_loop_t *loop) {
if (loop->data.connect_next_attempt_ns < 0) {
return -1;
}
long long diff = loop->data.connect_next_attempt_ns - (long long) us_internal_monotonic_ns();
return diff > 0 ? diff : 0;
}

#endif

/* ── Connection attempt delay (RFC 8305 §5) ─────────────────────────────── */

#ifdef LIBUS_USE_LIBUV
static void connect_timer_cb(struct us_internal_callback_t *cb) {
us_internal_connect_attempts_if_due(cb->loop);
}

/* Arm (or re-arm) the uv timer for loop->data.connect_next_attempt_ns. */
static void us_internal_connect_timer_arm(struct us_loop_t *loop) {
long long deadline = loop->data.connect_next_attempt_ns;
if (deadline < 0) {
us_timer_set(loop->data.connect_timer, (void (*)(struct us_timer_t *)) connect_timer_cb, 0, 0);
return;
}
/* Round up: uv measures from its cached loop time, so a truncated delay
* fires before the deadline. An early fire re-arms from the callback. */
long long ms = (deadline - (long long) us_internal_monotonic_ns() + 999999LL) / 1000000LL;
us_timer_set(loop->data.connect_timer, (void (*)(struct us_timer_t *)) connect_timer_cb, ms > 0 ? (int) ms : 1, 0);
}
#endif

void us_internal_connect_attempt_schedule(struct us_connecting_socket_t *c) {
if (c->addrinfo_head == NULL) {
c->next_attempt_ns = 0;
return;
}
struct us_loop_t *loop = c->loop;
c->next_attempt_ns = (long long) us_internal_monotonic_ns() + LIBUS_CONNECT_ATTEMPT_DELAY_NS;
/* Every delay is the same length, so a new deadline is never earlier than
* one already armed. */
if (loop->data.connect_next_attempt_ns < 0) {
loop->data.connect_next_attempt_ns = c->next_attempt_ns;
#ifdef LIBUS_USE_LIBUV
us_internal_connect_timer_arm(loop);
#endif
}
}

void us_internal_connect_attempts_if_due(struct us_loop_t *loop) {
if (loop->data.connect_next_attempt_ns < 0) {
return;
}
long long now = (long long) us_internal_monotonic_ns();
if (now < loop->data.connect_next_attempt_ns) {
#ifdef LIBUS_USE_LIBUV
us_internal_connect_timer_arm(loop);
#endif
return;
}
Comment thread
claude[bot] marked this conversation as resolved.
long long next = -1;
for (struct us_socket_group_t *group = loop->data.head; group; group = group->next) {
for (struct us_connecting_socket_t *c = group->head_connecting_sockets; c; c = c->next_pending) {
if (c->next_attempt_ns <= 0 || c->closed) {
continue;
}
/* A connecting socket with a scheduled attempt always has at least
* one attempt pending: the failure path of the last pending
* attempt starts the next address itself and decides the outcome.
* So nothing here can close `c` or dispatch to the owner, and the
* lists stay intact while they are walked. */
if (c->next_attempt_ns <= now && c->connecting_head != NULL) {
c->next_attempt_ns = 0;
if (us_internal_socket_start_next_attempt(c)) {
us_internal_connect_attempt_schedule(c);
}
Comment thread
robobun marked this conversation as resolved.
Comment on lines +195 to +199

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 (optional) Operators of a multi-address host that is dropping SYNs now get one pending connect per address from every Bun dial, not the base branch's 4. The timer walk at loop.c:198-203 opens another address every 250 ms while earlier attempts stay pending and never counts them. A 24-record name holds 24 sockets per dial after 6 s; bun install's default 256 concurrent requests make 6144 retransmitting SYN streams against the struggling server. Fix: bound in-flight attempts per connecting socket on the timer path (close the timed-out attempt before dialing the next, as Node does, or cap in-flight at 2 like curl), keeping the immediate advance on failure at context.c:836. The description calls the missing cap intended; this amplification is its cost.

Why this was flagged

Trigger: a name with many addresses whose SYNs are dropped rather than refused, dialed through fetch(), Bun.connect(), WebSocket or bun install. Base branch: start_connections opened 4 attempts and refilled only when at most one was pending, so a dial never held more than 4 pending connects. New code: us_internal_socket_start_next_attempt (context.c:698) opens exactly one, us_internal_connect_attempt_schedule (loop.c:157) re-arms 250 ms later, and the walk at loop.c:198-203 starts the next address whenever c->connecting_head != NULL with no count of how many attempts are already pending. For registry.npmjs.org a dial reaches 24 pending sockets after 23 x 250 ms = 5.75 s, each retransmitting SYNs for ~2 min. bun install runs up to MAX_SIMULTANEOUS_REQUESTS = 256 (src/http/AsyncHTTP.rs:70) such dials, so an overloaded registry sees up to 6144 SYN streams instead of 1024. The PR text states 'There is no cap on pending attempts' as a design choice; nothing in the code bounds it, and Node's autoSelectFamily and curl both bound it.

Verification: Triggering condition: a name with many addresses whose SYNs are dropped. On the base, start_connections (context.c:755) opened 4, so a dial never held more than 4 pending connects. In the new code packages/bun-usockets/src/loop.c:198-203 has one gate, at least one attempt still pending; nothing counts pending attempts, so a dial accumulates one pending connect() per resolved address, vs. 4 on the base.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed that this is a cost, and it is now listed under Downsides in the PR body: with no bound, a name whose N addresses all drop SYNs holds N pending connects per dial after (N-1) x 250 ms, where the old code held 4. A bound has to keep the later addresses reachable, or the stall this PR fixes comes back (four black holes in front of a live fifth address). The two candidates are to close the oldest pending attempt when a new one starts past the bound, or to give each attempt a timeout the way Node does. I have not picked one yet. This thread stays open until I do.

}
if (c->next_attempt_ns > 0 && (next < 0 || c->next_attempt_ns < next)) {
next = c->next_attempt_ns;
}
}
}
loop->data.connect_next_attempt_ns = next;
#ifdef LIBUS_USE_LIBUV
us_internal_connect_timer_arm(loop);
#endif
}


/* -1 if the wakeup async cannot be created; nothing is left allocated in loop->data. */
Expand All @@ -127,9 +216,11 @@ int us_internal_loop_data_init(struct us_loop_t *loop, void (*wakeup_cb)(struct
// We allocate with calloc, so we only need to initialize the specific fields in use.
#ifdef LIBUS_USE_LIBUV
loop->data.sweep_timer = us_create_timer(loop, 1, 0);
loop->data.connect_timer = us_create_timer(loop, 1, 0);
#else
loop->data.sweep_next_tick_ns = -1;
#endif
loop->data.connect_next_attempt_ns = -1;
loop->data.sweep_timer_count = 0;
loop->data.recv_buf = us_malloc(LIBUS_RECV_BUFFER_LENGTH + LIBUS_RECV_BUFFER_PADDING * 2);
loop->data.send_buf = us_malloc(LIBUS_SEND_BUFFER_LENGTH);
Expand All @@ -144,6 +235,7 @@ int us_internal_loop_data_init(struct us_loop_t *loop, void (*wakeup_cb)(struct
us_free(loop->data.send_buf);
#ifdef LIBUS_USE_LIBUV
us_timer_close(loop->data.sweep_timer, 0);
us_timer_close(loop->data.connect_timer, 0);
#endif
return -1;
}
Expand All @@ -167,6 +259,7 @@ void us_internal_loop_data_free(struct us_loop_t *loop) {

#ifdef LIBUS_USE_LIBUV
us_timer_close(loop->data.sweep_timer, 0);
us_timer_close(loop->data.connect_timer, 0);
if (loop->data.quic_timer) us_timer_close(loop->data.quic_timer, 0);
#endif
us_internal_async_close(loop->data.wakeup_async);
Expand Down
3 changes: 3 additions & 0 deletions packages/bun-usockets/src/socket.c
Original file line number Diff line number Diff line change
Expand Up @@ -210,6 +210,9 @@ void us_connecting_socket_free(struct us_connecting_socket_t *c) {
void us_connecting_socket_close(struct us_connecting_socket_t *c) {
if (c->closed) return;
c->closed = 1;
/* `c` stays linked until its owner has heard the error below, and the owner
* may run the loop from that callback. No address is tried from here on. */
c->next_attempt_ns = 0;
for (struct us_socket_t *s = c->connecting_head; s; s = s->connect_next) {
us_internal_socket_group_unlink_socket(s->group, s);

Expand Down
4 changes: 2 additions & 2 deletions src/runtime/dns_jsc/dns.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2516,8 +2516,8 @@ pub(crate) mod internal {
}

// Pack getaddrinfo results into one allocation with address families
// interleaved (RFC 8305 §4) so an unroutable family can never fill all
// CONCURRENT_CONNECTIONS parallel connect attempts. See #4938 / #33278.
// interleaved (RFC 8305 §4): usockets dials them one at a time, so the
// second attempt is the other family. See #4938 / #33278.
Comment thread
robobun marked this conversation as resolved.
fn process_results(info: *mut AddrInfo) -> Box<[ResultEntry]> {
let mut count: usize = 0;
let mut n_first: usize = 0;
Expand Down
4 changes: 4 additions & 0 deletions src/uws_sys/InternalLoopData.rs
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,10 @@ pub struct InternalLoopData {
pub low_prio_budget: i32,
pub dns_ready_head: *mut ConnectingSocket,
pub closed_connecting_head: *mut ConnectingSocket,
/// Earliest scheduled connect attempt (monotonic ns), or -1. Owned by `loop.c`.
pub(crate) connect_next_attempt_ns: i64,
#[cfg(windows)]
pub(crate) connect_timer: *mut Timer,
/// `bun.Mutex.ReleaseImpl.Type` — must match the C-side `zig_mutex_t`
/// (`packages/bun-usockets/src/internal/loop_data.h`). `Bun__lock`/`Bun__unlock`
/// are called on this field by C, and `loop.c` runtime-checks
Expand Down
Loading
Loading