diff --git a/BUGS.md b/BUGS.md index 5eb2657..4e71ee5 100644 --- a/BUGS.md +++ b/BUGS.md @@ -298,17 +298,6 @@ Fix: when a bump would leave change below `DUST_LIMIT_SATS`, the change output i dropped and its value goes to the fee (the replacement is then final); `RbfError` and every refused bump write a `rbf_bump_failed` audit row. -### B-88 — SQLite transactions are not what the code assumes - -pysqlite's legacy mode issues `BEGIN` only at the first DML, so reads before it are -not isolated, the "write-snapshot conflict" handled in `place_bet` cannot happen, and -lock waits surface as `database is locked` after 5s in arbitrary places. -`synchronous=NORMAL` can lose the last commits on power loss (e.g. a registration, -reusing its derivation index). - -Fix: `BEGIN IMMEDIATE` for every money-moving transaction, `synchronous=FULL`, -explicit handling of busy timeouts after a broadcast. - ### B-89 — Key problems surface only at first use Startup checks only that the key file exists and `XPRV_ENCRYPTION_KEY` is non-empty; diff --git a/CLAUDE.md b/CLAUDE.md index e2ad3fe..69e8b54 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -8,7 +8,7 @@ The user communicates in Italian in chat — reply to them in Italian. Everythin ## Project status -All 10 stages of the original build order are code-complete and unit-tested — 351 tests, all under `tests/unit/` (`tests/integration/` is an empty package). Beyond them: Docker + Caddy deployment, admin dashboard (`/admin`), static test UI (`/`), pending-inclusive balance display, an SSE push channel layered over the original polling, self-service password change + admin password reset, and the reconciliation/corroboration machinery below. +All 10 stages of the original build order are code-complete and unit-tested — 388 tests, all under `tests/unit/` (`tests/integration/` is an empty package). Beyond them: Docker + Caddy deployment, admin dashboard (`/admin`), static test UI (`/`), pending-inclusive balance display, an SSE push channel layered over the original polling, self-service password change + admin password reset, and the reconciliation/corroboration machinery below. Verified on mainnet with real money: registration + address derivation, deposit crediting (1-conf), a real 10 PLM bet (broadcast → confirmed → change credited back), and one full round cycle (close → draw on a real block hash → 70/30 payout with sat math checked against the broadcast tx → confirmation → close → next round auto-opened). **Withdrawal and the RBF bump path have never been exercised against a live broadcast** — unit-tested only. @@ -33,7 +33,7 @@ PYTHONPATH=. python scripts/decrypt_master_key.py # ops recovery: decrypt+pr PYTHONPATH=. python scripts/encrypt_master_key.py # ops bootstrap: import an externally-generated xprv (--overwrite to replace) PYTHONPATH=. python scripts/electrum_smoke_test.py # manual check: connect, handshake, subscribe to headers, print the tip -python -m pytest # all 351 tests +python -m pytest # all 388 tests python -m pytest tests/unit/test_hd.py # one file python -m pytest tests/unit/test_hd.py::test_derivation_is_deterministic # one test ``` @@ -199,6 +199,8 @@ Everything that spends money is written **before** it is broadcast and resolved - A refused broadcast releases the UTXOs, restores the balance, removes the participant (or marks the withdrawal `failed`), audit-logs, and raises `broadcast_failed` → **502**, since the network refused it, not the caller. - `tx/reconcile.py` asks the chain about anything still `building`/`pending`: present → promote; positively unknown → `failed` with a `failure_reason`, inputs released, domain row rolled back, `pending_tx_abandoned` logged. Grace differs by state (120s `building`, 6h `pending`, so the bumper gets its attempts first). A *transport* failure never abandons anything — only a server that positively doesn't know the tx, established by the tx's txid being absent from `get_history` of the address it spends from/pays back to (`app/tx/pending_address.py:own_address_for`) — B-41 replaced the old substring-matching on error text. A tx still in the server's mempool is never abandoned, so in practice that is what bounds the `closing` wait. +**SQLite transactions are real ones (B-88).** `_register_sqlite_pragmas` sets `synchronous=FULL` and switches pysqlite's implicit `BEGIN` off (`isolation_level = None`); an engine `begin` event emits `BEGIN` itself, so every read is inside a transaction. Every transaction that moves money or fixes its state opens with `await begin_immediate(session)` (`app/db/base.py`) — `BEGIN IMMEDIATE`, i.e. the write lock *before* the first read (it commits the caller's read-only transaction first and refuses a session with unflushed changes). A lock not granted within `busy_timeout` raises `OperationalError`: **before** a broadcast that means "refuse" (`service_busy`, 503, nothing built or sent); **after** one nothing is undone — the row stays `building` and the reconciler resolves it. New money-moving write code must call `begin_immediate` first, and tests of that must use a file DB whose engine went through `_register_sqlite_pragmas` (an in-memory engine without it never emits `BEGIN IMMEDIATE`). + `UtxoEvent.spent_txid` must always equal the tx's *current* txid, so `bump_fee` retargets it along with `RoundParticipant.bet_txid`, `Withdrawal.txid` and `Round.payout_txid` on every bump. `broadcast_at` is the *first* broadcast and is never rewritten (the reconciler's abandon clock measures from it); `last_broadcast_at` is what a bump updates and `should_bump` reads. Confirmation handlers key off immutable ids (`round_id`/`user_id`, `withdrawal_id`), never the txid, which changes under them. **Payouts retry, and are guarded against paying twice.** Every tick re-examines a `paying_out` round: `_retry_payout_if_due` throttles to one attempt per 60s using the latest `payout_failed` audit entry as its clock (a build failure leaves no DB row to throttle on), and every early return in `_trigger_payout` writes one, so `/admin` shows *why* a round is stuck. Before building, `_trigger_payout` refuses if a `building`/`pending` payout already exists for the round, and `_reserved_payout_outpoints` excludes pool UTXOs claimed by any unresolved payout — without both, a retry would pay the winner twice. diff --git a/README.md b/README.md index aee27bc..ba7b174 100644 --- a/README.md +++ b/README.md @@ -61,7 +61,7 @@ python -m pytest # all tests python -m pytest tests/unit/test_hd.py # one file ``` -351 unit tests cover HD derivation, PSBT building, the Electrum client, bets, +388 unit tests cover HD derivation, PSBT building, the Electrum client, bets, deposits, withdrawals, the round/draw engine, RBF fee-bumping, admin config, the pending-inclusive balance calculation, and the SSE push channel. No automated integration tests against a live Electrum connection — mainnet diff --git a/app/api/routes/bets.py b/app/api/routes/bets.py index 84a4f3d..8ef7b2c 100644 --- a/app/api/routes/bets.py +++ b/app/api/routes/bets.py @@ -41,6 +41,8 @@ async def create_bet( code = ( status.HTTP_502_BAD_GATEWAY if exc.code == "broadcast_failed" + else status.HTTP_503_SERVICE_UNAVAILABLE # B-88: write lock not granted, nothing sent + if exc.code == "service_busy" else status.HTTP_400_BAD_REQUEST ) raise from_api_error(code, exc) from exc diff --git a/app/api/routes/withdrawals.py b/app/api/routes/withdrawals.py index bf4eee4..36e38f9 100644 --- a/app/api/routes/withdrawals.py +++ b/app/api/routes/withdrawals.py @@ -47,6 +47,8 @@ async def create_withdrawal( code = ( status.HTTP_502_BAD_GATEWAY # the network refused it, not the caller (B-07) if exc.code == "broadcast_failed" + else status.HTTP_503_SERVICE_UNAVAILABLE # B-88: write lock not granted, nothing sent + if exc.code == "service_busy" else status.HTTP_400_BAD_REQUEST ) raise from_api_error(code, exc) from exc diff --git a/app/auth/routes.py b/app/auth/routes.py index 9664261..1a20ffe 100644 --- a/app/auth/routes.py +++ b/app/auth/routes.py @@ -1,7 +1,7 @@ from fastapi import APIRouter, Depends, Request, status from pydantic import BaseModel, Field from sqlalchemy import func, select -from sqlalchemy.exc import IntegrityError +from sqlalchemy.exc import IntegrityError, OperationalError from sqlalchemy.ext.asyncio import AsyncSession from app.api.client_ip import client_ip as _client_ip @@ -13,6 +13,7 @@ from app.auth.security import ( hash_password_async, verify_password_async, ) +from app.db.base import begin_immediate from app.db.models import User from app.db.session import get_session from app.wallet.hd import derive_user_address @@ -89,6 +90,18 @@ async def register( password_hash = await hash_password_async(body.password) for _ in range(_MAX_REGISTER_RETRIES): + # B-88: the max(derivation_index) read and the insert that uses it are one + # transaction under the write lock — with a plain BEGIN two concurrent + # registrations both read the same maximum and one of them failed on the + # unique index. A lock that cannot be had is a "try again", not a 500. + try: + await begin_immediate(session) + except OperationalError as exc: + raise http_error( + status.HTTP_503_SERVICE_UNAVAILABLE, + "service_busy", + "the service is busy right now, nothing was sent — please try again", + ) from exc max_index = await session.scalar(select(func.max(User.derivation_index))) next_index = 0 if max_index is None else max_index + 1 address = derive_user_address(next_index) diff --git a/app/bets/service.py b/app/bets/service.py index 4ebd480..d138f49 100644 --- a/app/bets/service.py +++ b/app/bets/service.py @@ -1,3 +1,4 @@ +import logging from datetime import datetime, timezone from embit import script @@ -7,6 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.api.errors import ApiError from app.audit.log import write_audit_log +from app.db.base import begin_immediate from app.db.models import PendingTransaction, Round, RoundParticipant, User, UtxoEvent from app.electrum.client import ElectrumClient from app.rounds.config import get_round_config @@ -23,6 +25,9 @@ from app.wallet.psbt_builder import ( ) +logger = logging.getLogger(__name__) + + class BetError(ApiError): pass @@ -33,6 +38,17 @@ class _RoundClosedDuringBuild(Exception): async def place_bet(session: AsyncSession, client: ElectrumClient, user: User) -> RoundParticipant: + # B-88 (DAT-74, BET-11): take SQLite's write lock before the first read, so the + # balance, the free UTXOs, the participants and the round's status read below are + # the state the phase-1 write lands on — a plain BEGIN only starts at the first + # write, leaving every read before it outside any transaction. Failing to get the + # lock within busy_timeout is not a conflict about this bet: nothing has been + # built or broadcast yet, so the player is simply asked to retry. + try: + await begin_immediate(session) + except OperationalError as exc: + raise _service_busy() from exc + round_ = await open_new_round_if_needed(session) if round_ is None: raise BetError("no_round_open", "no round open right now, please try again shortly") @@ -148,15 +164,19 @@ async def place_bet(session: AsyncSession, client: ElectrumClient, user: User) - if guard.rowcount != 1: raise _RoundClosedDuringBuild await session.commit() - except (_RoundClosedDuringBuild, OperationalError) as exc: - # OperationalError here is SQLite's write-snapshot conflict: the round row - # changed under us, which is the same situation as the guard matching nothing. + except _RoundClosedDuringBuild as exc: # Nothing has been broadcast yet, so the rollback undoes phase 1 entirely — # the UTXOs stay unspent and no participant row survives. await session.rollback() raise BetError( "round_closing", "the current round is closing, please try again shortly" ) from exc + except OperationalError as exc: + # B-88: SQLite has no write-snapshot conflicts — an OperationalError here is + # the lock wait (busy_timeout) expiring, which says nothing about the round. + # Same rollback, but the honest answer is "busy", not "closing". + await session.rollback() + raise _service_busy() from exc # --- Phase 2: broadcast, then promote both rows to their live state --------- try: @@ -165,9 +185,26 @@ async def place_bet(session: AsyncSession, client: ElectrumClient, user: User) - # The node refused it (fee too low, dust, mempool conflict, or simply an # unreachable server) — nothing is on-chain, so undo phase 1 completely and # give the user a translatable failure instead of a bare 500 (B-07). - await _release_failed_bet(session, participant, pending, built, user.id, str(exc)) + try: + await _release_failed_bet(session, participant, pending, built, user.id, str(exc)) + except OperationalError: + # B-88: the database is busy, so the release could not be written. The + # rows stay "building" — the state phase 1 committed — and the + # reconciler resolves them against the chain. + await session.rollback() + logger.error("could not release the failed bet %s: database busy, left to the reconciler", built.txid) raise BetError("broadcast_failed", f"the network refused the transaction: {exc}") from exc + # B-88: after the broadcast nothing may be undone. If the lock cannot be had the + # rows stay "building" (the tx is on the network) and the reconciler promotes + # them; the caller still gets the participant, with the status the DB holds. + try: + await begin_immediate(session) + except OperationalError: + await session.rollback() + logger.error("bet %s broadcast but the database is busy: left 'building' for the reconciler", built.txid) + await session.refresh(participant) # reports the status the DB really holds + return participant participant.status = "broadcast" pending.status = "pending" await write_audit_log( @@ -177,7 +214,6 @@ async def place_bet(session: AsyncSession, client: ElectrumClient, user: User) - user_id=user.id, round_id=round_.id, ) - await session.commit() await session.refresh(participant) broadcaster.publish() # participant_count/jackpot changed — nudge every dashboard to refetch @@ -195,6 +231,7 @@ async def _release_failed_bet( """Undo phase 1 after a failed broadcast: free the UTXOs the build reserved, drop the two rows, and restore the balance. Same shape as what the reconciler does for a tx that turns out never to have made it onto the chain.""" + await begin_immediate(session) # B-88 for spent in built.spent_utxos: row = await session.scalar( select(UtxoEvent).where(UtxoEvent.txid == spent.txid, UtxoEvent.vout == spent.vout) @@ -218,6 +255,13 @@ async def _release_failed_bet( broadcaster.publish() +def _service_busy() -> BetError: + return BetError( + "service_busy", + "the service is busy right now, nothing was sent — please try again", + ) + + def _pending_transaction( round_id: int, user_id: int, built: BuiltTransaction, fee_rate_sat_vb: int ) -> PendingTransaction: diff --git a/app/db/base.py b/app/db/base.py index fa0dbaa..629ae94 100644 --- a/app/db/base.py +++ b/app/db/base.py @@ -1,5 +1,5 @@ from sqlalchemy import event -from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.orm import DeclarativeBase from app.config import settings @@ -12,6 +12,11 @@ from app.config import settings _SQLITE_BUSY_TIMEOUT_MS = 5000 +# Execution option read by the `begin` event below: which BEGIN a connection's next +# transaction opens with. Set through `begin_immediate`, never by hand. +_BEGIN_MODE_OPTION = "sqlite_begin_mode" + + def _register_sqlite_pragmas(target_engine: AsyncEngine) -> None: """Without WAL, SQLite's default (rollback-journal) mode lets a writer block every reader for the duration of its transaction, and a second writer arriving @@ -21,22 +26,69 @@ def _register_sqlite_pragmas(target_engine: AsyncEngine) -> None: busy_timeout gives a second writer a real window to wait for the first instead of an instant `OperationalError`. + B-88: two more things, both about the code's transactions being what it + assumes. `synchronous=FULL`, because with WAL `NORMAL` can lose the last + commits on power loss — a registration, and with it a derivation index that + would then be handed to the next user. And real transaction control: pysqlite's + legacy mode emits `BEGIN` only at the first INSERT/UPDATE/DELETE, so every read + before it ran outside any transaction and the lock wait surfaced as `database is + locked` at an arbitrary later statement. `isolation_level = None` turns the + driver's implicit BEGIN off and the `begin` event emits it ourselves — a plain + `BEGIN` by default, `BEGIN IMMEDIATE` for a transaction opened through + `begin_immediate`, which takes the write lock *before* the first read so that + what a money-moving transaction reads is what it then writes on top of. + No-op for any dialect other than sqlite (e.g. a future PostgreSQL - DATABASE_URL), which neither needs nor understands these pragmas. + DATABASE_URL), which neither needs nor understands any of this. """ if target_engine.dialect.name != "sqlite": return @event.listens_for(target_engine.sync_engine, "connect") def _set_sqlite_pragmas(dbapi_connection, connection_record) -> None: + # Before the pragmas: journal_mode cannot change inside a transaction, and in + # legacy mode the driver may have one open. + dbapi_connection.isolation_level = None cursor = dbapi_connection.cursor() try: cursor.execute("PRAGMA journal_mode=WAL") - cursor.execute("PRAGMA synchronous=NORMAL") + cursor.execute("PRAGMA synchronous=FULL") # B-88 cursor.execute(f"PRAGMA busy_timeout={_SQLITE_BUSY_TIMEOUT_MS}") finally: cursor.close() + @event.listens_for(target_engine.sync_engine, "begin") + def _emit_begin(conn) -> None: # B-88 + mode = conn.get_execution_options().get(_BEGIN_MODE_OPTION, "DEFERRED") + conn.exec_driver_sql("BEGIN IMMEDIATE" if mode == "IMMEDIATE" else "BEGIN") + + +async def begin_immediate(session: AsyncSession) -> None: + """B-88: start the session's next transaction with `BEGIN IMMEDIATE`, i.e. take + SQLite's write lock now, before the first read. + + Every transaction that moves money or fixes its state calls this first thing, so + that what it reads (balance, free UTXOs, participants, the round's status) cannot + change under it before its own writes, and so that the wait for the lock happens + here — at the very start, before anything has been broadcast — instead of at the + first write. Raises `sqlalchemy.exc.OperationalError` ("database is locked") when + the lock is not granted within `busy_timeout`; the caller decides whether that + means "refuse" (before a broadcast) or "leave it to the reconciler" (after). + + A transaction the session already has open is committed first — a lock cannot be + upgraded in place, and the callers open the transaction with nothing but reads + behind them (typically the authentication lookup). Pending, unflushed changes + make that unsafe to do silently, so they are an error. + + Harmless on a session whose engine did not go through `_register_sqlite_pragmas` + (the option is simply never read) and on a non-sqlite dialect. + """ + if session.new or session.dirty or session.deleted: + raise RuntimeError("begin_immediate called with unflushed changes on the session") + if session.in_transaction(): + await session.commit() + await session.connection(execution_options={_BEGIN_MODE_OPTION: "IMMEDIATE"}) + engine = create_async_engine(settings.database_url) _register_sqlite_pragmas(engine) diff --git a/app/deposits/service.py b/app/deposits/service.py index 3dde2dd..d67444d 100644 --- a/app/deposits/service.py +++ b/app/deposits/service.py @@ -4,6 +4,7 @@ from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.log import write_audit_log +from app.db.base import begin_immediate from app.db.models import UtxoEvent from app.rounds.events import broadcaster from app.wallet.balance import recompute_balance @@ -45,7 +46,12 @@ async def credit_confirmed_utxos(session: AsyncSession, user_id: int, entries: l entries: [{"tx_hash": ..., "tx_pos": ..., "height": ..., "value": ...}, ...] height <= 0 means unconfirmed (mempool) per the Electrum protocol convention — skipped, since the spec requires 1 confirmation before crediting. + + B-88: opens with BEGIN IMMEDIATE, so the "already recorded?" read and the inserts + it guards are one transaction — the unique (txid, vout) constraint stays the + backstop, not the mechanism. """ + await begin_immediate(session) existing_keys = { (txid, vout) for txid, vout in ( @@ -98,6 +104,20 @@ async def reinstate_reappeared_utxos(session: AsyncSession, user_id: int, entrie """ current_keys = {(entry["tx_hash"], entry["tx_pos"]) for entry in entries} + # B-88: the lock is only worth taking when there is something to write, and this + # runs on every refresh — so look first (plain columns), then take it and read the + # rows again under it. + flagged = ( + await session.execute( + select(UtxoEvent.txid, UtxoEvent.vout).where( + UtxoEvent.user_id == user_id, UtxoEvent.spent_txid == _EXTERNAL_SPEND_SENTINEL + ) + ) + ).all() + if not any((txid, vout) in current_keys for txid, vout in flagged): + return 0 + await begin_immediate(session) + marked_rows = ( await session.scalars( select(UtxoEvent).where( @@ -178,6 +198,7 @@ async def mark_utxos_spent_externally(session: AsyncSession, user_id: int, utxo_ may have resolved it (a legitimate platform spend, or a prior refresh) between when the caller read the candidate list and finished corroborating it. """ + await begin_immediate(session) # B-88: that re-check and the mark are one transaction marked = 0 for utxo_id in utxo_ids: row = await session.get(UtxoEvent, utxo_id) diff --git a/app/rounds/scheduler.py b/app/rounds/scheduler.py index 12d2240..4add837 100644 --- a/app/rounds/scheduler.py +++ b/app/rounds/scheduler.py @@ -5,9 +5,11 @@ from datetime import datetime, timedelta, timezone from embit import script from embit.transaction import Transaction from sqlalchemy import func, select +from sqlalchemy.exc import OperationalError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from app.audit.log import write_audit_log +from app.db.base import begin_immediate from app.db.models import AuditLog, PendingTransaction, Round, RoundParticipant, User from app.electrum.listener import ElectrumListener from app.electrum.scripthash import address_to_scripthash @@ -63,6 +65,10 @@ class RoundScheduler: return async with self._session_factory() as session: + # B-88: every state transition below starts with BEGIN IMMEDIATE (DAT-74). + # A busy database raises out of the tick, which run() logs and retries at + # the next one — nothing has been written yet. + await begin_immediate(session) round_ = await open_new_round_if_needed(session) await session.commit() if round_ is None: @@ -90,6 +96,7 @@ class RoundScheduler: return async with self._session_factory() as session: + await begin_immediate(session) # B-88 round_ = await session.get(Round, round_id) # No new bets from here on, regardless of how long the pending-bet # wait below takes — flip to "closing" immediately so it's observable @@ -120,6 +127,7 @@ class RoundScheduler: async def _close_and_draw(self, round_id: int) -> None: async with self._session_factory() as session: + await begin_immediate(session) # B-88 round_ = await session.get(Round, round_id) # B-53: re-check for in-flight bets in the *same* session that snapshots @@ -178,6 +186,7 @@ class RoundScheduler: winner_address = draw_winner(addresses, block_hash) async with self._session_factory() as session: + await begin_immediate(session) # B-88 round_ = await session.get(Round, round_id) round_.draw_block_height = block_height round_.draw_block_hash = block_hash @@ -361,6 +370,12 @@ class RoundScheduler: # --- Phase 1: read (session closed before any network I/O) --------------- async with self._session_factory() as session: + try: + await begin_immediate(session) # B-88 + except OperationalError: + # Nothing built or sent yet: try again at the next tick. + logger.warning("round %s payout deferred: database busy", round_id) + return round_ = await session.get(Round, round_id) already_in_flight = await session.scalar( select(PendingTransaction).where( @@ -450,6 +465,13 @@ class RoundScheduler: # --- Phase 3: persist the intent, *then* broadcast (B-25) ----------------- async with self._session_factory() as session: + try: + await begin_immediate(session) # B-88 + except OperationalError: + # Not written, not broadcast: the payout is simply rebuilt on the retry. + logger.warning("round %s payout deferred before broadcast: database busy", round_id) + await self._log_payout_failure(round_id, winner_user_id, "database busy") + return round_ = await session.get(Round, round_id) round_.winner_amount_sats = built.winner_sats round_.fee_amount_sats = built.commission_sats @@ -481,6 +503,17 @@ class RoundScheduler: return async with self._session_factory() as session: + try: + await begin_immediate(session) # B-88 + except OperationalError: + # The tx is on the network; the row stays "building" and the + # reconciler finds it on the chain and promotes it (TX-28). + logger.error( + "round %s payout %s broadcast but the database is busy: left 'building'", + round_id, + built.txid, + ) + return pending = await session.get(PendingTransaction, pending_id) pending.status = "pending" await write_audit_log( diff --git a/app/static/i18n.js b/app/static/i18n.js index a0e765c..df848d1 100644 --- a/app/static/i18n.js +++ b/app/static/i18n.js @@ -159,6 +159,7 @@ const TRANSLATIONS = { 'error.invalid_request': 'Invalid request, please check the entered data.', 'error.invalid_amount': 'Enter an amount greater than zero.', 'error.broadcast_failed': 'The network refused the transaction. Please try again shortly.', + 'error.service_busy': 'The service is busy right now and nothing was sent. Please try again in a moment.', 'error.amount_below_dust_limit': 'The amount is too small to be sent.', 'error.too_many_inputs': 'Your balance is split across too many small deposits to be spent in a single transaction (max {max_inputs}). Please contact support to consolidate it.', 'error.withdrawal_to_own_address': 'That is your own deposit address — withdraw to an external wallet.', @@ -324,6 +325,7 @@ const TRANSLATIONS = { 'error.invalid_request': 'Richiesta non valida, controlla i dati inseriti.', 'error.invalid_amount': 'Inserisci un importo maggiore di zero.', 'error.broadcast_failed': 'La rete ha rifiutato la transazione. Riprova tra poco.', + 'error.service_busy': 'Il servizio è occupato in questo momento e non è stato inviato nulla. Riprova tra poco.', 'error.amount_below_dust_limit': "L'importo è troppo basso per essere inviato.", 'error.too_many_inputs': 'Il tuo saldo è suddiviso in troppi piccoli depositi per essere speso in una sola transazione (max {max_inputs}). Contatta l\'assistenza per consolidarlo.', 'error.withdrawal_to_own_address': 'Questo è il tuo indirizzo di deposito — preleva verso un wallet esterno.', @@ -489,6 +491,7 @@ const TRANSLATIONS = { 'error.invalid_request': 'Solicitud no válida, revisa los datos introducidos.', 'error.invalid_amount': 'Introduce un importe mayor que cero.', 'error.broadcast_failed': 'La red rechazó la transacción. Inténtalo de nuevo en un momento.', + 'error.service_busy': 'El servicio está ocupado en este momento y no se ha enviado nada. Inténtalo de nuevo en un momento.', 'error.amount_below_dust_limit': 'El importe es demasiado pequeño para enviarse.', 'error.too_many_inputs': 'Tu saldo está repartido en demasiados depósitos pequeños para gastarse en una sola transacción (máx. {max_inputs}). Contacta con soporte para consolidarlo.', 'error.withdrawal_to_own_address': 'Esa es tu propia dirección de depósito — retira a una cartera externa.', @@ -654,6 +657,7 @@ const TRANSLATIONS = { 'error.invalid_request': 'Requête invalide, vérifiez les données saisies.', 'error.invalid_amount': 'Saisissez un montant supérieur à zéro.', 'error.broadcast_failed': 'Le réseau a refusé la transaction. Veuillez réessayer dans un instant.', + 'error.service_busy': 'Le service est occupé pour le moment et rien n\'a été envoyé. Veuillez réessayer dans un instant.', 'error.amount_below_dust_limit': "Le montant est trop faible pour être envoyé.", 'error.too_many_inputs': 'Votre solde est réparti sur trop de petits dépôts pour être dépensé en une seule transaction (max {max_inputs}). Contactez le support pour le consolider.', 'error.withdrawal_to_own_address': "C'est votre propre adresse de dépôt — retirez vers un portefeuille externe.", @@ -819,6 +823,7 @@ const TRANSLATIONS = { 'error.invalid_request': 'Ungültige Anfrage, bitte überprüfe die eingegebenen Daten.', 'error.invalid_amount': 'Gib einen Betrag größer als null ein.', 'error.broadcast_failed': 'Das Netzwerk hat die Transaktion abgelehnt. Bitte versuche es in Kürze erneut.', + 'error.service_busy': 'Der Dienst ist gerade ausgelastet und es wurde nichts gesendet. Bitte versuche es gleich noch einmal.', 'error.amount_below_dust_limit': 'Der Betrag ist zu klein, um gesendet zu werden.', 'error.too_many_inputs': 'Ihr Guthaben ist auf zu viele kleine Einzahlungen verteilt, um in einer einzigen Transaktion ausgegeben zu werden (max. {max_inputs}). Bitte wenden Sie sich an den Support, um es zusammenzufassen.', 'error.withdrawal_to_own_address': 'Das ist deine eigene Einzahlungsadresse — zahle auf eine externe Wallet aus.', @@ -984,6 +989,7 @@ const TRANSLATIONS = { 'error.invalid_request': 'Некорректный запрос, проверьте введённые данные.', 'error.invalid_amount': 'Введите сумму больше нуля.', 'error.broadcast_failed': 'Сеть отклонила транзакцию. Попробуйте ещё раз через минуту.', + 'error.service_busy': 'Сервис сейчас занят, ничего не было отправлено. Попробуйте ещё раз через минуту.', 'error.amount_below_dust_limit': 'Сумма слишком мала для отправки.', 'error.too_many_inputs': 'Ваш баланс разбит на слишком много мелких депозитов, чтобы потратить его одной транзакцией (максимум {max_inputs}). Обратитесь в поддержку для консолидации.', 'error.withdrawal_to_own_address': 'Это ваш собственный адрес для депозита — выводите на внешний кошелёк.', @@ -1149,6 +1155,7 @@ const TRANSLATIONS = { 'error.invalid_request': '请求无效，请检查填写的内容。', 'error.invalid_amount': '请输入大于零的金额。', 'error.broadcast_failed': '网络拒绝了该交易，请稍后重试。', + 'error.service_busy': '服务当前繁忙，没有发送任何内容，请稍后重试。', 'error.amount_below_dust_limit': '金额过小，无法发送。', 'error.too_many_inputs': '您的余额分散在过多的小额存款中，无法在一笔交易中花费（最多 {max_inputs} 笔）。请联系客服进行归集。', 'error.withdrawal_to_own_address': '这是你自己的充值地址 — 请提现到外部钱包。', diff --git a/app/tx/broadcast.py b/app/tx/broadcast.py index 4540a99..5dd494c 100644 --- a/app/tx/broadcast.py +++ b/app/tx/broadcast.py @@ -9,6 +9,7 @@ from embit.finalizer import finalize_psbt from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker +from app.db.base import begin_immediate from app.db.models import PendingTransaction, Round, RoundParticipant, User, UtxoEvent, Withdrawal from app.electrum.client import ElectrumClient from app.rounds.config import get_round_config @@ -175,6 +176,7 @@ async def bump_fee( # --- Phase 3: persist the outcome ---------------------------------------------- async with session_factory() as session: + await begin_immediate(session) # B-88 pending = await session.get(PendingTransaction, pending_id) old_txid = pending.current_txid pending.replaced_by_txid = old_txid # points backwards: what current_txid replaced diff --git a/app/tx/confirmation.py b/app/tx/confirmation.py index ab9a53a..1ac4356 100644 --- a/app/tx/confirmation.py +++ b/app/tx/confirmation.py @@ -5,6 +5,7 @@ from collections.abc import Awaitable, Callable from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker +from app.db.base import begin_immediate from app.db.models import PendingTransaction from app.electrum.client import ElectrumClient from app.electrum.scripthash import address_to_scripthash @@ -80,6 +81,7 @@ async def poll_once(session_factory: async_sessionmaker, client: ElectrumClient) continue async with session_factory() as session: + await begin_immediate(session) # B-88: read the row and write its outcome under one lock row = await session.get(PendingTransaction, pending_id) if row is None or row.status != "pending": continue diff --git a/app/tx/reconcile.py b/app/tx/reconcile.py index 2382252..83bcfe4 100644 --- a/app/tx/reconcile.py +++ b/app/tx/reconcile.py @@ -34,6 +34,7 @@ from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from app.audit.log import write_audit_log +from app.db.base import begin_immediate from app.db.models import PendingTransaction, Round, RoundParticipant, UtxoEvent, Withdrawal from app.electrum.client import ElectrumClient from app.electrum.scripthash import address_to_scripthash @@ -122,6 +123,7 @@ async def reconcile_once(session_factory: async_sessionmaker, client: ElectrumCl continue async with session_factory() as session: + await begin_immediate(session) # B-88: the re-check and the resolution under one lock row = await session.get(PendingTransaction, row_id) if row is None or row.status != status: continue # something else moved it while we were asking diff --git a/app/withdrawals/service.py b/app/withdrawals/service.py index 9987333..2d156a5 100644 --- a/app/withdrawals/service.py +++ b/app/withdrawals/service.py @@ -1,9 +1,13 @@ +import logging + from embit import script from sqlalchemy import select +from sqlalchemy.exc import OperationalError from sqlalchemy.ext.asyncio import AsyncSession from app.api.errors import ApiError from app.audit.log import write_audit_log +from app.db.base import begin_immediate from app.db.models import PendingTransaction, User, UtxoEvent, Withdrawal from app.electrum.client import ElectrumClient from app.rounds.config import get_round_config @@ -19,6 +23,9 @@ from app.wallet.psbt_builder import ( ) +logger = logging.getLogger(__name__) + + class WithdrawalError(ApiError): pass @@ -43,6 +50,18 @@ async def request_withdrawal( "that is your own deposit address — withdraw to an external wallet instead", ) + # B-88 (DAT-74, TX-28): the write lock before the first read that decides + # anything, so the UTXOs selected below are the ones phase 1 then marks spent. + # Taken only after the checks that need no database at all, so a malformed + # request never queues for it. No lock within busy_timeout means nothing has been + # built or broadcast: the caller is asked to retry. + try: + await begin_immediate(session) + except OperationalError as exc: + raise WithdrawalError( + "service_busy", "the service is busy right now, nothing was sent — please try again" + ) from exc + config = await get_round_config(session) if amount_sats < config.bet_amount_sats: raise WithdrawalError( @@ -130,11 +149,31 @@ async def request_withdrawal( try: await client.broadcast(built.raw_hex) except Exception as exc: - await _release_failed_withdrawal(session, withdrawal, pending, built, user.id, str(exc)) + try: + await _release_failed_withdrawal(session, withdrawal, pending, built, user.id, str(exc)) + except OperationalError: + # B-88: the release could not be written (database busy). The rows stay + # "building", the state committed before the broadcast, and the + # reconciler resolves them against the chain. + await session.rollback() + logger.error( + "could not release the failed withdrawal %s: database busy, left to the reconciler", built.txid + ) raise WithdrawalError( "broadcast_failed", f"the network refused the transaction: {exc}" ) from exc + # B-88: after the broadcast nothing is undone. Without the lock the rows stay + # "building" and the reconciler promotes them; the caller sees the real status. + try: + await begin_immediate(session) + except OperationalError: + await session.rollback() + logger.error( + "withdrawal %s broadcast but the database is busy: left 'building' for the reconciler", built.txid + ) + await session.refresh(withdrawal) + return withdrawal withdrawal.status = "broadcast" pending.status = "pending" await write_audit_log( @@ -162,6 +201,7 @@ async def _release_failed_withdrawal( The Withdrawal row is kept (marked "failed") rather than deleted: unlike a bet, a withdrawal is an instruction the user gave, and they should be able to see that it didn't go through.""" + await begin_immediate(session) # B-88 for spent in built.spent_utxos: row = await session.scalar( select(UtxoEvent).where(UtxoEvent.txid == spent.txid, UtxoEvent.vout == spent.vout) diff --git a/docs/api.md b/docs/api.md index 958de60..daea976 100644 --- a/docs/api.md +++ b/docs/api.md @@ -205,8 +205,11 @@ esiste una API separata per l'amministrazione, solo un'autenticazione diversa. `app/api/errors.py:from_api_error` - **[API-13]** Gli errori di `POST /bets` e `POST /withdrawals` sollevati dal - servizio rispondono `400`, **tranne** `broadcast_failed`, che risponde `502`: - è la rete ad aver rifiutato la transazione, non il chiamante. + servizio rispondono `400`, **tranne** `broadcast_failed`, che risponde `502` + (è la rete ad aver rifiutato la transazione, non il chiamante), e + `service_busy`, che risponde `503` (il database non ha concesso il lock di + scrittura entro il timeout, [DAT-74](modello-dati.md); nulla è stato + costruito né trasmesso). `broadcast_failed` si usa **solo** per un rifiuto esplicito del server (un errore JSON-RPC in risposta al broadcast), e significa che tutto è stato annullato. Un esito **ambiguo** del broadcast (timeout, connessione caduta @@ -453,8 +456,8 @@ esiste una API separata per l'amministrazione, solo un'autenticazione diversa. `insufficient_balance` (`params.required_sats`), `too_many_inputs` (`params.max_inputs`), `balance_leaves_no_change` (`params.required_extra_sats`), `amount_below_network_fee`, - `amount_below_dust_limit`, `502 broadcast_failed` (solo rifiuto - esplicito). Il significato di ciascuno è nel + `amount_below_dust_limit`, `503 service_busy`, `502 broadcast_failed` + (solo rifiuto esplicito). Il significato di ciascuno è nel [catalogo](#catalogo-dei-codici-di-errore). - **Serializzazione:** la richiesta tiene il blocco per-utente condiviso con `POST /withdrawals` per tutta la costruzione e la trasmissione. @@ -506,8 +509,8 @@ esiste una API separata per l'amministrazione, solo un'autenticazione diversa. prelievo dell'intero saldo, B-96), `balance_pending_confirmation` (`params.pending_sats`), `insufficient_balance` (`params.required_sats`), `too_many_inputs` (`params.max_inputs`), `amount_below_network_fee`, - `amount_below_dust_limit`; `502 broadcast_failed` (solo rifiuto - esplicito). + `amount_below_dust_limit`; `503 service_busy`; `502 broadcast_failed` + (solo rifiuto esplicito). - **Serializzazione:** stesso blocco per-utente di `POST /bets`. - **Effetti:** segna le monete spese, crea prelievo e transazione in sospeso, trasmette al server attivo e poi, *best effort*, a tutti i @@ -1063,7 +1066,7 @@ dei parametri operativi in | `amount_below_minimum` | 400 | `app/withdrawals/service.py:request_withdrawal` | `minimum_sats` | prelievo sotto il costo attuale della scommessa | sì | | `internal_error` | 500 | `app/main.py:log_unhandled_exception` | — | eccezione non gestita | sì | | `withdrawal_to_platform_address` | 400 | `app/withdrawals/service.py:request_withdrawal` | — | destinazione uguale a un indirizzo di montepremi, di riserva o delle commissioni — *da implementare (B-80)* | no — da aggiungere | -| `service_busy` | 503 | `app/bets/service.py:place_bet` | — | il database non ha concesso il lock di scrittura entro il timeout; nulla è stato trasmesso, si può riprovare — *da implementare (B-88)* | no — da aggiungere | +| `service_busy` | 503 | `app/bets/service.py:place_bet`, `app/withdrawals/service.py:request_withdrawal`, `app/auth/routes.py:register` | — | il database non ha concesso il lock di scrittura entro il timeout; nulla è stato trasmesso, si può riprovare | sì | | `registration_unavailable` | 503 | `app/auth/routes.py:register` | — | registrazione sospesa finché il controllo degli indici di derivazione dopo un ripristino non è concluso o risolto ([registrazione-e-depositi.md](registrazione-e-depositi.md), B-90) — *da implementare* | no — da aggiungere | | `not_authenticated` | 401 | dipendenza Bearer di FastAPI | — | header `Authorization` assente — oggi `detail` stringa — *da implementare (B-94)* | no — da aggiungere | | `invalid_admin_token` | 403 | `app/api/routes/admin.py:require_admin` | — | token admin assente o errato — oggi `detail` stringa — *da implementare (B-94)* | no — da aggiungere | diff --git a/docs/architettura.md b/docs/architettura.md index 98143bf..8957c6a 100644 --- a/docs/architettura.md +++ b/docs/architettura.md @@ -121,11 +121,8 @@ flowchart LR (la riga resta al riconciliatore). Con un dialetto diverso da SQLite le pragma non vengono applicate. *Codice:* `app/db/base.py:_register_sqlite_pragmas` · - *Test:* `tests/unit/test_db_base.py` - *Stato:* da implementare (B-88) · *Oggi:* `synchronous=NORMAL`, e pysqlite - in modalità legacy emette `BEGIN` solo alla prima scrittura, quindi le - letture precedenti non sono isolate e l'attesa del lock può scattare in - punti qualsiasi. + `app/db/base.py:begin_immediate` · + *Test:* `tests/unit/test_db_base.py` · `tests/unit/test_sqlite_transactions.py` - **[ARC-08]** Lo schema è definito in `app/db/models.py` ed evolve solo tramite migrazioni Alembic in `migrations/versions/` (generate con `alembic revision --autogenerate`); `migrations/env.py` prende l'URL del diff --git a/docs/decisioni-e-limiti.md b/docs/decisioni-e-limiti.md index bea3e98..4a956b2 100644 --- a/docs/decisioni-e-limiti.md +++ b/docs/decisioni-e-limiti.md @@ -354,10 +354,8 @@ Formato di ogni voce: **decisione** in grassetto, poi *Motivazione*, invariante che non dipende dal processo singolo è «al più un round attivo», garantito da un indice ([DEC-22]). - *Codice:* `app/db/base.py:_register_sqlite_pragmas` · `app/tx/locks.py:UserLocks` · `app/rounds/events.py:RoundEventBroadcaster` · - *Test:* `tests/unit/test_db_base.py` - *Stato:* da implementare (B-88) · *Oggi:* `synchronous=NORMAL` e `BEGIN` - emesso solo alla prima scrittura (modalità legacy di pysqlite). + *Codice:* `app/db/base.py:_register_sqlite_pragmas` · `app/db/base.py:begin_immediate` · `app/tx/locks.py:UserLocks` · `app/rounds/events.py:RoundEventBroadcaster` · + *Test:* `tests/unit/test_db_base.py` · `tests/unit/test_sqlite_transactions.py` ### Economia del gioco diff --git a/docs/modello-dati.md b/docs/modello-dati.md index 10f8c9c..a131364 100644 --- a/docs/modello-dati.md +++ b/docs/modello-dati.md @@ -483,11 +483,8 @@ restano nella tabella come storico. `busy_timeout` scaduto (`database is locked`), non un conflitto di snapshot: la scommessa è rifiutata allo stesso modo, perché nulla è ancora stato trasmesso. - *Codice:* `app/bets/service.py:place_bet` · *Test:* `tests/unit/test_bets.py` - *Stato:* da implementare (B-88) · *Oggi:* la transazione inizia solo alla - prima scrittura (modalità legacy di pysqlite), le letture precedenti non - sono isolate, e il codice tratta l'`OperationalError` come un conflitto di - snapshot, che in SQLite non può verificarsi. + *Codice:* `app/bets/service.py:place_bet` · *Test:* `tests/unit/test_bets.py` · + `tests/unit/test_sqlite_transactions.py` - **[DAT-17]** Al massimo **un round attivo** alla volta, come vincolo del database: `ix_rounds_single_active` è un indice unico sull'espressione @@ -1371,10 +1368,7 @@ letti con una scansione completa. dialetto non viene eseguito nulla. `PRAGMA foreign_keys` non è tra questi, da cui [DAT-04]. *Codice:* `app/db/base.py:_register_sqlite_pragmas` · - *Test:* `tests/unit/test_db_base.py` - *Stato:* da implementare (B-88) · *Oggi:* `synchronous=NORMAL`, con cui in - modalità WAL gli ultimi commit possono andare persi in caso di interruzione - di corrente. + *Test:* `tests/unit/test_db_base.py` · `tests/unit/test_sqlite_transactions.py` - **[DAT-74]** Ogni transazione del database che sposta denaro o ne fissa lo stato inizia con `BEGIN IMMEDIATE`, che prende il lock di scrittura di SQLite @@ -1383,22 +1377,28 @@ letti con una scansione completa. `pending_transactions`) e le scritture che ne seguono vedono lo stesso stato, senza che un altro scrittore si inserisca fra le due. Riguarda almeno le fasi di scrittura di `place_bet`, `request_withdrawal`, `_trigger_payout`, - `bump_fee`, l'accredito dei depositi, le transizioni di stato dello - scheduler, i gestori di conferma e il riconciliatore. Un `OperationalError` + `bump_fee`, l'accredito dei depositi (accredito, marcatura come speso + all'esterno, reintegro: quest'ultimo solo quando c'è qualcosa da reintegrare, + perché gira a ogni aggiornamento), le transizioni di stato dello scheduler, + i gestori di conferma, il riconciliatore e l'allocazione dell'indice di + derivazione alla registrazione (senza il lock due registrazioni simultanee + leggevano lo stesso massimo). Si ottiene con `begin_immediate(session)`, che + chiude l'eventuale transazione di sola lettura già aperta (di solito la + ricerca dell'utente autenticato) e apre quella nuova con `BEGIN IMMEDIATE`; + le altre transazioni, in sola lettura, restano `BEGIN` semplice e in WAL non + aspettano nessuno. Un `OperationalError` `database is locked` è il `busy_timeout` di [DAT-52] scaduto: prima di un broadcast la transazione viene annullata e l'operazione rifiutata o rimandata al giro successivo; **dopo** un broadcast non si annulla nulla sulla rete, la riga resta nello stato scritto prima del broadcast (`building`) e il riconciliatore la risolve ([transazioni-e-riconciliazione.md](transazioni-e-riconciliazione.md) B-88). - *Codice:* `app/db/base.py` · `app/db/session.py:get_session` · + *Codice:* `app/db/base.py:begin_immediate` · `app/db/base.py:_register_sqlite_pragmas` · `app/bets/service.py:place_bet` · `app/withdrawals/service.py:request_withdrawal` · - *Test:* — - *Stato:* da implementare (B-88) · *Oggi:* pysqlite in modalità legacy emette - `BEGIN` solo alla prima istruzione di scrittura, quindi le letture che la - precedono non sono nella stessa transazione delle scritture; un `database - is locked` può emergere dopo 5 s in un punto qualunque, anche dopo un - broadcast. + `app/rounds/scheduler.py:RoundScheduler._trigger_payout` · `app/tx/broadcast.py:bump_fee` · + `app/tx/confirmation.py:poll_once` · `app/tx/reconcile.py:reconcile_once` · + `app/deposits/service.py:credit_confirmed_utxos` · `app/auth/routes.py:register` · + *Test:* `tests/unit/test_sqlite_transactions.py` ## Migrazioni @@ -1537,13 +1537,12 @@ scrive la stringa vuota). per file; un secondo scrittore aspetta il lock fino al `busy_timeout` ([DAT-52]) e poi riceve `OperationalError: database is locked`. Non esistono conflitti fra snapshot: un `OperationalError` è sempre un'attesa scaduta. - Nello schema atteso le transazioni che spostano denaro prendono il lock con - `BEGIN IMMEDIATE` prima di leggere ([DAT-74]), quindi il compare-and-set di - [DAT-16] e le letture che lo precedono stanno nella stessa transazione. - Oggi pysqlite, in modalità legacy, emette `BEGIN` solo alla prima scrittura: - le letture precedenti non sono isolate, e il codice di `place_bet` descrive - l'`OperationalError` come un conflitto di snapshot, cosa che in SQLite non - accade (B-88). + Le transazioni che spostano denaro prendono il lock con `BEGIN IMMEDIATE` + prima di leggere ([DAT-74]), quindi il compare-and-set di [DAT-16] e le + letture che lo precedono stanno nella stessa transazione. Il driver + (pysqlite) non apre più le transazioni da solo (`isolation_level = None`): + è l'evento `begin` del motore a emettere `BEGIN` o `BEGIN IMMEDIATE`, così + ogni lettura sta dentro una transazione vera (B-88). - **Indice parziale su espressione.** `ix_rounds_single_active` usa sia un'espressione (`(1)`) sia una clausola `WHERE`; SQLite e PostgreSQL supportano entrambe, e il modello dichiara la clausola per tutti e due (`sqlite_where`, diff --git a/docs/scommesse-e-prelievi.md b/docs/scommesse-e-prelievi.md index 0e55fe9..bb16a84 100644 --- a/docs/scommesse-e-prelievi.md +++ b/docs/scommesse-e-prelievi.md @@ -108,10 +108,7 @@ ancora `open` oltre la scadenza. `service_busy` (HTTP 503), perché il round non è necessariamente in chiusura. *Codice:* `app/bets/service.py:place_bet` · - *Test:* `tests/unit/test_bets.py` - *Stato:* da implementare (B-88) · *Oggi:* la transazione si apre con la - prima scrittura, non con `BEGIN IMMEDIATE`; un `OperationalError` viene - trattato come un conflitto di scrittura e risponde `round_closing`. + *Test:* `tests/unit/test_bets.py` · `tests/unit/test_sqlite_transactions.py` - **[BET-22]** `MAX_PARTICIPANTS_PER_ROUND` è minore di `MAX_PAYOUT_TX_INPUTS`. Il pagamento spende una moneta del montepremi per ogni scommessa, e una singola transazione di pagamento può spenderne al diff --git a/docs/sicurezza.md b/docs/sicurezza.md index 1a7382a..0af3baa 100644 --- a/docs/sicurezza.md +++ b/docs/sicurezza.md @@ -818,8 +818,7 @@ sicurezza. - **Commit persi.** Con `synchronous=FULL` un commit confermato al chiamante è sul disco ([DAT-52](modello-dati.md), B-88): un'interruzione di corrente non può più perdere una registrazione e far riassegnare il suo indice di - derivazione. *Oggi:* `synchronous=NORMAL`, con cui in modalità WAL gli - ultimi commit possono andare persi. + derivazione. ## Dati conservati e log diff --git a/docs/storico-bug.md b/docs/storico-bug.md index 31c6862..b97db5c 100644 --- a/docs/storico-bug.md +++ b/docs/storico-bug.md @@ -236,7 +236,7 @@ cifrava di nuovo il file sbagliato (vedi [operativita.md](operativita.md)). | B-85 | Le riorganizzazioni della catena non vengono né seguite né segnalate | Alta | **aperto** | — | ELC, RND | | B-86 | Un server connesso ma fermo non viene mai sostituito | Alta | **aperto** | — | ELC, OPS | | B-87 | L'RBF può lasciare un resto sotto la soglia di dust | Alta | **aperto** | — | TX | -| B-88 | Le transazioni SQLite non sono quelle che il codice presuppone; `synchronous=NORMAL` può perdere gli ultimi commit | Alta | **aperto** | — | DAT, ARC | +| B-88 | Le transazioni SQLite non sono quelle che il codice presuppone; `synchronous=NORMAL` può perdere gli ultimi commit | Alta | *da registrare* | `test_sqlite_transactions.py`, `test_db_base.py` | DAT, ARC | | B-89 | Problemi della chiave rilevati solo al primo uso; gli script accettano qualunque xprv e scrivono file leggibili da tutti | Alta | **aperto** | — | SEC, OPS | | B-90 | Il ripristino di un backup riapre round già pagati e riassegna indici di derivazione | Alta | **aperto** | — | OPS, REG | | B-91 | Container eseguito come root, permessi larghi, dipendenze e immagini non fissate | Alta | **aperto** | — | SEC, OPS | @@ -252,8 +252,9 @@ cifrava di nuovo il file sbagliato (vedi [operativita.md](operativita.md)). - **Corretti:** B-01 … B-70, tranne B-16, che è chiuso nella sostanza (il crash è eliminato e `/guida` serve un segnaposto) ma lascia la guida vera - come limite accettato. + come limite accettato; dalla revisione del 2026-09-28: B-88. - **Aperti:** B-71, B-72 (gravità bassa, operativi o di documentazione) e - B-73 … B-98 dalla revisione del 2026-09-28 (10 critici, 9 alti, 7 medi), - da implementare nell'ordine di [piano-correzioni.md](piano-correzioni.md). + B-73 … B-98 dalla revisione del 2026-09-28 tranne i corretti (10 critici, + 8 alti, 7 medi), da implementare nell'ordine di + [piano-correzioni.md](piano-correzioni.md). - **Prossimo numero libero:** `B-99`. diff --git a/docs/transazioni-e-riconciliazione.md b/docs/transazioni-e-riconciliazione.md index c9c8b9c..7d451a3 100644 --- a/docs/transazioni-e-riconciliazione.md +++ b/docs/transazioni-e-riconciliazione.md @@ -86,12 +86,8 @@ stateDiagram-v2 riuscito le righe restano `building` e vale [TX-20]. La durabilità dei commit (`synchronous=FULL`) è descritta in [architettura.md](architettura.md) (B-88). - *Codice:* `app/db/base.py:_register_sqlite_pragmas`, `app/bets/service.py:place_bet`, `app/withdrawals/service.py:request_withdrawal` · - *Test:* — - *Stato:* da implementare (B-88) · *Oggi:* pysqlite in modalità legacy apre - la transazione solo alla prima scrittura, quindi le letture precedenti non - sono isolate; un'attesa del blocco oltre 5 secondi emerge come - `OperationalError` in un punto qualunque, anche dopo un invio riuscito. + *Codice:* `app/db/base.py:begin_immediate`, `app/bets/service.py:place_bet`, `app/withdrawals/service.py:request_withdrawal`, `app/rounds/scheduler.py:RoundScheduler._trigger_payout`, `app/tx/broadcast.py:bump_fee`, `app/tx/reconcile.py:reconcile_once` · + *Test:* `tests/unit/test_sqlite_transactions.py` ### Invio rifiutato diff --git a/tests/unit/test_db_base.py b/tests/unit/test_db_base.py index 3a172c8..7f7f3cc 100644 --- a/tests/unit/test_db_base.py +++ b/tests/unit/test_db_base.py @@ -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): diff --git a/tests/unit/test_sqlite_transactions.py b/tests/unit/test_sqlite_transactions.py new file mode 100644 index 0000000..92c033c --- /dev/null +++ b/tests/unit/test_sqlite_transactions.py @@ -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