diff --git a/examples/advanced_faults/ipc_failure_during_stream.py b/examples/advanced_faults/ipc_failure_during_stream.py index 4242a543..1a17c335 100644 --- a/examples/advanced_faults/ipc_failure_during_stream.py +++ b/examples/advanced_faults/ipc_failure_during_stream.py @@ -1,111 +1,78 @@ ''' -Complex edge case where during real-time streaming the IPC tranport -channels are wiped out (purposely in this example though it could have -been an outage) and we want to ensure that despite being in debug mode -(or not) the user can sent SIGINT once they notice the hang and the -actor tree will eventually be cancelled without leaving any zombies. +Detect a closed IPC transport during real-time streaming. + +Detected closure must fail an unfinished context promptly, including +when the peer sent a stream stop first. Only the no-break example +simulates a user pressing Ctrl-C to cancel a healthy actor tree. ''' from contextlib import asynccontextmanager as acm -from functools import partial from tractor import ( open_nursery, context, Context, - ContextCancelled, - MsgStream, _testing, - trionics, - TransportClosed, ) import trio -import pytest - - -async def break_ipc_then_error( - stream: MsgStream, - break_ipc_with: str|None = None, - pre_close: bool = False, -): - await _testing.break_ipc( - stream=stream, - method=break_ipc_with, - pre_close=pre_close, - ) - async for msg in stream: - await stream.send(msg) - - assert 0 - - -async def iter_ipc_stream( - stream: MsgStream, - break_ipc_with: str|None = None, - pre_close: bool = False, -): - async for msg in stream: - await stream.send(msg) @context async def recv_and_spawn_net_killers( - ctx: Context, break_ipc_after: bool|int = False, pre_close: bool = False, - ) -> None: ''' - Receive stream msgs and spawn some IPC killers mid-stream. + Echo messages until a single, sequence-numbered IPC break. + + Break before echoing the trigger so the caller is waiting for + that reply, rather than racing another send against peer close. + Keep this context unfinished: stream stop is not a final result. ''' - broke_ipc: bool = False await ctx.started() - async with ( - ctx.open_stream() as stream, - trionics.collapse_eg(), - trio.open_nursery() as tn, - ): + async with ctx.open_stream() as stream: async for i in stream: - print(f'child echoing {i}') - if not broke_ipc: - await stream.send(i) - else: - await trio.sleep(0.01) - if ( break_ipc_after and i >= break_ipc_after ): - broke_ipc = True - tn.start_soon( - iter_ipc_stream, - stream, - ) - tn.start_soon( - partial( - break_ipc_then_error, - stream=stream, - pre_close=pre_close, - ) + if pre_close: + await stream.aclose() + # Protocol Stop closes the stream, not its IPC. + assert stream.closed + assert ctx.chan.connected() + await _testing.break_ipc( + stream=stream, + pre_close=False, ) + await trio.sleep_forever() + + print(f'child echoing {i}') + await stream.send(i) @acm async def stuff_hangin_ctlc(timeout: float = 1) -> None: + ''' + Simulate Ctrl-C when a healthy stream makes no progress. + Detected-fault runs use a failing deadline instead, so a timeout + cannot masquerade as successful transport-failure propagation. + + ''' with trio.move_on_after(timeout) as cs: yield timeout if cs.cancelled_caught: - # pretend to be a user seeing no streaming action - # thinking it's a hang, and then hitting ctl-c.. + # Pretend to be a user seeing no streaming action, + # thinking it's a hang, and then hitting Ctrl-C. print( - f"i'm a user on the PARENT side and thingz hangin " - f'after timeout={timeout} ???\n\n' - 'MASHING CTlR-C..!?\n' + f'PARENT streaming stalled after {timeout}s\n' + f'\n' + f'MASHING Ctrl-C..!\n' ) raise KeyboardInterrupt @@ -115,153 +82,85 @@ async def main( start_method: str = 'trio', loglevel: str = 'cancel', - # by default we break the parent IPC first (if configured to break - # at all), but this can be changed so the child does first (even if - # both are set to break). + # By default the parent breaks first, if configured to break + # at all. Reverse the thresholds to make the child break first. break_parent_ipc_after: int|bool = False, break_child_ipc_after: int|bool = False, pre_close: bool = False, tpt_proto: str = 'tcp', - + reg_addr: tuple[str, int|str]|None = None, ) -> None: + ''' + Run a bounded close scenario or demonstrate Ctrl-C cancellation. - async with ( - open_nursery( - start_method=start_method, + Tests pass their unique registrar address explicitly. Standalone + invocation retains the runtime's default address selection. - # NOTE: even debugger is used we shouldn't get - # a hang since it never engages due to broken IPC - debug_mode=debug_mode, - loglevel=loglevel, - enable_transports=[tpt_proto], + ''' + registry_kwargs = {} + if reg_addr is not None: + registry_kwargs = { + 'registry_addrs': [reg_addr], + 'ensure_registry': True, + } - ) as an, - ): - sub_name: str = 'chitty_hijo' + async with open_nursery( + start_method=start_method, + # Even with debugging enabled, broken IPC must not hang + # trying to engage a debugger through the closed channel. + debug_mode=debug_mode, + loglevel=loglevel, + enable_transports=[tpt_proto], + **registry_kwargs, + ) as an: portal = await an.start_actor( - sub_name, + 'chitty_hijo', enable_modules=[__name__], ) + breaking = ( + break_parent_ipc_after + or + break_child_ipc_after + ) + async with portal.open_context( + recv_and_spawn_net_killers, + break_ipc_after=break_child_ipc_after, + pre_close=pre_close, + ) as (ctx, sent): + # A transport failure is not a remote cancellation RPC; + # do not require ContextCancelled or invent a canceller. + with trio.fail_after(3): + async with ctx.open_stream() as stream: + for i in range(1000): + if ( + break_parent_ipc_after + and + i >= break_parent_ipc_after + ): + # TODO: other methods? See break_ipc(). + # await ctx.chan.send(None) + socket = ctx.chan.transport.stream + # await socket.send_eof() + await socket.aclose() + assert not ctx.chan.connected() + await trio.sleep_forever() - async with ( - stuff_hangin_ctlc(timeout=2) as timeout, - _testing.expect_ctxc( - yay=( - break_parent_ipc_after - or - break_child_ipc_after - ), - # TODO: we CAN'T remove this right? - # since we need the ctxc to bubble up from either - # the stream API after the `None` msg is sent - # (which actually implicitly cancels all remote - # tasks in the hijo) or from simluated - # KBI-mash-from-user - # or should we expect that a KBI triggers the ctxc - # and KBI in an eg? - reraise=True, - ), - - portal.open_context( - recv_and_spawn_net_killers, - break_ipc_after=break_child_ipc_after, - pre_close=pre_close, - ) as (ctx, sent), - ): - rx_eoc: bool = False - ipc_break_sent: bool = False - async with ctx.open_stream() as stream: - for i in range(1000): - - if ( - break_parent_ipc_after - and - i > break_parent_ipc_after - and - not ipc_break_sent - ): - print( - '#################################\n' - 'Simulating PARENT-side IPC BREAK!\n' - '#################################\n' - ) - - # TODO: other methods? see break func above. - # await stream._ctx.chan.send(None) - # await stream._ctx.chan.transport.stream.send_eof() - await stream._ctx.chan.transport.stream.aclose() - ipc_break_sent = True - - # it actually breaks right here in the - # mp_spawn/forkserver backends and thus the - # zombie reaper never even kicks in? - try: - print(f'parent sending {i}') await stream.send(i) - except ContextCancelled as ctxc: - print( - 'parent received ctxc on `stream.send()`\n' - f'{ctxc}\n' - ) - assert 'root' in ctxc.canceller - assert sub_name in ctx.canceller - - # TODO: is this needed or no? - raise - - except ( - trio.ClosedResourceError, - TransportClosed, - ) as _tpt_err: - # NOTE: don't send if we already broke the - # connection to avoid raising a closed-error - # such that we drop through to the ctl-c - # mashing by user. - with trio.CancelScope(shield=True): - await trio.sleep(0.01) - - # timeout: int = 1 - # with trio.move_on_after(timeout) as cs: - async with stuff_hangin_ctlc() as timeout: - print( - f'PARENT `stream.receive()` with timeout={timeout}\n' - ) - # NOTE: in the parent side IPC failure case this - # will raise an ``EndOfChannel`` after the child - # is killed and sends a stop msg back to it's - # caller/this-parent. - try: + if breaking: rx = await stream.receive() - print( - "I'm a happy PARENT user and echoed to me is\n" - f'{rx}\n' - ) - except trio.EndOfChannel: - rx_eoc: bool = True - print('MsgStream got EoC for PARENT') - raise + else: + async with stuff_hangin_ctlc(): + rx = await stream.receive() + assert rx == i - print( - 'Streaming finished and we got Eoc.\n' - 'Canceling `.open_context()` in root with\n' - 'CTlR-C..' - ) - if rx_eoc: - assert stream.closed - try: - await stream.send(i) - pytest.fail('stream not closed?') - except ( - TransportClosed, - trio.ClosedResourceError, - trio.EndOfChannel, - ) as send_err: - if rx_eoc: - assert send_err is stream._eoc - else: - assert send_err is stream._closed + if breaking: + # open_stream() absorbs a peer's stop/EoC, but + # open_context() still needs a result. EOF must + # wake this wait without a later user SIGINT. + await ctx.wait_for_result() + raise AssertionError('broken ctx returned?') - raise KeyboardInterrupt + raise KeyboardInterrupt if __name__ == '__main__': diff --git a/tests/test_advanced_faults.py b/tests/test_advanced_faults.py index 009e25b0..505d44e8 100644 --- a/tests/test_advanced_faults.py +++ b/tests/test_advanced_faults.py @@ -3,7 +3,7 @@ Sketchy network blackoutz, ugly byzantine gens, puedes eschuchar la cancelacion?.. ''' -from functools import partial +from contextlib import asynccontextmanager as acm from types import ModuleType import pytest @@ -11,57 +11,184 @@ from _pytest.pathlib import import_path import trio import tractor from tractor import ( + Portal, TransportClosed, ) +from tractor.runtime._supervise import ActorNursery from tractor._testing import ( examples_dir, break_ipc, + assert_child_reaped, ) +@pytest.fixture +def reaped_children(monkeypatch: pytest.MonkeyPatch): + ''' + Retain spawned process handles and check synchronous reaping. + + Snapshot each ActorNursery._children entry at publication, before + failure teardown removes it. Check both backend completion and + PID disappearance after trio.run(), without an external reaper. + + ''' + procs = [] + start_actor = ActorNursery.start_actor + + async def record_child(an, *args, **kwargs): + ''' + Record only children spawned by this test's actor nursery. + + ''' + portal = await start_actor(an, *args, **kwargs) + procs.append(an._children[portal.channel.aid.uid][1]) + return portal + + monkeypatch.setattr(ActorNursery, 'start_actor', record_child) + yield procs + for proc in procs: + assert_child_reaped(proc) + + +@pytest.fixture +def fault_contexts(monkeypatch: pytest.MonkeyPatch): + ''' + Observe fault state without altering cancellation or propagation. + + Collect the entered `tractor.Context`s in a `list` delivered + to the consuming test. This allows introspecting a ctx's recorded + fault even if teardown incorrectly swallows that failure. + + On teardown ensure transport-loss does not set + `ContextStatus._canceller`: connection loss alone supplies no + remote cancellation or task-error provenance. For the simulated + user-interrupt cases, ensure `Context.cancel_called` is True + because teardown requested cancellation of the remote task. + + ''' + contexts: list[tractor.Context] = [] + open_context = Portal.open_context + + @acm + async def record_context(portal, func, **kwargs): + ''' + Hook into all `Portal.open_context() as pair` calls, stashing + their contexts in a `list` for consumption by dependent + tests for introspection. + + ''' + ctx: tractor.Context|None = None + try: + async with open_context( + portal, + func, + **kwargs, + ) as (ctx, sent): + if func.__name__ in { + 'recv_and_spawn_net_killers', + 'break_ipc_after_started', + }: + contexts.append(ctx) + yield (ctx, sent) + finally: + if ( + ctx in contexts + and + isinstance(ctx.maybe_error, TransportClosed) + ): + # MsgStream closure is distinct from transport loss. + # After teardown, send prioritizes the ctx error, + # while receive retains the stream's terminal marker. + stream = ctx._stream + assert stream.closed + assert not ctx.chan.connected() + with pytest.raises(TransportClosed) as err: + await stream.send('after-close') + assert err.value is ctx.maybe_error + terminal = stream._eoc or stream._closed + assert isinstance(terminal, BaseException) + with pytest.raises(type(terminal)) as err: + await stream.receive() + assert err.value is terminal + + monkeypatch.setattr(Portal, 'open_context', record_context) + yield contexts + for ctx in contexts: + if isinstance(ctx.maybe_error, TransportClosed): + assert ctx.canceller is None + elif isinstance(ctx._status._local_error, KeyboardInterrupt): + assert ctx.cancel_called + + +def assert_transport_closed( + exc: TransportClosed|BaseExceptionGroup, + local: bool, +) -> tuple[TransportClosed, ...]: + ''' + Validate every failure leaf, including its socket-level source. + + Parent close interrupts receive with ClosedResourceError. + Peer close can report clean EOF or a reset, depending on buffered + traffic. No arbitrary exception or remote cancellation qualifies. + + ''' + if isinstance(exc, BaseExceptionGroup): + return tuple( + fault + for leaf in exc.exceptions + for fault in assert_transport_closed(leaf, local=local) + ) + + assert isinstance(exc, TransportClosed) + cause = exc.__cause__ + assert exc.src_exc is cause + if local: + assert isinstance(cause, trio.ClosedResourceError) + assert 'closed locally' in str(exc) + elif cause is None: + assert 'gracefully closed' in str(exc) + else: + assert 'closed by peer' in str(exc) + assert isinstance( + cause, + (ValueError, trio.BrokenResourceError, + ConnectionResetError), + ) + if isinstance(cause, ValueError): + assert 'unclean EOF' in str(cause) + + # TODO, should we expect a certain exc-message (per tpt) too? + # e.g. cause.args[0] == 'another task closed this fd' + return (exc,) + + @pytest.mark.parametrize( 'pre_aclose_msgstream', - [ - False, - True, - ], - ids=[ - 'no_msgstream_aclose', - 'pre_aclose_msgstream', - ], + [False, True], + ids=['no_msgstream_aclose', 'pre_aclose_msgstream'], ) @pytest.mark.parametrize( 'ipc_break', [ - # no breaks - { - 'break_parent_ipc_after': False, - 'break_child_ipc_after': False, - }, + # No breaks: retain the healthy Ctrl-C cancellation contract. + dict(break_parent_ipc_after=False, + break_child_ipc_after=False), - # only parent breaks - { - 'break_parent_ipc_after': 500, - 'break_child_ipc_after': False, - }, + # Only parent breaks. + dict(break_parent_ipc_after=500, + break_child_ipc_after=False), - # only child breaks - { - 'break_parent_ipc_after': False, - 'break_child_ipc_after': 500, - }, + # Only child breaks. + dict(break_parent_ipc_after=False, + break_child_ipc_after=500), - # both: break parent first - { - 'break_parent_ipc_after': 500, - 'break_child_ipc_after': 800, - }, - # both: break child first - { - 'break_parent_ipc_after': 800, - 'break_child_ipc_after': 500, - }, + # Both configured: parent breaks first. + dict(break_parent_ipc_after=500, + break_child_ipc_after=800), + # Both configured: child breaks first. + dict(break_parent_ipc_after=800, + break_child_ipc_after=500), ], ids=[ 'no_break', @@ -75,238 +202,170 @@ def test_ipc_channel_break_during_stream( debug_mode: bool, loglevel: str, spawn_backend: str, - ipc_break: dict|None, + ipc_break: dict, pre_aclose_msgstream: bool, tpt_proto: str, + reg_addr: tuple[str, int|str], + reaped_children, + fault_contexts: list[tractor.Context], ): ''' - Ensure we can have an IPC channel break its connection during - streaming and it's still possible for the (simulated) user to kill - the actor tree using SIGINT. + Detect local/peer closure without waiting for a later Ctrl-C. - We also verify the type of connection error expected in the parent - depending on which side if the IPC breaks first. + The old scenario required ContextCancelled and a hang even when + EOF had already notified the unfinished context. Sequence-number + thresholds choose the first breaker; the child closes before its + trigger reply, preventing the parent from advancing past it. + + With pre-close, MsgStream absorbs the peer's `Stop`, but pending + Context.wait_for_result() must fail on transport loss. Check each + exception leaf and its source, bound teardown, and verify the + child is gone synchronously. No-break runs still raise synthetic + KeyboardInterrupt and reap the child through normal cancellation. + + If Context._scope absorbs EOF-driven task cancellation, teardown + must raise the recorded fault despite a disconnected channel. + That path must not manufacture a cancel request or peer ack, and + must preserve the original TransportClosed object and its cause. + + The fixture checks stream closure separately from transport + connectivity after unwinding, then retries send and receive to + verify their distinct error precedence and terminal identities. ''' - if spawn_backend != 'trio': - if debug_mode: - pytest.skip('`debug_mode` only supported on `trio` spawner') - - # non-`trio` spawners should never hit the hang condition that - # requires the user to do ctl-c to cancel the actor tree. - # expect_final_exc = trio.ClosedResourceError - expect_final_exc = TransportClosed + if spawn_backend != 'trio' and debug_mode: + pytest.skip('debug_mode only supported on trio spawner') mod: ModuleType = import_path( - examples_dir() - / 'advanced_faults' + examples_dir() / 'advanced_faults' / 'ipc_failure_during_stream.py', root=examples_dir(), consider_namespace_packages=False, ) - - # by def we expect KBI from user after a simulated "hang - # period" wherein the user eventually hits ctl-c to kill the - # root-actor tree. - expect_final_exc: BaseException = KeyboardInterrupt - expect_final_cause: BaseException|None = None - - if ( - # only expect EoC if trans is broken on the child side, - ipc_break['break_child_ipc_after'] is not False + parent = ipc_break['break_parent_ipc_after'] + child = ipc_break['break_child_ipc_after'] + breaking = parent or child + local = bool( + parent and - # AND we tell the child to call `MsgStream.aclose()`. - pre_aclose_msgstream - ): - # expect_final_exc = trio.EndOfChannel - # ^XXX NOPE! XXX^ since now `.open_stream()` absorbs this - # gracefully! - expect_final_exc = KeyboardInterrupt - - # NOTE when ONLY the child breaks or it breaks BEFORE the - # parent we expect the parent to get a closed resource error - # on the next `MsgStream.receive()` and then fail out and - # cancel the child from there. - # - # ONLY CHILD breaks - if ( - ipc_break['break_child_ipc_after'] - and - ipc_break['break_parent_ipc_after'] is False - ): - # NOTE: we DO NOT expect this any more since - # the child side's channel will be broken silently - # and nothing on the parent side will indicate this! - # expect_final_exc = trio.ClosedResourceError - - # NOTE: child will send a 'stop' msg before it breaks - # the transport channel BUT, that will be absorbed by the - # `ctx.open_stream()` block and thus the `.open_context()` - # should hang, after which the test script simulates - # a user sending ctl-c by raising a KBI. - if pre_aclose_msgstream: - expect_final_exc = KeyboardInterrupt - - # XXX OLD XXX - # if child calls `MsgStream.aclose()` then expect EoC. - # ^ XXX not any more ^ since eoc is always absorbed - # gracefully and NOT bubbled to the `.open_context()` - # block! - # expect_final_exc = trio.EndOfChannel - - # BOTH but, CHILD breaks FIRST - elif ( - ipc_break['break_child_ipc_after'] is not False - and ( - ipc_break['break_parent_ipc_after'] - > - ipc_break['break_child_ipc_after'] + ( + (not child) + or + (parent < child) ) - ): - if pre_aclose_msgstream: - expect_final_exc = KeyboardInterrupt + ) - # NOTE when the parent IPC side dies (even if the child does as well - # but the child fails BEFORE the parent) we always expect the - # IPC layer to raise a closed-resource, NEVER do we expect - # a stop msg since the parent-side ctx apis will error out - # IMMEDIATELY before the child ever sends any 'stop' msg. - # - # ONLY PARENT breaks - elif ( - ipc_break['break_parent_ipc_after'] - and - ipc_break['break_child_ipc_after'] is False - ): - expect_final_exc = tractor.TransportClosed - expect_final_cause = trio.ClosedResourceError + async def run(): + ''' + Bound the example, including actor nursery teardown. - # BOTH but, PARENT breaks FIRST - elif ( - ipc_break['break_parent_ipc_after'] is not False - and ( - ipc_break['break_child_ipc_after'] - > - ipc_break['break_parent_ipc_after'] - ) - ): - expect_final_exc = tractor.TransportClosed - expect_final_cause = trio.ClosedResourceError - - with pytest.raises( - expected_exception=( - expect_final_exc, - ExceptionGroup, - ), - ) as excinfo: - try: - trio.run( - partial( - mod.main, - debug_mode=debug_mode, - start_method=spawn_backend, - loglevel=loglevel, - pre_close=pre_aclose_msgstream, - tpt_proto=tpt_proto, - **ipc_break, - ) - ) - except KeyboardInterrupt as _kbi: - kbi = _kbi - if expect_final_exc is not KeyboardInterrupt: - pytest.fail( - 'Rxed unexpected KBI !?\n' - f'{repr(kbi)}' - ) - - raise - - except tractor.TransportClosed as _tc: - tc = _tc - if expect_final_exc is KeyboardInterrupt: - pytest.fail( - 'Unexpected transport failure !?\n' - f'{repr(tc)}' - ) - cause: Exception = tc.__cause__ - assert ( - # type(cause) is trio.ClosedResourceError - type(cause) is expect_final_cause - - # TODO, should we expect a certain exc-message (per - # tpt) as well?? - # and - # cause.args[0] == 'another task closed this fd' + ''' + with trio.fail_after(10): + await mod.main( + debug_mode=debug_mode, + start_method=spawn_backend, + loglevel=loglevel, + pre_close=pre_aclose_msgstream, + tpt_proto=tpt_proto, + reg_addr=reg_addr, + **ipc_break, ) - raise - - # get raw instance from pytest wrapper - value = excinfo.value - if isinstance(value, ExceptionGroup): - excs: tuple[Exception] = value.exceptions - assert ( - len(excs) <= 2 - and - all( - isinstance(exc, TransportClosed) - for exc in excs + if not breaking: + with pytest.raises(KeyboardInterrupt): + trio.run(run) + else: + with pytest.raises((TransportClosed, ExceptionGroup)) as err: + trio.run(run) + ctx = fault_contexts[0] + pytest.fail( + 'Transport failure was swallowed:\n' + f'{ctx.repr_outcome()}\n' + f'error={ctx.maybe_error!r}' ) - ) - final_exc = excs[0] - assert isinstance(final_exc, expect_final_exc) + faults = assert_transport_closed(err.value, local=local) + ctx = fault_contexts[0] + assert any(fault is ctx.maybe_error for fault in faults) + if ctx._scope.cancelled_caught: + error = ctx._status._remote_error + assert any(fault is error for fault in faults) + assert ctx._status._local_error is error + assert not ctx.cancel_called + assert not ctx.cancel_acked + assert len(reaped_children) == len(fault_contexts) == 1 @tractor.context async def break_ipc_after_started( ctx: tractor.Context, ) -> None: + ''' + Wait for the caller's stream barrier, then close unfinished IPC. + + ''' await ctx.started() async with ctx.open_stream() as stream: + assert await stream.receive() == 'ready' # TODO: make a test which verifies the error # for this, i.e. raises a `MsgTypeError` # await ctx.chan.send(None) - await break_ipc( - stream=stream, - pre_close=True, - ) - print('child broke IPC and terminating') + await break_ipc(stream=stream, pre_close=True) + await trio.sleep_forever() -def test_stream_closed_right_after_ipc_break_and_zombie_lord_engages(): +def test_stream_closed_right_after_ipc_break_and_zombie_lord_engages( + reg_addr: tuple[str, int|str], + tpt_proto: str, + spawn_backend: str, + loglevel: str, + reaped_children, + fault_contexts, +): ''' - Verify that is a subactor's IPC goes down just after bringing up - a stream the parent can trigger a SIGINT and the child will be - reaped out-of-IPC by the localhost process supervision machinery: - aka "zombie lord". + Reap a child after detected close without a later user SIGINT. + + Previously the parent slept, timed out its context, then raised + KeyboardInterrupt to engage process supervision ("zombie lord"). + EOF now notifies the unfinished context immediately. A ready + message ensures both streams exist before the child sends `Stop` + and closes IPC; the caller then waits for the missing result. + + A bounded TransportClosed exit proves fault-driven teardown, and + reaped_children checks process completion and PID disappearance + synchronously. Timer-driven cancellation cannot pass this test. ''' async def main(): - with trio.fail_after(3): - async with tractor.open_nursery() as an: + ''' + Await transport failure and bound the entire nursery exit. + + ''' + with trio.fail_after(10): + async with tractor.open_nursery( + registry_addrs=[reg_addr], + ensure_registry=True, + enable_transports=[tpt_proto], + start_method=spawn_backend, + loglevel=loglevel, + ) as an: portal = await an.start_actor( 'ipc_breaker', enable_modules=[__name__], ) + async with portal.open_context( + break_ipc_after_started, + ) as (ctx, sent): + async with ctx.open_stream() as stream: + await stream.send('ready') + await stream.receive() + await ctx.wait_for_result() + pytest.fail('broken context returned a result') - with trio.move_on_after(1): - async with ( - portal.open_context( - break_ipc_after_started - ) as (ctx, sent), - ): - async with ctx.open_stream(): - await trio.sleep(0.5) - - print('parent waiting on context') - - print( - 'parent exited context\n' - 'parent raising KBI..\n' - ) - raise KeyboardInterrupt - - with pytest.raises(KeyboardInterrupt): + with pytest.raises((TransportClosed, ExceptionGroup)) as err: trio.run(main) + faults = assert_transport_closed(err.value, local=False) + ctx = fault_contexts[0] + assert any(fault is ctx.maybe_error for fault in faults) + assert len(reaped_children) == len(fault_contexts) == 1 diff --git a/tests/test_ipc_eof.py b/tests/test_ipc_eof.py index 2cc30ca7..4587247c 100644 --- a/tests/test_ipc_eof.py +++ b/tests/test_ipc_eof.py @@ -19,7 +19,8 @@ _streams: dict[str, tractor.Context] = {} @tractor.context async def stream_peer( ctx: tractor.Context, -) -> None: + result_value: None|bool|int|str = None, +) -> None|bool|int|str: ''' Publish a live stream and echo until its caller finishes. @@ -32,6 +33,7 @@ async def stream_peer( if value == 'finish': break await stream.send(value) + return result_value async def send_peer_eof(cid: str) -> None: @@ -52,6 +54,7 @@ async def consume_peer( owner: tractor.Context, wait_on: str, reg_addr, + result_value: None|bool|int|str, ) -> bool: ''' Exercise a sibling connection outside child supervision. @@ -114,7 +117,10 @@ async def consume_peer( ): try: async with ( - peer.open_context(stream_peer) as (ctx, _), + peer.open_context( + stream_peer, + result_value=result_value, + ) as (ctx, _), ctx.open_stream() as stream, ): assert await stream.receive() == 'ready' @@ -139,7 +145,7 @@ async def consume_peer( if wait_on == 'returned': assert ( await ctx.wait_for_result() - ) is None + ) == result_value # Export the cid only after stream startup # and any graceful Return delivery. The root # uses it to select the faulted connection. @@ -153,6 +159,11 @@ async def consume_peer( await paused.wait() if not graceful: pytest.fail('resumed without EOF') + if graceful: + assert ctx._status._result == result_value + assert type(ctx._status._result) is type( + result_value, + ) except TransportClosed as exc: assert not graceful assert exc is ctx._status._remote_error @@ -212,28 +223,35 @@ async def consume_peer( @pytest.mark.parametrize( - 'wait_on', + 'wait_on, result_value', [ # Unfinished context blocked in its receive API: wake it # without requiring an outbound write to discover EOF. - 'receive', + ('receive', None), # Unfinished context blocked on an application Event: # exercise the proposed interruption of non-IPC awaits. - 'event', + ('event', None), # Final Return already consumed: preserve success when the # connection closes after its remote task has completed. - 'returned', + ('returned', None), + ('returned', False), + ('returned', 0), + ('returned', ''), # Final Return received but not drained: preserve the same # success even though the owner has not requested its result. - 'buffered', + ('buffered', None), + ('buffered', False), + ('buffered', 0), + ('buffered', ''), ], ) def test_discovered_peer_eof( reg_addr, wait_on: str, + result_value: None|bool|int|str, ) -> None: ''' Exercise the proposed EOF-driven cancellation of context owners. @@ -257,6 +275,10 @@ def test_discovered_peer_eof( value or leaving it buffered while the owner pauses. Observing delivery and the EOF report synchronizes those races without a timing sleep. Neither completed owner may be cancelled by EOF. + Falsey payloads must remain completed outcomes in both states; + only the `Unresolved` sentinel permits EOF notification. Check + the final value and type after teardown so False and 0 cannot + accidentally pass as each other. ''' async def main(): @@ -285,6 +307,7 @@ def test_discovered_peer_eof( consume_peer, wait_on=wait_on, reg_addr=reg_addr, + result_value=result_value, ) as (ctx, cid): # consume_peer exports cid after both owners are # blocked; only then break their shared connection. diff --git a/tractor/_context.py b/tractor/_context.py index 6a4023c7..a0c3cd21 100644 --- a/tractor/_context.py +++ b/tractor/_context.py @@ -223,6 +223,23 @@ class ContextStatus(pretty_struct.Struct): ''' return self._result is not Unresolved + def _outcome_is_unfinished(self) -> bool: + ''' + Neither a received outcome nor a decoded result is available. + + Shared by `process_messages()` on EOF notification and + `open_context_from_portal()` after absorbed scope + cancellation. + A buffered `Return` precedes decoding its payload; falsey + decoded values are still results, not `Unresolved`. + + ''' + return ( + self._outcome_msg is Unresolved + and + self._result is Unresolved + ) + def pformat( self, field_indent: int = 2, @@ -2623,24 +2640,46 @@ async def open_context_from_portal( raise # duh - # no local scope error, the "clean exit with a result" case. + # No escaped scope error: either a result or absorbed task cancel. else: - if ctx.chan.connected(): + if ( + ctx.chan.connected() + or + ( + # EOF cancels Context._scope, whose nursery absorbs + # its own trio.Cancelled. A disconnected channel must + # not bypass the recorded fault on that exit path. + # Share ContextStatus._outcome_is_unfinished() with + # process_messages(): preserve buffered Return and + # self-cancel ack state. Unlike that notification + # site, teardown checks for an already recorded + # TransportClosed rather than an absent error. + isinstance( + ctx._status._remote_error, + TransportClosed, + ) + and + ctx._status._outcome_is_unfinished() + ) + ): log.runtime( 'Waiting on final context result for\n' f'peer: {uid}\n' f'|_{ctx._task}\n' ) # XXX NOTE XXX: the below call to - # `Context.wait_for_result()` will ALWAYS raise - # a `ContextCancelled` (via an embedded call to - # `Context._maybe_raise_remote_err()`) IFF + # `Context.wait_for_result()` raises a recorded error + # (via an embedded call to + # `Context._maybe_raise_remote_err()`) when # a `ContextStatus._remote_error` was set by the runtime # via a call to # `Context._maybe_cancel_and_set_remote_error()`. - # As per `Context._deliver_msg()`, that error IS - # ALWAYS SET any time "child" side fails and causes - # "parent side" cancellation via a `ContextCancelled` here. + # As per `Context._deliver_msg()`, that error is set + # when the "child" side fails and causes "parent side" + # cancellation. `process_messages()` also sets it for + # local transport loss: then we raise `TransportClosed`, + # not `ContextCancelled`, without waiting on IPC. + # The original cause and self-cancel ack policy survive. try: result_or_err: Exception|Any = await ctx.wait_for_result() except BaseException as berr: diff --git a/tractor/_testing/__init__.py b/tractor/_testing/__init__.py index 8b906d11..b45ad39b 100644 --- a/tractor/_testing/__init__.py +++ b/tractor/_testing/__init__.py @@ -24,6 +24,10 @@ from contextlib import ( ) import os import pathlib +from multiprocessing.process import BaseProcess + +import psutil +import trio import tractor from tractor.devx.debug import ( @@ -37,6 +41,24 @@ from .fault_simulation import ( ) +def assert_child_reaped( + proc: trio.Process|BaseProcess, +) -> None: + ''' + Check a test-owned child's completion without doing any cleanup. + + The caller retains the handle from its own actor nursery. A + return code alone does not prove PID disappearance; check both. + Unlike `_reap.reap()`, this never discovers or signals processes. + + ''' + if isinstance(proc, trio.Process): + assert proc.returncode is not None + else: + assert proc.exitcode is not None + assert not psutil.pid_exists(proc.pid) + + # TODO, use dulwhich for this instead? # -> we're going to likely need it (or something similar) # for supporting hot-coad reload feats eventually anyway! diff --git a/tractor/runtime/_rpc.py b/tractor/runtime/_rpc.py index d030a570..2f60c17b 100644 --- a/tractor/runtime/_rpc.py +++ b/tractor/runtime/_rpc.py @@ -46,7 +46,6 @@ from trio import ( from ..ipc import Channel from .._context import ( Context, - Unresolved, ) from .._exceptions import ( ContextCancelled, @@ -1369,13 +1368,13 @@ async def process_messages( ) except TransportClosed as tc: - # MsgpackTransport._iter_pkts() raises TransportClosed for + # MsgpackTransport._iter_packets() raises TransportClosed for # a socket EOF (including an empty frame header), or wraps # a reset/read error. This is not MsgStream's EndOfChannel # after a protocol Stop, nor a Context's final Return. # - # Proposed policy, pending the transport-fault review: - # an unfinished parent Context has lost its connection, + # Once this channel is terminally closed, an unfinished + # parent Context has lost its connection, # NOT necessarily its remote task. The call below cancels # Context._scope even when its owner is blocked in unrelated # application code. That checkpoint sees trio.Cancelled; @@ -1385,16 +1384,17 @@ async def process_messages( # # Actor._contexts can include multiple connections to one # peer. Match Context.chan by identity to leave other channels - # alone. Preserve received outcomes, including a Return which - # its owner has not drained, instead of replacing them by EOF. + # alone. ContextStatus._outcome_is_unfinished() is shared with + # context teardown to preserve received outcomes, including + # a Return its owner has not drained. Here only notify if no + # error is recorded yet; teardown instead raises that error. for ctx in tuple(actor._contexts.values()): if ( ctx.chan is chan and ctx.side == 'parent' and ctx._scope is not None and ctx._status._remote_error is None - and ctx._status._outcome_msg is Unresolved - and not ctx._status._final_result_is_set() + and ctx._status._outcome_is_unfinished() ): ctx._maybe_cancel_and_set_remote_error(tc) @@ -1407,7 +1407,7 @@ async def process_messages( # TODO: maybe add a teardown handshake? and, # -[x] don't show this msg if it's an ephemeral discovery ep call? # |_ see the below `.report_n_maybe_raise()` impl as well as - # tc-exc input details in `MsgpackTCPStream._iter_pkts()` + # tc-exc input details in `MsgpackTCPStream._iter_packets()` # for different read-failure cases. # -[ ] figure out how this will break with other transports? tc.report_n_maybe_raise(