diff --git a/migrations/versions/c5d7e9f1a3b5_add_pending_swaps_table.py b/migrations/versions/c5d7e9f1a3b5_add_pending_swaps_table.py deleted file mode 100644 index 61d7d3e2..00000000 --- a/migrations/versions/c5d7e9f1a3b5_add_pending_swaps_table.py +++ /dev/null @@ -1,46 +0,0 @@ -"""add pending_swaps table - -Revision ID: c5d7e9f1a3b5 -Revises: b4f7a1c9d2e3 -Create Date: 2026-08-18 00:00:00.000000 -""" - -import sqlalchemy as sa -from alembic import op - -# revision identifiers, used by Alembic. -revision = "c5d7e9f1a3b5" -down_revision = "b4f7a1c9d2e3" -branch_labels = None -depends_on = None - - -def upgrade() -> None: - op.create_table( - "pending_swaps", - sa.Column("id", sa.String(), primary_key=True), - sa.Column("created_at", sa.Integer(), nullable=False), - sa.Column("updated_at", sa.Integer(), nullable=False), - sa.Column("source_mint", sa.String(), nullable=False), - sa.Column("source_unit", sa.String(), nullable=False), - sa.Column("melt_quote_id", sa.String(), nullable=False), - sa.Column("dest_mint", sa.String(), nullable=False), - sa.Column("dest_unit", sa.String(), nullable=False), - sa.Column("mint_quote_id", sa.String(), nullable=False), - sa.Column("minted_amount", sa.Integer(), nullable=False), - sa.Column("key_hashed_key", sa.String(), nullable=True), - sa.Column("token", sa.String(), nullable=True), - sa.Column("state", sa.String(), nullable=False, server_default="pending"), - sa.Column("attempts", sa.Integer(), nullable=False, server_default="0"), - sa.Column("last_error", sa.String(), nullable=True), - ) - op.create_index( - "ix_pending_swaps_key_hashed_key", "pending_swaps", ["key_hashed_key"] - ) - op.create_index("ix_pending_swaps_state", "pending_swaps", ["state"]) - - -def downgrade() -> None: - op.drop_index("ix_pending_swaps_state", table_name="pending_swaps") - op.drop_index("ix_pending_swaps_key_hashed_key", table_name="pending_swaps") - op.drop_table("pending_swaps") diff --git a/routstr/core/db.py b/routstr/core/db.py index b025b2d5..d7679933 100644 --- a/routstr/core/db.py +++ b/routstr/core/db.py @@ -503,46 +503,6 @@ class CashuTransaction(SQLModel, table=True): # type: ignore ) -class PendingSwap(SQLModel, table=True): # type: ignore - """Durable checkpoint for an in-flight cross-mint swap. - - Written just before the source-mint melt is dispatched — the point after - which a failure becomes ambiguous (the Lightning payment may still settle). - A background reconciler polls these rows: a PAID melt is completed (mint on - destination + credit the API key) and an UNPAID/expired one is dropped so - the token can be retried. Rows for swaps that finish in-line are deleted - immediately, so the table only holds unresolved ambiguity. - """ - - __tablename__ = "pending_swaps" - - id: str = Field(primary_key=True, default_factory=lambda: uuid.uuid4().hex) - created_at: int = Field(default_factory=lambda: int(time.time())) - updated_at: int = Field(default_factory=lambda: int(time.time())) - source_mint: str = Field(description="Mint holding the melted proofs") - source_unit: str = Field(description="Unit of the source token") - melt_quote_id: str = Field(description="Melt quote id on the source mint") - dest_mint: str = Field(description="Destination mint for the swap") - dest_unit: str = Field(description="Unit minted on the destination") - mint_quote_id: str = Field(description="Mint quote id on the destination mint") - minted_amount: int = Field(description="Amount to mint, in dest_unit") - key_hashed_key: str | None = Field( - default=None, - index=True, - description="API key to credit once the swap completes", - ) - token: str | None = Field( - default=None, description="Original serialized Cashu token, for history" - ) - state: str = Field( - default="pending", - index=True, - description="pending | melt_confirmed | stale", - ) - attempts: int = Field(default=0) - last_error: str | None = Field(default=None) - - async def store_cashu_transaction( token: str, amount: int, diff --git a/routstr/core/main.py b/routstr/core/main.py index 20473202..979f5cdb 100644 --- a/routstr/core/main.py +++ b/routstr/core/main.py @@ -30,12 +30,7 @@ from ..proxy import initialize_upstreams, proxy_router, refresh_model_maps_perio from ..upstream.auto_topup import periodic_auto_topup from ..upstream.deepseek_v4_pricing_shim import register_deepseek_v4_pricing from ..upstream.litellm_routing import configure_litellm -from ..wallet import ( - periodic_payout, - periodic_refund_sweep, - periodic_routstr_fee_payout, - periodic_swap_reconciliation, -) +from ..wallet import periodic_payout, periodic_refund_sweep, periodic_routstr_fee_payout from .admin import admin_router from .db import create_session, init_db, run_migrations from .exceptions import general_exception_handler, http_exception_handler @@ -71,7 +66,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]: refund_sweep_task = None routstr_fee_task = None invoice_watcher_task = None - swap_reconciliation_task = None try: # Apply litellm-wide settings (drop_params, chat-completions URL, @@ -157,7 +151,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]: refund_sweep_task = asyncio.create_task(periodic_refund_sweep()) routstr_fee_task = asyncio.create_task(periodic_routstr_fee_payout()) invoice_watcher_task = asyncio.create_task(periodic_invoice_watcher()) - swap_reconciliation_task = asyncio.create_task(periodic_swap_reconciliation()) yield @@ -205,8 +198,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]: routstr_fee_task.cancel() if invoice_watcher_task is not None: invoice_watcher_task.cancel() - if swap_reconciliation_task is not None: - swap_reconciliation_task.cancel() try: tasks_to_wait = [] diff --git a/routstr/wallet.py b/routstr/wallet.py index 4247b1e9..383c79c8 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -64,12 +64,6 @@ _WALLET_OPERATION_LOCK = Path(".wallet") / ".routstr-operation.lock" _wallet_operation_depth: ContextVar[int] = ContextVar( "wallet_operation_depth", default=0 ) -# (key_hashed_key, serialized_token) of the credit_balance call currently on -# the stack, so swap_to_trusted_mint can checkpoint who to credit if the swap -# is interrupted between melt dispatch and the balance update. -_swap_credit_context: ContextVar[tuple[str, str] | None] = ContextVar( - "swap_credit_context", default=None -) @asynccontextmanager @@ -1298,86 +1292,6 @@ async def _confirm_melt_paid( return await _reconcile_ambiguous_melt(wallet, quote_id, proofs) -async def _persist_pending_swap( - *, - source_mint: str, - source_unit: str, - melt_quote_id: str, - dest_mint: str, - dest_unit: str, - mint_quote_id: str, - minted_amount: int, -) -> str | None: - """Checkpoint an about-to-be-dispatched melt. Best-effort: a persistence - failure must not block the swap, it only loses the automatic-recovery net.""" - credit_ctx = _swap_credit_context.get() - try: - row = db.PendingSwap( - source_mint=source_mint, - source_unit=source_unit, - melt_quote_id=melt_quote_id, - dest_mint=dest_mint, - dest_unit=dest_unit, - mint_quote_id=mint_quote_id, - minted_amount=minted_amount, - key_hashed_key=credit_ctx[0] if credit_ctx else None, - token=credit_ctx[1] if credit_ctx else None, - ) - async with db.create_session() as session: - session.add(row) - await session.commit() - return row.id - except Exception as e: - logger.warning( - "pending_swap: failed to persist checkpoint before melt dispatch", - extra={"error": str(e), "melt_quote_id": melt_quote_id}, - ) - return None - - -async def _mark_pending_swap( - swap_id: str | None, *, state: str, error: str | None = None -) -> None: - if swap_id is None: - return - try: - async with db.create_session() as session: - values: dict[str, object] = { - "state": state, - "updated_at": int(time.time()), - } - if error is not None: - values["last_error"] = error[:500] - stmt = ( - update(db.PendingSwap) - .where(col(db.PendingSwap.id) == swap_id) - .values(**values) - ) - await session.exec(stmt) # type: ignore[call-overload] - await session.commit() - except Exception as e: - logger.warning( - "pending_swap: failed to update checkpoint state", - extra={"error": str(e), "swap_id": swap_id, "state": state}, - ) - - -async def _delete_pending_swap(swap_id: str | None) -> None: - if swap_id is None: - return - try: - async with db.create_session() as session: - row = await session.get(db.PendingSwap, swap_id) - if row is not None: - await session.delete(row) - await session.commit() - except Exception as e: - logger.warning( - "pending_swap: failed to delete checkpoint", - extra={"error": str(e), "swap_id": swap_id}, - ) - - async def swap_to_trusted_mint( token_obj: Token, token_wallet: Wallet, @@ -1451,7 +1365,6 @@ async def swap_to_trusted_mint( # amount recomputed from the fees the mint actually demands. observed_extra_fee = 0 attempt = 0 - pending_swap_id: str | None = None dest_wallet = primary_wallet dest_mint_url = settings.primary_mint while True: @@ -1573,18 +1486,6 @@ async def swap_to_trusted_mint( minted_amount = recomputed continue - # Durable checkpoint: from the moment the melt hits the wire, a crash - # or ambiguous failure leaves real money in flight. The reconciler - # completes or discards this row if we never get to delete it in-line. - pending_swap_id = await _persist_pending_swap( - source_mint=token_obj.mint, - source_unit=token_obj.unit, - melt_quote_id=melt_quote.quote, - dest_mint=dest_mint_url, - dest_unit=settings.primary_mint_unit, - mint_quote_id=mint_quote.quote, - minted_amount=int(minted_amount), - ) try: melt_response = await run_mint_operation( lambda: token_wallet.melt( @@ -1606,7 +1507,6 @@ async def swap_to_trusted_mint( if isinstance(e, TokenConsumedError): raise if _melt_definitively_failed(e): - await _delete_pending_swap(pending_swap_id) raise ValueError( f"Failed to melt token from foreign mint {token_obj.mint}: {e}" ) from e @@ -1647,11 +1547,9 @@ async def swap_to_trusted_mint( "attempts": attempt, }, ) - await _delete_pending_swap(pending_swap_id) raise ValueError( f"Failed to melt token from foreign mint {token_obj.mint}: {e}" ) from e - await _delete_pending_swap(pending_swap_id) logger.warning( "swap_to_trusted_mint: mint demanded more than quoted at melt, retrying", extra={ @@ -1665,7 +1563,6 @@ async def swap_to_trusted_mint( break - await _mark_pending_swap(pending_swap_id, state="melt_confirmed") logger.info( "Source melt succeeded; minting on destination", extra={ @@ -1718,11 +1615,6 @@ async def swap_to_trusted_mint( # Recovery scan ran but did NOT restore the orphaned proofs # (mint reports them as spent — they're stuck). Refuse to # credit the API key balance for proofs we don't actually hold. - await _mark_pending_swap( - pending_swap_id, - state="stale", - error="outputs signed but proofs unrecoverable", - ) raise TokenConsumedError( f"Swap recovery failed: mint signed outputs but proofs are " f"unrecoverable (mint reports them spent). " @@ -1767,8 +1659,6 @@ async def swap_to_trusted_mint( }, ) - await _delete_pending_swap(pending_swap_id) - return int(minted_amount), settings.primary_mint_unit, dest_mint_url @@ -1783,11 +1673,7 @@ async def credit_balance( cashu_token: str, key: db.ApiKey, session: db.AsyncSession ) -> int: async with wallet_operation_guard(): - ctx_token = _swap_credit_context.set((key.hashed_key, cashu_token)) - try: - return await _credit_balance_locked(cashu_token, key, session) - finally: - _swap_credit_context.reset(ctx_token) + return await _credit_balance_locked(cashu_token, key, session) async def _credit_balance_locked( @@ -2518,232 +2404,6 @@ async def refund_sweep_once() -> None: await _refund_sweep_once(cutoff) -_SWAP_RECONCILE_INTERVAL_SECONDS = 5 * 60 -_SWAP_RECONCILE_MIN_AGE_SECONDS = 120 -_SWAP_RECONCILE_MAX_ATTEMPTS = 96 -_SWAP_UNPAID_GIVEUP_AGE_SECONDS = 60 * 60 - - -async def _bump_pending_swap(row: "db.PendingSwap", *, error: str | None) -> None: - """Record one more unresolved reconciliation attempt; park as stale at cap.""" - if row.attempts + 1 >= _SWAP_RECONCILE_MAX_ATTEMPTS: - logger.critical( - "swap_reconciliation: giving up on checkpoint; manual reconciliation required", - extra={ - "swap_id": row.id, - "source_mint": row.source_mint, - "melt_quote_id": row.melt_quote_id, - "dest_mint": row.dest_mint, - "mint_quote_id": row.mint_quote_id, - "minted_amount": row.minted_amount, - "last_error": error, - }, - ) - try: - async with db.create_session() as session: - values: dict[str, object] = { - "attempts": db.PendingSwap.attempts + 1, - "updated_at": int(time.time()), - } - if error is not None: - values["last_error"] = error[:500] - if row.attempts + 1 >= _SWAP_RECONCILE_MAX_ATTEMPTS: - values["state"] = "stale" - stmt = ( - update(db.PendingSwap) - .where(col(db.PendingSwap.id) == row.id) - .values(**values) - ) - await session.exec(stmt) # type: ignore[call-overload] - await session.commit() - except Exception as e: - logger.warning( - "swap_reconciliation: failed to bump checkpoint attempts", - extra={"error": str(e), "swap_id": row.id}, - ) - - -async def _credit_reconciled_swap(row: "db.PendingSwap") -> None: - amount_msat = ( - _sats_to_msats(row.minted_amount) - if row.dest_unit == "sat" - else row.minted_amount - ) - credited_key = row.key_hashed_key - if credited_key: - async with db.create_session() as session: - stmt = ( - update(db.ApiKey) - .where(col(db.ApiKey.hashed_key) == credited_key) - .values(balance=db.ApiKey.balance + amount_msat) - ) - result = await session.exec(stmt) # type: ignore[call-overload] - if (getattr(result, "rowcount", 0) or 0) == 0: - # The failed request rolled back the freshly-created key row. - # Recreate it: auth hashes the same bearer token back to this - # hashed_key, so the user reaches this balance by re-presenting - # the token they already paid with. - session.add( - db.ApiKey( - hashed_key=credited_key, - balance=amount_msat, - refund_mint_url=row.dest_mint, - refund_currency=row.dest_unit, - ) - ) - await session.commit() - if row.token: - await store_cashu_transaction( - token=row.token, - amount=row.minted_amount, - unit=row.dest_unit, - mint_url=row.dest_mint, - typ="in", - source="apikey", - api_key_hashed_key=credited_key, - ) - else: - logger.warning( - "swap_reconciliation: recovered swap has no API key to credit; " - "minted funds remain in the node wallet", - extra={"swap_id": row.id, "minted_amount": row.minted_amount}, - ) - await _delete_pending_swap(row.id) - logger.info( - "swap_reconciliation: recovered interrupted swap", - extra={ - "event": "cashu_swap_reconciled", - "swap_id": row.id, - "credited_key": (credited_key or "")[:8], - "amount_msat": amount_msat, - "dest_mint": row.dest_mint, - }, - ) - - -async def _reconcile_pending_swap(row: "db.PendingSwap") -> None: - async with wallet_operation_guard(): - state = row.state - if state == "pending": - source_wallet = await get_wallet( - row.source_mint, unit=row.source_unit, load=False - ) - try: - quote = await run_mint_operation( - lambda: source_wallet.get_melt_quote(row.melt_quote_id), - op_name="reconcile_pending_swap_melt_quote", - mint_url=row.source_mint, - retry_timeouts=False, - ) - except Exception as e: - await _bump_pending_swap(row, error=str(e)) - return - quote_state = getattr(quote, "state", None) - if quote is not None and quote_state == MeltQuoteState.paid: - await _mark_pending_swap(row.id, state="melt_confirmed") - state = "melt_confirmed" - elif quote is None or quote_state == MeltQuoteState.unpaid: - # Still UNPAID long after dispatch: the Lightning payment never - # happened, the source proofs were not consumed, and the user - # can safely re-present the token. Drop the checkpoint. - if int(time.time()) - row.created_at >= _SWAP_UNPAID_GIVEUP_AGE_SECONDS: - logger.info( - "swap_reconciliation: melt never settled; dropping checkpoint", - extra={ - "swap_id": row.id, - "source_mint": row.source_mint, - "melt_quote_id": row.melt_quote_id, - }, - ) - await _delete_pending_swap(row.id) - else: - await _bump_pending_swap(row, error="melt quote still unpaid") - return - else: - await _bump_pending_swap( - row, - error=f"melt quote state: {getattr(quote_state, 'value', 'unknown')}", - ) - return - - if state == "melt_confirmed": - dest_wallet = await get_wallet(row.dest_mint, unit=row.dest_unit) - try: - await run_mint_operation( - lambda: dest_wallet.mint( - row.minted_amount, quote_id=row.mint_quote_id - ), - op_name="reconcile_pending_swap_mint", - mint_url=row.dest_mint, - retry_timeouts=False, - ) - except Exception as e: - msg = str(e) - lowered = msg.lower() - if ( - "11003" in msg - or "outputs already signed" in lowered - or "already issued" in lowered - ): - # The quote was (partially) used by the crashed in-line - # attempt; blind re-minting risks double counting. Park for - # manual reconciliation. - logger.critical( - "swap_reconciliation: destination quote already used; " - "manual reconciliation required", - extra={ - "swap_id": row.id, - "dest_mint": row.dest_mint, - "mint_quote_id": row.mint_quote_id, - "error": msg, - }, - ) - await _mark_pending_swap(row.id, state="stale", error=msg) - else: - await _bump_pending_swap(row, error=msg) - return - await _credit_reconciled_swap(row) - - -async def reconcile_pending_swaps_once() -> None: - now = int(time.time()) - async with db.create_session() as session: - result = await session.exec( - select(db.PendingSwap).where( - col(db.PendingSwap.state).in_(["pending", "melt_confirmed"]) - ) - ) - rows = list(result.all()) - for row in rows: - # Give the in-line swap time to finish (and delete its row) before - # a second worker starts poking the mints about it. - if now - row.created_at < _SWAP_RECONCILE_MIN_AGE_SECONDS: - continue - try: - await _reconcile_pending_swap(row) - except Exception as e: - logger.error( - "swap_reconciliation: error processing checkpoint", - extra={ - "swap_id": row.id, - "error": str(e), - "error_type": type(e).__name__, - }, - ) - - -async def periodic_swap_reconciliation() -> None: - while True: - await asyncio.sleep(_SWAP_RECONCILE_INTERVAL_SECONDS) - try: - await reconcile_pending_swaps_once() - except Exception as e: - logger.error( - "Error in periodic swap reconciliation", - extra={"error": str(e), "error_type": type(e).__name__}, - ) - - async def periodic_refund_sweep() -> None: while True: await asyncio.sleep(60 * 60) # every hour diff --git a/tests/unit/test_pending_swap_reconciliation.py b/tests/unit/test_pending_swap_reconciliation.py deleted file mode 100644 index 1ef48035..00000000 --- a/tests/unit/test_pending_swap_reconciliation.py +++ /dev/null @@ -1,264 +0,0 @@ -"""Reconciliation of interrupted cross-mint swaps (pending_swaps checkpoints).""" - -import time -from collections.abc import AsyncGenerator -from contextlib import asynccontextmanager -from types import SimpleNamespace -from typing import Any - -import pytest -from cashu.core.base import MeltQuoteState -from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine -from sqlmodel import SQLModel, select -from sqlmodel.ext.asyncio.session import AsyncSession - -from routstr import wallet -from routstr.core import db - -SOURCE_MINT = "https://source.mint.test" -DEST_MINT = "https://dest.mint.test" -KEY_HASH = "hash-abc123" - - -@asynccontextmanager -async def _null_guard() -> AsyncGenerator[None, None]: - yield - - -async def _passthrough_mint_op(fn: Any, **_kwargs: Any) -> Any: - return await fn() - - -@pytest.fixture -async def engine() -> AsyncGenerator[AsyncEngine, None]: - engine = create_async_engine("sqlite+aiosqlite://") - async with engine.begin() as connection: - await connection.run_sync(SQLModel.metadata.create_all) - yield engine - await engine.dispose() - - -@pytest.fixture -def patched_env(engine: AsyncEngine, monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]: - @asynccontextmanager - async def _create_session() -> AsyncGenerator[AsyncSession, None]: - async with AsyncSession(engine, expire_on_commit=False) as session: - yield session - - monkeypatch.setattr(db, "create_session", _create_session) - monkeypatch.setattr(wallet, "wallet_operation_guard", _null_guard) - monkeypatch.setattr(wallet, "run_mint_operation", _passthrough_mint_op) - - stored_transactions: list[dict[str, Any]] = [] - - async def _store_tx(**kwargs: Any) -> None: - stored_transactions.append(kwargs) - - monkeypatch.setattr(wallet, "store_cashu_transaction", _store_tx) - - state: dict[str, Any] = { - "melt_state": MeltQuoteState.paid, - "mint_error": None, - "mint_calls": [], - "transactions": stored_transactions, - } - - class FakeSourceWallet: - async def get_melt_quote(self, quote_id: str) -> Any: - return SimpleNamespace(state=state["melt_state"], quote=quote_id) - - class FakeDestWallet: - async def mint(self, amount: int, quote_id: str | None = None) -> Any: - if state["mint_error"] is not None: - raise state["mint_error"] - state["mint_calls"].append((amount, quote_id)) - return [] - - async def _fake_get_wallet(mint_url: str, unit: str = "sat", **_kw: Any) -> Any: - return FakeSourceWallet() if mint_url == SOURCE_MINT else FakeDestWallet() - - monkeypatch.setattr(wallet, "get_wallet", _fake_get_wallet) - return state - - -def _pending_row(**overrides: Any) -> db.PendingSwap: - defaults: dict[str, Any] = { - "source_mint": SOURCE_MINT, - "source_unit": "sat", - "melt_quote_id": "melt-quote-1", - "dest_mint": DEST_MINT, - "dest_unit": "sat", - "mint_quote_id": "mint-quote-1", - "minted_amount": 90, - "key_hashed_key": KEY_HASH, - "token": "cashuAtest", - "state": "pending", - "created_at": int(time.time()) - 3600, - } - defaults.update(overrides) - return db.PendingSwap(**defaults) - - -async def _seed(engine: AsyncEngine, *rows: Any) -> None: - async with AsyncSession(engine, expire_on_commit=False) as session: - for row in rows: - session.add(row) - await session.commit() - - -async def _all_rows(engine: AsyncEngine) -> list[db.PendingSwap]: - async with AsyncSession(engine, expire_on_commit=False) as session: - result = await session.exec(select(db.PendingSwap)) - return list(result.all()) - - -@pytest.mark.asyncio -async def test_paid_melt_mints_and_credits_existing_key( - engine: AsyncEngine, patched_env: dict[str, Any] -) -> None: - await _seed( - engine, - db.ApiKey(hashed_key=KEY_HASH, balance=1_000), - _pending_row(), - ) - - await wallet.reconcile_pending_swaps_once() - - assert patched_env["mint_calls"] == [(90, "mint-quote-1")] - async with AsyncSession(engine, expire_on_commit=False) as session: - key = await session.get(db.ApiKey, KEY_HASH) - assert key is not None - assert key.balance == 1_000 + 90_000 # 90 sat credited as msats - assert await _all_rows(engine) == [] - assert len(patched_env["transactions"]) == 1 - assert patched_env["transactions"][0]["api_key_hashed_key"] == KEY_HASH - - -@pytest.mark.asyncio -async def test_paid_melt_recreates_rolled_back_key( - engine: AsyncEngine, patched_env: dict[str, Any] -) -> None: - await _seed(engine, _pending_row()) - - await wallet.reconcile_pending_swaps_once() - - async with AsyncSession(engine, expire_on_commit=False) as session: - key = await session.get(db.ApiKey, KEY_HASH) - assert key is not None - assert key.balance == 90_000 - assert key.refund_mint_url == DEST_MINT - assert key.refund_currency == "sat" - assert await _all_rows(engine) == [] - - -@pytest.mark.asyncio -async def test_old_unpaid_melt_drops_checkpoint_without_credit( - engine: AsyncEngine, patched_env: dict[str, Any] -) -> None: - patched_env["melt_state"] = MeltQuoteState.unpaid - await _seed( - engine, - db.ApiKey(hashed_key=KEY_HASH, balance=1_000), - _pending_row(), - ) - - await wallet.reconcile_pending_swaps_once() - - assert patched_env["mint_calls"] == [] - async with AsyncSession(engine, expire_on_commit=False) as session: - key = await session.get(db.ApiKey, KEY_HASH) - assert key is not None - assert key.balance == 1_000 - assert await _all_rows(engine) == [] - - -@pytest.mark.asyncio -async def test_recent_unpaid_melt_keeps_waiting( - engine: AsyncEngine, patched_env: dict[str, Any] -) -> None: - patched_env["melt_state"] = MeltQuoteState.unpaid - await _seed(engine, _pending_row(created_at=int(time.time()) - 300)) - - await wallet.reconcile_pending_swaps_once() - - rows = await _all_rows(engine) - assert len(rows) == 1 - assert rows[0].state == "pending" - assert rows[0].attempts == 1 - - -@pytest.mark.asyncio -async def test_pending_melt_state_only_bumps_attempts( - engine: AsyncEngine, patched_env: dict[str, Any] -) -> None: - patched_env["melt_state"] = MeltQuoteState.pending - await _seed(engine, _pending_row()) - - await wallet.reconcile_pending_swaps_once() - - rows = await _all_rows(engine) - assert len(rows) == 1 - assert rows[0].state == "pending" - assert rows[0].attempts == 1 - assert patched_env["mint_calls"] == [] - - -@pytest.mark.asyncio -async def test_transient_mint_failure_retries_later( - engine: AsyncEngine, patched_env: dict[str, Any] -) -> None: - patched_env["mint_error"] = RuntimeError("mint briefly down") - await _seed(engine, _pending_row(state="melt_confirmed")) - - await wallet.reconcile_pending_swaps_once() - - rows = await _all_rows(engine) - assert len(rows) == 1 - assert rows[0].state == "melt_confirmed" - assert rows[0].attempts == 1 - assert rows[0].last_error == "mint briefly down" - - -@pytest.mark.asyncio -async def test_already_issued_quote_parks_as_stale( - engine: AsyncEngine, patched_env: dict[str, Any] -) -> None: - patched_env["mint_error"] = RuntimeError("quote already issued") - await _seed(engine, _pending_row(state="melt_confirmed")) - - await wallet.reconcile_pending_swaps_once() - - rows = await _all_rows(engine) - assert len(rows) == 1 - assert rows[0].state == "stale" - - -@pytest.mark.asyncio -async def test_fresh_checkpoint_left_for_inline_swap( - engine: AsyncEngine, patched_env: dict[str, Any] -) -> None: - await _seed(engine, _pending_row(created_at=int(time.time()))) - - await wallet.reconcile_pending_swaps_once() - - rows = await _all_rows(engine) - assert len(rows) == 1 - assert rows[0].attempts == 0 - assert patched_env["mint_calls"] == [] - - -@pytest.mark.asyncio -async def test_attempt_cap_parks_checkpoint_as_stale( - engine: AsyncEngine, patched_env: dict[str, Any] -) -> None: - patched_env["melt_state"] = MeltQuoteState.pending - await _seed( - engine, - _pending_row(attempts=wallet._SWAP_RECONCILE_MAX_ATTEMPTS - 1), - ) - - await wallet.reconcile_pending_swaps_once() - - rows = await _all_rows(engine) - assert len(rows) == 1 - assert rows[0].state == "stale"