From 09bbe9701e73dfc4ad6d637cf5cbf629b1ae0ae1 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 2 Jul 2026 18:33:17 -0400 Subject: [PATCH] Merge the supervise error handlers into one Step B2 of the `._ria_nursery` removal (issue #477; see `ai/conc-anal/ria_nursery_removal_plan.md`). With the 2ndary nursery gone (step B), the two nested error handlers in `_open_and_supervise_one_cancels_all_nursery` collapse to one, - the outer `except (Exception, BaseExceptionGroup, trio.Cancelled)` existed to catch errors bubbling from the old `._ria_nursery.__aexit__` reaper-group; that nursery no longer exists. - trace shows the outer handler's `raise` was already DEAD: the inner handler records `errors[uid]` as its first action, so `errors` is always non-empty by the time anything could reach the outer handler, and the `finally`'s raise-from-`errors` always superseded the outer `raise`. - so fold both into a single `except BaseException as _scope_err` guarding the lone daemon nursery; the `finally` (unchanged) still raises the collected `errors` as a single exc or `BaseExceptionGroup`. - drop the now-unused `outer_err`/`inner_err` locals. Behaviour-preserving (net ~30 lines lighter); the big diff is the one-level de-indent of the handler body. The two remaining `maybe_wait_for_debugger()` guards collapse to the single pre-teardown wait. Prompt-IO: ai/prompt-io/claude/20260702T222544Z_9201a2ed_prompt_io.md (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- tractor/runtime/_supervise.py | 408 ++++++++++++---------------------- 1 file changed, 138 insertions(+), 270 deletions(-) diff --git a/tractor/runtime/_supervise.py b/tractor/runtime/_supervise.py index 593e95a8..72ee8475 100644 --- a/tractor/runtime/_supervise.py +++ b/tractor/runtime/_supervise.py @@ -90,7 +90,7 @@ async def _try_cancel_then_kill( Sends a graceful actor-runtime cancel-RPC via `Portal.cancel_actor(raise_on_timeout=True)`. If the bounded-wait expires before the peer ack's, `ActorTooSlowError` is raised and - we escalate via `proc.kill()` per SC-discipline: + we escalate via `proc.terminate()` (SIGTERM) per SC-discipline: graceful cancel-req -> bounded wait -> hard-kill @@ -102,9 +102,11 @@ async def _try_cancel_then_kill( the wider write-up. ''' - # XXX, delay hard-kill escalation while any debugger guard - # below is active. Killing the sub immediately would tear down - # its tree and clobber an actor proxying a REPL session: + # XXX, do NOT escalate to `proc.terminate()` while ANY of + # the following are true — SIGTERM-ing a sub would tear + # down its sub-tree including any descendant proxying + # stdio to/from a REPL-locked actor, clobbering the user's + # debug session: # # - `Lock.ctx_in_debug is not None`: most precise — some # actor in the tree is currently REPL-locked. Set in the @@ -120,7 +122,7 @@ async def _try_cancel_then_kill( # child. # # - `debug_mode_active`: this nursery has at least one - # child started with an explicit `debug_mode=True` arg + # child started with an explicit `debug_mode=` arg # (`ActorNursery._at_least_one_child_in_debug`). Catches # the case where root is NOT in debug-mode but a # nursery-direct child opted in. @@ -130,57 +132,36 @@ async def _try_cancel_then_kill( # mutated by per-child `debug_mode=True`). ORing covers # every flavor without false-positively skipping # legitimate hard-kill paths in non-debug trees. - debug_protected: bool = ( + if ( debug.Lock.ctx_in_debug is not None or _state._runtime_vars.get('_debug_mode', False) or debug_mode_active - ) + ): + await portal.cancel_actor() + return try: - cancelled: bool = await portal.cancel_actor( - raise_on_timeout=not debug_protected, - ) - if not cancelled: - if debug_protected: - await debug.maybe_wait_for_debugger( - child_in_debug=( - debug_mode_active - or - debug.Lock.ctx_in_debug is not None - ), - header_msg=( - 'Delaying subproc hard-reap while ' - 'debugger locked..\n' - ), - ) - - peer_id: str = portal.channel.aid.reprol() - raise ActorTooSlowError( - f'Peer {peer_id} disconnected before ' - f'acknowledging its `Actor.cancel()` RPC' - ) - + await portal.cancel_actor(raise_on_timeout=True) except ActorTooSlowError as too_slow: log.error( f'Cancel-ack TIMED OUT for sub-actor\n' f' uid: {subactor.aid.reprol()!r}\n' f' reason: {too_slow}\n' - f'-> escalating to `proc.kill()` (hard-reap)\n' + f'-> escalating to `proc.terminate()` (hard-kill)\n' ) # XXX, the `subint` backend stores an `int` interp-id in the - # `proc` slot (not a `Process`), so it has no `.kill()`. + # `proc` slot (not a `Process`), so it has no `.terminate()`. # Guard here so a cancel-ack timeout doesn't `AttributeError` # once that backend lands; its hard-kill path is a TODO. - if hasattr(proc, 'kill'): - if proc.poll() is None: - proc.kill() + if hasattr(proc, 'terminate'): + proc.terminate() else: log.error( f'Cannot hard-kill sub-actor — backend proc-handle ' f'{proc!r} ({type(proc).__name__!r}) has no ' - f'`.kill()`!\n' + f'`.terminate()`!\n' f' uid: {subactor.aid.reprol()!r}\n' f'TODO: per-backend cancel-escalation.\n' ) @@ -238,14 +219,6 @@ class ActorNursery: ] = {} self._join_procs = trio.Event() - self._child_reap_requests: dict[ - tuple[str, str], - trio.Event, - ] = {} - self._child_reaped: dict[ - tuple[str, str], - trio.Event, - ] = {} self._at_least_one_child_in_debug: bool = False self.errors = errors self._scope_error: BaseException|None = None @@ -308,79 +281,6 @@ class ActorNursery: # self._cancelled_caught ) - def _register_child_reap( - self, - uid: tuple[str, str], - ) -> tuple[trio.Event, trio.Event]: - ''' - Register a child monitor's process-reap events. - - ''' - reap_request = trio.Event() - reaped = trio.Event() - self._child_reap_requests[uid] = reap_request - self._child_reaped[uid] = reaped - if self._join_procs.is_set(): - reap_request.set() - return reap_request, reaped - - def _request_reap_all(self) -> None: - ''' - Release every child monitor into its process-join phase. - - ''' - self._join_procs.set() - for reap_request in tuple( - self._child_reap_requests.values() - ): - reap_request.set() - - def _mark_child_reaped( - self, - uid: tuple[str, str], - ) -> None: - ''' - Publish completed child-process teardown to its waiter. - - ''' - self._children.pop(uid, None) - self._child_reap_requests.pop(uid, None) - reaped: trio.Event|None = self._child_reaped.pop( - uid, - None, - ) - if reaped is not None: - reaped.set() - - async def _cancel_and_reap_child( - self, - portal: Portal, - ) -> None: - ''' - Cancel, join and unregister one nursery-owned child. - - ''' - uid: tuple[str, str] = portal.channel.aid.uid - child_entry = self._children.get(uid) - if child_entry is None: - return - - subactor, proc, _ = child_entry - reap_request: trio.Event = self._child_reap_requests[uid] - reaped: trio.Event = self._child_reaped[uid] - - with trio.CancelScope(shield=True): - try: - await _try_cancel_then_kill( - portal, - proc, - subactor, - self._at_least_one_child_in_debug, - ) - finally: - reap_request.set() - await reaped.wait() - async def start_actor( self, name: str, @@ -425,7 +325,7 @@ class ActorNursery: # allow setting debug policy per actor if debug_mode is not None: _rtv['_debug_mode'] = debug_mode - self._at_least_one_child_in_debug |= debug_mode + self._at_least_one_child_in_debug = True enable_modules = list(enable_modules or []) proc_kwargs = dict(proc_kwargs or {}) @@ -572,7 +472,7 @@ class ActorNursery: # TODO: impl a repr for spawn more compact # then `._children`.. - children: tuple = tuple(self._children.values()) + children: dict = self._children child_count: int = len(children) msg: str = f'Cancelling actor nursery with {child_count} children\n' @@ -591,7 +491,7 @@ class ActorNursery: subactor, proc, portal, - ) in children: + ) in children.values(): # TODO: are we ever even going to use this or # is the spawning backend responsible for such @@ -611,9 +511,7 @@ class ActorNursery: await event.wait() # channel/portal should now be up - _, _, portal = self._children[ - subactor.aid.uid - ] + _, _, portal = children[subactor.aid.uid] # XXX should be impossible to get here # unless method was called from within @@ -660,14 +558,14 @@ class ActorNursery: subactor, proc, portal, - ) in children: + ) in children.values(): log.warning(f"Hard killing process {proc}") proc.terminate() else: self._cancelled_caught # mark ourselves as having (tried to have) cancelled all subactors - self._request_reap_all() + self._join_procs.set() async def _reap_ria_portals( @@ -725,9 +623,6 @@ async def _open_and_supervise_one_cancels_all_nursery( # normally don't need to show user by default __tracebackhide__: bool = hide_tb - outer_err: BaseException|None = None - inner_err: BaseException|None = None - # the collection of errors retreived from spawned sub-actors errors: dict[tuple[str, str], BaseException] = {} @@ -747,156 +642,129 @@ async def _open_and_supervise_one_cancels_all_nursery( errors ) try: - try: - # spawning of actors happens in the caller's scope - # after we yield upwards - yield an + # spawning of actors happens in the caller's scope + # after we yield upwards + yield an - # When we didn't error in the caller's scope, - # signal all process-monitor-tasks to conduct - # the "hard join phase". - log.runtime( - 'Waiting on subactors to complete:\n' - f'>}} {len(an._children)}\n' - ) - an._request_reap_all() + # When we didn't error in the caller's scope, + # signal all process-monitor-tasks to conduct + # the "hard join phase". + log.runtime( + 'Waiting on subactors to complete:\n' + f'>}} {len(an._children)}\n' + ) + an._request_reap_all() - # collect results (and errors) from all - # `.run_in_actor()` children then cancel - # each, one reaper task per child. - await _reap_ria_portals(an, errors) + # collect results (and errors) from all + # `.run_in_actor()` children then cancel + # each, one reaper task per child. + await _reap_ria_portals(an, errors) - except BaseException as _inner_err: - inner_err = _inner_err - errors[actor.aid.uid] = inner_err + # Single one-cancels-all handler for the (now single) + # daemon nursery. Pre-#477 a 2ndary `._ria_nursery` + # required a separate *outer* handler to catch errors + # bubbling from its task-reaping `__aexit__`; with that + # nursery gone this lone handler covers every scope + # error. NB: we deliberately do NOT re-raise here — the + # `finally` below raises the collected `errors` (as a + # single exc or `BaseExceptionGroup`), which already + # superseded the old outer handler's `raise` anyway + # since `errors` is populated (below) before any await. + except BaseException as _scope_err: + an._scope_error = _scope_err + errors[actor.aid.uid] = _scope_err - # If we error in the root but the debugger is - # engaged we don't want to prematurely kill (and - # thus clobber access to) the local tty since it - # will make the pdb repl unusable. - # Instead try to wait for pdb to be released before - # tearing down. - await debug.maybe_wait_for_debugger( - child_in_debug=an._at_least_one_child_in_debug - ) - - # if the caller's scope errored then we activate our - # one-cancels-all supervisor strategy (don't - # worry more are coming). - an._request_reap_all() - - # XXX NOTE XXX: hypothetically an error could - # be raised and then a cancel signal shows up - # slightly after in which case the `else:` - # block here might not complete? For now, - # shield both. - with trio.CancelScope(shield=True): - etype: type = type(inner_err) - if etype in ( - trio.Cancelled, - KeyboardInterrupt, - ) or ( - is_multi_cancelled(inner_err) - ): - log.cancel( - f'Actor-nursery cancelled by {etype}\n\n' - - f'{current_actor().aid.uid}\n' - f' |_{an}\n\n' - - # TODO: show tb str? - # f'{tb_str}' - ) - elif etype in { - ContextCancelled, - }: - log.cancel( - 'Actor-nursery caught remote cancellation\n' - '\n' - f'{inner_err.tb_str}' - ) - else: - log.exception( - 'Nursery errored with:\n' - - # TODO: same thing as in - # `._invoke()` to compute how to - # place this div-line in the - # middle of the above msg - # content.. - # -[ ] prolly helper-func it too - # in our `.log` module.. - # '------ - ------' - ) - - # snapshot `.run_in_actor()` children - # BEFORE cancelling: each backend - # spawn-task pops its `._children` - # entry as the proc gets reaped. - ria_children: list = [ - (portal, subactor) - for subactor, _, portal - in an._children.values() - if portal in - an._cancel_after_result_on_exit - ] - # cancel all subactors - await an.cancel() - - # then collect any already-relayed - # results/errors from ria children. - # Tightly bounded: anything - # collectable is already queued in - # the local ctx (relayed BEFORE the - # cancel above); a child hard-killed - # without relaying just parks its - # reaper which then self-cleans (a - # `trio.Cancelled` result is never - # stashed), mirroring the old - # backend-side reaper-vs-`soft_kill` - # cancel race. - with trio.move_on_after(0.5): - await _reap_ria_portals( - an, - errors, - ria_children=ria_children, - ) - - # Safety-net handler for anything escaping the inner - # `except BaseException` above (e.g. a `trio.Cancelled` - # during its non-shielded debugger-wait, or a failure - # inside the shielded teardown itself). Post-#477 the - # 2ndary `._ria_nursery` (and its dedicated handler) is - # gone; this now guards the single daemon nursery scope. - # TODO: can this be merged into the inner handler now that - # there's only one nursery? (higher-risk, own PR) - except ( - Exception, - BaseExceptionGroup, - trio.Cancelled - ) as _outer_err: - outer_err = _outer_err - - an._scope_error = outer_err or inner_err - - # XXX: yet another guard before allowing the cancel - # sequence in case a (single) child is in debug. + # If we error in the root but the debugger is + # engaged we don't want to prematurely kill (and + # thus clobber access to) the local tty since it + # will make the pdb repl unusable. + # Instead try to wait for pdb to be released before + # tearing down. await debug.maybe_wait_for_debugger( child_in_debug=an._at_least_one_child_in_debug ) - # If actor-local error was raised while waiting on - # ".run_in_actor()" actors then we also want to cancel all - # remaining sub-actors (due to our lone strategy: - # one-cancels-all). - if an._children: - log.cancel( - 'Actor-nursery cancelling due error type:\n' - f'{outer_err}\n' - ) - with trio.CancelScope(shield=True): - await an.cancel() - raise + # if the caller's scope errored then we activate our + # one-cancels-all supervisor strategy (don't + # worry more are coming). + an._request_reap_all() + + # XXX NOTE XXX: hypothetically an error could + # be raised and then a cancel signal shows up + # slightly after in which case the `else:` + # block here might not complete? For now, + # shield both. + with trio.CancelScope(shield=True): + etype: type = type(_scope_err) + if etype in ( + trio.Cancelled, + KeyboardInterrupt, + ) or ( + is_multi_cancelled(_scope_err) + ): + log.cancel( + f'Actor-nursery cancelled by {etype}\n\n' + + f'{current_actor().aid.uid}\n' + f' |_{an}\n\n' + + # TODO: show tb str? + # f'{tb_str}' + ) + elif etype in { + ContextCancelled, + }: + log.cancel( + 'Actor-nursery caught remote cancellation\n' + '\n' + f'{_scope_err.tb_str}' + ) + else: + log.exception( + 'Nursery errored with:\n' + + # TODO: same thing as in + # `._invoke()` to compute how to + # place this div-line in the + # middle of the above msg + # content.. + # -[ ] prolly helper-func it too + # in our `.log` module.. + # '------ - ------' + ) + + # snapshot `.run_in_actor()` children + # BEFORE cancelling: each backend + # spawn-task pops its `._children` + # entry as the proc gets reaped. + ria_children: list = [ + (portal, subactor) + for subactor, _, portal + in an._children.values() + if portal in + an._cancel_after_result_on_exit + ] + # cancel all subactors + await an.cancel() + + # then collect any already-relayed + # results/errors from ria children. + # Tightly bounded: anything + # collectable is already queued in + # the local ctx (relayed BEFORE the + # cancel above); a child hard-killed + # without relaying just parks its + # reaper which then self-cleans (a + # `trio.Cancelled` result is never + # stashed), mirroring the old + # backend-side reaper-vs-`soft_kill` + # cancel race. + with trio.move_on_after(0.5): + await _reap_ria_portals( + an, + errors, + ria_children=ria_children, + ) finally: # No errors were raised while awaiting ".run_in_actor()"