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

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,
@@ -305,15 +310,94 @@ async def store_cashu_transaction(
             session.add(tx)
             await session.commit()
     except Exception:
-        logger.critical(
-            "Failed to store Cashu transaction",
-            extra={"type": typ, "request_id": request_id, "source": source},
-            exc_info=True,
-        )
+        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
     __tablename__ = "upstream_providers"
     __table_args__ = (

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

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
 

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