diff --git a/ai/prompt-io/opencode/20260812T150027Z_c2a6ccef_prompt_io.md b/ai/prompt-io/opencode/20260812T150027Z_c2a6ccef_prompt_io.md new file mode 100644 index 00000000..1e9784c5 --- /dev/null +++ b/ai/prompt-io/opencode/20260812T150027Z_c2a6ccef_prompt_io.md @@ -0,0 +1,33 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +session: ses_0799212ebffe42arY96czXn89F +timestamp: 2026-08-12T15:00:27Z +git_ref: c2a6ccef +scope: code +substantive: true +raw_file: 20260812T150027Z_c2a6ccef_prompt_io.raw.md +--- + +## Prompt + +For Tractor PR 490 item three, prevent subscriber closure from waking another +receiver's shared event or stranding peers. Preserve close-safe source-read +ownership handoff, then review, verify and return a complete commit plan. + +## Response summary + +Introduced receiver-local wait/source cancellation, owner-specific handoff, +and close precedence over shielded source values/errors/EOC. Clarified that +private scope cancellation means explicit close while outer task cancellation +remains `trio.Cancelled` for both source-read and peer-wait scopes, with +deterministic receiver-close regressions. + +## Files changed + +- `tractor/trionics/_broadcast.py` - receiver-local close and ownership scopes. +- `tests/test_task_broadcasting.py` - peer close and owner handoff regressions. + +## Human edits + +None - generated output follows the requested third iterative item. diff --git a/ai/prompt-io/opencode/20260812T150027Z_c2a6ccef_prompt_io.raw.md b/ai/prompt-io/opencode/20260812T150027Z_c2a6ccef_prompt_io.raw.md new file mode 100644 index 00000000..4e9e6d83 --- /dev/null +++ b/ai/prompt-io/opencode/20260812T150027Z_c2a6ccef_prompt_io.raw.md @@ -0,0 +1,51 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +timestamp: 2026-08-12T15:00:27Z +git_ref: c2a6ccef +diff_cmd: git diff HEAD~1..HEAD +--- + +The user requested the third iterative refinement for Tractor PR 490: closing +one broadcast subscriber must not set another receiver owner's shared event +and create a runnable hot loop, then stop for a complete commit plan. + +> `git diff HEAD~1..HEAD -- tractor/trionics/_broadcast.py` + +Added receiver-local wait cancellation and source-read ownership scopes. +Closing a non-owner waiting behind another source reader cancels only that +receiver's private wait and maps it to `ClosedResourceError`; the shared event +remains untouched. Closing the active source owner cancels only its private +source-read scope, wakes peers after cleanup, and lets one peer take ownership. + +Source outcomes are captured inside the owner scope and classified only after +checking close/cancel state. A cancellation-shielding source therefore cannot +publish a returned value, ordinary error, or EOC after its owner was closed. +The private scope's `cancel_called` bit is asserted to imply receiver closure; +outer task cancellation remains `trio.Cancelled` and is not translated into +`ClosedResourceError`. Owner-key comments document that only the receiver +identified by `recv_ready[0]` may cancel the shared source-read scope. The +same explicit-close invariant is enforced symmetrically for private peer-wait +scope cancellation. + +> `git diff HEAD~1..HEAD -- tests/test_task_broadcasting.py` + +Added deterministic bounded regressions for both close positions. The +non-owner test places two peers behind an active source read, closes one and +proves only that peer exits while the shared event stays unset. The owner test +closes a source owner whose receive shields cancellation and parameterizes a +returned value, `RuntimeError`, and `EndOfChannel`; each discarded outcome +hands the next source receive to the waiting root without terminal-state or +EOC publication. + +Verification output: + +```text +................. [100%] +17 passed in 5.84s +``` + +Python compilation and `git diff --check` passed. Iterative adversarial review +caught a waiting non-owner hang, cancellation-shielded source returns, and +shielded source exceptions. All were fixed. Final review found no actionable +issues. diff --git a/tests/test_task_broadcasting.py b/tests/test_task_broadcasting.py index 73cb9fd2..8d331f31 100644 --- a/tests/test_task_broadcasting.py +++ b/tests/test_task_broadcasting.py @@ -402,6 +402,7 @@ def test_broadcast_statistics_report_queued_counts() -> None: child.key: 0, } assert stats['tasks_waiting'] == 0 + state.recv_ready = None trio.run(main) @@ -588,6 +589,156 @@ def test_control_flow_exit_wakes_broadcast_peer() -> None: trio.run(main) +def test_closing_non_owner_preserves_source_wait() -> None: + ''' + Closing one subscriber must not wake another receiver's peers. + + `BroadcastReceiver.aclose()` previously set the one shared + `BroadcastState.recv_ready` event even when a different receiver + owned the source read. Waiting peers then repeatedly awaited an + already-set event until the source produced another value, + creating a runnable hot loop on idle streams. + + Block one child in the source receive, then place both the root + and a closing child behind its event. Close only that waiting + child and prove it gets `ClosedResourceError` without setting the + shared event. Both remaining receivers must still get the same + value after the source is released. + + ''' + async def main() -> None: + tx, rx = trio.open_memory_channel(1) + brx = broadcast_receiver(rx, 3) + owner_value: list[int] = [] + root_value: list[int] = [] + closing_closed = trio.Event() + + async with ( + brx.subscribe() as owner, + brx.subscribe() as closing, + ): + async def receive_owner() -> None: + owner_value.append(await owner.receive()) + + async def receive_root() -> None: + root_value.append(await brx.receive()) + + async def receive_closing() -> None: + with pytest.raises(trio.ClosedResourceError): + await closing.receive() + closing_closed.set() + + with trio.fail_after(1): + async with trio.open_nursery() as nursery: + nursery.start_soon(receive_owner) + while brx._state.recv_ready is None: + await trio.lowlevel.checkpoint() + + nursery.start_soon(receive_root) + nursery.start_soon(receive_closing) + _, event = brx._state.recv_ready + while event.statistics().tasks_waiting < 2: + await trio.lowlevel.checkpoint() + + await closing.aclose() + await closing_closed.wait() + assert not event.is_set() + await tx.send(1) + + assert owner_value == [1] + assert root_value == [1] + + trio.run(main) + + +@pytest.mark.parametrize( + 'first_outcome', + [ + 1, + RuntimeError('discarded source error'), + trio.EndOfChannel(), + ], +) +def test_closing_source_owner_hands_read_to_peer( + first_outcome: int|Exception, +) -> None: + ''' + Closing the source-read owner must transfer ownership to a peer. + + Merely suppressing the old shared-event wake would leave peers + blocked behind an externally closed receiver that still owned an + idle source read. Script a first receive which blocks until its + private scope is cancelled and a second which returns immediately. + Close that owner only after the root is waiting behind it. Cover + a shielded value, ordinary error and EOC from the cancelled source + read. The owner must always get `ClosedResourceError`, while the + awakened root takes the second source read without publishing the + discarded source outcome. + + ''' + class HandoffReceiver: + ''' + Block the first source read and satisfy the second. + + ''' + def __init__(self) -> None: + self.calls: int = 0 + self.first_started = trio.Event() + self.release_first = trio.Event() + + async def receive(self) -> int: + ''' + Drive one cancelled read followed by one value. + + ''' + self.calls += 1 + if self.calls == 1: + self.first_started.set() + with trio.CancelScope(shield=True): + await self.release_first.wait() + if isinstance(first_outcome, BaseException): + raise first_outcome + return first_outcome + + return 2 + + async def main() -> None: + source = HandoffReceiver() + brx = broadcast_receiver(source, 3) + owner_closed = trio.Event() + root_value: list[int] = [] + + async with brx.subscribe() as owner: + async def receive_owner() -> None: + with pytest.raises(trio.ClosedResourceError): + await owner.receive() + owner_closed.set() + + async def receive_root() -> None: + root_value.append(await brx.receive()) + + with trio.fail_after(1): + async with trio.open_nursery() as nursery: + nursery.start_soon(receive_owner) + await source.first_started.wait() + nursery.start_soon(receive_root) + + _, event = brx._state.recv_ready + while not event.statistics().tasks_waiting: + await trio.lowlevel.checkpoint() + + await owner.aclose() + source.release_first.set() + await owner_closed.wait() + + assert source.calls == 2 + assert root_value == [2] + assert brx._state.receive_exc is None + assert not brx._state.eoc + + trio.run(main) + + def test_ensure_slow_consumers_lag_out( reg_addr, start_method, @@ -729,6 +880,7 @@ def test_first_recver_is_cancelled(): async with brx.subscribe() as bc: async for value in bc: print(value) + assert cs.cancelled_caught async def cancel_and_send(): await trio.sleep(0.2) diff --git a/tractor/trionics/_broadcast.py b/tractor/trionics/_broadcast.py index 990b84b5..5ac21922 100644 --- a/tractor/trionics/_broadcast.py +++ b/tractor/trionics/_broadcast.py @@ -107,6 +107,13 @@ class BroadcastReceiveError(Exception): ''' +class _BroadcastReceiverClosed(Exception): + ''' + An active receiver was closed while owning the source read. + + ''' + + class BroadcastState(Struct): ''' Common state to all receivers of a broadcast. @@ -122,6 +129,7 @@ class BroadcastState(Struct): # broadcast event to wake up all sleeping consumer tasks # on a newly produced value from the sender. recv_ready: tuple[int, trio.Event]|None = None + recv_scope: trio.CancelScope|None = None # if a ``trio.EndOfChannel`` is received on any # consumer all consumers should be placed in this state @@ -209,6 +217,7 @@ class BroadcastReceiver(ReceiveChannel): self._recv = receive_afunc or rx_chan.receive self._closed: bool = False self._raise_on_lag = raise_on_lag + self._wait_scope: trio.CancelScope|None = None def receive_nowait( self, @@ -301,14 +310,34 @@ class BroadcastReceiver(ReceiveChannel): raise trio.ClosedResourceError event = trio.Event() + recv_scope = trio.CancelScope() assert state.recv_ready is None + assert state.recv_scope is None state.recv_ready = key, event + state.recv_scope = recv_scope try: # if we're cancelled here it should be # fine to bail without affecting any other consumers # right? - value = await self._recv() + receive_exc: BaseException|None = None + with recv_scope: + try: + value = await self._recv() + except BaseException as exc: + receive_exc = exc + + # Only this receiver's `aclose()` cancels its private + # source-read scope, and it marks the receiver closed + # first without a checkpoint. Outer task cancellation does + # not set `recv_scope.cancel_called`; it remains a real + # `trio.Cancelled` and follows the handler below. + if recv_scope.cancel_called: + assert self._closed + if self._closed: + raise _BroadcastReceiverClosed + if receive_exc is not None: + raise receive_exc # items with lower indices are "newer" # NOTE: ``collections.deque`` implicitly takes care of @@ -348,6 +377,14 @@ class BroadcastReceiver(ReceiveChannel): event.set() raise + except _BroadcastReceiverClosed: + # `aclose()` cancelled this receiver's source-read scope. + # Wake peers so one of them can take ownership after this + # task clears `recv_ready` in `finally`. + if event.statistics().tasks_waiting: + event.set() + raise trio.ClosedResourceError + except ( trio.Cancelled, ): @@ -384,6 +421,7 @@ class BroadcastReceiver(ReceiveChannel): # was cancelled to avoid the next consumer from blocking on # an event that won't be set! state.recv_ready = None + state.recv_scope = None async def receive(self) -> ReceiveType: key = self.key @@ -411,7 +449,23 @@ class BroadcastReceiver(ReceiveChannel): # seq = state.subs[key] # assert seq == -1 # sanity _, ev = state.recv_ready - await ev.wait() + wait_scope = trio.CancelScope() + self._wait_scope = wait_scope + try: + with wait_scope: + await ev.wait() + + # As with `recv_scope`, only this receiver's + # `aclose()` cancels its private peer-wait scope + # after marking the receiver closed. Outer task + # cancellation remains `trio.Cancelled`. + if wait_scope.cancel_called: + assert self._closed + if self._closed: + raise trio.ClosedResourceError + finally: + self._wait_scope = None + try: return self.receive_nowait( _key=key, @@ -489,18 +543,34 @@ class BroadcastReceiver(ReceiveChannel): if self._closed: return - # if there are sleeping consumers wake - # them on closure. - rr = self._state.recv_ready - if rr: - _, event = rr - event.set() - # XXX: leaving it like this consumers can still get values # up to the last received that still reside in the queue. - self._state.subs.pop(self.key) + state = self._state + state.subs.pop(self.key) self._closed = True + # A non-owner close must not wake peers waiting behind some + # other receiver's source read. If this receiver owns that + # read, cancel only its private scope; the owner task wakes + # peers after cancellation is delivered and state is ready + # for a clean ownership handoff. + rr = state.recv_ready + if ( + rr is not None + + # `recv_ready[0]` identifies the receiver which currently + # owns the one shared source read. Only that receiver may + # cancel `BroadcastState.recv_scope`; closing any other + # subscriber must not disturb the owner or its peer tasks. + and + rr[0] == self.key + ): + recv_scope = state.recv_scope + assert recv_scope is not None + recv_scope.cancel() + elif (wait_scope := self._wait_scope) is not None: + wait_scope.cancel() + def broadcast_receiver(