Merge pull request #387 from goodboy/the_finally_footgun

Coping with "`finally` footguns": avoiding `trio.Cancelled` exc masking as best we can..

tests/devx/test_debugger.py

@@ -317,6 +317,7 @@ def test_subactor_breakpoint(
 
     assert in_prompt_msg(
         child, [
+        'MessagingError:',
         'RemoteActorError:',
          "('breakpoint_forever'",
          'bdb.BdbQuit',

tests/devx/test_tooling.py

@@ -121,11 +121,9 @@ def test_shield_pause(
         child.pid,
         signal.SIGINT,
     )
-    from tractor._supervise import _shutdown_msg
     expect(
         child,
-        # 'Shutting down actor runtime',
+        'Shutting down actor runtime',
-        _shutdown_msg,
         timeout=6,
     )
     assert_before(

tests/test_inter_peer_cancellation.py

@@ -410,6 +410,7 @@ def test_peer_canceller(
     '''
     async def main():
         async with tractor.open_nursery(
+            # NOTE: to halt the peer tasks on ctxc, uncomment this.
             debug_mode=debug_mode,
         ) as an:
             canceller: Portal = await an.start_actor(

tests/test_resource_cache.py

@@ -72,13 +72,11 @@ def test_resource_only_entered_once(key_on):
         with trio.move_on_after(0.5):
             async with (
                 tractor.open_root_actor(),
-                trio.open_nursery() as tn,
+                trio.open_nursery() as n,
             ):
+
                 for i in range(10):
-                    tn.start_soon(
+                    n.start_soon(enter_cached_mngr, f'task_{i}')
-                        enter_cached_mngr,
-                        f'task_{i}',
-                    )
                     await trio.sleep(0.001)
 
     trio.run(main)
@@ -100,34 +98,23 @@ async def streamer(
 
 
 @acm
-async def open_stream() -> Awaitable[
+async def open_stream() -> Awaitable[tractor.MsgStream]:
-    tuple[
+
-        tractor.ActorNursery,
-        tractor.MsgStream,
-    ]
-]:
     try:
         async with tractor.open_nursery() as an:
             portal = await an.start_actor(
                 'streamer',
                 enable_modules=[__name__],
             )
-            try:
+            async with (
-                async with (
+                portal.open_context(streamer) as (ctx, first),
-                    portal.open_context(streamer) as (ctx, first),
+                ctx.open_stream() as stream,
-                    ctx.open_stream() as stream,
+            ):
-                ):
+                yield stream
-                    print('Entered open_stream() caller')
-                    yield an, stream
-                    print('Exited open_stream() caller')
 
-            finally:
+            print('Cancelling streamer')
-                print(
+            await portal.cancel_actor()
-                    'Cancelling streamer with,\n'
+            print('Cancelled streamer')
-                    '=> `Portal.cancel_actor()`'
-                )
-                await portal.cancel_actor()
-                print('Cancelled streamer')
 
     except Exception as err:
         print(
@@ -143,12 +130,8 @@ async def maybe_open_stream(taskname: str):
     async with tractor.trionics.maybe_open_context(
         # NOTE: all secondary tasks should cache hit on the same key
         acm_func=open_stream,
-    ) as (
+    ) as (cache_hit, stream):
-        cache_hit,
+
-        (an, stream)
-    ):
-        # when the actor + portal + ctx + stream has already been
-        # allocated we want to just bcast to this task.
         if cache_hit:
             print(f'{taskname} loaded from cache')
 
@@ -156,43 +139,10 @@ async def maybe_open_stream(taskname: str):
             # if this feed is already allocated by the first
             # task that entereed
             async with stream.subscribe() as bstream:
-                yield an, bstream
+                yield bstream
-                print(
-                    f'cached task exited\n'
-                    f')>\n'
-                    f' |_{taskname}\n'
-                )
-
-            # we should always unreg the "cloned" bcrc for this
-            # consumer-task
-            assert id(bstream) not in bstream._state.subs
-
         else:
             # yield the actual stream
-            try:
+            yield stream
-                yield an, stream
-            finally:
-                print(
-                    f'NON-cached task exited\n'
-                    f')>\n'
-                    f' |_{taskname}\n'
-                )
-
-        first_bstream = stream._broadcaster
-        bcrx_state = first_bstream._state
-        subs: dict[int, int] = bcrx_state.subs
-        if len(subs) == 1:
-            assert id(first_bstream) in subs
-            # ^^TODO! the bcrx should always de-allocate all subs,
-            # including the implicit first one allocated on entry
-            # by the first subscribing peer task, no?
-            #
-            # -[ ] adjust `MsgStream.subscribe()` to do this mgmt!
-            #  |_ allows reverting `MsgStream.receive()` to the
-            #    non-bcaster method.
-            #  |_ we can decide whether to reset `._broadcaster`?
-            #
-            # await tractor.pause(shield=True)
 
 
 def test_open_local_sub_to_stream(
@@ -209,24 +159,16 @@ def test_open_local_sub_to_stream(
 
     if debug_mode:
         timeout = 999
-        print(f'IN debug_mode, setting large timeout={timeout!r}..')
 
     async def main():
 
         full = list(range(1000))
-        an: tractor.ActorNursery|None = None
-        num_tasks: int = 10
 
         async def get_sub_and_pull(taskname: str):
 
-            nonlocal an
-
             stream: tractor.MsgStream
             async with (
-                maybe_open_stream(taskname) as (
+                maybe_open_stream(taskname) as stream,
-                    an,
-                    stream,
-                ),
             ):
                 if '0' in taskname:
                     assert isinstance(stream, tractor.MsgStream)
@@ -238,70 +180,34 @@ def test_open_local_sub_to_stream(
 
                 first = await stream.receive()
                 print(f'{taskname} started with value {first}')
-                seq: list[int] = []
+                seq = []
                 async for msg in stream:
                     seq.append(msg)
 
                 assert set(seq).issubset(set(full))
-
-            # end of @acm block
             print(f'{taskname} finished')
 
-        root: tractor.Actor
         with trio.fail_after(timeout) as cs:
             # TODO: turns out this isn't multi-task entrant XD
             # We probably need an indepotent entry semantic?
             async with tractor.open_root_actor(
                 debug_mode=debug_mode,
-                # maybe_enable_greenback=True,
+            ):
-                #
-                # ^TODO? doesn't seem to mk breakpoint() usage work
-                # bc each bg task needs to open a portal??
-                # - [ ] we should consider making this part of
-                #      our taskman defaults?
-                #   |_see https://github.com/goodboy/tractor/pull/363
-                #
-            ) as root:
-                assert root.is_registrar
-
                 async with (
                     trio.open_nursery() as tn,
                 ):
-                    for i in range(num_tasks):
+                    for i in range(10):
                         tn.start_soon(
                             get_sub_and_pull,
                             f'task_{i}',
                         )
                         await trio.sleep(0.001)
 
-                print('all consumer tasks finished!')
+                print('all consumer tasks finished')
-
-                # ?XXX, ensure actor-nursery is shutdown or we might
-                # hang here due to a minor task deadlock/race-condition?
-                #
-                # - seems that all we need is a checkpoint to ensure
-                #   the last suspended task, which is inside
-                #   `.maybe_open_context()`, can do the
-                #   `Portal.cancel_actor()` call?
-                #
-                # - if that bg task isn't resumed, then this blocks
-                #   timeout might hit before that?
-                #
-                if root.ipc_server.has_peers():
-                    await trio.lowlevel.checkpoint()
-
-                    # alt approach, cancel the entire `an`
-                    # await tractor.pause()
-                    # await an.cancel()
-
-            # end of runtime scope
-            print('root actor terminated.')
 
         if cs.cancelled_caught:
             pytest.fail(
                 'Should NOT time out in `open_root_actor()` ?'
             )
 
-        print('exiting main.')
-
     trio.run(main)

tests/test_task_broadcasting.py

@@ -67,6 +67,7 @@ async def ensure_sequence(
 
 @acm
 async def open_sequence_streamer(
+
     sequence: list[int],
     reg_addr: tuple[str, int],
     start_method: str,
@@ -95,43 +96,39 @@ async def open_sequence_streamer(
 
 
 def test_stream_fan_out_to_local_subscriptions(
-    debug_mode: bool,
+    reg_addr,
-    reg_addr: tuple,
     start_method,
 ):
 
     sequence = list(range(1000))
 
     async def main():
-        with trio.fail_after(9):
-            async with open_sequence_streamer(
-                sequence,
-                reg_addr,
-                start_method,
-            ) as stream:
 
-                async with (
+        async with open_sequence_streamer(
-                    collapse_eg(),
+            sequence,
-                    trio.open_nursery() as tn,
+            reg_addr,
-                ):
+            start_method,
-                    for i in range(10):
+        ) as stream:
-                        tn.start_soon(
-                            ensure_sequence,
-                            stream,
-                            sequence.copy(),
-                            name=f'consumer_{i}',
-                        )
 
-                    await stream.send(tuple(sequence))
+            async with trio.open_nursery() as n:
+                for i in range(10):
+                    n.start_soon(
+                        ensure_sequence,
+                        stream,
+                        sequence.copy(),
+                        name=f'consumer_{i}',
+                    )
 
-                    async for value in stream:
+                await stream.send(tuple(sequence))
-                        print(f'source stream rx: {value}')
-                        assert value == sequence[0]
-                        sequence.remove(value)
 
-                        if not sequence:
+                async for value in stream:
-                            # fully consumed
+                    print(f'source stream rx: {value}')
-                            break
+                    assert value == sequence[0]
+                    sequence.remove(value)
+
+                    if not sequence:
+                        # fully consumed
+                        break
 
     trio.run(main)
 
@@ -154,69 +151,67 @@ def test_consumer_and_parent_maybe_lag(
         sequence = list(range(300))
         parent_delay, sub_delay = task_delays
 
-        # TODO, maybe mak a cm-deco for main()s?
+        async with open_sequence_streamer(
-        with trio.fail_after(3):
+            sequence,
-            async with open_sequence_streamer(
+            reg_addr,
-                sequence,
+            start_method,
-                reg_addr,
+        ) as stream:
-                start_method,
-            ) as stream:
 
-                try:
+            try:
-                    async with (
+                async with (
-                        collapse_eg(),
+                    collapse_eg(),
-                        trio.open_nursery() as tn,
+                    trio.open_nursery() as tn,
-                    ):
+                ):
 
-                        tn.start_soon(
+                    tn.start_soon(
-                            ensure_sequence,
+                        ensure_sequence,
-                            stream,
+                        stream,
-                            sequence.copy(),
+                        sequence.copy(),
-                            sub_delay,
+                        sub_delay,
-                            name='consumer_task',
+                        name='consumer_task',
-                        )
+                    )
 
-                        await stream.send(tuple(sequence))
+                    await stream.send(tuple(sequence))
 
-                        # async for value in stream:
+                    # async for value in stream:
-                        lagged = False
+                    lagged = False
-                        lag_count = 0
+                    lag_count = 0
 
-                        while True:
+                    while True:
-                            try:
+                        try:
-                                value = await stream.receive()
+                            value = await stream.receive()
-                                print(f'source stream rx: {value}')
+                            print(f'source stream rx: {value}')
 
-                                if lagged:
+                            if lagged:
-                                    # re set the sequence starting at our last
+                                # re set the sequence starting at our last
-                                    # value
+                                # value
-                                    sequence = sequence[sequence.index(value) + 1:]
+                                sequence = sequence[sequence.index(value) + 1:]
-                                else:
+                            else:
-                                    assert value == sequence[0]
+                                assert value == sequence[0]
-                                    sequence.remove(value)
+                                sequence.remove(value)
 
-                                lagged = False
+                            lagged = False
 
-                            except Lagged:
+                        except Lagged:
-                                lagged = True
+                            lagged = True
-                                print(f'source stream lagged after {value}')
+                            print(f'source stream lagged after {value}')
-                                lag_count += 1
+                            lag_count += 1
-                                continue
+                            continue
 
-                            # lag the parent
+                        # lag the parent
-                            await trio.sleep(parent_delay)
+                        await trio.sleep(parent_delay)
 
-                            if not sequence:
+                        if not sequence:
-                                # fully consumed
+                            # fully consumed
-                                break
+                            break
-                        print(f'parent + source stream lagged: {lag_count}')
+                    print(f'parent + source stream lagged: {lag_count}')
 
-                        if parent_delay > sub_delay:
+                    if parent_delay > sub_delay:
-                            assert lag_count > 0
+                        assert lag_count > 0
 
-                except Lagged:
+            except Lagged:
-                    # child was lagged
+                # child was lagged
-                    assert parent_delay < sub_delay
+                assert parent_delay < sub_delay
 
     trio.run(main)
 
@@ -290,11 +285,7 @@ def test_faster_task_to_recv_is_cancelled_by_slower(
 
 
 def test_subscribe_errors_after_close():
-    '''
-    Verify after calling `BroadcastReceiver.aclose()` you can't
-    "re-open" it via `.subscribe()`.
 
-    '''
     async def main():
 
         size = 1
@@ -302,8 +293,6 @@ def test_subscribe_errors_after_close():
         async with broadcast_receiver(rx, size) as brx:
             pass
 
-        assert brx.key not in brx._state.subs
-
         try:
             # open and close
             async with brx.subscribe():
@@ -313,7 +302,7 @@ def test_subscribe_errors_after_close():
             assert brx.key not in brx._state.subs
 
         else:
-            pytest.fail('brx.subscribe() never raised!?')
+            assert 0
 
     trio.run(main)
 

tractor/_context.py

@@ -101,9 +101,6 @@ from ._state import (
     debug_mode,
     _ctxvar_Context,
 )
-from .trionics import (
-    collapse_eg,
-)
 # ------ - ------
 if TYPE_CHECKING:
     from ._portal import Portal
@@ -743,8 +740,6 @@ class Context:
             # cancelled, NOT their reported canceller. IOW in the
             # latter case we're cancelled by someone else getting
             # cancelled.
-            #
-            # !TODO, switching to `Actor.aid` here!
             if (canc := error.canceller) == self._actor.uid:
                 whom: str = 'us'
                 self._canceller = canc
@@ -945,7 +940,7 @@ class Context:
         self.cancel_called = True
 
         header: str = (
-            f'Cancelling ctx from {side!r}-side\n'
+            f'Cancelling ctx from {side.upper()}-side\n'
         )
         reminfo: str = (
             # ' =>\n'
@@ -953,7 +948,7 @@ class Context:
             f'\n'
             f'c)=> {self.chan.uid}\n'
             f'   |_[{self.dst_maddr}\n'
-            f'     >> {self.repr_rpc}\n'
+            f'     >>{self.repr_rpc}\n'
             # f'    >> {self._nsf}() -> {codec}[dict]:\n\n'
             # TODO: pull msg-type from spec re #320
         )
@@ -2028,8 +2023,10 @@ async def open_context_from_portal(
     ctxc_from_callee: ContextCancelled|None = None
     try:
         async with (
-            collapse_eg(),
+            trio.open_nursery(
-            trio.open_nursery() as tn,
+                strict_exception_groups=False,
+            ) as tn,
+
             msgops.maybe_limit_plds(
                 ctx=ctx,
                 spec=ctx_meta.get('pld_spec'),

tractor/_discovery.py

@@ -28,10 +28,7 @@ from typing import (
 from contextlib import asynccontextmanager as acm
 
 from tractor.log import get_logger
-from .trionics import (
+from .trionics import gather_contexts
-    gather_contexts,
-    collapse_eg,
-)
 from .ipc import _connect_chan, Channel
 from ._addr import (
     UnwrappedAddress,
@@ -90,6 +87,7 @@ async def get_registry(
             yield regstr_ptl
 
 
+
 @acm
 async def get_root(
     **kwargs,
@@ -255,12 +253,9 @@ async def find_actor(
         for addr in registry_addrs
     )
     portals: list[Portal]
-    async with (
+    async with gather_contexts(
-        collapse_eg(),
+        mngrs=maybe_portals,
-        gather_contexts(
+    ) as portals:
-            mngrs=maybe_portals,
-        ) as portals,
-    ):
         # log.runtime(
         #     'Gathered portals:\n'
         #     f'{portals}'

tractor/_entry.py

@@ -21,7 +21,7 @@ Sub-process entry points.
 from __future__ import annotations
 from functools import partial
 import multiprocessing as mp
-# import os
+import os
 from typing import (
     Any,
     TYPE_CHECKING,
@@ -38,7 +38,6 @@ from .devx import (
     _frame_stack,
     pformat,
 )
-# from .msg import pretty_struct
 from .to_asyncio import run_as_asyncio_guest
 from ._addr import UnwrappedAddress
 from ._runtime import (
@@ -128,13 +127,20 @@ def _trio_main(
 
     if actor.loglevel is not None:
         get_console_log(actor.loglevel)
+        actor_info: str = (
+            f'|_{actor}\n'
+            f'  uid: {actor.uid}\n'
+            f'  pid: {os.getpid()}\n'
+            f'  parent_addr: {parent_addr}\n'
+            f'  loglevel: {actor.loglevel}\n'
+        )
         log.info(
-            f'Starting `trio` subactor from parent @ '
+            'Starting new `trio` subactor\n'
-            f'{parent_addr}\n'
             +
             pformat.nest_from_op(
                 input_op='>(',  # see syntax ideas above
-                text=f'{actor}',
+                text=actor_info,
+                nest_indent=2,  # since "complete"
             )
         )
     logmeth = log.info
@@ -143,7 +149,7 @@ def _trio_main(
         +
         pformat.nest_from_op(
             input_op=')>',  # like a "closed-to-play"-icon from super perspective
-            text=f'{actor}',
+            text=actor_info,
             nest_indent=1,
         )
     )
@@ -161,7 +167,7 @@ def _trio_main(
             +
             pformat.nest_from_op(
                 input_op='c)>',  # closed due to cancel (see above)
-                text=f'{actor}',
+                text=actor_info,
             )
         )
     except BaseException as err:
@@ -171,7 +177,7 @@ def _trio_main(
             +
             pformat.nest_from_op(
                 input_op='x)>',  # closed by error
-                text=f'{actor}',
+                text=actor_info,
             )
         )
         # NOTE since we raise a tb will already be shown on the

tractor/_portal.py

@@ -39,10 +39,7 @@ import warnings
 
 import trio
 
-from .trionics import (
+from .trionics import maybe_open_nursery
-    maybe_open_nursery,
-    collapse_eg,
-)
 from ._state import (
     current_actor,
 )
@@ -118,10 +115,6 @@ class Portal:
 
     @property
     def chan(self) -> Channel:
-        '''
-        Ref to this ctx's underlying `tractor.ipc.Channel`.
-
-        '''
         return self._chan
 
     @property
@@ -181,17 +174,10 @@ class Portal:
 
         # not expecting a "main" result
         if self._expect_result_ctx is None:
-            peer_id: str = f'{self.channel.aid.reprol()!r}'
             log.warning(
-                f'Portal to peer {peer_id} will not deliver a final result?\n'
+                f"Portal for {self.channel.aid} not expecting a final"
-                f'\n'
+                " result?\nresult() should only be called if subactor"
-                f'Context.result() can only be called by the parent of '
+                " was spawned with `ActorNursery.run_in_actor()`")
-                f'a sub-actor when it was spawned with '
-                f'`ActorNursery.run_in_actor()`'
-                f'\n'
-                f'Further this `ActorNursery`-method-API will deprecated in the'
-                f'near fututre!\n'
-            )
             return NoResult
 
         # expecting a "main" result
@@ -224,7 +210,6 @@ class Portal:
         typname: str = type(self).__name__
         log.warning(
             f'`{typname}.result()` is DEPRECATED!\n'
-            f'\n'
             f'Use `{typname}.wait_for_result()` instead!\n'
         )
         return await self.wait_for_result(
@@ -236,10 +221,8 @@ class Portal:
         # terminate all locally running async generator
         # IPC calls
         if self._streams:
-            peer_id: str = f'{self.channel.aid.reprol()!r}'
+            log.cancel(
-            report: str = (
+                f"Cancelling all streams with {self.channel.aid}")
-                f'Cancelling all msg-streams with {peer_id}\n'
-            )
             for stream in self._streams.copy():
                 try:
                     await stream.aclose()
@@ -248,18 +231,10 @@ class Portal:
                     # (unless of course at some point down the road we
                     # won't expect this to always be the case or need to
                     # detect it for respawning purposes?)
-                    report += (
+                    log.debug(f"{stream} was already closed.")
-                        f'->) {stream!r} already closed\n'
-                    )
-
-            log.cancel(report)
 
     async def aclose(self):
-        log.debug(
+        log.debug(f"Closing {self}")
-            f'Closing portal\n'
-            f'>}}\n'
-            f'|_{self}\n'
-        )
         # TODO: once we move to implementing our own `ReceiveChannel`
         # (including remote task cancellation inside its `.aclose()`)
         # we'll need to .aclose all those channels here
@@ -285,22 +260,23 @@ class Portal:
         __runtimeframe__: int = 1  # noqa
 
         chan: Channel = self.channel
-        peer_id: str = f'{self.channel.aid.reprol()!r}'
         if not chan.connected():
             log.runtime(
-                'Peer {peer_id} is already disconnected\n'
+                'This channel is already closed, skipping cancel request..'
-                '-> skipping cancel request..\n'
             )
             return False
 
+        reminfo: str = (
+            f'c)=> {self.channel.aid}\n'
+            f'  |_{chan}\n'
+        )
         log.cancel(
-            f'Sending actor-runtime-cancel-req to peer\n'
+            f'Requesting actor-runtime cancel for peer\n\n'
-            f'\n'
+            f'{reminfo}'
-            f'c)=> {peer_id}\n'
         )
 
         # XXX the one spot we set it?
-        chan._cancel_called: bool = True
+        self.channel._cancel_called: bool = True
         try:
             # send cancel cmd - might not get response
             # XXX: sure would be nice to make this work with
@@ -321,9 +297,8 @@ class Portal:
                 # may timeout and we never get an ack (obvi racy)
                 # but that doesn't mean it wasn't cancelled.
                 log.debug(
-                    f'May have failed to cancel peer?\n'
+                    'May have failed to cancel peer?\n'
-                    f'\n'
+                    f'{reminfo}'
-                    f'c)=?> {peer_id}\n'
                 )
 
             # if we get here some weird cancellation case happened
@@ -341,22 +316,22 @@ class Portal:
 
             TransportClosed,
         ) as tpt_err:
-            ipc_borked_report: str = (
+            report: str = (
-                f'IPC for actor already closed/broken?\n\n'
+                f'IPC chan for actor already closed or broken?\n\n'
-                f'\n'
+                f'{self.channel.aid}\n'
-                f'c)=x> {peer_id}\n'
+                f' |_{self.channel}\n'
             )
             match tpt_err:
                 case TransportClosed():
-                    log.debug(ipc_borked_report)
+                    log.debug(report)
                 case _:
-                    ipc_borked_report += (
+                    report += (
                         f'\n'
                         f'Unhandled low-level transport-closed/error during\n'
                         f'Portal.cancel_actor()` request?\n'
                         f'<{type(tpt_err).__name__}( {tpt_err} )>\n'
                     )
-                    log.warning(ipc_borked_report)
+                    log.warning(report)
 
             return False
 
@@ -513,13 +488,10 @@ class Portal:
                 with trio.CancelScope(shield=True):
                     await ctx.cancel()
 
-            except trio.ClosedResourceError as cre:
+            except trio.ClosedResourceError:
                 # if the far end terminates before we send a cancel the
                 # underlying transport-channel may already be closed.
-                log.cancel(
+                log.cancel(f'Context {ctx} was already closed?')
-                    f'Context.cancel() -> {cre!r}\n'
-                    f'cid: {ctx.cid!r} already closed?\n'
-                )
 
             # XXX: should this always be done?
             # await recv_chan.aclose()
@@ -586,13 +558,14 @@ async def open_portal(
     assert actor
     was_connected: bool = False
 
-    async with (
+    async with maybe_open_nursery(
-        collapse_eg(),
+        tn,
-        maybe_open_nursery(
+        shield=shield,
-            tn,
+        strict_exception_groups=False,
-            shield=shield,
+        # ^XXX^ TODO? soo roll our own then ??
-        ) as tn,
+        # -> since we kinda want the "if only one `.exception` then
-    ):
+        # just raise that" interface?
+    ) as tn:
 
         if not channel.connected():
             await channel.connect()

tractor/_root.py

@@ -37,11 +37,16 @@ import warnings
 
 import trio
 
-from . import _runtime
+from ._runtime import (
+    Actor,
+    Arbiter,
+    # TODO: rename and make a non-actor subtype?
+    # Arbiter as Registry,
+    async_main,
+)
 from .devx import (
     debug,
     _frame_stack,
-    pformat as _pformat,
 )
 from . import _spawn
 from . import _state
@@ -58,7 +63,6 @@ from ._addr import (
 )
 from .trionics import (
     is_multi_cancelled,
-    collapse_eg,
 )
 from ._exceptions import (
     RuntimeFailure,
@@ -97,7 +101,7 @@ async def maybe_block_bp(
     ):
         logger.info(
             f'Found `greenback` installed @ {maybe_mod}\n'
-            f'Enabling `tractor.pause_from_sync()` support!\n'
+            'Enabling `tractor.pause_from_sync()` support!\n'
         )
         os.environ['PYTHONBREAKPOINT'] = (
             'tractor.devx.debug._sync_pause_from_builtin'
@@ -192,19 +196,13 @@ async def open_root_actor(
     # read-only state to sublayers?
     # extra_rt_vars: dict|None = None,
 
-) -> _runtime.Actor:
+) -> Actor:
     '''
-    Initialize the `tractor` runtime by starting a "root actor" in
+    Runtime init entry point for ``tractor``.
-    a parent-most Python process.
-
-    All (disjoint) actor-process-trees-as-programs are created via
-    this entrypoint.
 
     '''
     # XXX NEVER allow nested actor-trees!
-    if already_actor := _state.current_actor(
+    if already_actor := _state.current_actor(err_on_no_runtime=False):
-        err_on_no_runtime=False,
-    ):
         rtvs: dict[str, Any] = _state._runtime_vars
         root_mailbox: list[str, int] = rtvs['_root_mailbox']
         registry_addrs: list[list[str, int]] = rtvs['_registry_addrs']
@@ -274,20 +272,14 @@ async def open_root_actor(
                 DeprecationWarning,
                 stacklevel=2,
             )
-            uw_reg_addrs = [arbiter_addr]
+            registry_addrs = [arbiter_addr]
 
-        uw_reg_addrs = registry_addrs
+        if not registry_addrs:
-        if not uw_reg_addrs:
+            registry_addrs: list[UnwrappedAddress] = default_lo_addrs(
-            uw_reg_addrs: list[UnwrappedAddress] = default_lo_addrs(
                 enable_transports
             )
 
-        # must exist by now since all below code is dependent
+        assert registry_addrs
-        assert uw_reg_addrs
-        registry_addrs: list[Address] = [
-            wrap_address(uw_addr)
-            for uw_addr in uw_reg_addrs
-        ]
 
         loglevel = (
             loglevel
@@ -336,10 +328,10 @@ async def open_root_actor(
             enable_stack_on_sig()
 
         # closed into below ping task-func
-        ponged_addrs: list[Address] = []
+        ponged_addrs: list[UnwrappedAddress] = []
 
         async def ping_tpt_socket(
-            addr: Address,
+            addr: UnwrappedAddress,
             timeout: float = 1,
         ) -> None:
             '''
@@ -359,22 +351,17 @@ async def open_root_actor(
                 # be better to eventually have a "discovery" protocol
                 # with basic handshake instead?
                 with trio.move_on_after(timeout):
-                    async with _connect_chan(addr.unwrap()):
+                    async with _connect_chan(addr):
                         ponged_addrs.append(addr)
 
             except OSError:
-                # ?TODO, make this a "discovery" log level?
+                # TODO: make this a "discovery" log level?
                 logger.info(
-                    f'No root-actor registry found @ {addr!r}\n'
+                    f'No actor registry found @ {addr}\n'
                 )
 
-        # !TODO, this is basically just another (abstract)
-        # happy-eyeballs, so we should try for formalize it somewhere
-        # in a `.[_]discovery` ya?
-        #
         async with trio.open_nursery() as tn:
-            for uw_addr in uw_reg_addrs:
+            for addr in registry_addrs:
-                addr: Address = wrap_address(uw_addr)
                 tn.start_soon(
                     ping_tpt_socket,
                     addr,
@@ -396,35 +383,31 @@ async def open_root_actor(
                 f'Registry(s) seem(s) to exist @ {ponged_addrs}'
             )
 
-            actor = _runtime.Actor(
+            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
+            # DO NOT use the registry_addrs as the transport server
-            # ipc-transport-server's bind-addrs as this is
+            # addrs for this new non-registar, root-actor.
-            # a new NON-registrar, ROOT-actor.
-            #
-            # XXX INSTEAD, bind random addrs using the same tpt
-            # proto.
             for addr in ponged_addrs:
+                waddr: Address = wrap_address(addr)
                 trans_bind_addrs.append(
-                    addr.get_random(
+                    waddr.get_random(bindspace=waddr.bindspace)
-                        bindspace=addr.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 = uw_reg_addrs
+            trans_bind_addrs = registry_addrs
 
             # - it is normally desirable for any registrar to stay up
             #   indefinitely until either all registered (child/sub)
@@ -435,8 +418,7 @@ async def open_root_actor(
             # https://github.com/goodboy/tractor/pull/348
             # https://github.com/goodboy/tractor/issues/296
 
-            # TODO: rename as `RootActor` or is that even necessary?
+            actor = Arbiter(
-            actor = _runtime.Arbiter(
                 name=name or 'registrar',
                 uuid=mk_uuid(),
                 registry_addrs=registry_addrs,
@@ -448,16 +430,6 @@ async def open_root_actor(
             # `.trio.run()`.
             actor._infected_aio = _state._runtime_vars['_is_infected_aio']
 
-        # NOTE, only set the loopback addr for the
-        # process-tree-global "root" mailbox since all sub-actors
-        # should be able to speak to their root actor over that
-        # channel.
-        raddrs: list[Address] = _state._runtime_vars['_root_addrs']
-        raddrs.extend(trans_bind_addrs)
-        # TODO, remove once we have also removed all usage;
-        # eventually all (root-)registry apis should expect > 1 addr.
-        _state._runtime_vars['_root_mailbox'] = raddrs[0]
-
         # Start up main task set via core actor-runtime nurseries.
         try:
             # assign process-local actor
@@ -465,27 +437,21 @@ async def open_root_actor(
 
             # start local channel-server and fake the portal API
             # NOTE: this won't block since we provide the nursery
-            report: str = f'Starting actor-runtime for {actor.aid.reprol()!r}\n'
+            ml_addrs_str: str = '\n'.join(
-            if reg_addrs := actor.registry_addrs:
+                f'@{addr}' for addr in trans_bind_addrs
-                report += (
+            )
-                    '-> Opening new registry @ '
+            logger.info(
-                    +
+                f'Starting local {actor.uid} on the following transport addrs:\n'
-                    '\n'.join(
+                f'{ml_addrs_str}'
-                        f'{addr}' for addr in reg_addrs
+            )
-                    )
-                )
-            logger.info(f'{report}\n')
 
-            # start runtime in a bg sub-task, yield to caller.
+            # start the actor runtime in a new task
-            async with (
+            async with trio.open_nursery(
-                collapse_eg(),
+                strict_exception_groups=False,
-                trio.open_nursery() as root_tn,
+                # ^XXX^ TODO? instead unpack any RAE as per "loose" style?
+            ) as nursery:
 
-                # XXX, finally-footgun below?
+                # ``_runtime.async_main()`` creates an internal nursery
-                # -> see note on why shielding.
-                # maybe_raise_from_masking_exc(),
-            ):
-                # `_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
@@ -493,9 +459,9 @@ async def open_root_actor(
                 # "actor runtime" primitives are SC-compat and thus all
                 # transitively spawned actors/processes must be as
                 # well.
-                await root_tn.start(
+                await nursery.start(
                     partial(
-                        _runtime.async_main,
+                        async_main,
                         actor,
                         accept_addrs=trans_bind_addrs,
                         parent_addr=None
@@ -543,7 +509,7 @@ async def open_root_actor(
                     raise
 
                 finally:
-                    # NOTE/TODO?, not sure if we'll ever need this but it's
+                    # 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..')
@@ -552,21 +518,12 @@ async def open_root_actor(
                     #     for an in nurseries:
                     #         tempn.start_soon(an.exited.wait)
 
-                    op_nested_actor_repr: str = _pformat.nest_from_op(
-                        input_op='>) ',
-                        text=actor.pformat(),
-                        nest_prefix='|_',
-                    )
                     logger.info(
                         f'Closing down root actor\n'
-                        f'{op_nested_actor_repr}'
+                        f'>)\n'
+                        f'|_{actor}\n'
                     )
-                    # XXX, THIS IS A *finally-footgun*!
+                    await actor.cancel(None)  # self cancel
-                    # -> though already shields iternally it can
-                    # taskc here and mask underlying errors raised in
-                    # the try-block above?
-                    with trio.CancelScope(shield=True):
-                        await actor.cancel(None)  # self cancel
         finally:
             # revert all process-global runtime state
             if (
@@ -579,16 +536,10 @@ async def open_root_actor(
             _state._current_actor = None
             _state._last_actor_terminated = actor
 
-            sclang_repr: str = _pformat.nest_from_op(
+            logger.runtime(
-                input_op=')>',
-                text=actor.pformat(),
-                nest_prefix='|_',
-                nest_indent=1,
-            )
-
-            logger.info(
                 f'Root actor terminated\n'
-                f'{sclang_repr}'
+                f')>\n'
+                f' |_{actor}\n'
             )
 
 

tractor/_rpc.py

@@ -64,7 +64,6 @@ from .trionics import (
 from .devx import (
     debug,
     add_div,
-    pformat as _pformat,
 )
 from . import _state
 from .log import get_logger
@@ -73,7 +72,7 @@ from .msg import (
     MsgCodec,
     PayloadT,
     NamespacePath,
-    pretty_struct,
+    # pretty_struct,
     _ops as msgops,
 )
 from tractor.msg.types import (
@@ -221,18 +220,11 @@ async def _invoke_non_context(
             task_status.started(ctx)
             result = await coro
             fname: str = func.__name__
-
-            op_nested_task: str = _pformat.nest_from_op(
-                input_op=f')> cid: {ctx.cid!r}',
-                text=f'{ctx._task}',
-                nest_indent=1,  # under >
-            )
             log.runtime(
-                f'RPC task complete\n'
+                'RPC complete:\n'
-                f'\n'
+                f'task: {ctx._task}\n'
-                f'{op_nested_task}\n'
+                f'|_cid={ctx.cid}\n'
-                f'\n'
+                f'|_{fname}() -> {pformat(result)}\n'
-                f')> {fname}() -> {pformat(result)}\n'
             )
 
             # NOTE: only send result if we know IPC isn't down
@@ -672,8 +664,7 @@ async def _invoke(
                 ctx._result = res
                 log.runtime(
                     f'Sending result msg and exiting {ctx.side!r}\n'
-                    f'\n'
+                    f'{return_msg}\n'
-                    f'{pretty_struct.pformat(return_msg)}\n'
                 )
                 await chan.send(return_msg)
 
@@ -765,6 +756,7 @@ async def _invoke(
             BaseExceptionGroup,
             BaseException,
             trio.Cancelled,
+
         ) as _scope_err:
             scope_err = _scope_err
             if (
@@ -840,12 +832,12 @@ async def _invoke(
                 else:
                     descr_str += f'\n{merr!r}\n'
             else:
-                descr_str += f'\nwith final result {ctx.outcome!r}\n'
+                descr_str += f'\nand final result {ctx.outcome!r}\n'
 
             logmeth(
-                f'{message}\n'
+                message
-                f'\n'
+                +
-                f'{descr_str}\n'
+                descr_str
             )
 
 
@@ -1012,6 +1004,8 @@ async def process_messages(
                         cid=cid,
                         kwargs=kwargs,
                     ):
+                        kwargs |= {'req_chan': chan}
+
                         # XXX NOTE XXX don't start entire actor
                         # runtime cancellation if this actor is
                         # currently in debug mode!
@@ -1030,14 +1024,14 @@ async def process_messages(
                                 cid,
                                 chan,
                                 actor.cancel,
-                                kwargs | {'req_chan': chan},
+                                kwargs,
                                 is_rpc=False,
                                 return_msg_type=CancelAck,
                             )
 
                         log.runtime(
-                            'Cancelling RPC-msg-loop with peer\n'
+                            'Cancelling IPC transport msg-loop with peer:\n'
-                            f'->c}} {chan.aid.reprol()}@[{chan.maddr}]\n'
+                            f'|_{chan}\n'
                         )
                         loop_cs.cancel()
                         break
@@ -1050,7 +1044,7 @@ async def process_messages(
                     ):
                         target_cid: str = kwargs['cid']
                         kwargs |= {
-                            'requesting_aid': chan.aid,
+                            'requesting_uid': chan.uid,
                             'ipc_msg': msg,
 
                             # XXX NOTE! ONLY the rpc-task-owning
@@ -1086,34 +1080,21 @@ async def process_messages(
                         ns=ns,
                         func=funcname,
                         kwargs=kwargs,  # type-spec this? see `msg.types`
-                        uid=actor_uuid,
+                        uid=actorid,
                     ):
-                        if actor_uuid != chan.aid.uid:
-                            raise RuntimeError(
-                                f'IPC <Start> msg <-> chan.aid mismatch!?\n'
-                                f'Channel.aid = {chan.aid!r}\n'
-                                f'Start.uid = {actor_uuid!r}\n'
-                            )
-                        # await debug.pause()
-                        op_repr: str = 'Start <=) '
-                        req_repr: str = _pformat.nest_from_op(
-                            input_op=op_repr,
-                            op_suffix='',
-                            nest_prefix='',
-                            text=f'{chan}',
-
-                            nest_indent=len(op_repr)-1,
-                            rm_from_first_ln='<',
-                            # ^XXX, subtract -1 to account for
-                            # <Channel
-                            # ^_chevron to be stripped
-                        )
                         start_status: str = (
-                            'Handling RPC request\n'
+                            'Handling RPC `Start` request\n'
-                            f'{req_repr}\n'
+                            f'<= peer: {actorid}\n\n'
-                            f'\n'
+                            f'  |_{chan}\n'
-                            f'->{{ ipc-context-id: {cid!r}\n'
+                            f'  |_cid: {cid}\n\n'
-                            f'->{{ nsp for fn: `{ns}.{funcname}({kwargs})`\n'
+                            # f'  |_{ns}.{funcname}({kwargs})\n'
+                            f'>> {actor.uid}\n'
+                            f'  |_{actor}\n'
+                            f'   -> nsp: `{ns}.{funcname}({kwargs})`\n'
+
+                            # f'  |_{ns}.{funcname}({kwargs})\n\n'
+
+                            # f'{pretty_struct.pformat(msg)}\n'
                         )
 
                         # runtime-internal endpoint: `Actor.<funcname>`
@@ -1142,6 +1123,10 @@ async def process_messages(
                                 await chan.send(err_msg)
                                 continue
 
+                        start_status += (
+                            f'   -> func: {func}\n'
+                        )
+
                         # schedule a task for the requested RPC function
                         # in the actor's main "service nursery".
                         #
@@ -1149,7 +1134,7 @@ async def process_messages(
                         # supervision isolation? would avoid having to
                         # manage RPC tasks individually in `._rpc_tasks`
                         # table?
-                        start_status += '->( scheduling new task..\n'
+                        start_status += '   -> scheduling new task..\n'
                         log.runtime(start_status)
                         try:
                             ctx: Context = await actor._service_n.start(
@@ -1233,24 +1218,12 @@ async def process_messages(
             # END-OF `async for`:
             # IPC disconnected via `trio.EndOfChannel`, likely
             # due to a (graceful) `Channel.aclose()`.
-
-            chan_op_repr: str = '<=x] '
-            chan_repr: str = _pformat.nest_from_op(
-                input_op=chan_op_repr,
-                op_suffix='',
-                nest_prefix='',
-                text=chan.pformat(),
-                nest_indent=len(chan_op_repr)-1,
-                rm_from_first_ln='<',
-            )
             log.runtime(
-                f'IPC channel disconnected\n'
+                f'channel for {chan.uid} disconnected, cancelling RPC tasks\n'
-                f'{chan_repr}\n'
+                f'|_{chan}\n'
-                f'\n'
-                f'->c) cancelling RPC tasks.\n'
             )
             await actor.cancel_rpc_tasks(
-                req_aid=actor.aid,
+                req_uid=actor.uid,
                 # a "self cancel" in terms of the lifetime of the
                 # IPC connection which is presumed to be the
                 # source of any requests for spawned tasks.
@@ -1322,37 +1295,13 @@ async def process_messages(
     finally:
         # msg debugging for when he machinery is brokey
         if msg is None:
-            message: str = 'Exiting RPC-loop without receiving a msg?'
+            message: str = 'Exiting IPC msg loop without receiving a msg?'
         else:
-            task_op_repr: str = ')>'
-            task: trio.Task = trio.lowlevel.current_task()
-
-            # maybe add cancelled opt prefix
-            if task._cancel_status.effectively_cancelled:
-                task_op_repr = 'c' + task_op_repr
-
-            task_repr: str = _pformat.nest_from_op(
-                input_op=task_op_repr,
-                text=f'{task!r}',
-                nest_indent=1,
-            )
-            # chan_op_repr: str = '<=} '
-            # chan_repr: str = _pformat.nest_from_op(
-            #     input_op=chan_op_repr,
-            #     op_suffix='',
-            #     nest_prefix='',
-            #     text=chan.pformat(),
-            #     nest_indent=len(chan_op_repr)-1,
-            #     rm_from_first_ln='<',
-            # )
             message: str = (
-                f'Exiting RPC-loop with final msg\n'
+                'Exiting IPC msg loop with final msg\n\n'
-                f'\n'
+                f'<= peer: {chan.uid}\n'
-                # f'{chan_repr}\n'
+                f'  |_{chan}\n\n'
-                f'{task_repr}\n'
+                # f'{pretty_struct.pformat(msg)}'
-                f'\n'
-                f'{pretty_struct.pformat(msg)}'
-                f'\n'
             )
 
         log.runtime(message)

tractor/_runtime.py

@@ -55,7 +55,6 @@ from typing import (
     TYPE_CHECKING,
 )
 import uuid
-import textwrap
 from types import ModuleType
 import warnings
 
@@ -74,9 +73,6 @@ from tractor.msg import (
     pretty_struct,
     types as msgtypes,
 )
-from .trionics import (
-    collapse_eg,
-)
 from .ipc import (
     Channel,
     # IPCServer,  # causes cycles atm..
@@ -101,10 +97,7 @@ from ._exceptions import (
     MsgTypeError,
     unpack_error,
 )
-from .devx import (
+from .devx import debug
-    debug,
-    pformat as _pformat
-)
 from ._discovery import get_registry
 from ._portal import Portal
 from . import _state
@@ -213,7 +206,7 @@ class Actor:
         *,
         enable_modules: list[str] = [],
         loglevel: str|None = None,
-        registry_addrs: list[Address]|None = None,
+        registry_addrs: list[UnwrappedAddress]|None = None,
         spawn_method: str|None = None,
 
         # TODO: remove!
@@ -234,7 +227,7 @@ class Actor:
 
         # state
         self._cancel_complete = trio.Event()
-        self._cancel_called_by: tuple[str, tuple]|None = None
+        self._cancel_called_by_remote: tuple[str, tuple]|None = None
         self._cancel_called: bool = False
 
         # retreive and store parent `__main__` data which
@@ -256,12 +249,11 @@ class Actor:
         if arbiter_addr is not None:
             warnings.warn(
                 '`Actor(arbiter_addr=<blah>)` is now deprecated.\n'
-                'Use `registry_addrs: list[Address]` instead.',
+                'Use `registry_addrs: list[tuple]` instead.',
                 DeprecationWarning,
                 stacklevel=2,
             )
-
+            registry_addrs: list[UnwrappedAddress] = [arbiter_addr]
-            registry_addrs: list[Address] = [wrap_address(arbiter_addr)]
 
         # marked by the process spawning backend at startup
         # will be None for the parent most process started manually
@@ -300,10 +292,8 @@ class Actor:
         # input via the validator.
         self._reg_addrs: list[UnwrappedAddress] = []
         if registry_addrs:
-            _state._runtime_vars['_registry_addrs'] = self.reg_addrs = [
+            self.reg_addrs: list[UnwrappedAddress] = registry_addrs
-                addr.unwrap()
+            _state._runtime_vars['_registry_addrs'] = registry_addrs
-                for addr in registry_addrs
-            ]
 
     @property
     def aid(self) -> msgtypes.Aid:
@@ -349,125 +339,46 @@ class Actor:
     def pid(self) -> int:
         return self._aid.pid
 
-    @property
+    def pformat(self) -> str:
-    def repr_state(self) -> str:
+        ds: str = '='
-        if self.cancel_complete:
+        parent_uid: tuple|None = None
-            return 'cancelled'
-
-        elif canceller := self.cancel_caller:
-                return f' and cancel-called by {canceller}'
-
-        else:
-            return 'running'
-
-    def pformat(
-        self,
-        ds: str = ': ',
-        indent: int = 0,
-        privates: bool = False,
-    ) -> str:
-
-        fmtstr: str = f'|_id: {self.aid.reprol()!r}\n'
-        if privates:
-            aid_nest_prefix: str = '|_aid='
-            aid_field_repr: str = _pformat.nest_from_op(
-                input_op='',
-                text=pretty_struct.pformat(
-                    struct=self.aid,
-                    field_indent=2,
-                ),
-                op_suffix='',
-                nest_prefix=aid_nest_prefix,
-                nest_indent=0,
-            )
-            fmtstr: str = f'{aid_field_repr}'
-
         if rent_chan := self._parent_chan:
-            fmtstr += (
+            parent_uid = rent_chan.uid
-                f"|_parent{ds}{rent_chan.aid.reprol()}\n"
-            )
 
+        peers: list = []
         server: _server.IPCServer = self.ipc_server
         if server:
-            if privates:
+            peers: list[tuple] = list(server._peer_connected)
-                server_repr: str = self._ipc_server.pformat(
-                    privates=privates,
-                )
-                # create field ln as a key-header indented under
-                # and up to the section's key prefix.
-                # ^XXX if we were to indent `repr(Server)` to
-                # '<key>: '
-                #  _here_^
-                server_repr: str = _pformat.nest_from_op(
-                    input_op='',  # nest as sub-obj
-                    op_suffix='',
-                    text=server_repr,
-                )
-                fmtstr += (
-                    f"{server_repr}"
-                )
-            else:
-                fmtstr += (
-                    f'|_ipc: {server.repr_state!r}\n'
-                )
 
-        fmtstr += (
+        fmtstr: str = (
-            f'|_rpc: {len(self._rpc_tasks)} active tasks\n'
+            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"   ipc_server{ds}{self._ipc_server}\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 (
-        # TODO, actually fix the .repr_state impl/output?
+            '<Actor(\n'
-        # append ipc-ctx state summary
+            +
-        # ctxs: dict = self._contexts
+            fmtstr
-        # if ctxs:
+            +
-        #     ctx_states: dict[str, int] = {}
+            ')>\n'
-        #     for ctx in self._contexts.values():
-        #         ctx_state: str = ctx.repr_state
-        #         cnt = ctx_states.setdefault(ctx_state, 0)
-        #         ctx_states[ctx_state] = cnt + 1
-
-        #     fmtstr += (
-        #         f"  ctxs{ds}{ctx_states}\n"
-        #     )
-
-        # runtime-state
-        task_name: str = '<dne>'
-        if task := self._task:
-            task_name: str = task.name
-        fmtstr += (
-            # TODO, this just like ctx?
-            f'|_state: {self.repr_state!r}\n'
-            f'  task: {task_name}\n'
-            f'  loglevel: {self.loglevel!r}\n'
-            f'  subactors_spawned: {len(self._actoruid2nursery)}\n'
         )
-        if not _state.is_root_process():
-            fmtstr += f'  spawn_method: {self._spawn_method!r}\n'
-
-        if privates:
-            fmtstr += (
-                # f'  actoruid2nursery{ds}{self._actoruid2nursery}\n'
-                f'  cancel_complete{ds}{self._cancel_complete}\n'
-                f'  cancel_called_by_remote{ds}{self._cancel_called_by}\n'
-                f'  cancel_called{ds}{self._cancel_called}\n'
-            )
-
-        if fmtstr:
-            fmtstr: str = textwrap.indent(
-                text=fmtstr,
-                prefix=' '*(1 + indent),
-            )
-
-        _repr: str = (
-            f'<{type(self).__name__}(\n'
-            f'{fmtstr}'
-            f')>\n'
-        )
-        if indent:
-            _repr: str = textwrap.indent(
-                text=_repr,
-                prefix=' '*indent,
-            )
-        return _repr
 
     __repr__ = pformat
 
@@ -475,11 +386,7 @@ class Actor:
     def reg_addrs(self) -> list[UnwrappedAddress]:
         '''
         List of (socket) addresses for all known (and contactable)
-        registry-service actors in "unwrapped" (i.e. IPC interchange
+        registry actors.
-        wire-compat) form.
-
-        If you are looking for the "wrapped" address form, use
-        `.registry_addrs` instead.
 
         '''
         return self._reg_addrs
@@ -498,14 +405,8 @@ class Actor:
 
         self._reg_addrs = addrs
 
-    @property
-    def registry_addrs(self) -> list[Address]:
-        return [wrap_address(uw_addr)
-                for uw_addr in self.reg_addrs]
-
     def load_modules(
         self,
-
     ) -> None:
         '''
         Load explicitly enabled python modules from local fs after
@@ -552,14 +453,6 @@ class Actor:
             )
             raise
 
-    # ?TODO, factor this meth-iface into a new `.rpc` subsys primitive?
-    # - _get_rpc_func(),
-    # - _deliver_ctx_payload(),
-    # - get_context(),
-    # - start_remote_task(),
-    # - cancel_rpc_tasks(),
-    # - _cancel_task(),
-    #
     def _get_rpc_func(self, ns, funcname):
         '''
         Try to lookup and return a target RPC func from the
@@ -603,11 +496,11 @@ class Actor:
         queue.
 
         '''
-        aid: msgtypes.Aid = chan.aid
+        uid: tuple[str, str] = chan.uid
-        assert aid, f"`chan.aid` can't be {aid}"
+        assert uid, f"`chan.uid` can't be {uid}"
         try:
             ctx: Context = self._contexts[(
-                aid.uid,
+                uid,
                 cid,
 
                 # TODO: how to determine this tho?
@@ -618,7 +511,7 @@ class Actor:
                 'Ignoring invalid IPC msg!?\n'
                 f'Ctx seems to not/no-longer exist??\n'
                 f'\n'
-                f'<=? {aid.reprol()!r}\n'
+                f'<=? {uid}\n'
                 f'  |_{pretty_struct.pformat(msg)}\n'
             )
             match msg:
@@ -667,7 +560,6 @@ class Actor:
           msging session's lifetime.
 
         '''
-        # ?TODO, use Aid here as well?
         actor_uid = chan.uid
         assert actor_uid
         try:
@@ -1016,22 +908,6 @@ class Actor:
             None,  # self cancel all rpc tasks
         )
 
-    @property
-    def cancel_complete(self) -> bool:
-        return self._cancel_complete.is_set()
-
-    @property
-    def cancel_called(self) -> bool:
-        '''
-        Was this actor requested to cancel by a remote peer actor.
-
-        '''
-        return self._cancel_called_by is not None
-
-    @property
-    def cancel_caller(self) -> msgtypes.Aid|None:
-        return self._cancel_called_by
-
     async def cancel(
         self,
 
@@ -1056,18 +932,20 @@ class Actor:
 
         '''
         (
-            requesting_aid,  # Aid
+            requesting_uid,
-            requester_type,  # str
+            requester_type,
             req_chan,
             log_meth,
         ) = (
-            req_chan.aid,
+            req_chan.uid,
             'peer',
             req_chan,
             log.cancel,
+
         ) if req_chan else (
+
             # a self cancel of ALL rpc tasks
-            self.aid,
+            self.uid,
             'self',
             self,
             log.runtime,
@@ -1075,14 +953,14 @@ class Actor:
         # TODO: just use the new `Context.repr_rpc: str` (and
         # other) repr fields instead of doing this all manual..
         msg: str = (
-            f'Actor-runtime cancel request from {requester_type!r}\n'
+            f'Actor-runtime cancel request from {requester_type}\n\n'
+            f'<=c) {requesting_uid}\n'
+            f'  |_{self}\n'
             f'\n'
-            f'<=c)\n'
-            f'{self}'
         )
 
         # TODO: what happens here when we self-cancel tho?
-        self._cancel_called_by: tuple = requesting_aid
+        self._cancel_called_by_remote: tuple = requesting_uid
         self._cancel_called = True
 
         # cancel all ongoing rpc tasks
@@ -1110,7 +988,7 @@ class Actor:
 
             # self-cancel **all** ongoing RPC tasks
             await self.cancel_rpc_tasks(
-                req_aid=requesting_aid,
+                req_uid=requesting_uid,
                 parent_chan=None,
             )
 
@@ -1127,11 +1005,19 @@ class Actor:
         self._cancel_complete.set()
         return True
 
+    # XXX: hard kill logic if needed?
+    # def _hard_mofo_kill(self):
+    #     # If we're the root actor or zombied kill everything
+    #     if self._parent_chan is None:  # TODO: more robust check
+    #         root = trio.lowlevel.current_root_task()
+    #         for n in root.child_nurseries:
+    #             n.cancel_scope.cancel()
+
     async def _cancel_task(
         self,
         cid: str,
         parent_chan: Channel,
-        requesting_aid: msgtypes.Aid|None,
+        requesting_uid: tuple[str, str]|None,
 
         ipc_msg: dict|None|bool = False,
 
@@ -1169,7 +1055,7 @@ class Actor:
             log.runtime(
                 'Cancel request for invalid RPC task.\n'
                 'The task likely already completed or was never started!\n\n'
-                f'<= canceller: {requesting_aid}\n'
+                f'<= canceller: {requesting_uid}\n'
                 f'=> {cid}@{parent_chan.uid}\n'
                 f'  |_{parent_chan}\n'
             )
@@ -1177,12 +1063,9 @@ class Actor:
 
         log.cancel(
             'Rxed cancel request for RPC task\n'
-            f'{ctx._task!r} <=c) {requesting_aid}\n'
+            f'<=c) {requesting_uid}\n'
-            f'|_>> {ctx.repr_rpc}\n'
+            f' |_{ctx._task}\n'
-
+            f'    >> {ctx.repr_rpc}\n'
-            # f'|_{ctx._task}\n'
-            # f'   >> {ctx.repr_rpc}\n'
-
             # f'=> {ctx._task}\n'
             # f'  >> Actor._cancel_task() => {ctx._task}\n'
             # f'  |_ {ctx._task}\n\n'
@@ -1203,9 +1086,9 @@ class Actor:
         )
         if (
             ctx._canceller is None
-            and requesting_aid
+            and requesting_uid
         ):
-            ctx._canceller: tuple = requesting_aid.uid
+            ctx._canceller: tuple = requesting_uid
 
         # TODO: pack the RPC `{'cmd': <blah>}` msg into a ctxc and
         # then raise and pack it here?
@@ -1231,7 +1114,7 @@ class Actor:
 
         # wait for _invoke to mark the task complete
         flow_info: str = (
-            f'<= canceller: {requesting_aid}\n'
+            f'<= canceller: {requesting_uid}\n'
             f'=> ipc-parent: {parent_chan}\n'
             f'|_{ctx}\n'
         )
@@ -1248,7 +1131,7 @@ class Actor:
 
     async def cancel_rpc_tasks(
         self,
-        req_aid: msgtypes.Aid,
+        req_uid: tuple[str, str],
 
         # NOTE: when None is passed we cancel **all** rpc
         # tasks running in this actor!
@@ -1265,7 +1148,7 @@ class Actor:
         if not tasks:
             log.runtime(
                 'Actor has no cancellable RPC tasks?\n'
-                f'<= canceller: {req_aid.reprol()}\n'
+                f'<= canceller: {req_uid}\n'
             )
             return
 
@@ -1305,7 +1188,7 @@ class Actor:
         )
         log.cancel(
             f'Cancelling {descr} RPC tasks\n\n'
-            f'<=c) {req_aid} [canceller]\n'
+            f'<=c) {req_uid} [canceller]\n'
             f'{rent_chan_repr}'
             f'c)=> {self.uid} [cancellee]\n'
             f'  |_{self} [with {len(tasks)} tasks]\n'
@@ -1333,7 +1216,7 @@ class Actor:
             await self._cancel_task(
                 cid,
                 task_caller_chan,
-                requesting_aid=req_aid,
+                requesting_uid=req_uid,
             )
 
         if tasks:
@@ -1361,13 +1244,25 @@ class Actor:
         '''
         return self.accept_addrs[0]
 
-    # TODO, this should delegate ONLY to the
+    def get_parent(self) -> Portal:
-    # `._spawn_spec._runtime_vars: dict` / `._state` APIs?
+        '''
-    #
+        Return a `Portal` to our parent.
-    # XXX, AH RIGHT that's why..
+
-    #   it's bc we pass this as a CLI flag to the child.py precisely
+        '''
-    #   bc we need the bootstrapping pre `async_main()`.. but maybe
+        assert self._parent_chan, "No parent channel for this actor?"
-    #   keep this as an impl deat and not part of the pub iface impl?
+        return Portal(self._parent_chan)
+
+    def get_chans(
+        self,
+        uid: tuple[str, str],
+
+    ) -> list[Channel]:
+        '''
+        Return all IPC channels to the actor with provided `uid`.
+
+        '''
+        return self._peers[uid]
+
     def is_infected_aio(self) -> bool:
         '''
         If `True`, this actor is running `trio` in guest mode on
@@ -1378,23 +1273,6 @@ class Actor:
         '''
         return self._infected_aio
 
-    # ?TODO, is this the right type for this method?
-    def get_parent(self) -> Portal:
-        '''
-        Return a `Portal` to our parent.
-
-        '''
-        assert self._parent_chan, "No parent channel for this actor?"
-        return Portal(self._parent_chan)
-
-    # XXX: hard kill logic if needed?
-    # def _hard_mofo_kill(self):
-    #     # If we're the root actor or zombied kill everything
-    #     if self._parent_chan is None:  # TODO: more robust check
-    #         root = trio.lowlevel.current_root_task()
-    #         for n in root.child_nurseries:
-    #             n.cancel_scope.cancel()
-
 
 async def async_main(
     actor: Actor,
@@ -1438,8 +1316,6 @@ async def async_main(
         # establish primary connection with immediate parent
         actor._parent_chan: Channel|None = None
 
-        # is this a sub-actor?
-        # get runtime info from parent.
         if parent_addr is not None:
             (
                 actor._parent_chan,
@@ -1474,18 +1350,18 @@ async def async_main(
         # parent is kept alive as a resilient service until
         # cancellation steps have (mostly) occurred in
         # a deterministic way.
-        root_tn: trio.Nursery
+        async with trio.open_nursery(
-        async with (
+            strict_exception_groups=False,
-            collapse_eg(),
+        ) as root_nursery:
-            trio.open_nursery() as root_tn,
+            actor._root_n = root_nursery
-        ):
-            actor._root_n = root_tn
             assert actor._root_n
 
             ipc_server: _server.IPCServer
             async with (
-                collapse_eg(),
+                trio.open_nursery(
-                trio.open_nursery() as service_nursery,
+                    strict_exception_groups=False,
+                ) as service_nursery,
+
                 _server.open_ipc_server(
                     parent_tn=service_nursery,
                     stream_handler_tn=service_nursery,
@@ -1536,6 +1412,9 @@ async def async_main(
 
                 # TODO: why is this not with the root nursery?
                 try:
+                    log.runtime(
+                        'Booting IPC server'
+                    )
                     eps: list = await ipc_server.listen_on(
                         accept_addrs=accept_addrs,
                         stream_handler_nursery=service_nursery,
@@ -1567,6 +1446,18 @@ async def async_main(
                 # TODO, just read direct from ipc_server?
                 accept_addrs: list[UnwrappedAddress] = actor.accept_addrs
 
+                # NOTE: only set the loopback addr for the 
+                # process-tree-global "root" mailbox since
+                # all sub-actors should be able to speak to
+                # their root actor over that channel.
+                if _state._runtime_vars['_is_root']:
+                    raddrs: list[Address] = _state._runtime_vars['_root_addrs']
+                    for addr in accept_addrs:
+                        waddr: Address = wrap_address(addr)
+                        raddrs.append(addr)
+                    else:
+                        _state._runtime_vars['_root_mailbox'] = raddrs[0]
+
                 # Register with the arbiter if we're told its addr
                 log.runtime(
                     f'Registering `{actor.name}` => {pformat(accept_addrs)}\n'
@@ -1584,7 +1475,6 @@ async def async_main(
                     except AssertionError:
                         await debug.pause()
 
-                    # !TODO, get rid of the local-portal crap XD
                     async with get_registry(addr) as reg_portal:
                         for accept_addr in accept_addrs:
                             accept_addr = wrap_address(accept_addr)
@@ -1609,7 +1499,7 @@ async def async_main(
                 # start processing parent requests until our channel
                 # server is 100% up and running.
                 if actor._parent_chan:
-                    await root_tn.start(
+                    await root_nursery.start(
                         partial(
                             _rpc.process_messages,
                             chan=actor._parent_chan,
@@ -1621,9 +1511,8 @@ async def async_main(
                     # 'Blocking on service nursery to exit..\n'
                 )
             log.runtime(
-                'Service nursery complete\n'
+                "Service nursery complete\n"
-                '\n'
+                "Waiting on root nursery to complete"
-                '->} waiting on root nursery to complete..\n'
             )
 
         # Blocks here as expected until the root nursery is
@@ -1678,7 +1567,6 @@ async def async_main(
     finally:
         teardown_report: str = (
             'Main actor-runtime task completed\n'
-            '\n'
         )
 
         # ?TODO? should this be in `._entry`/`._root` mods instead?
@@ -1720,8 +1608,7 @@ async def async_main(
         # Unregister actor from the registry-sys / registrar.
         if (
             is_registered
-            and
+            and not actor.is_registrar
-            not actor.is_registrar
         ):
             failed: bool = False
             for addr in actor.reg_addrs:
@@ -1756,30 +1643,23 @@ async def async_main(
             ipc_server.has_peers(check_chans=True)
         ):
             teardown_report += (
-                f'-> Waiting for remaining peers to clear..\n'
+                f'-> Waiting for remaining peers {ipc_server._peers} to clear..\n'
-                f'   {pformat(ipc_server._peers)}'
             )
             log.runtime(teardown_report)
-            await ipc_server.wait_for_no_more_peers()
+            await ipc_server.wait_for_no_more_peers(
+                shield=True,
+            )
 
         teardown_report += (
-            '-]> all peer channels are complete.\n'
+            '-> All peer channels are complete\n'
         )
 
-    # op_nested_actor_repr: str = _pformat.nest_from_op(
-    #     input_op=')>',
-    #     text=actor.pformat(),
-    #     nest_prefix='|_',
-    #     nest_indent=1,  # under >
-    # )
     teardown_report += (
-        '-)> actor runtime main task exit.\n'
+        'Actor runtime exiting\n'
-        # f'{op_nested_actor_repr}'
+        f'>)\n'
+        f'|_{actor}\n'
     )
-    # if _state._runtime_vars['_is_root']:
+    log.info(teardown_report)
-    #     log.info(teardown_report)
-    # else:
-    log.runtime(teardown_report)
 
 
 # TODO: rename to `Registry` and move to `.discovery._registry`!

tractor/_spawn.py

@@ -34,9 +34,9 @@ from typing import (
 import trio
 from trio import TaskStatus
 
-from .devx import (
+from .devx.debug import (
-    debug,
+    maybe_wait_for_debugger,
-    pformat as _pformat
+    acquire_debug_lock,
 )
 from tractor._state import (
     current_actor,
@@ -51,17 +51,14 @@ from tractor._portal import Portal
 from tractor._runtime import Actor
 from tractor._entry import _mp_main
 from tractor._exceptions import ActorFailure
-from tractor.msg import (
+from tractor.msg.types import (
-    types as msgtypes,
+    Aid,
-    pretty_struct,
+    SpawnSpec,
 )
 
 
 if TYPE_CHECKING:
-    from ipc import (
+    from ipc import IPCServer
-        _server,
-        Channel,
-    )
     from ._supervise import ActorNursery
     ProcessType = TypeVar('ProcessType', mp.Process, trio.Process)
 
@@ -331,21 +328,20 @@ async def soft_kill(
     see `.hard_kill()`).
 
     '''
-    chan: Channel = portal.channel
+    peer_aid: Aid = portal.channel.aid
-    peer_aid: msgtypes.Aid = chan.aid
     try:
         log.cancel(
             f'Soft killing sub-actor via portal request\n'
             f'\n'
-            f'c)=> {peer_aid.reprol()}@[{chan.maddr}]\n'
+            f'(c=> {peer_aid}\n'
-            f'   |_{proc}\n'
+            f'  |_{proc}\n'
         )
         # wait on sub-proc to signal termination
         await wait_func(proc)
 
     except trio.Cancelled:
         with trio.CancelScope(shield=True):
-            await debug.maybe_wait_for_debugger(
+            await maybe_wait_for_debugger(
                 child_in_debug=_runtime_vars.get(
                     '_debug_mode', False
                 ),
@@ -469,7 +465,7 @@ async def trio_proc(
         "--uid",
         # TODO, how to pass this over "wire" encodings like
         # cmdline args?
-        # -[ ] maybe we can add an `msgtypes.Aid.min_tuple()` ?
+        # -[ ] maybe we can add an `Aid.min_tuple()` ?
         str(subactor.uid),
         # Address the child must connect to on startup
         "--parent_addr",
@@ -487,14 +483,13 @@ async def trio_proc(
 
     cancelled_during_spawn: bool = False
     proc: trio.Process|None = None
-    ipc_server: _server.Server = actor_nursery._actor.ipc_server
+    ipc_server: IPCServer = actor_nursery._actor.ipc_server
     try:
         try:
             proc: trio.Process = await trio.lowlevel.open_process(spawn_cmd, **proc_kwargs)
             log.runtime(
-                f'Started new child subproc\n'
+                'Started new child\n'
-                f'(>\n'
+                f'|_{proc}\n'
-                f' |_{proc}\n'
             )
 
             # wait for actor to spawn and connect back to us
@@ -512,10 +507,10 @@ async def trio_proc(
                 with trio.CancelScope(shield=True):
                     # don't clobber an ongoing pdb
                     if is_root_process():
-                        await debug.maybe_wait_for_debugger()
+                        await maybe_wait_for_debugger()
 
                     elif proc is not None:
-                        async with debug.acquire_debug_lock(subactor.uid):
+                        async with acquire_debug_lock(subactor.uid):
                             # soft wait on the proc to terminate
                             with trio.move_on_after(0.5):
                                 await proc.wait()
@@ -533,19 +528,14 @@ async def trio_proc(
 
         # send a "spawning specification" which configures the
         # initial runtime state of the child.
-        sspec = msgtypes.SpawnSpec(
+        sspec = SpawnSpec(
             _parent_main_data=subactor._parent_main_data,
             enable_modules=subactor.enable_modules,
             reg_addrs=subactor.reg_addrs,
             bind_addrs=bind_addrs,
             _runtime_vars=_runtime_vars,
         )
-        log.runtime(
+        log.runtime(f'Sending spawn spec: {str(sspec)}')
-            f'Sending spawn spec to child\n'
-            f'{{}}=> {chan.aid.reprol()!r}\n'
-            f'\n'
-            f'{pretty_struct.pformat(sspec)}\n'
-        )
         await chan.send(sspec)
 
         # track subactor in current nursery
@@ -573,7 +563,7 @@ async def trio_proc(
             # condition.
             await soft_kill(
                 proc,
-                trio.Process.wait,  # XXX, uses `pidfd_open()` below.
+                trio.Process.wait,
                 portal
             )
 
@@ -581,7 +571,8 @@ async def trio_proc(
             # tandem if not done already
             log.cancel(
                 'Cancelling portal result reaper task\n'
-                f'c)> {subactor.aid.reprol()!r}\n'
+                f'>c)\n'
+                f' |_{subactor.uid}\n'
             )
             nursery.cancel_scope.cancel()
 
@@ -590,24 +581,21 @@ async def trio_proc(
         # allowed! Do this **after** cancellation/teardown to avoid
         # killing the process too early.
         if proc:
-            reap_repr: str = _pformat.nest_from_op(
-                input_op='>x)',
-                text=subactor.pformat(),
-            )
             log.cancel(
                 f'Hard reap sequence starting for subactor\n'
-                f'{reap_repr}'
+                f'>x)\n'
+                f' |_{subactor}@{subactor.uid}\n'
             )
 
             with trio.CancelScope(shield=True):
                 # don't clobber an ongoing pdb
                 if cancelled_during_spawn:
                     # Try again to avoid TTY clobbering.
-                    async with debug.acquire_debug_lock(subactor.uid):
+                    async with acquire_debug_lock(subactor.uid):
                         with trio.move_on_after(0.5):
                             await proc.wait()
 
-                await debug.maybe_wait_for_debugger(
+                await maybe_wait_for_debugger(
                     child_in_debug=_runtime_vars.get(
                         '_debug_mode', False
                     ),
@@ -636,7 +624,7 @@ async def trio_proc(
                 #     acquire the lock and get notified of who has it,
                 #     check that uid against our known children?
                 # this_uid: tuple[str, str] = current_actor().uid
-                # await debug.acquire_debug_lock(this_uid)
+                # await acquire_debug_lock(this_uid)
 
                 if proc.poll() is None:
                     log.cancel(f"Attempting to hard kill {proc}")
@@ -739,7 +727,7 @@ async def mp_proc(
 
     log.runtime(f"Started {proc}")
 
-    ipc_server: _server.Server = actor_nursery._actor.ipc_server
+    ipc_server: IPCServer = actor_nursery._actor.ipc_server
     try:
         # wait for actor to spawn and connect back to us
         # channel should have handshake completed by the

tractor/_streaming.py

@@ -102,9 +102,6 @@ class MsgStream(trio.abc.Channel):
         self._eoc: bool|trio.EndOfChannel = False
         self._closed: bool|trio.ClosedResourceError = False
 
-    def is_eoc(self) -> bool|trio.EndOfChannel:
-        return self._eoc
-
     @property
     def ctx(self) -> Context:
         '''
@@ -191,14 +188,7 @@ class MsgStream(trio.abc.Channel):
 
         return pld
 
-    # XXX NOTE, this is left private because in `.subscribe()` usage
+    async def receive(
-    # we rebind the public `.recieve()` to a `BroadcastReceiver` but
-    # on `.subscribe().__aexit__()`, for the first task which enters,
-    # we want to revert to this msg-stream-instance's method since
-    # mult-task-tracking provided by the b-caster is then no longer
-    # necessary.
-    #
-    async def _receive(
         self,
         hide_tb: bool = False,
     ):
@@ -323,8 +313,6 @@ class MsgStream(trio.abc.Channel):
 
             raise src_err
 
-    receive = _receive
-
     async def aclose(self) -> list[Exception|dict]:
         '''
         Cancel associated remote actor task and local memory channel on
@@ -540,15 +528,10 @@ class MsgStream(trio.abc.Channel):
         receiver wrapper.
 
         '''
-        # XXX NOTE, This operation was originally implemented as
+        # NOTE: This operation is indempotent and non-reversible, so be
-        # indempotent and non-reversible, so you had to be **VERY**
+        # sure you can deal with any (theoretical) overhead of the the
-        # aware of any (theoretical) overhead from the allocated
+        # allocated ``BroadcastReceiver`` before calling this method for
-        # `BroadcastReceiver.receive()`.
+        # the first time.
-        #
-        # HOWEVER, NOw we do revert and de-alloc the ._broadcaster
-        # when the final caller (task) exits.
-        #
-        bcast: BroadcastReceiver|None = None
         if self._broadcaster is None:
 
             bcast = self._broadcaster = broadcast_receiver(
@@ -558,60 +541,29 @@ class MsgStream(trio.abc.Channel):
 
                 # TODO: can remove this kwarg right since
                 # by default behaviour is to do this anyway?
-                receive_afunc=self._receive,
+                receive_afunc=self.receive,
             )
 
-            # XXX NOTE, we override the original stream instance's
+            # NOTE: we override the original stream instance's receive
-            # receive method to instead delegate to the broadcaster's
+            # method to now delegate to the broadcaster's ``.receive()``
-            # `.receive()` such that new subscribers (multiple
+            # such that new subscribers will be copied received values
-            # `trio.Task`s) will be copied received values and the
+            # and this stream doesn't have to expect it's original
-            # *first* task to enter here doesn't have to expect its original consumer(s)
+            # consumer(s) to get a new broadcast rx handle.
-            # to get a new broadcast rx handle; everything happens
-            # underneath this iface seemlessly.
-            #
             self.receive = bcast.receive  # type: ignore
-            # seems there's no graceful way to type this with `mypy`?
+            # seems there's no graceful way to type this with ``mypy``?
             # https://github.com/python/mypy/issues/708
 
-        # TODO, prevent re-entrant sub scope?
+        async with self._broadcaster.subscribe() as bstream:
-        # if self._broadcaster._closed:
+            assert bstream.key != self._broadcaster.key
-        #     raise RuntimeError(
+            assert bstream._recv == self._broadcaster._recv
-        #         'This stream
 
-        try:
+            # NOTE: we patch on a `.send()` to the bcaster so that the
-            aenter = self._broadcaster.subscribe()
+            # caller can still conduct 2-way streaming using this
-            async with aenter as bstream:
+            # ``bstream`` handle transparently as though it was the msg
-                # ?TODO, move into test suite?
+            # stream instance.
-                assert bstream.key != self._broadcaster.key
+            bstream.send = self.send  # type: ignore
-                assert bstream._recv == self._broadcaster._recv
 
-                # NOTE: we patch on a `.send()` to the bcaster so that the
+            yield bstream
-                # caller can still conduct 2-way streaming using this
-                # ``bstream`` handle transparently as though it was the msg
-                # stream instance.
-                bstream.send = self.send  # type: ignore
-
-                # newly-allocated instance
-                yield bstream
-
-        finally:
-            # XXX, the first-enterer task should, like all other
-            # subs, close the first allocated bcrx, which adjusts the
-            # common `bcrx.state`
-            with trio.CancelScope(shield=True):
-                if bcast is not None:
-                    await bcast.aclose()
-
-                # XXX, when the bcrx.state reports there are no more subs
-                # we can revert to this obj's method, removing any
-                # delegation overhead!
-                if (
-                    (orig_bcast := self._broadcaster)
-                    and
-                    not orig_bcast.state.subs
-                ):
-                    self.receive = self._receive
-                    # self._broadcaster = None
 
     async def send(
         self,

tractor/_supervise.py

@@ -21,6 +21,7 @@
 from contextlib import asynccontextmanager as acm
 from functools import partial
 import inspect
+from pprint import pformat
 from typing import (
     TYPE_CHECKING,
 )
@@ -30,10 +31,7 @@ import warnings
 import trio
 
 
-from .devx import (
+from .devx.debug import maybe_wait_for_debugger
-    debug,
-    pformat as _pformat,
-)
 from ._addr import (
     UnwrappedAddress,
     mk_uuid,
@@ -44,7 +42,6 @@ from ._runtime import Actor
 from ._portal import Portal
 from .trionics import (
     is_multi_cancelled,
-    collapse_eg,
 )
 from ._exceptions import (
     ContextCancelled,
@@ -117,6 +114,7 @@ class ActorNursery:
             ]
         ] = {}
 
+        self.cancelled: bool = False
         self._join_procs = trio.Event()
         self._at_least_one_child_in_debug: bool = False
         self.errors = errors
@@ -134,53 +132,10 @@ class ActorNursery:
         # TODO: remove the `.run_in_actor()` API and thus this 2ndary
         # nursery when that API get's moved outside this primitive!
         self._ria_nursery = ria_nursery
-
-        # TODO, factor this into a .hilevel api!
-        #
         # portals spawned with ``run_in_actor()`` are
         # cancelled when their "main" result arrives
         self._cancel_after_result_on_exit: set = set()
 
-        # trio.Nursery-like cancel (request) statuses
-        self._cancelled_caught: bool = False
-        self._cancel_called: bool = False
-
-    @property
-    def cancel_called(self) -> bool:
-        '''
-        Records whether cancellation has been requested for this
-        actor-nursery by a call to  `.cancel()` either due to,
-        - an explicit call by some actor-local-task,
-        - an implicit call due to an error/cancel emited inside
-          the `tractor.open_nursery()` block.
-
-        '''
-        return self._cancel_called
-
-    @property
-    def cancelled_caught(self) -> bool:
-        '''
-        Set when this nursery was able to cance all spawned subactors
-        gracefully via an (implicit) call to `.cancel()`.
-
-        '''
-        return self._cancelled_caught
-
-    # TODO! remove internal/test-suite usage!
-    @property
-    def cancelled(self) -> bool:
-        warnings.warn(
-            "`ActorNursery.cancelled` is now deprecated, use "
-            " `.cancel_called` instead.",
-            DeprecationWarning,
-            stacklevel=2,
-        )
-        return (
-            self._cancel_called
-            # and
-            # self._cancelled_caught
-        )
-
     async def start_actor(
         self,
         name: str,
@@ -244,7 +199,7 @@ class ActorNursery:
             loglevel=loglevel,
 
             # verbatim relay this actor's registrar addresses
-            registry_addrs=current_actor().registry_addrs,
+            registry_addrs=current_actor().reg_addrs,
         )
         parent_addr: UnwrappedAddress = self._actor.accept_addr
         assert parent_addr
@@ -358,7 +313,7 @@ class ActorNursery:
 
         '''
         __runtimeframe__: int = 1  # noqa
-        self._cancel_called = True
+        self.cancelled = True
 
         # TODO: impl a repr for spawn more compact
         # then `._children`..
@@ -369,10 +324,9 @@ class ActorNursery:
         server: IPCServer = self._actor.ipc_server
 
         with trio.move_on_after(3) as cs:
-            async with (
+            async with trio.open_nursery(
-                collapse_eg(),
+                strict_exception_groups=False,
-                trio.open_nursery() as tn,
+            ) as tn:
-            ):
 
                 subactor: Actor
                 proc: trio.Process
@@ -436,8 +390,6 @@ class ActorNursery:
             ) 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._join_procs.set()
@@ -467,10 +419,10 @@ async def _open_and_supervise_one_cancels_all_nursery(
     # `ActorNursery.start_actor()`).
 
     # errors from this daemon actor nursery bubble up to caller
-    async with (
+    async with trio.open_nursery(
-        collapse_eg(),
+        strict_exception_groups=False,
-        trio.open_nursery() as da_nursery,
+        # ^XXX^ TODO? instead unpack any RAE as per "loose" style?
-    ):
+    ) as da_nursery:
         try:
             # This is the inner level "run in actor" nursery. It is
             # awaited first since actors spawned in this way (using
@@ -480,10 +432,11 @@ async def _open_and_supervise_one_cancels_all_nursery(
             # immediately raised for handling by a supervisor strategy.
             # As such if the strategy propagates any error(s) upwards
             # the above "daemon actor" nursery will be notified.
-            async with (
+            async with trio.open_nursery(
-                collapse_eg(),
+                strict_exception_groups=False,
-                trio.open_nursery() as ria_nursery,
+                # ^XXX^ TODO? instead unpack any RAE as per "loose" style?
-            ):
+            ) as ria_nursery:
+
                 an = ActorNursery(
                     actor,
                     ria_nursery,
@@ -500,7 +453,7 @@ async def _open_and_supervise_one_cancels_all_nursery(
                     # the "hard join phase".
                     log.runtime(
                         'Waiting on subactors to complete:\n'
-                        f'>}} {len(an._children)}\n'
+                        f'{pformat(an._children)}\n'
                     )
                     an._join_procs.set()
 
@@ -514,7 +467,7 @@ async def _open_and_supervise_one_cancels_all_nursery(
                     # will make the pdb repl unusable.
                     # Instead try to wait for pdb to be released before
                     # tearing down.
-                    await debug.maybe_wait_for_debugger(
+                    await maybe_wait_for_debugger(
                         child_in_debug=an._at_least_one_child_in_debug
                     )
 
@@ -590,7 +543,7 @@ async def _open_and_supervise_one_cancels_all_nursery(
 
             # XXX: yet another guard before allowing the cancel
             # sequence in case a (single) child is in debug.
-            await debug.maybe_wait_for_debugger(
+            await maybe_wait_for_debugger(
                 child_in_debug=an._at_least_one_child_in_debug
             )
 
@@ -639,14 +592,9 @@ async def _open_and_supervise_one_cancels_all_nursery(
     # final exit
 
 
-_shutdown_msg: str = (
-    'Actor-runtime-shutdown'
-)
-
-# @api_frame
 @acm
+# @api_frame
 async def open_nursery(
-    *,  # named params only!
     hide_tb: bool = True,
     **kwargs,
     # ^TODO, paramspec for `open_root_actor()`
@@ -731,26 +679,17 @@ async def open_nursery(
         ):
             __tracebackhide__: bool = False
 
-
+        msg: str = (
-        op_nested_an_repr: str = _pformat.nest_from_op(
+            'Actor-nursery exited\n'
-            input_op=')>',
+            f'|_{an}\n'
-            text=f'{an}',
-            # nest_prefix='|_',
-            nest_indent=1,  # under >
         )
-        an_msg: str = (
-            f'Actor-nursery exited\n'
-            f'{op_nested_an_repr}\n'
-        )
-        # keep noise low during std operation.
-        log.runtime(an_msg)
 
         if implicit_runtime:
             # shutdown runtime if it was started and report noisly
             # that we're did so.
-            msg: str = (
+            msg += '=> Shutting down actor runtime <=\n'
-                '\n'
-                '\n'
-                f'{_shutdown_msg} )>\n'
-            )
             log.info(msg)
+
+        else:
+            # keep noise low during std operation.
+            log.runtime(msg)

tractor/ipc/_chan.py

@@ -101,27 +101,11 @@ class Channel:
         # ^XXX! ONLY set if a remote actor sends an `Error`-msg
         self._closed: bool = False
 
-        # flag set by `Portal.cancel_actor()` indicating remote
+        # flag set by ``Portal.cancel_actor()`` indicating remote
-        # (possibly peer) cancellation of the far end actor runtime.
+        # (possibly peer) cancellation of the far end actor
+        # runtime.
         self._cancel_called: bool = False
 
-    @property
-    def closed(self) -> bool:
-        '''
-        Was `.aclose()` successfully called?
-
-        '''
-        return self._closed
-
-    @property
-    def cancel_called(self) -> bool:
-        '''
-        Set when `Portal.cancel_actor()` is called on a portal which
-        wraps this IPC channel.
-
-        '''
-        return self._cancel_called
-
     @property
     def uid(self) -> tuple[str, str]:
         '''
@@ -187,23 +171,11 @@ class Channel:
         )
         assert transport.raddr == addr
         chan = Channel(transport=transport)
-
+        log.runtime(
-        # ?TODO, compact this into adapter level-methods?
+            f'Connected channel IPC transport\n'
-        # -[ ] would avoid extra repr-calcs if level not active?
+            f'[>\n'
-        #   |_ how would the `calc_if_level` look though? func?
+            f' |_{chan}\n'
-        if log.at_least_level('runtime'):
+        )
-            from tractor.devx import (
-                pformat as _pformat,
-            )
-            chan_repr: str = _pformat.nest_from_op(
-                input_op='[>',
-                text=chan.pformat(),
-                nest_indent=1,
-            )
-            log.runtime(
-                f'Connected channel IPC transport\n'
-                f'{chan_repr}'
-            )
         return chan
 
     @cm
@@ -224,12 +196,9 @@ class Channel:
             self._transport.codec = orig
 
     # TODO: do a .src/.dst: str for maddrs?
-    def pformat(
+    def pformat(self) -> str:
-        self,
-        privates: bool = False,
-    ) -> str:
         if not self._transport:
-            return '<Channel( with inactive transport? )>'
+            return '<Channel with inactive transport?>'
 
         tpt: MsgTransport = self._transport
         tpt_name: str = type(tpt).__name__
@@ -237,35 +206,26 @@ class Channel:
             'connected' if self.connected()
             else 'closed'
         )
-        repr_str: str = (
+        return (
             f'<Channel(\n'
             f' |_status: {tpt_status!r}\n'
-        ) + (
             f'   _closed={self._closed}\n'
             f'   _cancel_called={self._cancel_called}\n'
-            if privates else ''
+            f'\n'
-        ) + (  # peer-actor (processs) section
+            f' |_peer: {self.aid}\n'
-            f' |_peer: {self.aid.reprol()!r}\n'
+            f'\n'
-            if self.aid else ' |_peer: <unknown>\n'
-        ) + (
             f' |_msgstream: {tpt_name}\n'
-            f'   maddr: {tpt.maddr!r}\n'
+            f'   proto={tpt.laddr.proto_key!r}\n'
-            f'   proto: {tpt.laddr.proto_key!r}\n'
+            f'   layer={tpt.layer_key!r}\n'
-            f'   layer: {tpt.layer_key!r}\n'
+            f'   laddr={tpt.laddr}\n'
-            f'   codec: {tpt.codec_key!r}\n'
+            f'   raddr={tpt.raddr}\n'
-            f'   .laddr={tpt.laddr}\n'
+            f'   codec={tpt.codec_key!r}\n'
-            f'   .raddr={tpt.raddr}\n'
+            f'   stream={tpt.stream}\n'
-        ) + (
+            f'   maddr={tpt.maddr!r}\n'
-            f'   ._transport.stream={tpt.stream}\n'
+            f'   drained={tpt.drained}\n'
-            f'   ._transport.drained={tpt.drained}\n'
-            if privates else ''
-        ) + (
             f'   _send_lock={tpt._send_lock.statistics()}\n'
-            if privates else ''
+            f')>\n'
-        ) + (
-            ')>\n'
         )
-        return repr_str
 
     # NOTE: making this return a value that can be passed to
     # `eval()` is entirely **optional** FYI!
@@ -287,10 +247,6 @@ class Channel:
     def raddr(self) -> Address|None:
         return self._transport.raddr if self._transport else None
 
-    @property
-    def maddr(self) -> str:
-        return self._transport.maddr if self._transport else '<no-tpt>'
-
     # TODO: something like,
     # `pdbp.hideframe_on(errors=[MsgTypeError])`
     # instead of the `try/except` hack we have rn..
@@ -478,8 +434,8 @@ class Channel:
         await self.send(aid)
         peer_aid: Aid = await self.recv()
         log.runtime(
-            f'Received hanshake with peer\n'
+            f'Received hanshake with peer actor,\n'
-            f'<= {peer_aid.reprol(sin_uuid=False)}\n'
+            f'{peer_aid}\n'
         )
         # NOTE, we always are referencing the remote peer!
         self.aid = peer_aid

tractor/ipc/_mp_bs.py

@@ -17,16 +17,9 @@
 Utils to tame mp non-SC madeness
 
 '''
-
-# !TODO! in 3.13 this can be disabled (the-same/similarly) using
-# a flag,
-# - [ ] soo if it works like this, drop this module entirely for
-#   3.13+ B)
-#  |_https://docs.python.org/3/library/multiprocessing.shared_memory.html
-#
 def disable_mantracker():
     '''
-    Disable all `multiprocessing` "resource tracking" machinery since
+    Disable all ``multiprocessing``` "resource tracking" machinery since
     it's an absolute multi-threaded mess of non-SC madness.
 
     '''

tractor/ipc/_server.py

@@ -26,7 +26,7 @@ from contextlib import (
 from functools import partial
 from itertools import chain
 import inspect
-import textwrap
+from pprint import pformat
 from types import (
     ModuleType,
 )
@@ -43,10 +43,7 @@ from trio import (
     SocketListener,
 )
 
-from ..devx.pformat import (
+# from ..devx import debug
-    ppfmt,
-    nest_from_op,
-)
 from .._exceptions import (
     TransportClosed,
 )
@@ -144,8 +141,9 @@ async def maybe_wait_on_canced_subs(
 
     ):
         log.cancel(
-            'Waiting on cancel request to peer\n'
+            'Waiting on cancel request to peer..\n'
-            f'c)=> {chan.aid.reprol()}@[{chan.maddr}]\n'
+            f'c)=>\n'
+            f'  |_{chan.aid}\n'
         )
 
         # XXX: this is a soft wait on the channel (and its
@@ -181,7 +179,7 @@ async def maybe_wait_on_canced_subs(
                 log.warning(
                     'Draining msg from disconnected peer\n'
                     f'{chan_info}'
-                    f'{ppfmt(msg)}\n'
+                    f'{pformat(msg)}\n'
                 )
                 # cid: str|None = msg.get('cid')
                 cid: str|None = msg.cid
@@ -250,7 +248,7 @@ async def maybe_wait_on_canced_subs(
                 if children := local_nursery._children:
                     # indent from above local-nurse repr
                     report += (
-                        f'   |_{ppfmt(children)}\n'
+                        f'   |_{pformat(children)}\n'
                     )
 
                 log.warning(report)
@@ -281,9 +279,8 @@ async def maybe_wait_on_canced_subs(
                     log.runtime(
                         f'Peer IPC broke but subproc is alive?\n\n'
 
-                        f'<=x {chan.aid.reprol()}@[{chan.maddr}]\n'
+                        f'<=x {chan.aid}@{chan.raddr}\n'
-                        f'\n'
+                        f'   |_{proc}\n'
-                        f'{proc}\n'
                     )
 
     return local_nursery
@@ -327,10 +324,9 @@ async def handle_stream_from_peer(
 
     chan = Channel.from_stream(stream)
     con_status: str = (
-        f'New inbound IPC transport connection\n'
+        'New inbound IPC connection <=\n'
-        f'<=( {stream!r}\n'
+        f'|_{chan}\n'
     )
-    con_status_steps: str = ''
 
     # initial handshake with peer phase
     try:
@@ -376,7 +372,7 @@ async def handle_stream_from_peer(
     if _pre_chan := server._peers.get(uid):
         familiar: str = 'pre-existing-peer'
     uid_short: str = f'{uid[0]}[{uid[1][-6:]}]'
-    con_status_steps += (
+    con_status += (
         f' -> Handshake with {familiar} `{uid_short}` complete\n'
     )
 
@@ -401,7 +397,7 @@ async def handle_stream_from_peer(
         None,
     )
     if event:
-        con_status_steps += (
+        con_status += (
             ' -> Waking subactor spawn waiters: '
             f'{event.statistics().tasks_waiting}\n'
             f' -> Registered IPC chan for child actor {uid}@{chan.raddr}\n'
@@ -412,7 +408,7 @@ async def handle_stream_from_peer(
         event.set()
 
     else:
-        con_status_steps += (
+        con_status += (
             f' -> Registered IPC chan for peer actor {uid}@{chan.raddr}\n'
         )  # type: ignore
 
@@ -426,15 +422,8 @@ async def handle_stream_from_peer(
     # TODO: can we just use list-ref directly?
     chans.append(chan)
 
-    con_status_steps += ' -> Entering RPC msg loop..\n'
+    con_status += ' -> Entering RPC msg loop..\n'
-    log.runtime(
+    log.runtime(con_status)
-        con_status
-        +
-        textwrap.indent(
-            con_status_steps,
-            prefix=' '*3,  # align to first-ln
-        )
-    )
 
     # Begin channel management - respond to remote requests and
     # process received reponses.
@@ -467,67 +456,41 @@ async def handle_stream_from_peer(
             disconnected=disconnected,
         )
 
-        # `Channel` teardown and closure sequence
+        # ``Channel`` teardown and closure sequence
         # drop ref to channel so it can be gc-ed and disconnected
-        #
-        # -[x]TODO mk this be like
-        # <=x Channel(
-        #     |_field: blah
-        # )>
-        op_repr: str = '<=x '
-        chan_repr: str = nest_from_op(
-            input_op=op_repr,
-            op_suffix='',
-            nest_prefix='',
-            text=chan.pformat(),
-            nest_indent=len(op_repr)-1,
-            rm_from_first_ln='<',
-        )
-
         con_teardown_status: str = (
-            f'IPC channel disconnect\n'
+            f'IPC channel disconnected:\n'
-            f'\n'
+            f'<=x uid: {chan.aid}\n'
-            f'{chan_repr}\n'
+            f'   |_{pformat(chan)}\n\n'
-            f'\n'
         )
-
         chans.remove(chan)
 
         # TODO: do we need to be this pedantic?
         if not chans:
             con_teardown_status += (
-                f'-> No more channels with {chan.aid.reprol()!r}\n'
+                f'-> No more channels with {chan.aid}'
             )
             server._peers.pop(uid, None)
 
-        if peers := list(server._peers.values()):
+        peers_str: str = ''
-            peer_cnt: int = len(peers)
+        for uid, chans in server._peers.items():
-            if (
+            peers_str += (
-                (first := peers[0][0]) is not chan
+                f'uid: {uid}\n'
-                and
+            )
-                not disconnected
+            for i, chan in enumerate(chans):
-                and
+                peers_str += (
-                peer_cnt > 1
+                    f' |_[{i}] {pformat(chan)}\n'
-            ):
-                con_teardown_status += (
-                    f'-> Remaining IPC {peer_cnt-1!r} peers:\n'
                 )
-                for chans in server._peers.values():
+
-                    first: Channel = chans[0]
+        con_teardown_status += (
-                    if not (
+            f'-> Remaining IPC {len(server._peers)} peers: {peers_str}\n'
-                        first is chan
+        )
-                        and
-                        disconnected
-                    ):
-                        con_teardown_status += (
-                            f'  |_{first.aid.reprol()!r} -> {len(chans)!r} chans\n'
-                        )
 
         # No more channels to other actors (at all) registered
         # as connected.
         if not server._peers:
             con_teardown_status += (
-                '-> Signalling no more peer connections!\n'
+                'Signalling no more peer channel connections'
             )
             server._no_more_peers.set()
 
@@ -616,10 +579,10 @@ async def handle_stream_from_peer(
 
 class Endpoint(Struct):
     '''
-    An instance of an IPC "bound" address where the lifetime of an
+    An instance of an IPC "bound" address where the lifetime of the
-    "ability to accept connections" and handle the subsequent
+    "ability to accept connections" (from clients) and then handle
-    sequence-of-packets (maybe oriented as sessions) is determined by
+    those inbound sessions or sequences-of-packets is determined by
-    the underlying nursery scope(s).
+    a (maybe pair of) nurser(y/ies).
 
     '''
     addr: Address
@@ -637,24 +600,6 @@ class Endpoint(Struct):
         MsgTransport,  # handle to encoded-msg transport stream
     ] = {}
 
-    def pformat(
-        self,
-        indent: int = 0,
-        privates: bool = False,
-    ) -> str:
-        type_repr: str = type(self).__name__
-        fmtstr: str = (
-            # !TODO, always be ns aware!
-            # f'|_netns: {netns}\n'
-            f' |.addr: {self.addr!r}\n'
-            f' |_peers: {len(self.peer_tpts)}\n'
-        )
-        return (
-            f'<{type_repr}(\n'
-            f'{fmtstr}'
-            f')>'
-        )
-
     async def start_listener(self) -> SocketListener:
         tpt_mod: ModuleType = inspect.getmodule(self.addr)
         lstnr: SocketListener = await tpt_mod.start_listener(
@@ -694,13 +639,11 @@ class Endpoint(Struct):
 class Server(Struct):
     _parent_tn: Nursery
     _stream_handler_tn: Nursery
-
     # level-triggered sig for whether "no peers are currently
     # connected"; field is **always** set to an instance but
     # initialized with `.is_set() == True`.
     _no_more_peers: trio.Event
 
-    # active eps as allocated by `.listen_on()`
     _endpoints: list[Endpoint] = []
 
     # connection tracking & mgmt
@@ -708,19 +651,12 @@ class Server(Struct):
         str,  # uaid
         list[Channel],  # IPC conns from peer
     ] = defaultdict(list)
-
-    # events-table with entries registered unset while the local
-    # actor is waiting on a new actor to inbound connect, often
-    # a parent waiting on its child just after spawn.
     _peer_connected: dict[
         tuple[str, str],
         trio.Event,
     ] = {}
 
     # syncs for setup/teardown sequences
-    # - null when not yet booted,
-    # - unset when active,
-    # - set when fully shutdown with 0 eps active.
     _shutdown: trio.Event|None = None
 
     # TODO, maybe just make `._endpoints: list[Endpoint]` and
@@ -728,6 +664,7 @@ class Server(Struct):
     # @property
     # def addrs2eps(self) -> dict[Address, Endpoint]:
     #     ...
+
     @property
     def proto_keys(self) -> list[str]:
         return [
@@ -753,7 +690,7 @@ class Server(Struct):
             # TODO: obvi a different server type when we eventually
             # support some others XD
             log.runtime(
-                f'Cancelling server(s) for tpt-protos\n'
+                f'Cancelling server(s) for\n'
                 f'{self.proto_keys!r}\n'
             )
             self._parent_tn.cancel_scope.cancel()
@@ -780,14 +717,6 @@ class Server(Struct):
                 f'protos: {tpt_protos!r}\n'
             )
 
-    def len_peers(
-        self,
-    ) -> int:
-        return len([
-            chan.connected()
-            for chan in chain(*self._peers.values())
-        ])
-
     def has_peers(
         self,
         check_chans: bool = False,
@@ -801,11 +730,13 @@ class Server(Struct):
             has_peers
             and
             check_chans
-            and
-            (peer_cnt := self.len_peers())
         ):
             has_peers: bool = (
-                peer_cnt > 0
+                any(chan.connected()
+                    for chan in chain(
+                        *self._peers.values()
+                    )
+                )
                 and
                 has_peers
             )
@@ -814,14 +745,10 @@ class Server(Struct):
 
     async def wait_for_no_more_peers(
         self,
-        # XXX, should this even be allowed?
+        shield: bool = False,
-        # -> i've seen it cause hangs on teardown
-        #    in `test_resource_cache.py`
-        # _shield: bool = False,
     ) -> None:
-        await self._no_more_peers.wait()
+        with trio.CancelScope(shield=shield):
-        # with trio.CancelScope(shield=_shield):
+            await self._no_more_peers.wait()
-        #     await self._no_more_peers.wait()
 
     async def wait_for_peer(
         self,
@@ -876,66 +803,30 @@ class Server(Struct):
 
         return ev.is_set()
 
-    @property
+    def pformat(self) -> str:
-    def repr_state(self) -> str:
-        '''
-        A `str`-status describing the current state of this
-        IPC server in terms of the current operating "phase".
-
-        '''
-        status = 'server is active'
-        if self.has_peers():
-            peer_cnt: int = self.len_peers()
-            status: str = (
-                f'{peer_cnt!r} peer chans'
-            )
-        else:
-            status: str = 'No peer chans'
-
-        if self.is_shutdown():
-            status: str = 'server-shutdown'
-
-        return status
-
-    def pformat(
-        self,
-        privates: bool = False,
-    ) -> str:
         eps: list[Endpoint] = self._endpoints
 
-        # state_repr: str = (
+        state_repr: str = (
-        #     f'{len(eps)!r} endpoints active'
+            f'{len(eps)!r} IPC-endpoints active'
-        # )
-        fmtstr = (
-            f' |_state: {self.repr_state!r}\n'
         )
-        if privates:
+        fmtstr = (
-            fmtstr += f'   no_more_peers: {self.has_peers()}\n'
+            f' |_state: {state_repr}\n'
-
+            f'   no_more_peers: {self.has_peers()}\n'
-            if self._shutdown is not None:
+        )
-                shutdown_stats: EventStatistics = self._shutdown.statistics()
+        if self._shutdown is not None:
-                fmtstr += (
+            shutdown_stats: EventStatistics = self._shutdown.statistics()
-                    f'   task_waiting_on_shutdown: {shutdown_stats}\n'
-                )
-
-        if eps := self._endpoints:
-            addrs: list[tuple] = [
-                ep.addr for ep in eps
-            ]
-            repr_eps: str = ppfmt(addrs)
-
             fmtstr += (
-                f' |_endpoints: {repr_eps}\n'
+                f'   task_waiting_on_shutdown: {shutdown_stats}\n'
-                # ^TODO? how to indent closing ']'..
             )
 
-        if peers := self._peers:
+        fmtstr += (
-            fmtstr += (
+            # TODO, use the `ppfmt()` helper from `modden`!
-                f' |_peers: {len(peers)} connected\n'
+            f' |_endpoints: {pformat(self._endpoints)}\n'
-            )
+            f' |_peers: {len(self._peers)} connected\n'
+        )
 
         return (
-            f'<Server(\n'
+            f'<IPCServer(\n'
             f'{fmtstr}'
             f')>\n'
         )
@@ -994,8 +885,8 @@ class Server(Struct):
             )
 
         log.runtime(
-            f'Binding endpoints\n'
+            f'Binding to endpoints for,\n'
-            f'{ppfmt(accept_addrs)}\n'
+            f'{accept_addrs}\n'
         )
         eps: list[Endpoint] = await self._parent_tn.start(
             partial(
@@ -1005,19 +896,13 @@ class Server(Struct):
                 listen_addrs=accept_addrs,
             )
         )
-        self._endpoints.extend(eps)
-
-        serv_repr: str = nest_from_op(
-            input_op='(>',
-            text=self.pformat(),
-            nest_indent=1,
-        )
         log.runtime(
-            f'Started IPC server\n'
+            f'Started IPC endpoints\n'
-            f'{serv_repr}'
+            f'{eps}\n'
         )
 
-        # XXX, a little sanity on new ep allocations
+        self._endpoints.extend(eps)
+        # XXX, just a little bit of sanity
         group_tn: Nursery|None = None
         ep: Endpoint
         for ep in eps:
@@ -1071,13 +956,9 @@ async def _serve_ipc_eps(
                     stream_handler_tn=stream_handler_tn,
                 )
                 try:
-                    ep_sclang: str = nest_from_op(
-                        input_op='>[',
-                        text=f'{ep.pformat()}',
-                    )
                     log.runtime(
                         f'Starting new endpoint listener\n'
-                        f'{ep_sclang}\n'
+                        f'{ep}\n'
                     )
                     listener: trio.abc.Listener = await ep.start_listener()
                     assert listener is ep._listener
@@ -1115,6 +996,17 @@ async def _serve_ipc_eps(
                     handler_nursery=stream_handler_tn
                 )
             )
+            # TODO, wow make this message better! XD
+            log.runtime(
+                'Started server(s)\n'
+                +
+                '\n'.join([f'|_{addr}' for addr in listen_addrs])
+            )
+
+            log.runtime(
+                f'Started IPC endpoints\n'
+                f'{eps}\n'
+            )
             task_status.started(
                 eps,
             )
@@ -1157,7 +1049,8 @@ async def open_ipc_server(
         try:
             yield ipc_server
             log.runtime(
-                'Server-tn running until terminated\n'
+                f'Waiting on server to shutdown or be cancelled..\n'
+                f'{ipc_server}'
             )
             # TODO? when if ever would we want/need this?
             # with trio.CancelScope(shield=True):

tractor/ipc/_shm.py

@@ -789,11 +789,6 @@ def open_shm_list(
         readonly=readonly,
     )
 
-    # TODO, factor into a @actor_fixture acm-API?
-    # -[ ] also `@maybe_actor_fixture()` which inludes
-    #     the .current_actor() convenience check?
-    #   |_ orr can that just be in the sin-maybe-version?
-    #
     # "close" attached shm on actor teardown
     try:
         actor = tractor.current_actor()

tractor/ipc/_tcp.py

@@ -160,9 +160,10 @@ async def start_listener(
     Start a TCP socket listener on the given `TCPAddress`.
 
     '''
-    log.runtime(
+    log.info(
-        f'Trying socket bind\n'
+        f'Attempting to bind TCP socket\n'
-        f'>[ {addr}\n'
+        f'>[\n'
+        f'|_{addr}\n'
     )
     # ?TODO, maybe we should just change the lower-level call this is
     # using internall per-listener?
@@ -177,10 +178,11 @@ async def start_listener(
     assert len(listeners) == 1
     listener = listeners[0]
     host, port = listener.socket.getsockname()[:2]
-    bound_addr: TCPAddress = type(addr).from_addr((host, port))
+
     log.info(
         f'Listening on TCP socket\n'
-        f'[> {bound_addr}\n'
+        f'[>\n'
+        f' |_{addr}\n'
     )
     return listener
 

tractor/log.py

@@ -81,35 +81,10 @@ BOLD_PALETTE = {
 }
 
 
-def at_least_level(
-    log: Logger|LoggerAdapter,
-    level: int|str,
-) -> bool:
-    '''
-    Predicate to test if a given level is active.
-
-    '''
-    if isinstance(level, str):
-        level: int = CUSTOM_LEVELS[level.upper()]
-
-    if log.getEffectiveLevel() <= level:
-        return True
-    return False
-
-
 # TODO: this isn't showing the correct '{filename}'
 # as it did before..
 class StackLevelAdapter(LoggerAdapter):
 
-    def at_least_level(
-        self,
-        level: str,
-    ) -> bool:
-        return at_least_level(
-            log=self,
-            level=level,
-        )
-
     def transport(
         self,
         msg: str,
@@ -426,3 +401,19 @@ def get_loglevel() -> str:
 
 # global module logger for tractor itself
 log: StackLevelAdapter = get_logger('tractor')
+
+
+def at_least_level(
+    log: Logger|LoggerAdapter,
+    level: int|str,
+) -> bool:
+    '''
+    Predicate to test if a given level is active.
+
+    '''
+    if isinstance(level, str):
+        level: int = CUSTOM_LEVELS[level.upper()]
+
+    if log.getEffectiveLevel() <= level:
+        return True
+    return False

tractor/msg/_ops.py

@@ -210,14 +210,12 @@ class PldRx(Struct):
         match msg:
             case Return()|Error():
                 log.runtime(
-                    f'Rxed final-outcome msg\n'
+                    f'Rxed final outcome msg\n'
-                    f'\n'
                     f'{msg}\n'
                 )
             case Stop():
                 log.runtime(
                     f'Rxed stream stopped msg\n'
-                    f'\n'
                     f'{msg}\n'
                 )
                 if passthrough_non_pld_msgs:
@@ -263,9 +261,8 @@ class PldRx(Struct):
         if (
             type(msg) is Return
         ):
-            log.runtime(
+            log.info(
                 f'Rxed final result msg\n'
-                f'\n'
                 f'{msg}\n'
             )
         return self.decode_pld(
@@ -307,13 +304,10 @@ class PldRx(Struct):
                 try:
                     pld: PayloadT = self._pld_dec.decode(pld)
                     log.runtime(
-                        f'Decoded payload for\n'
+                        'Decoded msg payload\n\n'
-                        # f'\n'
                         f'{msg}\n'
-                        # ^TODO?, ideally just render with `,
+                        f'where payload decoded as\n'
-                        # pld={decode}` in the `msg.pformat()`??
+                        f'|_pld={pld!r}\n'
-                        f'where, '
-                        f'{type(msg).__name__}.pld={pld!r}\n'
                     )
                     return pld
                 except TypeError as typerr:
@@ -500,8 +494,7 @@ def limit_plds(
 
     finally:
         log.runtime(
-            f'Reverted to previous payload-decoder\n'
+            'Reverted to previous payload-decoder\n\n'
-            f'\n'
             f'{orig_pldec}\n'
         )
         # sanity on orig settings
@@ -636,8 +629,7 @@ async def drain_to_final_msg(
                     (local_cs := rent_n.cancel_scope).cancel_called
                 ):
                     log.cancel(
-                        f'RPC-ctx cancelled by local-parent scope during drain!\n'
+                        'RPC-ctx cancelled by local-parent scope during drain!\n\n'
-                        f'\n'
                         f'c}}>\n'
                         f' |_{rent_n}\n'
                         f'   |_.cancel_scope = {local_cs}\n'
@@ -671,8 +663,7 @@ async def drain_to_final_msg(
             # final result arrived!
             case Return():
                 log.runtime(
-                    f'Context delivered final draining msg\n'
+                    'Context delivered final draining msg:\n'
-                    f'\n'
                     f'{pretty_struct.pformat(msg)}'
                 )
                 ctx._result: Any = pld
@@ -706,14 +697,12 @@ async def drain_to_final_msg(
                 ):
                     log.cancel(
                         'Cancelling `MsgStream` drain since '
-                        f'{reason}\n'
+                        f'{reason}\n\n'
-                        f'\n'
                         f'<= {ctx.chan.uid}\n'
-                        f'  |_{ctx._nsf}()\n'
+                        f'  |_{ctx._nsf}()\n\n'
-                        f'\n'
                         f'=> {ctx._task}\n'
-                        f'  |_{ctx._stream}\n'
+                        f'  |_{ctx._stream}\n\n'
-                        f'\n'
+
                         f'{pretty_struct.pformat(msg)}\n'
                     )
                     break
@@ -750,8 +739,7 @@ async def drain_to_final_msg(
             case Stop():
                 pre_result_drained.append(msg)
                 log.runtime(  # normal/expected shutdown transaction
-                    f'Remote stream terminated due to "stop" msg\n'
+                    'Remote stream terminated due to "stop" msg:\n\n'
-                    f'\n'
                     f'{pretty_struct.pformat(msg)}\n'
                 )
                 continue
@@ -826,8 +814,7 @@ async def drain_to_final_msg(
 
     else:
         log.cancel(
-            f'Skipping `MsgStream` drain since final outcome is set\n'
+            'Skipping `MsgStream` drain since final outcome is set\n\n'
-            f'\n'
             f'{ctx.outcome}\n'
         )
 

tractor/msg/types.py

@@ -154,39 +154,6 @@ class Aid(
     #     should also include at least `.pid` (equiv to port for tcp)
     #     and/or host-part always?
 
-    @property
-    def uid(self) -> tuple[str, str]:
-        '''
-        Legacy actor "unique-id" pair format.
-
-        '''
-        return (
-            self.name,
-            self.uuid,
-        )
-
-    def reprol(
-        self,
-        sin_uuid: bool = True,
-    ) -> str:
-        if not sin_uuid:
-            return (
-                f'{self.name}[{self.uuid[:6]}]@{self.pid!r}'
-            )
-        return (
-            f'{self.name}@{self.pid!r}'
-        )
-
-    # mk hashable via `.uuid`
-    def __hash__(self) -> int:
-        return hash(self.uuid)
-
-    def __eq__(self, other: Aid) -> bool:
-        return self.uuid == other.uuid
-
-    # use pretty fmt since often repr-ed for console/log
-    __repr__ = pretty_struct.Struct.__repr__
-
 
 class SpawnSpec(
     pretty_struct.Struct,

tractor/trionics/_beg.py

@@ -31,7 +31,7 @@ import trio
 
 def maybe_collapse_eg(
     beg: BaseExceptionGroup,
-) -> BaseException|bool:
+) -> BaseException:
     '''
     If the input beg can collapse to a single non-eg sub-exception,
     return it instead.
@@ -40,13 +40,11 @@ def maybe_collapse_eg(
     if len(excs := beg.exceptions) == 1:
         return excs[0]
 
-    return False
+    return beg
 
 
 @acm
-async def collapse_eg(
+async def collapse_eg():
-    hide_tb: bool = True,
-):
     '''
     If `BaseExceptionGroup` raised in the body scope is
     "collapse-able" (in the same way that
@@ -54,16 +52,12 @@ async def collapse_eg(
     only raise the lone emedded non-eg in in place.
 
     '''
-    __tracebackhide__: bool = hide_tb
     try:
         yield
     except* BaseException as beg:
         if (
             exc := maybe_collapse_eg(beg)
-        ):
+        ) is not beg:
-            if cause := exc.__cause__:
-                raise exc from cause
-
             raise exc
 
         raise beg

tractor/trionics/_broadcast.py

@@ -100,32 +100,6 @@ class Lagged(trio.TooSlowError):
     '''
 
 
-def wrap_rx_for_eoc(
-    rx: AsyncReceiver,
-) -> AsyncReceiver:
-
-    match rx:
-        case trio.MemoryReceiveChannel():
-
-            # XXX, taken verbatim from .receive_nowait()
-            def is_eoc() -> bool:
-                if not rx._state.open_send_channels:
-                    return trio.EndOfChannel
-
-                return False
-
-            rx.is_eoc = is_eoc
-
-        case _:
-            # XXX, ensure we define a private field!
-            # case tractor.MsgStream:
-            assert (
-                getattr(rx, '_eoc', False) is not None
-            )
-
-    return rx
-
-
 class BroadcastState(Struct):
     '''
     Common state to all receivers of a broadcast.
@@ -212,23 +186,11 @@ class BroadcastReceiver(ReceiveChannel):
         state.subs[self.key] = -1
 
         # underlying for this receiver
-        self._rx = wrap_rx_for_eoc(rx_chan)
+        self._rx = rx_chan
         self._recv = receive_afunc or rx_chan.receive
         self._closed: bool = False
         self._raise_on_lag = raise_on_lag
 
-    @property
-    def state(self) -> BroadcastState:
-        '''
-        Read-only access to this receivers internal `._state`
-        instance ref.
-
-        If you just want to read the high-level state metrics,
-        use `.state.statistics()`.
-
-        '''
-        return self._state
-
     def receive_nowait(
         self,
         _key: int | None = None,
@@ -253,23 +215,7 @@ class BroadcastReceiver(ReceiveChannel):
         try:
             seq = state.subs[key]
         except KeyError:
-            # from tractor import pause_from_sync
-            # pause_from_sync(shield=True)
-            if (
-                (rx_eoc := self._rx.is_eoc())
-                or
-                self.state.eoc
-            ):
-                raise trio.EndOfChannel(
-                    'Broadcast-Rx underlying already ended!'
-                ) from rx_eoc
-
             if self._closed:
-                # if (rx_eoc := self._rx._eoc):
-                #     raise trio.EndOfChannel(
-                #         'Broadcast-Rx underlying already ended!'
-                #     ) from rx_eoc
-
                 raise trio.ClosedResourceError
 
             raise RuntimeError(
@@ -507,9 +453,8 @@ class BroadcastReceiver(ReceiveChannel):
         self._closed = True
 
 
-# NOTE, this can we use as an `@acm` since `BroadcastReceiver`
-# derives from `ReceiveChannel`.
 def broadcast_receiver(
+
     recv_chan: AsyncReceiver,
     max_buffer_size: int,
     receive_afunc: Callable[[], Awaitable[Any]]|None = None,

tractor/trionics/_mngrs.py

@@ -40,7 +40,7 @@ from typing import (
 import trio
 from tractor._state import current_actor
 from tractor.log import get_logger
-from ._beg import collapse_eg
+# from ._beg import collapse_eg
 
 
 if TYPE_CHECKING:
@@ -152,8 +152,13 @@ async def gather_contexts(
         )
 
     async with (
-        collapse_eg(),
+        # collapse_eg(),
-        trio.open_nursery() as tn,
+        trio.open_nursery(
+            strict_exception_groups=False,
+            # ^XXX^ TODO? soo roll our own then ??
+            # -> since we kinda want the "if only one `.exception` then
+            # just raise that" interface?
+        ) as tn,
     ):
         for mngr in mngrs:
             tn.start_soon(
@@ -289,10 +294,7 @@ async def maybe_open_context(
         )
         _Cache.users += 1
         lock.release()
-        yield (
+        yield False, yielded
-            False,  # cache_hit = "no"
-            yielded,
-        )
 
     else:
         _Cache.users += 1
@@ -306,10 +308,7 @@ async def maybe_open_context(
             # f'{ctx_key!r} -> {yielded!r}\n'
         )
         lock.release()
-        yield (
+        yield True, yielded
-            True,  # cache_hit = "yes"
-            yielded,
-        )
 
     finally:
         _Cache.users -= 1

tractor/trionics/_taskc.py

@@ -60,8 +60,8 @@ def find_masked_excs(
     return None
 
 
-# XXX, relevant ish discussion @ `trio`-core,
+# XXX, relevant discussion @ `trio`-core,
-# https://github.com/python-trio/trio/issues/455#issuecomment-2785122216
+# https://github.com/python-trio/trio/issues/455
 #
 @acm
 async def maybe_raise_from_masking_exc(
@@ -113,7 +113,6 @@ async def maybe_raise_from_masking_exc(
     )
     matching: list[BaseException]|None = None
     maybe_eg: ExceptionGroup|None
-    maybe_eg: ExceptionGroup|None
 
     if tn:
         try:  # handle egs