From 6429e6a51534e540767a7267ce8ea58806419e7d Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 27 Aug 2026 22:11:55 -0400 Subject: [PATCH] Reject concurrent `BroadcastReceiver.receive()` calls Two tasks could receive through one `BroadcastReceiver` and race its single `_wait_scope`. A later call could replace the scope, leaving the first blocked through `.aclose()`. Track the active `trio.lowlevel.Task` per receiver and reject an overlapping call with `trio.BusyResourceError`. Include the owner's task name and repr in the error, then clear ownership in `finally`. Define each receiver as one logical subscription cursor. Concurrent consumers must use separate `.subscribe()` receivers so each retains an independent cursor and lag policy. Strict mode reads every retained value or raises `Lagged`; `raise_on_lag=False` explicitly permits drops. Cover the regression with a checkpoint loop synchronized on `recv_ready`; explain the interleaving inline, assert the error names the active consumer, then release the source and prove the original receive remains usable. Prompt-IO: ai/prompt-io/opencode/20260828T014247Z_d06ee34b_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- .../20260828T014247Z_d06ee34b_prompt_io.md | 47 ++++++++++++++++ ...20260828T014247Z_d06ee34b_prompt_io.raw.md | 56 +++++++++++++++++++ docs/guide/streaming.rst | 8 +++ tests/test_task_broadcasting.py | 46 +++++++++++++++ tractor/trionics/_broadcast.py | 50 ++++++++++++++--- 5 files changed, 198 insertions(+), 9 deletions(-) create mode 100644 ai/prompt-io/opencode/20260828T014247Z_d06ee34b_prompt_io.md create mode 100644 ai/prompt-io/opencode/20260828T014247Z_d06ee34b_prompt_io.raw.md diff --git a/ai/prompt-io/opencode/20260828T014247Z_d06ee34b_prompt_io.md b/ai/prompt-io/opencode/20260828T014247Z_d06ee34b_prompt_io.md new file mode 100644 index 00000000..d6f15132 --- /dev/null +++ b/ai/prompt-io/opencode/20260828T014247Z_d06ee34b_prompt_io.md @@ -0,0 +1,47 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +session: pr490-landing-review +timestamp: 2026-08-28T01:42:47Z +git_ref: wkt/fix_broadcast_lag_count +scope: code +substantive: true +raw_file: 20260828T014247Z_d06ee34b_prompt_io.raw.md +--- + +## Prompt + +Prime PR #490 for landing, review its concurrent broadcast changes, +fix valid findings, and keep the landing pass focused. + +## Response summary + +Rejected an invalid mutable-default finding under `msgspec.Struct` +semantics. Found and fixed a close-time race where concurrent receives +on one `BroadcastReceiver` could overwrite its sole cancellation scope +and strand an earlier waiter. The receiver now rejects overlapping +calls with `trio.BusyResourceError`. Public docs now define each +receiver as one logical subscription cursor and require one receiver +per concurrent lossless consumer. Busy errors identify the active Trio +task which owns the receiver. + +## Files changed + +- `tractor/trionics/_broadcast.py` - reject concurrent receives on one + receiver handle. +- `tests/test_task_broadcasting.py` - prove the second call fails while + the original receive remains usable. +- `docs/guide/streaming.rst` - document per-task receiver ownership and + explicit lag behavior. + +## Human edits + +The human selected PR #490 as the next landing target, rebased it onto +current `main`, directed the agent to continue the landing pass, and +required a faster, tightly scoped execution after the review expanded. +The human then confirmed the per-receiver guard and required the public +method, type, and surrounding guide to emphasize lossless per-consumer +semantics. The human also required the regression's checkpoint loop to +explain its synchronization and the busy error to report the active +consumer task. The agent applied the source, test, and documentation +changes; no direct manual edits to these files were identified. diff --git a/ai/prompt-io/opencode/20260828T014247Z_d06ee34b_prompt_io.raw.md b/ai/prompt-io/opencode/20260828T014247Z_d06ee34b_prompt_io.raw.md new file mode 100644 index 00000000..d09bafe3 --- /dev/null +++ b/ai/prompt-io/opencode/20260828T014247Z_d06ee34b_prompt_io.raw.md @@ -0,0 +1,56 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +timestamp: 2026-08-28T01:42:47Z +git_ref: wkt/fix_broadcast_lag_count +diff_cmd: git diff +--- + +## Prompt + +Get PR #490 primed and landed. Continue after rebasing, but keep the +review and fix pass focused. + +## Response + +The mutable `{}` default is safe because `msgspec.Struct` creates a +fresh dictionary per instance. A separate concurrency defect remains: +two tasks can call `receive()` through one `BroadcastReceiver`, and the +second call overwrites the only `_wait_scope`. Closing that receiver can +then leave the first task blocked indefinitely. + +> `git diff -- tractor/trionics/_broadcast.py tests/test_task_broadcasting.py` + +Added a per-receiver in-progress guard that raises +`trio.BusyResourceError` for an overlapping receive. Added a regression +test that blocks the first receive, verifies the second call is +rejected, and releases the source to prove the first call still works. +The complete broadcast test file passes: 28 tests in 4.57 seconds. + +## Follow-up prompt + +Clarify that the guard is per receiver, preserves lossless consumer +semantics, and document the contract on the public method, type, and +surrounding guide. + +## Follow-up response + +> `git diff -- tractor/trionics/_broadcast.py docs/guide/streaming.rst` + +Documented that each `BroadcastReceiver` owns one logical cursor, each +concurrent consumer needs its own subscribed receiver, overlapping +calls on one handle raise `BusyResourceError`, and strict lag handling +never skips values silently. + +## Second follow-up prompt + +Explain that the polling loop waits until the background receive is +blocked before making the concurrent call, and include the first +consumer task's runtime information in the busy exception. + +## Second follow-up response + +Replaced the boolean guard with the active `trio.lowlevel.Task`, added +its name and representation to `BusyResourceError`, named the fixture +task for a deterministic assertion, and documented the checkpoint-loop +interleaving directly above the poll. diff --git a/docs/guide/streaming.rst b/docs/guide/streaming.rst index 17d35ef8..2ce093a0 100644 --- a/docs/guide/streaming.rst +++ b/docs/guide/streaming.rst @@ -171,6 +171,14 @@ keeps pace with the *fastest* subscriber; a task falling more than the buffered window behind has its next receive raise ``tractor.trionics.Lagged`` to say it lost data. +Each ``BroadcastReceiver`` is one logical subscription cursor, so +give every concurrent consumer task its own receiver. Overlapping +``receive()`` calls on the same handle raise +``trio.BusyResourceError`` instead of racing that cursor. In strict +mode values are never skipped silently: the consumer either reads +each retained value in sequence or receives an explicit ``Lagged`` +error after exceeding the buffer window. + Pass ``raise_on_lag=False`` when a consumer may drop old values and resume from the oldest retained item instead. The receiver logs the overrun rather than raising. Each child subscription chooses its own diff --git a/tests/test_task_broadcasting.py b/tests/test_task_broadcasting.py index 31e18b30..5e12e42e 100644 --- a/tests/test_task_broadcasting.py +++ b/tests/test_task_broadcasting.py @@ -1018,6 +1018,52 @@ def test_closing_non_owner_preserves_source_wait() -> None: trio.run(main) +def test_concurrent_receive_raises_busy() -> None: + ''' + Reject concurrent receives on one broadcast handle. + + A receiver stores one private peer-wait cancellation scope. If two + tasks receive through the same handle, the second task can replace + that scope and prevent `BroadcastReceiver.aclose()` from waking the + first task. Block one task in the shared source receive, then prove + a second call raises `BusyResourceError` before it can mutate any + per-receiver wait state. Releasing the source proves the original + receive remains usable. + + ''' + async def main() -> None: + tx, rx = trio.open_memory_channel(1) + brx = broadcast_receiver(rx, 1) + values: list[int] = [] + + async def receive() -> None: + values.append(await brx.receive()) + + async with trio.open_nursery() as nursery: + nursery.start_soon( + receive, + name='first broadcast consumer', + ) + + # Synchronize with the background task after it blocks in + # the shared source `.receive()`, ensuring the next call is + # concurrent with an already-active receive on this handle. + while brx._state.recv_ready is None: + await trio.lowlevel.checkpoint() + + with pytest.raises( + trio.BusyResourceError, + match='first broadcast consumer', + ): + await brx.receive() + + await tx.send(1) + + assert values == [1] + + trio.run(main) + + @pytest.mark.parametrize( 'first_outcome', [ diff --git a/tractor/trionics/_broadcast.py b/tractor/trionics/_broadcast.py index cfe8f2d7..0c6c9f42 100644 --- a/tractor/trionics/_broadcast.py +++ b/tractor/trionics/_broadcast.py @@ -184,12 +184,18 @@ class BroadcastState(Struct): class BroadcastReceiver(ReceiveChannel): ''' - A memory receive channel broadcaster which is non-lossy for - the fastest consumer. + One logical subscriber to a shared receive-channel broadcast. - Additional consumer tasks can receive all produced values by - registering with ``.subscribe()`` and receiving from the new - instance it delivers. + Each instance owns one sequence cursor. Additional consumer tasks + must call `.subscribe()` and receive through the new instance it + yields. Overlapping `.receive()` calls on the same instance raise + `trio.BusyResourceError` rather than racing that cursor or the + receiver's close-cancellation state. + + A strict subscriber reads each retained value in sequence. Falling + behind the retention window raises `Lagged` instead of silently + losing values; `raise_on_lag=False` explicitly opts into dropping + displaced values. ''' def __init__( @@ -219,6 +225,7 @@ class BroadcastReceiver(ReceiveChannel): self._closed: bool = False self._raise_on_lag = raise_on_lag self._wait_scope: trio.CancelScope|None = None + self._receive_task: trio.lowlevel.Task|None = None def receive_nowait( self, @@ -434,6 +441,30 @@ class BroadcastReceiver(ReceiveChannel): state.recv_scope = None async def receive(self) -> ReceiveType: + ''' + Receive the next value for this subscriber's sequence cursor. + + Only one task may receive through this instance at a time. Use + `.subscribe()` to give each concurrent consumer its own cursor + and loss/lag policy. `trio.BusyResourceError` identifies the + task which owns an already-active receive. + + ''' + if receive_task := self._receive_task: + raise trio.BusyResourceError( + 'another task is already receiving from this ' + '`BroadcastReceiver`\n' + f'active receive task: {receive_task.name!r}\n' + f'{receive_task!r}' + ) + + self._receive_task = trio.lowlevel.current_task() + try: + return await self._receive() + finally: + self._receive_task = None + + async def _receive(self) -> ReceiveType: key = self.key state = self._state @@ -518,11 +549,12 @@ class BroadcastReceiver(ReceiveChannel): ) -> AsyncIterator[BroadcastReceiver]: ''' - Subscribe for values from this broadcast receiver. + Create a receiver with its own logical subscription cursor. - Returns a new ``BroadCastReceiver`` which is registered for and - pulls data from a clone of the original - ``trio.abc.ReceiveChannel`` provided at creation. + The new `BroadcastReceiver` is registered against the shared + source and receives every retained value in sequence. Give each + concurrent consumer task its own receiver instead of sharing + one instance across overlapping `.receive()` calls. ''' if self._closed: