diff --git a/ai/prompt-io/opencode/20261028T180856Z_032580bf_timeout_phase.md b/ai/prompt-io/opencode/20261028T180856Z_032580bf_timeout_phase.md new file mode 100644 index 00000000..d1bd99d5 --- /dev/null +++ b/ai/prompt-io/opencode/20261028T180856Z_032580bf_timeout_phase.md @@ -0,0 +1,54 @@ +--- +created_at: 2026-10-28T18:08:56Z +generated_by: + harness: opencode + provider: openai + model: openai/gpt-6.1-sol +git_ref: 032580bf +scope: code +substantive: true +raw_file: 20261028T180856Z_032580bf_timeout_phase.raw.md +--- + +## Prompt + +The human reported repeated readiness without first-message relay, +specifically after first attempting an invalid .usdm. FQME and then +the corrected .usdtm. one. They requested distinct log messages for +receive starvation versus local backpressure. + +## Response summary + +Add phase-specific WARNING messages for each elapsed relay-loop +deadline, with URL, timeout duration, accumulated count and reset +threshold. Leave the existing shared receive/send deadline and +counter/reset semantics intact. Do not claim that connected/ready +sockets alone establish parsed quote recovery. + +Read-only invalid-venue tracing found the usd substring check in +Binance get_mkt_info() accepts USDM while retaining it in MktPair. +Client.bars() later uses that market identity against canonical +USDTM keys. Backend stream startup precedes history startup and +sampler consumption. A relationship to the corrected-feed stall +still needs runtime evidence; neither venue handling nor task +lifetime behavior was changed by this diagnostic patch. + +## Files changed + +- `piker/data/_web_bs.py`: track and report the timed-out operation. +- `tests/test_websocket.py`: real silent-source and blocked-sink cases. +- This prompt provenance pair. + +## Verification + +15 selected Binance/websocket cases passed. The new cases exercise +real TCP, websocket handshake and relay behavior, replacing only +log observation. A zero-capacity relay deliberately forces the +local-send case; the source-silent case never receives a payload. +git diff --check passed. No live provider/daemon operations run. + +## Human edits + +No direct source edits observed. The human supplied the critical +invalid-symbol-then-corrected-symbol trigger and requested the +phase distinction; live testing and task acceptance remain theirs. diff --git a/ai/prompt-io/opencode/20261028T180856Z_032580bf_timeout_phase.raw.md b/ai/prompt-io/opencode/20261028T180856Z_032580bf_timeout_phase.raw.md new file mode 100644 index 00000000..84788da7 --- /dev/null +++ b/ai/prompt-io/opencode/20261028T180856Z_032580bf_timeout_phase.raw.md @@ -0,0 +1,34 @@ +--- +created_at: 2026-10-28T18:08:56Z +generated_by: + harness: opencode + provider: openai + model: openai/gpt-6.1-sol +git_ref: 032580bf +diff_cmd: git diff HEAD~1..HEAD +--- + +Added distinct timeout diagnostics for websocket receive starvation +versus local relay backpressure, preserving the existing timeout, +counter and reconnect threshold policy. + +> `git diff HEAD~1..HEAD -- piker/data/_web_bs.py` + +Track whether proxy_msgs() is awaiting ws.get_message() or +snd.send(). Report that phase, URL, configured deadline, timeout +count and reset threshold for every timeout, before the existing +threshold/reset diagnostic. Capture the configured deadline once +per operation so the report matches the deadline that elapsed. + +> `git diff HEAD~1..HEAD -- tests/test_websocket.py` + +Two real loopback variants use a silent server versus one payload +and an unconsumed zero-capacity Trio relay. Require the appropriate +phase and URL on the actual first timeout. All 15 selected +Binance/websocket cases pass. No live provider was contacted. + +Read-only tracing found a permissive invalid-venue path: get_mkt_info +accepts USDM via its usd substring check and preserves that venue +in MktPair, whereas Client.bars() indexes canonical USDTM pair keys. +This identifies a potential late failed-startup path, not proof of +the corrected-chart stall or of leaked producer ownership. diff --git a/piker/data/_web_bs.py b/piker/data/_web_bs.py index 0a981900..87343368 100644 --- a/piker/data/_web_bs.py +++ b/piker/data/_web_bs.py @@ -197,12 +197,15 @@ async def _reconnect_forever( first_message: bool = True while True: + waiting_on: str = 'websocket receive' + timeout: float = nobsws._timeout with trio.move_on_after( # can be dynamically changed by user code - nobsws._timeout, + timeout, ) as cs: try: msg: Any = await ws.get_message() + waiting_on = 'local relay send' await snd.send(msg) if first_message: log.info( @@ -224,6 +227,12 @@ async def _reconnect_forever( if cs.cancelled_caught: timeouts += 1 + log.warning( + f'Websocket timeout waiting on {waiting_on}\n' + f'url: {url!r}\n' + f'timeout: {timeout}s; count: {timeouts}; ' + f'reset after: {reset_after}\n' + ) if timeouts > reset_after: log.error( f'{src_mod}\n' diff --git a/tests/test_websocket.py b/tests/test_websocket.py index a6f12512..6dd40fd9 100644 --- a/tests/test_websocket.py +++ b/tests/test_websocket.py @@ -302,3 +302,73 @@ def test_reader_ping_close_recovery( message.startswith('Websocket first message relayed:') for message in messages ) == 1 + + +@pytest.mark.parametrize('message_arrives', [False, True]) +def test_timeout_identifies_blocked_operation( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + message_arrives: bool, +) -> None: + ''' + Receive starvation and local backpressure shared one reset log. + + The reconnect deadline covers both ws.get_message() and the + Trio relay send, so a connected-but-stalled feed could not be + diagnosed from the old timeout report. Use real loopback sockets + with either a silent server or a server sending one payload into + a zero-capacity local channel whose consumer never receives. + Require the actual timed-out operation and URL in the diagnostic. + Stop after the first timeout, before the unchanged reset budget. + + ''' + async def main() -> None: + observed = trio.Event() + warning = _web_bs.log.warning + + def observe_timeout(message, *args, **kwargs) -> None: + warning(message, *args, **kwargs) + if message.startswith('Websocket timeout waiting on '): + observed.set() + + monkeypatch.setattr(_web_bs.log, 'warning', observe_timeout) + + async def serve(request) -> None: + ws = await request.accept() + if message_arrives: + await ws.send_message('{"sequence": 1}') + await trio.sleep_forever() + + async with trio.open_nursery() as tn: + server = await tn.start( + serve_websocket, serve, '127.0.0.1', 0, None, + ) + url = f'ws://127.0.0.1:{server.port}/' + send, receive = trio.open_memory_channel(0) + async with send, receive: + wrapper = _web_bs.NoBsWs( + url, receive, msg_recv_timeout=0.05, + ) + await tn.start( + _web_bs._reconnect_forever, + url, send, wrapper, 3, + ) + with trio.fail_after(5): + await observed.wait() + waiting_on = ( + 'local relay send' + if message_arrives else 'websocket receive' + ) + messages = [ + record.getMessage() for record in caplog.records + if record.getMessage().startswith( + 'Websocket timeout waiting on ', + ) + ] + assert len(messages) == 1 + assert f'waiting on {waiting_on}' in messages[0] + assert f'url: {url!r}' in messages[0] + assert 'count: 1; reset after: 3' in messages[0] + tn.cancel_scope.cancel() + + trio.run(main)