Files
plm-lottery/app/tx/reconcile.py
T
davideandClaude Sonnet 5.5 9520c80dcf Close the smaller state inconsistencies (B-96, BET-02, BET-07, WDR-01, WDR-15, TX-16, TX-18, TX-27, TX-29, DEP-15, DEP-16, AUD-30, AUD-46, DAT-22)
- bet/withdrawal routes read listener.client after taking the user lock
- a reduced full-balance withdrawal is checked against the minimum again
- a bet covered only by pending balance answers balance_pending_confirmation
- a recovered bet/withdrawal/payout writes bet_placed/withdrawal_sent/payout_sent
- an abandoned payout clears winner/fee/reserve amounts with payout_txid
- the reconciler pings SSE only for rows it changed
- closing_stalled every 30 minutes past the deadline; closing_waiting_since on /rounds/current
- one refresh per user at a time; a failed refresh no longer ends the Electrum session

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-18 01:08:14 +02:00

511 lines
23 KiB
Python

"""Resolves in-flight transactions against the chain.
Everything else in this codebase assumes a broadcast either confirms or gets
fee-bumped until it does. Neither is guaranteed: an RBF bump raises RbfError
whenever there's no change output big enough to absorb it (see tx/broadcast.py),
a node can drop a low-fee tx from its mempool, and the process can die between
building a transaction and broadcasting it. Without this module those cases were
permanent: `spent_txid` was set at build time and never cleared, so the coins
stayed spendable on-chain while the database considered them gone — the user's
balance simply lost them, with no path back short of editing the DB by hand
(B-04, and the "building" half of B-08).
What it does, per PendingTransaction that isn't already terminal:
* status "building" — we crashed (or were killed) between writing the row and
broadcasting. Ask the network: if any server knows one of the row's txids,
promote everything to its live state.
* status "pending" — broadcast, still unconfirmed. Left alone until it has been
unconfirmed for `_ABANDON_AFTER_SECONDS`, and even then only examined.
and then *nothing is given up without proof* (B-78, ADR-51, TX-17). A row is
abandoned — its coins released — only when
(a) one of its inputs is spent **in a block** by a transaction that is none of its
own versions, and the other servers agree; or
(b) no server of the quorum knows any txid of the row **and** sending its latest
raw transaction to every server is refused, explicitly and by all of them, for
missing or already-spent inputs.
Absence from one server's history proves nothing — after a restart onto a different
server that is the normal state of a tx the new server has not seen — and neither
does the age of the row. A refusal for fee reasons is no proof either: such a row is
promoted (building) or left to the RBF bumper (pending). A transport failure never
resolves anything. Failed rows stay watched for `FAILED_WATCH_DAYS`: if one of them
confirms after all, the lottery is paused for the operator (TX-23).
"""
import asyncio
import logging
from collections.abc import Callable
from datetime import datetime, timedelta, timezone
from embit.transaction import Transaction
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from app.audit.log import write_audit_log
from app.db.base import begin_immediate
from app.db.models import (
PendingTransaction,
PendingTransactionTxid,
Round,
RoundParticipant,
UtxoEvent,
Withdrawal,
)
from app.electrum.client import ElectrumClient
from app.electrum.quorum import ServerAnswers
from app.electrum.scripthash import address_to_scripthash
from app.rounds.events import broadcaster
from app.rounds.pause import pause_automatically
from app.tx.broadcast import adopt_version
from app.tx.outcome import is_already_known
from app.tx.pending_address import own_address_for
from app.tx.versions import txids_by_row, txids_of
from app.wallet.balance import recompute_balance
logger = logging.getLogger(__name__)
_POLL_INTERVAL_SECONDS = 120
# A "building" row means we never got confirmation that the broadcast happened, so
# it only needs long enough to rule out a request still in flight.
_BUILDING_GRACE_SECONDS = 120
# A "pending" row was accepted by a node once. Give it a wide margin — the RBF
# bumper gets several attempts inside this window — before it is even examined.
_ABANDON_AFTER_SECONDS = 6 * 60 * 60
# How long a failed row keeps being looked at (B-78, DAT-72): a transaction given up
# for dead can still arrive in a block.
FAILED_WATCH_DAYS = 14
# How many distinct (txid, height) claims of one failed row are put to the quorum per pass.
_MAX_RESURRECTION_CLAIMS = 3
# What a node says when the transaction's inputs do not exist or are spent already.
_MISSING_INPUTS_MARKERS = ("missing-inputs", "missingorspent", "missing inputs", "inputs-missing")
# What it says when only the fee is the problem: not proof of anything.
_FEE_MARKERS = ("min relay fee", "mempool min fee", "insufficient fee", "fee not met", "fee too low", "fee rate")
class PendingTransactionReconciler:
def __init__(
self,
session_factory: async_sessionmaker,
get_client: Callable[[], ElectrumClient | None],
network=None,
):
self._session_factory = session_factory
self._get_client = get_client
self._network = network # the ElectrumListener: histories of every server, rebroadcast, spenders
async def run(self) -> None:
# Runs once promptly at startup: a crash mid-broadcast is exactly the case
# that leaves a "building" row, and the restart is when we can clear it.
while True:
client = self._get_client()
if client is not None:
try:
await reconcile_once(self._session_factory, client, self._network)
except asyncio.CancelledError:
raise
except Exception:
logger.exception("pending-transaction reconciliation failed")
await asyncio.sleep(_POLL_INTERVAL_SECONDS)
class _Candidate:
__slots__ = ("row_id", "status", "txid", "scripthash", "known_txids", "raw_hex")
def __init__(self, row_id, status, txid, scripthash, known_txids, raw_hex):
self.row_id = row_id
self.status = status
self.txid = txid
self.scripthash = scripthash
self.known_txids = known_txids
self.raw_hex = raw_hex
async def reconcile_once(
session_factory: async_sessionmaker, client: ElectrumClient, network=None
) -> int:
"""Returns how many rows were resolved (promoted, abandoned, realigned or
resurrected).
Existence is decided by asking every server of the quorum for the history of the
row's own address (blockchain.scripthash.get_history) and looking for **every** txid
the row has had, mempool or mined (B-41, B-77, B-78) — never for a verbose
blockchain.transaction.get reply, which several servers reject outright, and never
on error text. History is cached per scripthash within one pass, since the payout
rows of one round share its pool address (B-80). With no `network` only the active
server is asked, which can promote a row but never abandon one.
"""
now = datetime.now(timezone.utc)
async with session_factory() as session:
rows = (
await session.scalars(
select(PendingTransaction).where(PendingTransaction.status.in_(("building", "pending", "failed")))
)
).all()
txids_by_id = await txids_by_row(session, {row.id: row.current_txid for row in rows})
candidates: list[_Candidate] = []
for row in rows:
if not _is_due(row, now):
continue
try:
address = await own_address_for(session, row.kind, row.user_id, row.round_id)
scripthash = address_to_scripthash(address)
except Exception:
logger.exception("could not derive the address for pending_transaction %s", row.id)
continue
candidates.append(
_Candidate(row.id, row.status, row.current_txid, scripthash, txids_by_id[row.id], row.raw_tx_hex)
)
resolved = 0
history_cache: dict[str, ServerAnswers] = {}
for candidate in candidates:
try:
if candidate.scripthash not in history_cache:
history_cache[candidate.scripthash] = await _histories(client, network, candidate.scripthash)
answers = history_cache[candidate.scripthash]
decision = await _decide(network, candidate, answers)
except Exception:
# Transport/server problem — say nothing about this tx and try again on
# the next pass rather than abandoning a tx that may be perfectly alive.
logger.warning(
"could not check pending_transaction %s (txid %s) against the chain",
candidate.row_id,
candidate.txid,
exc_info=True,
)
continue
if decision is None:
continue
async with session_factory() as session:
await begin_immediate(session) # B-88: the re-check and the resolution under one lock
row = await session.get(PendingTransaction, candidate.row_id)
# TX-26: the status *and* the current txid must still be what they were when
# the network was asked — a confirmation or an RBF in the meantime changes
# what the answer means.
if row is None or row.status != candidate.status or row.current_txid != candidate.txid:
continue
kind, payload = decision
changed = False
if kind == "resurrected":
await _resurrect(session, row, *payload)
changed = True
elif kind == "present":
if payload and payload != row.current_txid:
# Another version of the row is the one on the network (B-77, TX-19):
# the row follows it, so the inputs and the domain row point at it.
await adopt_version(session, row, payload)
changed = True
if row.status == "building":
await _promote(session, row)
changed = True
elif kind == "promote":
await _promote(session, row)
changed = True
elif kind == "abandon":
await _abandon(session, row, payload)
changed = True
await session.commit()
if changed: # B-96 (TX-27): a row that was only examined moved nothing to refetch
resolved += 1
broadcaster.publish()
return resolved
async def _histories(client: ElectrumClient, network, scripthash: str) -> ServerAnswers:
if network is not None:
return await network.get_histories(scripthash)
# No quorum to ask: the active server alone can show a transaction exists, but its
# silence can never be taken for absence (`complete` stays false).
return ServerAnswers(await client.get_history(scripthash), (), 1)
def _entries_for(answers: ServerAnswers, known_txids: set[str]) -> list[tuple[str, int, bool]]:
"""(txid, height, from_active) for every history entry of every answering server that
is one of the row's txids."""
found: list[tuple[str, int, bool]] = []
if answers.active is not None:
found += [(e["tx_hash"], e.get("height") or 0, True) for e in answers.active if e.get("tx_hash") in known_txids]
for history in answers.answered_others:
found += [(e["tx_hash"], e.get("height") or 0, False) for e in history if e.get("tx_hash") in known_txids]
return found
def _confirmed_candidates(answers: ServerAnswers, known_txids: set[str]) -> list[tuple[str, int]]:
"""(txid, height) of every txid of the row that any answering server lists in a block,
the one most servers agree on first. They are only *claims*: each is proven by the
quorum's merkle proofs before it is believed (B-83, TX-22)."""
votes: dict[tuple[str, int], int] = {}
for history in answers.all:
for entry in history:
if entry.get("tx_hash") in known_txids and (entry.get("height") or 0) > 0:
key = (entry["tx_hash"], entry["height"])
votes[key] = votes.get(key, 0) + 1
return [key for key, _ in sorted(votes.items(), key=lambda item: -item[1])]
async def _decide(network, candidate: _Candidate, answers: ServerAnswers):
"""What the network says about one row, as ("present", txid | None), ("promote", None),
("abandon", reason), ("resurrected", (txid, height)) or None for "nothing to do or
not enough to tell". Pure reads — nothing is written here."""
if candidate.status == "failed":
if network is None:
return None # nothing can be proven without the quorum
for txid, height in _confirmed_candidates(answers, candidate.known_txids)[:_MAX_RESURRECTION_CLAIMS]:
if await network.confirmation_proven(txid, height):
return ("resurrected", (txid, height))
return None
found = _entries_for(answers, candidate.known_txids)
if found:
# Prefer a version that is mined, then the current one, then any.
mined = [txid for txid, height, _ in found if height > 0]
present = {txid for txid, _, _ in found}
chosen = mined[0] if mined else (candidate.txid if candidate.txid in present else found[0][0])
return ("present", chosen)
if not answers.complete:
logger.warning(
"pending_transaction %s (txid %s): too few servers answered to say the tx is unknown — left as is",
candidate.row_id,
candidate.txid,
)
return None
if network is None:
return None
# No server of the quorum knows any txid of the row. That alone proves nothing (TX-15).
tx = Transaction.parse(bytes.fromhex(candidate.raw_hex))
outpoints = {(vin.txid.hex(), vin.vout) for vin in tx.vin}
spender = await network.confirmed_spender(candidate.scripthash, candidate.known_txids, outpoints)
if spender is not None:
return ("abandon", f"input spent in a block by {spender}")
rebroadcast: ServerAnswers = await network.rebroadcast_everywhere(candidate.raw_hex)
verdicts = rebroadcast.all
if any(v[0] == "accepted" or (v[0] == "refused" and is_already_known(v[1])) for v in verdicts):
# It was alive after all (or is again): the same as having found it.
return ("present", None)
refusals = [v[1].lower() for v in verdicts if v[0] == "refused"]
if rebroadcast.complete and len(refusals) == len(verdicts) and all(
any(marker in message for marker in _MISSING_INPUTS_MARKERS) for message in refusals
):
return ("abandon", f"rebroadcast refused for missing inputs: {refusals[0]}")
if candidate.status == "building" and any(any(marker in m for marker in _FEE_MARKERS) for m in refusals):
return ("promote", None) # not proof: let the RBF bumper accelerate it (TX-17)
return None
def _is_due(row: PendingTransaction, now: datetime) -> bool:
if row.status == "failed":
# B-78 (TX-23, DAT-72): watched every pass for FAILED_WATCH_DAYS from failed_at.
failed_at = row.failed_at or row.updated_at
return failed_at is not None and now < failed_at.replace(tzinfo=timezone.utc) + timedelta(days=FAILED_WATCH_DAYS)
# Deliberately broadcast_at (the *first* broadcast), not last_broadcast_at: an
# RBF bump used to overwrite this same field, which reset this grace period on
# every bump and meant a repeatedly-bumped-but-never-mined tx was never
# abandoned (B-27). tx/broadcast.py:bump_fee now only ever touches
# last_broadcast_at, so this keeps measuring from when the tx first appeared,
# no matter how many times it's since been bumped. The period only decides from
# when a row is *examined*; giving it up still needs proof (TX-14, TX-17).
grace = _BUILDING_GRACE_SECONDS if row.status == "building" else _ABANDON_AFTER_SECONDS
return now >= row.broadcast_at.replace(tzinfo=timezone.utc) + timedelta(seconds=grace)
async def _resurrect(session: AsyncSession, row: PendingTransaction, txid: str, height: int) -> None:
"""A row given up for dead was mined (B-78, TX-23, DAT-72). The row becomes
`confirmed` and the lottery is paused for the operator: the domain rows the
abandonment already rolled back (a deleted participant, a failed withdrawal, released
coins, a rebuilt payout) are *not* put back by the system, which cannot know what has
happened to them since."""
if txid != row.current_txid:
version_raw = None
version = await session.scalar(
select(PendingTransactionTxid).where(
PendingTransactionTxid.txid == txid, PendingTransactionTxid.pending_transaction_id == row.id
)
)
if version is not None and version.raw_hex:
version_raw = version.raw_hex
row.current_txid = txid
if version_raw:
row.raw_tx_hex = version_raw
row.status = "confirmed"
await write_audit_log(
session,
"pending_tx_resurrected",
{"pending_transaction_id": row.id, "kind": row.kind, "txid": txid, "block_height": height},
user_id=row.user_id,
round_id=row.round_id,
)
await pause_automatically(session, "resurrected_tx")
logger.error(
"%s pending_transaction %s was given up as failed but tx %s is in block %s — lottery paused for the operator",
row.kind,
row.id,
txid,
height,
)
async def _promote(session: AsyncSession, row: PendingTransaction) -> None:
"""The tx did make it onto the chain before we died — finish what phase 2 of
place_bet/request_withdrawal would have done."""
row.status = "pending"
# B-96 (TX-16, AUD-30): besides pending_tx_recovered, the event the successful send would have
# written — a recovered bet, withdrawal or payout is otherwise absent from the audit trail.
sent_event: tuple[str, dict, int | None] | None = None
if row.kind == "bet":
participant = await session.scalar(
select(RoundParticipant).where(
RoundParticipant.round_id == row.round_id, RoundParticipant.user_id == row.user_id
)
)
if participant is not None and participant.status == "building":
participant.status = "broadcast"
sent_event = (
"bet_placed",
{"txid": row.current_txid, "amount_sats": participant.bet_amount_sats},
row.user_id,
)
elif row.kind == "withdrawal" and row.withdrawal_id is not None:
withdrawal = await session.get(Withdrawal, row.withdrawal_id)
if withdrawal is not None and withdrawal.status == "building":
withdrawal.status = "broadcast"
sent_event = (
"withdrawal_sent",
{
"txid": row.current_txid,
"amount_sent_sats": withdrawal.amount_sent_sats,
"external_address": withdrawal.external_address,
},
row.user_id,
)
elif row.kind == "payout" and row.round_id is not None:
round_ = await session.get(Round, row.round_id)
if round_ is not None:
sent_event = (
"payout_sent",
{
"txid": row.current_txid,
"winner_sats": round_.winner_amount_sats,
"commission_sats": round_.fee_amount_sats,
},
round_.winner_user_id,
)
if sent_event is not None:
event, payload, event_user_id = sent_event
await write_audit_log(session, event, payload, user_id=event_user_id, round_id=row.round_id)
await write_audit_log(
session,
"pending_tx_recovered",
{"pending_transaction_id": row.id, "kind": row.kind, "txid": row.current_txid},
user_id=row.user_id,
round_id=row.round_id,
)
logger.info("recovered %s pending_transaction %s: tx %s is on-chain", row.kind, row.id, row.current_txid)
async def _abandon(session: AsyncSession, row: PendingTransaction, reason: str) -> None:
"""The tx is gone for good. Release whatever it reserved so the funds come back,
and roll the domain row back to something truthful."""
reason = reason[:128]
row.status = "failed"
row.failure_reason = reason
row.failed_at = datetime.now(timezone.utc) # B-78: starts the 14-day watch
released = await _release_inputs(session, row, await txids_of(session, row))
if row.kind == "bet":
participant = await session.scalar(
select(RoundParticipant).where(
RoundParticipant.round_id == row.round_id, RoundParticipant.user_id == row.user_id
)
)
if participant is not None and participant.status in ("building", "broadcast"):
# The bet never happened, so the user is not in this round. Removing the
# row also unblocks the scheduler, which waits for every non-confirmed
# participant before closing the round.
await session.delete(participant)
elif row.kind == "withdrawal" and row.withdrawal_id is not None:
withdrawal = await session.get(Withdrawal, row.withdrawal_id)
if withdrawal is not None and withdrawal.status in ("building", "broadcast"):
withdrawal.status = "failed"
withdrawal.txid = None
elif row.kind == "payout":
# Payout funds come from the pool address, which isn't tracked in
# utxo_events, so there's nothing to release. Clearing payout_txid leaves the
# round in "paying_out" with no tx attached, which is exactly the state the
# scheduler's payout retry picks up (B-26: `_retry_payout_if_due`, one attempt
# per 60s), so an abandoned payout is rebuilt on its own rather than waiting
# for an operator — the log line below is the alert, not the recovery path.
round_ = await session.get(Round, row.round_id) if row.round_id else None
if round_ is not None and round_.status == "paying_out":
round_.payout_txid = None
# B-96 (TX-18, DAT-22): the amounts belong to the attempt that no longer exists; the
# next attempt writes its own.
round_.winner_amount_sats = None
round_.fee_amount_sats = None
round_.bump_reserve_sats = None
logger.error(
"round %s payout tx %s vanished — round needs operator attention", round_.id, row.current_txid
)
if row.user_id is not None:
await recompute_balance(session, row.user_id)
await write_audit_log(
session,
"pending_tx_abandoned",
{
"pending_transaction_id": row.id,
"kind": row.kind,
"txid": row.current_txid,
"reason": reason,
"utxos_released": released,
},
user_id=row.user_id,
round_id=row.round_id,
)
logger.warning(
"abandoned %s pending_transaction %s (txid %s): %s — released %s UTXO(s)",
row.kind,
row.id,
row.current_txid,
reason,
released,
)
async def _release_inputs(session: AsyncSession, row: PendingTransaction, own_txids: set[str] | None = None) -> int:
"""Clear spent_txid on every UTXO this transaction consumed, so the balance
counts them again. The inputs come from the stored raw tx, which is kept current
across RBF bumps, so this works for a bumped tx too."""
try:
tx = Transaction.parse(bytes.fromhex(row.raw_tx_hex))
except Exception:
logger.exception("could not parse raw tx of pending_transaction %s; inputs not released", row.id)
return 0
released = 0
for vin in tx.vin:
utxo = await session.scalar(
select(UtxoEvent).where(UtxoEvent.txid == vin.txid.hex(), UtxoEvent.vout == vin.vout)
)
# Only release what this tx actually reserved: if another tx has since spent
# the same UTXO, its claim is the live one and must not be cleared. "This tx"
# is any version of the row (B-77, DAT-13), not only its current one.
if utxo is not None and utxo.spent_txid in (own_txids or {row.current_txid}):
utxo.spent_txid = None
released += 1
return released