tests/_helpers.py

@@ -6,12 +6,11 @@ from pathlib import Path
 from types import TracebackType
 
 import tractor
-import trio
 
 
 class CancellationMarkers:
     '''
-    Mark a test endpoint and require cancellation-driven teardown.
+    Mark a test endpoint's lifetime without cleanup checkpoints.
 
     '''
     def __init__(
@@ -31,8 +30,6 @@ class CancellationMarkers:
         exc_value: BaseException|None,
         traceback: TracebackType|None,
     ) -> None:
-        assert exc_type is trio.Cancelled
-        assert isinstance(exc_value, trio.Cancelled)
         Path(self.cancelled_path).touch()
 
 

tests/devx/test_debugger.py

@@ -1359,8 +1359,6 @@ def test_ctxep_pauses_n_maybe_ipc_breaks(
                 expect_prompt=False,
             )
             child.expect(EOF)
-            before += ansi_strip(child.before.decode())
-            assert 'KeyboardInterrupt' in before
             assert child.flag_eof
             assert not child.isalive()
 

tests/ipc/test_each_tpt.py

@@ -33,19 +33,18 @@ def test_cancelled_transport_send_completes_frame():
 
     A cancelled `send_all()` may leave an arbitrary frame prefix on the
     wire. Closing the actor-wide stream avoids decoder corruption but
-    also destroys unrelated contexts using that channel. On its first
-    call, the fake stream publishes two header bytes and blocks until
-    the parent test releases it. This lets the parent cancel the sender
-    while frame publication is suspended. The sender must remain inside
-    `send_first()` until the complete frame is written, then observe
-    pending cancellation; a second sender proves sibling contexts can
-    safely reuse the still frame-aligned stream.
+    also destroys unrelated contexts using that channel. The fake
+    stream publishes two header bytes and blocks, letting this test
+    cancel the sender inside frame publication. The sender must remain
+    blocked until the complete frame is written, then observe pending
+    cancellation; a second complete frame proves channel reuse remains
+    safe.
 
     '''
     class PartialSendStream:
         def __init__(self) -> None:
-            self.send_all_entered = trio.Event()
-            self.send_all_release = trio.Event()
+            self.send_entered = trio.Event()
+            self.release = trio.Event()
             self.closed = False
             self.wire = bytearray()
 
@@ -56,8 +55,8 @@ def test_cancelled_transport_send_completes_frame():
             assert data
             if not self.wire:
                 self.wire.extend(data[:2])
-                self.send_all_entered.set()
-                await self.send_all_release.wait()
+                self.send_entered.set()
+                await self.release.wait()
                 self.wire.extend(data[2:])
             else:
                 self.wire.extend(data)
@@ -116,29 +115,21 @@ def test_cancelled_transport_send_completes_frame():
             tn.start_soon(
                 send_first,
             )
-            await stream.send_all_entered.wait()
+            await stream.send_entered.wait()
             sender_scopes[0].cancel()
             await wait_all_tasks_blocked()
 
             assert not stream.closed
-            # Cancellation is pending, but complete-frame shielding
-            # keeps `send_first()` suspended in `.send_all()`.
             assert not sender_done.is_set()
 
-            # Let the underlying frame write finish after the parent
-            # has requested sender cancellation.
-            stream.send_all_release.set()
+            stream.release.set()
             await sender_done.wait()
 
             assert cancelled_caught
             assert not stream.closed
-            # The initial two-byte prefix was completed into one valid
-            # frame before cancellation reached `send_first()`.
             assert count_frames(stream.wire) == 1
 
             await transport.send(second_msg)
-            # A sibling sender can append and decode another frame only
-            # because the first cancellation preserved stream alignment.
             assert count_frames(stream.wire) == 2
 
             tn.cancel_scope.cancel()
@@ -148,15 +139,14 @@ def test_cancelled_transport_send_completes_frame():
 
 def test_transport_send_deadline_closes_partial_frame():
     '''
-    Destroy a stalled partial frame before another sender can append.
+    Bound one shielded frame without exposing a corrupt stream.
 
     Ordinary cancellation cannot interrupt complete-frame publication.
-    Bounded actor/context cancellation instead passes its absolute
-    deadline into this operation. The fake stream writes a partial
-    header and stalls; when the send's own deadline fires, the transport
-    must close the stream before releasing its shared send lock. This
-    prevents the next sender from appending bytes which a decoder would
-    treat as the remainder of the corrupt first frame.
+    Actor-wide cancellation instead passes its absolute deadline into
+    this operation. The fake stream writes a partial header and stalls;
+    when the send's own deadline fires, the transport must close the
+    stream before releasing its shared send lock and report the channel
+    unusable.
 
     '''
     class StalledStream:
@@ -196,9 +186,9 @@ def test_transport_send_deadline_closes_partial_frame():
                 send_deadline=1,
             )
 
-        assert stream.closed  # partial-frame timeout destroys stream
-        assert len(stream.wire) == 2  # only a header fragment was sent
-        assert not transport._send_lock.locked()  # cleanup released lock
+        assert stream.closed
+        assert len(stream.wire) == 2
+        assert not transport._send_lock.locked()
 
     trio.run(
         main,
@@ -210,35 +200,30 @@ def test_cancelled_transport_send_preserves_cancellation():
     '''
     Prefer sender cancellation when teardown closes the stream.
 
-    `MsgpackTransport.send()` shields frame publication at
-    `tractor.ipc._transport:MsgpackTransport.send`. Before this fix, an
-    outer `move_on_after()`/cancel scope could cancel `Channel.send()`
-    while actor teardown closed the shared stream. The resulting
-    `ClosedResourceError` escaped from the transport handler instead of
-    its `checkpoint_if_cancelled()` redelivering pending cancellation.
+    `MsgpackTransport.send()` shields frame publication. Before this
+    regression fix, if an outer scope cancelled the sender and actor
+    teardown then made `send_all()` raise `ClosedResourceError`, the
+    transport error escaped instead of the pending cancellation. That
+    defeated `move_on_after()` and failed otherwise orderly teardown.
 
     The fake stream blocks inside the shield until the test cancels the
-    sender. Parent-controlled release then simulates actor teardown
-    closing the socket and raises the `ClosedResourceError` observed on
-    macOS UDS. `CancelScope.cancelled_caught` proves the handler's
-    checkpoint preserved cancellation as the primary outcome instead of
-    leaking that secondary close error.
+    sender, then raises the same close error seen on macOS UDS. Observing
+    `CancelScope.cancelled_caught` proves cancellation wins once the
+    shield unwinds.
 
     '''
     class ClosingStream:
         def __init__(self) -> None:
-            self.send_all_entered = trio.Event()
-            self.send_all_release = trio.Event()
+            self.send_entered = trio.Event()
+            self.release = trio.Event()
 
         async def send_all(
             self,
             data: bytes,
         ) -> None:
             assert data
-            self.send_all_entered.set()
-            await self.send_all_release.wait()
-            # Model actor teardown closing the shared transport while
-            # this sender is still inside the complete-frame shield.
+            self.send_entered.set()
+            await self.release.wait()
             raise trio.ClosedResourceError(
                 'this socket was already closed'
             )
@@ -271,12 +256,12 @@ def test_cancelled_transport_send_preserves_cancellation():
 
         async with trio.open_nursery() as tn:
             tn.start_soon(send)
-            await stream.send_all_entered.wait()
+            await stream.send_entered.wait()
             sender_scopes[0].cancel()
             await wait_all_tasks_blocked()
 
             assert not sender_done.is_set()
-            stream.send_all_release.set()
+            stream.release.set()
             await sender_done.wait()
 
         assert cancelled_caught

tests/test_context_stream_semantics.py

@@ -86,7 +86,7 @@ from ._helpers import (
 
 def test_overrun_error_send_tolerates_transport_close(
     monkeypatch: pytest.MonkeyPatch,
-) -> None:
+):
     '''
     Preserve a stream overrun when its error can not be shipped.
 
@@ -108,13 +108,10 @@ def test_overrun_error_send_tolerates_transport_close(
     )
     packed: dict[str, object] = {}
 
-    # Spy on the generated `StreamOverrun` and return a stable wire
-    # `Error`; the real packer adds traceback/relay details unrelated to
-    # this test's secondary transport-close contract.
     def pack_overrun(
         local_err: BaseException,
         cid: str,
-        **kwargs: object,
+        **kwargs,
     ) -> tractor.msg.Error:
         packed['local_err'] = local_err
         packed['cid'] = cid
@@ -278,7 +275,7 @@ async def simple_setup_teardown(
         _state = False
 
 
-async def assert_state(value: bool) -> None:
+async def assert_state(value: bool):
     global _state
     assert _state == value
 
@@ -289,7 +286,7 @@ async def test_cancel_during_context_startup(
     tmp_path: Path,
     start_method: str,
     debug_mode: bool,
-) -> None:
+):
     '''
     Cancel a context after sending `Start` but before its ack.
 
@@ -306,7 +303,6 @@ async def test_cancel_during_context_startup(
     started_path = tmp_path / 'startup_started'
     cancelled_path = tmp_path / 'startup_cancelled'
     start_sent = trio.Event()
-    start_funcs: list[str] = []
     original_send = tractor.Channel.send
 
     async def delay_after_start(
@@ -321,11 +317,7 @@ async def test_cancel_during_context_startup(
             hide_tb=hide_tb,
             send_deadline=send_deadline,
         )
-        # The patched method keeps `Channel.send()`'s broad message
-        # contract. It sees both the requested endpoint `Start` and the
-        # internal `self._cancel_task` startup RPC used for cleanup.
         if isinstance(payload, tractor.msg.Start):
-            start_funcs.append(payload.func)
             if payload.func == 'startup_cancel_target':
                 start_sent.set()
             await trio.sleep_forever()
@@ -341,7 +333,7 @@ async def test_cancel_during_context_startup(
             raise AssertionError('context startup should be cancelled')
 
     async with tractor.open_nursery() as an:
-        actor: Actor = tractor.current_actor()
+        actor = tractor.current_actor()
         portal: tractor.Portal = await an.start_actor(
             'startup_cancel_worker',
             enable_modules=[__name__],
@@ -373,10 +365,6 @@ async def test_cancel_during_context_startup(
             'return_one',
         ) == 1
         assert non_registration_contexts(actor) == contexts_before
-        assert start_funcs == [
-            'startup_cancel_target',
-            '_cancel_task',
-        ]
         await portal.cancel_actor()
 
 
@@ -384,7 +372,7 @@ async def test_cancel_during_context_startup(
 async def test_start_serialization_error_cleans_context(
     start_method: str,
     debug_mode: bool,
-) -> None:
+):
     '''
     Deallocate caller state when `Start` can not be serialized.
 
@@ -397,7 +385,7 @@ async def test_start_serialization_error_cleans_context(
 
     '''
     async with tractor.open_nursery() as an:
-        actor: Actor = tractor.current_actor()
+        actor = tractor.current_actor()
         portal: tractor.Portal = await an.start_actor(
             'serialization_error_worker',
             enable_modules=[__name__],
@@ -426,7 +414,7 @@ async def test_start_serialization_error_cleans_context(
 async def test_start_module_error_cleans_context(
     start_method: str,
     debug_mode: bool,
-) -> None:
+):
     '''
     Deallocate caller state after a remote startup rejection.
 
@@ -439,7 +427,7 @@ async def test_start_module_error_cleans_context(
 
     '''
     async with tractor.open_nursery() as an:
-        actor: Actor = tractor.current_actor()
+        actor = tractor.current_actor()
         portal: tractor.Portal = await an.start_actor(
             'module_error_worker',
         )

tests/test_to_actor.py

@@ -77,9 +77,9 @@ async def mark_task_cancellation(
 
 
 async def echo_startup_control(
-    cancel_on_startup: str,
+    _cancel_on_startup: str,
 ) -> str:
-    return cancel_on_startup
+    return _cancel_on_startup
 
 
 async def collect_args(
@@ -88,6 +88,13 @@ async def collect_args(
     return args
 
 
+async def collect_call(
+    *args: object,
+    **kwargs: object,
+) -> tuple[tuple[object, ...], dict[str, object]]:
+    return args, kwargs
+
+
 def test_public_module_alias() -> None:
     '''
     Keep the public trampoline alias separate from its private module.
@@ -580,12 +587,9 @@ async def test_reuse_existing_actor_via_portal(
     Pass `portal=` to schedule the one-shot task in an
     already-running actor; no spawn, no implicit reap.
 
-    The low-level call uses `__name__` to select the remote module and
-    `'echo_startup_control'` to select its function. Public
-    `Portal.run_from_ns()` packages `cancel_on_startup` inside the
-    target `kwargs` passed to private `Portal._run_from_ns()`. Receiving
-    `'target_value'` back proves the value reached the target instead of
-    binding the private Boolean startup-cancellation policy parameter.
+    The low-level `Portal.run_from_ns()` assertion also proves its
+    target kwargs remain separate from the private startup-cancel
+    policy used by context cleanup.
 
     '''
     async with tractor.open_nursery() as an:
@@ -608,7 +612,7 @@ async def test_reuse_existing_actor_via_portal(
         assert await portal.run_from_ns(
             __name__,
             'echo_startup_control',
-            cancel_on_startup='target_value',
+            _cancel_on_startup='target_value',
         ) == 'target_value'
         assert non_registration_contexts(actor) == contexts_before
 
@@ -627,8 +631,6 @@ async def test_concurrent_one_shots_from_task_nursery(
     nursery scheduling multiple one-shot calls against
     a shared caller-managed actor-nursery; error
     collection thus lives entirely in caller-code.
-    A proposed distilled task-manager API is tracked in #485:
-    https://github.com/goodboy/tractor/issues/485
 
     '''
     results: dict[int, int] = {}
@@ -656,7 +658,7 @@ async def test_concurrent_one_shots_from_task_nursery(
     }
 
 
-def test_rejects_sync_fn() -> None:
+def test_rejects_sync_fn():
     '''
     Non-async callables error BEFORE any spawn (or even
     runtime-boot) happens.
@@ -665,10 +667,7 @@ def test_rejects_sync_fn() -> None:
     def not_async() -> None:
         ...
 
-    with pytest.raises(
-        TypeError,
-        match='must be a non-streaming async function',
-    ):
+    with pytest.raises(TypeError):
         trio.run(
             partial(
                 to_actor.run,
@@ -677,7 +676,7 @@ def test_rejects_sync_fn() -> None:
         )
 
 
-def test_rejects_streaming_fn() -> None:
+def test_rejects_streaming_fn():
     '''
     Async-gen (streaming) fns are not one-shot-able,
     same constraint as `Portal.run()`.
@@ -686,10 +685,7 @@ def test_rejects_streaming_fn() -> None:
     async def agen():
         yield 1
 
-    with pytest.raises(
-        TypeError,
-        match='must be a non-streaming async function',
-    ):
+    with pytest.raises(TypeError):
         trio.run(
             partial(
                 to_actor.run,
@@ -742,7 +738,7 @@ def test_partial_placeholder_normalization(
         )
 
 
-def test_nested_partial_normalization() -> None:
+def test_nested_partial_normalization():
     '''
     Flatten every retained `functools.partial` layer before RPC.
 
@@ -754,12 +750,6 @@ def test_nested_partial_normalization() -> None:
     direct nested-partial call.
 
     '''
-    async def collect_call(
-        *args: object,
-        **kwargs: object,
-    ) -> tuple[tuple[object, ...], dict[str, object]]:
-        return args, kwargs
-
     inner = partial(
         collect_call,
         1,
@@ -781,15 +771,13 @@ def test_nested_partial_normalization() -> None:
     assert kwargs == {'label': 'outer'}
 
 
-def test_rejects_portal_and_an_combo() -> None:
+def test_rejects_portal_and_an_combo():
     '''
-    `portal=` and `an=` are mutually exclusive actor-lifetime handles.
+    `portal=` and `an=` are mutually exclusive
+    placement options.
 
     '''
-    with pytest.raises(
-        ValueError,
-        match='Pass at most ONE of `portal` or `an`',
-    ):
+    with pytest.raises(ValueError):
         trio.run(
             partial(
                 to_actor.run,
@@ -802,7 +790,7 @@ def test_rejects_portal_and_an_combo() -> None:
 
 
 @pytest.mark.parametrize(
-    'lifetime_mode',
+    'placement',
     ['an', 'portal'],
 )
 @pytest.mark.parametrize(
@@ -813,31 +801,28 @@ def test_rejects_portal_and_an_combo() -> None:
     ],
     ids=['empty', 'configured'],
 )
-def test_rejects_runtime_kwargs_with_lifetime_mode(
-    lifetime_mode: str,
+def test_rejects_runtime_kwargs_with_placement(
+    placement: str,
     runtime_kwargs: dict,
-) -> None:
+):
     '''
     `runtime_kwargs` only applies when the call opens
     its own private actor-nursery; passing it alongside
-    an actor-lifetime handle is an error, never silently
+    a placement opt is an error, never silently
     ignored. In particular, an empty dict still means the
     caller provided this mutually exclusive option; testing
-    both lifetime modes prevents truthiness checks from
+    both placement modes prevents truthiness checks from
     accepting it before any actor runtime is started.
 
     '''
-    with pytest.raises(
-        ValueError,
-        match='`runtime_kwargs` only applies',
-    ):
+    with pytest.raises(ValueError):
         trio.run(
             partial(
                 to_actor.run,
                 add_one,
                 1,
                 **{
-                    lifetime_mode: object(),
+                    placement: object(),
                     'runtime_kwargs': runtime_kwargs,
                 },
             )
@@ -895,13 +880,12 @@ async def test_portal_task_cancelled_with_local_caller(
     Couple a reused portal's remote task to its local caller.
 
     The former `Portal.run()` path abandoned its remote task when the
-    local `to_actor.run()` caller was cancelled. `CancellationMarkers`
-    writes one file after entry and writes the second only after its
-    synchronous exit verifies the remote task received `trio.Cancelled`.
-    Cancelling the local task nursery and observing that marker proves
-    `Portal.open_context()` propagated cancellation before the caller
-    exited. A subsequent call proves the caller-owned actor was not
-    cancelled with that task.
+    local `to_actor.run()` caller was cancelled. The target writes
+    one file after starting and another from its cancellation
+    `finally`. Cancelling the local task nursery and observing the
+    second file proves `Portal.open_context()` propagated
+    cancellation before the caller exited. A subsequent call proves
+    the caller-owned actor was not cancelled with that task.
 
     '''
     started_path = tmp_path / 'started'
@@ -934,16 +918,12 @@ async def test_portal_task_cancelled_with_local_caller(
             tn.cancel_scope.cancel()
 
         assert cancelled_path.exists()
-        # Remote-task cancellation must restore the exact application
-        # context snapshot captured before the one-shot call.
         assert non_registration_contexts(actor) == contexts_before
         assert await to_actor.run(
             add_one,
             1,
             portal=portal,
         ) == 2
-        # Reusing the actor for a later successful call must also leave
-        # no local or remote context registry entries behind.
         assert non_registration_contexts(actor) == contexts_before
 
         await portal.cancel_actor()
@@ -978,10 +958,7 @@ async def test_context_trampoline_preserves_module_allowlist(
                 portal=portal,
             )
 
-        err = excinfo.value
-        assert err.boxed_type is tractor.ModuleNotExposed
-        assert add_one.__module__ in str(err)
-        assert 'Make sure you exposed the target module' in str(err)
+        assert excinfo.value.boxed_type is tractor.ModuleNotExposed
         assert non_registration_contexts(actor) == contexts_before
         await portal.cancel_actor()
 

tractor/_context.py

@@ -1113,10 +1113,6 @@ class Context:
                 # NOTE: we're telling the far end actor to cancel a task
                 # corresponding to *this actor*. The far end local channel
                 # instance is passed to `Actor._cancel_task()` implicitly.
-                # Use private `Portal._run_from_ns()` because cancellation
-                # needs its internal `cancel_on_startup=False` policy and
-                # the transaction's shared absolute `send_deadline`; public
-                # `run_from_ns()` exposes neither control.
                 await self._portal._run_from_ns(
                     'self',
                     '_cancel_task',
@@ -1959,9 +1955,6 @@ class Context:
         # the sender; the main motivation is that using bp can block the
         # msg handling loop which calls into this method!
         except trio.WouldBlock:
-            # `send_chan.send_nowait(msg)` found the local receive feeder
-            # full. With overruns disabled below, report that primary
-            # local overflow to the far-end sender as `StreamOverrun`.
 
             # XXX: always push an error even if the local receiver
             # is in overrun state - i.e. if an 'error' msg is
@@ -2034,13 +2027,12 @@ class Context:
                     await chan.send(err_msg)
                     return True
 
-                # The `StreamOverrun` shipment can fail secondarily when
-                # context/channel teardown has already closed shared IPC.
-                # Local stream closure may surface as
-                # `BrokenResourceError`; either peer closing the transport
-                # can surface as `TransportClosed`. In both cases teardown
-                # owns far-end cancellation and the primary overrun can no
-                # longer be delivered.
+                # XXX: the local consumer may have closed its side of
+                # the IPC, in which case context/channel teardown owns
+                # cancellation of the far-end streaming task. The same
+                # shipment can raise `TransportClosed` when either peer
+                # has already closed the shared IPC channel. In both
+                # cases the primary overrun can no longer be reported.
                 except (
                     TransportClosed,
                     trio.BrokenResourceError,

tractor/ipc/_transport.py

@@ -448,11 +448,9 @@ class MsgpackTransport(MsgTransport):
         If `strict_types == True` then a `MsgTypeError` will be raised on any
         invalid msg type
 
-        `send_deadline` bounds publication of one complete
-        length-prefixed frame. If it expires after a prefix or payload
-        fragment reaches the wire, a later sender would append bytes the
-        peer decoder treats as the remainder of that corrupt frame. The
-        stream is therefore destroyed before releasing `._send_lock`.
+        `send_deadline` bounds publication of this complete frame. A
+        timeout destroys the stream because a partial prefix may have
+        reached the wire.
 
         '''
         __tracebackhide__: bool = hide_tb
@@ -507,17 +505,18 @@ class MsgpackTransport(MsgTransport):
             try:
                 # Every IPC msg is length-prefixed and all contexts
                 # on this actor pair share one transport stream. If
-                # the send deadline interrupts `send_all()`, an unknown
-                # frame prefix may already be on the wire; allowing the
-                # next sender to append would corrupt framing.
+                # cancellation interrupts `send_all()`, an unknown
+                # frame prefix may already be on the wire; allowing
+                # the next sender to append would corrupt framing.
+                # Closing the stream avoids that corruption but lets
+                # one context-local cancellation destroy every sibling
+                # context using the channel.
                 #
-                # The enclosing `async with self._send_lock` retains the
-                # lock through shielded publication and any forced-close
-                # cleanup. Context-manager exit releases it only after a
-                # complete frame or destruction of the corrupt stream.
-                # Ordinary outer cancellation remains shielded until
-                # complete publication; only expiry of `send_deadline`
-                # intentionally closes a partial stream here.
+                # Keep the `._send_lock` and defer cancellation only
+                # for complete frame publication. Broken/closed stream
+                # failures still escape to the handlers below. Once
+                # the frame is complete, the explicit checkpoint
+                # immediately delivers any pending cancellation.
                 #
                 # Ordinary sends may delay cancellation while the remote
                 # peer actor is not reading. Bounded actor/context cancel
@@ -540,8 +539,6 @@ class MsgpackTransport(MsgTransport):
                         f'deadline of {send_deadline!r}'
                     )
 
-                # The frame is complete and still aligned. Redeliver any
-                # pending outer cancellation before normal lock release.
                 await trio.lowlevel.checkpoint_if_cancelled()
                 return None
 

tractor/runtime/_runtime.py

@@ -794,9 +794,8 @@ class Actor:
         cancel_on_startup: bool = True,
 
         # Optional absolute deadline for publishing this exact `Start`
-        # frame. Used by bounded actor/context cancel RPCs whose outer
-        # timeout cannot penetrate `Channel.send()` forwarding into the
-        # shield in `MsgpackTransport.send()`.
+        # frame. Used by actor-wide cancel RPCs whose outer timeout
+        # cannot penetrate complete-frame transport shielding.
         send_deadline: float = float('inf'),
 
     ) -> Context: