Persist the payout before broadcasting it (B-25)

_trigger_payout used to broadcast the payout transaction and only
afterwards write payout_txid and its PendingTransaction. A crash in
that window (docker-compose.yml auto-restarts on crash) left a payout
on-chain with zero record: the round stuck in paying_out, nothing for
the reconciler to resolve, and a manual retry that would have paid the
winner a second time. This mirrors B-08, which already fixed the same
gap for place_bet/request_withdrawal.

_trigger_payout now has four phases: read, build (network read only,
no write), persist the intent as a PendingTransaction(kind="payout",
status="building") and commit, then broadcast and promote to
"pending". A rejected broadcast now leaves that "building" row for
tx/reconcile.py to resolve — its existing building/pending handling
already covers a payout kind correctly, including clearing
payout_txid on abandonment, so reconcile.py needed no changes.

Since pool UTXOs aren't tracked in utxo_events and so can never be
reserved/released the way a user's own UTXOs are, two guards go along
with the two-phase write: _trigger_payout now refuses to build a
second payout for a round that already has a non-terminal
PendingTransaction, and the payout builder excludes any UTXO already
referenced by any non-terminal payout transaction
(_reserved_payout_outpoints) so a stale payout from an earlier round
the reconciler hasn't abandoned yet can't be double-spent by a fresh
attempt.

This makes a payout retry safe; making one happen automatically is
B-26, still open. BUGS.md moves B-25 to "Previously fixed" with the
fix description; the suite grows from 139 to 143 tests.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

tests/unit/test_scheduler.py

@@ -4,9 +4,10 @@ import pytest
 from sqlalchemy import select
 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 Round, RoundConfig
-from app.rounds.scheduler import RoundScheduler
+from app.db.models import AuditLog, PendingTransaction, Round, RoundConfig, User
+from app.rounds.scheduler import RoundScheduler, _reserved_payout_outpoints
 
 
 class FakeListener:
@@ -52,3 +53,197 @@ async def test_tick_closes_round_with_no_participants_once_due(session_factory,
     async with session_factory() as session:
         round_ = (await session.scalars(select(Round))).one()
         assert round_.status == "closed"
+
+
+# --- B-25: the payout must be persisted before it is broadcast, like bets/withdrawals ---
+
+# A real, reusable PLM bech32 address so build_payout_transaction's
+# script.Script.from_address(...) succeeds — this is not a value the scheduler
+# validates itself (that's the admin panel's job for fee_address), it just needs to
+# actually decode.
+_WINNER_ADDRESS = "plm1q0xcqpzrky6eff2g52qdye53xkk9jxkvraxkkwd"
+_FEE_ADDRESS = "plm1q5x25wd6q463mfhckjraaedgjg0lyu73qfcj43n"
+
+_POOL_AMOUNT_SATS = 10_000_000_000  # 100 PLM
+
+
+class FakePayoutClient:
+    def __init__(self, entries, *, fail_broadcast=False):
+        self._entries = entries
+        self._fail_broadcast = fail_broadcast
+        self.broadcasted: list[str] = []
+
+    async def listunspent(self, scripthash):
+        return self._entries
+
+    async def broadcast(self, raw_tx_hex):
+        if self._fail_broadcast:
+            raise RuntimeError("node rejected the transaction")
+        self.broadcasted.append(raw_tx_hex)
+        return "network-txid"
+
+
+class FakePayoutListener:
+    def __init__(self, client):
+        self.client = client
+
+
+@pytest.fixture
+async def payout_session_factory(tmp_path, monkeypatch):
+    """Same master-key bootstrap as test_broadcast.py's fixture: _trigger_payout
+    needs a real pool key to sign with."""
+    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 _seed_paying_out_round(session_factory, round_id: int = 1) -> int:
+    async with session_factory() as session:
+        winner = User(username="winner", password_hash="x", derivation_index=0, address=_WINNER_ADDRESS)
+        session.add(winner)
+        await session.flush()
+        session.add(RoundConfig(fee_address=_FEE_ADDRESS, fee_rate_sat_vb=1))
+        session.add(
+            Round(
+                id=round_id,
+                status="paying_out",
+                pool_amount_sats=_POOL_AMOUNT_SATS,
+                winner_user_id=winner.id,
+            )
+        )
+        await session.commit()
+        return winner.id
+
+
+async def test_trigger_payout_persists_before_broadcasting(payout_session_factory):
+    """The happy path: payout_txid and a PendingTransaction must exist once the
+    broadcast succeeds, promoted from "building" to "pending" — the two-phase write
+    that used to be missing entirely (B-25)."""
+    await _seed_paying_out_round(payout_session_factory)
+    entries = [{"tx_hash": "33" * 32, "tx_pos": 0, "height": 10, "value": _POOL_AMOUNT_SATS + 100_000}]
+    client = FakePayoutClient(entries)
+    scheduler = RoundScheduler(payout_session_factory, FakePayoutListener(client))
+
+    await scheduler._trigger_payout(1)
+
+    assert client.broadcasted
+    async with payout_session_factory() as session:
+        round_ = await session.get(Round, 1)
+        assert round_.payout_txid is not None
+        assert round_.winner_amount_sats and round_.fee_amount_sats
+
+        pending = (await session.scalars(select(PendingTransaction))).one()
+        assert pending.kind == "payout"
+        assert pending.status == "pending"
+        assert pending.current_txid == round_.payout_txid
+
+        events = [e.event_type for e in (await session.scalars(select(AuditLog))).all()]
+        assert "payout_sent" in events
+
+
+async def test_trigger_payout_broadcast_failure_leaves_a_recoverable_row(payout_session_factory):
+    """Before B-25, a broadcast rejection here left nothing behind — no payout_txid,
+    no PendingTransaction — because everything was persisted only after the
+    broadcast. Now the intent is already durable, so the reconciler has something to
+    resolve instead of the round being stuck with zero trace of what was attempted."""
+    await _seed_paying_out_round(payout_session_factory)
+    entries = [{"tx_hash": "44" * 32, "tx_pos": 0, "height": 10, "value": _POOL_AMOUNT_SATS + 100_000}]
+    client = FakePayoutClient(entries, fail_broadcast=True)
+    scheduler = RoundScheduler(payout_session_factory, FakePayoutListener(client))
+
+    await scheduler._trigger_payout(1)
+
+    assert not client.broadcasted
+    async with payout_session_factory() as session:
+        round_ = await session.get(Round, 1)
+        assert round_.payout_txid is not None  # durable, even though the broadcast failed
+
+        pending = (await session.scalars(select(PendingTransaction))).one()
+        assert pending.kind == "payout"
+        assert pending.status == "building"  # not lost — the reconciler resolves this
+        assert pending.current_txid == round_.payout_txid
+
+        events = [e.event_type for e in (await session.scalars(select(AuditLog))).all()]
+        assert "payout_failed" in events
+
+
+async def test_trigger_payout_skips_when_already_in_flight(payout_session_factory):
+    """A second call for a round that already has a non-terminal payout
+    PendingTransaction must not build (and broadcast) another one — that would pay
+    the winner twice."""
+    winner_id = await _seed_paying_out_round(payout_session_factory)
+    async with payout_session_factory() as session:
+        round_ = await session.get(Round, 1)
+        round_.payout_txid = "already-sent-txid"
+        session.add(
+            PendingTransaction(
+                kind="payout",
+                round_id=1,
+                current_txid="already-sent-txid",
+                fee_rate_sat_vb=1,
+                raw_tx_hex="00",
+                status="pending",
+            )
+        )
+        await session.commit()
+
+    entries = [{"tx_hash": "55" * 32, "tx_pos": 0, "height": 10, "value": _POOL_AMOUNT_SATS + 100_000}]
+    client = FakePayoutClient(entries)
+    scheduler = RoundScheduler(payout_session_factory, FakePayoutListener(client))
+
+    await scheduler._trigger_payout(1)
+
+    assert not client.broadcasted
+    async with payout_session_factory() as session:
+        assert (await session.scalars(select(PendingTransaction))).all()  # still just the one seeded
+        rows = (await session.scalars(select(PendingTransaction))).all()
+        assert len(rows) == 1
+        assert rows[0].current_txid == "already-sent-txid"
+
+
+async def test_reserved_payout_outpoints_excludes_utxos_claimed_by_a_stale_payout(payout_session_factory):
+    """A payout still "building"/"pending" for some round — most plausibly a stale
+    one the reconciler hasn't abandoned yet — must keep its inputs off the table for
+    a fresh payout attempt, or the same pool coins could be spent twice."""
+    from embit import script
+    from embit.transaction import Transaction, TransactionInput, TransactionOutput
+
+    reserved_txid = "66" * 32
+    raw_tx = (
+        Transaction(
+            vin=[TransactionInput(bytes.fromhex(reserved_txid), 2)],
+            vout=[TransactionOutput(1_000_000, script.Script.from_address(_WINNER_ADDRESS))],
+        )
+        .serialize()
+        .hex()
+    )
+    async with payout_session_factory() as session:
+        session.add(
+            PendingTransaction(
+                kind="payout",
+                round_id=99,
+                current_txid="stale-payout-txid",
+                fee_rate_sat_vb=1,
+                raw_tx_hex=raw_tx,
+                status="building",
+            )
+        )
+        await session.commit()
+
+        reserved = await _reserved_payout_outpoints(session)
+
+    assert reserved == {(reserved_txid, 2)}