From ebf2258b4fcb1d48b380c401c5c22d8fc38f6fd8 Mon Sep 17 00:00:00 2001 From: goodboy Date: Fri, 14 Aug 2026 12:28:56 -0400 Subject: [PATCH] Keep accepted RPCs alive on `TransportClosed` An RPC caller can close its channel after the callee creates the endpoint coro but before its `StartAck` or final response lands. Treat those response-send failures as terminal delivery failures so the accepted endpoint still runs and application errors stay local instead of cancelling the shared service nursery. Register each cancellable RPC in `Actor._rpc_tasks` before publishing its `Context` through `TaskStatus.started()`. This closes the checkpoint-free completion race without a cross-task handoff and keeps `Actor._ongoing_rpc_tasks` balanced through existing cleanup. Add regressions for caller disconnects at `StartAck` and error shipment, plus registration-before-execution and cleanup checks. Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/test_rpc.py | 127 ++++++++++++++++++++++++++++++++++++++++ tractor/runtime/_rpc.py | 92 +++++++++++++++++++++-------- 2 files changed, 195 insertions(+), 24 deletions(-) diff --git a/tests/test_rpc.py b/tests/test_rpc.py index 6e2b414c..440e14c6 100644 --- a/tests/test_rpc.py +++ b/tests/test_rpc.py @@ -4,11 +4,18 @@ related API and error checks. ''' import itertools +from unittest.mock import ( + AsyncMock, + Mock, +) import pytest import tractor import trio +from tractor._exceptions import TransportClosed +from tractor.runtime import _rpc + async def sleep_back_actor( actor_name, @@ -46,6 +53,126 @@ async def short_sleep(): await trio.sleep(0) +def test_rpc_runs_after_startack_disconnect(): + ''' + Complete an accepted RPC when its caller closes before `StartAck`. + + Registrar teardown opens short-lived `unregister_actor` RPCs. A + loaded caller can close its channel while the registrar sends the + acknowledgement; normalized `TransportClosed` previously escaped + into the shared service nursery before the already-created + coroutine was awaited. This fake fails the first response send and + proves the RPC side effect still runs with no later send attempt. + + ''' + async def main(): + rpc_ran = trio.Event() + + chan = Mock() + chan.send = AsyncMock( + side_effect=TransportClosed( + message='caller closed before StartAck', + ), + ) + chan.connected.return_value = False + ctx = Mock( + chan=chan, + cid='rpc-cid', + _scope=None, + _task='rpc-task', + ) + actor = Mock() + actor.get_context.return_value = ctx + actor._rpc_tasks = {} + actor._ongoing_rpc_tasks = trio.Event() + actor._ongoing_rpc_tasks.set() + + async def rpc_func(): + assert (chan, ctx.cid) in actor._rpc_tasks + rpc_ran.set() + + async def invoke(task_status): + await _rpc._invoke( + actor=actor, + cid=ctx.cid, + chan=chan, + func=rpc_func, + kwargs={}, + task_status=task_status, + ) + + async with trio.open_nursery() as nursery: + started_ctx = await nursery.start(invoke) + assert started_ctx is ctx + await rpc_ran.wait() + + assert rpc_ran.is_set() + assert not actor._rpc_tasks + assert actor._ongoing_rpc_tasks.is_set() + chan.send.assert_awaited_once() + + trio.run(main) + + +def test_error_shipment_ignores_closed_response_channel(monkeypatch): + ''' + Preserve an application error when its response channel is closed. + + A caller can disconnect after submitting an RPC but before its + error response. Normalized `TransportClosed` from that final send + is terminal response failure, not a new actor-wide service error. + This test proves error shipment logs and returns without replacing + the original application exception. + + ''' + chan = Mock() + chan.send = AsyncMock( + side_effect=[ + None, + TransportClosed( + message='caller closed before Error response', + ), + ], + ) + error_msg = Mock(boxed_type_str='ValueError') + monkeypatch.setattr( + _rpc, + 'pack_error', + Mock(return_value=error_msg), + ) + ctx = Mock( + chan=chan, + cid='rpc-cid', + _scope=None, + _task='rpc-task', + ) + actor = Mock() + actor.get_context.return_value = ctx + actor._rpc_tasks = {} + actor._ongoing_rpc_tasks = trio.Event() + actor._ongoing_rpc_tasks.set() + + async def failing_rpc(): + raise ValueError('application failure') + + async def main(): + async with trio.open_nursery() as nursery: + started_ctx = await nursery.start( + _rpc._invoke, + actor, + ctx.cid, + chan, + failing_rpc, + {}, + ) + assert started_ctx is ctx + + trio.run(main) + assert chan.send.await_count == 2 + assert not actor._rpc_tasks + assert actor._ongoing_rpc_tasks.is_set() + + @pytest.mark.parametrize( 'to_call', [ ([], 'short_sleep', tractor.RemoteActorError), diff --git a/tractor/runtime/_rpc.py b/tractor/runtime/_rpc.py index d2c9ea30..d90cb658 100644 --- a/tractor/runtime/_rpc.py +++ b/tractor/runtime/_rpc.py @@ -94,6 +94,33 @@ if TYPE_CHECKING: log = get_logger('tractor') +def _register_rpc_task( + actor: Actor, + chan: Channel, + func: Callable, + is_rpc: bool, + task_status: TaskStatus[ + Context | BaseException + ], + ctx: Context, +) -> None: + ''' + Register an RPC task before publishing it to `Nursery.start()`. + + ''' + if is_rpc: + if not actor._rpc_tasks: + actor._ongoing_rpc_tasks = trio.Event() + + actor._rpc_tasks[(chan, ctx.cid)] = ( + ctx, + func, + trio.Event(), + ) + + task_status.started(ctx) + + # ?TODO? move to a `tractor.lowlevel._rpc` with the below # func-type-cases implemented "on top of" `@context` defs: # -[ ] std async func helper decorated with `@rpc_func`? @@ -142,7 +169,14 @@ async def _invoke_non_context( # is propagated! with cancel_scope as cs: ctx._scope = cs - task_status.started(ctx) + _register_rpc_task( + actor, + chan, + func, + is_rpc, + task_status, + ctx, + ) async with aclosing(coro) as agen: async for item in agen: # TODO: can we send values back in here? @@ -178,7 +212,14 @@ async def _invoke_non_context( ) with cancel_scope as cs: ctx._scope = cs - task_status.started(ctx) + _register_rpc_task( + actor, + chan, + func, + is_rpc, + task_status, + ctx, + ) await coro if not cs.cancelled_caught: @@ -202,23 +243,29 @@ async def _invoke_non_context( ) await chan.send(ack) except ( + TransportClosed, trio.ClosedResourceError, trio.BrokenResourceError, BrokenPipeError, ) as ipc_err: failed_resp = True - if is_rpc: - raise ipc_err - else: - log.exception( - f'Failed to ack runtime RPC request\n\n' - f'{func} x=> {ctx.chan}\n\n' - f'{ack}\n' - ) + log.warning( + f'Failed to ack runtime RPC request\n\n' + f'{func} x=> {ctx.chan}\n\n' + f'{ack}\n' + f' |_{ipc_err!r}\n' + ) with cancel_scope as cs: ctx._scope: CancelScope = cs - task_status.started(ctx) + _register_rpc_task( + actor, + chan, + func, + is_rpc, + task_status, + ctx, + ) result = await coro fname: str = func.__name__ @@ -247,6 +294,7 @@ async def _invoke_non_context( ) await chan.send(ret_msg) except ( + TransportClosed, BrokenPipeError, trio.BrokenResourceError, ): @@ -673,7 +721,14 @@ async def _invoke( ): ctx._scope_nursery = tn rpc_ctx_cs = ctx._scope = tn.cancel_scope - task_status.started(ctx) + _register_rpc_task( + actor, + chan, + func, + is_rpc, + task_status, + ctx, + ) # invoke user endpoint fn. res: Any|PayloadT = await coro @@ -920,6 +975,7 @@ async def try_ship_error_to_remote( # downward should be mostly wrapping such cases in a # tpt-closed; the `.critical()` usage is warranted. except ( + TransportClosed, trio.ClosedResourceError, trio.BrokenResourceError, BrokenPipeError, @@ -1221,18 +1277,6 @@ async def process_messages( ) continue - else: - # mark our global state with ongoing rpc tasks - actor._ongoing_rpc_tasks = trio.Event() - - # store cancel scope such that the rpc task can be - # cancelled gracefully if requested - actor._rpc_tasks[(chan, cid)] = ( - ctx, - func, - trio.Event(), - ) - # XXX RUNTIME-SCOPED! remote (likely internal) error # (^- bc no `Error.cid` -^) #