diff --git a/ai/prompt-io/opencode/20260811T233833Z_7cbd64ee_prompt_io.md b/ai/prompt-io/opencode/20260811T233833Z_7cbd64ee_prompt_io.md new file mode 100644 index 00000000..b6060d0a --- /dev/null +++ b/ai/prompt-io/opencode/20260811T233833Z_7cbd64ee_prompt_io.md @@ -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. diff --git a/ai/prompt-io/opencode/20260811T233833Z_7cbd64ee_prompt_io.raw.md b/ai/prompt-io/opencode/20260811T233833Z_7cbd64ee_prompt_io.raw.md new file mode 100644 index 00000000..0af477f1 --- /dev/null +++ b/ai/prompt-io/opencode/20260811T233833Z_7cbd64ee_prompt_io.raw.md @@ -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. diff --git a/tests/test_task_broadcasting.py b/tests/test_task_broadcasting.py index 9ea251d2..20965cbe 100644 --- a/tests/test_task_broadcasting.py +++ b/tests/test_task_broadcasting.py @@ -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, diff --git a/tractor/trionics/_broadcast.py b/tractor/trionics/_broadcast.py index eb3894cf..4bc2e3b8 100644 --- a/tractor/trionics/_broadcast.py +++ b/tractor/trionics/_broadcast.py @@ -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