fix trial -j / ctrl-c interaction

src/twisted/newsfragments/10340.bugfix

@@ -0,0 +1 @@
+`trial -j` no longer causes a traceback on exit when stopped with control-C.

src/twisted/newsfragments/11707.misc

@@ -0,0 +1 @@
+

src/twisted/trial/_dist/disttrial.py

@@ -441,15 +441,35 @@ async def runAndReport(n: int) -> DistReporter:
             await startedPool.join()
 
     def _run(self, test: Union[TestCase, TestSuite], untilFailure: bool) -> IReporter:
-        result: Union[Failure, DistReporter]
+        result: Union[Failure, DistReporter, None] = None
+        reactorStopping: bool = False
+        testsInProgress: Deferred[object]
 
-        def capture(r):
+        def capture(r: Union[Failure, DistReporter]) -> None:
             nonlocal result
             result = r
 
-        d = Deferred.fromCoroutine(self.runAsync(test, untilFailure))
-        d.addBoth(capture)
-        d.addBoth(lambda ignored: self._reactor.stop())
+        def maybeStopTests() -> None | Deferred[object]:
+            nonlocal reactorStopping
+            reactorStopping = True
+            if result is None:
+                testsInProgress.cancel()
+                return testsInProgress
+            return None
+
+        def maybeStopReactor(result: object) -> object:
+            if not reactorStopping:
+                self._reactor.stop()
+            return result
+
+        self._reactor.addSystemEventTrigger("before", "shutdown", maybeStopTests)
+
+        testsInProgress = (
+            Deferred.fromCoroutine(self.runAsync(test, untilFailure))
+            .addBoth(capture)
+            .addBoth(maybeStopReactor)
+        )
+
         self._reactor.run()
 
         if isinstance(result, Failure):
@@ -458,7 +478,7 @@ def capture(r):
         # mypy can't see that raiseException raises an exception so we can
         # only get here if result is not a Failure, so tell mypy result is
         # certainly a DistReporter at this point.
-        assert isinstance(result, DistReporter)
+        assert isinstance(result, DistReporter), f"{result} is not DistReporter"
 
         # Unwrap the DistReporter to give the caller some regular IReporter
         # object.  DistReporter isn't type annotated correctly so fix it here.

src/twisted/trial/_dist/test/test_disttrial.py

@@ -31,7 +31,7 @@
 
 from twisted.internet import interfaces
 from twisted.internet.base import ReactorBase
-from twisted.internet.defer import Deferred, succeed
+from twisted.internet.defer import CancelledError, Deferred, succeed
 from twisted.internet.error import ProcessDone
 from twisted.internet.protocol import ProcessProtocol, Protocol
 from twisted.internet.test.modulehelpers import AlternateReactor
@@ -125,6 +125,12 @@ def stop(self):
         See L{IReactorCore.stop}.
         """
         MemoryReactorClock.stop(self)
+        # TODO: implementing this more comprehensively in MemoryReactor would
+        # be nice, this is rather hard-coded to disttrial's current
+        # implementation.
+        if "before" in self.triggers:
+            if "shutdown" in self.triggers["before"]:
+                self.triggers["before"]["shutdown"][0][0]()
         self.stopCount += 1
 
     def run(self):
@@ -139,6 +145,9 @@ def run(self):
 
         for f, args, kwargs in self.whenRunningHooks:
             f(*args, **kwargs)
+        self.stop()
+        # do not count internal 'stop' against trial-initiated .stop() count
+        self.stopCount -= 1
 
 
 class CountingReactorTests(SynchronousTestCase):
@@ -461,6 +470,15 @@ def test_runUnexpectedError(self) -> None:
         assert_that(errors, has_length(1))
         assert_that(errors[0][1].type, equal_to(WorkerPoolBroken))
 
+    def test_runUnexpectedErrorCtrlC(self) -> None:
+        """
+        If the reactor is stopped by C-c (i.e. `run` returns before the test
+        case's Deferred has been fired) we should cancel the pending test run.
+        """
+        runner = self.getRunner(workerPoolFactory=LocalWorkerPool)
+        with self.assertRaises(CancelledError):
+            runner.run(self.suite)
+
     def test_runUnexpectedWorkerError(self) -> None:
         """
         If for some reason the worker process cannot run a test, the error is