diff --git a/packages/bun-usockets/src/bsd.c b/packages/bun-usockets/src/bsd.c index fd9fb92ed7d0..1b45c2a9958f 100644 --- a/packages/bun-usockets/src/bsd.c +++ b/packages/bun-usockets/src/bsd.c @@ -1507,21 +1507,6 @@ LIBUS_SOCKET_DESCRIPTOR bsd_create_udp_socket(const char *host, int port, int op } #endif -#if defined(__linux__) - /* Linux suppresses ICMP errors (port unreachable, host unreachable, TTL - * exceeded, etc.) on unconnected UDP sockets by default. Enabling - * IP_RECVERR/IPV6_RECVERR surfaces them as errors on the next send/recv, - * rather than silently dropping them. Matches libuv. */ -#ifdef IP_RECVERR - setsockopt(listenFd, IPPROTO_IP, IP_RECVERR, &enabled, sizeof(enabled)); -#endif -#ifdef IPV6_RECVERR - if (listenAddr->ai_family == AF_INET6) { - setsockopt(listenFd, IPPROTO_IPV6, IPV6_RECVERR, &enabled, sizeof(enabled)); - } -#endif -#endif - /* We bind here as well */ if (bind(listenFd, listenAddr->ai_addr, (socklen_t) listenAddr->ai_addrlen)) { if (err != NULL) { @@ -1543,6 +1528,28 @@ LIBUS_SOCKET_DESCRIPTOR bsd_create_udp_socket(const char *host, int port, int op return listenFd; } +/* Linux drops ICMP errors (port unreachable, host unreachable, TTL exceeded) + * on unconnected UDP sockets unless IP_RECVERR is set, and node relies on that: + * only a connected socket has a peer to attribute the error to. Disabling it + * purges whatever the kernel already queued. */ +static void bsd_udp_socket_set_recverr(LIBUS_SOCKET_DESCRIPTOR fd, int enabled) { +#if defined(__linux__) +#ifdef IP_RECVERR + setsockopt(fd, IPPROTO_IP, IP_RECVERR, &enabled, sizeof(enabled)); +#endif +#ifdef IPV6_RECVERR + int domain = AF_INET; + socklen_t domain_len = sizeof(domain); + if (getsockopt(fd, SOL_SOCKET, SO_DOMAIN, &domain, &domain_len) == 0 && domain == AF_INET6) { + setsockopt(fd, IPPROTO_IPV6, IPV6_RECVERR, &enabled, sizeof(enabled)); + } +#endif +#else + (void) fd; + (void) enabled; +#endif +} + int bsd_connect_udp_socket(LIBUS_SOCKET_DESCRIPTOR fd, const char *host, int port) { struct addrinfo hints, *result; memset(&hints, 0, sizeof(struct addrinfo)); @@ -1566,6 +1573,7 @@ int bsd_connect_udp_socket(LIBUS_SOCKET_DESCRIPTOR fd, const char *host, int por for (struct addrinfo *rp = result; rp != NULL; rp = rp->ai_next) { if (connect(fd, rp->ai_addr, rp->ai_addrlen) == 0) { freeaddrinfo(result); + bsd_udp_socket_set_recverr(fd, 1); return 0; } } @@ -1591,6 +1599,7 @@ int bsd_disconnect_udp_socket(LIBUS_SOCKET_DESCRIPTOR fd) { errno == EAFNOSUPPORT #endif ) { + bsd_udp_socket_set_recverr(fd, 0); return 0; } else { return -1; diff --git a/packages/bun-usockets/src/loop.c b/packages/bun-usockets/src/loop.c index c602732bb98a..a40c070afc82 100644 --- a/packages/bun-usockets/src/loop.c +++ b/packages/bun-usockets/src/loop.c @@ -744,15 +744,15 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in } #if defined(__linux__) - /* On Linux with IP_RECVERR, EPOLLERR fires when an ICMP error - * (port unreachable, host unreachable, TTL exceeded, ...) is - * queued on the socket's error queue. For an *unconnected* UDP - * socket regular recvmmsg does NOT dequeue these — only - * recvmsg(MSG_ERRQUEUE) does — so EPOLLERR stays level-triggered - * until we drain it explicitly. Do that here, surfacing each - * errno via on_recv_error; the socket stays open. On other - * platforms (kqueue EV_ERROR, Windows) an error event is fatal — - * preserve close-on-error there. */ + /* On Linux, connected UDP sockets have IP_RECVERR armed (see + * bsd_connect_udp_socket), so EPOLLERR fires with the ICMP error + * (port unreachable, host unreachable, TTL exceeded, ...) sitting + * on the socket's error queue, where only recvmsg(MSG_ERRQUEUE) + * dequeues it — EPOLLERR stays level-triggered until we drain it + * explicitly. Do that here, surfacing each errno via + * on_recv_error; the socket stays open. On other platforms (kqueue + * EV_ERROR, Windows) an error event is fatal — preserve + * close-on-error there. */ int recv_error_surfaced = 0; int recv_would_block_only = 0; if (error) { @@ -778,6 +778,20 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in u->on_recv_error(u, ee ? ee : ECONNREFUSED); } } + /* sk_err outlives the error queue: disabling IP_RECVERR on + * disconnect purges the queue but not sk_err, and a socket + * whose IP_RECVERR setup failed never queues at all. Consume + * it so EPOLLERR clears instead of closing the socket below. */ + if (!recv_error_surfaced && !u->closed) { + int so_error = 0; + socklen_t so_error_len = sizeof(so_error); + if (getsockopt(us_poll_fd(p), SOL_SOCKET, SO_ERROR, &so_error, &so_error_len) == 0 && so_error) { + recv_error_surfaced = 1; + if (u->on_recv_error) { + u->on_recv_error(u, so_error); + } + } + } } #endif diff --git a/src/runtime/socket/udp_socket.rs b/src/runtime/socket/udp_socket.rs index 90fe1b32e510..77bab929470c 100644 --- a/src/runtime/socket/udp_socket.rs +++ b/src/runtime/socket/udp_socket.rs @@ -86,12 +86,13 @@ extern "C" fn on_close(socket: *mut uws::udp::Socket) { } extern "C" fn on_recv_error(socket: *mut uws::udp::Socket, errno: c_int) { - // Only called on Linux via IP_RECVERR — loop.c guards the recv-on-error - // path with #if defined(__linux__) to preserve the pre-existing - // close-on-error behavior on kqueue/Windows (where an error event is a - // fatal socket condition, not a drainable error queue). Builds a - // SystemError from the ICMP errno (ECONNREFUSED, EHOSTUNREACH, - // ENETUNREACH, EMSGSIZE, ...) and dispatches through the 'error' handler. + // Only called on Linux. loop.c surfaces the errno here when EPOLLERR fires: + // from the IP_RECVERR error queue (armed on connect in bsd_connect_udp_socket) + // or from SO_ERROR when that queue is already empty, e.g. right after a + // disconnect purged it. The recv-on-error path is guarded with + // #if defined(__linux__) to preserve the pre-existing close-on-error + // behavior on kqueue/Windows (where an error event is a fatal socket + // condition, not a drainable error queue). let this: &UDPSocket = UDPSocket::from_uws(socket); let sys_err = bun_sys::Error::from_code_int(errno, bun_sys::Tag::recv); let global_this = this.global_this.get(); diff --git a/test/js/bun/udp/dgram-unconnected-icmp-fixture.ts b/test/js/bun/udp/dgram-unconnected-icmp-fixture.ts new file mode 100644 index 000000000000..6b25ed886432 --- /dev/null +++ b/test/js/bun/udp/dgram-unconnected-icmp-fixture.ts @@ -0,0 +1,68 @@ +import dgram from "node:dgram"; + +function bind(socket: dgram.Socket): Promise { + const { promise, resolve, reject } = Promise.withResolvers(); + const onError = (err: Error) => reject(err); + socket.once("error", onError); + socket.bind(0, "127.0.0.1", () => { + // Leave no "error" listener behind: node never emits one for an unconnected + // socket, so real dgram code does not install one either. + socket.removeListener("error", onError); + resolve(socket.address().port); + }); + return promise; +} + +function close(socket: dgram.Socket): Promise { + const { promise, resolve } = Promise.withResolvers(); + socket.close(() => resolve()); + return promise; +} + +const live = dgram.createSocket("udp4"); +const livePort = await bind(live); + +const sender = dgram.createSocket("udp4"); +await bind(sender); + +// Bind and close a probe to get a port nothing is listening on. It is bound +// last so the kernel cannot hand the same port to `live` or `sender`. +const probe = dgram.createSocket("udp4"); +const deadPort = await bind(probe); +await close(probe); + +const sendErrors: string[] = []; +const received: string[] = []; + +for (let i = 0; i < 5; i++) { + const { promise: settled, resolve: done } = Promise.withResolvers(); + const onMessage = (msg: Buffer) => { + received.push(msg.toString()); + done(); + }; + live.once("message", onMessage); + + // The peer on deadPort is gone, so localhost answers with ICMP port + // unreachable while this sendto() is still in the kernel. It must not reach + // the socket's "error" event, and it must not land in the callback of the + // unrelated send that immediately follows. + sender.send("dead", deadPort, "127.0.0.1", err => { + if (err) sendErrors.push(`dead:${(err as NodeJS.ErrnoException).code}`); + }); + sender.send(`live-${i}`, livePort, "127.0.0.1", err => { + if (err) { + sendErrors.push(`live:${(err as NodeJS.ErrnoException).code}`); + // This datagram never left, so no message is coming: unblock the loop + // instead of hanging. + done(); + } + }); + + await settled; + live.removeListener("message", onMessage); +} + +console.log(JSON.stringify({ sendErrors, received })); + +await close(sender); +await close(live); diff --git a/test/js/bun/udp/dgram.test.ts b/test/js/bun/udp/dgram.test.ts index 106c2acb402a..9e62193c0ae3 100644 --- a/test/js/bun/udp/dgram.test.ts +++ b/test/js/bun/udp/dgram.test.ts @@ -1,7 +1,7 @@ import { describe, expect, jest, test } from "bun:test"; import { createSocket } from "dgram"; -import { disableAggressiveGCScope } from "harness"; +import { bunEnv, bunExe, disableAggressiveGCScope, isLinux } from "harness"; import path from "path"; import { nodeDataCases } from "./testdata"; @@ -304,3 +304,117 @@ describe("bind()", () => { expect(onError).not.toHaveBeenCalled(); }); }); + +// An unconnected socket has no single peer, so node never reports ICMP errors +// on one: sending to a peer that has gone away is a no-op, and nothing listens +// for 'error'. +describe("ICMP errors on an unconnected socket", () => { + test("a vanished peer does not emit 'error'", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const dgram = require("node:dgram"); + // No 'error' listener, which is what node's contract allows. + const socket = dgram.createSocket("udp4"); + socket.bind(0, "127.0.0.1", () => { + // Bind and close a probe to get a port nothing is listening on. It + // is bound after 'socket' so the kernel cannot hand it the same one. + const probe = dgram.createSocket("udp4"); + probe.bind(0, "127.0.0.1", () => { + const deadPort = probe.address().port; + probe.close(() => { + let sent = 0; + const tick = () => { + socket.send("ping", deadPort, "127.0.0.1"); + if (++sent === 20) { + console.log("sent " + sent); + socket.close(); + return; + } + setImmediate(tick); + }; + tick(); + }); + }); + }); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).not.toContain("ECONNREFUSED"); + expect(stdout.trim()).toBe("sent 20"); + expect(exitCode).toBe(0); + }); + + test("a vanished peer does not poison the next send()", async () => { + await using proc = Bun.spawn({ + cmd: [bunExe(), path.join(import.meta.dir, "dgram-unconnected-icmp-fixture.ts")], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).not.toContain("ECONNREFUSED"); + expect(JSON.parse(stdout)).toEqual({ + sendErrors: [], + received: ["live-0", "live-1", "live-2", "live-3", "live-4"], + }); + expect(exitCode).toBe(0); + }); + + // A connected socket records the ICMP error in the kernel's sk_err, which + // outlives the error queue that disconnecting purges. Stranding it there + // leaves the poll permanently in an error state. + test.skipIf(!isLinux)("disconnect() with a peer's ICMP error pending keeps the socket usable", async () => { + function bind(socket: ReturnType): Promise { + const { promise, resolve, reject } = Promise.withResolvers(); + const onError = (err: Error) => reject(err); + socket.once("error", onError); + socket.bind(0, "127.0.0.1", () => { + socket.removeListener("error", onError); + resolve(socket.address().port); + }); + return promise; + } + + await using live = createSocket("udp4"); + const livePort = await bind(live); + + await using socket = createSocket("udp4"); + await bind(socket); + + // Bound last so the kernel cannot hand this port to one of the sockets above. + const probe = createSocket("udp4"); + const deadPort = await bind(probe); + await new Promise(resolve => probe.close(() => resolve())); + + const { promise: connected, resolve: onConnect } = Promise.withResolvers(); + socket.connect(deadPort, "127.0.0.1", onConnect); + await connected; + + // The error belongs to the peer we were connected to, so node still reports + // it after disconnect(). What must not happen is the socket quietly closing. + const { promise: errored, resolve: onError } = Promise.withResolvers(); + socket.once("error", onError); + + // localhost answers this sendto() with ICMP port unreachable before it + // returns, so the error is already pending when disconnect() runs. + socket.send("dead"); + socket.disconnect(); + expect((await errored).code).toBe("ECONNREFUSED"); + + const { promise: received, resolve: onMessage } = Promise.withResolvers(); + live.once("message", msg => onMessage(msg.toString())); + const { promise: sent, resolve: onSent, reject: onSendError } = Promise.withResolvers(); + socket.send("alive", livePort, "127.0.0.1", err => (err ? onSendError(err) : onSent())); + await sent; + expect(await received).toBe("alive"); + }); +}); 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 e7f98e27455c..8fa3fcb3d7f0 100644 --- a/test/js/bun/udp/udp_socket_recv_flags.test.ts +++ b/test/js/bun/udp/udp_socket_recv_flags.test.ts @@ -1,6 +1,6 @@ // Coverage for the fifth parameter of Bun.udpSocket's `data` callback // (`ReceiveFlags.truncated` from MSG_TRUNC) and for Linux's IP_RECVERR -// surfacing ICMP errors as `error` events on the socket. +// surfacing ICMP errors as `error` events on a connected socket. import { udpSocket } from "bun"; import { describe, expect, test } from "bun:test"; @@ -36,24 +36,18 @@ 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. + // IP_RECVERR is Linux-specific. On BSDs and Windows, ICMP errors on a + // connected UDP socket are delivered through different channels that we + // don't currently surface. test.skipIf(!isLinux)( - "surfaces ECONNREFUSED from ICMP port unreachable (IP_RECVERR) and keeps the socket usable", + "surfaces ECONNREFUSED from ICMP port unreachable (IP_RECVERR) on a connected socket and keeps it 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()); - }, - }, - }); const sender = await udpSocket({ + // Nothing is listening on port 1 of localhost, so the kernel replies + // with ICMP port unreachable. + connect: { hostname: "127.0.0.1", port: 1 }, socket: { error(err: Error & { code?: string }) { resolveErr(err); @@ -61,12 +55,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,21 +68,60 @@ 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. + // An ICMP error is one-shot, not a fatal socket state. 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(); } }, ); + + // An unconnected socket has no single peer to attribute an ICMP error to, so + // Linux drops it unless IP_RECVERR is set and node relies on that. Surfacing + // it kills processes that (per node's contract) installed no error handler, + // and the pending error poisons the socket's next send(). + test("does not surface ICMP errors on an unconnected socket", async () => { + const { promise: errored, reject: rejectErr } = Promise.withResolvers(); + const { promise: delivered, resolve: resolveMsg } = Promise.withResolvers(); + + const receiver = await udpSocket({ + hostname: "127.0.0.1", + socket: { + data(_socket, data) { + resolveMsg(data.toString()); + }, + }, + }); + + const sender = await udpSocket({ + hostname: "127.0.0.1", + socket: { + error(err: Error & { code?: string }) { + rejectErr(err); + }, + }, + }); + + function sendRec() { + if (sender.closed || receiver.closed) return; + try { + sender.send("dead", 1, "127.0.0.1"); + // The ICMP reply for the datagram above is already queued by the time + // this runs: it must not be reported as a failure of this send. + sender.send("alive", receiver.port, "127.0.0.1"); + } catch (err) { + rejectErr(err as Error); + return; + } + setTimeout(sendRec, 10); + } + sendRec(); + + try { + expect(await Promise.race([delivered, errored])).toBe("alive"); + } finally { + sender.close(); + receiver.close(); + } + }); });