davide/plm-lottery
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
88221b0dbd73eed75c0678c3fd3e5a693b48c010
plm-lottery/app/rounds/scheduler.py
T
davideandClaude Opus 5.5 61efbb0e37 Replace flowchart/ with a traceable specification in docs/ 
The two Mermaid diagrams under flowchart/ described the intended flow but
had drifted from the code and carried no link back to it. docs/ now holds
the specification instead: 15 Italian documents (flows, architecture, data
model, API, security, audit log, operations, decisions and limits, bug
history) with 472 numbered requirements, each pointing at the function
that implements it and the test that covers it. docs/processo-di-modifica.md
defines the ID convention and the spec -> code -> test change workflow, and
CLAUDE.md now names docs/ as authoritative.

Writing the spec from the code also corrected stale claims in CLAUDE.md
(B-41's get_history check, BUGS.md still tracking B-71/B-72, Cache-Control
scope, the error contract's bare-string exceptions) and in the operator
guides (payouts do retry, only `drawing` fails to resume, rounds do not
open without fee_address). The B-70 doc test now pins the prose in
docs/scommesse-e-prelievi.md instead of the deleted diagram.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-28 12:16:01 +02:00

538 lines
27 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, round_deadline, 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 the round lifecycle (docs/ciclo-di-vita-del-round.md):
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 = round_.id, round_.status
# B-61: this round's own snapshotted deadline, not one recomputed from
# whatever the config says now — an operator lowering the duration
# mid-round used to close the round on the spot.
deadline = round_deadline(round_)
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":
if datetime.now(timezone.utc) < deadline:
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.
B-63: `tip_at_close` of 0 means the tip was *unknown* when the round closed,
not that the chain was at height zero — and "the first block we hear about"
is then not necessarily a block mined after the close. Rather than seed the
draw from a hash that may already have been public while bets were open, the
first height we do learn becomes the baseline and this waits for a block
strictly after it. Since the Electrum listener now only publishes its client
once a header has been applied, and _tick won't run without one, this should
be unreachable — it stays as the local statement of what the draw actually
requires, since nothing else in this function would notice if that stopped
holding.
"""
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)
if tip_at_close <= 0:
tip_at_close = await self._adopt_baseline_tip(round_id)
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 _adopt_baseline_tip(self, round_id: int) -> int:
"""B-63: the height the draw must find a *later* block than, for the case
where the tip wasn't known at closing time. Waits for a header to arrive and
takes that height as the baseline — the block it describes may predate the
close, which is exactly why it is used as the floor rather than as the seed —
and records why, since a draw that waits one extra block should be explainable
from /admin rather than looking like a stall.
"""
while not self._listener.tip_header_hex or self._listener.tip_height <= 0:
await asyncio.sleep(_TICK_INTERVAL_SECONDS)
height = self._listener.tip_height
logger.warning(
"round %s: chain tip was unknown at closing time; using height %s as the draw baseline "
"and waiting for a further block",
round_id,
height,
)
async with self._session_factory() as session:
await write_audit_log(
session,
"draw_baseline_tip_unknown",
{"baseline_height": height},
round_id=round_id,
)
await session.commit()
return 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