fix(kqueue): call update_now in kqueue timer_next (#214)

This fixes some behavior I observed using libxev on macOS. Setting a
timer fires before the expected time, apparently because `timer_next`
uses the existing cached time (`self.cached_now`). If that cached time
is outdated because of a long wait, the timer will fire almost
immediately rather than after the desired time. I've included a test
which fails on `main` that shows this behavior. This test delays loop
wakeup, arms a timer from the callback, and asserts elapsed time is at
least `timer_delay_ms`.

The timer API describes `next_ms` as “from now”, so scheduling should
use current loop time, not the potentially stale cached time. I looked
through the existing issues and PRs and didn't see this referenced, so
I'm not sure if this in fact expected behavior. If it is, am I
misunderstanding the behavior of the timer API on Kqueue?

This is solved in the `io_uring` backend with this line: `if
(self.flags.now_outdated) self.update_now();`
https://github.com/mitchellh/libxev/blob/78781eef639f9eb9935b067b4bbc1e69254bda3d/src/backend/io_uring.zig#L337
which updates the cached time if we're `now_outdated`. For simplicity, I
opted to not add this flag and unconditionally update the cached time on
Kqueue.

Author: mitchellh

src/backend/kqueue.zig

@@ -659,7 +659,7 @@ pub const Loop = struct {
         }
     }
 
-    fn timer_next(self: Loop, next_ms: u64) posix.timespec {
+    fn timer_next(self: *Loop, next_ms: u64) posix.timespec {
         // Get the timestamp of the absolute time that we'll execute this timer.
         // There are lots of failure scenarios here in math. If we see any
         // of them we just use the maximum value.
@@ -675,6 +675,8 @@ pub const Loop = struct {
             (next_ms % std.time.ms_per_s) * std.time.ns_per_ms,
         ) orelse return max;
 
+        self.update_now();
+
         return .{
             .sec = std.math.add(isize, self.cached_now.sec, next_s) catch
                 return max,
@@ -2674,6 +2676,116 @@ test "kqueue: mach port" {
     try testing.expect(!called);
 }
 
+test "kqueue: timer armed from delayed callback must not fire early" {
+    if (builtin.os.tag != .macos) return error.SkipZigTest;
+
+    const testing = std.testing;
+
+    const send_delay_ms: u64 = 50;
+    const timer_delay_ms: u64 = 20;
+
+    var loop = try Loop.init(.{});
+    defer loop.deinit();
+
+    // Allocate the port used to wake the loop after a delayed send.
+    const mach_self = posix.system.mach_task_self();
+    var mach_port: posix.system.mach_port_name_t = undefined;
+    try testing.expectEqual(
+        darwin.KernE.SUCCESS,
+        darwin.getKernError(posix.system.mach_port_allocate(
+            mach_self,
+            @intFromEnum(posix.system.MACH_PORT_RIGHT.RECEIVE),
+            &mach_port,
+        )),
+    );
+    defer _ = posix.system.mach_port_deallocate(mach_self, mach_port);
+
+    const State = struct {
+        timer_started_ns: i128 = 0,
+        timer_fired_ns: i128 = 0,
+        timer_trigger: ?TimerTrigger = null,
+        timer_completion: Completion = undefined,
+    };
+
+    var state: State = .{};
+
+    const timer_cb: Callback = (struct {
+        fn callback(
+            ud: ?*anyopaque,
+            _: *Loop,
+            _: *Completion,
+            r: Result,
+        ) CallbackAction {
+            const s: *State = @ptrCast(@alignCast(ud.?));
+            s.timer_fired_ns = std.time.nanoTimestamp();
+            s.timer_trigger = r.timer catch unreachable;
+            return .disarm;
+        }
+    }).callback;
+
+    var c_wait: Completion = .{
+        .op = .{
+            .machport = .{
+                .port = mach_port,
+                .buffer = .{ .array = undefined },
+            },
+        },
+
+        .userdata = &state,
+        .callback = (struct {
+            fn callback(
+                ud: ?*anyopaque,
+                l: *Loop,
+                _: *Completion,
+                r: Result,
+            ) CallbackAction {
+                _ = r.machport catch unreachable;
+                const s: *State = @ptrCast(@alignCast(ud.?));
+                s.timer_started_ns = std.time.nanoTimestamp();
+                l.timer(&s.timer_completion, timer_delay_ms, s, timer_cb);
+                return .disarm;
+            }
+        }).callback,
+    };
+    loop.add(&c_wait);
+
+    // Send to the mach port only after the loop has been blocked for a while.
+    const sender = try std.Thread.spawn(.{}, (struct {
+        fn run(port: posix.system.mach_port_name_t) void {
+            std.Thread.sleep(send_delay_ms * std.time.ns_per_ms);
+
+            var msg: darwin.mach_msg_header_t = .{
+                .msgh_bits = @intFromEnum(posix.system.MACH_MSG_TYPE.MAKE_SEND_ONCE),
+                .msgh_size = @sizeOf(darwin.mach_msg_header_t),
+                .msgh_remote_port = port,
+                .msgh_local_port = darwin.MACH_PORT_NULL,
+                .msgh_voucher_port = undefined,
+                .msgh_id = undefined,
+            };
+
+            const rc = darwin.mach_msg(
+                &msg,
+                darwin.MACH_SEND_MSG,
+                msg.msgh_size,
+                0,
+                darwin.MACH_PORT_NULL,
+                darwin.MACH_MSG_TIMEOUT_NONE,
+                darwin.MACH_PORT_NULL,
+            );
+            assert(darwin.getMachMsgError(rc) == darwin.MachMsgE.SUCCESS);
+        }
+    }).run, .{mach_port});
+    defer sender.join();
+
+    try loop.run(.until_done);
+
+    try testing.expect(state.timer_trigger.? == .expiration);
+
+    const elapsed_ns = state.timer_fired_ns - state.timer_started_ns;
+    const elapsed_ms: i128 = @divFloor(elapsed_ns, std.time.ns_per_ms);
+    try testing.expect(elapsed_ms >= @as(i128, @intCast(timer_delay_ms)));
+}
+
 test "kqueue: socket accept/cancel cancellation should decrease active count" {
     const mem = std.mem;
     const net = std.net;