mirror of
https://github.com/Routstr/routstr-core.git
synced 2026-10-05 20:28:23 +00:00
Revert "feat: persistent reconciliation for interrupted cross-mint swaps"
This reverts commit d5528682ae.
This commit is contained in:
9qeklajc
@@ -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")
|
||||
@@ -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,
|
||||
|
||||
+1
-10
@@ -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 = []
|
||||
|
||||
+1
-341
@@ -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
|
||||
|
||||
@@ -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"
|
||||
Reference in New Issue
Block a user