fix: retry Cashu transaction storage

migrations/versions/d7e8f9a0b1c2_unique_token_type_cashu_transactions.py

@@ -0,0 +1,45 @@
+"""Add unique token/type index to Cashu transactions.
+
+Revision ID: d7e8f9a0b1c2
+Revises: c6d7e8f9a0b1
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "d7e8f9a0b1c2"
+down_revision = "c6d7e8f9a0b1"
+branch_labels = None
+depends_on = None
+
+_INDEX_NAME = "uq_cashu_transactions_token_type"
+
+
+def upgrade() -> None:
+    connection = op.get_bind()
+    connection.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 row_number "
+            "    FROM cashu_transactions"
+            "  ) WHERE row_number > 1"
+            ")"
+        )
+    )
+    op.create_index(
+        _INDEX_NAME,
+        "cashu_transactions",
+        ["token", "type"],
+        unique=True,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(_INDEX_NAME, table_name="cashu_transactions")

routstr/core/db.py

@@ -1,3 +1,4 @@
+import asyncio
 import os
 import pathlib
 import sqlite3
@@ -10,7 +11,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
@@ -105,8 +106,7 @@ async def reset_all_reserved_balances(session: AsyncSession) -> None:
 async def release_stale_reservations(
     session: AsyncSession, max_age_seconds: int
 ) -> int:
-    """Release reservations whose last reserve is older than max_age_seconds.
-    """
+    """Release reservations whose last reserve is older than max_age_seconds."""
     cutoff = int(time.time()) - max_age_seconds
     stmt = (
         update(ApiKey)
@@ -154,9 +154,7 @@ async def prune_dead_api_keys(session: AsyncSession, min_age_seconds: int) -> in
         .where(col(ApiKey.total_spent) == 0)
         .where(col(ApiKey.total_requests) == 0)
         .where(col(ApiKey.parent_key_hash).is_(None))
-        .where(
-            (col(ApiKey.created_at).is_(None)) | (col(ApiKey.created_at) < cutoff)
-        )
+        .where((col(ApiKey.created_at).is_(None)) | (col(ApiKey.created_at) < cutoff))
         .where(~pending_invoice)
         .where(~has_children)
     )
@@ -246,6 +244,9 @@ 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,
@@ -276,6 +277,23 @@ class CashuTransaction(SQLModel, table=True):  # type: ignore
     )
 
 
+async def _insert_cashu_transaction(transaction: CashuTransaction) -> None:
+    async with create_session() as session:
+        session.add(transaction)
+        await session.commit()
+
+
+async def _cashu_transaction_exists(token: str, typ: str) -> bool:
+    async with create_session() as session:
+        result = await session.exec(
+            select(CashuTransaction).where(
+                CashuTransaction.token == token,
+                CashuTransaction.type == typ,
+            )
+        )
+        return result.first() is not None
+
+
 async def store_cashu_transaction(
     token: str,
     amount: int,
@@ -287,28 +305,62 @@ async def store_cashu_transaction(
     created_at: int | None = None,
     source: str = "x-cashu",
     api_key_hashed_key: str | None = None,
-) -> None:
-    try:
-        async with create_session() as session:
-            tx = CashuTransaction(
-                token=token,
-                amount=amount,
-                unit=unit,
-                mint_url=mint_url,
-                type=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,
+    max_attempts: int = 3,
+) -> bool:
+    """Store a Cashu transaction, retrying transient database failures."""
+    transaction = CashuTransaction(
+        token=token,
+        amount=amount,
+        unit=unit,
+        mint_url=mint_url,
+        type=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,
+    )
+
+    for attempt in range(1, max_attempts + 1):
+        try:
+            await _insert_cashu_transaction(transaction)
+            return True
+        except IntegrityError:
+            if await _cashu_transaction_exists(token, typ):
+                return True
+            break
+        except OperationalError as error:
+            if attempt == max_attempts:
+                break
+            delay = 0.25 * (2 ** (attempt - 1))
+            logger.warning(
+                "Transient database failure storing Cashu transaction; retrying",
+                extra={
+                    "error": str(error),
+                    "type": typ,
+                    "request_id": request_id,
+                    "attempt": attempt,
+                    "max_attempts": max_attempts,
+                    "retry_delay_seconds": delay,
+                },
             )
-            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},
-        )
+            await asyncio.sleep(delay)
+        except Exception:
+            break
+
+    logger.critical(
+        "Cashu transaction could not be stored after bounded retries",
+        extra={
+            "type": typ,
+            "request_id": request_id,
+            "amount": amount,
+            "unit": unit,
+            "mint_url": mint_url,
+            "token": token,
+            "max_attempts": max_attempts,
+        },
+    )
+    return False
 
 
 class UpstreamProviderRow(SQLModel, table=True):  # type: ignore
@@ -358,9 +410,7 @@ class CliToken(SQLModel, table=True):  # type: ignore
     """Long-lived authorization token for CLI/agent use against admin endpoints."""
 
     __tablename__ = "cli_tokens"
-    id: str = Field(
-        primary_key=True, default_factory=lambda: uuid.uuid4().hex
-    )
+    id: str = Field(primary_key=True, default_factory=lambda: uuid.uuid4().hex)
     token: str = Field(unique=True, index=True, description="Bearer token value")
     name: str = Field(description="Human-readable label for this token")
     created_at: int = Field(default_factory=lambda: int(time.time()))

tests/unit/test_cashu_transaction_storage.py

@@ -0,0 +1,73 @@
+from unittest.mock import AsyncMock
+
+import pytest
+from sqlalchemy.exc import IntegrityError, OperationalError
+
+from routstr.core import db
+
+
+@pytest.mark.asyncio
+async def test_store_cashu_transaction_retries_transient_database_failures(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    insert = AsyncMock(
+        side_effect=[
+            OperationalError("insert", {}, Exception("database is locked")),
+            None,
+        ]
+    )
+    sleep = AsyncMock()
+    monkeypatch.setattr(db, "_insert_cashu_transaction", insert)
+    monkeypatch.setattr(db.asyncio, "sleep", sleep)
+
+    stored = await db.store_cashu_transaction(
+        token="cashuAretry", amount=100, unit="sat", typ="out"
+    )
+
+    assert stored is True
+    assert insert.await_count == 2
+    sleep.assert_awaited_once_with(0.25)
+
+
+@pytest.mark.asyncio
+async def test_store_cashu_transaction_stops_after_bounded_retries(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    insert = AsyncMock(
+        side_effect=OperationalError("insert", {}, Exception("database is locked"))
+    )
+    sleep = AsyncMock()
+    critical = AsyncMock()
+    monkeypatch.setattr(db, "_insert_cashu_transaction", insert)
+    monkeypatch.setattr(db.asyncio, "sleep", sleep)
+    monkeypatch.setattr(db.logger, "critical", critical)
+
+    stored = await db.store_cashu_transaction(
+        token="cashuAfailed", amount=100, unit="sat", typ="out"
+    )
+
+    assert stored is False
+    assert insert.await_count == 3
+    assert sleep.await_count == 2
+    critical.assert_called_once()
+    assert critical.call_args.kwargs["extra"]["token"] == "cashuAfailed"
+
+
+@pytest.mark.asyncio
+async def test_store_cashu_transaction_treats_duplicate_as_success(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    insert = AsyncMock(
+        side_effect=IntegrityError("insert", {}, Exception("unique constraint"))
+    )
+    exists = AsyncMock(return_value=True)
+    monkeypatch.setattr(db, "_insert_cashu_transaction", insert)
+    monkeypatch.setattr(db, "_cashu_transaction_exists", exists)
+
+    stored = await db.store_cashu_transaction(
+        token="cashuAduplicate", amount=100, unit="sat", typ="out"
+    )
+
+    assert stored is True
+    insert.assert_awaited_once()
+    exists.assert_awaited_once_with("cashuAduplicate", "out")