Add transaction broadcast, confirmation polling and RBF fee-bump

Shared per-user locking to serialize bet/withdrawal PSBT builds (tx/locks.py), a confirmation poller for pending outgoing transactions, and the timeout->fee-bump->rebroadcast loop used by bets, payouts and withdrawals alike (tx/broadcast.py). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-21 10:25:57 +02:00
parent 107e592704
commit fc2aadbc7e
6 changed files with 515 additions and 0 deletions
@@ -0,0 +1,163 @@
+import asyncio
+import logging
+from datetime import datetime, timedelta, timezone
+
+from embit import script
+from embit.psbt import PSBT
+from embit.transaction import Transaction, TransactionInput, TransactionOutput
+from embit.finalizer import finalize_psbt
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from app.config import settings
+from app.db.models import PendingTransaction, User
+from app.electrum.client import ElectrumClient
+from app.wallet.hd import derive_pool_key, derive_user_key
+from app.wallet.plm_network import PLM_MAINNET
+from app.wallet.psbt_builder import RBF_SEQUENCE, estimate_vsize
+
+logger = logging.getLogger(__name__)
+
+_POLL_INTERVAL_SECONDS = 30
+_FEE_RATE_INCREMENT = 1  # minimum relay-policy-friendly bump per BIP125
+
+
+class RbfError(Exception):
+    pass
+
+
+def should_bump(pending: PendingTransaction, now: datetime, timeout_seconds: int | None = None) -> bool:
+    """Pure decision: has this pending tx been unconfirmed for longer than the
+    configured timeout? Kept separate from the I/O-heavy bump_fee() so it's
+    trivially unit-testable."""
+    timeout = timeout_seconds if timeout_seconds is not None else settings.rbf_timeout_seconds
+    if pending.status != "pending":
+        return False
+    return now >= pending.broadcast_at.replace(tzinfo=timezone.utc) + timedelta(seconds=timeout)
+
+
+async def _signing_context(session: AsyncSession, pending: PendingTransaction) -> tuple:
+    """Returns (signing_key, own_script, own_address) for the single sender that
+    controls every input of this tx — a user for bet/withdrawal, the pool for
+    payout. All our builders only ever spend one address's UTXOs per tx."""
+    if pending.kind == "payout":
+        key = derive_pool_key()
+    else:
+        user = await session.get(User, pending.user_id)
+        key = derive_user_key(user.derivation_index)
+    own_script = script.p2wpkh(key.to_public())
+    own_address = own_script.address(network=PLM_MAINNET)
+    return key, own_script, own_address
+
+
+async def _prevout_amount(client: ElectrumClient, vin: TransactionInput) -> int:
+    txid_hex = vin.txid.hex()
+    tx = await client.get_transaction(txid_hex, verbose=True)
+    value_coins = tx["vout"][vin.vout]["value"]
+    return round(value_coins * 100_000_000)
+
+
+def _find_change_output(tx: Transaction, change_address: str) -> int | None:
+    for i, out in enumerate(tx.vout):
+        if out.script_pubkey.address(network=PLM_MAINNET) == change_address:
+            return i
+    return None
+
+
+async def bump_fee(session: AsyncSession, client: ElectrumClient, pending: PendingTransaction) -> str:
+    """Rebuild `pending`'s transaction with a higher fee (same inputs, same
+    recipient outputs, the extra fee taken from the change output) and
+    rebroadcast. Returns the new txid.
+
+    Only handles the common case: exactly one change output paying back to the
+    tx's own sender address, large enough to absorb the increase. If there's no
+    such output (e.g. an exact-amount bet with no change), this raises RbfError —
+    bumping such a tx would require selecting additional inputs, which isn't
+    implemented for the MVP; it needs manual operator intervention.
+    """
+    old_tx = Transaction.parse(bytes.fromhex(pending.raw_tx_hex))
+    signing_key, own_script, own_address = await _signing_context(session, pending)
+
+    input_amounts = [await _prevout_amount(client, vin) for vin in old_tx.vin]
+    total_in = sum(input_amounts)
+    old_fee = total_in - sum(o.value for o in old_tx.vout)
+
+    new_fee_rate = pending.fee_rate_sat_vb + _FEE_RATE_INCREMENT
+    new_fee = estimate_vsize(len(old_tx.vin), len(old_tx.vout)) * new_fee_rate
+    fee_delta = new_fee - old_fee
+    if fee_delta <= 0:
+        fee_delta = _FEE_RATE_INCREMENT  # already above the new target vsize*rate; bump by a token amount
+
+    change_index = _find_change_output(old_tx, own_address)
+    if change_index is None or old_tx.vout[change_index].value <= fee_delta:
+        raise RbfError(f"pending_transaction {pending.id}: no change output large enough to absorb a fee bump")
+
+    new_vout = list(old_tx.vout)
+    bumped_change = new_vout[change_index].value - fee_delta
+    new_vout[change_index] = TransactionOutput(bumped_change, new_vout[change_index].script_pubkey)
+
+    new_vin = [TransactionInput(v.txid, v.vout, sequence=RBF_SEQUENCE) for v in old_tx.vin]
+    new_tx = Transaction(vin=new_vin, vout=new_vout)
+    psbt = PSBT(new_tx)
+    for i, amount in enumerate(input_amounts):
+        psbt.inputs[i].witness_utxo = TransactionOutput(amount, own_script)
+
+    signed = psbt.sign_with(signing_key)
+    if signed != len(new_vin):
+        raise RuntimeError(f"expected {len(new_vin)} signatures, got {signed}")
+
+    final_tx = finalize_psbt(psbt)
+    if final_tx is None:
+        raise RuntimeError("failed to finalize bumped PSBT")
+
+    raw_hex = final_tx.serialize().hex()
+    new_txid = final_tx.txid().hex()
+    await client.broadcast(raw_hex)
+
+    pending.current_txid = new_txid
+    pending.raw_tx_hex = raw_hex
+    pending.fee_rate_sat_vb = new_fee_rate
+    pending.attempt_count += 1
+    pending.broadcast_at = datetime.now(timezone.utc)
+    await session.commit()
+
+    logger.info("bumped %s pending_transaction %s: %s -> %s", pending.kind, pending.id, pending.current_txid, new_txid)
+    return new_txid
+
+
+class RbfBumper:
+    def __init__(self, session_factory: async_sessionmaker, get_client):
+        self._session_factory = session_factory
+        self._get_client = get_client
+
+    async def run(self) -> None:
+        while True:
+            client = self._get_client()
+            if client is not None:
+                try:
+                    await self._tick(client)
+                except asyncio.CancelledError:
+                    raise
+                except Exception:
+                    logger.exception("RBF bump tick failed")
+            await asyncio.sleep(_POLL_INTERVAL_SECONDS)
+
+    async def _tick(self, client: ElectrumClient) -> None:
+        now = datetime.now(timezone.utc)
+        async with self._session_factory() as session:
+            candidates = (
+                await session.scalars(select(PendingTransaction).where(PendingTransaction.status == "pending"))
+            ).all()
+            due = [p for p in candidates if should_bump(p, now)]
+
+        for pending in due:
+            async with self._session_factory() as session:
+                row = await session.get(PendingTransaction, pending.id)
+                if row is None or row.status != "pending":
+                    continue
+                try:
+                    await bump_fee(session, client, row)
+                except RbfError:
+                    logger.exception("could not bump pending_transaction %s", row.id)
+                except Exception:
+                    logger.exception("unexpected error bumping pending_transaction %s", row.id)
@@ -0,0 +1,67 @@
+import asyncio
+import logging
+from collections.abc import Awaitable, Callable
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from app.db.models import PendingTransaction
+from app.electrum.client import ElectrumClient
+
+logger = logging.getLogger(__name__)
+
+_POLL_INTERVAL_SECONDS = 10
+
+ConfirmationHandler = Callable[[AsyncSession, PendingTransaction], Awaitable[None]]
+_handlers: dict[str, ConfirmationHandler] = {}
+
+
+def register_handler(kind: str, handler: ConfirmationHandler) -> None:
+    """Domain modules (bets, rounds, withdrawals) register here so this generic
+    poller can notify them when one of their outgoing txs gets its 1st
+    confirmation, without this module importing them directly."""
+    _handlers[kind] = handler
+
+
+async def poll_once(session_factory: async_sessionmaker, client: ElectrumClient) -> int:
+    async with session_factory() as session:
+        pending = (
+            await session.scalars(select(PendingTransaction).where(PendingTransaction.status == "pending"))
+        ).all()
+        pending_ids = [p.id for p in pending]
+
+    confirmed = 0
+    for pending_id, txid, kind in [(p.id, p.current_txid, p.kind) for p in pending]:
+        tx = await client.get_transaction(txid, verbose=True)
+        if not tx or tx.get("confirmations", 0) < 1:
+            continue
+        async with session_factory() as session:
+            row = await session.get(PendingTransaction, pending_id)
+            if row is None or row.status != "pending":
+                continue
+            row.status = "confirmed"
+            handler = _handlers.get(kind)
+            if handler is not None:
+                await handler(session, row)
+            await session.commit()
+        confirmed += 1
+
+    return confirmed
+
+
+class ConfirmationPoller:
+    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:
+        while True:
+            client = self._get_client()
+            if client is not None:
+                try:
+                    await poll_once(self._session_factory, client)
+                except asyncio.CancelledError:
+                    raise
+                except Exception:
+                    logger.exception("confirmation poll failed")
+            await asyncio.sleep(_POLL_INTERVAL_SECONDS)
@@ -0,0 +1,22 @@
+import asyncio
+from contextlib import asynccontextmanager
+
+
+class UserLocks:
+    """Per-user asyncio.Lock registry, shared by PLAY and WITHDRAW so a user can
+    never have a bet-build and a withdrawal-build in flight at once (both would
+    otherwise spend from the same UTXO set on the user's dedicated address).
+
+    Single-process-only by design (an in-memory dict of asyncio.Lock) — this is an
+    accepted MVP constraint; a multi-process deployment would need a DB or Redis
+    lock instead (e.g. a Postgres advisory lock).
+    """
+
+    def __init__(self) -> None:
+        self._locks: dict[int, asyncio.Lock] = {}
+
+    @asynccontextmanager
+    async def acquire(self, user_id: int):
+        lock = self._locks.setdefault(user_id, asyncio.Lock())
+        async with lock:
+            yield
@@ -0,0 +1,181 @@
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from embit import script
+from embit.bip32 import HDKey
+from embit.transaction import Transaction
+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 PendingTransaction, User
+from app.tx.broadcast import RbfError, bump_fee, should_bump
+from app.wallet.plm_network import PLM_MAINNET
+from app.wallet.psbt_builder import Utxo, build_signed_transaction
+
+
+def _key(seed_byte: int) -> HDKey:
+    root = HDKey.from_seed(bytes([seed_byte]) * 32, version=PLM_MAINNET["xprv"])
+    return root.derive("m/84h/746h/0h/0/0")
+
+
+def test_should_bump_false_before_timeout():
+    pending = PendingTransaction(
+        kind="bet", current_txid="x", fee_rate_sat_vb=1, raw_tx_hex="00", status="pending",
+        broadcast_at=datetime.now(timezone.utc),
+    )
+    assert should_bump(pending, datetime.now(timezone.utc), timeout_seconds=900) is False
+
+
+def test_should_bump_true_after_timeout():
+    pending = PendingTransaction(
+        kind="bet", current_txid="x", fee_rate_sat_vb=1, raw_tx_hex="00", status="pending",
+        broadcast_at=datetime.now(timezone.utc) - timedelta(seconds=1000),
+    )
+    assert should_bump(pending, datetime.now(timezone.utc), timeout_seconds=900) is True
+
+
+def test_should_bump_false_when_not_pending():
+    pending = PendingTransaction(
+        kind="bet", current_txid="x", fee_rate_sat_vb=1, raw_tx_hex="00", status="confirmed",
+        broadcast_at=datetime.now(timezone.utc) - timedelta(seconds=1000),
+    )
+    assert should_bump(pending, datetime.now(timezone.utc), timeout_seconds=900) is False
+
+
+class FakeClient:
+    def __init__(self, prevout_values: dict[str, int]):
+        self._prevout_values = prevout_values
+        self.broadcasted: list[str] = []
+
+    async def get_transaction(self, txid: str, verbose: bool = False) -> dict:
+        return {"vout": {0: {"value": self._prevout_values[txid] / 100_000_000}}}
+
+    async def broadcast(self, raw_tx_hex: str) -> str:
+        self.broadcasted.append(raw_tx_hex)
+        return "network-txid"
+
+
+@pytest.fixture
+async def session_factory(tmp_path, monkeypatch):
+    monkeypatch.setattr(settings, "master_key_path", str(tmp_path / "master.xprv.enc"))
+    monkeypatch.setattr(
+        settings,
+        "xprv_encryption_key",
+        __import__("cryptography.fernet", fromlist=["Fernet"]).Fernet.generate_key().decode(),
+    )
+    from app.wallet import hd
+
+    hd._account_key = None
+    hd.generate_master_key()
+
+    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()
+    hd._account_key = None
+
+
+async def test_bump_fee_shrinks_change_and_rebroadcasts(session_factory):
+    from app.wallet.hd import derive_user_address, derive_user_key
+
+    signer = derive_user_key(0)
+    my_address = derive_user_address(0)
+    from_script = script.p2wpkh(signer.to_public())
+    to_address = script.p2wpkh(_key(99).to_public()).address(network=PLM_MAINNET)
+
+    utxo_amount = 150_000_000
+    utxo_txid = "11" * 32
+    built = build_signed_transaction(
+        signing_key=signer,
+        from_script=from_script,
+        utxos=[Utxo(utxo_txid, 0, utxo_amount)],
+        to_address=to_address,
+        amount_sats=10_000_000,
+        change_address=my_address,
+        fee_rate_sat_vb=1,
+    )
+
+    async with session_factory() as session:
+        user = User(username="alice", password_hash="x", derivation_index=0, address=my_address)
+        session.add(user)
+        await session.commit()
+        pending = PendingTransaction(
+            kind="bet",
+            user_id=user.id,
+            current_txid=built.txid,
+            fee_rate_sat_vb=1,
+            raw_tx_hex=built.raw_hex,
+            status="pending",
+            broadcast_at=datetime.now(timezone.utc) - timedelta(seconds=1000),
+        )
+        session.add(pending)
+        await session.commit()
+        pending_id = pending.id
+
+    client = FakeClient({utxo_txid: utxo_amount})
+
+    async with session_factory() as session:
+        row = await session.get(PendingTransaction, pending_id)
+        new_txid = await bump_fee(session, client, row)
+
+    assert client.broadcasted
+    assert new_txid != built.txid
+
+    new_tx = Transaction.parse(bytes.fromhex(client.broadcasted[0]))
+    old_tx = Transaction.parse(bytes.fromhex(built.raw_hex))
+    old_change = next(o.value for o in old_tx.vout if o.script_pubkey.address(network=PLM_MAINNET) == my_address)
+    new_change = next(o.value for o in new_tx.vout if o.script_pubkey.address(network=PLM_MAINNET) == my_address)
+    assert new_change < old_change  # fee bump came out of the change output
+
+    async with session_factory() as session:
+        row = await session.get(PendingTransaction, pending_id)
+        assert row.current_txid == new_txid
+        assert row.fee_rate_sat_vb == 2
+        assert row.attempt_count == 2
+
+
+async def test_bump_fee_raises_when_no_change_output(session_factory):
+    from app.wallet.hd import derive_user_address, derive_user_key
+
+    signer = derive_user_key(0)
+    my_address = derive_user_address(0)
+    from_script = script.p2wpkh(signer.to_public())
+    to_address = script.p2wpkh(_key(98).to_public()).address(network=PLM_MAINNET)
+
+    utxo_amount = 10_000_000  # exact amount, no change output
+    utxo_txid = "22" * 32
+    built = build_signed_transaction(
+        signing_key=signer,
+        from_script=from_script,
+        utxos=[Utxo(utxo_txid, 0, utxo_amount)],
+        to_address=to_address,
+        amount_sats=10_000_000,
+        change_address=my_address,
+        fee_rate_sat_vb=1,
+    )
+
+    async with session_factory() as session:
+        user = User(username="bob", password_hash="x", derivation_index=0, address=my_address)
+        session.add(user)
+        await session.commit()
+        pending = PendingTransaction(
+            kind="bet",
+            user_id=user.id,
+            current_txid=built.txid,
+            fee_rate_sat_vb=1,
+            raw_tx_hex=built.raw_hex,
+            status="pending",
+            broadcast_at=datetime.now(timezone.utc) - timedelta(seconds=1000),
+        )
+        session.add(pending)
+        await session.commit()
+        pending_id = pending.id
+
+    client = FakeClient({utxo_txid: utxo_amount})
+
+    async with session_factory() as session:
+        row = await session.get(PendingTransaction, pending_id)
+        with pytest.raises(RbfError):
+            await bump_fee(session, client, row)
@@ -0,0 +1,82 @@
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+import app.bets.confirmation  # noqa: F401  (registers the "bet" handler)
+import app.rounds.confirmation  # noqa: F401  (registers the "payout" handler)
+from app.db.base import Base
+from app.db.models import PendingTransaction, Round, RoundParticipant
+from app.tx.confirmation import poll_once
+
+
+class FakeClient:
+    def __init__(self, confirmations_by_txid: dict[str, int]):
+        self._confirmations = confirmations_by_txid
+
+    async def get_transaction(self, txid: str, verbose: bool = False) -> dict:
+        return {"confirmations": self._confirmations.get(txid, 0)}
+
+
+@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_bet_confirmation_marks_participant_confirmed(session_factory):
+    async with session_factory() as session:
+        session.add(Round(id=1, status="open"))
+        session.add(
+            RoundParticipant(
+                round_id=1, user_id=1, bet_amount_sats=1_000, bet_txid="tx1", status="broadcast"
+            )
+        )
+        session.add(
+            PendingTransaction(kind="bet", round_id=1, user_id=1, current_txid="tx1", fee_rate_sat_vb=1, raw_tx_hex="00", status="pending")
+        )
+        await session.commit()
+
+    client = FakeClient({"tx1": 1})
+    confirmed = await poll_once(session_factory, client)
+    assert confirmed == 1
+
+    async with session_factory() as session:
+        participant = (await session.scalars(select(RoundParticipant))).one()
+        assert participant.status == "confirmed"
+        assert participant.confirmed_at is not None
+        pending = (await session.scalars(select(PendingTransaction))).one()
+        assert pending.status == "confirmed"
+
+
+async def test_unconfirmed_tx_is_left_pending(session_factory):
+    async with session_factory() as session:
+        session.add(Round(id=2, status="open"))
+        session.add(RoundParticipant(round_id=2, user_id=1, bet_amount_sats=1_000, bet_txid="tx2", status="broadcast"))
+        session.add(PendingTransaction(kind="bet", round_id=2, user_id=1, current_txid="tx2", fee_rate_sat_vb=1, raw_tx_hex="00", status="pending"))
+        await session.commit()
+
+    client = FakeClient({"tx2": 0})
+    confirmed = await poll_once(session_factory, client)
+    assert confirmed == 0
+
+    async with session_factory() as session:
+        participant = (await session.scalars(select(RoundParticipant))).one()
+        assert participant.status == "broadcast"
+
+
+async def test_payout_confirmation_closes_round(session_factory):
+    async with session_factory() as session:
+        session.add(Round(id=3, status="paying_out", payout_txid="tx3"))
+        session.add(PendingTransaction(kind="payout", round_id=3, current_txid="tx3", fee_rate_sat_vb=1, raw_tx_hex="00", status="pending"))
+        await session.commit()
+
+    client = FakeClient({"tx3": 2})
+    confirmed = await poll_once(session_factory, client)
+    assert confirmed == 1
+
+    async with session_factory() as session:
+        round_ = await session.get(Round, 3)
+        assert round_.status == "closed"