davide/plm-lottery
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
64f62291d2c0cbfe4e2724df409361ed624ecd85
plm-lottery/app/rounds/scheduler.py
T
davideandClaude Opus 5 64f62291d2 Close the window where a bet pays into a round it was left out of (B-53) 
place_bet commits its participant row as "building" before broadcasting (B-08's
two-phase write), while the scheduler flips the round "open" -> "closing" in one
transaction and counts in-flight participants in another. A bet whose deadline
check passed just before that flip could commit in between: the count saw zero,
so the round drew and paid out over the "confirmed" participants only, while the
bet confirmed normally and its sats landed in the pool address — credited to no
round, to no participant, with no refund path, silently improving the next
round's payout change.

Two locks on the same door:

- place_bet re-checks the deadline after building and signing (the first check
  happens before the UTXO scan, so a slow build could carry a bet past it), then
  commits the participant row behind a compare-and-set on the round's own row,
  UPDATE rounds ... WHERE status = 'open'. That UPDATE takes SQLite's write lock,
  so the two transactions can no longer interleave: either the bet commits first
  and the scheduler's in-flight count sees it, or the flip commits first and the
  guard matches zero rows and refuses the bet with round_closing before anything
  is broadcast. A write-snapshot conflict (OperationalError) is the same
  situation and gets the same answer. Nothing has been broadcast at that point,
  so the rollback releases the UTXOs and leaves no rows behind.

- _close_and_draw re-counts in-flight bets in the same session it snapshots the
  participants from, and returns with the round still "closing" if it finds any.
  Redundant given the CAS, and cheap: it fails safe and the next tick retries.

No new error code — a bet refused this way is exactly the "round is closing"
case the user already sees.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-03 22:06:21 +02:00

496 lines
24 KiB
Python
Raw Blame History

import asyncio
import logging
from datetime import datetime, timedelta, timezone
from embit import script
from embit.transaction import Transaction
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from app.audit.log import write_audit_log
from app.db.models import AuditLog, 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.events import broadcaster
from app.rounds.service import open_new_round_if_needed, winner_share
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
# B-26: how long to wait after a payout failure before automatically retrying it.
# Long enough that a persistently-broken payout (misconfigured fee_address,
# insufficient pool UTXOs) doesn't re-attempt — and re-write a payout_failed audit
# entry — every _TICK_INTERVAL_SECONDS; short enough that a transient failure
# (a dropped Electrum connection, a momentarily-empty pool) self-heals quickly.
_PAYOUT_RETRY_INTERVAL_SECONDS = 60
# B-36: _wait_for_next_block has no timeout of its own — a round can legitimately
# wait several PLM blocks (120s each) for its draw entropy, and re-waits on a
# corroboration failure. These only make an already-long wait *observable*, they
# never cut it short.
_DRAW_PROGRESS_LOG_INTERVAL_SECONDS = 60
_DRAW_STALL_THRESHOLD_SECONDS = 360 # a few multiples of PLM's 120s block time
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()
if round_ is None:
return # still in the cooldown window after the last round closed
round_id, status, opened_at = round_.id, round_.status, round_.opened_at
round_duration_seconds = (await get_round_config(session)).round_duration_seconds
if status == "paying_out":
# B-26: _trigger_payout used to run exactly once, from _close_and_draw —
# any failure after that (no Electrum client, insufficient pool UTXOs, a
# rejected broadcast) or a process restart while paying_out left the round
# wedged here forever. Every tick now re-checks and retries, throttled by
# _retry_payout_if_due so a persistent failure doesn't retry on every tick.
await self._retry_payout_if_due(round_id)
return
if status not in ("open", "closing"):
return # "drawing" — progress happens inside the in-flight _close_and_draw call
if status == "open":
opened_at = opened_at.replace(tzinfo=timezone.utc)
if datetime.now(timezone.utc) < opened_at + timedelta(seconds=round_duration_seconds):
return
async with self._session_factory() as session:
round_ = await session.get(Round, round_id)
# No new bets from here on, regardless of how long the pending-bet
# wait below takes — flip to "closing" immediately so it's observable
# via /rounds/current (e.g. "round closed, waiting for jackpot
# confirmation") instead of silently staying "open" past the deadline.
round_.status = "closing"
round_.closed_at = datetime.now(timezone.utc)
await session.commit()
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.in_(("building", "broadcast")),
)
)
if pending_count:
return # wait for in-flight bets to confirm before closing; stays "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)
# B-53: re-check for in-flight bets in the *same* session that snapshots
# the participants. _tick's check ran in a session of its own, so a bet
# committing its "building" row in between was counted by neither: the
# round drew and paid out without it, while its sats still landed in the
# pool. place_bet's compare-and-set on the round row is what makes that
# window unreachable; this is the cheap second lock on the same door, and
# it fails safe — the round stays "closing" and the next tick retries.
pending_count = await session.scalar(
select(func.count())
.select_from(RoundParticipant)
.where(
RoundParticipant.round_id == round_id,
RoundParticipant.status.in_(("building", "broadcast")),
)
)
if pending_count:
logger.info(
"round %s: %s bet(s) still in flight at close time, waiting", round_id, pending_count
)
return
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()
broadcaster.publish()
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"
drawing_started_at = datetime.now(timezone.utc)
round_.drawing_started_at = drawing_started_at
await session.commit()
broadcaster.publish()
tip_at_close = self._listener.tip_height
block_height, block_hash = await self._wait_for_next_block(round_id, tip_at_close, drawing_started_at)
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()
broadcaster.publish()
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, round_id: int, tip_at_close: int, waiting_since: datetime
) -> tuple[int, str]:
"""Waits for a block after tip_at_close and, before handing it back as the
draw's entropy source, requires it to be corroborated by the other
configured Electrum servers (B-28) — our own active connection is exactly
the thing a hostile server or a MITM would control, so its header alone is
not enough to seed a payout. A candidate that fails corroboration is never
used: this keeps waiting for a further block and tries corroborating that
one instead, logging why every time so a stuck draw is visible in
/admin's audit log rather than a silent, unexplained wait.
This wait has no timeout — it can't, since the draw's entropy genuinely
depends on a future block. B-36: what it lacked was *visibility*, so a
connection that stopped advancing the tip left the round silently frozen
in "drawing" with nothing in the logs or /admin to explain why. Progress
is now logged periodically, and past _DRAW_STALL_THRESHOLD_SECONDS a
draw_stalled audit entry is written (and re-written every threshold
interval for as long as the stall continues) so the wait shows up next
to the draw_header_corroboration_failed entries above.
"""
next_progress_log_at = waiting_since + timedelta(seconds=_DRAW_PROGRESS_LOG_INTERVAL_SECONDS)
next_stall_audit_at = waiting_since + timedelta(seconds=_DRAW_STALL_THRESHOLD_SECONDS)
while True:
while self._listener.tip_height <= tip_at_close or not self._listener.tip_header_hex:
now = datetime.now(timezone.utc)
if now >= next_progress_log_at:
logger.info(
"round %s: still waiting for a block past height %s (%.0fs since drawing started)",
round_id,
tip_at_close,
(now - waiting_since).total_seconds(),
)
next_progress_log_at = now + timedelta(seconds=_DRAW_PROGRESS_LOG_INTERVAL_SECONDS)
if now >= next_stall_audit_at:
async with self._session_factory() as session:
await write_audit_log(
session,
"draw_stalled",
{
"tip_at_close": tip_at_close,
"current_tip_height": self._listener.tip_height,
"elapsed_seconds": int((now - waiting_since).total_seconds()),
},
round_id=round_id,
)
await session.commit()
next_stall_audit_at = now + timedelta(seconds=_DRAW_STALL_THRESHOLD_SECONDS)
await asyncio.sleep(_TICK_INTERVAL_SECONDS)
height = self._listener.tip_height
block_hash = header_hex_to_block_hash(self._listener.tip_header_hex)
if await self._listener.corroborate_header(height, block_hash):
return height, block_hash
logger.error(
"round %s: block %s header %s could not be corroborated by other Electrum servers; "
"waiting for a further block",
round_id,
height,
block_hash,
)
async with self._session_factory() as session:
await write_audit_log(
session,
"draw_header_corroboration_failed",
{"height": height, "reported_hash": block_hash},
round_id=round_id,
)
await session.commit()
tip_at_close = height
async def _retry_payout_if_due(self, round_id: int) -> None:
"""B-26: whether a "paying_out" round is due for another payout attempt.
Throttled by the most recent payout_failed audit entry for this round
(written by _log_payout_failure on every early return in _trigger_payout,
including ones that used to fail silently) rather than by any new DB state,
since a failed attempt doesn't necessarily leave a PendingTransaction behind
(a build failure like a missing fee_address never gets that far). No entry
yet means this round hasn't failed before — either it's a fresh "paying_out"
(the very first call already happened from _close_and_draw and hasn't had a
chance to fail yet) or the process restarted before ever recording one —
either way it's due immediately.
"""
async with self._session_factory() as session:
last_failure_at = await session.scalar(
select(AuditLog.created_at)
.where(AuditLog.event_type == "payout_failed", AuditLog.round_id == round_id)
.order_by(AuditLog.id.desc())
.limit(1)
)
if last_failure_at is not None:
last_failure_at = last_failure_at.replace(tzinfo=timezone.utc)
if datetime.now(timezone.utc) < last_failure_at + timedelta(seconds=_PAYOUT_RETRY_INTERVAL_SECONDS):
return # too soon — avoid hammering a persistently-broken payout
await self._trigger_payout(round_id)
async def _trigger_payout(self, round_id: int) -> None:
"""Four phases, so no DB session is held across a network call (B-18): read
what's needed, build the tx, persist the intent, then broadcast.
The persist happens *before* the broadcast (B-25) — the same two-phase shape
as place_bet/request_withdrawal (B-08): a crash between building the payout
and recording it used to leave money on-chain with zero trace in the DB (no
payout_txid, no PendingTransaction), so a manual retry would have paid the
winner a second time. Now the worst case is a "building" PendingTransaction
the reconciler (app/tx/reconcile.py) can resolve either way by asking the
chain whether the tx exists, exactly like it already does for bets and
withdrawals.
"""
client = self._listener.client
if client is None:
logger.error("round %s payout deferred: not connected", round_id)
await self._log_payout_failure(round_id, None, "electrum client not connected")
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)
already_in_flight = await session.scalar(
select(PendingTransaction).where(
PendingTransaction.round_id == round_id,
PendingTransaction.kind == "payout",
PendingTransaction.status.in_(("building", "pending")),
)
)
if already_in_flight is not None:
# A payout for this round is already building or broadcast — this
# must not build a second one, or a retry (manual, or a future
# automatic one) would pay the winner twice. Confirmation/
# reconciliation already owns resolving that row.
logger.info(
"round %s payout already in flight (pending_transaction %s), skipping",
round_id,
already_in_flight.id,
)
return
reserved_outpoints = await _reserved_payout_outpoints(session)
config = await get_round_config(session)
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
if not fee_address:
logger.error(
"round %s payout blocked: no fee_address configured (set it via the admin endpoint)", round_id
)
await self._log_payout_failure(round_id, winner_user_id, "no fee_address configured")
return
if winner_address is None:
logger.error("round %s payout blocked: winner user %s not found", round_id, winner_user_id)
await self._log_payout_failure(round_id, winner_user_id, "winner user not found")
return
winner_sats = winner_share(pool_amount_sats)
commission_share = pool_amount_sats - winner_sats # remainder from rounding goes to fees
# --- Phase 2: build (network read only, no DB write yet) -----------------
try:
pool_key = derive_pool_key()
pool_script_obj = script.p2wpkh(pool_key.to_public())
pool_address = pool_script_obj.address(network=PLM_MAINNET)
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 and (e["tx_hash"], e["tx_pos"]) not in reserved_outpoints
]
built = build_payout_transaction(
signing_key=pool_key,
from_script=pool_script_obj,
utxos=utxos,
winner_address=winner_address,
winner_share_sats=winner_sats,
fee_address=fee_address,
commission_sats=commission_share,
change_address=pool_address,
fee_rate_sat_vb=fee_rate,
)
except InsufficientFundsError as exc:
# Includes the "too_many_inputs" case: the pool holds enough, but spread over
# more UTXOs than one transaction may spend, so /admin has to say which. Since
# B-52 that means MAX_PAYOUT_TX_INPUTS, and participants are capped below it at
# bet time (bets/service.py), so reaching it now takes pool change accumulated
# over many rounds rather than one busy round — an operator consolidation job,
# not a dead end for the bets of the round in progress.
reason = "insufficient pool UTXOs" if exc.code == "insufficient_balance" else exc.code
logger.exception("round %s payout failed: %s", round_id, reason)
await self._log_payout_failure(round_id, winner_user_id, reason)
return
except Exception:
# Anything else — a malformed fee_address (EmbitError) or similar. 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). _retry_payout_if_due (B-26) is what turns this recorded
# failure into an automatic retry instead of a dead end.
logger.exception("round %s payout build failed", round_id)
await self._log_payout_failure(round_id, winner_user_id, "payout build failed")
return
# --- Phase 3: persist the intent, *then* broadcast (B-25) -----------------
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
pending = PendingTransaction(
kind="payout",
round_id=round_id,
current_txid=built.txid,
fee_rate_sat_vb=fee_rate,
raw_tx_hex=built.raw_hex,
# "building" until the broadcast succeeds, exactly like place_bet's
# two phases — see the reconciler, which gives this a short grace
# period before asking the chain whether it made it out after all.
status="building",
)
session.add(pending)
await session.commit()
pending_id = pending.id
# --- Phase 4: broadcast, then promote the pending row --------------------
try:
await client.broadcast(built.raw_hex)
except Exception:
# The row stays "building": the reconciler will ask the chain about it
# and, finding nothing, abandon it and clear payout_txid (B-25) — instead
# of the round being stuck with a payout_txid that never went anywhere.
logger.exception("round %s payout broadcast failed", round_id)
await self._log_payout_failure(round_id, winner_user_id, "broadcast rejected")
return
async with self._session_factory() as session:
pending = await session.get(PendingTransaction, pending_id)
pending.status = "pending"
await write_audit_log(
session,
"payout_sent",
{"txid": built.txid, "winner_sats": built.winner_sats, "commission_sats": built.commission_sats},
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, reason: str) -> 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. Called from every
early return in _trigger_payout (B-26), not just the generic exception
branch as before, so _retry_payout_if_due always has an entry to throttle
against and /admin always shows *why* a round is stuck rather than just
that it is."""
try:
async with self._session_factory() as session:
await write_audit_log(
session,
"payout_failed",
{"round_id": round_id, "reason": reason},
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)
async def _reserved_payout_outpoints(session: AsyncSession) -> set[tuple[str, int]]:
"""Pool UTXOs already claimed by a payout that hasn't resolved yet — this
round's own in-flight payout (guarded against separately in _trigger_payout) or
a stale one from an earlier round the reconciler hasn't abandoned yet (B-25).
These must be excluded from selection, or a retry would double-spend the same
coins into two payouts before the reconciler gets a chance to release them."""
rows = (
await session.scalars(
select(PendingTransaction).where(
PendingTransaction.kind == "payout",
PendingTransaction.status.in_(("building", "pending")),
)
)
).all()
reserved: set[tuple[str, int]] = set()
for row in rows:
tx = Transaction.parse(bytes.fromhex(row.raw_tx_hex))
for vin in tx.vin:
reserved.add((vin.txid.hex(), vin.vout))
return reserved
Reference in New Issue View Git Blame Copy Permalink