davideandClaude Sonnet 5.5
54dbffb552
A user could bet and then double-spend the 1-conference deposit that funded the bet with a private chain: the bet left the chain but its participant stayed `confirmed`, and the payout failed "insufficient pool UTXOs" for ever with everyone else's money stuck. Coinbase outputs were also credited at 1 confirmation (maturity is 120). Deposits - `round_config.deposit_confirmations` (default 6, 1-100, editable from /admin) is the threshold for a deposit; an output of a platform transaction (any txid of a pending_transactions row, RBF versions included) keeps 1; a coinbase output needs 120, recognised from the raw transaction, which must hash to the txid (`is_coinbase`, cached). `read_deposit_policy` / `classify_new_deposits` run between the read and the write, so no session is open across the lookup. - The quorum must also place the coin deep enough on our own tip (ELC-17). - Confirmed outputs short of their threshold are pending deposits (memory only, replaced at every refresh): shown in `pending_balance_sats` / `has_pending` (also for DEP-13's withdrawal message), never spendable. No scripthash notification announces a maturing deposit, so each new block refreshes only the users that have one, one task per user (DEP-09 now says so). Payout - `_trigger_payout` verifies the round's confirmed participants' bet outputs with the other servers (`verify_pool_outpoints`: strict majority of >= 2 answers per outpoint, a mempool-only coin or a tie gives no verdict) and builds from exactly those outpoints (RND-22), never from whatever else sits at the address. No quorum: `payout_failed`, `pool verification unavailable`, nothing changes. Bets that are gone become `reorged_out`, one `participant_reorged_out` row each, `pool_amount_sats` rewritten (only downward) in the same transaction, and the 70/30 split is computed on what is left. - Winner's own bet gone: nothing is built; `rounds.payout_blocked_at` (Alembic f2c64a8d1e93, with `deposit_confirmations`, plain add_column) is set, `payout_blocked` and the pause with `paused_reason = 'payout_blocked'` are written in one transaction; `POST /admin/resume` clears the mark and the next attempt verifies again. /admin/rounds returns it. - Rounds opened before B-80 (pool = the shared reserve address) keep the old active-server `listunspent` choice, since that address holds no single round's bets. Ambiguities resolved: the winner's row stays `confirmed` while held (AUD-45 says every participant whose bet is gone becomes `reorged_out`, but OPS-30/DAT-68 need the verification to repeat after the resume, which only a `confirmed` row allows); `/`'s deposit hint no longer says "1 confirmation" (7 languages, same keys); guides updated. Tests changed because the spec changed: listener tests that credited height-100 coins at tip 0 now set a tip past every threshold; the payout fixtures seed confirmed participants (the new source of the inputs) instead of an active-server `listunspent`, FakeNetwork answers the verification, and test_round_pool_address asserts the quorum is asked about the round's address. New: thresholds at exactly 6/5, configured value, coinbase at 119/120 and a mismatching or missing raw tx, platform outputs at 1, dust, quorum height, waiting-deposit display and disappearance, block-triggered refresh, resume, verification verdicts, gone bets excluded with the split recomputed, winner gone held and paused, held payouts untouched until the resume, re-verification after it, no quorum, a payout with 1 and with 400 participants against reserve+fee+dust, admin config bounds and migration. Mutation-checked: threshold 1, ignored coinbase, no height bound, no verification and no winner hold each fail tests. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
893 lines
37 KiB
Python
893 lines
37 KiB
Python
"""B-88 (DAT-52, DAT-74, ARC-07, TX-28, BET-11, DAT-16): SQLite transactions are the
|
|
ones the code assumes.
|
|
|
|
Everything here runs on a real file database whose engine went through
|
|
`_register_sqlite_pragmas` — an in-memory engine without it would not exercise
|
|
`BEGIN`/`BEGIN IMMEDIATE`, `synchronous` or the lock at all — with the lock wait
|
|
shortened to 100 ms so a busy database is provoked by simply holding the write
|
|
lock from a second session.
|
|
"""
|
|
|
|
from contextlib import asynccontextmanager
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import Response
|
|
from cryptography.fernet import Fernet
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy import func, select, text
|
|
from sqlalchemy.exc import OperationalError
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.bets.service import BetError, place_bet
|
|
from app.config import settings
|
|
from tests.unit.fake_network import FakeNetwork, rpc_error
|
|
from app.db import base as db_base
|
|
from app.db.base import Base, begin_immediate
|
|
from app.db.models import (
|
|
AuditLog,
|
|
PendingTransaction,
|
|
RoundConfig,
|
|
RoundParticipant,
|
|
User,
|
|
UtxoEvent,
|
|
Withdrawal,
|
|
)
|
|
from app.deposits.service import (
|
|
credit_confirmed_utxos,
|
|
mark_utxos_spent_externally,
|
|
reinstate_reappeared_utxos,
|
|
)
|
|
from app.wallet.hd import derive_user_address
|
|
from app.withdrawals.service import WithdrawalError, request_withdrawal
|
|
|
|
_FEE_ADDRESS = "plm1q5x25wd6q463mfhckjraaedgjg0lyu73qfcj43n"
|
|
_EXTERNAL_ADDRESS = "plm1qqph9qup2mp7w7g5nlsdhdc9m2pp44ampzw0ctx"
|
|
|
|
|
|
@pytest.fixture
|
|
async def engine(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(db_base, "_SQLITE_BUSY_TIMEOUT_MS", 100)
|
|
eng = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/b88.db")
|
|
db_base._register_sqlite_pragmas(eng)
|
|
async with eng.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield eng
|
|
await eng.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
def session_factory(engine):
|
|
return async_sessionmaker(engine, expire_on_commit=False)
|
|
|
|
|
|
@pytest.fixture
|
|
def keyed(tmp_path, monkeypatch):
|
|
"""A real master key, so the HD wallet can derive and sign."""
|
|
monkeypatch.setattr(settings, "master_key_path", str(tmp_path / "master.xprv.enc"))
|
|
monkeypatch.setattr(settings, "xprv_encryption_key", Fernet.generate_key().decode())
|
|
from app.wallet import hd
|
|
|
|
hd._account_key = None
|
|
hd.generate_master_key()
|
|
yield
|
|
hd._account_key = None
|
|
|
|
|
|
@pytest.fixture
|
|
async def funded(session_factory, keyed):
|
|
"""Fee address + one user holding 15 PLM in a single UTXO."""
|
|
async with session_factory() as session:
|
|
session.add(RoundConfig(fee_address=_FEE_ADDRESS))
|
|
user = User(username="u0", password_hash="x", derivation_index=0, address=derive_user_address(0))
|
|
session.add(user)
|
|
await session.commit()
|
|
session.add(UtxoEvent(user_id=user.id, txid="00" * 32, vout=0, amount_sats=1_500_000_000, confirmed_height=100))
|
|
await session.commit()
|
|
user_id = user.id
|
|
yield user_id
|
|
|
|
|
|
@asynccontextmanager
|
|
async def _holding_write_lock(session_factory):
|
|
"""Another writer holding SQLite's write lock: every begin_immediate elsewhere
|
|
times out after the (shortened) busy_timeout."""
|
|
async with session_factory() as holder:
|
|
await begin_immediate(holder)
|
|
try:
|
|
yield
|
|
finally:
|
|
await holder.rollback()
|
|
|
|
|
|
class _Client:
|
|
"""Fake Electrum client. `on_broadcast` runs first (to grab the lock *after* the
|
|
coins have gone out); `fail` makes the broadcast raise."""
|
|
|
|
def __init__(self, on_broadcast=None, fail=False):
|
|
self.broadcasted: list[str] = []
|
|
self._on_broadcast = on_broadcast
|
|
self._fail = fail
|
|
|
|
async def broadcast(self, raw_tx_hex: str) -> str:
|
|
self.broadcasted.append(raw_tx_hex)
|
|
if self._on_broadcast is not None:
|
|
await self._on_broadcast()
|
|
if self._fail:
|
|
raise rpc_error("mempool conflict")
|
|
return "txid"
|
|
|
|
|
|
# --- the engine setup: ARC-07 / DAT-52 ---------------------------------------------
|
|
|
|
|
|
async def test_every_connection_is_full_sync_wal_with_no_driver_transactions(engine):
|
|
"""DAT-52: synchronous=FULL, WAL, busy_timeout — and the driver's implicit BEGIN
|
|
switched off (isolation_level None), so the transaction is the one we emit."""
|
|
async with engine.connect() as conn:
|
|
assert (await conn.exec_driver_sql("PRAGMA synchronous")).scalar() == 2 # FULL
|
|
assert (await conn.exec_driver_sql("PRAGMA journal_mode")).scalar().lower() == "wal"
|
|
assert (await conn.exec_driver_sql("PRAGMA busy_timeout")).scalar() == 100
|
|
raw = await conn.get_raw_connection()
|
|
assert raw.driver_connection.isolation_level is None
|
|
|
|
|
|
async def test_a_read_starts_a_real_transaction_with_a_stable_snapshot(session_factory):
|
|
"""DAT-74: with the legacy driver mode a SELECT ran outside any transaction and
|
|
saw whatever had been committed since the previous statement; now the first
|
|
statement opens BEGIN and the reads that follow share one snapshot."""
|
|
async with session_factory() as setup:
|
|
setup.add(User(username="a", password_hash="x", derivation_index=0, address="a"))
|
|
await setup.commit()
|
|
|
|
async with session_factory() as reader, session_factory() as writer:
|
|
assert await reader.scalar(select(func.count()).select_from(User)) == 1
|
|
raw = await (await reader.connection()).get_raw_connection()
|
|
assert raw.driver_connection.in_transaction
|
|
|
|
writer.add(User(username="b", password_hash="x", derivation_index=1, address="b"))
|
|
await writer.commit()
|
|
|
|
assert await reader.scalar(select(func.count()).select_from(User)) == 1 # snapshot held
|
|
await reader.commit()
|
|
assert await reader.scalar(select(func.count()).select_from(User)) == 2
|
|
|
|
|
|
async def test_commit_and_rollback_still_work_without_the_drivers_transactions(session_factory):
|
|
async with session_factory() as session:
|
|
session.add(User(username="kept", password_hash="x", derivation_index=0, address="k"))
|
|
await session.commit()
|
|
session.add(User(username="gone", password_hash="x", derivation_index=1, address="g"))
|
|
await session.flush()
|
|
await session.rollback()
|
|
async with session_factory() as session:
|
|
names = (await session.scalars(select(User.username))).all()
|
|
assert names == ["kept"]
|
|
|
|
|
|
# --- begin_immediate: DAT-74 ------------------------------------------------------
|
|
|
|
|
|
async def test_begin_immediate_takes_the_write_lock_before_any_read(session_factory):
|
|
async with session_factory() as first, session_factory() as second:
|
|
await begin_immediate(first)
|
|
with pytest.raises(OperationalError, match="locked"):
|
|
await begin_immediate(second)
|
|
await first.rollback()
|
|
await begin_immediate(second) # free again once the holder is done
|
|
|
|
|
|
async def test_a_plain_reader_is_not_blocked_by_the_lock_holder(session_factory):
|
|
"""WAL: only writers queue; BEGIN IMMEDIATE must not turn every GET into one."""
|
|
async with session_factory() as holder, session_factory() as reader:
|
|
await begin_immediate(holder)
|
|
assert await reader.scalar(select(func.count()).select_from(User)) == 0
|
|
|
|
|
|
async def test_begin_immediate_closes_the_readonly_transaction_that_preceded_it(session_factory):
|
|
"""The callers arrive with the authentication lookup already in a transaction."""
|
|
async with session_factory() as session:
|
|
await session.scalar(select(func.count()).select_from(User))
|
|
assert session.in_transaction()
|
|
await begin_immediate(session)
|
|
assert session.in_transaction()
|
|
raw = await (await session.connection()).get_raw_connection()
|
|
assert raw.driver_connection.in_transaction
|
|
# and it really holds the write lock
|
|
async with session_factory() as other:
|
|
with pytest.raises(OperationalError):
|
|
await begin_immediate(other)
|
|
|
|
|
|
async def test_begin_immediate_keeps_loaded_objects_usable(session_factory, funded):
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
await begin_immediate(session)
|
|
assert user.username == "u0" # no lazy load needed after the commit it did
|
|
|
|
|
|
async def test_begin_immediate_refuses_a_session_with_unflushed_changes(session_factory):
|
|
async with session_factory() as session:
|
|
session.add(User(username="x", password_hash="x", derivation_index=0, address="x"))
|
|
with pytest.raises(RuntimeError, match="unflushed"):
|
|
await begin_immediate(session)
|
|
session.expunge_all()
|
|
|
|
async with session_factory() as session:
|
|
user = User(username="y", password_hash="x", derivation_index=1, address="y")
|
|
session.add(user)
|
|
await session.commit()
|
|
user.username = "z" # dirty
|
|
with pytest.raises(RuntimeError, match="unflushed"):
|
|
await begin_immediate(session)
|
|
await session.rollback()
|
|
|
|
|
|
async def test_begin_immediate_is_harmless_on_an_engine_without_the_sqlite_setup():
|
|
"""The service tests run on plain in-memory engines: the option is never read."""
|
|
eng = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
async with eng.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
async with async_sessionmaker(eng, expire_on_commit=False)() as session:
|
|
await begin_immediate(session)
|
|
session.add(User(username="a", password_hash="x", derivation_index=0, address="a"))
|
|
await session.commit()
|
|
await eng.dispose()
|
|
|
|
|
|
# --- place_bet: BET-11, DAT-16, TX-28 ---------------------------------------------
|
|
|
|
|
|
async def _bet_state(session_factory, user_id):
|
|
async with session_factory() as session:
|
|
participants = (await session.scalars(select(RoundParticipant))).all()
|
|
pendings = (await session.scalars(select(PendingTransaction))).all()
|
|
utxo = await session.scalar(select(UtxoEvent).where(UtxoEvent.user_id == user_id))
|
|
return participants, pendings, utxo
|
|
|
|
|
|
async def test_place_bet_on_a_real_database_commits_and_promotes(session_factory, funded):
|
|
client = _Client()
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
participant = await place_bet(session, client, user)
|
|
assert participant.status == "broadcast"
|
|
participants, pendings, utxo = await _bet_state(session_factory, funded)
|
|
assert [p.status for p in participants] == ["broadcast"]
|
|
assert [p.status for p in pendings] == ["pending"]
|
|
assert utxo.spent_txid == participants[0].bet_txid
|
|
assert len(client.broadcasted) == 1
|
|
|
|
|
|
async def test_place_bet_answers_service_busy_when_the_lock_is_not_granted(session_factory, funded):
|
|
"""BET-11 / api.md `service_busy`: refused before anything is built or sent."""
|
|
client = _Client()
|
|
async with _holding_write_lock(session_factory):
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
with pytest.raises(BetError) as excinfo:
|
|
await place_bet(session, client, user)
|
|
assert excinfo.value.code == "service_busy"
|
|
assert client.broadcasted == []
|
|
participants, pendings, utxo = await _bet_state(session_factory, funded)
|
|
assert participants == [] and pendings == []
|
|
assert utxo.spent_txid is None
|
|
|
|
|
|
async def test_place_bet_can_be_retried_after_a_busy_refusal(session_factory, funded):
|
|
client = _Client()
|
|
async with _holding_write_lock(session_factory):
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
with pytest.raises(BetError):
|
|
await place_bet(session, client, user)
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
participant = await place_bet(session, client, user)
|
|
assert participant.status == "broadcast"
|
|
|
|
|
|
async def test_place_bet_maps_a_lock_timeout_at_the_phase_one_commit_to_service_busy_not_round_closing(
|
|
session_factory, funded
|
|
):
|
|
"""DAT-16: an OperationalError at the compare-and-set / commit is the lock wait
|
|
expiring, never a snapshot conflict (SQLite has none) — so not `round_closing`.
|
|
Rolled back, nothing sent."""
|
|
client = _Client()
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
real_commit = session.commit
|
|
calls = {"n": 0}
|
|
|
|
async def flaky_commit():
|
|
calls["n"] += 1
|
|
if calls["n"] == 1: # the phase-1 commit (begin_immediate's own commit is a no-op here)
|
|
raise OperationalError("COMMIT", {}, Exception("database is locked"))
|
|
await real_commit()
|
|
|
|
session.commit = flaky_commit
|
|
with pytest.raises(BetError) as excinfo:
|
|
await place_bet(session, client, user)
|
|
assert excinfo.value.code == "service_busy"
|
|
assert client.broadcasted == []
|
|
participants, pendings, utxo = await _bet_state(session_factory, funded)
|
|
assert participants == [] and pendings == [] and utxo.spent_txid is None
|
|
|
|
|
|
async def test_place_bet_after_a_broadcast_never_undoes_it_when_the_lock_is_lost(session_factory, funded):
|
|
"""TX-28 / DAT-74: the tx is on the network, so a busy database afterwards leaves
|
|
the rows `building` for the reconciler instead of failing or releasing anything;
|
|
the caller still gets the bet, with the status the database holds."""
|
|
holder_cm = _holding_write_lock(session_factory)
|
|
|
|
async def grab_lock():
|
|
await holder_cm.__aenter__()
|
|
|
|
client = _Client(on_broadcast=grab_lock)
|
|
try:
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
participant = await place_bet(session, client, user)
|
|
finally:
|
|
await holder_cm.__aexit__(None, None, None)
|
|
|
|
assert participant.status == "building"
|
|
assert len(client.broadcasted) == 1
|
|
participants, pendings, utxo = await _bet_state(session_factory, funded)
|
|
assert [p.status for p in participants] == ["building"]
|
|
assert [p.status for p in pendings] == ["building"]
|
|
assert utxo.spent_txid == participants[0].bet_txid # still reserved
|
|
async with session_factory() as session:
|
|
events = (await session.scalars(select(AuditLog.event_type))).all()
|
|
assert "bet_placed" not in events and "bet_broadcast_failed" not in events
|
|
|
|
|
|
async def test_place_bet_refused_broadcast_with_a_busy_database_leaves_the_rows_building(session_factory, funded):
|
|
"""A refused broadcast whose release cannot be written stays `building` (the
|
|
reconciler decides), and the caller still gets `broadcast_failed`."""
|
|
holder_cm = _holding_write_lock(session_factory)
|
|
|
|
async def grab_lock():
|
|
await holder_cm.__aenter__()
|
|
|
|
client = _Client(on_broadcast=grab_lock, fail=True)
|
|
try:
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
with pytest.raises(BetError) as excinfo:
|
|
await place_bet(session, client, user, network=FakeNetwork(known=False))
|
|
finally:
|
|
await holder_cm.__aexit__(None, None, None)
|
|
|
|
assert excinfo.value.code == "broadcast_failed"
|
|
participants, pendings, utxo = await _bet_state(session_factory, funded)
|
|
assert [p.status for p in participants] == ["building"]
|
|
assert [p.status for p in pendings] == ["building"]
|
|
assert utxo.spent_txid == participants[0].bet_txid
|
|
|
|
|
|
async def test_place_bet_refused_broadcast_still_releases_when_the_database_is_free(session_factory, funded):
|
|
client = _Client(fail=True)
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
with pytest.raises(BetError) as excinfo:
|
|
await place_bet(session, client, user, network=FakeNetwork(known=False))
|
|
assert excinfo.value.code == "broadcast_failed"
|
|
participants, pendings, utxo = await _bet_state(session_factory, funded)
|
|
assert participants == [] and pendings == [] and utxo.spent_txid is None
|
|
|
|
|
|
async def test_concurrent_bets_of_one_user_cannot_both_select_the_same_utxo(session_factory, funded):
|
|
"""DAT-74: with the lock taken before the reads, two place_bet calls for the same
|
|
user (which the per-user lock normally forbids) cannot both read the UTXO as free."""
|
|
import asyncio
|
|
|
|
client = _Client()
|
|
|
|
async def attempt():
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
return await place_bet(session, client, user)
|
|
|
|
results = await asyncio.gather(attempt(), attempt(), return_exceptions=True)
|
|
ok = [r for r in results if not isinstance(r, Exception)]
|
|
assert len(ok) == 1
|
|
failures = [r for r in results if isinstance(r, Exception)]
|
|
assert len(failures) == 1 and isinstance(failures[0], BetError)
|
|
assert len(client.broadcasted) == 1
|
|
|
|
|
|
# --- request_withdrawal ----------------------------------------------------------
|
|
|
|
|
|
async def _withdrawal_state(session_factory, user_id):
|
|
async with session_factory() as session:
|
|
return (
|
|
(await session.scalars(select(Withdrawal))).all(),
|
|
(await session.scalars(select(PendingTransaction))).all(),
|
|
await session.scalar(select(UtxoEvent).where(UtxoEvent.user_id == user_id)),
|
|
)
|
|
|
|
|
|
async def test_withdrawal_on_a_real_database_commits_and_promotes(session_factory, funded):
|
|
client = _Client()
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
withdrawal = await request_withdrawal(session, client, user, _EXTERNAL_ADDRESS, 1_000_000_000)
|
|
assert withdrawal.status == "broadcast"
|
|
withdrawals, pendings, utxo = await _withdrawal_state(session_factory, funded)
|
|
assert [w.status for w in withdrawals] == ["broadcast"]
|
|
assert [p.status for p in pendings] == ["pending"]
|
|
assert utxo.spent_txid == withdrawals[0].txid
|
|
|
|
|
|
async def test_withdrawal_answers_service_busy_when_the_lock_is_not_granted(session_factory, funded):
|
|
client = _Client()
|
|
async with _holding_write_lock(session_factory):
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
with pytest.raises(WithdrawalError) as excinfo:
|
|
await request_withdrawal(session, client, user, _EXTERNAL_ADDRESS, 1_000_000_000)
|
|
assert excinfo.value.code == "service_busy"
|
|
assert client.broadcasted == []
|
|
withdrawals, pendings, utxo = await _withdrawal_state(session_factory, funded)
|
|
assert withdrawals == [] and pendings == [] and utxo.spent_txid is None
|
|
|
|
|
|
async def test_withdrawal_validation_errors_do_not_wait_for_the_lock(session_factory, funded):
|
|
"""The checks that need no database answer immediately, lock or not."""
|
|
client = _Client()
|
|
async with _holding_write_lock(session_factory):
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
with pytest.raises(WithdrawalError) as excinfo:
|
|
await request_withdrawal(session, client, user, "not-an-address", 1_000_000_000)
|
|
assert excinfo.value.code == "invalid_address"
|
|
|
|
|
|
async def test_withdrawal_after_a_broadcast_never_undoes_it_when_the_lock_is_lost(session_factory, funded):
|
|
holder_cm = _holding_write_lock(session_factory)
|
|
|
|
async def grab_lock():
|
|
await holder_cm.__aenter__()
|
|
|
|
client = _Client(on_broadcast=grab_lock)
|
|
try:
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
withdrawal = await request_withdrawal(session, client, user, _EXTERNAL_ADDRESS, 1_000_000_000)
|
|
finally:
|
|
await holder_cm.__aexit__(None, None, None)
|
|
|
|
assert withdrawal.status == "building"
|
|
withdrawals, pendings, utxo = await _withdrawal_state(session_factory, funded)
|
|
assert [w.status for w in withdrawals] == ["building"]
|
|
assert [p.status for p in pendings] == ["building"]
|
|
assert utxo.spent_txid == withdrawals[0].txid
|
|
|
|
|
|
async def test_withdrawal_refused_broadcast_with_a_busy_database_leaves_the_rows_building(session_factory, funded):
|
|
holder_cm = _holding_write_lock(session_factory)
|
|
|
|
async def grab_lock():
|
|
await holder_cm.__aenter__()
|
|
|
|
client = _Client(on_broadcast=grab_lock, fail=True)
|
|
try:
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
with pytest.raises(WithdrawalError) as excinfo:
|
|
await request_withdrawal(
|
|
session, client, user, _EXTERNAL_ADDRESS, 1_000_000_000, network=FakeNetwork(known=False)
|
|
)
|
|
finally:
|
|
await holder_cm.__aexit__(None, None, None)
|
|
|
|
assert excinfo.value.code == "broadcast_failed"
|
|
withdrawals, pendings, utxo = await _withdrawal_state(session_factory, funded)
|
|
assert [w.status for w in withdrawals] == ["building"]
|
|
assert [p.status for p in pendings] == ["building"]
|
|
assert utxo.spent_txid == withdrawals[0].txid
|
|
|
|
|
|
async def test_withdrawal_refused_broadcast_still_releases_when_the_database_is_free(session_factory, funded):
|
|
client = _Client(fail=True)
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
with pytest.raises(WithdrawalError):
|
|
await request_withdrawal(
|
|
session, client, user, _EXTERNAL_ADDRESS, 1_000_000_000, network=FakeNetwork(known=False)
|
|
)
|
|
withdrawals, pendings, utxo = await _withdrawal_state(session_factory, funded)
|
|
assert [w.status for w in withdrawals] == ["failed"]
|
|
assert [p.status for p in pendings] == ["failed"]
|
|
assert utxo.spent_txid is None
|
|
|
|
|
|
# --- the HTTP mapping: api.md `service_busy` -> 503 ---------------------------------
|
|
|
|
|
|
def _fake_request(client):
|
|
listener = SimpleNamespace(client=client)
|
|
|
|
class _Locks:
|
|
@asynccontextmanager
|
|
async def acquire(self, user_id):
|
|
yield
|
|
|
|
return SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(electrum_listener=listener, user_locks=_Locks())))
|
|
|
|
|
|
async def test_bet_route_answers_503_for_service_busy(session_factory, funded):
|
|
from fastapi import HTTPException
|
|
|
|
from app.api.routes.bets import create_bet
|
|
|
|
async with _holding_write_lock(session_factory):
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
with pytest.raises(HTTPException) as excinfo:
|
|
await create_bet(_fake_request(_Client()), Response(), user=user, session=session)
|
|
assert excinfo.value.status_code == 503
|
|
assert excinfo.value.detail["code"] == "service_busy"
|
|
|
|
|
|
async def test_withdrawal_route_answers_503_for_service_busy(session_factory, funded):
|
|
from fastapi import HTTPException
|
|
|
|
from app.api.routes.withdrawals import WithdrawalRequest, create_withdrawal
|
|
|
|
body = WithdrawalRequest(external_address=_EXTERNAL_ADDRESS, amount_sats=1_000_000_000)
|
|
async with _holding_write_lock(session_factory):
|
|
async with session_factory() as session:
|
|
user = await session.get(User, funded)
|
|
with pytest.raises(HTTPException) as excinfo:
|
|
await create_withdrawal(body, _fake_request(_Client()), Response(), user=user, session=session)
|
|
assert excinfo.value.status_code == 503
|
|
assert excinfo.value.detail["code"] == "service_busy"
|
|
|
|
|
|
# --- registration ----------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
async def auth_client(session_factory, funded, monkeypatch):
|
|
from fastapi import FastAPI
|
|
|
|
from app.auth.routes import router
|
|
from app.db.session import get_session
|
|
|
|
monkeypatch.setattr(settings, "jwt_secret", "x" * 32)
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
|
|
async def _session():
|
|
async with session_factory() as session:
|
|
yield session
|
|
|
|
app.dependency_overrides[get_session] = _session
|
|
app.state.electrum_listener = SimpleNamespace(address_for_new_user=lambda *a, **k: None)
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://t") as ac:
|
|
yield ac
|
|
|
|
|
|
async def test_concurrent_registrations_get_distinct_derivation_indexes(auth_client, session_factory):
|
|
"""DAT-74: the max(derivation_index) read and the insert are one transaction under
|
|
the write lock, so two simultaneous signups cannot pick the same index."""
|
|
import asyncio
|
|
|
|
responses = await asyncio.gather(
|
|
*(auth_client.post("/auth/register", json={"username": f"racer{i}", "password": "long-enough-1"}) for i in range(4))
|
|
)
|
|
assert [r.status_code for r in responses] == [201] * 4
|
|
async with session_factory() as session:
|
|
indexes = (await session.scalars(select(User.derivation_index))).all()
|
|
assert len(indexes) == len(set(indexes)) == 5 # the funded fixture user plus four
|
|
|
|
|
|
async def test_registration_answers_503_service_busy_when_the_lock_is_not_granted(auth_client, session_factory):
|
|
async with _holding_write_lock(session_factory):
|
|
resp = await auth_client.post("/auth/register", json={"username": "late", "password": "long-enough-1"})
|
|
assert resp.status_code == 503
|
|
assert resp.json()["detail"]["code"] == "service_busy"
|
|
async with session_factory() as session:
|
|
assert await session.scalar(select(User).where(User.username == "late")) is None
|
|
|
|
|
|
# --- deposits --------------------------------------------------------------------
|
|
|
|
|
|
async def test_credit_confirmed_utxos_needs_the_write_lock(session_factory, funded):
|
|
entries = [{"tx_hash": "ab" * 32, "tx_pos": 0, "height": 100, "value": 5_000_000}]
|
|
async with _holding_write_lock(session_factory):
|
|
async with session_factory() as session:
|
|
with pytest.raises(OperationalError):
|
|
await credit_confirmed_utxos(session, funded, entries)
|
|
async with session_factory() as session:
|
|
assert await credit_confirmed_utxos(session, funded, entries) == 1
|
|
|
|
|
|
async def test_mark_utxos_spent_externally_needs_the_write_lock(session_factory, funded):
|
|
async with session_factory() as session:
|
|
utxo_id = (await session.scalar(select(UtxoEvent.id).where(UtxoEvent.user_id == funded)))
|
|
async with _holding_write_lock(session_factory):
|
|
async with session_factory() as session:
|
|
with pytest.raises(OperationalError):
|
|
await mark_utxos_spent_externally(session, funded, [utxo_id])
|
|
async with session_factory() as session:
|
|
assert await mark_utxos_spent_externally(session, funded, [utxo_id]) == 1
|
|
|
|
|
|
async def test_reinstate_takes_the_lock_only_when_there_is_something_to_reinstate(session_factory, funded):
|
|
"""It runs on every refresh: with nothing flagged it must not queue for the lock."""
|
|
async with session_factory() as session:
|
|
utxo = await session.scalar(select(UtxoEvent).where(UtxoEvent.user_id == funded))
|
|
utxo.spent_txid = "external-spend"
|
|
await session.commit()
|
|
entry = {"tx_hash": utxo.txid, "tx_pos": utxo.vout, "height": 100, "value": utxo.amount_sats}
|
|
|
|
async with _holding_write_lock(session_factory):
|
|
async with session_factory() as session:
|
|
# nothing flagged reappears -> no lock needed, no error
|
|
assert await reinstate_reappeared_utxos(session, funded, []) == 0
|
|
async with session_factory() as session:
|
|
with pytest.raises(OperationalError):
|
|
await reinstate_reappeared_utxos(session, funded, [entry])
|
|
|
|
async with session_factory() as session:
|
|
assert await reinstate_reappeared_utxos(session, funded, [entry]) == 1
|
|
utxo = await session.scalar(select(UtxoEvent).where(UtxoEvent.user_id == funded))
|
|
assert utxo.spent_txid is None
|
|
|
|
|
|
# --- background writers: scheduler, poller, reconciler, RBF -------------------------
|
|
|
|
|
|
async def test_scheduler_tick_state_transition_needs_the_write_lock_and_is_retried(session_factory, funded):
|
|
"""DAT-74: a state transition under a busy database raises out of the tick (run()
|
|
logs it and tries again five seconds later) instead of half-writing."""
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from app.db.models import Round
|
|
from app.rounds.scheduler import RoundScheduler
|
|
|
|
async with session_factory() as session:
|
|
session.add(Round(status="open", opened_at=datetime.now(timezone.utc) - timedelta(seconds=99999), duration_seconds=1))
|
|
await session.commit()
|
|
|
|
scheduler = RoundScheduler(session_factory, SimpleNamespace(client=object(), tip_height=1, tip_header_hex="00"))
|
|
async with _holding_write_lock(session_factory):
|
|
with pytest.raises(OperationalError):
|
|
await scheduler._tick()
|
|
async with session_factory() as session:
|
|
assert (await session.scalar(select(Round.status))) == "open" # untouched
|
|
|
|
await scheduler._tick() # lock free: the same tick now goes through
|
|
async with session_factory() as session:
|
|
assert (await session.scalar(select(Round.status))) == "closed" # no participants
|
|
|
|
|
|
def _payout_scheduler(session_factory, client):
|
|
from app.rounds.scheduler import RoundScheduler
|
|
from tests.unit.test_scheduler import FakePayoutListener
|
|
|
|
listener = FakePayoutListener(client)
|
|
# B-82: the network phase of a payout is the quorum's verification of the round's coins,
|
|
# so that is where a test injects "the lock is lost while the network is being used".
|
|
listener.on_verify = getattr(client, "_hook_listunspent", None)
|
|
return RoundScheduler(session_factory, listener)
|
|
|
|
|
|
class _PayoutClient:
|
|
def __init__(self, entries, hook_listunspent=None, hook_broadcast=None):
|
|
self._entries = entries
|
|
self._hook_listunspent = hook_listunspent
|
|
self._hook_broadcast = hook_broadcast
|
|
self.broadcasted: list[str] = []
|
|
|
|
async def listunspent(self, scripthash):
|
|
if self._hook_listunspent:
|
|
await self._hook_listunspent()
|
|
return self._entries
|
|
|
|
async def broadcast(self, raw_tx_hex):
|
|
self.broadcasted.append(raw_tx_hex)
|
|
if self._hook_broadcast:
|
|
await self._hook_broadcast()
|
|
return "txid"
|
|
|
|
|
|
async def _seed_payout(session_factory):
|
|
from tests.unit.test_scheduler import _POOL_AMOUNT_SATS, _seed_paying_out_round
|
|
|
|
await _seed_paying_out_round(session_factory)
|
|
return [{"tx_hash": "33" * 32, "tx_pos": 0, "height": 10, "value": _POOL_AMOUNT_SATS + 100_000}]
|
|
|
|
|
|
async def test_trigger_payout_defers_when_the_database_is_busy_before_reading(session_factory, keyed):
|
|
from app.db.models import Round
|
|
|
|
entries = await _seed_payout(session_factory)
|
|
client = _PayoutClient(entries)
|
|
async with _holding_write_lock(session_factory):
|
|
await _payout_scheduler(session_factory, client)._trigger_payout(1)
|
|
assert client.broadcasted == []
|
|
async with session_factory() as session:
|
|
assert (await session.scalars(select(PendingTransaction))).all() == []
|
|
assert (await session.get(Round, 1)).payout_txid is None
|
|
|
|
|
|
async def test_trigger_payout_does_not_broadcast_when_it_cannot_persist_the_intent(session_factory, keyed):
|
|
"""The lock is lost between building and persisting: nothing may be sent."""
|
|
entries = await _seed_payout(session_factory)
|
|
holder_cm = _holding_write_lock(session_factory)
|
|
|
|
async def grab():
|
|
await holder_cm.__aenter__()
|
|
|
|
client = _PayoutClient(entries, hook_listunspent=grab)
|
|
try:
|
|
await _payout_scheduler(session_factory, client)._trigger_payout(1)
|
|
finally:
|
|
await holder_cm.__aexit__(None, None, None)
|
|
assert client.broadcasted == []
|
|
async with session_factory() as session:
|
|
assert (await session.scalars(select(PendingTransaction))).all() == []
|
|
|
|
|
|
async def test_trigger_payout_leaves_the_row_building_when_the_lock_is_lost_after_the_broadcast(
|
|
session_factory, keyed
|
|
):
|
|
entries = await _seed_payout(session_factory)
|
|
holder_cm = _holding_write_lock(session_factory)
|
|
|
|
async def grab():
|
|
await holder_cm.__aenter__()
|
|
|
|
client = _PayoutClient(entries, hook_broadcast=grab)
|
|
try:
|
|
await _payout_scheduler(session_factory, client)._trigger_payout(1)
|
|
finally:
|
|
await holder_cm.__aexit__(None, None, None)
|
|
assert len(client.broadcasted) == 1
|
|
async with session_factory() as session:
|
|
pending = (await session.scalars(select(PendingTransaction))).one()
|
|
assert pending.status == "building" # the reconciler promotes it (TX-28)
|
|
events = (await session.scalars(select(AuditLog.event_type))).all()
|
|
assert "payout_sent" not in events
|
|
|
|
|
|
async def _seed_pending_withdrawal(session_factory, funded, status="pending", **kw):
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
async with session_factory() as session:
|
|
wd = Withdrawal(user_id=funded, external_address=_EXTERNAL_ADDRESS, amount_requested_sats=1, amount_sent_sats=1, txid="abcd", status="broadcast")
|
|
session.add(wd)
|
|
await session.flush()
|
|
row = PendingTransaction(
|
|
kind="withdrawal",
|
|
withdrawal_id=wd.id,
|
|
user_id=funded,
|
|
current_txid="abcd",
|
|
fee_rate_sat_vb=1,
|
|
raw_tx_hex="00",
|
|
status=status,
|
|
broadcast_at=datetime.now(timezone.utc) - timedelta(days=1),
|
|
**kw,
|
|
)
|
|
session.add(row)
|
|
await session.commit()
|
|
return row.id
|
|
|
|
|
|
async def test_confirmation_poller_writes_under_the_lock(session_factory, funded):
|
|
from app.tx.confirmation import poll_once
|
|
import app.withdrawals.confirmation # noqa: F401 (registers the handler)
|
|
|
|
row_id = await _seed_pending_withdrawal(session_factory, funded)
|
|
|
|
class Client:
|
|
async def get_history(self, scripthash):
|
|
return [{"tx_hash": "abcd", "height": 7}]
|
|
|
|
async with _holding_write_lock(session_factory):
|
|
with pytest.raises(OperationalError):
|
|
await poll_once(session_factory, Client(), FakeNetwork())
|
|
async with session_factory() as session:
|
|
assert (await session.get(PendingTransaction, row_id)).status == "pending"
|
|
|
|
assert await poll_once(session_factory, Client(), FakeNetwork()) == 1
|
|
async with session_factory() as session:
|
|
assert (await session.get(PendingTransaction, row_id)).status == "confirmed"
|
|
assert (await session.scalar(select(Withdrawal.status))) == "confirmed"
|
|
|
|
|
|
async def test_reconciler_resolves_under_the_lock(session_factory, funded):
|
|
from app.tx.reconcile import reconcile_once
|
|
|
|
row_id = await _seed_pending_withdrawal(session_factory, funded, status="building")
|
|
|
|
class Client:
|
|
async def get_history(self, scripthash):
|
|
return [{"tx_hash": "abcd", "height": 0}]
|
|
|
|
async with _holding_write_lock(session_factory):
|
|
with pytest.raises(OperationalError):
|
|
await reconcile_once(session_factory, Client())
|
|
async with session_factory() as session:
|
|
assert (await session.get(PendingTransaction, row_id)).status == "building"
|
|
|
|
assert await reconcile_once(session_factory, Client()) == 1
|
|
async with session_factory() as session:
|
|
assert (await session.get(PendingTransaction, row_id)).status == "pending"
|
|
|
|
|
|
async def test_bump_fee_persists_under_the_lock(session_factory, funded):
|
|
"""The persist phase of an RBF bump is a money-moving transaction like any other
|
|
(DAT-74): under a busy database it raises out of bump_fee rather than committing
|
|
a half-applied retarget."""
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from embit import script
|
|
from embit.transaction import Transaction, TransactionInput, TransactionOutput
|
|
|
|
from app.tx.broadcast import bump_fee
|
|
from app.wallet.hd import derive_user_key
|
|
from app.wallet.psbt_builder import Utxo, build_signed_transaction
|
|
|
|
signer = derive_user_key(0)
|
|
built = build_signed_transaction(
|
|
signing_key=signer,
|
|
from_script=script.p2wpkh(signer.to_public()),
|
|
utxos=[Utxo("11" * 32, 0, 150_000_000)],
|
|
to_address=_EXTERNAL_ADDRESS,
|
|
amount_sats=10_000_000,
|
|
change_address=derive_user_address(0),
|
|
fee_rate_sat_vb=1,
|
|
)
|
|
async with session_factory() as session:
|
|
session.add(
|
|
PendingTransaction(
|
|
kind="bet",
|
|
user_id=funded,
|
|
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),
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
class Client:
|
|
broadcasted: list = []
|
|
|
|
async def get_transaction(self, txid, verbose=False):
|
|
tx = Transaction(
|
|
vin=[TransactionInput(b"\x00" * 32, 0)],
|
|
vout=[TransactionOutput(150_000_000, script.Script(b"\x00\x14" + b"\x00" * 20))],
|
|
)
|
|
return tx.serialize().hex()
|
|
|
|
async def broadcast(self, raw_hex):
|
|
self.broadcasted.append(raw_hex)
|
|
return "x"
|
|
|
|
holder_cm = _holding_write_lock(session_factory)
|
|
|
|
class HookedClient(Client):
|
|
async def broadcast(self, raw_hex):
|
|
await super().broadcast(raw_hex)
|
|
await holder_cm.__aenter__() # the lock is taken right after the send
|
|
|
|
client = HookedClient()
|
|
try:
|
|
with pytest.raises(OperationalError):
|
|
await bump_fee(session_factory, client, 1)
|
|
finally:
|
|
await holder_cm.__aexit__(None, None, None)
|
|
async with session_factory() as session:
|
|
row = await session.get(PendingTransaction, 1)
|
|
assert row.current_txid == built.txid # nothing half-applied
|
|
assert len(client.broadcasted) == 1
|