diff --git a/ai/prompt-io/opencode/20260812T030608Z_1095e7f7_prompt_io.md b/ai/prompt-io/opencode/20260812T030608Z_1095e7f7_prompt_io.md new file mode 100644 index 00000000..fcaae8e7 --- /dev/null +++ b/ai/prompt-io/opencode/20260812T030608Z_1095e7f7_prompt_io.md @@ -0,0 +1,33 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +session: ses_0799212ebffe42arY96czXn89F +timestamp: 2026-08-12T03:06:08Z +git_ref: 1095e7f7 +scope: code +substantive: true +raw_file: 20260812T030608Z_1095e7f7_prompt_io.raw.md +--- + +## Prompt + +For Tractor PR 490 item two, make shared underlying receive failures wake and +terminate every broadcast subscriber without losing retained values. Review, +verify and return a complete commit plan before continuing. + +## Response summary + +Published ordinary receive failures as terminal broadcast state, introduced +a public chained peer exception, kept control-flow exits transient while +waking peers, documented the contract, and added deterministic regressions. + +## Files changed + +- `tractor/trionics/_broadcast.py` - terminal failure and peer wake protocol. +- `tractor/trionics/__init__.py` - public peer exception export. +- `docs/api/trionics.rst` - failure-delivery API contract. +- `tests/test_task_broadcasting.py` - terminal and transient failure tests. + +## Human edits + +None - generated output follows the requested second iterative item. diff --git a/ai/prompt-io/opencode/20260812T030608Z_1095e7f7_prompt_io.raw.md b/ai/prompt-io/opencode/20260812T030608Z_1095e7f7_prompt_io.raw.md new file mode 100644 index 00000000..44c6ef19 --- /dev/null +++ b/ai/prompt-io/opencode/20260812T030608Z_1095e7f7_prompt_io.raw.md @@ -0,0 +1,49 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +timestamp: 2026-08-12T03:06:08Z +git_ref: 1095e7f7 +diff_cmd: git diff HEAD~1..HEAD +--- + +The user requested the second iterative refinement for Tractor PR 490: ensure +non-EOC failures from a shared underlying broadcast receiver do not leave peer +subscribers blocked forever, then stop for a complete commit plan. + +> `git diff HEAD~1..HEAD -- tractor/trionics/_broadcast.py` + +Added shared terminal failure publication for ordinary `Exception` values. +The receive owner gets the original exception; peers may drain retained +values and then get a fresh `BroadcastReceiveError` chained from the original. +Late subscribers observe the same terminal state without retrying the failed +underlying receiver. Process-control and cancellation-like `BaseException` +values wake peers but are re-raised without becoming durable channel state. + +> `git diff HEAD~1..HEAD -- tractor/trionics/__init__.py` + +Exported `BroadcastReceiveError` as the public peer-delivery exception. + +> `git diff HEAD~1..HEAD -- docs/api/trionics.rst` + +Documented `BroadcastReceiveError` and the owner-versus-peer delivery +contract, including retained-value draining and late subscribers. + +> `git diff HEAD~1..HEAD -- tests/test_task_broadcasting.py` + +Added deterministic bounded regressions. One scripts a successful receive +followed by `RuntimeError`, proving the root drains retained data, all current +and late receivers observe terminal failure, and the source is not retried. +The second scripts a custom `BaseException`, proving peers wake and take over +the next source receive without retaining control-flow state. + +Verification output: + +```text +............. [100%] +13 passed in 5.62s +``` + +Python compilation and `git diff --check` passed. Iterative adversarial review +drove independent peer exception wrappers, ordinary-versus-control-flow +classification, bounded test completion, public docs, and the final catch-all +peer wake. Final review found no issues. diff --git a/docs/api/trionics.rst b/docs/api/trionics.rst index a92a290a..09827b44 100644 --- a/docs/api/trionics.rst +++ b/docs/api/trionics.rst @@ -40,6 +40,9 @@ Broadcast fan-out .. autoexception:: Lagged :show-inheritance: +.. autoexception:: BroadcastReceiveError + :show-inheritance: + A single-producer, many-consumer broadcast layer over any ``trio``-style receive channel: non-lossy for the *fastest* consumer while slower consumers raise :class:`Lagged` (a @@ -48,6 +51,13 @@ internal ring. This is exactly the machinery behind :meth:`tractor.MsgStream.subscribe` — see ``examples/streaming_broadcast_fanout.py``. +If the shared underlying receiver raises an ordinary exception, the +subscriber which owned that receive gets the original failure. +Waiting peers drain their retained values and then raise +:class:`BroadcastReceiveError`, with the original failure available +as ``__cause__``. Later subscribers observe the same terminal state +without retrying the failed underlying receiver. + ExceptionGroup helpers ---------------------- diff --git a/tests/test_task_broadcasting.py b/tests/test_task_broadcasting.py index b5d0d8aa..73cb9fd2 100644 --- a/tests/test_task_broadcasting.py +++ b/tests/test_task_broadcasting.py @@ -17,6 +17,7 @@ from trio.lowlevel import current_task import tractor from tractor.trionics import ( broadcast_receiver, + BroadcastReceiveError, Lagged, collapse_eg, ) @@ -405,6 +406,188 @@ def test_broadcast_statistics_report_queued_counts() -> None: trio.run(main) +def test_underlying_receive_failure_wakes_all_subscribers() -> None: + ''' + A shared receive failure must terminate every broadcast receiver. + + Previously, only `EndOfChannel` and receiver cancellation woke + peer tasks waiting on `BroadcastState.recv_ready`. If the shared + underlying receiver raised another error, its owner propagated + the failure and cleared the event while every peer remained + blocked forever. + + Script one successful receive followed by a controlled + `RuntimeError`. Let a fast child own both underlying receives + while the root first drains its retained value and then waits on + the child's second receive. Release the failure only after both + tasks have reached those positions. Both exact errors prove the + peer was awakened without losing buffered data. A later + subscriber proves the terminal failure remains published for new + receivers instead of retrying the failed underlying channel. + + ''' + class FailingReceiver: + ''' + Return one value, then fail after deterministic release. + + ''' + def __init__(self) -> None: + self.calls: int = 0 + self.failure_started = trio.Event() + self.release_failure = trio.Event() + + async def receive(self) -> int: + ''' + Drive the scripted success-then-failure sequence. + + ''' + self.calls += 1 + if self.calls == 1: + return 1 + + self.failure_started.set() + await self.release_failure.wait() + raise RuntimeError('underlying receive failed') + + async def main() -> None: + source = FailingReceiver() + brx = broadcast_receiver(source, 3) + child_error: list[RuntimeError] = [] + root_error: list[BroadcastReceiveError] = [] + late_error: list[BroadcastReceiveError] = [] + root_drained = trio.Event() + + async def receive_child() -> None: + async with brx.subscribe() as child: + assert await child.receive() == 1 + try: + await child.receive() + except RuntimeError as exc: + child_error.append(exc) + + async def receive_root() -> None: + assert await brx.receive() == 1 + root_drained.set() + try: + await brx.receive() + except BroadcastReceiveError as exc: + root_error.append(exc) + + with trio.fail_after(1): + async with trio.open_nursery() as nursery: + nursery.start_soon(receive_child) + await source.failure_started.wait() + + nursery.start_soon(receive_root) + await root_drained.wait() + + source.release_failure.set() + + assert source.calls == 2 + assert [str(exc) for exc in child_error] == [ + 'underlying receive failed', + ] + assert [str(exc) for exc in root_error] == [ + 'Shared broadcast receiver failed', + ] + assert child_error[0] is not root_error[0] + assert root_error[0].__cause__ is child_error[0] + + async with brx.subscribe() as late: + with pytest.raises( + BroadcastReceiveError, + match='Shared broadcast receiver failed', + ) as exc_info: + await late.receive() + late_error.append(exc_info.value) + assert late_error[0] is not child_error[0] + assert late_error[0] is not root_error[0] + assert late_error[0].__cause__ is child_error[0] + assert source.calls == 2 + + trio.run(main) + + +def test_control_flow_exit_wakes_broadcast_peer() -> None: + ''' + Non-terminal control flow must wake peers without being retained. + + Process-control and cancellation-like `BaseException` values + should remain local to the task which receives them, but the old + owner still has to wake subscribers blocked on its shared event. + Make one child own a controlled `BaseException` receive while the + root waits behind it. After release, prove the child gets that + exact exit and the root takes ownership of the next underlying + receive instead of hanging or replaying the control-flow event. + + ''' + class ReceiveExit(BaseException): + ''' + Model a non-terminal process-control receive exit. + + ''' + + class ControlFlowReceiver: + ''' + Raise one controlled exit, then return a value. + + ''' + def __init__(self) -> None: + self.calls: int = 0 + self.exit_started = trio.Event() + self.release_exit = trio.Event() + + async def receive(self) -> int: + ''' + Drive the scripted control-flow-then-value sequence. + + ''' + self.calls += 1 + if self.calls == 1: + self.exit_started.set() + await self.release_exit.wait() + raise ReceiveExit + + return 2 + + async def main() -> None: + source = ControlFlowReceiver() + brx = broadcast_receiver(source, 3) + child_exit: list[ReceiveExit] = [] + root_value: list[int] = [] + + async def receive_child() -> None: + async with brx.subscribe() as child: + try: + await child.receive() + except ReceiveExit as exc: + child_exit.append(exc) + + 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_child) + await source.exit_started.wait() + nursery.start_soon(receive_root) + + while True: + _, event = brx._state.recv_ready + if event.statistics().tasks_waiting: + break + await trio.lowlevel.checkpoint() + + source.release_exit.set() + + assert len(child_exit) == 1 + assert root_value == [2] + assert source.calls == 2 + assert brx._state.receive_exc is None + + trio.run(main) + + def test_ensure_slow_consumers_lag_out( reg_addr, start_method, diff --git a/tractor/trionics/__init__.py b/tractor/trionics/__init__.py index 6cf57b7b..e36bf599 100644 --- a/tractor/trionics/__init__.py +++ b/tractor/trionics/__init__.py @@ -26,6 +26,7 @@ from ._mngrs import ( from ._broadcast import ( AsyncReceiver as AsyncReceiver, broadcast_receiver as broadcast_receiver, + BroadcastReceiveError as BroadcastReceiveError, BroadcastReceiver as BroadcastReceiver, Lagged as Lagged, ) diff --git a/tractor/trionics/_broadcast.py b/tractor/trionics/_broadcast.py index 24aeb24f..990b84b5 100644 --- a/tractor/trionics/_broadcast.py +++ b/tractor/trionics/_broadcast.py @@ -100,6 +100,13 @@ class Lagged(trio.TooSlowError): ''' +class BroadcastReceiveError(Exception): + ''' + A shared underlying receiver failed in another subscriber task. + + ''' + + class BroadcastState(Struct): ''' Common state to all receivers of a broadcast. @@ -122,6 +129,11 @@ class BroadcastState(Struct): # For now, this is solely for testing/debugging purposes. eoc: bool = False + # Any non-EOC failure from the shared underlying receiver is + # terminal for every subscriber. Retained values remain readable + # before this failure is replayed at each receiver's boundary. + receive_exc: Exception | None = None + # If the broadcaster was cancelled, we might as well track it cancelled: dict[int, Task] = {} @@ -267,6 +279,15 @@ class BroadcastReceiver(ReceiveChannel): state.subs[key] -= 1 return value + receive_exc = state.receive_exc + if receive_exc is not None: + # Re-raising one shared exception mutates its traceback on + # every delivery. Give each receiver a stable wrapper while + # retaining the original failure as its cause. + raise BroadcastReceiveError( + 'Shared broadcast receiver failed' + ) from receive_exc + raise trio.WouldBlock async def _receive_from_underlying( @@ -339,6 +360,24 @@ class BroadcastReceiver(ReceiveChannel): event.set() raise + except Exception as receive_exc: + # The underlying receiver is shared by every subscriber, + # so any non-EOC failure terminates the entire broadcast. + # Publish it before waking peers so they can drain their + # retained values and then observe the same failure. + state.receive_exc = receive_exc + if event.statistics().tasks_waiting: + event.set() + raise + + except BaseException: + # Process-control and cancellation-like exceptions must + # not become durable broadcast state, but peers still + # need waking before `recv_ready` is cleared. + if event.statistics().tasks_waiting: + event.set() + raise + finally: # Reset receiver waiter task event for next blocking condition. # this MUST be reset even if the above ``.recv()`` call