Sometimes notify_closing can be called on dead events

newsfragments/3502.bugfix.rst

@@ -0,0 +1 @@
+Ensure `trio.lowlevel.notify_closing` works even on events that have been deleted, e.g. because they were registered as one shot events.

src/trio/_core/_io_kqueue.py

@@ -280,7 +280,13 @@ def notify_closing(self, fd: int | _HasFileNo) -> None:
 
             if type(receiver) is _core.Task:
                 event = select.kevent(fd, filter_, select.KQ_EV_DELETE)
-                self._kqueue.control([event], 0)
+                try:
+                    self._kqueue.control([event], 0)
+                except OSError as e:
+                    if e.errno == errno.ENOENT:  # pragma: no branch
+                        # the event isn't in kqueue
+                        continue
+                    raise  # pragma: no cover
                 exc = _core.ClosedResourceError("another task closed this fd")
                 _core.reschedule(receiver, outcome.Error(exc))
                 del self._registered[key]

src/trio/_core/_tests/test_guest_mode.py

@@ -753,3 +753,24 @@ async def trio_main() -> None:
     aiotrio_run(trio_main, host_uses_signal_set_wakeup_fd=True)
 
     assert record == {("asyncio", "asyncio", True), ("trio", "trio", True)}
+
+
+def test_notify_closing_after_events() -> None:
+    # inspired by wrong repro in https://github.com/python-trio/trio/pull/3502
+    # either the program should silently pass or wait_writable should fail.
+    pair = socket.socketpair()
+    for sock in pair:
+        sock.setblocking(False)
+
+    async def trio_main(in_host: InHost) -> None:
+        in_host(uh_oh)
+        with contextlib.suppress(trio.ClosedResourceError):
+            await trio.lowlevel.wait_writable(pair[0])  # blocks
+
+    def uh_oh() -> None:
+        # this will run after trio gets events but before they are processed
+        trio.lowlevel.notify_closing(pair[0])
+
+    trivial_guest_run(trio_main)
+    for sock in pair:
+        sock.close()