fix build

Remove require_parameters injection from OpenRouter provider

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")

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",

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"
 
@@ -246,6 +323,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,
@@ -287,9 +369,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,
@@ -304,11 +396,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

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)

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",

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
 
@@ -734,7 +734,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,
@@ -744,7 +744,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",

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"),

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

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,

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()),
     )

tests/unit/test_wallet.py

@@ -1062,9 +1062,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"
@@ -1078,8 +1078,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)