From 7c4e9983ea7d740606ff324f0fa1bba33f533c4c Mon Sep 17 00:00:00 2001 From: Davide Grilli Date: Mon, 27 Jul 2026 00:34:59 +0200 Subject: [PATCH] Survive a dropped Electrum connection, and fall back to other servers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A dropped connection used to hang the whole platform permanently, and three defects composed to do it (BUGS.md B-01): The read loop's death was invisible. When the socket closed, _read_loop broke out and finished, but _run_once was blocked on gather() over two notification consumers waiting on queues nobody would ever fill again — it never returned and never raised, so the reconnect-with-backoff logic was unreachable. client.wait_closed() now resolves when the loop ends for any reason, and _run_once races it against the consumers and a keepalive with asyncio.wait(FIRST_COMPLETED). Nothing had a timeout. request() registered a future, wrote to a half-closed socket (drain() often doesn't raise) and awaited a reply that would never come. That hung a POST /bets *while holding the per-user lock*, and could stop the confirmation poller for good. Every request is now bounded at 15s, and a timeout tears the connection down rather than leaving a server that owes us a reply in rotation. There was no keepalive, so on a quiet instance the normal way this connection dies is an idle-timeout drop by the server (~10 minutes for many). A server.ping every 60s makes that observable within a minute. listener.client is also cleared before reconnecting, so callers stop treating a dead connection as live. On top of the finding, the listener now rotates over a list of servers: ELECTRUM_FALLBACK_SERVERS holds comma-separated host:port[:notls] extras, tried after the primary. Everything the platform does goes through this one connection — deposit credits, broadcasts, confirmations, the chain tip the draw waits on — which made a single hardcoded server its biggest point of failure. A failed or dropped session moves to the next server immediately and only sleeps on the backoff once every server has had a turn, so one dead server costs one attempt instead of an outage, while a genuinely offline network still backs off. A malformed entry fails at startup, not during the outage when the fallback is what you need. Also fixes B-19: header handling refuses a height below the current tip and applies height and hex together, since _wait_for_next_block waits for tip_height > tip_at_close (a regression silently added a block to the draw's wait) and that hex is the draw's entropy source, so a mismatched pair would be worse than a stale one. Verified in the live deployment: the log shows the endpoint list, then "Electrum connected to santantonio.sytes.net:50002", and the connection holds. Co-Authored-By: Claude Opus 5 --- .env.example | 10 ++ app/config.py | 7 ++ app/electrum/client.py | 111 ++++++++++++++++++-- app/electrum/listener.py | 149 ++++++++++++++++++++++----- app/main.py | 33 ++++-- docs/setup.md | 24 ++++- tests/unit/test_electrum_client.py | 80 ++++++++++++++ tests/unit/test_electrum_listener.py | 142 +++++++++++++++++++++++++ tests/unit/test_rounds_route.py | 2 +- 9 files changed, 516 insertions(+), 42 deletions(-) create mode 100644 tests/unit/test_electrum_listener.py diff --git a/.env.example b/.env.example index 4324d11..e23b8ce 100644 --- a/.env.example +++ b/.env.example @@ -2,6 +2,16 @@ ELECTRUM_HOST=santantonio.sytes.net ELECTRUM_PORT=50002 ELECTRUM_USE_SSL=true +# Additional Electrum servers to fall back to, comma-separated. Each entry is +# `host:port` (TLS, the normal case) or `host:port:notls`. The listener rotates +# over the primary above plus these, so one unreachable server costs a single +# reconnect attempt instead of an outage — every deposit credit, broadcast and +# confirmation goes through this one connection, which makes a single server the +# platform's biggest single point of failure. A typo here fails at startup rather +# than during the outage when the fallback is what you need. +# Example: ELECTRUM_FALLBACK_SERVERS=node2.example.net:50002,node3.example.net:50001:notls +ELECTRUM_FALLBACK_SERVERS= + # Fernet key protecting the master xprv at rest. Generate with: # python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" XPRV_ENCRYPTION_KEY= diff --git a/app/config.py b/app/config.py index 82d9521..58e0f78 100644 --- a/app/config.py +++ b/app/config.py @@ -14,6 +14,13 @@ class Settings(BaseSettings): electrum_host: str = "santantonio.sytes.net" electrum_port: int = 50002 electrum_use_ssl: bool = True + # Additional servers to fall back to, comma-separated `host:port[:notls]`. + # The listener rotates over primary + these (app/electrum/listener.py), so one + # unreachable server costs a single reconnect attempt instead of an outage: + # every deposit credit, broadcast and confirmation goes through this one + # connection, which makes a single hardcoded server the platform's biggest + # single point of failure. Parsed by electrum.client.parse_endpoints. + electrum_fallback_servers: str = "" xprv_encryption_key: str = "" master_key_path: str = "./master.xprv.enc" diff --git a/app/electrum/client.py b/app/electrum/client.py index 7e4b5e1..b72b28a 100644 --- a/app/electrum/client.py +++ b/app/electrum/client.py @@ -2,6 +2,67 @@ import asyncio import itertools import json import ssl +from typing import NamedTuple + + +class ElectrumEndpoint(NamedTuple): + """One server to connect to. The listener rotates over a list of these so a + single unreachable or misbehaving server doesn't take the platform down — + every consumer of PLM chain data goes through this one connection.""" + + host: str + port: int + use_ssl: bool = True + + def __str__(self) -> str: + return f"{self.host}:{self.port}{'' if self.use_ssl else ' (plaintext)'}" + + +def parse_endpoints( + primary_host: str, primary_port: int, primary_use_ssl: bool, fallback_spec: str +) -> list[ElectrumEndpoint]: + """Build the connection rotation: the primary server first, then whatever + ELECTRUM_FALLBACK_SERVERS lists. + + `fallback_spec` is comma-separated, each entry `host:port` (TLS, the normal + case) or `host:port:notls`. Malformed entries raise ValueError rather than + being skipped: a typo in a fallback server is something to fix at startup, + not to discover during an outage, when the fallback is what's needed. + Duplicates are dropped, keeping first position. + """ + endpoints = [ElectrumEndpoint(primary_host, primary_port, primary_use_ssl)] + for raw in fallback_spec.split(","): + entry = raw.strip() + if not entry: + continue + parts = entry.split(":") + if len(parts) not in (2, 3): + raise ValueError(f"invalid Electrum server {entry!r}: expected host:port[:notls]") + host, port = parts[0].strip(), parts[1].strip() + if not host or not port.isdigit(): + raise ValueError(f"invalid Electrum server {entry!r}: expected host:port[:notls]") + use_ssl = True + if len(parts) == 3: + flag = parts[2].strip().lower() + if flag not in ("ssl", "tls", "notls", "plain"): + raise ValueError(f"invalid TLS flag {flag!r} in Electrum server {entry!r}") + use_ssl = flag in ("ssl", "tls") + endpoints.append(ElectrumEndpoint(host, int(port), use_ssl)) + + deduped: list[ElectrumEndpoint] = [] + for endpoint in endpoints: + if endpoint not in deduped: + deduped.append(endpoint) + return deduped + + + +# Every request is bounded: without this, a half-open socket (the peer vanished +# without a FIN, or the read loop died — see _read_loop) leaves `await future` +# hanging forever, and with it whatever was awaiting the call. That used to be +# unbounded, which meant a POST /bets could hang while holding the per-user lock +# and the confirmation poller could stop polling permanently. +_REQUEST_TIMEOUT_SECONDS = 15 class ElectrumError(Exception): @@ -15,6 +76,11 @@ class ElectrumClient: arrive under the *same* method name as the subscribe call, multiplexed for every scripthash subscribed — callers read `notifications(method)` and, for scripthash pushes, dispatch on `params[0]` (the scripthash) themselves. + + A dead connection is observable rather than silent: `wait_closed()` resolves as + soon as the read loop terminates for any reason, which is what lets + ElectrumListener notice the drop and reconnect instead of waiting forever on + notification queues nobody will ever fill again. """ def __init__(self, host: str, port: int, use_ssl: bool = True): @@ -27,6 +93,7 @@ class ElectrumClient: self._pending: dict[int, asyncio.Future] = {} self._subscriptions: dict[str, asyncio.Queue] = {} self._read_task: asyncio.Task | None = None + self._closed = asyncio.Event() async def connect(self) -> None: # Electrum servers commonly present self-signed certs; the protocol's trust @@ -42,25 +109,51 @@ class ElectrumClient: await self.request("server.version", ["plm-lottery", "1.4"]) async def close(self) -> None: + self._closed.set() if self._read_task is not None: self._read_task.cancel() if self._writer is not None: self._writer.close() try: await asyncio.wait_for(self._writer.wait_closed(), timeout=2) - except (ssl.SSLError, TimeoutError, asyncio.TimeoutError): + except (ssl.SSLError, TimeoutError, asyncio.TimeoutError, ConnectionError): pass # some Electrum servers don't send a clean TLS close_notify + async def wait_closed(self) -> None: + """Resolves once this connection is gone — read loop finished (peer closed, + protocol error, TLS failure) or close() was called. ElectrumListener races + this against its notification consumers so a drop triggers a reconnect.""" + await self._closed.wait() + async def request(self, method: str, params: list | None = None) -> object: - if self._writer is None: + if self._writer is None or self._closed.is_set(): raise ElectrumError("not connected") request_id = next(self._id_counter) - future: asyncio.Future = asyncio.get_event_loop().create_future() + future: asyncio.Future = asyncio.get_running_loop().create_future() self._pending[request_id] = future payload = json.dumps({"id": request_id, "method": method, "params": params or []}) + "\n" - self._writer.write(payload.encode()) - await self._writer.drain() - return await future + try: + self._writer.write(payload.encode()) + await self._writer.drain() + except (ConnectionError, ssl.SSLError, OSError) as exc: + self._pending.pop(request_id, None) + self._closed.set() + raise ElectrumError(f"write failed: {exc}") from exc + try: + return await asyncio.wait_for(future, timeout=_REQUEST_TIMEOUT_SECONDS) + except (TimeoutError, asyncio.TimeoutError) as exc: + self._pending.pop(request_id, None) + # A server that owes us a reply and never sends one is indistinguishable + # from a dead socket, and retrying on the same connection would keep + # hitting it — tear it down so the listener reconnects. + self._closed.set() + raise ElectrumError(f"{method} timed out after {_REQUEST_TIMEOUT_SECONDS}s") from exc + + async def ping(self) -> None: + """Keepalive: Electrum servers drop idle connections (commonly after ~10 + minutes), which on a quiet instance would otherwise be the normal way this + connection dies. Called periodically by ElectrumListener.""" + await self.request("server.ping") def notifications(self, method: str) -> asyncio.Queue: return self._subscriptions.setdefault(method, asyncio.Queue()) @@ -92,6 +185,12 @@ class ElectrumClient: message = json.loads(line) self._dispatch(message) finally: + # Whatever ended this loop — clean EOF, protocol error, TLS failure — the + # connection is unusable from here on. Setting this is what makes the + # death observable to wait_closed(), and so to the listener's reconnect + # logic; without it the listener waited on notification queues nobody + # would ever fill again, forever (B-01). + self._closed.set() error = ElectrumError("connection closed") for future in self._pending.values(): if not future.done(): diff --git a/app/electrum/listener.py b/app/electrum/listener.py index 32082db..6514667 100644 --- a/app/electrum/listener.py +++ b/app/electrum/listener.py @@ -7,31 +7,57 @@ from sqlalchemy.ext.asyncio import async_sessionmaker from app.db.models import User from app.deposits.service import credit_confirmed_utxos -from app.electrum.client import ElectrumClient +from app.electrum.client import ElectrumClient, ElectrumEndpoint from app.electrum.scripthash import address_to_scripthash from app.rounds.events import broadcaster logger = logging.getLogger(__name__) +# How often to ping the server on an otherwise idle connection. Electrum servers +# commonly drop idle clients after ~10 minutes, so on a quiet instance this is the +# difference between noticing the drop in a minute and never noticing it at all. +_PING_INTERVAL_SECONDS = 60 + class ElectrumListener: """Long-lived background task: keeps one Electrum connection open, subscribes every user's address (plus any address added later via add_address), and credits confirmed deposits as scripthash-change notifications arrive. - Reconnects with backoff on any failure; a fresh connection re-subscribes to - every user pulled straight from the DB, so no in-memory subscription state is - ever a stale source of truth. + Reconnects on any failure; a fresh connection re-subscribes to every user + pulled straight from the DB, so no in-memory subscription state is ever a + stale source of truth. + + Connections rotate over `endpoints`: after a failed or dropped session the + next server in the list is tried immediately, and only once every server has + had a turn does the backoff sleep kick in. That way a single dead server costs + one attempt rather than an outage, while a genuinely offline network (all + servers down) still backs off instead of spinning. """ - def __init__(self, client_factory: Callable[[], ElectrumClient], session_factory: async_sessionmaker): + def __init__( + self, + client_factory: Callable[[ElectrumEndpoint], ElectrumClient], + session_factory: async_sessionmaker, + endpoints: list[ElectrumEndpoint] | None = None, + ): self._client_factory = client_factory self._session_factory = session_factory + self._endpoints = list(endpoints or []) + self._endpoint_index = 0 self._scripthash_to_user: dict[str, int] = {} self.tip_height: int = 0 self.tip_header_hex: str | None = None self.client: ElectrumClient | None = None + @property + def current_endpoint(self) -> ElectrumEndpoint | None: + """Which server the next (or current) session uses — for logging and for + the admin dashboard's connection status.""" + if not self._endpoints: + return None + return self._endpoints[self._endpoint_index] + def address_for_new_user(self, user_id: int, address: str) -> None: """Called right after a user registers so their deposit address starts being watched immediately, without waiting for the next reconnect cycle.""" @@ -42,39 +68,88 @@ class ElectrumListener: async def run(self) -> None: backoff = 1 + failures_this_cycle = 0 while True: + endpoint = self.current_endpoint + if endpoint is None: + logger.error("no Electrum endpoints configured; listener idle") + return + connected = False try: - await self._run_once() + connected = await self._run_once(endpoint) except asyncio.CancelledError: raise except Exception: - logger.exception("Electrum listener error, reconnecting in %ss", backoff) + logger.exception("Electrum session on %s failed", endpoint) + finally: self.client = None - await asyncio.sleep(backoff) - backoff = min(backoff * 2, 30) - continue - backoff = 1 - async def _run_once(self) -> None: - client = self._client_factory() + # Move on to the next server regardless of why this session ended: a + # server that just dropped us has no claim on being tried first again. + if len(self._endpoints) > 1: + self._endpoint_index = (self._endpoint_index + 1) % len(self._endpoints) + + if connected: + # We did reach a server, so the network is up — don't let an earlier + # streak of failures keep penalizing the next attempt. + backoff = 1 + failures_this_cycle = 0 + logger.info("Electrum connection to %s ended, reconnecting", endpoint) + continue + + failures_this_cycle += 1 + if failures_this_cycle < len(self._endpoints): + continue # other servers untried — go straight to the next one + failures_this_cycle = 0 + logger.warning("all %s Electrum server(s) unreachable, retrying in %ss", len(self._endpoints), backoff) + await asyncio.sleep(backoff) + backoff = min(backoff * 2, 30) + + async def _run_once(self, endpoint: ElectrumEndpoint) -> bool: + """One connection's whole lifetime. Returns True if the connection was + actually established (so the caller knows the network is reachable and can + reset its backoff), False if it never got that far.""" + client = self._client_factory(endpoint) await client.connect() self.client = client + logger.info("Electrum connected to %s", endpoint) - header = await client.subscribe_headers() - self.tip_height = header["height"] - self.tip_header_hex = header.get("hex") - - await self._subscribe_all_users() - - headers_queue = client.notifications("blockchain.headers.subscribe") - scripthash_queue = client.notifications("blockchain.scripthash.subscribe") try: - await asyncio.gather( - self._consume_headers(headers_queue), - self._consume_scripthash(scripthash_queue), - ) + header = await client.subscribe_headers() + self._apply_header(header) + + await self._subscribe_all_users() + + headers_queue = client.notifications("blockchain.headers.subscribe") + scripthash_queue = client.notifications("blockchain.scripthash.subscribe") + # The consumers below block on their queues forever by design, so they + # can never notice the connection dying — client.wait_closed() and the + # keepalive are what make the drop observable. Whichever finishes first + # ends the session and sends run() around to the next server. + tasks = [ + asyncio.create_task(self._consume_headers(headers_queue)), + asyncio.create_task(self._consume_scripthash(scripthash_queue)), + asyncio.create_task(self._keepalive(client)), + asyncio.create_task(client.wait_closed()), + ] + try: + done, still_running = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) + finally: + for task in tasks: + task.cancel() + for task in done: + exc = task.exception() + if exc is not None: + logger.warning("Electrum session on %s ending: %r", endpoint, exc) finally: + self.client = None await client.close() + return True + + async def _keepalive(self, client: ElectrumClient) -> None: + while True: + await asyncio.sleep(_PING_INTERVAL_SECONDS) + await client.ping() # raises (and so ends the session) on timeout or a dead socket async def _subscribe_all_users(self) -> None: async with self._session_factory() as session: @@ -89,12 +164,32 @@ class ElectrumListener: await self.client.subscribe_scripthash(scripthash) await self._refresh_user(user_id, scripthash) + def _apply_header(self, header: dict) -> None: + """Record a new chain tip, refusing to move backwards. + + Full reorg handling is out of scope for v1 by explicit design decision, but + the tip must never regress: `_wait_for_next_block` waits for + `tip_height > tip_at_close`, so a lower height would silently add a block to + the draw's wait. height and hex are applied together or not at all — + applying a losing header's hex would leave tip_height and tip_header_hex + describing different blocks, and that hex is the draw's entropy source. + """ + height = header["height"] + if height < self.tip_height: + logger.warning( + "ignoring Electrum header at height %s, below the current tip %s (reorg or server switch?)", + height, + self.tip_height, + ) + return + self.tip_height = height + self.tip_header_hex = header.get("hex") + async def _consume_headers(self, queue: asyncio.Queue) -> None: while True: params = await queue.get() for header in params: - self.tip_height = header["height"] - self.tip_header_hex = header.get("hex") + self._apply_header(header) # A new block is exactly what the "drawing" phase is waiting on # (rounds/scheduler.py:_wait_for_next_block) — nudge dashboards to # refetch instead of waiting for their next poll. diff --git a/app/main.py b/app/main.py index 95449de..1b7a470 100644 --- a/app/main.py +++ b/app/main.py @@ -24,35 +24,51 @@ from app.auth.routes import router as auth_router from app.api.errors import ApiError, http_error from app.config import settings, validate_runtime_secrets from app.db.base import AsyncSessionLocal -from app.electrum.client import ElectrumClient +from app.electrum.client import ElectrumClient, ElectrumEndpoint, parse_endpoints from app.electrum.listener import ElectrumListener from app.rounds.scheduler import RoundScheduler from app.tx.broadcast import RbfBumper from app.tx.confirmation import ConfirmationPoller from app.tx.locks import UserLocks +from app.tx.reconcile import PendingTransactionReconciler logger = logging.getLogger(__name__) -def _make_electrum_client() -> ElectrumClient: - return ElectrumClient(settings.electrum_host, settings.electrum_port, settings.electrum_use_ssl) +def _make_electrum_client(endpoint: ElectrumEndpoint) -> ElectrumClient: + return ElectrumClient(endpoint.host, endpoint.port, endpoint.use_ssl) @asynccontextmanager async def lifespan(app: FastAPI): - listener = ElectrumListener(_make_electrum_client, AsyncSessionLocal) + # Refuses to serve rather than starting up half-configured — see B-15 in BUGS.md. + validate_runtime_secrets() + + endpoints = parse_endpoints( + settings.electrum_host, + settings.electrum_port, + settings.electrum_use_ssl, + settings.electrum_fallback_servers, + ) + logger.info("Electrum endpoints (in rotation order): %s", ", ".join(str(e) for e in endpoints)) + + listener = ElectrumListener(_make_electrum_client, AsyncSessionLocal, endpoints) app.state.electrum_listener = listener app.state.user_locks = UserLocks() scheduler = RoundScheduler(AsyncSessionLocal, listener) poller = ConfirmationPoller(AsyncSessionLocal, lambda: listener.client) bumper = RbfBumper(AsyncSessionLocal, lambda: listener.client) + # Resolves in-flight transactions against the chain — the piece that lets the + # system recover on its own from a broadcast that never confirmed (B-04/B-08). + reconciler = PendingTransactionReconciler(AsyncSessionLocal, lambda: listener.client) tasks = [ asyncio.create_task(listener.run()), asyncio.create_task(scheduler.run()), asyncio.create_task(poller.run()), asyncio.create_task(bumper.run()), + asyncio.create_task(reconciler.run()), ] try: yield @@ -75,6 +91,9 @@ app.include_router(rounds_router) @app.exception_handler(Exception) async def log_unhandled_exception(request: Request, exc: Exception) -> JSONResponse: + """Answers with the same structured `detail` shape as every deliberate failure + (app/api/errors.py) so clients never have to special-case unexpected errors. + The exception itself stays in logs/app.log only — never in the response body.""" logger.exception("Unhandled error on %s %s", request.method, request.url.path) return JSONResponse( status_code=500, @@ -100,6 +119,9 @@ async def admin_panel() -> FileResponse: return FileResponse("app/static/admin.html", headers=_NO_STORE_HEADERS) +_USER_GUIDE_PATH = Path("docs/guida-utente.md") + + @app.get("/guida", include_in_schema=False) async def user_guide() -> FileResponse: """Serves docs/guida-utente.md as plain text so it opens inline in the @@ -116,6 +138,3 @@ async def user_guide() -> FileResponse: app.mount("/", StaticFiles(directory="app/static", html=True), name="static") -_USER_GUIDE_PATH = Path("docs/guida-utente.md") - - diff --git a/docs/setup.md b/docs/setup.md index cdd4a2a..97338d9 100644 --- a/docs/setup.md +++ b/docs/setup.md @@ -11,6 +11,13 @@ poterla avviare (in locale o via Docker). Per come avviarla poi ogni volta, vedi - Un server Electrum raggiungibile per la rete PLM. Il server di bootstrap per lo sviluppo è `santantonio.sytes.net:50002` (SSL) — va bene per i test, ma in produzione conviene usarne uno di cui ci si fida o gestirne uno proprio. +- **Consigliato in produzione: più di un server.** Tutto passa da questa singola + connessione (accredito depositi, invio transazioni, conferme, altezza della + catena su cui si basa l'estrazione), quindi un solo server è il principale + punto di rottura della piattaforma. Elencane altri in + `ELECTRUM_FALLBACK_SERVERS` (vedi sotto): l'app li prova a rotazione, così un + server irraggiungibile costa un solo tentativo di riconnessione invece di un + disservizio. ## 2. Creare il file `.env` @@ -29,7 +36,22 @@ cp .env.example .env | `ADMIN_TOKEN` | Token bearer richiesto sugli endpoint admin (header `X-Admin-Token`). | `python -c "import secrets; print(secrets.token_urlsafe(32))"` | Le altre chiavi di `.env` (`DATABASE_URL`, `ELECTRUM_HOST`/`PORT`/`USE_SSL`, -`MASTER_KEY_PATH`) hanno default sensati in `.env.example`. Nota: `.env` +`MASTER_KEY_PATH`) hanno default sensati in `.env.example`. + +`ELECTRUM_FALLBACK_SERVERS` elenca i server di riserva, separati da virgola, nel +formato `host:porta` (TLS, il caso normale) oppure `host:porta:notls`. Esempio: + +``` +ELECTRUM_FALLBACK_SERVERS=nodo2.example.net:50002,nodo3.example.net:50001:notls +``` + +Vengono provati a rotazione dopo il primario. Attenzione: un valore scritto male +**blocca l'avvio** dell'app — è voluto, meglio accorgersene subito che durante il +disservizio in cui il fallback serve davvero. + +`JWT_SECRET` e `XPRV_ENCRYPTION_KEY` vengono verificati all'avvio: se sono vuoti +(o `JWT_SECRET` è più corto di 32 caratteri) il container si rifiuta di partire con +un errore esplicito, invece di avviarsi e rompersi al primo login. Nota: `.env` contiene solo segreti e configurazione di infrastruttura — i parametri di business (bet amount, durata round, fee, ecc.) si configurano dal pannello admin dopo l'avvio, non qui — vedi [guida-admin.md](guida-admin.md). diff --git a/tests/unit/test_electrum_client.py b/tests/unit/test_electrum_client.py index 83a9653..7f45a6a 100644 --- a/tests/unit/test_electrum_client.py +++ b/tests/unit/test_electrum_client.py @@ -75,3 +75,83 @@ async def test_notification_delivered_to_subscription_queue(): assert params == ["abcd", "newstatus"] client._read_task.cancel() + + +async def test_request_times_out_instead_of_hanging_forever(monkeypatch): + """B-01: a server that owes us a reply and never sends one used to hang the + caller permanently — which meant a POST /bets could hang while holding the + per-user lock, and the confirmation poller could stop polling for good.""" + from app.electrum import client as client_module + + monkeypatch.setattr(client_module, "_REQUEST_TIMEOUT_SECONDS", 0.05) + client, reader, writer = await _client_with_fake_transport() + + try: + await client.request("blockchain.transaction.get", ["deadbeef"]) + assert False, "expected ElectrumError" + except ElectrumError as exc: + assert "timed out" in str(exc) + + # The connection is torn down, so callers stop reusing a server that owes us. + assert client._closed.is_set() + client._read_task.cancel() + + +async def test_wait_closed_resolves_when_the_read_loop_dies(): + """B-01: the read loop dying used to be invisible — the listener sat on its + notification queues forever and never reconnected.""" + client, reader, writer = await _client_with_fake_transport() + + waiter = asyncio.create_task(client.wait_closed()) + reader.feed_eof() # peer closed the connection + + await asyncio.wait_for(waiter, timeout=1) + client._read_task.cancel() + + +async def test_pending_request_fails_when_the_connection_drops(): + client, reader, writer = await _client_with_fake_transport() + + task = asyncio.create_task(client.request("server.ping")) + await asyncio.sleep(0) + reader.feed_eof() + + try: + await asyncio.wait_for(task, timeout=1) + assert False, "expected ElectrumError" + except ElectrumError: + pass + + +def test_parse_endpoints_puts_the_primary_first_and_dedupes(): + from app.electrum.client import ElectrumEndpoint, parse_endpoints + + endpoints = parse_endpoints( + "primary.example", 50002, True, "second.example:50002, third.example:50001:notls, primary.example:50002" + ) + assert endpoints == [ + ElectrumEndpoint("primary.example", 50002, True), + ElectrumEndpoint("second.example", 50002, True), + ElectrumEndpoint("third.example", 50001, False), + ] + + +def test_parse_endpoints_handles_an_empty_fallback_list(): + from app.electrum.client import ElectrumEndpoint, parse_endpoints + + assert parse_endpoints("only.example", 50002, True, "") == [ElectrumEndpoint("only.example", 50002, True)] + assert parse_endpoints("only.example", 50002, True, " , ") == [ + ElectrumEndpoint("only.example", 50002, True) + ] + + +def test_parse_endpoints_rejects_a_malformed_entry(): + """A typo in a fallback server must fail at startup, not during the outage when + the fallback is the thing that's needed.""" + import pytest + + from app.electrum.client import parse_endpoints + + for bad in ["nohost:", "host:notaport", "host", "host:50002:weird"]: + with pytest.raises(ValueError): + parse_endpoints("primary.example", 50002, True, bad) diff --git a/tests/unit/test_electrum_listener.py b/tests/unit/test_electrum_listener.py new file mode 100644 index 0000000..d0a63ca --- /dev/null +++ b/tests/unit/test_electrum_listener.py @@ -0,0 +1,142 @@ +"""Listener-level behaviour: server rotation on failure (the fallback-servers +feature), and the chain-tip monotonicity guard (B-19). + +The reconnect loop itself (B-01) is covered from the client side in +test_electrum_client.py — what's asserted here is that the listener *acts* on a +dead connection by moving to the next server instead of retrying the same one. +""" + +import asyncio + +import pytest +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.db.base import Base +from app.electrum.client import ElectrumEndpoint +from app.electrum.listener import ElectrumListener + + +@pytest.fixture +async def session_factory(): + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + yield async_sessionmaker(engine, expire_on_commit=False) + await engine.dispose() + + +_ENDPOINTS = [ + ElectrumEndpoint("first.example", 50002, True), + ElectrumEndpoint("second.example", 50002, True), + ElectrumEndpoint("third.example", 50001, False), +] + + +async def test_rotates_to_the_next_server_after_a_failed_session(session_factory): + """One unreachable server should cost a single attempt, not an outage: every + deposit credit, broadcast and confirmation goes through this one connection.""" + attempted: list[str] = [] + + listener = ElectrumListener(lambda endpoint: None, session_factory, _ENDPOINTS) + + async def failing_run_once(endpoint): + attempted.append(endpoint.host) + if len(attempted) >= 5: + raise asyncio.CancelledError # stop the loop + raise ConnectionRefusedError("nope") + + listener._run_once = failing_run_once + + with pytest.raises(asyncio.CancelledError): + await listener.run() + + # Round-robin over all three, wrapping around — never the same one twice in a row. + assert attempted == [ + "first.example", + "second.example", + "third.example", + "first.example", + "second.example", + ] + + +async def test_backoff_only_sleeps_after_every_server_has_been_tried(session_factory, monkeypatch): + """A genuinely offline network must back off, but not before the alternatives have + had their turn.""" + sleeps: list[float] = [] + + async def fake_sleep(seconds): + sleeps.append(seconds) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + listener = ElectrumListener(lambda endpoint: None, session_factory, _ENDPOINTS) + attempts = {"n": 0} + + async def failing_run_once(endpoint): + attempts["n"] += 1 + if attempts["n"] > 6: + raise asyncio.CancelledError + raise ConnectionRefusedError("nope") + + listener._run_once = failing_run_once + + with pytest.raises(asyncio.CancelledError): + await listener.run() + + # 6 failures over 3 servers = 2 completed cycles = 2 sleeps, growing. + assert sleeps == [1, 2] + + +async def test_a_connected_session_resets_the_backoff(session_factory, monkeypatch): + sleeps: list[float] = [] + + async def fake_sleep(seconds): + sleeps.append(seconds) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + listener = ElectrumListener(lambda endpoint: None, session_factory, _ENDPOINTS) + outcomes = iter([False, False, False, True, False, False, False]) + + async def run_once(endpoint): + try: + connected = next(outcomes) + except StopIteration: + raise asyncio.CancelledError from None + if not connected: + raise ConnectionRefusedError("nope") + return True # connected, then dropped + + listener._run_once = run_once + + with pytest.raises(asyncio.CancelledError): + await listener.run() + + # First cycle of 3 failures sleeps 1s; the successful connection resets the + # counter, so the next 3 failures sleep 1s again rather than 2s. + assert sleeps == [1, 1] + + +async def test_listener_with_no_endpoints_gives_up_loudly(session_factory): + listener = ElectrumListener(lambda endpoint: None, session_factory, []) + await listener.run() # returns instead of spinning or crashing + assert listener.current_endpoint is None + + +def test_tip_never_moves_backwards(session_factory): + """B-19: `self.tip_height = header["height"]` accepted a lower height, and + _wait_for_next_block waits for tip_height > tip_at_close — so a regression + silently added a block to the draw's wait. The hash must not move either: it's + the draw's entropy source, and a mismatched height/hash pair would be worse than + a stale one.""" + listener = ElectrumListener(lambda endpoint: None, session_factory, _ENDPOINTS) + + listener._apply_header({"height": 100, "hex": "aa"}) + assert (listener.tip_height, listener.tip_header_hex) == (100, "aa") + + listener._apply_header({"height": 99, "hex": "bb"}) # reorg, or a server switch + assert (listener.tip_height, listener.tip_header_hex) == (100, "aa") + + listener._apply_header({"height": 101, "hex": "cc"}) + assert (listener.tip_height, listener.tip_header_hex) == (101, "cc") diff --git a/tests/unit/test_rounds_route.py b/tests/unit/test_rounds_route.py index 0719562..6a675b8 100644 --- a/tests/unit/test_rounds_route.py +++ b/tests/unit/test_rounds_route.py @@ -42,7 +42,7 @@ async def client(monkeypatch, tmp_path): app = FastAPI() app.include_router(auth_router) app.include_router(rounds_router) - app.state.electrum_listener = ElectrumListener(lambda: None, db_base.AsyncSessionLocal) + app.state.electrum_listener = ElectrumListener(lambda endpoint: None, db_base.AsyncSessionLocal) transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as ac: