Wake broadcast peers on shared receive failures

Only `EndOfChannel` and direct cancellation woke tasks waiting
behind the subscriber which owned the underlying receive. Any other
failure cleared `BroadcastState.recv_ready` while peers remained
blocked on its unreachable event.

Publish ordinary receive exceptions as terminal broadcast state.
The owner keeps the original failure while peers drain retained
values and then raise `BroadcastReceiveError` from that cause. Also
wake peers on process-control exits without retaining them as state.

Document the public owner/peer contract and cover current, late and
control-flow subscribers with deterministic bounded regressions.

Prompt-IO: ai/prompt-io/opencode/20260812T030608Z_1095e7f7_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))

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.

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.

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

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,

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

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