laantungir/routstr-core
1
0
Fork 0
mirror of https://github.com/Routstr/routstr-core.git synced 2026-10-05 20:28:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: bound Routstr auto-topup spend with a durable claim

Browse Source 
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.
This commit is contained in:
9qeklajc
2026-08-24 01:53:16 +02:00
parent 78a7632df3
commit c1b7f78a02
4 changed files with 1057 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files
routstr/core/admin.py
+111 -23
View File
@@ -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)
routstr/upstream/auto_topup.py
+536 -16
View File
@@ -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")
tests/integration/test_routstr_auto_topup_claim.py
+351
View File
@@ -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())
tests/unit/test_auto_topup.py
+59 -118
View File
@@ -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