.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;wkt/land_binance_usdm_feed97eb8347deliberately 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:97eb834799Original-equivalent:d50ab6124cRelated-history:3d83b61f3fPrompt-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`))
parent
7242dcb9e1
commit
b95e7c796e
|
|
@ -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.
|
||||
|
|
@ -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.
|
||||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Reference in New Issue