diff --git a/routstr/core/db.py b/routstr/core/db.py index 55693d41..e3bdbd3a 100644 --- a/routstr/core/db.py +++ b/routstr/core/db.py @@ -367,6 +367,12 @@ async def prune_dead_api_keys(session: AsyncSession, min_age_seconds: int) -> in ) ).exists() + has_refund_claim = ( + select(Refund.id).where( + col(Refund.api_key_hashed_key) == col(ApiKey.hashed_key) + ) + ).exists() + eligible_hashes = ( select(ApiKey.hashed_key) .where(col(ApiKey.balance) == 0) @@ -375,6 +381,8 @@ async def prune_dead_api_keys(session: AsyncSession, min_age_seconds: int) -> in .where(col(ApiKey.total_requests) == 0) .where((col(ApiKey.created_at).is_(None)) | (col(ApiKey.created_at) < cutoff)) .where(~settleable_invoice) + # refunds holds a non-null FK to the key. + .where(~has_refund_claim) ) # Unlink transactions rather than cascade-deleting them, so the financial @@ -561,12 +569,7 @@ _REFUND_OPEN_PREDICATE = "status IN ('pending', 'ambiguous')" class Refund(SQLModel, table=True): # type: ignore - """A durable claim on an API key's balance for a single payout. - - The partial unique index is the double-refund guarantee: a key can have at - most one open claim, so a Cashu refund cannot start while a Lightning - refund is in flight, and neither survives a crash without a record. - """ + """One payout claim; the partial unique index allows one open claim per key.""" __tablename__ = "refunds" __table_args__ = ( diff --git a/routstr/refund.py b/routstr/refund.py index 75c2a0db..0ff8b9a5 100644 --- a/routstr/refund.py +++ b/routstr/refund.py @@ -1,12 +1,8 @@ -"""Durable, mutually exclusive refund claims for API key balances. - -A claim debits the balance and records the payout in one transaction, so a key -can never have two payouts in flight and no crash can leave a debited balance -without a record of why. -""" +"""Refund claims: one open payout per API key, recorded before it is paid.""" import asyncio import time +from typing import Any from fastapi import HTTPException from sqlalchemy.exc import IntegrityError @@ -47,14 +43,12 @@ def amount_in_unit(amount_msats: int, unit: str) -> int: def refund_mint(key: ApiKey) -> str: - """Persisted mint preferences must not outlive the trusted-mint config.""" if key.refund_mint_url and key.refund_mint_url in settings.cashu_mints: return key.refund_mint_url return settings.primary_mint async def validate_lightning_destination(destination: str) -> None: - """Resolve the destination before claiming, so a bad address never debits.""" try: await get_lnurl_data(destination) except LNURLError as e: @@ -70,12 +64,7 @@ async def open_claim( method: str, destination: str | None, ) -> Refund: - """Debit the balance to zero and record the claim in one transaction. - - The claim starts leased (``claimed_at``) to the request that opened it, so - the reconciler leaves it alone until ``refund_claim_timeout_seconds`` have - passed; a payout still in flight is never released underneath itself. - """ + """Zero the balance and insert the claim in one transaction.""" unit = refund_unit(key) refund = Refund( api_key_hashed_key=key.hashed_key, @@ -118,29 +107,40 @@ async def open_claim( return refund -async def _close(session: AsyncSession, refund: Refund, **values: object) -> bool: - result = await session.exec( # type: ignore[call-overload] +async def _close( + session: AsyncSession, + refund: Refund, + *, + require_no_quote: bool = False, + **values: object, +) -> bool: + stmt = ( update(Refund) .where(col(Refund.id) == refund.id) .where(col(Refund.status).in_(REFUND_OPEN_STATUSES)) - .values(claimed_at=None, updated_at=int(time.time()), **values) + ) + if require_no_quote: + # A quote recorded since the row was read means a melt may be in flight. + stmt = stmt.where(col(Refund.quote_id).is_(None)) + result = await session.exec( # type: ignore[call-overload] + stmt.values(claimed_at=None, updated_at=int(time.time()), **values) ) return bool(result.rowcount) async def record_quote(refund: Refund, quote_id: str) -> None: - """Persist the melt quote before the melt is dispatched. - - Once the quote is on disk the reconciler can ask the mint what became of - it, so a crash after this point can never be mistaken for "never sent". - Raises if the claim is no longer open, which aborts the payout. - """ + """Store the melt quote before the melt is sent; raises if the claim closed.""" async with create_session() as session: result = await session.exec( # type: ignore[call-overload] update(Refund) .where(col(Refund.id) == refund.id) .where(col(Refund.status).in_(REFUND_OPEN_STATUSES)) - .values(quote_id=quote_id, updated_at=int(time.time())) + # Renew the lease so the reconciler leaves the payout alone. + .values( + quote_id=quote_id, + claimed_at=int(time.time()), + updated_at=int(time.time()), + ) ) await session.commit() if not result.rowcount: @@ -156,7 +156,7 @@ async def settle( token: str | None = None, mint_url: str | None = None, ) -> bool: - values: dict[str, object] = {"status": "paid"} + values: dict[str, Any] = {"status": "paid"} if quote_id is not None: values["quote_id"] = quote_id if token is not None: @@ -173,12 +173,14 @@ async def settle( return settled -async def release(session: AsyncSession, refund: Refund) -> bool: - """Close the claim and return the debited balance in the same transaction.""" - if not await _close(session, refund, status="failed"): - # Nothing changed; commit rather than roll back so the session stays - # usable (an async rollback after an ORM-enabled UPDATE expires the - # identity map and later loads fail outside the greenlet). +async def release( + session: AsyncSession, refund: Refund, *, require_no_quote: bool = False +) -> bool: + """Mark the claim failed and restore the balance.""" + if not await _close( + session, refund, require_no_quote=require_no_quote, status="failed" + ): + # Commit, not rollback: rollback after an ORM UPDATE breaks later loads. await session.commit() return False await session.exec( # type: ignore[call-overload] @@ -199,18 +201,12 @@ async def release(session: AsyncSession, refund: Refund) -> bool: async def hold(session: AsyncSession, refund: Refund, quote_id: str | None) -> None: - """Keep the debit and the claim open until reconciliation resolves it.""" await _close(session, refund, status="ambiguous", quote_id=quote_id) await session.commit() async def latest_terminal(session: AsyncSession, key: ApiKey) -> Refund | None: - """Most recent paid Lightning refund, for idempotent re-requests. - - Cashu payouts are deliberately excluded: their token lives in - ``cashu_transactions`` whose ``collected``/``swept`` flags decide whether - it may still be handed out. - """ + """Latest paid Lightning refund. Cashu is served from cashu_transactions.""" result = await session.exec( select(Refund) .where(Refund.api_key_hashed_key == key.hashed_key) @@ -235,7 +231,6 @@ def describe(refund: Refund) -> dict[str, str]: async def execute(session: AsyncSession, refund: Refund) -> dict[str, str]: - """Pay out an open claim, closing it on every outcome the mint makes known.""" amount = amount_in_unit(refund.amount_msats, refund.unit) quote_id: str | None = None @@ -338,11 +333,9 @@ async def _lease(refund_id: str, now: int, lease_cutoff: int) -> bool: return bool(result.rowcount) -async def _reconcile(refund: Refund) -> None: +async def _reconcile(refund: Refund, now: int) -> None: if refund.method != "lightning": - # A cashu payout leaves no quote to query: the token either reached the - # client or was lost with the process. Close the claim as ``stuck`` so - # the balance stays withheld and the operator is told exactly once. + # No quote to query for cashu; withhold the balance and alert once. async with create_session() as session: if await _close(session, refund, status="stuck"): await session.commit() @@ -357,9 +350,13 @@ async def _reconcile(refund: Refund) -> None: return if refund.quote_id is None: - # No melt quote exists, so the mint was never asked to pay. + # Never sent, unless a quote appeared since the row was read. async with create_session() as session: - await release(session, refund) + if not await release(session, refund, require_no_quote=True): + logger.info( + "refund gained a melt quote during reconciliation; left open", + extra={"refund_id": refund.id}, + ) return status = await check_bolt11_payment_status( @@ -369,6 +366,13 @@ async def _reconcile(refund: Refund) -> None: async with create_session() as session: await settle(session, refund) elif status == "unpaid": + # A fresh melt can report unpaid briefly; trust it only after a timeout. + if refund.updated_at > now - settings.refund_claim_timeout_seconds: + logger.info( + "refund unpaid at the mint but too recent to release; waiting", + extra={"refund_id": refund.id, "updated_at": refund.updated_at}, + ) + return async with create_session() as session: await release(session, refund) else: @@ -379,11 +383,7 @@ async def _reconcile(refund: Refund) -> None: async def reconcile_once() -> None: - """Resolve open claims whose lease has lapsed. - - A fresh claim is leased to the request paying it out; ``hold`` drops that - lease so an ambiguous outcome is queried on the next pass. - """ + """Resolve open claims whose lease has lapsed.""" now = int(time.time()) lease_cutoff = now - settings.refund_claim_timeout_seconds async with create_session() as session: @@ -403,7 +403,7 @@ async def reconcile_once() -> None: if not await _lease(refund.id, now, lease_cutoff): continue try: - await _reconcile(refund) + await _reconcile(refund, now) except Exception as e: logger.error( "refund reconciliation failed", diff --git a/tests/integration/test_prune_dead_api_keys.py b/tests/integration/test_prune_dead_api_keys.py index aaacecad..d4b54570 100644 --- a/tests/integration/test_prune_dead_api_keys.py +++ b/tests/integration/test_prune_dead_api_keys.py @@ -293,3 +293,33 @@ async def test_periodic_prune_disabled_returns_immediately( await auth.periodic_dead_key_prune() sleep_mock.assert_not_called() + + +@pytest.mark.asyncio +async def test_refund_claim_protects_key(patched_db_engine: None) -> None: + """A key with a refund row is the audit anchor for its payout and holds a + non-null FK, so the janitor must leave it alone.""" + from routstr.core.db import Refund + + key = _dead_key(LONG_AGO) + async with create_session() as session: + session.add(key) + await session.commit() + session.add( + Refund( + api_key_hashed_key=key.hashed_key, + method="lightning", + destination="user@ln.example.com", + amount_msats=1000, + unit="sat", + mint_url="https://mint.example.com", + status="paid", + ) + ) + await session.commit() + + async with create_session() as session: + pruned = await prune_dead_api_keys(session, OLD) + + assert pruned == 0 + assert await _exists(key.hashed_key) diff --git a/tests/integration/test_refund_claims.py b/tests/integration/test_refund_claims.py index 0ba4d7fa..dac2f70a 100644 --- a/tests/integration/test_refund_claims.py +++ b/tests/integration/test_refund_claims.py @@ -57,6 +57,7 @@ async def _age_claim(session: AsyncSession, refund_id: str, seconds: int) -> Non row = await _load_refund(session, refund_id) row.claimed_at = int(time.time()) - seconds row.created_at = int(time.time()) - seconds + row.updated_at = int(time.time()) - seconds session.add(row) await session.commit() @@ -314,6 +315,7 @@ async def test_reconcile_ambiguous_claims( expected_balance: int, ) -> None: claim = await _open_ambiguous(integration_session, "quote-123") + await _age_claim(integration_session, claim.id, 600) with patch( "routstr.refund.check_bolt11_payment_status", AsyncMock(return_value=mint_status), @@ -330,7 +332,8 @@ async def test_reconcile_ambiguous_claims( async def test_reconcile_credits_balance_once_across_passes( integration_session: AsyncSession, patched_db_engine: None, short_timeout: None ) -> None: - await _open_ambiguous(integration_session, "quote-123") + claim = await _open_ambiguous(integration_session, "quote-123") + await _age_claim(integration_session, claim.id, 600) with patch( "routstr.refund.check_bolt11_payment_status", AsyncMock(return_value="unpaid") ): @@ -339,6 +342,59 @@ async def test_reconcile_credits_balance_once_across_passes( assert (await _load_key(integration_session)).balance == BALANCE_MSATS +@pytest.mark.asyncio +async def test_reconcile_waits_before_trusting_recent_unpaid( + integration_session: AsyncSession, patched_db_engine: None, short_timeout: None +) -> None: + """A just-dispatched melt can report unpaid before it turns pending, so an + unpaid quote is only final once the claim has been quiet for a timeout.""" + claim = await _open_ambiguous(integration_session, "quote-123") + with patch( + "routstr.refund.check_bolt11_payment_status", AsyncMock(return_value="unpaid") + ): + await refund.reconcile_once() + row = await _load_refund(integration_session, claim.id) + assert row.status == "ambiguous" + assert (await _load_key(integration_session)).balance == 0 + + await _age_claim(integration_session, claim.id, 600) + with patch( + "routstr.refund.check_bolt11_payment_status", AsyncMock(return_value="unpaid") + ): + await refund.reconcile_once() + row = await _load_refund(integration_session, claim.id) + assert row.status == "failed" + assert (await _load_key(integration_session)).balance == BALANCE_MSATS + + +@pytest.mark.asyncio +async def test_reconcile_keeps_claim_that_gained_quote_mid_pass( + integration_session: AsyncSession, patched_db_engine: None, short_timeout: None +) -> None: + """The payout records its quote between the reconciler's read and its + release: the stale ``quote_id is None`` must not restore the balance.""" + key = await _seed_key(integration_session) + claim = await refund.open_claim( + integration_session, key, method="lightning", destination=ADDRESS + ) + await _age_claim(integration_session, claim.id, 600) + + real_lease = refund._lease + + async def lease_then_quote(refund_id: str, now: int, cutoff: int) -> bool: + leased = await real_lease(refund_id, now, cutoff) + await refund.record_quote(claim, "late-quote") + return leased + + with patch("routstr.refund._lease", lease_then_quote): + await refund.reconcile_once() + + row = await _load_refund(integration_session, claim.id) + assert row.status == "pending" + assert row.quote_id == "late-quote" + assert (await _load_key(integration_session)).balance == 0 + + @pytest.mark.asyncio async def test_reconcile_leaves_fresh_pending_claim_alone( integration_session: AsyncSession, patched_db_engine: None, short_timeout: None