From c1b7f78a024e7e08bd1c506bc660c2b7435f99a7 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Mon, 24 Aug 2026 01:53:16 +0200 Subject: [PATCH] fix: bound Routstr auto-topup spend with a durable claim Routstr-to-Routstr auto top-up had no spend bound. Admin settings were accepted without range or type validation, and every low-balance cycle independently minted a Cashu token and handed it to the configured peer. A malicious, buggy, or persistently non-crediting peer therefore received a fresh bearer token every sixty seconds; nothing in the worker noticed that the previous one had never been credited, and nothing survived a restart, so the bleed was limited only by the owner's mint balance. Auto top-up now mirrors the PPQ claim machinery that already guards the Lightning path. Each provider gets one durable claim row keyed by its id, so a second worker (or the same worker after a restart) loses the insert or the ownership-fenced update instead of paying twice. The claim moves to "sent" before the network call, and only a peer balance that reaches the pre-topup balance plus the top-up amount clears it: an uncredited token holds the slot rather than being retried. Repeated non-credit walks the claim through exponential backoff to a halt that needs an admin release, and a rolling 24h cap bounds the total even when every attempt looks successful. Settings validation now rejects non-positive, non-finite, boolean, huge, and non-integer amounts, amounts outside the per-transaction range, and a missing mint URL, at the admin API as well as in the worker. The claim row is a CashuTransaction like the PPQ one, so no migration is needed; provider delete and type change refuse to orphan it. --- routstr/core/admin.py | 134 ++++- routstr/upstream/auto_topup.py | 552 +++++++++++++++++- .../test_routstr_auto_topup_claim.py | 351 +++++++++++ tests/unit/test_auto_topup.py | 177 ++---- 4 files changed, 1057 insertions(+), 157 deletions(-) create mode 100644 tests/integration/test_routstr_auto_topup_claim.py diff --git a/routstr/core/admin.py b/routstr/core/admin.py index 985f9c46..2da3baf1 100644 --- a/routstr/core/admin.py +++ b/routstr/core/admin.py @@ -907,27 +907,43 @@ class UpstreamProviderUpdateBySlug(BaseModel): provider_settings: dict | None = None -async def _active_ppq_claim_in_session(session: AsyncSession, provider_id: int) -> bool: +async def _active_auto_topup_claim_in_session( + session: AsyncSession, provider_id: int, provider_type: str +) -> bool: """Check for an active claim inside the caller's transaction. Must share the transaction of whatever destructive write it is guarding — a check in its own session leaves a window for a worker to create the claim between the check and the commit. """ - from ..upstream.auto_topup import _ppq_state_id_for_provider + from ..upstream.auto_topup import ( + _ppq_state_id_for_provider, + _routstr_state_id_for_provider, + ) - claim = await session.get(CashuTransaction, _ppq_state_id_for_provider(provider_id)) + state_id = ( + _ppq_state_id_for_provider(provider_id) + if provider_type == "ppqai" + else _routstr_state_id_for_provider(provider_id) + ) + claim = await session.get(CashuTransaction, state_id) return claim is not None and not claim.collected and not claim.swept -def _require_valid_ppq_auto_topup(provider_type: str, settings: dict | None) -> None: - """Reject PPQ auto top-up settings the worker would later refuse.""" - if provider_type != "ppqai": +def _require_valid_auto_topup(provider_type: str, settings: dict | None) -> None: + """Reject auto top-up settings the worker would later refuse.""" + from ..upstream.auto_topup import ( + validate_ppq_auto_topup_settings, + validate_routstr_auto_topup_settings, + ) + + if provider_type == "ppqai": + problem = validate_ppq_auto_topup_settings(settings) + elif provider_type == "routstr": + problem = validate_routstr_auto_topup_settings(settings) + else: return - from ..upstream.auto_topup import validate_ppq_auto_topup_settings - - problem = validate_ppq_auto_topup_settings(settings) if problem is not None: raise HTTPException(status_code=400, detail=problem) @@ -952,16 +968,18 @@ async def _apply_provider_update( ) if ( provider_type_changed - and provider.provider_type == "ppqai" + and provider.provider_type in ("ppqai", "routstr") and provider.id is not None - and await _active_ppq_claim_in_session(session, provider.id) + and await _active_auto_topup_claim_in_session( + session, provider.id, provider.provider_type + ) ): - # Changing the type would orphan the claim: the PPQ endpoints refuse - # non-ppqai providers, so nobody could ever inspect or release it. + # Changing the type would orphan the claim: the claim endpoints refuse + # providers of the wrong type, so nobody could inspect or release it. raise HTTPException( status_code=409, detail=( - "This provider has an active PPQ auto top-up claim. Release " + "This provider has an active auto top-up claim. Release " "it before changing the provider type" ), ) @@ -1012,7 +1030,7 @@ async def _apply_provider_update( except (json.JSONDecodeError, TypeError): effective_settings = None if effective_settings is not None: - _require_valid_ppq_auto_topup(provider.provider_type, effective_settings) + _require_valid_auto_topup(provider.provider_type, effective_settings) if payload.provider_settings is not None: provider.provider_settings = json.dumps(payload.provider_settings) @@ -1052,7 +1070,7 @@ async def create_upstream_provider( else: slug = await allocate_unique_provider_slug(session, payload.provider_type) - _require_valid_ppq_auto_topup(payload.provider_type, payload.provider_settings) + _require_valid_auto_topup(payload.provider_type, payload.provider_settings) provider = UpstreamProviderRow( slug=slug, @@ -1148,16 +1166,19 @@ async def delete_upstream_provider(provider_id: str) -> dict[str, object]: # re-reads the provider inside its own transaction, so these two # writes serialise — either the claim lands first and this 409s, or # the delete lands first and the worker refuses to claim. - if provider.provider_type == "ppqai" and await _active_ppq_claim_in_session( - session, deleted_id + if provider.provider_type in ( + "ppqai", + "routstr", + ) and await _active_auto_topup_claim_in_session( + session, deleted_id, provider.provider_type ): # Deleting now would orphan the claim and any funds it tracks: - # the PPQ endpoints 404 without the provider row, so the claim + # the claim endpoints 404 without the provider row, so the claim # could never again be inspected or released. raise HTTPException( status_code=409, detail=( - "This provider has an active PPQ auto top-up claim. " + "This provider has an active auto top-up claim. " "Resolve and release it before deleting the provider" ), ) @@ -1842,6 +1863,72 @@ async def release_ppq_auto_topup_api( return {"ok": True, "released": True} +_ROUTSTR_RELEASE_ERRORS = { + "no_active_claim": "No active Routstr auto top-up claim to release", + "stale_state": "The claim changed since it was reviewed; reload and check again", + "claim_changed": ( + "The claim changed while the release was being applied; reload and check again" + ), +} + + +class ReleaseRoutstrAutoTopupRequest(BaseModel): + confirmed_peer_reconciled: bool + state_token: str | None = None + + +async def _require_routstr_provider(provider_id: int) -> UpstreamProviderRow: + async with create_session() as session: + provider = await session.get(UpstreamProviderRow, provider_id) + if provider is None: + raise HTTPException(status_code=404, detail="Provider not found") + if provider.provider_type != "routstr": + raise HTTPException(status_code=400, detail="Provider is not a Routstr node") + return provider + + +@admin_router.get( + "/api/upstream-providers/{provider_id}/routstr-auto-topup", + dependencies=[Depends(require_admin_api)], +) +async def get_routstr_auto_topup_api(provider_id: int) -> dict[str, object]: + await _require_routstr_provider(provider_id) + from ..upstream.auto_topup import get_routstr_auto_topup_state + + return {"ok": True, **await get_routstr_auto_topup_state(provider_id)} + + +@admin_router.post( + "/api/upstream-providers/{provider_id}/routstr-auto-topup/release", + dependencies=[Depends(require_admin_api)], +) +async def release_routstr_auto_topup_api( + provider_id: int, payload: ReleaseRoutstrAutoTopupRequest +) -> dict[str, object]: + await _require_routstr_provider(provider_id) + if not payload.confirmed_peer_reconciled: + raise HTTPException( + status_code=400, + detail="Confirm the peer credited or returned the token before releasing", + ) + + from ..upstream.auto_topup import release_routstr_auto_topup_state + + outcome = await release_routstr_auto_topup_state( + provider_id, state_token=payload.state_token + ) + if not outcome.released: + raise HTTPException( + status_code=409, detail=_ROUTSTR_RELEASE_ERRORS[outcome.reason] + ) + + logger.warning( + "Admin released Routstr auto top-up claim after manual reconciliation", + extra={"provider_id": provider_id, "state_token": payload.state_token}, + ) + return {"ok": True, "released": True} + + def _transaction_status(tx: CashuTransaction) -> str: """An outgoing admin withdrawal ends at "issued": the node hands the bearer token over and never learns whether it was redeemed, so its flags stay false @@ -1869,10 +1956,11 @@ async def get_transactions_api( async with create_session() as session: from sqlmodel import col, func - # Hide only the deterministic PPQ claim-lock rows. Append-only PPQ - # payment rows remain visible as the audit trail for irreversible melts. + # Hide only the deterministic claim-lock rows. Append-only PPQ payment + # rows and auto-topup token rows remain visible as the audit trail. base = select(CashuTransaction).where( - ~col(CashuTransaction.id).like("ppq-auto-topup-%") + ~col(CashuTransaction.id).like("ppq-auto-topup-%"), + ~col(CashuTransaction.id).like("routstr-auto-topup-%"), ) if type: base = base.where(CashuTransaction.type == type) diff --git a/routstr/upstream/auto_topup.py b/routstr/upstream/auto_topup.py index c8eadb6e..6dd5ac7d 100644 --- a/routstr/upstream/auto_topup.py +++ b/routstr/upstream/auto_topup.py @@ -58,6 +58,33 @@ PPQ_MAX_TOPUP_USD = 500 # the damage instead of letting the worker drain the owner's mint funds one # per-transaction-capped payment at a time. PPQ_MAX_DAILY_TOPUP_USD = 300 +# Routstr-to-Routstr claim lifecycle. "claimed" holds the slot while the token +# is being minted; nothing has left the wallet yet. "sent" means a bearer token +# was handed to the peer and only the peer's balance can say whether it landed. +# "backoff" holds the failure count between attempts, and "halted" stops the +# provider entirely until an admin releases it. +ROUTSTR_PHASE_CLAIMED = "claimed" +ROUTSTR_PHASE_SENT = "sent" +ROUTSTR_PHASE_BACKOFF = "backoff" +ROUTSTR_PHASE_HALTED = "halted" +ROUTSTR_PHASES = frozenset( + { + ROUTSTR_PHASE_CLAIMED, + ROUTSTR_PHASE_SENT, + ROUTSTR_PHASE_BACKOFF, + ROUTSTR_PHASE_HALTED, + } +) +ROUTSTR_PENDING_TTL_SECONDS = 15 * 60 +ROUTSTR_BACKOFF_BASE_SECONDS = 15 * 60 +ROUTSTR_MAX_TOPUP_FAILURES = 3 +ROUTSTR_MIN_TOPUP_SATS = 1 +ROUTSTR_MAX_TOPUP_SATS = 1_000_000 +# Rolling 24h ceiling on total Routstr auto top-up spend across all peers. The +# per-attempt claim already stops a peer from being paid twice for the same +# uncredited token; this bounds the total even when every attempt is credited +# and the peer simply keeps reporting a below-threshold balance. +ROUTSTR_MAX_DAILY_TOPUP_SATS = 2_000_000 async def periodic_auto_topup() -> None: @@ -139,7 +166,7 @@ async def _reconcile_all_ppq_claims() -> set[int]: return active_provider_ids -def _invalid_ppq_number(value: object, *, integer: bool = False) -> bool: +def _invalid_topup_number(value: object, *, integer: bool = False) -> bool: if isinstance(value, bool) or not isinstance(value, (int, float)): return True try: @@ -160,9 +187,9 @@ def validate_ppq_auto_topup_settings(settings: dict | None) -> str | None: threshold = settings.get("topup_threshold") amount = settings.get("topup_amount_limit") - if _invalid_ppq_number(threshold): + if _invalid_topup_number(threshold): return "PPQ auto top-up threshold must be a positive number" - if _invalid_ppq_number(amount, integer=True): + if _invalid_topup_number(amount, integer=True): return "PPQ auto top-up amount must be a positive whole number" amount_usd = int(typing.cast(int | float, amount)) if not PPQ_MIN_TOPUP_USD <= amount_usd <= PPQ_MAX_TOPUP_USD: @@ -173,6 +200,29 @@ def validate_ppq_auto_topup_settings(settings: dict | None) -> str | None: return None +def validate_routstr_auto_topup_settings(settings: dict | None) -> str | None: + """Return why enabled Routstr auto top-up settings are invalid, if anything.""" + if not settings or not settings.get("auto_topup"): + return None + + threshold = settings.get("topup_threshold") + amount = settings.get("topup_amount_limit") + mint_url = settings.get("topup_mint_url") + if _invalid_topup_number(threshold): + return "Routstr auto top-up threshold must be a positive number" + if _invalid_topup_number(amount, integer=True): + return "Routstr auto top-up amount must be a positive whole number" + amount_sats = int(typing.cast(int | float, amount)) + if not ROUTSTR_MIN_TOPUP_SATS <= amount_sats <= ROUTSTR_MAX_TOPUP_SATS: + return ( + f"Routstr auto top-up amount must be between {ROUTSTR_MIN_TOPUP_SATS} " + f"and {ROUTSTR_MAX_TOPUP_SATS} sats" + ) + if not isinstance(mint_url, str) or not mint_url.strip(): + return "Routstr auto top-up requires a mint URL" + return None + + async def _check_and_topup_ppq_from_row(row: UpstreamProviderRow) -> None: settings: dict = {} if row.provider_settings: @@ -212,22 +262,18 @@ async def _check_and_topup(row: UpstreamProviderRow) -> None: if not settings.get("auto_topup"): return - threshold = settings.get("topup_threshold") - amount = settings.get("topup_amount_limit") - mint_url = settings.get("topup_mint_url") - - if not threshold or not amount or not mint_url: + problem = validate_routstr_auto_topup_settings(settings) + if problem is not None: logger.warning( - "Auto top-up enabled but missing configuration", - extra={ - "provider_id": row.id, - "has_threshold": bool(threshold), - "has_amount": bool(amount), - "has_mint": bool(mint_url), - }, + "Auto top-up enabled but its configuration is invalid", + extra={"provider_id": row.id, "problem": problem}, ) return + threshold = float(settings["topup_threshold"]) + amount = int(settings["topup_amount_limit"]) + mint_url = str(settings["topup_mint_url"]) + if not row.api_key: return @@ -235,9 +281,12 @@ async def _check_and_topup(row: UpstreamProviderRow) -> None: provider = RoutstrUpstreamProvider.from_db_row(row) if provider is None: return + if await _reconcile_routstr_state(row, provider): + return + balance = await provider.get_balance() - if balance is None: + if balance is None or not math.isfinite(balance) or balance < 0: logger.warning( "Could not fetch balance for auto top-up", extra={"provider_id": row.id, "base_url": row.base_url}, @@ -247,6 +296,27 @@ async def _check_and_topup(row: UpstreamProviderRow) -> None: if balance >= threshold * 1000: return + spent_24h_sats = await _routstr_spent_last_24h_sats() + if spent_24h_sats + amount > ROUTSTR_MAX_DAILY_TOPUP_SATS: + logger.critical( + "Auto top-up skipped: rolling 24h spend cap reached", + extra={ + "provider_id": row.id, + "spent_24h_sats": spent_24h_sats, + "topup_amount": amount, + "daily_cap_sats": ROUTSTR_MAX_DAILY_TOPUP_SATS, + }, + ) + return + + # The balance the peer must report before another token may be sent. Any + # shortfall is treated as "not credited": the token is a bearer instrument + # and a peer that took one without crediting it must not be handed another. + expected_sats = math.floor(balance) + amount + operation_id = await _claim_routstr_topup(row, expected_sats=expected_sats) + if operation_id is None: + return + # Balance is below threshold - create token and top up logger.info( "Auto top-up triggered", @@ -271,6 +341,7 @@ async def _check_and_topup(row: UpstreamProviderRow) -> None: "error": str(e), }, ) + await _release_routstr_claim(row, operation_id) return actual_mint_url = token_mint_url(token, mint_url) @@ -305,8 +376,21 @@ async def _check_and_topup(row: UpstreamProviderRow) -> None: "Auto-topup token was released after persistence failed", extra={"provider_id": row.id, "mint_url": actual_mint_url}, ) + await _release_routstr_claim(row, operation_id) return + # Move the claim before the network call, not after: a worker that dies + # mid-request must leave behind a claim that says a token may already be + # with the peer. + await _mark_routstr_sent( + row, + operation_id, + expected_sats=expected_sats, + token=token, + amount=amount, + mint_url=actual_mint_url, + ) + result = await provider.topup(token) if "error" in result: @@ -348,6 +432,442 @@ async def _check_and_topup(row: UpstreamProviderRow) -> None: ) +def _routstr_state_id(row: UpstreamProviderRow) -> str: + if row.id is None: + raise ValueError("Routstr auto top-up requires a persisted provider row") + return _routstr_state_id_for_provider(row.id) + + +def _routstr_state_id_for_provider(provider_id: int | str) -> str: + return f"routstr-auto-topup-{provider_id}" + + +class RoutstrClaim(typing.NamedTuple): + operation_id: str + # Worker lease for "claimed"/"sent", retry-not-before for "backoff", and + # meaningless for "halted". + deadline: int + phase: str + # Peer balance in sats that proves this attempt was credited. + expected_sats: int + failures: int + + +def _routstr_request_id( + operation_id: str, + deadline: int, + phase: str, + expected_sats: int, + failures: int, +) -> str: + return f"routstr:{operation_id}:{deadline}:{phase}:{expected_sats}:{failures}" + + +def _parse_routstr_request_id(request_id: str | None) -> RoutstrClaim | None: + parts = (request_id or "").split(":", 5) + if len(parts) != 6 or parts[0] != "routstr" or parts[3] not in ROUTSTR_PHASES: + return None + try: + deadline = int(parts[2]) + expected_sats = int(parts[4]) + failures = int(parts[5]) + except (TypeError, ValueError): + return None + return RoutstrClaim(parts[1], deadline, parts[3], expected_sats, failures) + + +async def _routstr_spent_last_24h_sats() -> int: + """Total sats committed to Routstr auto top-ups in the last 24 hours. + + Uncollected rows count too: a token whose delivery is unconfirmed is spent + for capping purposes. Rows marked ``collected=False, swept=True`` record + tokens that were provably returned to the wallet and are excluded. + """ + cutoff = int(time.time()) - 24 * 60 * 60 + async with create_session() as session: + rows = ( + await session.exec( + select(CashuTransaction.amount, CashuTransaction.unit).where( + col(CashuTransaction.source) == "auto_topup", + col(CashuTransaction.type) == "out", + col(CashuTransaction.created_at) >= cutoff, + or_( + col(CashuTransaction.collected) == True, # noqa: E712 + col(CashuTransaction.swept) == False, # noqa: E712 + ), + ) + ) + ).all() + return sum( + amount if unit == "sat" else math.ceil(amount / 1000) for amount, unit in rows + ) + + +async def _routstr_provider_is_claimable( + session: AsyncSession, provider_id: int | str | None +) -> bool: + """Re-read the provider inside the claim transaction. + + Same reasoning as :func:`_ppq_provider_is_claimable`: a provider deleted or + retyped concurrently must either be visible here or lose the race against + the claim we are about to write. + """ + if provider_id is None: + return False + current = await session.get(UpstreamProviderRow, provider_id) + return current is not None and current.provider_type == "routstr" + + +async def _claim_routstr_topup( + row: UpstreamProviderRow, *, expected_sats: int +) -> str | None: + """Acquire the provider's single durable auto top-up slot. + + An expired backoff hands its failure count to the new attempt, so repeated + non-crediting peers still walk towards the halt instead of resetting the + counter every cycle. + """ + state_id = _routstr_state_id(row) + operation_id = uuid.uuid4().hex + deadline = int(time.time()) + ROUTSTR_PENDING_TTL_SECONDS + + async with create_session() as session: + if not await _routstr_provider_is_claimable(session, row.id): + return None + existing = await session.get(CashuTransaction, state_id) + if existing is not None: + failures = 0 + if not (existing.collected or existing.swept): + claim = _parse_routstr_request_id(existing.request_id) + if ( + claim is None + or claim.phase != ROUTSTR_PHASE_BACKOFF + or time.time() < claim.deadline + ): + return None + failures = claim.failures + result = await session.exec( # type: ignore[call-overload] + update(CashuTransaction) + .where( + col(CashuTransaction.id) == state_id, + # Fence on the exact row that was read: any concurrent + # writer that moved the claim must win instead of us. + col(CashuTransaction.request_id) == existing.request_id, + ) + .values( + token="pending", + amount=0, + unit="sat", + mint_url=None, + request_id=_routstr_request_id( + operation_id, + deadline, + ROUTSTR_PHASE_CLAIMED, + expected_sats, + failures, + ), + collected=False, + swept=False, + created_at=int(time.time()), + source="routstr_auto_topup_claim", + ) + ) + await session.commit() + if (getattr(result, "rowcount", 0) or 0) != 1: + return None + return operation_id + + try: + async with create_session() as session: + if not await _routstr_provider_is_claimable(session, row.id): + return None + session.add( + CashuTransaction( + id=state_id, + token="pending", + amount=0, + unit="sat", + type="out", + request_id=_routstr_request_id( + operation_id, + deadline, + ROUTSTR_PHASE_CLAIMED, + expected_sats, + 0, + ), + collected=False, + source="routstr_auto_topup_claim", + ) + ) + await session.commit() + except IntegrityError: + return None + return operation_id + + +async def _advance_routstr_claim( + row: UpstreamProviderRow, + operation_id: str, + *, + deadline: int, + phase: str, + expected_sats: int, + failures: int, + token: str | None = None, + amount: int | None = None, + mint_url: str | None = None, +) -> bool: + """Move this worker's claim to another phase, if it still owns it.""" + values: dict[str, object] = { + "request_id": _routstr_request_id( + operation_id, deadline, phase, expected_sats, failures + ) + } + if token is not None: + values.update(token=token, amount=amount, mint_url=mint_url) + + async with create_session() as session: + result = await session.exec( # type: ignore[call-overload] + update(CashuTransaction) + .where( + col(CashuTransaction.id) == _routstr_state_id(row), + col(CashuTransaction.request_id).like(f"routstr:{operation_id}:%"), + col(CashuTransaction.collected) == False, # noqa: E712 + col(CashuTransaction.swept) == False, # noqa: E712 + ) + .values(**values) + ) + await session.commit() + return (getattr(result, "rowcount", 0) or 0) == 1 + + +async def _set_routstr_state_terminal( + row: UpstreamProviderRow, operation_id: str, *, collected: bool, swept: bool +) -> bool: + """Finish an attempt only if this worker still owns the claim.""" + async with create_session() as session: + result = await session.exec( # type: ignore[call-overload] + update(CashuTransaction) + .where( + col(CashuTransaction.id) == _routstr_state_id(row), + col(CashuTransaction.request_id).like(f"routstr:{operation_id}:%"), + col(CashuTransaction.collected) == False, # noqa: E712 + col(CashuTransaction.swept) == False, # noqa: E712 + ) + .values(collected=collected, swept=swept) + ) + await session.commit() + return (getattr(result, "rowcount", 0) or 0) == 1 + + +async def _release_routstr_claim(row: UpstreamProviderRow, operation_id: str) -> None: + """Hand back a claim whose token never left the wallet.""" + if not await _set_routstr_state_terminal( + row, operation_id, collected=False, swept=True + ): + logger.warning( + "Could not release the auto top-up claim after a pre-send failure; " + "it is owned by another attempt", + extra={"provider_id": row.id}, + ) + + +async def _mark_routstr_sent( + row: UpstreamProviderRow, + operation_id: str, + *, + expected_sats: int, + token: str, + amount: int, + mint_url: str, +) -> None: + claim = await _current_routstr_claim(row) + failures = claim.failures if claim else 0 + if not await _advance_routstr_claim( + row, + operation_id, + deadline=int(time.time()) + ROUTSTR_PENDING_TTL_SECONDS, + phase=ROUTSTR_PHASE_SENT, + expected_sats=expected_sats, + failures=failures, + token=token, + amount=amount, + mint_url=mint_url, + ): + raise RuntimeError("Routstr auto top-up claim ownership was lost") + + +async def _current_routstr_claim(row: UpstreamProviderRow) -> RoutstrClaim | None: + async with create_session() as session: + transaction = await session.get(CashuTransaction, _routstr_state_id(row)) + if transaction is None: + return None + return _parse_routstr_request_id(transaction.request_id) + + +async def _reconcile_routstr_state( + row: UpstreamProviderRow, provider: RoutstrUpstreamProvider +) -> bool: + """Return True while a prior attempt must suppress a new payment.""" + async with create_session() as session: + transaction = await session.get(CashuTransaction, _routstr_state_id(row)) + if transaction is None or transaction.collected or transaction.swept: + return False + + claim = _parse_routstr_request_id(transaction.request_id) + if claim is None: + logger.critical( + "Malformed auto top-up state; suppressing duplicate payment", + extra={"provider_id": row.id}, + ) + return True + + now = time.time() + if claim.phase == ROUTSTR_PHASE_HALTED: + return True + if claim.phase == ROUTSTR_PHASE_BACKOFF: + return now < claim.deadline + if claim.phase == ROUTSTR_PHASE_CLAIMED: + # Nothing left the wallet, so a dead worker's slot is free to reuse. + if now < claim.deadline: + return True + return not await _set_routstr_state_terminal( + row, claim.operation_id, collected=False, swept=True + ) + + balance = await provider.get_balance() + if ( + balance is not None + and math.isfinite(balance) + and balance >= claim.expected_sats + ): + if not await _set_routstr_state_terminal( + row, claim.operation_id, collected=True, swept=False + ): + logger.critical( + "Auto top-up was credited but its claim was already released; " + "a duplicate top-up is possible on the next cycle", + extra={"provider_id": row.id}, + ) + return True + if now < claim.deadline: + return True + + failures = claim.failures + 1 + if failures >= ROUTSTR_MAX_TOPUP_FAILURES: + await _advance_routstr_claim( + row, + claim.operation_id, + deadline=claim.deadline, + phase=ROUTSTR_PHASE_HALTED, + expected_sats=claim.expected_sats, + failures=failures, + ) + logger.critical( + "Auto top-up halted: the peer repeatedly failed to credit a token", + extra={ + "provider_id": row.id, + "base_url": row.base_url, + "failures": failures, + "admin_action": ( + f"POST /admin/api/upstream-providers/{row.id}" + "/routstr-auto-topup/release" + ), + }, + ) + return True + + await _advance_routstr_claim( + row, + claim.operation_id, + deadline=int(now) + ROUTSTR_BACKOFF_BASE_SECONDS * 2 ** (failures - 1), + phase=ROUTSTR_PHASE_BACKOFF, + expected_sats=claim.expected_sats, + failures=failures, + ) + logger.warning( + "Auto top-up was not credited by the peer; backing off", + extra={ + "provider_id": row.id, + "base_url": row.base_url, + "expected_sats": claim.expected_sats, + "failures": failures, + }, + ) + return True + + +async def get_routstr_auto_topup_state(provider_id: int) -> dict[str, object]: + """Return admin-safe state for a provider's durable Routstr claim.""" + async with create_session() as session: + transaction = await session.get( + CashuTransaction, _routstr_state_id_for_provider(provider_id) + ) + if transaction is None or transaction.collected or transaction.swept: + return {"active": False} + + claim = _parse_routstr_request_id(transaction.request_id) + return { + "active": True, + # Echoed back verbatim on release so a claim that moved on since the + # admin reviewed it fails the write instead of being swept unseen. + "state_token": transaction.request_id, + "operation_id": claim.operation_id if claim else None, + "phase": claim.phase if claim else None, + "expected_sats": claim.expected_sats if claim else None, + "failures": claim.failures if claim else None, + "deadline": claim.deadline if claim else None, + "created_at": transaction.created_at, + "amount": transaction.amount, + "unit": transaction.unit, + "mint_url": transaction.mint_url, + "malformed": claim is None, + } + + +class RoutstrReleaseOutcome(typing.NamedTuple): + released: bool + reason: str + + +async def release_routstr_auto_topup_state( + provider_id: int, *, state_token: str | None +) -> RoutstrReleaseOutcome: + """Clear a halted or stuck claim after an admin reconciles the peer. + + Unlike a Lightning melt there is no in-flight window to protect: the token + is already with the peer or still in the wallet either way. What the fence + does protect is the admin's decision — the row must be byte-identical to + the one they reviewed, so a claim that advanced in the meantime is not + swept on the strength of stale information. + """ + state_id = _routstr_state_id_for_provider(provider_id) + async with create_session() as session: + transaction = await session.get(CashuTransaction, state_id) + + if transaction is None or transaction.collected or transaction.swept: + return RoutstrReleaseOutcome(False, "no_active_claim") + if transaction.request_id != state_token: + return RoutstrReleaseOutcome(False, "stale_state") + + async with create_session() as session: + result = await session.exec( # type: ignore[call-overload] + update(CashuTransaction) + .where( + col(CashuTransaction.id) == state_id, + col(CashuTransaction.request_id) == state_token, + col(CashuTransaction.collected) == False, # noqa: E712 + col(CashuTransaction.swept) == False, # noqa: E712 + ) + .values(swept=True) + ) + if (getattr(result, "rowcount", 0) or 0) == 1: + await session.commit() + return RoutstrReleaseOutcome(True, "released") + await session.rollback() + return RoutstrReleaseOutcome(False, "claim_changed") + + def _ppq_state_id(row: UpstreamProviderRow) -> str: if row.id is None: raise ValueError("PPQ auto top-up requires a persisted provider row") diff --git a/tests/integration/test_routstr_auto_topup_claim.py b/tests/integration/test_routstr_auto_topup_claim.py new file mode 100644 index 00000000..25b827a9 --- /dev/null +++ b/tests/integration/test_routstr_auto_topup_claim.py @@ -0,0 +1,351 @@ +"""Real-database tests for the Routstr-to-Routstr auto top-up spend bound. + +The bound has to survive a process restart and concurrent workers, so these +run against actual SQL instead of mocked sessions: an in-memory counter would +pass a mocked test and still let a non-crediting peer drain the wallet. +""" + +import importlib +import time +from contextlib import ExitStack +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from sqlmodel import select + +from routstr.core.db import CashuTransaction, UpstreamProviderRow, create_session +from routstr.upstream import auto_topup as auto_topup_module +from routstr.upstream.auto_topup import ( + ROUTSTR_MAX_DAILY_TOPUP_SATS, + ROUTSTR_MAX_TOPUP_FAILURES, + ROUTSTR_PHASE_BACKOFF, + ROUTSTR_PHASE_HALTED, + ROUTSTR_PHASE_SENT, + _check_and_topup, + _claim_routstr_topup, + _parse_routstr_request_id, + _routstr_spent_last_24h_sats, + _routstr_state_id_for_provider, + get_routstr_auto_topup_state, + release_routstr_auto_topup_state, +) + +pytestmark = pytest.mark.asyncio + +TOPUP_SATS = 50 + + +async def _seed_provider(provider_id: int = 1) -> UpstreamProviderRow: + import json + + row = UpstreamProviderRow( + id=provider_id, + slug=f"peer-{provider_id}", + provider_type="routstr", + base_url="https://peer.test", + api_key="secret", + enabled=True, + provider_settings=json.dumps( + { + "auto_topup": True, + "topup_threshold": 1, + "topup_amount_limit": TOPUP_SATS, + "topup_mint_url": "https://mint.test", + } + ), + ) + async with create_session() as session: + session.add(row) + await session.commit() + await session.refresh(row) + return row + + +def _peer(balance: float, *, topup: object = None) -> MagicMock: + provider = MagicMock() + provider.get_balance = AsyncMock(return_value=balance) + provider.topup = AsyncMock(return_value=topup or {"balance": balance}) + return provider + + +def _patch_wallet(module: Any, peer: MagicMock, token: str) -> ExitStack: + stack = ExitStack() + stack.enter_context( + patch.object(module.RoutstrUpstreamProvider, "from_db_row", return_value=peer) + ) + stack.enter_context( + patch.object(module, "send_token", AsyncMock(return_value=token)) + ) + stack.enter_context( + patch.object(module, "token_mint_url", return_value="https://mint.test") + ) + return stack + + +async def _claim_state(provider_id: int = 1) -> CashuTransaction | None: + async with create_session() as session: + return await session.get( + CashuTransaction, _routstr_state_id_for_provider(provider_id) + ) + + +async def _sent_tokens() -> list[CashuTransaction]: + async with create_session() as session: + return list( + ( + await session.exec( + select(CashuTransaction).where( + CashuTransaction.source == "auto_topup" + ) + ) + ).all() + ) + + +async def test_second_worker_cannot_claim_while_the_first_holds_one( + patched_db_engine: Any, +) -> None: + row = await _seed_provider() + assert await _claim_routstr_topup(row, expected_sats=TOPUP_SATS) is not None + assert await _claim_routstr_topup(row, expected_sats=TOPUP_SATS) is None + + +async def test_token_is_persisted_before_it_reaches_the_peer( + patched_db_engine: Any, +) -> None: + row = await _seed_provider() + seen: list[CashuTransaction] = [] + + async def _record_then_accept(token: str) -> dict: + seen.extend(await _sent_tokens()) + return {"balance": TOPUP_SATS} + + peer = _peer(0.0) + peer.topup = AsyncMock(side_effect=_record_then_accept) + + with _patch_wallet(auto_topup_module, peer, "cashu-token-1"): + await _check_and_topup(row) + + assert [tx.token for tx in seen] == ["cashu-token-1"] + assert seen[0].collected is False + assert (await _sent_tokens())[0].collected is True + + +async def test_untracked_token_is_returned_and_never_sent( + patched_db_engine: Any, +) -> None: + row = await _seed_provider() + peer = _peer(0.0) + + with ( + _patch_wallet(auto_topup_module, peer, "cashu-token-1"), + patch.object( + auto_topup_module, + "store_cashu_transaction", + AsyncMock(side_effect=RuntimeError("database unavailable")), + ), + patch.object( + auto_topup_module, "release_token_reservation", AsyncMock() + ) as reclaim, + ): + await _check_and_topup(row) + + reclaim.assert_awaited_once_with("cashu-token-1") + peer.topup.assert_not_awaited() + # Nothing left the wallet, so the slot must be free again immediately. + state = await _claim_state() + assert state is not None and state.swept is True + + +async def test_failed_topup_keeps_token_uncollected_and_suppresses_new_claims( + patched_db_engine: Any, +) -> None: + row = await _seed_provider() + peer = _peer(0.0, topup={"error": "rejected"}) + + with _patch_wallet(auto_topup_module, peer, "cashu-token-1"): + await _check_and_topup(row) + + tokens = await _sent_tokens() + assert len(tokens) == 1 + assert tokens[0].collected is False + + claim = _parse_routstr_request_id((await _claim_state()).request_id) # type: ignore[union-attr] + assert claim is not None and claim.phase == ROUTSTR_PHASE_SENT + + peer.topup.reset_mock() + with _patch_wallet(auto_topup_module, peer, "cashu-token-2"): + await _check_and_topup(row) + + peer.topup.assert_not_awaited() + assert len(await _sent_tokens()) == 1 + + +async def test_claim_blocks_a_restarted_process(patched_db_engine: Any) -> None: + row = await _seed_provider() + peer = _peer(0.0, topup={"error": "rejected"}) + + with _patch_wallet(auto_topup_module, peer, "cashu-token-1"): + await _check_and_topup(row) + + # A fresh module drops every process-local variable; only a durable row + # can still stop the next payment. + reloaded = importlib.reload(auto_topup_module) + try: + peer.topup.reset_mock() + with _patch_wallet(reloaded, peer, "cashu-token-2"): + await reloaded._check_and_topup(row) + peer.topup.assert_not_awaited() + finally: + importlib.reload(auto_topup_module) + + assert len(await _sent_tokens()) == 1 + + +async def _uncredited_attempt( + row: UpstreamProviderRow, peer: MagicMock, token: str +) -> None: + """One full payment attempt against a peer that never credits it.""" + with _patch_wallet(auto_topup_module, peer, token): + await _check_and_topup(row) + await _expire_claim() + with _patch_wallet(auto_topup_module, peer, f"{token}-retry"): + await _check_and_topup(row) + await _expire_claim() + + +async def test_non_crediting_peer_is_halted_after_repeated_failures( + patched_db_engine: Any, +) -> None: + row = await _seed_provider() + peer = _peer(0.0) + + for attempt in range(ROUTSTR_MAX_TOPUP_FAILURES): + await _uncredited_attempt(row, peer, f"cashu-token-{attempt}") + + claim = _parse_routstr_request_id((await _claim_state()).request_id) # type: ignore[union-attr] + assert claim is not None and claim.phase == ROUTSTR_PHASE_HALTED + + peer.topup.reset_mock() + with _patch_wallet(auto_topup_module, peer, "cashu-token-after-halt"): + await _check_and_topup(row) + peer.topup.assert_not_awaited() + assert len(await _sent_tokens()) == ROUTSTR_MAX_TOPUP_FAILURES + + +async def _expire_claim(provider_id: int = 1) -> None: + """Age the claim's deadline so the reconciler treats it as timed out.""" + async with create_session() as session: + state = await session.get( + CashuTransaction, _routstr_state_id_for_provider(provider_id) + ) + assert state is not None and state.request_id is not None + claim = _parse_routstr_request_id(state.request_id) + assert claim is not None + state.request_id = auto_topup_module._routstr_request_id( + claim.operation_id, + int(time.time()) - 1, + claim.phase, + claim.expected_sats, + claim.failures, + ) + session.add(state) + await session.commit() + + +async def test_crediting_peer_releases_the_claim_for_a_later_topup( + patched_db_engine: Any, +) -> None: + row = await _seed_provider() + peer = _peer(0.0) + + with _patch_wallet(auto_topup_module, peer, "cashu-token-1"): + await _check_and_topup(row) + + tokens = await _sent_tokens() + assert len(tokens) == 1 and tokens[0].collected is True + + peer.get_balance = AsyncMock(return_value=float(TOPUP_SATS)) + with _patch_wallet(auto_topup_module, peer, "cashu-token-2"): + await _check_and_topup(row) + + state = await _claim_state() + assert state is not None and state.collected is True + + +async def test_rolling_budget_refuses_a_topup_that_would_exceed_it( + patched_db_engine: Any, +) -> None: + row = await _seed_provider() + async with create_session() as session: + session.add( + CashuTransaction( + id="prior-spend", + token="cashu-prior", + amount=ROUTSTR_MAX_DAILY_TOPUP_SATS, + unit="sat", + type="out", + source="auto_topup", + ) + ) + await session.commit() + + assert await _routstr_spent_last_24h_sats() == ROUTSTR_MAX_DAILY_TOPUP_SATS + + peer = _peer(0.0) + with _patch_wallet(auto_topup_module, peer, "cashu-token-1"): + await _check_and_topup(row) + + peer.topup.assert_not_awaited() + assert await _claim_state() is None + + +async def test_admin_release_is_fenced_on_the_state_it_reviewed( + patched_db_engine: Any, +) -> None: + row = await _seed_provider() + peer = _peer(0.0, topup={"error": "rejected"}) + with _patch_wallet(auto_topup_module, peer, "cashu-token-1"): + await _check_and_topup(row) + + state = await get_routstr_auto_topup_state(1) + assert state["active"] is True + assert state["phase"] == ROUTSTR_PHASE_SENT + + stale = await release_routstr_auto_topup_state( + 1, state_token="routstr:other:0:sent:0:0" + ) + assert stale.released is False and stale.reason == "stale_state" + + released = await release_routstr_auto_topup_state( + 1, state_token=str(state["state_token"]) + ) + assert released.released is True + + peer.topup.reset_mock() + with _patch_wallet(auto_topup_module, peer, "cashu-token-2"): + await _check_and_topup(row) + peer.topup.assert_awaited_once() + + +async def test_backoff_suppresses_retries_until_its_deadline( + patched_db_engine: Any, +) -> None: + row = await _seed_provider() + peer = _peer(0.0) + + with _patch_wallet(auto_topup_module, peer, "cashu-token-1"): + await _check_and_topup(row) + await _expire_claim() + + # First reconciliation after the lease: the peer never credited, so the + # claim moves to backoff rather than paying again immediately. + peer.topup.reset_mock() + with _patch_wallet(auto_topup_module, peer, "cashu-token-2"): + await _check_and_topup(row) + peer.topup.assert_not_awaited() + + claim = _parse_routstr_request_id((await _claim_state()).request_id) # type: ignore[union-attr] + assert claim is not None and claim.phase == ROUTSTR_PHASE_BACKOFF + assert claim.deadline > int(time.time()) diff --git a/tests/unit/test_auto_topup.py b/tests/unit/test_auto_topup.py index 0db7408f..d28f939c 100644 --- a/tests/unit/test_auto_topup.py +++ b/tests/unit/test_auto_topup.py @@ -3,12 +3,12 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest -from routstr.core.db import CashuTransaction from routstr.upstream.auto_topup import ( _check_and_topup, _parse_ppq_request_id, _run_auto_topup_cycle, validate_ppq_auto_topup_settings, + validate_routstr_auto_topup_settings, ) from routstr.upstream.ppqai import PPQAIUpstreamProvider from routstr.wallet import Bolt11PaymentAmbiguous, Bolt11PaymentNotAttempted @@ -46,81 +46,18 @@ def _row() -> MagicMock: return row -class _Session: - def __init__(self, transaction: CashuTransaction) -> None: - self.transaction = transaction - self.commit = AsyncMock() - - async def __aenter__(self) -> "_Session": - return self - - async def __aexit__(self, *args: object) -> None: - return None - - async def exec(self, query: object) -> MagicMock: - result = MagicMock() - result.first.return_value = self.transaction - return result - - def add(self, transaction: CashuTransaction) -> None: - self.transaction = transaction - - @pytest.mark.asyncio -async def test_auto_topup_persists_before_sending_and_marks_success_collected() -> None: +async def test_auto_topup_refuses_invalid_settings_before_touching_the_wallet() -> None: provider = MagicMock() - provider.get_balance = AsyncMock(return_value=0) - provider.topup = AsyncMock(return_value={"balance": 50}) - transaction = CashuTransaction( - token="cashu-token", amount=50, unit="sat", source="auto_topup" - ) - session = _Session(transaction) - - with ( - patch( - "routstr.upstream.auto_topup.RoutstrUpstreamProvider.from_db_row", - return_value=provider, - ), - patch( - "routstr.upstream.auto_topup.send_token", - AsyncMock(return_value="cashu-token"), - ), - patch( - "routstr.upstream.auto_topup.store_cashu_transaction", - AsyncMock(return_value=True), - ) as store, - patch( - "routstr.upstream.auto_topup.token_mint_url", - return_value="https://fallback-mint.test", - ), - patch("routstr.upstream.auto_topup.create_session", return_value=session), - ): - await _check_and_topup(_row()) - - store.assert_awaited_once_with( - token="cashu-token", - amount=50, - unit="sat", - mint_url="https://fallback-mint.test", - typ="out", - collected=False, - source="auto_topup", - ) - provider.topup.assert_awaited_once_with("cashu-token") - assert transaction.collected is True - session.commit.assert_awaited_once() - - -@pytest.mark.asyncio -@pytest.mark.parametrize("outcome", [{"error": "rejected"}, RuntimeError("network")]) -async def test_auto_topup_failure_leaves_persisted_token_uncollected( - outcome: object, -) -> None: - provider = MagicMock() - provider.get_balance = AsyncMock(return_value=0) - provider.topup = AsyncMock( - side_effect=outcome if isinstance(outcome, Exception) else None, - return_value=outcome, + provider.get_balance = AsyncMock() + row = _row() + row.provider_settings = json.dumps( + { + "auto_topup": True, + "topup_threshold": 100, + "topup_amount_limit": 10**9, + "topup_mint_url": "https://mint.test", + } ) with ( @@ -128,52 +65,12 @@ async def test_auto_topup_failure_leaves_persisted_token_uncollected( "routstr.upstream.auto_topup.RoutstrUpstreamProvider.from_db_row", return_value=provider, ), - patch( - "routstr.upstream.auto_topup.send_token", - AsyncMock(return_value="cashu-token"), - ), - patch( - "routstr.upstream.auto_topup.store_cashu_transaction", - AsyncMock(return_value=True), - ), - patch("routstr.upstream.auto_topup.create_session") as create_session, + patch("routstr.upstream.auto_topup.send_token", AsyncMock()) as send, ): - if isinstance(outcome, Exception): - with pytest.raises(RuntimeError): - await _check_and_topup(_row()) - else: - await _check_and_topup(_row()) + await _check_and_topup(row) - create_session.assert_not_called() - - -@pytest.mark.asyncio -async def test_auto_topup_does_not_send_untracked_token() -> None: - provider = MagicMock() - provider.get_balance = AsyncMock(return_value=0) - provider.topup = AsyncMock() - with ( - patch( - "routstr.upstream.auto_topup.RoutstrUpstreamProvider.from_db_row", - return_value=provider, - ), - patch( - "routstr.upstream.auto_topup.send_token", - AsyncMock(return_value="cashu-token"), - ), - patch( - "routstr.upstream.auto_topup.store_cashu_transaction", - AsyncMock(side_effect=RuntimeError("database unavailable")), - ), - patch( - "routstr.upstream.auto_topup.release_token_reservation", - AsyncMock(), - ) as reclaim, - ): - await _check_and_topup(_row()) - - reclaim.assert_awaited_once_with("cashu-token") - provider.topup.assert_not_awaited() + provider.get_balance.assert_not_awaited() + send.assert_not_awaited() def _ppq_row() -> MagicMock: @@ -734,3 +631,47 @@ def test_ppq_auto_topup_settings_validation_survives_huge_json_integers() -> Non {"auto_topup": True, "topup_threshold": 10**400, "topup_amount_limit": 10} ) assert problem is not None and "threshold" in problem + + +def _routstr_settings(**overrides: object) -> dict: + settings = { + "auto_topup": True, + "topup_threshold": 1, + "topup_amount_limit": 50, + "topup_mint_url": "https://mint.test", + } + settings.update(overrides) + return settings + + +@pytest.mark.parametrize( + ("settings", "expected"), + [ + ({"auto_topup": False, "topup_threshold": -1}, None), + (_routstr_settings(), None), + (_routstr_settings(topup_threshold=None), "threshold"), + (_routstr_settings(topup_threshold=True), "threshold"), + (_routstr_settings(topup_threshold=float("inf")), "threshold"), + (_routstr_settings(topup_amount_limit=0), "positive"), + (_routstr_settings(topup_amount_limit=True), "positive"), + (_routstr_settings(topup_amount_limit=1.5), "whole number"), + (_routstr_settings(topup_amount_limit=10**9), "between"), + (_routstr_settings(topup_mint_url=""), "mint URL"), + (_routstr_settings(topup_mint_url=True), "mint URL"), + ], +) +def test_routstr_auto_topup_settings_validation( + settings: dict, expected: str | None +) -> None: + problem = validate_routstr_auto_topup_settings(settings) + if expected is None: + assert problem is None + else: + assert problem is not None and expected in problem + + +def test_routstr_auto_topup_settings_validation_survives_huge_json_integers() -> None: + problem = validate_routstr_auto_topup_settings( + _routstr_settings(topup_amount_limit=10**400) + ) + assert problem is not None and "positive" in problem