fix: make storage retries idempotent

routstr/core/db.py

@@ -1,4 +1,5 @@
 import asyncio
+import hashlib
 import os
 import pathlib
 import sqlite3
@@ -11,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
@@ -288,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,
@@ -306,15 +310,21 @@ async def store_cashu_transaction(
             session.add(tx)
             await session.commit()
     except Exception:
-        logger.critical(
+        if log_failure:
-            "Failed to store Cashu transaction",
+            logger.critical(
-            extra={"type": typ, "request_id": request_id, "source": source},
+                "Failed to store Cashu transaction",
-            exc_info=True,
+                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,
@@ -329,6 +339,7 @@ async def store_cashu_transaction_with_retry(
     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:
@@ -343,9 +354,21 @@ async def store_cashu_transaction_with_retry(
                 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))

tests/unit/test_cashu_transaction_storage_retry.py

@@ -1,6 +1,10 @@
+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
 
@@ -25,6 +29,44 @@ async def test_cashu_transaction_storage_retries_then_succeeds() -> None:
     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")