plm-lottery/app/tx/reconcile.py

"""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 chain: if the tx is there after all, promote everything
    to its live state; if it isn't, release the UTXOs and undo the intent.
  * status "pending" — broadcast, still unconfirmed. Left alone until it has been
    unconfirmed for `_ABANDON_AFTER_SECONDS`, since absence from one server's
    mempool is not proof of death; only then is it abandoned like the above.

Deliberately conservative: it never touches a tx the chain knows about, and the
grace period is long (multiples of the RBF timeout) so a slow-but-alive tx is
bumped by RbfBumper rather than abandoned here.
"""

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, Round, RoundParticipant, UtxoEvent, Withdrawal
from app.electrum.client import ElectrumClient
from app.electrum.scripthash import address_to_scripthash
from app.rounds.events import broadcaster
from app.tx.broadcast import adopt_version
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 concluding it's gone.
_ABANDON_AFTER_SECONDS = 6 * 60 * 60


class PendingTransactionReconciler:
    def __init__(self, session_factory: async_sessionmaker, get_client: Callable[[], ElectrumClient | None]):
        self._session_factory = session_factory
        self._get_client = get_client

    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)
                except asyncio.CancelledError:
                    raise
                except Exception:
                    logger.exception("pending-transaction reconciliation failed")
            await asyncio.sleep(_POLL_INTERVAL_SECONDS)


async def reconcile_once(session_factory: async_sessionmaker, client: ElectrumClient) -> int:
    """Returns how many rows were resolved (promoted or abandoned).

    Existence is decided by checking whether a row's own address's history
    (blockchain.scripthash.get_history) includes its txid at all — mempool or
    mined — rather than asking blockchain.transaction.get for a verbose reply
    (B-41): several Electrum server implementations and versions reject the
    verbose flag outright, and the previous substring-matching on the error
    text (looking for "missing", "not found", ...) was fragile as the basis for
    a decision that releases funds. A transport failure fetching history still
    raises and leaves the row alone until next time — get_history not
    returning our txid is the only thing that means "gone". History is cached
    per scripthash within one pass, since the payout rows of one round share its
    pool address (B-80).
    """
    now = datetime.now(timezone.utc)
    async with session_factory() as session:
        rows = (
            await session.scalars(
                select(PendingTransaction).where(PendingTransaction.status.in_(("building", "pending")))
            )
        ).all()
        candidates = []
        # B-77 (TX-15, TX-19): a row's every txid, not only the current one.
        txids_by_id = await txids_by_row(session, {row.id: row.current_txid for row in rows})
        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((row.id, row.status, row.current_txid, scripthash, txids_by_id[row.id]))

    resolved = 0
    history_cache: dict[str, list[dict]] = {}
    for row_id, status, txid, scripthash, known_txids in candidates:
        try:
            if scripthash not in history_cache:
                history_cache[scripthash] = await client.get_history(scripthash)
            found = [entry["tx_hash"] for entry in history_cache[scripthash] if entry.get("tx_hash") in known_txids]
            exists = bool(found)
        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", row_id, txid)
            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, row_id)
            if row is None or row.status != status:
                continue  # something else moved it while we were asking
            if exists:
                if txid not in found:
                    # 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, found[0])
                    resolved += 1
                if row.status == "building":
                    await _promote(session, row)
                    resolved += 1
            else:
                await _abandon(session, row, "not found on chain")
                resolved += 1
            await session.commit()
        broadcaster.publish()

    return resolved


def _is_due(row: PendingTransaction, now: datetime) -> bool:
    # 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.
    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 _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"
    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"
    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"
    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."""
    row.status = "failed"
    row.failure_reason = reason[:128]

    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
            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