Decouple actor-state from low-level ipc-server

As much as is possible given we currently do some graceful
cancellation join-waiting on any connected sub-actors whenever an active
`local_nursery: AcrtorNursery` in the post-rpc teardown sequence of
`handle_stream_from_peer()` is detected. In such cases we try to allow
the higher level inter-actor (task) context(s) to fully cancelled-ack
before conducting IPC machinery shutdown.

The main immediate motivation for all this is to support unit testing
the `.ipc._server` APIs but in the future may be useful for anyone
wanting to use our modular IPC transport layer sin-"actors".

Impl deats,
- drop passing an `actor: Actor` ref from as many routines in
  `.ipc._server` as possible instead opting to use
  `._state.current_actor()` where abs needed; thus the fns dropping an
  `actor` input param are:
  - `open_ipc_server()`
  - `IPCServer.listen_on()`
  - `._serve_ipc_eps()`
  - `.handle_stream_from_peer()`
- factor the above mentioned graceful remote-cancel-ack waiting into
  a new `maybe_wait_on_canced_subs()` which is called from
  `handle_stream_from_peer()` and delivers a
  maybe-`local_nursery: ActorNursery` for downstream logic; it's this
  new fn which primarily still needs to call `current_actor()`.
- in `handle_stream_from_peer()` also use `current_actor()` to check if
  a handshake is needed (or if it was called as part of some
  actor-runtime-less operation like our unit test suite!).
- also don't pass an `actor` to `._rpc.process_messages()` see how-n-why
  below..

Surrounding ipc-server client/caller adjustments,
- `._rpc.process_messages()` no longer takes an `actor` input and
  now calls `current_actor()` instead.
- `._portal.open_portal()` is adjusted to ^.
- `._runtime.async_main()` is adjusted to the `.ipc._server`'s removal
  of `actor` ref passing.

Also,
- drop some server `log.info()`s to `.runtime()`

Author: goodboy

tractor/_portal.py

@@ -582,8 +582,7 @@ async def open_portal(
             msg_loop_cs = await tn.start(
                 partial(
                     _rpc.process_messages,
-                    actor,
-                    channel,
+                    chan=channel,
                     # if the local task is cancelled we want to keep
                     # the msg loop running until our block ends
                     shield=True,

tractor/_rpc.py

@@ -869,7 +869,6 @@ async def try_ship_error_to_remote(
 
 
 async def process_messages(
-    actor: Actor,
     chan: Channel,
     shield: bool = False,
     task_status: TaskStatus[CancelScope] = trio.TASK_STATUS_IGNORED,
@@ -907,6 +906,7 @@ async def process_messages(
       (as utilized inside `Portal.cancel_actor()` ).
 
     '''
+    actor: Actor = _state.current_actor()
     assert actor._service_n  # runtime state sanity
 
     # TODO: once `trio` get's an "obvious way" for req/resp we

tractor/_runtime.py

@@ -1262,6 +1262,10 @@ async def async_main(
     the actor's "runtime" and all thus all ongoing RPC tasks.
 
     '''
+    # XXX NOTE, `_state._current_actor` **must** be set prior to
+    # calling this core runtime entrypoint!
+    assert actor is _state.current_actor()
+
     actor._task: trio.Task = trio.lowlevel.current_task()
 
     # attempt to retreive ``trio``'s sigint handler and stash it
@@ -1321,7 +1325,6 @@ async def async_main(
                 ) as service_nursery,
 
                 _server.open_ipc_server(
-                    actor=actor,
                     parent_tn=service_nursery,
                     stream_handler_tn=service_nursery,
                 ) as ipc_server,
@@ -1375,7 +1378,6 @@ async def async_main(
                         'Booting IPC server'
                     )
                     eps: list = await ipc_server.listen_on(
-                        actor=actor,
                         accept_addrs=accept_addrs,
                         stream_handler_nursery=service_nursery,
                     )
@@ -1460,8 +1462,7 @@ async def async_main(
                     await root_nursery.start(
                         partial(
                             _rpc.process_messages,
-                            actor,
-                            actor._parent_chan,
+                            chan=actor._parent_chan,
                             shield=True,
                         )
                     )