io(kqueue): unregister before handing the owner off on close, and bit-test EV_ERROR

src/io/lib.rs

@@ -1117,17 +1117,15 @@ impl IoRequestLoop {
                     match (request.callback)(request) {
                         Action::Readable(readable) => {
                             Poll::apply_kqueue(
-                                ApplyAction::Readable,
-                                readable.tag,
+                                ApplyAction::Readable(readable.tag),
                                 readable.poll,
                                 readable.fd,
                                 add_one(&mut events_list),
                             );
                         }
                         Action::Writable(writable) => {
                             Poll::apply_kqueue(
-                                ApplyAction::Writable,
-                                writable.tag,
+                                ApplyAction::Writable(writable.tag),
                                 writable.poll,
                                 writable.fd,
                                 add_one(&mut events_list),
@@ -1139,12 +1137,15 @@ impl IoRequestLoop {
                             {
                                 Poll::apply_kqueue(
                                     ApplyAction::Cancel,
-                                    close.tag,
                                     close.poll,
                                     close.fd,
                                     add_one(&mut events_list),
                                 );
                             }
+                            // From here on another thread may finish and free
+                            // the owner, before the kevent() below has even
+                            // submitted the cancel; safe only because the cancel
+                            // addresses nobody (see `ApplyAction::Cancel`).
                             (close.on_done)(close.ctx);
                         }
                     }
@@ -1158,10 +1159,10 @@ impl IoRequestLoop {
                 self.pollfd().native(),
                 events_list.as_ptr(),
                 c_int::try_from(change_count).expect("int cast"),
-                // The same array may be used for the changelist and eventlist.
+                // The same array may be used for the changelist and eventlist;
+                // a rejected change comes back through it as an EV_ERROR entry
+                // (see `on_update_kqueue`) instead of failing the call.
                 events_list.as_mut_ptr(),
-                // we set 0 here so that if we get an error on
-                // registration, it becomes errno
                 c_int::try_from(capacity).expect("int cast"),
                 core::ptr::null(),
             );
@@ -1353,11 +1354,12 @@ pub struct FileAction<'a> {
     pub on_error: fn(*mut (), &sys::Error),
 }
 
+/// No [`PollableTag`], unlike [`FileAction`]: `on_done` hands the owner to
+/// another thread, so nothing a close submits may dispatch back into it.
 pub struct CloseAction<'a> {
     pub fd: Fd,
     pub poll: &'a mut Poll,
     pub ctx: *mut (),
-    pub tag: PollableTag,
     pub on_done: fn(*mut ()),
 }
 
@@ -1496,8 +1498,15 @@ pub type FlagsSet = enumset::EnumSet<Flags>;
 #[cfg(any(target_os = "macos", target_os = "freebsd"))]
 #[derive(PartialEq, Eq, Clone, Copy)]
 enum ApplyAction {
-    Readable,
-    Writable,
+    /// One-shot `EV_ADD`; its event, or the `EV_ERROR` entry if the add is
+    /// rejected, dispatches to the owner the tag names.
+    Readable(PollableTag),
+    Writable(PollableTag),
+    /// `EV_DELETE` of whatever the poll still has armed, submitted with
+    /// `udata = 0` so that `on_update_kqueue` drops its reply like the waker's.
+    /// The reply is the rule, not the exception (the one-shot knote is usually
+    /// gone already: ENOENT; or the owner closed the fd first: EBADF), and by
+    /// the time it arrives `tick_kqueue` has handed the owner to another thread.
     Cancel,
 }
 
@@ -1506,31 +1515,38 @@ impl Poll {
     #[inline]
     pub(crate) fn apply_kqueue(
         action: ApplyAction,
-        tag: PollableTag,
         poll: &mut Poll,
         fd: Fd,
         kqueue_event: &mut KEvent,
     ) {
         log!(
             "register({}, {})",
             match action {
-                ApplyAction::Readable => "readable",
-                ApplyAction::Writable => "writable",
+                ApplyAction::Readable(_) => "readable",
+                ApplyAction::Writable(_) => "writable",
                 ApplyAction::Cancel => "cancel",
             },
             fd
         );
 
         let one_shot_flag = libc::EV_ONESHOT;
-        let udata: usize = Pollable::init(tag, std::ptr::from_mut::<Poll>(poll)).ptr() as usize;
-        let (filter, flags_): (i16, u16) = match action {
-            ApplyAction::Readable => (libc::EVFILT_READ, libc::EV_ADD | one_shot_flag),
-            ApplyAction::Writable => (libc::EVFILT_WRITE, libc::EV_ADD | one_shot_flag),
+        let poll_ptr = std::ptr::from_mut::<Poll>(poll);
+        let (filter, flags_, udata): (i16, u16, usize) = match action {
+            ApplyAction::Readable(tag) => (
+                libc::EVFILT_READ,
+                libc::EV_ADD | one_shot_flag,
+                Pollable::init(tag, poll_ptr).ptr() as usize,
+            ),
+            ApplyAction::Writable(tag) => (
+                libc::EVFILT_WRITE,
+                libc::EV_ADD | one_shot_flag,
+                Pollable::init(tag, poll_ptr).ptr() as usize,
+            ),
             ApplyAction::Cancel => {
                 if poll.flags.contains(Flags::PollReadable) {
-                    (libc::EVFILT_READ, libc::EV_DELETE)
+                    (libc::EVFILT_READ, libc::EV_DELETE, 0)
                 } else if poll.flags.contains(Flags::PollWritable) {
-                    (libc::EVFILT_WRITE, libc::EV_DELETE)
+                    (libc::EVFILT_WRITE, libc::EV_DELETE, 0)
                 } else {
                     unreachable!()
                 }
@@ -1561,10 +1577,10 @@ impl Poll {
         }
 
         match action {
-            ApplyAction::Readable => {
+            ApplyAction::Readable(_) => {
                 poll.flags.insert(Flags::PollReadable);
             }
-            ApplyAction::Writable => {
+            ApplyAction::Writable(_) => {
                 poll.flags.insert(Flags::PollWritable);
             }
             ApplyAction::Cancel => {
@@ -1611,16 +1627,21 @@ impl Poll {
 
         let pollable = Pollable::from(event.udata as u64);
         let tag = pollable.tag();
-        // The waker is registered with udata=0 → tag=.empty. The wakeup exists
-        // only to unblock kevent() so the pending queue drains.
+        // udata=0 → tag=.empty is both the waker (registered only to unblock
+        // kevent() so the pending queue drains) and the EV_ERROR reply to an
+        // `ApplyAction::Cancel`, whose owner may already be gone.
         if tag == PollableTag::Empty {
             return;
         }
         let poll = pollable.poll();
         // CYCLEBREAK: owner (ReadFile/WriteFile) is T6; dispatch via link-time
         // `extern "Rust"` defined in `bun_runtime::dispatch`. The
         // container_of(io_poll) recovery happens there.
-        if event.flags == libc::EV_ERROR {
+        //
+        // A rejected change comes back with EV_ERROR set in `flags`; xnu ORs it
+        // into the bits we submitted (EV_ADD|EV_ONESHOT|EV_ERROR) while FreeBSD
+        // replaces them, so test the bit rather than the whole word.
+        if (event.flags & libc::EV_ERROR) != 0 {
             log!("error({}) = {}", event.ident, event.data);
             // SAFETY: poll is the `io_poll` field of a live owner; link-time
             // extern body matches on `tag`.

src/runtime/webcore/Blob.rs

@@ -7031,8 +7031,6 @@ pub trait FileOpener: Sized {
 
 // TODO: move to bun_sys?
 pub trait FileCloser: Sized {
-    const IO_TAG: bun_io::Tag;
-
     fn opened_fd(&self) -> Fd;
     fn set_opened_fd(&mut self, fd: Fd);
     fn close_after_io(&self) -> bool;
@@ -7109,15 +7107,13 @@ pub trait FileCloser: Sized {
 }
 
 /// Implements [`FileCloser`] for a task struct with the standard field set
-/// (`opened_fd`, `close_after_io`, `state`, `io_request`, `io_poll`, `task`),
-/// an inherent `update()`, and a [`bun_io::Tag`] variant named after the type.
-/// The type must also carry `bun_threading::intrusive_work_task!` and
-/// `bun_io::intrusive_io_request!`, which provide the parent-pointer recovery
-/// used by the two trampolines.
+/// (`opened_fd`, `close_after_io`, `state`, `io_request`, `io_poll`, `task`)
+/// and an inherent `update()`. The type must also carry
+/// `bun_threading::intrusive_work_task!` and `bun_io::intrusive_io_request!`,
+/// which provide the parent-pointer recovery used by the two trampolines.
 macro_rules! impl_file_closer {
     ($T:ident) => {
         impl crate::webcore::blob::FileCloser for $T {
-            const IO_TAG: ::bun_io::Tag = ::bun_io::Tag::$T;
             fn opened_fd(&self) -> ::bun_sys::Fd {
                 self.opened_fd
             }
@@ -7164,7 +7160,6 @@ macro_rules! impl_file_closer {
                     fd,
                     poll,
                     ctx: this.cast::<()>(),
-                    tag: <Self as crate::webcore::blob::FileCloser>::IO_TAG,
                     on_done,
                 })
             }

src/runtime/webcore/blob/read_file.rs

@@ -960,7 +960,6 @@ impl<'a> FileOpener for ReadFileUV<'a> {
 
 #[cfg(windows)]
 impl<'a> FileCloser for ReadFileUV<'a> {
-    const IO_TAG: bun_io::Tag = bun_io::Tag::ReadFile;
     fn opened_fd(&self) -> Fd {
         self.opened_fd
     }

test/internal/source-lints/kevent-ev-error-equality.test.ts

@@ -0,0 +1,116 @@
+import { file } from "bun";
+import { expect, test } from "bun:test";
+import { realpathSync } from "fs";
+import path from "path";
+import { globAllSources } from "../../../scripts/glob-sources.ts";
+
+// A kevent changelist entry the kernel rejects is handed back in the eventlist
+// with EV_ERROR set in `flags` and the errno in `data`. How it is "set" differs
+// between the two kqueue kernels we build for:
+//
+//   xnu      (bsd/kern/kern_event.c, kevent_register)   kev->flags |= EV_ERROR;
+//   FreeBSD  (sys/kern/kern_event.c, kqueue_kevent)      kevp->flags = EV_ERROR;
+//
+// so on macOS the reply to a failed `EV_ADD|EV_ONESHOT` reads 0x4011, not
+// 0x4000, and `flags == EV_ERROR` silently classifies it as a ready event.
+// For the io request loop that meant a rejected registration was dispatched
+// as "readable", which re-reads, gets EAGAIN, re-registers, and spins; for
+// FilePoll (posix_event_loop.rs) it meant change errors were swallowed. Both
+// kernels agree on the bit, so that is what gets tested:
+//
+//   flags == EV_ERROR / flags != EV::ERROR   →   (flags & EV_ERROR) != 0
+//
+// The Zig original carried the equality in both kqueue users; the port fixed
+// FilePoll's and this lint keeps the other from coming back.
+
+const root = path.resolve(import.meta.dir, "..", "..", "..");
+const rustSources = globAllSources().rust.filter(abs => abs.endsWith(".rs"));
+
+// Only scan files tracked in HEAD (a `git stash` round-trip can leave stray
+// `.rs` files in the working tree; CI runs on a clean checkout).
+const tracked: Set<string> | null = (() => {
+  const r = Bun.spawnSync({
+    cmd: ["git", "-C", root, "ls-tree", "-r", "--name-only", "-z", "HEAD"],
+    stdout: "pipe",
+    stderr: "ignore",
+  });
+  if (!r.success) return null;
+  return new Set(r.stdout.toString().split("\0").filter(Boolean));
+})();
+
+// The constant in any of its spellings: `EV_ERROR`, `libc::EV_ERROR`,
+// `bun_sys::darwin::EV::ERROR`, ...
+const CONSTANT = String.raw`(?:\w+::)*EV(?:_ERROR|::ERROR)\b`;
+// `x & EV_ERROR` is the bit test this lint asks for. It is erased before
+// looking for comparisons so that `(flags & EV_ERROR) == 0`, and Rust's
+// `flags & EV_ERROR == 0` (`&` binds tighter than `==` in Rust), do not
+// register as comparing the whole word. `(?<!&)&(?!&)` keeps `&&` out of it.
+const BIT_TEST = new RegExp(String.raw`(?<!&)&(?!&)\s*${CONSTANT}`, "g");
+// The constant as either operand of `==` / `!=`, optionally parenthesized.
+const COMPARED = new RegExp(String.raw`(?:==|!=)\s*\(?\s*${CONSTANT}|\b${CONSTANT}\s*\)?\s*(?:==|!=)`);
+
+function comparesWholeWord(line: string): boolean {
+  return COMPARED.test(line.replace(BIT_TEST, ""));
+}
+
+const offenders: string[] = [];
+let scanned = 0;
+let bitTests = 0;
+for (const abs of rustSources) {
+  const source = path.relative(root, abs).replaceAll(path.sep, "/");
+  // `src/cli` is a symlink into `src/runtime/cli`; count each file once under
+  // its canonical path.
+  if (path.relative(root, realpathSync(abs)).replaceAll(path.sep, "/") !== source) continue;
+  if (tracked !== null && !tracked.has(source)) continue;
+  scanned++;
+  const content = await file(abs).text();
+  if (!content.includes("EV_ERROR") && !content.includes("EV::ERROR")) continue;
+  // Drop comments (the ones in posix_event_loop.rs and io/lib.rs describe this
+  // very hazard) without disturbing line numbers: block comments keep their
+  // newlines, line comments are cut at the `//`.
+  const stripped = content.replace(/\/\*[\s\S]*?\*\//g, m => m.replace(/[^\n]/g, "")).replace(/\/\/.*$/gm, "");
+  for (const [index, line] of stripped.split("\n").entries()) {
+    BIT_TEST.lastIndex = 0;
+    if (BIT_TEST.test(line)) bitTests++;
+    if (comparesWholeWord(line)) offenders.push(`${source}:${index + 1}: ${line.trim()}`);
+  }
+}
+
+test("scans a non-empty set of tracked Rust sources", () => {
+  // Guards against the tracked/realpath filters above over-firing and leaving
+  // nothing to scan, which would make the ban below pass vacuously.
+  expect(scanned).toBeGreaterThan(0);
+});
+
+test("the scan still sees the tree's EV_ERROR bit tests", () => {
+  // FilePoll's register/unregister (src/io/posix_event_loop.rs) and the
+  // process reaper (src/spawn/process.rs) test the bit; if none are seen, the
+  // file set or the comment stripping is broken and the ban below is vacuous.
+  expect(bitTests).toBeGreaterThan(0);
+});
+
+test("the pattern recognizes the spellings it claims to", () => {
+  const banned = [
+    "if event.flags == libc::EV_ERROR {",
+    "if changelist[0].flags == EV::ERROR {",
+    "if ev.flags != bun_sys::darwin::EV::ERROR {",
+    "if libc::EV_ERROR == event.flags {",
+    "let failed = event.flags == (EV::ERROR);",
+    "if event.flags == EV_ERROR && event.data != 0 {",
+  ];
+  const allowed = [
+    "if (event.flags & libc::EV_ERROR) != 0 {",
+    "if (changelist[0].flags & EV::ERROR) != 0 && changelist[0].data != 0 {",
+    "if (changelist[i].flags & EV::ERROR) == 0 || changelist[i].data == 0 {",
+    "if r.flags & libc::EV_ERROR == 0 || r.data == 0 {",
+    "if event.data != 0 && event.flags & EV_ERROR != 0 {",
+    "pub const ERROR: u16 = libc::EV_ERROR;",
+    "let is_error = event.flags & EV::ERROR != 0;",
+  ];
+  expect(banned.filter(s => !comparesWholeWord(s))).toEqual([]);
+  expect(allowed.filter(comparesWholeWord)).toEqual([]);
+});
+
+test("kevent EV_ERROR is tested as a bit, never compared against the whole flags word", () => {
+  expect(offenders).toEqual([]);
+});