Slice C: discovery deadlock, shared join deadline, shutdown hardening — v0.10.1

CLAUDE.md

@@ -59,7 +59,7 @@ pip install -r requirements-dev.txt
 - `monitor_thread.py` — `MonitorThread`: shared daemon-loop base for the stall watchdog and resource guard.
 - `resource_guard.py` — `ResourceGuard`: opt-in low-resource (disk / commit space) automatic soft stop.
 - `pytest_process.py` — `PytestProcess`: spawns one `pytest` subprocess per test module, attaches a `ProcessMonitor` (a daemon child, so it can never block the test process's exit). Coverage finalization and live-output reads are guarded so the final result record is still written when they fail.
-- `test_list.py` — `GetTests` process: discovers tests via `pytest --collect-only`.
+- `test_list.py` — `GetTests` process: discovers tests via `pytest --collect-only`. Callers wait via `collect(abort_event)`, which drains the result queue while the child runs — a join-before-drain deadlocks permanently past ~500 modules (full queue pipe blocks the child's exit).
 - `process_monitor.py` — `ProcessMonitor` subprocess: samples CPU/memory of the test process tree; `SubtreeCpuSampler` (shared persistent-handle CPU sampling).
 - `system_monitor.py` — `SystemMonitor` subprocess: system-wide CPU/memory/commit/disk/network sampling for the Run tab charts.
 - `commit_memory.py` — Windows commit-charge readers and psutil subtree helpers.

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "pytest-fly"
 description = "pytest runner and observer"
-version = "0.10.0"
+version = "0.10.1"
 readme = "README.md"
 requires-python = ">=3.12"
 authors = [

src/pytest_fly/gui/about_tab/about.py

@@ -10,7 +10,7 @@
 from PySide6.QtWidgets import QSizePolicy, QVBoxLayout, QWidget
 
 from ...__version__ import application_name
-from ...logger import get_log_directory
+from ...logger import get_log_directory, get_logger
 from ...platform.platform_info import get_performance_core_count, get_platform_info
 from ...preferences import get_active_put_path, get_preferences_db_path
 from ...project_info import get_project_info
@@ -24,6 +24,8 @@
 # Display order for pytest-fly's own metadata fields.
 _PYTEST_FLY_FIELD_ORDER = ("name", "version", "description", "author", "license", "home_url", "repository_url")
 
+log = get_logger()
+
 
 class AboutDataWorker(QObject):
     """
@@ -121,3 +123,16 @@ def update_about_box(self, text):
         self.about_box.set_text(text)
         self._thread.quit()
         self._thread.wait()
+
+    def shutdown(self, timeout_ms: int = 10_000) -> None:
+        """Stop the background data thread before the widget is destroyed.
+
+        Destroying a QThread that is still running is a Qt *fatal* error — a fail-fast
+        abort (exit 0xC0000409 on Windows) with no Python traceback — and this thread
+        runs git, which can take seconds. Called from the main window's closeEvent;
+        closing the app before the About data arrived previously crashed it at exit.
+        """
+        if self._thread.isRunning():
+            self._thread.quit()
+            if not self._thread.wait(timeout_ms):
+                log.warning(f"About data thread did not finish within {timeout_ms / 1000.0:.0f} s at shutdown")

src/pytest_fly/gui/gui_main.py

@@ -57,6 +57,11 @@ def __init__(self, data_dir: Path):
         # automation (screenshot capture, auto-quit-on-done) where a modal prompt would block.
         self._suppress_close_confirmation = False
 
+        # Re-entrancy latch for closeEvent: its processEvents() calls can deliver a second
+        # close request (double-click on the window's X), which would run the whole teardown
+        # — dialogs, runner stop, monitor join — a second time, nested inside the first.
+        self._closing = False
+
         super().__init__()
 
         # set monospace font
@@ -152,6 +157,12 @@ def closeEvent(self, event, /):
 
         log.info(f"{self.__class__.__name__}.closeEvent() - entering")
 
+        if self._closing:
+            # Nested close (delivered by a processEvents() below) — the outer invocation
+            # owns the teardown; just accept.
+            event.accept()
+            return
+
         # If a run is in progress (or being prepared), confirm with the user before tearing it
         # down. Skipped under automation, where a modal prompt would block the programmatic close.
         control = self.run_tab.control_window
@@ -170,6 +181,12 @@ def closeEvent(self, event, /):
                 event.ignore()
                 return
 
+        self._closing = True
+        # Stop the refresh timer first: the processEvents() calls below would otherwise run
+        # full _update_tick passes (and reconcile_process_count on a stopping runner) nested
+        # inside this teardown.
+        self.timer.stop()
+
         pref = get_pref()
 
         # Save window geometry via Qt's own serialization (frame, size, and maximized state), so
@@ -192,6 +209,9 @@ def closeEvent(self, event, /):
         self._system_monitor.request_stop()
         self._system_monitor.join(5.0)
 
+        # Wind down the About tab's QThread — destroying it mid-run is a Qt fatal abort.
+        self.about.shutdown()
+
         event.accept()
 
     def _force_stop_single_test(self, test_name: str):

src/pytest_fly/gui/run_tab/control_window.py

@@ -18,6 +18,7 @@
 from pathlib import Path
 from threading import Event, Thread
 
+import shiboken6
 from PySide6.QtCore import Qt, Signal
 from PySide6.QtWidgets import QGroupBox, QSizePolicy, QVBoxLayout
 from typeguard import typechecked
@@ -347,7 +348,21 @@ def _prepare_run(self, config: _RunPrepConfig, prior_runner: PytestRunner | None
         except (OSError, RuntimeError, ValueError, sqlite3.OperationalError) as e:
             log.error(f"run preparation failed: {e}", exc_info=True)
         finally:
-            self.run_prep_finished.emit(result)
+            # The window can be destroyed while this daemon thread is still preparing
+            # (closeEvent's bounded abort wait gave up). Emitting on a destroyed QObject
+            # is a use-after-free in C++ if it races the destructor, so check first —
+            # and if nobody is left to adopt the runner, stop it instead of orphaning
+            # its pytest subprocesses.
+            try:
+                if shiboken6.isValid(self):
+                    self.run_prep_finished.emit(result)
+                elif result is not None:
+                    log.warning("run preparation finished after the window was destroyed; stopping the prepared runner")
+                    result.runner.stop()
+            except RuntimeError as e:  # "Signal source has been deleted" — destroyed between check and emit
+                log.warning(f"could not deliver run preparation result: {e}")
+                if result is not None:
+                    result.runner.stop()
 
     def _build_runner(self, config: _RunPrepConfig, prior_runner: PytestRunner | None) -> "_RunPrepResult | None":
         """Prepare a run: discovery, RESUME handling, ordering — and start the runner.
@@ -357,27 +372,28 @@ def _build_runner(self, config: _RunPrepConfig, prior_runner: PytestRunner | Non
         """
         put_version_info = detect_put_version(config.project_root)
         log.info(f"PUT detected: {put_version_info}")
+        if self._run_prep_abort.is_set():
+            return None
 
         get_tests = GetTests(test_dir=config.project_root)
         get_tests.start()
 
-        # Wind down any previous runner while discovery proceeds. Bounded join: a wedged
-        # worker thread must not hang preparation forever (and since this is no longer on
-        # the GUI thread, it cannot freeze the UI either way).
+        # Wind down any previous runner while discovery proceeds. Bounded and abort-aware:
+        # a wedged worker thread must not hang preparation forever, and a closing window
+        # (abort set) must not wait out the full wind-down.
         if prior_runner is not None and prior_runner.is_running():
             prior_runner.stop()
-            if not prior_runner.join(120.0):
+            wind_down_deadline = time.monotonic() + 120.0
+            while prior_runner.is_running() and time.monotonic() < wind_down_deadline and not self._run_prep_abort.is_set():
+                prior_runner.join(1.0)
+            if prior_runner.is_running() and not self._run_prep_abort.is_set():
                 log.warning(f"previous run did not wind down within 120 s; starting the new run anyway ({config.run_guid=})")
 
-        while get_tests.is_alive():
-            get_tests.join(1.0)
-            if self._run_prep_abort.is_set():
-                get_tests.terminate()
-                get_tests.join(5.0)
-                return None
-        get_tests.join()
-
-        tests = get_tests.get_tests()
+        # Drains the discovery queue while waiting (a full queue pipe otherwise deadlocks
+        # the child's exit — see GetTests.collect) and honors the abort event.
+        tests = get_tests.collect(self._run_prep_abort)
+        if tests is None:
+            return None
 
         # Query prior results once (used by RESUME filtering, failed-first ordering, and
         # never-run prioritization). Read-only access; outputs are included because RESUME
@@ -392,6 +408,11 @@ def _build_runner(self, config: _RunPrepConfig, prior_runner: PytestRunner | Non
         if config.run_mode == RunMode.CHECK:
             effective_mode = self._resolve_check_mode(prior_results, put_version_info)
 
+        # Last side-effect-free abort point: past here preparation mutates state (deletes
+        # stale coverage data, copies RESUME records into the new run, starts the runner).
+        if self._run_prep_abort.is_set():
+            return None
+
         # Clear stale coverage data before any PytestProcess starts writing into
         # coverage/. Done here (before pytest_runner.start) rather than from a periodic
         # GUI tick so we cannot delete the directory while a still-running PytestProcess

src/pytest_fly/pytest_runner/pytest_runner.py

@@ -358,25 +358,27 @@ def join(self, timeout_seconds: float | None = None) -> bool:
         within the timeout. Waits for the pool to be spun up first, so calling right after
         :meth:`start` is safe.
 
-        :param timeout_seconds: Per-thread join timeout, or ``None`` to wait indefinitely.
+        :param timeout_seconds: One shared deadline for the whole join, or ``None`` to wait
+            indefinitely. (This was previously applied per thread, so a wedged pool of N
+            workers could hold a caller — including ``closeEvent`` on the GUI thread — for
+            N × timeout.)
         :return: ``True`` if all workers and the runner thread have exited.
         """
+        deadline = None if timeout_seconds is None else time.monotonic() + timeout_seconds
+
+        def remaining() -> float | None:
+            return None if deadline is None else max(0.0, deadline - time.monotonic())
 
         # in case join is called right after .start(), wait until .run() has started all workers
-        if timeout_seconds is not None:
-            start = time.time()
-            while not self._started_event.is_set() and time.time() - start < timeout_seconds:
-                time.sleep(0.1)
-        else:
-            self._started_event.wait()
+        self._started_event.wait(remaining())
 
         with self._pool_lock:
             test_runners = list(self._test_runners.values())
         for test_runner in test_runners:
-            test_runner.join(timeout_seconds)
+            test_runner.join(remaining())
         # Also join the runner thread itself so soft-stop finalization (marking the
         # remaining queue STOPPED) is complete when join() returns.
-        Thread.join(self, timeout_seconds)
+        Thread.join(self, remaining())
         return all(not test_runner.is_alive() for test_runner in test_runners) and not self.is_alive()
 
     def stop(self):

src/pytest_fly/pytest_runner/test_list.py

@@ -9,6 +9,7 @@
 from multiprocessing import Process, Queue
 from pathlib import Path
 from queue import Empty
+from threading import Event
 
 import pytest
 from typeguard import typechecked
@@ -101,16 +102,43 @@ def run(self):
 
         log.info(f'Discovered {len(pytest_tests)} pytest tests in "{self.test_dir}"')
 
-    def get_tests(self) -> list[ScheduledTest]:
-        """
-        Returns the list of scheduled tests after the process has run.
+    def collect(self, abort_event: Event | None = None, poll_seconds: float = 1.0) -> "list[ScheduledTest] | None":
+        """Wait for discovery to finish, draining results as they arrive, and return them.
+
+        Draining *while* the child is alive is what makes completion possible at all: a
+        multiprocessing child cannot exit until its queue's feeder thread has flushed
+        everything to the pipe, and the pipe blocks once its buffer (~64 KB, a few hundred
+        test modules) is full. Joining before draining therefore deadlocked run preparation
+        permanently on any suite past that size.
+
+        :param abort_event: When set, discovery is terminated and ``None`` is returned.
+        :param poll_seconds: Join/drain poll interval while the child runs.
+        :return: All discovered tests (sorted), or ``None`` on abort.
         """
+        while self.is_alive():
+            self.join(poll_seconds)
+            self._drain()
+            if abort_event is not None and abort_event.is_set():
+                self.terminate()
+                self.join(5.0)
+                return None
+        self.join()
+        return self.get_tests()
+
+    def _drain(self) -> None:
+        """Move everything currently on the result queue into ``scheduled_tests``."""
         try:
             while test := self._scheduled_tests_queue.get(False):
                 self.scheduled_tests.append(test)
         except Empty:
             pass
 
+    def get_tests(self) -> list[ScheduledTest]:
+        """
+        Returns the list of scheduled tests after the process has run.
+        """
+        self._drain()
+
         # Deterministic discovery order — the final execution order is decided
         # later by :func:`pytest_fly.pytest_runner.ordering.apply_ordering_aspects`.
         self.scheduled_tests.sort(key=lambda t: t.node_id)

tests/test_about_shutdown.py

@@ -0,0 +1,23 @@
+"""About tab thread teardown — destroying a running QThread is a Qt fatal abort."""
+
+from pytest_fly.gui.about_tab.about import About
+
+from .paths import get_temp_dir
+
+
+def test_about_shutdown_stops_data_thread(app, qtbot):
+    """shutdown() must leave the data thread finished, however early it is called.
+
+    Without it, closing the app before the About data arrived (git-based PUT detection can
+    take seconds) destroyed a running QThread: "QThread: Destroyed while thread is still
+    running", exit 0xC0000409, no Python traceback.
+    """
+    about = About(None, get_temp_dir("test_about_shutdown"))
+    qtbot.addWidget(about)
+
+    about.shutdown()  # immediately — the worker may still be mid-detect_put_version
+
+    assert not about._thread.isRunning()
+    # Idempotent: a second call (e.g. two close paths) is a no-op.
+    about.shutdown()
+    assert not about._thread.isRunning()

tests/test_pytest_runner/test_pytest_runner_robustness.py

@@ -12,6 +12,7 @@
 """
 
 import os
+import time
 from queue import Queue
 from threading import Event
 
@@ -118,3 +119,27 @@ def test_worker_clears_process_reference_between_tests(app):
 def test_process_monitor_is_daemon():
     """A daemon monitor can never block its parent test process's exit."""
     assert ProcessMonitor("run-guid", "tests/test_x.py", 1234, 1.0).daemon is True
+
+
+def test_join_timeout_is_a_shared_deadline(app):
+    """join(t) must bound the whole call at ~t, not t per worker thread.
+
+    Previously a wedged pool of N workers held the caller — including closeEvent on the
+    GUI thread — for N x t (minutes of "Not Responding" that users end with a kill).
+    """
+    data_dir = get_temp_dir("test_join_shared_deadline")
+    run_guid = generate_uuid()
+
+    long_tests = _scheduled("tests/test_long_operation.py", "tests/test_3_sec_operation.py", "tests/test_sleep.py")
+    runner = PytestRunner(run_guid, long_tests, 3, data_dir, update_rate=0.5)
+    runner.start()
+    assert runner.join(0.1) is False  # wait for the pool to spin up
+
+    start = time.monotonic()
+    finished = runner.join(1.0)
+    elapsed = time.monotonic() - start
+    assert finished is False  # the long tests are still running
+    assert elapsed < 3.0, f"join(1.0) with 3 busy workers took {elapsed:.1f}s — timeout applied per thread, not shared"
+
+    runner.stop()
+    assert runner.join(60.0)

tests/test_test_list.py

@@ -1,7 +1,9 @@
 """Tests for pytest_runner.test_list.GetTests."""
 
+import time
 from pathlib import Path
 from tempfile import TemporaryDirectory
+from threading import Event
 
 from pytest_fly.interfaces import ScheduledTest
 from pytest_fly.pytest_runner.test_list import GetTests
@@ -60,3 +62,50 @@ def test_get_tests_empty_dir():
         collector.join(60.0)
 
         assert collector.get_tests() == []
+
+
+class _ManyResults(GetTests):
+    """Discovery stand-in that produces far more results than the queue pipe can buffer.
+
+    ~5000 pickled ScheduledTests is well past the ~64 KB pipe buffer, so the child cannot
+    exit until the parent drains — the deadlock GetTests.collect exists to prevent.
+    """
+
+    def run(self) -> None:
+        for n in range(5000):
+            self._scheduled_tests_queue.put(ScheduledTest(f"tests/test_{n:05d}.py", False, None, None))
+
+
+class _NeverFinishes(GetTests):
+    """Discovery stand-in that never completes (wedged collection)."""
+
+    def run(self) -> None:
+        time.sleep(600.0)
+
+
+def test_collect_drains_large_result_sets_without_deadlock():
+    """collect() must drain while the child runs; join-before-drain hung forever at ~500 modules."""
+    collector = _ManyResults()
+    collector.start()
+    start = time.monotonic()
+    tests = collector.collect(Event(), poll_seconds=0.1)
+    elapsed = time.monotonic() - start
+    assert tests is not None
+    assert len(tests) == 5000
+    assert tests == sorted(tests, key=lambda t: t.node_id)
+    assert not collector.is_alive()
+    assert elapsed < 30.0, f"collect took {elapsed:.1f}s"
+
+
+def test_collect_aborts_wedged_discovery():
+    """A set abort event must terminate discovery promptly and return None."""
+    collector = _NeverFinishes()
+    collector.start()
+    abort = Event()
+    abort.set()
+    start = time.monotonic()
+    result = collector.collect(abort, poll_seconds=0.1)
+    elapsed = time.monotonic() - start
+    assert result is None
+    assert not collector.is_alive()
+    assert elapsed < 15.0, f"abort took {elapsed:.1f}s"