From 23e26b5b320caa5907ccda4fd373e52abbb80d5f Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 20 Aug 2026 22:52:20 -0400 Subject: [PATCH] Bound `Portal.cancel_actor()` frame sends A cancel RPC could stall forever in complete-frame transport shielding before the peer received it, bypassing the outer ack timeout and blocking graceful supervision. - thread one absolute deadline from `Portal.cancel_actor()` through the private `Start` publication path - force-close a partial-frame stream before releasing its send lock - keep ordinary sends unbounded and preserve pending cancellation - document the current `Start -> StartAck -> CancelAck` exchange and link the dedicated `Cancel` msg follow-up in #506 - cover partial publication and the shared send/ack timeout budget Prompt-IO: ai/prompt-io/opencode/20260821T023537Z_ae6f2ac3_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- .../20260821T023537Z_ae6f2ac3_prompt_io.md | 65 ++++++++++++++ ...20260821T023537Z_ae6f2ac3_prompt_io.raw.md | 41 +++++++++ tests/ipc/test_each_tpt.py | 64 +++++++++++++- tests/test_to_actor.py | 63 ++++++++++++++ tractor/ipc/_chan.py | 19 ++++- tractor/ipc/_transport.py | 27 +++++- tractor/runtime/_portal.py | 85 ++++++++++++------- tractor/runtime/_runtime.py | 13 ++- 8 files changed, 337 insertions(+), 40 deletions(-) create mode 100644 ai/prompt-io/opencode/20260821T023537Z_ae6f2ac3_prompt_io.md create mode 100644 ai/prompt-io/opencode/20260821T023537Z_ae6f2ac3_prompt_io.raw.md diff --git a/ai/prompt-io/opencode/20260821T023537Z_ae6f2ac3_prompt_io.md b/ai/prompt-io/opencode/20260821T023537Z_ae6f2ac3_prompt_io.md new file mode 100644 index 00000000..1e60c94c --- /dev/null +++ b/ai/prompt-io/opencode/20260821T023537Z_ae6f2ac3_prompt_io.md @@ -0,0 +1,65 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +session: 76c5d31c-5a2f-4503-9b16-410ee7f4fab3 +timestamp: 2026-08-21T02:35:37Z +git_ref: ae6f2ac3 +scope: code +substantive: true +raw_file: 20260821T023537Z_ae6f2ac3_prompt_io.raw.md +--- + +## Prompt + +Simplify bounded actor cancellation by passing an explicit absolute +deadline from `Portal.cancel_actor()` through `_run_from_ns()`, +`Actor.start_remote_task()`, and `Channel.send()` into +`MsgpackTransport.send()`. Avoid a `ContextVar`, watcher tasks, shared +status, coalescing, and waiter state. After tracing the current +`Start -> StartAck -> CancelAck` transaction, rename the local result to +`cancel_ack_received`, document its exact semantics, and link a focused +follow-up for a dedicated `Cancel -> CancelAck` protocol. + +## Response summary + +Threaded one absolute Trio deadline through the existing private +actor-cancel RPC path. The transport retains complete-frame shielding +for ordinary sends, while a cancel-control send that overruns its +deadline force-closes the potentially corrupted stream before releasing +the send lock. The outer actor-cancel scope uses the same deadline for +ack waiting and redelivers pending caller cancellation afterward. + +Renamed the completion flag to `cancel_ack_received` and documented that +the current private call consumes `StartAck`, then receives a real +`CancelAck` after `Actor.cancel()` completes; this does not establish +that the OS process exited. Added a source TODO linking issue #506 for +the future first-class `Cancel -> CancelAck` transaction. + +Focused transport and actor-cancel verification passed all four tests. + +## Files changed + +- `tractor/runtime/_portal.py` - own the absolute deadline, accurately + record ack receipt, and link the dedicated cancellation protocol. +- `tractor/runtime/_runtime.py` - forward the optional deadline for the + exact private `Start` publication. +- `tractor/ipc/_chan.py` - pass the operation-specific deadline to the + transport without changing ordinary sends. +- `tractor/ipc/_transport.py` - bound the shielded frame publication and + close a partial-frame stream before unlocking it. +- `tests/ipc/test_each_tpt.py` - cover deadline expiry after a partial + frame prefix reaches the stream. +- `tests/test_to_actor.py` - prove actor-cancel publication and ack + waiting share one absolute timeout budget. + +## Human edits + +The human rejected the initial watcher-task, shared `_SendStatus`, cancel +coalescing, and per-waiter design as unnecessary complexity. They also +rejected `ContextVar` propagation in favor of explicit functional +threading, selected a single absolute deadline for publication and ack +waiting, and required item 2 to remain separate from the item-3 child +reaping work. After reviewing the result, they requested the precise +`cancel_ack_received` name, a detailed protocol-trace comment, a focused +follow-up issue, and a linked source TODO. No direct source-line edits +were made by the human. diff --git a/ai/prompt-io/opencode/20260821T023537Z_ae6f2ac3_prompt_io.raw.md b/ai/prompt-io/opencode/20260821T023537Z_ae6f2ac3_prompt_io.raw.md new file mode 100644 index 00000000..ce468a24 --- /dev/null +++ b/ai/prompt-io/opencode/20260821T023537Z_ae6f2ac3_prompt_io.raw.md @@ -0,0 +1,41 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +timestamp: 2026-08-21T02:35:37Z +git_ref: ae6f2ac3 +diff_cmd: git diff HEAD~1..HEAD +--- + +Replace the actor-cancel timeout watcher/status experiment with one +explicit absolute deadline threaded through the existing private call +path. Do not use a `ContextVar`, shared result state, waiter +coalescing, or polling tasks. + +> `git diff HEAD~1..HEAD -- tractor/runtime/_portal.py` + +`Portal.cancel_actor()` computes one absolute deadline and uses it for +both `Start` frame publication and the subsequent cancel-ack wait. + +> `git diff HEAD~1..HEAD -- tractor/runtime/_runtime.py` + +> `git diff HEAD~1..HEAD -- tractor/ipc/_chan.py` + +The private RPC path forwards the operation-specific deadline. Lower +layers preserve the ordinary infinite-deadline call shape. + +> `git diff HEAD~1..HEAD -- tractor/ipc/_transport.py` + +`MsgpackTransport.send()` applies the deadline inside its complete-frame +shield. If the deadline expires after partial publication, it closes +the unusable stream before releasing the send lock. + +> `git diff HEAD~1..HEAD -- tests/ipc/test_each_tpt.py` + +> `git diff HEAD~1..HEAD -- tests/test_to_actor.py` + +Focused regressions prove a partial-frame timeout closes the stream and +that actor-cancel publication and acknowledgement share one budget. + +The implementation removes the earlier `_SendStatus`, watcher task, +coalescing, shared cancel result, and per-waiter state. Four focused +transport and actor-cancel tests pass. diff --git a/tests/ipc/test_each_tpt.py b/tests/ipc/test_each_tpt.py index 83bd299a..73debc62 100644 --- a/tests/ipc/test_each_tpt.py +++ b/tests/ipc/test_each_tpt.py @@ -15,7 +15,10 @@ from unittest.mock import Mock import pytest import trio -from trio.testing import wait_all_tasks_blocked +from trio.testing import ( + MockClock, + wait_all_tasks_blocked, +) import tractor from tractor import Actor from tractor.discovery import _addr @@ -134,6 +137,65 @@ def test_cancelled_transport_send_completes_frame(): trio.run(main) +def test_transport_send_deadline_closes_partial_frame(): + ''' + Bound one shielded frame without exposing a corrupt stream. + + Ordinary cancellation cannot interrupt complete-frame publication. + 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: + def __init__(self) -> None: + self.closed = False + self.wire = bytearray() + + async def send_all( + self, + data: bytes, + ) -> None: + self.wire.extend(data[:2]) + await trio.sleep_forever() + + async def aclose(self) -> None: + self.closed = True + + async def main() -> None: + stream = StalledStream() + transport = object.__new__(MsgpackTransport) + transport.stream = stream + transport._send_lock = trio.StrictFIFOLock() + msg = tractor.msg.Start( + ns=__name__, + func='add_one', + kwargs={'n': 1}, + uid=('root', 'test'), + cid='deadline-send', + ) + + with pytest.raises( + tractor.TransportClosed, + match='frame publication exceeded', + ): + await transport.send( + msg, + send_deadline=1, + ) + + assert stream.closed + assert len(stream.wire) == 2 + assert not transport._send_lock.locked() + + trio.run( + main, + clock=MockClock(autojump_threshold=0), + ) + + def test_cancelled_transport_send_preserves_cancellation(): ''' Prefer sender cancellation when teardown closes the stream. diff --git a/tests/test_to_actor.py b/tests/test_to_actor.py index a163c2b6..28443d06 100644 --- a/tests/test_to_actor.py +++ b/tests/test_to_actor.py @@ -11,12 +11,14 @@ from pathlib import Path import pytest import trio +from trio.testing import MockClock import tractor from tractor import ( RemoteActorError, to_actor, ) from tractor._testing import tractor_test +from tractor._exceptions import ActorTooSlowError from tractor.msg import ptr as msgptr from tractor.msg.ptr import NamespacePath from tractor.to_actor import _api as to_actor_api @@ -254,6 +256,67 @@ async def test_cancel_ack_failure_hard_reaps_child( assert not an._children +def test_cancel_actor_timeout_closes_blocked_send(): + ''' + Thread one absolute cancel deadline into shielded frame publication. + + The cancel RPC's outer timeout cannot penetrate a complete-frame + shield. The fake private RPC applies the forwarded send deadline to + its own shielded wait, then checkpoints into the outer scope. A + bounded `ActorTooSlowError` and the recorded absolute deadline prove + publication and acknowledgement share one timeout budget. + + ''' + class ConnectedChannel: + def __init__(self) -> None: + self._cancel_called = False + self.aid = tractor.msg.Aid( + name='blocked_peer', + uuid='test', + ) + + def connected(self) -> bool: + return True + + async def main() -> None: + channel = ConnectedChannel() + portal = object.__new__(tractor.Portal) + portal._chan = channel + deadlines: list[float] = [] + + async def blocked_cancel( + namespace: str, + function: str, + kwargs: dict[str, object], + cancel_on_startup: bool, + send_deadline: float, + ) -> None: + assert (namespace, function) == ('self', 'cancel') + assert kwargs == {} + assert not cancel_on_startup + deadlines.append(send_deadline) + with trio.CancelScope( + deadline=send_deadline, + shield=True, + ): + await trio.sleep_forever() + await trio.lowlevel.checkpoint_if_cancelled() + + portal._run_from_ns = blocked_cancel + with pytest.raises(ActorTooSlowError): + await portal.cancel_actor( + timeout=1, + raise_on_timeout=True, + ) + + assert deadlines == [1.] + + trio.run( + main, + clock=MockClock(autojump_threshold=0), + ) + + def test_late_child_reap_registration_is_released(): ''' Preserve a nursery-wide reap request across child startup. diff --git a/tractor/ipc/_chan.py b/tractor/ipc/_chan.py index 6eadf381..f07caedc 100644 --- a/tractor/ipc/_chan.py +++ b/tractor/ipc/_chan.py @@ -310,6 +310,7 @@ class Channel: payload: Any, hide_tb: bool = False, + send_deadline: float = float('inf'), ) -> None: ''' @@ -320,6 +321,9 @@ class Channel: expected-graceful cases, normally ephemercal (re/dis)connects. + `send_deadline` is an absolute Trio clock deadline forwarded + only to transports that support bounded frame publication. + ''' __tracebackhide__: bool = hide_tb try: @@ -330,10 +334,17 @@ class Channel: f'{pformat(payload)}\n' ) # assert self._transport # but why typing? - await self._transport.send( - payload, - hide_tb=hide_tb, - ) + if send_deadline == float('inf'): + await self._transport.send( + payload, + hide_tb=hide_tb, + ) + else: + await self._transport.send( + payload, + hide_tb=hide_tb, + send_deadline=send_deadline, + ) except ( BaseException, MsgTypeError, diff --git a/tractor/ipc/_transport.py b/tractor/ipc/_transport.py index b30e0f55..3ebaf61d 100644 --- a/tractor/ipc/_transport.py +++ b/tractor/ipc/_transport.py @@ -439,6 +439,7 @@ class MsgpackTransport(MsgTransport): strict_types: bool = True, hide_tb: bool = True, + send_deadline: float = float('inf'), ) -> None: ''' @@ -447,6 +448,10 @@ class MsgpackTransport(MsgTransport): If `strict_types == True` then a `MsgTypeError` will be raised on any invalid msg type + `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 @@ -513,12 +518,26 @@ class MsgpackTransport(MsgTransport): # the frame is complete, the explicit checkpoint # immediately delivers any pending cancellation. # - # This can delay cancellation while a peer is not - # reading; peer/channel teardown must close the stream - # to unblock a permanently stalled socket write. - with trio.CancelScope(shield=True): + # Ordinary sends may delay cancellation while a peer is + # not reading. Actor-wide cancel requests pass their own + # deadline so this operation can close a stalled stream. + with trio.CancelScope( + deadline=send_deadline, + shield=True, + ) as send_cs: await self.stream.send_all(size + bytes_data) + if send_cs.cancelled_caught: + # This frame may be partial. Destroy the stream + # before releasing `_send_lock` so no later sender + # can append bytes to a corrupted frame. + await trio.aclose_forcefully(self.stream) + await trio.lowlevel.checkpoint_if_cancelled() + raise TransportClosed( + 'IPC frame publication exceeded its ' + f'deadline of {send_deadline!r}' + ) + await trio.lowlevel.checkpoint_if_cancelled() return None diff --git a/tractor/runtime/_portal.py b/tractor/runtime/_portal.py index 7c94d74c..fffaf98b 100644 --- a/tractor/runtime/_portal.py +++ b/tractor/runtime/_portal.py @@ -319,45 +319,61 @@ class Portal: or self.cancel_timeout ) + cancel_deadline: float = ( + trio.current_time() + + + cancel_timeout + ) + # NOTE: Actor-runtime cancellation currently rides the normal + # RPC envelope: + # + # `Start(self.cancel)` -> `StartAck` -> `CancelAck`. + # + # `Actor.start_remote_task()` consumes the `StartAck`, then + # `._run_from_ns()` returns only after `PldRx.recv_pld()` + # decodes the final `CancelAck`. Thus this flag means that ack + # reached this portal after the peer's `Actor.cancel()` routine + # completed; it does not prove the peer OS process has exited. + # A dedicated `Cancel` request msg can eventually replace the + # internal `Start` RPC envelope and its extra `StartAck`. + cancel_ack_received: bool = False try: - # send cancel cmd - might not get response - # XXX: sure would be nice to make this work with - # a proper shield - with trio.move_on_after(cancel_timeout) as cs: + with trio.move_on_at(cancel_deadline) as cs: cs.shield: bool = True - await self.run_from_ns( + await self._run_from_ns( 'self', 'cancel', + kwargs={}, + cancel_on_startup=False, + send_deadline=cancel_deadline, ) - return True + cancel_ack_received = True - # `move_on_after` fired — peer didn't ack within + # Preserve shielded actor teardown, then immediately + # redeliver any cancellation pending from an outer scope. + await trio.lowlevel.checkpoint_if_cancelled() + + # `move_on_at` fired — peer didn't ack within # bounded window. Behaviour depends on # `raise_on_timeout`: - if ( - cs.cancelled_caught - and - raise_on_timeout - ): - raise ActorTooSlowError( - f'Peer {peer_id} did not ack its ' - f'`Actor.cancel()` RPC within bounded wait ' - f'of {cancel_timeout!r}s' - ) + if cs.cancelled_caught: + if raise_on_timeout: + raise ActorTooSlowError( + f'Peer {peer_id} did not ack its ' + f'`Actor.cancel()` RPC within bounded wait ' + f'of {cancel_timeout!r}s' + ) - # legacy fire-and-forget path: log + return False so - # the caller can decide whether to escalate. - # - # NOTE, we also land here in the (unexpected) case where - # the shielded `move_on_after` block exits WITHOUT - # `return True` and WITHOUT the deadline firing — prefer - # a soft `False` over an `assert`-crash mid-teardown. - log.debug( - f'May have failed to cancel peer?\n' - f'\n' - f'c)=?> {peer_id}\n' - ) - return False + # Legacy fire-and-forget callers decide whether to + # escalate the missed acknowledgement themselves. + log.debug( + f'May have failed to cancel peer?\n' + f'\n' + f'c)=?> {peer_id}\n' + ) + return False + + return cancel_ack_received except TransportClosed as tpt_err: ipc_borked_report: str = ( @@ -379,16 +395,24 @@ class Portal: return False + # TODO: Replace actor-runtime cancellation's internal + # `Start -> StartAck -> CancelAck` RPC with a dedicated + # `Cancel -> CancelAck` transaction: + # https://github.com/goodboy/tractor/issues/506 async def _run_from_ns( self, namespace_path: str, function_name: str, kwargs: dict[str, Any], cancel_on_startup: bool = True, + send_deadline: float = float('inf'), ) -> Any: ''' Run a namespace target with local startup policy controls. + `send_deadline` bounds only publication of the `Start` frame; + the caller owns any larger RPC/acknowledgement deadline. + ''' nsf = NamespacePath( f'{namespace_path}:{function_name}' @@ -399,6 +423,7 @@ class Portal: kwargs=kwargs, portal=self, cancel_on_startup=cancel_on_startup, + send_deadline=send_deadline, ) try: return await ctx._pld_rx.recv_pld( diff --git a/tractor/runtime/_runtime.py b/tractor/runtime/_runtime.py index f56ff83a..7cc6371c 100644 --- a/tractor/runtime/_runtime.py +++ b/tractor/runtime/_runtime.py @@ -793,6 +793,11 @@ class Actor: ack_timeout: float = float('inf'), cancel_on_startup: bool = True, + # Optional absolute deadline for publishing this exact `Start` + # frame. Used by actor-wide cancel RPCs whose outer timeout + # cannot penetrate complete-frame transport shielding. + send_deadline: float = float('inf'), + ) -> Context: ''' Send a `'cmd'` msg to a remote actor, which requests the @@ -845,7 +850,13 @@ class Actor: ) start_published: bool = False try: - await chan.send(msg) + if send_deadline == float('inf'): + await chan.send(msg) + else: + await chan.send( + msg, + send_deadline=send_deadline, + ) start_published = True # NOTE wait on first `StartAck` response msg and validate;