merge: combine PR #632 with broader pool-exhaustion fixes

routstr/core/db.py

@@ -15,6 +15,7 @@ from alembic.util.exc import CommandError
 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 import AsyncEngine
 from sqlalchemy.ext.asyncio.engine import create_async_engine
 from sqlalchemy.orm import aliased
 from sqlmodel import Field, Relationship, SQLModel, col, func, select, update
@@ -27,58 +28,43 @@ logger = get_logger(__name__)
 DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite+aiosqlite:///keys.db")
 
 
-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 create_db_engine(database_url: str = DATABASE_URL) -> AsyncEngine:
+    """Build an async engine from validated, environment-only pool settings."""
+    from .settings import settings
 
-
-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
+    options: dict[str, int | float | bool] = {
+        "pool_pre_ping": settings.database_pool_pre_ping,
+    }
+    if not is_memory_sqlite:
+        options.update(
+            pool_size=settings.database_pool_size,
+            max_overflow=settings.database_max_overflow,
+            pool_timeout=settings.database_pool_timeout,
+            pool_recycle=settings.database_pool_recycle,
+        )
 
-    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,
+    logger.info(
+        "Database pool configured",
+        extra={
+            "database_url_backend": url.get_backend_name(),
+            "in_memory_sqlite": is_memory_sqlite,
+            **options,
+        },
     )
-    return options
+    return create_async_engine(database_url, echo=False, **options)
 
 
-engine = create_async_engine(DATABASE_URL, echo=False, **_engine_options(DATABASE_URL))
+engine = create_db_engine()
 
-_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")
+from .settings import settings as _runtime_settings  # noqa: E402
+
+_POOL_HOLD_WARN_SECONDS = _runtime_settings.database_pool_hold_warn_seconds
 
 
 @event.listens_for(engine.sync_engine, "checkout")
@@ -441,6 +427,10 @@ class CashuTransaction(SQLModel, table=True):  # type: ignore
     )
     collected: bool = Field(default=False)
     swept: bool = Field(default=False)
+    sweep_started_at: int | None = Field(
+        default=None,
+        description="Unix timestamp for a recoverable refund-sweep claim",
+    )
     source: str = Field(
         default="x-cashu",
         description="Payment source: x-cashu or apikey",
@@ -793,20 +783,12 @@ async def complete_routstr_fee_payout(
     return result.rowcount == 1
 
 
-async def balances_for_mint_and_unit(
-    db_session: AsyncSession, mint_url: str, unit: str
-) -> int:
-    query = select(func.sum(ApiKey.balance)).where(
-        ApiKey.refund_mint_url == mint_url, ApiKey.refund_currency == unit
-    )
-    result = await db_session.exec(query)
-    return result.one() or 0
-
-
 async def balances_by_mint_and_unit(
-    db_session: AsyncSession,
+    db_session: AsyncSession, mint_urls: list[str], units: list[str]
 ) -> dict[tuple[str, str], int]:
-    """Return all user liabilities in one query, grouped by mint and unit."""
+    """Return requested user liabilities grouped by mint and unit."""
+    if not mint_urls or not units:
+        return {}
     query = (
         select(
             col(ApiKey.refund_mint_url),
@@ -814,8 +796,8 @@ async def balances_by_mint_and_unit(
             func.sum(ApiKey.balance),
         )
         .where(
-            col(ApiKey.refund_mint_url).is_not(None),
-            col(ApiKey.refund_currency).is_not(None),
+            col(ApiKey.refund_mint_url).in_(mint_urls),
+            col(ApiKey.refund_currency).in_(units),
         )
         .group_by(col(ApiKey.refund_mint_url), col(ApiKey.refund_currency))
     )

routstr/core/settings.py

@@ -102,6 +102,21 @@ class Settings(BaseSettings):
     enable_models_refresh: bool = Field(default=True, env="ENABLE_MODELS_REFRESH")
     refund_cache_ttl_seconds: int = Field(default=3600, env="REFUND_CACHE_TTL_SECONDS")
     refund_sweep_ttl_seconds: int = Field(default=604800, env="REFUND_SWEEP_TTL_SECONDS")
+    refund_sweep_claim_timeout_seconds: int = Field(
+        default=900, gt=0, env="REFUND_SWEEP_CLAIM_TIMEOUT_SECONDS"
+    )
+
+    # Database connection-pool controls (advanced). Keep the default pool
+    # bounded and fail quickly: increasing it can hide leaked connections and
+    # makes SQLite lock contention worse. These fields are env-only below.
+    database_pool_size: int = Field(default=5, ge=1, env="DATABASE_POOL_SIZE")
+    database_max_overflow: int = Field(default=0, ge=0, env="DATABASE_MAX_OVERFLOW")
+    database_pool_timeout: float = Field(default=5.0, gt=0, env="DATABASE_POOL_TIMEOUT")
+    database_pool_recycle: int = Field(default=1800, ge=0, env="DATABASE_POOL_RECYCLE")
+    database_pool_pre_ping: bool = Field(default=True, env="DATABASE_POOL_PRE_PING")
+    database_pool_hold_warn_seconds: float = Field(
+        default=10.0, gt=0, env="DATABASE_POOL_HOLD_WARN_SECONDS"
+    )
 
     # Logging
     log_level: str = Field(default="INFO", env="LOG_LEVEL")
@@ -147,10 +162,32 @@ def _normalize_settings_data(data: dict[str, Any]) -> dict[str, Any]:
 # ``routstr.core.vault``.
 SECRET_FIELDS = frozenset({"admin_password", "nsec"})
 
+# Infrastructure the node needs *before* it can open a DB session — so it can
+# never be configured from the DB (chicken-and-egg) and stays env-only. Unlike
+# secrets (owned by bootstrap), these are excluded so the DB settings blob can
+# neither store nor shadow them; env is always authoritative.
+ENV_ONLY_FIELDS = frozenset(
+    {
+        "database_pool_size",
+        "database_max_overflow",
+        "database_pool_timeout",
+        "database_pool_recycle",
+        "database_pool_pre_ping",
+        "database_pool_hold_warn_seconds",
+    }
+)
+
+_NON_PERSISTED_FIELDS = SECRET_FIELDS | ENV_ONLY_FIELDS
+
 
 def _strip_secret_fields(data: dict[str, Any]) -> dict[str, Any]:
-    """Return a copy of ``data`` without any secret fields (for persistence)."""
-    return {k: v for k, v in data.items() if k not in SECRET_FIELDS}
+    """Return a copy of ``data`` without secret or env-only fields.
+
+    Both are kept out of the persisted settings blob: secrets for confidentiality,
+    env-only fields (e.g. DB pool sizing) because they must never be sourced from
+    the database.
+    """
+    return {k: v for k, v in data.items() if k not in _NON_PERSISTED_FIELDS}
 
 
 def _apply_to_live_settings(data: dict[str, Any]) -> None:
@@ -330,7 +367,13 @@ class SettingsService:
             valid_fields = set(env_resolved.dict().keys())
             merged_dict: dict[str, Any] = dict(env_resolved.dict())
             merged_dict.update(
-                {k: v for k, v in db_json.items() if v not in (None, "", [], {}) and k in valid_fields}
+                {
+                    k: v
+                    for k, v in db_json.items()
+                    if v not in (None, "", [], {})
+                    and k in valid_fields
+                    and k not in ENV_ONLY_FIELDS
+                }
             )
             merged_dict = Settings(**merged_dict).dict()
 
@@ -397,8 +440,13 @@ class SettingsService:
                 )
             )
             await db_session.commit()
-            # Update in-place
+            # Update in-place. Env-only fields (e.g. DB pool sizing) are never
+            # applied here: the engine pool is already built at boot from env,
+            # so letting an update mutate the live value would only make it
+            # diverge from the running pool.
             for k, v in candidate.dict().items():
+                if k in ENV_ONLY_FIELDS:
+                    continue
                 setattr(settings, k, v)
             cls._current = settings
             return settings

routstr/wallet.py

@@ -836,8 +836,22 @@ async def fetch_all_balances(
     if units is None:
         units = ["sat", "msat"]
 
-    async with db.create_session() as session:
-        user_balances = await db.balances_by_mint_and_unit(session)
+    # Received tokens are stored against primary_mint even when cashu_mints is
+    # empty, so include it in both the liability query and mint fan-out.
+    mint_urls: list[str] = list(settings.cashu_mints)
+    if settings.primary_mint and settings.primary_mint not in mint_urls:
+        mint_urls.append(settings.primary_mint)
+
+    user_balances: dict[tuple[str, str], int] = {}
+    liabilities_error: str | None = None
+    try:
+        async with db.create_session() as session:
+            user_balances = await db.balances_by_mint_and_unit(
+                session, mint_urls, units
+            )
+    except Exception as e:
+        logger.error("Error reading user balances", extra={"error": str(e)})
+        liabilities_error = str(e)
 
     mint_check_limit = asyncio.Semaphore(settings.mint_operation_concurrency)
 
@@ -874,40 +888,37 @@ async def fetch_all_balances(
             }
             return error_result
 
-    # Build the set of mints to inspect. Received tokens are stored against
-    # ``primary_mint`` (which defaults to a real mint even when ``cashu_mints``
-    # is empty), so include it as a fallback — otherwise a node that accepts
-    # payments would still report empty balances when ``cashu_mints`` is unset.
-    mint_urls: list[str] = list(settings.cashu_mints)
-    if settings.primary_mint and settings.primary_mint not in mint_urls:
-        mint_urls.append(settings.primary_mint)
-
     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
     total_user_balance_sats = 0
-
     for detail in balance_details:
-        if not detail.get("error"):
-            # Convert to sats for total calculation
-            unit = detail["unit"]
-            proofs_balance_sats = (
-                detail["wallet_balance"]
-                if unit == "sat"
-                else detail["wallet_balance"] // 1000
-            )
-            user_balance_sats = (
+        if detail.get("error"):
+            continue
+        unit = detail["unit"]
+        total_wallet_balance_sats += (
+            detail["wallet_balance"]
+            if unit == "sat"
+            else detail["wallet_balance"] // 1000
+        )
+        if liabilities_error is None:
+            total_user_balance_sats += (
                 detail["user_balance"]
                 if unit == "sat"
                 else detail["user_balance"] // 1000
             )
 
-            total_wallet_balance_sats += proofs_balance_sats
-            total_user_balance_sats += user_balance_sats
-
-    owner_balance = total_wallet_balance_sats - total_user_balance_sats
+    if liabilities_error is None:
+        owner_balance = total_wallet_balance_sats - total_user_balance_sats
+    else:
+        # Custody remains knowable when the DB read fails, but the user/owner
+        # split does not. Never report unknown liabilities as owner profit.
+        owner_balance = 0
+        for detail in balance_details:
+            detail["user_balance"] = 0
+            detail["owner_balance"] = 0
+            detail.setdefault("error", liabilities_error)
 
     return (
         balance_details,
@@ -947,9 +958,10 @@ async def periodic_payout() -> None:
                         # 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
+                            balances = await db.balances_by_mint_and_unit(
+                                session, [mint_url], [unit]
                             )
+                        user_balance = balances.get((mint_url, unit), 0)
                         if unit == "sat":
                             user_balance = user_balance // 1000
                         proofs_balance = sum(proof.amount for proof in proofs)
@@ -993,20 +1005,37 @@ async def periodic_payout() -> None:
             )
 
 
+async def _set_refund_sweep_state(refund_id: str, **values: object) -> None:
+    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(**values)
+        )
+        await session.commit()
+
+
 async def _refund_sweep_once(cutoff: int) -> None:
+    claim_cutoff = (
+        int(time.time()) - settings.refund_sweep_claim_timeout_seconds
+    )
+    claim_available = col(db.CashuTransaction.sweep_started_at).is_(None) | (
+        col(db.CashuTransaction.sweep_started_at) < claim_cutoff
+    )
     async with db.create_session() as session:
         stmt = select(db.CashuTransaction).where(
             db.CashuTransaction.type == "out",
             db.CashuTransaction.collected == False,  # noqa: E712
             db.CashuTransaction.swept == False,  # noqa: E712
             db.CashuTransaction.created_at < cutoff,
+            claim_available,
         )
         results = await session.exec(stmt)
         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.
+        reclaimed_stale_claim = refund.sweep_started_at is not None
+        claim_started_at = int(time.time())
         async with db.create_session() as session:
             claim = await session.exec(  # type: ignore[call-overload]
                 update(db.CashuTransaction)
@@ -1014,8 +1043,9 @@ async def _refund_sweep_once(cutoff: int) -> None:
                     col(db.CashuTransaction.id) == refund.id,
                     col(db.CashuTransaction.swept) == False,  # noqa: E712
                     col(db.CashuTransaction.collected) == False,  # noqa: E712
+                    claim_available,
                 )
-                .values(swept=True)
+                .values(sweep_started_at=claim_started_at)
             )
             await session.commit()
         if claim.rowcount != 1:
@@ -1023,6 +1053,9 @@ async def _refund_sweep_once(cutoff: int) -> None:
 
         try:
             await recieve_token(refund.token)
+            await _set_refund_sweep_state(
+                refund.id, swept=True, sweep_started_at=None
+            )
             logger.info(
                 "Swept uncollected refund",
                 extra={
@@ -1031,38 +1064,42 @@ async def _refund_sweep_once(cutoff: int) -> None:
                     "unit": refund.unit,
                 },
             )
-        except Exception as e:
+        except BaseException 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, swept=False)
+            if isinstance(e, Exception) and "already spent" in error_msg:
+                if reclaimed_stale_claim:
+                    # A prior worker may have redeemed the token and crashed
+                    # before finalizing. Treat the ambiguous stale claim as a
+                    # completed sweep rather than misreporting client collection.
+                    await _set_refund_sweep_state(
+                        refund.id, swept=True, sweep_started_at=None
+                    )
+                else:
+                    await _set_refund_sweep_state(
+                        refund.id,
+                        collected=True,
+                        swept=False,
+                        sweep_started_at=None,
                     )
-                    await session.commit()
                 logger.info(
-                    "Refund already spent (client collected), marking swept",
-                    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",
+                    "Refund token was already spent",
                     extra={
                         "id": refund.id,
-                        "error": str(e),
+                        "reclaimed_stale_claim": reclaimed_stale_claim,
                     },
                 )
+            else:
+                # Cancellation is a known-no-send outcome at this boundary, so
+                # release the lease under shield and let a later sweep retry.
+                await asyncio.shield(
+                    _set_refund_sweep_state(refund.id, sweep_started_at=None)
+                )
+                if not isinstance(e, Exception):
+                    raise
+                logger.warning(
+                    "Failed to sweep refund",
+                    extra={"id": refund.id, "error": str(e)},
+                )
 
 
 async def refund_sweep_once() -> None:
@@ -1112,15 +1149,21 @@ async def periodic_routstr_fee_payout() -> None:
                 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
 
+            # Wallet/proof preparation cannot send funds, so do it before the
+            # durable checkpoint. A preparation failure must not strand an
+            # in-progress payout that requires manual reconciliation.
             wallet = await get_wallet(settings.primary_mint, "sat")
             proofs = get_proofs_per_mint_and_unit(
                 wallet, settings.primary_mint, "sat", not_reserved=True
             )
+
+            async with db.create_session() as session:
+                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,