diff --git a/routstr/core/db.py b/routstr/core/db.py index d2d1fe30..20faade5 100644 --- a/routstr/core/db.py +++ b/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( - "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, @@ -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)) diff --git a/tests/unit/test_cashu_transaction_storage_retry.py b/tests/unit/test_cashu_transaction_storage_retry.py index 5c0f3a40..e9c52fe6 100644 --- a/tests/unit/test_cashu_transaction_storage_retry.py +++ b/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")