From 5ce49d7b88552a23988e7f88a9bdd865a6475a80 Mon Sep 17 00:00:00 2001 From: Davide Grilli Date: Tue, 21 Jul 2026 10:26:18 +0200 Subject: [PATCH] Add round lifecycle, draw algorithm and scheduler Periodic scheduler (configurable round duration) that closes a round only once all broadcast bets confirm, waits for the next block after closing, draws a winner via block-hash-seeded modulo over participants ordered by broadcast time, triggers the 70/30 payout, and only opens the next round once that payout confirms. Draw logic is isolated in draw.py as a deliberately simple, replaceable component. Co-Authored-By: Claude Sonnet 5 --- app/rounds/__init__.py | 0 app/rounds/config.py | 17 +++ app/rounds/confirmation.py | 16 +++ app/rounds/draw.py | 21 +++ app/rounds/scheduler.py | 206 ++++++++++++++++++++++++++++++ app/rounds/service.py | 27 ++++ tests/unit/test_draw.py | 32 +++++ tests/unit/test_rounds_service.py | 58 +++++++++ tests/unit/test_scheduler.py | 55 ++++++++ 9 files changed, 432 insertions(+) create mode 100644 app/rounds/__init__.py create mode 100644 app/rounds/config.py create mode 100644 app/rounds/confirmation.py create mode 100644 app/rounds/draw.py create mode 100644 app/rounds/scheduler.py create mode 100644 app/rounds/service.py create mode 100644 tests/unit/test_draw.py create mode 100644 tests/unit/test_rounds_service.py create mode 100644 tests/unit/test_scheduler.py diff --git a/app/rounds/__init__.py b/app/rounds/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/rounds/config.py b/app/rounds/config.py new file mode 100644 index 0000000..55bee3f --- /dev/null +++ b/app/rounds/config.py @@ -0,0 +1,17 @@ +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import settings +from app.db.models import RoundConfig + + +async def get_round_config(session: AsyncSession) -> RoundConfig: + """Single-row operational config, lazily seeded from settings defaults on + first use. fee_address starts empty until an operator sets it (admin + endpoint, stage 10) — payouts must refuse to run until it's set.""" + config = await session.scalar(select(RoundConfig)) + if config is None: + config = RoundConfig(fee_address="", bet_amount_sats=settings.bet_amount_sats) + session.add(config) + await session.flush() + return config diff --git a/app/rounds/confirmation.py b/app/rounds/confirmation.py new file mode 100644 index 0000000..c11991c --- /dev/null +++ b/app/rounds/confirmation.py @@ -0,0 +1,16 @@ +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import PendingTransaction, Round +from app.tx.confirmation import register_handler + + +async def _on_payout_confirmed(session: AsyncSession, pending: PendingTransaction) -> None: + round_ = await session.scalar(select(Round).where(Round.payout_txid == pending.current_txid)) + if round_ is not None and round_.status == "paying_out": + round_.status = "closed" + # The winner's own address is already watched by the Electrum listener, so + # their balance is credited by the normal deposit path once this confirms. + + +register_handler("payout", _on_payout_confirmed) diff --git a/app/rounds/draw.py b/app/rounds/draw.py new file mode 100644 index 0000000..9e26ffc --- /dev/null +++ b/app/rounds/draw.py @@ -0,0 +1,21 @@ +import hashlib + + +def header_hex_to_block_hash(header_hex: str) -> str: + """Block hash from a raw Electrum header: sha256d, byte-reversed, hex. + Verified against a real mainnet block (blockchain.transaction.get's own + reported blockhash) during development.""" + header_bytes = bytes.fromhex(header_hex) + digest = hashlib.sha256(hashlib.sha256(header_bytes).digest()).digest() + return digest[::-1].hex() + + +def draw_winner(participants: list[str], block_hash_hex: str) -> str: + """v1 draw algorithm (flowchart.mmd, node R): seed = block hash as an integer, + index = seed mod participant_count, winner = participants[index]. Anyone can + recompute and verify it from public data. Deliberately simple/replaceable.""" + if not participants: + raise ValueError("no participants to draw from") + seed = int(block_hash_hex, 16) + index = seed % len(participants) + return participants[index] diff --git a/app/rounds/scheduler.py b/app/rounds/scheduler.py new file mode 100644 index 0000000..34db08c --- /dev/null +++ b/app/rounds/scheduler.py @@ -0,0 +1,206 @@ +import asyncio +import logging +from datetime import datetime, timedelta, timezone + +from embit import script +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import async_sessionmaker + +from app.audit.log import write_audit_log +from app.config import settings +from app.db.models import PendingTransaction, Round, RoundParticipant, User +from app.electrum.listener import ElectrumListener +from app.electrum.scripthash import address_to_scripthash +from app.rounds.config import get_round_config +from app.rounds.draw import draw_winner, header_hex_to_block_hash +from app.rounds.service import open_new_round_if_needed +from app.wallet.hd import derive_pool_key +from app.wallet.plm_network import PLM_MAINNET +from app.wallet.psbt_builder import InsufficientFundsError, Utxo, build_payout_transaction + +logger = logging.getLogger(__name__) + +_TICK_INTERVAL_SECONDS = 5 + + +class RoundScheduler: + """Background task implementing flowchart.mmd's DRAW subgraph: closes the + round on its timer (once any in-flight bets have confirmed), draws a winner + from the next confirmed block, and broadcasts the payout. The next round only + opens once this one is fully closed (rounds/service.get_active_round).""" + + def __init__(self, session_factory: async_sessionmaker, listener: ElectrumListener): + self._session_factory = session_factory + self._listener = listener + + async def run(self) -> None: + while True: + try: + await self._tick() + except asyncio.CancelledError: + raise + except Exception: + logger.exception("round scheduler tick failed") + await asyncio.sleep(_TICK_INTERVAL_SECONDS) + + async def _tick(self) -> None: + if self._listener.client is None: + return + + async with self._session_factory() as session: + round_ = await open_new_round_if_needed(session) + await session.commit() + round_id, status, opened_at = round_.id, round_.status, round_.opened_at + + if status != "open": + return # already closing/drawing/paying_out; progress happens elsewhere + + opened_at = opened_at.replace(tzinfo=timezone.utc) + if datetime.now(timezone.utc) < opened_at + timedelta(seconds=settings.round_duration_seconds): + return + + async with self._session_factory() as session: + pending_count = await session.scalar( + select(func.count()) + .select_from(RoundParticipant) + .where(RoundParticipant.round_id == round_id, RoundParticipant.status == "broadcast") + ) + if pending_count: + return # wait for in-flight bets to confirm before closing + + await self._close_and_draw(round_id) + + async def _close_and_draw(self, round_id: int) -> None: + async with self._session_factory() as session: + round_ = await session.get(Round, round_id) + round_.status = "closing" + round_.closed_at = datetime.now(timezone.utc) + + participants = ( + await session.scalars( + select(RoundParticipant) + .where(RoundParticipant.round_id == round_id, RoundParticipant.status == "confirmed") + .order_by(RoundParticipant.broadcast_at) + ) + ).all() + + if not participants: + round_.status = "closed" + await write_audit_log(session, "round_closed", {"participants": 0}, round_id=round_id) + await session.commit() + logger.info("round %s closed with no participants", round_id) + return + + pool_amount = sum(p.bet_amount_sats for p in participants) + addresses: list[str] = [] + user_by_address: dict[str, int] = {} + for p in participants: + user = await session.get(User, p.user_id) + addresses.append(user.address) + user_by_address[user.address] = user.id + + round_.status = "drawing" + await session.commit() + + tip_at_close = self._listener.tip_height + block_height, block_hash = await self._wait_for_next_block(tip_at_close) + winner_address = draw_winner(addresses, block_hash) + + async with self._session_factory() as session: + round_ = await session.get(Round, round_id) + round_.draw_block_height = block_height + round_.draw_block_hash = block_hash + round_.seed_int = str(int(block_hash, 16)) + round_.winner_user_id = user_by_address[winner_address] + round_.pool_amount_sats = pool_amount + round_.status = "paying_out" + await write_audit_log( + session, + "winner_drawn", + { + "winner_address": winner_address, + "pool_amount_sats": pool_amount, + "block_height": block_height, + "block_hash": block_hash, + "participants": len(addresses), + }, + user_id=user_by_address[winner_address], + round_id=round_id, + ) + await session.commit() + + logger.info("round %s: winner=%s pool=%s", round_id, winner_address, pool_amount) + await self._trigger_payout(round_id) + + async def _wait_for_next_block(self, tip_at_close: int) -> tuple[int, str]: + while self._listener.tip_height <= tip_at_close or not self._listener.tip_header_hex: + await asyncio.sleep(_TICK_INTERVAL_SECONDS) + return self._listener.tip_height, header_hex_to_block_hash(self._listener.tip_header_hex) + + async def _trigger_payout(self, round_id: int) -> None: + client = self._listener.client + if client is None: + logger.error("round %s payout deferred: not connected", round_id) + return + + 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 + + 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 + + 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) + 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=settings.fee_rate_sat_vb, + ) + except InsufficientFundsError: + logger.exception("round %s payout failed: insufficient pool UTXOs", round_id) + return + + await client.broadcast(built.raw_hex) + + round_.winner_amount_sats = built.winner_sats + round_.fee_amount_sats = built.commission_sats + round_.payout_txid = built.txid + session.add( + PendingTransaction( + kind="payout", + round_id=round_id, + current_txid=built.txid, + fee_rate_sat_vb=settings.fee_rate_sat_vb, + raw_tx_hex=built.raw_hex, + status="pending", + ) + ) + await write_audit_log( + session, + "payout_sent", + {"txid": built.txid, "winner_sats": built.winner_sats, "commission_sats": built.commission_sats}, + user_id=round_.winner_user_id, + round_id=round_id, + ) + await session.commit() + + logger.info("round %s payout broadcast: txid=%s", round_id, built.txid) diff --git a/app/rounds/service.py b/app/rounds/service.py new file mode 100644 index 0000000..e7c219c --- /dev/null +++ b/app/rounds/service.py @@ -0,0 +1,27 @@ +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import Round + +_ACTIVE_STATUSES = ("open", "closing", "drawing", "paying_out") + + +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())) + + +async def open_new_round_if_needed(session: AsyncSession) -> Round: + """Returns the active round if one exists (whatever its status), otherwise + opens a fresh one. Callers that need to attach a bet must additionally check + the returned round's status == "open" — a round in closing/drawing/paying_out + isn't accepting new bets, but a new round can't open until it's done.""" + active = await get_active_round(session) + if active is not None: + return active + round_ = Round(status="open") + session.add(round_) + await session.flush() + return round_ diff --git a/tests/unit/test_draw.py b/tests/unit/test_draw.py new file mode 100644 index 0000000..4f1bea6 --- /dev/null +++ b/tests/unit/test_draw.py @@ -0,0 +1,32 @@ +import pytest + +from app.rounds.draw import draw_winner, header_hex_to_block_hash + + +def test_header_hex_to_block_hash_matches_known_mainnet_block(): + # Real PLM mainnet block 477486: header from blockchain.block.header, hash + # cross-checked against the blockhash reported by blockchain.transaction.get + # for a tx confirmed in that block. + header_hex = ( + "0020ed30fbc39ee45200d10214f5107c5f36e7bf753032fd1d3279810c170000000000" + "009a4ea723f732be4c538f3a2490837dd6560103d73ece606aa7d578a66700447b28875" + "e6a47a61b1ad8012582" + ) + known_block_hash = "00000000000008788b55ade13b74d54ceffda9e54315b802411be1ca65064e86" + assert header_hex_to_block_hash(header_hex) == known_block_hash + + +def test_draw_winner_is_deterministic_and_within_range(): + participants = ["addrA", "addrB", "addrC"] + block_hash = "00" * 31 + "05" # seed = 5, index = 5 % 3 = 2 + assert draw_winner(participants, block_hash) == "addrC" + + +def test_draw_winner_single_participant_always_wins(): + block_hash = "ff" * 32 + assert draw_winner(["only"], block_hash) == "only" + + +def test_draw_winner_raises_on_empty_participants(): + with pytest.raises(ValueError): + draw_winner([], "00" * 32) diff --git a/tests/unit/test_rounds_service.py b/tests/unit/test_rounds_service.py new file mode 100644 index 0000000..0ef8937 --- /dev/null +++ b/tests/unit/test_rounds_service.py @@ -0,0 +1,58 @@ +import pytest +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.db.base import Base +from app.db.models import Round +from app.rounds.service import get_active_round, open_new_round_if_needed + + +@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 test_opens_a_round_when_none_exists(session_factory): + async with session_factory() as session: + round_ = await open_new_round_if_needed(session) + await session.commit() + assert round_.status == "open" + + +async def test_reuses_existing_open_round(session_factory): + async with session_factory() as session: + first = await open_new_round_if_needed(session) + await session.commit() + first_id = first.id + + async with session_factory() as session: + second = await open_new_round_if_needed(session) + assert second.id == first_id + + +@pytest.mark.parametrize("status", ["closing", "drawing", "paying_out"]) +async def test_does_not_open_new_round_while_previous_is_in_progress(session_factory, status): + async with session_factory() as session: + session.add(Round(status=status)) + await session.commit() + + async with session_factory() as session: + active = await get_active_round(session) + assert active is not None + assert active.status == status + # open_new_round_if_needed must return the in-progress round, not open a new one + returned = await open_new_round_if_needed(session) + assert returned.status == status + + +async def test_opens_new_round_after_previous_is_closed(session_factory): + async with session_factory() as session: + session.add(Round(status="closed")) + await session.commit() + + async with session_factory() as session: + round_ = await open_new_round_if_needed(session) + assert round_.status == "open" diff --git a/tests/unit/test_scheduler.py b/tests/unit/test_scheduler.py new file mode 100644 index 0000000..05b2791 --- /dev/null +++ b/tests/unit/test_scheduler.py @@ -0,0 +1,55 @@ +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.config import settings +from app.db.base import Base +from app.db.models import Round +from app.rounds.scheduler import RoundScheduler + + +class FakeListener: + client = object() # truthy sentinel; _tick only checks "is not None" + tip_height = 100 + tip_header_hex = "00" + + +@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 test_tick_survives_sqlite_naive_datetime_roundtrip(session_factory, monkeypatch): + """Regression test: SQLite drops tzinfo on round-trip, so opened_at comes back + naive even though it was written as an aware UTC datetime. A prior bug compared + it directly against datetime.now(timezone.utc) and crashed with + "can't compare offset-naive and offset-aware datetimes" on every tick once a + round existed — this must not happen.""" + monkeypatch.setattr(settings, "round_duration_seconds", 3600) # not due yet + async with session_factory() as session: + session.add(Round(status="open", opened_at=datetime.now(timezone.utc))) + await session.commit() + + scheduler = RoundScheduler(session_factory, FakeListener()) + await scheduler._tick() # must not raise + + +async def test_tick_closes_round_with_no_participants_once_due(session_factory, monkeypatch): + monkeypatch.setattr(settings, "round_duration_seconds", 1) + past = datetime.now(timezone.utc) - timedelta(seconds=10) + async with session_factory() as session: + session.add(Round(status="open", opened_at=past)) + await session.commit() + + scheduler = RoundScheduler(session_factory, FakeListener()) + await scheduler._tick() + + async with session_factory() as session: + round_ = (await session.scalars(select(Round))).one() + assert round_.status == "closed"