add a test for crash() reliability

src/twisted/internet/base.py

@@ -769,6 +769,8 @@ def crash(self) -> None:
         """
         self._started = False
         self.running = False
+        # TODO: test for calling crash() multiple times, this system event
+        # trigger should not be added multiple times
         self.addSystemEventTrigger("during", "startup", self._reallyStartRunning)
 
     def sigInt(self, number: int, frame: Optional[FrameType] = None) -> None:

src/twisted/internet/cfreactor.py

@@ -372,6 +372,14 @@ def _moveCallLaterSooner(self, tple):
         self._scheduleSimulate()
         return result
 
+    def startRunning(self, installSignalHandlers: bool = True) -> None:
+        """
+        Start running the reactor, then kick off the timer that advances
+        Twisted's clock to keep pace with CFRunLoop's.
+        """
+        super().startRunning(installSignalHandlers)
+        self._scheduleSimulate()
+
     _inCFLoop = False
 
     def mainLoop(self):
@@ -384,6 +392,7 @@ def mainLoop(self):
             self._runner()
         finally:
             self._inCFLoop = False
+            self._stopSimulating()
 
         assert not self.running, (
             self._stopped,
@@ -392,9 +401,18 @@ def mainLoop(self):
             self._startedBefore,
         )
 
-    _currentSimulator = None
+    _currentSimulator: object | None = None
+
+    def _stopSimulating(self) -> None:
+        """
+        If we have a CFRunLoopTimer registered with the CFRunLoop, invalidate
+        it and set it to None.
+        """
+        if self._currentSimulator is not None:
+            CFRunLoopTimerInvalidate(self._currentSimulator)
+            self._currentSimulator = None
 
-    def _scheduleSimulate(self, force=False):
+    def _scheduleSimulate(self, force: bool = False) -> None:
         """
         Schedule a call to C{self.runUntilCurrent}.  This will cancel the
         currently scheduled call if it is already scheduled.
@@ -408,9 +426,12 @@ def _scheduleSimulate(self, force=False):
 
         @type force: C{bool}
         """
-        if self._currentSimulator is not None:
-            CFRunLoopTimerInvalidate(self._currentSimulator)
-            self._currentSimulator = None
+        self._stopSimulating()
+        if not self.running:
+            # If the reactor is not running (e.g. we are scheduling callLater
+            # calls before starting the reactor) we should not be scheduling
+            # CFRunLoopTimers against the global CFRunLoop.
+            return
         timeout = self.timeout()
         if force:
             timeout = 0.0

src/twisted/internet/test/test_cfreactor.py

@@ -0,0 +1,46 @@
+from typing import TYPE_CHECKING
+
+from twisted.trial.unittest import SynchronousTestCase
+from .reactormixins import ReactorBuilder
+
+if TYPE_CHECKING:
+    fakeBase = SynchronousTestCase
+else:
+    fakeBase = object
+
+
+class CoreFoundationSpecificTests(ReactorBuilder, fakeBase):
+    """
+    Tests for platform interactions of the CoreFoundation-based reactor.
+    """
+
+    _reactors = ["twisted.internet.cfreactor.CFReactor"]
+
+    def test_whiteboxStopSimulating(self) -> None:
+        """
+        CFReactor's simulation timer is None after CFReactor crashes.
+        """
+        r = self.buildReactor()
+        r.callLater(0, r.crash)
+        r.callLater(100, lambda: None)
+        self.runReactor(r)
+        self.assertIs(r._currentSimulator, None)
+
+    def test_callLaterLeakage(self) -> None:
+        """
+        callLater should not leak global state into CoreFoundation which will
+        be invoked by a different reactor running the main loop.
+
+        @note: this test may actually be usable for other reactors as well, so
+            we may wish to promote it to ensure this invariant across other
+            foreign-main-loop reactors.
+        """
+        r = self.buildReactor()
+        delayed = r.callLater(0, lambda: None)
+        r2 = self.buildReactor()
+        r2.callLater(0, r2.callLater, 0, r2.stop)
+        self.runReactor(r2)
+        self.assertEqual(r.getDelayedCalls(), [delayed])
+
+
+globals().update(CoreFoundationSpecificTests.makeTestCaseClasses())