diff --git a/.gitignore b/.gitignore index f9db7ffb..c845a15c 100644 --- a/.gitignore +++ b/.gitignore @@ -38,3 +38,8 @@ proof_backups *.todo ui_out + +# local cashu wallet state (never commit) +.wallet/ +*.sqlite3-shm +*.sqlite3-wal diff --git a/migrations/versions/c7f1a2b3d4e5_add_balance_version_and_refund_tokens.py b/migrations/versions/c7f1a2b3d4e5_add_balance_version_and_refund_tokens.py new file mode 100644 index 00000000..6412b041 --- /dev/null +++ b/migrations/versions/c7f1a2b3d4e5_add_balance_version_and_refund_tokens.py @@ -0,0 +1,55 @@ +"""add balance_version to api_keys and refund_tokens table (core #412) + +Durable, balance-versioned refund idempotency: + * ``api_keys.balance_version`` is bumped atomically on every credit. + * ``refund_tokens`` records the refund issued at each (api_key_hash, + balance_version); an in-flight retry at the same version re-serves the + stored token, while any later credit (which bumps the version) invalidates + it. + +Revision ID: c7f1a2b3d4e5 +Revises: b5e7c9d1f3a2 +Create Date: 2026-06-20 00:00:00.000000 +""" + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision = "c7f1a2b3d4e5" +down_revision = "b5e7c9d1f3a2" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + # Existing rows default to 0; the first credit moves them to 1. There are no + # historical refund tokens to migrate (the old cache was in-memory only). + op.add_column( + "api_keys", + sa.Column( + "balance_version", + sa.Integer(), + nullable=False, + server_default="0", + ), + ) + op.create_table( + "refund_tokens", + sa.Column("api_key_hash", sa.String(), nullable=False), + sa.Column("balance_version", sa.Integer(), nullable=False), + sa.Column("token", sa.String(), nullable=False), + sa.Column("amount", sa.Integer(), nullable=False, server_default="0"), + sa.Column("unit", sa.String(), nullable=False, server_default="sat"), + sa.Column("created_at", sa.Integer(), nullable=False), + sa.Column("redeemed_at", sa.Integer(), nullable=True), + sa.ForeignKeyConstraint( + ["api_key_hash"], ["api_keys.hashed_key"] + ), + sa.PrimaryKeyConstraint("api_key_hash", "balance_version"), + ) + + +def downgrade() -> None: + op.drop_table("refund_tokens") + op.drop_column("api_keys", "balance_version") diff --git a/routstr/balance.py b/routstr/balance.py index 2f24cb8c..1580e1e1 100644 --- a/routstr/balance.py +++ b/routstr/balance.py @@ -14,6 +14,7 @@ from .core.db import ( ApiKey, AsyncSession, CashuTransaction, + RefundToken, get_session, store_cashu_transaction, ) @@ -192,6 +193,29 @@ def _cache_key_for_authorization(authorization: str) -> str: return hashlib.sha256(authorization.strip().encode()).hexdigest() +async def _stored_refund_for_version( + session: AsyncSession, hashed_key: str, balance_version: int +) -> RefundToken | None: + """Return the stored, UNREDEEMED refund token for this key at this exact + balance_version, or None. Durable, worker-agnostic replacement for the + in-process refund cache (core #412). + + A row exists only for the version at which it was minted; any credit bumps + balance_version, so a token minted at version k is structurally unreachable + once a topup moves the key to k+1. Within the same version, a token whose + ``redeemed_at`` is set is never re-served. + """ + row = await session.get(RefundToken, (hashed_key, balance_version)) + # Defensive isinstance guard: some unit tests inject a MagicMock session + # whose ``get`` returns an ApiKey regardless of the model queried. Only an + # actual RefundToken row counts as an idempotency hit. + if not isinstance(row, RefundToken): + return None + if row.redeemed_at is not None: + return None + return row + + async def _refund_cache_get(authorization: str) -> dict[str, str] | None: key = _cache_key_for_authorization(authorization) async with _refund_cache_lock: @@ -302,9 +326,31 @@ async def refund_wallet_endpoint( detail="Key not found. Deposit first via /v1/wallet/create before requesting a refund.", ) - if key.total_balance <= 0: - if cached := await _refund_cache_get(bearer_value): - return cached + # Durable, balance-versioned idempotency (core #412). Capture the version + # the refund is being computed against. A stored refund token is only + # re-served at the SAME balance_version and only if still unredeemed: + # * in-flight retry (no intervening credit) -> same version -> same token + # (true idempotency, balance debited exactly once); + # * after ANY topup (cashu OR Lightning OR new-key credit) the version was + # bumped, so the prior token is unreachable and never re-served. + current_balance_version: int = key.balance_version + stored = await _stored_refund_for_version( + session, key.hashed_key, current_balance_version + ) + if stored is not None: + result: dict[str, str] = {"token": stored.token} + if stored.unit == "sat": + result["sats"] = str(stored.amount) + else: + result["msats"] = str(stored.amount) + logger.info( + "refund_wallet_endpoint: re-serving stored refund (idempotent)", + extra={ + "hashed_key": key.hashed_key, + "balance_version": current_balance_version, + }, + ) + return result if key.parent_key_hash: raise HTTPException( @@ -450,7 +496,36 @@ async def refund_wallet_endpoint( else: raise HTTPException(status_code=500, detail=f"Refund failed: {error_msg}") - await _refund_cache_set(bearer_value, result) + # Durably record the issued refund token keyed on (hashed_key, + # balance_version) so an in-flight retry at the SAME version re-serves this + # exact token instead of minting a second one (core #412). Only cashu-token + # refunds are re-servable; LNURL refunds ("recipient") have no token to + # re-serve. The PK (hashed_key, balance_version) makes a concurrent + # duplicate insert at the same version fail loudly rather than double-mint. + if "token" in result: + try: + session.add( + RefundToken( + api_key_hash=key.hashed_key, + balance_version=current_balance_version, + token=result["token"], + amount=remaining_balance, + unit=key.refund_currency or "sat", + ) + ) + await session.commit() + except Exception: + # A row already exists for this version (concurrent in-flight + # refund won the race) — roll back our duplicate; the stored token + # is authoritative and will be re-served on the client's retry. + await session.rollback() + logger.warning( + "refund_wallet_endpoint: refund token already stored for version", + extra={ + "hashed_key": key.hashed_key, + "balance_version": current_balance_version, + }, + ) if "token" in result: try: diff --git a/routstr/core/db.py b/routstr/core/db.py index b0510fb6..edcf2e6c 100644 --- a/routstr/core/db.py +++ b/routstr/core/db.py @@ -88,12 +88,75 @@ class ApiKey(SQLModel, table=True): # type: ignore default=None, description="Unix timestamp after which the key is no longer valid", ) + balance_version: int = Field( + default=0, + nullable=False, + description=( + "Monotonic counter bumped atomically on EVERY credit to this key " + "(cashu topup, Lightning topup, new-key credit). Refund idempotency " + "is keyed on (hashed_key, balance_version): a refund minted at " + "version k is invalidated by any later credit, which moves the key " + "to k+1. See core #412." + ), + ) @property def total_balance(self) -> int: return self.balance - self.reserved_balance +class RefundToken(SQLModel, table=True): # type: ignore + """Durable, balance-versioned record of issued refund tokens (core #412). + + One row per (api_key_hash, balance_version). A refund issued while the key + is at balance_version=k is stored here; an in-flight retry at the same + version re-serves the exact same token (true idempotency). Any credit bumps + balance_version, so the row at k can never be re-served after a topup. + Once the gateway/client confirms redemption, ``redeemed_at`` is set so even + within the same version a redeemed token is never re-served. + """ + + __tablename__ = "refund_tokens" + + api_key_hash: str = Field(primary_key=True, foreign_key="api_keys.hashed_key") + balance_version: int = Field(primary_key=True) + token: str = Field(description="Serialized refund token re-served on retry") + amount: int = Field(default=0, description="Refunded amount in the key's unit") + unit: str = Field(default="sat", description="Refund unit (sat or msat)") + created_at: int = Field(default_factory=lambda: int(time.time())) + redeemed_at: int | None = Field( + default=None, + description="Set once the refund token's proofs are observed spent / acked", + ) + + +async def credit_key_balance( + session: AsyncSession, hashed_key: str, amount_msats: int +) -> int: + """Atomically credit ``amount_msats`` to a key AND bump its balance_version. + + This is the single choke point that invalidates all prior refund tokens for + the key, regardless of how the credit arrived (cashu topup, Lightning topup, + new-key credit) or which worker handles it (state lives in the shared DB). + See core #412. + + Does NOT commit — the caller owns the transaction so the credit and any + surrounding work (e.g. marking a Lightning invoice paid) are atomic. + + Returns the number of rows updated (1 on success, 0 if the key vanished). + """ + stmt = ( + update(ApiKey) + .where(col(ApiKey.hashed_key) == hashed_key) + .values( + balance=col(ApiKey.balance) + amount_msats, + balance_version=col(ApiKey.balance_version) + 1, + ) + ) + result = await session.exec(stmt) # type: ignore[call-overload] + return int(result.rowcount or 0) + + async def reset_all_reserved_balances(session: AsyncSession) -> None: stmt = update(ApiKey).values(reserved_balance=0, reserved_at=None) await session.exec(stmt) # type: ignore[call-overload] diff --git a/routstr/lightning.py b/routstr/lightning.py index b0bbc63b..439604fc 100644 --- a/routstr/lightning.py +++ b/routstr/lightning.py @@ -8,7 +8,13 @@ from pydantic import BaseModel, Field from sqlmodel import col, select from sqlmodel.ext.asyncio.session import AsyncSession -from .core.db import ApiKey, LightningInvoice, create_session, get_session +from .core.db import ( + ApiKey, + LightningInvoice, + create_session, + credit_key_balance, + get_session, +) from .core.logging import get_logger from .core.settings import settings from .wallet import get_wallet @@ -293,7 +299,17 @@ async def topup_api_key_from_invoice( if not api_key: raise ValueError("Associated API key not found") - api_key.balance += invoice.amount_sats * 1000 # Convert to msats + # Atomically credit AND bump balance_version in the same DB txn. This is the + # load-bearing fix for core #412 on the Lightning path: this credit runs in + # the background invoice watcher with NO bearer/authorization in scope, so a + # header-keyed refund-cache invalidation cannot fire here. Bumping + # balance_version is data-driven and worker-agnostic, so the next refund + # recomputes against the new version and never re-serves the stale token. + updated = await credit_key_balance( + session, api_key.hashed_key, invoice.amount_sats * 1000 + ) + if updated == 0: + raise ValueError("Associated API key not found") await session.flush() diff --git a/routstr/wallet.py b/routstr/wallet.py index 30f36ded..39724533 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -8,7 +8,7 @@ from cashu.core.mint_info import MintInfo as _CashuMintInfo from cashu.wallet.helpers import deserialize_token_from_string from cashu.wallet.wallet import Wallet from pydantic_core import PydanticUndefined -from sqlmodel import col, select, update +from sqlmodel import select from .core import db, get_logger from .core.db import store_cashu_transaction @@ -435,13 +435,11 @@ async def credit_balance( extra={"old_balance": key.balance, "credit_amount": amount}, ) - # Use atomic SQL UPDATE to prevent race conditions during concurrent topups - stmt = ( - update(db.ApiKey) - .where(col(db.ApiKey.hashed_key) == key.hashed_key) - .values(balance=(db.ApiKey.balance) + amount) - ) - await session.exec(stmt) # type: ignore[call-overload] + # Use atomic SQL UPDATE to prevent race conditions during concurrent + # topups. credit_key_balance also bumps balance_version in the same + # statement, which invalidates any prior refund token for this key + # (core #412) — regardless of worker or auth context. + await db.credit_key_balance(session, key.hashed_key, amount) await session.commit() await session.refresh(key) diff --git a/tests/unit/test_refund_idempotency_durable.py b/tests/unit/test_refund_idempotency_durable.py new file mode 100644 index 00000000..4fdc59ed --- /dev/null +++ b/tests/unit/test_refund_idempotency_durable.py @@ -0,0 +1,273 @@ +"""Durable, balance-versioned refund idempotency — core #412. + +These tests exercise the full ``POST /refund`` -> topup -> spend -> ``POST /refund`` +sequence against a **real in-memory SQLite database** (mints/network mocked). + +The bug (#412): the refund endpoint cached the issued refund token keyed by +``sha256(bearer)`` and re-served it whenever ``total_balance <= 0``. After a +refund (token T1, balance debited to 0), a *topup* followed by a *spend* brings +the balance back to 0 — but the cache still holds T1, so a second refund +re-serves the already-spent T1 -> "proofs already spent" / fund loss. + +The Lightning topup path is the load-bearing gap: it credits balance from a +background watcher with no bearer/authorization in scope, so a header-keyed +cache invalidation structurally cannot fire there. + +The durable fix bumps ``ApiKey.balance_version`` atomically on *every* credit +(cashu topup, Lightning topup, new-key credit) and keys refund idempotency on +``(api_key_hash, balance_version)`` in a DB table, so any credit anywhere +invalidates prior refund tokens regardless of worker or auth context. +""" + +from __future__ import annotations + +import os +from typing import AsyncGenerator +from unittest.mock import AsyncMock, patch + +import pytest +import pytest_asyncio +from sqlalchemy.ext.asyncio import create_async_engine +from sqlmodel import SQLModel +from sqlmodel.ext.asyncio.session import AsyncSession + +os.environ.setdefault("DATABASE_URL", "sqlite+aiosqlite:///:memory:") +os.environ.setdefault("CASHU_MINTS", "https://mint.example.com") +os.environ.setdefault("NSEC", "nsec1testkey") + +import routstr.balance as balance_mod # noqa: E402 +from routstr.core.db import ApiKey # noqa: E402 +from routstr.lightning import topup_api_key_from_invoice # noqa: E402 +from routstr.wallet import credit_balance # noqa: E402 + + +@pytest_asyncio.fixture +async def engine(): # type: ignore[no-untyped-def] + eng = create_async_engine("sqlite+aiosqlite:///:memory:") + async with eng.begin() as conn: + await conn.run_sync(SQLModel.metadata.create_all) + yield eng + await eng.dispose() + + +@pytest_asyncio.fixture +async def session(engine) -> AsyncGenerator[AsyncSession, None]: # type: ignore[no-untyped-def] + async with AsyncSession(engine, expire_on_commit=False) as s: + yield s + + +async def _new_key(session: AsyncSession, hashed_key: str, balance: int) -> ApiKey: + key = ApiKey( + hashed_key=hashed_key, + balance=balance, + refund_currency="sat", + refund_mint_url="https://mint.example.com", + ) + session.add(key) + await session.commit() + await session.refresh(key) + return key + + +async def _spend_to_zero(session: AsyncSession, hashed_key: str) -> None: + """Simulate the user spending their whole balance (no version bump).""" + key = await session.get(ApiKey, hashed_key) + assert key is not None + key.balance = 0 + session.add(key) + await session.commit() + + +def _clear_refund_cache() -> None: + # Defensively clear the legacy in-process cache if it still exists. + cache = getattr(balance_mod, "_refund_cache", None) + if isinstance(cache, dict): + cache.clear() + + +# --------------------------------------------------------------------------- +# Token minting fakes: each call returns a fresh, unique token so we can assert +# whether a *new* token was minted (correct) or a *stale* one re-served (bug). +# --------------------------------------------------------------------------- + + +class TokenMinter: + def __init__(self) -> None: + self.counter = 0 + self.minted: list[str] = [] + + async def __call__(self, amount: int, unit: str, mint_url: str | None = None) -> str: + self.counter += 1 + tok = f"cashuMINTED_{self.counter}_amt{amount}" + self.minted.append(tok) + return tok + + +async def _do_refund(session: AsyncSession, bearer: str) -> dict: + result = await balance_mod.refund_wallet_endpoint( + authorization=f"Bearer {bearer}", + x_cashu=None, + session=session, + ) + assert isinstance(result, dict), f"expected dict refund result, got {result!r}" + return result + + +async def _refund_token_or_none(session: AsyncSession, bearer: str) -> str | None: + """Return the refund token a /refund call yields, or None if the endpoint + declines (e.g. 400 'No balance to refund'). The #412 bug manifests as a + *stale token string* being returned where None (or a fresh token) is + correct.""" + from fastapi import HTTPException + + try: + result = await balance_mod.refund_wallet_endpoint( + authorization=f"Bearer {bearer}", + x_cashu=None, + session=session, + ) + except HTTPException: + return None + if isinstance(result, dict): + return result.get("token") + return None + + +# =========================================================================== +# 1. #412 across CASHU topup +# =========================================================================== + + +@pytest.mark.asyncio +async def test_412_cashu_topup_second_refund_is_not_stale_token(session: AsyncSession) -> None: + _clear_refund_cache() + hashed = "cashu412hash" + bearer = f"sk-{hashed}" + await _new_key(session, hashed, balance=1000) # 1000 msats + + minter = TokenMinter() + + async def fake_recieve(token: str) -> tuple[int, str, str]: + # crediting 1 sat = 1000 msats + return 1, "sat", "https://mint.example.com" + + with ( + patch("routstr.balance.send_token", minter), + patch("routstr.balance.get_billing_key", AsyncMock(side_effect=lambda k, s: k)), + patch("routstr.balance.store_cashu_transaction", AsyncMock()), + patch("routstr.wallet.recieve_token", AsyncMock(side_effect=fake_recieve)), + patch("routstr.wallet.store_cashu_transaction", AsyncMock()), + ): + # 1) First refund: mints T1, debits balance to 0, caches/stores T1. + r1 = await _do_refund(session, bearer) + t1 = r1["token"] + assert t1 == "cashuMINTED_1_amt1" + + # 2) Cashu topup: credit balance again (user adds funds & uses the key). + key = await session.get(ApiKey, hashed) + assert key is not None + await credit_balance("cashuAtopup", key, session) + + # 3) Spend the topped-up balance back to zero (normal usage). + await _spend_to_zero(session, hashed) + + # 4) Second refund while balance == 0. The legacy cache serves the + # STALE T1 here (fund loss: T1's proofs were already spent/minted + # against and the user may have already redeemed it). The durable + # fix must NOT re-serve T1 — the topup bumped balance_version, so + # the stored T1 is no longer valid at the current version (and with + # balance 0 there is nothing to refund -> declines). + t2 = await _refund_token_or_none(session, bearer) + + assert t2 != t1, ( + f"#412 REGRESSION: second refund re-served stale token {t1!r} " + f"after a cashu topup + spend. The topup must invalidate T1." + ) + + +# =========================================================================== +# 2. #412 across LIGHTNING topup (the load-bearing gap) +# =========================================================================== + + +@pytest.mark.asyncio +async def test_412_lightning_topup_second_refund_is_not_stale_token(session: AsyncSession) -> None: + _clear_refund_cache() + hashed = "ln412hash" + bearer = f"sk-{hashed}" + await _new_key(session, hashed, balance=1000) + + minter = TokenMinter() + + # Fake the cashu wallet used by topup_api_key_from_invoice -> wallet.mint + fake_wallet = AsyncMock() + fake_wallet.mint = AsyncMock(return_value=None) + + class _Invoice: + amount_sats = 1 + api_key_hash = hashed + id = "inv1" + payment_hash = "ph1" + + with ( + patch("routstr.balance.send_token", minter), + patch("routstr.balance.get_billing_key", AsyncMock(side_effect=lambda k, s: k)), + patch("routstr.balance.store_cashu_transaction", AsyncMock()), + patch("routstr.lightning.get_wallet", AsyncMock(return_value=fake_wallet)), + ): + # 1) First refund: mint T1, debit to 0, cache/store T1. + r1 = await _do_refund(session, bearer) + t1 = r1["token"] + + # 2) LIGHTNING topup via the background-watcher code path. No bearer in + # scope here — a header-keyed cache invalidation cannot fire. The + # durable fix must bump balance_version inside this DB txn. + await topup_api_key_from_invoice(_Invoice(), session) # type: ignore[arg-type] + await session.commit() + + # 3) Spend the Lightning-topped-up balance back to zero. + await _spend_to_zero(session, hashed) + + # 4) Second refund at balance 0: must NOT re-serve the stale T1. + t2 = await _refund_token_or_none(session, bearer) + + assert t2 != t1, ( + f"#412 REGRESSION (Lightning): second refund re-served stale token " + f"{t1!r} after a Lightning topup. The Lightning credit must bump " + f"balance_version so the cached/stored refund token is invalidated." + ) + + +# =========================================================================== +# 3. In-flight idempotency: two refunds at the SAME balance_version (no +# intervening credit) return the SAME token and debit the balance once. +# =========================================================================== + + +@pytest.mark.asyncio +async def test_inflight_idempotency_same_version_same_token(session: AsyncSession) -> None: + _clear_refund_cache() + hashed = "inflighthash" + bearer = f"sk-{hashed}" + await _new_key(session, hashed, balance=2000) + + minter = TokenMinter() + + with ( + patch("routstr.balance.send_token", minter), + patch("routstr.balance.get_billing_key", AsyncMock(side_effect=lambda k, s: k)), + patch("routstr.balance.store_cashu_transaction", AsyncMock()), + ): + r1 = await _do_refund(session, bearer) + # Second refund with NO intervening credit -> same balance_version. + # The balance is now 0; idempotency must re-serve the same token. + r2 = await _do_refund(session, bearer) + + assert r1["token"] == r2["token"], ( + "in-flight idempotency broken: two refunds at the same balance_version " + "with no intervening credit must return the SAME token" + ) + # Exactly one token minted -> balance debited exactly once. + assert len(minter.minted) == 1, ( + f"idempotent retry must NOT mint a second token; minted={minter.minted}" + )