Recover from a btcdata restart without reloading the page
pull-request / validate (pull_request) Successful in 24s
pull-request / validate (pull_request) Successful in 24s
htmx's WebSocket extension reconnects only after close codes 1006/1011/1012/1013. _ws_proxy closed the browser socket with a normal 1000 on every upstream exit, so htmx read a btcdata restart as a deliberate shutdown and left both panels frozen until the visitor reloaded. New browser connections always worked, which hid the defect from non-browser checks. The proxy now reconnects upstream behind the still-open browser socket on a bounded schedule, resets that schedule after a connection that lasted long enough to count as healthy, and closes with 1012 only once the schedule runs out, so recovery falls back to htmx's own backoff. A concurrent reader detects a departed browser, which a dead upstream would otherwise be indistinguishable from, and an unrenderable payload is logged and skipped rather than ending the stream. tests/test_ws_proxy.py covers this against a real loopback WebSocket server: same socket recovery across an upstream restart for both streams, the retryable close code when the upstream stays down, and skip-on-unrenderable. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,142 @@
|
||||
"""Proxy recovery tests.
|
||||
|
||||
htmx reconnects a browser WebSocket only after an abnormal close, so a btcdata
|
||||
restart must either be invisible to the browser or arrive as a retryable close
|
||||
code. These tests run a real loopback WebSocket server as the stand-in btcdata:
|
||||
the recovery being tested is connection lifecycle behaviour, which a stubbed
|
||||
client would not exercise.
|
||||
"""
|
||||
import asyncio
|
||||
import json
|
||||
import socket
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
import websockets
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app import main
|
||||
|
||||
PRICE = json.dumps({"price": 67432.1, "prev_price": 67400, "recorded_at": ""})
|
||||
LATER_PRICE = json.dumps({"price": 67500.0, "prev_price": 67432.1, "recorded_at": ""})
|
||||
|
||||
|
||||
class FakeBtcdata:
|
||||
"""Serves one list of messages per incoming connection.
|
||||
|
||||
Every session but the last closes as soon as its messages are sent, which is
|
||||
what a btcdata restart looks like from the proxy's side. The last session
|
||||
stays open so the test can assert on a live stream.
|
||||
"""
|
||||
|
||||
def __init__(self, sessions):
|
||||
self._sessions = [list(session) for session in sessions]
|
||||
self._ready = threading.Event()
|
||||
self.connections = 0
|
||||
self.port = None
|
||||
|
||||
async def _handler(self, connection):
|
||||
self.connections += 1
|
||||
session = self._sessions.pop(0) if self._sessions else []
|
||||
for message in session:
|
||||
await connection.send(message)
|
||||
if not self._sessions:
|
||||
await connection.wait_closed()
|
||||
|
||||
async def _start(self):
|
||||
# websockets.serve needs a running loop, so it cannot be called from
|
||||
# run_until_complete's argument list.
|
||||
self._server = await websockets.serve(self._handler, "127.0.0.1", 0)
|
||||
self.port = self._server.sockets[0].getsockname()[1]
|
||||
|
||||
async def _stop(self):
|
||||
self._server.close()
|
||||
await self._server.wait_closed()
|
||||
|
||||
def _run(self):
|
||||
self._loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(self._loop)
|
||||
self._loop.run_until_complete(self._start())
|
||||
self._ready.set()
|
||||
self._loop.run_forever()
|
||||
self._loop.run_until_complete(self._stop())
|
||||
self._loop.close()
|
||||
|
||||
def __enter__(self):
|
||||
self._thread = threading.Thread(target=self._run, daemon=True)
|
||||
self._thread.start()
|
||||
assert self._ready.wait(10), "fake btcdata did not start"
|
||||
return self
|
||||
|
||||
def stop(self):
|
||||
"""Stop listening. Idempotent, so a test may stop the upstream early."""
|
||||
if self._thread.is_alive():
|
||||
self._loop.call_soon_threadsafe(self._loop.stop)
|
||||
self._thread.join(10)
|
||||
|
||||
def __exit__(self, *exception):
|
||||
self.stop()
|
||||
return False
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def quick_retries(monkeypatch):
|
||||
"""Keep the real schedule's shape without waiting out its real delays."""
|
||||
monkeypatch.setattr(main, "UPSTREAM_RETRY_DELAYS_SECONDS", (0.01, 0.01, 0.01))
|
||||
|
||||
|
||||
def closed_port():
|
||||
with socket.socket() as probe:
|
||||
probe.bind(("127.0.0.1", 0))
|
||||
return probe.getsockname()[1]
|
||||
|
||||
|
||||
def test_upstream_restart_recovers_on_the_same_browser_socket(monkeypatch):
|
||||
with FakeBtcdata([[PRICE], [LATER_PRICE]]) as upstream:
|
||||
monkeypatch.setattr(main, "BTCDATA_URL", f"http://127.0.0.1:{upstream.port}")
|
||||
with TestClient(main.app).websocket_connect("/ws/price") as browser:
|
||||
assert "$67.432,10" in browser.receive_text()
|
||||
# Same browser socket, no reload: the proxy reconnected upstream.
|
||||
assert "$67.500,00" in browser.receive_text()
|
||||
assert upstream.connections == 2
|
||||
|
||||
|
||||
def test_browser_is_asked_to_reconnect_when_upstream_stays_down(monkeypatch):
|
||||
monkeypatch.setattr(main, "BTCDATA_URL", f"http://127.0.0.1:{closed_port()}")
|
||||
with TestClient(main.app).websocket_connect("/ws/price") as browser:
|
||||
with pytest.raises(main.WebSocketDisconnect) as disconnect:
|
||||
browser.receive_text()
|
||||
# 1000 is the regression: htmx treats it as "done" and never reconnects.
|
||||
assert disconnect.value.code == main.BROWSER_RETRY_CLOSE_CODE
|
||||
assert disconnect.value.code in (1006, 1011, 1012, 1013)
|
||||
|
||||
|
||||
def test_browser_is_asked_to_reconnect_after_a_lost_upstream(monkeypatch):
|
||||
with FakeBtcdata([[PRICE]]) as upstream:
|
||||
port = upstream.port
|
||||
monkeypatch.setattr(main, "BTCDATA_URL", f"http://127.0.0.1:{port}")
|
||||
with TestClient(main.app).websocket_connect("/ws/price") as browser:
|
||||
assert "$67.432,10" in browser.receive_text()
|
||||
# The fake stops listening entirely, so in-proxy retries cannot win.
|
||||
upstream.stop()
|
||||
with pytest.raises(main.WebSocketDisconnect) as disconnect:
|
||||
browser.receive_text()
|
||||
assert disconnect.value.code == main.BROWSER_RETRY_CLOSE_CODE
|
||||
|
||||
|
||||
def test_unrenderable_message_does_not_end_the_stream(monkeypatch):
|
||||
with FakeBtcdata([["not json", "", PRICE]]) as upstream:
|
||||
monkeypatch.setattr(main, "BTCDATA_URL", f"http://127.0.0.1:{upstream.port}")
|
||||
with TestClient(main.app).websocket_connect("/ws/price") as browser:
|
||||
assert "$67.432,10" in browser.receive_text()
|
||||
assert upstream.connections == 1
|
||||
|
||||
|
||||
def test_strike_stream_recovers_on_the_same_browser_socket(monkeypatch):
|
||||
rates = json.dumps({"BTC/USDT": 67432.1, "BTC/EUR": 61000.5, "quote-sats": 1476})
|
||||
with FakeBtcdata([[rates], [rates]]) as upstream:
|
||||
monkeypatch.setattr(main, "BTCDATA_URL", f"http://127.0.0.1:{upstream.port}")
|
||||
with TestClient(main.app).websocket_connect("/ws/strike") as browser:
|
||||
assert "1.476 <span" in browser.receive_text()
|
||||
assert 'id="strike-display"' in browser.receive_text()
|
||||
assert upstream.connections == 2
|
||||
Reference in New Issue
Block a user