laantungir/routstr-core
1
0
Fork 0
mirror of https://github.com/Routstr/routstr-core.git synced 2026-08-09 02:54:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

ppq-auto-topup

Browse Source
This commit is contained in:
9qeklajc
2026-07-26 02:38:07 +02:00
parent 1d4b8d7cb2
commit 423e2cba73
11 changed files with 2536 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
routstr/core/admin.py
+161 -1
View File
@@ -863,6 +863,32 @@ class UpstreamProviderUpdateBySlug(BaseModel):
provider_settings: dict | None = None
async def _active_ppq_claim_in_session(
session: AsyncSession, provider_id: int
) -> 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
claim = await session.get(
CashuTransaction, _ppq_state_id_for_provider(provider_id)
)
return claim is not None and not claim.collected and not claim.swept
def _require_valid_auto_topup(provider_type: str, settings: dict | None) -> None:
"""Reject auto top-up settings the worker would later refuse to act on."""
from ..upstream.auto_topup import validate_auto_topup_settings
problem = validate_auto_topup_settings(provider_type, settings)
if problem is not None:
raise HTTPException(status_code=400, detail=problem)
async def _apply_provider_update(
session: AsyncSession,
provider: UpstreamProviderRow,
@@ -874,6 +900,23 @@ async def _apply_provider_update(
await _ensure_unique_slug(session, validated, exclude_id=provider.id)
provider.slug = validated
if (
payload.provider_type is not None
and payload.provider_type != provider.provider_type
and provider.provider_type == "ppqai"
and provider.id is not None
and await _active_ppq_claim_in_session(session, provider.id)
):
# Changing the type would orphan the claim: the PPQ endpoints refuse
# non-ppqai providers, so nobody could ever inspect or release it.
raise HTTPException(
status_code=409,
detail=(
"This provider has an active PPQ auto top-up claim. Release "
"it before changing the provider type"
),
)
if payload.provider_type is not None:
provider.provider_type = payload.provider_type
if payload.base_url is not None:
@@ -886,6 +929,22 @@ async def _apply_provider_update(
provider.enabled = payload.enabled
if payload.provider_fee is not None:
provider.provider_fee = payload.provider_fee
# Validate against the effective type and effective settings: a type
# change without new settings must not leave stored settings that the
# worker will refuse, and new settings must fit the new type.
effective_settings = payload.provider_settings
if effective_settings is None and payload.provider_type is not None:
try:
effective_settings = (
json.loads(provider.provider_settings)
if provider.provider_settings
else None
)
except (json.JSONDecodeError, TypeError):
effective_settings = None
if effective_settings is not None:
_require_valid_auto_topup(provider.provider_type, effective_settings)
if payload.provider_settings is not None:
provider.provider_settings = json.dumps(payload.provider_settings)
@@ -925,6 +984,8 @@ async def create_upstream_provider(
else:
slug = await allocate_unique_provider_slug(session, payload.provider_type)
_require_valid_auto_topup(payload.provider_type, payload.provider_settings)
provider = UpstreamProviderRow(
slug=slug,
provider_type=payload.provider_type,
@@ -1013,6 +1074,25 @@ async def delete_upstream_provider(provider_id: str) -> dict[str, object]:
async with create_session() as session:
provider = await _get_upstream_provider_by_ref(session, provider_id)
deleted_id = _provider_pk(provider)
# Checked inside the delete transaction: the worker's claim creation
# 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
):
# Deleting now would orphan the claim and any funds it tracks:
# the PPQ 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. "
"Resolve and release it before deleting the provider"
),
)
await session.delete(provider)
await session.commit()
await reinitialize_upstreams()
@@ -1621,6 +1701,83 @@ async def get_log_dates_api(request: Request) -> dict[str, object]:
return {"dates": dates}
_PPQ_RELEASE_ERRORS = {
"no_active_claim": "No active PPQ claim to release",
"stale_state": (
"The claim changed since it was reviewed; reload and check again"
),
"payment_in_flight": (
"A Lightning payment is still in flight for this claim. Wait for it to "
"finish or expire before releasing"
),
"claim_changed": (
"The claim changed while the release was being applied; reload and "
"check again"
),
}
class ReleasePPQAutoTopupRequest(BaseModel):
confirmed_safe_to_retry: bool
# Echoes the state_token the admin reviewed — the claim's full versioned
# state, not just its operation id. Any change since the review (a new
# attempt, a phase change, a renewed lease) fails the match, so the
# release cannot land on a state the admin never saw.
state_token: str | None = None
async def _require_ppq_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 != "ppqai":
raise HTTPException(status_code=400, detail="Provider is not PPQ")
return provider
@admin_router.get(
"/api/upstream-providers/{provider_id}/ppq-auto-topup",
dependencies=[Depends(require_admin_api)],
)
async def get_ppq_auto_topup_api(provider_id: int) -> dict[str, object]:
await _require_ppq_provider(provider_id)
from ..upstream.auto_topup import get_ppq_auto_topup_state
return {"ok": True, **await get_ppq_auto_topup_state(provider_id)}
@admin_router.post(
"/api/upstream-providers/{provider_id}/ppq-auto-topup/release",
dependencies=[Depends(require_admin_api)],
)
async def release_ppq_auto_topup_api(
provider_id: int, payload: ReleasePPQAutoTopupRequest
) -> dict[str, object]:
await _require_ppq_provider(provider_id)
if not payload.confirmed_safe_to_retry:
raise HTTPException(
status_code=400,
detail="Confirm the Lightning payment outcome is safe before releasing",
)
from ..upstream.auto_topup import release_ppq_auto_topup_state
outcome = await release_ppq_auto_topup_state(
provider_id, state_token=payload.state_token
)
if not outcome.released:
raise HTTPException(
status_code=409, detail=_PPQ_RELEASE_ERRORS[outcome.reason]
)
logger.warning(
"Admin released PPQ auto top-up claim after manual reconciliation",
extra={"provider_id": provider_id, "state_token": payload.state_token},
)
return {"ok": True, "released": True}
@admin_router.get("/api/transactions", dependencies=[Depends(require_admin_api)])
async def get_transactions_api(
type: str | None = None,
@@ -1633,7 +1790,10 @@ async def get_transactions_api(
async with create_session() as session:
from sqlmodel import col, func
base = select(CashuTransaction)
base = select(CashuTransaction).where(
(CashuTransaction.source != "ppq_auto_topup")
| (CashuTransaction.source == None) # noqa: E711
)
if type:
base = base.where(CashuTransaction.type == type)
if source:
routstr/upstream/auto_topup.py
+781 -88
View File
@@ -1,35 +1,50 @@
import asyncio
import json
import math
import time
import typing
import uuid
from sqlmodel import select
from sqlalchemy.exc import IntegrityError
from sqlmodel import col, or_, select, update
from ..core import get_logger
from ..core.db import (
CashuTransaction,
UpstreamProviderRow,
create_session,
)
from ..core.db import CashuTransaction, UpstreamProviderRow, create_session
from ..core.db import (
store_cashu_transaction_with_retry as store_cashu_transaction,
)
from ..wallet import send_token
from ..payment.price import sats_usd_price
from ..wallet import (
Bolt11PaymentNotAttempted,
check_bolt11_payment_status,
execute_bolt11_payment,
prepare_bolt11_payment,
send_token,
)
from .ppqai import PPQAIUpstreamProvider
from .routstr import RoutstrUpstreamProvider
logger = get_logger(__name__)
# Check every 60 seconds
AUTO_TOPUP_INTERVAL_SECONDS = 60
# Claim lifecycle. "claimed" holds the slot while the invoice is being created
# and priced; nothing has been spent yet, so it is always safe to release.
# "in_flight" means proofs are committed to a mint and the outcome is unknown.
# "reconcile" means the worker gave up and an admin must decide.
PPQ_PHASE_CLAIMED = "claimed"
PPQ_PHASE_IN_FLIGHT = "in_flight"
PPQ_PHASE_RECONCILE = "reconcile"
PPQ_PHASES = frozenset({PPQ_PHASE_CLAIMED, PPQ_PHASE_IN_FLIGHT, PPQ_PHASE_RECONCILE})
PPQ_SETTLEMENT_ATTEMPTS = 5
PPQ_SETTLEMENT_POLL_SECONDS = 2
PPQ_PENDING_TTL_SECONDS = 15 * 60
PPQ_MAX_INVOICE_PREMIUM = 1.10
PPQ_MIN_TOPUP_USD = 1
PPQ_MAX_TOPUP_USD = 500
async def periodic_auto_topup() -> None:
"""Background task that monitors Routstr provider balances and auto-tops up when below threshold.
For each Routstr provider with auto_topup enabled in provider_settings:
1. Checks the upstream balance via get_balance()
2. If balance < topup_threshold, creates a cashu token from the configured mint
3. Sends the token to the upstream provider via topup()
"""
# Wait for initial startup to complete
"""Monitor enabled Routstr and PPQ providers and fund low balances."""
await asyncio.sleep(30)
logger.info("Auto top-up worker started")
@@ -41,19 +56,19 @@ async def periodic_auto_topup() -> None:
"Auto top-up cycle failed",
extra={"error": str(e), "error_type": type(e).__name__},
)
await asyncio.sleep(AUTO_TOPUP_INTERVAL_SECONDS)
async def _run_auto_topup_cycle() -> None:
"""Single cycle: check all eligible providers and top up if needed."""
"""Check all eligible providers once, isolating failures by provider."""
await _reconcile_all_ppq_claims()
async with create_session() as session:
query = select(UpstreamProviderRow).where(
UpstreamProviderRow.provider_type == "routstr",
col(UpstreamProviderRow.provider_type).in_(["routstr", "ppqai"]),
UpstreamProviderRow.enabled == True, # noqa: E712
)
result = await session.exec(query)
providers = result.all()
providers = (await session.exec(query)).all()
for row in providers:
try:
@@ -63,6 +78,7 @@ async def _run_auto_topup_cycle() -> None:
"Auto top-up failed for provider",
extra={
"provider_id": row.id,
"provider_type": row.provider_type,
"base_url": row.base_url,
"error": str(e),
"error_type": type(e).__name__,
@@ -70,62 +86,156 @@ async def _run_auto_topup_cycle() -> None:
)
async def _check_and_topup(row: UpstreamProviderRow) -> None:
"""Check a single provider's balance and top up if below threshold."""
# Parse provider settings
async def _reconcile_all_ppq_claims() -> None:
"""Reconcile every active PPQ claim, ignoring top-up eligibility.
A claim tracks money already committed, so it must keep reconciling even
after the provider is disabled, auto top-up is switched off, or the API
key is removed — none of which change what happened to the payment. Only
new top-ups depend on eligibility.
"""
async with create_session() as session:
rows = (
await session.exec(
select(UpstreamProviderRow).where(
col(UpstreamProviderRow.provider_type) == "ppqai"
)
)
).all()
for row in rows:
try:
if row.id is None:
continue
state = await get_ppq_auto_topup_state(row.id)
if not state.get("active"):
continue
# Without an API key the PPQ side cannot be polled, but mint
# reconciliation still can — pass provider as None.
provider = (
PPQAIUpstreamProvider.from_db_row(row) if row.api_key else None
)
await _reconcile_ppq_state(row, provider)
except Exception as e:
logger.error(
"PPQ claim reconciliation failed",
extra={"provider_id": row.id, "error": str(e)},
)
def _invalid_number(value: object, *, integer: bool = False) -> bool:
if isinstance(value, bool) or not isinstance(value, (int, float)):
return True
try:
# JSON accepts arbitrarily large integers; float() raises
# OverflowError past ~1e308 rather than returning inf.
as_float = float(value)
except OverflowError:
return True
if not math.isfinite(as_float) or value <= 0:
return True
return integer and not as_float.is_integer()
def validate_auto_topup_settings(
provider_type: str, settings: dict | None
) -> str | None:
"""Return why these auto top-up settings are unusable, or None if they are.
Shared by the admin write path and the worker so a configuration that the
UI accepted cannot silently fail to top up hours later. Settings with
auto top-up switched off are always valid.
"""
if not settings or not settings.get("auto_topup"):
return None
if _invalid_number(settings.get("topup_threshold")):
return "Auto top-up threshold must be a positive number"
if _invalid_number(settings.get("topup_amount_limit"), integer=True):
return "Auto top-up amount must be a positive whole number"
if provider_type == "routstr":
mint_url = settings.get("topup_mint_url")
if not isinstance(mint_url, str) or not mint_url.strip():
return "Auto top-up requires a mint URL"
elif provider_type == "ppqai":
amount = int(settings["topup_amount_limit"])
if not PPQ_MIN_TOPUP_USD <= amount <= PPQ_MAX_TOPUP_USD:
return (
f"PPQ auto top-up amount must be between {PPQ_MIN_TOPUP_USD} "
f"and {PPQ_MAX_TOPUP_USD} USD"
)
return None
def _get_auto_topup_settings(row: UpstreamProviderRow) -> dict | None:
settings: dict = {}
if row.provider_settings:
try:
settings = json.loads(row.provider_settings)
except (json.JSONDecodeError, TypeError):
return
return None
if not settings.get("auto_topup"):
if not isinstance(settings, dict) or not settings.get("auto_topup"):
return None
# Re-checked here even though the admin API validates on write: rows
# predate that check, and the database is not the only way in.
problem = validate_auto_topup_settings(row.provider_type, settings)
if problem is not None:
logger.warning(
"Auto top-up enabled but its configuration is invalid",
extra={"provider_id": row.id, "problem": problem},
)
return None
return settings
async def _check_and_topup(row: UpstreamProviderRow) -> None:
"""Dispatch one provider row to its funding strategy."""
settings = _get_auto_topup_settings(row)
if settings is None or not row.api_key:
return
threshold = settings.get("topup_threshold")
amount = settings.get("topup_amount_limit")
mint_url = settings.get("topup_mint_url")
if row.provider_type == "routstr":
await _check_and_topup_routstr(row, settings)
elif row.provider_type == "ppqai":
await _check_and_topup_ppq(row, settings)
if not threshold or not amount or not mint_url:
async def _check_and_topup_routstr(
row: UpstreamProviderRow, settings: dict
) -> None:
threshold = float(settings["topup_threshold"])
amount = int(settings["topup_amount_limit"])
mint_url = settings.get("topup_mint_url")
if not isinstance(mint_url, str) or not mint_url:
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),
},
"Routstr auto top-up enabled but no mint is configured",
extra={"provider_id": row.id},
)
return
if not row.api_key:
return
# Instantiate provider and check balance
provider = RoutstrUpstreamProvider.from_db_row(row)
if provider is None:
return
balance = await provider.get_balance()
if balance is None:
logger.warning(
"Could not fetch balance for auto top-up",
"Could not fetch Routstr balance for auto top-up",
extra={"provider_id": row.id, "base_url": row.base_url},
)
return
if balance >= threshold * 1000:
if balance >= threshold:
return
# Balance is below threshold - create token and top up
logger.info(
"Auto top-up triggered",
"Routstr auto top-up triggered",
extra={
"provider_id": row.id,
"balance": balance,
"threshold": threshold,
"topup_amount": amount,
"balance_sats": balance,
"threshold_sats": threshold,
"topup_sats": amount,
"mint_url": mint_url,
},
)
@@ -134,13 +244,8 @@ async def _check_and_topup(row: UpstreamProviderRow) -> None:
token = await send_token(amount, "sat", mint_url)
except Exception as e:
logger.error(
"Failed to create cashu token for auto top-up",
extra={
"provider_id": row.id,
"amount": amount,
"mint_url": mint_url,
"error": str(e),
},
"Failed to create Cashu token for Routstr auto top-up",
extra={"provider_id": row.id, "mint_url": mint_url, "error": str(e)},
)
return
@@ -156,47 +261,635 @@ async def _check_and_topup(row: UpstreamProviderRow) -> None:
)
except Exception:
logger.critical(
"Aborting auto top-up because its cashu token could not be persisted",
"Aborting Routstr auto top-up because its token was not persisted",
extra={"provider_id": row.id, "mint_url": mint_url},
)
return
result = await provider.topup(token)
if "error" in result:
logger.error(
"Auto top-up upstream call failed",
extra={
"provider_id": row.id,
"error": result["error"],
},
"Routstr auto top-up call failed",
extra={"provider_id": row.id, "error": result["error"]},
)
else:
async with create_session() as session:
transaction = (
await session.exec(
select(CashuTransaction).where(
CashuTransaction.token == token,
CashuTransaction.type == "out",
CashuTransaction.source == "auto_topup",
)
)
).first()
if transaction is None:
logger.critical(
"Completed auto top-up transaction is missing from the database",
extra={"provider_id": row.id, "mint_url": mint_url},
)
else:
transaction.collected = True
session.add(transaction)
await session.commit()
return
logger.info(
"Auto top-up completed successfully",
await _mark_routstr_topup_collected(row, token, mint_url)
logger.info(
"Routstr auto top-up completed",
extra={"provider_id": row.id, "amount_sats": amount},
)
async def _mark_routstr_topup_collected(
row: UpstreamProviderRow, token: str, mint_url: str
) -> None:
async with create_session() as session:
transaction = (
await session.exec(
select(CashuTransaction).where(
CashuTransaction.token == token,
CashuTransaction.type == "out",
CashuTransaction.source == "auto_topup",
)
)
).first()
if transaction is None:
logger.critical(
"Completed Routstr auto top-up is missing from the database",
extra={"provider_id": row.id, "mint_url": mint_url},
)
return
transaction.collected = True
session.add(transaction)
await session.commit()
def _ppq_state_id(row: UpstreamProviderRow) -> str:
if row.id is None:
raise ValueError("PPQ auto top-up requires a persisted provider row")
return _ppq_state_id_for_provider(row.id)
def _ppq_state_id_for_provider(provider_id: int | str) -> str:
return f"ppq-auto-topup-{provider_id}"
class PPQClaim(typing.NamedTuple):
operation_id: str
# Worker lease, not the BOLT11 invoice expiry: the invoice can expire
# while melt() is still running, and only the lease says whether the
# owning worker can still be alive.
lease_expires_at: int
phase: str
invoice_id: str
# Cashu melt quote id, "none" until a payment plan exists. This is what
# lets an ambiguous payment be reconciled against the mint later.
quote_id: str
def _ppq_request_id(
operation_id: str,
lease_expires_at: int,
phase: str,
invoice_id: str,
quote_id: str = "none",
) -> str:
return f"ppq:{operation_id}:{lease_expires_at}:{phase}:{invoice_id}:{quote_id}"
def _parse_ppq_request_id(request_id: str | None) -> PPQClaim | None:
parts = (request_id or "").split(":", 5)
if len(parts) != 6 or parts[0] != "ppq" or parts[3] not in PPQ_PHASES:
return None
try:
lease_expires_at = int(parts[2])
except (TypeError, ValueError):
return None
return PPQClaim(parts[1], lease_expires_at, parts[3], parts[4], parts[5])
def _ppq_claim_is_releasable(claim: PPQClaim | None, now: float) -> bool:
"""Whether an admin may sweep this claim without risking a double payment.
A claim in ``in_flight`` is owned by a worker that is somewhere between
reserving proofs and hearing back from the mint. Releasing it there frees
the next cycle to pay a second invoice, which is exactly what the claim
exists to prevent — so it is releasable only once its lease has passed,
which means the owning worker died rather than that it is still working.
"""
if claim is None:
# Corrupt state cannot be reasoned about and has no operation id to
# fence on. Leaving it unreleasable would strand the provider forever.
return True
if claim.phase == PPQ_PHASE_IN_FLIGHT:
return now >= claim.lease_expires_at
return True
async def get_ppq_auto_topup_state(provider_id: int) -> dict[str, object]:
"""Return admin-safe state for a provider's durable PPQ claim."""
async with create_session() as session:
transaction = await session.get(
CashuTransaction, _ppq_state_id_for_provider(provider_id)
)
if transaction is None or transaction.collected or transaction.swept:
return {"active": False}
claim = _parse_ppq_request_id(transaction.request_id)
return {
"active": True,
# The exact version of the claim the admin is looking at. A release
# must echo it back verbatim: the operation id alone is stable across
# phase changes, so it cannot distinguish "the state I reviewed" from
# "the same attempt after its payment turned ambiguous".
"state_token": transaction.request_id,
"operation_id": claim.operation_id if claim else None,
"phase": claim.phase if claim else None,
"releasable": _ppq_claim_is_releasable(claim, time.time()),
"expires_at": claim.lease_expires_at if claim else None,
"invoice_id": (
claim.invoice_id if claim and claim.invoice_id != "pending" else None
),
"created_at": transaction.created_at,
"amount": transaction.amount,
"unit": transaction.unit,
"mint_url": transaction.mint_url,
"malformed": claim is None,
}
class PPQReleaseOutcome(typing.NamedTuple):
released: bool
reason: str
async def release_ppq_auto_topup_state(
provider_id: int, *, state_token: str | None
) -> PPQReleaseOutcome:
"""Force-release an active claim after an admin reconciles payment status.
``state_token`` is the ``state_token`` the caller read from
:func:`get_ppq_auto_topup_state` — the claim's full ``request_id``. Two
things are enforced with it. The claim must not be inside an unexpired
``in_flight`` lease, because a worker between reserving proofs and hearing
from the mint still owns the outcome. And the row must be byte-identical
to the one the caller reviewed: the update fences on the whole token, so
any phase change, lease renewal, or new attempt since the review fails the
write instead of sweeping a state the admin never saw.
"""
state_id = _ppq_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 PPQReleaseOutcome(False, "no_active_claim")
if transaction.request_id != state_token:
return PPQReleaseOutcome(False, "stale_state")
claim = _parse_ppq_request_id(transaction.request_id)
if not _ppq_claim_is_releasable(claim, time.time()):
return PPQReleaseOutcome(False, "payment_in_flight")
async with create_session() as session:
result = await session.exec( # type: ignore[call-overload]
update(CashuTransaction)
.where(
col(CashuTransaction.id) == state_id,
col(CashuTransaction.source) == "ppq_auto_topup",
# Fence on the token itself, not the row read above: a change
# between the read and this write must lose the race.
col(CashuTransaction.request_id) == state_token,
col(CashuTransaction.collected) == False, # noqa: E712
col(CashuTransaction.swept) == False, # noqa: E712
)
.values(swept=True)
)
await session.commit()
if (getattr(result, "rowcount", 0) or 0) == 1:
return PPQReleaseOutcome(True, "released")
return PPQReleaseOutcome(False, "claim_changed")
async def _set_ppq_state_terminal(
row: UpstreamProviderRow,
operation_id: str,
*,
collected: bool,
swept: bool,
) -> bool:
"""Finish a PPQ 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) == _ppq_state_id(row),
col(CashuTransaction.request_id).like(f"ppq:{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 _reconcile_ppq_state(
row: UpstreamProviderRow, provider: PPQAIUpstreamProvider | None
) -> bool:
"""Return True while a prior PPQ attempt must suppress a new payment.
``provider`` may be ``None`` when the API key is gone: PPQ settlement
cannot be polled then, but mint-side reconciliation still runs.
"""
async with create_session() as session:
transaction = await session.get(CashuTransaction, _ppq_state_id(row))
if transaction is None or transaction.collected or transaction.swept:
return False
claim = _parse_ppq_request_id(transaction.request_id)
if claim is None:
logger.critical(
"Malformed PPQ auto top-up state; suppressing duplicate payment",
extra={"provider_id": row.id},
)
return True
if claim.invoice_id != "pending":
if provider is not None and await provider.check_topup_status(
claim.invoice_id
):
if not await _set_ppq_state_terminal(
row, claim.operation_id, collected=True, swept=False
):
# An admin release won the race against a settlement that
# turned out to have succeeded. The next cycle may pay again;
# the balance check is the only remaining guard, so shout.
logger.critical(
"PPQ invoice settled but its claim was already released; "
"a duplicate top-up is possible on the next cycle",
extra={
"provider_id": row.id,
"invoice_id": claim.invoice_id,
},
)
return True
# PPQ has not credited the invoice. Ask the mint what became of the
# melt — the durable reconciliation path for a payment whose worker
# died or whose melt call never returned. cashu settles the wallet
# database as a side effect: "unpaid" releases the reserved proofs.
if (
claim.quote_id != "none"
and transaction.mint_url
and time.time() >= claim.lease_expires_at
):
status = await check_bolt11_payment_status(
transaction.mint_url, transaction.unit, claim.quote_id
)
if status == "unpaid":
# Provably never paid, funds recovered — safe to retry.
released = await _set_ppq_state_terminal(
row, claim.operation_id, collected=False, swept=True
)
if released:
logger.warning(
"PPQ auto top-up melt was never paid; claim released",
extra={
"provider_id": row.id,
"invoice_id": claim.invoice_id,
},
)
return not released
# "paid" means the mint paid but PPQ has not credited yet: keep
# waiting on PPQ. "pending"/"unknown" stay locked for the admin.
return True
if time.time() < claim.lease_expires_at:
return True
released = await _set_ppq_state_terminal(
row, claim.operation_id, collected=False, swept=True
)
return not released
async def _ppq_provider_is_claimable(
session: object, provider_id: int | None
) -> bool:
"""Re-read the provider inside the claim transaction.
SQLite serialises write transactions, so checking here — rather than
trusting the row the cycle loaded earlier — means a concurrent provider
deletion or type change either commits before us (we see it and refuse)
or after us (its own claim check sees our claim and refuses). Without
this the worker could create a claim for a provider that no longer
exists, orphaning it forever.
"""
if provider_id is None:
return False
current = await session.get(UpstreamProviderRow, provider_id) # type: ignore[attr-defined]
return current is not None and current.provider_type == "ppqai"
async def _claim_ppq_topup(row: UpstreamProviderRow) -> str | None:
"""Acquire a durable, ownership-fenced per-provider claim."""
state_id = _ppq_state_id(row)
operation_id = uuid.uuid4().hex
expires_at = int(time.time()) + PPQ_PENDING_TTL_SECONDS
request_id = _ppq_request_id(
operation_id, expires_at, PPQ_PHASE_CLAIMED, "pending"
)
async with create_session() as session:
if not await _ppq_provider_is_claimable(session, row.id):
return None
existing = await session.get(CashuTransaction, state_id)
if existing is not None:
result = await session.exec( # type: ignore[call-overload]
update(CashuTransaction)
.where(
col(CashuTransaction.id) == state_id,
or_(
CashuTransaction.collected == True, # noqa: E712
CashuTransaction.swept == True, # noqa: E712
),
)
.values(
token="pending",
amount=0,
unit="sat",
mint_url=None,
request_id=request_id,
collected=False,
swept=False,
created_at=int(time.time()),
)
)
await session.commit()
if (getattr(result, "rowcount", 0) or 0) != 1:
return None
return operation_id
try:
async with create_session() as session:
# Same fencing as the update path: the provider must still exist
# inside the transaction that creates the claim.
if not await _ppq_provider_is_claimable(session, row.id):
return None
session.add(
CashuTransaction(
id=state_id,
token="pending",
amount=0,
unit="sat",
type="out",
request_id=request_id,
collected=False,
source="ppq_auto_topup",
)
)
await session.commit()
except IntegrityError:
return None
return operation_id
async def _record_ppq_invoice(
row: UpstreamProviderRow,
operation_id: str,
*,
invoice: str,
invoice_id: str,
quote_id: str,
amount: int,
unit: str,
mint_url: str,
) -> int:
"""Move the claim to in_flight and return its fresh worker lease.
The lease is minted here, at the start of the payment, and deliberately
not derived from the BOLT11 invoice's expiry: the invoice can expire
while melt() is still running, and the lease answers a different question
— can the worker that owns this claim still be alive?
"""
lease_expires_at = int(time.time()) + PPQ_PENDING_TTL_SECONDS
async with create_session() as session:
result = await session.exec( # type: ignore[call-overload]
update(CashuTransaction)
.where(
col(CashuTransaction.id) == _ppq_state_id(row),
col(CashuTransaction.request_id).like(
f"ppq:{operation_id}:%:{PPQ_PHASE_CLAIMED}:pending:none"
),
col(CashuTransaction.collected) == False, # noqa: E712
col(CashuTransaction.swept) == False, # noqa: E712
)
.values(
token=invoice,
# Moves the claim to in_flight: from here the proofs are
# committed and an admin may not release it until the lease
# runs out.
request_id=_ppq_request_id(
operation_id,
lease_expires_at,
PPQ_PHASE_IN_FLIGHT,
invoice_id,
quote_id,
),
amount=amount,
unit=unit,
mint_url=mint_url,
)
)
await session.commit()
if (getattr(result, "rowcount", 0) or 0) != 1:
raise RuntimeError("PPQ auto top-up claim ownership was lost")
return lease_expires_at
async def _mark_ppq_reconcile(
row: UpstreamProviderRow,
operation_id: str,
lease_expires_at: int,
invoice_id: str,
quote_id: str,
) -> None:
"""Move an in_flight claim to reconcile so an admin may release it.
Without this an ambiguous payment stays in_flight, and in_flight is only
releasable once its lease expires — the admin would have to wait out the
lease before they could act on an alert that already fired.
"""
async with create_session() as session:
result = await session.exec( # type: ignore[call-overload]
update(CashuTransaction)
.where(
col(CashuTransaction.id) == _ppq_state_id(row),
col(CashuTransaction.request_id)
== _ppq_request_id(
operation_id,
lease_expires_at,
PPQ_PHASE_IN_FLIGHT,
invoice_id,
quote_id,
),
col(CashuTransaction.collected) == False, # noqa: E712
col(CashuTransaction.swept) == False, # noqa: E712
)
.values(
request_id=_ppq_request_id(
operation_id,
lease_expires_at,
PPQ_PHASE_RECONCILE,
invoice_id,
quote_id,
)
)
)
await session.commit()
if (getattr(result, "rowcount", 0) or 0) != 1:
logger.warning(
"Could not flag the PPQ claim for reconciliation; "
"it is no longer owned by this attempt",
extra={"provider_id": row.id, "invoice_id": invoice_id},
)
async def _check_and_topup_ppq(row: UpstreamProviderRow, settings: dict) -> None:
threshold_usd = float(settings["topup_threshold"])
amount_usd = int(settings["topup_amount_limit"])
provider = PPQAIUpstreamProvider.from_db_row(row)
if provider is None or await _reconcile_ppq_state(row, provider):
return
balance = await provider.get_balance()
if balance is None or not math.isfinite(balance) or balance < 0:
logger.warning(
"Could not fetch a valid PPQ balance for auto top-up",
extra={"provider_id": row.id},
)
return
if balance >= threshold_usd:
return
operation_id = await _claim_ppq_topup(row)
if operation_id is None:
return
logger.info(
"PPQ auto top-up triggered",
extra={
"provider_id": row.id,
"balance_usd": balance,
"threshold_usd": threshold_usd,
"topup_usd": amount_usd,
},
)
try:
topup = await provider.initiate_topup(amount_usd)
if topup.currency.upper() != "USD" or topup.amount != amount_usd:
raise ValueError("PPQ top-up response amount or currency does not match")
now = int(time.time())
# The invoice's own expiry is a pre-payment sanity check only; the
# claim's lease is minted separately in _record_ppq_invoice.
invoice_expires_at = topup.expires_at or now + PPQ_PENDING_TTL_SECONDS
if invoice_expires_at > 10**12:
invoice_expires_at //= 1000
if invoice_expires_at <= now:
raise ValueError("PPQ returned an expired Lightning invoice")
plan = await prepare_bolt11_payment(topup.payment_request)
max_invoice_sats = math.ceil(
amount_usd / sats_usd_price() * PPQ_MAX_INVOICE_PREMIUM
)
if plan.maximum_spend_sats > max_invoice_sats:
raise ValueError("PPQ Lightning invoice exceeds the USD spending cap")
lease_expires_at = await _record_ppq_invoice(
row,
operation_id,
invoice=topup.payment_request,
invoice_id=topup.invoice_id,
quote_id=str(plan.quote.quote),
amount=int(plan.quote.amount + plan.quote.fee_reserve),
unit=plan.unit,
mint_url=plan.mint_url,
)
except Exception:
# Nothing has been paid yet, so the claim can be handed back. If the
# release does not land, the claim is no longer ours to reason about.
if not await _set_ppq_state_terminal(
row, operation_id, collected=False, swept=True
):
logger.warning(
"Could not release the PPQ auto top-up claim after a "
"pre-payment failure; it is owned by another attempt",
extra={"provider_id": row.id},
)
raise
try:
paid_amount, mint_url, unit = await execute_bolt11_payment(plan)
except Bolt11PaymentNotAttempted:
# The mint's own answer rules out a settlement and any reserved proofs
# were handed back, so this claim is safe to retry next cycle.
if not await _set_ppq_state_terminal(
row, operation_id, collected=False, swept=True
):
logger.warning(
"Could not release the PPQ auto top-up claim after a payment "
"that was never attempted; it is owned by another attempt",
extra={"provider_id": row.id},
)
logger.warning(
"PPQ Lightning payment was not attempted; claim released for retry",
extra={"provider_id": row.id, "invoice_id": topup.invoice_id},
exc_info=True,
)
raise
except Exception:
await _mark_ppq_reconcile(
row,
operation_id,
lease_expires_at,
topup.invoice_id,
str(plan.quote.quote),
)
logger.critical(
"PPQ auto top-up payment outcome is ambiguous; claim remains locked until admin reconciliation",
extra={
"provider_id": row.id,
"amount": amount,
"new_balance_approx": balance + amount,
"invoice_id": topup.invoice_id,
"admin_action": f"POST /admin/api/upstream-providers/{row.id}/ppq-auto-topup/release",
},
exc_info=True,
)
raise
settled = False
for attempt in range(PPQ_SETTLEMENT_ATTEMPTS):
if await provider.check_topup_status(topup.invoice_id):
settled = True
break
if attempt + 1 < PPQ_SETTLEMENT_ATTEMPTS:
await asyncio.sleep(PPQ_SETTLEMENT_POLL_SECONDS)
if not settled:
# The payment left the wallet, so the claim must stay. Flag it for
# reconciliation: _reconcile_ppq_state keeps polling PPQ, and an admin
# can step in without waiting out the lease.
await _mark_ppq_reconcile(
row,
operation_id,
lease_expires_at,
topup.invoice_id,
str(plan.quote.quote),
)
logger.critical(
"PPQ Lightning payment completed but credit settlement is unconfirmed",
extra={"provider_id": row.id, "invoice_id": topup.invoice_id},
)
return
if not await _set_ppq_state_terminal(
row, operation_id, collected=True, swept=False
):
# Something else finished this claim while the payment was in flight —
# an admin release, most likely. The next cycle is now free to pay
# again, so surface it rather than completing quietly.
logger.critical(
"PPQ auto top-up settled but its claim was already released; "
"a duplicate top-up is possible on the next cycle",
extra={"provider_id": row.id, "invoice_id": topup.invoice_id},
)
logger.info(
"PPQ auto top-up completed",
extra={
"provider_id": row.id,
"topup_usd": amount_usd,
"cashu_paid": paid_amount,
"cashu_unit": unit,
"mint_url": mint_url,
},
)
routstr/upstream/ppqai.py
+1 -1
View File
@@ -441,7 +441,7 @@ class PPQAIUpstreamProvider(BaseUpstreamProvider):
"""
data = await self.check_balance()
balance = data.get("balance")
if isinstance(balance, (int, float)):
if isinstance(balance, (int, float)) and not isinstance(balance, bool):
return float(balance)
return None
routstr/wallet.py
+203 -1
View File
@@ -3,10 +3,11 @@ import re
import socket
import time
import typing
from dataclasses import dataclass
from typing import TypedDict
import httpx
from cashu.core.base import Proof, Token
from cashu.core.base import MeltQuote, Proof, Token
from cashu.core.mint_info import MintInfo as _CashuMintInfo
from cashu.wallet.helpers import deserialize_token_from_string
from cashu.wallet.wallet import Wallet
@@ -282,6 +283,203 @@ async def send_token(amount: int, unit: str, mint_url: str | None = None) -> str
return token
class Bolt11PaymentNotAttempted(Exception):
"""The invoice was definitively not paid, so the attempt can be retried.
Raised only where the mint's own answer rules out a settlement: coin
selection never reached ``melt``, or ``melt`` returned an explicit unpaid
state. Any proofs reserved along the way are released before this is
raised.
"""
class Bolt11PaymentAmbiguous(Exception):
"""The payment may or may not have settled, so it must not be retried.
Raised when ``melt`` errored, timed out, or came back pending. The selected
proofs stay reserved: the mint may still complete the payment with them,
and spending them elsewhere would be a double spend.
"""
@dataclass
class Bolt11PaymentPlan:
invoice: str
wallet: Wallet
proofs: list[Proof]
quote: MeltQuote
mint_url: str
unit: str
@property
def invoice_amount_sats(self) -> int:
amount = int(self.quote.amount)
return amount if self.unit == "sat" else (amount + 999) // 1000
@property
def maximum_spend_sats(self) -> int:
maximum = (
int(self.quote.amount)
+ int(self.quote.fee_reserve)
+ int(self.wallet.get_fees_for_proofs(self.proofs))
)
return maximum if self.unit == "sat" else (maximum + 999) // 1000
async def prepare_bolt11_payment(invoice: str) -> Bolt11PaymentPlan:
"""Choose the sufficiently funded configured mint with most balance.
Candidate discovery only reads balances and melt quotes. Coin selection,
which may split proofs, is deferred until the winning mint is known.
"""
mint_urls = list(dict.fromkeys([*settings.cashu_mints, settings.primary_mint]))
candidates: list[tuple[int, Wallet, list[Proof], MeltQuote, str, str]] = []
for mint_url in mint_urls:
if not mint_url:
continue
for unit in ("sat", "msat"):
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)
balance = sum(proof.amount for proof in proofs)
if balance <= 0:
continue
quote = await wallet.melt_quote(invoice=invoice)
# select_to_send runs with include_fees=True, so the input fee
# has to be part of sufficiency too. Without it a mint passes
# this filter and then fails coin selection.
required = (
quote.amount
+ quote.fee_reserve
+ wallet.get_fees_for_proofs(proofs)
)
if balance < required:
continue
balance_msats = balance * 1000 if unit == "sat" else balance
candidates.append(
(balance_msats, wallet, proofs, quote, mint_url, unit)
)
except Exception as e:
logger.debug(
"Cashu mint cannot fund BOLT11 invoice",
extra={"mint_url": mint_url, "unit": unit, "error": str(e)},
)
if not candidates:
raise ValueError("No configured Cashu mint has enough balance to pay invoice")
_, wallet, proofs, quote, mint_url, unit = max(
candidates, key=lambda item: item[0]
)
return Bolt11PaymentPlan(invoice, wallet, proofs, quote, mint_url, unit)
async def execute_bolt11_payment(plan: Bolt11PaymentPlan) -> tuple[int, str, str]:
"""Execute a prepared payment, separating retryable from ambiguous failure.
Raises ``Bolt11PaymentNotAttempted`` when the invoice provably did not
settle, and ``Bolt11PaymentAmbiguous`` when the outcome is unknown. Callers
may safely retry the first and must never retry the second.
"""
# Select unreserved, mirroring send_token: a selection failure must not
# strand proofs that were never handed to the mint.
try:
selected, _ = await plan.wallet.select_to_send(
plan.proofs,
plan.quote.amount + plan.quote.fee_reserve,
set_reserved=False,
include_fees=True,
)
except Exception as e:
raise Bolt11PaymentNotAttempted(f"Coin selection failed: {e}") from e
await plan.wallet.set_reserved_for_send(selected, reserved=True)
try:
result = await asyncio.wait_for(
plan.wallet.melt(
proofs=selected,
invoice=plan.invoice,
fee_reserve_sat=plan.quote.fee_reserve,
quote_id=plan.quote.quote,
),
timeout=60,
)
except Exception as e:
# The mint may still be settling with these proofs, so they must stay
# reserved — but cashu's melt() un-reserves them itself on a mint
# transport error, the exact ambiguous case. Re-reserve with the melt
# quote id, not as a send: get_melt_quote() finds the proofs to settle
# by melt_id, so a send-style reservation would strand them — paid
# proofs never invalidated, unpaid ones never released.
try:
await plan.wallet.set_reserved_for_melt(
selected, reserved=True, quote_id=plan.quote.quote
)
except Exception:
logger.critical(
"Could not re-reserve proofs after an ambiguous melt",
extra={"mint_url": plan.mint_url, "quote_id": plan.quote.quote},
)
raise Bolt11PaymentAmbiguous(f"Cashu melt did not return: {e}") from e
raw_state = getattr(result, "state", None)
state = str(raw_state).lower().rsplit(".", 1)[-1] if raw_state is not None else ""
if state == "paid" or getattr(result, "paid", None) is True:
change = getattr(result, "change", None) or []
paid = sum(proof.amount for proof in selected) - sum(
int(item.amount) for item in change
)
return paid, plan.mint_url, plan.unit
if state == "unpaid":
# The mint is telling us it did not pay, so the proofs are ours again.
await plan.wallet.set_reserved_for_send(selected, reserved=False)
raise Bolt11PaymentNotAttempted("Cashu mint reported the melt as unpaid")
raise Bolt11PaymentAmbiguous(
f"Cashu melt did not reach a final state: {state or 'unknown'}"
)
async def pay_bolt11_invoice(invoice: str) -> tuple[int, str, str]:
"""Prepare and pay a BOLT11 invoice from the best-funded Cashu mint."""
return await execute_bolt11_payment(await prepare_bolt11_payment(invoice))
async def check_bolt11_payment_status(
mint_url: str, unit: str, quote_id: str
) -> str:
"""Ask the mint what became of an earlier melt attempt.
Returns ``"paid"``, ``"unpaid"``, ``"pending"``, or ``"unknown"``. This is
the durable reconciliation path for an ambiguous payment: cashu's
``get_melt_quote`` also settles the wallet database — invalidating the
proofs on ``paid`` and releasing their reservation on ``unpaid`` — so a
caller that sees ``"unpaid"`` may safely retry with the same funds.
"""
try:
wallet = await get_wallet(mint_url, unit)
quote = await wallet.get_melt_quote(quote_id)
except Exception as e:
logger.warning(
"Could not query the mint for a melt quote's status",
extra={"mint_url": mint_url, "quote_id": quote_id, "error": str(e)},
)
return "unknown"
if quote is None:
return "unknown"
state = str(getattr(quote, "state", "")).lower().rsplit(".", 1)[-1]
if state in ("paid", "unpaid", "pending"):
return state
return "unknown"
# A foreign mint's fee_reserve is a non-binding estimate (NUT-05): the mint may
# demand more when re-quoting or at melt execution. Instead of padding the
# estimate with a safety buffer (which strands the margin at the foreign mint
@@ -999,6 +1197,10 @@ async def _refund_sweep_once(cutoff: int) -> None:
db.CashuTransaction.type == "out",
db.CashuTransaction.collected == False, # noqa: E712
db.CashuTransaction.swept == False, # noqa: E712
# NULL != 'x' is NULL in SQL, so legacy rows with no source would
# drop out of the sweep without the explicit IS NULL arm.
(db.CashuTransaction.source != "ppq_auto_topup")
| (db.CashuTransaction.source == None), # noqa: E711
db.CashuTransaction.created_at < cutoff,
)
results = await session.exec(stmt)
tests/integration/test_ppq_auto_topup_claim.py
+516
View File
@@ -0,0 +1,516 @@
"""Real-database tests for the PPQ auto top-up claim lifecycle.
These exercise the claim against actual SQL rather than mocked sessions,
because the guarantees under test are all about what the database will and
will not let two concurrent writers do.
"""
import time
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from sqlmodel import select
from routstr.core.db import CashuTransaction, create_session
from routstr.upstream.auto_topup import (
PPQ_PHASE_CLAIMED,
PPQ_PHASE_IN_FLIGHT,
PPQ_PHASE_RECONCILE,
_claim_ppq_topup,
_ppq_request_id,
_ppq_state_id_for_provider,
_record_ppq_invoice,
_set_ppq_state_terminal,
get_ppq_auto_topup_state,
release_ppq_auto_topup_state,
)
pytestmark = pytest.mark.asyncio
def _row(provider_id: int = 1) -> MagicMock:
row = MagicMock()
row.id = provider_id
return row
async def _seed_provider(provider_id: int = 1, slug: str = "ppq") -> None:
"""Claim creation is fenced on the provider row existing; seed it."""
from routstr.core.db import UpstreamProviderRow
async with create_session() as session:
session.add(
UpstreamProviderRow(
id=provider_id,
slug=slug,
provider_type="ppqai",
base_url="https://api.ppq.ai",
api_key="secret",
enabled=True,
)
)
await session.commit()
async def _state_row(provider_id: int = 1) -> CashuTransaction | None:
async with create_session() as session:
return await session.get(
CashuTransaction, _ppq_state_id_for_provider(provider_id)
)
async def _seed_claim(
provider_id: int,
phase: str,
invoice_id: str,
lease_expires_at: int,
quote_id: str = "quote-1",
) -> str:
"""Seed a claim row and return its state token (the full request_id)."""
token = _ppq_request_id(
"operation-1", lease_expires_at, phase, invoice_id, quote_id
)
async with create_session() as session:
session.add(
CashuTransaction(
id=_ppq_state_id_for_provider(provider_id),
token="lnbc-invoice",
amount=102,
unit="sat",
type="out",
request_id=token,
mint_url="https://mint.test",
collected=False,
source="ppq_auto_topup",
)
)
await session.commit()
return token
async def test_second_claim_is_refused_while_the_first_is_active(
patched_db_engine: Any,
) -> None:
await _seed_provider()
assert await _claim_ppq_topup(_row()) is not None
# The whole point of the claim: a concurrent cycle must not get one.
assert await _claim_ppq_topup(_row()) is None
async with create_session() as session:
rows = (await session.exec(select(CashuTransaction))).all()
assert len(rows) == 1
async def test_claim_is_reusable_once_the_previous_attempt_finished(
patched_db_engine: Any,
) -> None:
await _seed_provider()
first = await _claim_ppq_topup(_row())
assert first is not None
assert await _set_ppq_state_terminal(
_row(), first, collected=True, swept=False
)
second = await _claim_ppq_topup(_row())
assert second is not None and second != first
async def test_recording_the_invoice_moves_the_claim_in_flight(
patched_db_engine: Any,
) -> None:
await _seed_provider()
operation_id = await _claim_ppq_topup(_row())
assert operation_id is not None
state = await get_ppq_auto_topup_state(1)
assert state["phase"] == PPQ_PHASE_CLAIMED
assert state["releasable"] is True
assert state["invoice_id"] is None
lease = await _record_ppq_invoice(
_row(),
operation_id,
invoice="lnbc-invoice",
invoice_id="invoice-1",
quote_id="quote-1",
amount=102,
unit="sat",
mint_url="https://mint.test",
)
assert lease > int(time.time())
state = await get_ppq_auto_topup_state(1)
assert state["phase"] == PPQ_PHASE_IN_FLIGHT
assert state["invoice_id"] == "invoice-1"
# A payment is committed to a mint, so an admin must not sweep it.
assert state["releasable"] is False
# The raw BOLT11 invoice must never reach the admin API.
assert "token" not in state
async def test_release_refuses_an_in_flight_claim(patched_db_engine: Any) -> None:
token = await _seed_claim(
1, PPQ_PHASE_IN_FLIGHT, "invoice-1", int(time.time()) + 900
)
outcome = await release_ppq_auto_topup_state(1, state_token=token)
assert outcome.released is False
assert outcome.reason == "payment_in_flight"
row = await _state_row()
assert row is not None and row.swept is False
async def test_release_refuses_a_stale_state_token(patched_db_engine: Any) -> None:
await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900)
outcome = await release_ppq_auto_topup_state(1, state_token="ppq:stale:token")
assert outcome.released is False
assert outcome.reason == "stale_state"
row = await _state_row()
assert row is not None and row.swept is False
async def test_release_accepts_a_reconcile_claim(patched_db_engine: Any) -> None:
token = await _seed_claim(
1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900
)
outcome = await release_ppq_auto_topup_state(1, state_token=token)
assert outcome.released is True
row = await _state_row()
assert row is not None and row.swept is True
async def test_expired_in_flight_claim_becomes_releasable(
patched_db_engine: Any,
) -> None:
# A worker that died mid-payment must not lock the provider forever.
token = await _seed_claim(
1, PPQ_PHASE_IN_FLIGHT, "invoice-1", int(time.time()) - 1
)
assert (await get_ppq_auto_topup_state(1))["releasable"] is True
outcome = await release_ppq_auto_topup_state(1, state_token=token)
assert outcome.released is True
async def test_release_reports_no_active_claim_once_swept(
patched_db_engine: Any,
) -> None:
token = await _seed_claim(
1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900
)
assert (await release_ppq_auto_topup_state(1, state_token=token)).released
outcome = await release_ppq_auto_topup_state(1, state_token=token)
assert outcome.released is False
assert outcome.reason == "no_active_claim"
async def test_terminal_write_fails_after_the_claim_was_released(
patched_db_engine: Any,
) -> None:
"""The symptom an admin release leaves behind for the owning worker."""
token = await _seed_claim(
1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900
)
assert (await release_ppq_auto_topup_state(1, state_token=token)).released
assert (
await _set_ppq_state_terminal(
_row(), "operation-1", collected=True, swept=False
)
is False
)
async def test_ppq_claim_rows_are_excluded_from_the_admin_transaction_list(
patched_db_engine: Any,
) -> None:
from routstr.core.admin import get_transactions_api
await _claim_ppq_topup(_row())
async with create_session() as session:
session.add(
CashuTransaction(
id="real-transaction",
token="cashuAreal",
amount=50,
unit="sat",
type="out",
source="x-cashu",
)
)
await session.commit()
result = await get_transactions_api()
ids = {t["id"] for t in result["transactions"]} # type: ignore[index,union-attr]
assert "real-transaction" in ids
assert _ppq_state_id_for_provider(1) not in ids
async def test_reconcile_settles_a_recorded_invoice(patched_db_engine: Any) -> None:
from routstr.upstream.auto_topup import _reconcile_ppq_state
await _seed_claim(1, PPQ_PHASE_IN_FLIGHT, "invoice-1", int(time.time()) + 900)
provider = MagicMock()
provider.check_topup_status = AsyncMock(return_value=True)
# Still suppresses this cycle, but the claim is now finished.
assert await _reconcile_ppq_state(_row(), provider) is True
row = await _state_row()
assert row is not None and row.collected is True
async def test_stale_token_from_before_a_phase_change_cannot_release(
patched_db_engine: Any,
) -> None:
"""The blocker scenario: admin reviews `claimed`, payment turns ambiguous.
The operation id is identical in both states, so an id-based fence would
let the stale confirmation land. The full state token must not.
"""
await _seed_provider()
operation_id = await _claim_ppq_topup(_row())
assert operation_id is not None
reviewed = await get_ppq_auto_topup_state(1)
assert reviewed["phase"] == PPQ_PHASE_CLAIMED
# Worker records the invoice: same operation, new phase, proofs committed.
await _record_ppq_invoice(
_row(),
operation_id,
invoice="lnbc-invoice",
invoice_id="invoice-1",
quote_id="quote-1",
amount=102,
unit="sat",
mint_url="https://mint.test",
)
outcome = await release_ppq_auto_topup_state(
1, state_token=str(reviewed["state_token"])
)
assert outcome.released is False
assert outcome.reason == "stale_state"
row = await _state_row()
assert row is not None and row.swept is False
async def test_concurrent_claims_only_one_wins(patched_db_engine: Any) -> None:
import asyncio
await _seed_provider()
results = await asyncio.gather(
*(_claim_ppq_topup(_row()) for _ in range(5)), return_exceptions=True
)
winners = [r for r in results if isinstance(r, str)]
assert len(winners) == 1
async with create_session() as session:
rows = (await session.exec(select(CashuTransaction))).all()
assert len(rows) == 1
async def test_reconcile_releases_claim_when_mint_reports_unpaid(
patched_db_engine: Any,
) -> None:
from routstr.upstream.auto_topup import _reconcile_ppq_state
# Lease expired, PPQ never credited: only the mint's own "unpaid" answer
# may hand the claim back.
await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) - 1)
provider = MagicMock()
provider.check_topup_status = AsyncMock(return_value=False)
with patch(
"routstr.upstream.auto_topup.check_bolt11_payment_status",
AsyncMock(return_value="unpaid"),
) as status:
suppressed = await _reconcile_ppq_state(_row(), provider)
status.assert_awaited_once_with("https://mint.test", "sat", "quote-1")
assert suppressed is False
row = await _state_row()
assert row is not None and row.swept is True
async def test_reconcile_keeps_claim_when_mint_answer_is_not_final(
patched_db_engine: Any,
) -> None:
from routstr.upstream.auto_topup import _reconcile_ppq_state
await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) - 1)
provider = MagicMock()
provider.check_topup_status = AsyncMock(return_value=False)
for answer in ("paid", "pending", "unknown"):
with patch(
"routstr.upstream.auto_topup.check_bolt11_payment_status",
AsyncMock(return_value=answer),
):
assert await _reconcile_ppq_state(_row(), provider) is True
row = await _state_row()
assert row is not None and row.swept is False, answer
async def test_release_endpoint_maps_refusals_to_409(patched_db_engine: Any) -> None:
from fastapi import HTTPException
from routstr.core.admin import (
ReleasePPQAutoTopupRequest,
release_ppq_auto_topup_api,
)
provider_row = MagicMock()
provider_row.provider_type = "ppqai"
token = await _seed_claim(
1, PPQ_PHASE_IN_FLIGHT, "invoice-1", int(time.time()) + 900
)
with patch(
"routstr.core.admin._require_ppq_provider",
AsyncMock(return_value=provider_row),
):
with pytest.raises(HTTPException) as excinfo:
await release_ppq_auto_topup_api(
1,
ReleasePPQAutoTopupRequest(
confirmed_safe_to_retry=True, state_token=token
),
)
assert excinfo.value.status_code == 409
assert "in flight" in excinfo.value.detail
with pytest.raises(HTTPException) as excinfo:
await release_ppq_auto_topup_api(
1,
ReleasePPQAutoTopupRequest(
confirmed_safe_to_retry=True, state_token="ppq:wrong"
),
)
assert excinfo.value.status_code == 409
assert "changed since" in excinfo.value.detail
async def test_provider_delete_is_blocked_by_an_active_claim(
patched_db_engine: Any,
) -> None:
from fastapi import HTTPException
from routstr.core.admin import delete_upstream_provider
from routstr.core.db import UpstreamProviderRow
async with create_session() as session:
session.add(
UpstreamProviderRow(
id=1,
slug="ppq",
provider_type="ppqai",
base_url="https://api.ppq.ai",
api_key="secret",
enabled=True,
)
)
await session.commit()
await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900)
with pytest.raises(HTTPException) as excinfo:
await delete_upstream_provider("1")
assert excinfo.value.status_code == 409
# Provider must still exist.
async with create_session() as session:
assert await session.get(UpstreamProviderRow, 1) is not None
async def test_claim_is_refused_when_the_provider_row_is_gone(
patched_db_engine: Any,
) -> None:
"""The worker's half of the delete race: no provider row, no claim."""
assert await _claim_ppq_topup(_row()) is None
async with create_session() as session:
rows = (await session.exec(select(CashuTransaction))).all()
assert rows == []
async def test_claim_is_refused_after_a_provider_type_change(
patched_db_engine: Any,
) -> None:
from routstr.core.db import UpstreamProviderRow
await _seed_provider()
async with create_session() as session:
provider = await session.get(UpstreamProviderRow, 1)
assert provider is not None
provider.provider_type = "openai"
session.add(provider)
await session.commit()
assert await _claim_ppq_topup(_row()) is None
async def test_disabled_provider_with_claim_still_reconciles(
patched_db_engine: Any,
) -> None:
"""A claim tracks committed money; eligibility must not stop reconciling."""
from routstr.core.db import UpstreamProviderRow
from routstr.upstream.auto_topup import _reconcile_all_ppq_claims
await _seed_provider()
async with create_session() as session:
provider = await session.get(UpstreamProviderRow, 1)
assert provider is not None
provider.enabled = False
session.add(provider)
await session.commit()
await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900)
ppq = MagicMock()
ppq.check_topup_status = AsyncMock(return_value=True)
with patch(
"routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
return_value=ppq,
):
await _reconcile_all_ppq_claims()
row = await _state_row()
assert row is not None and row.collected is True
async def test_claim_without_api_key_still_reconciles_via_the_mint(
patched_db_engine: Any,
) -> None:
from routstr.core.db import UpstreamProviderRow
from routstr.upstream.auto_topup import _reconcile_all_ppq_claims
await _seed_provider()
async with create_session() as session:
provider = await session.get(UpstreamProviderRow, 1)
assert provider is not None
provider.api_key = ""
session.add(provider)
await session.commit()
# Lease expired, so the mint may be consulted.
await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) - 1)
with patch(
"routstr.upstream.auto_topup.check_bolt11_payment_status",
AsyncMock(return_value="unpaid"),
) as status:
await _reconcile_all_ppq_claims()
# No API key: PPQ was never polled, but the mint was, and its definitive
# "unpaid" released the claim.
status.assert_awaited_once()
row = await _state_row()
assert row is not None and row.swept is True
tests/unit/test_auto_topup.py
+362 -1
View File
@@ -4,7 +4,24 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from routstr.core.db import CashuTransaction
from routstr.upstream.auto_topup import _check_and_topup
from routstr.upstream.auto_topup import (
_check_and_topup,
_parse_ppq_request_id,
validate_auto_topup_settings,
)
from routstr.upstream.ppqai import PPQAIUpstreamProvider
def test_ppq_claim_parser_rejects_invalid_expiry() -> None:
assert _parse_ppq_request_id("ppq:operation:not-a-timestamp:invoice") is None
@pytest.mark.asyncio
async def test_ppq_balance_rejects_boolean_api_value() -> None:
provider = PPQAIUpstreamProvider("secret")
provider.check_balance = AsyncMock(return_value={"balance": False}) # type: ignore[method-assign]
assert await provider.get_balance() is None
def _row() -> MagicMock:
@@ -12,6 +29,7 @@ def _row() -> MagicMock:
row.id = "provider-1"
row.base_url = "https://provider.test"
row.api_key = "secret"
row.provider_type = "routstr"
row.provider_settings = json.dumps(
{
"auto_topup": True,
@@ -141,3 +159,346 @@ async def test_auto_topup_does_not_send_untracked_token() -> None:
):
await _check_and_topup(_row())
provider.topup.assert_not_awaited()
def _ppq_row() -> MagicMock:
row = MagicMock()
row.id = "ppq-provider-1"
row.base_url = "https://api.ppq.ai"
row.api_key = "secret"
row.provider_type = "ppqai"
row.provider_settings = json.dumps(
{
"auto_topup": True,
"topup_threshold": 5.0,
"topup_amount_limit": 10,
}
)
return row
@pytest.mark.asyncio
async def test_ppq_auto_topup_pays_invoice_and_confirms_settlement() -> None:
provider = MagicMock()
provider.get_balance = AsyncMock(return_value=2.5)
provider.initiate_topup = AsyncMock(
return_value=MagicMock(
invoice_id="invoice-1",
payment_request="lnbc-invoice",
amount=10,
currency="USD",
expires_at=None,
)
)
provider.check_topup_status = AsyncMock(return_value=True)
plan = MagicMock()
plan.invoice_amount_sats = 100
plan.maximum_spend_sats = 102
plan.quote.amount = 100
plan.quote.fee_reserve = 2
plan.mint_url = "https://mint-rich.test"
plan.unit = "sat"
row = _ppq_row()
with (
patch(
"routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
return_value=provider,
),
patch(
"routstr.upstream.auto_topup._reconcile_ppq_state",
AsyncMock(return_value=False),
),
patch(
"routstr.upstream.auto_topup._claim_ppq_topup",
AsyncMock(return_value="operation-1"),
),
patch(
"routstr.upstream.auto_topup.prepare_bolt11_payment",
AsyncMock(return_value=plan),
) as prepare,
patch(
"routstr.upstream.auto_topup.execute_bolt11_payment",
AsyncMock(return_value=(101, "https://mint-rich.test", "sat")),
) as execute,
patch(
"routstr.upstream.auto_topup._record_ppq_invoice", AsyncMock()
) as record,
patch(
"routstr.upstream.auto_topup._set_ppq_state_terminal", AsyncMock()
) as terminal,
patch("routstr.upstream.auto_topup.sats_usd_price", return_value=0.001),
):
await _check_and_topup(row)
provider.initiate_topup.assert_awaited_once_with(10)
prepare.assert_awaited_once_with("lnbc-invoice")
execute.assert_awaited_once_with(plan)
record.assert_awaited_once()
provider.check_topup_status.assert_awaited_once_with("invoice-1")
terminal.assert_awaited_once_with(
row, "operation-1", collected=True, swept=False
)
@pytest.mark.asyncio
async def test_ppq_ambiguous_melt_keeps_claim_and_emits_critical_alert() -> None:
provider = MagicMock()
provider.get_balance = AsyncMock(return_value=2.5)
provider.initiate_topup = AsyncMock(
return_value=MagicMock(
invoice_id="invoice-1",
payment_request="lnbc-invoice",
amount=10,
currency="USD",
expires_at=None,
)
)
plan = MagicMock(maximum_spend_sats=102, mint_url="https://mint.test", unit="sat")
plan.quote.amount = 100
plan.quote.fee_reserve = 2
row = _ppq_row()
with (
patch(
"routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
return_value=provider,
),
patch(
"routstr.upstream.auto_topup._reconcile_ppq_state",
AsyncMock(return_value=False),
),
patch(
"routstr.upstream.auto_topup._claim_ppq_topup",
AsyncMock(return_value="operation-1"),
),
patch(
"routstr.upstream.auto_topup.prepare_bolt11_payment",
AsyncMock(return_value=plan),
),
patch(
"routstr.upstream.auto_topup.execute_bolt11_payment",
AsyncMock(side_effect=TimeoutError("ambiguous melt")),
),
patch(
"routstr.upstream.auto_topup._record_ppq_invoice",
AsyncMock(return_value=2_000_000_000),
),
patch(
"routstr.upstream.auto_topup._mark_ppq_reconcile", AsyncMock()
) as reconcile_mark,
patch(
"routstr.upstream.auto_topup._set_ppq_state_terminal", AsyncMock()
) as terminal,
patch("routstr.upstream.auto_topup.sats_usd_price", return_value=0.001),
patch("routstr.upstream.auto_topup.logger.critical") as critical,
):
with pytest.raises(TimeoutError, match="ambiguous melt"):
await _check_and_topup(row)
# The claim is never released — it moves to reconcile for the admin.
terminal.assert_not_awaited()
reconcile_mark.assert_awaited_once()
critical.assert_called_once()
assert "admin reconciliation" in critical.call_args.args[0]
@pytest.mark.asyncio
async def test_ppq_auto_topup_skips_when_balance_meets_threshold() -> None:
provider = MagicMock()
provider.get_balance = AsyncMock(return_value=5.0)
provider.initiate_topup = AsyncMock()
with (
patch(
"routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
return_value=provider,
),
patch(
"routstr.upstream.auto_topup._reconcile_ppq_state",
AsyncMock(return_value=False),
),
):
await _check_and_topup(_ppq_row())
provider.initiate_topup.assert_not_awaited()
@pytest.mark.asyncio
async def test_ppq_pending_attempt_suppresses_duplicate_topup() -> None:
provider = MagicMock()
provider.get_balance = AsyncMock()
with (
patch(
"routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
return_value=provider,
),
patch(
"routstr.upstream.auto_topup._reconcile_ppq_state",
AsyncMock(return_value=True),
),
):
await _check_and_topup(_ppq_row())
provider.get_balance.assert_not_awaited()
@pytest.mark.asyncio
async def test_ppq_auto_topup_rejects_non_finite_balance() -> None:
provider = MagicMock()
provider.get_balance = AsyncMock(return_value=float("nan"))
with (
patch(
"routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
return_value=provider,
),
patch(
"routstr.upstream.auto_topup._reconcile_ppq_state",
AsyncMock(return_value=False),
),
patch("routstr.upstream.auto_topup._claim_ppq_topup", AsyncMock()) as claim,
):
await _check_and_topup(_ppq_row())
claim.assert_not_awaited()
@pytest.mark.asyncio
async def test_routstr_threshold_is_compared_in_sats() -> None:
provider = MagicMock()
provider.get_balance = AsyncMock(return_value=100)
provider.topup = AsyncMock()
with (
patch(
"routstr.upstream.auto_topup.RoutstrUpstreamProvider.from_db_row",
return_value=provider,
),
patch("routstr.upstream.auto_topup.send_token", AsyncMock()) as send,
):
await _check_and_topup(_row())
send.assert_not_awaited()
provider.topup.assert_not_awaited()
@pytest.mark.asyncio
async def test_settled_topup_alerts_when_its_claim_was_already_released() -> None:
provider = MagicMock()
provider.get_balance = AsyncMock(return_value=2.5)
provider.initiate_topup = AsyncMock(
return_value=MagicMock(
invoice_id="invoice-1",
payment_request="lnbc-invoice",
amount=10,
currency="USD",
expires_at=None,
)
)
provider.check_topup_status = AsyncMock(return_value=True)
plan = MagicMock()
plan.maximum_spend_sats = 102
plan.quote.amount = 100
plan.quote.fee_reserve = 2
plan.mint_url = "https://mint-rich.test"
plan.unit = "sat"
with (
patch(
"routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
return_value=provider,
),
patch(
"routstr.upstream.auto_topup._reconcile_ppq_state",
AsyncMock(return_value=False),
),
patch(
"routstr.upstream.auto_topup._claim_ppq_topup",
AsyncMock(return_value="operation-1"),
),
patch(
"routstr.upstream.auto_topup.prepare_bolt11_payment",
AsyncMock(return_value=plan),
),
patch(
"routstr.upstream.auto_topup.execute_bolt11_payment",
AsyncMock(return_value=(101, "https://mint-rich.test", "sat")),
),
patch("routstr.upstream.auto_topup._record_ppq_invoice", AsyncMock()),
patch(
"routstr.upstream.auto_topup._set_ppq_state_terminal",
AsyncMock(return_value=False),
),
patch("routstr.upstream.auto_topup.sats_usd_price", return_value=0.001),
patch("routstr.upstream.auto_topup.logger") as log,
):
await _check_and_topup(_ppq_row())
assert any(
"claim was already released" in call.args[0]
for call in log.critical.call_args_list
)
@pytest.mark.parametrize(
("provider_type", "settings", "expected"),
[
("ppqai", {"auto_topup": False, "topup_threshold": -1}, None),
("ppqai", {"auto_topup": True, "topup_threshold": 5, "topup_amount_limit": 10}, None),
(
"ppqai",
{"auto_topup": True, "topup_threshold": None, "topup_amount_limit": 10},
"threshold",
),
(
"ppqai",
{"auto_topup": True, "topup_threshold": 5, "topup_amount_limit": 0.5},
"whole number",
),
(
"ppqai",
{"auto_topup": True, "topup_threshold": 5, "topup_amount_limit": 5000},
"between",
),
(
"ppqai",
{"auto_topup": True, "topup_threshold": True, "topup_amount_limit": 10},
"threshold",
),
(
"routstr",
{"auto_topup": True, "topup_threshold": 5, "topup_amount_limit": 10},
"mint URL",
),
(
"routstr",
{
"auto_topup": True,
"topup_threshold": 5,
"topup_amount_limit": 10,
"topup_mint_url": "https://mint.test",
},
None,
),
],
)
def test_auto_topup_settings_validation(
provider_type: str, settings: dict, expected: str | None
) -> None:
problem = validate_auto_topup_settings(provider_type, settings)
if expected is None:
assert problem is None
else:
assert problem is not None and expected in problem
def test_auto_topup_settings_validation_survives_huge_json_integers() -> None:
# json.loads happily produces integers past float range; float() raises
# OverflowError there instead of returning inf.
problem = validate_auto_topup_settings(
"ppqai",
{"auto_topup": True, "topup_threshold": 10**400, "topup_amount_limit": 10},
)
assert problem is not None and "threshold" in problem
tests/unit/test_wallet.py
+179 -1
View File
@@ -1,19 +1,24 @@
import base64
import json
import socket
from unittest.mock import AsyncMock, Mock, patch
from unittest.mock import AsyncMock, MagicMock, Mock, patch
import httpx
import pytest
from routstr.core.db import ApiKey
from routstr.wallet import (
Bolt11PaymentAmbiguous,
Bolt11PaymentNotAttempted,
MintConnectionError,
TokenConsumedError,
classify_redemption_error,
credit_balance,
execute_bolt11_payment,
get_balance,
is_mint_connection_error,
pay_bolt11_invoice,
prepare_bolt11_payment,
recieve_token,
send,
send_token,
@@ -1389,3 +1394,176 @@ async def test_swap_melt_transport_error_raises_mint_connection_error() -> None:
await swap_to_primary_mint(mock_token, mock_token_wallet)
assert mock_token_wallet.melt.call_count == 1
@pytest.mark.asyncio
async def test_pay_bolt11_invoice_uses_sufficient_mint_with_highest_balance() -> None:
from routstr.core.settings import settings
low_wallet = MagicMock()
high_wallet = MagicMock()
low_proof = MagicMock(amount=200)
high_proof = MagicMock(amount=500)
low_wallet.proofs = [low_proof]
high_wallet.proofs = [high_proof]
for wallet in (low_wallet, high_wallet):
wallet.melt_quote = AsyncMock(
return_value=MagicMock(amount=100, fee_reserve=2, quote="quote-1")
)
wallet.get_fees_for_proofs = Mock(return_value=0)
wallet.select_to_send = AsyncMock(
side_effect=lambda proofs, *args, **kwargs: (proofs, 0)
)
wallet.melt = AsyncMock(
return_value=MagicMock(
state="PAID", change=[MagicMock(amount=399)]
)
)
wallet.set_reserved_for_send = AsyncMock()
async def get_wallet(mint_url: str, unit: str = "sat") -> MagicMock:
if unit == "msat":
raise ValueError("unit unsupported")
return high_wallet if mint_url == "https://high.test" else low_wallet
with (
patch.object(
settings,
"cashu_mints",
["https://low.test", "https://high.test"],
),
patch.object(settings, "primary_mint", "https://low.test"),
patch("routstr.wallet.get_wallet", side_effect=get_wallet),
patch(
"routstr.wallet.get_proofs_per_mint_and_unit",
side_effect=lambda wallet, *args, **kwargs: wallet.proofs,
),
patch(
"routstr.wallet.slow_filter_spend_proofs",
side_effect=lambda proofs, wallet: proofs,
),
):
amount, mint_url, unit = await pay_bolt11_invoice("lnbc-invoice")
assert (amount, mint_url, unit) == (101, "https://high.test", "sat")
high_wallet.melt.assert_awaited_once()
low_wallet.melt.assert_not_awaited()
@pytest.mark.asyncio
async def test_execute_bolt11_payment_rejects_unpaid_melt_state() -> None:
plan = MagicMock()
plan.proofs = [MagicMock(amount=110)]
plan.quote.amount = 100
plan.quote.fee_reserve = 10
plan.quote.quote = "quote-1"
plan.invoice = "lnbc-invoice"
plan.wallet.select_to_send = AsyncMock(return_value=(plan.proofs, 0))
plan.wallet.set_reserved_for_send = AsyncMock()
plan.wallet.melt = AsyncMock(return_value=MagicMock(state="UNPAID", change=[]))
with pytest.raises(Bolt11PaymentNotAttempted):
await execute_bolt11_payment(plan)
# An explicit unpaid answer means the proofs are ours again.
plan.wallet.set_reserved_for_send.assert_awaited_with(plan.proofs, reserved=False)
@pytest.mark.asyncio
async def test_execute_bolt11_payment_accepts_legacy_paid_response() -> None:
plan = MagicMock()
plan.proofs = [MagicMock(amount=110)]
plan.quote.amount = 100
plan.quote.fee_reserve = 10
plan.quote.quote = "quote-1"
plan.invoice = "lnbc-invoice"
plan.mint_url = "https://mint.test"
plan.unit = "sat"
plan.wallet.select_to_send = AsyncMock(return_value=(plan.proofs, 0))
plan.wallet.set_reserved_for_send = AsyncMock()
plan.wallet.melt = AsyncMock(
return_value=MagicMock(state=None, paid=True, change=[])
)
assert await execute_bolt11_payment(plan) == (
110,
"https://mint.test",
"sat",
)
@pytest.mark.asyncio
async def test_execute_bolt11_payment_keeps_proofs_reserved_when_melt_errors() -> None:
plan = MagicMock()
plan.proofs = [MagicMock(amount=110)]
plan.quote.amount = 100
plan.quote.fee_reserve = 10
plan.quote.quote = "quote-1"
plan.invoice = "lnbc-invoice"
plan.wallet.select_to_send = AsyncMock(return_value=(plan.proofs, 0))
plan.wallet.set_reserved_for_send = AsyncMock()
plan.wallet.set_reserved_for_melt = AsyncMock()
plan.wallet.melt = AsyncMock(side_effect=TimeoutError("no answer"))
with pytest.raises(Bolt11PaymentAmbiguous):
await execute_bolt11_payment(plan)
# The mint may still settle with these proofs. cashu's own melt()
# un-reserves them on a mint transport error, so the ambiguous path must
# re-reserve — and it must do so with the melt quote id, because
# get_melt_quote() finds the proofs to settle by melt_id.
plan.wallet.set_reserved_for_melt.assert_awaited_once_with(
plan.proofs, reserved=True, quote_id="quote-1"
)
@pytest.mark.asyncio
async def test_execute_bolt11_payment_does_not_reserve_when_selection_fails() -> None:
plan = MagicMock()
plan.proofs = [MagicMock(amount=110)]
plan.quote.amount = 100
plan.quote.fee_reserve = 10
plan.wallet.select_to_send = AsyncMock(side_effect=ValueError("insufficient"))
plan.wallet.set_reserved_for_send = AsyncMock()
plan.wallet.melt = AsyncMock()
with pytest.raises(Bolt11PaymentNotAttempted):
await execute_bolt11_payment(plan)
plan.wallet.set_reserved_for_send.assert_not_awaited()
plan.wallet.melt.assert_not_awaited()
@pytest.mark.asyncio
async def test_prepare_bolt11_payment_counts_input_fees_in_sufficiency() -> None:
from routstr.core.settings import settings
wallet = MagicMock()
wallet.proofs = [MagicMock(amount=105)]
wallet.melt_quote = AsyncMock(
return_value=MagicMock(amount=100, fee_reserve=2, quote="quote-1")
)
# Balance covers amount + fee_reserve (102) but not the 5 sat input fee.
wallet.get_fees_for_proofs = Mock(return_value=5)
async def get_wallet(mint_url: str, unit: str = "sat") -> MagicMock:
if unit == "msat":
raise ValueError("unit unsupported")
return wallet
with (
patch.object(settings, "cashu_mints", ["https://only.test"]),
patch.object(settings, "primary_mint", "https://only.test"),
patch("routstr.wallet.get_wallet", side_effect=get_wallet),
patch(
"routstr.wallet.get_proofs_per_mint_and_unit",
side_effect=lambda wallet, *args, **kwargs: wallet.proofs,
),
patch(
"routstr.wallet.slow_filter_spend_proofs",
side_effect=lambda proofs, wallet: proofs,
),
pytest.raises(ValueError, match="enough balance"),
):
await prepare_bolt11_payment("lnbc-invoice")
ui/components/provider-card.tsx
+131 -1
View File
@@ -1,3 +1,5 @@
import { AdminService } from '@/lib/api/services/admin';
import type { PPQAutoTopupState } from '@/lib/api/services/admin';
import type {
AdminModel,
ProviderModels,
@@ -20,12 +22,15 @@ import {
Trash2,
Key,
RotateCcw,
AlertTriangle,
Unlock,
Loader2,
} from 'lucide-react';
import { ProviderBalance } from '@/components/provider-balance';
import { ProviderModelsPanel } from '@/components/provider-models-panel';
import { RoutstrCreateKeySection } from '@/components/providers/RoutstrCreateKeySection';
import { RoutstrProviderService } from '@/lib/api/services/routstr-provider';
import { useMutation, useQueryClient } from '@tanstack/react-query';
import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
import { useState } from 'react';
import { toast } from 'sonner';
import { cn } from '@/lib/utils';
@@ -36,6 +41,16 @@ import {
DialogHeader,
DialogTitle,
} from '@/components/ui/dialog';
import {
AlertDialog,
AlertDialogAction,
AlertDialogCancel,
AlertDialogContent,
AlertDialogDescription,
AlertDialogFooter,
AlertDialogHeader,
AlertDialogTitle,
} from '@/components/ui/alert-dialog';
interface ProviderCardProps {
provider: UpstreamProvider;
@@ -77,8 +92,60 @@ export function ProviderCard({
}: ProviderCardProps) {
const queryClient = useQueryClient();
const [isKeyModalOpen, setIsKeyModalOpen] = useState(false);
const [isReleaseDialogOpen, setIsReleaseDialogOpen] = useState(false);
// Snapshot of the claim as it looked when the admin opened the dialog.
// The mutation sends this token, never the live query data: a background
// refetch must not swap in a state the admin never reviewed.
const [reviewedState, setReviewedState] = useState<PPQAutoTopupState | null>(
null
);
const hasDetails = Boolean(provider.api_version) || isExpanded;
const isRoutstr = provider.provider_type === 'routstr';
const isPPQ = provider.provider_type === 'ppqai';
const { data: ppqAutoTopupState } = useQuery({
queryKey: ['ppq-auto-topup-state', provider.id],
queryFn: () => AdminService.getPPQAutoTopupState(provider.id),
enabled: isPPQ,
refetchInterval: 30000,
});
// A claim the server will not let us release: a worker is between reserving
// proofs and hearing back from the mint, and sweeping it would let the next
// cycle pay a second invoice.
const isPPQPaymentInFlight =
Boolean(ppqAutoTopupState?.active) &&
ppqAutoTopupState?.releasable === false;
const openReleaseDialog = () => {
setReviewedState(ppqAutoTopupState ?? null);
setIsReleaseDialogOpen(true);
};
const releasePPQMutation = useMutation({
mutationFn: () =>
AdminService.releasePPQAutoTopup(
provider.id,
reviewedState?.state_token ?? null
),
onSuccess: () => {
queryClient.invalidateQueries({
queryKey: ['ppq-auto-topup-state', provider.id],
});
setIsReleaseDialogOpen(false);
setReviewedState(null);
toast.success('PPQ auto top-up claim released');
},
onError: (error: Error) => {
// A 409 usually means the claim changed since it was reviewed.
queryClient.invalidateQueries({
queryKey: ['ppq-auto-topup-state', provider.id],
});
setIsReleaseDialogOpen(false);
setReviewedState(null);
toast.error(`Failed to release PPQ claim: ${error.message}`);
},
});
const refundMutation = useMutation({
mutationFn: () => RoutstrProviderService.refundBalance(provider.id),
@@ -113,6 +180,26 @@ export function ProviderCard({
>
{provider.enabled ? 'Enabled' : 'Disabled'}
</Badge>
{ppqAutoTopupState?.active && (
<Badge
variant='outline'
className={cn(
'w-fit gap-1',
isPPQPaymentInFlight
? 'border-blue-500 text-blue-700 dark:text-blue-400'
: 'border-amber-500 text-amber-700 dark:text-amber-400'
)}
>
{isPPQPaymentInFlight ? (
<Loader2 className='h-3 w-3 animate-spin' />
) : (
<AlertTriangle className='h-3 w-3' />
)}
{isPPQPaymentInFlight
? 'Paying invoice'
: 'Auto top-up needs review'}
</Badge>
)}
</div>
<CardDescription className='break-all'>
{provider.base_url}
@@ -153,6 +240,19 @@ export function ProviderCard({
</Button>
)}
{isPPQ && ppqAutoTopupState?.active && !isPPQPaymentInFlight && (
<Button
variant='outline'
size='sm'
onClick={openReleaseDialog}
className='justify-center gap-1.5 border-amber-500 text-amber-700 dark:text-amber-400'
title='Release only after manually verifying the Lightning payment outcome'
>
<Unlock className='h-4 w-4' />
<span>Release top-up</span>
</Button>
)}
{isRoutstr && provider.api_key && (
<Button
variant='outline'
@@ -212,6 +312,36 @@ export function ProviderCard({
</div>
</CardHeader>
<AlertDialog
open={isReleaseDialogOpen}
onOpenChange={setIsReleaseDialogOpen}
>
<AlertDialogContent>
<AlertDialogHeader>
<AlertDialogTitle>Release PPQ auto top-up?</AlertDialogTitle>
<AlertDialogDescription>
Only continue after checking PPQ and the Cashu mint and confirming
the previous Lightning payment cannot later settle. Releasing an
ambiguous payment can allow a duplicate top-up.
{reviewedState?.invoice_id
? ` Invoice: ${reviewedState.invoice_id}`
: ''}
</AlertDialogDescription>
</AlertDialogHeader>
<AlertDialogFooter>
<AlertDialogCancel>Cancel</AlertDialogCancel>
<AlertDialogAction
onClick={() => releasePPQMutation.mutate()}
disabled={releasePPQMutation.isPending}
>
{releasePPQMutation.isPending
? 'Releasing...'
: 'I verified it is safe to retry'}
</AlertDialogAction>
</AlertDialogFooter>
</AlertDialogContent>
</AlertDialog>
<Dialog open={isKeyModalOpen} onOpenChange={setIsKeyModalOpen}>
<DialogContent className='max-h-[90dvh] overflow-y-auto sm:max-w-[500px]'>
<DialogHeader>
ui/components/provider-form-fields.tsx
+16
View File
@@ -17,6 +17,7 @@ import {
SelectValue,
} from '@/components/ui/select';
import { Switch } from '@/components/ui/switch';
import { PPQAutoTopupSettings } from '@/components/providers/PPQAutoTopupSettings';
import { RoutstrNodeSettings } from '@/components/providers/RoutstrNodeSettings';
import { RoutstrCreateKeySection } from '@/components/providers/RoutstrCreateKeySection';
@@ -78,6 +79,8 @@ export function ProviderFormFields({
provider_type: value,
base_url: getDefaultBaseUrl(value),
provider_fee: value === 'openrouter' ? 1.06 : 1.01,
provider_settings:
value === prev.provider_type ? prev.provider_settings : {},
}));
}}
>
@@ -118,6 +121,19 @@ export function ProviderFormFields({
/>
)}
{formData.provider_type === 'ppqai' && (
<PPQAutoTopupSettings
settings={formData.provider_settings || {}}
onSettingsChange={(settings) =>
setFormData((prev) => ({
...prev,
provider_settings: settings,
}))
}
idPrefix={mode === 'edit' ? 'edit' : ''}
/>
)}
<div className='grid gap-2'>
<Label htmlFor={`${idPrefix}slug`}>
Slug {mode === 'create' ? '(optional, auto-generated)' : ''}
ui/components/providers/PPQAutoTopupSettings.tsx
+136
View File
@@ -0,0 +1,136 @@
'use client';
import { Input } from '@/components/ui/input';
import { Label } from '@/components/ui/label';
import { Switch } from '@/components/ui/switch';
interface ProviderSettings {
auto_topup?: boolean;
topup_threshold?: number;
topup_amount_limit?: number;
[key: string]: unknown;
}
interface PPQAutoTopupSettingsProps {
settings: ProviderSettings;
onSettingsChange: (settings: ProviderSettings) => void;
idPrefix?: string;
}
export function PPQAutoTopupSettings({
settings,
onSettingsChange,
idPrefix = '',
}: PPQAutoTopupSettingsProps) {
const prefix = idPrefix ? `${idPrefix}_` : '';
const update = (patch: Partial<ProviderSettings>) =>
onSettingsChange({ ...settings, ...patch });
/**
* Clearing the field yields '' and parse* yields NaN, which JSON.stringify
* turns into null. Drop the key instead so the server rejects a missing
* value rather than storing a broken one.
*/
const updateNumber = (
key: 'topup_threshold' | 'topup_amount_limit',
raw: string,
parse: (value: string) => number
) => {
const next = { ...settings };
const parsed = parse(raw);
if (raw === '' || Number.isNaN(parsed)) {
delete next[key];
} else {
next[key] = parsed;
}
onSettingsChange(next);
};
const threshold = settings.topup_threshold;
const amount = settings.topup_amount_limit;
const thresholdError =
threshold !== undefined && threshold <= 0
? 'Must be greater than 0'
: undefined;
const amountError =
amount !== undefined && (amount < 1 || amount > 500)
? 'Must be between 1 and 500 USD'
: undefined;
return (
<div className='bg-muted/30 grid gap-4 rounded-lg border p-4'>
<Label className='text-sm font-semibold'>PPQ Auto Top-up</Label>
<div className='flex items-center justify-between'>
<Label htmlFor={`${prefix}ppq_auto_topup`} className='text-sm'>
Enable Auto Top-up
</Label>
<Switch
id={`${prefix}ppq_auto_topup`}
checked={!!settings.auto_topup}
onCheckedChange={(checked) => update({ auto_topup: checked })}
/>
</div>
{settings.auto_topup && (
<div className='border-primary/20 grid gap-4 border-l-2 pt-2 pl-4'>
<div className='grid gap-2'>
<Label
htmlFor={`${prefix}ppq_topup_threshold`}
className='text-xs font-medium'
>
When credits are below (USD)
</Label>
<Input
id={`${prefix}ppq_topup_threshold`}
type='number'
min='0.01'
step='0.01'
className='h-9'
placeholder='e.g. 5'
value={settings.topup_threshold ?? ''}
aria-invalid={Boolean(thresholdError)}
onChange={(e) =>
updateNumber('topup_threshold', e.target.value, parseFloat)
}
/>
{thresholdError && (
<p className='text-destructive text-[10px]'>{thresholdError}</p>
)}
</div>
<div className='grid gap-2'>
<Label
htmlFor={`${prefix}ppq_topup_amount_limit`}
className='text-xs font-medium'
>
Purchase this amount (USD)
</Label>
<Input
id={`${prefix}ppq_topup_amount_limit`}
type='number'
min='1'
max='500'
step='1'
className='h-9'
placeholder='e.g. 10'
value={settings.topup_amount_limit ?? ''}
aria-invalid={Boolean(amountError)}
onChange={(e) =>
updateNumber('topup_amount_limit', e.target.value, parseInt)
}
/>
{amountError && (
<p className='text-destructive text-[10px]'>{amountError}</p>
)}
</div>
<p className='text-muted-foreground text-[10px]'>
Pays PPQ&apos;s Lightning invoice from the sufficiently funded Cashu
mint with the highest available balance.
</p>
</div>
)}
</div>
);
}
ui/lib/api/services/admin.ts
+50
View File
@@ -1013,6 +1013,32 @@ export class AdminService {
}>(`/admin/api/upstream-providers/${providerId}/balance`);
}
static async getPPQAutoTopupState(
providerId: number
): Promise<PPQAutoTopupState> {
return await apiClient.get<PPQAutoTopupState>(
`/admin/api/upstream-providers/${providerId}/ppq-auto-topup`
);
}
/**
* `stateToken` must be the `state_token` snapshotted when the admin opened
* the confirmation — not re-read at submit time. The server rejects a
* release whose claim changed in any way since that snapshot.
*/
static async releasePPQAutoTopup(
providerId: number,
stateToken: string | null
): Promise<{
ok: boolean;
released: boolean;
}> {
return await apiClient.post<{ ok: boolean; released: boolean }>(
`/admin/api/upstream-providers/${providerId}/ppq-auto-topup/release`,
{ confirmed_safe_to_retry: true, state_token: stateToken }
);
}
// ── CLI Tokens ──
static async listCliTokens(): Promise<CliTokenListItem[]> {
@@ -1244,6 +1270,30 @@ export interface TransactionsResponse {
total: number;
}
export interface PPQAutoTopupState {
ok: boolean;
active: boolean;
/**
* Opaque version of the claim as reviewed. Echo it back verbatim to
* release; the server rejects a release whose claim changed in any way
* (new attempt, phase change, renewed lease) since this was read.
*/
state_token?: string | null;
/** Identifies the attempt currently holding the claim. Informational. */
operation_id?: string | null;
/** 'claimed' | 'in_flight' | 'reconcile'. Null when the claim is malformed. */
phase?: string | null;
/** False while a payment is in flight — the server rejects a release then. */
releasable?: boolean;
expires_at?: number | null;
invoice_id?: string | null;
created_at?: number;
amount?: number;
unit?: string;
mint_url?: string | null;
malformed?: boolean;
}
export interface LightningInvoice {
id: string;
bolt11: string;