Real SQLite transactions: BEGIN IMMEDIATE and synchronous=FULL (B-88, DAT-74, DAT-52, ARC-07, TX-28, BET-11, DAT-16)
pysqlite's legacy mode emitted BEGIN only at the first DML, so every read before it ran outside any transaction and a lock wait surfaced as "database is locked" at an arbitrary later statement; place_bet also mistook that error for a write-snapshot conflict, which SQLite cannot produce. - _register_sqlite_pragmas: synchronous=FULL, isolation_level=None and a `begin` event that emits BEGIN / BEGIN IMMEDIATE. - begin_immediate(session): takes the write lock before the first read; called first thing in place_bet, request_withdrawal, the scheduler transitions, _trigger_payout, the confirmation poller, the reconciler, bump_fee's persist phase, deposit crediting / external-spend marking / reinstating, and the derivation-index allocation at registration. - A lock not granted before a broadcast answers service_busy (503, all seven languages), nothing built or sent; after a broadcast nothing is undone, the rows stay `building` for the reconciler. - Spec, storico, BUGS.md and CLAUDE.md updated; 37 new tests on a real file DB. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
davideandClaude Sonnet 5.5 committed
1 parent
4be58709d1
commit
7ba8e22bb4
25 files changed
+1165
-82
No files matched your search
@@ -29,7 +29,7 @@ async def test_register_sqlite_pragmas_enables_wal_and_busy_timeout(sqlite_engin
|
||||
|
||||
assert (await _pragma(sqlite_engine, "journal_mode")).lower() == "wal"
|
||||
assert await _pragma(sqlite_engine, "busy_timeout") == _SQLITE_BUSY_TIMEOUT_MS
|
||||
assert await _pragma(sqlite_engine, "synchronous") == 1 # NORMAL
|
||||
assert await _pragma(sqlite_engine, "synchronous") == 2 # FULL (B-88, DAT-52)
|
||||
|
||||
|
||||
async def test_register_sqlite_pragmas_applies_to_every_new_connection(sqlite_engine):
|
||||
|
||||
@@ -0,0 +1,882 @@
|
||||
"""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 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 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 RuntimeError("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)
|
||||
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)
|
||||
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)
|
||||
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)
|
||||
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()), 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()), 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
|
||||
|
||||
return RoundScheduler(session_factory, FakePayoutListener(client))
|
||||
|
||||
|
||||
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())
|
||||
async with session_factory() as session:
|
||||
assert (await session.get(PendingTransaction, row_id)).status == "pending"
|
||||
|
||||
assert await poll_once(session_factory, Client()) == 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
|
||||
Reference in new issue
Block a user