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
14 changes: 14 additions & 0 deletions packages/bun-usockets/src/bsd.c
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,9 @@ extern int Bun__doesMacOSVersionSupportSendRecvMsgX();

/* We need to emulate sendmmsg, recvmmsg on platform who don't have it */
int bsd_sendmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_sendbuf* sendbuf, int flags) {
ssize_t injected = 0; int unused = 0;
if (US_FAULT_CHECK(US_FAULT_SENDMSG, fd, injected, unused)) return (int) injected;
(void)unused;
#if defined(_WIN32)// || defined(__APPLE__)
for (int i = 0; i < sendbuf->num; i++) {
while (1) {
Expand Down Expand Up @@ -1079,6 +1082,17 @@ int bsd_send_is_transient_error() {
#endif
}

/* libuv's UDP retry set is exactly EAGAIN/EWOULDBLOCK/ENOBUFS (unix/udp.c);
* ENOMEM stays a hard error like Node, so this is narrower than the TCP
* helper above. */
int bsd_udp_send_is_transient_error() {
#ifdef _WIN32
return WSAGetLastError() == WSAENOBUFS;
#else
return errno == ENOBUFS;
#endif
}

static int us_internal_bind_and_listen(LIBUS_SOCKET_DESCRIPTOR listenFd, struct sockaddr *listenAddr, socklen_t listenAddrLength, int backlog, int* error) {
int result;
do
Expand Down
1 change: 1 addition & 0 deletions packages/bun-usockets/src/internal/networking/bsd.h
Original file line number Diff line number Diff line change
Expand Up @@ -227,6 +227,7 @@ 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);
int bsd_would_block();
int bsd_send_is_transient_error();
int bsd_udp_send_is_transient_error();

// return LIBUS_SOCKET_ERROR or the fd that represents listen socket
// listen both on ipv6 and ipv4
Expand Down
7 changes: 5 additions & 2 deletions packages/bun-usockets/src/udp.c
Original file line number Diff line number Diff line change
Expand Up @@ -62,8 +62,11 @@ int us_udp_socket_send(struct us_udp_socket_t *s, void** payloads, size_t* lengt
if (sent < 0) {
/* Linux sendmmsg reports EAGAIN as -1 (per-msg-loop platforms
* report it as sent==0): re-arm writable so on_drain fires and
* report how many earlier batches went out. */
if (bsd_would_block()) {
* report how many earlier batches went out. ENOBUFS joins the
* would-block set because libuv classifies it as EAGAIN for UDP
* sends (macOS returns it when bursty sends fill the interface
* queue), so a later retry can succeed. */
if (bsd_would_block() || bsd_udp_send_is_transient_error()) {
us_poll_change((struct us_poll_t *) s, s->loop, LIBUS_SOCKET_READABLE | LIBUS_SOCKET_WRITABLE);
return total_sent;
}
Expand Down
138 changes: 138 additions & 0 deletions test/js/bun/udp/udp-syscall-fault.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,138 @@
import { socketFaultInjection as fault } from "bun:internal-for-testing";
import { afterEach, describe, expect, test } from "bun:test";
import { isWindows } from "harness";
import dgram from "node:dgram";
import { once } from "node:events";

// Every Bun.udpSocket / node:dgram send reaches the kernel through
// bsd_sendmmsg, which the "sendmsg" fault hook intercepts. Windows maps
// WSAENOBUFS the same way but its errno plumbing differs; land POSIX coverage
// first, like net-syscall-fault.test.ts.
const skip = !fault.available() || isWindows;

afterEach(() => fault.clear());

describe.skipIf(skip)("udp send under injected syscall faults", () => {
// libuv's UDP send retry set is exactly EAGAIN/EWOULDBLOCK/ENOBUFS (macOS
// returns ENOBUFS when bursty sends fill the interface queue). All of them
// must report backpressure (false) and arm the drain callback, not throw;
// the EWOULDBLOCK cell pins the baseline ENOBUFS is specified against.
test.each(["ENOBUFS", "EWOULDBLOCK"] as const)("udpSocket.send → %s is backpressure, not an error", async errno => {
const received = Promise.withResolvers<string>();
using server = await Bun.udpSocket({
port: 0,
hostname: "127.0.0.1",
socket: {
data(_sock, buf) {
received.resolve(buf.toString());
},
},
});
let drained = Promise.withResolvers<void>();
using client = await Bun.udpSocket({
port: 0,
hostname: "127.0.0.1",
socket: {
data() {},
drain() {
drained.resolve();
},
},
});

// The poll starts out writable, so one creation-time drain fires on the
// first loop turn. Consume it now; otherwise it would satisfy the await
// below even if the failed send never re-armed writable.
await drained.promise;
drained = Promise.withResolvers();

fault.set({ syscall: "sendmsg", action: "errno", errno, repeat: 1 });
expect(client.send("hello", server.port, "127.0.0.1")).toBe(false);

// The creation-time drain is spent, so this only resolves if the failed
// send re-armed writable.
await drained.promise;
expect(client.send("hello", server.port, "127.0.0.1")).toBe(true);
expect(await received.promise).toBe("hello");
});

// ENOMEM is deliberately not in the retry set: libuv surfaces it for UDP
// sends and so does Node (only the TCP path treats it as transient).
test.each(["EINVAL", "ENOMEM"] as const)("udpSocket.send still throws on a hard errno (%s)", async errno => {
using server = await Bun.udpSocket({ port: 0, hostname: "127.0.0.1", socket: { data() {} } });
using client = await Bun.udpSocket({ port: 0, hostname: "127.0.0.1", socket: { data() {} } });

fault.set({ syscall: "sendmsg", action: "errno", errno, repeat: 1 });
let thrown: any;
try {
client.send("x", server.port, "127.0.0.1");
} catch (e) {
thrown = e;
}
expect(thrown?.code).toBe(errno);
});

// node:dgram queues a backpressured send and retries it from the drain
// handler, so a transient ENOBUFS must complete the callback with no error
// once the kernel recovers (three consecutive failures here: the initial
// try plus two drain retries, so at least two of the drains require the
// failed send to re-arm writable).
test("dgram send callback succeeds after transient ENOBUFS", async () => {
await using server = dgram.createSocket("udp4");
const messageP = once(server, "message");
server.bind(0, "127.0.0.1");
await once(server, "listening");
await using client = dgram.createSocket("udp4");

fault.set({ syscall: "sendmsg", action: "errno", errno: "ENOBUFS", repeat: 3 });
const cbErr = Promise.withResolvers<Error | null>();
client.send("ping", server.address().port, "127.0.0.1", err => cbErr.resolve(err ?? null));

expect(await cbErr.promise).toBeNull();
const [msg] = await messageP;
expect(msg.toString()).toBe("ping");
});

// More packets than one 16 KiB sendbuf batch holds (~200 on Linux), so
// us_udp_socket_send issues several sendmmsg calls and the second one fails
// with ENOBUFS. The partial count must come back without an error and with
// writable re-armed, so the caller can send the tail from drain.
test("sendMany resumes after ENOBUFS at a batch boundary", async () => {
using server = await Bun.udpSocket({ port: 0, hostname: "127.0.0.1", socket: { data() {} } });
let drained = Promise.withResolvers<void>();
using client = await Bun.udpSocket({
port: 0,
hostname: "127.0.0.1",
socket: {
data() {},
drain() {
drained.resolve();
},
},
});

const N = 1000;
const parts: (string | number)[] = [];
for (let i = 0; i < N; i++) {
parts.push("x", server.port, "127.0.0.1");
}

// Consume the creation-time drain (see above) so the loop below only
// advances on drains the failed batch re-armed.
await drained.promise;
drained = Promise.withResolvers();

fault.set({ syscall: "sendmsg", action: "errno", errno: "ENOBUFS", after: 1, repeat: 1 });
const first = client.sendMany(parts);
expect(first).toBeGreaterThan(0);
expect(first).toBeLessThan(N);

let total = first;
while (total < N) {
await drained.promise;
drained = Promise.withResolvers();
total += client.sendMany(parts.slice(total * 3));
}
expect(total).toBe(N);
});
});