From 48844aa4ac3d34755c212549995d7d37044da003 Mon Sep 17 00:00:00 2001 From: goodboy Date: Tue, 25 Aug 2026 14:04:48 -0400 Subject: [PATCH] Clarify bounded IPC frame publication Older review threads left partial-frame scheduling, send-lock ownership, deadline-only stream destruction and cancellation precedence unclear in both transport tests and source comments. Document exact sender/parent ordering, name send events explicitly and explain why stream alignment controls sibling reuse. Clarify private context controls, overrun relay failure and transport shield boundaries. Review: PR #481 (goodboy) https://github.com/goodboy/tractor/pull/481#pullrequestreview-5012942328 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/ipc/test_each_tpt.py | 85 ++++++++++++++++++++++--------------- tractor/_context.py | 20 ++++++--- tractor/ipc/_transport.py | 31 ++++++++------ tractor/runtime/_runtime.py | 5 ++- 4 files changed, 84 insertions(+), 57 deletions(-) diff --git a/tests/ipc/test_each_tpt.py b/tests/ipc/test_each_tpt.py index 73debc62..9c2f7dfc 100644 --- a/tests/ipc/test_each_tpt.py +++ b/tests/ipc/test_each_tpt.py @@ -33,18 +33,19 @@ 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. 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. + 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. ''' class PartialSendStream: def __init__(self) -> None: - self.send_entered = trio.Event() - self.release = trio.Event() + self.send_all_entered = trio.Event() + self.send_all_release = trio.Event() self.closed = False self.wire = bytearray() @@ -55,8 +56,8 @@ def test_cancelled_transport_send_completes_frame(): assert data if not self.wire: self.wire.extend(data[:2]) - self.send_entered.set() - await self.release.wait() + self.send_all_entered.set() + await self.send_all_release.wait() self.wire.extend(data[2:]) else: self.wire.extend(data) @@ -115,21 +116,29 @@ def test_cancelled_transport_send_completes_frame(): tn.start_soon( send_first, ) - await stream.send_entered.wait() + await stream.send_all_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() - stream.release.set() + # Let the underlying frame write finish after the parent + # has requested sender cancellation. + stream.send_all_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() @@ -139,14 +148,15 @@ def test_cancelled_transport_send_completes_frame(): def test_transport_send_deadline_closes_partial_frame(): ''' - Bound one shielded frame without exposing a corrupt stream. + Destroy a stalled partial frame before another sender can append. 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. + 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. ''' class StalledStream: @@ -186,9 +196,9 @@ def test_transport_send_deadline_closes_partial_frame(): send_deadline=1, ) - assert stream.closed - assert len(stream.wire) == 2 - assert not transport._send_lock.locked() + 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 trio.run( main, @@ -200,30 +210,35 @@ def test_cancelled_transport_send_preserves_cancellation(): ''' Prefer sender cancellation when teardown closes the stream. - `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. + `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. The fake stream blocks inside the shield until the test cancels the - sender, then raises the same close error seen on macOS UDS. Observing - `CancelScope.cancelled_caught` proves cancellation wins once the - shield unwinds. + 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. ''' class ClosingStream: def __init__(self) -> None: - self.send_entered = trio.Event() - self.release = trio.Event() + self.send_all_entered = trio.Event() + self.send_all_release = trio.Event() async def send_all( self, data: bytes, ) -> None: assert data - self.send_entered.set() - await self.release.wait() + 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. raise trio.ClosedResourceError( 'this socket was already closed' ) @@ -256,12 +271,12 @@ def test_cancelled_transport_send_preserves_cancellation(): async with trio.open_nursery() as tn: tn.start_soon(send) - await stream.send_entered.wait() + await stream.send_all_entered.wait() sender_scopes[0].cancel() await wait_all_tasks_blocked() assert not sender_done.is_set() - stream.release.set() + stream.send_all_release.set() await sender_done.wait() assert cancelled_caught diff --git a/tractor/_context.py b/tractor/_context.py index 0b383d92..6a6ad087 100644 --- a/tractor/_context.py +++ b/tractor/_context.py @@ -1113,6 +1113,10 @@ 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', @@ -1955,6 +1959,9 @@ 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 @@ -2027,12 +2034,13 @@ class Context: await chan.send(err_msg) return True - # 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. + # 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. except ( TransportClosed, trio.BrokenResourceError, diff --git a/tractor/ipc/_transport.py b/tractor/ipc/_transport.py index 2ebd76b8..018b70c4 100644 --- a/tractor/ipc/_transport.py +++ b/tractor/ipc/_transport.py @@ -448,9 +448,11 @@ 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. + `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`. ''' __tracebackhide__: bool = hide_tb @@ -505,18 +507,17 @@ class MsgpackTransport(MsgTransport): try: # Every IPC msg is length-prefixed and all contexts # on this actor pair share one transport stream. If - # 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 send deadline interrupts `send_all()`, an unknown + # frame prefix may already be on the wire; allowing the + # next sender to append would corrupt framing. # - # 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. + # 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. # # Ordinary sends may delay cancellation while the remote # peer actor is not reading. Bounded actor/context cancel @@ -539,6 +540,8 @@ 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 diff --git a/tractor/runtime/_runtime.py b/tractor/runtime/_runtime.py index eb3eea57..27e861f8 100644 --- a/tractor/runtime/_runtime.py +++ b/tractor/runtime/_runtime.py @@ -794,8 +794,9 @@ class Actor: 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. + # frame. Used by bounded actor/context cancel RPCs whose outer + # timeout cannot penetrate `Channel.send()` forwarding into the + # shield in `MsgpackTransport.send()`. send_deadline: float = float('inf'), ) -> Context: