laantungir/routstr-core
1
0
Fork 0
mirror of https://github.com/Routstr/routstr-core.git synced 2026-07-30 15:26:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix-concurent-issue

Browse Source
This commit is contained in:
9qeklajc
2026-07-24 21:29:21 +02:00
parent b94d95fc2b
commit 7394b10e75
12 changed files with 952 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files
routstr/core/db.py
+102 -2
View File
@@ -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:
routstr/core/settings.py
+3
View File
@@ -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
routstr/lightning.py
+93 -50
View File
@@ -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
routstr/proxy.py
+15
View File
@@ -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
routstr/wallet.py
+140 -134
View File
@@ -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__}",
tests/integration/test_lightning_invoice_constraints.py
+143 -1
View File
@@ -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,
tests/unit/test_db_pool_config.py
+83
View File
@@ -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}
tests/unit/test_fee_payout_crash_safety.py
+22 -8
View File
@@ -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),
tests/unit/test_fetch_all_balances.py
+165 -9
View File
@@ -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:
tests/unit/test_periodic_payout.py
+105 -43
View File
@@ -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()
tests/unit/test_proxy_session_lifecycle.py
+43
View File
@@ -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"]
tests/unit/test_refund_sweep.py
+38
View File
@@ -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],