Merge pull request #622 from Routstr/fix/retry-cashu-storage-writes

fix: retry critical Cashu storage writes

routstr/balance.py

@@ -15,7 +15,9 @@ from .core.db import (
     AsyncSession,
     CashuTransaction,
     get_session,
-    store_cashu_transaction,
+)
+from .core.db import (
+    store_cashu_transaction_with_retry as store_cashu_transaction,
 )
 from .core.logging import get_logger
 from .core.settings import settings

routstr/core/admin.py

@@ -28,7 +28,9 @@ from .db import (
     ModelRow,
     UpstreamProviderRow,
     create_session,
-    store_cashu_transaction,
+)
+from .db import (
+    store_cashu_transaction_with_retry as store_cashu_transaction,
 )
 from .log_manager import log_manager
 from .logging import get_logger
@@ -434,15 +436,25 @@ async def withdraw(
     token = await send_token(
         withdraw_request.amount, withdraw_request.unit, effective_mint
     )
-    await store_cashu_transaction(
-        token=token,
-        amount=withdraw_request.amount,
-        unit=withdraw_request.unit,
-        mint_url=effective_mint,
-        typ="out",
-        collected=False,
-        source="admin",
-    )
+    try:
+        await store_cashu_transaction(
+            token=token,
+            amount=withdraw_request.amount,
+            unit=withdraw_request.unit,
+            mint_url=effective_mint,
+            typ="out",
+            collected=False,
+            source="admin",
+        )
+    except Exception:
+        logger.critical(
+            "Admin withdrawal token issued without a persisted audit record",
+            extra={
+                "amount": withdraw_request.amount,
+                "unit": withdraw_request.unit,
+                "mint_url": effective_mint,
+            },
+        )
     return {"token": token}
 
 

routstr/core/db.py

@@ -1,3 +1,5 @@
+import asyncio
+import hashlib
 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
@@ -287,10 +289,13 @@ async def store_cashu_transaction(
     created_at: int | None = None,
     source: str = "x-cashu",
     api_key_hashed_key: str | None = None,
+    transaction_id: str | None = None,
+    log_failure: bool = True,
 ) -> bool:
     try:
         async with create_session() as session:
             tx = CashuTransaction(
+                id=transaction_id or uuid.uuid4().hex,
                 token=token,
                 amount=amount,
                 unit=unit,
@@ -304,13 +309,93 @@ async def store_cashu_transaction(
             )
             session.add(tx)
             await session.commit()
-        return True
-    except Exception as e:
-        logger.warning(
-            f"Failed to store cashu transaction: {e} (type={typ})",
-            extra={"error": str(e), "type": typ},
-        )
-        return False
+    except Exception:
+        if log_failure:
+            logger.critical(
+                "Failed to store Cashu transaction",
+                extra={"type": typ, "request_id": request_id, "source": source},
+                exc_info=True,
+            )
+        raise
+    return True
+
+
+async def _cashu_transaction_exists(transaction_id: str) -> bool:
+    async with create_session() as session:
+        return await session.get(CashuTransaction, transaction_id) is not None
+
+
+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_attempts: int = 3,
+) -> bool:
+    """Retry a critical Cashu transaction write with bounded backoff."""
+    transaction_id = hashlib.sha256(f"{typ}\0{token}".encode()).hexdigest()
+    last_error: Exception | None = None
+    for attempt in range(1, max_attempts + 1):
+        try:
+            return 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,
+                transaction_id=transaction_id,
+                log_failure=False,
+            )
+        except IntegrityError as error:
+            try:
+                if await _cashu_transaction_exists(transaction_id):
+                    return True
+            except Exception as lookup_error:
+                last_error = lookup_error
+            else:
+                last_error = error
+        except Exception as error:
+            last_error = error
+
+        if last_error is not None:
+            if attempt == max_attempts:
+                break
+            delay = 0.25 * (2 ** (attempt - 1))
+            logger.warning(
+                "Cashu transaction storage failed; retrying",
+                extra={
+                    "type": typ,
+                    "request_id": request_id,
+                    "attempt": attempt,
+                    "max_attempts": max_attempts,
+                    "retry_delay_seconds": delay,
+                },
+            )
+            await asyncio.sleep(delay)
+
+    logger.critical(
+        "Cashu transaction storage failed after bounded retries",
+        extra={
+            "type": typ,
+            "request_id": request_id,
+            "attempts": max_attempts,
+            "error": str(last_error),
+        },
+    )
+    if last_error is None:
+        raise RuntimeError("Cashu transaction storage failed without an exception")
+    raise last_error
 
 
 class UpstreamProviderRow(SQLModel, table=True):  # type: ignore

routstr/upstream/auto_topup.py

@@ -8,7 +8,9 @@ from ..core.db import (
     CashuTransaction,
     UpstreamProviderRow,
     create_session,
-    store_cashu_transaction,
+)
+from ..core.db import (
+    store_cashu_transaction_with_retry as store_cashu_transaction,
 )
 from ..wallet import send_token
 from .routstr import RoutstrUpstreamProvider
@@ -142,16 +144,17 @@ async def _check_and_topup(row: UpstreamProviderRow) -> None:
         )
         return
 
-    stored = await store_cashu_transaction(
-        token=token,
-        amount=amount,
-        unit="sat",
-        mint_url=mint_url,
-        typ="out",
-        collected=False,
-        source="auto_topup",
-    )
-    if not stored:
+    try:
+        await store_cashu_transaction(
+            token=token,
+            amount=amount,
+            unit="sat",
+            mint_url=mint_url,
+            typ="out",
+            collected=False,
+            source="auto_topup",
+        )
+    except Exception:
         logger.critical(
             "Aborting auto top-up because its cashu token could not be persisted",
             extra={"provider_id": row.id, "mint_url": mint_url},

routstr/upstream/base.py

@@ -21,7 +21,9 @@ from ..core.db import (
     AsyncSession,
     UpstreamProviderRow,
     create_session,
-    store_cashu_transaction,
+)
+from ..core.db import (
+    store_cashu_transaction_with_retry as store_cashu_transaction,
 )
 from ..core.exceptions import UpstreamError
 from ..core.redaction import redact_org_ids

routstr/upstream/ehbp.py

@@ -23,7 +23,9 @@ from ..core.db import (
     ApiKey,
     AsyncSession,
     accumulate_routstr_fee,
-    store_cashu_transaction,
+)
+from ..core.db import (
+    store_cashu_transaction_with_retry as store_cashu_transaction,
 )
 from ..core.exceptions import UpstreamError
 from ..core.settings import settings

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 as store_cashu_transaction
 from .core.settings import settings
 from .payment.lnurl import raw_send_to_lnurl
 
@@ -745,11 +745,11 @@ async def credit_balance(
             )
         except Exception:
             pass
-
-        logger.debug(
-            "Cashu token successfully redeemed and stored",
-            extra={"amount": amount, "unit": unit, "mint_url": mint_url},
-        )
+        else:
+            logger.debug(
+                "Cashu token successfully redeemed and stored",
+                extra={"amount": amount, "unit": unit, "mint_url": mint_url},
+            )
         return amount
     except Exception as e:
         logger.error(

tests/unit/test_admin_withdraw.py

@@ -49,3 +49,34 @@ async def test_withdraw_uses_effective_mint_and_records_outgoing_transaction(
         collected=False,
         source="admin",
     )
+
+
+@pytest.mark.asyncio
+async def test_withdraw_returns_issued_token_when_audit_storage_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    mint = "https://primary.example"
+    proofs = [SimpleNamespace(amount=100)]
+    token = "cashuBrecoverable"
+
+    monkeypatch.setattr(admin, "get_wallet", AsyncMock(return_value=object()))
+    monkeypatch.setattr(
+        admin, "get_proofs_per_mint_and_unit", Mock(return_value=proofs)
+    )
+    monkeypatch.setattr(
+        admin, "slow_filter_spend_proofs", AsyncMock(return_value=proofs)
+    )
+    monkeypatch.setattr(admin, "send_token", AsyncMock(return_value=token))
+    monkeypatch.setattr(
+        admin,
+        "store_cashu_transaction",
+        AsyncMock(side_effect=RuntimeError("database unavailable")),
+    )
+    critical = Mock()
+    monkeypatch.setattr(admin.logger, "critical", critical)
+    monkeypatch.setattr(admin.settings, "primary_mint", mint)
+
+    result = await admin.withdraw(Mock(), admin.WithdrawRequest(amount=75))
+
+    assert result == {"token": token}
+    critical.assert_called_once()

tests/unit/test_auto_topup.py

@@ -136,7 +136,7 @@ async def test_auto_topup_does_not_send_untracked_token() -> None:
         ),
         patch(
             "routstr.upstream.auto_topup.store_cashu_transaction",
-            AsyncMock(return_value=False),
+            AsyncMock(side_effect=RuntimeError("database unavailable")),
         ),
     ):
         await _check_and_topup(_row())

tests/unit/test_cashu_transaction_storage_errors.py

@@ -0,0 +1,56 @@
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from routstr.core.db import store_cashu_transaction
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "error",
+    [
+        OSError("disk full"),
+        RuntimeError("connection lost"),
+        ConnectionRefusedError("database unavailable"),
+    ],
+)
+async def test_store_cashu_transaction_propagates_commit_errors(
+    error: Exception,
+) -> None:
+    session = AsyncMock()
+    session.commit.side_effect = error
+    session.__aenter__.return_value = session
+    session.__aexit__.return_value = None
+
+    with (
+        patch("routstr.core.db.create_session", return_value=session),
+        patch("routstr.core.db.logger.critical") as critical,
+    ):
+        with pytest.raises(type(error), match=str(error)):
+            await store_cashu_transaction(
+                token="cashuAtest",
+                amount=1_000,
+                unit="sat",
+                mint_url="https://mint.example",
+                typ="out",
+                request_id="request-1",
+            )
+
+    critical.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_store_cashu_transaction_returns_true_after_commit() -> None:
+    session = AsyncMock()
+    session.__aenter__.return_value = session
+    session.__aexit__.return_value = None
+
+    with patch("routstr.core.db.create_session", return_value=session):
+        stored = await store_cashu_transaction(
+            token="cashuAtest",
+            amount=1_000,
+            unit="sat",
+        )
+
+    assert stored is True
+    session.commit.assert_awaited_once()

tests/unit/test_cashu_transaction_storage_retry.py

@@ -0,0 +1,91 @@
+from typing import Any
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlmodel import SQLModel, select
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from routstr.core import db
+
+
+@pytest.mark.asyncio
+async def test_cashu_transaction_storage_retries_then_succeeds() -> None:
+    store = AsyncMock(side_effect=[OSError("database locked"), True])
+    sleep = AsyncMock()
+
+    with (
+        patch("routstr.core.db.store_cashu_transaction", store),
+        patch("routstr.core.db.asyncio.sleep", sleep),
+    ):
+        stored = await db.store_cashu_transaction_with_retry(
+            token="cashuAretry",
+            amount=100,
+            unit="sat",
+        )
+
+    assert stored is True
+    assert store.await_count == 2
+    sleep.assert_awaited_once_with(0.25)
+
+
+@pytest.mark.asyncio
+async def test_cashu_transaction_retry_is_idempotent_after_ambiguous_commit() -> None:
+    engine = create_async_engine("sqlite+aiosqlite://")
+    async with engine.begin() as connection:
+        await connection.run_sync(SQLModel.metadata.create_all)
+
+    original_store = db.store_cashu_transaction
+    attempts = 0
+
+    async def ambiguous_store(**kwargs: Any) -> bool:
+        nonlocal attempts
+        attempts += 1
+        stored = await original_store(**kwargs)
+        if attempts == 1:
+            raise OSError("connection dropped after commit")
+        return stored
+
+    with (
+        patch.object(db, "engine", engine),
+        patch("routstr.core.db.store_cashu_transaction", ambiguous_store),
+        patch("routstr.core.db.asyncio.sleep", AsyncMock()),
+    ):
+        stored = await db.store_cashu_transaction_with_retry(
+            token="cashuAambiguous",
+            amount=100,
+            unit="sat",
+        )
+
+        async with AsyncSession(engine) as session:
+            result = await session.exec(select(db.CashuTransaction))
+            transactions = result.all()
+
+    assert stored is True
+    assert attempts == 2
+    assert len(transactions) == 1
+    await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_cashu_transaction_storage_raises_after_bounded_retries() -> None:
+    error = OSError("database unavailable")
+    store = AsyncMock(side_effect=error)
+    sleep = AsyncMock()
+
+    with (
+        patch("routstr.core.db.store_cashu_transaction", store),
+        patch("routstr.core.db.asyncio.sleep", sleep),
+        patch("routstr.core.db.logger.critical") as critical,
+    ):
+        with pytest.raises(OSError, match="database unavailable"):
+            await db.store_cashu_transaction_with_retry(
+                token="cashuAfail",
+                amount=100,
+                unit="sat",
+                max_attempts=3,
+            )
+
+    assert store.await_count == 3
+    assert [call.args[0] for call in sleep.await_args_list] == [0.25, 0.5]
+    critical.assert_called_once()