diff --git a/ai/prompt-io/opencode/20260916_ib_disconnect_cleanup.raw.md b/ai/prompt-io/opencode/20260916_ib_disconnect_cleanup.raw.md new file mode 100644 index 00000000..017ecbb8 --- /dev/null +++ b/ai/prompt-io/opencode/20260916_ib_disconnect_cleanup.raw.md @@ -0,0 +1,26 @@ +--- +model: openai/gpt-6-astra +service: opencode +git_ref: 0d95a9e6 +diff_cmd: git diff HEAD +--- + +> git diff HEAD -- tractor/runtime/_rpc.py tractor/_context.py tests/test_ipc_eof.py + +TransportClosed now wakes established unfinished parent contexts on +the failed channel. Delivered error/result outcomes remain intact, +including results buffered before EOF. Original disconnect causes +survive context unwinding. Tests cover idle/receiving consumers, +unrelated peers, graceful completion and cause preservation. + +> git diff HEAD -- tractor/runtime/_runtime.py tests/test_cancel_soon.py + +Actor.cancel_soon schedules runtime cancellation in the root nursery, +then cancels its root scope after teardown. Repeated calls do not +spawn redundant tasks. Fresh-actor tests close the service nursery, +exercise native/infected execution and verify natural lifetime +finalization unlinks all three uniquely named test-owned shm segments. + +The actual IB/SIGINT/debugger sequence remains outside these focused +regressions. Runtime cancellation and idle EOF failure mechanisms were +reproduced deterministically; no installed dependencies were modified. diff --git a/tests/test_cancel_soon.py b/tests/test_cancel_soon.py new file mode 100644 index 00000000..c04dfdef --- /dev/null +++ b/tests/test_cancel_soon.py @@ -0,0 +1,248 @@ +''' +Synchronous cancellation after the RPC service nursery exits. + +''' +from contextlib import contextmanager +from pathlib import Path +import platform +from traceback import format_exc +from uuid import uuid4 + +import pytest +import trio +from trio.testing import wait_all_tasks_blocked + +import tractor +from tractor._testing import tractor_test +from tractor.ipc import _shm + + +async def start_late_cancel( + finalizer_path: str, + close_services: bool, + shm_key: str, + cancel_path: str, +) -> dict: + ''' + Arrange cancellation from outside the RPC service nursery. + + ''' + actor: tractor.Actor = tractor.current_actor() + root_tn: trio.Nursery = actor._root_tn + service_tn: trio.Nursery = actor._service_tn + checked: trio.Event = trio.Event() + result_path: Path = Path(finalizer_path) + + @contextmanager + def report_failure(): + ''' + Preserve child failures even after its IPC loop exits. + + ''' + try: + yield + except Exception: + if not result_path.exists(): + result_path.write_text(format_exc()) + # Unwind this fresh child even when the regression + # leaves its shielded parent-channel loop alive. + actor._parent_chan_cs.cancel() + root_tn.cancel_scope.cancel() + raise + + def finalize() -> None: + ''' + Prove runtime unwinding reached the lifetime callbacks. + + ''' + with report_failure(): + assert checked.is_set() + assert root_tn._closed + assert actor.cancel_complete + # Completed teardown must not schedule into the closed + # Actor._root_tn, including repeated finalizer calls. + assert actor.cancel_soon() is False + assert actor.cancel_soon() is False + result_path.write_text('finalized') + + actor.lifetime_stack.callback(finalize) + + # ExitStack is LIFO: allocator destroy/close callbacks must + # finish before finalize publishes its success marker. + with report_failure(): + shmarr = _shm.open_shm_ndarray( + size=4, + key=shm_key, + dtype=[('index', 'i8')], + ) + + async def cancel_after_rpc() -> None: + ''' + Let the setup RPC return before shutting down services. + + ''' + with ( + report_failure(), + trio.CancelScope(shield=True), + trio.fail_after(5), + ): + # The parent first proves every segment exists and + # closes its probe handles, then releases this gate. + while not Path(cancel_path).exists(): + await trio.sleep(0.01) + await wait_all_tasks_blocked() + if close_services: + service_tn.cancel_scope.cancel() + await wait_all_tasks_blocked() + assert service_tn._closed + + assert not root_tn._closed + assert not actor.cancel_complete + assert actor._parent_chan_cs.shield + # No checkpoint between calls: the first schedules the + # root-owned task; the duplicate must not schedule one. + assert actor.cancel_soon() is True + assert actor.cancel_soon() is False + await actor._cancel_complete.wait() + await wait_all_tasks_blocked() + assert actor._parent_chan_cs.cancel_called + assert root_tn.cancel_scope.cancel_called + assert actor.cancel_soon() is False + checked.set() + + root_tn.start_soon(cancel_after_rpc) + return shmarr.token + + +@pytest.mark.parametrize( + 'infect_asyncio', + [ + False, + pytest.param( + True, + marks=pytest.mark.skipif( + platform.system() == 'Windows', + reason='asyncio infection is unsupported on Windows', + ), + ), + ], +) +@pytest.mark.parametrize('close_services', [False, True]) +@tractor_test +async def test_cancel_soon_finishes_runtime( + reg_addr: tuple, + tmp_path: Path, + close_services: bool, + infect_asyncio: bool, +) -> None: + ''' + Late Actor.cancel_soon must finish teardown after services exit. + + An IB disconnect left Actor._service_tn closed while the root + nursery and shielded parent-channel loop survived. A late + synchronous cancellation then raised RuntimeError from + Nursery.start_soon, preventing the asyncio host from waiting + for its Trio guest and the actor's lifetime finalizers. + + A fresh child schedules a root-nursery controller. Trio's + wait_all_tasks_blocked lets its setup RPC finish, then proves + service-nursery exit before requesting cancellation. The open + service variant also guards ordinary self-cancellation. Run + both with native Trio and an asyncio-infected child runtime. + + Repeated requests before the canceller runs, after completion, + and from a lifetime callback must return False; only the first + request returns True. A shielded controller checks that + Actor.cancel actually cancels the + parent-channel scope and the root nursery. Only runtime exit + may write the finalizer marker; the parent waits for it before + leaving its actor nursery, so parent teardown cannot cause a + false pass. No signals or external services are involved. + + The child allocates a four-row array via open_shm_ndarray. + A file gate holds cancellation until the parent has opened and + closed all three data/first/last segments. The allocator's + lifetime destroy/close callbacks precede the success marker. + Before exiting its actor nursery, the parent proves all three + names are gone. No parent lifetime callback owns these segments; + a finally block cleans only this test's unique names if a + failure leaks them, after the assertions have already failed. + + ''' + pytest.importorskip('numpy') + finalizer_path: Path = tmp_path / 'finalized' + cancel_path: Path = tmp_path / 'cancel' + shm_key: str = f'cancel_soon_{uuid4().hex}' + token: _shm.NDToken = _shm._make_token( + key=shm_key, + size=4, + dtype=[('index', 'i8')], + ) + # Know the exact names even if allocation or the setup RPC + # fails partway through. _make_token also handles macOS names. + names: tuple[str, ...] = ( + token.shm_name, + token.shm_first_index_name, + token.shm_last_index_name, + ) + assert len(set(names)) == 3 + # Reject any pre-existing name before entering our cleanup + # block: a collision would not belong to this test invocation. + for name in names: + try: + probe = _shm.SharedMemory(name, create=False) + except FileNotFoundError: + continue + else: + probe.close() + pytest.fail(f'SHM name already exists: {name}') + + try: + with trio.fail_after(10): + async with tractor.open_nursery() as an: + portal: tractor.Portal = await an.start_actor( + 'late_cancel', + enable_modules=[__name__], + infect_asyncio=infect_asyncio, + ) + allocated: dict = await portal.run( + start_late_cancel, + finalizer_path=str(finalizer_path), + close_services=close_services, + shm_key=shm_key, + cancel_path=str(cancel_path), + ) + assert _shm.NDToken.from_msg(allocated) == token + for name in names: + probe = _shm.SharedMemory(name, create=False) + probe.close() + cancel_path.touch() + + while not finalizer_path.exists(): + await trio.sleep(0.01) + + result: str = finalizer_path.read_text() + assert result == 'finalized', result + + remaining: list[str] = [] + for name in names: + try: + probe = _shm.SharedMemory(name, create=False) + except FileNotFoundError: + continue + else: + probe.close() + remaining.append(name) + assert not remaining, remaining + finally: + # No scanning: only this invocation's three unique names. + # Run after child teardown, never before the absence check. + for name in names: + try: + leaked = _shm.SharedMemory(name, create=False) + except FileNotFoundError: + continue + try: + leaked.unlink() + finally: + leaked.close() diff --git a/tractor/runtime/_runtime.py b/tractor/runtime/_runtime.py index 27e861f8..a9f1c8f7 100644 --- a/tractor/runtime/_runtime.py +++ b/tractor/runtime/_runtime.py @@ -267,6 +267,7 @@ class Actor: self._cancel_complete = trio.Event() self._cancel_called_by: tuple[str, tuple]|None = None self._cancel_called: bool = False + self._cancel_soon_scheduled: bool = False # retrieve and store parent `__main__` data which # will be passed to children @@ -1165,14 +1166,49 @@ class Actor: await self.cancel(req_chan=None) # self cancel raise - def cancel_soon(self) -> None: + def cancel_soon(self) -> bool: ''' Cancel this actor asap; can be called from a sync context. - Schedules runtime cancellation via `Actor.cancel()` inside - the RPC service nursery. + Schedule `Actor.cancel()` in the longer-lived root nursery, + then cancel that nursery once runtime teardown completes. + + Return a `bool` indicating whether the runtime cancellation + was scheduled in the `Actor._root_tn` by this call. ''' + root_tn: Nursery = self._root_tn + assert root_tn + if self.cancel_complete: + # A prior cancel already tore down services and the + # shielded parent-channel loop. The root may still be + # waiting, or already closed: neither needs a new task. + log.runtime( + 'Actor cancellation already complete; cancelling ' + 'the root scope without scheduling another task' + ) + root_tn.cancel_scope.cancel() + return False + + if self._cancel_soon_scheduled: + log.runtime( + 'Actor cancellation already scheduled; skipping ' + 'duplicate cancel_soon() request' + ) + return False + + # Actor._service_tn may already be closed after a service + # failure. Keep the canceller in Actor._root_tn so teardown + # still executes, including the shielded parent-loop cancel. + async def cancel_root_tn_after_services() -> None: + await self.cancel(None) # self-cancel + log.cancel( + '>> cancelling `._root_tn`\n' + f'c>(\n' + f' |_{root_tn.cancel_scope!r}\n' + ) + root_tn.cancel_scope.cancel() + actor_repr: str = _pformat.nest_from_op( input_op='>c(', text=self.pformat(), @@ -1183,37 +1219,11 @@ class Actor: f'>> scheduling `Actor.cancel()`\n' f'{actor_repr}' ) - assert self._service_tn - self._service_tn.start_soon( - self.cancel, - None, # self cancel all rpc tasks - ) - - # schedule a "canceller task" in the `._root_tn` once the - # `._service_tn` is fully shutdown; task waits for child-ish - # scopes to fully exit then finally cancels its parent, - # root-most, scope. - async def cancel_root_tn_after_services(): - log.runtime( - 'Waiting on service-tn to cancel..\n' - f'c>)\n' - f'|_{self._service_tn.cancel_scope!r}\n' - ) - await self._cancel_complete.wait() - log.cancel( - f'`._service_tn` cancelled\n' - f'>c)\n' - f'|_{self._service_tn.cancel_scope!r}\n' - f'\n' - f'>> cancelling `._root_tn`\n' - f'c>(\n' - f' |_{self._root_tn.cancel_scope!r}\n' - ) - self._root_tn.cancel_scope.cancel() - - self._root_tn.start_soon( + root_tn.start_soon( cancel_root_tn_after_services ) + self._cancel_soon_scheduled = True + return True @property def cancel_complete(self) -> bool: