From b95e7c796e6d0e70359f58b056e3581aef477236 Mon Sep 17 00:00:00 2001 From: goodboy Date: Wed, 28 Oct 2026 12:48:06 -0400 Subject: [PATCH] .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: 97eb834799cd8922d4efb96f442201291281c08a Original-equivalent: d50ab6124cd0e1646c82fa4a85d7c13bfff82611 Related-history: 3d83b61f3f7b982e84a4230dba5777f1407c78f1 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`)) --- ...261028T160846Z_7242dcb9_websocket_close.md | 77 +++++++++++ ...28T160846Z_7242dcb9_websocket_close.raw.md | 36 +++++ piker/data/_web_bs.py | 22 +-- tests/test_websocket.py | 126 +++++++++++++++++- 4 files changed, 247 insertions(+), 14 deletions(-) create mode 100644 ai/prompt-io/opencode/20261028T160846Z_7242dcb9_websocket_close.md create mode 100644 ai/prompt-io/opencode/20261028T160846Z_7242dcb9_websocket_close.raw.md diff --git a/ai/prompt-io/opencode/20261028T160846Z_7242dcb9_websocket_close.md b/ai/prompt-io/opencode/20261028T160846Z_7242dcb9_websocket_close.md new file mode 100644 index 00000000..81a2e0cc --- /dev/null +++ b/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. diff --git a/ai/prompt-io/opencode/20261028T160846Z_7242dcb9_websocket_close.raw.md b/ai/prompt-io/opencode/20261028T160846Z_7242dcb9_websocket_close.raw.md new file mode 100644 index 00000000..1709e177 --- /dev/null +++ b/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. diff --git a/piker/data/_web_bs.py b/piker/data/_web_bs.py index af8db8cf..0cd11082 100644 --- a/piker/data/_web_bs.py +++ b/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 ( - HandshakeError, - ConnectionRejected, - ): - log.exception('Retrying connection') + except nobsws.recon_errors: + # The websocket reader can fail while its context exits, + # outside proxy_msgs() (e.g. Pong during LOCAL_CLOSING). + # Collapse its nursery group before classifying the error. + 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: - berr = _berr + except BaseException: 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() diff --git a/tests/test_websocket.py b/tests/test_websocket.py index 6bff49df..ad52a022 100644 --- a/tests/test_websocket.py +++ b/tests/test_websocket.py @@ -6,8 +6,13 @@ from contextlib import asynccontextmanager as acm import pytest import trio -from trio_websocket import ConnectionRejected -from wsproto.utilities import LocalProtocolError +from trio_websocket import ( + 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)