davideandClaude Opus 5
025754c860
The payout has to spend one pool UTXO per bet, so reusing MAX_TX_INPUTS (50) for it made any round past ~50 players unpayable: select_utxos raised too_many_inputs, the round stayed "paying_out" retrying every 60s forever, and since no new round may open while one is active, the whole lottery stopped with the pool stuck. The cap was being enforced on the payout side, i.e. discovered once the money was already committed and there was no way back. Two halves: - select_utxos takes the cap as a parameter. Bets and withdrawals keep MAX_TX_INPUTS = 50, which protects a user from a fee that eats into the amount they are moving; the payout uses MAX_PAYOUT_TX_INPUTS = 500, where that argument doesn't apply — 400 inputs at 1 sat/vB cost ~0.00027 PLM out of the winner's 70% share. What actually bounds it is relay policy: 500 inputs is ~34 kvB against the 100 kvB standardness limit, and signing that many measures ~0.4s, once per round, inside a background task. - place_bet refuses the 401st bet with a new round_full error (translated into all 7 languages), so "a round can always be paid out" is an invariant checked before any money moves. MAX_PARTICIPANTS_PER_ROUND sits below the input cap to leave the payout headroom for pool change from earlier rounds, and counts every participant row rather than only confirmed ones, since a failed bet frees a slot. A round already wedged past the old cap now pays out on the next retry tick. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
331 lines
14 KiB
Python
331 lines
14 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.bets.service import BetError, place_bet
|
|
from app.config import settings
|
|
from app.db.base import Base
|
|
from app.db.models import AuditLog, PendingTransaction, Round, RoundConfig, RoundParticipant, User, UtxoEvent
|
|
from app.rounds.events import broadcaster
|
|
from app.rounds.service import open_new_round_if_needed
|
|
from app.wallet.hd import derive_user_address
|
|
from app.wallet.psbt_builder import MAX_PARTICIPANTS_PER_ROUND, MAX_TX_INPUTS
|
|
|
|
|
|
class FakeElectrumClient:
|
|
def __init__(self):
|
|
self.broadcasted: list[str] = []
|
|
|
|
async def broadcast(self, raw_tx_hex: str) -> str:
|
|
self.broadcasted.append(raw_tx_hex)
|
|
return "fake-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 _make_funded_user(session_factory, index: int, funded_sats: int) -> int:
|
|
async with session_factory() as session:
|
|
address = derive_user_address(index)
|
|
user = User(username=f"user{index}", password_hash="x", derivation_index=index, address=address)
|
|
session.add(user)
|
|
await session.commit()
|
|
session.add(
|
|
UtxoEvent(
|
|
user_id=user.id,
|
|
txid=f"{index:02x}" * 32,
|
|
vout=0,
|
|
amount_sats=funded_sats,
|
|
confirmed_height=100,
|
|
)
|
|
)
|
|
await session.commit()
|
|
return user.id
|
|
|
|
|
|
async def test_place_bet_broadcasts_and_records_participant(session_factory):
|
|
user_id = await _make_funded_user(session_factory, 0, 1_500_000_000)
|
|
client = FakeElectrumClient()
|
|
|
|
async with session_factory() as session:
|
|
user = await session.get(User, user_id)
|
|
participant = await place_bet(session, client, user)
|
|
|
|
assert client.broadcasted # a raw tx was broadcast
|
|
assert participant.status == "broadcast"
|
|
assert participant.bet_txid
|
|
|
|
async with session_factory() as session:
|
|
utxo = (await session.scalars(select(UtxoEvent).where(UtxoEvent.user_id == user_id))).one()
|
|
assert utxo.spent_txid == participant.bet_txid
|
|
|
|
pending = (await session.scalars(select(PendingTransaction))).one()
|
|
assert pending.kind == "bet"
|
|
|
|
audit_events = (await session.scalars(select(AuditLog))).all()
|
|
assert any(e.event_type == "bet_placed" for e in audit_events)
|
|
assert pending.current_txid == participant.bet_txid
|
|
|
|
|
|
async def test_place_bet_rejects_insufficient_balance(session_factory):
|
|
user_id = await _make_funded_user(session_factory, 1, 1_000_000) # below bet_amount_sats
|
|
client = FakeElectrumClient()
|
|
|
|
async with session_factory() as session:
|
|
user = await session.get(User, user_id)
|
|
with pytest.raises(BetError, match="insufficient balance"):
|
|
await place_bet(session, client, user)
|
|
|
|
|
|
async def test_place_bet_reports_a_too_fragmented_balance_distinctly(session_factory): # B-48
|
|
# 100 x 0.15 PLM = 15 PLM, plenty for a 10 PLM bet, but the 50 largest inputs
|
|
# only add up to 7.5 PLM — so the build must fail with its own code, not with
|
|
# the "you have no funds" one, and must carry the cap for the translation.
|
|
user_id = await _make_funded_user(session_factory, 20, 15_000_000)
|
|
async with session_factory() as session:
|
|
for i in range(99):
|
|
session.add(
|
|
UtxoEvent(
|
|
user_id=user_id,
|
|
txid=f"{i:064x}",
|
|
vout=0,
|
|
amount_sats=15_000_000,
|
|
confirmed_height=100,
|
|
)
|
|
)
|
|
await session.commit()
|
|
client = FakeElectrumClient()
|
|
|
|
async with session_factory() as session:
|
|
user = await session.get(User, user_id)
|
|
with pytest.raises(BetError) as excinfo:
|
|
await place_bet(session, client, user)
|
|
|
|
assert excinfo.value.code == "too_many_inputs"
|
|
assert excinfo.value.params == {"max_inputs": MAX_TX_INPUTS}
|
|
assert not client.broadcasted
|
|
|
|
|
|
async def _fill_round_with_participants(session_factory, round_id: int, count: int) -> None:
|
|
"""Participant rows only, no real bets: what the cap counts is rows, and building
|
|
`count` genuine transactions would just make the test slow without exercising
|
|
anything the other tests don't already cover."""
|
|
async with session_factory() as session:
|
|
for i in range(count):
|
|
session.add(
|
|
RoundParticipant(
|
|
round_id=round_id,
|
|
user_id=10_000 + i, # placeholder ids; the cap check never joins users
|
|
bet_amount_sats=1_000_000_000,
|
|
bet_txid=f"{i:064x}",
|
|
status="confirmed",
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
|
|
async def test_place_bet_rejects_the_bet_past_the_participant_cap(session_factory): # B-52
|
|
"""The payout has to spend one pool UTXO per bet, so a round is only ever allowed
|
|
to grow to what a single payout transaction can drain. Enforced here, before the
|
|
player's money moves — not discovered at payout time, when the bets are already in
|
|
the pool and the round can no longer be paid at all."""
|
|
user_id = await _make_funded_user(session_factory, 30, 3_000_000_000)
|
|
client = FakeElectrumClient()
|
|
|
|
async with session_factory() as session:
|
|
round_ = await open_new_round_if_needed(session)
|
|
await session.commit()
|
|
round_id = round_.id
|
|
await _fill_round_with_participants(session_factory, round_id, MAX_PARTICIPANTS_PER_ROUND)
|
|
|
|
async with session_factory() as session:
|
|
user = await session.get(User, user_id)
|
|
with pytest.raises(BetError) as excinfo:
|
|
await place_bet(session, client, user)
|
|
|
|
assert excinfo.value.code == "round_full"
|
|
assert excinfo.value.params == {"max_participants": MAX_PARTICIPANTS_PER_ROUND}
|
|
assert not client.broadcasted
|
|
|
|
# Refused cleanly: no participant row, and the user's UTXO is still spendable.
|
|
async with session_factory() as session:
|
|
assert await session.scalar(
|
|
select(func.count()).select_from(RoundParticipant).where(RoundParticipant.round_id == round_id)
|
|
) == MAX_PARTICIPANTS_PER_ROUND
|
|
utxo = (await session.scalars(select(UtxoEvent).where(UtxoEvent.user_id == user_id))).one()
|
|
assert utxo.spent_txid is None
|
|
|
|
|
|
async def test_place_bet_still_accepts_the_last_slot_under_the_cap(session_factory): # B-52
|
|
user_id = await _make_funded_user(session_factory, 31, 3_000_000_000)
|
|
client = FakeElectrumClient()
|
|
|
|
async with session_factory() as session:
|
|
round_ = await open_new_round_if_needed(session)
|
|
await session.commit()
|
|
round_id = round_.id
|
|
await _fill_round_with_participants(session_factory, round_id, MAX_PARTICIPANTS_PER_ROUND - 1)
|
|
|
|
async with session_factory() as session:
|
|
user = await session.get(User, user_id)
|
|
participant = await place_bet(session, client, user)
|
|
|
|
assert participant.status == "broadcast"
|
|
assert client.broadcasted
|
|
|
|
|
|
async def test_place_bet_rejects_second_bet_same_round(session_factory):
|
|
user_id = await _make_funded_user(session_factory, 2, 3_000_000_000)
|
|
client = FakeElectrumClient()
|
|
|
|
async with session_factory() as session:
|
|
user = await session.get(User, user_id)
|
|
await place_bet(session, client, user)
|
|
|
|
async with session_factory() as session:
|
|
user = await session.get(User, user_id)
|
|
with pytest.raises(BetError, match="already"):
|
|
await place_bet(session, client, user)
|
|
|
|
async with session_factory() as session:
|
|
participants = (await session.scalars(select(RoundParticipant))).all()
|
|
assert len(participants) == 1
|
|
|
|
|
|
async def test_place_bet_rejects_after_timer_expires_even_if_still_open(session_factory):
|
|
"""The scheduler only flips status "open" -> "closing" on its next tick (up
|
|
to a few seconds late) — place_bet must independently refuse bets once the
|
|
round's own deadline has passed, so no new player can sneak in during that
|
|
gap (see rounds/service.round_accepts_bets)."""
|
|
user_id = await _make_funded_user(session_factory, 3, 3_000_000_000)
|
|
client = FakeElectrumClient()
|
|
|
|
async with session_factory() as session:
|
|
session.add(RoundConfig(fee_address="", round_duration_seconds=60))
|
|
round_ = await open_new_round_if_needed(session)
|
|
round_.opened_at = datetime.now(timezone.utc) - timedelta(seconds=61)
|
|
await session.commit()
|
|
|
|
async with session_factory() as session:
|
|
user = await session.get(User, user_id)
|
|
with pytest.raises(BetError, match="closing"):
|
|
await place_bet(session, client, user)
|
|
|
|
async with session_factory() as session:
|
|
participants = (await session.scalars(select(RoundParticipant))).all()
|
|
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_publishes_an_sse_update(session_factory): # B-49
|
|
"""The rollback moves as much state as the successful path does, so it must ping
|
|
the dashboards the same way — otherwise the phantom bet stays on screen until the
|
|
next poll."""
|
|
user_id = await _make_funded_user(session_factory, 21, 3_000_000_000)
|
|
async with session_factory() as session:
|
|
# Open the round up front: place_bet would otherwise open it itself, and that
|
|
# publish() would satisfy the assertion below whether or not the rollback ever
|
|
# published one of its own.
|
|
await open_new_round_if_needed(session)
|
|
await session.commit()
|
|
|
|
queue = broadcaster.subscribe()
|
|
try:
|
|
while not queue.empty():
|
|
queue.get_nowait()
|
|
|
|
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)
|
|
|
|
assert not queue.empty()
|
|
finally:
|
|
broadcaster.unsubscribe(queue)
|
|
|
|
|
|
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"]
|