tractor/tests/test_inter_peer_cancellation.py

'''
Codify the cancellation request semantics in terms
of one remote actor cancelling another.

'''
# from contextlib import asynccontextmanager as acm
import itertools

import pytest
import trio
import tractor
from tractor import (  # typing
    Actor,
    current_actor,
    open_nursery,
    Portal,
    Context,
    ContextCancelled,
    RemoteActorError,
)
from tractor._testing import (
    # tractor_test,
    expect_ctxc,
)

# XXX TODO cases:
# - [ ] peer cancelled itself - so other peers should
#   get errors reflecting that the peer was itself the .canceller?

# - [x] WE cancelled the peer and thus should not see any raised
#   `ContextCancelled` as it should be reaped silently?
#   => pretty sure `test_context_stream_semantics::test_caller_cancels()`
#      already covers this case?

# - [x] INTER-PEER: some arbitrary remote peer cancels via
#   Portal.cancel_actor().
#   => all other connected peers should get that cancel requesting peer's
#      uid in the ctx-cancelled error msg raised in all open ctxs
#      with that peer.

# - [ ] PEER-FAILS-BY-CHILD-ERROR: peer spawned a sub-actor which
#   (also) spawned a failing task which was unhandled and
#   propagated up to the immediate parent - the peer to the actor
#   that also spawned a remote task task in that same peer-parent.


# def test_self_cancel():
#     '''
#     2 cases:
#     - calls `Actor.cancel()` locally in some task
#     - calls LocalPortal.cancel_actor()` ?

#     '''
#     ...


@tractor.context
async def open_stream_then_sleep_forever(
    ctx: Context,
    expect_ctxc: bool = False,

) -> None:
    '''
    Sync the context, open a stream then just sleep.

    Allow checking for (context) cancellation locally.

    '''
    try:
        await ctx.started()

        # NOTE: the below means this child will send a `Stop`
        # to it's parent-side task despite that side never
        # opening a stream itself.
        async with ctx.open_stream():
            await trio.sleep_forever()

    except BaseException as berr:

        # TODO: it'd sure be nice to be able to inject our own
        # `ContextCancelled` here instead of of `trio.Cancelled`
        # so that our runtime can expect it and this "user code"
        # would be able to tell the diff between a generic trio
        # cancel and a tractor runtime-IPC cancel.
        if expect_ctxc:
            assert isinstance(berr, trio.Cancelled)

        raise


@tractor.context
async def error_before_started(
    ctx: Context,
) -> None:
    '''
    This simulates exactly an original bug discovered in:
    https://github.com/pikers/piker/issues/244

    Cancel a context **before** any underlying error is raised so
    as to trigger a local reception of a ``ContextCancelled`` which
    SHOULD NOT be re-raised in the local surrounding ``Context``
    *iff* the cancel was requested by **this** (callee)  side of
    the context.

    '''
    async with tractor.wait_for_actor('sleeper') as p2:
        async with (
            p2.open_context(open_stream_then_sleep_forever) as (peer_ctx, first),
            peer_ctx.open_stream(),
        ):
            # NOTE: this WAS inside an @acm body but i factored it
            # out and just put it inline here since i don't think
            # the mngr part really matters, though maybe it could?
            try:
                # XXX NOTE XXX: THIS sends an UNSERIALIZABLE TYPE which
                # should raise a `TypeError` and **NOT BE SWALLOWED** by
                # the surrounding try/finally (normally inside the
                # body of some acm)..
                await ctx.started(object())
                # yield
            finally:
                # XXX: previously this would trigger local
                # ``ContextCancelled`` to be received and raised in the
                # local context overriding any local error due to logic
                # inside ``_invoke()`` which checked for an error set on
                # ``Context._error`` and raised it in a cancellation
                # scenario.
                # ------
                # The problem is you can have a remote cancellation that
                # is part of a local error and we shouldn't raise
                # ``ContextCancelled`` **iff** we **were not** the side
                # of the context to initiate it, i.e.
                # ``Context._cancel_called`` should **NOT** have been
                # set. The special logic to handle this case is now
                # inside ``Context._maybe_raise_from_remote_msg()`` XD
                await peer_ctx.cancel()


def test_do_not_swallow_error_before_started_by_remote_contextcancelled(
    debug_mode: bool,
):
    '''
    Verify that an error raised in a remote context which itself
    opens YET ANOTHER remote context, which it then cancels, does not
    override the original error that caused the cancellation of the
    secondary context.

    '''
    async def main():
        async with tractor.open_nursery(
            debug_mode=debug_mode,
        ) as n:
            portal = await n.start_actor(
                'errorer',
                enable_modules=[__name__],
            )
            await n.start_actor(
                'sleeper',
                enable_modules=[__name__],
            )

            async with (
                portal.open_context(
                    error_before_started
                ) as (ctx, sent),
            ):
                await trio.sleep_forever()

    with pytest.raises(RemoteActorError) as excinfo:
        trio.run(main)

    rae = excinfo.value
    assert rae.boxed_type == TypeError


@tractor.context
async def sleep_a_bit_then_cancel_peer(
    ctx: Context,
    peer_name: str = 'sleeper',
    cancel_after: float = .5,

) -> None:
    '''
    Connect to peer, sleep as per input delay, cancel the peer.

    '''
    peer: Portal
    async with tractor.wait_for_actor(peer_name) as peer:
        await ctx.started()
        await trio.sleep(cancel_after)
        await peer.cancel_actor()

        # such that we're cancelled by our rent ctx-task
        await trio.sleep(3)
        print('CANCELLER RETURNING!')


@tractor.context
async def stream_ints(
    ctx: Context,
):
    await ctx.started()
    async with ctx.open_stream() as stream:
        for i in itertools.count():
            await stream.send(i)
            await trio.sleep(0.01)


@tractor.context
async def stream_from_peer(
    ctx: Context,
    debug_mode: bool,
    peer_name: str = 'sleeper',
) -> None:

    # sanity
    assert tractor._state.debug_mode() == debug_mode

    peer: Portal
    try:
        async with (
            tractor.wait_for_actor(peer_name) as peer,
            peer.open_context(stream_ints) as (peer_ctx, first),
            peer_ctx.open_stream() as stream,
        ):
            await ctx.started()
            # XXX QUESTIONS & TODO: for further details around this
            # in the longer run..
            # https://github.com/goodboy/tractor/issues/368
            # - should we raise `ContextCancelled` or `Cancelled` (rn
            #   it does latter) and should/could it be implemented
            #   as a general injection override for `trio` such
            #   that ANY next checkpoint would raise the "cancel
            #   error type" of choice?
            # - should the `ContextCancelled` bubble from
            #   all `Context` and `MsgStream` apis wherein it
            #   prolly makes the most sense to make it
            #   a `trio.Cancelled` subtype?
            # - what about IPC-transport specific errors, should
            #   they bubble from the async for and trigger
            #   other special cases?
            #
            # NOTE: current ctl flow:
            # - stream raises `trio.EndOfChannel` and
            #   exits the loop
            # - `.open_context()` will raise the ctxc received
            #   from the sleeper.
            async for msg in stream:
                assert msg is not None
                print(msg)

    # NOTE: cancellation of the (sleeper) peer should always cause
    # a `ContextCancelled` raise in this streaming actor.
    except ContextCancelled as _ctxc:
        ctxc = _ctxc

        # print("TRYING TO ENTER PAUSSE!!!")
        # await tractor.pause(shield=True)
        re: ContextCancelled = peer_ctx._remote_error

        # XXX YES XXX, remote error should be unpacked only once!
        assert (
            re
            is
            peer_ctx.maybe_error
            is
            ctxc
            is
            peer_ctx._local_error
        )
        # NOTE: these errors should all match!
        #   ------ - ------
        # XXX [2024-05-03] XXX
        #   ------ - ------
        # broke this due to a re-raise inside `.msg._ops.drain_to_final_msg()`
        # where the `Error()` msg was directly raising the ctxc
        # instead of just returning up to the caller inside
        # `Context.return()` which would results in a diff instance of
        # the same remote error bubbling out above vs what was
        # already unpacked and set inside `Context.
        assert (
            peer_ctx._remote_error.msgdata
            ==
            ctxc.msgdata
        )
        # ^-XXX-^ notice the data is of course the exact same.. so
        # the above larger assert makes sense to also always be true!

        # XXX YES XXX, bc should be exact same msg instances
        assert peer_ctx._remote_error._ipc_msg is ctxc._ipc_msg

        # XXX NO XXX, bc new one always created for property accesss
        assert peer_ctx._remote_error.ipc_msg != ctxc.ipc_msg

        # the peer ctx is the canceller even though it's canceller
        # is the "canceller" XD
        assert peer_name in peer_ctx.canceller

        assert "canceller" in ctxc.canceller

        # caller peer should not be the cancel requester
        assert not ctx.cancel_called
        assert not ctx.cancel_acked

        # XXX can NEVER BE TRUE since `._invoke` only
        # sets this AFTER the nursery block this task
        # was started in, exits.
        assert not ctx._scope.cancelled_caught

        # we never requested cancellation, it was the 'canceller'
        # peer.
        assert not peer_ctx.cancel_called
        assert not peer_ctx.cancel_acked

        # the `.open_context()` exit definitely caught
        # a cancellation in the internal `Context._scope` since
        # likely the runtime called `_deliver_msg()` after
        # receiving the remote error from the streaming task.
        assert not peer_ctx._scope.cancelled_caught

        # TODO / NOTE `.canceller` won't have been set yet
        # here because that machinery is inside
        # `Portal.open_context().__aexit__()` BUT, if we had
        # a way to know immediately (from the last
        # checkpoint) that cancellation was due to
        # a remote, we COULD assert this here..see,
        # https://github.com/goodboy/tractor/issues/368
        #
        # await tractor.pause()
        # assert 'canceller' in ctx.canceller

        # root/parent actor task should NEVER HAVE cancelled us!
        assert not ctx.canceller

        raise
        # TODO: IN THEORY we could have other cases depending on
        # who cancels first, the root actor or the canceller peer?.
        #
        # 1- when the peer request is first then the `.canceller`
        #   field should obvi be set to the 'canceller' uid,
        #
        # 2-if the root DOES req cancel then we should see the same
        #   `trio.Cancelled` implicitly raised
        # assert ctx.canceller[0] == 'root'
        # assert peer_ctx.canceller[0] == 'sleeper'

    raise RuntimeError('Never triggered local `ContextCancelled` ?!?')


@pytest.mark.parametrize(
    'error_during_ctxerr_handling',
    [False, True],
    ids=lambda item: f'rte_during_ctxerr={item}',
)
def test_peer_canceller(
    error_during_ctxerr_handling: bool,
    debug_mode: bool,
):
    '''
    Verify that a cancellation triggered by an in-actor-tree peer
    results in a cancelled errors with all other actors which have
    opened contexts to that same actor.

    legend:
    name>
        a "play button" that indicates a new runtime instance,
        an individual actor with `name`.

    .subname>
        a subactor who's parent should be on some previous
        line and be less indented.

    .actor0> ()-> .actor1>
        a inter-actor task context opened (by `async with
        `Portal.open_context()`) from actor0 *into* actor1.

    .actor0> ()<=> .actor1>
        a inter-actor task context opened (as above)
        from actor0 *into* actor1 which INCLUDES an additional
        stream open using `async with Context.open_stream()`.


    ------ - ------
    supervision view
    ------ - ------
    root>
     .sleeper> TODO: SOME SYNTAX SHOWING JUST SLEEPING
     .just_caller> ()=> .sleeper>
     .canceller> ()-> .sleeper>
                  TODO:  how define calling `Portal.cancel_actor()`

    In this case a `ContextCancelled` with `.errorer` set to the
    requesting actor, in this case 'canceller', should be relayed
    to all other actors who have also opened a (remote task)
    context with that now cancelled actor.

    ------ - ------
    task view
    ------ - ------
    So there are 5 context open in total with 3 from the root to
    its children and 2 from children to their peers:
    1. root> ()-> .sleeper>
    2. root> ()-> .streamer>
    3. root> ()-> .canceller>

    4. .streamer> ()<=> .sleep>
    5. .canceller> ()-> .sleeper>
        - calls `Portal.cancel_actor()`

    '''
    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(
                'canceller',
                enable_modules=[__name__],
            )
            sleeper: Portal = await an.start_actor(
                'sleeper',
                enable_modules=[__name__],
            )
            just_caller: Portal = await an.start_actor(
                'just_caller',  # but i just met her?
                enable_modules=[__name__],
            )
            root: Actor = current_actor()

            try:
                async with (
                    sleeper.open_context(
                        open_stream_then_sleep_forever,
                        expect_ctxc=True,
                    ) as (sleeper_ctx, sent),

                    just_caller.open_context(
                        stream_from_peer,
                        debug_mode=debug_mode,
                    ) as (caller_ctx, sent),

                    canceller.open_context(
                        sleep_a_bit_then_cancel_peer,
                    ) as (canceller_ctx, sent),

                ):
                    ctxs: dict[str, Context] = {
                        'sleeper': sleeper_ctx,
                        'caller': caller_ctx,
                        'canceller': canceller_ctx,
                    }

                    try:
                        print('PRE CONTEXT RESULT')
                        res = await sleeper_ctx.result()
                        assert res

                        # should never get here
                        pytest.fail(
                            'Context.result() did not raise ctx-cancelled?'
                        )

                    # should always raise since this root task does
                    # not request the sleeper cancellation ;)
                    except ContextCancelled as _ctxc:
                        ctxc = _ctxc
                        print(
                            'CAUGHT REMOTE CONTEXT CANCEL\n\n'
                            f'{ctxc}\n'
                        )

                        # canceller and caller peers should not
                        # have been remotely cancelled.
                        assert canceller_ctx.canceller is None
                        assert caller_ctx.canceller is None

                        # we were not the actor, our peer was
                        assert not sleeper_ctx.cancel_acked

                        assert ctxc.canceller[0] == 'canceller'

                        # XXX NOTE XXX: since THIS `ContextCancelled`
                        # HAS NOT YET bubbled up to the
                        # `sleeper.open_context().__aexit__()` this
                        # value is not yet set, however outside this
                        # block it should be.
                        assert not sleeper_ctx._scope.cancelled_caught

                        # CASE_1: error-during-ctxc-handling,
                        if error_during_ctxerr_handling:
                            raise RuntimeError('Simulated RTE re-raise during ctxc handling')

                        # CASE_2: standard teardown inside in `.open_context()` block
                        raise

                    # XXX SHOULD NEVER EVER GET HERE XXX
                    except BaseException as berr:
                        raise

                        # XXX if needed to debug failure
                        # _err = berr
                        # await tractor.pause()
                        # await trio.sleep_forever()

                        pytest.fail(
                            'did not rx ctxc ?!?\n\n'

                            f'{berr}\n'
                        )

                    else:
                        pytest.fail(
                            'did not rx ctxc ?!?\n\n'
                            f'{ctxs}\n'
                        )

            except (
                ContextCancelled,
                RuntimeError,
            )as loc_err:
                _loc_err = loc_err

                # NOTE: the main state to check on `Context` is:
                # - `.cancel_called` (bool of whether this side
                #    requested)
                # - `.cancel_acked` (bool of whether a ctxc
                #   response was received due to cancel req).
                # - `.maybe_error` (highest prio error to raise
                #    locally)
                # - `.outcome` (final error or result value)
                # - `.canceller` (uid of cancel-causing actor-task)
                # - `._remote_error` (any `RemoteActorError`
                #    instance from other side of context)
                # - `._local_error` (any error caught inside the
                #   `.open_context()` block).
                #
                # XXX: Deprecated and internal only
                # - `.cancelled_caught` (maps to nursery cs)
                #  - now just use `._scope.cancelled_caught`
                #    since it maps to the internal (maps to nursery cs)
                #
                # TODO: are we really planning to use this tho?
                # - `._cancel_msg` (any msg that caused the
                #    cancel)

                # CASE_1: error-during-ctxc-handling,
                # - far end cancels due to peer 'canceller',
                # - `ContextCancelled` relayed to this scope,
                # - inside `.open_context()` ctxc is caught and
                #   a rte raised instead
                #
                # => block should raise the rte but all peers
                #   should be cancelled by US.
                #
                if error_during_ctxerr_handling:
                    print(f'loc_err: {_loc_err}\n')
                    assert isinstance(loc_err, RuntimeError)

                    # since we do a rte reraise above, the
                    # `.open_context()` error handling should have
                    # raised a local rte, thus the internal
                    # `.open_context()` enterer task's
                    # cancel-scope should have raised the RTE, NOT
                    # a `trio.Cancelled` due to a local
                    # `._scope.cancel()` call.
                    assert not sleeper_ctx._scope.cancelled_caught
                    # assert sleeper_ctx._local_error is _loc_err
                    # assert sleeper_ctx._local_error is _loc_err
                    assert not (
                        loc_err
                        is sleeper_ctx.maybe_error
                        is sleeper_ctx.outcome
                        is sleeper_ctx._remote_error
                    )

                    # NOTE: this root actor task should have
                    # called `Context.cancel()` on the
                    # `.__aexit__()` to every opened ctx.
                    for name, ctx in ctxs.items():

                        # this root actor task should have
                        # cancelled all opened contexts except the
                        # sleeper which is obvi by the "canceller"
                        # peer.
                        re = ctx._remote_error
                        le = ctx._local_error

                        assert ctx.cancel_called

                        if ctx is sleeper_ctx:
                            assert 'canceller' in re.canceller
                            assert 'sleeper' in ctx.canceller

                        if ctx is canceller_ctx:
                            assert (
                                re.canceller
                                ==
                                root.uid
                            )

                        else:  # the other 2 ctxs
                            assert (
                                isinstance(re, ContextCancelled)
                                and (
                                    re.canceller
                                    ==
                                    canceller.channel.uid
                                )
                            )

                    # since the sleeper errors while handling a
                    # peer-cancelled (by ctxc) scenario, we expect
                    # that the `.open_context()` block DOES call
                    # `.cancel() (despite in this test case it
                    # being unecessary).
                    assert (
                        sleeper_ctx.cancel_called
                        and
                        not sleeper_ctx.cancel_acked
                    )

                # CASE_2: standard teardown inside in `.open_context()` block
                # - far end cancels due to peer 'canceller',
                # - `ContextCancelled` relayed to this scope and
                #   raised locally without any raise-during-handle,
                #
                # => inside `.open_context()` ctxc is raised and
                #   propagated
                #
                else:
                    # since sleeper_ctx.result() IS called above
                    # we should have (silently) absorbed the
                    # corresponding `ContextCancelled` for it and
                    # `._scope.cancel()` should never have been
                    # called.
                    assert not sleeper_ctx._scope.cancelled_caught

                    assert isinstance(loc_err, ContextCancelled)

                    # the received remote error's `.canceller`
                    # will of course be the "canceller" actor BUT
                    # the canceller set on the local handle to
                    # `sleeper_ctx` will be the "sleeper" uid
                    # since it's the actor that relayed us the
                    # error which was **caused** by the
                    # "canceller".
                    assert 'sleeper' in sleeper_ctx.canceller
                    assert 'canceller' == loc_err.canceller[0]

                    # the sleeper's remote error is the error bubbled
                    # out of the context-stack above!
                    final_err = sleeper_ctx.outcome
                    assert (
                        final_err is loc_err
                        is sleeper_ctx.maybe_error
                        is sleeper_ctx._remote_error
                    )

                    for name, ctx in ctxs.items():

                        re: BaseException|None = ctx._remote_error
                        le: BaseException|None = ctx._local_error
                        err = ctx.maybe_error
                        out = ctx.outcome

                        # every ctx should error!
                        assert out is err

                        # the recorded local erro should always be
                        # the same as the one raised by the
                        # `sleeper_ctx.result()` call
                        assert (
                            le
                            and
                            le is loc_err
                        )

                        # root doesn't cancel sleeper since it's
                        # cancelled by its peer.
                        if ctx is sleeper_ctx:
                            assert re
                            assert (
                                ctx._remote_error
                                is ctx.maybe_error
                                is ctx.outcome
                                is ctx._local_error
                            )

                            assert not ctx.cancel_called
                            assert not ctx.cancel_acked

                            # since sleeper_ctx.result() IS called
                            # above we should have (silently)
                            # absorbed the corresponding
                            # `ContextCancelled` for it and thus
                            # the logic inside `.cancelled_caught`
                            # should trigger!
                            assert not ctx._scope.cancelled_caught

                        elif ctx in (
                            caller_ctx,
                            canceller_ctx,
                        ):

                            assert not ctx._remote_error

                            # neither of the `caller/canceller_ctx` should
                            # have called `ctx.cancel()` bc the
                            # canceller's task internally issues
                            # a `Portal.cancel_actor()` to the
                            # sleeper and thus never should call
                            # `ctx.cancel()` per say UNLESS the
                            # sleeper's `.result()` call above
                            # ctxc exception results in the
                            # canceller's
                            # `.open_context().__aexit__()` error
                            # handling to kick in BEFORE a remote
                            # error is delivered - which since
                            # we're asserting what we are above,
                            # that should normally be the case
                            # right?
                            #
                            assert not ctx.cancel_called
                            #
                            # assert ctx.cancel_called
                            # orig ^

                            # TODO: figure out the details of this..?
                            # if you look the `._local_error` here
                            # is a multi of ctxc + 2 Cancelleds?
                            # assert not ctx._scope.cancelled_caught

                            assert (
                                not ctx.cancel_called
                                and not ctx.cancel_acked
                            )
                            assert not ctx._scope.cancelled_caught

                        # elif ctx is canceller_ctx:
                        #     assert not ctx._remote_error

                            # XXX NOTE XXX: ONLY the canceller
                            # will get a self-cancelled outcome
                            # whilst everyone else gets
                            # a peer-caused cancellation!
                            #
                            # TODO: really we should avoid calling
                            # .cancel() whenever an interpeer
                            # cancel takes place since each
                            # reception of a ctxc

                        else:
                            pytest.fail(
                                'Uhh wut ctx is this?\n'
                                f'{ctx}\n'
                            )

                        # TODO: do we even need this flag?
                        # -> each context should have received
                        # a silently absorbed context cancellation
                        # in its remote nursery scope.
                        # assert ctx.chan.uid == ctx.canceller

                    # NOTE: when an inter-peer cancellation
                    # occurred, we DO NOT expect this
                    # root-actor-task to have requested a cancel of
                    # the context since cancellation was caused by
                    # the "canceller" peer and thus
                    # `Context.cancel()` SHOULD NOT have been
                    # called inside
                    # `Portal.open_context().__aexit__()`.
                    assert not (
                        sleeper_ctx.cancel_called
                        or
                        sleeper_ctx.cancel_acked
                    )

                # XXX NOTE XXX: and see matching comment above but,
                # the `._scope` is only set by `trio` AFTER the
                # `.open_context()` block has exited and should be
                # set in both outcomes including the case where
                # ctx-cancel handling itself errors.
                assert not sleeper_ctx._scope.cancelled_caught
                assert _loc_err is sleeper_ctx._local_error
                assert (
                    sleeper_ctx.outcome
                    is sleeper_ctx.maybe_error
                    is sleeper_ctx._remote_error
                )

                raise  # always to ensure teardown

    if error_during_ctxerr_handling:
        with pytest.raises(RuntimeError) as excinfo:
            trio.run(main)
    else:

        with pytest.raises(ContextCancelled) as excinfo:
            trio.run(main)

        assert excinfo.value.boxed_type == ContextCancelled
        assert excinfo.value.canceller[0] == 'canceller'


@tractor.context
async def basic_echo_server(
    ctx: Context,
    peer_name: str = 'wittle_bruv',

    err_after: int|None = None,

) -> None:
    '''
    Just the simplest `MsgStream` echo server which resays what
    you told it but with its uid in front ;)

    '''
    actor: Actor = tractor.current_actor()
    uid: tuple = actor.uid
    await ctx.started(uid)
    async with ctx.open_stream() as ipc:
        async for msg in ipc:

            # repack msg pair with our uid
            # as first element.
            (
                client_uid,
                i,
            ) = msg
            resp: tuple = (
                uid,
                i,
            )
            # OOF! looks like my runtime-error is causing a lockup
            # assert 0
            await ipc.send(resp)

            if (
                err_after
                and i > err_after
            ):
                raise RuntimeError(
                    f'Simulated error in `{peer_name}`'
                )


@tractor.context
async def serve_subactors(
    ctx: Context,
    peer_name: str,
    debug_mode: bool,

) -> None:
    async with open_nursery() as an:

        # sanity
        assert tractor._state.debug_mode() == debug_mode

        await ctx.started(peer_name)
        async with ctx.open_stream() as ipc:
            async for msg in ipc:
                peer_name: str = msg
                peer: Portal = await an.start_actor(
                    name=peer_name,
                    enable_modules=[__name__],
                )
                print(
                    'Spawning new subactor\n'
                    f'{peer_name}\n'
                    f'|_{peer}\n'
                )
                await ipc.send((
                    peer.chan.uid,
                    peer.chan.raddr,
                ))

        print('Spawner exiting spawn serve loop!')


@tractor.context
async def client_req_subactor(
    ctx: Context,
    peer_name: str,
    debug_mode: bool,

    # used to simulate a user causing an error to be raised
    # directly in thread (like a KBI) to better replicate the
    # case where a `modden` CLI client would hang afer requesting
    # a `Context.cancel()` to `bigd`'s wks spawner.
    reraise_on_cancel: str|None = None,
    sub_err_after: int|None = None,

) -> None:
    # sanity
    if debug_mode:
        assert tractor._state.debug_mode()

    # TODO: other cases to do with sub lifetimes:
    # -[ ] test that we can have the server spawn a sub
    #   that lives longer then ctx with this client.
    # -[ ] test that

    # open ctx with peer spawn server and ask it to spawn a little
    # bro which we'll then connect and stream with.
    async with (
        tractor.find_actor(
            name='spawn_server',
            raise_on_none=True,

            # TODO: we should be isolating this from other runs!
            # => ideally so we can eventually use something like
            # `pytest-xdist` Bo
            # registry_addrs=bigd._reg_addrs,
        ) as spawner,

        spawner.open_context(
            serve_subactors,
            peer_name=peer_name,
            debug_mode=debug_mode,
        ) as (spawner_ctx, first),
    ):
        assert first == peer_name
        await ctx.started(
            'yup i had brudder',
        )

        async with spawner_ctx.open_stream() as reqs:

            # send single spawn request to the server
            await reqs.send(peer_name)
            with trio.fail_after(3):
                (
                    sub_uid,
                    sub_raddr,
                ) = await reqs.receive()


            await tell_little_bro(
                actor_name=sub_uid[0],
                caller='client',
                err_after=sub_err_after,
            )

            # TODO: test different scope-layers of
            # cancellation?
            # with trio.CancelScope() as cs:
            try:
                await trio.sleep_forever()

            # TODO: would be super nice to have a special injected
            # cancel type here (maybe just our ctxc) but using
            # some native mechanism in `trio` :p
            except trio.Cancelled as err:
                _err = err
                if reraise_on_cancel:
                    errtype = globals()['__builtins__'][reraise_on_cancel]
                    assert errtype
                    to_reraise: BaseException = errtype()
                    print(f'client re-raising on cancel: {repr(to_reraise)}')
                    raise err

                raise

            # if cs.cancelled_caught:
            #     print('client handling expected KBI!')
            #     await ctx.
            #     await trio.sleep(
            #     await tractor.pause()
            #     await spawner_ctx.cancel()

            # cancel spawned sub-actor directly?
            # await sub_ctx.cancel()

            # maybe cancel runtime?
            # await sub.cancel_actor()


async def tell_little_bro(
    actor_name: str,

    caller: str = '',
    err_after: int|None = None,
):
    # contact target actor, do a stream dialog.
    async with (
        tractor.wait_for_actor(
            name=actor_name
        ) as lb,
        lb.open_context(
            basic_echo_server,

            # XXX proxy any delayed err condition
            err_after=err_after,
        ) as (sub_ctx, first),

        sub_ctx.open_stream() as echo_ipc,
    ):
        actor: Actor = current_actor()
        uid: tuple = actor.uid
        for i in range(100):
            msg: tuple = (
                uid,
                i,
            )
            await echo_ipc.send(msg)
            resp = await echo_ipc.receive()
            print(
                f'{caller} => {actor_name}: {msg}\n'
                f'{caller} <= {actor_name}: {resp}\n'
            )
            (
                sub_uid,
                _i,
            ) = resp
            assert sub_uid != uid
            assert _i == i


@pytest.mark.parametrize(
    'raise_client_error',
    [None, 'KeyboardInterrupt'],
)
@pytest.mark.parametrize(
    'raise_sub_spawn_error_after',
    [None, 50],
)
def test_peer_spawns_and_cancels_service_subactor(
    debug_mode: bool,
    raise_client_error: str,
    reg_addr: tuple[str, int],
    raise_sub_spawn_error_after: int|None,
):
    # NOTE: this tests for the modden `mod wks open piker` bug
    # discovered as part of implementing workspace ctx
    # open-.pause()-ctx.cancel() as part of the CLI..

    # -> start actor-tree (server) that offers sub-actor spawns via
    #   context API
    # -> start another full actor-tree (client) which requests to the first to
    #   spawn over its `@context` ep / api.
    # -> client actor cancels the context and should exit gracefully
    #   and the server's spawned child should cancel and terminate!
    peer_name: str = 'little_bro'

    def check_inner_rte(rae: RemoteActorError):
        '''
        Validate the little_bro's relayed inception!

        '''
        assert rae.boxed_type is RemoteActorError
        assert rae.src_type is RuntimeError
        assert 'client' in rae.relay_uid
        assert peer_name in rae.src_uid

    async def main():
        async with tractor.open_nursery(
            # NOTE: to halt the peer tasks on ctxc, uncomment this.
            debug_mode=debug_mode,
            registry_addrs=[reg_addr],
        ) as an:
            server: Portal = await an.start_actor(
                (server_name := 'spawn_server'),
                enable_modules=[__name__],
            )
            print(f'Spawned `{server_name}`')

            client: Portal = await an.start_actor(
                client_name := 'client',
                enable_modules=[__name__],
            )
            print(f'Spawned `{client_name}`')

            try:
                async with (
                    server.open_context(
                        serve_subactors,
                        peer_name=peer_name,
                        debug_mode=debug_mode,

                    ) as (spawn_ctx, first),

                    client.open_context(
                        client_req_subactor,
                        peer_name=peer_name,
                        debug_mode=debug_mode,
                        reraise_on_cancel=raise_client_error,

                        # trigger for error condition in sub
                        # during streaming.
                        sub_err_after=raise_sub_spawn_error_after,

                    ) as (client_ctx, client_says),
                ):
                    root: Actor = current_actor()
                    spawner_uid: tuple = spawn_ctx.chan.uid
                    print(
                        f'Server says: {first}\n'
                        f'Client says: {client_says}\n'
                    )

                    # attach to client-requested-to-spawn
                    # (grandchild of this root actor) "little_bro"
                    # and ensure we can also use it as an echo
                    # server.
                    sub: Portal
                    async with tractor.wait_for_actor(
                        name=peer_name,
                    ) as sub:
                        assert sub

                    print(
                        'Sub-spawn came online\n'
                        f'portal: {sub}\n'
                        f'.uid: {sub.actor.uid}\n'
                        f'chan.raddr: {sub.chan.raddr}\n'
                    )

                    async with expect_ctxc(
                        yay=raise_sub_spawn_error_after,
                        reraise=False,
                    ):
                        await tell_little_bro(
                            actor_name=peer_name,
                            caller='root',
                        )

                    if not raise_sub_spawn_error_after:

                        # signal client to cancel and maybe raise a KBI
                        await client_ctx.cancel()
                        print(
                            '-> root cancelling client,\n'
                            '-> root checking `client_ctx.result()`,\n'
                            f'-> checking that sub-spawn {peer_name} is down\n'
                        )

                    try:
                        res = await client_ctx.result(hide_tb=False)

                        # in remote (relayed inception) error
                        # case, we should error on the line above!
                        if raise_sub_spawn_error_after:
                            pytest.fail(
                                'Never rxed proxied `RemoteActorError[RuntimeError]` !?'
                            )

                        assert isinstance(res, ContextCancelled)
                        assert client_ctx.cancel_acked
                        assert res.canceller == root.uid

                    except RemoteActorError as rae:
                        _err = rae
                        assert raise_sub_spawn_error_after

                        # since this is a "relayed error" via the client
                        # sub-actor, it is expected to be
                        # a `RemoteActorError` boxing another
                        # `RemoteActorError` otherwise known as
                        #  an "inception" (from `trio`'s parlance)
                        # ((or maybe a "Matryoshka" and/or "matron"
                        # in our own working parlance)) which
                        # contains the source error from the
                        # little_bro: a `RuntimeError`.
                        #
                        check_inner_rte(rae)
                        assert rae.relay_uid == client.chan.uid
                        assert rae.src_uid == sub.chan.uid

                        assert not client_ctx.cancel_acked
                        assert (
                            client_ctx.maybe_error
                            is client_ctx.outcome
                            is rae
                        )
                        raise
                        # await tractor.pause()

                    else:
                        assert not raise_sub_spawn_error_after

                        # cancelling the spawner sub should
                        # transitively cancel it's sub, the little
                        # bruv.
                        print('root cancelling server/client sub-actors')
                        await spawn_ctx.cancel()
                        async with tractor.find_actor(
                            name=peer_name,
                        ) as sub:
                            assert not sub

                    # await server.cancel_actor()

            except RemoteActorError as rae:
                # XXX more-or-less same as above handler
                # this is just making sure the error bubbles out
                # of the 
                _err = rae
                assert raise_sub_spawn_error_after
                raise

            # since we called `.cancel_actor()`, `.cancel_ack`
            # will not be set on the ctx bc `ctx.cancel()` was not
            # called directly fot this confext.
            except ContextCancelled as ctxc:
                _ctxc = ctxc
                print(
                    f'{root.uid} caught ctxc from ctx with {client_ctx.chan.uid}\n'
                    f'{repr(ctxc)}\n'
                )

                if not raise_sub_spawn_error_after:
                    assert ctxc.canceller == root.uid
                else:
                    assert ctxc.canceller == spawner_uid

                assert ctxc is spawn_ctx.outcome
                assert ctxc is spawn_ctx.maybe_error
                raise

            if raise_sub_spawn_error_after:
                pytest.fail(
                    'context block(s) in PARENT never raised?!?'
                )

            if not raise_sub_spawn_error_after:
                # assert spawn_ctx.cancel_acked
                assert spawn_ctx.cancel_acked
                assert client_ctx.cancel_acked

                await client.cancel_actor()
                await server.cancel_actor()

                # WOA WOA WOA! we need this to close..!!!??
                # that's super bad XD

                # TODO: why isn't this working!?!?
                # we're now outside the `.open_context()` block so
                # the internal `Context._scope: CancelScope` should be
                # gracefully "closed" ;)

                # assert spawn_ctx.cancelled_caught

    if raise_sub_spawn_error_after:
        with pytest.raises(RemoteActorError) as excinfo:
            trio.run(main)

        rae: RemoteActorError = excinfo.value
        check_inner_rte(rae)

    else:
        trio.run(main)