Add a ignore-masking-case script + suite

Demonstrating the guilty `trio.Lock.acquire()` impl which puts
a checkpoint inside its `trio.WouldBlock` handler and which will always
appear to mask the "sync path" case on (graceful) cancellation.

This first script draft demos the issue from within a `tractor.context`
ep bc that's where it was orig discovered, however i'm going to factor
out the `tractor` code and instead just use
a `.trionics.maybe_raise_from_masking_exc()` to demo its low-level
ignore-case feature.

Further, this script exposed a previously unhandled remote graceful
cancellation case which hangs:

- parent actor spawns child and opens a >1 ctxs with it,
- the parent then OoB (out-of-band) cancels the child actor (with
  `Portal.cancel_actor()`),
- since the open ctxs raise a ctxc with a `.canceller == parent.uid` the
  `Context._is_self_cancelled()` will eval `True`,
- the `Context._scope` will NOT be cancelled in
  `._maybe_cancel_and_set_remote_error()` resulting in any bg-task which
  is waiting on a `Portal.open_context()` to not be cancelled/unblocked.

So my plan is to factor this ^^ scenario into a standalone unit test
as well as another test which consumes from al low-level `trio`-only
version of **this** script-scenario to sanity check the interaction
of the unmasker-with-ignore-cases usage implicitly around a ctx ep.

examples/trio/lockacquire_not_unmasked.py

@@ -0,0 +1,97 @@
+from functools import partial
+
+import tractor
+import trio
+
+
+log = tractor.log.get_logger(
+    name=__name__
+)
+
+
+async def acquire_singleton_lock(
+    _lock = trio.Lock(),
+) -> None:
+    log.info('TRYING TO LOCK ACQUIRE')
+    await _lock.acquire()
+    log.info('ACQUIRED')
+
+
+@tractor.context
+async def acquire_actor_global_lock(
+    ctx: tractor.Context,
+
+    ignore_special_cases: bool,
+):
+    if not ignore_special_cases:
+        from tractor.trionics import _taskc
+        _taskc._mask_cases.clear()
+
+    await acquire_singleton_lock()
+    await ctx.started('locked')
+
+    # block til cancelled
+    await trio.sleep_forever()
+
+
+async def main(
+    ignore_special_cases: bool,
+    loglevel: str = 'info',
+    debug_mode: bool = True,
+
+    _fail_after: float = 2,
+):
+    tractor.log.get_console_log(level=loglevel)
+
+    with trio.fail_after(_fail_after):
+        async with (
+            tractor.trionics.collapse_eg(),
+            tractor.open_nursery(
+                debug_mode=debug_mode,
+                loglevel=loglevel,
+            ) as an,
+            trio.open_nursery() as tn,
+        ):
+            ptl = await an.start_actor(
+                'locker',
+                enable_modules=[__name__],
+            )
+
+            async def _open_ctx(
+                task_status=trio.TASK_STATUS_IGNORED,
+            ):
+                async with ptl.open_context(
+                    acquire_actor_global_lock,
+                    ignore_special_cases=ignore_special_cases,
+                ) as pair:
+                    task_status.started(pair)
+                    await trio.sleep_forever()
+
+            first_ctx, first = await tn.start(_open_ctx,)
+            assert first == 'locked'
+
+            with trio.move_on_after(0.5):# as cs:
+                await _open_ctx()
+
+            # await tractor.pause()
+            print('cancelling first IPC ctx!')
+            await first_ctx.cancel()
+
+            await ptl.cancel_actor()
+            # await tractor.pause()
+
+
+# XXX, manual test as script
+if __name__ == '__main__':
+    tractor.log.get_console_log(level='info')
+    for case in [False, True]:
+        log.info(
+            f'\n'
+            f'------ RUNNING SCRIPT TRIAL ------\n'
+            f'child_errors_midstream: {case!r}\n'
+        )
+        trio.run(partial(
+            main,
+            ignore_special_cases=case,
+            loglevel='info',
+        ))

tests/test_trioisms.py

@@ -265,3 +265,38 @@ def test_unmask_aclose_as_checkpoint_on_aexit(
             raise_unmasked=raise_unmasked,
             child_errors_mid_stream=child_errors_mid_stream,
         ))
+
+
+
+@pytest.mark.parametrize(
+    'ignore_special_cases', [
+        True,
+        pytest.param(
+            False,
+            marks=pytest.mark.xfail(
+                reason="see examples/trio/lockacquire_not_umasked.py"
+            )
+        ),
+    ]
+)
+def test_cancelled_lockacquire_in_ipctx_not_unmaskeed(
+    ignore_special_cases: bool,
+    loglevel: str,
+    debug_mode: bool,
+):
+    mod: ModuleType = pathlib.import_path(
+        examples_dir()
+        / 'trio'
+        / 'lockacquire_not_unmasked.py',
+        root=examples_dir(),
+        consider_namespace_packages=False,
+    )
+    async def _main():
+        with trio.fail_after(2):
+            await mod.main(
+                ignore_special_cases=ignore_special_cases,
+                loglevel=loglevel,
+                debug_mode=debug_mode,
+            )
+
+    trio.run(_main)