Harden broadcast fan-out ownership and failures - #490
Open
goodboy wants to merge 6 commits into
Open
Conversation
`BroadcastReceiver.receive_nowait()` treated `seq` as a deque index but subtracted `BroadcastState.maxlen` without counting the first invalid index. A one-slot queue thus claimed it dropped zero values after its subscriber missed one. Include that first displaced value in the count. Preserve the existing Tokio-style reset to the oldest retained item. Also, cover exact loss reporting and recovery for one- and three-slot retention windows. Prompt-IO: ai/prompt-io/opencode/20260811T233833Z_7cbd64ee_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
`BroadcastState.subs` stores each receiver's next unread deque index, but `.statistics()` exposed that index as a queue length. A caught-up receiver looked correct by accident while every queued count was one short. Convert cursors to retained, receivable counts and clamp lagged receivers to the current queue length. Also avoid deprecated `trio.Event` truthiness when reporting waiter counts. Cover caught-up, queued, lagged and real-event states using actual broadcast sends and receives. Prompt-IO: ai/prompt-io/opencode/20260812T012324Z_06c4af17_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
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`))
Closing any subscriber set the shared `recv_ready` event, even when another receiver owned the source read. Waiting peers then looped until an idle source produced another value. Give each receiver private source-read and peer-wait cancellation scopes. Closing a waiting peer interrupts only that peer. Closing the source owner discards post-close source outcomes and then wakes peers for a clean ownership handoff. Keep outer task cancellation as `trio.Cancelled`; only explicit receiver close maps either private scope's cancellation to `ClosedResourceError`. Assert that scope cancellation implies the receiver is closed and document the source-owner key check. Prompt-IO: ai/prompt-io/opencode/20260812T150027Z_c2a6ccef_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
`MsgStream.subscribe()` and `LinkedTaskChannel.subscribe()` omitted `BroadcastReceiver.raise_on_lag`, forcing downstream consumers to mutate a private receiver attribute when overruns were acceptable. Add `raise_on_lag` to both public wrappers. The first subscription sets the irreversible root broadcaster's policy, while every child selects its own strict or warn/drop/resume behavior independently. Document both fan-out APIs. Cover policy forwarding plus real IPC and infected-asyncio paths. Prompt-IO: ai/prompt-io/opencode/20260812T213117Z_51185487_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
Bound `BroadcastState.cancelled` entries to receiver progress, terminal state and resource lifetime instead of retaining completed `Task`s indefinitely. Deats, - make EOC durable so awakened peers never re-enter a closed source. - close root broadcasters during explicit `MsgStream` and `LinkedTaskChannel` teardown without re-entrant EOC closure or breaking `MsgStream.aclose()` overrides. - reject non-positive fan-out retention capacity before constructing an unusable zero-length queue. - cover child/root cancellation cleanup, terminal peer wakeups, wrapper teardown, subclass compatibility and zero-buffer rejection. Prompt-IO: ai/prompt-io/opencode/20260813T181901Z_a2e0df4b_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
Owner
Author
|
The current macOS CI failure in |
goodboy
changed the title
BroadcastReceiver lag counts
goodboy
marked this pull request as ready for review
August 14, 2026 00:26
There was a problem hiding this comment.
Pull request overview
This PR hardens the Trio broadcast fan-out implementation so subscriber ownership, lag accounting, and terminal failures behave deterministically and don’t strand peers. It also exposes per-subscriber lag policy through public wrapper APIs and adds regressions + docs to lock in the intended semantics.
Changes:
- Make broadcast fan-out lifecycle/ownership explicit (private source-read + peer-wait cancel scopes; terminal EOC/receive-failure state is durable and wakes peers).
- Fix lag drop counts and broadcast statistics to report queued counts (not cursor indexes), and reject non-positive buffer sizes.
- Add
raise_on_lagtoMsgStream.subscribe()andLinkedTaskChannel.subscribe(), document behavior, and expand test coverage for lag/failure/close interleavings.
Reviewed changes
Copilot reviewed 20 out of 20 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| tractor/trionics/_broadcast.py | Hardens broadcast receiver state machine (lag counts, terminal failure/EOC propagation, close scoping, stats). |
| tractor/trionics/init.py | Exports BroadcastReceiveError as public API. |
| tractor/to_asyncio.py | Adds raise_on_lag to linked-channel subscriptions and closes wrapper-owned root broadcaster on teardown. |
| tractor/_streaming.py | Adds raise_on_lag to MsgStream.subscribe() and prevents EOC-driven internal close from re-entrantly closing the root broadcaster. |
| tests/test_task_broadcasting.py | Adds deterministic regressions for lag counts, stats reporting, terminal failure/EOC, cancellation diagnostics, and wrapper forwarding/teardown. |
| docs/guide/streaming.rst | Documents per-subscriber lag policy and the “first subscription fixes root policy” behavior. |
| docs/guide/asyncio.rst | Documents per-subscriber lag policy for asyncio-linked channels. |
| docs/api/trionics.rst | Documents BroadcastReceiveError semantics and how underlying failures are delivered to peers. |
| ai/prompt-io/opencode/20260813T181901Z_a2e0df4b_prompt_io.raw.md | Adds opencode prompt I/O artifact for this PR iteration (raw). |
| ai/prompt-io/opencode/20260813T181901Z_a2e0df4b_prompt_io.md | Adds opencode prompt I/O artifact for this PR iteration (summary). |
| ai/prompt-io/opencode/20260812T213117Z_51185487_prompt_io.raw.md | Adds opencode prompt I/O artifact for prior iteration (raw). |
| ai/prompt-io/opencode/20260812T213117Z_51185487_prompt_io.md | Adds opencode prompt I/O artifact for prior iteration (summary). |
| ai/prompt-io/opencode/20260812T150027Z_c2a6ccef_prompt_io.raw.md | Adds opencode prompt I/O artifact for prior iteration (raw). |
| ai/prompt-io/opencode/20260812T150027Z_c2a6ccef_prompt_io.md | Adds opencode prompt I/O artifact for prior iteration (summary). |
| ai/prompt-io/opencode/20260812T030608Z_1095e7f7_prompt_io.raw.md | Adds opencode prompt I/O artifact for prior iteration (raw). |
| ai/prompt-io/opencode/20260812T030608Z_1095e7f7_prompt_io.md | Adds opencode prompt I/O artifact for prior iteration (summary). |
| ai/prompt-io/opencode/20260812T012324Z_06c4af17_prompt_io.raw.md | Adds opencode prompt I/O artifact for prior iteration (raw). |
| ai/prompt-io/opencode/20260812T012324Z_06c4af17_prompt_io.md | Adds opencode prompt I/O artifact for prior iteration (summary). |
| ai/prompt-io/opencode/20260811T233833Z_7cbd64ee_prompt_io.raw.md | Adds opencode prompt I/O artifact for initial iteration (raw). |
| ai/prompt-io/opencode/20260811T233833Z_7cbd64ee_prompt_io.md | Adds opencode prompt I/O artifact for initial iteration (summary). |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Comment on lines
+145
to
147
| # Retain the latest interrupted source-reader task until its | ||
| # receiver next makes progress or closes. | ||
| cancelled: dict[int, Task] = {} |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Harden broadcast fan-out ownership and failures
Motivation
BroadcastReceiverfan-out had several edge cases where subscriberstate stopped matching actual retained data or task ownership. Lag
counts were off by one, statistics exposed cursor indexes, shared
receive failures could strand peers, and closing one subscriber could
wake or cancel another subscriber's source-read path.
Downstream users also had no public way to select non-strict lag
handling, which forced private receiver mutation. This branch makes
lag, terminal-state and close behavior explicit and gives stream
subscribers a public per-child lag policy.
Summary of changes
Report exact displaced-value counts and receivable queue depths for
caught-up, queued and lagged subscribers.
Publish shared source failures as terminal broadcast state, wake
all peers, and replay durable EOC without letting peers re-enter a
closed source.
Give each
BroadcastReceiverprivate source-read and peer-waitclose scopes so closing one child cannot disturb siblings.
Bound cancelled-task diagnostics to receiver progress, terminal
state and resource lifetime; reject non-positive retention
capacity.
Add
raise_on_lagtoMsgStream.subscribe()andLinkedTaskChannel.subscribe()for caller-selected strict orwarn/drop/resume behavior.
Close wrapper-owned broadcaster roots during explicit stream
teardown without re-entrant EOC closure or breaking
MsgStream.aclose()overrides.Add deterministic broadcast, real IPC and infected-asyncio
regressions for lag, failure, close and lifecycle interleavings.
TODOs before landing
cpu_perf_headroom()timeoutscaling fixes the unrelated macOS
test_lifetime_stack_wipes_tmpfileCI failure seen on this branch.Future follow up
downstream consumers do not need to graft a parent
MsgStream.ctxonto a
BroadcastReceiverfor send-side controls.Links
this branch.
and lifecycle contracts.
(this pr content was generated in some part by
opencodeusinggpt-5.6-sol(openai))