From b6316ed776f5533044bdefe138418fda44315000 Mon Sep 17 00:00:00 2001 From: goodboy Date: Fri, 30 Oct 2026 13:10:53 -0400 Subject: [PATCH] Avoid `raise None` after successful `asyncio` exit An asyncio-side task can handle the issued `TrioTaskExited` and return normally. With `suppress_graceful_exits=False`, the strict reporting branch in `translate_aio_errors()` nevertheless tried to raise `trio_to_raise=None`, replacing successful completion with TypeError('exceptions must derive from BaseException'). Require an actual pending exception before that strict-policy raise. Existing errors retain their reporting and cause handling; strict reporting does not manufacture an error after a successful asyncio-side return with no remaining trio-side error. Deats, - run the channel-exit matrix directly in an infected root as well as through sequential TCP RPCs in an infected child; - add direct/cached cases where the asyncio-side task suppresses the issued TTE and returns, under graceful and strict policies, with native and Greenback task execution; - test a child suppressing TTE and staying blocked: check channel startup state, then require receipt of that channel's exact issued TTE with trio-side exit set and asyncio still running; - publish the receipt marker only after those checks, before the responsive parent awaits `an.cancel()`; require nursery exit to join the retained child process handle; - bound that probe from another process, failing if its watchdog must rescue the test group; successful child exit does not prove that hard-kill escalation was needed; - explicitly `raise signal`, rename the caller-cancellation flag to `cancel_trio_side`, and format signal/callback diagnostics; - retain the maintainer's test layout and helper documentation. Qualification: expanded matrix before the guard: 8 failed / 33 passed; all eight failures are strict-policy successful returns. After the guard, selected exit/root/infected-asyncio/formatter suites: 70 passed / 18 deselected, with 24 fork deprecation warnings. Strengthened blocked-child state checks: 1 passed. Interactive SIGINT/debugger cases remain excluded. (this patch was generated in some part by `opencode` using `gpt-6.1-sol` (`openai`)) --- tests/test_aio_channel_exit.py | 392 ++++++++++++++++++++++++++++----- tractor/to_asyncio.py | 8 +- 2 files changed, 338 insertions(+), 62 deletions(-) diff --git a/tests/test_aio_channel_exit.py b/tests/test_aio_channel_exit.py index fc80408c..ae7ac7ae 100644 --- a/tests/test_aio_channel_exit.py +++ b/tests/test_aio_channel_exit.py @@ -3,6 +3,11 @@ Inter-loop channel teardown must handle the issued exit signal. ''' import asyncio +import os +from pathlib import Path +import signal +import subprocess +import sys from contextlib import asynccontextmanager as acm from collections.abc import AsyncIterator @@ -22,10 +27,11 @@ _actor_test_timeout_s: float = 10 async def exercise_trio_side_exit( cached: bool, - cancelled: bool, + cancel_trio_side: bool, suppress: bool, expect_aio_side_err_type: str|None, greenback_shim: bool, + suppress_aio_side_tte: bool = False, ) -> None: ''' Exit direct and cached inter-loop channels in an infected actor. @@ -38,12 +44,14 @@ async def exercise_trio_side_exit( Cases: - ``cached`` wraps the channel API in ``maybe_open_context()``. - - ``cancelled`` cancels the last consumer at a Trio checkpoint. + - ``cancel_trio_side`` cancels the caller at a Trio checkpoint. - ``suppress`` selects graceful or strict exit-signal reporting. - ``expect_aio_side_err_type`` is a class name or ``None``. Class names travel as RPC payloads; actual exception objects remain inside this actor for identity/cause assertions. - ``greenback_shim`` wraps the asyncio task without enabling PDB. + - ``suppress_aio_side_tte`` handles the issued signal and returns + normally instead of raising it or a distinct cleanup failure. Ordering and proof: - ``tn.start()`` waits for a sibling subscription; ``release`` @@ -108,10 +116,15 @@ async def exercise_trio_side_exit( except to_asyncio.TrioTaskExited as signal: signals.append(signal) + # Returning here tests successful asyncio-side completion + # after it handles the trio-side parent's exit signal. + if suppress_aio_side_tte: + return + # Healthy cleanup re-raises the very signal issued by # open_channel_from(); suppression uses object identity. if expect_aio_side_err_type is None: - raise + raise signal failure: RuntimeError|to_asyncio.TrioTaskExited if expect_aio_side_err_type == 'RuntimeError': @@ -196,7 +209,7 @@ async def exercise_trio_side_exit( ), ) -> None: ''' - Close one cached-channel subscription while another remains. + Close a subscription while another stays open in the caller. - Enter ``open_maybe_cached_aio_chan`` in another trio.Task. - Require ``expected`` identity to prove actual cache reuse. @@ -245,7 +258,7 @@ async def exercise_trio_side_exit( # Consume a real broadcast value before # exit or last-consumer cancellation. assert await bchan.receive() == 'tick' - if cancelled: + if cancel_trio_side: cs.cancel() await trio.lowlevel.checkpoint() @@ -253,10 +266,19 @@ async def exercise_trio_side_exit( caught = error assert tasks[task_index].done() - assert len(signals) == task_index + 1, ( - caught, - tasks[task_index].exception(), - signals, + num_signals: int = len(signals) + expect_num_signals: int = task_index + 1 + task_error: BaseException|None = ( + tasks[task_index].exception() + ) + assert num_signals == expect_num_signals, ( + f'Expected one issued TTE per asyncio task\n' + f'task index: {task_index}\n' + f'expected signals: {expect_num_signals}\n' + f'recorded signal count: {num_signals}\n' + f'caught trio-side error: {caught!r}\n' + f'asyncio task error: {task_error!r}\n' + f'recorded signals: {signals!r}\n' ) if expect_aio_side_err_type: # For TTE, matching the issued signal's type must NOT @@ -265,6 +287,10 @@ async def exercise_trio_side_exit( assert caught is failures[task_index] assert caught.__cause__ is signals[task_index] + elif suppress_aio_side_tte: + assert caught is None, repr(caught) + assert tasks[task_index].exception() is None + elif not suppress: # Strict reporting returns the exact issued signal, # not a new wrapper or an asyncio-created substitute. @@ -284,7 +310,23 @@ async def exercise_trio_side_exit( # Expected shutdown must have no unhandled callbacks. # Fault cases above test propagation and causes, without # asserting a broader callback-cleanliness policy. - assert not callback_errors, callback_errors + callback_report: str = '\n'.join( + f'callback {index}:\n' + f' message: {context.get("message")!r}\n' + f' exception: {context.get("exception")!r}\n' + f' handle: {context.get("handle")!r}\n' + f' task: {context.get("task")!r}\n' + f' future: {context.get("future")!r}' + for index, context in enumerate(callback_errors) + ) + assert not callback_errors, ( + f'Unhandled asyncio callbacks after channel exit\n' + f'cached: {cached}\n' + f'cancel_trio_side: {cancel_trio_side}\n' + f'suppress_graceful_exits: {suppress}\n' + f'greenback_shim: {greenback_shim}\n' + f'{callback_report}\n' + ) finally: loop.set_exception_handler(previous_handler) @@ -295,49 +337,76 @@ async def exercise_trio_side_exit( ids=['native', 'greenback'], ) @pytest.mark.parametrize( - 'cached,cancelled,suppress,expect_aio_side_err_type', + 'infect_root', + [False, True], + ids=['infected-child', 'infected-root'], +) +@pytest.mark.parametrize( + 'cached,cancel_trio_side,suppress,expect_aio_side_err_type,' + 'suppress_aio_side_tte', [ # Expected TTE ack: direct use, last cached caller # exit, and cancellation of that last cached caller. pytest.param( - False, False, True, None, + False, False, True, None, False, id='direct-close', ), pytest.param( - True, False, True, None, + True, False, True, None, False, id='last-cache-user', ), pytest.param( - True, True, True, None, + True, True, True, None, False, id='cancelled-last-cache-user', ), # Strict reporting must preserve the issued TTE's identity. pytest.param( - True, False, False, None, + True, False, False, None, False, id='explicit-exit-reporting', ), # New asyncio-side cleanup failures must escape suppression, # even when the exception class matches the issued signal. pytest.param( - True, False, True, 'RuntimeError', + True, False, True, 'RuntimeError', False, id='independent-cleanup-error', ), pytest.param( - True, False, True, 'TrioTaskExited', + True, False, True, 'TrioTaskExited', False, id='independent-same-type-error', ), + # The asyncio task may handle the TTE and finish normally. + pytest.param( + False, False, True, None, True, + id='direct-suppress-tte-and-return', + ), + pytest.param( + True, False, True, None, True, + id='cached-suppress-tte-and-return', + ), + # Strict trio-side policy must not manufacture a task error + # when the asyncio-side task handled its signal successfully. + pytest.param( + False, False, False, None, True, + id='strict-direct-suppress-tte-and-return', + ), + pytest.param( + True, False, False, None, True, + id='strict-cached-suppress-tte-and-return', + ), ], ) def test_channel_owner_exit( reg_addr: tuple[str, int], cached: bool, - cancelled: bool, + cancel_trio_side: bool, suppress: bool, expect_aio_side_err_type: str|None, greenback_shim: bool, + infect_root: bool, + suppress_aio_side_tte: bool, ) -> None: ''' - Channel teardown must leave the infected child actor RPC-capable. + Exercise trio-side channel exit in an infected root or child. Regression: - ``open_channel_from()`` signalled the running asyncio task when @@ -346,11 +415,13 @@ def test_channel_owner_exit( suppression and was also raised from the asyncio done callback. Proof boundary: - - Real TCP RPC invokes ``exercise_trio_side_exit`` in an infected - child, using the channel API and its broadcast subscriptions. - Cached cases additionally wrap it in ``maybe_open_context()``. - - Repeated RPCs in that same actor prove it remains usable after - the endpoint's asyncio tasks have finished. + - Infected-child cases invoke ``exercise_trio_side_exit`` by TCP + RPC. Infected-root cases call it directly in the guest Trio run + hosted by asyncio, with the root actor runtime open. + - Both use the actual channel API and broadcast subscriptions; + cached cases additionally wrap it in ``maybe_open_context()``. + - Sequential repeats check continued actor usability after prior + asyncio tasks finish; the child path also checks RPC usability. - Strict reporting and independent faults retain their identity; broad exception-type suppression cannot satisfy the matrix. - The deadline bounds hangs; no live broker or debugger is used. @@ -361,45 +432,244 @@ def test_channel_owner_exit( application nor caching in its direct-channel case. ''' - async def main() -> None: - ''' - Own the infected child and bound the complete RPC exercise. + if not infect_root: + async def main() -> None: + ''' + Run the selected case directly in a root or via child + RPCs. - - Capture ``reg_addr`` for the test's isolated TCP registry. - - Forward the selected case to each RPC in the same child. - - A callback could fail after the RPC body finishes and then - terminate the actor. A second RPC checks continued use; - the callback-drain assertion inside the endpoint separately - checks the failure directly, without relying on RPC timing. - - Explicitly cancel the child after the probes: a successful - RPC does not itself end the actor's service lifetime. + - reg_addr isolates the runtime's TCP registry. + - Each call finishes before the next starts. + - The child path explicitly cancels the actor after its + RPCs; completing an RPC does not end the actor's + lifetime. + - A delayed done-callback failure could terminate the + child after an RPC body finishes. Another RPC checks + usability; the endpoint also inspects callback + exceptions directly. - ''' - an: tractor.ActorNursery - with trio.fail_after(_actor_test_timeout_s): - async with tractor.open_nursery( - registry_addrs=[reg_addr], - ) as an: - portal: tractor.Portal = await an.start_actor( - 'aio_channel_owner', - enable_modules=[__name__], - infect_asyncio=True, - ) - # portal.run() waits for this remote task's final - # result. These RPCs are sequential, not concurrent; - # only the endpoint's subscription tasks overlap. - for _ in range(_rpc_repetitions): - await portal.run( - exercise_trio_side_exit, - cached=cached, - cancelled=cancelled, - suppress=suppress, - expect_aio_side_err_type=( - expect_aio_side_err_type - ), - greenback_shim=greenback_shim, + ''' + an: tractor.ActorNursery + with trio.fail_after(_actor_test_timeout_s): + async with tractor.open_nursery( + registry_addrs=[reg_addr], + ) as an: + portal: tractor.Portal = await an.start_actor( + 'aio_channel_owner', + enable_modules=[__name__], + infect_asyncio=True, ) + # portal.run() waits for this remote task's final + # result. These RPCs are sequential, not concurrent; + # only the endpoint's subscription tasks overlap. + for _ in range(_rpc_repetitions): + await portal.run( + exercise_trio_side_exit, + cached=cached, + cancel_trio_side=cancel_trio_side, + suppress=suppress, + expect_aio_side_err_type=( + expect_aio_side_err_type + ), + greenback_shim=greenback_shim, + suppress_aio_side_tte=suppress_aio_side_tte, + ) - await portal.cancel_actor() + await portal.cancel_actor() - trio.run(main) + trio.run(main) + + else: # infected-root case + async def infected_root_main(): + ''' + Same as `main` above except invoke + `exercise_trio_side_exit` in-task with no subactor spawn. + + ''' + # run_as_asyncio_guest() supplies the host asyncio loop. + # No RPC/child actor is involved: call the same endpoint + # directly with an infected root runtime open. + with trio.fail_after(_actor_test_timeout_s): + async with tractor.open_root_actor( + registry_addrs=[reg_addr], + ): + for _ in range(_rpc_repetitions): + await exercise_trio_side_exit( + cached=cached, + cancel_trio_side=cancel_trio_side, + suppress=suppress, + expect_aio_side_err_type=( + expect_aio_side_err_type + ), + greenback_shim=greenback_shim, + suppress_aio_side_tte=( + suppress_aio_side_tte + ), + ) + + to_asyncio.run_as_asyncio_guest( + trio_main=infected_root_main, + ) + + +async def suppress_tte_and_block(marker: str) -> None: + ''' + Leave channel teardown waiting on a task that handled its TTE. + + ``marker`` is a test-owned file shared with the parent process. + Write it ONLY after receiving the issued signal, so cancellation + of the actor nursery cannot race ahead of the blocked teardown. + + ''' + async def block_until_trio_side_exit( + chan: to_asyncio.LinkedTaskChannel, + ) -> None: + ''' + Handle the issued TTE but keep the asyncio task running. + + Capture marker to acknowledge that the hang precondition was + reached. Ordinary later asyncio cancellation remains enabled; + the assertion is actor termination, not mandatory hard-kill. + + ''' + chan.started_nowait() + try: + await asyncio.Event().wait() + + except to_asyncio.TrioTaskExited as signal: + # Publish the marker only for this channel's issued TTE, + # after trio-side context exit has actually begun. + assert signal is chan._aio_owner_exit + assert chan._trio_exited + assert chan._aio_task is asyncio.current_task() + assert not chan._aio_task.done() + Path(marker).write_text(str(os.getpid())) + await asyncio.Event().wait() + + chan: to_asyncio.LinkedTaskChannel + async with to_asyncio.open_channel_from( + block_until_trio_side_exit, + ) as (chan, first): + assert first is None + assert chan._aio_task is not None + assert not chan._aio_task.done() + assert not chan._trio_exited + + # Leaving this body immediately sends TTE. Context exit then + # waits for the asyncio task, which handles TTE and blocks + # again; the RPC cannot finish without external cancellation. + + +async def cancel_an_with_blocked_aio( + marker: str, + host: str, + port: int, +) -> None: + ''' + Cancel the actor nursery after its child suppresses channel TTE. + + Called in a separate parent process by the watchdog test below. + ActorNursery.cancel() must terminate the child and nursery exit + must join it. A deadline in the pytest process detects failure + independently of either actor's event loop. + + ''' + async def invoke_blocked_rpc() -> None: + ''' + Keep one RPC pending while the parent cancels its actor tree. + + The parent joins this `trio` task through `tn` after + `an.cancel()`. + + ''' + await portal.run( + suppress_tte_and_block, + marker=marker, + ) + + an: tractor.ActorNursery + async with tractor.open_nursery( + registry_addrs=[(host, port)], + debug_mode=False, + ) as an: + portal: tractor.Portal = await an.start_actor( + 'blocked_aio_channel', + enable_modules=[__name__], + infect_asyncio=True, + debug_mode=False, + ) + # Retain the runtime's process handle across nursery cleanup; + # successful cancel-RPC delivery alone does not prove exit. + child, = an._children.values() + proc: trio.Process = child[1] + tn: trio.Nursery + async with trio.open_nursery() as tn: + tn.start_soon(invoke_blocked_rpc) + while not Path(marker).exists(): + await trio.sleep(0.01) + + await an.cancel() + tn.cancel_scope.cancel() + + assert proc.poll() is not None, ( + 'ActorNursery exited without joining the blocked child' + ) + + +def test_cancel_an_reaps_blocked_aio( + tmp_path: Path, + reg_addr: tuple[str, int], + _reap_watchdog_s: float = 30, +) -> None: + ''' + Nursery cancellation must terminate a TTE-suppressing child. + + - Start a dedicated parent process with its own process group. + - Wait inside that process for the child's TTE receipt marker, + then call an.cancel() and check its process handle after exit. + - Expected termination uses Tractor's own cancellation/reaping. + Watchdog rescue kills this test's group and FAILS the test. + It cannot count as evidence of successful runtime teardown. + + ''' + if os.name != 'posix': + pytest.skip('The watchdog uses POSIX process groups') + + marker: Path = tmp_path / 'tte-suppressed' + code: str = ( + 'import sys, trio; ' + 'from tests.test_aio_channel_exit ' + 'import cancel_an_with_blocked_aio; ' + 'trio.run(cancel_an_with_blocked_aio, ' + 'sys.argv[1], sys.argv[2], int(sys.argv[3]))' + ) + parent: subprocess.Popen = subprocess.Popen( + [ + sys.executable, + '-c', + code, + str(marker), + reg_addr[0], + str(reg_addr[1]), + ], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + start_new_session=True, + ) + try: + output, _ = parent.communicate(timeout=_reap_watchdog_s) + + except subprocess.TimeoutExpired: + os.killpg(parent.pid, signal.SIGKILL) + output, _ = parent.communicate() + pytest.fail( + f'Watchdog rescued blocked actor-nursery teardown\n' + f'TTE received: {marker.exists()}\n' + f'{output}' + ) + + assert marker.exists(), f'Child never received TTE\n{output}' + assert parent.returncode == 0, ( + f'Blocked-child cancellation failed\n{output}' + ) diff --git a/tractor/to_asyncio.py b/tractor/to_asyncio.py index c9a81fa1..f8bd6729 100644 --- a/tractor/to_asyncio.py +++ b/tractor/to_asyncio.py @@ -1479,7 +1479,13 @@ async def translate_aio_errors( ) raise aio_err - if not suppress_graceful_exits: + # Strict reporting preserves an existing error; it must not + # raise None when the asyncio-side task handled TTE and + # returned successfully (and no trio-side error remains). + if ( + not suppress_graceful_exits + and trio_to_raise is not None + ): raise trio_to_raise from raise_from if trio_to_raise: