From 7394b10e75cc908f790a3e935b7be8b1dcbc839c Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Fri, 24 Jul 2026 21:29:21 +0200 Subject: [PATCH 1/3] fix-concurent-issue --- routstr/core/db.py | 104 ++++++- routstr/core/settings.py | 3 + routstr/lightning.py | 143 +++++---- routstr/proxy.py | 15 + routstr/wallet.py | 274 +++++++++--------- .../test_lightning_invoice_constraints.py | 144 ++++++++- tests/unit/test_db_pool_config.py | 83 ++++++ tests/unit/test_fee_payout_crash_safety.py | 30 +- tests/unit/test_fetch_all_balances.py | 174 ++++++++++- tests/unit/test_periodic_payout.py | 148 +++++++--- tests/unit/test_proxy_session_lifecycle.py | 43 +++ tests/unit/test_refund_sweep.py | 38 +++ 12 files changed, 952 insertions(+), 247 deletions(-) create mode 100644 tests/unit/test_db_pool_config.py create mode 100644 tests/unit/test_proxy_session_lifecycle.py diff --git a/routstr/core/db.py b/routstr/core/db.py index 16cfedfd..ab556cc5 100644 --- a/routstr/core/db.py +++ b/routstr/core/db.py @@ -12,7 +12,8 @@ from typing import AsyncGenerator from alembic import command from alembic.config import Config from alembic.util.exc import CommandError -from sqlalchemy import Index, UniqueConstraint, case, delete, or_ +from sqlalchemy import Index, UniqueConstraint, case, delete, event, or_ +from sqlalchemy.engine import make_url from sqlalchemy.exc import IntegrityError, OperationalError from sqlalchemy.ext.asyncio.engine import create_async_engine from sqlalchemy.orm import aliased @@ -26,7 +27,82 @@ logger = get_logger(__name__) DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite+aiosqlite:///keys.db") -engine = create_async_engine(DATABASE_URL, echo=False) # echo=True for debugging SQL +def _env_bool(name: str, default: bool) -> bool: + raw = os.environ.get(name) + if raw is None: + return default + normalized = raw.strip().lower() + if normalized in {"1", "true", "yes", "on"}: + return True + if normalized in {"0", "false", "no", "off"}: + return False + raise ValueError(f"{name} must be a boolean") + + +def _engine_options(database_url: str) -> dict[str, int | float | bool]: + """Build a bounded pool configuration, preserving SQLite memory semantics.""" + options: dict[str, int | float | bool] = { + "pool_pre_ping": _env_bool("DATABASE_POOL_PRE_PING", True) + } + url = make_url(database_url) + is_memory_sqlite = url.get_backend_name() == "sqlite" and url.database in { + None, + "", + ":memory:", + } + if is_memory_sqlite: + return options + + pool_size = int(os.environ.get("DATABASE_POOL_SIZE", "5")) + max_overflow = int(os.environ.get("DATABASE_MAX_OVERFLOW", "0")) + pool_timeout = float(os.environ.get("DATABASE_POOL_TIMEOUT", "5")) + pool_recycle = int(os.environ.get("DATABASE_POOL_RECYCLE", "1800")) + if pool_size < 1: + raise ValueError("DATABASE_POOL_SIZE must be at least 1") + if max_overflow < 0: + raise ValueError("DATABASE_MAX_OVERFLOW cannot be negative") + if pool_timeout <= 0: + raise ValueError("DATABASE_POOL_TIMEOUT must be positive") + if pool_recycle < 0: + raise ValueError("DATABASE_POOL_RECYCLE cannot be negative") + options.update( + pool_size=pool_size, + max_overflow=max_overflow, + pool_timeout=pool_timeout, + pool_recycle=pool_recycle, + ) + return options + + +engine = create_async_engine(DATABASE_URL, echo=False, **_engine_options(DATABASE_URL)) + +_POOL_HOLD_WARN_SECONDS = float(os.environ.get("DATABASE_POOL_HOLD_WARN_SECONDS", "10")) +if _POOL_HOLD_WARN_SECONDS <= 0: + raise ValueError("DATABASE_POOL_HOLD_WARN_SECONDS must be positive") + + +@event.listens_for(engine.sync_engine, "checkout") +def _record_pool_checkout( + dbapi_connection: object, connection_record: object, proxy: object +) -> None: + connection_record.info["routstr_checked_out_at"] = time.monotonic() # type: ignore[attr-defined] + + +@event.listens_for(engine.sync_engine, "checkin") +def _record_pool_checkin(dbapi_connection: object, connection_record: object) -> None: + checked_out_at = connection_record.info.pop("routstr_checked_out_at", None) # type: ignore[attr-defined] + if checked_out_at is None: + return + held_seconds = time.monotonic() - checked_out_at + if held_seconds >= _POOL_HOLD_WARN_SECONDS: + logger.warning( + "Database connection held longer than threshold", + extra={ + "held_seconds": round(held_seconds, 3), + "threshold_seconds": _POOL_HOLD_WARN_SECONDS, + "pool_status": engine.pool.status(), + }, + ) class ApiKey(SQLModel, table=True): # type: ignore @@ -727,6 +803,30 @@ async def balances_for_mint_and_unit( return result.one() or 0 +async def balances_by_mint_and_unit( + db_session: AsyncSession, +) -> dict[tuple[str, str], int]: + """Return all user liabilities in one query, grouped by mint and unit.""" + query = ( + select( + col(ApiKey.refund_mint_url), + col(ApiKey.refund_currency), + func.sum(ApiKey.balance), + ) + .where( + col(ApiKey.refund_mint_url).is_not(None), + col(ApiKey.refund_currency).is_not(None), + ) + .group_by(col(ApiKey.refund_mint_url), col(ApiKey.refund_currency)) + ) + result = await db_session.exec(query) + return { + (mint_url, unit): int(balance or 0) + for mint_url, unit, balance in result.all() + if mint_url is not None and unit is not None + } + + async def init_db() -> None: """Initializes the database and creates tables if they don't exist.""" async with engine.begin() as conn: diff --git a/routstr/core/settings.py b/routstr/core/settings.py index a0b5d05c..991c7554 100644 --- a/routstr/core/settings.py +++ b/routstr/core/settings.py @@ -41,6 +41,9 @@ class Settings(BaseSettings): receive_ln_address: str = Field(default="", env="RECEIVE_LN_ADDRESS") primary_mint: str = Field(default="", env="PRIMARY_MINT_URL") primary_mint_unit: str = Field(default="sat", env="PRIMARY_MINT_UNIT") + mint_operation_concurrency: int = Field( + default=4, ge=1, env="MINT_OPERATION_CONCURRENCY" + ) # Lightning payout configuration # Minimum available balance (in satoshis) before profit is paid out over diff --git a/routstr/lightning.py b/routstr/lightning.py index b0bbc63b..30aa7918 100644 --- a/routstr/lightning.py +++ b/routstr/lightning.py @@ -5,7 +5,7 @@ import time from fastapi import APIRouter, Depends, Header, HTTPException from pydantic import BaseModel, Field -from sqlmodel import col, select +from sqlmodel import col, select, update from sqlmodel.ext.asyncio.session import AsyncSession from .core.db import ApiKey, LightningInvoice, create_session, get_session @@ -222,62 +222,114 @@ async def recover_invoice( async def check_invoice_payment( invoice: LightningInvoice, session: AsyncSession ) -> None: + minted = False + invoice_id = invoice.id + invoice_purpose = invoice.purpose try: + # A preceding invoice lookup starts a transaction. End it before the + # potentially slow mint request so it cannot pin a pool connection. + await session.commit() + wallet = await get_wallet(settings.primary_mint, "sat") - mint_status = await wallet.get_mint_quote(invoice.payment_hash) + if not mint_status.paid: + return - if mint_status.paid: - invoice.status = "paid" - invoice.paid_at = int(time.time()) + # The mint enforces single-use quotes, so a concurrent checker that + # races us here fails inside wallet.mint rather than double-minting. + await wallet.mint(invoice.amount_sats, quote_id=invoice.payment_hash) + minted = True - if invoice.purpose == "create": - api_key = await create_api_key_from_invoice(invoice, session) - invoice.api_key_hash = api_key.hashed_key - elif invoice.purpose == "topup" and invoice.api_key_hash: - await topup_api_key_from_invoice(invoice, session) + if invoice_purpose == "create": + api_key = await _create_api_key_record(invoice, session) + invoice.api_key_hash = api_key.hashed_key + elif invoice_purpose == "topup": + await _credit_topup_record(invoice, session) - await session.commit() - - logger.info( - "Lightning invoice paid", - extra={ - "invoice_id": invoice.id, - "amount_sats": invoice.amount_sats, - "purpose": invoice.purpose, - "api_key_hash": invoice.api_key_hash[:8] + "..." - if invoice.api_key_hash - else None, - }, + # Conditional transition guards against double-credit: the credit + # above and this status flip commit atomically, and a lost race + # rolls both back. + paid_at = int(time.time()) + finalized = await session.exec( # type: ignore[call-overload] + update(LightningInvoice) + .where( + col(LightningInvoice.id) == invoice_id, + col(LightningInvoice.status) == "pending", ) + .values( + status="paid", + paid_at=paid_at, + api_key_hash=invoice.api_key_hash, + ) + ) + if finalized.rowcount != 1: + await session.rollback() + await session.refresh(invoice) + return + await session.commit() + invoice.status = "paid" + invoice.paid_at = paid_at + logger.info( + "Lightning invoice paid", + extra={ + "invoice_id": invoice_id, + "amount_sats": invoice.amount_sats, + "purpose": invoice_purpose, + "api_key_hash": invoice.api_key_hash[:8] + "..." + if invoice.api_key_hash + else None, + }, + ) except Exception as e: + await session.rollback() + if minted: + logger.critical( + "Invoice mint succeeded but DB finalization failed; reconciliation required", + extra={"invoice_id": invoice_id, "purpose": invoice_purpose}, + ) logger.error(f"Failed to check invoice payment: {e}") +async def _create_api_key_record( + invoice: LightningInvoice, session: AsyncSession +) -> ApiKey: + dummy_token = f"invoice-{invoice.id}-{invoice.payment_hash}" + hashed_key = hashlib.sha256(dummy_token.encode()).hexdigest() + api_key = ApiKey( + hashed_key=hashed_key, + balance=invoice.amount_sats * 1000, + refund_currency="sat", + refund_mint_url=settings.primary_mint, + balance_limit=invoice.balance_limit, + balance_limit_reset=invoice.balance_limit_reset, + validity_date=invoice.validity_date, + ) + session.add(api_key) + await session.flush() + return api_key + + +async def _credit_topup_record( + invoice: LightningInvoice, session: AsyncSession +) -> None: + if not invoice.api_key_hash: + raise ValueError("No API key associated with topup invoice") + credited = await session.exec( # type: ignore[call-overload] + update(ApiKey) + .where(col(ApiKey.hashed_key) == invoice.api_key_hash) + .values(balance=col(ApiKey.balance) + invoice.amount_sats * 1000) + ) + if credited.rowcount != 1: + raise ValueError("Associated API key not found") + + async def create_api_key_from_invoice( invoice: LightningInvoice, session: AsyncSession ) -> ApiKey: wallet = await get_wallet(settings.primary_mint, "sat") await wallet.mint(invoice.amount_sats, quote_id=invoice.payment_hash) - - dummy_token = f"invoice-{invoice.id}-{invoice.payment_hash}" - hashed_key = hashlib.sha256(dummy_token.encode()).hexdigest() - - api_key = ApiKey( - hashed_key=hashed_key, - balance=invoice.amount_sats * 1000, # Convert to msats - refund_currency="sat", - refund_mint_url=settings.primary_mint, - balance_limit=invoice.balance_limit, - balance_limit_reset=invoice.balance_limit_reset, - validity_date=invoice.validity_date, - ) - - session.add(api_key) - await session.flush() - - return api_key + return await _create_api_key_record(invoice, session) async def topup_api_key_from_invoice( @@ -285,16 +337,7 @@ async def topup_api_key_from_invoice( ) -> None: wallet = await get_wallet(settings.primary_mint, "sat") await wallet.mint(invoice.amount_sats, quote_id=invoice.payment_hash) - - if not invoice.api_key_hash: - raise ValueError("No API key associated with topup invoice") - - api_key = await session.get(ApiKey, invoice.api_key_hash) - if not api_key: - raise ValueError("Associated API key not found") - - api_key.balance += invoice.amount_sats * 1000 # Convert to msats - await session.flush() + await _credit_topup_record(invoice, session) INVOICE_WATCH_INTERVAL_SECONDS = 5 diff --git a/routstr/proxy.py b/routstr/proxy.py index c0b8794f..46596077 100644 --- a/routstr/proxy.py +++ b/routstr/proxy.py @@ -1,4 +1,5 @@ import asyncio +import inspect import json from typing import Any @@ -220,6 +221,20 @@ _API_PATH_PREFIXES = ( @proxy_router.api_route("/{path:path}", methods=["GET", "POST"], response_model=None) async def proxy( request: Request, path: str, session: AsyncSession = Depends(get_session) +) -> Response | StreamingResponse: + """Run proxy setup in a short request session, never across response streaming.""" + try: + return await _proxy(request, path, session) + finally: + # FastAPI yield dependencies normally close after the response body is + # sent. Close explicitly so a long stream cannot retain DB resources. + close_result = session.close() + if inspect.isawaitable(close_result): + await close_result + + +async def _proxy( + request: Request, path: str, session: AsyncSession ) -> Response | StreamingResponse: # GET requests must hit a known API prefix; otherwise return a 404 (HTML # for browsers, JSON for API clients). POST requests are always forwarded diff --git a/routstr/wallet.py b/routstr/wallet.py index dd92d913..24b11fe9 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -836,16 +836,20 @@ async def fetch_all_balances( if units is None: units = ["sat", "msat"] - async def fetch_balance( - session: db.AsyncSession, mint_url: str, unit: str - ) -> BalanceDetail: + async with db.create_session() as session: + user_balances = await db.balances_by_mint_and_unit(session) + + mint_check_limit = asyncio.Semaphore(settings.mint_operation_concurrency) + + async def fetch_balance(mint_url: str, unit: str) -> BalanceDetail: try: - wallet = await get_wallet(mint_url, unit) - proofs = get_proofs_per_mint_and_unit( - wallet, mint_url, unit, not_reserved=True - ) - proofs = await slow_filter_spend_proofs(proofs, wallet) - user_balance = await db.balances_for_mint_and_unit(session, mint_url, unit) + async with mint_check_limit: + wallet = await get_wallet(mint_url, unit) + proofs = get_proofs_per_mint_and_unit( + wallet, mint_url, unit, not_reserved=True + ) + proofs = await slow_filter_spend_proofs(proofs, wallet) + user_balance = user_balances.get((mint_url, unit), 0) if unit == "sat": user_balance = user_balance // 1000 proofs_balance = sum(proof.amount for proof in proofs) @@ -878,16 +882,8 @@ async def fetch_all_balances( if settings.primary_mint and settings.primary_mint not in mint_urls: mint_urls.append(settings.primary_mint) - # Create tasks for all mint/unit combinations - async with db.create_session() as session: - tasks = [ - fetch_balance(session, mint_url, unit) - for mint_url in mint_urls - for unit in units - ] - - # Run all tasks concurrently - balance_details = list(await asyncio.gather(*tasks)) + tasks = [fetch_balance(mint_url, unit) for mint_url in mint_urls for unit in units] + balance_details = list(await asyncio.gather(*tasks)) # Calculate totals total_wallet_balance_sats = 0 @@ -936,56 +932,56 @@ async def periodic_payout() -> None: mint_urls.append(settings.primary_mint) async with db.create_session() as session: - for mint_url in mint_urls: - for unit in ["sat", "msat"]: - # Isolate failures per mint/unit so one slow or failing - # mint does not abort payout for every other mint/unit. - try: - wallet = await get_wallet(mint_url, unit) - proofs = get_proofs_per_mint_and_unit( - wallet, mint_url, unit, not_reserved=True + user_balances = await db.balances_by_mint_and_unit(session) + + for mint_url in mint_urls: + for unit in ["sat", "msat"]: + # Isolate failures per mint/unit so one slow or failing + # mint does not abort payout for every other mint/unit. + try: + wallet = await get_wallet(mint_url, unit) + proofs = get_proofs_per_mint_and_unit( + wallet, mint_url, unit, not_reserved=True + ) + proofs = await slow_filter_spend_proofs(proofs, wallet) + await asyncio.sleep(5) + user_balance = user_balances.get((mint_url, unit), 0) + if unit == "sat": + user_balance = user_balance // 1000 + proofs_balance = sum(proof.amount for proof in proofs) + available_balance = proofs_balance - user_balance + # Threshold is configured in sats; convert for msat wallets. + min_amount = ( + settings.min_payout_sat + if unit == "sat" + else settings.min_payout_sat * 1000 + ) + if available_balance > min_amount: + amount_received = await raw_send_to_lnurl( + wallet, + proofs, + settings.receive_ln_address, + unit, + amount=available_balance, ) - proofs = await slow_filter_spend_proofs(proofs, wallet) - await asyncio.sleep(5) - user_balance = await db.balances_for_mint_and_unit( - session, mint_url, unit - ) - if unit == "sat": - user_balance = user_balance // 1000 - proofs_balance = sum(proof.amount for proof in proofs) - available_balance = proofs_balance - user_balance - # Threshold is configured in sats; convert for msat wallets. - min_amount = ( - settings.min_payout_sat - if unit == "sat" - else settings.min_payout_sat * 1000 - ) - if available_balance > min_amount: - amount_received = await raw_send_to_lnurl( - wallet, - proofs, - settings.receive_ln_address, - unit, - amount=available_balance, - ) - logger.info( - "Payout sent successfully", - extra={ - "mint_url": mint_url, - "unit": unit, - "balance": available_balance, - "amount_received": amount_received, - }, - ) - except Exception as e: - logger.error( - f"Error sending payout: {type(e).__name__}", + logger.info( + "Payout sent successfully", extra={ - "error": str(e), "mint_url": mint_url, "unit": unit, + "balance": available_balance, + "amount_received": amount_received, }, ) + except Exception as e: + logger.error( + f"Error sending payout: {type(e).__name__}", + extra={ + "error": str(e), + "mint_url": mint_url, + "unit": unit, + }, + ) except Exception as e: logger.error( f"Error in periodic payout cycle: {type(e).__name__}", @@ -1004,39 +1000,49 @@ async def _refund_sweep_once(cutoff: int) -> None: results = await session.exec(stmt) refunds = results.all() - for refund in refunds: - try: - await recieve_token(refund.token) - refund.swept = True - session.add(refund) + for refund in refunds: + try: + await recieve_token(refund.token) + async with db.create_session() as session: + await session.exec( # type: ignore[call-overload] + update(db.CashuTransaction) + .where( + col(db.CashuTransaction.id) == refund.id, + col(db.CashuTransaction.swept) == False, # noqa: E712 + ) + .values(swept=True) + ) + await session.commit() + logger.info( + "Swept uncollected refund", + extra={ + "id": refund.id, + "amount": refund.amount, + "unit": refund.unit, + }, + ) + except Exception as e: + error_msg = str(e).lower() + if "already spent" in error_msg: + async with db.create_session() as session: + await session.exec( # type: ignore[call-overload] + update(db.CashuTransaction) + .where(col(db.CashuTransaction.id) == refund.id) + .values(collected=True) + ) + await session.commit() logger.info( - "Swept uncollected refund", + "Refund already spent (client collected), marking swept", + extra={"id": refund.id}, + ) + else: + logger.warning( + "Failed to sweep refund", extra={ "id": refund.id, - "amount": refund.amount, - "unit": refund.unit, + "error": str(e), }, ) - except Exception as e: - error_msg = str(e).lower() - if "already spent" in error_msg: - refund.collected = True - session.add(refund) - logger.info( - "Refund already spent (client collected), marking swept", - extra={ - "id": refund.id, - }, - ) - else: - logger.warning( - "Failed to sweep refund", - extra={ - "id": refund.id, - "error": str(e), - }, - ) - await session.commit() async def refund_sweep_once() -> None: @@ -1083,52 +1089,52 @@ async def periodic_routstr_fee_payout() -> None: continue accumulated_sats = fee.accumulated_msats // 1000 - if accumulated_sats >= ROUTSTR_FEE_DEFAULT_PAYOUT: - wallet = await get_wallet(settings.primary_mint, "sat") - proofs = get_proofs_per_mint_and_unit( - wallet, settings.primary_mint, "sat", not_reserved=True - ) - paid_msats = accumulated_sats * 1000 - payout_checkpointed = await db.reset_routstr_fee( - session, paid_msats - ) - if not payout_checkpointed: - logger.warning("Routstr fee payout was already claimed") - continue + if accumulated_sats < ROUTSTR_FEE_DEFAULT_PAYOUT: + continue + paid_msats = accumulated_sats * 1000 + payout_checkpointed = await db.reset_routstr_fee(session, paid_msats) + if not payout_checkpointed: + logger.warning("Routstr fee payout was already claimed") + continue - try: - amount_received = await raw_send_to_lnurl( - wallet, - proofs, - ROUTSTR_LN_ADDRESS, - "sat", - amount=accumulated_sats, - ) - except Exception: - logger.critical( - "Routstr fee payout outcome is unknown; manual reconciliation required", - extra={"payout_in_progress_msats": paid_msats}, - exc_info=True, - ) - continue + wallet = await get_wallet(settings.primary_mint, "sat") + proofs = get_proofs_per_mint_and_unit( + wallet, settings.primary_mint, "sat", not_reserved=True + ) + try: + amount_received = await raw_send_to_lnurl( + wallet, + proofs, + ROUTSTR_LN_ADDRESS, + "sat", + amount=accumulated_sats, + ) + except Exception: + logger.critical( + "Routstr fee payout outcome is unknown; manual reconciliation required", + extra={"payout_in_progress_msats": paid_msats}, + exc_info=True, + ) + continue - payout_completed = await db.complete_routstr_fee_payout( - session, paid_msats - ) - if not payout_completed: - logger.critical( - "Routstr fee payout sent but checkpoint was not completed", - extra={"payout_in_progress_msats": paid_msats}, - ) - continue + async with db.create_session() as session: + payout_completed = await db.complete_routstr_fee_payout( + session, paid_msats + ) + if not payout_completed: + logger.critical( + "Routstr fee payout sent but checkpoint was not completed", + extra={"payout_in_progress_msats": paid_msats}, + ) + continue - logger.info( - "Routstr fee payout sent", - extra={ - "accumulated_sats": accumulated_sats, - "amount_received": amount_received, - }, - ) + logger.info( + "Routstr fee payout sent", + extra={ + "accumulated_sats": accumulated_sats, + "amount_received": amount_received, + }, + ) except Exception as e: logger.error( f"Error in Routstr fee payout: {type(e).__name__}", diff --git a/tests/integration/test_lightning_invoice_constraints.py b/tests/integration/test_lightning_invoice_constraints.py index a26b9083..8c8e15d9 100644 --- a/tests/integration/test_lightning_invoice_constraints.py +++ b/tests/integration/test_lightning_invoice_constraints.py @@ -9,10 +9,12 @@ Covers two things: from __future__ import annotations +import asyncio import time -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest +from sqlalchemy.ext.asyncio import AsyncEngine from sqlmodel.ext.asyncio.session import AsyncSession from routstr.core.db import ApiKey, LightningInvoice @@ -48,6 +50,7 @@ def mock_wallet_mint() -> object: # Persistence # --------------------------------------------------------------------------- + @pytest.mark.asyncio async def test_invoice_persists_balance_limit( integration_session: AsyncSession, @@ -92,6 +95,7 @@ async def test_invoice_persists_validity_date( # Propagation to ApiKey # --------------------------------------------------------------------------- + @pytest.mark.asyncio async def test_created_key_receives_balance_limit( integration_session: AsyncSession, @@ -141,6 +145,144 @@ async def test_created_key_receives_validity_date( assert stored_key.validity_date == expiry +@pytest.mark.asyncio +async def test_payment_check_releases_connection_during_mint_quote( + integration_engine: AsyncEngine, +) -> None: + invoice = _make_invoice(id="inv_slow_quote", status="pending", paid_at=None) + async with AsyncSession(integration_engine, expire_on_commit=False) as setup: + setup.add(invoice) + await setup.commit() + + async with AsyncSession(integration_engine, expire_on_commit=False) as session: + stored = await session.get(LightningInvoice, invoice.id) + assert stored is not None + + async def quote_status(*args: object, **kwargs: object) -> MagicMock: + assert integration_engine.pool.checkedout() == 0 # type: ignore[attr-defined] + return MagicMock(paid=False) + + wallet = MagicMock() + wallet.get_mint_quote = AsyncMock(side_effect=quote_status) + with patch("routstr.lightning.get_wallet", AsyncMock(return_value=wallet)): + from routstr.lightning import check_invoice_payment + + await check_invoice_payment(stored, session) + + +@pytest.mark.asyncio +async def test_concurrent_payment_checks_mint_and_credit_invoice_once( + integration_engine: AsyncEngine, +) -> None: + invoice = _make_invoice(id="inv_concurrent", status="pending", paid_at=None) + async with AsyncSession(integration_engine, expire_on_commit=False) as setup: + setup.add(invoice) + await setup.commit() + + wallet = MagicMock() + wallet.get_mint_quote = AsyncMock(return_value=MagicMock(paid=True)) + + mint_calls = 0 + + async def single_use_mint(*args: object, **kwargs: object) -> list[object]: + # Real mints enforce single-use quotes: the second concurrent minter + # gets rejected at the mint, mirroring cashu quote semantics. + nonlocal mint_calls + mint_calls += 1 + call_number = mint_calls + await asyncio.sleep(0.05) + if call_number > 1: + raise Exception("quote already issued") + return [] + + wallet.mint = AsyncMock(side_effect=single_use_mint) + + async with ( + AsyncSession(integration_engine, expire_on_commit=False) as first, + AsyncSession(integration_engine, expire_on_commit=False) as second, + ): + first_invoice = await first.get(LightningInvoice, invoice.id) + second_invoice = await second.get(LightningInvoice, invoice.id) + assert first_invoice is not None + assert second_invoice is not None + + with patch("routstr.lightning.get_wallet", AsyncMock(return_value=wallet)): + from routstr.lightning import check_invoice_payment + + await asyncio.gather( + check_invoice_payment(first_invoice, first), + check_invoice_payment(second_invoice, second), + ) + + async with AsyncSession(integration_engine, expire_on_commit=False) as verify: + stored_invoice = await verify.get(LightningInvoice, invoice.id) + assert stored_invoice is not None + assert stored_invoice.status == "paid" + assert stored_invoice.api_key_hash is not None + stored_key = await verify.get(ApiKey, stored_invoice.api_key_hash) + assert stored_key is not None + assert stored_key.balance == invoice.amount_sats * 1000 + + +@pytest.mark.asyncio +async def test_failed_mint_keeps_invoice_pending_for_retry( + integration_engine: AsyncEngine, +) -> None: + invoice = _make_invoice(id="inv_mint_failure", status="pending", paid_at=None) + async with AsyncSession(integration_engine, expire_on_commit=False) as setup: + setup.add(invoice) + await setup.commit() + + wallet = MagicMock() + wallet.get_mint_quote = AsyncMock(return_value=MagicMock(paid=True)) + wallet.mint = AsyncMock(side_effect=TimeoutError("mint unavailable")) + async with AsyncSession(integration_engine, expire_on_commit=False) as session: + stored = await session.get(LightningInvoice, invoice.id) + assert stored is not None + with patch("routstr.lightning.get_wallet", AsyncMock(return_value=wallet)): + from routstr.lightning import check_invoice_payment + + await check_invoice_payment(stored, session) + + async with AsyncSession(integration_engine, expire_on_commit=False) as verify: + stored = await verify.get(LightningInvoice, invoice.id) + assert stored is not None + assert stored.status == "pending" + + +@pytest.mark.asyncio +async def test_post_mint_db_failure_keeps_invoice_pending_for_reconciliation( + integration_engine: AsyncEngine, +) -> None: + invoice = _make_invoice(id="inv_finalize_failure", status="pending", paid_at=None) + async with AsyncSession(integration_engine, expire_on_commit=False) as setup: + setup.add(invoice) + await setup.commit() + + wallet = MagicMock() + wallet.get_mint_quote = AsyncMock(return_value=MagicMock(paid=True)) + wallet.mint = AsyncMock(return_value=[]) + async with AsyncSession(integration_engine, expire_on_commit=False) as session: + stored = await session.get(LightningInvoice, invoice.id) + assert stored is not None + with ( + patch("routstr.lightning.get_wallet", AsyncMock(return_value=wallet)), + patch( + "routstr.lightning._create_api_key_record", + AsyncMock(side_effect=RuntimeError("database unavailable")), + ), + ): + from routstr.lightning import check_invoice_payment + + await check_invoice_payment(stored, session) + + assert wallet.mint.await_count == 1 + async with AsyncSession(integration_engine, expire_on_commit=False) as verify: + stored = await verify.get(LightningInvoice, invoice.id) + assert stored is not None + assert stored.status == "pending" + + @pytest.mark.asyncio async def test_created_key_without_constraints_has_none_fields( integration_session: AsyncSession, diff --git a/tests/unit/test_db_pool_config.py b/tests/unit/test_db_pool_config.py new file mode 100644 index 00000000..ccfba575 --- /dev/null +++ b/tests/unit/test_db_pool_config.py @@ -0,0 +1,83 @@ +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from routstr.core import db +from routstr.core.db import _engine_options + + +def test_engine_options_use_bounded_fail_fast_pool(monkeypatch) -> None: # type: ignore[no-untyped-def] + monkeypatch.delenv("DATABASE_POOL_SIZE", raising=False) + monkeypatch.delenv("DATABASE_MAX_OVERFLOW", raising=False) + monkeypatch.delenv("DATABASE_POOL_TIMEOUT", raising=False) + monkeypatch.delenv("DATABASE_POOL_RECYCLE", raising=False) + monkeypatch.delenv("DATABASE_POOL_PRE_PING", raising=False) + + options = _engine_options("postgresql+asyncpg://db/routstr") + + assert options == { + "pool_size": 5, + "max_overflow": 0, + "pool_timeout": 5.0, + "pool_recycle": 1800, + "pool_pre_ping": True, + } + + +def test_engine_options_are_operator_configurable(monkeypatch) -> None: # type: ignore[no-untyped-def] + monkeypatch.setenv("DATABASE_POOL_SIZE", "12") + monkeypatch.setenv("DATABASE_MAX_OVERFLOW", "3") + monkeypatch.setenv("DATABASE_POOL_TIMEOUT", "2.5") + monkeypatch.setenv("DATABASE_POOL_RECYCLE", "900") + monkeypatch.setenv("DATABASE_POOL_PRE_PING", "false") + + assert _engine_options("sqlite+aiosqlite:///keys.db") == { + "pool_size": 12, + "max_overflow": 3, + "pool_timeout": 2.5, + "pool_recycle": 900, + "pool_pre_ping": False, + } + + +@pytest.mark.parametrize( + ("name", "value"), + [ + ("DATABASE_POOL_SIZE", "0"), + ("DATABASE_MAX_OVERFLOW", "-1"), + ("DATABASE_POOL_TIMEOUT", "0"), + ("DATABASE_POOL_RECYCLE", "-1"), + ("DATABASE_POOL_PRE_PING", "maybe"), + ], +) +def test_engine_options_reject_invalid_values( + monkeypatch: pytest.MonkeyPatch, + name: str, + value: str, +) -> None: + monkeypatch.setenv(name, value) + with pytest.raises(ValueError): + _engine_options("postgresql+asyncpg://db/routstr") + + +def test_pool_observer_warns_when_connection_is_held_too_long( + monkeypatch: pytest.MonkeyPatch, +) -> None: + record = SimpleNamespace(info={}) + monotonic = MagicMock(side_effect=[100.0, 112.5]) + monkeypatch.setattr(db.time, "monotonic", monotonic) + monkeypatch.setattr(db, "_POOL_HOLD_WARN_SECONDS", 10.0) + + with patch.object(db.logger, "warning") as warning: + db._record_pool_checkout(None, record, None) + db._record_pool_checkin(None, record) + + warning.assert_called_once() + assert warning.call_args.kwargs["extra"]["held_seconds"] == 12.5 + + +def test_memory_sqlite_keeps_dialect_static_pool(monkeypatch) -> None: # type: ignore[no-untyped-def] + monkeypatch.delenv("DATABASE_POOL_PRE_PING", raising=False) + + assert _engine_options("sqlite+aiosqlite://") == {"pool_pre_ping": True} diff --git a/tests/unit/test_fee_payout_crash_safety.py b/tests/unit/test_fee_payout_crash_safety.py index 0939baa4..a1b18edb 100644 --- a/tests/unit/test_fee_payout_crash_safety.py +++ b/tests/unit/test_fee_payout_crash_safety.py @@ -1,6 +1,4 @@ import asyncio -from collections.abc import AsyncIterator -from contextlib import asynccontextmanager from types import SimpleNamespace from unittest.mock import AsyncMock, Mock, patch @@ -13,9 +11,19 @@ from routstr import wallet from routstr.core import db -@asynccontextmanager -async def _session_context(session: Mock) -> AsyncIterator[Mock]: - yield session +class _SessionContext: + def __init__(self, session: Mock) -> None: + self.session = session + + async def __aenter__(self) -> Mock: + return self.session + + async def __aexit__(self, *args: object) -> None: + return None + + +def _session_context(session: Mock) -> _SessionContext: + return _SessionContext(session) @pytest.mark.asyncio @@ -77,7 +85,9 @@ async def test_fee_payout_checkpoints_before_sending() -> None: "routstr.wallet.asyncio.sleep", AsyncMock(side_effect=[None, asyncio.CancelledError()]), ), - patch("routstr.wallet.db.create_session", return_value=_session_context(session)), + patch( + "routstr.wallet.db.create_session", return_value=_session_context(session) + ), patch("routstr.wallet.db.get_routstr_fee", AsyncMock(return_value=fee)), patch("routstr.wallet.db.reset_routstr_fee", side_effect=checkpoint), patch("routstr.wallet.db.complete_routstr_fee_payout", side_effect=complete), @@ -107,7 +117,9 @@ async def test_fee_payout_does_not_retry_an_unresolved_checkpoint() -> None: "routstr.wallet.asyncio.sleep", AsyncMock(side_effect=[None, asyncio.CancelledError()]), ), - patch("routstr.wallet.db.create_session", return_value=_session_context(session)), + patch( + "routstr.wallet.db.create_session", return_value=_session_context(session) + ), patch("routstr.wallet.db.get_routstr_fee", AsyncMock(return_value=fee)), patch("routstr.wallet.db.reset_routstr_fee", AsyncMock()) as checkpoint, patch("routstr.wallet.get_wallet", AsyncMock()) as get_wallet, @@ -141,7 +153,9 @@ async def test_fee_payout_keeps_checkpoint_when_send_outcome_is_unknown() -> Non "routstr.wallet.asyncio.sleep", AsyncMock(side_effect=[None, asyncio.CancelledError()]), ), - patch("routstr.wallet.db.create_session", return_value=_session_context(session)), + patch( + "routstr.wallet.db.create_session", return_value=_session_context(session) + ), patch("routstr.wallet.db.get_routstr_fee", AsyncMock(return_value=fee)), patch("routstr.wallet.db.reset_routstr_fee", AsyncMock(return_value=True)), patch("routstr.wallet.db.complete_routstr_fee_payout", complete), diff --git a/tests/unit/test_fetch_all_balances.py b/tests/unit/test_fetch_all_balances.py index 433e84d4..1d69e31e 100644 --- a/tests/unit/test_fetch_all_balances.py +++ b/tests/unit/test_fetch_all_balances.py @@ -1,7 +1,13 @@ +import asyncio +from collections.abc import AsyncGenerator from contextlib import asynccontextmanager +from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch import pytest +from sqlalchemy.ext.asyncio import create_async_engine +from sqlmodel import SQLModel +from sqlmodel.ext.asyncio.session import AsyncSession from routstr.wallet import fetch_all_balances @@ -26,8 +32,10 @@ def _patches( # type: ignore[no-untyped-def] AsyncMock(side_effect=lambda proofs, wallet: proofs), ), patch( - "routstr.wallet.db.balances_for_mint_and_unit", - AsyncMock(return_value=user_balance_msats), + "routstr.wallet.db.balances_by_mint_and_unit", + AsyncMock( + return_value={("http://primary:3338", "sat"): user_balance_msats} + ), ), patch("routstr.wallet.db.create_session", _fake_session), ] @@ -38,8 +46,9 @@ async def test_fetch_all_balances_falls_back_to_primary_mint() -> None: """With empty cashu_mints, balances are still fetched for primary_mint.""" from routstr.core.settings import settings - with patch.object(settings, "cashu_mints", []), patch.object( - settings, "primary_mint", "http://primary:3338" + with ( + patch.object(settings, "cashu_mints", []), + patch.object(settings, "primary_mint", "http://primary:3338"), ): for p in _patches(proof_amount=1000): p.start() @@ -54,13 +63,159 @@ async def test_fetch_all_balances_falls_back_to_primary_mint() -> None: assert total_wallet == 1000 +@pytest.mark.asyncio +async def test_fetch_all_balances_closes_db_session_before_concurrent_mint_io() -> None: + """Slow mint checks must never run while the balance DB session is open.""" + from routstr.core.settings import settings + + session_open = False + mint_calls = 0 + + @asynccontextmanager + async def tracked_session(): # type: ignore[no-untyped-def] + nonlocal session_open + session_open = True + try: + yield MagicMock() + finally: + session_open = False + + async def slow_filter(proofs, wallet): # type: ignore[no-untyped-def] + nonlocal mint_calls + assert session_open is False + mint_calls += 1 + await asyncio.sleep(0) + return proofs + + with ( + patch.object(settings, "cashu_mints", ["http://one:3338", "http://two:3338"]), + patch.object(settings, "primary_mint", "http://one:3338"), + patch("routstr.wallet.db.create_session", tracked_session), + patch( + "routstr.wallet.db.balances_by_mint_and_unit", + AsyncMock(return_value={}), + create=True, + ), + patch("routstr.wallet.get_wallet", AsyncMock(return_value=MagicMock())), + patch( + "routstr.wallet.get_proofs_per_mint_and_unit", + MagicMock(return_value=[MagicMock(amount=1)]), + ), + patch( + "routstr.wallet.slow_filter_spend_proofs", + AsyncMock(side_effect=slow_filter), + ), + ): + details, *_ = await fetch_all_balances(units=["sat", "msat"]) + + assert mint_calls == 4 + assert all("error" not in detail for detail in details) + + +@pytest.mark.asyncio +async def test_fetch_all_balances_bounds_parallel_mint_checks() -> None: + """A slow mint fleet cannot create an unbounded external-I/O fan-out.""" + from routstr.core.settings import settings + + active = 0 + peak = 0 + + async def slow_filter(proofs, wallet): # type: ignore[no-untyped-def] + nonlocal active, peak + active += 1 + peak = max(peak, active) + await asyncio.sleep(0.01) + active -= 1 + return proofs + + with ( + patch.object( + settings, + "cashu_mints", + [f"http://mint-{index}:3338" for index in range(8)], + ), + patch.object(settings, "primary_mint", ""), + patch.object(settings, "mint_operation_concurrency", 2), + patch("routstr.wallet.db.create_session", _fake_session), + patch( + "routstr.wallet.db.balances_by_mint_and_unit", + AsyncMock(return_value={}), + ), + patch("routstr.wallet.get_wallet", AsyncMock(return_value=MagicMock())), + patch( + "routstr.wallet.get_proofs_per_mint_and_unit", + MagicMock(return_value=[]), + ), + patch( + "routstr.wallet.slow_filter_spend_proofs", + AsyncMock(side_effect=slow_filter), + ), + ): + details, *_ = await fetch_all_balances(units=["sat"]) + + assert len(details) == 8 + assert peak == 2 + + +@pytest.mark.asyncio +async def test_slow_mints_do_not_exhaust_a_single_connection_pool( + tmp_path: Path, +) -> None: + """Concurrent slow balance refreshes release the sole DB connection promptly.""" + from routstr.core.settings import settings + + engine = create_async_engine( + f"sqlite+aiosqlite:///{tmp_path / 'pool-pressure.db'}", + pool_size=1, + max_overflow=0, + pool_timeout=0.2, + ) + async with engine.begin() as connection: + await connection.run_sync(SQLModel.metadata.create_all) + + @asynccontextmanager + async def single_pool_session() -> AsyncGenerator[AsyncSession, None]: + async with AsyncSession(engine, expire_on_commit=False) as session: + yield session + + async def slow_filter(proofs, wallet): # type: ignore[no-untyped-def] + await asyncio.sleep(0.3) + return proofs + + try: + with ( + patch.object(settings, "cashu_mints", ["http://slow:3338"]), + patch.object(settings, "primary_mint", "http://slow:3338"), + patch.object(settings, "mint_operation_concurrency", 1), + patch("routstr.wallet.db.create_session", single_pool_session), + patch("routstr.wallet.get_wallet", AsyncMock(return_value=MagicMock())), + patch( + "routstr.wallet.get_proofs_per_mint_and_unit", + MagicMock(return_value=[]), + ), + patch( + "routstr.wallet.slow_filter_spend_proofs", + AsyncMock(side_effect=slow_filter), + ), + ): + results = await asyncio.gather( + *(fetch_all_balances(units=["sat"]) for _ in range(6)) + ) + + assert all("error" not in result[0][0] for result in results) + assert engine.pool.checkedout() == 0 # type: ignore[attr-defined] + finally: + await engine.dispose() + + @pytest.mark.asyncio async def test_fetch_all_balances_reports_liability_when_wallet_is_empty() -> None: """An empty wallet must not hide outstanding user liabilities.""" from routstr.core.settings import settings - with patch.object(settings, "cashu_mints", []), patch.object( - settings, "primary_mint", "http://primary:3338" + with ( + patch.object(settings, "cashu_mints", []), + patch.object(settings, "primary_mint", "http://primary:3338"), ): for p in _patches(proof_amount=0, user_balance_msats=5000): p.start() @@ -84,9 +239,10 @@ async def test_fetch_all_balances_no_duplicate_primary_mint() -> None: """primary_mint already in cashu_mints is not inspected twice.""" from routstr.core.settings import settings - with patch.object( - settings, "cashu_mints", ["http://primary:3338"] - ), patch.object(settings, "primary_mint", "http://primary:3338"): + with ( + patch.object(settings, "cashu_mints", ["http://primary:3338"]), + patch.object(settings, "primary_mint", "http://primary:3338"), + ): for p in _patches(proof_amount=1000): p.start() try: diff --git a/tests/unit/test_periodic_payout.py b/tests/unit/test_periodic_payout.py index 54ebadb4..77d0a7cf 100644 --- a/tests/unit/test_periodic_payout.py +++ b/tests/unit/test_periodic_payout.py @@ -59,23 +59,28 @@ async def test_periodic_payout_includes_primary_mint_not_in_cashu_mints() -> Non get_wallet = AsyncMock(return_value=MagicMock()) raw_send = AsyncMock(return_value=1000) - with patch.object(settings, "cashu_mints", []), patch.object( - settings, "primary_mint", "http://primary:3338" - ), patch.object(settings, "receive_ln_address", "owner@ln.tld"), patch.object( - settings, "payout_interval_seconds", _INTERVAL - ), patch.object(settings, "min_payout_sat", 10), patch( - "routstr.wallet.asyncio.sleep", _one_cycle_sleep() - ), patch("routstr.wallet.db.create_session", _fake_session), patch( - "routstr.wallet.get_wallet", get_wallet - ), patch( - "routstr.wallet.get_proofs_per_mint_and_unit", - MagicMock(return_value=[MagicMock(amount=100_000)]), - ), patch( - "routstr.wallet.slow_filter_spend_proofs", - AsyncMock(side_effect=lambda proofs, wallet: proofs), - ), patch( - "routstr.wallet.db.balances_for_mint_and_unit", AsyncMock(return_value=0) - ), patch("routstr.wallet.raw_send_to_lnurl", raw_send): + with ( + patch.object(settings, "cashu_mints", []), + patch.object(settings, "primary_mint", "http://primary:3338"), + patch.object(settings, "receive_ln_address", "owner@ln.tld"), + patch.object(settings, "payout_interval_seconds", _INTERVAL), + patch.object(settings, "min_payout_sat", 10), + patch("routstr.wallet.asyncio.sleep", _one_cycle_sleep()), + patch("routstr.wallet.db.create_session", _fake_session), + patch("routstr.wallet.get_wallet", get_wallet), + patch( + "routstr.wallet.get_proofs_per_mint_and_unit", + MagicMock(return_value=[MagicMock(amount=100_000)]), + ), + patch( + "routstr.wallet.slow_filter_spend_proofs", + AsyncMock(side_effect=lambda proofs, wallet: proofs), + ), + patch( + "routstr.wallet.db.balances_by_mint_and_unit", AsyncMock(return_value={}) + ), + patch("routstr.wallet.raw_send_to_lnurl", raw_send), + ): with pytest.raises(_LoopBreak): await periodic_payout() @@ -84,6 +89,58 @@ async def test_periodic_payout_includes_primary_mint_not_in_cashu_mints() -> Non assert raw_send.await_count >= 1 +@pytest.mark.asyncio +async def test_periodic_payout_releases_session_before_slow_mint_send() -> None: + """The DB connection is returned before the external LNURL call starts.""" + from routstr.core.settings import settings + + session_open = False + sends_completed = 0 + + @asynccontextmanager + async def tracked_session(): # type: ignore[no-untyped-def] + nonlocal session_open + session_open = True + try: + yield MagicMock() + finally: + session_open = False + + async def raw_send(*args: object, **kwargs: object) -> int: + nonlocal sends_completed + assert session_open is False + sends_completed += 1 + return 1000 + + with ( + patch.object(settings, "cashu_mints", []), + patch.object(settings, "primary_mint", "http://primary:3338"), + patch.object(settings, "receive_ln_address", "owner@ln.tld"), + patch.object(settings, "payout_interval_seconds", _INTERVAL), + patch.object(settings, "min_payout_sat", 10), + patch("routstr.wallet.asyncio.sleep", _one_cycle_sleep()), + patch("routstr.wallet.db.create_session", tracked_session), + patch("routstr.wallet.get_wallet", AsyncMock(return_value=MagicMock())), + patch( + "routstr.wallet.get_proofs_per_mint_and_unit", + MagicMock(return_value=[MagicMock(amount=100_000)]), + ), + patch( + "routstr.wallet.slow_filter_spend_proofs", + AsyncMock(side_effect=lambda proofs, wallet: proofs), + ), + patch( + "routstr.wallet.db.balances_by_mint_and_unit", + AsyncMock(return_value={}), + ), + patch("routstr.wallet.raw_send_to_lnurl", AsyncMock(side_effect=raw_send)), + ): + with pytest.raises(_LoopBreak): + await periodic_payout() + + assert sends_completed == 2 + + @pytest.mark.asyncio async def test_periodic_payout_isolates_failing_mint() -> None: """A failing mint does not prevent payout for the other mints.""" @@ -97,23 +154,28 @@ async def test_periodic_payout_isolates_failing_mint() -> None: get_wallet = AsyncMock(side_effect=_get_wallet) raw_send = AsyncMock(return_value=1000) - with patch.object( - settings, "cashu_mints", ["http://bad:3338", "http://good:3338"] - ), patch.object(settings, "primary_mint", "http://good:3338"), patch.object( - settings, "receive_ln_address", "owner@ln.tld" - ), patch.object(settings, "payout_interval_seconds", _INTERVAL), patch.object( - settings, "min_payout_sat", 10 - ), patch("routstr.wallet.asyncio.sleep", _one_cycle_sleep()), patch( - "routstr.wallet.db.create_session", _fake_session - ), patch("routstr.wallet.get_wallet", get_wallet), patch( - "routstr.wallet.get_proofs_per_mint_and_unit", - MagicMock(return_value=[MagicMock(amount=100_000)]), - ), patch( - "routstr.wallet.slow_filter_spend_proofs", - AsyncMock(side_effect=lambda proofs, wallet: proofs), - ), patch( - "routstr.wallet.db.balances_for_mint_and_unit", AsyncMock(return_value=0) - ), patch("routstr.wallet.raw_send_to_lnurl", raw_send): + with ( + patch.object(settings, "cashu_mints", ["http://bad:3338", "http://good:3338"]), + patch.object(settings, "primary_mint", "http://good:3338"), + patch.object(settings, "receive_ln_address", "owner@ln.tld"), + patch.object(settings, "payout_interval_seconds", _INTERVAL), + patch.object(settings, "min_payout_sat", 10), + patch("routstr.wallet.asyncio.sleep", _one_cycle_sleep()), + patch("routstr.wallet.db.create_session", _fake_session), + patch("routstr.wallet.get_wallet", get_wallet), + patch( + "routstr.wallet.get_proofs_per_mint_and_unit", + MagicMock(return_value=[MagicMock(amount=100_000)]), + ), + patch( + "routstr.wallet.slow_filter_spend_proofs", + AsyncMock(side_effect=lambda proofs, wallet: proofs), + ), + patch( + "routstr.wallet.db.balances_by_mint_and_unit", AsyncMock(return_value={}) + ), + patch("routstr.wallet.raw_send_to_lnurl", raw_send), + ): with pytest.raises(_LoopBreak): await periodic_payout() @@ -134,15 +196,15 @@ async def test_periodic_payout_handles_session_creation_failure() -> None: create_session = MagicMock(side_effect=RuntimeError("db unavailable")) logger = MagicMock() - with patch.object(settings, "cashu_mints", ["http://mint:3338"]), patch.object( - settings, "primary_mint", "http://mint:3338" - ), patch.object(settings, "receive_ln_address", "owner@ln.tld"), patch.object( - settings, "payout_interval_seconds", _INTERVAL - ), patch( - "routstr.wallet.asyncio.sleep", _one_cycle_sleep() - ), patch( - "routstr.wallet.db.create_session", create_session - ), patch("routstr.wallet.logger", logger): + with ( + patch.object(settings, "cashu_mints", ["http://mint:3338"]), + patch.object(settings, "primary_mint", "http://mint:3338"), + patch.object(settings, "receive_ln_address", "owner@ln.tld"), + patch.object(settings, "payout_interval_seconds", _INTERVAL), + patch("routstr.wallet.asyncio.sleep", _one_cycle_sleep()), + patch("routstr.wallet.db.create_session", create_session), + patch("routstr.wallet.logger", logger), + ): with pytest.raises(_LoopBreak): await periodic_payout() diff --git a/tests/unit/test_proxy_session_lifecycle.py b/tests/unit/test_proxy_session_lifecycle.py new file mode 100644 index 00000000..5d0416d5 --- /dev/null +++ b/tests/unit/test_proxy_session_lifecycle.py @@ -0,0 +1,43 @@ +from collections.abc import AsyncIterator +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi.responses import StreamingResponse + +from routstr import proxy as proxy_module + + +@pytest.mark.asyncio +async def test_proxy_closes_request_session_before_returning_response() -> None: + """Route completion must release DB resources before response delivery.""" + request = MagicMock() + request.method = "GET" + request.headers = {"accept": "application/json"} + request.url.path = "/not-an-api-route" + request.state.request_id = "test-request" + session = AsyncMock() + + response = await proxy_module.proxy(request, "not-an-api-route", session=session) + + assert response.status_code == 404 + session.close.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_proxy_session_is_closed_before_first_stream_chunk() -> None: + request = MagicMock() + session = AsyncMock() + + async def stream() -> AsyncIterator[bytes]: + session.close.assert_awaited_once() + yield b"chunk" + + upstream_response = StreamingResponse(stream()) + with patch("routstr.proxy._proxy", AsyncMock(return_value=upstream_response)): + response = await proxy_module.proxy( + request, "v1/chat/completions", session=session + ) + + assert isinstance(response, StreamingResponse) + chunks = [chunk async for chunk in response.body_iterator] + assert chunks == [b"chunk"] diff --git a/tests/unit/test_refund_sweep.py b/tests/unit/test_refund_sweep.py index e82c2111..64b36efc 100644 --- a/tests/unit/test_refund_sweep.py +++ b/tests/unit/test_refund_sweep.py @@ -1,4 +1,5 @@ from collections.abc import AsyncIterator +from contextlib import asynccontextmanager from pathlib import Path from unittest.mock import AsyncMock, patch @@ -39,6 +40,43 @@ async def _load( return {row.token: row for row in result.all()} +@pytest.mark.asyncio +async def test_refund_sweep_releases_db_session_during_token_redemption( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _insert( + session_factory, + CashuTransaction( + token="eligible", amount=1, unit="sat", type="out", created_at=800 + ), + ) + session_open = False + + @asynccontextmanager + async def tracked_session() -> AsyncIterator[AsyncSession]: + nonlocal session_open + async with session_factory() as session: + session_open = True + try: + yield session + finally: + session_open = False + + async def receive_token(token: str) -> None: + assert token == "eligible" + assert session_open is False + + with ( + patch("routstr.wallet.db.create_session", tracked_session), + patch("routstr.wallet.settings.refund_sweep_ttl_seconds", 100), + patch("routstr.wallet.time.time", return_value=1000), + patch("routstr.wallet.recieve_token", AsyncMock(side_effect=receive_token)), + ): + await refund_sweep_once() + + assert (await _load(session_factory))["eligible"].swept is True + + @pytest.mark.asyncio async def test_refund_sweep_only_processes_expired_eligible_outgoing_tokens( session_factory: async_sessionmaker[AsyncSession], From 1b09639265de7b9957f76e7bd4553afb41152505 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Fri, 24 Jul 2026 21:44:11 +0200 Subject: [PATCH 2/3] docs: document DB pool and mint concurrency env vars in .env.example --- .env.example | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/.env.example b/.env.example index 8ff04b35..8c3e2657 100644 --- a/.env.example +++ b/.env.example @@ -22,6 +22,13 @@ ROUTSTR_SECRET_KEY= # Database # DATABASE_URL=sqlite+aiosqlite:///keys.db +# Keep total pool capacity across all workers below the database connection limit. +# DATABASE_POOL_SIZE=5 +# DATABASE_MAX_OVERFLOW=0 +# DATABASE_POOL_TIMEOUT=5 +# DATABASE_POOL_RECYCLE=1800 +# DATABASE_POOL_PRE_PING=true +# DATABASE_POOL_HOLD_WARN_SECONDS=10 # Node Information # NAME=My Routstr Node @@ -31,6 +38,7 @@ ROUTSTR_SECRET_KEY= # RELAYS="wss://relay.damus.io,wss://relay.nostr.band,wss://eden.nostr.land,wss://relay.routstr.com" # ENABLE_ANALYTICS_SHARING=true # CASHU_MINTS="https://mint.minibits.cash/Bitcoin,https://mint.cubabitcoin.org,https://ecashmint.otrta.me" +# MINT_OPERATION_CONCURRENCY=4 # RECEIVE_LN_ADDRESS= # Custom Pricing Configuration From 2410a4a6ce0729a6e9f0bb471cb17a51adb4a7cf Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Fri, 24 Jul 2026 22:14:57 +0200 Subject: [PATCH 3/3] fix: address review findings on payout liability staleness, sweep races, and cancellation safety - periodic_payout: fetch liability per mint/unit right before computing available balance, so a concurrent top-up can only shrink the payout - refund sweep: atomically claim each refund before redeeming; release the claim on failure so retries still happen and concurrent sweeps cannot misreport a sweep as client-collected - check_invoice_payment: catch BaseException so task cancellation after a successful mint still emits the reconciliation alert - tests: DB-guard race test where both mints succeed (exactly one credit); pool_size=1 test proving the fee payout releases its connection during the external send --- routstr/lightning.py | 11 +++- routstr/wallet.py | 50 ++++++++++++----- .../test_lightning_invoice_constraints.py | 55 +++++++++++++++++++ tests/unit/test_fee_payout_crash_safety.py | 49 +++++++++++++++++ tests/unit/test_periodic_payout.py | 29 +++++++--- 5 files changed, 168 insertions(+), 26 deletions(-) diff --git a/routstr/lightning.py b/routstr/lightning.py index 30aa7918..5bba7593 100644 --- a/routstr/lightning.py +++ b/routstr/lightning.py @@ -281,13 +281,20 @@ async def check_invoice_payment( else None, }, ) - except Exception as e: - await session.rollback() + except BaseException as e: + # BaseException so task cancellation (e.g. client disconnect) after a + # successful mint still triggers the reconciliation alert. + try: + await asyncio.shield(session.rollback()) + except Exception: + logger.warning("Rollback failed during invoice check cleanup") if minted: logger.critical( "Invoice mint succeeded but DB finalization failed; reconciliation required", extra={"invoice_id": invoice_id, "purpose": invoice_purpose}, ) + if not isinstance(e, Exception): + raise logger.error(f"Failed to check invoice payment: {e}") diff --git a/routstr/wallet.py b/routstr/wallet.py index 24b11fe9..d61bbd4c 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -931,9 +931,6 @@ async def periodic_payout() -> None: if settings.primary_mint and settings.primary_mint not in mint_urls: mint_urls.append(settings.primary_mint) - async with db.create_session() as session: - user_balances = await db.balances_by_mint_and_unit(session) - for mint_url in mint_urls: for unit in ["sat", "msat"]: # Isolate failures per mint/unit so one slow or failing @@ -945,7 +942,14 @@ async def periodic_payout() -> None: ) proofs = await slow_filter_spend_proofs(proofs, wallet) await asyncio.sleep(5) - user_balance = user_balances.get((mint_url, unit), 0) + # Fetch the liability AFTER the proofs snapshot and the + # settle delay: a concurrent top-up then only inflates + # the liability, shrinking the payout — never sending + # customer-backed funds as profit. + async with db.create_session() as session: + user_balance = await db.balances_for_mint_and_unit( + session, mint_url, unit + ) if unit == "sat": user_balance = user_balance // 1000 proofs_balance = sum(proof.amount for proof in proofs) @@ -1001,18 +1005,24 @@ async def _refund_sweep_once(cutoff: int) -> None: refunds = results.all() for refund in refunds: + # Claim the refund atomically before redeeming so a concurrent sweep + # cannot redeem the same token and misreport it as client-collected. + async with db.create_session() as session: + claim = await session.exec( # type: ignore[call-overload] + update(db.CashuTransaction) + .where( + col(db.CashuTransaction.id) == refund.id, + col(db.CashuTransaction.swept) == False, # noqa: E712 + col(db.CashuTransaction.collected) == False, # noqa: E712 + ) + .values(swept=True) + ) + await session.commit() + if claim.rowcount != 1: + continue + try: await recieve_token(refund.token) - async with db.create_session() as session: - await session.exec( # type: ignore[call-overload] - update(db.CashuTransaction) - .where( - col(db.CashuTransaction.id) == refund.id, - col(db.CashuTransaction.swept) == False, # noqa: E712 - ) - .values(swept=True) - ) - await session.commit() logger.info( "Swept uncollected refund", extra={ @@ -1024,11 +1034,13 @@ async def _refund_sweep_once(cutoff: int) -> None: except Exception as e: error_msg = str(e).lower() if "already spent" in error_msg: + # We held the claim, so nobody else swept it: the client + # really collected the token. async with db.create_session() as session: await session.exec( # type: ignore[call-overload] update(db.CashuTransaction) .where(col(db.CashuTransaction.id) == refund.id) - .values(collected=True) + .values(collected=True, swept=False) ) await session.commit() logger.info( @@ -1036,6 +1048,14 @@ async def _refund_sweep_once(cutoff: int) -> None: extra={"id": refund.id}, ) else: + # Release the claim so the next sweep retries this refund. + async with db.create_session() as session: + await session.exec( # type: ignore[call-overload] + update(db.CashuTransaction) + .where(col(db.CashuTransaction.id) == refund.id) + .values(swept=False) + ) + await session.commit() logger.warning( "Failed to sweep refund", extra={ diff --git a/tests/integration/test_lightning_invoice_constraints.py b/tests/integration/test_lightning_invoice_constraints.py index 8c8e15d9..a7fa1384 100644 --- a/tests/integration/test_lightning_invoice_constraints.py +++ b/tests/integration/test_lightning_invoice_constraints.py @@ -299,3 +299,58 @@ async def test_created_key_without_constraints_has_none_fields( assert stored_key.balance_limit is None assert stored_key.balance_limit_reset is None assert stored_key.validity_date is None + + +@pytest.mark.asyncio +async def test_db_guard_credits_once_when_both_mints_succeed( + integration_engine: AsyncEngine, +) -> None: + """Even if the mint fails to enforce single-use quotes and both racers + mint successfully, the conditional status update must credit exactly once.""" + key = ApiKey(hashed_key="race-key", balance=1_000) + invoice = _make_invoice( + id="inv_db_guard", + status="pending", + paid_at=None, + purpose="topup", + api_key_hash="race-key", + ) + async with AsyncSession(integration_engine, expire_on_commit=False) as setup: + setup.add(key) + setup.add(invoice) + await setup.commit() + + wallet = MagicMock() + wallet.get_mint_quote = AsyncMock(return_value=MagicMock(paid=True)) + + async def always_succeeding_mint(*args: object, **kwargs: object) -> list[object]: + await asyncio.sleep(0.05) + return [] + + wallet.mint = AsyncMock(side_effect=always_succeeding_mint) + + async with ( + AsyncSession(integration_engine, expire_on_commit=False) as first, + AsyncSession(integration_engine, expire_on_commit=False) as second, + ): + first_invoice = await first.get(LightningInvoice, invoice.id) + second_invoice = await second.get(LightningInvoice, invoice.id) + assert first_invoice is not None + assert second_invoice is not None + + with patch("routstr.lightning.get_wallet", AsyncMock(return_value=wallet)): + from routstr.lightning import check_invoice_payment + + await asyncio.gather( + check_invoice_payment(first_invoice, first), + check_invoice_payment(second_invoice, second), + ) + + assert wallet.mint.await_count == 2 + async with AsyncSession(integration_engine, expire_on_commit=False) as verify: + stored_invoice = await verify.get(LightningInvoice, invoice.id) + assert stored_invoice is not None + assert stored_invoice.status == "paid" + stored_key = await verify.get(ApiKey, "race-key") + assert stored_key is not None + assert stored_key.balance == 1_000 + invoice.amount_sats * 1000 diff --git a/tests/unit/test_fee_payout_crash_safety.py b/tests/unit/test_fee_payout_crash_safety.py index a1b18edb..19b95ec3 100644 --- a/tests/unit/test_fee_payout_crash_safety.py +++ b/tests/unit/test_fee_payout_crash_safety.py @@ -1,4 +1,6 @@ import asyncio +from collections.abc import AsyncGenerator +from contextlib import asynccontextmanager from types import SimpleNamespace from unittest.mock import AsyncMock, Mock, patch @@ -172,3 +174,50 @@ async def test_fee_payout_keeps_checkpoint_when_send_outcome_is_unknown() -> Non complete.assert_not_awaited() critical.assert_called_once() + + +@pytest.mark.asyncio +async def test_fee_payout_releases_db_connection_during_send(tmp_path: object) -> None: + """With pool_size=1, the payout must not hold a connection while the + external LNURL send is in flight, or the completion step would starve.""" + engine = create_async_engine( + f"sqlite+aiosqlite:///{tmp_path}/payout.db", pool_size=1, max_overflow=0 + ) + async with engine.begin() as connection: + await connection.run_sync(SQLModel.metadata.create_all) + async with AsyncSession(engine) as session: + session.add(db.RoutstrFee(id=1, accumulated_msats=5_000_000)) + await session.commit() + + @asynccontextmanager + async def create_session() -> AsyncGenerator[AsyncSession, None]: + async with AsyncSession(engine, expire_on_commit=False) as session: + yield session + + async def send(*_args: object, **_kwargs: object) -> int: + assert engine.pool.checkedout() == 0 # type: ignore[attr-defined] + return 5 + + try: + with ( + patch("routstr.auth.ROUTSTR_FEE_DEFAULT_PAYOUT", 1), + patch("routstr.auth.ROUTSTR_FEE_PAYOUT_INTERVAL_SECONDS", 1), + patch("routstr.auth.ROUTSTR_LN_ADDRESS", "fees@example.com"), + patch( + "routstr.wallet.asyncio.sleep", + AsyncMock(side_effect=[None, asyncio.CancelledError()]), + ), + patch("routstr.wallet.db.create_session", create_session), + patch("routstr.wallet.get_wallet", AsyncMock(return_value=Mock())), + patch("routstr.wallet.get_proofs_per_mint_and_unit", return_value=[]), + patch("routstr.wallet.raw_send_to_lnurl", side_effect=send), + ): + with pytest.raises(asyncio.CancelledError): + await wallet.periodic_routstr_fee_payout() + + async with AsyncSession(engine) as session: + fee = await db.get_routstr_fee(session) + assert fee.payout_in_progress_msats == 0 + assert fee.total_paid_msats == 5_000_000 + finally: + await engine.dispose() diff --git a/tests/unit/test_periodic_payout.py b/tests/unit/test_periodic_payout.py index 77d0a7cf..c1768dc7 100644 --- a/tests/unit/test_periodic_payout.py +++ b/tests/unit/test_periodic_payout.py @@ -77,7 +77,7 @@ async def test_periodic_payout_includes_primary_mint_not_in_cashu_mints() -> Non AsyncMock(side_effect=lambda proofs, wallet: proofs), ), patch( - "routstr.wallet.db.balances_by_mint_and_unit", AsyncMock(return_value={}) + "routstr.wallet.db.balances_for_mint_and_unit", AsyncMock(return_value=0) ), patch("routstr.wallet.raw_send_to_lnurl", raw_send), ): @@ -130,8 +130,8 @@ async def test_periodic_payout_releases_session_before_slow_mint_send() -> None: AsyncMock(side_effect=lambda proofs, wallet: proofs), ), patch( - "routstr.wallet.db.balances_by_mint_and_unit", - AsyncMock(return_value={}), + "routstr.wallet.db.balances_for_mint_and_unit", + AsyncMock(return_value=0), ), patch("routstr.wallet.raw_send_to_lnurl", AsyncMock(side_effect=raw_send)), ): @@ -172,7 +172,7 @@ async def test_periodic_payout_isolates_failing_mint() -> None: AsyncMock(side_effect=lambda proofs, wallet: proofs), ), patch( - "routstr.wallet.db.balances_by_mint_and_unit", AsyncMock(return_value={}) + "routstr.wallet.db.balances_for_mint_and_unit", AsyncMock(return_value=0) ), patch("routstr.wallet.raw_send_to_lnurl", raw_send), ): @@ -190,7 +190,7 @@ async def test_periodic_payout_isolates_failing_mint() -> None: @pytest.mark.asyncio async def test_periodic_payout_handles_session_creation_failure() -> None: - """A db.create_session failure is logged and the payout loop continues.""" + """A db.create_session failure is logged per mint/unit and the loop continues.""" from routstr.core.settings import settings create_session = MagicMock(side_effect=RuntimeError("db unavailable")) @@ -203,14 +203,25 @@ async def test_periodic_payout_handles_session_creation_failure() -> None: patch.object(settings, "payout_interval_seconds", _INTERVAL), patch("routstr.wallet.asyncio.sleep", _one_cycle_sleep()), patch("routstr.wallet.db.create_session", create_session), + patch("routstr.wallet.get_wallet", AsyncMock(return_value=MagicMock())), + patch( + "routstr.wallet.get_proofs_per_mint_and_unit", + MagicMock(return_value=[MagicMock(amount=100_000)]), + ), + patch( + "routstr.wallet.slow_filter_spend_proofs", + AsyncMock(side_effect=lambda proofs, wallet: proofs), + ), patch("routstr.wallet.logger", logger), ): with pytest.raises(_LoopBreak): await periodic_payout() - create_session.assert_called_once() - logger.error.assert_called_once() + # The liability session is opened per mint/unit (sat + msat), and each + # failure is isolated to its own iteration rather than aborting the cycle. + assert create_session.call_count == 2 + assert logger.error.call_count == 2 message = logger.error.call_args.args[0] extra = logger.error.call_args.kwargs["extra"] - assert message == "Error in periodic payout cycle: RuntimeError" - assert extra == {"error": "db unavailable"} + assert message == "Error sending payout: RuntimeError" + assert extra["error"] == "db unavailable"