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

@@ -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),

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:
+            log.warning(
-                raise ipc_err
+                f'Failed to ack runtime RPC request\n\n'
-            else:
+                f'{func} x=> {ctx.chan}\n\n'
-                log.exception(
+                f'{ack}\n'
-                    f'Failed to ack runtime RPC request\n\n'
+                f' |_{ipc_err!r}\n'
-                    f'{func} x=> {ctx.chan}\n\n'
+            )
-                    f'{ack}\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` -^)
                     #