diff --git a/BUGS.md b/BUGS.md index 2bf7f96..663f542 100644 --- a/BUGS.md +++ b/BUGS.md @@ -137,24 +137,6 @@ the intermediate headers (`blockchain.block.headers`) and validate the chain; on linkage mismatch walk back up to 10 headers to find the fork point and follow the reorg instead of ending the session (reorg handling: B-85). -### B-76 — An ambiguous broadcast is rolled back as if refused - -`place_bet`/`request_withdrawal` release everything on *any* broadcast exception, -including the 15s timeout and a drop after the write. The tx may have propagated: the -bet's sats land in the pool with no participant, the withdrawal leaves while its row -says `failed`. - -Fix: only an explicit JSON-RPC rejection can release, and only after verifying it: -`blockchain.transaction.get(txid)` is asked of the active server and every fallback -(10s each). Any server returning the tx → accepted; every responder denying it, with -the active server and the quorum minimum among them → released; too few answers → -uncertain (below). Matching "already known / already in mempool" messages is only a -shortcut that skips the check, so an unrecognised message can never release a tx -that is on the network (spec: TX-04). A timeout or disconnect leaves the rows -`building` for the reconciler and the API answers 202 with the row in `building`. -After the active server accepts, the raw tx is also sent to every fallback -(best effort). - ### B-77 — Only the latest txid is tracked, and `bump_fee` broadcasts before it persists The poller and the reconciler match `current_txid` only (`replaced_by_txid` keeps diff --git a/CLAUDE.md b/CLAUDE.md index 8547b1f..a5bd7f5 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 — 425 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 — 463 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. @@ -49,7 +49,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 425 tests +python -m pytest # all 463 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 ``` @@ -212,7 +212,7 @@ Both SPAs refresh on an `update` message *or* on `open` — the latter fires on Everything that spends money is written **before** it is broadcast and resolved against the chain afterwards; this is what makes the system recover without manual DB edits. `PendingTransaction.status`: `building` → `pending` → `confirmed`, or `failed`. - `building` is written first, UTXOs already marked `spent_txid`, and committed *before* the broadcast (`bets/service.py`, `withdrawals/service.py`, and `scheduler.py:_trigger_payout` — the same shape in four phases, so no DB session is ever held across a network call). A crash in that window leaves evidence, not coins spent on-chain with no record. -- 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. +- **Nothing is released on doubt (B-76).** `tx/outcome.py:broadcast_with_verdict` sorts a broadcast into accepted / refused / uncertain. Only an explicit JSON-RPC error (`ElectrumRpcError`) that `ElectrumListener.transaction_known` then confirms — the active server and the quorum minimum of the others all say they do not know the tx — releases the UTXOs, restores the balance, removes the participant (or marks the withdrawal `failed`), audit-logs, and raises `broadcast_failed` → **502**. If any server returns the tx, or the message is an "already known" one, it counts as accepted. A timeout, a dropped connection or a refusal that could not be verified leaves the rows `building` for the reconciler and the API answers **202**. An accepted tx is also fanned out to every other server (`fan_out_transaction`, best effort). - `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`). diff --git a/README.md b/README.md index 8816070..cbfc6d9 100644 --- a/README.md +++ b/README.md @@ -62,7 +62,7 @@ python -m pytest # all tests python -m pytest tests/unit/test_hd.py # one file ``` -425 unit tests cover HD derivation, PSBT building, the Electrum client, bets, +463 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 8ef7b2c..52a7d1f 100644 --- a/app/api/routes/bets.py +++ b/app/api/routes/bets.py @@ -1,4 +1,4 @@ -from fastapi import APIRouter, Depends, Request, status +from fastapi import APIRouter, Depends, Request, Response, status from pydantic import BaseModel from sqlalchemy.ext.asyncio import AsyncSession @@ -21,6 +21,7 @@ class BetResponse(BaseModel): @router.post("", response_model=BetResponse, status_code=status.HTTP_201_CREATED) async def create_bet( request: Request, + response: Response, user: User = Depends(get_current_user), session: AsyncSession = Depends(get_session), ) -> BetResponse: @@ -34,7 +35,7 @@ async def create_bet( async with request.app.state.user_locks.acquire(user.id): try: - participant = await place_bet(session, listener.client, user) + participant = await place_bet(session, listener.client, user, network=listener) except BetError as exc: # A rejected broadcast isn't the client's fault — it's the network # refusing our transaction, so it answers 502 rather than 400 (B-07). @@ -47,6 +48,11 @@ async def create_bet( ) raise from_api_error(code, exc) from exc + if participant.status == "building": + # B-76 (TX-20, BET-23): the outcome of the broadcast is uncertain — the tx may + # already be on the network — so the bet is accepted for processing, not + # confirmed as sent, and the reconciler settles it. + response.status_code = status.HTTP_202_ACCEPTED return BetResponse( round_id=participant.round_id, bet_txid=participant.bet_txid, diff --git a/app/api/routes/withdrawals.py b/app/api/routes/withdrawals.py index 36e38f9..f5c6494 100644 --- a/app/api/routes/withdrawals.py +++ b/app/api/routes/withdrawals.py @@ -1,4 +1,4 @@ -from fastapi import APIRouter, Depends, Request, status +from fastapi import APIRouter, Depends, Request, Response, status from pydantic import BaseModel from sqlalchemy.ext.asyncio import AsyncSession @@ -27,6 +27,7 @@ class WithdrawalResponse(BaseModel): async def create_withdrawal( body: WithdrawalRequest, request: Request, + response: Response, user: User = Depends(get_current_user), session: AsyncSession = Depends(get_session), ) -> WithdrawalResponse: @@ -41,7 +42,7 @@ async def create_withdrawal( async with request.app.state.user_locks.acquire(user.id): try: withdrawal = await request_withdrawal( - session, listener.client, user, body.external_address, body.amount_sats + session, listener.client, user, body.external_address, body.amount_sats, network=listener ) except WithdrawalError as exc: code = ( @@ -53,6 +54,10 @@ async def create_withdrawal( ) raise from_api_error(code, exc) from exc + if withdrawal.status == "building": + # B-76 (TX-20, WDR-16): uncertain broadcast — accepted for processing, the + # reconciler decides whether the transaction left. + response.status_code = status.HTTP_202_ACCEPTED return WithdrawalResponse( txid=withdrawal.txid, amount_requested_sats=withdrawal.amount_requested_sats, diff --git a/app/bets/service.py b/app/bets/service.py index f575e4b..9132e1f 100644 --- a/app/bets/service.py +++ b/app/bets/service.py @@ -14,6 +14,7 @@ from app.electrum.client import ElectrumClient from app.rounds.config import get_round_config from app.rounds.events import broadcaster from app.rounds.service import open_new_round_if_needed, round_accepts_bets +from app.tx.outcome import BroadcastOutcome, TransactionNetwork, broadcast_with_verdict from app.wallet.balance import recompute_balance from app.wallet.hd import derive_user_key from app.wallet.psbt_builder import ( @@ -37,7 +38,12 @@ class _RoundClosedDuringBuild(Exception): being built. Never leaves place_bet — it becomes a `round_closing` BetError.""" -async def place_bet(session: AsyncSession, client: ElectrumClient, user: User) -> RoundParticipant: +async def place_bet( + session: AsyncSession, client: ElectrumClient, user: User, network: TransactionNetwork | None = None +) -> RoundParticipant: + """`network` (the ElectrumListener) is what verifies a refused broadcast and fans an + accepted one out (B-76); without it a refusal can never be proven, so it never + releases anything.""" # 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 @@ -184,21 +190,29 @@ async def place_bet(session: AsyncSession, client: ElectrumClient, user: User) - raise _service_busy() from exc # --- Phase 2: broadcast, then promote both rows to their live state --------- - try: - await client.broadcast(built.raw_hex) - except Exception as exc: - # 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). + # B-76 (TX-04, TX-20): only a *verified* refusal undoes phase 1. A timeout, a + # dropped connection or a refusal nobody could verify leaves the tx possibly on the + # network, and releasing the coins then sent the bet's sats into the pool with no + # participant behind them — so those rows stay "building" for the reconciler. + verdict = await broadcast_with_verdict(client, network, built.raw_hex, built.txid) + if verdict.outcome is BroadcastOutcome.REFUSED: + # The network refused it (fee too low, dust, mempool conflict) and no server + # knows it — nothing is on-chain, so undo phase 1 completely and give the user + # a translatable failure instead of a bare 500 (B-07). try: - await _release_failed_bet(session, participant, pending, built, user.id, str(exc)) + await _release_failed_bet(session, participant, pending, built, user.id, verdict.reason) 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 + raise BetError("broadcast_failed", f"the network refused the transaction: {verdict.reason}") + if verdict.outcome is BroadcastOutcome.UNCERTAIN: + logger.warning("bet %s: broadcast outcome uncertain (%s), left 'building' for the reconciler", built.txid, verdict.reason) + await session.refresh(participant) # reports the status the DB really holds: building + broadcaster.publish() # an in-flight bet counts towards the round's figures + return participant # 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 diff --git a/app/electrum/client.py b/app/electrum/client.py index 7cce201..8b4dfb4 100644 --- a/app/electrum/client.py +++ b/app/electrum/client.py @@ -128,6 +128,20 @@ class ElectrumError(Exception): pass +class ElectrumRpcError(ElectrumError): + """The server *answered* with a JSON-RPC `error` object (B-76, TX-04). It is the + only failure that says anything about the request itself: a timeout, a dropped + connection or a write error (plain ElectrumError) leaves it unknown whether the + server ever acted on it. `args[0]` is the error object as the server sent it.""" + + @property + def message(self) -> str: + detail = self.args[0] if self.args else "" + if isinstance(detail, dict): + return str(detail.get("message", detail)) + return str(detail) + + class ElectrumClient: """Minimal asyncio Electrum protocol client: line-delimited JSON-RPC over TLS. @@ -266,6 +280,13 @@ class ElectrumClient: async def get_transaction(self, txid: str, verbose: bool = False) -> object: return await self.request("blockchain.transaction.get", [txid, verbose]) + async def get_merkle(self, txid: str, height: int) -> dict: + """The merkle branch proving `txid` is in the block at `height` (B-83, ELC-26).""" + return await self.request("blockchain.transaction.get_merkle", [txid, height]) + + async def get_block_header(self, height: int) -> str: + return await self.request("blockchain.block.header", [height]) + async def _read_loop(self) -> None: """Frame the stream into lines by hand rather than with StreamReader.readline: readline raises on a line over the limit, and that exception used to end this @@ -350,7 +371,7 @@ class ElectrumClient: if future.done(): return if message.get("error"): - future.set_exception(ElectrumError(message["error"])) + future.set_exception(ElectrumRpcError(message["error"])) else: future.set_result(message.get("result")) elif "method" in message: diff --git a/app/electrum/listener.py b/app/electrum/listener.py index 875e6f2..df21da7 100644 --- a/app/electrum/listener.py +++ b/app/electrum/listener.py @@ -2,6 +2,7 @@ import asyncio import logging from collections.abc import Callable +from embit.transaction import Transaction from sqlalchemy import select from sqlalchemy.ext.asyncio import async_sessionmaker @@ -14,7 +15,7 @@ from app.deposits.service import ( reinstate_reappeared_utxos, ) from app.config import MIN_QUORUM_RESPONSES -from app.electrum.client import ElectrumClient, ElectrumEndpoint +from app.electrum.client import ElectrumClient, ElectrumEndpoint, ElectrumRpcError from app.electrum.scripthash import address_to_scripthash from app.rounds.draw import ( HeaderValidationError, @@ -403,6 +404,99 @@ class ElectrumListener: agreements = sum(1 for result in responded if agrees(result)) return agreements * 2 > len(responded) + # --- Asking every server (B-76, TX-04, TX-21) --------------------------------- + + def _other_endpoints(self) -> list[ElectrumEndpoint]: + return [endpoint for endpoint in self._endpoints if endpoint != self.current_endpoint] + + def _minimum_other_answers(self) -> int: + """How many *other* servers must answer before their silence or agreement + counts as the network's: MIN_QUORUM_RESPONSES, or the reduced development rule + of ALLOW_NO_QUORUM (one answer, none if no other server is configured).""" + if not self._allow_no_quorum: + return MIN_QUORUM_RESPONSES + return 1 if self._other_endpoints() else 0 + + async def _ask_other_servers(self, ask: Callable[[ElectrumClient], "asyncio.Future"]) -> list: + """`ask(client)` on a fresh connection to every endpoint but the active one, in + parallel, each bounded by _CORROBORATION_TIMEOUT_SECONDS; a server that cannot be + reached or does not answer in time yields None. `ask` itself decides what an + answer is (it may return None for an unusable one).""" + + async def _one(endpoint: ElectrumEndpoint): + client = self._client_factory(endpoint) + try: + await asyncio.wait_for(client.connect(), timeout=_CORROBORATION_TIMEOUT_SECONDS) + return await asyncio.wait_for(ask(client), timeout=_CORROBORATION_TIMEOUT_SECONDS) + except Exception: + return None + finally: + await client.close() + + return list(await asyncio.gather(*(_one(endpoint) for endpoint in self._other_endpoints()))) + + async def _ask_active_server(self, ask: Callable[[ElectrumClient], "asyncio.Future"]): + client = self.client + if client is None: + return None + try: + return await asyncio.wait_for(ask(client), timeout=_CORROBORATION_TIMEOUT_SECONDS) + except Exception: + return None + + async def transaction_known(self, txid: str) -> bool | None: + """Does the network know `txid`? Asked of the active server and every other + configured one with `blockchain.transaction.get` (B-76, TX-04): + + * True — at least one server returns the transaction: it is on the network, + whatever the server that refused it said; + * False — every server that answered says it does not know it, and the active + server and the minimum number of the others (see _minimum_other_answers) are + among them: nobody has it; + * None — too few answers to tell. Nothing may be released on this. + """ + + async def _ask(client: ElectrumClient) -> bool | None: + try: + raw = await client.get_transaction(txid, False) + except ElectrumRpcError: + return False # an explicit "no such transaction" + except Exception: + return None + try: + if isinstance(raw, str) and Transaction.parse(bytes.fromhex(raw)).txid().hex() == txid: + return True + except Exception: + pass + return None # a reply that is not this transaction proves nothing either way + + active = await self._ask_active_server(_ask) + others = await self._ask_other_servers(_ask) + if active is True or any(answer is True for answer in others): + return True + answered_others = [answer for answer in others if answer is not None] + if active is False and len(answered_others) >= self._minimum_other_answers(): + return False + return None + + def fan_out_transaction(self, raw_tx_hex: str) -> None: + """Send an accepted transaction to every other configured server too (B-76, + TX-21): a second route onto the network in case the active server's own relay + is weak. Best effort and off the caller's path — a failure is only logged.""" + + async def _send() -> None: + async def _broadcast(client: ElectrumClient) -> str: + return await client.broadcast(raw_tx_hex) + + results = await self._ask_other_servers(_broadcast) + logger.info( + "fanned a transaction out to %s other server(s), %s answered", len(results), sum(r is not None for r in results) + ) + + task = asyncio.create_task(_send()) + self._background_tasks.add(task) + task.add_done_callback(self._background_tasks.discard) + async def corroborate_header(self, height: int, expected_hash: str) -> bool: """B-28: is `expected_hash` — the header our own active connection reported for `height` — corroborated by other configured servers before diff --git a/app/rounds/scheduler.py b/app/rounds/scheduler.py index 48644ae..eb51589 100644 --- a/app/rounds/scheduler.py +++ b/app/rounds/scheduler.py @@ -17,6 +17,7 @@ from app.rounds.config import get_round_config from app.rounds.draw import draw_winner, header_hex_to_block_hash from app.rounds.events import broadcaster from app.rounds.service import open_new_round_if_needed, round_deadline, winner_share +from app.tx.outcome import BroadcastOutcome, broadcast_with_verdict from app.wallet.hd import derive_round_pool_key from app.wallet.plm_network import PLM_MAINNET from app.wallet.psbt_builder import InsufficientFundsError, Utxo, build_payout_transaction @@ -496,13 +497,14 @@ class RoundScheduler: pending_id = pending.id # --- Phase 4: broadcast, then promote the pending row -------------------- - try: - await client.broadcast(built.raw_hex) - except Exception: - # The row stays "building": the reconciler will ask the chain about it - # and, finding nothing, abandon it and clear payout_txid (B-25) — instead - # of the round being stuck with a payout_txid that never went anywhere. - logger.exception("round %s payout broadcast failed", round_id) + # B-76 (TX-05): the same three outcomes as a bet's broadcast. Accepted — the + # server took it, or refused with "already known" / a server of the network + # holds it — promotes the row; a refusal and an uncertain outcome both leave it + # "building" with a payout_failed entry, since which of the two it really was + # is the reconciler's call, made against the chain (TX-16, TX-17). + verdict = await broadcast_with_verdict(client, self._listener, built.raw_hex, built.txid) + if verdict.outcome is not BroadcastOutcome.ACCEPTED: + logger.error("round %s payout broadcast not accepted (%s): %s", round_id, verdict.outcome.value, verdict.reason) await self._log_payout_failure(round_id, winner_user_id, "broadcast rejected") return diff --git a/app/tx/outcome.py b/app/tx/outcome.py new file mode 100644 index 0000000..b8c6373 --- /dev/null +++ b/app/tx/outcome.py @@ -0,0 +1,101 @@ +"""What a broadcast attempt amounted to (B-76, TX-04, TX-20, TX-21). + +Money may only be released on proof that a transaction never left. The text of an +Electrum error cannot give that proof (servers word "already in the mempool" in many +ways and there is no standard code), and a timeout or a dropped connection says +nothing about whether the server acted before it went quiet. So a broadcast ends in +one of three ways, and only one of them frees coins: + + ACCEPTED the server answered with the txid, or refused but at least one server of + the network (the active one or any other) returns the transaction; + REFUSED the server answered with an explicit JSON-RPC error *and* the active + server plus the quorum minimum of the others all say they do not know the + transaction; + UNCERTAIN anything else — a timeout, a dropped connection, a write error, a + refusal whose verification could not get enough answers. +""" + +import logging +from enum import Enum +from typing import NamedTuple, Protocol + +from app.electrum.client import ElectrumRpcError + +logger = logging.getLogger(__name__) + +# Messages by which a node says it already has the transaction. Recognising one skips +# the verification below; *not* recognising a message never releases anything, it only +# costs one more round of questions (TX-04). +_ALREADY_KNOWN_MARKERS = ( + "already known", + "already in mempool", + "already in the mempool", + "txn-already-known", + "txn-already-in-mempool", + "already in block chain", + "already in blockchain", +) + + +class BroadcastOutcome(Enum): + ACCEPTED = "accepted" + REFUSED = "refused" + UNCERTAIN = "uncertain" + + +class BroadcastVerdict(NamedTuple): + outcome: BroadcastOutcome + reason: str # the server's own wording or the failure's, for logs and audit rows + + +class TransactionNetwork(Protocol): + """The part of ElectrumListener a broadcast needs (see electrum/listener.py).""" + + async def transaction_known(self, txid: str) -> bool | None: ... + + def fan_out_transaction(self, raw_tx_hex: str) -> None: ... + + +def is_already_known(message: str) -> bool: + lowered = message.lower() + return any(marker in lowered for marker in _ALREADY_KNOWN_MARKERS) + + +async def broadcast_with_verdict(client, network: TransactionNetwork | None, raw_tx_hex: str, txid: str) -> BroadcastVerdict: + """Broadcast `raw_tx_hex` through the active `client` and say what happened. + + With no `network` to ask, an explicit refusal cannot be verified and therefore + comes back UNCERTAIN: the doubt never frees coins. An accepted transaction is + sent to every other server too (best effort, TX-21).""" + try: + await client.broadcast(raw_tx_hex) + except ElectrumRpcError as exc: + reason = exc.message + if is_already_known(reason): + known: bool | None = True + elif network is None: + known = None + else: + known = await network.transaction_known(txid) + if known is True: + _fan_out(network, raw_tx_hex) + return BroadcastVerdict(BroadcastOutcome.ACCEPTED, reason) + if known is False: + return BroadcastVerdict(BroadcastOutcome.REFUSED, reason) + logger.warning("broadcast of %s was refused (%s) but the refusal could not be verified", txid, reason) + return BroadcastVerdict(BroadcastOutcome.UNCERTAIN, reason) + except Exception as exc: + # Timeout, dropped connection, write error: the server may have relayed it. + logger.warning("broadcast of %s ended without an answer (%r): outcome uncertain", txid, exc) + return BroadcastVerdict(BroadcastOutcome.UNCERTAIN, str(exc) or type(exc).__name__) + _fan_out(network, raw_tx_hex) + return BroadcastVerdict(BroadcastOutcome.ACCEPTED, "") + + +def _fan_out(network: TransactionNetwork | None, raw_tx_hex: str) -> None: + if network is None: + return + try: + network.fan_out_transaction(raw_tx_hex) + except Exception: + logger.exception("could not start the fan-out of an accepted transaction") diff --git a/app/withdrawals/service.py b/app/withdrawals/service.py index 5906eef..88f400b 100644 --- a/app/withdrawals/service.py +++ b/app/withdrawals/service.py @@ -12,6 +12,7 @@ from app.db.models import PendingTransaction, Round, User, UtxoEvent, Withdrawal from app.electrum.client import ElectrumClient from app.rounds.config import get_round_config from app.rounds.events import broadcaster +from app.tx.outcome import BroadcastOutcome, TransactionNetwork, broadcast_with_verdict from app.wallet.address import is_valid_plm_address from app.wallet.balance import compute_pending_balance, recompute_balance from app.wallet.hd import derive_pool_address, derive_user_key @@ -31,7 +32,12 @@ class WithdrawalError(ApiError): async def request_withdrawal( - session: AsyncSession, client: ElectrumClient, user: User, external_address: str, amount_sats: int + session: AsyncSession, + client: ElectrumClient, + user: User, + external_address: str, + amount_sats: int, + network: TransactionNetwork | None = None, ) -> Withdrawal: # Checked before anything else: an address from another chain parses fine as a # witness program (see wallet/address.py), so without this the tx would build, @@ -162,11 +168,14 @@ async def request_withdrawal( session.add(pending) await session.commit() - try: - await client.broadcast(built.raw_hex) - except Exception as exc: + # B-76 (TX-04, TX-20): only a verified refusal releases the coins. A timeout, a + # drop after the write, or a refusal nobody could verify may mean the transaction + # left: the withdrawal would then be paid out while its row said "failed" and its + # coins were spendable again. Those outcomes leave the rows "building". + verdict = await broadcast_with_verdict(client, network, built.raw_hex, built.txid) + if verdict.outcome is BroadcastOutcome.REFUSED: try: - await _release_failed_withdrawal(session, withdrawal, pending, built, user.id, str(exc)) + await _release_failed_withdrawal(session, withdrawal, pending, built, user.id, verdict.reason) except OperationalError: # B-88: the release could not be written (database busy). The rows stay # "building", the state committed before the broadcast, and the @@ -175,9 +184,16 @@ async def request_withdrawal( 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 + raise WithdrawalError("broadcast_failed", f"the network refused the transaction: {verdict.reason}") + if verdict.outcome is BroadcastOutcome.UNCERTAIN: + logger.warning( + "withdrawal %s: broadcast outcome uncertain (%s), left 'building' for the reconciler", + built.txid, + verdict.reason, + ) + await session.refresh(withdrawal) + broadcaster.publish() + return withdrawal # 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. diff --git a/docs/adr/ADR-31-tutto-cio-spende-denaro-scrive-prima.md b/docs/adr/ADR-31-tutto-cio-spende-denaro-scrive-prima.md index abe5b34..428a309 100644 --- a/docs/adr/ADR-31-tutto-cio-spende-denaro-scrive-prima.md +++ b/docs/adr/ADR-31-tutto-cio-spende-denaro-scrive-prima.md @@ -1,6 +1,6 @@ # ADR-31 — Tutto ciò che spende denaro si scrive prima di inviarlo -- **Stato:** Accettata — implementazione da completare (B-76, B-77) +- **Stato:** Accettata — implementazione da completare (B-77) - **Area:** Denaro in movimento - **Fonte di verità per:** la scelta di progetto descritta qui sotto; i requisiti (`PREFISSO-NN`) che la attuano stanno nei documenti di dominio, indicati nelle *Conseguenze*. @@ -19,7 +19,6 @@ riconciliatore e risponde **202** ([ADR-51](ADR-51-dubbio-liberano-mai-fondi.md) [transazioni-e-riconciliazione.md](../transazioni-e-riconciliazione.md). *Codice:* `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_bets.py::test_bet_is_persisted_before_it_is_broadcast`, `tests/unit/test_scheduler.py::test_trigger_payout_persists_before_broadcasting`, `tests/unit/test_reconcile.py` -*Stato:* da implementare (B-76, B-77) · *Oggi:* qualunque eccezione -dell'invio, timeout compreso, annulla tutto con 502; `bump_fee` invia la -sostituzione prima di salvarla. +*Test:* `tests/unit/test_bets.py::test_bet_is_persisted_before_it_is_broadcast`, `tests/unit/test_broadcast_outcome.py`, `tests/unit/test_scheduler.py::test_trigger_payout_persists_before_broadcasting`, `tests/unit/test_reconcile.py` +*Stato:* da implementare (B-77), per le sostituzioni RBF · *Oggi:* +`bump_fee` invia la sostituzione prima di salvarla. diff --git a/docs/adr/ADR-51-dubbio-liberano-mai-fondi.md b/docs/adr/ADR-51-dubbio-liberano-mai-fondi.md index 48709a1..9ba0ca2 100644 --- a/docs/adr/ADR-51-dubbio-liberano-mai-fondi.md +++ b/docs/adr/ADR-51-dubbio-liberano-mai-fondi.md @@ -1,6 +1,6 @@ # ADR-51 — Nel dubbio non si liberano mai i fondi -- **Stato:** Accettata — implementazione da completare (B-76, B-77, B-78) +- **Stato:** Accettata — implementazione da completare (B-77, B-78) - **Area:** Denaro in movimento - **Fonte di verità per:** la scelta di progetto descritta qui sotto; i requisiti (`PREFISSO-NN`) che la attuano stanno nei documenti di dominio, indicati nelle *Conseguenze*. @@ -36,8 +36,8 @@ pausa la lotteria ([ADR-28](ADR-28-pausa-manutenzione-ferma-round-successivo.md) congelato e visibile, mai duplicato. *Codice:* `app/tx/reconcile.py:reconcile_once` · `app/tx/reconcile.py:_is_due` · `app/bets/service.py:place_bet` · `app/withdrawals/service.py:request_withdrawal` · `app/tx/confirmation.py:poll_once` · `app/tx/broadcast.py:bump_fee` · -*Test:* — -*Stato:* da implementare (B-76, B-77, B-78) · *Oggi:* l'abbandono avviene +*Test:* `tests/unit/test_broadcast_outcome.py` +*Stato:* da implementare (B-77, B-78) · *Oggi:* l'abbandono avviene quando la cronologia del solo server attivo non contiene il txid corrente, dopo 120s per `building` e 6 ore per `pending`; si segue solo l'ultimo txid; le righe `failed` non sono più controllate. diff --git a/docs/api.md b/docs/api.md index 87fe814..095a65b 100644 --- a/docs/api.md +++ b/docs/api.md @@ -220,10 +220,7 @@ esiste una API separata per l'amministrazione, solo un'autenticazione diversa. "già noto / già nel mempool" vale come successo (`201`). *Codice:* `app/api/routes/bets.py:create_bet`, `app/api/routes/withdrawals.py:create_withdrawal` · - *Test:* `tests/unit/test_bets.py` - *Stato:* da implementare (B-76) · *Oggi:* qualunque eccezione del broadcast, - compresi il timeout di 15 s e la caduta della connessione, annulla tutto e - risponde `502 broadcast_failed`. + *Test:* `tests/unit/test_bets.py`, `tests/unit/test_broadcast_outcome.py` - **[API-14]** Qualunque eccezione non gestita risponde `500` con lo stesso formato strutturato, codice `internal_error`, messaggio @@ -471,9 +468,7 @@ esiste una API separata per l'amministrazione, solo un'autenticazione diversa. *Codice:* `app/api/routes/bets.py:create_bet`, `app/bets/service.py:place_bet` · *Test:* `tests/unit/test_bets.py` - *Stato:* da implementare (B-76, B-95, B-96) · *Oggi:* ogni esito diverso - dal successo è `502 broadcast_failed` con tutto annullato; la trasmissione - va solo al server attivo; un costo coperto solo da resto non confermato + *Stato:* da implementare (B-95, B-96) · *Oggi:* un costo coperto solo da resto non confermato risponde `insufficient_balance`; non c'è tetto al montepremi; il costo è quello configurato al momento della scommessa, non quello del round. @@ -521,9 +516,7 @@ esiste una API separata per l'amministrazione, solo un'autenticazione diversa. *Codice:* `app/api/routes/withdrawals.py:create_withdrawal`, `app/withdrawals/service.py:request_withdrawal` · *Test:* `tests/unit/test_withdrawals.py` - *Stato:* da implementare (B-76, B-96) · *Oggi:* ogni esito diverso - dal successo è `502 broadcast_failed` con il prelievo `failed`; la - trasmissione va solo al server attivo; l'importo ridotto di un prelievo + *Stato:* da implementare (B-96) · *Oggi:* l'importo ridotto di un prelievo dell'intero saldo non è riconfrontato con il minimo. ### Round diff --git a/docs/audit-log.md b/docs/audit-log.md index cd1c85b..6bd1cc3 100644 --- a/docs/audit-log.md +++ b/docs/audit-log.md @@ -352,9 +352,7 @@ flowchart LR 200 caratteri. `user_id` sì, `round_id` **no** (a differenza di `bet_placed`). Stessa transazione del ripristino. *Codice:* `app/bets/service.py:_release_failed_bet` · - *Test:* `tests/unit/test_bets.py` - *Stato:* da implementare (B-76) · *Oggi:* scritto per qualunque errore - d'invio, compresi timeout e disconnessioni. + *Test:* `tests/unit/test_bets.py`, `tests/unit/test_broadcast_outcome.py` - **[AUD-21]** `withdrawal_sent` — la rete ha accettato la transazione di un prelievo. Payload `{"txid", "amount_sent_sats", "external_address"}`: @@ -373,9 +371,7 @@ flowchart LR (`reason` troncato a 200 caratteri). `user_id` sì, `round_id` no. Stessa transazione del ripristino. *Codice:* `app/withdrawals/service.py:_release_failed_withdrawal` · - *Test:* nessun test verifica la riga di audit. - *Stato:* da implementare (B-76) · *Oggi:* scritto per qualunque errore - d'invio, compresi timeout e disconnessioni. + *Test:* `tests/unit/test_broadcast_outcome.py` ### Round diff --git a/docs/connessione-electrum.md b/docs/connessione-electrum.md index bade6c5..5884f25 100644 --- a/docs/connessione-electrum.md +++ b/docs/connessione-electrum.md @@ -349,7 +349,8 @@ L'**invio** di una transazione non è una corroborazione ma usa gli stessi server: dopo che il server attivo l'ha accettata, la transazione grezza viene inviata anche a ogni server di riserva, al meglio. La regola è in [transazioni-e-riconciliazione.md](transazioni-e-riconciliazione.md) (B-76, -B-83); oggi l'invio va al solo server attivo. +B-83); oggi l'invio va anche a tutti gli altri server, salvo la sostituzione +RBF di `bump_fee` (B-77). ## Riconnessione diff --git a/docs/modello-dati.md b/docs/modello-dati.md index b0dfe4d..7e1198f 100644 --- a/docs/modello-dati.md +++ b/docs/modello-dati.md @@ -392,10 +392,9 @@ del database sono tutti uguali, perché arrivano tutti dallo stesso percorso *Codice:* le funzioni in tabella · *Test:* `tests/unit/test_deposits.py`, `tests/unit/test_bets.py`, `tests/unit/test_withdrawals.py`, `tests/unit/test_reconcile.py` - *Stato:* da implementare (B-76, B-78, B-83) · *Oggi:* il rilascio dopo il - broadcast avviene su qualunque eccezione (anche un timeout), l'abbandono - sul silenzio del solo server attivo, il reintegro sulla parola del solo - server attivo. + *Stato:* da implementare (B-78, B-83) · *Oggi:* l'abbandono avviene sul + silenzio del solo server attivo, il reintegro sulla parola del solo server + attivo. - **[DAT-13]** Il riconciliatore libera un input solo se il suo `spent_txid` è uno dei txid della riga che sta abbandonando (`current_txid` o una sua @@ -975,11 +974,9 @@ Le funzioni citate sono `app/bets/service.py:place_bet`, *Test:* `tests/unit/test_confirmation.py`, `tests/unit/test_reconcile.py`, `tests/unit/test_scheduler.py`, `tests/unit/test_bets.py`, `tests/unit/test_withdrawals.py` - *Stato:* da implementare (B-76, B-78) · *Oggi:* qualunque eccezione del - broadcast di una scommessa o di un prelievo è trattata come un rifiuto; - l'abbandono avviene quando la cronologia del solo server attivo non - contiene `current_txid`; il poller guarda solo le righe `pending` e una - riga `failed` non viene più guardata. + *Stato:* da implementare (B-78) · *Oggi:* l'abbandono avviene quando la + cronologia del solo server attivo non contiene `current_txid`; il poller + guarda solo le righe `pending` e una riga `failed` non viene più guardata. - **[DAT-34]** Dopo la creazione, `current_txid`, `raw_tx_hex`, `fee_rate_sat_vb`, `replaced_by_txid`, `attempt_count` e `last_broadcast_at` @@ -1157,8 +1154,8 @@ dall'utente. ([DAT-72]), il prelievo resta `failed` finché l'operatore non lo riallinea. *Codice:* le funzioni in tabella · *Test:* `tests/unit/test_withdrawals.py`, `tests/unit/test_reconcile.py` - *Stato:* da implementare (B-76, B-78) · *Oggi:* qualunque eccezione del - broadcast porta a `failed`, e l'abbandono non richiede una prova. + *Stato:* da implementare (B-78) · *Oggi:* l'abbandono non richiede una + prova. - **[DAT-40]** `amount_requested_sats` è ciò che l'utente ha chiesto; `amount_sent_sats` è l'output effettivo verso il destinatario: l'importo diff --git a/docs/scommesse-e-prelievi.md b/docs/scommesse-e-prelievi.md index ce0825a..6e10186 100644 --- a/docs/scommesse-e-prelievi.md +++ b/docs/scommesse-e-prelievi.md @@ -203,11 +203,7 @@ flowchart TB registro di audit, notifica SSE; la risposta è `broadcast_failed` (HTTP 502). *Codice:* `app/bets/service.py:_release_failed_bet`, `app/api/routes/bets.py:create_bet` · - *Test:* `tests/unit/test_bets.py` - *Stato:* da implementare (B-76) · *Oggi:* tutto viene annullato per - qualunque errore dell'invio, anche un timeout o una connessione caduta - dopo la scrittura, quando la transazione può essere già in rete: i suoi - sat arrivano al montepremi senza partecipante. + *Test:* `tests/unit/test_bets.py`, `tests/unit/test_broadcast_outcome.py` - **[BET-23]** Se l'invio ha un **esito incerto** ([TX-20](transazioni-e-riconciliazione.md#invio-rifiutato)), partecipante e transazione in corso restano `building` con le monete ancora segnate come @@ -216,9 +212,7 @@ flowchart TB limite di [BET-06] e tiene il round in `closing` finché il riconciliatore non lo promuove o lo abbandona. *Codice:* `app/bets/service.py:place_bet`, `app/api/routes/bets.py:create_bet` · - *Test:* — - *Stato:* da implementare (B-76) · *Oggi:* un esito incerto è trattato come - un rifiuto ([BET-16]). + *Test:* `tests/unit/test_broadcast_outcome.py` - **[BET-17]** Se l'invio è accettato — anche con un rifiuto "già nota" ([TX-04](transazioni-e-riconciliazione.md#invio-rifiutato)) — il partecipante passa a `broadcast`, la transazione in corso a `pending`, si @@ -229,9 +223,7 @@ flowchart TB scommessa promossa dal riconciliatore scrive `bet_placed` in quel momento ([TX-16](transazioni-e-riconciliazione.md#riconciliazione)). *Codice:* `app/bets/service.py:place_bet`, `app/api/routes/bets.py:create_bet` · - *Test:* `tests/unit/test_bets.py` - *Stato:* da implementare (B-76) · *Oggi:* un "già nota" è trattato come un - rifiuto e la transazione va solo al server attivo. + *Test:* `tests/unit/test_bets.py`, `tests/unit/test_broadcast_outcome.py` - **[BET-18]** Alla conferma il partecipante passa a `confirmed` ([TX-07](transazioni-e-riconciliazione.md#conferma)). Solo le scommesse **confermate** partecipano all'estrazione; il round, allo scadere, aspetta @@ -324,12 +316,7 @@ solo se la rete rifiuta esplicitamente, RBF se non conferma in tempo. `failure_reason`); si scrive `withdrawal_broadcast_failed`, si invia una notifica SSE e la risposta è `broadcast_failed` (HTTP 502). *Codice:* `app/withdrawals/service.py:_release_failed_withdrawal`, `app/api/routes/withdrawals.py:create_withdrawal` · - *Test:* `tests/unit/test_withdrawals.py` (nessun test verifica l'evento - `withdrawal_broadcast_failed`) - *Stato:* da implementare (B-76) · *Oggi:* tutto viene annullato per - qualunque errore dell'invio, anche un timeout o una connessione caduta - dopo la scrittura: la transazione può essere partita mentre il prelievo - risulta `failed` e le monete di nuovo spendibili. + *Test:* `tests/unit/test_withdrawals.py`, `tests/unit/test_broadcast_outcome.py` - **[WDR-16]** Se l'invio ha un **esito incerto** ([TX-20](transazioni-e-riconciliazione.md#invio-rifiutato)), prelievo e transazione in corso restano `building` con le monete ancora segnate come @@ -337,9 +324,7 @@ solo se la rete rifiuta esplicitamente, RBF se non conferma in tempo. `amount_sent_sats` e `status` = `building`; sarà il riconciliatore a promuoverlo o abbandonarlo. *Codice:* `app/withdrawals/service.py:request_withdrawal`, `app/api/routes/withdrawals.py:create_withdrawal` · - *Test:* — - *Stato:* da implementare (B-76) · *Oggi:* un esito incerto è trattato come - un rifiuto ([WDR-11]). + *Test:* `tests/unit/test_broadcast_outcome.py` - **[WDR-12]** Se l'invio è accettato — anche con un rifiuto "già nota" ([TX-04](transazioni-e-riconciliazione.md#invio-rifiutato)) — il prelievo passa a `broadcast`, la transazione in corso a `pending`, si scrive @@ -350,9 +335,7 @@ solo se la rete rifiuta esplicitamente, RBF se non conferma in tempo. `status`. Alla conferma il prelievo passa a `confirmed` ([TX-07](transazioni-e-riconciliazione.md#conferma)). *Codice:* `app/withdrawals/service.py:request_withdrawal`, `app/withdrawals/confirmation.py:_on_withdrawal_confirmed` · - *Test:* `tests/unit/test_withdrawals.py` - *Stato:* da implementare (B-76) · *Oggi:* un "già nota" è trattato come un - rifiuto e la transazione va solo al server attivo. + *Test:* `tests/unit/test_withdrawals.py`, `tests/unit/test_broadcast_outcome.py` ## Scommessa e prelievo insieme diff --git a/docs/storico-bug.md b/docs/storico-bug.md index b4407cb..d05c50c 100644 --- a/docs/storico-bug.md +++ b/docs/storico-bug.md @@ -224,7 +224,7 @@ cifrava di nuovo il file sbagliato (vedi [operativita.md](operativita.md)). | B-73 | L'estrazione dipende solo dall'hash del blocco: un minatore con biglietti o il server Electrum attivo possono pilotarla; ordine dei partecipanti non verificabile | Critica | **aperto** | — | RND, ELC | | B-74 | Header accettati senza controllo della difficoltà minima; un salto di più blocchi non viene collegato; una riorganizzazione di un blocco chiude la sessione | Critica | **aperto** | — | ELC | | B-75 | Una risposta Electrum oltre 64 KiB chiude la connessione, e la cronologia dell'indirizzo del montepremi la supera dopo ~800 transazioni | Critica | *da registrare* | `test_electrum_client.py`, `test_deposits.py`, `test_psbt_builder.py` | ELC, TX | -| B-76 | Un invio ambiguo (timeout, connessione caduta) viene annullato come se fosse rifiutato | Critica | **aperto** | — | TX, BET, WDR | +| B-76 | Un invio ambiguo (timeout, connessione caduta) viene annullato come se fosse rifiutato | Critica | *da registrare* | `test_broadcast_outcome.py`, `test_electrum_client.py` | TX, BET, WDR | | B-77 | Si ricorda solo l'ultimo txid di una transazione accelerata, e la sostituzione viene inviata prima di essere salvata | Critica | **aperto** | — | TX | | B-78 | Una transazione viene abbandonata sulla base del silenzio di un solo server | Critica | **aperto** | — | TX, ELC | | B-79 | Il pagamento di norma non ha resto, quindi non si può accelerare e può bloccare la lotteria | Critica | **aperto** | — | RND, TX | @@ -252,9 +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; dalla revisione del 2026-09-28: B-88, B-75, B-97, B-80, B-84. + come limite accettato; dalla revisione del 2026-09-28: B-88, B-75, B-97, B-80, B-84, B-76. - **Aperti:** B-71, B-72 (gravità bassa, operativi o di documentazione) e - B-73 … B-98 dalla revisione del 2026-09-28 tranne i corretti (8 critici, + B-73 … B-98 dalla revisione del 2026-09-28 tranne i corretti (7 critici, 7 alti, 6 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 472eb1e..6907c7a 100644 --- a/docs/transazioni-e-riconciliazione.md +++ b/docs/transazioni-e-riconciliazione.md @@ -127,12 +127,8 @@ monete: `transaction already in block chain`) resta solo una scorciatoia che salta la verifica: un messaggio non riconosciuto non può più far liberare le monete di una transazione partita, al massimo costa una verifica in più. - *Codice:* `app/bets/service.py:_release_failed_bet`, `app/withdrawals/service.py:_release_failed_withdrawal`, `app/electrum/client.py:ElectrumClient.broadcast`, `app/electrum/client.py:ElectrumClient.get_transaction` · - *Test:* `tests/unit/test_bets.py`, `tests/unit/test_withdrawals.py` - *Stato:* da implementare (B-76) · *Oggi:* qualunque eccezione dell'invio, - compresi il timeout di 15 secondi, la connessione caduta dopo la scrittura - e il "già nota", annulla tutto e risponde 502, anche se la transazione può - essere già in rete. + *Codice:* `app/tx/outcome.py:broadcast_with_verdict`, `app/electrum/client.py:ElectrumRpcError`, `app/electrum/listener.py:ElectrumListener.transaction_known`, `app/bets/service.py:_release_failed_bet`, `app/withdrawals/service.py:_release_failed_withdrawal` · + *Test:* `tests/unit/test_broadcast_outcome.py`, `tests/unit/test_electrum_client.py`, `tests/unit/test_bets.py`, `tests/unit/test_withdrawals.py` - **[TX-20]** **Invio incerto.** Se l'invio di una scommessa o di un prelievo finisce senza una risposta del server (timeout, connessione caduta, errore di scrittura), o se il commit della promozione successiva a un invio @@ -142,21 +138,18 @@ monete: [TX-17]). L'utente riceve HTTP **202** con l'oggetto nello stato `building` ([BET-23](scommesse-e-prelievi.md#cosa-succede), [WDR-16](scommesse-e-prelievi.md#prelievo)) e si invia una notifica SSE. - *Codice:* `app/bets/service.py:place_bet`, `app/withdrawals/service.py:request_withdrawal` · - *Test:* — - *Stato:* da implementare (B-76) · *Oggi:* l'invio incerto è trattato come - un rifiuto ([TX-04]); un commit fallito dopo un invio accettato solleva un - errore 500 lasciando le righe `building`. + *Codice:* `app/bets/service.py:place_bet`, `app/withdrawals/service.py:request_withdrawal`, `app/api/routes/bets.py:create_bet`, `app/api/routes/withdrawals.py:create_withdrawal` · + *Test:* `tests/unit/test_broadcast_outcome.py`, `tests/unit/test_sqlite_transactions.py` - **[TX-21]** **Invio a tutti i server.** Dopo che il server attivo ha accettato una transazione ([TX-04]) — scommessa, prelievo, pagamento o sostituzione RBF — la stessa transazione viene inviata anche a ogni altro server configurato. È un tentativo in più, non una condizione: un errore di questi invii viene solo registrato nei log, non cambia lo stato della riga e non ritarda la risposta all'utente. - *Codice:* `app/electrum/listener.py:ElectrumListener`, `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` · - *Test:* — - *Stato:* da implementare (B-76) · *Oggi:* ogni transazione viene inviata - solo al server attivo. + *Codice:* `app/electrum/listener.py:ElectrumListener.fan_out_transaction`, `app/tx/outcome.py:broadcast_with_verdict`, `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` · + *Test:* `tests/unit/test_broadcast_outcome.py` + *Stato:* da implementare (B-77), per la sostituzione RBF · *Oggi:* la + sostituzione di `bump_fee` viene inviata solo al server attivo. - **[TX-05]** Se l'invio di un pagamento non è accettato — rifiuto esplicito o esito incerto — la riga resta `building` e si scrive `payout_failed` (`broadcast rejected`): è il riconciliatore a decidere, interrogando la @@ -164,10 +157,7 @@ monete: "già nota" ([TX-04]) vale come invio accettato: la riga passa a `pending` e si scrive `payout_sent`. *Codice:* `app/rounds/scheduler.py:RoundScheduler._trigger_payout` · - *Test:* `tests/unit/test_scheduler.py` - *Stato:* da implementare (B-76) · *Oggi:* anche il "già nota" lascia la - riga `building` con `payout_failed`, finché il riconciliatore non la - promuove. + *Test:* `tests/unit/test_scheduler.py`, `tests/unit/test_broadcast_outcome.py` ## Conferma diff --git a/tests/unit/fake_network.py b/tests/unit/fake_network.py new file mode 100644 index 0000000..2b1baae --- /dev/null +++ b/tests/unit/fake_network.py @@ -0,0 +1,26 @@ +"""Test doubles for what the services ask of the network (B-76 and after).""" + +from app.electrum.client import ElectrumRpcError + + +def rpc_error(message: str, code: int = -26) -> ElectrumRpcError: + """A server that answered with a JSON-RPC `error` object — an explicit refusal.""" + return ElectrumRpcError({"code": code, "message": message}) + + +class FakeNetwork: + """The ElectrumListener surface a broadcast needs: `known` is what + `transaction_known` answers (True: some server has it, False: every server denies + it, None: too few answers to tell); fanned-out transactions are recorded.""" + + def __init__(self, known: bool | None = False): + self.known = known + self.asked: list[str] = [] + self.fanned_out: list[str] = [] + + async def transaction_known(self, txid: str) -> bool | None: + self.asked.append(txid) + return self.known + + def fan_out_transaction(self, raw_tx_hex: str) -> None: + self.fanned_out.append(raw_tx_hex) diff --git a/tests/unit/test_bets.py b/tests/unit/test_bets.py index afdfe90..a1a0914 100644 --- a/tests/unit/test_bets.py +++ b/tests/unit/test_bets.py @@ -12,6 +12,7 @@ from app.rounds.events import broadcaster from app.rounds.service import open_new_round_if_needed from app.wallet.hd import derive_user_address from app.wallet.psbt_builder import MAX_PARTICIPANTS_PER_ROUND, MAX_TX_INPUTS +from tests.unit.fake_network import FakeNetwork, rpc_error _FEE_ADDRESS = "plm1q5x25wd6q463mfhckjraaedgjg0lyu73qfcj43n" @@ -250,7 +251,7 @@ class RejectingElectrumClient: conflict, or simply an unreachable server.""" async def broadcast(self, raw_tx_hex: str) -> str: - raise RuntimeError("min relay fee not met") + raise rpc_error("min relay fee not met") async def test_failed_broadcast_leaves_nothing_behind(session_factory): @@ -262,7 +263,7 @@ async def test_failed_broadcast_leaves_nothing_behind(session_factory): async with session_factory() as session: user = await session.get(User, user_id) with pytest.raises(BetError, match="refused"): - await place_bet(session, RejectingElectrumClient(), user) + await place_bet(session, RejectingElectrumClient(), user, network=FakeNetwork(known=False)) async with session_factory() as session: utxo = (await session.scalars(select(UtxoEvent).where(UtxoEvent.user_id == user_id))).one() @@ -296,7 +297,7 @@ async def test_failed_broadcast_publishes_an_sse_update(session_factory): # B-4 async with session_factory() as session: user = await session.get(User, user_id) with pytest.raises(BetError, match="refused"): - await place_bet(session, RejectingElectrumClient(), user) + await place_bet(session, RejectingElectrumClient(), user, network=FakeNetwork(known=False)) assert not queue.empty() finally: @@ -309,7 +310,7 @@ async def test_failed_broadcast_reports_the_broadcast_failed_code(session_factor async with session_factory() as session: user = await session.get(User, user_id) try: - await place_bet(session, RejectingElectrumClient(), user) + await place_bet(session, RejectingElectrumClient(), user, network=FakeNetwork(known=False)) assert False, "expected BetError" except BetError as exc: assert exc.code == "broadcast_failed" diff --git a/tests/unit/test_broadcast_outcome.py b/tests/unit/test_broadcast_outcome.py new file mode 100644 index 0000000..357fa9f --- /dev/null +++ b/tests/unit/test_broadcast_outcome.py @@ -0,0 +1,543 @@ +"""B-76 (TX-04, TX-20, TX-21, TX-05, BET-16, BET-17, BET-23, WDR-11, WDR-12, WDR-16, API-09): +an ambiguous broadcast is never rolled back as if it had been refused. + +Only an explicit JSON-RPC refusal that the whole network then confirms it does not know +may release coins. A timeout, a dropped connection, or a refusal nobody could verify +leaves the rows `building` for the reconciler. Each of these tests fails against the +old code, where *any* exception from the broadcast undid everything. +""" + +import asyncio +from types import SimpleNamespace + +import pytest +from fastapi import Response +from sqlalchemy import select + +from app.bets.service import BetError, place_bet +from app.db.models import AuditLog, PendingTransaction, RoundParticipant, User, UtxoEvent, Withdrawal +from app.electrum.client import ElectrumEndpoint, ElectrumError +from app.electrum.listener import ElectrumListener +from app.rounds.events import broadcaster +from app.rounds.scheduler import RoundScheduler +from app.tx.outcome import BroadcastOutcome, broadcast_with_verdict +from app.withdrawals.service import WithdrawalError, request_withdrawal +from tests.unit.fake_network import FakeNetwork, rpc_error +from tests.unit.test_bets import _make_funded_user, session_factory # noqa: F401 (fixture) +from tests.unit.test_scheduler import ( # noqa: F401 (fixtures and helpers shared with the payout tests) + FakePayoutClient, + FakePayoutListener, + _POOL_AMOUNT_SATS, + _seed_paying_out_round, + payout_session_factory, +) + +_EXTERNAL = "plm1qqph9qup2mp7w7g5nlsdhdc9m2pp44ampzw0ctx" +_FUNDS = 3_000_000_000 + + +class _FailingClient: + """A client whose broadcast fails the way `exc` says.""" + + def __init__(self, exc: BaseException): + self._exc = exc + self.broadcasted: list[str] = [] + + async def broadcast(self, raw_tx_hex: str) -> str: + self.broadcasted.append(raw_tx_hex) + raise self._exc + + +_UNCERTAIN_FAILURES = [ + pytest.param(ElectrumError("blockchain.transaction.broadcast timed out after 15s"), id="timeout-after-write"), + pytest.param(asyncio.TimeoutError(), id="bare-timeout"), + pytest.param(ElectrumError("connection closed"), id="disconnect"), + pytest.param(ElectrumError("write failed: broken pipe"), id="write-error"), + pytest.param(ConnectionResetError("reset by peer"), id="connection-reset"), +] + + +async def _bet_rows(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.scalars(select(UtxoEvent).where(UtxoEvent.user_id == user_id))).one() + events = [e.event_type for e in (await session.scalars(select(AuditLog))).all()] + return participants, pendings, utxo, events + + +# --- bets ---------------------------------------------------------------------- + + +@pytest.mark.parametrize("failure", _UNCERTAIN_FAILURES) +async def test_a_bet_whose_broadcast_ends_without_an_answer_is_left_building(session_factory, failure): + """TX-20 / BET-23: nothing is released, the player gets the `building` participant, + and the network is never asked to verify (there was no refusal to verify).""" + user_id = await _make_funded_user(session_factory, 4, _FUNDS) + network = FakeNetwork(known=False) # would say "release" if asked — it must not be + queue = broadcaster.subscribe() + try: + while not queue.empty(): + queue.get_nowait() + async with session_factory() as session: + user = await session.get(User, user_id) + participant = await place_bet(session, _FailingClient(failure), user, network=network) + assert participant.status == "building" + assert not queue.empty() # SSE + finally: + broadcaster.unsubscribe(queue) + + participants, pendings, utxo, events = await _bet_rows(session_factory, user_id) + assert [p.status for p in participants] == ["building"] + assert [p.status for p in pendings] == ["building"] + assert utxo.spent_txid == participants[0].bet_txid # the coins stay committed to the tx + assert "bet_broadcast_failed" not in events and "bet_placed" not in events + assert network.asked == [] and network.fanned_out == [] + async with session_factory() as session: + assert (await session.get(User, user_id)).cached_balance_sats == 0 # not given back + + +async def test_a_refused_bet_that_another_server_knows_counts_as_accepted(session_factory): + """TX-04: the active server refused ("txn-mempool-conflict", whatever it said) but a + server of the network returns the transaction — it is on the network, so the bet + stands and is promoted. Releasing here sent the sats to the pool without a participant.""" + user_id = await _make_funded_user(session_factory, 4, _FUNDS) + network = FakeNetwork(known=True) + client = _FailingClient(rpc_error("some wording we have never seen")) + async with session_factory() as session: + user = await session.get(User, user_id) + participant = await place_bet(session, client, user, network=network) + + assert participant.status == "broadcast" + participants, pendings, utxo, events = await _bet_rows(session_factory, user_id) + assert [p.status for p in pendings] == ["pending"] + assert "bet_placed" in events and "bet_broadcast_failed" not in events + assert network.asked == [participants[0].bet_txid] + assert len(network.fanned_out) == 1 # TX-21: sent to the other servers too + + +async def test_a_refused_bet_nobody_knows_is_released(session_factory): + """TX-04 / BET-16: a verified refusal still undoes everything and answers 502.""" + user_id = await _make_funded_user(session_factory, 4, _FUNDS) + network = FakeNetwork(known=False) + async with session_factory() as session: + user = await session.get(User, user_id) + with pytest.raises(BetError) as excinfo: + await place_bet(session, _FailingClient(rpc_error("min relay fee not met")), user, network=network) + assert excinfo.value.code == "broadcast_failed" + participants, pendings, utxo, events = await _bet_rows(session_factory, user_id) + assert participants == [] and pendings == [] and utxo.spent_txid is None + assert "bet_broadcast_failed" in events + assert network.fanned_out == [] + + +async def test_a_refusal_the_network_cannot_verify_releases_nothing(session_factory): + """TX-04: too few servers answered the verification — the outcome is uncertain, so + the bet stays `building` instead of being rolled back on the refusal's word alone.""" + user_id = await _make_funded_user(session_factory, 4, _FUNDS) + async with session_factory() as session: + user = await session.get(User, user_id) + participant = await place_bet( + session, _FailingClient(rpc_error("min relay fee not met")), user, network=FakeNetwork(known=None) + ) + assert participant.status == "building" + participants, pendings, utxo, events = await _bet_rows(session_factory, user_id) + assert [p.status for p in pendings] == ["building"] and utxo.spent_txid == participants[0].bet_txid + assert "bet_broadcast_failed" not in events + + +async def test_a_refusal_with_no_network_to_verify_it_releases_nothing(session_factory): + user_id = await _make_funded_user(session_factory, 4, _FUNDS) + async with session_factory() as session: + user = await session.get(User, user_id) + participant = await place_bet(session, _FailingClient(rpc_error("min relay fee not met")), user) + assert participant.status == "building" + + +@pytest.mark.parametrize( + "message", + [ + "txn-already-known", + "txn-already-in-mempool", + "Transaction already in block chain", + "the transaction was rejected by network rules.\n\ntxn-already-in-mempool", + ], +) +async def test_already_known_is_a_shortcut_that_skips_the_verification(session_factory, message): + """TX-04: a recognised "already known" message counts as accepted without asking + anyone; it is a shortcut, never a condition of safety.""" + user_id = await _make_funded_user(session_factory, 4, _FUNDS) + network = FakeNetwork(known=False) + async with session_factory() as session: + user = await session.get(User, user_id) + participant = await place_bet(session, _FailingClient(rpc_error(message)), user, network=network) + assert participant.status == "broadcast" + assert network.asked == [] + assert len(network.fanned_out) == 1 + + +async def test_the_bet_route_answers_202_for_an_uncertain_outcome_and_201_for_success(session_factory): + """API-09 / BET-23: `building` is HTTP 202; a promoted bet keeps 201.""" + from contextlib import asynccontextmanager + + from app.api.routes.bets import create_bet + + class _Locks: + @asynccontextmanager + async def acquire(self, user_id): + yield + + def _request(client, network): + listener = SimpleNamespace(client=client, **{"transaction_known": network.transaction_known, + "fan_out_transaction": network.fan_out_transaction}) + return SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(electrum_listener=listener, user_locks=_Locks()))) + + first = await _make_funded_user(session_factory, 4, _FUNDS) + second = await _make_funded_user(session_factory, 5, _FUNDS) + + response = Response() + async with session_factory() as session: + user = await session.get(User, first) + body = await create_bet( + _request(_FailingClient(ElectrumError("connection closed")), FakeNetwork()), response, user=user, session=session + ) + assert response.status_code == 202 and body.status == "building" + + class _OkClient: + async def broadcast(self, raw_tx_hex): + return "txid" + + response = Response() + async with session_factory() as session: + user = await session.get(User, second) + body = await create_bet(_request(_OkClient(), FakeNetwork()), response, user=user, session=session) + assert response.status_code != 202 and body.status == "broadcast" # the route's declared 201 stands + + +# --- withdrawals ----------------------------------------------------------------- + + +async def _withdrawal_rows(session_factory, user_id): + async with session_factory() as session: + withdrawals = (await session.scalars(select(Withdrawal))).all() + pendings = (await session.scalars(select(PendingTransaction))).all() + utxo = (await session.scalars(select(UtxoEvent).where(UtxoEvent.user_id == user_id))).one() + events = [e.event_type for e in (await session.scalars(select(AuditLog))).all()] + return withdrawals, pendings, utxo, events + + +@pytest.mark.parametrize("failure", _UNCERTAIN_FAILURES) +async def test_a_withdrawal_whose_broadcast_ends_without_an_answer_is_left_building(session_factory, failure): + """TX-20 / WDR-16: the transaction may have left; the withdrawal must not be + marked failed with its coins spendable again (the money would go out twice).""" + user_id = await _make_funded_user(session_factory, 7, _FUNDS) + network = FakeNetwork(known=False) + async with session_factory() as session: + user = await session.get(User, user_id) + withdrawal = await request_withdrawal( + session, _FailingClient(failure), user, _EXTERNAL, 1_000_000_000, network=network + ) + assert withdrawal.status == "building" + withdrawals, pendings, utxo, events = await _withdrawal_rows(session_factory, user_id) + assert [w.status for w in withdrawals] == ["building"] + assert [p.status for p in pendings] == ["building"] + assert utxo.spent_txid == withdrawals[0].txid + assert "withdrawal_broadcast_failed" not in events and "withdrawal_sent" not in events + assert network.asked == [] + + +async def test_a_refused_withdrawal_that_another_server_knows_counts_as_accepted(session_factory): + user_id = await _make_funded_user(session_factory, 7, _FUNDS) + network = FakeNetwork(known=True) + async with session_factory() as session: + user = await session.get(User, user_id) + withdrawal = await request_withdrawal( + session, _FailingClient(rpc_error("unrecognised")), user, _EXTERNAL, 1_000_000_000, network=network + ) + assert withdrawal.status == "broadcast" + withdrawals, pendings, utxo, events = await _withdrawal_rows(session_factory, user_id) + assert [p.status for p in pendings] == ["pending"] and "withdrawal_sent" in events + assert len(network.fanned_out) == 1 + + +async def test_a_refused_withdrawal_nobody_knows_is_released_and_marked_failed(session_factory): + user_id = await _make_funded_user(session_factory, 7, _FUNDS) + async with session_factory() as session: + user = await session.get(User, user_id) + with pytest.raises(WithdrawalError) as excinfo: + await request_withdrawal( + session, + _FailingClient(rpc_error("min relay fee not met")), + user, + _EXTERNAL, + 1_000_000_000, + network=FakeNetwork(known=False), + ) + assert excinfo.value.code == "broadcast_failed" + withdrawals, pendings, utxo, events = await _withdrawal_rows(session_factory, user_id) + assert [w.status for w in withdrawals] == ["failed"] and utxo.spent_txid is None + assert "withdrawal_broadcast_failed" in events + + +async def test_an_unverifiable_withdrawal_refusal_releases_nothing(session_factory): + user_id = await _make_funded_user(session_factory, 7, _FUNDS) + async with session_factory() as session: + user = await session.get(User, user_id) + withdrawal = await request_withdrawal( + session, _FailingClient(rpc_error("x")), user, _EXTERNAL, 1_000_000_000, network=FakeNetwork(known=None) + ) + assert withdrawal.status == "building" + + +async def test_the_withdrawal_route_answers_202_for_an_uncertain_outcome(session_factory): + from contextlib import asynccontextmanager + + from app.api.routes.withdrawals import WithdrawalRequest, create_withdrawal + + class _Locks: + @asynccontextmanager + async def acquire(self, user_id): + yield + + user_id = await _make_funded_user(session_factory, 7, _FUNDS) + network = FakeNetwork() + listener = SimpleNamespace( + client=_FailingClient(ElectrumError("connection closed")), + transaction_known=network.transaction_known, + fan_out_transaction=network.fan_out_transaction, + ) + request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(electrum_listener=listener, user_locks=_Locks()))) + response = Response() + async with session_factory() as session: + user = await session.get(User, user_id) + body = await create_withdrawal( + WithdrawalRequest(external_address=_EXTERNAL, amount_sats=1_000_000_000), request, response, user=user, session=session + ) + assert response.status_code == 202 and body.status == "building" + + +# --- payouts --------------------------------------------------------------------- + + +def _payout_entries(): + return [{"tx_hash": "44" * 32, "tx_pos": 0, "height": 10, "value": _POOL_AMOUNT_SATS + 100_000}] + + +class _RaisingPayoutClient(FakePayoutClient): + def __init__(self, entries, exc): + super().__init__(entries) + self._exc = exc + + async def broadcast(self, raw_tx_hex): + raise self._exc + + +@pytest.mark.parametrize( + "exc, known", + [ + pytest.param(ElectrumError("connection closed"), False, id="uncertain"), + pytest.param(rpc_error("min relay fee not met"), False, id="verified-refusal"), + pytest.param(rpc_error("min relay fee not met"), None, id="unverifiable-refusal"), + ], +) +async def test_a_payout_that_is_not_accepted_stays_building_with_a_payout_failed_entry(payout_session_factory, exc, known): + """TX-05: refusal or uncertainty alike leave the row `building` — the reconciler, + asking the chain, decides whether it left (TX-16, TX-17).""" + await _seed_paying_out_round(payout_session_factory) + scheduler = RoundScheduler(payout_session_factory, FakePayoutListener(_RaisingPayoutClient(_payout_entries(), exc), known=known)) + + await scheduler._trigger_payout(1) + + async with payout_session_factory() as session: + pending = (await session.scalars(select(PendingTransaction))).one() + assert pending.status == "building" + events = [e.event_type for e in (await session.scalars(select(AuditLog))).all()] + assert "payout_failed" in events and "payout_sent" not in events + + +async def test_a_payout_refused_as_already_known_is_promoted(payout_session_factory): + """TX-05: "already known" is an accepted broadcast — `pending` + `payout_sent`.""" + await _seed_paying_out_round(payout_session_factory) + listener = FakePayoutListener(_RaisingPayoutClient(_payout_entries(), rpc_error("txn-already-in-mempool"))) + await RoundScheduler(payout_session_factory, listener)._trigger_payout(1) + + async with payout_session_factory() as session: + pending = (await session.scalars(select(PendingTransaction))).one() + assert pending.status == "pending" + events = [e.event_type for e in (await session.scalars(select(AuditLog))).all()] + assert "payout_sent" in events + assert len(listener.fanned_out) == 1 + + +async def test_an_accepted_payout_is_fanned_out_to_the_other_servers(payout_session_factory): + """TX-21.""" + await _seed_paying_out_round(payout_session_factory) + client = FakePayoutClient(_payout_entries()) + listener = FakePayoutListener(client) + await RoundScheduler(payout_session_factory, listener)._trigger_payout(1) + assert listener.fanned_out == client.broadcasted + + +# --- the verdict itself ------------------------------------------------------------- + + +async def test_the_verdict_distinguishes_the_three_outcomes(): + class _Ok: + async def broadcast(self, raw): + return "txid" + + ok = await broadcast_with_verdict(_Ok(), FakeNetwork(), "00", "aa") + refused = await broadcast_with_verdict(_FailingClient(rpc_error("bad")), FakeNetwork(known=False), "00", "aa") + uncertain = await broadcast_with_verdict(_FailingClient(ElectrumError("timed out")), FakeNetwork(known=False), "00", "aa") + assert (ok.outcome, refused.outcome, uncertain.outcome) == ( + BroadcastOutcome.ACCEPTED, + BroadcastOutcome.REFUSED, + BroadcastOutcome.UNCERTAIN, + ) + + +# --- the listener: transaction_known / fan_out_transaction ---------------------------- + + +_ENDPOINTS = [ + ElectrumEndpoint("active.example", 50002, True), + ElectrumEndpoint("first.example", 50002, True), + ElectrumEndpoint("second.example", 50002, True), + ElectrumEndpoint("third.example", 50002, True), +] + + +def _real_tx(): + """A real serialised transaction and its txid — transaction_known checks the + returned bytes hash to the txid asked about.""" + from embit.transaction import Transaction, TransactionInput, TransactionOutput + from embit import script + + tx = Transaction( + vin=[TransactionInput(bytes.fromhex("11" * 32), 0)], + vout=[TransactionOutput(1000, script.p2wpkh(__import__("embit").ec.PrivateKey(b"\x01" * 32).get_public_key()))], + ) + return tx.serialize().hex(), tx.txid().hex() + + +class _ServerClient: + """A fake connection to one server: `answer` is a raw tx hex (it knows the tx), the + string "unknown" (an RPC error: no such transaction), or None (unreachable).""" + + def __init__(self, answer, log): + self._answer = answer + self._log = log + + async def connect(self): + if self._answer is None: + raise ConnectionRefusedError("down") + + async def get_transaction(self, txid, verbose=False): + if self._answer == "unknown": + raise rpc_error("No such mempool or blockchain transaction", code=-5) + return self._answer + + async def broadcast(self, raw): + self._log.append(raw) + return "txid" + + async def close(self): + pass + + +def _listener(session_factory, answers: dict, active_answer, log=None, allow_no_quorum=False, endpoints=None): + log = log if log is not None else [] + listener = ElectrumListener( + lambda endpoint: _ServerClient(answers[endpoint.host], log), + session_factory, + endpoints or _ENDPOINTS, + allow_no_quorum=allow_no_quorum, + ) + listener._endpoint_index = 0 + listener.client = None if active_answer is Ellipsis else _ServerClient(active_answer, log) + return listener + + +async def test_transaction_known_is_true_when_any_server_returns_it(session_factory): + raw, txid = _real_tx() + listener = _listener( + session_factory, + {"first.example": "unknown", "second.example": raw, "third.example": None}, + active_answer="unknown", + ) + assert await listener.transaction_known(txid) is True + + +async def test_transaction_known_is_true_when_only_the_active_server_has_it(session_factory): + raw, txid = _real_tx() + listener = _listener( + session_factory, + {"first.example": "unknown", "second.example": "unknown", "third.example": "unknown"}, + active_answer=raw, + ) + assert await listener.transaction_known(txid) is True + + +async def test_transaction_known_is_false_only_when_the_active_server_and_a_quorum_deny_it(session_factory): + _, txid = _real_tx() + listener = _listener( + session_factory, + {"first.example": "unknown", "second.example": "unknown", "third.example": None}, + active_answer="unknown", + ) + assert await listener.transaction_known(txid) is False + + +async def test_transaction_known_is_uncertain_with_too_few_answers(session_factory): + """Only one of the other servers answered: below MIN_QUORUM_RESPONSES, so the + denial proves nothing.""" + _, txid = _real_tx() + listener = _listener( + session_factory, + {"first.example": "unknown", "second.example": None, "third.example": None}, + active_answer="unknown", + ) + assert await listener.transaction_known(txid) is None + + +async def test_transaction_known_is_uncertain_when_the_active_server_does_not_answer(session_factory): + _, txid = _real_tx() + listener = _listener( + session_factory, + {"first.example": "unknown", "second.example": "unknown", "third.example": "unknown"}, + active_answer=Ellipsis, # no active connection + ) + assert await listener.transaction_known(txid) is None + + +async def test_transaction_known_ignores_a_reply_that_is_not_the_transaction(session_factory): + """A server answering with some other transaction does not make this one known — + and does not count as a denial either.""" + listener = _listener( + session_factory, + {"first.example": "unknown", "second.example": "00", "third.example": "unknown"}, + active_answer="unknown", + ) + assert await listener.transaction_known("ab" * 32) is False # 2 real denials, 1 ignored reply + + +async def test_transaction_known_follows_the_reduced_rule_under_allow_no_quorum(session_factory): + _, txid = _real_tx() + single = [ElectrumEndpoint("active.example", 50002, True)] + listener = _listener(session_factory, {}, active_answer="unknown", allow_no_quorum=True, endpoints=single) + assert await listener.transaction_known(txid) is False + strict = _listener(session_factory, {}, active_answer="unknown", endpoints=single) + assert await strict.transaction_known(txid) is None + + +async def test_fan_out_sends_to_every_other_server_and_never_to_the_active_one(session_factory): + log: list[str] = [] + listener = _listener( + session_factory, + {"first.example": "unknown", "second.example": "unknown", "third.example": None}, + active_answer="unknown", + log=log, + ) + listener.fan_out_transaction("deadbeef") + await asyncio.gather(*listener._background_tasks) + assert log == ["deadbeef", "deadbeef"] # first + second; third unreachable; active untouched diff --git a/tests/unit/test_electrum_client.py b/tests/unit/test_electrum_client.py index 28a2ce8..5078be4 100644 --- a/tests/unit/test_electrum_client.py +++ b/tests/unit/test_electrum_client.py @@ -380,3 +380,36 @@ async def test_main_builds_clients_with_the_endpoint_pin(): assert _make_electrum_client(ElectrumEndpoint("h", 1, True, _PIN)).pin == _PIN assert _make_electrum_client(ElectrumEndpoint("h", 1, True)).pin is None + + +async def test_only_a_server_error_object_is_an_explicit_refusal(): + """B-76 / TX-04: a JSON-RPC `error` reply is an ElectrumRpcError carrying the + server's wording (string or object); a timeout or a dropped connection is a plain + ElectrumError — they say nothing about whether the server acted.""" + from app.electrum import client as client_module + from app.electrum.client import ElectrumRpcError + + client, reader, writer = await _client_with_fake_transport() + task = asyncio.create_task(client.request("blockchain.transaction.broadcast", ["00"])) + await asyncio.sleep(0) + sent = json.loads(writer.written.decode()) + reader.feed_data((json.dumps({"id": sent["id"], "error": {"code": -26, "message": "txn-mempool-conflict"}}) + "\n").encode()) + try: + await task + assert False, "expected ElectrumRpcError" + except ElectrumRpcError as exc: + assert exc.message == "txn-mempool-conflict" + client._read_task.cancel() + + original = client_module._REQUEST_TIMEOUT_SECONDS + client_module._REQUEST_TIMEOUT_SECONDS = 0.05 + try: + silent, _reader, _writer = await _client_with_fake_transport() + try: + await silent.request("blockchain.transaction.broadcast", ["00"]) + assert False, "expected ElectrumError" + except ElectrumError as exc: + assert not isinstance(exc, ElectrumRpcError) + silent._read_task.cancel() + finally: + client_module._REQUEST_TIMEOUT_SECONDS = original diff --git a/tests/unit/test_scheduler.py b/tests/unit/test_scheduler.py index 1aea8ba..d6ac9e7 100644 --- a/tests/unit/test_scheduler.py +++ b/tests/unit/test_scheduler.py @@ -7,6 +7,7 @@ from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.config import settings from app.db.base import Base +from tests.unit.fake_network import FakeNetwork from app.db.models import AuditLog, PendingTransaction, Round, RoundConfig, RoundParticipant, User from app.rounds.scheduler import RoundScheduler, _reserved_payout_outpoints from app.wallet.hd import derive_pool_address @@ -87,8 +88,12 @@ class FakePayoutClient: return "network-txid" -class FakePayoutListener: - def __init__(self, client): +class FakePayoutListener(FakeNetwork): + """The listener as the scheduler sees it: the active `client`, plus the network + surface a broadcast verdict uses (B-76).""" + + def __init__(self, client, known: bool | None = False): + super().__init__(known=known) self.client = client diff --git a/tests/unit/test_sqlite_transactions.py b/tests/unit/test_sqlite_transactions.py index 92c033c..abb9e6a 100644 --- a/tests/unit/test_sqlite_transactions.py +++ b/tests/unit/test_sqlite_transactions.py @@ -12,6 +12,7 @@ from contextlib import asynccontextmanager from types import SimpleNamespace import pytest +from fastapi import Response from cryptography.fernet import Fernet from httpx import ASGITransport, AsyncClient from sqlalchemy import func, select, text @@ -20,6 +21,7 @@ from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.bets.service import BetError, place_bet from app.config import settings +from tests.unit.fake_network import FakeNetwork, rpc_error from app.db import base as db_base from app.db.base import Base, begin_immediate from app.db.models import ( @@ -112,7 +114,7 @@ class _Client: if self._on_broadcast is not None: await self._on_broadcast() if self._fail: - raise RuntimeError("mempool conflict") + raise rpc_error("mempool conflict") return "txid" @@ -353,7 +355,7 @@ async def test_place_bet_refused_broadcast_with_a_busy_database_leaves_the_rows_ async with session_factory() as session: user = await session.get(User, funded) with pytest.raises(BetError) as excinfo: - await place_bet(session, client, user) + await place_bet(session, client, user, network=FakeNetwork(known=False)) finally: await holder_cm.__aexit__(None, None, None) @@ -369,7 +371,7 @@ async def test_place_bet_refused_broadcast_still_releases_when_the_database_is_f async with session_factory() as session: user = await session.get(User, funded) with pytest.raises(BetError) as excinfo: - await place_bet(session, client, user) + await place_bet(session, client, user, network=FakeNetwork(known=False)) assert excinfo.value.code == "broadcast_failed" participants, pendings, utxo = await _bet_state(session_factory, funded) assert participants == [] and pendings == [] and utxo.spent_txid is None @@ -475,7 +477,9 @@ async def test_withdrawal_refused_broadcast_with_a_busy_database_leaves_the_rows 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) + await request_withdrawal( + session, client, user, _EXTERNAL_ADDRESS, 1_000_000_000, network=FakeNetwork(known=False) + ) finally: await holder_cm.__aexit__(None, None, None) @@ -491,7 +495,9 @@ async def test_withdrawal_refused_broadcast_still_releases_when_the_database_is_ 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) + await request_withdrawal( + session, client, user, _EXTERNAL_ADDRESS, 1_000_000_000, network=FakeNetwork(known=False) + ) withdrawals, pendings, utxo = await _withdrawal_state(session_factory, funded) assert [w.status for w in withdrawals] == ["failed"] assert [p.status for p in pendings] == ["failed"] @@ -521,7 +527,7 @@ async def test_bet_route_answers_503_for_service_busy(session_factory, funded): 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) + await create_bet(_fake_request(_Client()), Response(), user=user, session=session) assert excinfo.value.status_code == 503 assert excinfo.value.detail["code"] == "service_busy" @@ -536,7 +542,7 @@ async def test_withdrawal_route_answers_503_for_service_busy(session_factory, fu 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) + await create_withdrawal(body, _fake_request(_Client()), Response(), user=user, session=session) assert excinfo.value.status_code == 503 assert excinfo.value.detail["code"] == "service_busy" diff --git a/tests/unit/test_withdrawals.py b/tests/unit/test_withdrawals.py index 47954e0..1282b74 100644 --- a/tests/unit/test_withdrawals.py +++ b/tests/unit/test_withdrawals.py @@ -7,6 +7,7 @@ from app.config import settings from app.db.base import Base from app.db.models import PendingTransaction, RoundConfig, User, UtxoEvent, Withdrawal from app.rounds.events import broadcaster +from tests.unit.fake_network import FakeNetwork, rpc_error from app.wallet.hd import derive_user_address from app.withdrawals.service import WithdrawalError, request_withdrawal @@ -190,7 +191,7 @@ async def test_failed_broadcast_publishes_an_sse_update(session_factory): # B-4 class RejectingClient: async def broadcast(self, raw_tx_hex: str) -> str: - raise RuntimeError("min relay fee not met") + raise rpc_error("min relay fee not met") queue = broadcaster.subscribe() try: @@ -200,7 +201,7 @@ async def test_failed_broadcast_publishes_an_sse_update(session_factory): # B-4 async with session_factory() as session: user = await session.get(User, user_id) with pytest.raises(WithdrawalError, match="refused"): - await request_withdrawal(session, RejectingClient(), user, derive_user_address(98), 1_000_000_000) + await request_withdrawal(session, RejectingClient(), user, derive_user_address(98), 1_000_000_000, network=FakeNetwork(known=False)) assert not queue.empty() finally: @@ -214,13 +215,13 @@ async def test_failed_broadcast_marks_the_withdrawal_failed_and_frees_the_coins( class RejectingClient: async def broadcast(self, raw_tx_hex: str) -> str: - raise RuntimeError("min relay fee not met") + raise rpc_error("min relay fee not met") async with session_factory() as session: user = await session.get(User, user_id) external = derive_user_address(99) with pytest.raises(WithdrawalError, match="refused"): - await request_withdrawal(session, RejectingClient(), user, external, 1_000_000_000) + await request_withdrawal(session, RejectingClient(), user, external, 1_000_000_000, network=FakeNetwork(known=False)) async with session_factory() as session: withdrawal = (await session.scalars(select(Withdrawal))).one()