From 4d07965a2a1aa83cb78ffa56984bb911c438019b Mon Sep 17 00:00:00 2001 From: goodboy Date: Tue, 18 Aug 2026 20:21:10 -0400 Subject: [PATCH] Close interrupted `MsgTransport.send()` streams `SendStream.send_all()` can raise `trio.Cancelled` after writing an arbitrary prefix of the four-byte length header and payload. The peer can no longer distinguish a following msg boundary. Close the stream under a shield before propagating cancellation so callers can not append another msg to an indeterminate byte stream. Caught-during: review remediation Found-via: prospective P2 cancellation review Review: PR #481 (opencode) https://github.com/goodboy/tractor/pull/481#pullrequestreview-4956692120 Prompt-IO: ai/prompt-io/opencode/20260818T193001Z_bf06b4f8_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- .../20260818T193001Z_bf06b4f8_prompt_io.md | 36 ++++++++++++ ...20260818T193001Z_bf06b4f8_prompt_io.raw.md | 19 +++++++ tests/ipc/test_each_tpt.py | 56 +++++++++++++++++++ tractor/ipc/_transport.py | 6 ++ 4 files changed, 117 insertions(+) create mode 100644 ai/prompt-io/opencode/20260818T193001Z_bf06b4f8_prompt_io.md create mode 100644 ai/prompt-io/opencode/20260818T193001Z_bf06b4f8_prompt_io.raw.md diff --git a/ai/prompt-io/opencode/20260818T193001Z_bf06b4f8_prompt_io.md b/ai/prompt-io/opencode/20260818T193001Z_bf06b4f8_prompt_io.md new file mode 100644 index 00000000..111c31d2 --- /dev/null +++ b/ai/prompt-io/opencode/20260818T193001Z_bf06b4f8_prompt_io.md @@ -0,0 +1,36 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +session: ses_3e4c90d3eafeqHEtRYSIHgHhpA +timestamp: 2026-08-18T19:30:01Z +git_ref: bf06b4f8 +scope: code +substantive: true +raw_file: 20260818T193001Z_bf06b4f8_prompt_io.raw.md +--- + +## Prompt + +Fix cancellation during an indeterminate transport write without +allowing a second RPC msg to corrupt the stream. Keep the transport +fix separate from the higher-level `to_actor` remediation and explain +the length-prefixed protocol rather than using unexplained "framed +send" terminology. + +## Response summary + +Close a msgpack transport stream when cancellation interrupts its +length-prefixed `send_all()` operation. The stream can no longer be +safely reused because Trio cannot report how many bytes were written. + +## Files changed + +- `tractor/ipc/_transport.py` - close an interrupted send stream. +- `tests/ipc/test_each_tpt.py` - cover cancellation during the write. + +## Human edits + +The human required this transport edge-case fix to land as its own +behavioral commit with a detailed message. During staged review, the +human also rejected the unexplained "framed send" wording and asked +for terminology tied directly to the actual transport operation. diff --git a/ai/prompt-io/opencode/20260818T193001Z_bf06b4f8_prompt_io.raw.md b/ai/prompt-io/opencode/20260818T193001Z_bf06b4f8_prompt_io.raw.md new file mode 100644 index 00000000..81389498 --- /dev/null +++ b/ai/prompt-io/opencode/20260818T193001Z_bf06b4f8_prompt_io.raw.md @@ -0,0 +1,19 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +timestamp: 2026-08-18T19:30:01Z +git_ref: bf06b4f8 +diff_cmd: git diff HEAD~1..HEAD +--- + +Prospective review found that cancellation can interrupt +`MsgpackTransport.send()` after `send_all()` writes only part of its +length-prefixed msg. Sending a cancellation request afterward can +append another msg to the indeterminate stream and desynchronize the +peer decoder. + +> `git diff HEAD~1..HEAD -- tractor/ipc/_transport.py tests/ipc/test_each_tpt.py` + +Close the stream under a cancellation shield when `send_all()` is +cancelled. Cover the behavior with a fake stream that checkpoints +inside the write and records forced closure. diff --git a/tests/ipc/test_each_tpt.py b/tests/ipc/test_each_tpt.py index 512300a6..7a7715f4 100644 --- a/tests/ipc/test_each_tpt.py +++ b/tests/ipc/test_each_tpt.py @@ -17,9 +17,65 @@ import trio import tractor from tractor import Actor from tractor.discovery import _addr +from tractor.ipc._transport import MsgpackTransport from tractor.runtime import _state + +def test_cancelled_transport_send_closes_stream(): + ''' + Discard a transport after cancellation interrupts a framed send. + + Trio's `SendStream.send_all()` may write an arbitrary frame prefix + before raising `Cancelled`. Sending another IPC msg afterward + would append a second frame and desynchronize the peer decoder. + The fake stream checkpoints after recording send entry; cancelling + its nursery deterministically interrupts that unknown-publication + window. Its close assertion proves the transport is made unusable + before another framed msg can be attempted. + + ''' + class PartialSendStream: + def __init__(self) -> None: + self.send_entered = trio.Event() + self.closed = False + + async def send_all( + self, + data: bytes, + ) -> None: + assert data + self.send_entered.set() + await trio.sleep_forever() + + async def aclose(self) -> None: + self.closed = True + + async def main() -> None: + stream = PartialSendStream() + transport = object.__new__(MsgpackTransport) + transport.stream = stream + transport._send_lock = trio.StrictFIFOLock() + + async with trio.open_nursery() as tn: + tn.start_soon( + transport.send, + tractor.msg.Start( + ns=__name__, + func='add_one', + kwargs={'n': 1}, + uid=('root', 'test'), + cid='partial-send', + ), + ) + await stream.send_entered.wait() + tn.cancel_scope.cancel() + + assert stream.closed + + trio.run(main) + + @pytest.fixture def bindspace_dir_str() -> str: diff --git a/tractor/ipc/_transport.py b/tractor/ipc/_transport.py index dfa36696..c65c9d4a 100644 --- a/tractor/ipc/_transport.py +++ b/tractor/ipc/_transport.py @@ -499,6 +499,12 @@ class MsgpackTransport(MsgTransport): size: bytes = struct.pack("