Handle kqueue cancellation after event fetch

newsfragments/3500.bugfix.rst

@@ -0,0 +1 @@
+Fixed a race on kqueue backends where cancelling an I/O wait in guest mode after its event had been fetched could raise :exc:`TrioInternalError`.

src/trio/_core/_io_kqueue.py

@@ -38,6 +38,7 @@ class KqueueIOManager:
     _registered: dict[tuple[int, int], Task | UnboundedQueue[select.kevent]] = (
         attrs.Factory(dict)
     )
+    _cancelled_keys: set[tuple[int, int]] = attrs.Factory(set)
     _force_wakeup: WakeupSocketpair = attrs.Factory(WakeupSocketpair)
     _force_wakeup_fd: int | None = None
 
@@ -85,12 +86,20 @@ def get_events(self, timeout: float) -> EventResult:
         return events
 
     def process_events(self, events: EventResult) -> None:
+        cancelled_keys = self._cancelled_keys
+        self._cancelled_keys = set()
         for event in events:
             key = (event.ident, event.filter)
             if event.ident == self._force_wakeup_fd:
                 self._force_wakeup.drain()
                 continue
-            receiver = self._registered[key]
+            try:
+                receiver = self._registered[key]
+            except KeyError:
+                if key in cancelled_keys:
+                    # See test_cancel_io_after_events_fetched.
+                    continue
+                raise
             if event.flags & select.KQ_EV_ONESHOT:  # TODO: test this branch
                 del self._registered[key]
             if isinstance(receiver, _core.Task):
@@ -161,6 +170,7 @@ async def wait_kevent(
         def abort(raise_cancel: RaiseCancelT) -> Abort:
             r = abort_func(raise_cancel)
             if r is _core.Abort.SUCCEEDED:  # TODO: test this branch
+                self._cancelled_keys.add(key)
                 del self._registered[key]
             return r
 

src/trio/_core/_tests/test_guest_mode.py

@@ -774,3 +774,19 @@ def uh_oh() -> None:
     trivial_guest_run(trio_main)
     for sock in pair:
         sock.close()
+
+
+@pytest.mark.skipif(
+    sys.platform in {"win32", "linux"},
+    reason="requires the kqueue backend",
+)
+def test_cancel_io_after_events_fetched() -> None:
+    async def trio_main(in_host: InHost) -> None:
+        left, right = socket.socketpair()
+        with left, right, trio.CancelScope() as scope:
+            in_host(scope.cancel)
+            await trio.lowlevel.wait_writable(left)
+
+        assert scope.cancelled_caught
+
+    trivial_guest_run(trio_main)

src/trio/_core/_tests/test_io.py

@@ -422,6 +422,27 @@ def check(
             check(expected_monitors=0, expected_readers=1, expected_writers=0)
 
 
+@pytest.mark.skipif(
+    sys.platform in {"win32", "linux"},
+    reason="requires the kqueue backend",
+)
+def test_kqueue_process_events_rejects_unknown_key() -> None:
+    from .._io_kqueue import (  # type: ignore[attr-defined, unused-ignore]
+        KqueueIOManager,
+    )
+
+    manager = KqueueIOManager()
+    try:
+        event = select.kevent(  # type: ignore[attr-defined, unused-ignore]
+            1_000_000,
+            select.KQ_FILTER_READ,  # type: ignore[attr-defined, unused-ignore]
+        )
+        with pytest.raises(KeyError):
+            manager.process_events([event])
+    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