diff --git a/migrations/versions/d7e8f9a0b1c2_unique_token_type_cashu_transactions.py b/migrations/versions/d7e8f9a0b1c2_unique_token_type_cashu_transactions.py new file mode 100644 index 00000000..08487eef --- /dev/null +++ b/migrations/versions/d7e8f9a0b1c2_unique_token_type_cashu_transactions.py @@ -0,0 +1,75 @@ +"""unique (token, type) constraint on cashu_transactions + +Revision ID: d7e8f9a0b1c2 +Revises: c6d7e8f9a0b1 +Create Date: 2026-07-10 00:00:00.000000 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision = "d7e8f9a0b1c2" +down_revision = "c6d7e8f9a0b1" +branch_labels = None +depends_on = None + + +_CONSTRAINT_NAME = "uq_cashu_transactions_token_type" + + +def _dedup_rows(conn: sa.Connection) -> None: + """Delete duplicate (token, type) rows, keeping the oldest per group.""" + conn.execute( + sa.text( + "DELETE FROM cashu_transactions " + "WHERE id IN (" + " SELECT id FROM (" + " SELECT id, ROW_NUMBER() OVER (" + " PARTITION BY token, type " + " ORDER BY created_at ASC, id ASC" + " ) AS rn " + " FROM cashu_transactions" + " ) WHERE rn > 1" + ")" + ) + ) + + +def upgrade() -> None: + conn = op.get_bind() + inspector = sa.inspect(conn) + + _dedup_rows(conn) + + existing_indexes = { + idx["name"] for idx in inspector.get_indexes("cashu_transactions") + } + existing_constraints = { + uc["name"] + for uc in inspector.get_unique_constraints("cashu_transactions") + } + if _CONSTRAINT_NAME not in existing_indexes and _CONSTRAINT_NAME not in existing_constraints: + op.create_index( + _CONSTRAINT_NAME, + "cashu_transactions", + ["token", "type"], + unique=True, + ) + + +def downgrade() -> None: + conn = op.get_bind() + inspector = sa.inspect(conn) + existing_indexes = { + idx["name"] for idx in inspector.get_indexes("cashu_transactions") + } + existing_constraints = { + uc["name"] + for uc in inspector.get_unique_constraints("cashu_transactions") + } + if _CONSTRAINT_NAME in existing_indexes: + op.drop_index(_CONSTRAINT_NAME, table_name="cashu_transactions") + elif _CONSTRAINT_NAME in existing_constraints: + op.drop_constraint(_CONSTRAINT_NAME, "cashu_transactions") diff --git a/routstr/balance.py b/routstr/balance.py index cc37d089..67edb4d0 100644 --- a/routstr/balance.py +++ b/routstr/balance.py @@ -15,7 +15,7 @@ from .core.db import ( AsyncSession, CashuTransaction, get_session, - store_cashu_transaction, + store_cashu_transaction_with_retry, ) from .core.logging import get_logger from .core.settings import settings @@ -457,19 +457,16 @@ async def refund_wallet_endpoint( await _refund_cache_set(bearer_value, result) if "token" in result: - try: - await store_cashu_transaction( - token=result["token"], - amount=remaining_balance, - unit=key.refund_currency or "sat", - mint_url=key.refund_mint_url, - typ="out", - collected=False, - source="apikey", - api_key_hashed_key=key.hashed_key, - ) - except Exception: - pass # store_cashu_transaction already logs + await store_cashu_transaction_with_retry( + token=result["token"], + amount=remaining_balance, + unit=key.refund_currency or "sat", + mint_url=key.refund_mint_url, + typ="out", + collected=False, + source="apikey", + api_key_hashed_key=key.hashed_key, + ) logger.info( "refund_wallet_endpoint: refund successful", diff --git a/routstr/core/db.py b/routstr/core/db.py index 6cd5b593..5c5e6ba1 100644 --- a/routstr/core/db.py +++ b/routstr/core/db.py @@ -1,3 +1,5 @@ +import asyncio +import json import os import pathlib import sqlite3 @@ -10,7 +12,7 @@ from alembic import command from alembic.config import Config from alembic.util.exc import CommandError from sqlalchemy import UniqueConstraint, delete -from sqlalchemy.exc import OperationalError +from sqlalchemy.exc import IntegrityError, OperationalError from sqlalchemy.ext.asyncio.engine import create_async_engine from sqlalchemy.orm import aliased from sqlmodel import Field, Relationship, SQLModel, col, func, select, update @@ -26,6 +28,81 @@ DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite+aiosqlite:///keys.db") engine = create_async_engine(DATABASE_URL, echo=False) # echo=True for debugging SQL +# Durable JSONL outbox for cashu transactions when the DB is down. +_cashu_outbox_lock = asyncio.Lock() + + +def _cashu_outbox_path() -> pathlib.Path: + """Resolve the per-process cashu outbox file path.""" + override = os.environ.get("CASHU_OUTBOX_PATH") + if override: + return pathlib.Path(override) + + filename = f"cashu_outbox.{os.getpid()}.jsonl" + + if DATABASE_URL.startswith("sqlite"): + raw = DATABASE_URL.split("///", 1)[-1] + if not raw or raw == ":memory:": + return pathlib.Path("/tmp").resolve() / filename + db_file = pathlib.Path(raw) + return db_file.parent.resolve() / filename + + return pathlib.Path(filename).resolve() + + +async def append_to_cashu_outbox( + *, + token: str, + amount: int, + unit: str, + mint_url: str | None, + typ: str, + request_id: str | None, + collected: bool, + created_at: int | None, + source: str, + api_key_hashed_key: str | None, +) -> None: + """Append a transaction payload to the durable outbox. Never raises.""" + entry = { + "outbox_id": uuid.uuid4().hex, + "queued_at": int(time.time()), + "token": token, + "amount": amount, + "unit": unit, + "mint_url": mint_url, + "type": typ, + "request_id": request_id, + "collected": collected, + "created_at": created_at, + "source": source, + "api_key_hashed_key": api_key_hashed_key, + } + line = json.dumps(entry, separators=(",", ":")) + "\n" + async with _cashu_outbox_lock: + try: + path = _cashu_outbox_path() + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "a", encoding="utf-8") as fh: + fh.write(line) + fh.flush() + os.fsync(fh.fileno()) + logger.warning( + "Cashu transaction spooled to durable outbox", + extra={"outbox_path": str(path), "type": typ, "request_id": request_id}, + ) + except Exception as outbox_exc: + logger.critical( + "cashu outbox append failed; token may be unrecoverable", + extra={ + "error": str(outbox_exc), + "type": typ, + "request_id": request_id, + "token": token, + }, + ) + + class ApiKey(SQLModel, table=True): # type: ignore __tablename__ = "api_keys" @@ -249,6 +326,11 @@ class LightningInvoice(SQLModel, table=True): # type: ignore class CashuTransaction(SQLModel, table=True): # type: ignore __tablename__ = "cashu_transactions" + __table_args__ = ( + UniqueConstraint( + "token", "type", name="uq_cashu_transactions_token_type" + ), + ) id: str = Field( primary_key=True, @@ -290,9 +372,19 @@ async def store_cashu_transaction( created_at: int | None = None, source: str = "x-cashu", api_key_hashed_key: str | None = None, -) -> None: +) -> bool: + """Persist a cashu transaction; idempotent on (token, type). Returns True on success.""" try: async with create_session() as session: + existing = await session.exec( + select(CashuTransaction).where( + CashuTransaction.token == token, + CashuTransaction.type == typ, + ) + ) + if existing.first() is not None: + return True + tx = CashuTransaction( token=token, amount=amount, @@ -307,11 +399,225 @@ async def store_cashu_transaction( ) session.add(tx) await session.commit() - except Exception as e: - logger.warning( - f"Failed to store cashu transaction: {e} (type={typ})", - extra={"error": str(e), "type": typ}, + return True + except IntegrityError: + # A concurrent insert of the same (token, type) won the race. + async with create_session() as session: + existing = await session.exec( + select(CashuTransaction).where( + CashuTransaction.token == token, + CashuTransaction.type == typ, + ) + ) + if existing.first() is not None: + return True + logger.error( + f"Integrity error storing cashu transaction: non-duplicate violation (type={typ})", + extra={ + "error": "non-duplicate IntegrityError", + "type": typ, + "request_id": request_id, + "amount": amount, + "unit": unit, + "mint_url": mint_url, + }, ) + return False + except Exception as e: + logger.error( + f"Failed to store cashu transaction: {e} (type={typ})", + extra={ + "error": str(e), + "type": typ, + "request_id": request_id, + "amount": amount, + "unit": unit, + "mint_url": mint_url, + "token_preview": (token[:30] + "...") if len(token) > 30 else token, + }, + ) + return False + + +async def store_cashu_transaction_with_retry( + token: str, + amount: int, + unit: str, + mint_url: str | None = None, + typ: str = "out", + request_id: str | None = None, + collected: bool = False, + created_at: int | None = None, + source: str = "x-cashu", + api_key_hashed_key: str | None = None, + max_retries: int = 3, +) -> bool: + """Retry ``store_cashu_transaction`` with backoff; spool to the outbox on exhaustion.""" + for attempt in range(max_retries): + ok = await store_cashu_transaction( + token=token, + amount=amount, + unit=unit, + mint_url=mint_url, + typ=typ, + request_id=request_id, + collected=collected, + created_at=created_at, + source=source, + api_key_hashed_key=api_key_hashed_key, + ) + if ok: + return True + if attempt < max_retries - 1: + backoff = 0.5 * (2 ** attempt) + logger.warning( + "Retrying store_cashu_transaction", + extra={ + "attempt": attempt + 1, + "backoff_seconds": backoff, + "type": typ, + "request_id": request_id, + }, + ) + await asyncio.sleep(backoff) + + logger.critical( + "Cashu transaction spooled to outbox after retries exhausted", + extra={ + "type": typ, + "request_id": request_id, + "amount": amount, + "unit": unit, + "mint_url": mint_url, + "token": token, + }, + ) + await append_to_cashu_outbox( + token=token, + amount=amount, + unit=unit, + mint_url=mint_url, + typ=typ, + request_id=request_id, + collected=collected, + created_at=created_at or int(time.time()), + source=source, + api_key_hashed_key=api_key_hashed_key, + ) + return False + + +async def replay_cashu_outbox(path: pathlib.Path | None = None) -> int: + """Replay spooled outbox entries into the DB. Returns the count persisted.""" + if path is None: + path = _cashu_outbox_path() + if not path.exists(): + return 0 + + # Hold the lock for the full read-replay-rewrite cycle so a concurrent + # appender cannot slip a new line between the read and the rewrite. + async with _cashu_outbox_lock: + try: + raw_lines = path.read_text(encoding="utf-8").splitlines() + except Exception as e: + logger.error("Failed to read cashu outbox", extra={"error": str(e)}) + return 0 + + entries: list[dict] = [] + for line in raw_lines: + line = line.strip() + if not line: + continue + try: + entries.append(json.loads(line)) + except json.JSONDecodeError as e: + logger.warning( + "Skipping malformed outbox line", + extra={"error": str(e), "line_preview": line[:80]}, + ) + + if not entries: + return 0 + + persisted = 0 + remaining: list[dict] = [] + for entry in entries: + ok = await store_cashu_transaction( + token=entry["token"], + amount=entry["amount"], + unit=entry["unit"], + mint_url=entry.get("mint_url"), + typ=entry.get("type", "out"), + request_id=entry.get("request_id"), + collected=entry.get("collected", False), + created_at=entry.get("created_at"), + source=entry.get("source", "x-cashu"), + api_key_hashed_key=entry.get("api_key_hashed_key"), + ) + if ok: + persisted += 1 + logger.info( + "Outbox entry replayed into DB", + extra={ + "outbox_id": entry.get("outbox_id"), + "type": entry.get("type"), + "request_id": entry.get("request_id"), + }, + ) + else: + remaining.append(entry) + + try: + if remaining: + tmp = path.with_suffix(path.suffix + ".tmp") + with open(tmp, "w", encoding="utf-8") as fh: + for entry in remaining: + fh.write(json.dumps(entry, separators=(",", ":")) + "\n") + fh.flush() + os.fsync(fh.fileno()) + os.replace(tmp, path) + else: + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.write_text("") + os.replace(tmp, path) + except Exception as e: + logger.error( + "Failed to rewrite cashu outbox after replay", + extra={"error": str(e), "remaining": len(remaining)}, + ) + + return persisted + + +async def replay_all_outbox_files() -> int: + """Replay every cashu outbox file in the outbox directory.""" + override = os.environ.get("CASHU_OUTBOX_PATH") + if override: + return await replay_cashu_outbox(pathlib.Path(override)) + + outbox_dir = _cashu_outbox_path().parent + + total = 0 + for file in sorted(outbox_dir.glob("cashu_outbox.*.jsonl")): + total += await replay_cashu_outbox(file) + return total + + +async def periodic_cashu_outbox_replay() -> None: + """Background loop that drains the cashu outbox into the database.""" + interval = float(os.environ.get("CASHU_OUTBOX_REPLAY_INTERVAL", "30")) + logger.info("Starting cashu outbox replay loop", extra={"interval": interval}) + while True: + try: + await asyncio.sleep(interval) + await replay_all_outbox_files() + except asyncio.CancelledError: + raise + except Exception as e: + logger.error( + "Outbox replay iteration failed", + extra={"error": str(e), "error_type": type(e).__name__}, + ) class UpstreamProviderRow(SQLModel, table=True): # type: ignore diff --git a/routstr/core/main.py b/routstr/core/main.py index 903d6d1a..aa278db0 100644 --- a/routstr/core/main.py +++ b/routstr/core/main.py @@ -32,7 +32,7 @@ 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 from .admin import admin_router -from .db import create_session, init_db, run_migrations +from .db import create_session, init_db, periodic_cashu_outbox_replay, run_migrations from .exceptions import general_exception_handler, http_exception_handler from .logging import get_logger, setup_logging from .middleware import LoggingMiddleware @@ -65,6 +65,7 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]: refund_sweep_task = None routstr_fee_task = None invoice_watcher_task = None + cashu_outbox_task = None try: # Apply litellm-wide settings (drop_params, chat-completions URL, @@ -142,6 +143,7 @@ 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()) + cashu_outbox_task = asyncio.create_task(periodic_cashu_outbox_replay()) yield @@ -187,6 +189,8 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]: routstr_fee_task.cancel() if invoice_watcher_task is not None: invoice_watcher_task.cancel() + if cashu_outbox_task is not None: + cashu_outbox_task.cancel() try: tasks_to_wait = [] @@ -220,6 +224,8 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]: tasks_to_wait.append(routstr_fee_task) if invoice_watcher_task is not None: tasks_to_wait.append(invoice_watcher_task) + if cashu_outbox_task is not None: + tasks_to_wait.append(cashu_outbox_task) if tasks_to_wait: await asyncio.gather(*tasks_to_wait, return_exceptions=True) diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index f8ea2d4a..e0905596 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -20,7 +20,7 @@ from ..core.db import ( AsyncSession, UpstreamProviderRow, create_session, - store_cashu_transaction, + store_cashu_transaction_with_retry, ) from ..core.exceptions import UpstreamError from ..core.redaction import redact_org_ids @@ -3286,7 +3286,7 @@ class BaseUpstreamProvider: ) try: - await store_cashu_transaction( + await store_cashu_transaction_with_retry( token=refund_token, amount=amount, unit=unit, @@ -3294,8 +3294,25 @@ class BaseUpstreamProvider: typ="out", request_id=request_id, ) - except Exception: - pass # store_cashu_transaction already logs + except Exception as store_exc: + # store_cashu_transaction_with_retry returns False (and + # spools to the outbox) on normal DB failures, so this only + # fires on an unexpected raise. Catch it here so the outer + # retry loop does NOT re-mint a second token (which would + # double-spend). The full token is logged so it can be + # recovered manually if the outbox write also failed. + logger.critical( + "send_refund: store_cashu_transaction_with_retry raised — " + "refund token is minted; spooled to outbox if possible", + extra={ + "error": str(store_exc), + "amount": amount, + "unit": unit, + "mint": mint, + "request_id": request_id, + "token": refund_token, + }, + ) return refund_token except Exception as e: @@ -3640,17 +3657,14 @@ class BaseUpstreamProvider: emergency_refund = amount refund_token = await send_token(emergency_refund, unit=unit, mint_url=mint) response.headers["X-Cashu"] = refund_token - try: - await store_cashu_transaction( - token=refund_token, - amount=emergency_refund, - unit=unit, - mint_url=mint, - typ="out", - request_id=request_id, - ) - except Exception: - pass + await store_cashu_transaction_with_retry( + token=refund_token, + amount=emergency_refund, + unit=unit, + mint_url=mint, + typ="out", + request_id=request_id, + ) logger.warning( "Emergency refund issued due to JSON parse error", @@ -4002,18 +4016,15 @@ class BaseUpstreamProvider: headers = self.prepare_headers(dict(request.headers)) request_id = getattr(request.state, "request_id", None) - try: - await store_cashu_transaction( - token=x_cashu_token, - amount=amount, - unit=unit, - mint_url=mint, - typ="in", - request_id=request_id, - collected=True, - ) - except Exception: - pass + await store_cashu_transaction_with_retry( + token=x_cashu_token, + amount=amount, + unit=unit, + mint_url=mint, + typ="in", + request_id=request_id, + collected=True, + ) logger.info( "X-Cashu token redeemed for Responses API", @@ -4604,17 +4615,14 @@ class BaseUpstreamProvider: emergency_refund = amount refund_token = await send_token(emergency_refund, unit=unit, mint_url=mint) response.headers["X-Cashu"] = refund_token - try: - await store_cashu_transaction( - token=refund_token, - amount=emergency_refund, - unit=unit, - mint_url=mint, - typ="out", - request_id=request_id, - ) - except Exception: - pass + await store_cashu_transaction_with_retry( + token=refund_token, + amount=emergency_refund, + unit=unit, + mint_url=mint, + typ="out", + request_id=request_id, + ) logger.warning( "Emergency refund issued for Responses API due to JSON parse error", @@ -4678,18 +4686,15 @@ class BaseUpstreamProvider: headers = self.prepare_headers(dict(request.headers)) request_id = getattr(request.state, "request_id", None) - try: - await store_cashu_transaction( - token=x_cashu_token, - amount=amount, - unit=unit, - mint_url=mint, - typ="in", - request_id=request_id, - collected=True, - ) - except Exception: - pass + await store_cashu_transaction_with_retry( + token=x_cashu_token, + amount=amount, + unit=unit, + mint_url=mint, + typ="in", + request_id=request_id, + collected=True, + ) logger.info( "X-Cashu token redeemed successfully", diff --git a/routstr/wallet.py b/routstr/wallet.py index a5a4e8cd..0d7e82c0 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -14,7 +14,7 @@ from pydantic_core import PydanticUndefined from sqlmodel import col, select, update from .core import db, get_logger -from .core.db import store_cashu_transaction +from .core.db import store_cashu_transaction_with_retry from .core.settings import settings from .payment.lnurl import raw_send_to_lnurl @@ -988,7 +988,7 @@ async def credit_balance( ) try: - await store_cashu_transaction( + await store_cashu_transaction_with_retry( token=cashu_token, amount=original_amount, unit=original_unit, @@ -998,7 +998,7 @@ async def credit_balance( api_key_hashed_key=key.hashed_key, ) except Exception: - pass + pass # store_cashu_transaction_with_retry already logs + spools to outbox logger.debug( "Cashu token successfully redeemed and stored", diff --git a/tests/unit/test_balance.py b/tests/unit/test_balance.py index 609e2557..5e136e06 100644 --- a/tests/unit/test_balance.py +++ b/tests/unit/test_balance.py @@ -235,7 +235,7 @@ async def test_apikey_refund_stores_cashu_transaction_with_apikey_source() -> No with ( patch("routstr.balance.get_billing_key", AsyncMock(return_value=key)), patch("routstr.balance.send_token", AsyncMock(return_value=refund_token)), - patch("routstr.balance.store_cashu_transaction", AsyncMock()) as mock_store, + patch("routstr.balance.store_cashu_transaction_with_retry", AsyncMock(return_value=True)) as mock_store, patch("routstr.balance._refund_cache_get", AsyncMock(return_value=None)), patch("routstr.balance._refund_cache_set", AsyncMock()), ): @@ -270,7 +270,7 @@ async def test_apikey_refund_logs_token() -> None: with ( patch("routstr.balance.get_billing_key", AsyncMock(return_value=key)), patch("routstr.balance.send_token", AsyncMock(return_value=refund_token)), - patch("routstr.balance.store_cashu_transaction", AsyncMock()), + patch("routstr.balance.store_cashu_transaction_with_retry", AsyncMock(return_value=True)), patch("routstr.balance._refund_cache_get", AsyncMock(return_value=None)), patch("routstr.balance._refund_cache_set", AsyncMock()), patch("routstr.balance.logger") as mock_logger, @@ -299,7 +299,7 @@ async def test_apikey_refund_log_includes_path() -> None: with ( patch("routstr.balance.get_billing_key", AsyncMock(return_value=key)), patch("routstr.balance.send_token", AsyncMock(return_value=refund_token)), - patch("routstr.balance.store_cashu_transaction", AsyncMock()), + patch("routstr.balance.store_cashu_transaction_with_retry", AsyncMock(return_value=True)), patch("routstr.balance._refund_cache_get", AsyncMock(return_value=None)), patch("routstr.balance._refund_cache_set", AsyncMock()), patch("routstr.balance.logger") as mock_logger, @@ -338,7 +338,7 @@ async def test_apikey_refund_rejects_on_concurrent_balance_change() -> None: with ( patch("routstr.balance.get_billing_key", AsyncMock(return_value=key)), patch("routstr.balance.send_token", mock_send_token), - patch("routstr.balance.store_cashu_transaction", AsyncMock()), + patch("routstr.balance.store_cashu_transaction_with_retry", AsyncMock(return_value=True)), patch("routstr.balance._refund_cache_get", AsyncMock(return_value=None)), patch("routstr.balance._refund_cache_set", AsyncMock()), ): @@ -367,7 +367,10 @@ async def test_credit_balance_stores_apikey_transaction_history() -> None: "routstr.wallet.recieve_token", AsyncMock(return_value=(100, "sat", "https://mint.example")), ), - patch("routstr.wallet.store_cashu_transaction", AsyncMock()) as mock_store, + patch( + "routstr.wallet.store_cashu_transaction_with_retry", + AsyncMock(return_value=True), + ) as mock_store, ): amount = await credit_balance("cashuAtopup_token", key, session) @@ -402,7 +405,7 @@ async def test_apikey_refund_restores_balance_on_mint_failure() -> None: "routstr.balance.send_token", AsyncMock(side_effect=MintConnectionError("raw mint outage detail")), ), - patch("routstr.balance.store_cashu_transaction", AsyncMock()), + patch("routstr.balance.store_cashu_transaction_with_retry", AsyncMock(return_value=True)), patch("routstr.balance._refund_cache_get", AsyncMock(return_value=None)), patch("routstr.balance._refund_cache_set", AsyncMock()), patch("routstr.balance.logger"), @@ -437,7 +440,7 @@ async def test_apikey_refund_generic_failure_is_sanitized_500() -> None: with ( patch("routstr.balance.get_billing_key", AsyncMock(return_value=key)), patch("routstr.balance.send_token", AsyncMock(side_effect=RuntimeError(raw_error))), - patch("routstr.balance.store_cashu_transaction", AsyncMock()), + patch("routstr.balance.store_cashu_transaction_with_retry", AsyncMock(return_value=True)), patch("routstr.balance._refund_cache_get", AsyncMock(return_value=None)), patch("routstr.balance._refund_cache_set", AsyncMock()), patch("routstr.balance.logger"), diff --git a/tests/unit/test_cashu_outbox.py b/tests/unit/test_cashu_outbox.py new file mode 100644 index 00000000..d8a13288 --- /dev/null +++ b/tests/unit/test_cashu_outbox.py @@ -0,0 +1,411 @@ +"""Unit tests for the Cashu outbox + retry persistence path. + +These cover the guarantees introduced by the refund-token persistence fix: + * ``store_cashu_transaction`` is idempotent (no duplicate rows on retry) + * ``store_cashu_transaction_with_retry`` retries, then spools the full + token to a durable outbox when the DB stays unavailable + * ``replay_cashu_outbox`` drains the outbox into the DB and is idempotent +""" +import asyncio +import json +import os +from collections.abc import AsyncIterator +from pathlib import Path +from typing import Any + +import pytest +from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine +from sqlalchemy.pool import StaticPool +from sqlmodel import SQLModel, select + +import routstr.core.db as db +from routstr.core.db import ( + CashuTransaction, + create_session, + replay_all_outbox_files, + replay_cashu_outbox, + store_cashu_transaction, + store_cashu_transaction_with_retry, +) + + +def _in_memory_engine() -> AsyncEngine: + return create_async_engine( + "sqlite+aiosqlite://", + poolclass=StaticPool, + connect_args={"check_same_thread": False}, + ) + + +@pytest.fixture +async def fresh_engine( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> AsyncIterator[tuple[AsyncEngine, Path]]: + """Point db.engine at an isolated in-memory DB and the outbox at a tmp file.""" + engine = _in_memory_engine() + async with engine.begin() as conn: + await conn.run_sync(SQLModel.metadata.create_all) + monkeypatch.setattr(db, "engine", engine) + # create_session() looks up module-global `engine` at call time. + monkeypatch.setattr(db, "create_session", create_session, raising=False) + outbox = tmp_path / "cashu_outbox.jsonl" + monkeypatch.setenv("CASHU_OUTBOX_PATH", str(outbox)) + try: + yield engine, outbox + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_store_is_idempotent_on_duplicate( + fresh_engine: tuple[AsyncEngine, Path] +) -> None: + """Inserting the same (token, type) twice yields exactly one row.""" + _, _ = fresh_engine + ok1 = await store_cashu_transaction( + token="cashuAtokenA", amount=100, unit="sat", typ="out", request_id="req-1" + ) + ok2 = await store_cashu_transaction( + token="cashuAtokenA", amount=100, unit="sat", typ="out", request_id="req-1" + ) + assert ok1 is True + assert ok2 is True # duplicate is treated as success, not a new row + + async with create_session() as session: + rows = (await session.exec(select(CashuTransaction))).all() + assert len(rows) == 1 + assert rows[0].token == "cashuAtokenA" + + +@pytest.mark.asyncio +async def test_with_retry_succeeds_first_try( + fresh_engine: tuple[AsyncEngine, Path] +) -> None: + _, _ = fresh_engine + ok = await store_cashu_transaction_with_retry( + token="cashuAtokenB", amount=50, unit="sat", typ="out", request_id="req-2" + ) + assert ok is True + async with create_session() as session: + rows = (await session.exec(select(CashuTransaction))).all() + assert len(rows) == 1 + + +@pytest.mark.asyncio +async def test_with_retry_succeeds_after_transient_failure( + fresh_engine: tuple[AsyncEngine, Path], monkeypatch: pytest.MonkeyPatch +) -> None: + """First attempt fails, second succeeds; no outbox spool, one row.""" + _, _ = fresh_engine + calls = {"n": 0} + real_store = store_cashu_transaction + + async def flaky_store(*args: Any, **kwargs: Any) -> bool: + calls["n"] += 1 + if calls["n"] == 1: + return False + return await real_store(*args, **kwargs) + + monkeypatch.setattr(db, "store_cashu_transaction", flaky_store) + # zero backoff so the test is fast + monkeypatch.setattr(db.asyncio, "sleep", lambda *_a, **_k: _asyncio_sleep_zero()) + + ok = await store_cashu_transaction_with_retry( + token="cashuAtokenC", amount=50, unit="sat", typ="out", request_id="req-3", + max_retries=3, + ) + assert ok is True + assert calls["n"] == 2 + async with create_session() as session: + rows = (await session.exec(select(CashuTransaction))).all() + assert len(rows) == 1 + + +@pytest.mark.asyncio +async def test_with_retry_spools_full_token_to_outbox_on_exhaustion( + fresh_engine: tuple[AsyncEngine, Path], monkeypatch: pytest.MonkeyPatch +) -> None: + """When the DB stays down, the full token is spooled to the outbox file.""" + _, outbox = fresh_engine + + async def always_fail(*args: Any, **kwargs: Any) -> bool: + return False + + monkeypatch.setattr(db, "store_cashu_transaction", always_fail) + monkeypatch.setattr(db.asyncio, "sleep", lambda *_a, **_k: _asyncio_sleep_zero()) + + full_token = "cashuAtokenD_super_secret_full_value_xyz" + ok = await store_cashu_transaction_with_retry( + token=full_token, amount=77, unit="sat", typ="out", request_id="req-4", + max_retries=3, + ) + assert ok is False + assert outbox.exists() + lines = outbox.read_text(encoding="utf-8").strip().splitlines() + assert len(lines) == 1 + entry = json.loads(lines[0]) + # The full token must be recoverable from the outbox, not a preview. + assert entry["token"] == full_token + assert entry["amount"] == 77 + assert entry["request_id"] == "req-4" + assert entry["type"] == "out" + + +@pytest.mark.asyncio +async def test_replay_drains_outbox_into_db( + fresh_engine: tuple[AsyncEngine, Path] +) -> None: + """Outbox entries are persisted on replay and the outbox is cleared.""" + _, outbox = fresh_engine + entry = { + "outbox_id": "abc123", + "queued_at": 1, + "token": "cashuAtokenE", + "amount": 200, + "unit": "sat", + "mint_url": None, + "type": "out", + "request_id": "req-5", + "collected": False, + "created_at": None, + "source": "x-cashu", + "api_key_hashed_key": None, + } + outbox.write_text(json.dumps(entry) + "\n", encoding="utf-8") + + persisted = await replay_cashu_outbox() + assert persisted == 1 + # outbox should now be empty (truncated) + assert outbox.read_text(encoding="utf-8") == "" + + async with create_session() as session: + rows = (await session.exec(select(CashuTransaction))).all() + assert len(rows) == 1 + assert rows[0].token == "cashuAtokenE" + + +@pytest.mark.asyncio +async def test_replay_is_idempotent(fresh_engine: tuple[AsyncEngine, Path]) -> None: + """Replaying an outbox whose entries already exist in the DB drops them.""" + _, outbox = fresh_engine + # Pre-seed the DB row. + await store_cashu_transaction( + token="cashuAtokenF", amount=10, unit="sat", typ="out", request_id="req-6" + ) + # Outbox references the same token. + entry = { + "outbox_id": "xyz", + "queued_at": 1, + "token": "cashuAtokenF", + "amount": 10, + "unit": "sat", + "mint_url": None, + "type": "out", + "request_id": "req-6", + "collected": False, + "created_at": None, + "source": "x-cashu", + "api_key_hashed_key": None, + } + outbox.write_text(json.dumps(entry) + "\n", encoding="utf-8") + + persisted = await replay_cashu_outbox() + assert persisted == 1 # counted as persisted (already-present = success) + assert outbox.read_text(encoding="utf-8") == "" + + async with create_session() as session: + rows = (await session.exec(select(CashuTransaction))).all() + assert len(rows) == 1 # no duplicate + + +@pytest.mark.asyncio +async def test_replay_keeps_entries_that_still_fail( + fresh_engine: tuple[AsyncEngine, Path], monkeypatch: pytest.MonkeyPatch +) -> None: + """If replay still can't persist an entry, it stays in the outbox.""" + _, outbox = fresh_engine + monkeypatch.setattr(db, "store_cashu_transaction", lambda *a, **k: _afail()) + monkeypatch.setattr(db.asyncio, "sleep", lambda *_a, **_k: _asyncio_sleep_zero()) + + entry = { + "outbox_id": "keepme", + "queued_at": 1, + "token": "cashuAtokenG", + "amount": 5, + "unit": "sat", + "mint_url": None, + "type": "out", + "request_id": "req-7", + "collected": False, + "created_at": None, + "source": "x-cashu", + "api_key_hashed_key": None, + } + outbox.write_text(json.dumps(entry) + "\n", encoding="utf-8") + + persisted = await replay_cashu_outbox() + assert persisted == 0 + remaining = outbox.read_text(encoding="utf-8").strip().splitlines() + assert len(remaining) == 1 + assert json.loads(remaining[0])["outbox_id"] == "keepme" + + +@pytest.mark.asyncio +async def test_replay_drops_corrupt_lines_and_persists_valid_ones( + fresh_engine: tuple[AsyncEngine, Path] +) -> None: + """Corrupt JSONL lines are dropped; valid entries are persisted.""" + _, outbox = fresh_engine + valid_entry = { + "outbox_id": "good1", + "queued_at": 1, + "token": "cashuAtokenH", + "amount": 300, + "unit": "sat", + "mint_url": None, + "type": "out", + "request_id": "req-8", + "collected": False, + "created_at": None, + "source": "x-cashu", + "api_key_hashed_key": None, + } + outbox.write_text( + json.dumps(valid_entry) + "\n" + '{"token": "broken\n', + encoding="utf-8", + ) + + persisted = await replay_cashu_outbox() + assert persisted == 1 + assert outbox.read_text(encoding="utf-8") == "" + + async with create_session() as session: + rows = (await session.exec(select(CashuTransaction))).all() + assert len(rows) == 1 + assert rows[0].token == "cashuAtokenH" + + +@pytest.mark.asyncio +async def test_store_concurrent_duplicate_insert_yields_one_row( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Two concurrent stores of the same (token, type) yield one row, both True.""" + db_file = tmp_path / "concurrent.db" + engine = create_async_engine(f"sqlite+aiosqlite:///{db_file}") + async with engine.begin() as conn: + await conn.run_sync(SQLModel.metadata.create_all) + monkeypatch.setattr(db, "engine", engine) + monkeypatch.setattr(db, "create_session", create_session, raising=False) + monkeypatch.setenv("CASHU_OUTBOX_PATH", str(tmp_path / "outbox.jsonl")) + + try: + results = await asyncio.gather( + store_cashu_transaction( + token="cashuAtokenI", amount=100, unit="sat", typ="out", request_id="req-9" + ), + store_cashu_transaction( + token="cashuAtokenI", amount=100, unit="sat", typ="out", request_id="req-9" + ), + ) + assert all(results) + async with create_session() as session: + rows = (await session.exec(select(CashuTransaction))).all() + assert len(rows) == 1 + assert rows[0].token == "cashuAtokenI" + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_replay_under_concurrent_replay_is_idempotent( + fresh_engine: tuple[AsyncEngine, Path] +) -> None: + """Two concurrent replay calls on a seeded outbox yield one row, outbox empty.""" + _, outbox = fresh_engine + entry = { + "outbox_id": "concurrent1", + "queued_at": 1, + "token": "cashuAtokenJ", + "amount": 42, + "unit": "sat", + "mint_url": None, + "type": "out", + "request_id": "req-10", + "collected": False, + "created_at": None, + "source": "x-cashu", + "api_key_hashed_key": None, + } + outbox.write_text(json.dumps(entry) + "\n", encoding="utf-8") + + results = await asyncio.gather(replay_cashu_outbox(), replay_cashu_outbox()) + # At least one call persisted the entry; both may report success. + assert sum(results) >= 1 + assert outbox.read_text(encoding="utf-8") == "" + async with create_session() as session: + rows = (await session.exec(select(CashuTransaction))).all() + assert len(rows) == 1 + assert rows[0].token == "cashuAtokenJ" + + +@pytest.mark.asyncio +async def test_replay_all_drains_orphaned_outbox_files( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """replay_all_outbox_files drains every per-PID outbox file, including orphans.""" + engine = _in_memory_engine() + async with engine.begin() as conn: + await conn.run_sync(SQLModel.metadata.create_all) + monkeypatch.setattr(db, "engine", engine) + monkeypatch.setattr(db, "create_session", create_session, raising=False) + # No CASHU_OUTBOX_PATH override; _cashu_outbox_path resolves into tmp_path. + monkeypatch.delenv("CASHU_OUTBOX_PATH", raising=False) + monkeypatch.setattr( + db, "_cashu_outbox_path", + lambda: tmp_path / f"cashu_outbox.{os.getpid()}.jsonl", + ) + + def _make_entry(token: str, oid: str) -> dict[str, object]: + return { + "outbox_id": oid, + "queued_at": 1, + "token": token, + "amount": 100, + "unit": "sat", + "mint_url": None, + "type": "out", + "request_id": f"req-{oid}", + "collected": False, + "created_at": None, + "source": "x-cashu", + "api_key_hashed_key": None, + } + + orphan_a = tmp_path / "cashu_outbox.11111.jsonl" + orphan_b = tmp_path / "cashu_outbox.22222.jsonl" + orphan_a.write_text(json.dumps(_make_entry("cashuOrphanA", "a1")) + "\n", encoding="utf-8") + orphan_b.write_text(json.dumps(_make_entry("cashuOrphanB", "b2")) + "\n", encoding="utf-8") + + try: + total = await replay_all_outbox_files() + assert total == 2 + assert orphan_a.read_text(encoding="utf-8") == "" + assert orphan_b.read_text(encoding="utf-8") == "" + async with create_session() as session: + rows = (await session.exec(select(CashuTransaction))).all() + tokens = {r.token for r in rows} + assert tokens == {"cashuOrphanA", "cashuOrphanB"} + finally: + await engine.dispose() + + +# --- tiny async helpers (can't use `await` inside lambdas) --- + + +async def _asyncio_sleep_zero() -> None: + """No-op async sleep replacement for fast tests.""" + return None + + +async def _afail() -> bool: + return False diff --git a/tests/unit/test_messages_litellm_dispatch.py b/tests/unit/test_messages_litellm_dispatch.py index d7f919a7..d4ba7de5 100644 --- a/tests/unit/test_messages_litellm_dispatch.py +++ b/tests/unit/test_messages_litellm_dispatch.py @@ -1481,7 +1481,7 @@ async def test_x_cashu_transport_error_after_redemption_is_not_retryable( "routstr.upstream.base.recieve_token", new=AsyncMock(return_value=(5_000, "sat", "https://mint")), ), - patch("routstr.upstream.base.store_cashu_transaction", new=AsyncMock()), + patch("routstr.upstream.base.store_cashu_transaction_with_retry", new=AsyncMock(return_value=True)), patch.object( provider, forward_attr, @@ -1601,7 +1601,7 @@ async def test_x_cashu_zero_value_rejected_not_forwarded( "routstr.upstream.base.recieve_token", new=AsyncMock(return_value=(amount, "sat", "https://mint")), ), - patch("routstr.upstream.base.store_cashu_transaction", new=AsyncMock()), + patch("routstr.upstream.base.store_cashu_transaction_with_retry", new=AsyncMock(return_value=True)), patch.object( provider, forward_attr, diff --git a/tests/unit/test_stale_reservations.py b/tests/unit/test_stale_reservations.py index 8c7d7c8f..4687b172 100644 --- a/tests/unit/test_stale_reservations.py +++ b/tests/unit/test_stale_reservations.py @@ -215,7 +215,7 @@ async def test_reset_all_reserved_balances_clears_reserved_at(session: AsyncSess def _refund_patches(refund_token: str = "cashuArefund"): # type: ignore[no-untyped-def] return ( patch("routstr.balance.send_token", AsyncMock(return_value=refund_token)), - patch("routstr.balance.store_cashu_transaction", AsyncMock()), + patch("routstr.balance.store_cashu_transaction_with_retry", AsyncMock(return_value=True)), patch("routstr.balance._refund_cache_get", AsyncMock(return_value=None)), patch("routstr.balance._refund_cache_set", AsyncMock()), ) diff --git a/tests/unit/test_wallet.py b/tests/unit/test_wallet.py index 24381f9a..2d552c76 100644 --- a/tests/unit/test_wallet.py +++ b/tests/unit/test_wallet.py @@ -1076,9 +1076,9 @@ async def test_credit_balance_msat_unit_not_converted() -> None: @pytest.mark.asyncio async def test_credit_balance_survives_audit_store_failure() -> None: """A failure writing the CashuTransaction history record must not undo the - already-committed balance credit. (The silent swallow is a known - audit-trail gap slated for its own fix — this test pins the financial - invariant that the user keeps their credit, not the swallow itself.)""" + already-committed balance credit. (Persistence is spooled to the outbox on + failure; this test pins the financial invariant that the user keeps their + credit, not the spool behavior itself.)""" mock_key = Mock() mock_key.balance = 0 mock_key.hashed_key = "test_hash" @@ -1092,8 +1092,8 @@ async def test_credit_balance_survives_audit_store_failure() -> None: return_value=(1000, "sat", "http://mint:3338"), ): with patch( - "routstr.wallet.store_cashu_transaction", - side_effect=Exception("history table locked"), + "routstr.wallet.store_cashu_transaction_with_retry", + AsyncMock(return_value=False), ): amount = await credit_balance("cashuAtest", mock_key, mock_session)