From daf66fd6bc367a60272502dd4d35aeb6dfa1ff82 Mon Sep 17 00:00:00 2001 From: Davide Grilli Date: Mon, 27 Jul 2026 00:32:22 +0200 Subject: [PATCH] Fix the round-open race, the advertised jackpot, and payout error handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round opening (B-09). open_new_round_if_needed now handles the IntegrityError from ix_rounds_single_active (previous commit) by rolling back and using the winner's round. Deviation from the plan in BUGS.md, which proposed making the scheduler the only writer: that would mean the first bet after a cooldown couldn't open a round, so both callers stay and a bounded retry was added instead — a conflict where nothing is active yet just means the winner hadn't committed, and a bet must not fail on that timing. get_active_round also logs loudly if it ever sees more than one active round rather than silently picking the newest. The jackpot (B-11). It was participant_count * the *current* bet_amount_sats, which overstated the pool (each stored bet is already net of that bet's network fee) and silently rewrote the advertised jackpot of a round in progress whenever an operator edited the bet amount. It now sums the participants' stored bet_amount_sats. The remaining imprecision — the payout tx's own fee, deducted from the winner's share and unknowable until the payout is built — is documented in the code rather than promised away, since the comment there claimed exactness. Payout (B-05, B-18). _trigger_payout is split into read / build+broadcast / persist, so no DB session is held across a network call (on SQLite that meant holding the write lock for two unbounded round-trips). That restructuring is also what makes the error handling placeable: it now catches Exception around the chain work and writes a payout_failed audit entry, where a malformed fee_address used to raise EmbitError all the way to the scheduler's catch-all, leaving the round stuck in paying_out with nothing recorded about why. Automatic payout retry remains an open gap. The scheduler also counts "building" participants as in-flight when deciding whether a round may close, matching the two-phase bet write. Co-Authored-By: Claude Opus 5 --- app/api/routes/rounds.py | 19 ++++-- app/rounds/scheduler.py | 105 +++++++++++++++++++++--------- app/rounds/service.py | 66 +++++++++++++++---- tests/unit/test_rounds_route.py | 67 +++++++++++++++++++ tests/unit/test_rounds_service.py | 66 +++++++++++++++++++ 5 files changed, 279 insertions(+), 44 deletions(-) diff --git a/app/api/routes/rounds.py b/app/api/routes/rounds.py index f48d591..7445dea 100644 --- a/app/api/routes/rounds.py +++ b/app/api/routes/rounds.py @@ -118,6 +118,16 @@ async def current_round( participant_count = await session.scalar( select(func.count()).select_from(RoundParticipant).where(RoundParticipant.round_id == round_.id) ) or 0 + # The pool is the sum of what the participants' bets actually paid into the pool + # address — each one is already net of that bet's network fee. Deriving it from + # participant_count * the *current* bet_amount_sats instead overstated it, and + # silently changed the advertised jackpot of a round in progress whenever an + # operator edited the bet amount (B-11). + pool_amount_sats = await session.scalar( + select(func.coalesce(func.sum(RoundParticipant.bet_amount_sats), 0)).where( + RoundParticipant.round_id == round_.id + ) + ) or 0 opened_at = round_.opened_at.replace(tzinfo=timezone.utc) closes_at = opened_at + timedelta(seconds=config.round_duration_seconds) @@ -137,10 +147,11 @@ async def current_round( await session.commit() - # Shown to players as "jackpot": the winner's 70% share of the pool (same - # split rounds/scheduler.py applies at payout time), not the full pool — - # what's displayed should match what the winner actually receives. - pool_amount_sats = participant_count * config.bet_amount_sats + # Shown to players as "jackpot": the winner's 70% share of the pool (same split + # rounds/scheduler.py applies at payout time), not the full pool. It remains an + # upper bound by the payout tx's own fee, which is deducted from the winner's + # share and isn't knowable until the payout is built — a few hundred sat on a + # 1 sat/vB payout, i.e. invisible at PLM amounts, but it is not exact. jackpot_sats = pool_amount_sats * 70 // 100 return CurrentRoundResponse( diff --git a/app/rounds/scheduler.py b/app/rounds/scheduler.py index 1993f7d..614eec7 100644 --- a/app/rounds/scheduler.py +++ b/app/rounds/scheduler.py @@ -75,10 +75,17 @@ class RoundScheduler: broadcaster.publish() async with self._session_factory() as session: + # "building" counts as in-flight too: it's a bet mid-broadcast (see + # bets/service.py's two-phase write). A bet that never confirms is + # eventually removed by app/tx/reconcile.py, which is what stops this + # wait from being unbounded. pending_count = await session.scalar( select(func.count()) .select_from(RoundParticipant) - .where(RoundParticipant.round_id == round_id, RoundParticipant.status == "broadcast") + .where( + RoundParticipant.round_id == round_id, + RoundParticipant.status.in_(("building", "broadcast")), + ) ) if pending_count: return # wait for in-flight bets to confirm before closing; stays "closing" @@ -154,49 +161,73 @@ class RoundScheduler: return self._listener.tip_height, header_hex_to_block_hash(self._listener.tip_header_hex) async def _trigger_payout(self, round_id: int) -> None: + """Three phases, so no DB session is held across a network call (B-18): read + what's needed, do the chain work, then persist the outcome.""" client = self._listener.client if client is None: logger.error("round %s payout deferred: not connected", round_id) return + # --- Phase 1: read (session closed before any network I/O) --------------- async with self._session_factory() as session: round_ = await session.get(Round, round_id) config = await get_round_config(session) - if not config.fee_address: - logger.error( - "round %s payout blocked: no fee_address configured (set it via the admin endpoint)", round_id - ) - return + fee_address = config.fee_address + fee_rate = config.fee_rate_sat_vb + pool_amount_sats = round_.pool_amount_sats + winner_user_id = round_.winner_user_id + winner = await session.get(User, winner_user_id) + winner_address = winner.address if winner is not None else None + await session.commit() # get_round_config may have created the row - winner = await session.get(User, round_.winner_user_id) - winner_share = round_.pool_amount_sats * 70 // 100 - commission_share = round_.pool_amount_sats - winner_share # remainder from rounding goes to fees + if not fee_address: + logger.error( + "round %s payout blocked: no fee_address configured (set it via the admin endpoint)", round_id + ) + return + if winner_address is None: + logger.error("round %s payout blocked: winner user %s not found", round_id, winner_user_id) + return + winner_share = pool_amount_sats * 70 // 100 + commission_share = pool_amount_sats - winner_share # remainder from rounding goes to fees + + # --- Phase 2: build and broadcast ---------------------------------------- + try: pool_key = derive_pool_key() pool_script_obj = script.p2wpkh(pool_key.to_public()) pool_address = pool_script_obj.address(network=PLM_MAINNET) - pool_scripthash = address_to_scripthash(pool_address) - entries = await client.listunspent(pool_scripthash) + entries = await client.listunspent(address_to_scripthash(pool_address)) utxos = [Utxo(e["tx_hash"], e["tx_pos"], e["value"]) for e in entries if e["height"] > 0] - try: - built = build_payout_transaction( - signing_key=pool_key, - from_script=pool_script_obj, - utxos=utxos, - winner_address=winner.address, - winner_share_sats=winner_share, - fee_address=config.fee_address, - commission_sats=commission_share, - change_address=pool_address, - fee_rate_sat_vb=config.fee_rate_sat_vb, - ) - except InsufficientFundsError: - logger.exception("round %s payout failed: insufficient pool UTXOs", round_id) - return - + built = build_payout_transaction( + signing_key=pool_key, + from_script=pool_script_obj, + utxos=utxos, + winner_address=winner_address, + winner_share_sats=winner_share, + fee_address=fee_address, + commission_sats=commission_share, + change_address=pool_address, + fee_rate_sat_vb=fee_rate, + ) await client.broadcast(built.raw_hex) + except InsufficientFundsError: + logger.exception("round %s payout failed: insufficient pool UTXOs", round_id) + return + except Exception: + # Anything else — a malformed fee_address (EmbitError), a rejected + # broadcast, a dead connection. This used to escape all the way to + # run()'s catch-all, which logged it without recording anything, leaving + # no trace of *why* the round was stuck (B-05). The round stays in + # "paying_out" either way: automatic payout retry is still an open gap. + logger.exception("round %s payout failed", round_id) + await self._log_payout_failure(round_id, winner_user_id) + return + # --- Phase 3: persist ----------------------------------------------------- + async with self._session_factory() as session: + round_ = await session.get(Round, round_id) round_.winner_amount_sats = built.winner_sats round_.fee_amount_sats = built.commission_sats round_.payout_txid = built.txid @@ -205,7 +236,7 @@ class RoundScheduler: kind="payout", round_id=round_id, current_txid=built.txid, - fee_rate_sat_vb=config.fee_rate_sat_vb, + fee_rate_sat_vb=fee_rate, raw_tx_hex=built.raw_hex, status="pending", ) @@ -214,9 +245,25 @@ class RoundScheduler: session, "payout_sent", {"txid": built.txid, "winner_sats": built.winner_sats, "commission_sats": built.commission_sats}, - user_id=round_.winner_user_id, + user_id=winner_user_id, round_id=round_id, ) await session.commit() logger.info("round %s payout broadcast: txid=%s", round_id, built.txid) + + async def _log_payout_failure(self, round_id: int, winner_user_id: int | None) -> None: + """Leaves an operator-visible trace in the audit log for a round stuck in + "paying_out" — the logs alone don't show up in /admin.""" + try: + async with self._session_factory() as session: + await write_audit_log( + session, + "payout_failed", + {"round_id": round_id}, + user_id=winner_user_id, + round_id=round_id, + ) + await session.commit() + except Exception: + logger.exception("could not record the payout failure of round %s", round_id) diff --git a/app/rounds/service.py b/app/rounds/service.py index c6c25ef..c8e9c7a 100644 --- a/app/rounds/service.py +++ b/app/rounds/service.py @@ -1,20 +1,43 @@ +import logging from datetime import datetime, timedelta, timezone from sqlalchemy import select +from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.db.models import Round from app.rounds.config import get_round_config from app.rounds.events import broadcaster +logger = logging.getLogger(__name__) + _ACTIVE_STATUSES = ("open", "closing", "drawing", "paying_out") +# Bounded: a conflict means someone else is opening a round right now, so a couple +# of retries is plenty. Unbounded retries could spin if the invariant were ever +# broken in a way we don't anticipate. +_OPEN_ROUND_ATTEMPTS = 3 + async def get_active_round(session: AsyncSession) -> Round | None: """The round currently in progress (in any non-closed state), if any. Rounds never overlap: a new round only opens once the previous one is fully closed - (payout confirmed, or no participants to pay out).""" - return await session.scalar(select(Round).where(Round.status.in_(_ACTIVE_STATUSES)).order_by(Round.id.desc())) + (payout confirmed, or no participants to pay out). + + The database enforces "at most one active round" (ix_rounds_single_active, see + app/db/models.py), so the ordering below is belt-and-braces; if it ever does + see two, that's a broken invariant and worth a loud log rather than silently + picking one.""" + active = ( + await session.scalars(select(Round).where(Round.status.in_(_ACTIVE_STATUSES)).order_by(Round.id.desc())) + ).all() + if len(active) > 1: + logger.error( + "invariant violated: %s rounds are active at once (ids=%s) — using the newest", + len(active), + [r.id for r in active], + ) + return active[0] if active else None def round_accepts_bets(round_: Round, round_duration_seconds: int) -> bool: @@ -55,12 +78,33 @@ async def open_new_round_if_needed(session: AsyncSession) -> Round | None: if datetime.now(timezone.utc) < closed_at + timedelta(seconds=config.round_cooldown_seconds): return None - round_ = Round(status="open") - session.add(round_) - await session.flush() - # Published pre-commit (the caller commits right after) — acceptable: this - # only tells subscribers "go refetch", and by the time an SSE client's - # refetch request actually lands, this in-process commit (microseconds - # away) has essentially always already happened. - broadcaster.publish() - return round_ + for attempt in range(_OPEN_ROUND_ATTEMPTS): + round_ = Round(status="open") + session.add(round_) + try: + await session.flush() + except IntegrityError: + # Another caller (the scheduler tick, or a concurrent place_bet) got + # there first — ix_rounds_single_active turns what used to be two live + # rounds into a clean failure here. Roll our insert back and use theirs. + # Safe to roll back: this runs before its callers have written anything + # else in this session. + await session.rollback() + existing = await get_active_round(session) + if existing is not None: + logger.info("lost the race to open a round; using round %s", existing.id) + return existing + # Nothing active *and* the insert conflicted: the winner's transaction + # hadn't committed yet when we looked. Try again rather than failing the + # caller — a bet shouldn't 500 because of a scheduler tick's timing. + logger.info("round-open conflict with nothing active yet (attempt %s), retrying", attempt + 1) + continue + # Published pre-commit (the caller commits right after) — acceptable: this + # only tells subscribers "go refetch", and by the time an SSE client's + # refetch request actually lands, this in-process commit (microseconds + # away) has essentially always already happened. + broadcaster.publish() + return round_ + + logger.error("could not open a round after %s attempts", _OPEN_ROUND_ATTEMPTS) + return await get_active_round(session) diff --git a/tests/unit/test_rounds_route.py b/tests/unit/test_rounds_route.py index 1ba59bd..0719562 100644 --- a/tests/unit/test_rounds_route.py +++ b/tests/unit/test_rounds_route.py @@ -94,3 +94,70 @@ async def test_user_played_true_only_for_participants(client): resp = await ac.get("/rounds/current") # no auth at all — logged-out chain-only view assert resp.status_code == 200 assert resp.json()["user_played"] is False + + +async def test_jackpot_comes_from_the_participants_actual_bets(client): + """B-11: the jackpot was participant_count * the *current* bet_amount_sats, which + overstated it (each stored bet is already net of that bet's network fee) and + silently rewrote the advertised jackpot of a round in progress whenever an + operator edited the bet amount.""" + from sqlalchemy import select + + from app.db.models import Round, RoundConfig, RoundParticipant + + ac, session_factory = client + + async with session_factory() as session: + session.add(RoundConfig(fee_address="", bet_amount_sats=1_000_000_000)) + session.add(Round(id=50, status="open")) + await session.flush() + # Two bets that actually paid 999_800_000 each (fee deducted), not 1_000_000_000. + session.add( + RoundParticipant(round_id=50, user_id=1, bet_amount_sats=999_800_000, bet_txid="a", status="confirmed") + ) + session.add( + RoundParticipant(round_id=50, user_id=2, bet_amount_sats=999_800_000, bet_txid="b", status="confirmed") + ) + await session.commit() + + body = (await ac.get("/rounds/current")).json() + assert body["participant_count"] == 2 + assert body["jackpot_sats"] == (999_800_000 * 2) * 70 // 100 + + # Changing the configured bet amount must not move a running round's jackpot. + async with session_factory() as session: + config = (await session.scalars(select(RoundConfig))).one() + config.bet_amount_sats = 5_000_000_000 + await session.commit() + + body = (await ac.get("/rounds/current")).json() + assert body["jackpot_sats"] == (999_800_000 * 2) * 70 // 100 + + +async def test_unhandled_errors_use_the_structured_detail_shape(client): + """B-24: the catch-all handler answered with a bare-string `detail`, while + app/api/errors.py documents detail as {"code", "message", "params"}. Clients then + had to special-case exactly the responses they understand least.""" + from fastapi import FastAPI + from httpx import ASGITransport, AsyncClient + + from app.main import log_unhandled_exception + + app = FastAPI() + app.add_exception_handler(Exception, log_unhandled_exception) + + @app.get("/boom") + async def boom(): + raise RuntimeError("secret internal detail") + + transport = ASGITransport(app=app, raise_app_exceptions=False) + async with AsyncClient(transport=transport, base_url="http://test") as ac: + resp = await ac.get("/boom") + + assert resp.status_code == 500 + detail = resp.json()["detail"] + assert detail["code"] == "internal_error" + assert detail["message"] == "internal server error" + assert detail["params"] == {} + # The exception text belongs in logs/app.log, never in the response body. + assert "secret internal detail" not in resp.text diff --git a/tests/unit/test_rounds_service.py b/tests/unit/test_rounds_service.py index f5bb8aa..45608b4 100644 --- a/tests/unit/test_rounds_service.py +++ b/tests/unit/test_rounds_service.py @@ -1,6 +1,7 @@ from datetime import datetime, timedelta, timezone import pytest +from sqlalchemy import select from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.db.base import Base @@ -105,3 +106,68 @@ async def test_pause_does_not_interrupt_a_round_in_progress(session_factory): returned = await open_new_round_if_needed(session) assert returned is not None assert returned.status == "drawing" + + +async def test_losing_the_open_race_reuses_the_winning_round(session_factory, monkeypatch): + """B-09: open_new_round_if_needed was a read-then-insert with no lock, called from + both the scheduler and every place_bet, so two callers could both see "no active + round" and insert one — and a second stuck "open" row blocks every future round, + since get_active_round matches on status. + + The race is forced deterministically: the round already exists and is committed, + but this caller's first look is made to miss it (exactly what the loser of the + race sees). The insert then hits ix_rounds_single_active, and the caller must + recover by using the winner's round instead of raising at its caller — a bet must + not fail because a scheduler tick beat it by a millisecond. + """ + from app.rounds import service as service_module + + async with session_factory() as session: + session.add(Round(status="open")) + await session.commit() + + real_get_active_round = service_module.get_active_round + calls = {"n": 0} + + async def blind_first_look(session): + calls["n"] += 1 + if calls["n"] == 1: + return None # what the loser of the race sees + return await real_get_active_round(session) + + monkeypatch.setattr(service_module, "get_active_round", blind_first_look) + + async with session_factory() as session: + round_ = await service_module.open_new_round_if_needed(session) + await session.commit() + + assert round_ is not None # recovered, didn't raise + async with session_factory() as session: + rounds = (await session.scalars(select(Round))).all() + assert len(rounds) == 1, f"expected one round, got {[(r.id, r.status) for r in rounds]}" + assert round_.id == rounds[0].id # the winner's round, not a second one + + +async def test_the_database_refuses_a_second_active_round(session_factory): + """The guarantee itself, independent of the application code path.""" + from sqlalchemy.exc import IntegrityError + + async with session_factory() as session: + session.add(Round(status="open")) + await session.commit() + + async with session_factory() as session: + session.add(Round(status="drawing")) + with pytest.raises(IntegrityError): + await session.commit() + + +async def test_closed_rounds_can_coexist_with_an_active_one(session_factory): + async with session_factory() as session: + session.add(Round(status="closed")) + session.add(Round(status="closed")) + session.add(Round(status="open")) + await session.commit() + + async with session_factory() as session: + assert len((await session.scalars(select(Round))).all()) == 3