plm-lottery/tests/unit/test_tx_versions.py

"""B-77 (TX-19, TX-12, TX-13, TX-21, TX-06, TX-15, DAT-13, DAT-14, DAT-34, DAT-38, DAT-70, DAT-71):
every version of a bumped transaction is tracked, and an RBF replacement is written
before it is broadcast.

Before the fix only `current_txid` was recognised: if an earlier version of a bumped
transaction confirmed, nothing noticed, and six hours later the reconciler abandoned the
row — a confirmed bet's participant deleted, a sent withdrawal marked failed, a paid
payout rebuilt. And `bump_fee` broadcast first and persisted afterwards, unconditionally.
"""

from datetime import datetime, timedelta, timezone

import pytest
from embit import script
from sqlalchemy import select
from sqlalchemy.exc import OperationalError

import app.bets.confirmation  # noqa: F401  (registers the "bet" handler)
import app.rounds.confirmation  # noqa: F401  (registers the "payout" handler)
import app.withdrawals.confirmation  # noqa: F401  (registers the "withdrawal" handler)
from app.bets.service import place_bet
from app.db.models import (
    AuditLog,
    PendingTransaction,
    PendingTransactionTxid,
    Round,
    RoundParticipant,
    User,
    UtxoEvent,
    Withdrawal,
)
from app.tx import broadcast as broadcast_module
from app.tx.broadcast import RbfError, bump_fee
from app.tx.confirmation import poll_once
from app.tx.reconcile import reconcile_once
from app.wallet.hd import derive_pool_address, derive_user_address, derive_user_key
from app.wallet.plm_network import PLM_MAINNET
from app.wallet.psbt_builder import Utxo, build_signed_transaction
from app.withdrawals.service import request_withdrawal
from tests.unit.fake_network import FakeNetwork, ScriptedNetwork, rpc_error
from tests.unit.test_bets import _make_funded_user
from tests.unit.test_bets import session_factory as bets_session_factory  # noqa: F401  (fixture)
from tests.unit.test_broadcast import FakeClient as ChainClient, session_factory  # noqa: F401  (fixture)

_UTXO_TXID = "11" * 32
_UTXO_AMOUNT = 150_000_000


async def _seed_pending_bet(session_factory, *, age_seconds: int = 1000) -> dict:
    """A user, a round, a participant and a `pending` bet row, with its first version
    recorded the way place_bet records it."""
    signer = derive_user_key(0)
    address = derive_user_address(0)
    built = build_signed_transaction(
        signing_key=signer,
        from_script=script.p2wpkh(signer.to_public()),
        utxos=[Utxo(_UTXO_TXID, 0, _UTXO_AMOUNT)],
        to_address=derive_pool_address(1),
        amount_sats=10_000_000,
        change_address=address,
        fee_rate_sat_vb=1,
    )
    broadcast_at = datetime.now(timezone.utc) - timedelta(seconds=age_seconds)
    async with session_factory() as session:
        user = User(username="alice", password_hash="x", derivation_index=0, address=address)
        session.add(user)
        await session.flush()
        session.add(Round(id=1, status="open", pool_address=derive_pool_address(1)))
        session.add(
            UtxoEvent(user_id=user.id, txid=_UTXO_TXID, vout=0, amount_sats=_UTXO_AMOUNT, confirmed_height=5, spent_txid=built.txid)
        )
        session.add(
            RoundParticipant(
                round_id=1, user_id=user.id, bet_amount_sats=10_000_000, bet_txid=built.txid, status="broadcast"
            )
        )
        pending = PendingTransaction(
            kind="bet",
            round_id=1,
            user_id=user.id,
            current_txid=built.txid,
            fee_rate_sat_vb=1,
            raw_tx_hex=built.raw_hex,
            status="pending",
            broadcast_at=broadcast_at,
            last_broadcast_at=broadcast_at,
        )
        session.add(pending)
        await session.flush()
        session.add(PendingTransactionTxid(pending_transaction_id=pending.id, txid=built.txid, raw_hex=built.raw_hex))
        await session.commit()
        return {"pending_id": pending.id, "user_id": user.id, "first_txid": built.txid, "first_raw": built.raw_hex}


class _Chain(ChainClient):
    """prevout lookups for bump_fee, plus a scriptable broadcast and history."""

    def __init__(self, on_broadcast=None, fail_with=None):
        super().__init__({_UTXO_TXID: _UTXO_AMOUNT})
        self.on_broadcast = on_broadcast
        self.fail_with = fail_with
        self.history: list[dict] = []

    async def broadcast(self, raw_tx_hex: str) -> str:
        if self.on_broadcast is not None:
            await self.on_broadcast(raw_tx_hex)
        if self.fail_with is not None:
            raise self.fail_with
        return await super().broadcast(raw_tx_hex)

    async def get_history(self, scripthash: str) -> list[dict]:
        return self.history


async def _versions(session_factory, pending_id):
    async with session_factory() as session:
        rows = (
            await session.scalars(
                select(PendingTransactionTxid)
                .where(PendingTransactionTxid.pending_transaction_id == pending_id)
                .order_by(PendingTransactionTxid.id)
            )
        ).all()
        return [(r.txid, r.raw_hex) for r in rows]


# --- the first version is written with the row (DAT-70) --------------------------------


async def test_a_bet_records_its_first_version_and_a_refusal_deletes_it_with_the_row(bets_session_factory):
    user_id = await _make_funded_user(bets_session_factory, 4, 3_000_000_000)

    class _Ok:
        async def broadcast(self, raw):
            return "x"

    async with bets_session_factory() as session:
        user = await session.get(User, user_id)
        participant = await place_bet(session, _Ok(), user, network=FakeNetwork())
    async with bets_session_factory() as session:
        pending = (await session.scalars(select(PendingTransaction))).one()
        version = (await session.scalars(select(PendingTransactionTxid))).one()
        assert (version.pending_transaction_id, version.txid) == (pending.id, participant.bet_txid)
        assert version.raw_hex == pending.raw_tx_hex

    other = await _make_funded_user(bets_session_factory, 5, 3_000_000_000)

    class _Refuse:
        async def broadcast(self, raw):
            raise rpc_error("min relay fee not met")

    from app.bets.service import BetError

    async with bets_session_factory() as session:
        user = await session.get(User, other)
        with pytest.raises(BetError):
            await place_bet(session, _Refuse(), user, network=FakeNetwork(known=False))
    async with bets_session_factory() as session:
        # DAT-38: the refused bet's row and its versions are gone; the first bet's stay.
        assert len((await session.scalars(select(PendingTransactionTxid))).all()) == 1


async def test_a_withdrawal_records_its_first_version(bets_session_factory):
    user_id = await _make_funded_user(bets_session_factory, 7, 3_000_000_000)

    class _Ok:
        async def broadcast(self, raw):
            return "x"

    async with bets_session_factory() as session:
        user = await session.get(User, user_id)
        withdrawal = await request_withdrawal(
            session, _Ok(), user, "plm1qqph9qup2mp7w7g5nlsdhdc9m2pp44ampzw0ctx", 1_000_000_000, network=FakeNetwork()
        )
    async with bets_session_factory() as session:
        pending = (await session.scalars(select(PendingTransaction))).one()
        version = (await session.scalars(select(PendingTransactionTxid))).one()
        assert (version.txid, version.raw_hex) == (withdrawal.txid, pending.raw_tx_hex)


async def test_a_payout_records_its_first_version(payout_session_factory):
    from app.rounds.scheduler import RoundScheduler
    from tests.unit.test_scheduler import FakePayoutClient, FakePayoutListener, _POOL_AMOUNT_SATS, _seed_paying_out_round

    await _seed_paying_out_round(payout_session_factory)
    entries = [{"tx_hash": "44" * 32, "tx_pos": 0, "height": 10, "value": _POOL_AMOUNT_SATS + 100_000}]
    await RoundScheduler(payout_session_factory, FakePayoutListener(FakePayoutClient(entries)))._trigger_payout(1)
    async with payout_session_factory() as session:
        pending = (await session.scalars(select(PendingTransaction))).one()
        version = (await session.scalars(select(PendingTransactionTxid))).one()
        assert (version.txid, version.raw_hex) == (pending.current_txid, pending.raw_tx_hex)


from tests.unit.test_scheduler import payout_session_factory  # noqa: E402,F401  (fixture)


# --- bump_fee: write first, then broadcast (TX-12, DAT-34) ------------------------------


async def test_the_replacement_is_recorded_as_a_candidate_before_it_is_broadcast(session_factory):
    seeded = await _seed_pending_bet(session_factory)
    seen: dict = {}

    async def observe(raw_tx_hex: str) -> None:
        # An independent session, mid-broadcast: what has been committed so far.
        async with session_factory() as probe:
            seen["versions"] = [
                v.txid for v in (await probe.scalars(select(PendingTransactionTxid))).all()
            ]
            seen["current"] = (await probe.get(PendingTransaction, seeded["pending_id"])).current_txid

    client = _Chain(on_broadcast=observe)
    new_txid = await bump_fee(session_factory, client, seeded["pending_id"])

    assert new_txid is not None and new_txid != seeded["first_txid"]
    assert seen["current"] == seeded["first_txid"]  # the row had not been switched yet...
    assert new_txid in seen["versions"]  # ...but the replacement was already on record
    assert [txid for txid, _ in await _versions(session_factory, seeded["pending_id"])] == [seeded["first_txid"], new_txid]
    async with session_factory() as session:
        row = await session.get(PendingTransaction, seeded["pending_id"])
        assert row.current_txid == new_txid and row.replaced_by_txid == seeded["first_txid"]
        assert row.attempt_count == 2 and row.fee_rate_sat_vb == 2
        participant = (await session.scalars(select(RoundParticipant))).one()
        utxo = (await session.scalars(select(UtxoEvent))).one()
        assert participant.bet_txid == utxo.spent_txid == new_txid


async def test_every_bump_adds_a_version_and_all_of_them_stay(session_factory):
    seeded = await _seed_pending_bet(session_factory)
    client = _Chain()
    second = await bump_fee(session_factory, client, seeded["pending_id"])
    third = await bump_fee(session_factory, client, seeded["pending_id"])
    assert [txid for txid, _ in await _versions(session_factory, seeded["pending_id"])] == [
        seeded["first_txid"],
        second,
        third,
    ]
    # DAT-70: the raw transaction of every version is kept.
    assert all(raw for _, raw in await _versions(session_factory, seeded["pending_id"]))


async def test_a_crash_between_the_broadcast_and_the_switch_over_leaves_the_replacement_on_record(
    session_factory, monkeypatch
):
    """The old phase 3 ran after the broadcast and was unconditional, so a crash or a
    `database is locked` there left a transaction on the network the DB did not know. Now
    the candidate is already committed, and the confirmation poller recognises it."""
    seeded = await _seed_pending_bet(session_factory)
    real = broadcast_module.begin_immediate
    calls = {"n": 0}

    async def flaky(session):
        # begin_immediate calls in bump_fee: the candidate write (1), then the switch-over
        # (2). Failing the second is the lock timing out right after the broadcast.
        calls["n"] += 1
        if calls["n"] == 2:
            raise OperationalError("BEGIN IMMEDIATE", {}, Exception("database is locked"))
        await real(session)

    monkeypatch.setattr(broadcast_module, "begin_immediate", flaky)
    client = _Chain()
    with pytest.raises(OperationalError):
        await bump_fee(session_factory, client, seeded["pending_id"])
    monkeypatch.setattr(broadcast_module, "begin_immediate", real)

    assert len(client.broadcasted) == 1  # the replacement is on the network
    versions = await _versions(session_factory, seeded["pending_id"])
    assert len(versions) == 2  # ...and in the database
    candidate = versions[1][0]
    async with session_factory() as session:
        row = await session.get(PendingTransaction, seeded["pending_id"])
        assert row.current_txid == seeded["first_txid"]  # not switched over

    # The network confirms the candidate: the poller recognises it and realigns the row.
    client.history = [{"tx_hash": candidate, "height": 101}]
    assert await poll_once(session_factory, client, FakeNetwork()) == 1
    async with session_factory() as session:
        row = await session.get(PendingTransaction, seeded["pending_id"])
        assert row.status == "confirmed" and row.current_txid == candidate
        assert (await session.scalars(select(UtxoEvent))).one().spent_txid == candidate
        participant = (await session.scalars(select(RoundParticipant))).one()
        assert participant.bet_txid == candidate and participant.status == "confirmed"


async def test_a_bump_retried_after_a_lost_switch_over_reuses_the_candidate(session_factory, monkeypatch):
    """The same replacement is rebuilt (deterministic signing): recording it again must
    not trip the unique txid."""
    seeded = await _seed_pending_bet(session_factory)
    real = broadcast_module.begin_immediate
    calls = {"n": 0}

    async def flaky(session):
        calls["n"] += 1
        if calls["n"] == 2:
            raise OperationalError("BEGIN IMMEDIATE", {}, Exception("database is locked"))
        await real(session)

    monkeypatch.setattr(broadcast_module, "begin_immediate", flaky)
    with pytest.raises(OperationalError):
        await bump_fee(session_factory, _Chain(), seeded["pending_id"])
    monkeypatch.setattr(broadcast_module, "begin_immediate", real)

    new_txid = await bump_fee(session_factory, _Chain(), seeded["pending_id"])
    assert new_txid is not None
    assert len(await _versions(session_factory, seeded["pending_id"])) == 2


# --- the conditional update (TX-13) ------------------------------------------------------


async def test_a_row_that_changed_during_the_broadcast_is_not_overwritten(session_factory):
    """An abandon or a confirmation that lands between the broadcast and the switch-over
    used to be silently overwritten. Now zero matched rows is a conflict: the row stays as
    it is, the candidate stays recorded."""
    seeded = await _seed_pending_bet(session_factory)

    async def confirmed_meanwhile(raw_tx_hex: str) -> None:
        async with session_factory() as session:
            row = await session.get(PendingTransaction, seeded["pending_id"])
            row.status = "confirmed"
            await session.commit()

    result = await bump_fee(session_factory, _Chain(on_broadcast=confirmed_meanwhile), seeded["pending_id"])

    assert result is None
    async with session_factory() as session:
        row = await session.get(PendingTransaction, seeded["pending_id"])
        assert row.status == "confirmed"
        assert row.current_txid == seeded["first_txid"] and row.attempt_count == 1
        assert (await session.scalars(select(UtxoEvent))).one().spent_txid == seeded["first_txid"]
    assert len(await _versions(session_factory, seeded["pending_id"])) == 2  # the candidate stays


async def test_a_row_rebumped_during_the_broadcast_is_not_overwritten(session_factory):
    seeded = await _seed_pending_bet(session_factory)

    async def rebumped_meanwhile(raw_tx_hex: str) -> None:
        async with session_factory() as session:
            row = await session.get(PendingTransaction, seeded["pending_id"])
            row.current_txid = "ee" * 32
            await session.commit()

    assert await bump_fee(session_factory, _Chain(on_broadcast=rebumped_meanwhile), seeded["pending_id"]) is None
    async with session_factory() as session:
        assert (await session.get(PendingTransaction, seeded["pending_id"])).current_txid == "ee" * 32


async def test_an_explicitly_refused_replacement_leaves_the_row_alone(session_factory):
    """TX-24: a verified refusal raises RbfError and the row keeps the version that is
    on the network."""
    seeded = await _seed_pending_bet(session_factory)
    client = _Chain(fail_with=rpc_error("insufficient fee, rejecting replacement"))
    with pytest.raises(RbfError, match="refused the replacement"):
        await bump_fee(session_factory, client, seeded["pending_id"], network=FakeNetwork(known=False))
    async with session_factory() as session:
        row = await session.get(PendingTransaction, seeded["pending_id"])
        assert row.current_txid == seeded["first_txid"] and row.attempt_count == 1


async def test_an_uncertain_replacement_is_adopted_and_an_accepted_one_is_fanned_out(session_factory):
    from app.electrum.client import ElectrumError

    seeded = await _seed_pending_bet(session_factory)
    network = FakeNetwork(known=False)
    new_txid = await bump_fee(
        session_factory, _Chain(fail_with=ElectrumError("connection closed")), seeded["pending_id"], network=network
    )
    assert new_txid is not None and network.fanned_out == []  # uncertain: not fanned out, but adopted
    async with session_factory() as session:
        assert (await session.get(PendingTransaction, seeded["pending_id"])).current_txid == new_txid

    ok_network = FakeNetwork()
    client = _Chain()
    third = await bump_fee(session_factory, client, seeded["pending_id"], network=ok_network)
    assert third is not None and ok_network.fanned_out == client.broadcasted  # TX-21


# --- an earlier version confirming (TX-19, DAT-14) -----------------------------------------


async def test_the_confirmation_of_an_earlier_version_confirms_the_row_and_retargets_everything(session_factory):
    """The case that used to be missed: v1 is bumped to v2 and v3, then *v1* confirms.
    Only v3 was looked for, so the row stayed pending, and six hours later was abandoned
    — the participant of a confirmed bet deleted."""
    seeded = await _seed_pending_bet(session_factory)
    client = _Chain()
    await bump_fee(session_factory, client, seeded["pending_id"])
    third = await bump_fee(session_factory, client, seeded["pending_id"])

    client.history = [{"tx_hash": seeded["first_txid"], "height": 77}]
    assert await poll_once(session_factory, client, FakeNetwork()) == 1

    async with session_factory() as session:
        row = await session.get(PendingTransaction, seeded["pending_id"])
        assert row.status == "confirmed"
        assert row.current_txid == seeded["first_txid"] != third
        assert row.raw_tx_hex == seeded["first_raw"]
        assert (await session.scalars(select(UtxoEvent))).one().spent_txid == seeded["first_txid"]
        participant = (await session.scalars(select(RoundParticipant))).one()
        assert participant.bet_txid == seeded["first_txid"] and participant.status == "confirmed"


async def test_an_unrelated_txid_in_the_history_does_not_confirm_the_row(session_factory):
    seeded = await _seed_pending_bet(session_factory)
    client = _Chain()
    client.history = [{"tx_hash": "ab" * 32, "height": 5}]
    assert await poll_once(session_factory, client, FakeNetwork()) == 0
    async with session_factory() as session:
        assert (await session.get(PendingTransaction, seeded["pending_id"])).status == "pending"


# --- the reconciler (TX-15, DAT-13) ----------------------------------------------------------


async def test_the_reconciler_does_not_abandon_a_row_whose_earlier_version_is_on_the_network(session_factory):
    """Before the fix a row older than 6h whose *current* txid the server did not list
    was abandoned — even though an earlier version of it was in the mempool or mined."""
    seeded = await _seed_pending_bet(session_factory, age_seconds=7 * 3600)
    client = _Chain()
    await bump_fee(session_factory, client, seeded["pending_id"])
    client.history = [{"tx_hash": seeded["first_txid"], "height": 0}]  # only the old version is known

    await reconcile_once(session_factory, client)

    async with session_factory() as session:
        row = await session.get(PendingTransaction, seeded["pending_id"])
        assert row.status == "pending"
        assert row.current_txid == seeded["first_txid"]  # the row follows the version on the network
        assert (await session.scalars(select(RoundParticipant))).one().bet_txid == seeded["first_txid"]
        assert "pending_tx_abandoned" not in [e.event_type for e in (await session.scalars(select(AuditLog))).all()]


async def test_abandoning_a_row_releases_inputs_claimed_by_any_of_its_versions(session_factory):
    """DAT-13: an input still marked with an earlier version's txid is this row's claim
    too, so abandoning the row frees it."""
    seeded = await _seed_pending_bet(session_factory, age_seconds=7 * 3600)
    client = _Chain()
    await bump_fee(session_factory, client, seeded["pending_id"])
    async with session_factory() as session:  # simulate a claim left on the earlier version
        (await session.scalars(select(UtxoEvent))).one().spent_txid = seeded["first_txid"]
        await session.commit()
    client.history = []  # nobody knows any version

    await reconcile_once(session_factory, client, ScriptedNetwork.proof_of_absence())

    async with session_factory() as session:
        assert (await session.get(PendingTransaction, seeded["pending_id"])).status == "failed"
        assert (await session.scalars(select(UtxoEvent))).one().spent_txid is None