From 13ffb13be5ad4562274f1fe608c52ab22bbf3b2b Mon Sep 17 00:00:00 2001 From: goodboy Date: Tue, 29 Sep 2026 18:05:55 -0400 Subject: [PATCH] Test SIGINT teardown with stalled IPC transports Add `examples.advanced_faults.ipc_stalls` and an external `RootMonitor` so SIGINT delivery and shutdown evidence stay independent of the stalled actor channels. Exercise six Linux TCP cases with the `trio` spawner: healthy IPC, context receive, send, both directions, context cancel send, and both directions plus the subactor `Portal` cancel send. Deats, - Split cancellable send gates from buffered receive gates, and require actual parked operations before publishing readiness. - Keep the OOB context `Channel` distinct from the child-control `Portal`; context-only faults retain cooperative child exit. - Require finite cancel-send expiry, runtime hard-reap logs, SIGKILL status and completed `waitpid` for the control fault. - Bound readiness and shutdown with an independent stdout watchdog; rescue cleanup cannot satisfy teardown assertions. - Use `get_rando_addr()` for unreserved registry port selection, raise `TimeoutError` on watchdog expiry, and retain structured shutdown evidence through `record_property`. These waits model cancellable runtime unresponsiveness, not kernel TCP loss or an uninterruptible syscall. Preserve human review decisions and historical working-tree verification in the prompt provenance; exact-tree plan checks remain pending. Prompt-IO: ai/prompt-io/opencode/20260923_ipc_fault_regressions.md (this commit msg was generated in some part by `opencode` using `gpt-6.1-sol` (`openai`)) --- .../20260923_ipc_fault_regressions.md | 84 +++ .../20260923_ipc_fault_regressions.raw.md | 25 + examples/advanced_faults/ipc_stalls.py | 514 ++++++++++++++++++ tests/test_ipc_stalls.py | 402 ++++++++++++++ 4 files changed, 1025 insertions(+) create mode 100644 ai/prompt-io/opencode/20260923_ipc_fault_regressions.md create mode 100644 ai/prompt-io/opencode/20260923_ipc_fault_regressions.raw.md create mode 100644 examples/advanced_faults/ipc_stalls.py create mode 100644 tests/test_ipc_stalls.py diff --git a/ai/prompt-io/opencode/20260923_ipc_fault_regressions.md b/ai/prompt-io/opencode/20260923_ipc_fault_regressions.md new file mode 100644 index 00000000..ea1e5435 --- /dev/null +++ b/ai/prompt-io/opencode/20260923_ipc_fault_regressions.md @@ -0,0 +1,84 @@ +--- +model: openai/gpt-6-astra +service: opencode +scope: code +substantive: true +git_ref: 9547c2f9 +raw_file: 20260923_ipc_fault_regressions.raw.md +--- + +## Human instructions + +Audit terminal transport-close context interruption and reconcile the +advanced-fault suite. Preserve its original stalled-IPC cancellation +invariant using controlled cancellable waits, rather than privileged +kernel DROP tooling immediately. Write a detailed future network-outage +handoff for TIPC/WireGuard and IB resiliency. Do not disturb live setups. + +## Changes + +- examples/advanced_faults/ipc_failure_during_stream.py: distinguish + detected close from stalled I/O and use explicit test registry inputs. +- tests/test_advanced_faults.py: validate transport outcomes, origin and + child reaping instead of obsolete expected hangs/cancel acknowledgements. +- examples/advanced_faults/ipc_stalls.py and tests/test_ipc_stalls.py: + controlled per-channel cancellable waits, actual test-root SIGINT, + finite-deadline cleanup synchronization, and stalled-subactor-Portal + hard-reap coverage. The example has a discoverable console entrypoint. + The review split the transport send wrapper from its buffered + receive-stream wrapper, moved the runnable scenario to examples, + and extracted external process monitoring into a context manager. +- tractor/_context.py: raise unfinished recorded transport errors even + after scope cancellation was absorbed and the channel is disconnected. +- plans/opencode/network-outage-resiliency-handoff.md: separate, + follow-up engineering handoff for kernel/network/application + resiliency; it is not part of the stall-test commit boundary. + +## Verification + +Advanced faults: 11 TCP and 11 UDS passed, including all 22 child-exit +checks. Existing EOF: 5 TCP and 5 UDS passed. Context-stream semantics: +65 TCP and 65 UDS passed. Final stall suite: 6 TCP cases passed, including +runtime hard reaping of a test-owned child whose control channel stalls. +Passing tests do not rely on forced watchdog cleanup. Reaper disabled; +legacy test registry defaults were overridden in memory for isolation. +These are working-tree checks, not commit-plan attestations. +After the review adjustments, the six stall cases passed again. +The third review retained those six cases after moving the runnable +scenario and isolating subprocess monitoring from assertions. +The next local review's six renamed TCP cases passed in 11.98 seconds +with the reaper plugin disabled; the example's CLI help also passed. + +## Human contributions + +The human identified the old close simulation and its intended Ctrl-C +contract, explicitly distinguished that from the newly accepted EOF +notification contract, and selected controlled runtime stalls as the +bounded first step. They supplied the Gateway/roaming incident traces +and required isolated worktrees rather than changes to live source. +The original unstaged EOF mask remains untouched in its own checkout. +During a second review, the human supplied example documentation, +asked for precise type/field references, pointed out the mixed +transport/receiver interface, and requested explicit parameter +types, method contracts, and test-mode descriptions. Those decisions +shape the final test helper and driver. +The human's third review requested the example-module location, +explicit parked-I/O documentation, code-quoted symbols and namespaces, +and a context manager for the external process monitor. They chose to +consider the longer-term network handoff in a follow-up commit. +The next local review requested explicit context/subactor Portal +distinctions, six descriptive mode names, structured case documentation +and per-branch shutdown evidence in tests/test_ipc_stalls.py and +examples/advanced_faults/ipc_stalls.py. It also requested TimeoutError +for watchdog expiry, a configurable signal with SIGINT as default, +reuse of tractor._testing.addr.get_rando_addr(), and follow-up TODOs +for a shared monitor/scenario interface and discovery daemon reuse. +The human explicitly authorized removal of the record_property TODO. + +## Scope limits + +No automatic IB reconnect, kernel DROP, suspend/resume monitor, native +debugger/Qt crash fix, or new production transport API is implemented. +Stall tests currently target Linux/Trio-spawner/TCP. The separate +detected-close fix was committed as `192411c4`; this stall-test +follow-up remains uncommitted. diff --git a/ai/prompt-io/opencode/20260923_ipc_fault_regressions.raw.md b/ai/prompt-io/opencode/20260923_ipc_fault_regressions.raw.md new file mode 100644 index 00000000..8fcfbd66 --- /dev/null +++ b/ai/prompt-io/opencode/20260923_ipc_fault_regressions.raw.md @@ -0,0 +1,25 @@ +--- +model: openai/gpt-6-astra +service: opencode +git_ref: 9547c2f9 +diff_cmd: git diff HEAD +--- + +> git diff HEAD -- tractor/_context.py tests/test_advanced_faults.py examples/advanced_faults/ipc_failure_during_stream.py tests/test_ipc_stalls.py tests/_ipc_stalls.py + +Separate terminal transport closure from unresponsive IPC in fault +coverage. Preserve original fault causes and cancellation provenance, +and test actual SIGINT only on externally controlled test-root processes. +Add normally cancellable gates with parked-operation handshakes. +An additional control-stall case requires runtime out-of-IPC hard reaping, +not cooperative cancellation through an unaffected control connection. + +The corrected matrix exposed a swallowed TransportClosed after the +context nursery absorbed its own cancellation. Route that unfinished +recorded outcome through wait_for_result even on a disconnected channel. +This raises without waiting on IPC and preserves normal final outcomes. + +Validation completed: 22 advanced-fault cases, 10 existing EOF cases, +130 context-stream cases, and 6 final controlled-stall cases passed. +No kernel network tools, user processes, live containers, or unrelated +shm resources were modified. diff --git a/examples/advanced_faults/ipc_stalls.py b/examples/advanced_faults/ipc_stalls.py new file mode 100644 index 00000000..d9086496 --- /dev/null +++ b/examples/advanced_faults/ipc_stalls.py @@ -0,0 +1,514 @@ +''' +Test-owned IPC stall simulation and a separately signalled root. + +These cancellable waits simulate runtime unresponsiveness, NOT +kernel TCP half-open connections, loss, or a shielded syscall stall. +Context-only faults wrap the OOB ``peer_ptl``'s separate ``Channel`` +to the child, preserving cancellation through the subactor +``staller_ptl`` returned by ``an.start_actor()``. The mode +``stalled_context_and_subactor_cancel_send`` parks both context +directions and the subactor ``Portal``'s cancel send. Registry +traffic and the independent stdout control pipe remain usable. + +''' +import argparse +from collections.abc import AsyncIterator +from functools import partial +import json +import os +from typing import Any + +import trio +import tractor +from tricycle import BufferedReceiveStream +from tractor._testing import assert_child_reaped +from tractor._testing.addr import get_rando_addr +from tractor.discovery._addr import UnwrappedAddress +from tractor.ipc._chan import _connect_chan +from tractor.ipc._transport import MsgpackTransport +from tractor.msg.types import ( + Start, + Yield, +) +from tractor.runtime._portal import open_portal + + +class StallableReceiveStream: + ''' + Simulate a stalled buffered read on one IPC transport. + + ``MsgpackTransport._iter_packets()`` reads each packet header + via ``MsgpackTransport.recv_stream.receive_exactly(4)``. + Replacing ``MsgpackTransport.recv_stream`` intercepts that call, + including reads satisfied without another socket operation. + The packet iterator reads this field for each new header and + payload. If an old call is already waiting inside the original + receiver, the peer's responses finish that call first; its + next read then goes through the replacement. + + ''' + def __init__( + self, + orig_recv: BufferedReceiveStream, + ) -> None: + ''' + Retain the receiver and allocate cancellable stall events. + + ''' + self.orig_recv: BufferedReceiveStream = orig_recv + self.enabled: bool = False + self.parked: trio.Event = trio.Event() + self.release: trio.Event = trio.Event() + self.receiving: int = 0 + + async def receive_exactly( + self, + size: int, + ) -> bytes: + ''' + Pause a buffered read until the test releases or cancels it. + + Count active calls and set + ``StallableReceiveStream.parked`` before waiting on + ``StallableReceiveStream.release``. Once released, read + from the original ``BufferedReceiveStream``. + The wait stays cancellable for the SIGINT teardown test. + + ''' + if self.enabled: + self.receiving += 1 + try: + self.parked.set() + await self.release.wait() + finally: + self.receiving -= 1 + return await self.orig_recv.receive_exactly(size) + + +class StallableTransport: + ''' + Simulate stalled IPC sends and receives on one ``Channel``. + + Replace ``Channel._transport`` to pause sends before its + production shield, and replace ``MsgpackTransport.recv_stream`` + with a ``StallableReceiveStream`` for buffered packet reads. + Other transport operations delegate to the original + ``MsgpackTransport``. + The test can inspect each paused operation independently while + the registry and unrelated actor channels remain usable. + + ''' + def __init__( + self, + channel: tractor.Channel, + control_marker: str|None = None, + ) -> None: + ''' + Wrap one ``Channel``'s transport and buffered receiver. + + ''' + self.orig_tpt: MsgpackTransport = channel._transport + self.recv_stream: StallableReceiveStream = ( + StallableReceiveStream( + self.orig_tpt.recv_stream + ) + ) + self.send_enabled: bool = False + self.send_parked: trio.Event = trio.Event() + self.release_send: trio.Event = trio.Event() + self.sending: int = 0 + self.finite_send: bool = False + self.cancel_request: bool = False + self.control_marker: str|None = control_marker + self.orig_tpt.recv_stream = self.recv_stream + channel._transport = self + + def __getattr__( + self, + name: str, + ) -> Any: + ''' + Delegate other transport operations to the original. + + ''' + return getattr(self.orig_tpt, name) + + def __aiter__(self) -> AsyncIterator[dict]: + ''' + Delegate packet iteration to ``StallableTransport.orig_tpt``. + + Its packet reader uses the replaced ``recv_stream`` field. + + ''' + return self.orig_tpt.__aiter__() + + async def send( + self, + msg: Any, + send_deadline: float = float('inf'), + **kwargs: Any, + ) -> None: + ''' + Park a send before ``MsgpackTransport.send()`` enters its + shield. + + ``StallableTransport.send_parked`` tells the controller + when a real send is blocked. The cancellable wait respects + ``send_deadline``; a stalled cancel request can time out + and trigger runtime hard-reaping. + + ''' + if self.send_enabled: + self.sending += 1 + try: + self.finite_send = send_deadline != float('inf') + self.cancel_request = ( + isinstance(msg, Start) + and + msg.ns == 'self' + and + msg.func == 'cancel' + ) + self.send_parked.set() + with trio.CancelScope(deadline=send_deadline) as cs: + await self.release_send.wait() + if cs.cancelled_caught: + raise tractor.TransportClosed( + 'Test IPC send deadline expired' + ) + finally: + self.sending -= 1 + if self.control_marker: + emit( + marker=self.control_marker, + event='control_send_exit', + deadline_expired=( + trio.current_time() >= send_deadline + ), + ) + await self.orig_tpt.send( + msg, + send_deadline=send_deadline, + **kwargs, + ) + + def snapshot(self) -> dict[str, bool]: + ''' + Report which transport operations are still parked. + + ''' + return { + 'receive': bool(self.recv_stream.receiving), + 'send': bool(self.sending), + 'finite_send': self.finite_send, + } + + +@tractor.context +async def probe_peer( + ctx: tractor.Context, +) -> None: + ''' + Keep the peer context alive while the caller stalls IPC. + + Two responses finish any pending read in the original + receiver. ``MsgpackTransport._iter_packets()`` then reads + again through the replaced ``recv_stream`` field, even when + the previous response used buffered bytes. Only the external + controller's SIGINT may terminate this test context. + + ''' + await ctx.started(os.getpid()) + async with ctx.open_stream() as stream: + assert await stream.receive() == 'probe' + await stream.send('probe-1') + await stream.send('probe-2') + await trio.sleep_forever() + raise AssertionError('Peer stream completed before cancellation') + + +async def peer_address() -> UnwrappedAddress: + ''' + Provide the peer address for a distinct OOB context connection. + + ''' + return tractor.current_actor().accept_addrs[0] + + +def emit( + marker: str, + event: str, + **fields: Any, +) -> None: + ''' + Publish a JSON test event on the root's stdout pipe. + + The controller in ``tests/test_ipc_stalls.py`` uses ``marker`` + to distinguish child identity, readiness and final results + from actor logs. + + ''' + message: str = marker + json.dumps({ + 'event': event, + **fields, + }) + print(message, flush=True) + + +async def root_main( + # !TODO, explain each of these inputs with comments.. + mode: str, # ``test_stalled_channel_sigint`` case to simulate + port: int, # isolated root actor registry port + marker: str, # prefix for controller-visible JSON records + state: dict[str, Any], # child and parked-I/O result evidence +) -> None: + ''' + Park selected IPC operations in a test-owned actor tree. + + ``mode`` selects ``healthy_ipc``, ``stalled_context_receive``, + ``stalled_context_send``, ``stalled_context_send_and_receive``, + ``stalled_context_cancel_send``, or + ``stalled_context_and_subactor_cancel_send``. The last case + parks both context directions and the subactor ``Portal``'s + cancel request. ``port`` selects this root's registry address; + ``marker`` frames stdout events; ``state`` retains child and + parked-I/O evidence for the SIGINT result. + + ''' + async with tractor.open_root_actor( + name='ipc-stall-root', + registry_addrs=[('127.0.0.1', port)], + enable_transports=['tcp'], + ensure_registry=True, + start_method='trio', + debug_mode=False, + loglevel='error', + ): + async with tractor.open_nursery() as an: + staller_ptl = await an.start_actor( + 'ipc-stall-peer', + enable_modules=[__name__], + ) + proc: trio.Process = next( + iter(an._children.values()) + )[1] + state['child'] = proc + emit( + marker=marker, + event='child', + pid=proc.pid, + ) + peer_addr: UnwrappedAddress = await staller_ptl.run( + peer_address + ) + + # Open an OOB context ``Channel`` via ``_connect_chan`` + # and ``open_portal`` as ``peer_ptl``. The subactor + # ``Portal``'s ``staller_ptl.channel`` is distinct. + # + # Most modes leave subactor ``Portal`` cancel usable. + # ``stalled_context_and_subactor_cancel_send`` also + # stalls its cancel send AND both context directions. + async with ( + _connect_chan(peer_addr) as peer_channel, + open_portal(peer_channel) as peer_ptl, + peer_ptl.open_context(probe_peer) as (ctx, pid), + ): + # Match the peer PID from ``Context.started()``. + assert pid == proc.pid + # The OOB context connection must not silently + # stall the subactor ``Portal``'s channel as well. + assert ctx.chan is not staller_ptl.channel + stall_tpt: StallableTransport = StallableTransport( + ctx.chan + ) + stall_tpt.recv_stream.enabled = mode in { + # Suspend context receive only. + 'stalled_context_receive', + # Suspend context send and receive. + 'stalled_context_send_and_receive', + # Also stall the subactor ``Portal`` cancel send. + 'stalled_context_and_subactor_cancel_send', + } + await ctx.chan.send( + Yield(cid=ctx.cid, pld='probe') + ) + # Wait for the packet reader to park for each + # receive-stalled mode, not just wrapper setup. + # ?TODO, in what cases is this set vs. not? + # sure would be handy to see that from this layer.. + # ``stalled_context_receive``, + # ``stalled_context_send_and_receive``, and + # ``stalled_context_and_subactor_cancel_send`` wait + # here; other cases consume both probe responses. + if stall_tpt.recv_stream.enabled: + await stall_tpt.recv_stream.parked.wait() + else: + # Drain the peer's two responses in modes that + # do not intentionally stall context reads. + for expected in ('probe-1', 'probe-2'): + pld: str = await ctx.pld_rx.recv_pld( + ipc=ctx, + expect_msg=Yield, + ) + assert pld == expected + + stall_tpt.send_enabled = mode in { + # Suspend context send only. + 'stalled_context_send', + # Suspend both context directions. + 'stalled_context_send_and_receive', + # Suspend ``Context.cancel()``'s timed send. + 'stalled_context_cancel_send', + # Also suspend the subactor ``Portal`` cancel. + 'stalled_context_and_subactor_cancel_send', + } + control_tpt: StallableTransport|None = None + if mode == ( + 'stalled_context_and_subactor_cancel_send' + ): + control_tpt = StallableTransport( + staller_ptl.channel, + control_marker=marker, + ) + control_tpt.send_enabled = True + + async def ready() -> None: + ''' + Report selected parked I/O to the controller. + + Readiness requires real blocked sends/receives; + simply installing a wrapper would not prove a + stalled connection before SIGINT arrives. + + ''' + if stall_tpt.send_enabled: + await stall_tpt.send_parked.wait() + if control_tpt is not None: + await control_tpt.send_parked.wait() + assert control_tpt.cancel_request + assert control_tpt.finite_send + assert control_tpt.sending + state['control_fault_active'] = ( + control_tpt.snapshot() + ) + snapshot: dict[str, bool] = stall_tpt.snapshot() + assert snapshot['receive'] == ( + stall_tpt.recv_stream.enabled + ) + assert snapshot['send'] == stall_tpt.send_enabled + if mode == 'stalled_context_cancel_send': + assert snapshot['finite_send'] + assert proc.returncode is None + assert ctx.chan.connected() + state['stalls_fault_active'] = snapshot + emit( + marker=marker, + event='ready', + stalls=snapshot, + control=state.get( + 'control_fault_active' + ), + ) + + async with trio.open_nursery() as tn: + tn.start_soon(ready) + if mode == 'stalled_context_cancel_send': + try: + pass + finally: + # Actual teardown publication, with its + # normal finite deadline, not a sleep. + await ctx.cancel(timeout=5) + raise AssertionError( + 'Cleanup finished before SIGINT' + ) + if stall_tpt.send_enabled: + tn.start_soon( + ctx.chan.send, + Yield(cid=ctx.cid, pld='blocked'), + ) + if control_tpt is not None: + await staller_ptl.cancel_actor(timeout=5) + raise AssertionError( + 'Control cancel finished before SIGINT' + ) + await trio.sleep_forever() + raise AssertionError('Context completed before SIGINT') + + +def main() -> None: + ''' + Run the root in a process separate from the test controller. + + ``tests/test_ipc_stalls.py`` passes ``mode``, ``port``, and + ``marker`` as positional CLI args to run this example in a + separate process. It sends SIGINT without signalling ``pytest`` + and observes JSON events and process exit through stdout and + its OS process handle, not a potentially stalled ``Channel``. + Running this module directly defaults to + ``stalled_context_receive`` and selects an isolated registry + port with ``tractor._testing.addr.get_rando_addr()``. + + ''' + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + 'mode', nargs='?', default='stalled_context_receive', + choices=( + 'healthy_ipc', + 'stalled_context_receive', + 'stalled_context_send', + 'stalled_context_send_and_receive', + 'stalled_context_cancel_send', + 'stalled_context_and_subactor_cancel_send', + ), + ) + parser.add_argument('port', type=int, nargs='?') + parser.add_argument( + 'marker', nargs='?', default='IPC-STALL:', + ) + args = parser.parse_args() + port: int|None = args.port + if port is None: + _, port = get_rando_addr('tcp') + + state: dict[str, Any] = {} + try: + trio.run( + partial( + root_main, + mode=args.mode, + port=port, + marker=args.marker, + state=state, + ), + restrict_keyboard_interrupt_to_checkpoints=True, + ) + except KeyboardInterrupt: + proc: trio.Process = state['child'] + assert_child_reaped(proc) + # A returncode alone is insufficient: prove the runtime + # waited for its child rather than leaving a zombie behind. + try: + waited = os.waitpid(proc.pid, os.WNOHANG) + except ChildProcessError: + reaped = True + else: + raise AssertionError( + ('Runtime did not reap child', waited) + ) + emit( + marker=args.marker, + event='result', + root_exit='KeyboardInterrupt', + stalls_fault_active=state['stalls_fault_active'], + child_reaped=reaped, + child_returncode=proc.returncode, + control_fault_active=state.get('control_fault_active'), + ) + else: + raise AssertionError('Root exited without KeyboardInterrupt') + + +if __name__ == '__main__': + main() diff --git a/tests/test_ipc_stalls.py b/tests/test_ipc_stalls.py new file mode 100644 index 00000000..1476ab9f --- /dev/null +++ b/tests/test_ipc_stalls.py @@ -0,0 +1,402 @@ +''' +Externally supervised tests for cancellable IPC unresponsiveness. + +No kernel packet filtering: these waits model a task making no +progress. EOF remains enabled in production and is never injected. +The watchdog signals only its own root process handle. + +''' +from collections.abc import ( + Callable, + Iterator, +) +from contextlib import ( + contextmanager as cm, + suppress, +) +from dataclasses import ( + dataclass, + field, +) +import json +import os +from pathlib import Path +import selectors +import signal +import subprocess +import sys +import time +from typing import Any +from uuid import uuid4 + +import psutil +import pytest +from tractor._testing.addr import get_rando_addr + + +@dataclass +class RootMonitor: + ''' + Observe one stalled-IPC root outside its actor/IPC tree. + + ``RootMonitor.proc`` is the test-owned root subprocess. + ``RootMonitor.transcript`` retains plain logs; only lines + prefixed with ``RootMonitor.marker`` contain JSON records + from ``examples.advanced_faults.ipc_stalls.emit()``. + ``RootMonitor.child`` is learned from the peer's PID event, + so cleanup never selects a process by global name. + + ''' + # TODO: reuse for the ``daemon`` remote registrar in + # ``tests/discovery/conftest.py`` to verify logged discovery + # transactions from outside the actor tree. + # TODO: colocate scenario and monitor behind a reusable + # ``tractor._testing`` interface for other suites. + proc: subprocess.Popen[bytes] + marker: str + child: psutil.Process|None = None + transcript: list[str] = field(default_factory=list) + events: list[dict[str, Any]] = field(default_factory=list) + deadline: float = field( + default_factory=lambda: time.monotonic() + 20 + ) + + def iter_events(self) -> Iterator[dict[str, Any]]: + ''' + Yield marker-prefixed records until root stdout closes. + + Poll the independent pipe using ``RootMonitor.deadline``. + Keep plain actor logs in ``RootMonitor.transcript`` and + parse only ``emit()`` records into ``RootMonitor.events``. + A root that cannot publish readiness or finish after + SIGINT fails the external watchdog, regardless of IPC. + + ''' + stdout = self.proc.stdout + assert stdout is not None + pending: bytes = b'' + with selectors.DefaultSelector() as poller: + poller.register(stdout, selectors.EVENT_READ) + while poller.get_map(): + remaining: float = ( + self.deadline - time.monotonic() + ) + if remaining <= 0: + raise TimeoutError('External watchdog expired') + if not poller.select(remaining): + raise TimeoutError( + 'External watchdog expired' + ) + chunk: bytes = os.read(stdout.fileno(), 65536) + if not chunk: + poller.unregister(stdout) + break + pending += chunk + + while b'\n' in pending: + line, pending = pending.split(b'\n', 1) + text: str = line.decode(errors='replace') + self.transcript.append(text) + if not text.startswith(self.marker): + continue + event: dict[str, Any] = json.loads( + text[len(self.marker):] + ) + self.events.append(event) + yield event + + def signal_root( + self, + sig: int = signal.SIGINT, + ) -> None: + ''' + Signal the owned root; bound the ensuing shutdown. + + Default to graceful SIGINT cancellation. ``sig`` also allows + future scenarios to exercise non-graceful termination. + + ''' + self.proc.send_signal(sig) + self.deadline = time.monotonic() + 20 + + +@cm +def open_root_monitor( + mode: str, +) -> Iterator[RootMonitor]: + ''' + Own one isolated actor program and rescue only failed runs. + + Launch ``examples.advanced_faults.ipc_stalls`` as a separate + process with a per-test registry port and stdout control pipe. + Keep ``pytest`` outside both potentially stalled IPC channels + so it can signal, time out, and inspect the root's OS exit. + Failure cleanup touches only handles created by this context; + watchdog rescue never counts as runtime teardown success. + + ''' + root: Path = Path(__file__).resolve().parents[1] + marker: str = f'IPC-STALL-{uuid4().hex}:' + # Pure address selection, without ``reg_addr`` runtime overrides. + # ``ensure_registry`` refuses a collision instead of attaching. + _, port = get_rando_addr('tcp') + + env: dict[str, str] = dict(os.environ) + env['PYTHONPATH'] = os.pathsep.join(filter(None, ( + str(root), env.get('PYTHONPATH'), + ))) + env['PYTHONDONTWRITEBYTECODE'] = '1' + # TODO: generalize the scenario entrypoint together with the + # ``daemon`` fixture's spawner in + # ``tests/discovery/conftest.py``. + proc: subprocess.Popen[bytes] = subprocess.Popen( + [ + sys.executable, + '-u', + '-m', 'examples.advanced_faults.ipc_stalls', + mode, + str(port), + marker, + ], + cwd=root, + env=env, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + monitor = RootMonitor(proc=proc, marker=marker) + try: + yield monitor + except BaseException as exc: + output: str = '\n'.join(monitor.transcript) + # XXX i tweaked this for newlines + exc.add_note( + f'Controller events:\n' + f'{monitor.events!r}\n' + f'{output}' + ) + raise + finally: + # Failure-only rescue: retained test-owned PIDs, never a + # global reaper, process group, or unrelated resource sweep. + owned: list[psutil.Process] = ( + [monitor.child] if monitor.child else [] + ) + if proc.poll() is None: + # Include children if root stalled before its PID event. + with suppress(psutil.NoSuchProcess): + owned.extend( + psutil.Process(proc.pid).children( + recursive=True, + ) + ) + + for descendant in set(owned): + with suppress(psutil.NoSuchProcess): + if descendant.is_running(): + descendant.kill() + + if proc.poll() is None: + proc.send_signal(signal.SIGINT) + try: + proc.wait(timeout=3) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=3) + assert proc.stdout is not None + proc.stdout.close() + + _, survivors = psutil.wait_procs(owned, timeout=3) + assert not survivors, ( + 'Watchdog cleanup survivors', survivors, + ) + + +@pytest.mark.skipif( + sys.platform != 'linux', + reason='Linux watchdog', +) +@pytest.mark.parametrize( + 'mode', + [ + # Baseline: both context and subactor ``Portal`` IPC work. + 'healthy_ipc', + # Stall directions on the separate OOB context ``Channel``. + 'stalled_context_receive', + 'stalled_context_send', + 'stalled_context_send_and_receive', + # Park ``Context.cancel()`` on that OOB context ``Channel``. + 'stalled_context_cancel_send', + # Both context directions plus subactor ``Portal`` cancel. + 'stalled_context_and_subactor_cancel_send', + ], +) +def test_stalled_channel_sigint( + mode: str, + # ``pytest`` provides ``record_property`` via _pytest.junitxml. + record_property: Callable[[str, object], None], +) -> None: + ''' + SIGINT must tear down a root with stalled child IPC. + + The example's ``StallableReceiveStream.receive_exactly()`` sets + ``StallableReceiveStream.parked`` before waiting on + ``StallableReceiveStream.release``. Peer responses complete any + pending read inside the original ``BufferedReceiveStream``. + The reader task's next header or payload read accesses the + replaced ``MsgpackTransport.recv_stream`` and parks on that + event; this does not flush buffered bytes. + ``StallableTransport.send()`` sets + ``StallableTransport.send_parked`` before waiting on + ``StallableTransport.release_send``. + + In ``examples.advanced_faults.ipc_stalls.root_main()``, the root + actor's ``ready()`` closure waits for selected parked-operation + events and emits ready JSON on stdout. ``RootMonitor`` checks + that record before delivering SIGINT; readiness does not depend + on communicating over a root ``Portal``. + + CASES: (parametrized by ``mode``) + + - Subactor ``Portal`` usable: ``staller_ptl`` comes from + ``an.start_actor()``. The distinct OOB context ``peer_ptl`` + uses a ``Channel`` from ``_connect_chan()`` and + ``open_portal()``. + These five cases allow cooperative child cancellation: + + - ``healthy_ipc``: probe traffic, with no stalled operations. + - ``stalled_context_receive``: park context packet reads. + - ``stalled_context_send``: park a context send. + - ``stalled_context_send_and_receive``: park both directions. + - ``stalled_context_cancel_send``: park the finite-deadline + ``Context.cancel()`` send during actual teardown. + + - Subactor ``Portal`` stalled: + + - ``stalled_context_and_subactor_cancel_send``: park both + context directions AND ``staller_ptl``'s finite-deadline + ``Start(self.cancel)`` send. Require deadline expiration, + runtime hard-reap logs, SIGKILL exit and completed waitpid. + + ``open_root_monitor()`` observes OS lifetime outside the actor + tree. A ``Portal`` within the root cannot independently enforce + a watchdog deadline or prove OS exit over stalled IPC. Missing + readiness, premature completion, root timeout or an unreaped + child fails this test; watchdog rescue cannot count as success. + Kernel TCP half-open/DROP scenarios remain follow-up work. + + ''' + signalled: bool = False + result: dict[str, Any]|None = None + stalls: dict[str, bool]|None = None + with open_root_monitor(mode=mode) as monitor: + proc: subprocess.Popen[bytes] = monitor.proc + for event in monitor.iter_events(): + if event['event'] == 'child': + # Retain one OS identity and verify root ownership; + # later exit checks must concern this exact child. + assert monitor.child is None + monitor.child = psutil.Process(event['pid']) + assert monitor.child.ppid() == proc.pid + + elif event['event'] == 'ready': + # Prove the child is live and selected operations + # really parked before sending the sole SIGINT. + assert not signalled + child = monitor.child + assert child is not None + assert child.is_running() + stalls = event['stalls'] + assert stalls['receive'] == ( + mode in { + 'stalled_context_receive', + 'stalled_context_send_and_receive', + 'stalled_context_and_subactor_cancel_send', + } + ) + assert stalls['send'] == ( + mode in { + 'stalled_context_send', + 'stalled_context_send_and_receive', + 'stalled_context_cancel_send', + 'stalled_context_and_subactor_cancel_send', + } + ) + assert stalls['finite_send'] == ( + mode == 'stalled_context_cancel_send' + ) + if mode == ( + 'stalled_context_and_subactor_cancel_send' + ): + assert event['control'] == { + 'receive': False, + 'send': True, + 'finite_send': True, + } + assert proc.poll() is None + monitor.signal_root() + signalled = True + + elif event['event'] == 'result': + # Accept exactly one shutdown result, only after + # SIGINT; voluntary early completion is a failure. + assert signalled + assert result is None + result = event + + elif event['event'] == 'control_send_exit': + # Only this case parks the subactor ``Portal`` send. + # Its finite-budget exit must ultimately time out; + # the final checks below require that evidence. + assert mode == ( + 'stalled_context_and_subactor_cancel_send' + ) + + else: + raise AssertionError(event) + + # The root caught SIGINT and reported runtime waitpid proof; + # these checks precede any failure-only watchdog rescue. + assert proc.wait(timeout=2) == 0 + assert signalled, 'Root completed before signal readiness' + assert result is not None + assert stalls is not None + child = monitor.child + assert child is not None + assert result['root_exit'] == 'KeyboardInterrupt' + assert result['child_reaped'] is True + control_expired: bool = any( + event.get('deadline_expired', False) + for event in monitor.events + ) + if mode == 'stalled_context_and_subactor_cancel_send': + # Both context directions and the subactor ``Portal`` + # cancel send stalled: require actual budget expiry, + # SIGKILL status and the runtime's hard-reap log. + assert control_expired + assert result['child_returncode'] == -signal.SIGKILL + assert result['control_fault_active'] == { + 'receive': False, + 'send': True, + 'finite_send': True, + } + assert any( + 'escalating to `proc.kill()` (hard-reap)' in line + for line in monitor.transcript + ), 'Missing runtime hard-reap evidence' + else: + # The baseline and context-only faults retain a usable + # subactor ``Portal``: cooperative child exit is clean. + assert result['child_returncode'] == 0 + assert result['stalls_fault_active'] == stalls + assert not child.is_running() + record_property('ipc_stall_result', json.dumps({ + **result, + 'mode': mode, + 'root_returncode': proc.returncode, + 'watchdog_forced_cleanup': False, + 'runtime_hard_reap': mode == ( + 'stalled_context_and_subactor_cancel_send' + ), + 'control_deadline_expired': control_expired, + }))