fix: resolve mint fallback review comments

This commit is contained in:
9qeklajc
2026-07-31 02:10:00 +02:00
parent 443c910b9e
commit 2ec6b27200
20 changed files with 957 additions and 612 deletions
+3
View File
@@ -45,6 +45,9 @@ ROUTSTR_SECRET_KEY=
# ENABLE_ANALYTICS_SHARING=true # ENABLE_ANALYTICS_SHARING=true
# CASHU_MINTS="https://mint.minibits.cash/Bitcoin,https://mint.cubabitcoin.org,https://ecashmint.otrta.me" # CASHU_MINTS="https://mint.minibits.cash/Bitcoin,https://mint.cubabitcoin.org,https://ecashmint.otrta.me"
# MINT_OPERATION_CONCURRENCY=4 # MINT_OPERATION_CONCURRENCY=4
# MINT_OPERATION_TIMEOUT_SECONDS=30
# MINT_MAX_CONCURRENCY=4
# MINT_RETRY_MAX_ATTEMPTS=3
# RECEIVE_LN_ADDRESS= # RECEIVE_LN_ADDRESS=
# REFUND_SWEEP_CLAIM_TIMEOUT_SECONDS=900 # REFUND_SWEEP_CLAIM_TIMEOUT_SECONDS=900
+9
View File
@@ -136,6 +136,10 @@ Use environment variables for:
| `NSEC` | Legacy seed for the Nostr private key (otherwise set from the admin UI) | — | | `NSEC` | Legacy seed for the Nostr private key (otherwise set from the admin UI) | — |
| `ENABLE_ANALYTICS_SHARING` | Enable usage analytics sharing to Nostr | `true` | | `ENABLE_ANALYTICS_SHARING` | Enable usage analytics sharing to Nostr | `true` |
| `CASHU_MINTS` | Comma-separated mint URLs | `https://mint.minibits.cash/Bitcoin` | | `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 | — | | `RECEIVE_LN_ADDRESS` | Lightning address for withdrawals | — |
| `MIN_PAYOUT_SAT` | Min payout balance in sats (applies to all mints) | `210` | | `MIN_PAYOUT_SAT` | Min payout balance in sats (applies to all mints) | `210` |
| `PAYOUT_INTERVAL_SECONDS` | Payout loop interval (seconds) | `900` | | `PAYOUT_INTERVAL_SECONDS` | Payout loop interval (seconds) | `900` |
@@ -143,6 +147,11 @@ Use environment variables for:
| `CORS_ORIGINS` | Allowed CORS origins | `*` | | `CORS_ORIGINS` | Allowed CORS origins | `*` |
| `RELAYS` | Nostr relays (comma-separated) | (default set) | | `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 ### 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. 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.
+2 -2
View File
@@ -215,7 +215,7 @@ async def topup_wallet_endpoint(
raise HTTPException(status_code=400, detail="Invalid token format") raise HTTPException(status_code=400, detail="Invalid token format")
source_mint = token_mint_url(cashu_token, "unknown") source_mint = token_mint_url(cashu_token, "unknown")
logger.warning( logger.info(
"Cashu wallet top-up started", "Cashu wallet top-up started",
extra={ extra={
"event": "cashu_topup_started", "event": "cashu_topup_started",
@@ -259,7 +259,7 @@ async def topup_wallet_endpoint(
) )
raise HTTPException(status_code=status_code, detail=message) raise HTTPException(status_code=status_code, detail=message)
logger.warning( logger.info(
"Cashu wallet top-up completed", "Cashu wallet top-up completed",
extra={ extra={
"event": "cashu_topup_completed", "event": "cashu_topup_completed",
+39 -13
View File
@@ -3,8 +3,9 @@ import hashlib
import re import re
import secrets import secrets
import time import time
from contextlib import asynccontextmanager
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any from typing import Any, AsyncGenerator
from fastapi import APIRouter, Depends, Header, HTTPException from fastapi import APIRouter, Depends, Header, HTTPException
from pydantic import BaseModel, Field 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.db import ApiKey, LightningInvoice, create_session, get_session
from .core.logging import get_logger from .core.logging import get_logger
from .core.settings import settings from .core.settings import settings
from .mint import (
is_mint_rate_limited,
mint_cooldown_remaining,
run_mint_operation,
)
from .wallet import ( from .wallet import (
MintConnectionError, MintConnectionError,
_is_mint_rate_limited,
_mint_cooldown_remaining,
_mint_operation,
get_wallet, get_wallet,
is_mint_connection_error, is_mint_connection_error,
wallet_operation_guard, wallet_operation_guard,
@@ -31,7 +34,31 @@ lightning_router = APIRouter(prefix="/lightning")
# Avoid duplicate work within one process. Cross-process credit fencing is done # Avoid duplicate work within one process. Cross-process credit fencing is done
# by the conditional pending -> paid update in _finalize_invoice_settlement(). # 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) @dataclass(frozen=True)
@@ -142,11 +169,11 @@ async def _request_mint_with_fallback(
configured = allowed_mints or [settings.primary_mint, *settings.cashu_mints] configured = allowed_mints or [settings.primary_mint, *settings.cashu_mints]
candidates = list(dict.fromkeys(configured)) candidates = list(dict.fromkeys(configured))
for mint_url in candidates: for mint_url in candidates:
cooldown = _mint_cooldown_remaining(mint_url) cooldown = mint_cooldown_remaining(mint_url)
if cooldown > 0: if cooldown > 0:
tried.append(f"{mint_url}: cooling down") tried.append(f"{mint_url}: cooling down")
logger.info( logger.info(
"Skipping rate-limited mint", "Skipping mint during cooldown",
extra={ extra={
"mint_url": mint_url, "mint_url": mint_url,
"cooldown_seconds": round(cooldown, 2), "cooldown_seconds": round(cooldown, 2),
@@ -156,7 +183,7 @@ async def _request_mint_with_fallback(
continue continue
try: try:
wallet = await get_wallet(mint_url, "sat", retry_on_rate_limit=False) 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), lambda: wallet.request_mint(amount_sats),
op_name="request_mint_invoice", op_name="request_mint_invoice",
mint_url=mint_url, mint_url=mint_url,
@@ -165,7 +192,7 @@ async def _request_mint_with_fallback(
return quote.request, quote.quote, mint_url return quote.request, quote.quote, mint_url
except Exception as e: except Exception as e:
tried.append(f"{mint_url}: {type(e).__name__}") 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 raise
logger.warning( logger.warning(
"request_mint failed, trying fallback mint", "request_mint failed, trying fallback mint",
@@ -352,8 +379,7 @@ async def recover_invoice(
async def check_invoice_payment( async def check_invoice_payment(
invoice: LightningInvoice, session: AsyncSession invoice: LightningInvoice, session: AsyncSession
) -> None: ) -> None:
lock = _invoice_settlement_locks.setdefault(invoice.id, asyncio.Lock()) async with _invoice_settlement_lock(invoice.id), wallet_operation_guard():
async with lock, wallet_operation_guard():
minted = False minted = False
try: try:
# Snapshot the row and end the caller's read transaction before any # 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 mint_url = settlement.mint_url or settings.primary_mint
wallet = await get_wallet(mint_url, "sat") 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), lambda: wallet.get_mint_quote(settlement.payment_hash),
op_name="get_mint_quote", op_name="get_mint_quote",
mint_url=mint_url, mint_url=mint_url,
@@ -483,7 +509,7 @@ async def _mint_invoice_quote(
return return
try: try:
await _mint_operation( await run_mint_operation(
lambda: wallet.mint(invoice.amount_sats, quote_id=invoice.payment_hash), lambda: wallet.mint(invoice.amount_sats, quote_id=invoice.payment_hash),
op_name=f"invoice_mint_{invoice.purpose}", op_name=f"invoice_mint_{invoice.purpose}",
mint_url=mint_url, mint_url=mint_url,
+338
View File
@@ -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")
-12
View File
@@ -18,18 +18,6 @@ from ..wallet import deserialize_token_from_string
logger = get_logger(__name__) 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: def check_token_balance(headers: dict, body: dict, max_cost_for_model: int) -> None:
if x_cashu := headers.get("x-cashu", None): if x_cashu := headers.get("x-cashu", None):
cashu_token = x_cashu cashu_token = x_cashu
+41 -25
View File
@@ -1,17 +1,13 @@
from __future__ import annotations from __future__ import annotations
import asyncio
import math import math
from typing import TypedDict from typing import TypedDict
import httpx import httpx
from cashu.core.base import MeltQuoteState
from cashu.wallet.wallet import Proof, Wallet from cashu.wallet.wallet import Proof, Wallet
# The Cashu library issues POST /v1/melt/bolt11 with timeout=None, so a hung or from ..mint import MINT_TRANSPORT_EXCEPTIONS, run_mint_operation
# 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
try: try:
from bech32 import bech32_decode, convertbits # type: ignore from bech32 import bech32_decode, convertbits # type: ignore
@@ -222,9 +218,7 @@ async def raw_send_to_lnurl(
lnurl_data["callback_url"], final_amount lnurl_data["callback_url"], final_amount
) )
from ..wallet import _mint_operation melt_quote_resp = await run_mint_operation(
melt_quote_resp = await _mint_operation(
lambda: wallet.melt_quote(invoice=bolt11_invoice), lambda: wallet.melt_quote(invoice=bolt11_invoice),
op_name="lnurl_melt_quote", op_name="lnurl_melt_quote",
mint_url=str(wallet.url), 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) proofs, _ = await wallet.select_to_send(proofs, amount, set_reserved=True)
try: try:
_ = await asyncio.wait_for( melt_response = await run_mint_operation(
_mint_operation( lambda: wallet.melt(
lambda: wallet.melt( proofs=proofs,
proofs=proofs, invoice=bolt11_invoice,
invoice=bolt11_invoice, fee_reserve_sat=melt_quote_resp.fee_reserve,
fee_reserve_sat=melt_quote_resp.fee_reserve, quote_id=melt_quote_resp.quote,
quote_id=melt_quote_resp.quote,
),
op_name="lnurl_melt",
mint_url=str(wallet.url),
retry_timeouts=False,
), ),
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( raise LNURLError(
f"Melt timed out after {MELT_TIMEOUT_SECONDS}s (mint unresponsive)" "Melt outcome is ambiguous; quote reconciliation failed and proofs "
) from e "must not be retried"
return final_amount ) 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
-5
View File
@@ -27,7 +27,6 @@ from .core.db import (
from .core.exceptions import UpstreamError from .core.exceptions import UpstreamError
from .core.not_found import build_not_found_response from .core.not_found import build_not_found_response
from .payment.helpers import ( from .payment.helpers import (
apply_mint_fee_allowance,
calculate_discounted_max_cost, calculate_discounted_max_cost,
check_token_balance, check_token_balance,
create_error_response, create_error_response,
@@ -354,7 +353,6 @@ async def _proxy(
max_cost_for_model = await calculate_discounted_max_cost( max_cost_for_model = await calculate_discounted_max_cost(
_max_cost_for_model, request_body_dict, model_obj=model_obj _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) 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 = await calculate_discounted_max_cost(
candidate_max, request_body_dict, model_obj=model_obj 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: if candidate_max > max_cost_for_model:
await revert_pay_for_request( await revert_pay_for_request(
key, session, max_cost_for_model, reservation_snapshot key, session, max_cost_for_model, reservation_snapshot
+190 -423
View File
@@ -2,16 +2,15 @@ import asyncio
import fcntl import fcntl
import os import os
import re import re
import socket
import time import time
import typing import typing
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from contextvars import ContextVar from contextvars import ContextVar
from pathlib import Path from pathlib import Path
from typing import Any, AsyncGenerator, Awaitable, Callable, TypedDict from typing import AsyncGenerator, TypedDict
import httpx 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.core.mint_info import MintInfo as _CashuMintInfo
from cashu.wallet.helpers import deserialize_token_from_string from cashu.wallet.helpers import deserialize_token_from_string
from cashu.wallet.wallet import Wallet as _CashuWallet 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 import db, get_logger
from .core.db import store_cashu_transaction_with_retry as store_cashu_transaction from .core.db import store_cashu_transaction_with_retry as store_cashu_transaction
from .core.settings import settings 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 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. # cashu still declares Optional[X] without explicit defaults on MintInfo.
# Under pydantic v2 those are required, but real mints omit many of them. # 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. # 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: except BlockingIOError:
await _scheduler_sleep(0.05) await _scheduler_sleep(0.05)
depth_token = _wallet_operation_depth.set(1) depth_token = _wallet_operation_depth.set(1)
yield async with fail_fast_mint_operations():
yield
finally: finally:
if depth_token is not None: if depth_token is not None:
_wallet_operation_depth.reset(depth_token) _wallet_operation_depth.reset(depth_token)
@@ -94,6 +113,20 @@ def _mints_to_inspect() -> list[str]:
return mint_urls 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): class MintConnectionError(Exception):
"""The mint could not be reached (network transport failure). """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: def is_source_mint_connection_error(error: BaseException) -> bool:
seen: set[int] = set() seen: set[int] = set()
current: BaseException | None = error current: BaseException | None = error
@@ -475,7 +175,7 @@ def is_mint_connection_error(error: BaseException) -> bool:
return False return False
if isinstance(current, MintConnectionError): if isinstance(current, MintConnectionError):
return True return True
if isinstance(current, _TRANSPORT_EXC_TYPES): if isinstance(current, MINT_TRANSPORT_EXCEPTIONS):
return True return True
current = current.__cause__ or current.__context__ current = current.__cause__ or current.__context__
return False 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", "The mint that issued this Cashu token is unreachable; the token cannot be redeemed at another mint",
"cashu_source_mint_unreachable", "cashu_source_mint_unreachable",
) )
if _is_mint_rate_limited(error): if is_mint_rate_limited(error):
return ( return (
"mint_rate_limited", "mint_rate_limited",
503, 503,
@@ -605,14 +305,14 @@ async def _redeem_same_mint(
drifts insolvent. drifts insolvent.
""" """
try: try:
await _mint_operation( await run_mint_operation(
lambda: wallet.load_mint(keyset_id=token_obj.keysets[0]), lambda: wallet.load_mint(keyset_id=token_obj.keysets[0]),
op_name="redeem_load_mint", op_name="redeem_load_mint",
mint_url=token_obj.mint, mint_url=token_obj.mint,
) )
wallet.verify_proofs_dleq(token_obj.proofs) wallet.verify_proofs_dleq(token_obj.proofs)
input_fees = wallet.get_fees_for_proofs(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), lambda: wallet.split(proofs=token_obj.proofs, amount=0, include_fees=True),
op_name="redeem_split", op_name="redeem_split",
mint_url=token_obj.mint, mint_url=token_obj.mint,
@@ -655,7 +355,7 @@ async def recieve_token(
destinations = list( destinations = list(
dict.fromkeys([settings.primary_mint, *settings.cashu_mints]) dict.fromkeys([settings.primary_mint, *settings.cashu_mints])
) )
logger.warning( logger.info(
"Cashu cross-mint swap required", "Cashu cross-mint swap required",
extra={ extra={
"event": "cashu_swap_started", "event": "cashu_swap_started",
@@ -667,7 +367,7 @@ async def recieve_token(
) )
return await swap_to_trusted_mint(token_obj, wallet) return await swap_to_trusted_mint(token_obj, wallet)
logger.warning( logger.info(
"Trying same-mint Cashu redemption", "Trying same-mint Cashu redemption",
extra={ extra={
"event": "cashu_same_mint_redemption", "event": "cashu_same_mint_redemption",
@@ -769,12 +469,12 @@ async def find_trusted_mint_with_funds(
balances: dict[str, int] = {} balances: dict[str, int] = {}
for mint_url in candidates: for mint_url in candidates:
if _mint_cooldown_remaining(mint_url) > 0: if mint_cooldown_remaining(mint_url) > 0:
continue continue
try: try:
wallet = await get_wallet(mint_url, unit, retry_on_rate_limit=False) wallet = await get_wallet(mint_url, unit, retry_on_rate_limit=False)
except Exception as error: 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 balances[mint_url] = 0
continue continue
raise raise
@@ -819,6 +519,17 @@ def _net_minted_amount(amount_msat: int, token_unit: str, fees: int) -> int:
return int(remaining_msat) 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: def _melt_insufficient_shortfall(error: Exception) -> int | None:
""" """
Classify a melt failure: return the observed shortfall (in the token unit) 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." f"Token value is too small after fee deduction or unit conversion."
) )
candidates = list(dict.fromkeys([settings.primary_mint, *settings.cashu_mints])) candidates = list(dict.fromkeys([settings.primary_mint, *settings.cashu_mints]))
logger.warning( logger.info(
"Trying trusted destination mints", "Trying trusted destination mints",
extra={ extra={
"event": "cashu_destination_candidates", "event": "cashu_destination_candidates",
@@ -883,7 +594,7 @@ async def _request_mint_with_fallback(
) )
tried: list[str] = [] tried: list[str] = []
for candidate_index, mint_url in enumerate(candidates, start=1): 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: if cooldown > 0:
tried.append(f"{mint_url}: cooling down") tried.append(f"{mint_url}: cooling down")
logger.warning( logger.warning(
@@ -898,7 +609,7 @@ async def _request_mint_with_fallback(
}, },
) )
continue continue
logger.warning( logger.info(
"Trying destination mint", "Trying destination mint",
extra={ extra={
"event": "cashu_destination_attempt", "event": "cashu_destination_attempt",
@@ -917,13 +628,13 @@ async def _request_mint_with_fallback(
settings.primary_mint_unit, settings.primary_mint_unit,
retry_on_rate_limit=False, retry_on_rate_limit=False,
) )
quote = await _mint_operation( quote = await run_mint_operation(
lambda: wallet.request_mint(amount), lambda: wallet.request_mint(amount),
op_name=op_name, op_name=op_name,
mint_url=mint_url, mint_url=mint_url,
retry_on_rate_limit=False, retry_on_rate_limit=False,
) )
logger.warning( logger.info(
"Destination mint selected", "Destination mint selected",
extra={ extra={
"event": "cashu_destination_selected", "event": "cashu_destination_selected",
@@ -937,12 +648,12 @@ async def _request_mint_with_fallback(
except Exception as error: except Exception as error:
tried.append(f"{mint_url}: {type(error).__name__}") tried.append(f"{mint_url}: {type(error).__name__}")
connection_failure = is_mint_connection_error(error) 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: if not connection_failure and not rate_limited:
raise raise
if connection_failure: if connection_failure:
_MintRateGuard.get(mint_url).apply_cooldown( MintRateGuard.get(mint_url).apply_cooldown(
_MINT_TRANSPORT_COOLDOWN_SECONDS, reason="unreachable" MINT_TRANSPORT_COOLDOWN_SECONDS, reason="unreachable"
) )
logger.warning( logger.warning(
"Destination mint failed", "Destination mint failed",
@@ -993,7 +704,7 @@ async def _calculate_swap_amount(
if token_mint_url == settings.primary_mint: if token_mint_url == settings.primary_mint:
logger.info( 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}, extra={"minted_amount": receive_amount},
) )
return int(receive_amount) return int(receive_amount)
@@ -1005,7 +716,7 @@ async def _calculate_swap_amount(
# logs. Guard early with full diagnostic context instead. # logs. Guard early with full diagnostic context instead.
if receive_amount <= 0: if receive_amount <= 0:
logger.error( 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={ extra={
"amount_msat": amount_msat, "amount_msat": amount_msat,
"token_unit": token_unit, "token_unit": token_unit,
@@ -1022,7 +733,7 @@ async def _calculate_swap_amount(
) )
logger.info( logger.info(
"swap_to_primary_mint: estimating fees", "swap_to_trusted_mint: estimating fees",
extra={ extra={
"dummy_amount": receive_amount, "dummy_amount": receive_amount,
"unit": settings.primary_mint_unit, "unit": settings.primary_mint_unit,
@@ -1040,7 +751,7 @@ async def _calculate_swap_amount(
primary_wallet=primary_wallet, primary_wallet=primary_wallet,
) )
stage = "source_fee_quote" 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), lambda: token_wallet.melt_quote(dummy_mint_quote.request),
op_name="swap_fee_est_melt_quote", op_name="swap_fee_est_melt_quote",
mint_url=token_mint_url, 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") raise ValueError(f"Fees ({total_fees} {token_unit}) exceed token amount")
logger.info( logger.info(
"swap_to_primary_mint: fee estimation result", "swap_to_trusted_mint: fee estimation result",
extra={ extra={
"token_amount_sat": _msats_to_sats(amount_msat), "token_amount_sat": _msats_to_sats(amount_msat),
"estimated_fee": total_fees, "estimated_fee": total_fees,
@@ -1105,10 +816,62 @@ async def _calculate_swap_amount(
raise ValueError(f"Failed to estimate fees: {e}") from e 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( async def swap_to_trusted_mint(
token_obj: Token, token_wallet: Wallet token_obj: Token, token_wallet: Wallet
) -> tuple[int, str, str]: ) -> tuple[int, str, str]:
logger.warning( logger.info(
"Starting Cashu cross-mint swap", "Starting Cashu cross-mint swap",
extra={ extra={
"event": "cashu_swap_started", "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. # NUT-02 input fee still applies; _redeem_same_mint accounts for it.
if token_obj.mint == settings.primary_mint: if token_obj.mint == settings.primary_mint:
logger.info( 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={ extra={
"mint": token_obj.mint, "mint": token_obj.mint,
"amount": token_amount, "amount": token_amount,
@@ -1166,7 +929,7 @@ async def swap_to_trusted_mint(
attempt += 1 attempt += 1
if minted_amount <= 0: if minted_amount <= 0:
logger.error( 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={ extra={
"minted_amount": minted_amount, "minted_amount": minted_amount,
"attempt": attempt, "attempt": attempt,
@@ -1188,7 +951,7 @@ async def swap_to_trusted_mint(
primary_wallet=primary_wallet, primary_wallet=primary_wallet,
) )
logger.info( logger.info(
"swap_to_primary_mint: mint quote received", "swap_to_trusted_mint: mint quote received",
extra={ extra={
"mint_quote_id": mint_quote.quote, "mint_quote_id": mint_quote.quote,
"attempt": attempt, "attempt": attempt,
@@ -1196,7 +959,7 @@ async def swap_to_trusted_mint(
}, },
) )
logger.warning( logger.info(
"Requesting melt quote from source mint", "Requesting melt quote from source mint",
extra={ extra={
"event": "cashu_source_melt_quote_attempt", "event": "cashu_source_melt_quote_attempt",
@@ -1206,7 +969,7 @@ async def swap_to_trusted_mint(
}, },
) )
try: try:
melt_quote = await _mint_operation( melt_quote = await run_mint_operation(
lambda: token_wallet.melt_quote(mint_quote.request), lambda: token_wallet.melt_quote(mint_quote.request),
op_name="swap_melt_quote", op_name="swap_melt_quote",
mint_url=token_obj.mint, 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) input_fees = token_wallet.get_fees_for_proofs(token_obj.proofs)
total_needed = melt_quote.amount + melt_quote.fee_reserve + input_fees total_needed = melt_quote.amount + melt_quote.fee_reserve + input_fees
logger.info( logger.info(
"swap_to_primary_mint: melt quote received", "swap_to_trusted_mint: melt quote received",
extra={ extra={
"melt_quote_id": melt_quote.quote, "melt_quote_id": melt_quote.quote,
"melt_amount": melt_quote.amount, "melt_amount": melt_quote.amount,
@@ -1252,7 +1015,7 @@ async def swap_to_trusted_mint(
) )
if attempt >= _MAX_SWAP_ATTEMPTS or recomputed <= 0: if attempt >= _MAX_SWAP_ATTEMPTS or recomputed <= 0:
logger.warning( logger.warning(
"swap_to_primary_mint: insufficient token amount for melt fees", "swap_to_trusted_mint: insufficient token amount for melt fees",
extra={ extra={
"token_amount": token_amount, "token_amount": token_amount,
"melt_amount": melt_quote.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})" f"(amount: {melt_quote.amount} + fee: {melt_quote.fee_reserve} + input_fees: {input_fees})"
) )
logger.warning( logger.warning(
"swap_to_primary_mint: melt quote exceeds token amount, retrying", "swap_to_trusted_mint: melt quote exceeds token amount, retrying",
extra={ extra={
"total_needed": total_needed, "total_needed": total_needed,
"token_amount": token_amount, "token_amount": token_amount,
@@ -1281,7 +1044,7 @@ async def swap_to_trusted_mint(
continue continue
try: try:
_ = await _mint_operation( melt_response = await run_mint_operation(
lambda: token_wallet.melt( lambda: token_wallet.melt(
proofs=token_obj.proofs, proofs=token_obj.proofs,
invoice=mint_quote.request, invoice=mint_quote.request,
@@ -1292,36 +1055,45 @@ async def swap_to_trusted_mint(
mint_url=token_obj.mint, mint_url=token_obj.mint,
retry_timeouts=False, retry_timeouts=False,
) )
await _confirm_melt_paid(
token_wallet, melt_quote.quote, token_obj.proofs, melt_response
)
except Exception as e: 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) shortfall = _melt_insufficient_shortfall(e)
recomputed = 0 if shortfall is None:
if shortfall is not None: if isinstance(e, TokenConsumedError):
observed_extra_fee += shortfall raise
recomputed = _net_minted_amount( if _melt_definitively_failed(e):
amount_msat, raise ValueError(
token_obj.unit, f"Failed to melt token from foreign mint {token_obj.mint}: {e}"
melt_quote.fee_reserve + input_fees + observed_extra_fee, ) from e
) if is_mint_connection_error(e):
if shortfall is None or attempt >= _MAX_SWAP_ATTEMPTS or recomputed <= 0: 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( logger.error(
"swap_to_primary_mint: melt failed", "swap_to_trusted_mint: melt failed",
extra={ extra={
"error": str(e), "error": str(e),
"error_type": type(e).__name__, "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}" f"Failed to melt token from foreign mint {token_obj.mint}: {e}"
) from e ) from e
logger.warning( 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={ extra={
"shortfall": shortfall, "shortfall": shortfall,
"retry_minted_amount": recomputed, "retry_minted_amount": recomputed,
@@ -1348,7 +1120,7 @@ async def swap_to_trusted_mint(
break break
logger.warning( logger.info(
"Source melt succeeded; minting on destination", "Source melt succeeded; minting on destination",
extra={ extra={
"event": "cashu_destination_mint_attempt", "event": "cashu_destination_mint_attempt",
@@ -1361,9 +1133,9 @@ async def swap_to_trusted_mint(
await dest_wallet.load_proofs(reload=True) await dest_wallet.load_proofs(reload=True)
pre_mint_balance = dest_wallet.available_balance.amount pre_mint_balance = dest_wallet.available_balance.amount
try: try:
_ = await _mint_operation( _ = await run_mint_operation(
lambda: dest_wallet.mint(minted_amount, quote_id=mint_quote.quote), 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, mint_url=dest_mint_url,
retry_timeouts=False, retry_timeouts=False,
) )
@@ -1373,7 +1145,7 @@ async def swap_to_trusted_mint(
# bump_secret_derivation ran locally. Recover orphaned proofs and # bump_secret_derivation ran locally. Recover orphaned proofs and
# advance the counter so the next request derives fresh secrets. # advance the counter so the next request derives fresh secrets.
logger.warning( logger.warning(
"swap_to_primary_mint: outputs already signed — recovering orphaned proofs", "swap_to_trusted_mint: outputs already signed — recovering orphaned proofs",
extra={ extra={
"mint_quote_id": mint_quote.quote, "mint_quote_id": mint_quote.quote,
"minted_amount": minted_amount, "minted_amount": minted_amount,
@@ -1388,7 +1160,7 @@ async def swap_to_trusted_mint(
post_recovery_balance = dest_wallet.available_balance.amount post_recovery_balance = dest_wallet.available_balance.amount
balance_gained = post_recovery_balance - pre_mint_balance balance_gained = post_recovery_balance - pre_mint_balance
logger.info( logger.info(
"swap_to_primary_mint: recovery scan completed", "swap_to_trusted_mint: recovery scan completed",
extra={ extra={
"pre_mint_balance": pre_mint_balance, "pre_mint_balance": pre_mint_balance,
"post_recovery_balance": post_recovery_balance, "post_recovery_balance": post_recovery_balance,
@@ -1411,7 +1183,7 @@ async def swap_to_trusted_mint(
raise raise
except Exception as recovery_err: except Exception as recovery_err:
logger.error( logger.error(
"swap_to_primary_mint: recovery failed", "swap_to_trusted_mint: recovery failed",
extra={"error": str(recovery_err)}, extra={"error": str(recovery_err)},
) )
raise TokenConsumedError( raise TokenConsumedError(
@@ -1419,7 +1191,7 @@ async def swap_to_trusted_mint(
) from e ) from e
else: else:
logger.error( 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={ extra={
"error": str(e), "error": str(e),
"error_type": type(e).__name__, "error_type": type(e).__name__,
@@ -1432,7 +1204,7 @@ async def swap_to_trusted_mint(
"Mint on primary failed after successful melt" "Mint on primary failed after successful melt"
) from e ) from e
logger.warning( logger.info(
"Cashu cross-mint swap completed", "Cashu cross-mint swap completed",
extra={ extra={
"event": "cashu_swap_completed", "event": "cashu_swap_completed",
@@ -1595,13 +1367,13 @@ async def get_wallet(
now = time.monotonic() now = time.monotonic()
last = _wallet_last_load.get(id) last = _wallet_last_load.get(id)
if last is None or now - last >= _WALLOAD_RELOAD_MIN_INTERVAL_SECONDS: if last is None or now - last >= _WALLOAD_RELOAD_MIN_INTERVAL_SECONDS:
await _mint_operation( await run_mint_operation(
lambda: _wallets[id].load_mint(), lambda: _wallets[id].load_mint(),
op_name="load_mint", op_name="load_mint",
mint_url=mint_url, mint_url=mint_url,
retry_on_rate_limit=retry_on_rate_limit, retry_on_rate_limit=retry_on_rate_limit,
) )
await _mint_operation( await run_mint_operation(
lambda: _wallets[id].load_proofs(reload=True), lambda: _wallets[id].load_proofs(reload=True),
op_name="load_proofs", op_name="load_proofs",
mint_url=mint_url, mint_url=mint_url,
@@ -1640,7 +1412,7 @@ async def slow_filter_spend_proofs(
batch_size = 1000 batch_size = 1000
for i in range(0, len(proofs), batch_size): for i in range(0, len(proofs), batch_size):
pb = proofs[i : i + 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), lambda: wallet.check_proof_state(pb),
op_name="check_proof_state", op_name="check_proof_state",
mint_url=str(wallet.url), mint_url=str(wallet.url),
@@ -1652,12 +1424,7 @@ async def slow_filter_spend_proofs(
else: else:
_spent_proofs.append(proof) _spent_proofs.append(proof)
if _spent_proofs: if _spent_proofs:
await _mint_operation( await wallet.set_reserved_for_send(_spent_proofs, reserved=True)
lambda: wallet.set_reserved_for_send(_spent_proofs, reserved=True),
op_name="set_reserved_spent_proofs",
mint_url=str(wallet.url),
retry_timeouts=False,
)
return _proofs return _proofs
@@ -1686,7 +1453,7 @@ async def _get_supported_mint_units(mint_url: str) -> list[str]:
return cached[1] return cached[1]
wallet = await get_wallet(mint_url, settings.primary_mint_unit, load=False) 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(), lambda: wallet._get_keysets(),
op_name="get_mint_keysets", op_name="get_mint_keysets",
mint_url=mint_url, mint_url=mint_url,
@@ -1750,7 +1517,7 @@ async def fetch_all_balances(
mint_units[mint_url] = await _get_supported_mint_units(mint_url) mint_units[mint_url] = await _get_supported_mint_units(mint_url)
except Exception as error: except Exception as error:
connection_failure = is_mint_connection_error(error) connection_failure = is_mint_connection_error(error)
rate_limited = _is_mint_rate_limited(error) rate_limited = is_mint_rate_limited(error)
error_code = ( error_code = (
"rate_limited" "rate_limited"
if rate_limited if rate_limited
@@ -1759,12 +1526,12 @@ async def fetch_all_balances(
else "mint_error" else "mint_error"
) )
if connection_failure: if connection_failure:
_MintRateGuard.get(mint_url).apply_cooldown( MintRateGuard.get(mint_url).apply_cooldown(
_BALANCE_FETCH_RETRY_SECONDS, reason="unreachable" _BALANCE_FETCH_RETRY_SECONDS, reason="unreachable"
) )
retry_delay = max( retry_delay = max(
_BALANCE_FETCH_RETRY_SECONDS, _BALANCE_FETCH_RETRY_SECONDS,
_mint_cooldown_remaining(mint_url), mint_cooldown_remaining(mint_url),
) )
discovery_errors.append( discovery_errors.append(
_balance_error( _balance_error(
@@ -1819,9 +1586,9 @@ async def fetch_all_balances(
retry_after_seconds=failure[0] - now, retry_after_seconds=failure[0] - now,
) )
cooldown = _mint_cooldown_remaining(mint_url) cooldown = mint_cooldown_remaining(mint_url)
if cooldown > 0: if cooldown > 0:
error_code = _mint_cooldown_reason(mint_url) or "cooldown" error_code = mint_cooldown_reason(mint_url) or "cooldown"
error = { error = {
"rate_limited": "Mint is rate limited", "rate_limited": "Mint is rate limited",
"unreachable": "Mint is unreachable", "unreachable": "Mint is unreachable",
@@ -1846,7 +1613,7 @@ async def fetch_all_balances(
proofs = await slow_filter_spend_proofs(proofs, wallet) proofs = await slow_filter_spend_proofs(proofs, wallet)
except Exception as error: except Exception as error:
connection_failure = is_mint_connection_error(error) connection_failure = is_mint_connection_error(error)
rate_limited = _is_mint_rate_limited(error) rate_limited = is_mint_rate_limited(error)
error_code = ( error_code = (
"rate_limited" "rate_limited"
if rate_limited if rate_limited
@@ -1855,16 +1622,16 @@ async def fetch_all_balances(
else "mint_error" else "mint_error"
) )
if rate_limited: if rate_limited:
_MintRateGuard.get(mint_url).apply_rate_limit_cooldown( MintRateGuard.get(mint_url).apply_rate_limit_cooldown(
_BALANCE_FETCH_RETRY_SECONDS _BALANCE_FETCH_RETRY_SECONDS
) )
elif connection_failure: elif connection_failure:
_MintRateGuard.get(mint_url).apply_cooldown( MintRateGuard.get(mint_url).apply_cooldown(
_BALANCE_FETCH_RETRY_SECONDS, reason=error_code _BALANCE_FETCH_RETRY_SECONDS, reason=error_code
) )
retry_delay = max( retry_delay = max(
_BALANCE_FETCH_RETRY_SECONDS, _BALANCE_FETCH_RETRY_SECONDS,
_mint_cooldown_remaining(mint_url), mint_cooldown_remaining(mint_url),
) )
_balance_fetch_failures[key] = ( _balance_fetch_failures[key] = (
time.monotonic() + retry_delay, time.monotonic() + retry_delay,
+18 -13
View File
@@ -208,25 +208,30 @@ async def test_pay_for_request_succeeds_when_balance_equals_cost(
@pytest.mark.asyncio @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, integration_session: AsyncSession,
) -> None: ) -> None:
from routstr.auth import pay_for_request, validate_bearer_key from routstr.auth import pay_for_request, validate_bearer_key
from routstr.payment.helpers import apply_mint_fee_allowance
key = _key(balance=95_000) short_key = _key(balance=95_000)
integration_session.add(key) exact_key = _key(balance=100_000)
integration_session.add(short_key)
integration_session.add(exact_key)
await integration_session.commit() await integration_session.commit()
admission_cost = apply_mint_fee_allowance(100_000) with pytest.raises(HTTPException) as insufficient:
validated = await validate_bearer_key( await validate_bearer_key(
f"sk-{key.hashed_key}", integration_session, min_cost=admission_cost f"sk-{short_key.hashed_key}", integration_session, min_cost=100_000
) )
await pay_for_request(validated, admission_cost, integration_session) assert insufficient.value.status_code == 402
await integration_session.refresh(key) validated = await validate_bearer_key(
assert admission_cost == 95_000 f"sk-{exact_key.hashed_key}", integration_session, min_cost=100_000
assert key.reserved_balance == 95_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"] error = body["detail"]["error"]
assert error["code"] == "insufficient_balance" assert error["code"] == "insufficient_balance"
assert error["type"] == "insufficient_quota" 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"] assert "20.32 sats (20320 msats) available" in error["message"]
# Balance must be completely untouched # Balance must be completely untouched
+5 -2
View File
@@ -20,6 +20,7 @@ from collections.abc import Callable
from unittest.mock import AsyncMock, Mock, patch from unittest.mock import AsyncMock, Mock, patch
import pytest import pytest
from cashu.core.base import MeltQuoteState
from httpx import AsyncClient, Response from httpx import AsyncClient, Response
from routstr.core.settings import settings 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() 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 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. " "Mint Error: not enough inputs provided for melt. "
"Provided: 179, needed: 180 (Code: 11000)" "Provided: 179, needed: 180 (Code: 11000)"
), ),
Mock(), Mock(state=MeltQuoteState.paid),
] ]
response = await _post_topup( response = await _post_topup(
+3 -3
View File
@@ -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.object(settings, "primary_mint", "http://mint:3338"),
patch("routstr.wallet.get_wallet", get_wallet), patch("routstr.wallet.get_wallet", get_wallet),
patch("routstr.wallet.db.create_session", _fake_session), 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, patch("routstr.wallet.logger.warning") as warning,
): ):
first = await fetch_all_balances(units=["sat"]) 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.object(settings, "primary_mint", "http://mint:3338"),
patch("routstr.wallet.get_wallet", get_wallet), patch("routstr.wallet.get_wallet", get_wallet),
patch("routstr.wallet.db.create_session", _fake_session), 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"), patch("routstr.wallet.logger.warning"),
): ):
await fetch_all_balances(units=["sat"]) 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.object(settings, "primary_mint", mint),
patch("routstr.wallet.get_wallet", get_wallet), patch("routstr.wallet.get_wallet", get_wallet),
patch("routstr.wallet.db.create_session", _fake_session), 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, patch("routstr.wallet.logger.warning") as warning,
): ):
details, *_ = await fetch_all_balances(units=["sat", "msat"]) details, *_ = await fetch_all_balances(units=["sat", "msat"])
+2
View File
@@ -155,6 +155,7 @@ async def test_non_pending_invoice_is_not_minted() -> None:
get_wallet.assert_not_awaited() get_wallet.assert_not_awaited()
session.commit.assert_awaited_once() session.commit.assert_awaited_once()
assert _invoice_settlement_locks == {}
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -212,3 +213,4 @@ async def test_concurrent_invoice_checks_finalize_once_in_process() -> None:
assert invoice.status == "paid" assert invoice.status == "paid"
finalize.assert_awaited_once() finalize.assert_awaited_once()
assert _invoice_settlement_locks == {}
+101 -43
View File
@@ -1,70 +1,127 @@
"""raw_send_to_lnurl() must not hang forever on an unresponsive mint. """LNURL melt attempts must not misclassify ambiguous payment outcomes."""
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.
"""
import asyncio import asyncio
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch from unittest.mock import AsyncMock, MagicMock, patch
import pytest 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 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)] proofs = [MagicMock(amount=1000)]
wallet = MagicMock(url="https://mint.test")
wallet = MagicMock()
wallet.melt_quote = AsyncMock(return_value=MagicMock(fee_reserve=1, quote="q")) wallet.melt_quote = AsyncMock(return_value=MagicMock(fee_reserve=1, quote="q"))
wallet.select_to_send = AsyncMock(return_value=(proofs, None)) 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: 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.melt = AsyncMock(side_effect=_hang)
wallet.get_melt_quote = AsyncMock(
return_value=MagicMock(state=MeltQuoteState.unpaid)
)
data_patch, invoice_patch = _lnurl_patches()
lnurl_data = { with (
"callback_url": "https://ln.tld/cb", patch.object(settings, "mint_operation_timeout_seconds", 0.05),
"min_sendable": 1_000, patch.object(settings, "mint_retry_max_attempts", 0),
"max_sendable": 100_000_000, data_patch,
} invoice_patch,
pytest.raises(LNURLError, match="outcome is ambiguous"),
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 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 @pytest.mark.asyncio
async def test_raw_send_to_lnurl_succeeds_within_timeout() -> None: async def test_raw_send_to_lnurl_timeout_reconciled_paid_is_success() -> None:
"""A prompt melt still returns the net amount, unaffected by the guard.""" wallet, proofs = _wallet()
proofs = [MagicMock(amount=1000)]
wallet = MagicMock() async def _hang(**kwargs: object) -> None:
wallet.melt_quote = AsyncMock(return_value=MagicMock(fee_reserve=1, quote="q")) await asyncio.sleep(5)
wallet.select_to_send = AsyncMock(return_value=(proofs, None))
wallet.melt = AsyncMock(return_value=MagicMock())
lnurl_data = { wallet.melt = AsyncMock(side_effect=_hang)
"callback_url": "https://ln.tld/cb", wallet.get_melt_quote = AsyncMock(
"min_sendable": 1_000, return_value=MagicMock(state=MeltQuoteState.paid)
"max_sendable": 100_000_000, )
} data_patch, invoice_patch = _lnurl_patches()
with patch.object(lnurl, "MELT_TIMEOUT_SECONDS", 5), patch( with (
"routstr.payment.lnurl.get_lnurl_data", AsyncMock(return_value=lnurl_data) patch.object(settings, "mint_operation_timeout_seconds", 0.05),
), patch( patch.object(settings, "mint_retry_max_attempts", 0),
"routstr.payment.lnurl.get_lnurl_invoice", data_patch,
AsyncMock(return_value=("lnbc1...", {})), 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( paid = await raw_send_to_lnurl(
wallet, proofs, "owner@ln.tld", "sat", amount=1000 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 assert paid > 0
wallet.melt.assert_awaited_once() wallet.melt.assert_awaited_once()
wallet.get_melt_quote.assert_not_awaited()
+91
View File
@@ -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()])
+65
View File
@@ -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)
+1 -15
View File
@@ -7,21 +7,7 @@ os.environ["UPSTREAM_BASE_URL"] = "http://test"
os.environ["UPSTREAM_API_KEY"] = "test" os.environ["UPSTREAM_API_KEY"] = "test"
from routstr.core.settings import settings # noqa: E402 from routstr.core.settings import settings # noqa: E402
from routstr.payment.helpers import ( # noqa: E402 from routstr.payment.helpers import get_max_cost_for_model # 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
async def test_get_max_cost_for_model_known() -> None: async def test_get_max_cost_for_model_known() -> None:
+1 -1
View File
@@ -422,4 +422,4 @@ async def test_proxy_reverts_reservation_on_client_disconnect() -> None:
with pytest.raises(asyncio.CancelledError): with pytest.raises(asyncio.CancelledError):
await proxy_module.proxy(request, "v1/chat/completions", session=session) 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)
+1 -1
View File
@@ -406,4 +406,4 @@ async def test_proxy_loop_surfaces_rate_limit_and_reverts_once() -> None:
assert RAW_ORG_ID not in serialized assert RAW_ORG_ID not in serialized
assert "org-[REDACTED]" in serialized assert "org-[REDACTED]" in serialized
# Single upstream failed -> reservation reverted exactly once (no double-charge). # 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)
+47 -54
View File
@@ -7,6 +7,7 @@ from unittest.mock import AsyncMock, Mock, patch
import httpx import httpx
import pytest import pytest
from cashu.core.base import MeltQuoteState
from routstr.core.db import ApiKey from routstr.core.db import ApiKey
from routstr.wallet import ( from routstr.wallet import (
@@ -639,7 +640,9 @@ def _make_swap_mocks(
quote=f"melt_quote_{invoice}", amount=invoice, fee_reserve=_next_fee() 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 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. " "Mint Error: not enough inputs provided for melt. "
"Provided: 179, needed: 180 (Code: 11000)" "Provided: 179, needed: 180 (Code: 11000)"
), ),
Mock(), Mock(state=MeltQuoteState.paid),
] ]
from routstr.core.settings import settings 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 = [ mock_token_wallet.melt.side_effect = [
Exception("Mint Error: Transaction unbalanced: 179, 178, 2 (Code: 11005)"), Exception("Mint Error: Transaction unbalanced: 179, 178, 2 (Code: 11005)"),
Mock(), Mock(state=MeltQuoteState.paid),
] ]
from routstr.core.settings import settings 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 @pytest.mark.asyncio
async def test_swap_melt_transport_error_raises_mint_connection_error() -> None: async def test_swap_melt_transport_error_is_never_reported_reusable() -> None:
"""A transport failure during melt is surfaced as MintConnectionError and """A timed-out melt remains ambiguous even when an immediate snapshot says
is NOT retried — the mint is down, not demanding higher fees.""" UNPAID/UNSPENT, so callers must not receive the original token for retry."""
from routstr.wallet import swap_to_primary_mint from routstr.wallet import swap_to_primary_mint
mock_token, mock_token_wallet, mock_primary_wallet = _make_swap_mocks( mock_token, mock_token_wallet, mock_primary_wallet = _make_swap_mocks(
1000, fee_reserves=[10, 10] 1000, fee_reserves=[10, 10]
) )
mock_token_wallet.melt = AsyncMock(side_effect=httpx.ConnectTimeout("timed out")) 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 from routstr.core.settings import settings
with patch.object(settings, "primary_mint", "http://primary:3338"): with patch.object(settings, "primary_mint", "http://primary:3338"):
with patch.object(settings, "primary_mint_unit", "sat"): with patch.object(settings, "primary_mint_unit", "sat"):
with patch("routstr.wallet.get_wallet", return_value=mock_primary_wallet): 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) await swap_to_primary_mint(mock_token, mock_token_wallet)
assert mock_token_wallet.melt.call_count == 1 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 guard._cooldown_until = 15.0
operation = AsyncMock(return_value="ok") operation = AsyncMock(return_value="ok")
with patch("routstr.wallet.time.monotonic", return_value=10.0): with patch("routstr.mint.time.monotonic", return_value=10.0):
with patch("routstr.wallet.asyncio.sleep", AsyncMock()) as sleep: with patch("routstr.mint.asyncio.sleep", AsyncMock()) as sleep:
assert await guard.run(operation) == "ok" assert await guard.run(operation) == "ok"
sleep.assert_awaited_once_with(5.0) 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] expected_delays = [60, 120, 240, 480, 960, 1920, 3840, 7680, 15360, 25200]
now = 0.0 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): for index, expected in enumerate(expected_delays, start=1):
monotonic.return_value = now monotonic.return_value = now
assert guard.apply_rate_limit_cooldown(60) == expected 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") operation = AsyncMock(return_value="ok")
with ( with (
patch.object(settings, "mint_max_concurrency", 0), patch.object(settings, "mint_max_concurrency", 0),
patch("routstr.wallet.time.monotonic", return_value=0), patch("routstr.mint.time.monotonic", return_value=0),
patch("routstr.wallet.asyncio.sleep", AsyncMock()) as sleep, patch("routstr.mint.asyncio.sleep", AsyncMock()) as sleep,
): ):
guard = _MintRateGuard.get("http://mint:3338") guard = _MintRateGuard.get("http://mint:3338")
guard.apply_cooldown(5) 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_retry_max_attempts", 1):
with patch.object(settings, "mint_operation_timeout_seconds", 0): with patch.object(settings, "mint_operation_timeout_seconds", 0):
with patch.object(settings, "mint_max_concurrency", 1): with patch.object(settings, "mint_max_concurrency", 1):
with patch("routstr.wallet.time.monotonic", return_value=0.1): with patch("routstr.mint.time.monotonic", return_value=0.1):
with patch("routstr.wallet.asyncio.sleep", sleep): with patch("routstr.mint.asyncio.sleep", sleep):
result = await _mint_operation( result = await _mint_operation(
factory, mint_url="http://mint:3338" factory, mint_url="http://mint:3338"
) )
@@ -1734,7 +1747,7 @@ async def test_mint_operation_timeout_excludes_adaptive_cooldown() -> None:
with ( with (
patch.object(settings, "mint_max_concurrency", 1), patch.object(settings, "mint_max_concurrency", 1),
patch.object(settings, "mint_operation_timeout_seconds", 0.01), 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 = _MintRateGuard.get("http://mint:3338")
guard.apply_cooldown(60) 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_retry_max_attempts", 3),
patch.object(settings, "mint_operation_timeout_seconds", 30), patch.object(settings, "mint_operation_timeout_seconds", 30),
patch.object(settings, "mint_max_concurrency", 1), 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" 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_retry_max_attempts", 2):
with patch.object(settings, "mint_operation_timeout_seconds", 0): 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" assert await _mint_operation(retrying) == "ok"
with pytest.raises(httpx.TimeoutException): with pytest.raises(httpx.TimeoutException):
await _mint_operation(non_retrying, retry_timeouts=False) 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: ) as create:
# A fresh wallet must load even when the host has been up for less than # A fresh wallet must load even when the host has been up for less than
# the reload interval. # 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( first, second = await asyncio.gather(
get_wallet("http://mint:3338"), get_wallet("http://mint:3338") 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("routstr.wallet.Wallet.with_db", AsyncMock(return_value=wallet)),
patch.object(settings, "mint_retry_max_attempts", 3), patch.object(settings, "mint_retry_max_attempts", 3),
patch.object(settings, "mint_operation_timeout_seconds", 0), 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): with pytest.raises(httpx.HTTPStatusError):
await get_wallet("http://mint:3338", retry_on_rate_limit=False) 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( melt_quote=AsyncMock(
return_value=Mock(quote="melt_q", amount=990, fee_reserve=10) 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") 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("asyncio.sleep", AsyncMock()),
patch("routstr.wallet.get_wallet", side_effect=mock_get), patch("routstr.wallet.get_wallet", side_effect=mock_get),
patch("routstr.wallet.logger.warning") as warning, 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) 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) assert any(call.args[0] == secondary for call in mock_get.await_args_list)
events = { events = {
call.kwargs["extra"]["event"] 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"] if "extra" in call.kwargs and "event" in call.kwargs["extra"]
} }
assert "cashu_destination_failed" in events 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, "cashu_mints", [primary, secondary]),
patch.object(settings, "mint_max_concurrency", 0), patch.object(settings, "mint_max_concurrency", 0),
patch.object(settings, "mint_operation_timeout_seconds", 0), patch.object(settings, "mint_operation_timeout_seconds", 0),
patch("routstr.wallet.time.monotonic", return_value=10), patch("routstr.mint.time.monotonic", return_value=10),
patch("routstr.wallet.asyncio.sleep", AsyncMock()) as sleep, patch("routstr.mint.asyncio.sleep", AsyncMock()) as sleep,
patch( patch(
"routstr.wallet.get_wallet", "routstr.wallet.get_wallet",
AsyncMock(side_effect=lambda mint, *args, **kwargs: wallets[mint]), 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 @pytest.mark.asyncio
async def test_probe_does_not_escalate_consecutive_rate_limits() -> None: async def test_probe_does_not_escalate_consecutive_rate_limits() -> None:
"""When a probe fails with a rate limit, _consecutive_rate_limits should from routstr.mint import MintRateGuard
NOT increment — the probe is a recovery check, not a new request."""
from routstr.wallet import _MINT_RATE_LIMIT_BASE_COOLDOWN_SECONDS, _MintRateGuard
guard = _MintRateGuard("http://mint", max_concurrency=0) guard = MintRateGuard("http://mint", max_concurrency=0)
# Simulate initial rate limit: apply_rate_limit_cooldown increments counter
guard.apply_rate_limit_cooldown() guard.apply_rate_limit_cooldown()
assert guard._consecutive_rate_limits == 1 guard._cooldown_until = 0.0
cooldown_before = guard._cooldown_until
assert cooldown_before > 0
# Simulate probe failure: _run_probe uses apply_cooldown, NOT with pytest.raises(httpx.HTTPStatusError):
# apply_rate_limit_cooldown, so the counter stays at 1. await guard.run(AsyncMock(side_effect=_http_429_error()))
guard.apply_cooldown(_MINT_RATE_LIMIT_BASE_COOLDOWN_SECONDS, reason="rate_limited")
assert guard._consecutive_rate_limits == 1 # unchanged! assert guard._consecutive_rate_limits == 1
assert guard._needs_probe is True assert guard._needs_probe is True
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_probe_recovery_resets_consecutive_rate_limits() -> None: async def test_probe_recovery_resets_consecutive_rate_limits() -> None:
"""A successful probe resets _consecutive_rate_limits to 0.""" from routstr.mint import MintRateGuard
from routstr.wallet import _MintRateGuard
guard = _MintRateGuard("http://mint", max_concurrency=0) guard = MintRateGuard("http://mint", max_concurrency=0)
# First rate limit: increments to 1, sets 60s cooldown.
guard.apply_rate_limit_cooldown() 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_until = 0.0
guard._cooldown_reason = None
# Second rate limit (after cooldown expired): increments to 2. assert await guard.run(AsyncMock(return_value="ok")) == "ok"
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 guard._consecutive_rate_limits == 0 assert guard._consecutive_rate_limits == 0
assert guard._needs_probe is False assert guard._needs_probe is False