Fix `BroadcastState.statistics()` queue counts

`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`))
wkt/fix_broadcast_lag_count
Gud Boi 2026-08-11 22:49:23 -04:00
parent 06c4af17e4
commit 1095e7f710
4 changed files with 133 additions and 2 deletions

View File

@ -0,0 +1,31 @@
---
model: openai/gpt-5.6-sol
service: opencode
session: ses_0799212ebffe42arY96czXn89F
timestamp: 2026-08-12T01:23:24Z
git_ref: 06c4af17
scope: code
substantive: true
raw_file: 20260812T012324Z_06c4af17_prompt_io.raw.md
---
## Prompt
Iteratively refine Tractor PR 490. For item one, correct broadcast statistics
queue counts and Trio event checks, verify and review the exact change, then
return a complete commit plan before proceeding.
## Response summary
Converted subscriber cursor indexes into clamped retained queue counts,
removed deprecated event truthiness, and added deterministic state and
deprecation regressions.
## Files changed
- `tractor/trionics/_broadcast.py` - accurate queue and waiter statistics.
- `tests/test_task_broadcasting.py` - retained-count and event regression.
## Human edits
None - generated output follows the requested first iterative item.

View File

@ -0,0 +1,37 @@
---
model: openai/gpt-5.6-sol
service: opencode
timestamp: 2026-08-12T01:23:24Z
git_ref: 06c4af17
diff_cmd: git diff HEAD~1..HEAD
---
After opening draft Tractor PR 490, the user requested an iterative pass over
additional broadcast subsystem findings. The first item was to correct
`BroadcastState.statistics()` queued counts and its deprecated Trio event
truthiness check, then stop for a complete commit plan.
> `git diff HEAD~1..HEAD -- tractor/trionics/_broadcast.py`
Changed `queued_len_by_task` from raw deque cursor indexes to retained,
receivable counts. Caught-up `-1` reports zero, valid indexes report index plus
one, and lagged cursors clamp to the current retained queue length. Replaced
`trio.Event` truthiness with an explicit `is not None` branch.
> `git diff HEAD~1..HEAD -- tests/test_task_broadcasting.py`
Added a deterministic statistics regression using actual sends and receives.
It verifies caught-up and one-queued states, drives a root receiver beyond a
three-slot retention window to prove clamping, and installs a real
`trio.Event` while treating deprecations as errors.
Verification output:
```text
........... [100%]
11 passed in 5.76s
```
Python compilation and `git diff --check` passed. Initial adversarial review
caught unclamped lagged cursors and an ineffective event test; both were
fixed. Final review found no actionable issues.

View File

@ -9,6 +9,7 @@ from functools import partial
from itertools import cycle
import time
from typing import Optional
import warnings
import pytest
import trio
@ -349,6 +350,61 @@ def test_lagged_reports_exact_drop_count(
trio.run(main)
def test_broadcast_statistics_report_queued_counts() -> None:
'''
`BroadcastState.statistics()` must report counts, not indexes.
Each `BroadcastState.subs` value is the deque index of a
receiver's next unread value, with `-1` meaning caught up. The
statistics API returned these indexes directly, so one queued
value appeared as zero and every positive count was one short.
Keep one root receiver idle while a child synchronously receives
four produced values. Prove the root count advances through one
and three retained values, then remains clamped to the three-slot
retention window after lagging.
Finally install an actual unwaited `trio.Event` in
`BroadcastState.recv_ready` while treating deprecations as errors.
This proves statistics checks `None` explicitly instead of using
deprecated `trio.Event` truthiness.
'''
async def main() -> None:
tx, rx = trio.open_memory_channel(3)
brx = broadcast_receiver(rx, 3)
async with brx.subscribe() as child:
state = brx._state
assert state.statistics()['queued_len_by_task'] == {
brx.key: 0,
child.key: 0,
}
await tx.send(0)
assert await child.receive() == 0
assert state.statistics()['queued_len_by_task'] == {
brx.key: 1,
child.key: 0,
}
for value in range(1, 4):
await tx.send(value)
assert await child.receive() == value
state.recv_ready = (child.key, trio.Event())
with warnings.catch_warnings():
warnings.simplefilter('error', DeprecationWarning)
stats = state.statistics()
assert stats['queued_len_by_task'] == {
brx.key: 3,
child.key: 0,
}
assert stats['tasks_waiting'] == 0
trio.run(main)
def test_ensure_slow_consumers_lag_out(
reg_addr,
start_method,

View File

@ -142,13 +142,20 @@ class BroadcastState(Struct):
qlens: dict[int, int] = {}
for tid, sz in subs.items():
qlens[tid] = sz if sz != -1 else 0
qlens[tid] = min(
sz + 1,
len(self.queue),
)
return {
'open_consumers': len(subs),
'queued_len_by_task': qlens,
'max_buffer_size': self.maxlen,
'tasks_waiting': ev.statistics().tasks_waiting if ev else 0,
'tasks_waiting': (
ev.statistics().tasks_waiting
if ev is not None
else 0
),
'tasks_cancelled': self.cancelled,
'next_value_receiver_id': key,
}