udp: do not let remote ICMP kill an unconnected socket

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 {

src/runtime/socket/udp_socket.rs

@@ -95,9 +95,19 @@
     // 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.
     let this: &UDPSocket = UDPSocket::from_uws(socket);
+    // Node never enables IP_RECVERR, so Linux drops ICMP for an unconnected
+    // UDP socket before it can reach sk_err (`__udp4_lib_err`: `if (!harderr
+    // || sk->sk_state != TCP_ESTABLISHED) goto out;`). Bun keeps IP_RECVERR
+    // on to drain the queue, but the only observable effect that matches
+    // Node is to discard those entries here — one layer below node:dgram's
+    // own filter so Bun.udpSocket inherits the same behavior and a reply
+    // server without an error handler is not killed by a vanished client.
+    // Connected sockets keep the error: Node surfaces that via recvmsg too.
+    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

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 <code>`.
 test.skipIf(isWindows)("connected send() failure reports Node's error shape", async () => {

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,100 @@
   });
 });
 
+// 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, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    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, turning a caught throw into a process kill.
+  test("oversized send() throws EMSGSIZE once and does not redeliver it asynchronously", async () => {
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `
+        const errs = [];
+        const s = await Bun.udpSocket({
+          port: 0,
+          hostname: "127.0.0.1",
+          socket: {
+            data() {},
+            error(err) { errs.push(err?.code ?? String(err)); },
+          },
+        });
+        let syncErr = null;
+        try {
+          s.send(new Uint8Array(65510), 1, "127.0.0.1");
+        } 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();
+        `,
+      ],
+      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(`{"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

test/js/bun/udp/udp_socket_recv_flags.test.ts

@@ -37,60 +37,43 @@
   });
 
   // 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.
   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 usable",
     async () => {
       const { promise: errPromise, resolve: resolveErr } = Promise.withResolvers<Error & { code?: string }>();
-      const { promise: msgPromise, resolve: resolveMsg } = Promise.withResolvers<string>();
 
-      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);
           },
         },
       });
 
-      // 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);
         }
       }
       sendDead();
 
       try {
         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();
       }
     },
   );
 });