From 2ec6b2720092511ffcd80a9b774612076571a965 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Fri, 31 Jul 2026 02:10:00 +0200 Subject: [PATCH] fix: resolve mint fallback review comments --- .env.example | 3 + docs/provider/configuration.md | 9 + routstr/balance.py | 4 +- routstr/lightning.py | 52 +- routstr/mint.py | 338 ++++++++++ routstr/payment/helpers.py | 12 - routstr/payment/lnurl.py | 66 +- routstr/proxy.py | 5 - routstr/wallet.py | 613 ++++++------------ .../integration/test_insufficient_balance.py | 31 +- tests/integration/test_swap_fee_retry.py | 7 +- tests/unit/test_fetch_all_balances.py | 6 +- tests/unit/test_lightning_settlement.py | 2 + tests/unit/test_lnurl_melt_timeout.py | 144 ++-- tests/unit/test_melt_reconciliation.py | 91 +++ tests/unit/test_mint.py | 65 ++ tests/unit/test_payment_helpers.py | 16 +- tests/unit/test_stale_reservations.py | 2 +- tests/unit/test_upstream_rate_limit.py | 2 +- tests/unit/test_wallet.py | 101 ++- 20 files changed, 957 insertions(+), 612 deletions(-) create mode 100644 routstr/mint.py create mode 100644 tests/unit/test_melt_reconciliation.py create mode 100644 tests/unit/test_mint.py diff --git a/.env.example b/.env.example index 9f0c0bbc..3d12ee5c 100644 --- a/.env.example +++ b/.env.example @@ -45,6 +45,9 @@ ROUTSTR_SECRET_KEY= # ENABLE_ANALYTICS_SHARING=true # CASHU_MINTS="https://mint.minibits.cash/Bitcoin,https://mint.cubabitcoin.org,https://ecashmint.otrta.me" # MINT_OPERATION_CONCURRENCY=4 +# MINT_OPERATION_TIMEOUT_SECONDS=30 +# MINT_MAX_CONCURRENCY=4 +# MINT_RETRY_MAX_ATTEMPTS=3 # RECEIVE_LN_ADDRESS= # REFUND_SWEEP_CLAIM_TIMEOUT_SECONDS=900 diff --git a/docs/provider/configuration.md b/docs/provider/configuration.md index 33efabb8..eee669c4 100644 --- a/docs/provider/configuration.md +++ b/docs/provider/configuration.md @@ -136,6 +136,10 @@ Use environment variables for: | `NSEC` | Legacy seed for the Nostr private key (otherwise set from the admin UI) | — | | `ENABLE_ANALYTICS_SHARING` | Enable usage analytics sharing to Nostr | `true` | | `CASHU_MINTS` | Comma-separated mint URLs | `https://mint.minibits.cash/Bitcoin` | +| `MINT_OPERATION_CONCURRENCY` | Concurrent mint/unit balance reads | `4` | +| `MINT_OPERATION_TIMEOUT_SECONDS` | Per-attempt timeout for mint network calls | `30` | +| `MINT_MAX_CONCURRENCY` | Concurrent operations allowed per mint (`0` disables the limit) | `4` | +| `MINT_RETRY_MAX_ATTEMPTS` | Retries after a timeout or HTTP 429 (`0` disables retries) | `3` | | `RECEIVE_LN_ADDRESS` | Lightning address for withdrawals | — | | `MIN_PAYOUT_SAT` | Min payout balance in sats (applies to all mints) | `210` | | `PAYOUT_INTERVAL_SECONDS` | Payout loop interval (seconds) | `900` | @@ -143,6 +147,11 @@ Use environment variables for: | `CORS_ORIGINS` | Allowed CORS origins | `*` | | `RELAYS` | Nostr relays (comma-separated) | (default set) | +Mint HTTP 429 responses create a per-mint cooldown. Operations that already hold +Routstr's wallet mutation lock fail fast during that cooldown instead of waiting +while blocking every other wallet mutation. Callers receive an error and may retry +later; the current response does not include the cooldown duration. + ### Priority Environment variables are read on startup. Dashboard settings override them and persist in the database. Once you change a setting in the dashboard, the env var is ignored for that setting. diff --git a/routstr/balance.py b/routstr/balance.py index 00894608..b106aebd 100644 --- a/routstr/balance.py +++ b/routstr/balance.py @@ -215,7 +215,7 @@ async def topup_wallet_endpoint( raise HTTPException(status_code=400, detail="Invalid token format") source_mint = token_mint_url(cashu_token, "unknown") - logger.warning( + logger.info( "Cashu wallet top-up started", extra={ "event": "cashu_topup_started", @@ -259,7 +259,7 @@ async def topup_wallet_endpoint( ) raise HTTPException(status_code=status_code, detail=message) - logger.warning( + logger.info( "Cashu wallet top-up completed", extra={ "event": "cashu_topup_completed", diff --git a/routstr/lightning.py b/routstr/lightning.py index 6fa4976e..41dc775f 100644 --- a/routstr/lightning.py +++ b/routstr/lightning.py @@ -3,8 +3,9 @@ import hashlib import re import secrets import time +from contextlib import asynccontextmanager from dataclasses import dataclass -from typing import Any +from typing import Any, AsyncGenerator from fastapi import APIRouter, Depends, Header, HTTPException from pydantic import BaseModel, Field @@ -15,11 +16,13 @@ from sqlmodel.ext.asyncio.session import AsyncSession from .core.db import ApiKey, LightningInvoice, create_session, get_session from .core.logging import get_logger from .core.settings import settings +from .mint import ( + is_mint_rate_limited, + mint_cooldown_remaining, + run_mint_operation, +) from .wallet import ( MintConnectionError, - _is_mint_rate_limited, - _mint_cooldown_remaining, - _mint_operation, get_wallet, is_mint_connection_error, wallet_operation_guard, @@ -31,7 +34,31 @@ lightning_router = APIRouter(prefix="/lightning") # Avoid duplicate work within one process. Cross-process credit fencing is done # by the conditional pending -> paid update in _finalize_invoice_settlement(). -_invoice_settlement_locks: dict[str, asyncio.Lock] = {} +@dataclass +class _InvoiceLockEntry: + lock: asyncio.Lock + users: int = 0 + + +_invoice_settlement_locks: dict[str, _InvoiceLockEntry] = {} + + +@asynccontextmanager +async def _invoice_settlement_lock(invoice_id: str) -> AsyncGenerator[None, None]: + """Serialize one invoice and remove its lock after the last waiter leaves.""" + + entry = _invoice_settlement_locks.get(invoice_id) + if entry is None: + entry = _InvoiceLockEntry(asyncio.Lock()) + _invoice_settlement_locks[invoice_id] = entry + entry.users += 1 + try: + async with entry.lock: + yield + finally: + entry.users -= 1 + if entry.users == 0 and _invoice_settlement_locks.get(invoice_id) is entry: + del _invoice_settlement_locks[invoice_id] @dataclass(frozen=True) @@ -142,11 +169,11 @@ async def _request_mint_with_fallback( configured = allowed_mints or [settings.primary_mint, *settings.cashu_mints] candidates = list(dict.fromkeys(configured)) for mint_url in candidates: - cooldown = _mint_cooldown_remaining(mint_url) + cooldown = mint_cooldown_remaining(mint_url) if cooldown > 0: tried.append(f"{mint_url}: cooling down") logger.info( - "Skipping rate-limited mint", + "Skipping mint during cooldown", extra={ "mint_url": mint_url, "cooldown_seconds": round(cooldown, 2), @@ -156,7 +183,7 @@ async def _request_mint_with_fallback( continue try: wallet = await get_wallet(mint_url, "sat", retry_on_rate_limit=False) - quote = await _mint_operation( + quote = await run_mint_operation( lambda: wallet.request_mint(amount_sats), op_name="request_mint_invoice", mint_url=mint_url, @@ -165,7 +192,7 @@ async def _request_mint_with_fallback( return quote.request, quote.quote, mint_url except Exception as e: tried.append(f"{mint_url}: {type(e).__name__}") - if not is_mint_connection_error(e) and not _is_mint_rate_limited(e): + if not is_mint_connection_error(e) and not is_mint_rate_limited(e): raise logger.warning( "request_mint failed, trying fallback mint", @@ -352,8 +379,7 @@ async def recover_invoice( async def check_invoice_payment( invoice: LightningInvoice, session: AsyncSession ) -> None: - lock = _invoice_settlement_locks.setdefault(invoice.id, asyncio.Lock()) - async with lock, wallet_operation_guard(): + async with _invoice_settlement_lock(invoice.id), wallet_operation_guard(): minted = False try: # Snapshot the row and end the caller's read transaction before any @@ -368,7 +394,7 @@ async def check_invoice_payment( mint_url = settlement.mint_url or settings.primary_mint wallet = await get_wallet(mint_url, "sat") - mint_status = await _mint_operation( + mint_status = await run_mint_operation( lambda: wallet.get_mint_quote(settlement.payment_hash), op_name="get_mint_quote", mint_url=mint_url, @@ -483,7 +509,7 @@ async def _mint_invoice_quote( return try: - await _mint_operation( + await run_mint_operation( lambda: wallet.mint(invoice.amount_sats, quote_id=invoice.payment_hash), op_name=f"invoice_mint_{invoice.purpose}", mint_url=mint_url, diff --git a/routstr/mint.py b/routstr/mint.py new file mode 100644 index 00000000..a676a320 --- /dev/null +++ b/routstr/mint.py @@ -0,0 +1,338 @@ +"""Shared policy for bounded, rate-aware Cashu mint API operations.""" + +from __future__ import annotations + +import asyncio +import socket +import time +from contextlib import asynccontextmanager +from contextvars import ContextVar +from typing import Any, AsyncGenerator, Awaitable, Callable + +import httpx + +from .core.logging import get_logger +from .core.settings import settings + +logger = get_logger(__name__) + +MINT_TRANSPORT_EXCEPTIONS: tuple[type[BaseException], ...] = ( + httpx.NetworkError, + httpx.TimeoutException, + ConnectionError, + socket.gaierror, + asyncio.TimeoutError, +) + +MINT_TRANSPORT_COOLDOWN_SECONDS = 30.0 +_MINT_RATE_LIMIT_BASE_COOLDOWN_SECONDS = 60.0 +_MINT_RATE_LIMIT_MAX_COOLDOWN_SECONDS = 7 * 60 * 60 + +_fail_fast_depth: ContextVar[int] = ContextVar("mint_fail_fast_depth", default=0) + + +class MintRateLimitedError(httpx.HTTPStatusError): + """Typed boundary error preserving a Cashu mint's HTTP 429 response.""" + + +class MintCooldownError(Exception): + """A mint is cooling down and this operation must not wait.""" + + def __init__(self, mint_url: str, retry_after_seconds: float): + self.mint_url = mint_url + self.retry_after_seconds = max(0.0, retry_after_seconds) + super().__init__( + f"Mint {mint_url} is cooling down; retry after " + f"{self.retry_after_seconds:.2f}s" + ) + + +@asynccontextmanager +async def fail_fast_mint_operations() -> AsyncGenerator[None, None]: + """Make mint cooldown/probe waits fail fast in the current task. + + Wallet mutation code holds a process-wide file lock. It enters this scope so + an existing mint cooldown can never turn that lock into a multi-hour wait. + """ + + token = _fail_fast_depth.set(_fail_fast_depth.get() + 1) + try: + yield + finally: + _fail_fast_depth.reset(token) + + +class MintRateGuard: + """Limit concurrency and remember per-mint cooldown/probe state.""" + + _guards: dict[str, "MintRateGuard"] = {} + + @classmethod + def get(cls, mint_url: str) -> "MintRateGuard": + concurrency = settings.mint_max_concurrency + guard = cls._guards.get(mint_url) + if guard is None or guard._max_concurrency != concurrency: + guard = cls(mint_url, concurrency) + cls._guards[mint_url] = guard + return guard + + def __init__(self, mint_url: str, max_concurrency: int): + self._mint_url = mint_url + self._max_concurrency = max_concurrency + self._semaphore = ( + asyncio.Semaphore(max_concurrency) if max_concurrency > 0 else None + ) + self._cooldown_until = 0.0 + self._cooldown_reason: str | None = None + self._consecutive_rate_limits = 0 + self._needs_probe = False + self._probe_lock = asyncio.Lock() + + def apply_cooldown(self, delay: float, *, reason: str | None = None) -> None: + deadline = time.monotonic() + max(0.0, delay) + if deadline >= self._cooldown_until: + self._cooldown_until = deadline + if reason is not None: + self._cooldown_reason = reason + elif self._cooldown_reason is None and reason is not None: + self._cooldown_reason = reason + self._needs_probe = True + + def apply_rate_limit_cooldown(self, retry_after: float | None = None) -> float: + remaining = self.cooldown_remaining() + if remaining > 0 and self._cooldown_reason == "rate_limited": + minimum = min( + _MINT_RATE_LIMIT_MAX_COOLDOWN_SECONDS, + max(_MINT_RATE_LIMIT_BASE_COOLDOWN_SECONDS, retry_after or 0.0), + ) + if minimum > remaining: + self.apply_cooldown(minimum, reason="rate_limited") + return minimum + return remaining + + self._consecutive_rate_limits += 1 + base = max(_MINT_RATE_LIMIT_BASE_COOLDOWN_SECONDS, retry_after or 0.0) + multiplier = 2 ** min(self._consecutive_rate_limits - 1, 10) + delay = min(_MINT_RATE_LIMIT_MAX_COOLDOWN_SECONDS, base * multiplier) + self.apply_cooldown(delay, reason="rate_limited") + return delay + + def cooldown_remaining(self) -> float: + return max(0.0, self._cooldown_until - time.monotonic()) + + def cooldown_reason(self) -> str | None: + return self._cooldown_reason if self.cooldown_remaining() > 0 else None + + def _raise_if_wait_forbidden(self) -> None: + if _fail_fast_depth.get() and ( + self._needs_probe or self.cooldown_remaining() > 0 + ): + raise MintCooldownError(self._mint_url, self.cooldown_remaining()) + + async def _wait_for_cooldown(self) -> None: + while True: + self._raise_if_wait_forbidden() + deadline = self._cooldown_until + wait = max(0.0, deadline - time.monotonic()) + if wait <= 0: + return + logger.debug( + "Mint rate guard: cooling down", + extra={"mint_url": self._mint_url, "wait_seconds": round(wait, 2)}, + ) + await asyncio.sleep(wait) + if self._cooldown_until <= deadline: + return + + async def _run_probe(self, factory: Callable[[], Awaitable[Any]]) -> Any: + await self._wait_for_cooldown() + logger.info( + "Mint cooldown ended; sending one probe request", + extra={"event": "mint_cooldown_probe_started", "mint_url": self._mint_url}, + ) + try: + result = await factory() + except Exception as error: + if is_mint_rate_limited(error): + retry_after = None + if isinstance(error, httpx.HTTPStatusError): + retry_after = parse_retry_after(error.response.headers) + delay = max( + _MINT_RATE_LIMIT_BASE_COOLDOWN_SECONDS, + retry_after or 0.0, + ) + self.apply_cooldown(delay, reason="rate_limited") + else: + self.apply_cooldown(1.0) + logger.warning( + "Mint cooldown probe failed", + extra={ + "event": "mint_cooldown_probe_failed", + "mint_url": self._mint_url, + "error": str(error), + "error_type": type(error).__name__, + "cooldown_seconds": round(self.cooldown_remaining(), 2), + "consecutive_rate_limits": self._consecutive_rate_limits, + }, + ) + raise + + self._needs_probe = False + self._cooldown_until = 0.0 + self._cooldown_reason = None + self._consecutive_rate_limits = 0 + logger.info( + "Mint cooldown probe succeeded; restoring normal concurrency", + extra={ + "event": "mint_cooldown_probe_succeeded", + "mint_url": self._mint_url, + }, + ) + return result + + async def run(self, factory: Callable[[], Awaitable[Any]]) -> Any: + while True: + self._raise_if_wait_forbidden() + if self._needs_probe or self.cooldown_remaining() > 0: + async with self._probe_lock: + self._raise_if_wait_forbidden() + if self.cooldown_remaining() > 0: + self._needs_probe = True + if self._needs_probe: + return await self._run_probe(factory) + continue + + if self._semaphore is None: + return await factory() + async with self._semaphore: + self._raise_if_wait_forbidden() + if self._needs_probe: + continue + return await factory() + + +def mint_cooldown_remaining(mint_url: str) -> float: + return MintRateGuard.get(mint_url).cooldown_remaining() + + +def mint_cooldown_reason(mint_url: str) -> str | None: + return MintRateGuard.get(mint_url).cooldown_reason() + + +def is_mint_rate_limited(error: BaseException) -> bool: + """Return whether an exception chain represents HTTP 429/cooldown.""" + + current: BaseException | None = error + seen: set[int] = set() + while current is not None and id(current) not in seen: + seen.add(id(current)) + if isinstance(current, MintCooldownError): + return True + if isinstance(current, httpx.HTTPStatusError): + if current.response.status_code == 429: + return True + current = current.__cause__ or current.__context__ + return False + + +def parse_retry_after(headers: Any) -> float | None: + raw = headers.get("retry-after") or headers.get("Retry-After") + if raw is None: + return None + try: + return float(str(raw).strip()) + except (TypeError, ValueError): + return None + + +async def run_mint_operation( + factory: Callable[[], Awaitable[Any]], + *, + op_name: str = "mint_operation", + mint_url: str = "", + retry_timeouts: bool = True, + retry_on_rate_limit: bool = True, +) -> Any: + """Run one mint operation with bounded concurrency and adaptive cooldown.""" + + guard = MintRateGuard.get(mint_url) if mint_url else None + timeout = settings.mint_operation_timeout_seconds + max_attempts = settings.mint_retry_max_attempts + 1 + + async def timed_factory() -> Any: + if timeout > 0: + return await asyncio.wait_for(factory(), timeout=timeout) + return await factory() + + async def invoke() -> Any: + if guard is not None: + return await guard.run(timed_factory) + return await timed_factory() + + for attempt in range(max_attempts): + try: + return await invoke() + except MintCooldownError: + raise + except (asyncio.TimeoutError, httpx.TimeoutException) as exc: + if retry_timeouts and attempt < max_attempts - 1: + backoff = (2**attempt) + (time.monotonic() % 1.0) + logger.warning( + "Mint operation timed out, retrying", + extra={ + "op_name": op_name, + "mint_url": mint_url, + "attempt": attempt + 1, + "backoff_seconds": round(backoff, 2), + }, + ) + await asyncio.sleep(backoff) + continue + raise httpx.TimeoutException( + f"{op_name} timed out (attempts: {attempt + 1})" + ) from exc + except Exception as exc: + if not is_mint_rate_limited(exc): + raise + + backoff = (2**attempt) + (time.monotonic() % 1.0) + if isinstance(exc, httpx.HTTPStatusError): + retry_after = parse_retry_after(exc.response.headers) + if retry_after is not None: + backoff = max(retry_after, backoff) + cooldown = backoff + if guard is not None: + cooldown = guard.apply_rate_limit_cooldown(backoff) + + if not retry_on_rate_limit: + logger.warning( + "Mint rate-limited, skipping retries for fallback", + extra={ + "op_name": op_name, + "mint_url": mint_url, + "cooldown_seconds": round(cooldown, 2), + "consecutive_rate_limits": guard._consecutive_rate_limits + if guard is not None + else attempt + 1, + }, + ) + raise + + if attempt >= max_attempts - 1: + raise + logger.warning( + "Mint rate-limited, applying cooldown", + extra={ + "op_name": op_name, + "mint_url": mint_url, + "attempt": attempt + 1, + "cooldown_seconds": round(cooldown, 2), + "consecutive_rate_limits": guard._consecutive_rate_limits + if guard is not None + else attempt + 1, + }, + ) + if guard is None: + await asyncio.sleep(cooldown) + + raise RuntimeError(f"{op_name}: exhausted retries unexpectedly") diff --git a/routstr/payment/helpers.py b/routstr/payment/helpers.py index 67a7284d..a3ab1fb0 100644 --- a/routstr/payment/helpers.py +++ b/routstr/payment/helpers.py @@ -18,18 +18,6 @@ from ..wallet import deserialize_token_from_string logger = get_logger(__name__) -# Interim policy: when Routstr must move value to another trusted mint, the -# cross-mint Lightning round trip can consume fees that are not visible to the -# client. Reserve 5% headroom until the fee-payer policy is made explicit. -_MINT_FEE_ALLOWANCE = 0.05 - - -def apply_mint_fee_allowance(cost_msat: int) -> int: - """Reserve headroom for possible trusted-mint fallback fees.""" - adjusted = math.ceil(cost_msat * (1 - _MINT_FEE_ALLOWANCE)) - return max(settings.min_request_msat, adjusted) - - def check_token_balance(headers: dict, body: dict, max_cost_for_model: int) -> None: if x_cashu := headers.get("x-cashu", None): cashu_token = x_cashu diff --git a/routstr/payment/lnurl.py b/routstr/payment/lnurl.py index fbd28586..c6c3ccb7 100644 --- a/routstr/payment/lnurl.py +++ b/routstr/payment/lnurl.py @@ -1,17 +1,13 @@ from __future__ import annotations -import asyncio import math from typing import TypedDict import httpx +from cashu.core.base import MeltQuoteState from cashu.wallet.wallet import Proof, Wallet -# The Cashu library issues POST /v1/melt/bolt11 with timeout=None, so a hung or -# very slow mint can block a melt (and any caller, e.g. the payout loop) -# indefinitely. _mint_operation (imported lazily in raw_send_to_lnurl to avoid -# a circular import with wallet.py) bounds it via MINT_OPERATION_TIMEOUT_SECONDS. -MELT_TIMEOUT_SECONDS = 60 +from ..mint import MINT_TRANSPORT_EXCEPTIONS, run_mint_operation try: from bech32 import bech32_decode, convertbits # type: ignore @@ -222,9 +218,7 @@ async def raw_send_to_lnurl( lnurl_data["callback_url"], final_amount ) - from ..wallet import _mint_operation - - melt_quote_resp = await _mint_operation( + melt_quote_resp = await run_mint_operation( lambda: wallet.melt_quote(invoice=bolt11_invoice), op_name="lnurl_melt_quote", mint_url=str(wallet.url), @@ -234,22 +228,44 @@ async def raw_send_to_lnurl( proofs, _ = await wallet.select_to_send(proofs, amount, set_reserved=True) try: - _ = await asyncio.wait_for( - _mint_operation( - lambda: wallet.melt( - proofs=proofs, - invoice=bolt11_invoice, - fee_reserve_sat=melt_quote_resp.fee_reserve, - quote_id=melt_quote_resp.quote, - ), - op_name="lnurl_melt", - mint_url=str(wallet.url), - retry_timeouts=False, + melt_response = await run_mint_operation( + lambda: wallet.melt( + proofs=proofs, + invoice=bolt11_invoice, + fee_reserve_sat=melt_quote_resp.fee_reserve, + quote_id=melt_quote_resp.quote, ), - timeout=MELT_TIMEOUT_SECONDS, + op_name="lnurl_melt", + mint_url=str(wallet.url), + retry_timeouts=False, ) - except (httpx.TimeoutException, asyncio.TimeoutError) as e: + except MINT_TRANSPORT_EXCEPTIONS as error: + melt_response = None + melt_error: BaseException | None = error + else: + melt_error = None + + if getattr(melt_response, "state", None) == MeltQuoteState.paid: + return final_amount + + try: + quote = await run_mint_operation( + lambda: wallet.get_melt_quote(melt_quote_resp.quote), + op_name="reconcile_lnurl_melt_quote", + mint_url=str(wallet.url), + retry_timeouts=False, + ) + except Exception as reconciliation_error: raise LNURLError( - f"Melt timed out after {MELT_TIMEOUT_SECONDS}s (mint unresponsive)" - ) from e - return final_amount + "Melt outcome is ambiguous; quote reconciliation failed and proofs " + "must not be retried" + ) from reconciliation_error + + if quote is not None and quote.state == MeltQuoteState.paid: + return final_amount + + state = getattr(getattr(quote, "state", None), "value", "unknown") + raise LNURLError( + "Melt outcome is ambiguous; proofs must not be retried " + f"(quote_state={state})" + ) from melt_error diff --git a/routstr/proxy.py b/routstr/proxy.py index 2bbe10ee..9b7d4077 100644 --- a/routstr/proxy.py +++ b/routstr/proxy.py @@ -27,7 +27,6 @@ from .core.db import ( from .core.exceptions import UpstreamError from .core.not_found import build_not_found_response from .payment.helpers import ( - apply_mint_fee_allowance, calculate_discounted_max_cost, check_token_balance, create_error_response, @@ -354,7 +353,6 @@ async def _proxy( max_cost_for_model = await calculate_discounted_max_cost( _max_cost_for_model, request_body_dict, model_obj=model_obj ) - max_cost_for_model = apply_mint_fee_allowance(max_cost_for_model) check_token_balance(headers, request_body_dict, max_cost_for_model) @@ -493,9 +491,6 @@ async def _proxy( candidate_max = await calculate_discounted_max_cost( candidate_max, request_body_dict, model_obj=model_obj ) - # Apply the same interim 5% trusted-mint fee headroom used for the - # first candidate; failover must not silently change admission. - candidate_max = apply_mint_fee_allowance(candidate_max) if candidate_max > max_cost_for_model: await revert_pay_for_request( key, session, max_cost_for_model, reservation_snapshot diff --git a/routstr/wallet.py b/routstr/wallet.py index a58d4c95..c533355e 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -2,16 +2,15 @@ import asyncio import fcntl import os import re -import socket import time import typing from contextlib import asynccontextmanager from contextvars import ContextVar from pathlib import Path -from typing import Any, AsyncGenerator, Awaitable, Callable, TypedDict +from typing import AsyncGenerator, TypedDict import httpx -from cashu.core.base import MintQuote, Proof, Token +from cashu.core.base import MeltQuoteState, MintQuote, Proof, Token from cashu.core.mint_info import MintInfo as _CashuMintInfo from cashu.wallet.helpers import deserialize_token_from_string from cashu.wallet.wallet import Wallet as _CashuWallet @@ -21,8 +20,27 @@ from sqlmodel import col, select, update from .core import db, get_logger from .core.db import store_cashu_transaction_with_retry as store_cashu_transaction from .core.settings import settings +from .mint import ( + MINT_TRANSPORT_COOLDOWN_SECONDS, + MINT_TRANSPORT_EXCEPTIONS, + MintRateGuard, + MintRateLimitedError, + fail_fast_mint_operations, + is_mint_rate_limited, + mint_cooldown_reason, + mint_cooldown_remaining, + run_mint_operation, +) from .payment.lnurl import raw_send_to_lnurl +# Backwards-compatible aliases for callers/tests that imported the former +# wallet-local policy. Production modules use the public routstr.mint API. +_MintRateGuard = MintRateGuard +_mint_operation = run_mint_operation +_mint_cooldown_remaining = mint_cooldown_remaining +_mint_cooldown_reason = mint_cooldown_reason +_is_mint_rate_limited = is_mint_rate_limited + # cashu still declares Optional[X] without explicit defaults on MintInfo. # Under pydantic v2 those are required, but real mints omit many of them. # Default Optional fields to None at import time so balance fetches don't 422. @@ -69,7 +87,8 @@ async def wallet_operation_guard() -> AsyncGenerator[None, None]: except BlockingIOError: await _scheduler_sleep(0.05) depth_token = _wallet_operation_depth.set(1) - yield + async with fail_fast_mint_operations(): + yield finally: if depth_token is not None: _wallet_operation_depth.reset(depth_token) @@ -94,6 +113,20 @@ def _mints_to_inspect() -> list[str]: return mint_urls +class Wallet(_CashuWallet): + """Cashu adapter that preserves HTTP 429 for Routstr's mint policy.""" + + @staticmethod + def raise_on_error_request(resp: httpx.Response) -> None: + if resp.status_code == 429: + raise MintRateLimitedError( + "Cashu mint rate limited", + request=resp.request, + response=resp, + ) + _CashuWallet.raise_on_error_request(resp) + + class MintConnectionError(Exception): """The mint could not be reached (network transport failure). @@ -116,339 +149,6 @@ class TokenConsumedError(Exception): """ -class MintRateLimitedError(httpx.HTTPStatusError): - """Typed boundary error preserving a Cashu mint's HTTP 429 response.""" - - -class Wallet(_CashuWallet): - """Cashu wallet adapter that preserves rate-limit status information. - - Cashu's default response adapter converts JSON error bodies into plain - ``Exception`` instances before calling ``raise_for_status``. Intercept 429 - here so Routstr's fallback and cooldown policy can use the real status - without unreliable message matching. - """ - - @staticmethod - def raise_on_error_request(resp: httpx.Response) -> None: - if resp.status_code == 429: - raise MintRateLimitedError( - "Cashu mint rate limited", - request=resp.request, - response=resp, - ) - _CashuWallet.raise_on_error_request(resp) - - -# httpx base classes cover their subclasses. HTTPStatusError is excluded on -# purpose — that means the mint answered, just with an error status. -_MINT_TRANSPORT_COOLDOWN_SECONDS = 30.0 -_MINT_RATE_LIMIT_BASE_COOLDOWN_SECONDS = 60.0 -_MINT_RATE_LIMIT_MAX_COOLDOWN_SECONDS = 7 * 60 * 60 - -_TRANSPORT_EXC_TYPES: tuple[type[BaseException], ...] = ( - httpx.NetworkError, - httpx.TimeoutException, - ConnectionError, # refused/reset/aborted - socket.gaierror, # DNS failure - asyncio.TimeoutError, -) - - -class _MintRateGuard: - """Limit concurrency and remember per-mint rate-limit cooldowns.""" - - _guards: dict[str, "_MintRateGuard"] = {} - - @classmethod - def get(cls, mint_url: str) -> "_MintRateGuard": - concurrency = settings.mint_max_concurrency - guard = cls._guards.get(mint_url) - if guard is None or guard._max_concurrency != concurrency: - guard = cls(mint_url, concurrency) - cls._guards[mint_url] = guard - return guard - - def __init__(self, mint_url: str, max_concurrency: int): - self._mint_url = mint_url - self._max_concurrency = max_concurrency - self._semaphore = ( - asyncio.Semaphore(max_concurrency) if max_concurrency > 0 else None - ) - self._cooldown_until = 0.0 - self._cooldown_reason: str | None = None - self._consecutive_rate_limits = 0 - self._needs_probe = False - self._probe_lock = asyncio.Lock() - - def apply_cooldown(self, delay: float, *, reason: str | None = None) -> None: - deadline = time.monotonic() + max(0.0, delay) - if deadline >= self._cooldown_until: - self._cooldown_until = deadline - if reason is not None: - self._cooldown_reason = reason - elif self._cooldown_reason is None and reason is not None: - self._cooldown_reason = reason - self._needs_probe = True - - def apply_rate_limit_cooldown(self, retry_after: float | None = None) -> float: - remaining = self.cooldown_remaining() - if remaining > 0 and self._cooldown_reason == "rate_limited": - minimum = min( - _MINT_RATE_LIMIT_MAX_COOLDOWN_SECONDS, - max(_MINT_RATE_LIMIT_BASE_COOLDOWN_SECONDS, retry_after or 0.0), - ) - if minimum > remaining: - self.apply_cooldown(minimum, reason="rate_limited") - return minimum - return remaining - - self._consecutive_rate_limits += 1 - base = max(_MINT_RATE_LIMIT_BASE_COOLDOWN_SECONDS, retry_after or 0.0) - multiplier = 2 ** min(self._consecutive_rate_limits - 1, 10) - delay = min(_MINT_RATE_LIMIT_MAX_COOLDOWN_SECONDS, base * multiplier) - self.apply_cooldown(delay, reason="rate_limited") - return delay - - def cooldown_remaining(self) -> float: - return max(0.0, self._cooldown_until - time.monotonic()) - - def cooldown_reason(self) -> str | None: - return self._cooldown_reason if self.cooldown_remaining() > 0 else None - - async def _wait_for_cooldown(self) -> None: - while True: - deadline = self._cooldown_until - wait = max(0.0, deadline - time.monotonic()) - if wait <= 0: - return - logger.debug( - "Mint rate guard: cooling down", - extra={"mint_url": self._mint_url, "wait_seconds": round(wait, 2)}, - ) - await asyncio.sleep(wait) - if self._cooldown_until <= deadline: - return - - async def _run_probe(self, factory: Callable[[], Awaitable[Any]]) -> Any: - await self._wait_for_cooldown() - logger.warning( - "Mint cooldown ended; sending one probe request", - extra={"event": "mint_cooldown_probe_started", "mint_url": self._mint_url}, - ) - try: - result = await factory() - except Exception as error: - # Keep queued callers behind the probe. On a rate-limit, - # re-apply the *same* cooldown the caller already set rather - # than calling apply_rate_limit_cooldown() — the probe is a - # recovery check, not a new request that should escalate the - # exponential backoff counter. - if _is_mint_rate_limited(error): - retry_after = None - if isinstance(error, httpx.HTTPStatusError): - retry_after = _parse_retry_after(error.response.headers) - delay = max( - _MINT_RATE_LIMIT_BASE_COOLDOWN_SECONDS, - retry_after or 0.0, - ) - self.apply_cooldown(delay, reason="rate_limited") - else: - self.apply_cooldown(1.0) - logger.warning( - "Mint cooldown probe failed", - extra={ - "event": "mint_cooldown_probe_failed", - "mint_url": self._mint_url, - "error": str(error), - "error_type": type(error).__name__, - "cooldown_seconds": round(self.cooldown_remaining(), 2), - "consecutive_rate_limits": self._consecutive_rate_limits, - }, - ) - raise - - self._needs_probe = False - self._cooldown_until = 0.0 - self._cooldown_reason = None - self._consecutive_rate_limits = 0 - logger.warning( - "Mint cooldown probe succeeded; restoring normal concurrency", - extra={ - "event": "mint_cooldown_probe_succeeded", - "mint_url": self._mint_url, - }, - ) - return result - - async def run(self, factory: Callable[[], Awaitable[Any]]) -> Any: - while True: - if self._needs_probe or self.cooldown_remaining() > 0: - async with self._probe_lock: - if self.cooldown_remaining() > 0: - self._needs_probe = True - if self._needs_probe: - return await self._run_probe(factory) - continue - - if self._semaphore is None: - return await factory() - async with self._semaphore: - if self._needs_probe: - continue - return await factory() - - -def _mint_cooldown_remaining(mint_url: str) -> float: - return _MintRateGuard.get(mint_url).cooldown_remaining() - - -def _mint_cooldown_reason(mint_url: str) -> str | None: - return _MintRateGuard.get(mint_url).cooldown_reason() - - -def _is_mint_rate_limited(error: BaseException) -> bool: - """True if the mint returned an HTTP 429 (Too Many Requests). - - Only matches ``httpx.HTTPStatusError`` with status code 429 — never - classifies based on the exception's message text. Substring matching - on ``"rate limit"`` / ``"too many requests"`` was removed because it - catches unrelated errors (e.g. a 503 whose body happens to mention - "database rate exceeded"), which triggers unnecessary exponential - backoff and can block state recovery indefinitely. - """ - current: BaseException | None = error - seen: set[int] = set() - while current is not None and id(current) not in seen: - seen.add(id(current)) - if isinstance(current, httpx.HTTPStatusError): - if current.response.status_code == 429: - return True - current = current.__cause__ or current.__context__ - return False - - -async def _mint_operation( - factory: Callable[[], Awaitable[Any]], - *, - op_name: str = "mint_operation", - mint_url: str = "", - retry_timeouts: bool = True, - retry_on_rate_limit: bool = True, -) -> Any: - """Run a mint operation with bounded concurrency and adaptive cooldown. - - The timeout applies to each network attempt. Queueing, cooldown, and retry - backoff are deliberately outside it so the shipped 60-second 429 cooldown - is not cancelled by the 30-second operation timeout. ``factory`` must - return a fresh coroutine for every retry. - - When ``retry_on_rate_limit`` is False a 429 is not retried in-place — the - cooldown is still applied to the per-mint guard (so subsequent operations on - that mint wait), but the exception is re-raised so the caller (typically - ``_request_mint_with_fallback``) can immediately try a different mint. - """ - guard = _MintRateGuard.get(mint_url) if mint_url else None - timeout = settings.mint_operation_timeout_seconds - max_attempts = settings.mint_retry_max_attempts + 1 - - async def timed_factory() -> Any: - if timeout > 0: - return await asyncio.wait_for(factory(), timeout=timeout) - return await factory() - - async def invoke() -> Any: - if guard is not None: - return await guard.run(timed_factory) - return await timed_factory() - - async def run_with_retries() -> Any: - for attempt in range(max_attempts): - try: - return await invoke() - except (asyncio.TimeoutError, httpx.TimeoutException) as exc: - if retry_timeouts and attempt < max_attempts - 1: - backoff = (2**attempt) + (time.monotonic() % 1.0) - logger.warning( - "Mint operation timed out, retrying", - extra={ - "op_name": op_name, - "mint_url": mint_url, - "attempt": attempt + 1, - "backoff_seconds": round(backoff, 2), - }, - ) - await asyncio.sleep(backoff) - continue - raise httpx.TimeoutException( - f"{op_name} timed out (attempts: {attempt + 1})" - ) from exc - except Exception as exc: - if not _is_mint_rate_limited(exc): - raise - - # Apply cooldown to the guard regardless — even when we're - # about to re-raise for fallback, the guard must remember that - # this mint is rate-limited for future operations. - backoff = (2**attempt) + (time.monotonic() % 1.0) - if isinstance(exc, httpx.HTTPStatusError): - retry_after = _parse_retry_after(exc.response.headers) - if retry_after is not None: - backoff = max(retry_after, backoff) - cooldown = backoff - if guard is not None: - cooldown = guard.apply_rate_limit_cooldown(backoff) - - # When the caller has a fallback strategy (trusted-mint - # list), re-raise immediately so the caller can try the next - # mint instead of waiting through this mint's cooldown. - if not retry_on_rate_limit: - logger.warning( - "Mint rate-limited, skipping retries for fallback", - extra={ - "op_name": op_name, - "mint_url": mint_url, - "cooldown_seconds": round(cooldown, 2), - "consecutive_rate_limits": guard._consecutive_rate_limits - if guard is not None - else attempt + 1, - }, - ) - raise - - if attempt >= max_attempts - 1: - raise - logger.warning( - "Mint rate-limited, applying cooldown", - extra={ - "op_name": op_name, - "mint_url": mint_url, - "attempt": attempt + 1, - "cooldown_seconds": round(cooldown, 2), - "consecutive_rate_limits": guard._consecutive_rate_limits - if guard is not None - else attempt + 1, - }, - ) - if guard is None: - await asyncio.sleep(cooldown) - - raise RuntimeError(f"{op_name}: exhausted retries unexpectedly") - - return await run_with_retries() - - -def _parse_retry_after(headers: Any) -> float | None: - """Parse a Retry-After header (delta-seconds form) into seconds.""" - raw = headers.get("retry-after") or headers.get("Retry-After") - if raw is None: - return None - try: - return float(str(raw).strip()) - except (TypeError, ValueError): - return None - - def is_source_mint_connection_error(error: BaseException) -> bool: seen: set[int] = set() current: BaseException | None = error @@ -475,7 +175,7 @@ def is_mint_connection_error(error: BaseException) -> bool: return False if isinstance(current, MintConnectionError): return True - if isinstance(current, _TRANSPORT_EXC_TYPES): + if isinstance(current, MINT_TRANSPORT_EXCEPTIONS): return True current = current.__cause__ or current.__context__ return False @@ -519,7 +219,7 @@ def classify_redemption_error( "The mint that issued this Cashu token is unreachable; the token cannot be redeemed at another mint", "cashu_source_mint_unreachable", ) - if _is_mint_rate_limited(error): + if is_mint_rate_limited(error): return ( "mint_rate_limited", 503, @@ -605,14 +305,14 @@ async def _redeem_same_mint( drifts insolvent. """ try: - await _mint_operation( + await run_mint_operation( lambda: wallet.load_mint(keyset_id=token_obj.keysets[0]), op_name="redeem_load_mint", mint_url=token_obj.mint, ) wallet.verify_proofs_dleq(token_obj.proofs) input_fees = wallet.get_fees_for_proofs(token_obj.proofs) - await _mint_operation( + await run_mint_operation( lambda: wallet.split(proofs=token_obj.proofs, amount=0, include_fees=True), op_name="redeem_split", mint_url=token_obj.mint, @@ -655,7 +355,7 @@ async def recieve_token( destinations = list( dict.fromkeys([settings.primary_mint, *settings.cashu_mints]) ) - logger.warning( + logger.info( "Cashu cross-mint swap required", extra={ "event": "cashu_swap_started", @@ -667,7 +367,7 @@ async def recieve_token( ) return await swap_to_trusted_mint(token_obj, wallet) - logger.warning( + logger.info( "Trying same-mint Cashu redemption", extra={ "event": "cashu_same_mint_redemption", @@ -769,12 +469,12 @@ async def find_trusted_mint_with_funds( balances: dict[str, int] = {} for mint_url in candidates: - if _mint_cooldown_remaining(mint_url) > 0: + if mint_cooldown_remaining(mint_url) > 0: continue try: wallet = await get_wallet(mint_url, unit, retry_on_rate_limit=False) except Exception as error: - if is_mint_connection_error(error) or _is_mint_rate_limited(error): + if is_mint_connection_error(error) or is_mint_rate_limited(error): balances[mint_url] = 0 continue raise @@ -819,6 +519,17 @@ def _net_minted_amount(amount_msat: int, token_unit: str, fees: int) -> int: return int(remaining_msat) +def _melt_definitively_failed(error: Exception) -> bool: + """Return whether the mint authoritatively rejected the Lightning payment. + + Cashu releases the reserved proofs for these responses, so the token remains + reusable. Transport failures and unknown errors are deliberately excluded: + after dispatch their payment outcome may still be pending or paid. + """ + message = str(error).strip() + return message.lower() == "could not pay invoice." or "(Code: 20004)" in message + + def _melt_insufficient_shortfall(error: Exception) -> int | None: """ Classify a melt failure: return the observed shortfall (in the token unit) @@ -871,7 +582,7 @@ async def _request_mint_with_fallback( f"Token value is too small after fee deduction or unit conversion." ) candidates = list(dict.fromkeys([settings.primary_mint, *settings.cashu_mints])) - logger.warning( + logger.info( "Trying trusted destination mints", extra={ "event": "cashu_destination_candidates", @@ -883,7 +594,7 @@ async def _request_mint_with_fallback( ) tried: list[str] = [] for candidate_index, mint_url in enumerate(candidates, start=1): - cooldown = _mint_cooldown_remaining(mint_url) + cooldown = mint_cooldown_remaining(mint_url) if cooldown > 0: tried.append(f"{mint_url}: cooling down") logger.warning( @@ -898,7 +609,7 @@ async def _request_mint_with_fallback( }, ) continue - logger.warning( + logger.info( "Trying destination mint", extra={ "event": "cashu_destination_attempt", @@ -917,13 +628,13 @@ async def _request_mint_with_fallback( settings.primary_mint_unit, retry_on_rate_limit=False, ) - quote = await _mint_operation( + quote = await run_mint_operation( lambda: wallet.request_mint(amount), op_name=op_name, mint_url=mint_url, retry_on_rate_limit=False, ) - logger.warning( + logger.info( "Destination mint selected", extra={ "event": "cashu_destination_selected", @@ -937,12 +648,12 @@ async def _request_mint_with_fallback( except Exception as error: tried.append(f"{mint_url}: {type(error).__name__}") connection_failure = is_mint_connection_error(error) - rate_limited = _is_mint_rate_limited(error) + rate_limited = is_mint_rate_limited(error) if not connection_failure and not rate_limited: raise if connection_failure: - _MintRateGuard.get(mint_url).apply_cooldown( - _MINT_TRANSPORT_COOLDOWN_SECONDS, reason="unreachable" + MintRateGuard.get(mint_url).apply_cooldown( + MINT_TRANSPORT_COOLDOWN_SECONDS, reason="unreachable" ) logger.warning( "Destination mint failed", @@ -993,7 +704,7 @@ async def _calculate_swap_amount( if token_mint_url == settings.primary_mint: logger.info( - "swap_to_primary_mint: skipping fee estimation (same mint)", + "swap_to_trusted_mint: skipping fee estimation (same mint)", extra={"minted_amount": receive_amount}, ) return int(receive_amount) @@ -1005,7 +716,7 @@ async def _calculate_swap_amount( # logs. Guard early with full diagnostic context instead. if receive_amount <= 0: logger.error( - "swap_to_primary_mint: receive_amount is zero or negative, cannot estimate fees", + "swap_to_trusted_mint: receive_amount is zero or negative, cannot estimate fees", extra={ "amount_msat": amount_msat, "token_unit": token_unit, @@ -1022,7 +733,7 @@ async def _calculate_swap_amount( ) logger.info( - "swap_to_primary_mint: estimating fees", + "swap_to_trusted_mint: estimating fees", extra={ "dummy_amount": receive_amount, "unit": settings.primary_mint_unit, @@ -1040,7 +751,7 @@ async def _calculate_swap_amount( primary_wallet=primary_wallet, ) stage = "source_fee_quote" - dummy_melt_quote = await _mint_operation( + dummy_melt_quote = await run_mint_operation( lambda: token_wallet.melt_quote(dummy_mint_quote.request), op_name="swap_fee_est_melt_quote", mint_url=token_mint_url, @@ -1055,7 +766,7 @@ async def _calculate_swap_amount( raise ValueError(f"Fees ({total_fees} {token_unit}) exceed token amount") logger.info( - "swap_to_primary_mint: fee estimation result", + "swap_to_trusted_mint: fee estimation result", extra={ "token_amount_sat": _msats_to_sats(amount_msat), "estimated_fee": total_fees, @@ -1105,10 +816,62 @@ async def _calculate_swap_amount( raise ValueError(f"Failed to estimate fees: {e}") from e +async def _reconcile_ambiguous_melt( + wallet: Wallet, quote_id: str, proofs: list[Proof] +) -> bool: + """Confirm a dispatched melt is paid or conservatively mark it ambiguous. + + A PAID quote is authoritative and does not require a proof-state lookup. + Every other immediate snapshot remains unsafe to retry: an in-flight + Lightning payment can still move UNPAID/UNSPENT to PENDING or PAID after the + cancelled HTTP request returns. + """ + try: + quote = await run_mint_operation( + lambda: wallet.get_melt_quote(quote_id), + op_name="reconcile_swap_melt_quote", + mint_url=str(wallet.url), + retry_timeouts=False, + ) + except Exception as error: + raise TokenConsumedError( + "Source melt outcome is unknown; reconciliation required" + ) from error + + if quote is not None and quote.state == MeltQuoteState.paid: + return True + + try: + proof_response = await run_mint_operation( + lambda: wallet.check_proof_state(proofs), + op_name="reconcile_swap_proofs", + mint_url=str(wallet.url), + retry_timeouts=False, + ) + proof_states = [state.state.value for state in proof_response.states] + except Exception: + proof_states = [] + + quote_state = getattr(getattr(quote, "state", None), "value", "unknown") + raise TokenConsumedError( + "Source melt outcome is ambiguous; reconciliation required " + f"(quote_state={quote_state}, proof_states={proof_states})" + ) + + +async def _confirm_melt_paid( + wallet: Wallet, quote_id: str, proofs: list[Proof], response: object +) -> bool: + """Accept a melt response only when PAID is explicit or reconciled.""" + if getattr(response, "state", None) == MeltQuoteState.paid: + return True + return await _reconcile_ambiguous_melt(wallet, quote_id, proofs) + + async def swap_to_trusted_mint( token_obj: Token, token_wallet: Wallet ) -> tuple[int, str, str]: - logger.warning( + logger.info( "Starting Cashu cross-mint swap", extra={ "event": "cashu_swap_started", @@ -1135,7 +898,7 @@ async def swap_to_trusted_mint( # NUT-02 input fee still applies; _redeem_same_mint accounts for it. if token_obj.mint == settings.primary_mint: logger.info( - "swap_to_primary_mint: token already on primary mint, skipping swap", + "swap_to_trusted_mint: token already on primary mint, skipping swap", extra={ "mint": token_obj.mint, "amount": token_amount, @@ -1166,7 +929,7 @@ async def swap_to_trusted_mint( attempt += 1 if minted_amount <= 0: logger.error( - "swap_to_primary_mint: minted_amount is zero or negative before requesting quote", + "swap_to_trusted_mint: minted_amount is zero or negative before requesting quote", extra={ "minted_amount": minted_amount, "attempt": attempt, @@ -1188,7 +951,7 @@ async def swap_to_trusted_mint( primary_wallet=primary_wallet, ) logger.info( - "swap_to_primary_mint: mint quote received", + "swap_to_trusted_mint: mint quote received", extra={ "mint_quote_id": mint_quote.quote, "attempt": attempt, @@ -1196,7 +959,7 @@ async def swap_to_trusted_mint( }, ) - logger.warning( + logger.info( "Requesting melt quote from source mint", extra={ "event": "cashu_source_melt_quote_attempt", @@ -1206,7 +969,7 @@ async def swap_to_trusted_mint( }, ) try: - melt_quote = await _mint_operation( + melt_quote = await run_mint_operation( lambda: token_wallet.melt_quote(mint_quote.request), op_name="swap_melt_quote", mint_url=token_obj.mint, @@ -1232,7 +995,7 @@ async def swap_to_trusted_mint( input_fees = token_wallet.get_fees_for_proofs(token_obj.proofs) total_needed = melt_quote.amount + melt_quote.fee_reserve + input_fees logger.info( - "swap_to_primary_mint: melt quote received", + "swap_to_trusted_mint: melt quote received", extra={ "melt_quote_id": melt_quote.quote, "melt_amount": melt_quote.amount, @@ -1252,7 +1015,7 @@ async def swap_to_trusted_mint( ) if attempt >= _MAX_SWAP_ATTEMPTS or recomputed <= 0: logger.warning( - "swap_to_primary_mint: insufficient token amount for melt fees", + "swap_to_trusted_mint: insufficient token amount for melt fees", extra={ "token_amount": token_amount, "melt_amount": melt_quote.amount, @@ -1269,7 +1032,7 @@ async def swap_to_trusted_mint( f"(amount: {melt_quote.amount} + fee: {melt_quote.fee_reserve} + input_fees: {input_fees})" ) logger.warning( - "swap_to_primary_mint: melt quote exceeds token amount, retrying", + "swap_to_trusted_mint: melt quote exceeds token amount, retrying", extra={ "total_needed": total_needed, "token_amount": token_amount, @@ -1281,7 +1044,7 @@ async def swap_to_trusted_mint( continue try: - _ = await _mint_operation( + melt_response = await run_mint_operation( lambda: token_wallet.melt( proofs=token_obj.proofs, invoice=mint_quote.request, @@ -1292,36 +1055,45 @@ async def swap_to_trusted_mint( mint_url=token_obj.mint, retry_timeouts=False, ) + await _confirm_melt_paid( + token_wallet, melt_quote.quote, token_obj.proofs, melt_response + ) except Exception as e: - # A down mint won't fix itself by retrying with a smaller amount. - if is_mint_connection_error(e): - logger.error( - "Source mint became unreachable during melt", - extra={ - "event": "cashu_source_mint_unreachable", - "stage": "source_melt", - "error": str(e), - "error_type": type(e).__name__, - "source_mint": token_obj.mint, - "destination_mint": dest_mint_url, - "attempt": attempt, - }, - ) - raise SourceMintConnectionError( - "Issuing Cashu mint is unreachable" - ) from e shortfall = _melt_insufficient_shortfall(e) - recomputed = 0 - if shortfall is not None: - observed_extra_fee += shortfall - recomputed = _net_minted_amount( - amount_msat, - token_obj.unit, - melt_quote.fee_reserve + input_fees + observed_extra_fee, - ) - if shortfall is None or attempt >= _MAX_SWAP_ATTEMPTS or recomputed <= 0: + if shortfall is None: + if isinstance(e, TokenConsumedError): + raise + if _melt_definitively_failed(e): + raise ValueError( + f"Failed to melt token from foreign mint {token_obj.mint}: {e}" + ) from e + if is_mint_connection_error(e): + await _reconcile_ambiguous_melt( + token_wallet, melt_quote.quote, token_obj.proofs + ) + logger.info( + "Source melt reconciled as paid; minting on destination", + extra={ + "event": "cashu_source_melt_reconciled_paid", + "source_mint": token_obj.mint, + "destination_mint": dest_mint_url, + "melt_quote_id": melt_quote.quote, + }, + ) + break + raise TokenConsumedError( + "Source melt failed after dispatch; outcome requires reconciliation" + ) from e + + observed_extra_fee += shortfall + recomputed = _net_minted_amount( + amount_msat, + token_obj.unit, + melt_quote.fee_reserve + input_fees + observed_extra_fee, + ) + if attempt >= _MAX_SWAP_ATTEMPTS or recomputed <= 0: logger.error( - "swap_to_primary_mint: melt failed", + "swap_to_trusted_mint: melt failed", extra={ "error": str(e), "error_type": type(e).__name__, @@ -1336,7 +1108,7 @@ async def swap_to_trusted_mint( f"Failed to melt token from foreign mint {token_obj.mint}: {e}" ) from e logger.warning( - "swap_to_primary_mint: mint demanded more than quoted at melt, retrying", + "swap_to_trusted_mint: mint demanded more than quoted at melt, retrying", extra={ "shortfall": shortfall, "retry_minted_amount": recomputed, @@ -1348,7 +1120,7 @@ async def swap_to_trusted_mint( break - logger.warning( + logger.info( "Source melt succeeded; minting on destination", extra={ "event": "cashu_destination_mint_attempt", @@ -1361,9 +1133,9 @@ async def swap_to_trusted_mint( await dest_wallet.load_proofs(reload=True) pre_mint_balance = dest_wallet.available_balance.amount try: - _ = await _mint_operation( + _ = await run_mint_operation( lambda: dest_wallet.mint(minted_amount, quote_id=mint_quote.quote), - op_name="swap_mint_on_primary", + op_name="swap_mint_on_destination", mint_url=dest_mint_url, retry_timeouts=False, ) @@ -1373,7 +1145,7 @@ async def swap_to_trusted_mint( # bump_secret_derivation ran locally. Recover orphaned proofs and # advance the counter so the next request derives fresh secrets. logger.warning( - "swap_to_primary_mint: outputs already signed — recovering orphaned proofs", + "swap_to_trusted_mint: outputs already signed — recovering orphaned proofs", extra={ "mint_quote_id": mint_quote.quote, "minted_amount": minted_amount, @@ -1388,7 +1160,7 @@ async def swap_to_trusted_mint( post_recovery_balance = dest_wallet.available_balance.amount balance_gained = post_recovery_balance - pre_mint_balance logger.info( - "swap_to_primary_mint: recovery scan completed", + "swap_to_trusted_mint: recovery scan completed", extra={ "pre_mint_balance": pre_mint_balance, "post_recovery_balance": post_recovery_balance, @@ -1411,7 +1183,7 @@ async def swap_to_trusted_mint( raise except Exception as recovery_err: logger.error( - "swap_to_primary_mint: recovery failed", + "swap_to_trusted_mint: recovery failed", extra={"error": str(recovery_err)}, ) raise TokenConsumedError( @@ -1419,7 +1191,7 @@ async def swap_to_trusted_mint( ) from e else: logger.error( - "swap_to_primary_mint: mint on primary failed after successful melt", + "swap_to_trusted_mint: mint on primary failed after successful melt", extra={ "error": str(e), "error_type": type(e).__name__, @@ -1432,7 +1204,7 @@ async def swap_to_trusted_mint( "Mint on primary failed after successful melt" ) from e - logger.warning( + logger.info( "Cashu cross-mint swap completed", extra={ "event": "cashu_swap_completed", @@ -1595,13 +1367,13 @@ async def get_wallet( now = time.monotonic() last = _wallet_last_load.get(id) if last is None or now - last >= _WALLOAD_RELOAD_MIN_INTERVAL_SECONDS: - await _mint_operation( + await run_mint_operation( lambda: _wallets[id].load_mint(), op_name="load_mint", mint_url=mint_url, retry_on_rate_limit=retry_on_rate_limit, ) - await _mint_operation( + await run_mint_operation( lambda: _wallets[id].load_proofs(reload=True), op_name="load_proofs", mint_url=mint_url, @@ -1640,7 +1412,7 @@ async def slow_filter_spend_proofs( batch_size = 1000 for i in range(0, len(proofs), batch_size): pb = proofs[i : i + batch_size] - proof_states = await _mint_operation( + proof_states = await run_mint_operation( lambda: wallet.check_proof_state(pb), op_name="check_proof_state", mint_url=str(wallet.url), @@ -1652,12 +1424,7 @@ async def slow_filter_spend_proofs( else: _spent_proofs.append(proof) if _spent_proofs: - await _mint_operation( - lambda: wallet.set_reserved_for_send(_spent_proofs, reserved=True), - op_name="set_reserved_spent_proofs", - mint_url=str(wallet.url), - retry_timeouts=False, - ) + await wallet.set_reserved_for_send(_spent_proofs, reserved=True) return _proofs @@ -1686,7 +1453,7 @@ async def _get_supported_mint_units(mint_url: str) -> list[str]: return cached[1] wallet = await get_wallet(mint_url, settings.primary_mint_unit, load=False) - keysets = await _mint_operation( + keysets = await run_mint_operation( lambda: wallet._get_keysets(), op_name="get_mint_keysets", mint_url=mint_url, @@ -1750,7 +1517,7 @@ async def fetch_all_balances( mint_units[mint_url] = await _get_supported_mint_units(mint_url) except Exception as error: connection_failure = is_mint_connection_error(error) - rate_limited = _is_mint_rate_limited(error) + rate_limited = is_mint_rate_limited(error) error_code = ( "rate_limited" if rate_limited @@ -1759,12 +1526,12 @@ async def fetch_all_balances( else "mint_error" ) if connection_failure: - _MintRateGuard.get(mint_url).apply_cooldown( + MintRateGuard.get(mint_url).apply_cooldown( _BALANCE_FETCH_RETRY_SECONDS, reason="unreachable" ) retry_delay = max( _BALANCE_FETCH_RETRY_SECONDS, - _mint_cooldown_remaining(mint_url), + mint_cooldown_remaining(mint_url), ) discovery_errors.append( _balance_error( @@ -1819,9 +1586,9 @@ async def fetch_all_balances( retry_after_seconds=failure[0] - now, ) - cooldown = _mint_cooldown_remaining(mint_url) + cooldown = mint_cooldown_remaining(mint_url) if cooldown > 0: - error_code = _mint_cooldown_reason(mint_url) or "cooldown" + error_code = mint_cooldown_reason(mint_url) or "cooldown" error = { "rate_limited": "Mint is rate limited", "unreachable": "Mint is unreachable", @@ -1846,7 +1613,7 @@ async def fetch_all_balances( proofs = await slow_filter_spend_proofs(proofs, wallet) except Exception as error: connection_failure = is_mint_connection_error(error) - rate_limited = _is_mint_rate_limited(error) + rate_limited = is_mint_rate_limited(error) error_code = ( "rate_limited" if rate_limited @@ -1855,16 +1622,16 @@ async def fetch_all_balances( else "mint_error" ) if rate_limited: - _MintRateGuard.get(mint_url).apply_rate_limit_cooldown( + MintRateGuard.get(mint_url).apply_rate_limit_cooldown( _BALANCE_FETCH_RETRY_SECONDS ) elif connection_failure: - _MintRateGuard.get(mint_url).apply_cooldown( + MintRateGuard.get(mint_url).apply_cooldown( _BALANCE_FETCH_RETRY_SECONDS, reason=error_code ) retry_delay = max( _BALANCE_FETCH_RETRY_SECONDS, - _mint_cooldown_remaining(mint_url), + mint_cooldown_remaining(mint_url), ) _balance_fetch_failures[key] = ( time.monotonic() + retry_delay, diff --git a/tests/integration/test_insufficient_balance.py b/tests/integration/test_insufficient_balance.py index 3acc5db9..a63c4251 100644 --- a/tests/integration/test_insufficient_balance.py +++ b/tests/integration/test_insufficient_balance.py @@ -208,25 +208,30 @@ async def test_pay_for_request_succeeds_when_balance_equals_cost( @pytest.mark.asyncio -async def test_five_percent_mint_fallback_headroom_is_admitted_and_reserved( +async def test_full_model_maximum_is_required_and_reserved( integration_session: AsyncSession, ) -> None: from routstr.auth import pay_for_request, validate_bearer_key - from routstr.payment.helpers import apply_mint_fee_allowance - key = _key(balance=95_000) - integration_session.add(key) + short_key = _key(balance=95_000) + exact_key = _key(balance=100_000) + integration_session.add(short_key) + integration_session.add(exact_key) await integration_session.commit() - admission_cost = apply_mint_fee_allowance(100_000) - validated = await validate_bearer_key( - f"sk-{key.hashed_key}", integration_session, min_cost=admission_cost - ) - await pay_for_request(validated, admission_cost, integration_session) + with pytest.raises(HTTPException) as insufficient: + await validate_bearer_key( + f"sk-{short_key.hashed_key}", integration_session, min_cost=100_000 + ) + assert insufficient.value.status_code == 402 - await integration_session.refresh(key) - assert admission_cost == 95_000 - assert key.reserved_balance == 95_000 + validated = await validate_bearer_key( + f"sk-{exact_key.hashed_key}", integration_session, min_cost=100_000 + ) + await pay_for_request(validated, 100_000, integration_session) + + await integration_session.refresh(exact_key) + assert exact_key.reserved_balance == 100_000 # --------------------------------------------------------------------------- @@ -288,7 +293,7 @@ async def test_http_402_response_shape_on_insufficient_balance( error = body["detail"]["error"] assert error["code"] == "insufficient_balance" assert error["type"] == "insufficient_quota" - assert "591.744 sats (591744 msats) required" in error["message"] + assert "622.888 sats (622888 msats) required" in error["message"] assert "20.32 sats (20320 msats) available" in error["message"] # Balance must be completely untouched diff --git a/tests/integration/test_swap_fee_retry.py b/tests/integration/test_swap_fee_retry.py index 138a4d4d..b27360a2 100644 --- a/tests/integration/test_swap_fee_retry.py +++ b/tests/integration/test_swap_fee_retry.py @@ -20,6 +20,7 @@ from collections.abc import Callable from unittest.mock import AsyncMock, Mock, patch import pytest +from cashu.core.base import MeltQuoteState from httpx import AsyncClient, Response from routstr.core.settings import settings @@ -81,7 +82,9 @@ def _make_swap_mocks( quote=f"melt_quote_{invoice}", amount=invoice, fee_reserve=_next_fee() ) ) - mock_token_wallet.melt = AsyncMock(return_value=Mock()) + mock_token_wallet.melt = AsyncMock( + return_value=Mock(state=MeltQuoteState.paid) + ) return mock_token, mock_token_wallet, mock_primary_wallet @@ -144,7 +147,7 @@ async def test_topup_retries_when_melt_demands_more_than_quoted( "Mint Error: not enough inputs provided for melt. " "Provided: 179, needed: 180 (Code: 11000)" ), - Mock(), + Mock(state=MeltQuoteState.paid), ] response = await _post_topup( diff --git a/tests/unit/test_fetch_all_balances.py b/tests/unit/test_fetch_all_balances.py index 7618ed47..ceeca1e4 100644 --- a/tests/unit/test_fetch_all_balances.py +++ b/tests/unit/test_fetch_all_balances.py @@ -161,7 +161,7 @@ async def test_fetch_all_balances_backs_off_after_connection_failure() -> None: patch.object(settings, "primary_mint", "http://mint:3338"), patch("routstr.wallet.get_wallet", get_wallet), patch("routstr.wallet.db.create_session", _fake_session), - patch("routstr.wallet.time.monotonic", return_value=10), + patch("routstr.mint.time.monotonic", return_value=10), patch("routstr.wallet.logger.warning") as warning, ): first = await fetch_all_balances(units=["sat"]) @@ -180,7 +180,7 @@ async def test_fetch_all_balances_backs_off_after_connection_failure() -> None: patch.object(settings, "primary_mint", "http://mint:3338"), patch("routstr.wallet.get_wallet", get_wallet), patch("routstr.wallet.db.create_session", _fake_session), - patch("routstr.wallet.time.monotonic", return_value=71), + patch("routstr.mint.time.monotonic", return_value=71), patch("routstr.wallet.logger.warning"), ): await fetch_all_balances(units=["sat"]) @@ -219,7 +219,7 @@ async def test_balance_failure_applies_mint_cooldown_to_other_units() -> None: patch.object(settings, "primary_mint", mint), patch("routstr.wallet.get_wallet", get_wallet), patch("routstr.wallet.db.create_session", _fake_session), - patch("routstr.wallet.time.monotonic", return_value=10), + patch("routstr.mint.time.monotonic", return_value=10), patch("routstr.wallet.logger.warning") as warning, ): details, *_ = await fetch_all_balances(units=["sat", "msat"]) diff --git a/tests/unit/test_lightning_settlement.py b/tests/unit/test_lightning_settlement.py index c44954f2..d8741b7e 100644 --- a/tests/unit/test_lightning_settlement.py +++ b/tests/unit/test_lightning_settlement.py @@ -155,6 +155,7 @@ async def test_non_pending_invoice_is_not_minted() -> None: get_wallet.assert_not_awaited() session.commit.assert_awaited_once() + assert _invoice_settlement_locks == {} @pytest.mark.asyncio @@ -212,3 +213,4 @@ async def test_concurrent_invoice_checks_finalize_once_in_process() -> None: assert invoice.status == "paid" finalize.assert_awaited_once() + assert _invoice_settlement_locks == {} diff --git a/tests/unit/test_lnurl_melt_timeout.py b/tests/unit/test_lnurl_melt_timeout.py index 47efbcf2..7311b4a2 100644 --- a/tests/unit/test_lnurl_melt_timeout.py +++ b/tests/unit/test_lnurl_melt_timeout.py @@ -1,70 +1,127 @@ -"""raw_send_to_lnurl() must not hang forever on an unresponsive mint. - -The Cashu library issues POST /v1/melt/bolt11 with timeout=None, so a hung -mint would block the melt (and the payout loop) indefinitely. raw_send_to_lnurl -now wraps wallet.melt() in asyncio.wait_for(MELT_TIMEOUT_SECONDS) and surfaces a -timeout as LNURLError instead of hanging. -""" +"""LNURL melt attempts must not misclassify ambiguous payment outcomes.""" import asyncio +from typing import Any from unittest.mock import AsyncMock, MagicMock, patch import pytest +from cashu.core.base import MeltQuoteState -from routstr.payment import lnurl +from routstr.core.settings import settings from routstr.payment.lnurl import LNURLError, raw_send_to_lnurl +LNURL_DATA = { + "callback_url": "https://ln.tld/cb", + "min_sendable": 1_000, + "max_sendable": 100_000_000, +} -@pytest.mark.asyncio -async def test_raw_send_to_lnurl_times_out_on_hung_melt() -> None: + +def _wallet() -> tuple[MagicMock, list[MagicMock]]: proofs = [MagicMock(amount=1000)] - - wallet = MagicMock() + wallet = MagicMock(url="https://mint.test") wallet.melt_quote = AsyncMock(return_value=MagicMock(fee_reserve=1, quote="q")) wallet.select_to_send = AsyncMock(return_value=(proofs, None)) + return wallet, proofs + + +def _lnurl_patches() -> tuple[Any, Any]: + return ( + patch( + "routstr.payment.lnurl.get_lnurl_data", + AsyncMock(return_value=LNURL_DATA), + ), + patch( + "routstr.payment.lnurl.get_lnurl_invoice", + AsyncMock(return_value=("lnbc1...", {})), + ), + ) + + +@pytest.mark.asyncio +async def test_raw_send_to_lnurl_timeout_keeps_unpaid_outcome_ambiguous() -> None: + wallet, proofs = _wallet() async def _hang(**kwargs: object) -> None: - await asyncio.sleep(5) # far longer than the patched timeout + await asyncio.sleep(5) wallet.melt = AsyncMock(side_effect=_hang) + wallet.get_melt_quote = AsyncMock( + return_value=MagicMock(state=MeltQuoteState.unpaid) + ) + data_patch, invoice_patch = _lnurl_patches() - lnurl_data = { - "callback_url": "https://ln.tld/cb", - "min_sendable": 1_000, - "max_sendable": 100_000_000, - } - - with patch.object(lnurl, "MELT_TIMEOUT_SECONDS", 0.05), patch( - "routstr.payment.lnurl.get_lnurl_data", AsyncMock(return_value=lnurl_data) - ), patch( - "routstr.payment.lnurl.get_lnurl_invoice", - AsyncMock(return_value=("lnbc1...", {})), + with ( + patch.object(settings, "mint_operation_timeout_seconds", 0.05), + patch.object(settings, "mint_retry_max_attempts", 0), + data_patch, + invoice_patch, + pytest.raises(LNURLError, match="outcome is ambiguous"), ): - with pytest.raises(LNURLError, match="Melt timed out"): - await raw_send_to_lnurl(wallet, proofs, "owner@ln.tld", "sat", amount=1000) + await raw_send_to_lnurl(wallet, proofs, "owner@ln.tld", "sat", amount=1000) + + wallet.get_melt_quote.assert_awaited_once_with("q") + wallet.set_reserved_for_melt.assert_not_called() @pytest.mark.asyncio -async def test_raw_send_to_lnurl_succeeds_within_timeout() -> None: - """A prompt melt still returns the net amount, unaffected by the guard.""" - proofs = [MagicMock(amount=1000)] +async def test_raw_send_to_lnurl_timeout_reconciled_paid_is_success() -> None: + wallet, proofs = _wallet() - wallet = MagicMock() - wallet.melt_quote = AsyncMock(return_value=MagicMock(fee_reserve=1, quote="q")) - wallet.select_to_send = AsyncMock(return_value=(proofs, None)) - wallet.melt = AsyncMock(return_value=MagicMock()) + async def _hang(**kwargs: object) -> None: + await asyncio.sleep(5) - lnurl_data = { - "callback_url": "https://ln.tld/cb", - "min_sendable": 1_000, - "max_sendable": 100_000_000, - } + wallet.melt = AsyncMock(side_effect=_hang) + wallet.get_melt_quote = AsyncMock( + return_value=MagicMock(state=MeltQuoteState.paid) + ) + data_patch, invoice_patch = _lnurl_patches() - with patch.object(lnurl, "MELT_TIMEOUT_SECONDS", 5), patch( - "routstr.payment.lnurl.get_lnurl_data", AsyncMock(return_value=lnurl_data) - ), patch( - "routstr.payment.lnurl.get_lnurl_invoice", - AsyncMock(return_value=("lnbc1...", {})), + with ( + patch.object(settings, "mint_operation_timeout_seconds", 0.05), + patch.object(settings, "mint_retry_max_attempts", 0), + data_patch, + invoice_patch, + ): + paid = await raw_send_to_lnurl( + wallet, proofs, "owner@ln.tld", "sat", amount=1000 + ) + + assert paid > 0 + wallet.get_melt_quote.assert_awaited_once_with("q") + + +@pytest.mark.asyncio +async def test_raw_send_to_lnurl_pending_response_stays_ambiguous() -> None: + wallet, proofs = _wallet() + wallet.melt = AsyncMock(return_value=MagicMock(state=MeltQuoteState.pending)) + wallet.get_melt_quote = AsyncMock( + return_value=MagicMock(state=MeltQuoteState.pending) + ) + data_patch, invoice_patch = _lnurl_patches() + + with ( + patch.object(settings, "mint_operation_timeout_seconds", 5), + data_patch, + invoice_patch, + pytest.raises(LNURLError, match="outcome is ambiguous"), + ): + await raw_send_to_lnurl(wallet, proofs, "owner@ln.tld", "sat", amount=1000) + + wallet.get_melt_quote.assert_awaited_once_with("q") + + +@pytest.mark.asyncio +async def test_raw_send_to_lnurl_succeeds_on_explicit_paid_response() -> None: + wallet, proofs = _wallet() + wallet.melt = AsyncMock(return_value=MagicMock(state=MeltQuoteState.paid)) + wallet.get_melt_quote = AsyncMock() + data_patch, invoice_patch = _lnurl_patches() + + with ( + patch.object(settings, "mint_operation_timeout_seconds", 5), + data_patch, + invoice_patch, ): paid = await raw_send_to_lnurl( wallet, proofs, "owner@ln.tld", "sat", amount=1000 @@ -72,3 +129,4 @@ async def test_raw_send_to_lnurl_succeeds_within_timeout() -> None: assert paid > 0 wallet.melt.assert_awaited_once() + wallet.get_melt_quote.assert_not_awaited() diff --git a/tests/unit/test_melt_reconciliation.py b/tests/unit/test_melt_reconciliation.py new file mode 100644 index 00000000..a68cb64b --- /dev/null +++ b/tests/unit/test_melt_reconciliation.py @@ -0,0 +1,91 @@ +from unittest.mock import AsyncMock, Mock + +import pytest +from cashu.core.base import MeltQuoteState, ProofSpentState + +from routstr.wallet import ( + TokenConsumedError, + _confirm_melt_paid, + _reconcile_ambiguous_melt, +) + + +@pytest.mark.asyncio +async def test_paid_quote_is_authoritative_when_proof_lookup_would_fail() -> None: + wallet = Mock( + url="http://source-mint:3338", + get_melt_quote=AsyncMock(return_value=Mock(state=MeltQuoteState.paid)), + check_proof_state=AsyncMock(side_effect=RuntimeError("proof API unavailable")), + ) + + assert await _reconcile_ambiguous_melt(wallet, "quote-1", [Mock()]) is True + wallet.check_proof_state.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_timeout_snapshot_unpaid_unspent_remains_non_retryable() -> None: + wallet = Mock( + url="http://source-mint:3338", + get_melt_quote=AsyncMock(return_value=Mock(state=MeltQuoteState.unpaid)), + check_proof_state=AsyncMock( + return_value=Mock(states=[Mock(state=ProofSpentState.unspent)]) + ), + ) + + with pytest.raises(TokenConsumedError, match="ambiguous"): + await _reconcile_ambiguous_melt(wallet, "quote-2", [Mock()]) + + +@pytest.mark.asyncio +async def test_successful_pending_melt_response_requires_reconciliation() -> None: + wallet = Mock( + url="http://source-mint:3338", + get_melt_quote=AsyncMock(return_value=Mock(state=MeltQuoteState.pending)), + check_proof_state=AsyncMock( + return_value=Mock(states=[Mock(state=ProofSpentState.pending)]) + ), + ) + + with pytest.raises(TokenConsumedError, match="ambiguous"): + await _confirm_melt_paid( + wallet, + "quote-pending", + [Mock()], + Mock(state=MeltQuoteState.pending), + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("quote_state", "proof_state"), + [ + (MeltQuoteState.pending, ProofSpentState.pending), + (MeltQuoteState.unpaid, ProofSpentState.spent), + (MeltQuoteState.unpaid, ProofSpentState.pending), + ], +) +async def test_ambiguous_or_consumed_melt_is_never_reported_unspent( + quote_state: MeltQuoteState, proof_state: ProofSpentState +) -> None: + wallet = Mock( + url="http://source-mint:3338", + get_melt_quote=AsyncMock(return_value=Mock(state=quote_state)), + check_proof_state=AsyncMock( + return_value=Mock(states=[Mock(state=proof_state)]) + ), + ) + + with pytest.raises(TokenConsumedError, match="reconciliation required"): + await _reconcile_ambiguous_melt(wallet, "quote-3", [Mock()]) + + +@pytest.mark.asyncio +async def test_failed_melt_reconciliation_is_non_retryable() -> None: + wallet = Mock( + url="http://source-mint:3338", + get_melt_quote=AsyncMock(side_effect=RuntimeError("mint unavailable")), + check_proof_state=AsyncMock(), + ) + + with pytest.raises(TokenConsumedError, match="outcome is unknown"): + await _reconcile_ambiguous_melt(wallet, "quote-4", [Mock()]) diff --git a/tests/unit/test_mint.py b/tests/unit/test_mint.py new file mode 100644 index 00000000..5a258a29 --- /dev/null +++ b/tests/unit/test_mint.py @@ -0,0 +1,65 @@ +from unittest.mock import AsyncMock, Mock, patch + +import httpx +import pytest +from cashu.core.base import Unit + +from routstr.mint import ( + MintCooldownError, + MintRateGuard, + MintRateLimitedError, + fail_fast_mint_operations, +) +from routstr.wallet import Wallet + + +@pytest.mark.asyncio +async def test_cooldown_fails_fast_while_wallet_mutation_scope_is_held() -> None: + guard = MintRateGuard("http://mint:3338", max_concurrency=1) + guard.apply_cooldown(3600, reason="rate_limited") + operation = AsyncMock(return_value="should not run") + + with ( + patch("routstr.mint.asyncio.sleep", AsyncMock()) as sleep, + pytest.raises(MintCooldownError) as caught, + ): + async with fail_fast_mint_operations(): + await guard.run(operation) + + assert caught.value.retry_after_seconds > 0 + operation.assert_not_awaited() + sleep.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_cashu_429_dispatches_through_wallet_override() -> None: + async def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 429, + request=request, + json={"detail": "too many requests", "code": 42900}, + ) + + wallet = object.__new__(Wallet) + wallet.url = "http://mint:3338" + wallet.db = Mock() + wallet.keysets = {"loaded": Mock()} + wallet.mint_info = Mock() + wallet.mint_info.requires_blind_auth_path.return_value = False + wallet.mint_info.requires_clear_auth_path.return_value = False + wallet.auth_db = None + wallet.auth_keyset_id = None + + real_client = httpx.AsyncClient + + def client_factory(*args: object, **kwargs: object) -> httpx.AsyncClient: + return real_client( + transport=httpx.MockTransport(handler), + base_url=str(kwargs["base_url"]), + ) + + with ( + patch("cashu.wallet.v1_api.httpx.AsyncClient", side_effect=client_factory), + pytest.raises(MintRateLimitedError), + ): + await wallet.mint_quote(1, Unit.sat) diff --git a/tests/unit/test_payment_helpers.py b/tests/unit/test_payment_helpers.py index fe0cd573..ef8dde63 100644 --- a/tests/unit/test_payment_helpers.py +++ b/tests/unit/test_payment_helpers.py @@ -7,21 +7,7 @@ os.environ["UPSTREAM_BASE_URL"] = "http://test" os.environ["UPSTREAM_API_KEY"] = "test" from routstr.core.settings import settings # noqa: E402 -from routstr.payment.helpers import ( # noqa: E402 - apply_mint_fee_allowance, - get_max_cost_for_model, -) - - -def test_mint_fee_allowance_reserves_five_percent_fallback_headroom() -> None: - # Interim policy: Routstr may pay hidden cross-mint Lightning fees when a - # trusted-mint fallback is required. - assert apply_mint_fee_allowance(124_886) == 118_642 - - -def test_mint_fee_allowance_never_drops_below_minimum() -> None: - with patch.object(settings, "min_request_msat", 100): - assert apply_mint_fee_allowance(50) == 100 +from routstr.payment.helpers import get_max_cost_for_model # noqa: E402 async def test_get_max_cost_for_model_known() -> None: diff --git a/tests/unit/test_stale_reservations.py b/tests/unit/test_stale_reservations.py index 512dfa31..31dd5767 100644 --- a/tests/unit/test_stale_reservations.py +++ b/tests/unit/test_stale_reservations.py @@ -422,4 +422,4 @@ async def test_proxy_reverts_reservation_on_client_disconnect() -> None: with pytest.raises(asyncio.CancelledError): await proxy_module.proxy(request, "v1/chat/completions", session=session) - revert_mock.assert_awaited_once_with(key, session, 950, reservation_snapshot) + revert_mock.assert_awaited_once_with(key, session, 1000, reservation_snapshot) diff --git a/tests/unit/test_upstream_rate_limit.py b/tests/unit/test_upstream_rate_limit.py index 65b50a73..495f1e57 100644 --- a/tests/unit/test_upstream_rate_limit.py +++ b/tests/unit/test_upstream_rate_limit.py @@ -406,4 +406,4 @@ async def test_proxy_loop_surfaces_rate_limit_and_reverts_once() -> None: assert RAW_ORG_ID not in serialized assert "org-[REDACTED]" in serialized # Single upstream failed -> reservation reverted exactly once (no double-charge). - revert_mock.assert_awaited_once_with(key, session, 950, reservation) + revert_mock.assert_awaited_once_with(key, session, 1000, reservation) diff --git a/tests/unit/test_wallet.py b/tests/unit/test_wallet.py index 257b6132..03123a60 100644 --- a/tests/unit/test_wallet.py +++ b/tests/unit/test_wallet.py @@ -7,6 +7,7 @@ from unittest.mock import AsyncMock, Mock, patch import httpx import pytest +from cashu.core.base import MeltQuoteState from routstr.core.db import ApiKey from routstr.wallet import ( @@ -639,7 +640,9 @@ def _make_swap_mocks( quote=f"melt_quote_{invoice}", amount=invoice, fee_reserve=_next_fee() ) ) - mock_token_wallet.melt = AsyncMock(return_value=Mock()) + mock_token_wallet.melt = AsyncMock( + return_value=Mock(state=MeltQuoteState.paid) + ) return mock_token, mock_token_wallet, mock_primary_wallet @@ -770,7 +773,7 @@ async def test_swap_retries_when_melt_demands_more_than_quoted() -> None: "Mint Error: not enough inputs provided for melt. " "Provided: 179, needed: 180 (Code: 11000)" ), - Mock(), + Mock(state=MeltQuoteState.paid), ] from routstr.core.settings import settings @@ -801,7 +804,7 @@ async def test_swap_retries_on_cdk_unbalanced_error() -> None: ) mock_token_wallet.melt.side_effect = [ Exception("Mint Error: Transaction unbalanced: 179, 178, 2 (Code: 11005)"), - Mock(), + Mock(state=MeltQuoteState.paid), ] from routstr.core.settings import settings @@ -1522,22 +1525,32 @@ async def test_swap_fee_estimation_transport_error_raises_mint_connection_error( @pytest.mark.asyncio -async def test_swap_melt_transport_error_raises_mint_connection_error() -> None: - """A transport failure during melt is surfaced as MintConnectionError and - is NOT retried — the mint is down, not demanding higher fees.""" +async def test_swap_melt_transport_error_is_never_reported_reusable() -> None: + """A timed-out melt remains ambiguous even when an immediate snapshot says + UNPAID/UNSPENT, so callers must not receive the original token for retry.""" from routstr.wallet import swap_to_primary_mint mock_token, mock_token_wallet, mock_primary_wallet = _make_swap_mocks( 1000, fee_reserves=[10, 10] ) mock_token_wallet.melt = AsyncMock(side_effect=httpx.ConnectTimeout("timed out")) + from cashu.core.base import MeltQuoteState, ProofSpentState + + mock_token_wallet.get_melt_quote = AsyncMock( + return_value=Mock(state=MeltQuoteState.unpaid) + ) + mock_token_wallet.check_proof_state = AsyncMock( + return_value=Mock( + states=[Mock(state=ProofSpentState.unspent) for _ in mock_token.proofs] + ) + ) from routstr.core.settings import settings with patch.object(settings, "primary_mint", "http://primary:3338"): with patch.object(settings, "primary_mint_unit", "sat"): with patch("routstr.wallet.get_wallet", return_value=mock_primary_wallet): - with pytest.raises(MintConnectionError): + with pytest.raises(TokenConsumedError, match="ambiguous"): await swap_to_primary_mint(mock_token, mock_token_wallet) assert mock_token_wallet.melt.call_count == 1 @@ -1595,8 +1608,8 @@ async def test_mint_rate_guard_waits_for_adaptive_cooldown() -> None: guard._cooldown_until = 15.0 operation = AsyncMock(return_value="ok") - with patch("routstr.wallet.time.monotonic", return_value=10.0): - with patch("routstr.wallet.asyncio.sleep", AsyncMock()) as sleep: + with patch("routstr.mint.time.monotonic", return_value=10.0): + with patch("routstr.mint.asyncio.sleep", AsyncMock()) as sleep: assert await guard.run(operation) == "ok" sleep.assert_awaited_once_with(5.0) @@ -1611,7 +1624,7 @@ async def test_mint_rate_guard_exponentially_backs_off_repeated_429s() -> None: expected_delays = [60, 120, 240, 480, 960, 1920, 3840, 7680, 15360, 25200] now = 0.0 - with patch("routstr.wallet.time.monotonic") as monotonic: + with patch("routstr.mint.time.monotonic") as monotonic: for index, expected in enumerate(expected_delays, start=1): monotonic.return_value = now assert guard.apply_rate_limit_cooldown(60) == expected @@ -1682,8 +1695,8 @@ async def test_mint_rate_guard_keeps_cooldown_when_concurrency_is_unlimited() -> operation = AsyncMock(return_value="ok") with ( patch.object(settings, "mint_max_concurrency", 0), - patch("routstr.wallet.time.monotonic", return_value=0), - patch("routstr.wallet.asyncio.sleep", AsyncMock()) as sleep, + patch("routstr.mint.time.monotonic", return_value=0), + patch("routstr.mint.asyncio.sleep", AsyncMock()) as sleep, ): guard = _MintRateGuard.get("http://mint:3338") guard.apply_cooldown(5) @@ -1715,8 +1728,8 @@ async def test_mint_operation_honors_retry_after_as_minimum() -> None: with patch.object(settings, "mint_retry_max_attempts", 1): with patch.object(settings, "mint_operation_timeout_seconds", 0): with patch.object(settings, "mint_max_concurrency", 1): - with patch("routstr.wallet.time.monotonic", return_value=0.1): - with patch("routstr.wallet.asyncio.sleep", sleep): + with patch("routstr.mint.time.monotonic", return_value=0.1): + with patch("routstr.mint.asyncio.sleep", sleep): result = await _mint_operation( factory, mint_url="http://mint:3338" ) @@ -1734,7 +1747,7 @@ async def test_mint_operation_timeout_excludes_adaptive_cooldown() -> None: with ( patch.object(settings, "mint_max_concurrency", 1), patch.object(settings, "mint_operation_timeout_seconds", 0.01), - patch("routstr.wallet.asyncio.sleep", AsyncMock()) as sleep, + patch("routstr.mint.asyncio.sleep", AsyncMock()) as sleep, ): guard = _MintRateGuard.get("http://mint:3338") guard.apply_cooldown(60) @@ -1763,7 +1776,7 @@ async def test_default_timeout_allows_retry_after_rate_limit_cooldown() -> None: patch.object(settings, "mint_retry_max_attempts", 3), patch.object(settings, "mint_operation_timeout_seconds", 30), patch.object(settings, "mint_max_concurrency", 1), - patch("routstr.wallet.asyncio.sleep", AsyncMock()), + patch("routstr.mint.asyncio.sleep", AsyncMock()), ): assert await _mint_operation(operation, mint_url="http://mint:3338") == "ok" @@ -1780,7 +1793,7 @@ async def test_mint_operation_retries_httpx_timeout_only_when_safe() -> None: with patch.object(settings, "mint_retry_max_attempts", 2): with patch.object(settings, "mint_operation_timeout_seconds", 0): - with patch("routstr.wallet.asyncio.sleep", AsyncMock()): + with patch("routstr.mint.asyncio.sleep", AsyncMock()): assert await _mint_operation(retrying) == "ok" with pytest.raises(httpx.TimeoutException): await _mint_operation(non_retrying, retry_timeouts=False) @@ -1802,7 +1815,7 @@ async def test_get_wallet_initializes_and_loads_once_concurrently() -> None: ) as create: # A fresh wallet must load even when the host has been up for less than # the reload interval. - with patch("routstr.wallet.time.monotonic", return_value=10.0): + with patch("routstr.mint.time.monotonic", return_value=10.0): first, second = await asyncio.gather( get_wallet("http://mint:3338"), get_wallet("http://mint:3338") ) @@ -1833,7 +1846,7 @@ async def test_get_wallet_can_surface_429_without_retrying() -> None: patch("routstr.wallet.Wallet.with_db", AsyncMock(return_value=wallet)), patch.object(settings, "mint_retry_max_attempts", 3), patch.object(settings, "mint_operation_timeout_seconds", 0), - patch("routstr.wallet.asyncio.sleep", AsyncMock()) as sleep, + patch("routstr.mint.asyncio.sleep", AsyncMock()) as sleep, ): with pytest.raises(httpx.HTTPStatusError): await get_wallet("http://mint:3338", retry_on_rate_limit=False) @@ -1965,7 +1978,7 @@ async def test_swap_falls_back_when_primary_wallet_cannot_load() -> None: melt_quote=AsyncMock( return_value=Mock(quote="melt_q", amount=990, fee_reserve=10) ), - melt=AsyncMock(return_value=Mock()), + melt=AsyncMock(return_value=Mock(state=MeltQuoteState.paid)), ) mint_quote = Mock(quote="mint_q_secondary", request="lnbc1secondary") @@ -1994,6 +2007,7 @@ async def test_swap_falls_back_when_primary_wallet_cannot_load() -> None: patch("asyncio.sleep", AsyncMock()), patch("routstr.wallet.get_wallet", side_effect=mock_get), patch("routstr.wallet.logger.warning") as warning, + patch("routstr.wallet.logger.info") as info, ): amount, unit, mint_url = await swap_to_primary_mint(token, source_wallet) @@ -2003,7 +2017,7 @@ async def test_swap_falls_back_when_primary_wallet_cannot_load() -> None: assert any(call.args[0] == secondary for call in mock_get.await_args_list) events = { call.kwargs["extra"]["event"] - for call in warning.call_args_list + for call in [*warning.call_args_list, *info.call_args_list] if "extra" in call.kwargs and "event" in call.kwargs["extra"] } assert "cashu_destination_failed" in events @@ -2181,8 +2195,8 @@ async def test_wallet_fallback_skips_mint_during_cooldown() -> None: patch.object(settings, "cashu_mints", [primary, secondary]), patch.object(settings, "mint_max_concurrency", 0), patch.object(settings, "mint_operation_timeout_seconds", 0), - patch("routstr.wallet.time.monotonic", return_value=10), - patch("routstr.wallet.asyncio.sleep", AsyncMock()) as sleep, + patch("routstr.mint.time.monotonic", return_value=10), + patch("routstr.mint.asyncio.sleep", AsyncMock()) as sleep, patch( "routstr.wallet.get_wallet", AsyncMock(side_effect=lambda mint, *args, **kwargs: wallets[mint]), @@ -2378,49 +2392,28 @@ def test_classify_500_with_rate_limit_text_is_not_mint_rate_limited() -> None: @pytest.mark.asyncio async def test_probe_does_not_escalate_consecutive_rate_limits() -> None: - """When a probe fails with a rate limit, _consecutive_rate_limits should - NOT increment — the probe is a recovery check, not a new request.""" - from routstr.wallet import _MINT_RATE_LIMIT_BASE_COOLDOWN_SECONDS, _MintRateGuard + from routstr.mint import MintRateGuard - guard = _MintRateGuard("http://mint", max_concurrency=0) - - # Simulate initial rate limit: apply_rate_limit_cooldown increments counter + guard = MintRateGuard("http://mint", max_concurrency=0) guard.apply_rate_limit_cooldown() - assert guard._consecutive_rate_limits == 1 - cooldown_before = guard._cooldown_until - assert cooldown_before > 0 + guard._cooldown_until = 0.0 - # Simulate probe failure: _run_probe uses apply_cooldown, NOT - # apply_rate_limit_cooldown, so the counter stays at 1. - guard.apply_cooldown(_MINT_RATE_LIMIT_BASE_COOLDOWN_SECONDS, reason="rate_limited") - assert guard._consecutive_rate_limits == 1 # unchanged! + with pytest.raises(httpx.HTTPStatusError): + await guard.run(AsyncMock(side_effect=_http_429_error())) + + assert guard._consecutive_rate_limits == 1 assert guard._needs_probe is True @pytest.mark.asyncio async def test_probe_recovery_resets_consecutive_rate_limits() -> None: - """A successful probe resets _consecutive_rate_limits to 0.""" - from routstr.wallet import _MintRateGuard + from routstr.mint import MintRateGuard - guard = _MintRateGuard("http://mint", max_concurrency=0) - - # First rate limit: increments to 1, sets 60s cooldown. + guard = MintRateGuard("http://mint", max_concurrency=0) guard.apply_rate_limit_cooldown() - assert guard._consecutive_rate_limits == 1 - - # Manually expire the cooldown so the next call creates a fresh one. guard._cooldown_until = 0.0 - guard._cooldown_reason = None - # Second rate limit (after cooldown expired): increments to 2. - guard.apply_rate_limit_cooldown() - assert guard._consecutive_rate_limits == 2 - - # Simulate a successful probe by resetting (as _run_probe does) - guard._needs_probe = False - guard._cooldown_until = 0.0 - guard._cooldown_reason = None - guard._consecutive_rate_limits = 0 + assert await guard.run(AsyncMock(return_value="ok")) == "ok" assert guard._consecutive_rate_limits == 0 assert guard._needs_probe is False