.data: reconnect after websocket reader close errors

A Binance Ping racing local socket closure makes trio-websocket's
reader send a Pong while wsproto is LOCAL_CLOSING. That reader
nursery error escapes proxy_msgs(), then tears down brokerd and
pikerd instead of restoring quote delivery.

Classify NoBsWs.recon_errors around the entire websocket context,
with collapse_eg() outside it so a singleton reader exit group
reaches the reconnect handler. Invalidate the stale connected
event and socket reference before retrying. Keep retry sleep
unshielded so parent cancellation stops recovery promptly.

Deats,
- Add real loopback websocket coverage: TCP, handshake, Ping,
  reader dispatch and wsproto's actual Pong validation.
- Control only the first client Ping's protocol-state ordering,
  withholding its Close frame to prevent a peer response racing
  the deterministic LOCAL_CLOSING reproducer.
- Require fresh data on a replacement connection, plus exactly
  one connection when cancelling at the reader or retry boundary.
- Use RuntimeError for the unexpected-failure control:
  LocalProtocolError is now a recoverable connection failure.
- Preserve the original unexpected exception with bare raise,
  and assert its identity in the test. Log recoverable errors
  with the websocket URL and the underlying traceback.

This is a local follow-up, not another downstream cherry-pick.
The earlier wrapper suppressed escaped communication failures;
97eb8347 deliberately re-raised them. Retain that outer error
visibility rather than restoring suppression without recovery.
The isolated baseline reproduces the live Pong failure; all 13
selected Binance/websocket cases pass with this fix. Human live
qualification remains separate from these loopback regressions.

Follow-up-to: 97eb834799
Original-equivalent: d50ab6124c
Related-history: 3d83b61f3f
Prompt-IO: ai/prompt-io/opencode/20261028T160846Z_7242dcb9_websocket_close.md

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

ai/prompt-io/opencode/20261028T160846Z_7242dcb9_websocket_close.md

@@ -0,0 +1,77 @@
+---
+created_at: 2026-10-28T16:08:46Z
+generated_by:
+  harness: opencode
+  provider: openai
+  model: openai/gpt-6.1-sol
+git_ref: 7242dcb9
+scope: code
+substantive: true
+raw_file: 20261028T160846Z_7242dcb9_websocket_close.raw.md
+---
+
+## Prompt
+
+The human reported repeated live Binance perp crashes after correct
+worktree startup: the real trio-websocket reader attempted a Pong
+while wsproto was LOCAL_CLOSING. They asked whether an existing
+downstream websocket fix should be reused, then requested:
+
+> ok add the fix now, follow by a /commit-plan and then i'll test
+> before saving it ya?
+
+They approved a real websocket reader/close regression covering
+recovery and intentional shutdown, and separately authorized
+reconciling this worktree's legacy workflow bookkeeping while
+preserving existing archives.
+
+## Response summary
+
+The older `3d83b61f3f7b982e84a4230dba5777f1407c78f1` wrapper
+suppressed escaped communication failures. The landing replay
+`97eb8347` of `d50ab6124cd0e1646c82fa4a85d7c13bfff82611`
+added propagation; it did not restart a reader-context failure.
+Existing `LocalProtocolError` membership in NoBsWs.recon_errors
+only protected the receive relay, not reader-nursery context exit.
+
+Classify these errors around the complete websocket context and
+retry instead of restoring outer suppression. Place collapse_eg
+outside that context, invalidate stale connection readiness and
+socket references, and leave the retry wait cancellable. Preserve
+unexpected failures and mixed exception groups for propagation.
+After review, use RuntimeError rather than ValueError for the
+test-only unexpected-failure sentinel. Its injected exception has
+no underlying cause. Production recovery logs retain the original
+traceback and now name the URL; unexpected exceptions use bare
+raise to preserve identity, existing causes and notes.
+
+The deterministic seam controls protocol-state ordering in the real
+client Ping handler. It does not replace the websocket transport,
+reader, context manager, reconnect task or wsproto exception.
+
+## Files changed
+
+- `piker/data/_web_bs.py`: reader-context failure recovery.
+- `tests/test_websocket.py`: loopback recovery and two shutdown
+  variants; retain unexpected-failure propagation control.
+- This provenance pair: live report, lineage and test scope.
+
+## Verification
+
+Baseline uses new tests against original HEAD websocket code in an
+isolated exact-tree checkout. The recovery case fails with the
+reported Pong-in-LOCAL_CLOSING LocalProtocolError.
+
+Fixed code passes 13 cases across tests/test_websocket.py and
+tests/test_binance.py. Exact-boundary verification and native-library
+configuration are retained in the ignored commit-plan package.
+Live testing remains human-owned and pending.
+
+## Human edits
+
+No direct manual source edits observed in this follow-up. The human
+supplied live failures, corrected the earlier daemon/worktree setup,
+set the narrow landing scope, requested source-fix reuse, and will
+test before reviewing and committing. Existing commits are retained.
+The human requested the RuntimeError sentinel and preservation of
+underlying error context. The raw partner retains the initial output.

ai/prompt-io/opencode/20261028T160846Z_7242dcb9_websocket_close.raw.md

@@ -0,0 +1,36 @@
+---
+created_at: 2026-10-28T16:08:46Z
+generated_by:
+  harness: opencode
+  provider: openai
+  model: openai/gpt-6.1-sol
+git_ref: 7242dcb9
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+Implemented a focused websocket-context reconnect fix and real
+loopback reader/close regressions in the Binance landing worktree.
+
+> `git diff HEAD~1..HEAD -- piker/data/_web_bs.py`
+
+Move exception-group collapse outside the websocket context so
+reader-nursery exit errors are classified by the reconnect loop.
+Retry NoBsWs.recon_errors there, clear the stale connected event
+and socket reference, and honor cancellation through unshielded
+retry sleep. Retain unexpected-error propagation at the outer API.
+
+> `git diff HEAD~1..HEAD -- tests/test_websocket.py`
+
+Exercise real TCP, handshake, Ping frames, reader dispatch and
+wsproto Pong validation. A narrow handler gate advances the real
+protocol state to LOCAL_CLOSING without sending the Close frame,
+preventing a peer reply from racing the controlled state. Require
+fresh application data from the replacement connection. Separate
+variants cancel at the reader gate and at the actual retry wait.
+Update the existing unexpected-failure control to use ValueError
+because LocalProtocolError is now intentionally recoverable.
+
+The isolated original-code/new-test baseline fails with the exact
+Pong-in-LOCAL_CLOSING error. The fixed isolated tree passes all 13
+selected Binance/websocket cases. No live provider, broker orders
+or production daemon operations were performed by the agent.

piker/data/_web_bs.py

@@ -261,8 +261,8 @@ async def _reconnect_forever(
         ws: WebSocketConnection
         try:
             async with (
-                open_websocket_url(url) as ws,
                 tractor.trionics.collapse_eg(),
+                open_websocket_url(url) as ws,
                 trio.open_nursery() as tn,
             ):
                 cs = nobsws._cs = tn.cancel_scope
@@ -298,11 +298,16 @@ async def _reconnect_forever(
                 nobsws._connected.set()
                 await trio.sleep_forever()
 
-        except (
+        except nobsws.recon_errors:
-            HandshakeError,
+            # The websocket reader can fail while its context exits,
-            ConnectionRejected,
+            # outside proxy_msgs() (e.g. Pong during LOCAL_CLOSING).
-        ):
+            # Collapse its nursery group before classifying the error.
-            log.exception('Retrying connection')
+            nobsws._connected = trio.Event()
+            nobsws._ws = None
+            # Unlike the old outer wrapper suppression, restart the
+            # connection. This checkpoint also honors parent shutdown;
+            # never shield a reconnect retry from cancellation.
+            log.exception(f'Retrying connection: {url!r}')
             await trio.sleep(0.5)  # throttle
             continue
 
@@ -314,14 +319,13 @@ async def _reconnect_forever(
         except trio.Cancelled:
             raise
 
-        except BaseException as _berr:
+        except BaseException:
-            berr = _berr
             log.exception(
                 'Reconnect-attempt failed ??\n'
             )
             with trio.CancelScope(shield=True):
                 await trio.sleep(0.2)  # throttle
-            raise berr
+            raise
 
         #|_ws & nursery block ends
         nobsws._connected = trio.Event()

tests/test_websocket.py

@@ -6,8 +6,13 @@ from contextlib import asynccontextmanager as acm
 
 import pytest
 import trio
-from trio_websocket import ConnectionRejected
+from trio_websocket import (
-from wsproto.utilities import LocalProtocolError
+    ConnectionClosed,
+    ConnectionRejected,
+    WebSocketConnection,
+    serve_websocket,
+)
+from wsproto.events import CloseConnection
 
 from piker.data import _web_bs
 
@@ -64,13 +69,16 @@ def test_immediate_websocket_failure_retries(
     unbound-local error.
 
     Make the real reconnect loop receive ``ConnectionRejected`` on
-    its first context entry and ``LocalProtocolError`` on its second.
+    its first context entry and ``RuntimeError`` on its second.
     Removing retry sleeps fixes their order. Two attempts and the
     propagated second error prove the first rejection restarts the
     loop without touching connection-local cancellation state.
+    Require the exact error object so wrapping cannot discard its
+    existing diagnostic context.
 
     '''
     attempts: int = 0
+    failure = RuntimeError('unexpected websocket failure')
 
     @acm
     async def reject_url(url: str):
@@ -79,7 +87,7 @@ def test_immediate_websocket_failure_retries(
         if attempts == 1:
             raise ConnectionRejected(503, [], b'unavailable')
 
-        raise LocalProtocolError('invalid websocket response')
+        raise failure
         yield
 
     async def no_throttle(delay: float) -> None:
@@ -97,7 +105,10 @@ def test_immediate_websocket_failure_retries(
     )
 
     async def main() -> None:
-        with pytest.raises(LocalProtocolError):
+        with pytest.raises(
+            RuntimeError,
+            match='unexpected websocket',
+        ) as caught:
             async with _web_bs.open_autorecon_ws(
                 url='wss://reject.invalid/ws',
                 connect_timeout=0.1,
@@ -105,6 +116,7 @@ def test_immediate_websocket_failure_retries(
                 pytest.fail(
                     'failed websocket unexpectedly connected'
                 )
+        assert caught.value is failure
 
     trio.run(main)
     assert attempts == 2
@@ -173,3 +185,107 @@ def test_reconnect_cancellation_bypasses_error_throttle(
 
     trio.run(main)
     assert sleep_calls == []
+
+
+@pytest.mark.parametrize('shutdown', ['recover', 'reader', 'retry'])
+def test_reader_ping_close_recovery(
+    monkeypatch: pytest.MonkeyPatch,
+    shutdown: str,
+) -> None:
+    '''
+    A real websocket reader's Pong error escaped reconnect ownership.
+
+    Binance pings arriving during local closure raised wsproto's
+    LocalProtocolError in trio-websocket's reader nursery, outside
+    proxy_msgs(). 97eb8347 propagated that error from the outer
+    wrapper, terminating the persistent broker instead of reconnecting.
+
+    Use real loopback TCP, handshake, Ping frames and reader dispatch.
+    At the first client Ping, synchronously advance its real wsproto
+    connection to LOCAL_CLOSING. Withhold that Close frame from the
+    wire so a peer Close reply cannot race this deterministic gate.
+    The original Ping handler then generates the actual Pong error.
+    Require a replacement connection to deliver fresh application data.
+
+    For shutdown, cancel either at the held Ping boundary or after
+    the actual error reaches the reconnect throttle. Require bounded
+    exit and exactly one connection, proving teardown never starts a
+    retry. Only the Ping handler ordering is controlled; contexts,
+    readers, cancellation, reconnect tasks and delivery remain real.
+
+    '''
+    async def main() -> None:
+        connections: int = 0
+        ping_entered = trio.Event()
+        handler_exited = trio.Event()
+        retry_entered = trio.Event()
+        ping_handler = WebSocketConnection._handle_ping_event
+        faulted: list[WebSocketConnection] = []
+        log_exception = _web_bs.log.exception
+
+        def observe_retry(message, *args, **kwargs) -> None:
+            if message.startswith('Retrying connection:'):
+                retry_entered.set()
+            log_exception(message, *args, **kwargs)
+
+        monkeypatch.setattr(_web_bs.log, 'exception', observe_retry)
+
+        async def gate_ping(ws, event) -> None:
+            if ws.is_client and not faulted:
+                faulted.append(ws)
+                ping_entered.set()
+                try:
+                    if shutdown == 'reader':
+                        await trio.sleep_forever()
+                    ws._wsproto.send(CloseConnection(code=1000))
+                    await ping_handler(ws, event)
+                finally:
+                    handler_exited.set()
+            else:
+                await ping_handler(ws, event)
+
+        monkeypatch.setattr(
+            WebSocketConnection, '_handle_ping_event', gate_ping,
+        )
+
+        async def serve(request) -> None:
+            nonlocal connections
+            ws = await request.accept()
+            connections += 1
+            if connections == 1:
+                try:
+                    await ws.ping(b'close-race')
+                except ConnectionClosed:
+                    pass
+            else:
+                await ws.send_message('{"generation": 2}')
+            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}/'
+            with trio.fail_after(5):
+                with trio.CancelScope() as owner:
+                    async with _web_bs.open_autorecon_ws(
+                        url, connect_timeout=2,
+                    ) as ws:
+                        await ping_entered.wait()
+                        if shutdown == 'retry':
+                            await retry_entered.wait()
+                        if shutdown != 'recover':
+                            owner.cancel()
+                            await trio.lowlevel.checkpoint()
+                        else:
+                            assert await ws.recv_msg() == {
+                                'generation': 2,
+                            }
+                await handler_exited.wait()
+                assert len(faulted) == 1
+                assert connections == (
+                    2 if shutdown == 'recover' else 1
+                )
+            tn.cancel_scope.cancel()
+
+    trio.run(main)