Distinguish websocket receive and relay timeouts

The `proxy_msgs()` deadline covers both `ws.get_message()` and
`snd.send()`. Repeated socket-ready reports with no first relay
could therefore mean either a silent source or a blocked local
consumer; the old generic reset log could not distinguish them.

Track the currently awaited operation and report its phase, URL,
deadline, accumulated timeout count and reset threshold on each
timeout. Snapshot `NoBsWs._timeout` once per iteration so the log
reflects the deadline that actually elapsed. Keep the existing
shared deadline, counter behavior and reconnect policy intact.

Add two real loopback regressions: a silent websocket server and
one payload blocked on an unconsumed zero-capacity Trio relay.
Require the matching timeout phase and URL before the reset budget
is exhausted. The 15 selected Binance/websocket cases pass.

The human observes this after invalid `.usdm.` startup followed
by corrected `.usdtm.` input. Metadata accepts the typo but later
history lookup expects canonical keys. Record that investigation
as context; this diagnostic patch does not fix FQME validation,
failed-startup cleanup or prove the cause of corrected-feed stalls.

Prompt-IO: ai/prompt-io/opencode/20261028T180856Z_032580bf_timeout_phase.md

(this patch was generated in some part by `opencode` using
`gpt-6.1-sol` (`openai`))

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.

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.

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'

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)