From 9d34b49e00d108801c92376d41d51476333b0b99 Mon Sep 17 00:00:00 2001
From: gx14ac <shinta@gx14ac.com>
Date: Mon, 14 Sep 2026 00:21:17 -0700
Subject: [PATCH] fix(kqueue): let UDP streams poll for readability

Stream.poll built its kqueue operation from options.read and hit
unreachable for streams without a read mode. UDPSendto, the kqueue UDP,
enables poll but has no read mode, so xev.UDP.poll panicked on macOS
and the BSDs while it worked on io_uring and epoll.

Both read modes produced the same zero-length read, which waits for
readability without consuming data, so use it regardless of the mode.

Adds a test that polls a UDP socket, then reads the datagram that
woke it.

Fixes #242
---
 src/watcher/stream.zig | 20 ++++------
 src/watcher/udp.zig    | 87 ++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 94 insertions(+), 13 deletions(-)

diff --git a/src/watcher/stream.zig b/src/watcher/stream.zig
index c00e631e..298ea194 100644
--- a/src/watcher/stream.zig
+++ b/src/watcher/stream.zig
@@ -263,19 +263,13 @@ fn Pollable(comptime xev: type, comptime T: type, comptime options: Options) typ
                         },
                     } },
 
-                    .kqueue => switch (options.read) {
-                        .none => unreachable,
-
-                        .recv => .{ .read = .{
-                            .fd = self.fd,
-                            .buffer = .{ .slice = &.{} },
-                        } },
-
-                        .read => .{ .read = .{
-                            .fd = self.fd,
-                            .buffer = .{ .slice = &.{} },
-                        } },
-                    },
+                    // A zero-length read waits for readability without
+                    // consuming anything, whatever read mode the stream has;
+                    // UDP streams have none.
+                    .kqueue => .{ .read = .{
+                        .fd = self.fd,
+                        .buffer = .{ .slice = &.{} },
+                    } },
 
                     .iocp,
                     .wasi_poll,
diff --git a/src/watcher/udp.zig b/src/watcher/udp.zig
index a480e9ed..bf8dc9e5 100644
--- a/src/watcher/udp.zig
+++ b/src/watcher/udp.zig
@@ -924,6 +924,93 @@ fn UDPTests(comptime xev: type, comptime Impl: type) type {
             }
         }
 
+        test "UDP: poll waits for a datagram without consuming it" {
+            if (builtin.os.tag == .freebsd) return error.SkipZigTest;
+            if (!@hasDecl(Impl, "poll")) return error.SkipZigTest;
+            const testing = std.testing;
+
+            var tpool = ThreadPool.init(.{});
+            defer tpool.deinit();
+            defer tpool.shutdown();
+            var loop = try xev.Loop.init(.{ .thread_pool = &tpool });
+            defer loop.deinit();
+
+            const address = try std.Io.net.IpAddress.parse("127.0.0.1", 3134);
+            const server = try Impl.init(address);
+            const client = try Impl.init(address);
+            try server.bind(address);
+
+            var polled = false;
+            var c_poll: xev.Completion = undefined;
+            server.poll(&loop, &c_poll, .read, bool, &polled, (struct {
+                fn callback(
+                    ud: ?*bool,
+                    _: *xev.Loop,
+                    _: *xev.Completion,
+                    _: Impl,
+                    r: xev.PollError!xev.PollEvent,
+                ) xev.CallbackAction {
+                    _ = r catch unreachable;
+                    ud.?.* = true;
+                    return .disarm;
+                }
+            }).callback);
+
+            var send_buf = [_]u8{ 1, 2, 3 };
+            var c_write: xev.Completion = undefined;
+            var s_write: Impl.State = undefined;
+            client.write(&loop, &c_write, &s_write, address, .{ .slice = &send_buf }, void, null, (struct {
+                fn callback(
+                    _: ?*void,
+                    _: *xev.Loop,
+                    _: *xev.Completion,
+                    _: *Impl.State,
+                    _: Impl,
+                    _: xev.WriteBuffer,
+                    r: xev.WriteError!usize,
+                ) xev.CallbackAction {
+                    _ = r catch unreachable;
+                    return .disarm;
+                }
+            }).callback);
+
+            try loop.run(.until_done);
+            try testing.expect(polled);
+
+            // The datagram is still there to read.
+            var c_read: xev.Completion = undefined;
+            var s_read: Impl.State = undefined;
+            var recv_buf: [16]u8 = undefined;
+            var recv_len: usize = 0;
+            server.read(&loop, &c_read, &s_read, .{ .slice = &recv_buf }, usize, &recv_len, (struct {
+                fn callback(
+                    ud: ?*usize,
+                    _: *xev.Loop,
+                    _: *xev.Completion,
+                    _: *Impl.State,
+                    _: std.Io.net.IpAddress,
+                    _: Impl,
+                    _: xev.ReadBuffer,
+                    r: xev.ReadError!usize,
+                ) xev.CallbackAction {
+                    ud.?.* = r catch unreachable;
+                    return .disarm;
+                }
+            }).callback);
+            try loop.run(.until_done);
+            try testing.expectEqualSlices(u8, &send_buf, recv_buf[0..recv_len]);
+
+            const onClose = (struct {
+                fn callback(_: ?*void, _: *xev.Loop, _: *xev.Completion, _: Impl, r: xev.CloseError!void) xev.CallbackAction {
+                    _ = r catch unreachable;
+                    return .disarm;
+                }
+            }).callback;
+            server.close(&loop, &c_read, void, null, onClose);
+            client.close(&loop, &c_write, void, null, onClose);
+            try loop.run(.until_done);
+        }
+
         test "UDP: read/write" {
             if (builtin.os.tag == .freebsd) return error.SkipZigTest;
             const testing = std.testing;