Harden `test_infected_asyncio` for fork spawners

Deats,
- `test_echoserver_detailed_mechanics`: add `is_forking_spawner`
  param, wrap `main()` in `fa_main()` with per-backend
  `trio.fail_after` (4s fork / 1s trio) to cap cancel-cascade
  teardown that compounds under forkserver.
- `test_sigint_closes_lifetime_stack`: swap `start_method` param
  for `is_forking_spawner`, pre-init `tmp_file`/`ctx` to `None` so
  KBI firing before `open_context` body doesn't `UnboundLocalError`,
  add `pytest.fail` guard for the spawn-time IPC race case, arm
  `signal.alarm` AFK-safety cap (10s) under fork backends

Also,
- `pytestmark`: add `track_orphaned_uds_per_test` +
  `detect_runaway_subactors_per_test` fixtures.
- `delay()`: hardcode `return 1e3` at top (debug override still in
  place).

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code

tests/msg/test_ext_types_msgspec.py

@@ -57,6 +57,7 @@ from tractor.msg._ops import (
     limit_plds,
 )
 
+
 def enc_nsp(obj: Any) -> Any:
     actor: Actor = tractor.current_actor(
         err_on_no_runtime=False,
@@ -617,6 +618,17 @@ def test_ext_types_over_ipc(
     debug_mode: bool,
     pld_spec: Union[Type],
     add_hooks: bool,
+
+    set_fork_aware_capture,
+    # ^^XXX? for forking spawners
+
+    # capfd: pytest.CaptureFixture,
+    # ^^NOTE, super interesting that if
+    # we disable this below then the tpt-layer
+    # suffers as an "unclean EOF"??
+    # ?TODO, determine why/how that mks sense when addressing,
+    # https://github.com/pytest-dev/pytest/issues/14444
+    #
 ):
     '''
     Ensure we can support extension types coverted using
@@ -725,18 +737,26 @@ def test_ext_types_over_ipc(
 
             await p.cancel_actor()
 
+    async def fa_main():
+        with (
+            trio.fail_after(2),
+            # ?TODO, investigate? see NOTE above..
+            # capfd.disabled(),
+        ):
+            await main()
+
     if (
         NamespacePath in pld_types
         and
         add_hooks
     ):
-        trio.run(main)
+        trio.run(fa_main)
 
     else:
         with pytest.raises(
             expected_exception=tractor.RemoteActorError,
         ) as excinfo:
-            trio.run(main)
+            trio.run(fa_main)
 
         exc = excinfo.value
         # bc `.started(nsp: NamespacePath)` will raise

tests/msg/test_pldrx_limiting.py

@@ -284,6 +284,11 @@ def test_basic_payload_spec(
     return_value: str|None,
     started_value: int|PldMsg,
     pld_check_started_value: bool,
+
+    set_fork_aware_capture,
+    # ^XXX TODO? for forking spawners, seems to prevent hangs when
+    # --capture=sys not set, but only for a while then the problem
+    # accumulates?
 ):
     '''
     Validate the most basic `PldRx` msg-type-spec semantics around

tests/test_advanced_streaming.py

@@ -147,6 +147,9 @@ def test_dynamic_pub_sub(
     test_log: tractor.log.StackLevelAdapter,
     reap_subactors_per_test: int,
     expect_cancel_exc: Type[BaseException],
+
+    is_forking_spawner: bool,
+    set_fork_aware_capture,
 ):
     failed_to_raise_report: str = (
         f'Never got a {expect_cancel_exc!r} ??'
@@ -157,30 +160,59 @@ def test_dynamic_pub_sub(
     from multiprocessing import cpu_count
     cpus = cpu_count()
 
-    # Hard safety cap via trio's own cancellation — see the
+    # Hard safety cap via trio's own cancellation. NOTE see the
-    # module-level NOTE on why we avoid `pytest-timeout` for
+    # module-level note on why we avoid `pytest-timeout` for this
-    # this test. Picked backend-aware: under `trio` backend
+    # test. Picked backend-aware: under `trio` backend spawn is
-    # spawn is cheap (~1s for `cpus` actors) but fork-based
+    # cheap (~1s for `cpus` actors) but fork-based backends pay
-    # backends pay a per-spawn cost (forkserver round-trip +
+    # a per-spawn cost (forkserver round-trip + IPC peer-handshake)
-    # IPC peer-handshake) that can stack up over `cpus - 1`
+    # that can stack up over `cpus - 1` sequential `n.run_in_actor()`
-    # sequential `n.run_in_actor()` calls — especially on UDS
+    # calls — especially on UDS under cross-pytest contention
-    # under cross-pytest contention (#451 / #452). Empirically
+    # (#451 / #452). Empirically a flat 15s flakes on
-    # 12s flakes on `main_thread_forkserver`; 30s gives
+    # `main_thread_forkserver` for many-cpu hosts (a single bad
-    # plenty of headroom while still failing-loud on a real
+    # spawn-stack puts total run-time at ~15.5s, just over);
-    # hang.
+    # 30s gives plenty of headroom while still failing-loud on
-    from tractor.spawn import _spawn as _spawn_mod
+    # a real hang.
+    #
+    # XXX caveat: this is an *inner* `trio.fail_after` — its
+    # `Cancelled` cannot reach a task parked in a shielded `await`
+    # (e.g. inside actor-nursery teardown). When the in-band cancel
+    # path is itself buggy (the bug-class-3 `raise KBI` swallow we're
+    # currently chasing) this guard does NOT fire and the test sits
+    # forever until external SIGINT. The `_DIAG_CAP_S` outer guard
+    # below is the AFK-safety counterpart.
     fail_after_s: int = (
-        30
+        4
-        if _spawn_mod._spawn_method == 'main_thread_forkserver'
+        if is_forking_spawner
         else 12
     )
 
+    # outer guard: when the inner fail_after fails to fire because of
+    # a shielded-await deadlock, this cap *aborts the trio run via
+    # signal.alarm → KBI* so AFK runs don't sit for >20min on the
+    # bug-class-3 hang. Slightly larger than `fail_after_s` so the
+    # trio-native path always wins when it works.
+    _DIAG_CAP_S: int = fail_after_s + 5
+
     async def main():
+        # bug-class-3 breadcrumb: tag each level of the cancel path
+        # so when the run hangs and we capture cancel-level logs, the
+        # *last* breadcrumb that fired names the swallow point.
+        test_log.cancel('test_dynamic_pub_sub: enter main()')
+        try:
             with trio.fail_after(fail_after_s):
+                test_log.cancel(
+                    f'test_dynamic_pub_sub: '
+                    f'enter `trio.fail_after({fail_after_s})` scope'
+                )
+                try:
                     async with tractor.open_nursery(
                         registry_addrs=[reg_addr],
                         debug_mode=debug_mode,
                     ) as n:
+                        test_log.cancel(
+                            'test_dynamic_pub_sub: '
+                            'actor nursery opened'
+                        )
 
                         # name of this actor will be same as target func
                         await n.run_in_actor(publisher)
@@ -208,8 +240,33 @@ def test_dynamic_pub_sub(
                             f'Raising user cancel exc: '
                             f'{expect_cancel_exc!r}'
                         )
+                        test_log.cancel(
+                            f'test_dynamic_pub_sub: '
+                            f'ABOUT TO RAISE {expect_cancel_exc!r}'
+                        )
                         raise expect_cancel_exc('simulate user cancel!')
+                finally:
+                    test_log.cancel(
+                        'test_dynamic_pub_sub: '
+                        'actor nursery `__aexit__` returned'
+                    )
+            test_log.cancel(
+                'test_dynamic_pub_sub: `fail_after` scope exited'
+            )
+        finally:
+            test_log.cancel(
+                'test_dynamic_pub_sub: leaving `main()`'
+            )
 
+    # outer signal-based guard — survives a shielded-await deadlock
+    # since `signal.alarm` raises in the main thread regardless of
+    # trio's scope state. ONLY armed under fork-based backends since
+    # the bug we're chasing is MTF-specific.
+    import signal
+    armed_alarm: bool = bool(is_forking_spawner)
+    if armed_alarm:
+        signal.alarm(_DIAG_CAP_S)
+    try:
         try:
             trio.run(main)
             pytest.fail(failed_to_raise_report)
@@ -230,6 +287,11 @@ def test_dynamic_pub_sub(
                 pytest.fail(failed_to_raise_report)
 
             test_log.exception('Got user-cancel exc AS EXPECTED')
+    finally:
+        # always disarm so a passing test doesn't get killed
+        # post-trio.run by a stale alarm.
+        if armed_alarm:
+            signal.alarm(0)
 
 
 @tractor.context
@@ -361,9 +423,12 @@ def test_sigint_both_stream_types():
                             resp = await stream.receive()
                             assert resp == msg
                             raise KeyboardInterrupt
+
+    # TODO, use pytest.raises() here instead?
+    # (why weren't we originally?)
     try:
         trio.run(main)
-        assert 0, "Didn't receive KBI!?"
+        pytest.fail("Didn't receive KBI!?")
     except KeyboardInterrupt:
         pass
 

tests/test_clustering.py

@@ -77,6 +77,7 @@ async def worker(
 @tractor_test
 async def test_streaming_to_actor_cluster(
     tpt_proto: str,
+    is_forking_spawner: bool,
 ):
     '''
     Open an actor "cluster" using the (experimental) `._clustering`
@@ -88,7 +89,11 @@ async def test_streaming_to_actor_cluster(
             f'Test currently fails with tpt-proto={tpt_proto!r}\n'
         )
 
-    with trio.fail_after(6):
+    delay: float = (
+        10 if is_forking_spawner
+        else 6
+    )
+    with trio.fail_after(delay):
         async with (
             open_actor_cluster(modules=[__name__]) as portals,
 

tests/test_context_stream_semantics.py

@@ -188,9 +188,16 @@ def test_simple_context(
     pointlessly_open_stream,
     reg_addr: tuple,
     debug_mode: bool,
+    is_forking_spawner: bool,
 ):
 
-    timeout = 1.5 if not platform.system() == 'Windows' else 4
+    timeout: float = 1.5
+    # windows and forking-spawner both have "slower but more
+    # deterministic" cancel teardown.
+    if platform.system() == 'Windows':
+        timeout = 4
+    elif is_forking_spawner:
+        timeout = 3
 
     async def main():
 

tests/test_infected_asyncio.py

@@ -45,6 +45,12 @@ from tractor._testing import expect_ctxc
 # `test_legacy_one_way_streaming`, etc.).
 pytestmark = pytest.mark.usefixtures(
     'reap_subactors_per_test',
+    # NOTE, asyncio cancel cascade has historically
+    # triggered both UDS sockfile leaks (SIGKILL path)
+    # AND the trio `WakeupSocketpair.drain()` busy-loop
+    # — see `test_aio_simple_error`'s history.
+    'track_orphaned_uds_per_test',
+    'detect_runaway_subactors_per_test',
 )
 
 
@@ -52,6 +58,7 @@ pytestmark = pytest.mark.usefixtures(
     scope='module',
 )
 def delay(debug_mode: bool) -> int:
+    return 1e3
     if debug_mode:
         return 999
     else:
@@ -826,13 +833,19 @@ async def trio_to_aio_echo_server(
 
 @pytest.mark.parametrize(
     'raise_error_mid_stream',
-    [False, Exception, KeyboardInterrupt],
+    [
+        False,
+        Exception,
+        KeyboardInterrupt,
+    ],
     ids='raise_error={}'.format,
 )
 def test_echoserver_detailed_mechanics(
     reg_addr: tuple[str, int],
     debug_mode: bool,
     raise_error_mid_stream,
+
+    is_forking_spawner: bool,
 ):
     async def main():
         async with tractor.open_nursery(
@@ -880,12 +893,34 @@ def test_echoserver_detailed_mechanics(
             # is cancelled by kbi or out of task cancellation
             await p.cancel_actor()
 
+    # NOTE: under fork-based backends the cancel-cascade
+    # path is structurally slower than `trio`'s subproc-exec
+    # (per-spawn forkserver-handshake compounds during
+    # teardown). Bump the cap so cross-test contamination
+    # doesn't flake this — see
+    # `ai/conc-anal/cancel_cascade_too_slow_under_main_thread_forkserver_issue.md`.
+    timeout: float = (
+        999 if tractor.debug_mode()
+        else 4 if is_forking_spawner
+        else 1
+    )
+    with_timeout: bool = (
+        True
+        # False
+    )
+    async def fa_main():
+        if with_timeout:
+            with trio.fail_after(timeout):
+                await main()
+        else:
+            await main()
+
     if raise_error_mid_stream:
         with pytest.raises(raise_error_mid_stream):
-            trio.run(main)
+            trio.run(fa_main)
 
     else:
-        trio.run(main)
+        trio.run(fa_main)
 
 
 @tractor.context
@@ -1038,7 +1073,7 @@ def test_sigint_closes_lifetime_stack(
     bg_aio_task: bool,
     trio_side_is_shielded: bool,
     send_sigint_to: str,
-    start_method: str,
+    is_forking_spawner: bool,
 ):
     '''
     Ensure that an infected child can use the `Actor.lifetime_stack`
@@ -1053,6 +1088,14 @@ def test_sigint_closes_lifetime_stack(
             if debug_mode
             else 1
         )
+        # pre-init so the `except (KeyboardInterrupt, ContextCancelled)`
+        # handler below doesn't `UnboundLocalError` if KBI fires BEFORE
+        # we ever enter the `as (ctx, first)` body (e.g. when
+        # `p.open_context().__aenter__` is hung waiting for the
+        # subactor's `StartAck` due to a fork-child IPC race —
+        # see `dynamic_pub_sub_spawn_time_transport_close_under_mtf_issue.md`).
+        tmp_file: Path|None = None
+        ctx: tractor.Context|None = None
         try:
             an: tractor.ActorNursery
             async with tractor.open_nursery(
@@ -1078,7 +1121,7 @@ def test_sigint_closes_lifetime_stack(
                 ) as (ctx, first):
 
                     path_str, cpid = first
-                    tmp_file: Path = Path(path_str)
+                    tmp_file = Path(path_str)
                     assert tmp_file.exists()
 
                     # XXX originally to simulate what (hopefully)
@@ -1129,7 +1172,7 @@ def test_sigint_closes_lifetime_stack(
                             if (
                                 send_sigint_to == 'child'
                                 and
-                                start_method == 'main_thread_forkserver'
+                                is_forking_spawner
                             ):
                                 pytest.xfail(
                                     reason=(
@@ -1156,6 +1199,21 @@ def test_sigint_closes_lifetime_stack(
             KeyboardInterrupt,
             ContextCancelled,
         ):
+            # If we got here BEFORE entering the ctx body (e.g.
+            # spawn-time IPC race hung `open_context.__aenter__` and
+            # the AFK-guard `signal.alarm` fired KBI from outside the
+            # trio loop), `tmp_file`/`ctx` are still `None` — surface
+            # that fact directly instead of `UnboundLocalError`.
+            if tmp_file is None:
+                pytest.fail(
+                    'KBI/ctxc fired BEFORE `p.open_context()` returned '
+                    "the child's `started` value — likely fork-child "
+                    'IPC race; see '
+                    '`ai/conc-anal/'
+                    'dynamic_pub_sub_spawn_time_transport_close_'
+                    'under_mtf_issue.md`'
+                )
+
             # XXX CASE 2: without the bug fixed, in the
             # KBI-raised-in-parent case, the actor teardown should
             # never get run (silently abaondoned by `asyncio`..) and
@@ -1163,7 +1221,32 @@ def test_sigint_closes_lifetime_stack(
             assert not tmp_file.exists()
             assert ctx.maybe_error
 
+    # outer signal-based AFK-safety guard. mirrors the pattern in
+    # `tests/test_advanced_streaming.py::test_dynamic_pub_sub`: when
+    # the in-band trio cancel path doesn't fire (e.g. parent is
+    # parked in a shielded `await` inside actor-nursery teardown, or
+    # `open_context.__aenter__` hangs waiting for a child's
+    # `StartAck` that never comes), `signal.alarm` raises KBI in the
+    # main thread regardless of trio's scope state. This caps the
+    # absolute wall-clock so an AFK run can't sit for an hour on a
+    # forkserver-launchpad-contamination hang. Only armed under fork-
+    # based backends since the bug class is MTF-specific.
+    _AFK_CAP_S: int = (
+        999 if debug_mode
+        else 10
+    )
+    armed_alarm: bool = (
+        not debug_mode
+        and
+        is_forking_spawner
+    )
+    if armed_alarm:
+        signal.alarm(_AFK_CAP_S)
+    try:
         trio.run(main)
+    finally:
+        if armed_alarm:
+            signal.alarm(0)
 
 
 

tests/test_inter_peer_cancellation.py

@@ -26,14 +26,30 @@ from tractor._testing import (
 
 from .conftest import cpu_scaling_factor
 
-pytestmark = pytest.mark.skipon_spawn_backend(
+pytestmark = [
+    pytest.mark.skipon_spawn_backend(
         'subint',
         reason=(
             'XXX SUBINT GIL-CONTENTION HANGING TEST XXX\n'
-        'See oustanding issue(s)\n'
+            'Inter-peer cancel cascades under '
-        # TODO, put issue link!
+            '`--spawn-backend=subint` trip the abandoned-subint '
-    )
+            'GIL-hostage class — see\n'
+            '  - `ai/conc-anal/subint_sigint_starvation_issue.md` '
+            '(GIL-hostage, SIGINT-unresponsive)\n'
+            '  - `ai/conc-anal/subint_cancel_delivery_hang_issue.md` '
+            '(sibling: parent parks on dead chan)\n'
+            '  - https://github.com/goodboy/tractor/issues/379 '
+            '(subint umbrella)\n'
         )
+    ),
+    # NOTE, inter-peer cancellation tests stress the
+    # multi-actor cancel cascade which under SIGKILL
+    # leaves UDS sock-files orphaned. Track per-test
+    # for blame attribution.
+    pytest.mark.usefixtures(
+        'track_orphaned_uds_per_test',
+    ),
+]
 
 # XXX TODO cases:
 # - [x] WE cancelled the peer and thus should not see any raised

tractor/_testing/pytest.py

@@ -653,6 +653,7 @@ def pytest_generate_tests(
     #         scope='module',
     #     )
 
+
 def _is_forking_spawner(
     start_method: str,
 ) -> bool:
@@ -670,7 +671,7 @@ def is_forking_spawner(
     Is the `pytest` run using a `fork()`ing process spawning-backend?
 
     '''
-    return _is_forking_spawner
+    return _is_forking_spawner(start_method)
 
 
 def maybe_xfail_for_spawner(