Fix `BroadcastReceiver` lag counts
`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`))wkt/fix_broadcast_lag_count
parent
83b3488455
commit
06c4af17e4
|
|
@ -0,0 +1,30 @@
|
|||
---
|
||||
model: openai/gpt-5.6-sol
|
||||
service: opencode
|
||||
session: ses_0799212ebffe42arY96czXn89F
|
||||
timestamp: 2026-08-11T23:38:33Z
|
||||
git_ref: 7cbd64ee
|
||||
scope: code
|
||||
substantive: true
|
||||
raw_file: 20260811T233833Z_7cbd64ee_prompt_io.raw.md
|
||||
---
|
||||
|
||||
## Prompt
|
||||
|
||||
Open a new isolated worktree in the local Tractor repository and draft a fix
|
||||
for `BroadcastReceiver` reporting that a lagged one-slot consumer dropped
|
||||
zero values when one value had actually been displaced.
|
||||
|
||||
## Response summary
|
||||
|
||||
Corrected the off-by-one lag count while preserving cursor recovery and added
|
||||
deterministic narrow- and wider-window regressions for exact loss reporting.
|
||||
|
||||
## Files changed
|
||||
|
||||
- `tractor/trionics/_broadcast.py` - exact broadcast overrun count.
|
||||
- `tests/test_task_broadcasting.py` - lag count and recovery regression.
|
||||
|
||||
## Human edits
|
||||
|
||||
None - generated output follows the user's diagnosed edge case.
|
||||
|
|
@ -0,0 +1,50 @@
|
|||
---
|
||||
model: openai/gpt-5.6-sol
|
||||
service: opencode
|
||||
timestamp: 2026-08-11T23:38:33Z
|
||||
git_ref: 7cbd64ee
|
||||
diff_cmd: git diff HEAD~1..HEAD
|
||||
---
|
||||
|
||||
The user asked for a Tractor fix in a new isolated worktree after a live piker
|
||||
failure reported:
|
||||
|
||||
```text
|
||||
tractor.trionics._broadcast.Lagged:
|
||||
Task `piker.brokers.ib.broker.handle_order_requests` overrun and
|
||||
dropped `0` values
|
||||
```
|
||||
|
||||
Inspection showed the lag exception was valid but its count was off by one.
|
||||
`BroadcastReceiver.receive_nowait()` treats `seq` as a deque index. With a
|
||||
one-entry queue, index zero is the only retained value and `seq == 1` already
|
||||
means one value was displaced. The old `seq - maxlen` calculation therefore
|
||||
reported zero instead of one.
|
||||
|
||||
> `git diff HEAD~1..HEAD -- tractor/trionics/_broadcast.py`
|
||||
|
||||
Adjusted the lag count to `seq - maxlen + 1` and documented why the first
|
||||
invalid deque index must be included. The existing Tokio-style cursor reset
|
||||
remains unchanged.
|
||||
|
||||
> `git diff HEAD~1..HEAD -- tests/test_task_broadcasting.py`
|
||||
|
||||
Added a deterministic parameterized regression covering a one-slot queue
|
||||
with one dropped value and a three-slot queue with two dropped values. The
|
||||
test keeps the root receiver idle while a child subscriber synchronously
|
||||
drains each produced value, asserts the exact `Lagged` message, and proves the
|
||||
next receive resumes at the oldest retained item.
|
||||
|
||||
Verification output:
|
||||
|
||||
```text
|
||||
.. [100%]
|
||||
2 passed in 0.04s
|
||||
|
||||
.......... [100%]
|
||||
10 passed in 4.65s
|
||||
```
|
||||
|
||||
The targeted import resolved to the new Tractor worktree. Python compilation
|
||||
and `git diff --check` passed. Adversarial review found no actionable issues;
|
||||
zero-capacity channels remain a pre-existing untested edge outside this fix.
|
||||
|
|
@ -307,6 +307,48 @@ def test_subscribe_errors_after_close():
|
|||
trio.run(main)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
('size', 'sent', 'dropped'),
|
||||
[
|
||||
(1, 2, 1),
|
||||
(3, 5, 2),
|
||||
],
|
||||
)
|
||||
def test_lagged_reports_exact_drop_count(
|
||||
size: int,
|
||||
sent: int,
|
||||
dropped: int,
|
||||
) -> None:
|
||||
'''
|
||||
`Lagged` must report every value outside the retained window.
|
||||
|
||||
`BroadcastReceiver.receive_nowait()` previously subtracted the
|
||||
queue length from an already-invalid deque index without counting
|
||||
that first displaced value. A one-slot queue therefore claimed it
|
||||
dropped zero values after two sends. Keep one root receiver idle
|
||||
while a child subscriber drains every produced value, then prove
|
||||
the lag error reports the exact overrun and positions the root at
|
||||
the oldest value still retained by `BroadcastState.queue`.
|
||||
|
||||
'''
|
||||
async def main() -> None:
|
||||
tx, rx = trio.open_memory_channel(size)
|
||||
brx = broadcast_receiver(rx, size)
|
||||
|
||||
async with brx.subscribe() as fast:
|
||||
for value in range(sent):
|
||||
await tx.send(value)
|
||||
assert await fast.receive() == value
|
||||
|
||||
match = rf'dropped `{dropped}` values'
|
||||
with pytest.raises(Lagged, match=match):
|
||||
await brx.receive()
|
||||
|
||||
assert await brx.receive() == sent - size
|
||||
|
||||
trio.run(main)
|
||||
|
||||
|
||||
def test_ensure_slow_consumers_lag_out(
|
||||
reg_addr,
|
||||
start_method,
|
||||
|
|
|
|||
|
|
@ -237,7 +237,10 @@ class BroadcastReceiver(ReceiveChannel):
|
|||
# https://docs.rs/tokio/1.11.0/tokio/sync/broadcast/index.html#lagging
|
||||
|
||||
mxln = state.maxlen
|
||||
lost = seq - mxln
|
||||
# `seq == mxln` is already one past the final
|
||||
# valid deque index, so include that first
|
||||
# displaced value in the loss count.
|
||||
lost = seq - mxln + 1
|
||||
|
||||
# decrement to the last value and expect
|
||||
# consumer to either handle the ``Lagged`` and come back
|
||||
|
|
|
|||
Loading…
Reference in New Issue