Add a periodic deposit reconciler, and stop losing subscribe tasks (B-30)

Deposits were credited exclusively by scripthash-change notifications,
with nothing re-verifying a user's balance against the chain if a
subscription was ever silently lost. address_for_new_user's subscribe
was fire-and-forget: the task wasn't retained, so it could be
garbage-collected mid-flight, and any failure (including self.client
turning None between the check and the task running) vanished into
asyncio's default unretrieved-exception handler instead of being
logged anywhere. On an otherwise healthy, long-lived connection there
may be no reconnect for days to re-subscribe everyone, so a user in
that state never saw their deposits.

address_for_new_user now retains the task and logs its exception if
it fails. New app/deposits/reconcile.py adds DepositReconciler, a
periodic sweep (every 5 minutes, gated on the Electrum client being
connected, same shape as tx/reconcile.py) that round-robins over every
user and calls the listener's own refresh_user (renamed from
_refresh_user since it's now called from outside the class) - so the
notification-driven and periodic paths can never behave differently.
Deliberately sweeps every user rather than only ones missing from the
in-memory scripthash map, since that map can't tell "never subscribed"
apart from "subscribed, but the server stopped delivering
notifications for it". Wired into app/main.py's lifespan alongside the
other three background reconcilers.

Suite grows from 176 to 182 tests. BUGS.md moves B-30 to Previously
fixed.

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
+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 20 are High/Medium/Low.
+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
+coverage — every fix lands with a regression test (the six fixes so far brought the suite from
-139 to 176).
+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
+no equivalent resume-after-restart or stall visibility (B-36).
-deposit-side reconciler independent of scripthash notifications (B-30).
 
 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,

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)

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.
 

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.

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

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 == []

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):
+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="plm1qtest")
+    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 = (