Use `delay=0` in pump loop..

Turns out it does work XD

Prior presumption was from before I had the fute poll-loop so makes
sense we needed more then one sched-tick's worth of context switch vs.
now we can just keep looping-n-pumping as fast possible until the
guest-run's main task completes.

Also,
- minimize the preface commentary (as per todo) now that we have tests
  codifying all the edge cases :finger_crossed:
- parameter-ize the pump-loop-cycle delay and default it to 0.

tractor/to_asyncio.py

@@ -558,6 +558,8 @@ def run_as_asyncio_guest(
     # normally `Actor._async_main()` as is passed by some boostrap
     # entrypoint like `._entry._trio_main()`.
 
+    _sigint_loop_pump_delay: float = 0,
+
 ) -> None:
 # ^-TODO-^ technically whatever `trio_main` returns.. we should
 # try to use func-typevar-params at leaast by 3.13!
@@ -598,7 +600,7 @@ def run_as_asyncio_guest(
 
         '''
         loop = asyncio.get_running_loop()
-        trio_done_fut = asyncio.Future()
+        trio_done_fute = asyncio.Future()
         startup_msg: str = (
             'Starting `asyncio` guest-loop-run\n'
             '-> got running loop\n'
@@ -633,13 +635,13 @@ def run_as_asyncio_guest(
                     f'{error}\n\n'
                     f'{tb_str}\n'
                 )
-                trio_done_fut.set_exception(error)
+                trio_done_fute.set_exception(error)
 
                 # raise inline
                 main_outcome.unwrap()
 
             else:
-                trio_done_fut.set_result(main_outcome)
+                trio_done_fute.set_result(main_outcome)
 
         startup_msg += (
             f'-> created {trio_done_callback!r}\n'
@@ -660,7 +662,7 @@ def run_as_asyncio_guest(
         )
         fute_err: BaseException|None = None
         try:
-            out: Outcome = await asyncio.shield(trio_done_fut)
+            out: Outcome = await asyncio.shield(trio_done_fute)
 
             # NOTE will raise (via `Error.unwrap()`) from any
             # exception packed into the guest-run's `main_outcome`.
@@ -697,83 +699,75 @@ def run_as_asyncio_guest(
                 f'  |_{actor}.cancel_soon()\n'
             )
 
-            # TODO: reduce this comment bloc since abandon issues are
+            # XXX WARNING XXX the next LOCs are super important, since
-            # now solved?
+            # without them, we can get guest-run abandonment cases
+            # where `asyncio` will not schedule or wait on the `trio`
+            # guest-run task before final shutdown! This is
+            # particularly true if the `trio` side has tasks doing
+            # shielded work when a SIGINT condition occurs.
             #
-            # XXX NOTE XXX the next LOC is super important!!!
+            # We now have the
-            #  => without it, we can get a guest-run abandonment case
+            # `test_infected_asyncio.test_sigint_closes_lifetime_stack()`
-            #  where asyncio will not trigger `trio` in a final event
+            # suite to ensure we do not suffer this issues
-            #  loop cycle!
+            # (hopefully) ever again.
             #
-            #  our test,
+            # The original abandonment issue surfaced as 2 different
-            #  `test_infected_asyncio.test_sigint_closes_lifetime_stack()`
+            # race-condition dependent types scenarios all to do with
-            #  demonstrates how if when we raise a SIGINT-signal in an infected
+            # `asyncio` handling SIGINT from the system:
-            #  child we get a variable race condition outcome where
-            #  either of the following can indeterminately happen,
             #
-            # - "silent-abandon": `asyncio` abandons the `trio`
+            # - "silent-abandon" (WORST CASE):
-            #   guest-run task silently and no `trio`-guest-run or
+            #  `asyncio` abandons the `trio` guest-run task silently
-            #   `tractor`-actor-runtime teardown happens whatsoever..
+            #  and no `trio`-guest-run or `tractor`-actor-runtime
-            #   this is the WORST (race) case outcome.
+            #  teardown happens whatsoever..
             #
-            # - OR, "loud-abandon": the guest run get's abaondoned "loudly" with
+            # - "loud-abandon" (BEST-ish CASE):
-            #   `trio` reporting a console traceback and further tbs of all
+            #   the guest run get's abaondoned "loudly" with `trio`
-            #   the failed shutdown routines also show on console..
+            #   reporting a console traceback and further tbs of all
+            #   the (failed) GC-triggered shutdown routines which
+            #   thankfully does get dumped to console..
             #
-            # our test can thus fail and (has been parametrized for)
+            # The abandonment is most easily reproduced if the `trio`
-            # the 2 cases:
+            # side has tasks doing shielded work where those tasks
+            # ignore the normal `Cancelled` condition and continue to
+            # run, but obviously `asyncio` isn't aware of this and at
+            # some point bails on the guest-run unless we take manual
+            # intervention..
             #
-            # - when the parent raises a KBI just after
+            # To repeat, *WITHOUT THIS* stuff below the guest-run can
-            #   signalling the child,
+            # get race-conditionally abandoned!!
-            #  |_silent-abandon => the `Actor.lifetime_stack` will
-            #    never be closed thus leaking a resource!
-            #   -> FAIL!
-            #  |_loud-abandon => despite the abandonment at least the
-            #    stack will be closed out..
-            #   -> PASS
             #
-            # - when the parent instead simply waits on `ctx.wait_for_result()`
+            # XXX SOLUTION XXX
-            #   (i.e. DOES not raise a KBI itself),
+            # ------ - ------
-            #  |_silent-abandon => test will just hang and thus the ctx
+            # XXX FIRST PART:
-            #    and actor will never be closed/cancelled/shutdown
+            # ------ - ------
-            #    resulting in leaking a (file) resource since the
+            # the obvious fix to the "silent-abandon" case is to
-            #    `trio`/`tractor` runtime never relays a ctxc back to
+            # explicitly cancel the actor runtime such that no
-            #    the parent; the test's timeout will trigger..
+            # runtime tasks are even left unaware that the guest-run
-            #   -> FAIL!
+            # should be terminated due to OS cancellation.
-            #  |_loud-abandon => this case seems to never happen??
             #
-            # XXX FIRST PART XXX, SO, this is a fix to the
-            # "silent-abandon" case, NOT the `trio`-guest-run
-            # abandonment issue in general, for which the NEXT LOC
-            # is apparently a working fix!
             actor.cancel_soon()
 
-            # XXX NOTE XXX pump the `asyncio` event-loop to allow
+            # ------ - ------
+            # XXX SECOND PART:
+            # ------ - ------
+            # Pump the `asyncio` event-loop to allow
             # `trio`-side to `trio`-guest-run to complete and
             # teardown !!
             #
-            # *WITHOUT THIS* the guest-run can get race-conditionally abandoned!!
+            # oh `asyncio`, how i don't miss you at all XD
-            # XD
+            while not trio_done_fute.done():
-            #
-            await asyncio.sleep(.1)  # `delay` can't be 0 either XD
-            while not trio_done_fut.done():
                 log.runtime(
                     'Waiting on main guest-run `asyncio` task to complete..\n'
-                    f'|_trio_done_fut: {trio_done_fut}\n'
+                    f'|_trio_done_fut: {trio_done_fute}\n'
                 )
-                await asyncio.sleep(.1)
+                await asyncio.sleep(_sigint_loop_pump_delay)
 
-                # XXX: don't actually need the shield.. seems to
+                # XXX is there any alt API/approach like the internal
-                # make no difference (??) and we know it spawns an
+                # call below but that doesn't block indefinitely..?
-                # internal task..
-                # await asyncio.shield(asyncio.sleep(.1))
-
-                # XXX alt approach but can block indefinitely..
-                # so don't use?
                 # loop._run_once()
 
             try:
-                return trio_done_fut.result()
+                return trio_done_fute.result()
             except asyncio.exceptions.InvalidStateError as state_err:
 
                 # XXX be super dupere noisy about abandonment issues!