From fc2aadbc7e9e73128889c020357638ca51b8eaf9 Mon Sep 17 00:00:00 2001 From: Davide Grilli Date: Tue, 21 Jul 2026 10:25:57 +0200 Subject: [PATCH] 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 --- app/tx/__init__.py | 0 app/tx/broadcast.py | 163 ++++++++++++++++++++++++++++ app/tx/confirmation.py | 67 ++++++++++++ app/tx/locks.py | 22 ++++ tests/unit/test_broadcast.py | 181 ++++++++++++++++++++++++++++++++ tests/unit/test_confirmation.py | 82 +++++++++++++++ 6 files changed, 515 insertions(+) create mode 100644 app/tx/__init__.py create mode 100644 app/tx/broadcast.py create mode 100644 app/tx/confirmation.py create mode 100644 app/tx/locks.py create mode 100644 tests/unit/test_broadcast.py create mode 100644 tests/unit/test_confirmation.py diff --git a/app/tx/__init__.py b/app/tx/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/tx/broadcast.py b/app/tx/broadcast.py new file mode 100644 index 0000000..821fe6e --- /dev/null +++ b/app/tx/broadcast.py @@ -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) diff --git a/app/tx/confirmation.py b/app/tx/confirmation.py new file mode 100644 index 0000000..edba732 --- /dev/null +++ b/app/tx/confirmation.py @@ -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) diff --git a/app/tx/locks.py b/app/tx/locks.py new file mode 100644 index 0000000..6439bb5 --- /dev/null +++ b/app/tx/locks.py @@ -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 diff --git a/tests/unit/test_broadcast.py b/tests/unit/test_broadcast.py new file mode 100644 index 0000000..072b483 --- /dev/null +++ b/tests/unit/test_broadcast.py @@ -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) diff --git a/tests/unit/test_confirmation.py b/tests/unit/test_confirmation.py new file mode 100644 index 0000000..24bf060 --- /dev/null +++ b/tests/unit/test_confirmation.py @@ -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"