Let the system recover from a broadcast that never confirms

The code treated a broadcast as final: money moved on-chain and the DB was
updated on the assumption it would either confirm or be fee-bumped until it
did. Neither is guaranteed, and every way that assumption broke was permanent
(BUGS.md B-02, B-03, B-04, B-07, B-08, B-20, B-21).

Persist before broadcasting. place_bet and request_withdrawal now write their
rows in a "building" state and commit, then broadcast, then promote to
broadcast/pending in a second commit. Before, a failure or crash between the
broadcast and the commit left the coins irreversibly spent with no trace: no
participant (so no entry in the draw), no pending row (so no RBF and no
confirmation tracking), and the UTXOs not even marked spent, so the next bet
would try to double-spend them. A refused broadcast now releases the reserved
UTXOs, restores the balance, removes the participant (or marks the withdrawal
failed), audit-logs it, and answers a translatable broadcast_failed — as 502,
since the network refused it, not the caller, where it used to be an opaque 500.

Reconcile what's in flight against the chain. New PendingTransactionReconciler
(app/tx/reconcile.py, every 120s and once at startup) asks whether each
non-terminal tx exists: present -> promote, gone -> mark failed with a reason,
release the inputs, roll the domain row back, audit-log it. Grace periods differ
by state (120s for "building", 6h for "pending", so the RBF bumper gets its
attempts first). It is deliberately biased to inaction: only a server that
positively doesn't know the tx counts as absent, and a transport failure never
abandons anything, because releasing a UTXO whose tx is actually alive would
invite a double-spend. Verified against the live server, which answers "No such
mempool or blockchain transaction" for an unknown txid.

Stop keying on a value that changes. An RBF bump changes the txid, and
_on_bet_confirmed looked the participant up by bet_txid — so a bumped bet
confirmed under a txid no participant carried, the row stayed "broadcast"
forever, and the scheduler waited on it forever: the round could never close and
the lottery stopped. Handlers now resolve by immutable ids (round_id/user_id,
withdrawal_id), and bump_fee retargets every stored txid — bet_txid,
Withdrawal.txid, Round.payout_txid and UtxoEvent.spent_txid — plus records the
previous one in replaced_by_txid, which was never written at all.

One bad row no longer blocks the rest. The confirmation poller's per-tx lookup
is guarded: a txid the server can't resolve used to abort the whole pass, so
nothing confirmed again until an operator intervened. It also selects plain
columns instead of hydrating entities that outlive their session.

Tests: 6 reconciler cases including "a broken connection must not release coins";
the bet-ordering test probes committed state from an independent session during
the broadcast, and caught a real mistake in the first draft of this change (the
_pending_transaction helper still hardcoded status="pending", so rows were born
already-broadcast and would have got the 6-hour grace instead of 120s).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-27 00:31:24 +02:00
co-authored by Claude Opus 5
parent cc88763a9d
commit d528c5b475
16 changed files with 988 additions and 36 deletions
+69
View File
@@ -133,3 +133,72 @@ async def test_place_bet_rejects_after_timer_expires_even_if_still_open(session_
assert len(participants) == 0
round_ = (await session.scalars(select(Round))).one()
assert round_.status == "open" # scheduler hasn't ticked — status is unchanged, only the check is deadline-aware
class RejectingElectrumClient:
"""A node that refuses the transaction — fee too low, dust output, mempool
conflict, or simply an unreachable server."""
async def broadcast(self, raw_tx_hex: str) -> str:
raise RuntimeError("min relay fee not met")
async def test_failed_broadcast_leaves_nothing_behind(session_factory):
"""B-07/B-08: the broadcast used to happen before anything was written, so a
rejection left the UTXOs marked spent with no rows to explain it, and the caller
got an opaque HTTP 500. Now it's a translatable error and a full rollback."""
user_id = await _make_funded_user(session_factory, 4, 3_000_000_000)
async with session_factory() as session:
user = await session.get(User, user_id)
with pytest.raises(BetError, match="refused"):
await place_bet(session, RejectingElectrumClient(), user)
async with session_factory() as session:
utxo = (await session.scalars(select(UtxoEvent).where(UtxoEvent.user_id == user_id))).one()
assert utxo.spent_txid is None # released, so the user can bet again
assert (await session.scalars(select(RoundParticipant))).all() == []
assert (await session.scalars(select(PendingTransaction))).all() == []
user = await session.get(User, user_id)
assert user.cached_balance_sats == 3_000_000_000
events = [e.event_type for e in (await session.scalars(select(AuditLog))).all()]
assert "bet_broadcast_failed" in events
assert "bet_placed" not in events
async def test_failed_broadcast_reports_the_broadcast_failed_code(session_factory):
user_id = await _make_funded_user(session_factory, 5, 3_000_000_000)
async with session_factory() as session:
user = await session.get(User, user_id)
try:
await place_bet(session, RejectingElectrumClient(), user)
assert False, "expected BetError"
except BetError as exc:
assert exc.code == "broadcast_failed"
async def test_bet_is_persisted_before_it_is_broadcast(session_factory):
"""The ordering guarantee behind B-08: by the time the network call happens, the
rows already exist, so a crash there is recoverable rather than silent."""
user_id = await _make_funded_user(session_factory, 6, 3_000_000_000)
seen: dict[str, object] = {}
class ObservingClient:
async def broadcast(self, raw_tx_hex: str) -> str:
# Read committed state from an independent session, mid-broadcast.
async with session_factory() as probe:
seen["pending"] = [
(p.kind, p.status) for p in (await probe.scalars(select(PendingTransaction))).all()
]
seen["participants"] = [
(p.status) for p in (await probe.scalars(select(RoundParticipant))).all()
]
return "network-txid"
async with session_factory() as session:
user = await session.get(User, user_id)
await place_bet(session, ObservingClient(), user)
assert seen["pending"] == [("bet", "building")]
assert seen["participants"] == ["building"]
+69
View File
@@ -179,3 +179,72 @@ async def test_bump_fee_raises_when_no_change_output(session_factory):
row = await session.get(PendingTransaction, pending_id)
with pytest.raises(RbfError):
await bump_fee(session, client, row)
async def test_bump_fee_retargets_every_stored_txid(session_factory):
"""B-02/B-20: a bump changes the txid, and everything that recorded the old one
has to follow — the participant's bet_txid (whose staleness used to wedge the
round forever), the UTXO's spent_txid (which the reconciler matches on), and
replaced_by_txid, which was never written at all."""
from app.db.models import Round, RoundParticipant, UtxoEvent
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 = 150_000_000
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)
session.add(Round(id=1, status="open"))
await session.flush()
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=built.recipient_sats,
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=datetime.now(timezone.utc) - timedelta(seconds=1000),
)
session.add(pending)
await session.commit()
pending_id = pending.id
async with session_factory() as session:
row = await session.get(PendingTransaction, pending_id)
new_txid = await bump_fee(session, FakeClient({utxo_txid: utxo_amount}), row)
async with session_factory() as session:
from sqlalchemy import select
row = await session.get(PendingTransaction, pending_id)
assert row.current_txid == new_txid
assert row.replaced_by_txid == built.txid # points backwards at what it replaced
participant = (await session.scalars(select(RoundParticipant))).one()
assert participant.bet_txid == new_txid
utxo = (await session.scalars(select(UtxoEvent))).one()
assert utxo.spent_txid == new_txid
+91
View File
@@ -80,3 +80,94 @@ async def test_payout_confirmation_closes_round(session_factory):
async with session_factory() as session:
round_ = await session.get(Round, 3)
assert round_.status == "closed"
class ExplodingClient:
"""Answers for one txid and raises for the other — a tx the server no longer
knows (dropped from the mempool, replaced by a bump)."""
def __init__(self, known: dict[str, int], exploding_txid: str):
self._known = known
self._exploding = exploding_txid
async def get_transaction(self, txid: str, verbose: bool = False) -> dict:
if txid == self._exploding:
raise RuntimeError("missing transaction")
return {"confirmations": self._known.get(txid, 0)}
async def test_one_unresolvable_txid_does_not_block_the_others(session_factory):
"""B-03: the lookup used to be unguarded, so a single unknown txid aborted the
whole pass — nothing confirmed again until an operator intervened, which in turn
meant no round could ever close."""
async with session_factory() as session:
session.add(Round(id=10, status="open"))
session.add(
RoundParticipant(round_id=10, user_id=1, bet_amount_sats=1_000, bet_txid="good", status="broadcast")
)
session.add(
PendingTransaction(
kind="bet", round_id=10, user_id=2, current_txid="gone", fee_rate_sat_vb=1, raw_tx_hex="00",
status="pending",
)
)
session.add(
PendingTransaction(
kind="bet", round_id=10, user_id=1, current_txid="good", fee_rate_sat_vb=1, raw_tx_hex="00",
status="pending",
)
)
await session.commit()
confirmed = await poll_once(session_factory, ExplodingClient({"good": 1}, exploding_txid="gone"))
assert confirmed == 1 # the healthy one still got processed
async with session_factory() as session:
participant = (await session.scalars(select(RoundParticipant).where(RoundParticipant.round_id == 10))).one()
assert participant.status == "confirmed"
rows = {p.current_txid: p.status for p in (await session.scalars(select(PendingTransaction))).all()}
assert rows["good"] == "confirmed"
assert rows["gone"] == "pending" # left for the reconciler to judge, not abandoned here
async def test_bet_confirms_after_an_rbf_bump_changed_the_txid(session_factory):
"""B-02: the handler used to match on bet_txid, so a bumped bet confirmed under
a txid no participant carried — the participant stayed "broadcast" forever and
the round could never close. It now resolves by (round_id, user_id)."""
async with session_factory() as session:
session.add(Round(id=11, status="open"))
session.add(
RoundParticipant(
round_id=11, user_id=7, bet_amount_sats=1_000, bet_txid="old-txid", status="broadcast"
)
)
session.add(
PendingTransaction(
kind="bet", round_id=11, user_id=7, current_txid="bumped-txid", fee_rate_sat_vb=2,
raw_tx_hex="00", status="pending", replaced_by_txid="old-txid",
)
)
await session.commit()
assert await poll_once(session_factory, FakeClient({"bumped-txid": 1})) == 1
async with session_factory() as session:
participant = (await session.scalars(select(RoundParticipant).where(RoundParticipant.round_id == 11))).one()
assert participant.status == "confirmed"
async def test_payout_confirms_after_an_rbf_bump_changed_the_txid(session_factory):
async with session_factory() as session:
session.add(Round(id=12, status="paying_out", payout_txid="old-payout"))
session.add(
PendingTransaction(
kind="payout", round_id=12, current_txid="bumped-payout", fee_rate_sat_vb=2,
raw_tx_hex="00", status="pending",
)
)
await session.commit()
assert await poll_once(session_factory, FakeClient({"bumped-payout": 1})) == 1
async with session_factory() as session:
assert (await session.get(Round, 12)).status == "closed"
+233
View File
@@ -0,0 +1,233 @@
"""Regression tests for B-04 (and the "building" half of B-08): a transaction that
never made it onto the chain must give the coins back instead of freezing them."""
import pytest
from embit import script
from embit.transaction import Transaction, TransactionInput, TransactionOutput
from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from app.db.base import Base
from app.db.models import AuditLog, PendingTransaction, RoundParticipant, User, UtxoEvent, Withdrawal
from app.tx.reconcile import reconcile_once
class UnknownTxClient:
"""A server that doesn't know any of the txids it's asked about."""
async def get_transaction(self, txid: str, verbose: bool = False):
raise RuntimeError(f"missing transaction {txid}")
class KnownTxClient:
async def get_transaction(self, txid: str, verbose: bool = False):
return {"txid": txid, "confirmations": 0}
class BrokenClient:
"""A transport failure — says nothing about whether the tx exists."""
async def get_transaction(self, txid: str, verbose: bool = False):
raise ConnectionResetError("connection reset")
@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()
# A real (unsigned) transaction spending one input, built rather than hand-written
# so it round-trips through Transaction.parse — that parse is how the reconciler
# discovers which UTXOs to release, so a fixture the parser rejects would test
# nothing.
_TX_INPUT_TXID = "11" * 32
_RAW_TX = (
Transaction(
vin=[TransactionInput(bytes.fromhex(_TX_INPUT_TXID), 0)],
vout=[
TransactionOutput(
999_000_000, script.Script.from_address("plm1q0xcqpzrky6eff2g52qdye53xkk9jxkvraxkkwd")
)
],
)
.serialize()
.hex()
)
async def _seed_bet(session_factory, *, pending_status: str, participant_status: str, age_seconds: int):
from datetime import datetime, timedelta, timezone
async with session_factory() as session:
user = User(username="u", password_hash="x", derivation_index=0, address="plm1qtest")
session.add(user)
await session.flush()
session.add(
UtxoEvent(
user_id=user.id,
txid=_TX_INPUT_TXID,
vout=0,
amount_sats=1_000_000_000,
confirmed_height=10,
spent_txid="betxid",
)
)
session.add(
RoundParticipant(
round_id=1,
user_id=user.id,
bet_amount_sats=999_000_000,
bet_txid="betxid",
status=participant_status,
)
)
session.add(
PendingTransaction(
kind="bet",
round_id=1,
user_id=user.id,
current_txid="betxid",
fee_rate_sat_vb=1,
raw_tx_hex=_RAW_TX,
status=pending_status,
broadcast_at=datetime.now(timezone.utc) - timedelta(seconds=age_seconds),
)
)
await session.commit()
return user.id
async def test_abandons_a_building_bet_and_gives_the_coins_back(session_factory):
"""The crash-mid-broadcast case: the tx isn't on the chain, so the UTXO must be
released, the participant removed (they never entered the round) and the balance
restored. Before this existed, spent_txid stayed set forever and the user simply
lost the coins."""
user_id = await _seed_bet(
session_factory, pending_status="building", participant_status="building", age_seconds=300
)
resolved = await reconcile_once(session_factory, UnknownTxClient())
assert resolved == 1
async with session_factory() as session:
utxo = (await session.scalars(select(UtxoEvent))).one()
assert utxo.spent_txid is None # spendable again
assert (await session.scalars(select(RoundParticipant))).all() == []
row = (await session.scalars(select(PendingTransaction))).one()
assert row.status == "failed"
assert row.failure_reason
user = await session.get(User, user_id)
assert user.cached_balance_sats == 1_000_000_000
events = [e.event_type for e in (await session.scalars(select(AuditLog))).all()]
assert "pending_tx_abandoned" in events
async def test_promotes_a_building_row_whose_tx_did_reach_the_chain(session_factory):
"""We died after the broadcast, not before: the tx is real, so the rows must be
finished rather than rolled back."""
await _seed_bet(
session_factory, pending_status="building", participant_status="building", age_seconds=300
)
resolved = await reconcile_once(session_factory, KnownTxClient())
assert resolved == 1
async with session_factory() as session:
row = (await session.scalars(select(PendingTransaction))).one()
assert row.status == "pending"
participant = (await session.scalars(select(RoundParticipant))).one()
assert participant.status == "broadcast"
utxo = (await session.scalars(select(UtxoEvent))).one()
assert utxo.spent_txid == "betxid" # still legitimately spent
async def test_leaves_a_young_building_row_alone(session_factory):
"""A row written seconds ago may just be a broadcast still in flight."""
await _seed_bet(
session_factory, pending_status="building", participant_status="building", age_seconds=5
)
assert await reconcile_once(session_factory, UnknownTxClient()) == 0
async with session_factory() as session:
assert (await session.scalars(select(PendingTransaction))).one().status == "building"
async def test_leaves_a_recently_broadcast_pending_row_alone(session_factory):
"""A broadcast tx gets a wide grace window — absence from one server's mempool
is not proof of death, and the RBF bumper should get its attempts first."""
await _seed_bet(
session_factory, pending_status="pending", participant_status="broadcast", age_seconds=3600
)
assert await reconcile_once(session_factory, UnknownTxClient()) == 0
async def test_transport_failure_never_abandons_anything(session_factory):
"""A dead connection says nothing about the transaction. Treating it as "gone"
would release coins for transactions that are perfectly alive."""
await _seed_bet(
session_factory, pending_status="building", participant_status="building", age_seconds=300
)
assert await reconcile_once(session_factory, BrokenClient()) == 0
async with session_factory() as session:
assert (await session.scalars(select(PendingTransaction))).one().status == "building"
assert (await session.scalars(select(UtxoEvent))).one().spent_txid == "betxid"
async def test_abandoned_withdrawal_is_marked_failed_and_kept(session_factory):
"""Unlike a bet, a withdrawal is an instruction the user gave: the row stays so
they can see it didn't go through."""
from datetime import datetime, timedelta, timezone
async with session_factory() as session:
user = User(username="w", password_hash="x", derivation_index=1, address="plm1qtest2")
session.add(user)
await session.flush()
session.add(
UtxoEvent(
user_id=user.id,
txid=_TX_INPUT_TXID,
vout=0,
amount_sats=500_000_000,
confirmed_height=10,
spent_txid="wdtxid",
)
)
withdrawal = Withdrawal(
user_id=user.id,
external_address="plm1qexternal",
amount_requested_sats=400_000_000,
amount_sent_sats=399_000_000,
txid="wdtxid",
status="broadcast",
)
session.add(withdrawal)
await session.flush()
session.add(
PendingTransaction(
kind="withdrawal",
withdrawal_id=withdrawal.id,
user_id=user.id,
current_txid="wdtxid",
fee_rate_sat_vb=1,
raw_tx_hex=_RAW_TX,
status="pending",
broadcast_at=datetime.now(timezone.utc) - timedelta(days=1),
)
)
await session.commit()
assert await reconcile_once(session_factory, UnknownTxClient()) == 1
async with session_factory() as session:
withdrawal = (await session.scalars(select(Withdrawal))).one()
assert withdrawal.status == "failed"
assert withdrawal.txid is None
assert (await session.scalars(select(UtxoEvent))).one().spent_txid is None
+42 -1
View File
@@ -4,7 +4,7 @@ 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, UtxoEvent
from app.db.models import PendingTransaction, User, UtxoEvent, Withdrawal
from app.wallet.hd import derive_user_address
from app.withdrawals.service import WithdrawalError, request_withdrawal
@@ -122,3 +122,44 @@ async def test_withdrawal_rejects_non_plm_address(session_factory, address):
assert exc_info.value.code == "invalid_address"
assert not client.broadcasted
async def test_withdrawal_to_own_address_is_rejected(session_factory):
"""B-17: allowed before, and it broke two things that assume the recipient and
the change are distinguishable by address — the RBF bump would shrink the
recipient output, and compute_pending_balance counted the amount twice."""
user_id = await _make_funded_user(session_factory, 8, 3_000_000_000)
client = FakeElectrumClient()
async with session_factory() as session:
user = await session.get(User, user_id)
with pytest.raises(WithdrawalError, match="own deposit address"):
await request_withdrawal(session, client, user, user.address, 1_000_000_000)
assert not client.broadcasted
async with session_factory() as session:
assert (await session.scalars(select(UtxoEvent).where(UtxoEvent.user_id == user_id))).one().spent_txid is None
async def test_failed_broadcast_marks_the_withdrawal_failed_and_frees_the_coins(session_factory):
"""B-07/B-08: the Withdrawal row is kept (unlike a bet) so the user can see the
instruction didn't go through, but the coins must come back."""
user_id = await _make_funded_user(session_factory, 9, 3_000_000_000)
class RejectingClient:
async def broadcast(self, raw_tx_hex: str) -> str:
raise RuntimeError("min relay fee not met")
async with session_factory() as session:
user = await session.get(User, user_id)
external = derive_user_address(99)
with pytest.raises(WithdrawalError, match="refused"):
await request_withdrawal(session, RejectingClient(), user, external, 1_000_000_000)
async with session_factory() as session:
withdrawal = (await session.scalars(select(Withdrawal))).one()
assert withdrawal.status == "failed"
assert withdrawal.txid is None
assert (await session.scalars(select(UtxoEvent).where(UtxoEvent.user_id == user_id))).one().spent_txid is None
user = await session.get(User, user_id)
assert user.cached_balance_sats == 3_000_000_000