diff --git a/packages/bun-usockets/src/loop.c b/packages/bun-usockets/src/loop.c index baba410af040..34a2cca97197 100644 --- a/packages/bun-usockets/src/loop.c +++ b/packages/bun-usockets/src/loop.c @@ -457,6 +457,46 @@ void us_internal_loop_post(struct us_loop_t *loop) { #define us_ioctl ioctl #endif +#if defined(__linux__) +/* Drain a UDP socket's MSG_ERRQUEUE. Each remote-origin (ICMP/ICMP6) entry is + * surfaced via on_recv_error with is_errqueue=1; SO_EE_ORIGIN_LOCAL entries are + * skipped — the failing syscall already returned that errno synchronously, so + * redelivering it here is a double report. After draining, SO_ERROR is read to + * clear any mirrored sk_err so EPOLLERR does not stay level-set with an empty + * queue (which would fall through to recvmmsg and surface the same errno again + * without the errqueue flag). Returns the number of entries dequeued. */ +static int us_internal_udp_drain_errqueue(struct us_udp_socket_t *u, LIBUS_SOCKET_DESCRIPTOR fd) { + int drained = 0; + struct msghdr eh; char ectrl[512]; char ebuf[1]; + struct iovec eiov = { ebuf, sizeof(ebuf) }; + while (!u->closed) { + memset(&eh, 0, sizeof(eh)); + eh.msg_iov = &eiov; eh.msg_iovlen = 1; + eh.msg_control = ectrl; eh.msg_controllen = sizeof(ectrl); + if (recvmsg(fd, &eh, MSG_ERRQUEUE) < 0) break; + drained++; + if (!u->on_recv_error) continue; + int ee = 0, origin = -1; + for (struct cmsghdr *cm = CMSG_FIRSTHDR(&eh); cm; cm = CMSG_NXTHDR(&eh, cm)) { + if ((cm->cmsg_level == IPPROTO_IP && cm->cmsg_type == IP_RECVERR) || + (cm->cmsg_level == IPPROTO_IPV6 && cm->cmsg_type == IPV6_RECVERR)) { + struct sock_extended_err *se = (struct sock_extended_err *) CMSG_DATA(cm); + ee = se->ee_errno; + origin = se->ee_origin; + break; + } + } + if (origin == SO_EE_ORIGIN_LOCAL) continue; + u->on_recv_error(u, ee ? ee : ECONNREFUSED, 1); + } + if (drained && !u->closed) { + int so_err = 0; socklen_t sl = sizeof(so_err); + getsockopt(fd, SOL_SOCKET, SO_ERROR, &so_err, &sl); + } + return drained; +} +#endif + void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, int events) { switch (us_internal_poll_type(p)) { case POLL_TYPE_CALLBACK: { @@ -904,28 +944,8 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in int recv_error_surfaced = 0; int recv_would_block_only = 0; if (error) { - struct msghdr eh; char ectrl[512]; char ebuf[1]; - struct iovec eiov = { ebuf, sizeof(ebuf) }; - while (!u->closed) { - memset(&eh, 0, sizeof(eh)); - eh.msg_iov = &eiov; eh.msg_iovlen = 1; - eh.msg_control = ectrl; eh.msg_controllen = sizeof(ectrl); - if (recvmsg(us_poll_fd(p), &eh, MSG_ERRQUEUE) < 0) break; - recv_error_surfaced = 1; - if (u->on_recv_error) { - /* The queued ICMP error is in sock_extended_err, - * not errno. */ - int ee = 0; - for (struct cmsghdr *cm = CMSG_FIRSTHDR(&eh); cm; cm = CMSG_NXTHDR(&eh, cm)) { - if ((cm->cmsg_level == IPPROTO_IP && cm->cmsg_type == IP_RECVERR) || - (cm->cmsg_level == IPPROTO_IPV6 && cm->cmsg_type == IPV6_RECVERR)) { - ee = ((struct sock_extended_err *) CMSG_DATA(cm))->ee_errno; - break; - } - } - u->on_recv_error(u, ee ? ee : ECONNREFUSED, 1); - } - } + recv_error_surfaced = + us_internal_udp_drain_errqueue(u, us_poll_fd(p)) > 0; } /* An EPOLLERR with no EPOLLIN and an empty error queue still has to @@ -967,6 +987,18 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in if (u->on_recv_error) { #if defined(__linux__) recv_error_surfaced = 1; + /* On Linux the kernel mirrors a queued + * ICMP into sk_err, so recvmmsg can return + * it here before EPOLLERR is dispatched + * (e.g. on_data just sent a reply to a + * port that already closed). If the error + * queue has the entry, that is the source + * and the recvmmsg errno is its mirror — + * surface the queue (is_errqueue=1), not a + * second un-flagged copy. */ + if (us_internal_udp_drain_errqueue(u, us_poll_fd(p)) > 0) { + break; + } #endif u->on_recv_error(u, recv_err, 0); } else { diff --git a/src/js/node/dgram.ts b/src/js/node/dgram.ts index 85295097a0a1..3b78d80bc0c5 100644 --- a/src/js/node/dgram.ts +++ b/src/js/node/dgram.ts @@ -724,12 +724,6 @@ function startBunSocket(self, state, createOptions) { }, error: error => { if (error?.syscall === "recv") { - // Drop errqueue-origin ICMP errors on unconnected sockets like - // Node (which never enables IP_RECVERR); always emit real - // recvmmsg failures — the errno namespaces overlap. - if (error.errqueue === true && state.connectState !== CONNECT_STATE_CONNECTED) { - return; - } // Node reports receive-path failures as `recvmsg` errors; keep // the native error so its code stays platform-correct. error.syscall = "recvmsg"; diff --git a/src/runtime/socket/udp_socket.rs b/src/runtime/socket/udp_socket.rs index e8c48b5e3b98..31cbf0e11199 100644 --- a/src/runtime/socket/udp_socket.rs +++ b/src/runtime/socket/udp_socket.rs @@ -95,9 +95,12 @@ extern "C" fn on_recv_error(socket: *mut uws::udp::Socket, errno: c_int, is_errq // Reached on every POSIX platform. `is_errqueue` distinguishes an ICMP // errno drained from Linux's MSG_ERRQUEUE from a real recvmmsg failure — // which on the BSDs (no error queue) is also how a connected socket's - // ICMP error (so_error) arrives. node:dgram must drop only the former on - // unconnected sockets, and the errno namespaces overlap. + // ICMP error (so_error) arrives. Node never enables IP_RECVERR, so its + // kernel drops ICMP for an unconnected socket; do the same here. let this: &UDPSocket = UDPSocket::from_uws(socket); + if is_errqueue != 0 && this.connect_info.get().is_none() { + return; + } let sys_err = bun_sys::Error::from_code_int(errno, bun_sys::Tag::recv); let global_this = this.global_this.get(); // A callback earlier in the same poll dispatch may have left a @@ -109,11 +112,7 @@ extern "C" fn on_recv_error(socket: *mut uws::udp::Socket, errno: c_int, is_errq if global_this.has_exception() { return; } - let err_value = sys_err.to_js(global_this); - if is_errqueue != 0 { - err_value.put(global_this, b"errqueue", JSValue::TRUE); - } - this.call_error_handler(JSValue::ZERO, err_value); + this.call_error_handler(JSValue::ZERO, sys_err.to_js(global_this)); } extern "C" fn on_drain(socket: *mut uws::udp::Socket) { diff --git a/test/js/bun/udp/dgram.test.ts b/test/js/bun/udp/dgram.test.ts index 0bddd52d6d90..375ebcff5d2f 100644 --- a/test/js/bun/udp/dgram.test.ts +++ b/test/js/bun/udp/dgram.test.ts @@ -2,7 +2,7 @@ import { describe, expect, jest, test } from "bun:test"; import { createSocket } from "dgram"; import { Worker } from "node:worker_threads"; -import { bunEnv, bunExe, disableAggressiveGCScope, isWindows } from "harness"; +import { bunEnv, bunExe, disableAggressiveGCScope, isLinux, isWindows } from "harness"; import path from "path"; import { nodeDataCases } from "./testdata"; @@ -439,6 +439,51 @@ test("setBroadcast()/setMulticastLoopback() before bind() throw EBADF", () => { socket.close(); }); +// An echo server in the Node docs shape (no 'error' listener) must survive a +// client that sends one datagram and closes before the reply arrives. Linux +// only: without IP_RECVERR the kernel never sets sk_err for an unconnected +// socket, so the failure mode this guards does not exist elsewhere. +test.skipIf(!isLinux)("unconnected echo server without an 'error' listener survives a vanished client", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + import dgram from "node:dgram"; + const server = dgram.createSocket("udp4"); + server.on("message", (msg, rinfo) => server.send(msg, rinfo.port, rinfo.address)); + await new Promise(r => server.bind(0, "127.0.0.1", r)); + const port = server.address().port; + // First client: send and close immediately so the reply hits a dead port. + const c1 = dgram.createSocket("udp4"); + await new Promise(r => c1.bind(0, "127.0.0.1", r)); + c1.send("gone", port, "127.0.0.1", () => c1.close()); + // Second client: must receive its echo after the ICMP. + const c2 = dgram.createSocket("udp4"); + await new Promise(r => c2.bind(0, "127.0.0.1", r)); + const got = await new Promise((resolve, reject) => { + let n = 0; + const t = setInterval(() => { + if (++n > 400) { clearInterval(t); return reject(new Error("timeout")); } + c2.send("probe", port, "127.0.0.1"); + }, 10); + c2.on("message", m => { clearInterval(t); resolve(m.toString()); }); + }); + console.log(JSON.stringify({ got })); + c2.close(); + server.close(); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout.trim()).toBe(`{"got":"probe"}`); + expect(exitCode).toBe(0); +}); + // An oversized datagram fails send(2) with EMSGSIZE; on the connected path no // address/port is known, so the error must match Node's bare `send `. test.skipIf(isWindows)("connected send() failure reports Node's error shape", async () => { diff --git a/test/js/bun/udp/udp_socket.test.ts b/test/js/bun/udp/udp_socket.test.ts index f49f07dba506..f33a1376eef7 100644 --- a/test/js/bun/udp/udp_socket.test.ts +++ b/test/js/bun/udp/udp_socket.test.ts @@ -1,7 +1,7 @@ import { udpSocket } from "bun"; import { heapStats } from "bun:jsc"; import { describe, expect, test } from "bun:test"; -import { bunEnv, bunExe, disableAggressiveGCScope, isWindows, randomPort } from "harness"; +import { bunEnv, bunExe, disableAggressiveGCScope, isLinux, isWindows, randomPort } from "harness"; import path from "node:path"; import { dataCases, dataTypes } from "./testdata"; @@ -585,6 +585,110 @@ describe("udpSocket()", () => { }); }); +// IP_RECVERR is Linux-only: on macOS/BSD the kernel never queues ICMP for an +// unconnected UDP socket, and on Windows SIO_UDP_CONNRESET is already off. +describe.skipIf(!isLinux)("unconnected socket vs. remote ICMP", () => { + // A reply server in the shape the docs show (no error handler) must not be + // killed by a client that sends one datagram and closes before the reply + // arrives: the reply hits a closed port, the kernel answers with ICMP port + // unreachable, and that used to surface as an uncaught ECONNREFUSED. + test("reply server without an error handler survives a vanished client", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + import dgram from "node:dgram"; + const server = await Bun.udpSocket({ + port: 0, + hostname: "127.0.0.1", + socket: { + data(socket, data, port, address) { socket.send(data, port, address); }, + }, + }); + // First client: send and close immediately so the reply hits a dead port. + const c1 = dgram.createSocket("udp4"); + await new Promise(r => c1.bind(0, "127.0.0.1", r)); + c1.send("gone", server.port, "127.0.0.1", () => c1.close()); + // Second client: the server MUST answer it after the ICMP. + const c2 = dgram.createSocket("udp4"); + await new Promise(r => c2.bind(0, "127.0.0.1", r)); + const got = await new Promise((resolve, reject) => { + let n = 0; + const t = setInterval(() => { + if (server.closed) { clearInterval(t); return reject(new Error("server closed")); } + if (++n > 400) { clearInterval(t); return reject(new Error("timeout")); } + c2.send("probe", server.port, "127.0.0.1"); + }, 10); + c2.on("message", m => { clearInterval(t); resolve(m.toString()); }); + }); + console.log(JSON.stringify({ got, closed: server.closed })); + c2.close(); + server.close(); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, rawStderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + const stderr = rawStderr + .split("\n") + .filter(l => l && !l.startsWith("WARNING: ASAN interferes")) + .join("\n"); + expect(stderr).toBe(""); + expect(stdout.trim()).toBe(`{"got":"probe","closed":false}`); + expect(exitCode).toBe(0); + }); + + // send() rejects an oversized payload synchronously with EMSGSIZE; the same + // error used to also be queued on the socket's error queue (SO_EE_ORIGIN_LOCAL) + // and redelivered asynchronously. A connected socket exercises the origin + // check directly (on an unconnected one the errqueue drop would mask it). + test("oversized send() throws EMSGSIZE once and does not redeliver it asynchronously", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const errs = []; + const peer = await Bun.udpSocket({ port: 0, hostname: "127.0.0.1" }); + const s = await Bun.udpSocket({ + connect: { hostname: "127.0.0.1", port: peer.port }, + socket: { + data() {}, + error(err) { errs.push(err?.code ?? String(err)); }, + }, + }); + let syncErr = null; + try { + s.send(new Uint8Array(65510)); + } catch (e) { + syncErr = e?.code ?? String(e); + } + // Drive the loop past the EPOLLERR that draining the local-origin + // errqueue entry produces. + for (let i = 0; i < 10; i++) await new Promise(r => setImmediate(r)); + console.log(JSON.stringify({ syncErr, asyncErrs: errs, closed: s.closed })); + s.close(); + peer.close(); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, rawStderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + const stderr = rawStderr + .split("\n") + .filter(l => l && !l.startsWith("WARNING: ASAN interferes")) + .join("\n"); + expect(stderr).toBe(""); + expect(stdout.trim()).toBe(`{"syncErr":"EMSGSIZE","asyncErrs":[],"closed":false}`); + expect(exitCode).toBe(0); + }); +}); + // us_udp_socket_send batches at most ~204 messages per sendmmsg; a >batch-size // sendMany must loop and report the TOTAL accepted. The pre-fix loop condition // compared against a decremented `num` and stopped after one batch, which every diff --git a/test/js/bun/udp/udp_socket_recv_flags.test.ts b/test/js/bun/udp/udp_socket_recv_flags.test.ts index 548a13307799..b52dec5fb021 100644 --- a/test/js/bun/udp/udp_socket_recv_flags.test.ts +++ b/test/js/bun/udp/udp_socket_recv_flags.test.ts @@ -37,23 +37,22 @@ describe("udpSocket() receive flags", () => { }); // IP_RECVERR is Linux-specific. On BSDs and Windows, ICMP errors on - // unconnected UDP sockets either propagate by default or are delivered - // through different channels that we don't currently surface. + // unconnected UDP sockets are not queued by the kernel at all. The + // I/O-after-ICMP round-trip is covered by the reply-server test in + // udp_socket.test.ts; Bun.udpSocket exposes no public reconnect to + // exercise it on the connected path. test.skipIf(!isLinux)( - "surfaces ECONNREFUSED from ICMP port unreachable (IP_RECVERR) and keeps the socket usable", + "connected socket surfaces ECONNREFUSED from ICMP port unreachable and stays open", async () => { const { promise: errPromise, resolve: resolveErr } = Promise.withResolvers(); - const { promise: msgPromise, resolve: resolveMsg } = Promise.withResolvers(); - const receiver = await udpSocket({ - socket: { - data(_socket, data) { - resolveMsg(data.toString()); - }, - }, - }); + // Bind then close a probe so the port is known-dead. + const probe = await udpSocket({}); + const deadPort = probe.port; + probe.close(); const sender = await udpSocket({ + connect: { hostname: "127.0.0.1", port: deadPort }, socket: { error(err: Error & { code?: string }) { resolveErr(err); @@ -61,12 +60,10 @@ describe("udpSocket() receive flags", () => { }, }); - // Send to a closed port on localhost. The kernel replies with ICMP - // port unreachable; with IP_RECVERR the next recv surfaces ECONNREFUSED. let gotError = false; function sendDead() { if (!gotError && !sender.closed) { - sender.send("dead", 1, "127.0.0.1"); + sender.send("dead"); setTimeout(sendDead, 10); } } @@ -76,20 +73,9 @@ describe("udpSocket() receive flags", () => { const err = await errPromise; gotError = true; expect(err?.code).toBe("ECONNREFUSED"); - // The sender socket must remain usable after an ICMP error. expect(sender.closed).toBe(false); - - function sendAlive() { - if (!sender.closed && !receiver.closed) { - sender.send("alive", receiver.port, "127.0.0.1"); - setTimeout(sendAlive, 10); - } - } - sendAlive(); - expect(await msgPromise).toBe("alive"); } finally { sender.close(); - receiver.close(); } }, );