HOLD (owner area #32): fix(net): stop and join pw-reaper threads at interpreter exit

passiveworkers/net/store.py

@@ -21,8 +21,11 @@
 
 from __future__ import annotations
 
+import atexit
 import sqlite3
 import threading
+import time
+import weakref
 
 from passiveworkers.ledger import ESCROW_ID
 from passiveworkers.net._store_assisted import _AssistedMixin
@@ -36,6 +39,31 @@
 # `from passiveworkers.net.store import _hash` — used directly by tests/test_security_privacy.py —
 # keeps working unchanged now that the helper itself lives in _store_base.py.
 
+# Every live Store, so its daemon reaper can be stopped and joined at interpreter exit. A reaper
+# still printing (its failed-pass log line) when the interpreter finalizes holds stdout's buffer
+# lock, and CPython aborts with "Fatal Python error: _enter_buffered_busy" (the same hazard #48
+# fixed for the pw-baseline thread). Weak, so a stopped Store can still be collected. Kept across
+# importlib.reload() (tests reload this module constantly): a fresh set would orphan every Store
+# made before the reload, and the hook below is registered once and reads the set at exit.
+_LIVE_STORES: weakref.WeakSet = globals().get("_LIVE_STORES", weakref.WeakSet())
+_REAPER_EXIT_JOIN_S = 5.0
+
+
+def _stop_reapers_at_exit() -> None:
+    """Stop every reaper, then join them all under ONE shared deadline (a reaper stuck in a pass
+    can delay exit by at most _REAPER_EXIT_JOIN_S, never wedge it)."""
+    stores = list(_LIVE_STORES)
+    for s in stores:
+        s._stop.set()
+    deadline = time.monotonic() + _REAPER_EXIT_JOIN_S
+    for s in stores:
+        s._reaper.join(max(0.0, deadline - time.monotonic()))
+
+
+if not globals().get("_EXIT_HOOK_REGISTERED"):
+    atexit.register(_stop_reapers_at_exit)
+    _EXIT_HOOK_REGISTERED = True
+
 
 class Store(_JobsMixin, _AssistedMixin, _LedgerMixin, _ReportingMixin):
     """Composes the four Store mixins (jobs/orchestration, assisted marketplace, ledger/
@@ -59,6 +87,7 @@ def __init__(self, path: str | None = None):
         self._stop = threading.Event()
         self._reaper = threading.Thread(target=self._reap_loop, daemon=True, name="pw-reaper")
         self._reaper.start()
+        _LIVE_STORES.add(self)
 
     # ------------------------------------------------------------------ integrity
     def ledger_integrity(self) -> dict:

tests/test_reaper_shutdown.py

@@ -0,0 +1,96 @@
+"""The pw-reaper daemon thread must never be left printing into interpreter shutdown.
+
+A reaper whose pass fails prints a log line to stdout. If it is mid-write when the interpreter
+finalizes, it holds stdout's buffer lock, the final flush can't take it, and CPython aborts with
+"Fatal Python error: _enter_buffered_busy ... possibly due to daemon threads": the same crash #48
+fixed for the pw-baseline thread (test (3.10), CI run 38124448503). store.py now stops and joins
+every reaper at exit.
+
+Each test runs a real child interpreter. The child shortens the reaper tick (it patches
+threading.Event.wait so the 10 s floor in _reap_loop doesn't apply) and makes every pass fail.
+The parent holds off reading the child's stdout, so the child's reaper is blocked inside its write
+(holding the lock) when the child's main thread returns. The parent then drains. On the old code
+this aborts every time (SIGABRT, rc -6) on 3.10-3.13; with the fix the child exits 0.
+No network: the store is a temp SQLite file.
+"""
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+_CHILD_PRELUDE = r"""
+import sys, threading, time
+_Event = threading.Event
+class _FastEvent(_Event):
+    def wait(self, timeout=None):
+        return super().wait(None if timeout is None else min(timeout, 0.001))
+threading.Event = _FastEvent   # shorten the reaper tick only; Thread.start's untimed wait is unchanged
+import passiveworkers.net.store as store_mod
+in_pass = _Event()
+"""
+
+# Every pass fails with a 2 MB message: one print overfills the 64 KB pipe and blocks in write().
+_CHILD_FAILING_PASS = _CHILD_PRELUDE + r"""
+def failing_pass(self):
+    in_pass.set()
+    raise RuntimeError("x" * 2_000_000)
+store_mod.Store._reap_once = failing_pass
+store = store_mod.Store(sys.argv[1])
+if sys.argv[2:] == ["reload"]:   # tests reload this module constantly; a Store made before must still stop
+    import importlib
+    importlib.reload(store_mod)
+in_pass.wait(10)
+time.sleep(0.3)   # the reaper is now blocked in print() holding stdout's lock
+sys.stderr.write("main-exiting\n"); sys.stderr.flush()
+"""
+
+# A pass that never returns: the exit join must be bounded, never wedge shutdown.
+_CHILD_STUCK_PASS = _CHILD_PRELUDE + r"""
+store_mod._REAPER_EXIT_JOIN_S = 0.5
+def stuck_pass(self):
+    in_pass.set()
+    _Event().wait()
+store_mod.Store._reap_once = stuck_pass
+store_mod.Store(sys.argv[1])
+in_pass.wait(10)
+sys.stderr.write("main-exiting\n"); sys.stderr.flush()
+"""
+
+
+def _run_child(code: str, db: Path, read_delay_s: float, *args: str):
+    env = {**os.environ, "PYTHONPATH": str(ROOT) + os.pathsep + os.environ.get("PYTHONPATH", "")}
+    p = subprocess.Popen([sys.executable, "-c", code, str(db), *args], stdout=subprocess.PIPE,
+                         stderr=subprocess.PIPE, env=env, cwd=str(ROOT))
+    try:
+        first = p.stderr.readline()      # wait until the child's main thread is about to return
+        time.sleep(read_delay_s)         # ...and keep its stdout pipe full meanwhile
+        _out, err = p.communicate(timeout=60)
+    finally:
+        if p.poll() is None:
+            p.kill()
+            p.communicate()
+    return p.returncode, (first + err).decode("utf-8", "replace")
+
+
+@pytest.mark.parametrize("reload", [False, True], ids=["plain", "after-module-reload"])
+def test_reaper_printing_at_exit_does_not_abort_the_interpreter(tmp_path, reload):
+    rc, err = _run_child(_CHILD_FAILING_PASS, tmp_path / "coord.db", 1.5, *(["reload"] if reload else []))
+    assert "main-exiting" in err, err[-2000:]
+    assert "Fatal Python error" not in err, err[-2000:]
+    assert rc == 0, (rc, err[-2000:])
+
+
+def test_a_stuck_reaper_cannot_wedge_interpreter_exit(tmp_path):
+    t0 = time.monotonic()
+    rc, err = _run_child(_CHILD_STUCK_PASS, tmp_path / "coord.db", 0.0)
+    elapsed = time.monotonic() - t0
+    assert "main-exiting" in err, err[-2000:]
+    assert rc == 0, (rc, err[-2000:])
+    assert elapsed < 30, elapsed     # bounded by _REAPER_EXIT_JOIN_S (0.5 s here) plus startup