diff --git a/AGENTS.md b/AGENTS.md index 74401dd..51b4e9e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -72,6 +72,27 @@ and `/api/strike/ws`, converts each JSON message to an HTML fragment (`price_jso `strike_json_to_html`) and forwards it. `BTCDATA_URL` is the single point of contact; `_btcdata_ws_url` rewrites its scheme to `ws`/`wss`. +### Recovering from a btcdata restart + +htmx's WebSocket extension reconnects only after an abnormal close +(1006/1011/1012/1013); a normal 1000 close means "the server is done" and the panel stays +frozen until the visitor reloads. So a btcdata restart must never reach the browser as a +normal close: + +- `_ws_proxy` reconnects upstream behind the still-open browser socket, on the bounded + `UPSTREAM_RETRY_DELAYS_SECONDS` schedule. An ordinary restart is invisible to the page — + the last values stay on screen and updates resume. +- A connection that lasted `UPSTREAM_STABLE_SECONDS` resets that schedule, so a long-lived + page treats a later restart as a fresh outage rather than a continuing one. +- When the schedule runs out, the proxy closes with `BROWSER_RETRY_CLOSE_CODE` (1012) and + htmx's own backoff takes over. Every server-side exit uses that code; only a browser that + has already left can see a plain close. +- A payload that cannot be rendered is logged and skipped, not treated as a dead stream. + +`tests/test_ws_proxy.py` covers this against a real loopback WebSocket server, including +same-socket recovery across an upstream restart and the retryable close code. The regression +to watch for is a 1000 close after upstream loss. + ## Source Layout btcprice acts as a WebSocket proxy between the browser and btcdata: diff --git a/CLAUDE.md b/CLAUDE.md index 20502b9..b06fb05 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -72,6 +72,27 @@ and `/api/strike/ws`, converts each JSON message to an HTML fragment (`price_jso `strike_json_to_html`) and forwards it. `BTCDATA_URL` is the single point of contact; `_btcdata_ws_url` rewrites its scheme to `ws`/`wss`. +### Recovering from a btcdata restart + +htmx's WebSocket extension reconnects only after an abnormal close +(1006/1011/1012/1013); a normal 1000 close means "the server is done" and the panel stays +frozen until the visitor reloads. So a btcdata restart must never reach the browser as a +normal close: + +- `_ws_proxy` reconnects upstream behind the still-open browser socket, on the bounded + `UPSTREAM_RETRY_DELAYS_SECONDS` schedule. An ordinary restart is invisible to the page — + the last values stay on screen and updates resume. +- A connection that lasted `UPSTREAM_STABLE_SECONDS` resets that schedule, so a long-lived + page treats a later restart as a fresh outage rather than a continuing one. +- When the schedule runs out, the proxy closes with `BROWSER_RETRY_CLOSE_CODE` (1012) and + htmx's own backoff takes over. Every server-side exit uses that code; only a browser that + has already left can see a plain close. +- A payload that cannot be rendered is logged and skipped, not treated as a dead stream. + +`tests/test_ws_proxy.py` covers this against a real loopback WebSocket server, including +same-socket recovery across an upstream restart and the retryable close code. The regression +to watch for is a 1000 close after upstream loss. + ## Source Layout btcprice acts as a WebSocket proxy between the browser and btcdata: diff --git a/app/main.py b/app/main.py index c813b22..84919b2 100644 --- a/app/main.py +++ b/app/main.py @@ -1,6 +1,8 @@ +import asyncio import json import logging import os +import time from datetime import datetime import httpx @@ -367,36 +369,112 @@ def strike_json_to_html(text: str) -> str: # --- WebSocket proxy --- +# htmx's WebSocket extension reconnects a browser socket only after an abnormal +# close (1006/1011/1012/1013). A normal 1000 close means "the server is done", +# so the extension never reconnects and the panel stays frozen until the visitor +# reloads. A btcdata restart must therefore never reach the browser as a normal +# close: the proxy reconnects upstream behind the still-open browser socket, and +# only if that keeps failing does it close with 1012 (Service Restart) and hand +# recovery to the page's own reconnect. +BROWSER_RETRY_CLOSE_CODE = 1012 +UPSTREAM_OPEN_TIMEOUT_SECONDS = 5.0 +UPSTREAM_RETRY_DELAYS_SECONDS = (0.5, 1.0, 2.0, 4.0, 8.0, 8.0) +UPSTREAM_STABLE_SECONDS = 30.0 + def _btcdata_ws_url(path: str) -> str: return BTCDATA_URL.replace("http", "ws", 1) + path -async def _ws_proxy(browser_ws: WebSocket, upstream_path: str, transform): +async def _await_browser_close(browser_ws: WebSocket) -> None: + """Return as soon as the browser side goes away. + + The browser sends nothing, but reading is the only way to notice that it + left while the proxy is waiting on a dead upstream rather than sending. + """ + try: + while True: + if (await browser_ws.receive())["type"] == "websocket.disconnect": + return + except (WebSocketDisconnect, RuntimeError): + return + + +async def _forward_upstream(browser_ws: WebSocket, upstream_url: str, transform) -> None: + """Forward one upstream connection to the browser until that connection ends.""" + async with websockets.connect( + upstream_url, ping_interval=None, open_timeout=UPSTREAM_OPEN_TIMEOUT_SECONDS + ) as upstream: + unrenderable = 0 + async for message in upstream: + if not message: + continue + try: + html = transform(message) + except Exception: + # One unexpected payload must not take down a working page. Log + # the first per connection, then stay quiet about the rest. + unrenderable += 1 + logger.log( + logging.WARNING if unrenderable == 1 else logging.DEBUG, + "Discarding an upstream message that could not be rendered", + exc_info=True, + ) + continue + await browser_ws.send_text(html) + + +async def _ws_proxy(browser_ws: WebSocket, upstream_path: str, transform) -> None: await browser_ws.accept() upstream_url = _btcdata_ws_url(upstream_path) logger.info("Browser connected, proxying to %s", upstream_url) + browser_closed = asyncio.create_task(_await_browser_close(browser_ws)) + attempt = 0 try: - async with websockets.connect(upstream_url, ping_interval=None) as upstream: - async for message in upstream: - if not message: - continue - try: - html = transform(message) - await browser_ws.send_text(html) - except WebSocketDisconnect: - break - except Exception: - logger.debug("Error forwarding to browser", exc_info=True) - break - except WebSocketDisconnect: - pass - except Exception: - logger.debug("Upstream connection error", exc_info=True) + while True: + started = time.monotonic() + forwarding = asyncio.create_task( + _forward_upstream(browser_ws, upstream_url, transform) + ) + done, _ = await asyncio.wait( + {forwarding, browser_closed}, return_when=asyncio.FIRST_COMPLETED + ) + if browser_closed in done: + # The browser left; a healthy upstream would otherwise keep this + # task forwarding into a socket nobody is reading. + forwarding.cancel() + await asyncio.gather(forwarding, return_exceptions=True) + logger.info("Browser disconnected from %s", upstream_url) + return + try: + forwarding.result() + logger.info("Upstream %s closed; reconnecting", upstream_url) + except Exception: + logger.info("Upstream %s unavailable; reconnecting", upstream_url) + logger.debug("Upstream connection error", exc_info=True) + if time.monotonic() - started >= UPSTREAM_STABLE_SECONDS: + # A connection that worked for a while means a restart, not a + # persistently broken upstream: start the schedule over. + attempt = 0 + if attempt >= len(UPSTREAM_RETRY_DELAYS_SECONDS): + logger.warning( + "Upstream %s still unavailable; asking the browser to reconnect", + upstream_url, + ) + return + await asyncio.wait({browser_closed}, timeout=UPSTREAM_RETRY_DELAYS_SECONDS[attempt]) + attempt += 1 + if browser_closed.done(): + logger.info("Browser disconnected from %s", upstream_url) + return finally: + browser_closed.cancel() + await asyncio.gather(browser_closed, return_exceptions=True) try: - await browser_ws.close() + # Every server-side exit, expected or not, closes with a code the + # page retries; a browser that already left ignores it. + await browser_ws.close(code=BROWSER_RETRY_CLOSE_CODE) except Exception: pass diff --git a/tests/test_ws_proxy.py b/tests/test_ws_proxy.py new file mode 100644 index 0000000..6d30586 --- /dev/null +++ b/tests/test_ws_proxy.py @@ -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