davide/plm-lottery
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Survive a dropped Electrum connection, and fall back to other servers

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
davide
2026-07-27 00:34:59 +02:00
co-authored by Claude Opus 5
parent 25f4a1c6b6
commit 7c4e9983ea
9 changed files with 516 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/electrum/client.py
+105 -6
View File
@@ -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():
app/electrum/listener.py
+122 -27
View File
@@ -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.