diff --git a/BUGS.md b/BUGS.md index 738bbb3..1ab436e 100644 --- a/BUGS.md +++ b/BUGS.md @@ -1,17 +1,16 @@ # Known bugs A second full-codebase audit on 2026-07-27 found **25 further issues** (4 critical, 6 high, -7 medium, 8 low), listed below as B-30 … B-49. B-25 through B-29 are fixed (see "Previously -fixed" below) — no Critical-severity finding remains open; the other 20 are High/Medium/Low. +7 medium, 8 low), listed below as B-31 … B-49. B-25 through B-30 are fixed (see "Previously +fixed" below) — no Critical-severity finding remains open; the other 19 are High/Medium/Low. The 139-test suite was green at the time of the audit, so none of these were caught by existing -coverage — every fix lands with a regression test (the five fixes so far brought the suite from -139 to 176). +coverage — every fix lands with a regression test (the six fixes so far brought the suite from +139 to 182). The recurring pattern across the open findings is worth stating once: the code is rigorous about the failure modes that have actually been hit, and silent about the ones that have not. The payout phase is now fully recoverable; the "drawing" phase (waiting on a block) still has -no equivalent resume-after-restart or stall visibility (B-36), and there's still no periodic -deposit-side reconciler independent of scripthash notifications (B-30). +no equivalent resume-after-restart or stall visibility (B-36). For limitations that are accepted by design rather than bugs (single-shared-token admin auth, single-process assumptions, no user-facing history, etc.), see "Known gaps / TODO" in @@ -21,26 +20,6 @@ single-process assumptions, no user-facing history, etc.), see "Known gaps / TOD ## High -### B-30 — No deposit-side reconciler: one missed subscription means deposits are never credited - -`electrum/listener.py:61-67` (`address_for_new_user`) fires `asyncio.create_task(...)` without -retaining the reference and without handling exceptions. If `self.client` becomes `None` -between the check and the task running, the `assert` at `:163` raises inside an orphan task -and the exception is swallowed. - -What makes this serious is what happens next: deposits are credited **exclusively** by -scripthash notifications. There is no periodic routine reconciling balances against the chain -(the reconciler only covers outgoing transactions). On a healthy keepalive'd connection there -are no reconnects, so a lost subscription is never recovered and that user **never sees their -deposits**, indefinitely. - -**Proposed fix.** Two parts. (a) Make the subscription reliable: retain the task, log its -exceptions, and retry with backoff instead of relying on a reconnect. (b) Add the missing -safety net — a periodic sweep (say every few minutes, similar in shape to -`PendingTransactionReconciler`) that re-runs `_refresh_user` for users whose scripthash is not -in `_scripthash_to_user`, or simply round-robins over all users so a missed notification is -always eventually caught. - ### B-31 — Reconnect costs O(users) sequential round-trips and stalls the draw In `_run_once` the order is: subscribe headers → `_subscribe_all_users()` → *then* start the @@ -285,9 +264,10 @@ already does. - **B-27** — an RBF bump reset the reconciler's own abandon clock, so a repeatedly-bumped tx was never abandoned - **B-28** — a hostile Electrum server (or a MITM) could single-handedly pick the round's winner - **B-29** — a UTXO absent from one server's `listunspent` was marked spent immediately, irreversibly, on a single unauthenticated reply +- **B-30** — a lost scripthash subscription meant a user's deposits were never credited, with no periodic safety net See git history for the fix-by-fix breakdown (commits `f13f685`, `50a43ae`, `933760e`, and the -B-28/B-29 fixes). Suite grew from 139 to 176 tests over the five. +B-28/B-29/B-30 fixes). Suite grew from 139 to 182 tests over the six. A full-codebase audit on 2026-07-26 (commit `d4e0974`) found 24 bugs across every Python module under `app/`, both static frontends, and the Docker/Caddy deployment — 5 critical, diff --git a/app/deposits/reconcile.py b/app/deposits/reconcile.py new file mode 100644 index 0000000..048d9d3 --- /dev/null +++ b/app/deposits/reconcile.py @@ -0,0 +1,69 @@ +"""Periodic safety net for deposit crediting and external-spend detection (B-30), +independent of scripthash-change notifications. + +Those notifications are the fast path, but nothing else re-verifies a user's +balance against the chain if one is ever silently lost: `address_for_new_user`'s +subscribe is best-effort (its own failure just logs, see electrum/listener.py), +and on an otherwise healthy, long-lived connection there may be no reconnect for +days — the only other event that re-subscribes everyone from scratch. Without +this, a single lost subscription meant that user's deposits were never credited, +indefinitely. + +This mirrors app/tx/reconcile.py's shape (a periodic sweep gated on the Electrum +client being connected) but reuses ElectrumListener.refresh_user directly rather +than re-implementing crediting/spend-detection, so the notification-driven and +periodic paths can never behave differently from each other. +""" + +import asyncio +import logging + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker + +from app.db.models import User +from app.electrum.listener import ElectrumListener +from app.electrum.scripthash import address_to_scripthash + +logger = logging.getLogger(__name__) + +_SWEEP_INTERVAL_SECONDS = 300 + + +class DepositReconciler: + def __init__(self, session_factory: async_sessionmaker, listener: ElectrumListener): + self._session_factory = session_factory + self._listener = listener + + async def run(self) -> None: + while True: + await asyncio.sleep(_SWEEP_INTERVAL_SECONDS) + if self._listener.client is None: + continue + try: + await self._sweep_once() + except asyncio.CancelledError: + raise + except Exception: + logger.exception("deposit reconciliation sweep failed") + + async def _sweep_once(self) -> None: + """Round-robins over every user's address rather than only ones missing + from the listener's in-memory `_scripthash_to_user` map: that map can't + tell "never subscribed" apart from "subscribed, but this server silently + stopped delivering notifications for it" — exactly the failure mode this + exists to catch. One user failing (a transient network hiccup) must not + stop the sweep from reaching the rest, mirroring poll_once's per-item + isolation in tx/confirmation.py. + """ + async with self._session_factory() as session: + users = (await session.scalars(select(User))).all() + + for user in users: + if self._listener.client is None: + return # connection dropped mid-sweep; the next reconnect's own _subscribe_all_users covers everyone + scripthash = address_to_scripthash(user.address) + try: + await self._listener.refresh_user(user.id, scripthash) + except Exception: + logger.exception("deposit reconciliation failed for user_id=%s", user.id) diff --git a/app/deposits/service.py b/app/deposits/service.py index 4b88aae..ef9a037 100644 --- a/app/deposits/service.py +++ b/app/deposits/service.py @@ -113,7 +113,7 @@ async def find_utxos_missing_from(session: AsyncSession, user_id: int, entries: Returning a row here is *not* proof it was actually spent — only that this one server's reply no longer lists it. A single broken, behind, or malicious server could otherwise zero a user's balance on one bad reply, which is why - the caller (electrum/listener.py:_refresh_user) must independently + the caller (electrum/listener.py:refresh_user) must independently corroborate each candidate against other configured servers before treating it as genuine, rather than this function marking anything itself. diff --git a/app/electrum/listener.py b/app/electrum/listener.py index f83daf0..55267cd 100644 --- a/app/electrum/listener.py +++ b/app/electrum/listener.py @@ -63,6 +63,11 @@ class ElectrumListener: self._endpoints = list(endpoints or []) self._endpoint_index = 0 self._scripthash_to_user: dict[str, int] = {} + # Retains address_for_new_user's fire-and-forget subscribe task so it + # can't be garbage-collected mid-flight, and so its exception (if any) is + # actually observed instead of only reaching asyncio's default "Task + # exception was never retrieved" handler (B-30). + self._background_tasks: set[asyncio.Task] = set() self.tip_height: int = 0 self.tip_header_hex: str | None = None self.client: ElectrumClient | None = None @@ -77,11 +82,32 @@ class ElectrumListener: 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.""" + being watched immediately, without waiting for the next reconnect cycle. + + Best-effort, not retried on its own: `self.client` can still become None + between the check below and the task actually running (the connection + drops in between), which used to raise an AssertionError inside an + untracked task and vanish silently (B-30). The exception is now logged + instead, and — since a failure here just means this one address stays + unsubscribed until the next reconnect's `_subscribe_all_users` or the + periodic `DepositReconciler` sweep (also B-30) catches it — that's an + acceptable, self-healing outcome rather than something worth its own + retry/backoff loop. + """ scripthash = address_to_scripthash(address) self._scripthash_to_user[scripthash] = user_id if self.client is not None: - asyncio.create_task(self._subscribe_and_refresh(scripthash, user_id)) + task = asyncio.create_task(self._subscribe_and_refresh(scripthash, user_id)) + self._background_tasks.add(task) + task.add_done_callback(self._background_tasks.discard) + task.add_done_callback(self._log_subscribe_task_failure) + + def _log_subscribe_task_failure(self, task: asyncio.Task) -> None: + if task.cancelled(): + return + exc = task.exception() + if exc is not None: + logger.warning("could not subscribe a newly-registered user's address: %r", exc) async def run(self) -> None: backoff = 1 @@ -179,7 +205,7 @@ class ElectrumListener: async def _subscribe_and_refresh(self, scripthash: str, user_id: int) -> None: assert self.client is not None await self.client.subscribe_scripthash(scripthash) - await self._refresh_user(user_id, scripthash) + await self.refresh_user(user_id, scripthash) def _apply_header(self, header: dict) -> None: """Record a new chain tip, refusing to move backwards. @@ -329,9 +355,9 @@ class ElectrumListener: scripthash, _status = await queue.get() user_id = self._scripthash_to_user.get(scripthash) if user_id is not None: - await self._refresh_user(user_id, scripthash) + await self.refresh_user(user_id, scripthash) - async def _refresh_user(self, user_id: int, scripthash: str) -> None: + async def refresh_user(self, user_id: int, scripthash: str) -> None: """Three phases, so no DB session is held across a network call (B-18), same shape as _trigger_payout: read what's needed, corroborate any candidate external spends against other servers (B-29), then persist. diff --git a/app/main.py b/app/main.py index 03627e6..cc5a5b5 100644 --- a/app/main.py +++ b/app/main.py @@ -23,6 +23,7 @@ from app.auth.routes import router as auth_router from app.api.errors import ApiError from app.config import settings, validate_runtime_secrets from app.db.base import AsyncSessionLocal +from app.deposits.reconcile import DepositReconciler from app.electrum.client import ElectrumClient, ElectrumEndpoint, parse_endpoints from app.electrum.listener import ElectrumListener from app.rounds.scheduler import RoundScheduler @@ -61,6 +62,10 @@ async def lifespan(app: FastAPI): # 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) + # Periodic safety net for deposit crediting/external-spend detection, + # independent of scripthash-change notifications — catches a subscription + # silently lost on an otherwise healthy connection (B-30). + deposit_reconciler = DepositReconciler(AsyncSessionLocal, listener) tasks = [ asyncio.create_task(listener.run()), @@ -68,6 +73,7 @@ async def lifespan(app: FastAPI): asyncio.create_task(poller.run()), asyncio.create_task(bumper.run()), asyncio.create_task(reconciler.run()), + asyncio.create_task(deposit_reconciler.run()), ] try: yield diff --git a/tests/unit/test_deposit_reconcile.py b/tests/unit/test_deposit_reconcile.py new file mode 100644 index 0000000..824e2e8 --- /dev/null +++ b/tests/unit/test_deposit_reconcile.py @@ -0,0 +1,98 @@ +"""Regression tests for B-30: a periodic sweep must catch a deposit whose +scripthash notification was silently lost, independent of whatever the +notification-driven path (electrum/listener.py:refresh_user) is doing.""" + +import pytest +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.db.base import Base +from app.db.models import User +from app.deposits.reconcile import DepositReconciler + + +@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() + + +async def _seed_users(session_factory, addresses: list[str]) -> list[int]: + async with session_factory() as session: + ids = [] + for i, address in enumerate(addresses): + user = User(username=f"user{i}", password_hash="x", derivation_index=i, address=address) + session.add(user) + await session.flush() + ids.append(user.id) + await session.commit() + return ids + + +# Real, decodable PLM bech32 addresses (address_to_scripthash actually parses +# them) — arbitrary otherwise. +_ADDRESSES = [ + "plm1q0xcqpzrky6eff2g52qdye53xkk9jxkvraxkkwd", + "plm1q5x25wd6q463mfhckjraaedgjg0lyu73qfcj43n", + "plm1qqph9qup2mp7w7g5nlsdhdc9m2pp44ampzw0ctx", +] + + +class FakeListener: + def __init__(self, *, fail_for: set[int] | None = None, disconnect_after: int | None = None): + self.client = object() # truthy: "connected" + self.refreshed: list[int] = [] + self._fail_for = fail_for or set() + self._disconnect_after = disconnect_after + + async def refresh_user(self, user_id: int, scripthash: str) -> None: + self.refreshed.append(user_id) + if self._disconnect_after is not None and len(self.refreshed) >= self._disconnect_after: + self.client = None + if user_id in self._fail_for: + raise RuntimeError(f"listunspent failed for user {user_id}") + + +async def test_sweep_once_refreshes_every_user(session_factory): + user_ids = await _seed_users(session_factory, _ADDRESSES) + listener = FakeListener() + reconciler = DepositReconciler(session_factory, listener) + + await reconciler._sweep_once() + + assert listener.refreshed == user_ids + + +async def test_sweep_once_continues_past_a_failing_user(session_factory): + """One user's refresh failing (a transient network hiccup) must not stop the + sweep from reaching the rest — mirrors poll_once's per-item isolation.""" + user_ids = await _seed_users(session_factory, _ADDRESSES) + listener = FakeListener(fail_for={user_ids[1]}) + reconciler = DepositReconciler(session_factory, listener) + + await reconciler._sweep_once() + + assert listener.refreshed == user_ids + + +async def test_sweep_once_stops_when_the_connection_drops_mid_sweep(session_factory): + """No point continuing once the connection is gone — the next reconnect's own + _subscribe_all_users will cover everyone anyway.""" + user_ids = await _seed_users(session_factory, _ADDRESSES) + listener = FakeListener(disconnect_after=1) + reconciler = DepositReconciler(session_factory, listener) + + await reconciler._sweep_once() + + assert listener.refreshed == user_ids[:1] + + +async def test_sweep_once_does_nothing_with_no_users(session_factory): + listener = FakeListener() + reconciler = DepositReconciler(session_factory, listener) + + await reconciler._sweep_once() # must not raise + + assert listener.refreshed == [] diff --git a/tests/unit/test_electrum_listener.py b/tests/unit/test_electrum_listener.py index 9d4568a..12e9afc 100644 --- a/tests/unit/test_electrum_listener.py +++ b/tests/unit/test_electrum_listener.py @@ -1,6 +1,7 @@ """Listener-level behaviour: server rotation on failure (the fallback-servers feature), the chain-tip monotonicity guard (B-19), header validation and -multi-server corroboration (B-28). +multi-server corroboration (B-28), and the new-user subscribe task's retention +and error logging (B-30). 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 @@ -8,6 +9,7 @@ dead connection by moving to the next server instead of retrying the same one. """ import asyncio +import logging import struct import pytest @@ -157,6 +159,40 @@ async def test_listener_with_no_endpoints_gives_up_loudly(session_factory): assert listener.current_endpoint is None +# --- B-30: address_for_new_user's subscribe task must be retained (not fire-and- +# forget) and its failure must be observable, not silently swallowed. ------------- + + +async def test_address_for_new_user_does_nothing_without_a_connection(session_factory): + listener = ElectrumListener(lambda endpoint: None, session_factory, _ENDPOINTS) + listener.address_for_new_user(1, "plm1q0xcqpzrky6eff2g52qdye53xkk9jxkvraxkkwd") # listener.client is None + assert listener._background_tasks == set() + + +async def test_address_for_new_user_retains_and_logs_a_failed_subscribe_task(session_factory, caplog): + """Before B-30, this task was fire-and-forget: an AssertionError (self.client + turning None mid-flight) or any other failure vanished into asyncio's default + unretrieved-exception handler instead of being logged anywhere the operator + could see, and nothing kept the task alive in the meantime.""" + + class FailingClient: + async def subscribe_scripthash(self, scripthash): + raise ConnectionResetError("dropped mid-subscribe") + + listener = ElectrumListener(lambda endpoint: None, session_factory, _ENDPOINTS) + listener.client = FailingClient() + + listener.address_for_new_user(1, "plm1q0xcqpzrky6eff2g52qdye53xkk9jxkvraxkkwd") + assert len(listener._background_tasks) == 1 # retained while in flight + + with caplog.at_level(logging.WARNING): + await asyncio.gather(*list(listener._background_tasks), return_exceptions=True) + await asyncio.sleep(0) # let the done_callbacks (scheduled via call_soon) run + + assert listener._background_tasks == set() # discarded once done + assert "could not subscribe" in caplog.text + + 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 @@ -375,7 +411,7 @@ async def test_corroborate_utxo_spent_false_when_nobody_responds(session_factory class _ActiveClient: """Stands in for `self.client`, the listener's one active connection — - _refresh_user only ever calls listunspent on it.""" + refresh_user only ever calls listunspent on it.""" def __init__(self, entries: list[dict]): self._entries = entries @@ -406,8 +442,8 @@ async def _seed_funded_user(session_factory, *, username: str, address: str) -> _UNRELATED_ENTRY = [{"tx_hash": "99" * 32, "tx_pos": 0, "height": 100, "value": 1_000_000}] -async def test_refresh_user_marks_a_utxo_spent_once_others_corroborate_it(session_factory): - user_id = await _seed_funded_user(session_factory, username="bob", address="plm1qtest") +async def testrefresh_user_marks_a_utxo_spent_once_others_corroborate_it(session_factory): + user_id = await _seed_funded_user(session_factory, username="bob", address="plm1q0xcqpzrky6eff2g52qdye53xkk9jxkvraxkkwd") others_factory = await _listunspent_client_factory( {"first.example": [], "second.example": [], "third.example": []} @@ -415,7 +451,7 @@ async def test_refresh_user_marks_a_utxo_spent_once_others_corroborate_it(sessio listener = ElectrumListener(others_factory, session_factory, _ENDPOINTS) listener.client = _ActiveClient(_UNRELATED_ENTRY) # our own connection no longer sees the UTXO either - await listener._refresh_user(user_id, "scripthash") + await listener.refresh_user(user_id, "scripthash") async with session_factory() as session: utxo = ( @@ -428,7 +464,7 @@ async def test_refresh_user_marks_a_utxo_spent_once_others_corroborate_it(sessio assert user.cached_balance_sats == 1_000_000 -async def test_refresh_user_does_not_mark_when_corroboration_fails(session_factory): +async def testrefresh_user_does_not_mark_when_corroboration_fails(session_factory): """The single most important case: our own connection alone reporting the UTXO missing must not be enough — before B-29 this zeroed the balance on one bad reply.""" @@ -441,7 +477,7 @@ async def test_refresh_user_does_not_mark_when_corroboration_fails(session_facto listener = ElectrumListener(others_factory, session_factory, _ENDPOINTS) listener.client = _ActiveClient(_UNRELATED_ENTRY) - await listener._refresh_user(user_id, "scripthash") + await listener.refresh_user(user_id, "scripthash") async with session_factory() as session: utxo = (