Merge pull request #606 from Routstr/fix/retry-cashu-storage-idempotent-index

Retry Cashu transaction storage with idempotent uniqueness migration

migrations/versions/d7e8f9a0b1c2_unique_token_type_cashu_transactions.py

@@ -0,0 +1,123 @@
+"""Add unique token/type index to Cashu transactions.
+
+Revision ID: d7e8f9a0b1c2
+Revises: c6d7e8f9a0b1
+"""
+
+from __future__ import annotations
+
+from itertools import groupby
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.engine import Connection, RowMapping
+
+revision = "d7e8f9a0b1c2"
+down_revision = "c6d7e8f9a0b1"
+branch_labels = None
+depends_on = None
+
+_INDEX_NAME = "uq_cashu_transactions_token_type"
+_NULLABLE_METADATA = ("request_id", "mint_url", "api_key_hashed_key")
+
+
+def _merge_duplicate_transactions(connection: Connection) -> None:
+    transactions = sa.Table(
+        "cashu_transactions",
+        sa.MetaData(),
+        autoload_with=connection,
+    )
+    duplicate_keys = (
+        sa.select(transactions.c.token, transactions.c.type)
+        .group_by(transactions.c.token, transactions.c.type)
+        .having(sa.func.count() > 1)
+        .subquery()
+    )
+    rows = (
+        connection.execute(
+            sa.select(transactions)
+            .join(
+                duplicate_keys,
+                sa.and_(
+                    transactions.c.token == duplicate_keys.c.token,
+                    transactions.c.type == duplicate_keys.c.type,
+                ),
+            )
+            .order_by(
+                transactions.c.token,
+                transactions.c.type,
+                transactions.c.created_at,
+                transactions.c.id,
+            )
+        )
+        .mappings()
+        .all()
+    )
+
+    def transaction_key(row: RowMapping) -> tuple[str, str]:
+        return row["token"], row["type"]
+
+    for _, grouped_rows in groupby(rows, key=transaction_key):
+        duplicates = list(grouped_rows)
+        if len(duplicates) < 2:
+            continue
+
+        keeper = duplicates[0]
+        updates: dict[str, object] = {
+            "collected": any(row["collected"] for row in duplicates),
+            "swept": any(row["swept"] for row in duplicates),
+        }
+        for column in _NULLABLE_METADATA:
+            if not keeper[column]:
+                updates[column] = next(
+                    (row[column] for row in duplicates if row[column]),
+                    keeper[column],
+                )
+        if not keeper["source"]:
+            updates["source"] = next(
+                (row["source"] for row in duplicates if row["source"]),
+                keeper["source"],
+            )
+
+        connection.execute(
+            transactions.update()
+            .where(transactions.c.id == keeper["id"])
+            .values(**updates)
+        )
+        connection.execute(
+            transactions.delete().where(
+                transactions.c.id.in_(row["id"] for row in duplicates[1:])
+            )
+        )
+
+
+def upgrade() -> None:
+    connection = op.get_bind()
+    _merge_duplicate_transactions(connection)
+
+    existing_index = next(
+        (
+            index
+            for index in sa.inspect(connection).get_indexes("cashu_transactions")
+            if index["name"] == _INDEX_NAME
+        ),
+        None,
+    )
+    if existing_index is not None:
+        if existing_index["unique"] and existing_index["column_names"] == [
+            "token",
+            "type",
+        ]:
+            return
+        op.drop_index(_INDEX_NAME, table_name="cashu_transactions")
+
+    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,80 @@ 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,
+    )
+
+    last_error_type: str | None = None
+    attempts_performed = 0
+    for attempt in range(1, max_attempts + 1):
+        attempts_performed = attempt
+        retry_error: OperationalError | None = None
+        try:
+            await _insert_cashu_transaction(transaction)
+            return True
+        except IntegrityError as error:
+            last_error_type = type(error).__name__
+            try:
+                if await _cashu_transaction_exists(token, typ):
+                    return True
+            except OperationalError as lookup_error:
+                retry_error = lookup_error
+            except Exception as lookup_error:
+                last_error_type = type(lookup_error).__name__
+                break
+            else:
+                break
+        except OperationalError as error:
+            retry_error = error
+        except Exception as error:
+            last_error_type = type(error).__name__
+            break
+
+        if retry_error is not None:
+            last_error_type = type(retry_error.orig).__name__
+            if attempt == max_attempts:
+                break
+            delay = 0.25 * (2 ** (attempt - 1))
+            logger.warning(
+                "Transient database failure storing Cashu transaction; retrying",
+                extra={
+                    "error_type": last_error_type,
+                    "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)
+
+    logger.critical(
+        "Cashu transaction could not be stored",
+        extra={
+            "error_type": last_error_type,
+            "type": typ,
+            "request_id": request_id,
+            "amount": amount,
+            "unit": unit,
+            "mint_url": mint_url,
+            "attempts_performed": attempts_performed,
+            "max_attempts": max_attempts,
+        },
+    )
+    return False
 
 
 class UpstreamProviderRow(SQLModel, table=True):  # type: ignore
@@ -358,9 +428,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()))

routstr/wallet.py

@@ -733,8 +733,9 @@ async def credit_balance(
             extra={"new_balance": key.balance},
         )
 
+        transaction_stored = False
         try:
-            await store_cashu_transaction(
+            transaction_stored = await store_cashu_transaction(
                 token=cashu_token,
                 amount=original_amount,
                 unit=original_unit,
@@ -747,8 +748,13 @@ async def credit_balance(
             pass
 
         logger.debug(
-            "Cashu token successfully redeemed and stored",
-            extra={"amount": amount, "unit": unit, "mint_url": mint_url},
+            "Cashu token successfully redeemed",
+            extra={
+                "amount": amount,
+                "unit": unit,
+                "mint_url": mint_url,
+                "transaction_stored": transaction_stored,
+            },
         )
         return amount
     except Exception as e:

tests/unit/test_cashu_transaction_storage.py

@@ -0,0 +1,181 @@
+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_with(
+        "Cashu transaction could not be stored",
+        extra={
+            "error_type": "Exception",
+            "type": "out",
+            "request_id": None,
+            "amount": 100,
+            "unit": "sat",
+            "mint_url": None,
+            "attempts_performed": 3,
+            "max_attempts": 3,
+        },
+    )
+    assert "cashuAfailed" not in repr(critical.call_args)
+
+
+@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")
+
+
+@pytest.mark.asyncio
+async def test_store_cashu_transaction_retries_duplicate_lookup_failure(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    insert = AsyncMock(
+        side_effect=IntegrityError("insert", {}, Exception("unique constraint"))
+    )
+    exists = AsyncMock(
+        side_effect=[
+            OperationalError("select", {}, Exception("database is locked")),
+            True,
+        ]
+    )
+    sleep = AsyncMock()
+    warning = AsyncMock()
+    monkeypatch.setattr(db, "_insert_cashu_transaction", insert)
+    monkeypatch.setattr(db, "_cashu_transaction_exists", exists)
+    monkeypatch.setattr(db.asyncio, "sleep", sleep)
+    monkeypatch.setattr(db.logger, "warning", warning)
+
+    stored = await db.store_cashu_transaction(
+        token="cashuAlookup-retry", amount=100, unit="sat", typ="out"
+    )
+
+    assert stored is True
+    assert insert.await_count == 2
+    assert exists.await_count == 2
+    sleep.assert_awaited_once_with(0.25)
+    assert "cashuAlookup-retry" not in repr(warning.call_args)
+    assert warning.call_args.kwargs["extra"]["error_type"] == "Exception"
+
+
+@pytest.mark.asyncio
+async def test_store_cashu_transaction_bounds_duplicate_lookup_failures(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    insert = AsyncMock(
+        side_effect=IntegrityError("insert", {}, Exception("unique constraint"))
+    )
+    exists = AsyncMock(
+        side_effect=OperationalError("select", {}, Exception("database is locked"))
+    )
+    sleep = AsyncMock()
+    critical = AsyncMock()
+    monkeypatch.setattr(db, "_insert_cashu_transaction", insert)
+    monkeypatch.setattr(db, "_cashu_transaction_exists", exists)
+    monkeypatch.setattr(db.asyncio, "sleep", sleep)
+    monkeypatch.setattr(db.logger, "critical", critical)
+
+    stored = await db.store_cashu_transaction(
+        token="cashuAlookup-failed", amount=100, unit="sat", typ="out"
+    )
+
+    assert stored is False
+    assert insert.await_count == 3
+    assert exists.await_count == 3
+    assert sleep.await_count == 2
+    critical.assert_called_once()
+    assert "cashuAlookup-failed" not in repr(critical.call_args)
+
+
+@pytest.mark.asyncio
+async def test_store_cashu_transaction_contains_non_transient_lookup_failure(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    insert = AsyncMock(
+        side_effect=IntegrityError("insert", {}, Exception("unique constraint"))
+    )
+    exists = AsyncMock(side_effect=RuntimeError("lookup failed"))
+    critical = AsyncMock()
+    monkeypatch.setattr(db, "_insert_cashu_transaction", insert)
+    monkeypatch.setattr(db, "_cashu_transaction_exists", exists)
+    monkeypatch.setattr(db.logger, "critical", critical)
+
+    stored = await db.store_cashu_transaction(
+        token="cashuAlookup-error", amount=100, unit="sat", typ="out"
+    )
+
+    assert stored is False
+    insert.assert_awaited_once()
+    exists.assert_awaited_once()
+    critical.assert_called_once_with(
+        "Cashu transaction could not be stored",
+        extra={
+            "error_type": "RuntimeError",
+            "type": "out",
+            "request_id": None,
+            "amount": 100,
+            "unit": "sat",
+            "mint_url": None,
+            "attempts_performed": 1,
+            "max_attempts": 3,
+        },
+    )

tests/unit/test_cashu_transaction_uniqueness_migration.py

@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+
+import sqlalchemy as sa
+
+_MIGRATION_PATH = (
+    Path(__file__).resolve().parents[2]
+    / "migrations"
+    / "versions"
+    / "d7e8f9a0b1c2_unique_token_type_cashu_transactions.py"
+)
+_spec = importlib.util.spec_from_file_location(
+    "cashu_transaction_uniqueness_migration", _MIGRATION_PATH
+)
+assert _spec is not None and _spec.loader is not None
+migration = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(migration)
+
+
+def test_duplicate_merge_preserves_state_and_missing_linkage() -> None:
+    engine = sa.create_engine("sqlite:///:memory:")
+    metadata = sa.MetaData()
+    transactions = sa.Table(
+        "cashu_transactions",
+        metadata,
+        sa.Column("id", sa.String, primary_key=True),
+        sa.Column("token", sa.String, nullable=False),
+        sa.Column("amount", sa.Integer, nullable=False),
+        sa.Column("unit", sa.String, nullable=False),
+        sa.Column("mint_url", sa.String),
+        sa.Column("type", sa.String, nullable=False),
+        sa.Column("request_id", sa.String),
+        sa.Column("created_at", sa.Integer, nullable=False),
+        sa.Column("collected", sa.Boolean, nullable=False),
+        sa.Column("swept", sa.Boolean, nullable=False),
+        sa.Column("source", sa.String, nullable=False),
+        sa.Column("api_key_hashed_key", sa.String),
+    )
+    metadata.create_all(engine)
+
+    with engine.begin() as connection:
+        connection.execute(
+            transactions.insert(),
+            [
+                {
+                    "id": "oldest",
+                    "token": "cashuAduplicate",
+                    "amount": 100,
+                    "unit": "sat",
+                    "mint_url": "",
+                    "type": "out",
+                    "request_id": "",
+                    "created_at": 1,
+                    "collected": False,
+                    "swept": False,
+                    "source": "",
+                    "api_key_hashed_key": "",
+                },
+                {
+                    "id": "newer",
+                    "token": "cashuAduplicate",
+                    "amount": 100,
+                    "unit": "sat",
+                    "mint_url": "https://mint.example",
+                    "type": "out",
+                    "request_id": "request-newer",
+                    "created_at": 2,
+                    "collected": True,
+                    "swept": False,
+                    "source": "apikey",
+                    "api_key_hashed_key": "hashed-key",
+                },
+                {
+                    "id": "newest",
+                    "token": "cashuAduplicate",
+                    "amount": 100,
+                    "unit": "sat",
+                    "mint_url": "https://other.example",
+                    "type": "out",
+                    "request_id": "request-newest",
+                    "created_at": 3,
+                    "collected": False,
+                    "swept": True,
+                    "source": "x-cashu",
+                    "api_key_hashed_key": "other-key",
+                },
+            ],
+        )
+
+        migration._merge_duplicate_transactions(connection)
+        rows = connection.execute(sa.select(transactions)).mappings().all()
+
+    assert rows == [
+        {
+            "id": "oldest",
+            "token": "cashuAduplicate",
+            "amount": 100,
+            "unit": "sat",
+            "mint_url": "https://mint.example",
+            "type": "out",
+            "request_id": "request-newer",
+            "created_at": 1,
+            "collected": True,
+            "swept": True,
+            "source": "apikey",
+            "api_key_hashed_key": "hashed-key",
+        }
+    ]
+
+
+def test_upgrade_is_idempotent_when_unique_index_already_exists(monkeypatch) -> None:
+    engine = sa.create_engine("sqlite:///:memory:")
+    metadata = sa.MetaData()
+    sa.Table(
+        "cashu_transactions",
+        metadata,
+        sa.Column("id", sa.String, primary_key=True),
+        sa.Column("token", sa.String, nullable=False),
+        sa.Column("type", sa.String, nullable=False),
+        sa.Column("created_at", sa.Integer, nullable=False),
+        sa.Column("collected", sa.Boolean, nullable=False),
+        sa.Column("swept", sa.Boolean, nullable=False),
+        sa.Column("source", sa.String, nullable=False),
+        sa.Column("request_id", sa.String),
+        sa.Column("mint_url", sa.String),
+        sa.Column("api_key_hashed_key", sa.String),
+    )
+    metadata.create_all(engine)
+
+    with engine.begin() as connection:
+        connection.exec_driver_sql(
+            "CREATE UNIQUE INDEX uq_cashu_transactions_token_type "
+            "ON cashu_transactions (token, type)"
+        )
+        monkeypatch.setattr(migration.op, "get_bind", lambda: connection)
+        monkeypatch.setattr(
+            migration.op,
+            "create_index",
+            lambda *args, **kwargs: connection.exec_driver_sql(
+                "CREATE UNIQUE INDEX uq_cashu_transactions_token_type "
+                "ON cashu_transactions (token, type)"
+            ),
+        )
+
+        migration.upgrade()
+
+        indexes = sa.inspect(connection).get_indexes("cashu_transactions")
+
+    assert indexes == [
+        {
+            "name": "uq_cashu_transactions_token_type",
+            "column_names": ["token", "type"],
+            "unique": 1,
+            "dialect_options": {},
+        }
+    ]

tests/unit/test_wallet.py

@@ -1060,31 +1060,71 @@ 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.)"""
-    mock_key = Mock()
-    mock_key.balance = 0
-    mock_key.hashed_key = "test_hash"
+async def test_credit_balance_reports_audit_store_false() -> None:
+    mock_key = Mock(balance=0, hashed_key="test_hash")
     mock_session = AsyncMock()
+    debug = Mock()
 
     from routstr.core.settings import settings
 
-    with patch.object(settings, "cashu_mints", ["http://mint:3338"]):
-        with patch(
+    with (
+        patch.object(settings, "cashu_mints", ["http://mint:3338"]),
+        patch(
             "routstr.wallet.recieve_token",
             return_value=(1000, "sat", "http://mint:3338"),
-        ):
-            with patch(
-                "routstr.wallet.store_cashu_transaction",
-                side_effect=Exception("history table locked"),
-            ):
-                amount = await credit_balance("cashuAtest", mock_key, mock_session)
+        ),
+        patch("routstr.wallet.store_cashu_transaction", return_value=False),
+        patch("routstr.wallet.logger.debug", debug),
+    ):
+        amount = await credit_balance("cashuAtest", mock_key, mock_session)
 
     assert amount == 1_000_000
     assert mock_session.commit.called
+    debug.assert_called_once_with(
+        "Cashu token successfully redeemed",
+        extra={
+            "amount": 1_000_000,
+            "unit": "sat",
+            "mint_url": "http://mint:3338",
+            "transaction_stored": False,
+        },
+    )
+
+
+@pytest.mark.asyncio
+async def test_credit_balance_survives_audit_store_failure() -> None:
+    """A failed CashuTransaction history write must not undo balance credit."""
+    mock_key = Mock(balance=0, hashed_key="test_hash")
+    mock_session = AsyncMock()
+    debug = Mock()
+
+    from routstr.core.settings import settings
+
+    with (
+        patch.object(settings, "cashu_mints", ["http://mint:3338"]),
+        patch(
+            "routstr.wallet.recieve_token",
+            return_value=(1000, "sat", "http://mint:3338"),
+        ),
+        patch(
+            "routstr.wallet.store_cashu_transaction",
+            side_effect=Exception("history table locked"),
+        ),
+        patch("routstr.wallet.logger.debug", debug),
+    ):
+        amount = await credit_balance("cashuAtest", mock_key, mock_session)
+
+    assert amount == 1_000_000
+    assert mock_session.commit.called
+    debug.assert_called_once_with(
+        "Cashu token successfully redeemed",
+        extra={
+            "amount": 1_000_000,
+            "unit": "sat",
+            "mint_url": "http://mint:3338",
+            "transaction_stored": False,
+        },
+    )
 
 
 @pytest.mark.asyncio