Skip to content
Merged
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
9 changes: 5 additions & 4 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -186,10 +186,11 @@ overflow-checks = false
# back to "warn" (their priority 0 beats the group's -1) where a warning level
# is intentional.
warnings = { level = "deny", priority = -1 }
# `bun_asan` / `bun_debug` are set via RUSTFLAGS (`--cfg=bun_*` +
# `--check-cfg=cfg(bun_*)`) by scripts/build/rust.ts; register them here so a
# plain `cargo build` / `cargo check` (without those flags) doesn't warn.
unexpected_cfgs = { level = "warn", check-cfg = ['cfg(bun_asan)', 'cfg(bun_debug)'] }
# `bun_asan` / `bun_debug` / `socket_fault_injection` are set via RUSTFLAGS
# (`--cfg=...` + `--check-cfg=cfg(...)`) by scripts/build/rust.ts; register
# them here so a plain `cargo build` / `cargo check` (without those flags)
# doesn't warn.
unexpected_cfgs = { level = "warn", check-cfg = ['cfg(bun_asan)', 'cfg(bun_debug)', 'cfg(socket_fault_injection)'] }
# link.exe unconditionally prints "Creating library X.dll.lib and object
# X.dll.exp" to stdout when linking each proc-macro DLL on Windows hosts;
# there is no linker flag to suppress it. The lint already exempts itself
Expand Down
24 changes: 24 additions & 0 deletions packages/bun-usockets/src/bsd.c
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@

#include "libusockets.h"
#include "internal/internal.h"
#include "internal/fault_inject.h"

#include <stdio.h>
#include <stdlib.h>
Expand Down Expand Up @@ -789,6 +790,10 @@ int bsd_addr_get_port(struct bsd_addr_t *addr) {
LIBUS_SOCKET_DESCRIPTOR bsd_accept_socket(LIBUS_SOCKET_DESCRIPTOR fd, struct bsd_addr_t *addr) {
LIBUS_SOCKET_DESCRIPTOR accepted_fd;

ssize_t injected = 0; int unused = 0;
if (US_FAULT_CHECK(US_FAULT_ACCEPT, fd, injected, unused)) return LIBUS_SOCKET_ERROR;
(void)injected; (void)unused;

while (1) {
addr->len = sizeof(addr->mem);

Expand Down Expand Up @@ -842,6 +847,8 @@ LIBUS_SOCKET_DESCRIPTOR bsd_accept_socket(LIBUS_SOCKET_DESCRIPTOR fd, struct bsd
}

ssize_t bsd_recv(LIBUS_SOCKET_DESCRIPTOR fd, void *buf, int length, int flags) {
ssize_t injected = 0;
if (US_FAULT_CHECK(US_FAULT_RECV, fd, injected, length)) return injected;
while (1) {
ssize_t ret = recv(fd, buf, length, flags);

Expand All @@ -866,6 +873,9 @@ ssize_t bsd_recv(LIBUS_SOCKET_DESCRIPTOR fd, void *buf, int length, int flags) {

#if !defined(_WIN32)
ssize_t bsd_recvmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct msghdr *msg, int flags) {
ssize_t injected = 0; int unused = 0;
if (US_FAULT_CHECK(US_FAULT_RECVMSG, fd, injected, unused)) return injected;
(void)unused;
while (1) {
ssize_t ret = recvmsg(fd, msg, flags);

Expand All @@ -882,6 +892,9 @@ ssize_t bsd_recvmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct msghdr *msg, int flags) {
#include <sys/uio.h>

ssize_t bsd_writev(LIBUS_SOCKET_DESCRIPTOR fd, const struct us_iovec_t *iov, int count) {
ssize_t injected = 0; int unused = 0;
if (US_FAULT_CHECK(US_FAULT_WRITEV, fd, injected, unused)) return injected;
(void)unused;
/* POSIX writev fails with EINVAL above IOV_MAX (1024 on Linux/macOS); cap and
* let the caller's partial-write handling carry the remainder. */
if (count > 1024) {
Expand All @@ -897,6 +910,9 @@ ssize_t bsd_writev(LIBUS_SOCKET_DESCRIPTOR fd, const struct us_iovec_t *iov, int
}

ssize_t bsd_write2(LIBUS_SOCKET_DESCRIPTOR fd, const char *header, int header_length, const char *payload, int payload_length) {
ssize_t injected = 0; int unused = 0;
if (US_FAULT_CHECK(US_FAULT_WRITEV, fd, injected, unused)) return injected;
(void)unused;
struct iovec chunks[2];

chunks[0].iov_base = (char *)header;
Expand Down Expand Up @@ -938,6 +954,8 @@ ssize_t bsd_write2(LIBUS_SOCKET_DESCRIPTOR fd, const char *header, int header_le
#endif

ssize_t bsd_send(LIBUS_SOCKET_DESCRIPTOR fd, const char *buf, int length) {
ssize_t injected = 0;
if (US_FAULT_CHECK(US_FAULT_SEND, fd, injected, length)) return injected;
while (1) {
// MSG_MORE (Linux), MSG_PARTIAL (Windows), TCP_NOPUSH (BSD)

Expand Down Expand Up @@ -969,6 +987,9 @@ ssize_t bsd_send(LIBUS_SOCKET_DESCRIPTOR fd, const char *buf, int length) {

#if !defined(_WIN32)
ssize_t bsd_sendmsg(LIBUS_SOCKET_DESCRIPTOR fd, const struct msghdr *msg, int flags) {
ssize_t injected = 0; int unused = 0;
if (US_FAULT_CHECK(US_FAULT_SENDMSG, fd, injected, unused)) return injected;
(void)unused;
while (1) {
ssize_t rc = sendmsg(fd, msg, flags);

Expand Down Expand Up @@ -1609,6 +1630,9 @@ int bsd_disconnect_udp_socket(LIBUS_SOCKET_DESCRIPTOR fd) {

static int bsd_do_connect_raw(LIBUS_SOCKET_DESCRIPTOR fd, struct sockaddr *addr, size_t namelen)
{
ssize_t injected = 0; int unused = 0;
if (US_FAULT_CHECK(US_FAULT_CONNECT, fd, injected, unused)) return errno;
(void)injected; (void)unused;
Comment thread
cirospaciari marked this conversation as resolved.
#ifdef _WIN32
while (1) {
if (connect(fd, (struct sockaddr *)addr, namelen) == 0) {
Expand Down
5 changes: 4 additions & 1 deletion packages/bun-usockets/src/eventing/epoll_kqueue.c
Original file line number Diff line number Diff line change
Expand Up @@ -203,7 +203,10 @@ static void us_internal_dispatch_ready_polls(struct us_loop_t *loop) {
continue;
}
int events = loop->ready_polls[loop->current_ready_poll].events;
const int error = events & EPOLLERR;
/* Normalize to 0/1 like the kqueue path's EV_ERROR: the value is
* forwarded as a libus close code, and a raw EPOLLERR (8) would
* read as errno 8 (ENOEXEC) in the JS error path. */
const int error = !!(events & EPOLLERR);
const int eof = events & EPOLLHUP;
events &= us_poll_events(poll);
if (events || error || eof) {
Expand Down
115 changes: 115 additions & 0 deletions packages/bun-usockets/src/fault_inject.c
Original file line number Diff line number Diff line change
@@ -0,0 +1,115 @@
#include "internal/internal.h"
#include "internal/fault_inject.h"

#if defined(LIBUS_SOCKET_FAULT_INJECTION) && LIBUS_SOCKET_FAULT_INJECTION

#include <errno.h>
#include <string.h>

int us_fault_armed = 0;

struct us_fault_slot {
struct us_fault_rule rule;
int calls_seen;
int fired;
};

/* Process-global so rules armed on the JS thread also affect the HTTP-client
* thread (fetch) and any worker threads. Per-socket isolation is provided by
* rule.target_fd instead. */
static struct us_fault_slot us_fault_state[US_FAULT_COUNT];

/* Guards every access to us_fault_state so a re-arm on the JS thread cannot
* tear the rule (or its counters) out from under a faulting I/O thread. Only
* taken once the lock-free us_fault_armed fast path in US_FAULT_CHECK has
* passed, so the disarmed hot path never touches it. */
static zig_mutex_t us_fault_lock;

/* Caller must hold us_fault_lock. The release store pairs with the acquire
* load in US_FAULT_CHECK; a reader that observes armed==1 then re-reads the
* rule under the lock in us_fault_hit(), so it never sees a torn rule. */
static void us_fault_recompute_armed(void) {
int any = 0;
for (int i = 0; i < US_FAULT_COUNT; i++) {
if (us_fault_state[i].rule.action != US_FAULT_NONE) {
any = 1;
break;
}
}
__atomic_store_n(&us_fault_armed, any, __ATOMIC_RELEASE);
}

void us_fault_set(int sc, const struct us_fault_rule *rule) {
if ((unsigned)sc >= US_FAULT_COUNT) return;
Bun__lock(&us_fault_lock);
us_fault_state[sc].rule = *rule;
us_fault_state[sc].calls_seen = 0;
us_fault_state[sc].fired = 0;
us_fault_recompute_armed();
Comment thread
cirospaciari marked this conversation as resolved.
Bun__unlock(&us_fault_lock);
}

void us_fault_clear(int sc) {
if ((unsigned)sc >= US_FAULT_COUNT) return;
Bun__lock(&us_fault_lock);
us_fault_state[sc].rule.action = US_FAULT_NONE;
us_fault_state[sc].calls_seen = 0;
us_fault_state[sc].fired = 0;
us_fault_recompute_armed();
Bun__unlock(&us_fault_lock);
}

void us_fault_clear_all(void) {
Bun__lock(&us_fault_lock);
for (int i = 0; i < US_FAULT_COUNT; i++) {
us_fault_state[i].rule.action = US_FAULT_NONE;
}
__atomic_store_n(&us_fault_armed, 0, __ATOMIC_RELEASE);
Bun__unlock(&us_fault_lock);
}

int us_fault_hit(int sc, int fd, ssize_t *out, int *clamp) {
if ((unsigned)sc >= US_FAULT_COUNT) return 0;
Bun__lock(&us_fault_lock);
struct us_fault_slot *slot = &us_fault_state[sc];
Comment thread
coderabbitai[bot] marked this conversation as resolved.
/* Snapshot under the lock so the post-release switch below acts on one
* coherent rule even if another thread swaps it right after we unlock.
* Single lock/unlock pair: every exit funnels through the one release. */
struct us_fault_rule rule = slot->rule;
int fire = 0;
if (rule.action != US_FAULT_NONE && (rule.target_fd < 0 || rule.target_fd == fd)) {
int seen = slot->calls_seen++;
if (seen >= rule.after_n_calls) {
int f = slot->fired++;
if (rule.repeat >= 0 && f >= rule.repeat) {
slot->rule.action = US_FAULT_NONE;
us_fault_recompute_armed();
} else {
fire = 1;
}
}
}
Bun__unlock(&us_fault_lock);
if (!fire) return 0;
switch (rule.action) {
case US_FAULT_ERRNO:
#ifdef _WIN32
WSASetLastError(rule.errno_value);
#endif
errno = rule.errno_value;
*out = -1;
return 1;
case US_FAULT_ZERO:
*out = 0;
return 1;
case US_FAULT_SHORT:
if (rule.clamp_bytes >= 0 && *clamp > rule.clamp_bytes) {
*clamp = rule.clamp_bytes;
}
return 0;
default:
return 0;
}
}

#endif /* LIBUS_SOCKET_FAULT_INJECTION */
94 changes: 94 additions & 0 deletions packages/bun-usockets/src/internal/fault_inject.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
/*
* Socket syscall fault injection for the bsd_* wrapper layer.
*
* Compiled in only when LIBUS_SOCKET_FAULT_INJECTION is defined (controlled
* by the `socketFaultInjection` Config field). When compiled out,
* US_FAULT_CHECK() expands to a constant 0 and the optimizer drops every
* reference. When compiled in but no rule is armed, the hot path is a single
* acquire atomic load + predicted-not-taken branch.
*
* Rules are process-global so a rule armed from the JS thread also affects
* the HTTP-client thread (fetch) and worker threads. Use rule.target_fd for
* per-socket isolation.
*/
// clang-format off
#pragma once
#ifndef LIBUS_FAULT_INJECT_H
#define LIBUS_FAULT_INJECT_H

#if defined(LIBUS_SOCKET_FAULT_INJECTION) && LIBUS_SOCKET_FAULT_INJECTION

#ifdef _WIN32
#include <BaseTsd.h>
typedef SSIZE_T ssize_t;
#else
#include <sys/types.h>
#endif

enum us_fault_syscall {
US_FAULT_RECV,
US_FAULT_SEND,
US_FAULT_WRITEV,
US_FAULT_SENDMSG,
US_FAULT_RECVMSG,
US_FAULT_CONNECT,
US_FAULT_ACCEPT,
/* Reserved: no bsd.c hooks yet, so the JS setter does not accept them. */
US_FAULT_SOCKET,
US_FAULT_CLOSE,
US_FAULT_SHUTDOWN,
US_FAULT_COUNT
};

enum us_fault_action {
US_FAULT_NONE,
/* return -1 and set errno = errno_value */
US_FAULT_ERRNO,
/* recv/send: clamp the length to clamp_bytes, then run the real syscall.
* Other syscalls have no length to clamp; the JS setter rejects them. */
US_FAULT_SHORT,
/* recv/recvmsg: return 0 (peer closed); send/sendmsg/writev: return 0
* (treated as backpressure). The JS setter rejects other syscalls. */
US_FAULT_ZERO,
};

struct us_fault_rule {
int action;
int errno_value;
int clamp_bytes;
/* skip the first N matching calls before triggering */
int after_n_calls;
/* fire this many times then disarm; -1 = forever */
int repeat;
/* match only this fd; -1 = any */
int target_fd;
};

#ifdef __cplusplus
extern "C" {
#endif

extern int us_fault_armed;

void us_fault_set(int syscall, const struct us_fault_rule *rule);
void us_fault_clear(int syscall);
void us_fault_clear_all(void);
int us_fault_hit(int syscall, int fd, ssize_t *out, int *clamp);

#ifdef __cplusplus
}
#endif

/* Hot-path macro. `out_lvalue` and `clamp_lvalue` must be lvalues. Returns 1
* when the call should short-circuit with *out_lvalue. clamp_lvalue may be
* shrunk in-place when the rule wants a partial read/write. */
#define US_FAULT_CHECK(sc, fd, out_lvalue, clamp_lvalue) \
(__builtin_expect(__atomic_load_n(&us_fault_armed, __ATOMIC_ACQUIRE), 0) \
&& us_fault_hit((sc), (int)(fd), &(out_lvalue), &(clamp_lvalue)))

#else /* !LIBUS_SOCKET_FAULT_INJECTION */

#define US_FAULT_CHECK(sc, fd, out_lvalue, clamp_lvalue) 0

#endif
#endif /* LIBUS_FAULT_INJECT_H */
20 changes: 16 additions & 4 deletions packages/bun-usockets/src/loop.c
Original file line number Diff line number Diff line change
Expand Up @@ -382,8 +382,13 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
}
case POLL_TYPE_SEMI_SOCKET: {
/* Both connect and listen sockets are semi-sockets
* but they poll for different events */
if (us_poll_events(p) == LIBUS_SOCKET_WRITABLE) {
* but they poll for different events. A connecting socket starts
* out polling WRITABLE only, but a write issued before the connect
* completes (the kernel buffers it) can partial-write and call
* us_poll_change(R|W) from us_socket_write*. Test the WRITABLE bit
* rather than exact equality so the connect-complete is still
* recognized; listen sockets only ever poll READABLE. */
if (us_poll_events(p) & LIBUS_SOCKET_WRITABLE) {
/* The connecting fd became writable with an error/HUP flag also
* set: the handshake may have completed and then been reset
* before we collected the event. Report the kernel's actual
Expand Down Expand Up @@ -698,8 +703,15 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
if (error && s) {
/* Peer-initiated error event — same rationale as the recv-error
* branch above: bypass us_internal_ssl_close so on_handshake
* isn't fired for a passive close. */
s = us_internal_socket_close_raw(s, error, NULL);
* isn't fired for a passive close. The poll flag only says THAT
* the socket failed; fetch the real errno (like the connect-error
* path) so the close code is ECONNRESET and not a poll bit that
* callers would either misread as an errno or drop entirely.
* Values 0..2 collide with the libus CloseCode enum that JS
* filters out as self-initiated; SO_ERROR can't be EPERM/ENOENT
* for an established TCP socket, so clamp them defensively. */
int socket_error = us_socket_get_error(s);
s = us_internal_socket_close_raw(s, socket_error > 2 ? socket_error : ECONNRESET, NULL);
return;
}
break;
Expand Down
8 changes: 6 additions & 2 deletions packages/bun-uws/src/WebSocketProtocol.h
Original file line number Diff line number Diff line change
Expand Up @@ -129,8 +129,12 @@ static inline CloseFrame parseClosePayload(char *src, size_t length) {
if (length >= 2) {
memcpy(&cf.code, src, 2);
cf = {cond_byte_swap<uint16_t>(cf.code), src + 2, length - 2};
if (cf.code < 1000 || cf.code > 4999 || (cf.code > 1011 && cf.code < 4000) ||
(cf.code >= 1004 && cf.code <= 1006) || !isValidUtf8((unsigned char *) cf.message, cf.length)) {
// RFC 6455 §7.4: 1000-1015 defined, 1016-2999 reserved (MUST NOT be
// used), 3000-3999 IANA-registered for libraries/frameworks, 4000-4999
// private use. 1004/1005/1006/1015 are not valid on the wire.
if (cf.code < 1000 || cf.code > 4999 || (cf.code > 1015 && cf.code < 3000) ||
(cf.code >= 1004 && cf.code <= 1006) || cf.code == 1015 ||
!isValidUtf8((unsigned char *) cf.message, cf.length)) {
/* Even though we got a WebSocket close frame, it in itself is abnormal */
return {1006, nullptr, 0};
}
Expand Down
1 change: 1 addition & 0 deletions scripts/build.ts
Original file line number Diff line number Diff line change
Expand Up @@ -384,6 +384,7 @@ function parseArgs(argv: string[]): CliArgs {
"tinycc",
"valgrind",
"fuzzilli",
"socketFaultInjection",
"unifiedSources",
"archiveDeps",
"timeTrace",
Expand Down
Loading
Loading