Add `psutil` to `--dev` / testing deps

Factor `breakpoint()` blocking into `@acm`
Call it `maybe_block_bp()` can wrap the `open_root_actor()` body with it. Main reason is to guarantee we can bp inside actor runtime bootup as needed when debugging internals! Prolly should factor this to another module tho? ALSO, ensure we RTE on recurrent entries to `open_root_actor()` from within an existing tree! There was actually `test_spawning` test somehow getting away with this!? Should never be possible or allowed!
2025-04-01 22:26:56 -04:00 · 2025-04-01 22:25:29 -04:00 · 2025-04-01 22:21:28 -04:00
5 changed files with 513 additions and 346 deletions
--- a/pyproject.toml
+++ b/pyproject.toml
@ -64,6 +64,7 @@ dev = [
  "pyperclip>=1.9.0",
  "prompt-toolkit>=3.0.50",
  "xonsh>=0.19.2",
  "psutil>=7.0.0",
 ]
 # TODO, add these with sane versions; were originally in
 # `requirements-docs.txt`..
--- a/tractor/_root.py
+++ b/tractor/_root.py
@ -18,7 +18,9 @@
 Root actor runtime ignition(s).
 '''
-from contextlib import asynccontextmanager as acm
+from contextlib import (
    asynccontextmanager as acm,
 )
 from functools import partial
 import importlib
 import inspect
@ -26,7 +28,10 @@ import logging
 import os
 import signal
 import sys
-from typing import Callable
+from typing import (
    Any,
    Callable,
 )
 import warnings
@ -47,21 +52,95 @@ from .ipc import (
    _connect_chan,
 )
 from ._addr import (
    Address,
    UnwrappedAddress,
    default_lo_addrs,
    mk_uuid,
    preferred_transport,
    wrap_address,
 )
-from ._exceptions import is_multi_cancelled
+from ._exceptions import (
    ActorFailure,
    is_multi_cancelled,
 )
 logger = log.get_logger('tractor')
 # TODO: stick this in a `@acm` defined in `devx._debug`?
 # -[ ] also maybe consider making this a `wrapt`-deco to
 #     save an indent level?
 #
@acm
 async def maybe_block_bp(
    debug_mode: bool,
    maybe_enable_greenback: bool,
 ) -> bool:
    # Override the global debugger hook to make it play nice with
    # ``trio``, see much discussion in:
    # https://github.com/python-trio/trio/issues/1155#issuecomment-742964018
    builtin_bp_handler: Callable = sys.breakpointhook
    orig_bp_path: str|None = os.environ.get(
        'PYTHONBREAKPOINT',
        None,
    )
    bp_blocked: bool
    if (
        debug_mode
        and maybe_enable_greenback
        and (
            maybe_mod := await _debug.maybe_init_greenback(
                raise_not_found=False,
            )
        )
    ):
        logger.info(
            f'Found `greenback` installed @ {maybe_mod}\n'
            'Enabling `tractor.pause_from_sync()` support!\n'
        )
        os.environ['PYTHONBREAKPOINT'] = (
            'tractor.devx._debug._sync_pause_from_builtin'
        )
        _state._runtime_vars['use_greenback'] = True
        bp_blocked = False
    else:
        # TODO: disable `breakpoint()` by default (without
        # `greenback`) since it will break any multi-actor
        # usage by a clobbered TTY's stdstreams!
        def block_bps(*args, **kwargs):
            raise RuntimeError(
                'Trying to use `breakpoint()` eh?\n\n'
                'Welp, `tractor` blocks `breakpoint()` built-in calls by default!\n'
                'If you need to use it please install `greenback` and set '
                '`debug_mode=True` when opening the runtime '
                '(either via `.open_nursery()` or `open_root_actor()`)\n'
            )
        sys.breakpointhook = block_bps
        # lol ok,
        # https://docs.python.org/3/library/sys.html#sys.breakpointhook
        os.environ['PYTHONBREAKPOINT'] = "0"
        bp_blocked = True
    try:
        yield bp_blocked
    finally:
        # restore any prior built-in `breakpoint()` hook state
        if builtin_bp_handler is not None:
            sys.breakpointhook = builtin_bp_handler
        if orig_bp_path is not None:
            os.environ['PYTHONBREAKPOINT'] = orig_bp_path
        else:
            # clear env back to having no entry
            os.environ.pop('PYTHONBREAKPOINT', None)
@acm
 async def open_root_actor(
    *,
    # defaults are above
    registry_addrs: list[UnwrappedAddress]|None = None,
@ -111,355 +190,323 @@ async def open_root_actor(
    Runtime init entry point for ``tractor``.
    '''
-    _debug.hide_runtime_frames()
+    # XXX NEVER allow nested actor-trees!
-    __tracebackhide__: bool = hide_tb
+    if already_actor := _state.current_actor(err_on_no_runtime=False):
-
+        rtvs: dict[str, Any] = _state._runtime_vars
-    # TODO: stick this in a `@cm` defined in `devx._debug`?
+        root_mailbox: list[str, int] = rtvs['_root_mailbox']
-    #
+        registry_addrs: list[list[str, int]] = rtvs['_registry_addrs']
-    # Override the global debugger hook to make it play nice with
+        raise ActorFailure(
-    # ``trio``, see much discussion in:
+            f'A current actor already exists !?\n'
-    # https://github.com/python-trio/trio/issues/1155#issuecomment-742964018
+            f'({already_actor}\n'
-    builtin_bp_handler: Callable = sys.breakpointhook
+            f'\n'
-    orig_bp_path: str|None = os.environ.get(
+            f'You can NOT open a second root actor from within '
-        'PYTHONBREAKPOINT',
+            f'an existing tree and the current root of this '
-        None,
+            f'already exists !!\n'
-    )
+            f'\n'
-    if (
+            f'_root_mailbox: {root_mailbox!r}\n'
-        debug_mode
+            f'_registry_addrs: {registry_addrs!r}\n'
        and maybe_enable_greenback
        and (
            maybe_mod := await _debug.maybe_init_greenback(
                raise_not_found=False,
            )
        )
    async with maybe_block_bp(
        debug_mode=debug_mode,
        maybe_enable_greenback=maybe_enable_greenback,
    ):
-        logger.info(
+        _debug.hide_runtime_frames()
-            f'Found `greenback` installed @ {maybe_mod}\n'
+        __tracebackhide__: bool = hide_tb
            'Enabling `tractor.pause_from_sync()` support!\n'
        )
        os.environ['PYTHONBREAKPOINT'] = (
            'tractor.devx._debug._sync_pause_from_builtin'
        )
        _state._runtime_vars['use_greenback'] = True
-    else:
+        # attempt to retreive ``trio``'s sigint handler and stash it
-        # TODO: disable `breakpoint()` by default (without
+        # on our debugger lock state.
-        # `greenback`) since it will break any multi-actor
+        _debug.DebugStatus._trio_handler = signal.getsignal(signal.SIGINT)
-        # usage by a clobbered TTY's stdstreams!
+
-        def block_bps(*args, **kwargs):
+        # mark top most level process as root actor
-            raise RuntimeError(
+        _state._runtime_vars['_is_root'] = True
-                'Trying to use `breakpoint()` eh?\n\n'
+
-                'Welp, `tractor` blocks `breakpoint()` built-in calls by default!\n'
+        # caps based rpc list
-                'If you need to use it please install `greenback` and set '
+        enable_modules = (
-                '`debug_mode=True` when opening the runtime '
+            enable_modules
-                '(either via `.open_nursery()` or `open_root_actor()`)\n'
+            or
            []
        )
        if rpc_module_paths:
            warnings.warn(
                "`rpc_module_paths` is now deprecated, use "
                " `enable_modules` instead.",
                DeprecationWarning,
                stacklevel=2,
            )
            enable_modules.extend(rpc_module_paths)
        if start_method is not None:
            _spawn.try_set_start_method(start_method)
        # TODO! remove this ASAP!
        if arbiter_addr is not None:
            warnings.warn(
                '`arbiter_addr` is now deprecated\n'
                'Use `registry_addrs: list[tuple]` instead..',
                DeprecationWarning,
                stacklevel=2,
            )
            registry_addrs = [arbiter_addr]
        if not registry_addrs:
            registry_addrs: list[UnwrappedAddress] = default_lo_addrs(
                enable_transports
            )
-        sys.breakpointhook = block_bps
+        assert registry_addrs
        # lol ok,
        # https://docs.python.org/3/library/sys.html#sys.breakpointhook
        os.environ['PYTHONBREAKPOINT'] = "0"
-    # attempt to retreive ``trio``'s sigint handler and stash it
+        loglevel = (
-    # on our debugger lock state.
+            loglevel
-    _debug.DebugStatus._trio_handler = signal.getsignal(signal.SIGINT)
+            or log._default_loglevel
        ).upper()
    # mark top most level process as root actor
    _state._runtime_vars['_is_root'] = True
    # caps based rpc list
    enable_modules = (
        enable_modules
        or
        []
    )
    if rpc_module_paths:
        warnings.warn(
            "`rpc_module_paths` is now deprecated, use "
            " `enable_modules` instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        enable_modules.extend(rpc_module_paths)
    if start_method is not None:
        _spawn.try_set_start_method(start_method)
    if arbiter_addr is not None:
        warnings.warn(
            '`arbiter_addr` is now deprecated\n'
            'Use `registry_addrs: list[tuple]` instead..',
            DeprecationWarning,
            stacklevel=2,
        )
        registry_addrs = [arbiter_addr]
    if not registry_addrs:
        registry_addrs: list[UnwrappedAddress] = default_lo_addrs(
            enable_transports
        )
    assert registry_addrs
    loglevel = (
        loglevel
        or log._default_loglevel
    ).upper()
    if (
        debug_mode
        and _spawn._spawn_method == 'trio'
    ):
        _state._runtime_vars['_debug_mode'] = True
        # expose internal debug module to every actor allowing for
        # use of ``await tractor.pause()``
        enable_modules.append('tractor.devx._debug')
        # if debug mode get's enabled *at least* use that level of
        # logging for some informative console prompts.
        if (
-            logging.getLevelName(
+            debug_mode
-                # lul, need the upper case for the -> int map?
+            and _spawn._spawn_method == 'trio'
                # sweet "dynamic function behaviour" stdlib...
                loglevel,
            ) > logging.getLevelName('PDB')
        ):
-            loglevel = 'PDB'
+            _state._runtime_vars['_debug_mode'] = True
            # expose internal debug module to every actor allowing for
            # use of ``await tractor.pause()``
            enable_modules.append('tractor.devx._debug')
            # if debug mode get's enabled *at least* use that level of
            # logging for some informative console prompts.
            if (
                logging.getLevelName(
                    # lul, need the upper case for the -> int map?
                    # sweet "dynamic function behaviour" stdlib...
                    loglevel,
                ) > logging.getLevelName('PDB')
            ):
                loglevel = 'PDB'
-    elif debug_mode:
+        elif debug_mode:
        raise RuntimeError(
            "Debug mode is only supported for the `trio` backend!"
        )
    assert loglevel
    _log = log.get_console_log(loglevel)
    assert _log
    # TODO: factor this into `.devx._stackscope`!!
    if (
        debug_mode
        and
        enable_stack_on_sig
    ):
        from .devx._stackscope import enable_stack_on_sig
        enable_stack_on_sig()
    # closed into below ping task-func
    ponged_addrs: list[UnwrappedAddress] = []
    async def ping_tpt_socket(
        addr: UnwrappedAddress,
        timeout: float = 1,
    ) -> None:
        '''
        Attempt temporary connection to see if a registry is
        listening at the requested address by a tranport layer
        ping.
        If a connection can't be made quickly we assume none no
        server is listening at that addr.
        '''
        try:
            # TODO: this connect-and-bail forces us to have to
            # carefully rewrap TCP 104-connection-reset errors as
            # EOF so as to avoid propagating cancel-causing errors
            # to the channel-msg loop machinery. Likely it would
            # be better to eventually have a "discovery" protocol
            # with basic handshake instead?
            with trio.move_on_after(timeout):
                async with _connect_chan(addr):
                    ponged_addrs.append(addr)
        except OSError:
            # TODO: make this a "discovery" log level?
            logger.info(
                f'No actor registry found @ {addr}\n'
            )
    async with trio.open_nursery() as tn:
        for addr in registry_addrs:
            tn.start_soon(
                ping_tpt_socket,
                addr,
            )
    trans_bind_addrs: list[UnwrappedAddress] = []
    # Create a new local root-actor instance which IS NOT THE
    # REGISTRAR
    if ponged_addrs:
        if ensure_registry:
            raise RuntimeError(
-                 f'Failed to open `{name}`@{ponged_addrs}: '
+                "Debug mode is only supported for the `trio` backend!"
                'registry socket(s) already bound'
            )
-        # we were able to connect to an arbiter
+        assert loglevel
-        logger.info(
+        _log = log.get_console_log(loglevel)
-            f'Registry(s) seem(s) to exist @ {ponged_addrs}'
+        assert _log
        )
-        actor = Actor(
+        # TODO: factor this into `.devx._stackscope`!!
            name=name or 'anonymous',
            uuid=mk_uuid(),
            registry_addrs=ponged_addrs,
            loglevel=loglevel,
            enable_modules=enable_modules,
        )
        # DO NOT use the registry_addrs as the transport server
        # addrs for this new non-registar, root-actor.
        for addr in ponged_addrs:
            waddr = wrap_address(addr)
            print(waddr)
            trans_bind_addrs.append(
                waddr.get_random(namespace=waddr.namespace)
            )
    # Start this local actor as the "registrar", aka a regular
    # actor who manages the local registry of "mailboxes" of
    # other process-tree-local sub-actors.
    else:
        # NOTE that if the current actor IS THE REGISTAR, the
        # following init steps are taken:
        # - the tranport layer server is bound to each addr
        #   pair defined in provided registry_addrs, or the default.
        trans_bind_addrs = registry_addrs
        # - it is normally desirable for any registrar to stay up
        #   indefinitely until either all registered (child/sub)
        #   actors are terminated (via SC supervision) or,
        #   a re-election process has taken place. 
        # NOTE: all of ^ which is not implemented yet - see:
        # https://github.com/goodboy/tractor/issues/216
        # https://github.com/goodboy/tractor/pull/348
        # https://github.com/goodboy/tractor/issues/296
        actor = Arbiter(
            name=name or 'registrar',
            uuid=mk_uuid(),
            registry_addrs=registry_addrs,
            loglevel=loglevel,
            enable_modules=enable_modules,
        )
        # XXX, in case the root actor runtime was actually run from
        # `tractor.to_asyncio.run_as_asyncio_guest()` and NOt
        # `.trio.run()`.
        actor._infected_aio = _state._runtime_vars['_is_infected_aio']
    # Start up main task set via core actor-runtime nurseries.
    try:
        # assign process-local actor
        _state._current_actor = actor
        # start local channel-server and fake the portal API
        # NOTE: this won't block since we provide the nursery
        ml_addrs_str: str = '\n'.join(
            f'@{addr}' for addr in trans_bind_addrs
        )
        logger.info(
            f'Starting local {actor.uid} on the following transport addrs:\n'
            f'{ml_addrs_str}'
        )
        # start the actor runtime in a new task
        async with trio.open_nursery(
            strict_exception_groups=False,
            # ^XXX^ TODO? instead unpack any RAE as per "loose" style?
        ) as nursery:
            # ``_runtime.async_main()`` creates an internal nursery
            # and blocks here until any underlying actor(-process)
            # tree has terminated thereby conducting so called
            # "end-to-end" structured concurrency throughout an
            # entire hierarchical python sub-process set; all
            # "actor runtime" primitives are SC-compat and thus all
            # transitively spawned actors/processes must be as
            # well.
            await nursery.start(
                partial(
                    async_main,
                    actor,
                    accept_addrs=trans_bind_addrs,
                    parent_addr=None
                )
            )
            try:
                yield actor
            except (
                Exception,
                BaseExceptionGroup,
            ) as err:
                # TODO, in beginning to handle the subsubactor with
                # crashed grandparent cases..
                #
                # was_locked: bool = await _debug.maybe_wait_for_debugger(
                #     child_in_debug=True,
                # )
                # XXX NOTE XXX see equiv note inside
                # `._runtime.Actor._stream_handler()` where in the
                # non-root or root-that-opened-this-mahually case we
                # wait for the local actor-nursery to exit before
                # exiting the transport channel handler.
                entered: bool = await _debug._maybe_enter_pm(
                    err,
                    api_frame=inspect.currentframe(),
                    debug_filter=debug_filter,
                )
                if (
                    not entered
                    and
                    not is_multi_cancelled(
                        err,
                    )
                ):
                    logger.exception('Root actor crashed\n')
                # ALWAYS re-raise any error bubbled up from the
                # runtime!
                raise
            finally:
                # NOTE: not sure if we'll ever need this but it's
                # possibly better for even more determinism?
                # logger.cancel(
                #     f'Waiting on {len(nurseries)} nurseries in root..')
                # nurseries = actor._actoruid2nursery.values()
                # async with trio.open_nursery() as tempn:
                #     for an in nurseries:
                #         tempn.start_soon(an.exited.wait)
                logger.info(
                    'Closing down root actor'
                )
                await actor.cancel(None)  # self cancel
    finally:
        _state._current_actor = None
        _state._last_actor_terminated = actor
        # restore built-in `breakpoint()` hook state
        if (
            debug_mode
            and
-            maybe_enable_greenback
+            enable_stack_on_sig
        ):
-            if builtin_bp_handler is not None:
+            from .devx._stackscope import enable_stack_on_sig
-                sys.breakpointhook = builtin_bp_handler
+            enable_stack_on_sig()
-            if orig_bp_path is not None:
+        # closed into below ping task-func
-                os.environ['PYTHONBREAKPOINT'] = orig_bp_path
+        ponged_addrs: list[UnwrappedAddress] = []
-            else:
+        async def ping_tpt_socket(
-                # clear env back to having no entry
+            addr: UnwrappedAddress,
-                os.environ.pop('PYTHONBREAKPOINT', None)
+            timeout: float = 1,
        ) -> None:
            '''
            Attempt temporary connection to see if a registry is
            listening at the requested address by a tranport layer
            ping.
-        logger.runtime("Root actor terminated")
+            If a connection can't be made quickly we assume none no
            server is listening at that addr.
            '''
            try:
                # TODO: this connect-and-bail forces us to have to
                # carefully rewrap TCP 104-connection-reset errors as
                # EOF so as to avoid propagating cancel-causing errors
                # to the channel-msg loop machinery. Likely it would
                # be better to eventually have a "discovery" protocol
                # with basic handshake instead?
                with trio.move_on_after(timeout):
                    async with _connect_chan(addr):
                        ponged_addrs.append(addr)
            except OSError:
                # TODO: make this a "discovery" log level?
                logger.info(
                    f'No actor registry found @ {addr}\n'
                )
        async with trio.open_nursery() as tn:
            for addr in registry_addrs:
                tn.start_soon(
                    ping_tpt_socket,
                    addr,
                )
        trans_bind_addrs: list[UnwrappedAddress] = []
        # Create a new local root-actor instance which IS NOT THE
        # REGISTRAR
        if ponged_addrs:
            if ensure_registry:
                raise RuntimeError(
                     f'Failed to open `{name}`@{ponged_addrs}: '
                    'registry socket(s) already bound'
                )
            # we were able to connect to an arbiter
            logger.info(
                f'Registry(s) seem(s) to exist @ {ponged_addrs}'
            )
            actor = Actor(
                name=name or 'anonymous',
                uuid=mk_uuid(),
                registry_addrs=ponged_addrs,
                loglevel=loglevel,
                enable_modules=enable_modules,
            )
            # DO NOT use the registry_addrs as the transport server
            # addrs for this new non-registar, root-actor.
            for addr in ponged_addrs:
                waddr: Address = wrap_address(addr)
                trans_bind_addrs.append(
                    waddr.get_random(bindspace=waddr.bindspace)
                )
        # Start this local actor as the "registrar", aka a regular
        # actor who manages the local registry of "mailboxes" of
        # other process-tree-local sub-actors.
        else:
            # NOTE that if the current actor IS THE REGISTAR, the
            # following init steps are taken:
            # - the tranport layer server is bound to each addr
            #   pair defined in provided registry_addrs, or the default.
            trans_bind_addrs = registry_addrs
            # - it is normally desirable for any registrar to stay up
            #   indefinitely until either all registered (child/sub)
            #   actors are terminated (via SC supervision) or,
            #   a re-election process has taken place.
            # NOTE: all of ^ which is not implemented yet - see:
            # https://github.com/goodboy/tractor/issues/216
            # https://github.com/goodboy/tractor/pull/348
            # https://github.com/goodboy/tractor/issues/296
            actor = Arbiter(
                name=name or 'registrar',
                uuid=mk_uuid(),
                registry_addrs=registry_addrs,
                loglevel=loglevel,
                enable_modules=enable_modules,
            )
            # XXX, in case the root actor runtime was actually run from
            # `tractor.to_asyncio.run_as_asyncio_guest()` and NOt
            # `.trio.run()`.
            actor._infected_aio = _state._runtime_vars['_is_infected_aio']
        # Start up main task set via core actor-runtime nurseries.
        try:
            # assign process-local actor
            _state._current_actor = actor
            # start local channel-server and fake the portal API
            # NOTE: this won't block since we provide the nursery
            ml_addrs_str: str = '\n'.join(
                f'@{addr}' for addr in trans_bind_addrs
            )
            logger.info(
                f'Starting local {actor.uid} on the following transport addrs:\n'
                f'{ml_addrs_str}'
            )
            # start the actor runtime in a new task
            async with trio.open_nursery(
                strict_exception_groups=False,
                # ^XXX^ TODO? instead unpack any RAE as per "loose" style?
            ) as nursery:
                # ``_runtime.async_main()`` creates an internal nursery
                # and blocks here until any underlying actor(-process)
                # tree has terminated thereby conducting so called
                # "end-to-end" structured concurrency throughout an
                # entire hierarchical python sub-process set; all
                # "actor runtime" primitives are SC-compat and thus all
                # transitively spawned actors/processes must be as
                # well.
                await nursery.start(
                    partial(
                        async_main,
                        actor,
                        accept_addrs=trans_bind_addrs,
                        parent_addr=None
                    )
                )
                try:
                    yield actor
                except (
                    Exception,
                    BaseExceptionGroup,
                ) as err:
                    # TODO, in beginning to handle the subsubactor with
                    # crashed grandparent cases..
                    #
                    # was_locked: bool = await _debug.maybe_wait_for_debugger(
                    #     child_in_debug=True,
                    # )
                    # XXX NOTE XXX see equiv note inside
                    # `._runtime.Actor._stream_handler()` where in the
                    # non-root or root-that-opened-this-mahually case we
                    # wait for the local actor-nursery to exit before
                    # exiting the transport channel handler.
                    entered: bool = await _debug._maybe_enter_pm(
                        err,
                        api_frame=inspect.currentframe(),
                        debug_filter=debug_filter,
                    )
                    if (
                        not entered
                        and
                        not is_multi_cancelled(
                            err,
                        )
                    ):
                        logger.exception(
                            'Root actor crashed\n'
                            f'>x)\n'
                            f' |_{actor}\n'
                        )
                    # ALWAYS re-raise any error bubbled up from the
                    # runtime!
                    raise
                finally:
                    # NOTE: not sure if we'll ever need this but it's
                    # possibly better for even more determinism?
                    # logger.cancel(
                    #     f'Waiting on {len(nurseries)} nurseries in root..')
                    # nurseries = actor._actoruid2nursery.values()
                    # async with trio.open_nursery() as tempn:
                    #     for an in nurseries:
                    #         tempn.start_soon(an.exited.wait)
                    logger.info(
                        f'Closing down root actor\n'
                        f'>)\n'
                        f'|_{actor}\n'
                    )
                    await actor.cancel(None)  # self cancel
        finally:
            _state._current_actor = None
            _state._last_actor_terminated = actor
            logger.runtime(
                f'Root actor terminated\n'
                f')>\n'
                f' |_{actor}\n'
            )
 def run_daemon(
--- a/tractor/_runtime.py
+++ b/tractor/_runtime.py
@ -200,9 +200,14 @@ class Actor:
        phase (aka before a new process is executed).
        '''
-        self.name = name
+        self._aid = msgtypes.Aid(
-        self.uid = (name, uuid)
+            name=name,
            uuid=uuid,
            pid=os.getpid(),
        )
        self._task: trio.Task|None = None
        # state
        self._cancel_complete = trio.Event()
        self._cancel_called_by_remote: tuple[str, tuple]|None = None
        self._cancel_called: bool = False
@ -281,6 +286,77 @@ class Actor:
            self.reg_addrs: list[UnwrappedAddress] = registry_addrs
            _state._runtime_vars['_registry_addrs'] = registry_addrs
    @property
    def aid(self) -> msgtypes.Aid:
        '''
        This process-singleton-actor's "unique ID" in struct form.
        '''
        return self._aid
    @property
    def name(self) -> str:
        return self._aid.name
    @property
    def uid(self) -> tuple[str, str]:
        '''
        This process-singleton's "unique (cross-host) ID".
        Delivered from the `.Aid.name/.uuid` fields as a `tuple` pair
        and should be multi-host unique despite a large distributed
        process plane.
        '''
        return (
            self._aid.name,
            self._aid.uuid,
        )
    @property
    def pid(self) -> int:
        return self._aid.pid
    def pformat(self) -> str:
        ds: str = '='
        parent_uid: tuple|None = None
        if rent_chan := self._parent_chan:
            parent_uid = rent_chan.uid
        peers: list[tuple] = list(self._peer_connected)
        listen_addrs: str = pformat(self._listen_addrs)
        fmtstr: str = (
            f' |_id: {self.aid!r}\n'
            # f"   aid{ds}{self.aid!r}\n"
            f"   parent{ds}{parent_uid}\n"
            f'\n'
            f' |_ipc: {len(peers)!r} connected peers\n'
            f"   peers{ds}{peers!r}\n"
            f"   _listen_addrs{ds}'{listen_addrs}'\n"
            f"   _listeners{ds}'{self._listeners}'\n"
            f'\n'
            f' |_rpc: {len(self._rpc_tasks)} tasks\n'
            f"   ctxs{ds}{len(self._contexts)}\n"
            f'\n'
            f' |_runtime: ._task{ds}{self._task!r}\n'
            f'   _spawn_method{ds}{self._spawn_method}\n'
            f'   _actoruid2nursery{ds}{self._actoruid2nursery}\n'
            f'   _forkserver_info{ds}{self._forkserver_info}\n'
            f'\n'
            f' |_state: "TODO: .repr_state()"\n'
            f'   _cancel_complete{ds}{self._cancel_complete}\n'
            f'   _cancel_called_by_remote{ds}{self._cancel_called_by_remote}\n'
            f'   _cancel_called{ds}{self._cancel_called}\n'
        )
        return (
            '<Actor(\n'
            +
            fmtstr
            +
            ')>\n'
        )
    __repr__ = pformat
    @property
    def reg_addrs(self) -> list[UnwrappedAddress]:
        '''
@ -421,12 +497,19 @@ class Actor:
        try:
            uid: tuple|None = await self._do_handshake(chan)
        except (
-            # we need this for ``msgspec`` for some reason?
+            TransportClosed,
-            # for now, it's been put in the stream backend.
+            # ^XXX NOTE, the above wraps `trio` exc types raised
            # during various `SocketStream.send/receive_xx()` calls
            # under different fault conditions such as,
            #
            # trio.BrokenResourceError,
            # trio.ClosedResourceError,
-
+            #
-            TransportClosed,
+            # Inside our `.ipc._transport` layer we absorb and
            # re-raise our own `TransportClosed` exc such that this
            # higher level runtime code can only worry one
            # "kinda-error" that we expect to tolerate during
            # discovery-sys related pings, queires, DoS etc.
        ):
            # XXX: This may propagate up from `Channel._aiter_recv()`
            # and `MsgpackStream._inter_packets()` on a read from the
@ -1205,7 +1288,8 @@ class Actor:
        task_status: TaskStatus[Nursery] = trio.TASK_STATUS_IGNORED,
    ) -> None:
        '''
-        Start the IPC transport server, begin listening for new connections.
+        Start the IPC transport server, begin listening/accepting new
        `trio.SocketStream` connections.
        This will cause an actor to continue living (and thus
        blocking at the process/OS-thread level) until
@ -1223,10 +1307,24 @@ class Actor:
        self._server_down = trio.Event()
        try:
            async with trio.open_nursery() as server_n:
-                listeners: list[trio.abc.Listener] = [
+
-                    await addr.open_listener()
+                listeners: list[trio.abc.Listener] = []
-                    for addr in listen_addrs
+                for addr in listen_addrs:
-                ]
+                    try:
                        listener: trio.abc.Listener = await addr.open_listener()
                    except OSError as oserr:
                        if (
                            '[Errno 98] Address already in use'
                            in
                            oserr.args[0]
                        ):
                            log.exception(
                                f'Address already in use?\n'
                                f'{addr}\n'
                            )
                            raise
                    listeners.append(listener)
                await server_n.start(
                    partial(
                        trio.serve_listeners,
@ -1249,8 +1347,10 @@ class Actor:
                task_status.started(server_n)
        finally:
            addr: Address
            for addr in listen_addrs:
-                await addr.close_listener()
+                addr.close_listener()
            # signal the server is down since nursery above terminated
            self._server_down.set()
@ -1717,6 +1817,8 @@ async def async_main(
    the actor's "runtime" and all thus all ongoing RPC tasks.
    '''
    actor._task: trio.Task = trio.lowlevel.current_task()
    # attempt to retreive ``trio``'s sigint handler and stash it
    # on our debugger state.
    _debug.DebugStatus._trio_handler = signal.getsignal(signal.SIGINT)
@ -1726,18 +1828,17 @@ async def async_main(
        # establish primary connection with immediate parent
        actor._parent_chan: Channel|None = None
        if parent_addr is not None:
        if parent_addr is not None:
            (
                actor._parent_chan,
                set_accept_addr_says_rent,
                maybe_preferred_transports_says_rent,
            ) = await actor._from_parent(parent_addr)
-
+            accept_addrs: list[UnwrappedAddress] = []
            # either it's passed in because we're not a child or
            # because we're running in mp mode
            accept_addrs: list[UnwrappedAddress] = []
            if (
                set_accept_addr_says_rent
                and
--- a/tractor/msg/types.py
+++ b/tractor/msg/types.py
@ -143,6 +143,7 @@ class Aid(
    '''
    name: str
    uuid: str
    pid: int|None = None
    # TODO? can/should we extend this field set?
    # -[ ] use built-in support for UUIDs? `uuid.UUID` which has
--- a/uv.lock
+++ b/uv.lock
@ -257,6 +257,21 @@ wheels = [
    { url = "https://files.pythonhosted.org/packages/e4/ea/d836f008d33151c7a1f62caf3d8dd782e4d15f6a43897f64480c2b8de2ad/prompt_toolkit-3.0.50-py3-none-any.whl", hash = "sha256:9b6427eb19e479d98acff65196a307c555eb567989e6d88ebbb1b509d9779198", size = 387816 },
 ]
 [[package]]
 name = "psutil"
 version = "7.0.0"
 source = { registry = "https://pypi.org/simple" }
 sdist = { url = "https://files.pythonhosted.org/packages/2a/80/336820c1ad9286a4ded7e845b2eccfcb27851ab8ac6abece774a6ff4d3de/psutil-7.0.0.tar.gz", hash = "sha256:7be9c3eba38beccb6495ea33afd982a44074b78f28c434a1f51cc07fd315c456", size = 497003 }
 wheels = [
    { url = "https://files.pythonhosted.org/packages/ed/e6/2d26234410f8b8abdbf891c9da62bee396583f713fb9f3325a4760875d22/psutil-7.0.0-cp36-abi3-macosx_10_9_x86_64.whl", hash = "sha256:101d71dc322e3cffd7cea0650b09b3d08b8e7c4109dd6809fe452dfd00e58b25", size = 238051 },
    { url = "https://files.pythonhosted.org/packages/04/8b/30f930733afe425e3cbfc0e1468a30a18942350c1a8816acfade80c005c4/psutil-7.0.0-cp36-abi3-macosx_11_0_arm64.whl", hash = "sha256:39db632f6bb862eeccf56660871433e111b6ea58f2caea825571951d4b6aa3da", size = 239535 },
    { url = "https://files.pythonhosted.org/packages/2a/ed/d362e84620dd22876b55389248e522338ed1bf134a5edd3b8231d7207f6d/psutil-7.0.0-cp36-abi3-manylinux_2_12_i686.manylinux2010_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:1fcee592b4c6f146991ca55919ea3d1f8926497a713ed7faaf8225e174581e91", size = 275004 },
    { url = "https://files.pythonhosted.org/packages/bf/b9/b0eb3f3cbcb734d930fdf839431606844a825b23eaf9a6ab371edac8162c/psutil-7.0.0-cp36-abi3-manylinux_2_12_x86_64.manylinux2010_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:4b1388a4f6875d7e2aff5c4ca1cc16c545ed41dd8bb596cefea80111db353a34", size = 277986 },
    { url = "https://files.pythonhosted.org/packages/eb/a2/709e0fe2f093556c17fbafda93ac032257242cabcc7ff3369e2cb76a97aa/psutil-7.0.0-cp36-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:a5f098451abc2828f7dc6b58d44b532b22f2088f4999a937557b603ce72b1993", size = 279544 },
    { url = "https://files.pythonhosted.org/packages/50/e6/eecf58810b9d12e6427369784efe814a1eec0f492084ce8eb8f4d89d6d61/psutil-7.0.0-cp37-abi3-win32.whl", hash = "sha256:ba3fcef7523064a6c9da440fc4d6bd07da93ac726b5733c29027d7dc95b39d99", size = 241053 },
    { url = "https://files.pythonhosted.org/packages/50/1b/6921afe68c74868b4c9fa424dad3be35b095e16687989ebbb50ce4fceb7c/psutil-7.0.0-cp37-abi3-win_amd64.whl", hash = "sha256:4cf3d4eb1aa9b348dec30105c55cd9b7d4629285735a102beb4441e38db90553", size = 244885 },
 ]
 [[package]]
 name = "ptyprocess"
 version = "0.7.0"
@ -373,6 +388,7 @@ dev = [
    { name = "greenback" },
    { name = "pexpect" },
    { name = "prompt-toolkit" },
    { name = "psutil" },
    { name = "pyperclip" },
    { name = "pytest" },
    { name = "stackscope" },
@ -396,6 +412,7 @@ dev = [
    { name = "greenback", specifier = ">=1.2.1,<2" },
    { name = "pexpect", specifier = ">=4.9.0,<5" },
    { name = "prompt-toolkit", specifier = ">=3.0.50" },
    { name = "psutil", specifier = ">=7.0.0" },
    { name = "pyperclip", specifier = ">=1.9.0" },
    { name = "pytest", specifier = ">=8.3.5" },
    { name = "stackscope", specifier = ">=0.2.2,<0.3" },