Ignore stale kqueue events for an already-deregistered key

newsfragments/3500.bugfix.rst

@@ -0,0 +1 @@
+On the kqueue backend (macOS/BSD), `~trio.lowlevel.notify_closing` could race with an already-fetched batch of kernel events under guest mode, causing `TrioInternalError: internal error in Trio - please file a bug!` wrapping a `KeyError` from inside `process_events`. Such stale events are now ignored instead of crashing the run.

src/trio/_core/_io_kqueue.py

@@ -90,7 +90,15 @@ def process_events(self, events: EventResult) -> None:
             if event.ident == self._force_wakeup_fd:
                 self._force_wakeup.drain()
                 continue
-            receiver = self._registered[key]
+            receiver = self._registered.get(key)
+            if receiver is None:
+                # In guest mode, get_events() can fetch a batch of kernel
+                # events ahead of when they actually reach process_events()
+                # on a later tick. notify_closing() (or a completed
+                # wait_kevent abort) can deregister this same key in that
+                # window, so by the time the stale event gets here there's
+                # nothing left to deliver it to.
+                continue
             if event.flags & select.KQ_EV_ONESHOT:  # TODO: test this branch
                 del self._registered[key]
             if isinstance(receiver, _core.Task):

src/trio/_core/_tests/test_io.py

@@ -422,6 +422,39 @@ def check(
             check(expected_monitors=0, expected_readers=1, expected_writers=0)
 
 
+@pytest.mark.skipif(
+    sys.platform == "win32" or sys.platform == "linux",
+    reason="only the kqueue backend has this failure mode",
+)
+async def test_kqueue_process_events_tolerates_stale_deregistered_key() -> None:
+    # In guest mode, GuestState.guest_tick can fetch a batch of ready kqueue
+    # events with get_events(0) ahead of when they're actually delivered to
+    # process_events() on a later tick. notify_closing() can run in that
+    # window and deregister the same key, since it's a plain synchronous
+    # function that isn't tied to a checkpoint the way task cancellation is.
+    # When the stale event then arrives, process_events() used to look it up
+    # with a bare `self._registered[key]` and blow up with a KeyError instead
+    # of just treating it as "nothing left to wake up here."
+    from trio._core._io_kqueue import KqueueIOManager
+
+    manager = KqueueIOManager()
+    try:
+        with stdlib_socket.socket() as s:
+            fileno = s.fileno()
+        # The socket is closed now, so this fileno is free, but we're not
+        # actually doing any real kqueue registration here, we just care
+        # about the dict lookup process_events() does against _registered.
+        key = (fileno, select.KQ_FILTER_WRITE)
+
+        manager._registered[key] = object()
+        del manager._registered[key]  # e.g. notify_closing() already ran
+
+        stale_event = select.kevent(fileno, select.KQ_FILTER_WRITE)
+        manager.process_events([stale_event])  # must not raise KeyError
+    finally:
+        manager.close()
+
+
 async def test_can_survive_unnotified_close() -> None:
     # An "unnotified" close is when the user closes an fd/socket/handle
     # directly, without calling notify_closing first. This should never happen