goodboy
/
tractor
1
0
Fork 2
Code Issues 5 Pull Requests 6 Packages Projects Releases Wiki Activity

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`))
wkt/to_actor_subpkg
Gud Boi 2026-08-25 14:04:48 -04:00
parent 760abc8268
commit 48844aa4ac
4 changed files with 84 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
tests/ipc/test_each_tpt.py
View File

@ -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

20
tractor/_context.py
View File

@ -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,

31
tractor/ipc/_transport.py
View File

@ -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

5
tractor/runtime/_runtime.py
View File

@ -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: