diff --git a/routstr/balance.py b/routstr/balance.py index cc37d08..4630c22 100644 --- a/routstr/balance.py +++ b/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 diff --git a/routstr/core/admin.py b/routstr/core/admin.py index 09f14f9..71a6e34 100644 --- a/routstr/core/admin.py +++ b/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 diff --git a/routstr/core/db.py b/routstr/core/db.py index d0d7805..d2d1fe3 100644 --- a/routstr/core/db.py +++ b/routstr/core/db.py @@ -1,3 +1,4 @@ +import asyncio import os import pathlib import sqlite3 @@ -314,6 +315,66 @@ async def store_cashu_transaction( return True +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.""" + 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, + ) + except Exception as error: + last_error = error + 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__ = ( diff --git a/routstr/upstream/auto_topup.py b/routstr/upstream/auto_topup.py index cb1ede9..a88a28f 100644 --- a/routstr/upstream/auto_topup.py +++ b/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 diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index 306aca1..c87e4ee 100644 --- a/routstr/upstream/base.py +++ b/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 diff --git a/routstr/upstream/ehbp.py b/routstr/upstream/ehbp.py index ba58aae..213be7a 100644 --- a/routstr/upstream/ehbp.py +++ b/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 diff --git a/routstr/wallet.py b/routstr/wallet.py index cd7798d..a79b1f9b 100644 --- a/routstr/wallet.py +++ b/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 diff --git a/tests/unit/test_cashu_transaction_storage_retry.py b/tests/unit/test_cashu_transaction_storage_retry.py new file mode 100644 index 0000000..5c0f3a4 --- /dev/null +++ b/tests/unit/test_cashu_transaction_storage_retry.py @@ -0,0 +1,49 @@ +from unittest.mock import AsyncMock, patch + +import pytest + +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_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()