From 8b3b59e176881e7df7bb4775b0f107f50d8fa3ec Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sat, 18 Jul 2026 14:16:30 +0200 Subject: [PATCH 1/4] fix: propagate Cashu transaction storage errors --- routstr/core/db.py | 37 ++++++-------- .../test_cashu_transaction_storage_errors.py | 51 +++++++++++++++++++ 2 files changed, 66 insertions(+), 22 deletions(-) create mode 100644 tests/unit/test_cashu_transaction_storage_errors.py diff --git a/routstr/core/db.py b/routstr/core/db.py index 9ebb87e9..6e629da1 100644 --- a/routstr/core/db.py +++ b/routstr/core/db.py @@ -288,29 +288,22 @@ async def store_cashu_transaction( source: str = "x-cashu", api_key_hashed_key: str | None = None, ) -> bool: - try: - async with create_session() as session: - tx = CashuTransaction( - token=token, - amount=amount, - unit=unit, - mint_url=mint_url, - type=typ, - request_id=request_id, - collected=collected, - created_at=created_at or int(time.time()), - source=source, - api_key_hashed_key=api_key_hashed_key, - ) - 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}, + async with create_session() as session: + tx = CashuTransaction( + token=token, + amount=amount, + unit=unit, + mint_url=mint_url, + type=typ, + request_id=request_id, + collected=collected, + created_at=created_at or int(time.time()), + source=source, + api_key_hashed_key=api_key_hashed_key, ) - return False + session.add(tx) + await session.commit() + return True class UpstreamProviderRow(SQLModel, table=True): # type: ignore diff --git a/tests/unit/test_cashu_transaction_storage_errors.py b/tests/unit/test_cashu_transaction_storage_errors.py new file mode 100644 index 00000000..cc6d5221 --- /dev/null +++ b/tests/unit/test_cashu_transaction_storage_errors.py @@ -0,0 +1,51 @@ +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): + 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", + ) + + +@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() From 90da3803c65adf8e819b5ac06be88f05d8fbe605 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sat, 18 Jul 2026 14:22:16 +0200 Subject: [PATCH 2/4] fix: preserve caller recovery on storage errors --- routstr/core/admin.py | 28 ++++++++++----- routstr/core/db.py | 36 +++++++++++-------- routstr/upstream/auto_topup.py | 21 +++++------ routstr/wallet.py | 10 +++--- tests/unit/test_admin_withdraw.py | 31 ++++++++++++++++ tests/unit/test_auto_topup.py | 2 +- .../test_cashu_transaction_storage_errors.py | 7 +++- 7 files changed, 95 insertions(+), 40 deletions(-) diff --git a/routstr/core/admin.py b/routstr/core/admin.py index c6738efc..09f14f98 100644 --- a/routstr/core/admin.py +++ b/routstr/core/admin.py @@ -434,15 +434,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} diff --git a/routstr/core/db.py b/routstr/core/db.py index 6e629da1..d0d7805e 100644 --- a/routstr/core/db.py +++ b/routstr/core/db.py @@ -288,21 +288,29 @@ async def store_cashu_transaction( source: str = "x-cashu", api_key_hashed_key: str | None = None, ) -> bool: - async with create_session() as session: - tx = CashuTransaction( - token=token, - amount=amount, - unit=unit, - mint_url=mint_url, - type=typ, - request_id=request_id, - collected=collected, - created_at=created_at or int(time.time()), - source=source, - api_key_hashed_key=api_key_hashed_key, + try: + async with create_session() as session: + tx = CashuTransaction( + token=token, + amount=amount, + unit=unit, + mint_url=mint_url, + type=typ, + request_id=request_id, + collected=collected, + created_at=created_at or int(time.time()), + source=source, + api_key_hashed_key=api_key_hashed_key, + ) + 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, ) - session.add(tx) - await session.commit() + raise return True diff --git a/routstr/upstream/auto_topup.py b/routstr/upstream/auto_topup.py index 3517be7d..cb1ede96 100644 --- a/routstr/upstream/auto_topup.py +++ b/routstr/upstream/auto_topup.py @@ -142,16 +142,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}, diff --git a/routstr/wallet.py b/routstr/wallet.py index 68f93cc2..cd7798d2 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -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( diff --git a/tests/unit/test_admin_withdraw.py b/tests/unit/test_admin_withdraw.py index 1c3a1919..e54f7d01 100644 --- a/tests/unit/test_admin_withdraw.py +++ b/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() diff --git a/tests/unit/test_auto_topup.py b/tests/unit/test_auto_topup.py index c05c85e5..2adb13e6 100644 --- a/tests/unit/test_auto_topup.py +++ b/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()) diff --git a/tests/unit/test_cashu_transaction_storage_errors.py b/tests/unit/test_cashu_transaction_storage_errors.py index cc6d5221..b26b4115 100644 --- a/tests/unit/test_cashu_transaction_storage_errors.py +++ b/tests/unit/test_cashu_transaction_storage_errors.py @@ -22,7 +22,10 @@ async def test_store_cashu_transaction_propagates_commit_errors( session.__aenter__.return_value = session session.__aexit__.return_value = None - with patch("routstr.core.db.create_session", return_value=session): + 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", @@ -33,6 +36,8 @@ async def test_store_cashu_transaction_propagates_commit_errors( request_id="request-1", ) + critical.assert_called_once() + @pytest.mark.asyncio async def test_store_cashu_transaction_returns_true_after_commit() -> None: From c9533c872a44b96ed409ed9b18023e20af9f5f0f Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sat, 18 Jul 2026 14:28:18 +0200 Subject: [PATCH 3/4] fix: retry critical Cashu storage writes --- routstr/balance.py | 4 +- routstr/core/admin.py | 4 +- routstr/core/db.py | 61 +++++++++++++++++++ routstr/upstream/auto_topup.py | 4 +- routstr/upstream/base.py | 4 +- routstr/upstream/ehbp.py | 4 +- routstr/wallet.py | 2 +- .../test_cashu_transaction_storage_retry.py | 49 +++++++++++++++ 8 files changed, 126 insertions(+), 6 deletions(-) create mode 100644 tests/unit/test_cashu_transaction_storage_retry.py diff --git a/routstr/balance.py b/routstr/balance.py index cc37d089..4630c224 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 09f14f98..71a6e348 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 d0d7805e..d2d1fe30 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 cb1ede96..a88a28fb 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 306aca14..c87e4ee4 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 ba58aae7..213be7af 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 cd7798d2..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 00000000..5c0f3a40 --- /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() From a3a4d69ed3585b29d107aad4cbe3542ebe8947ed Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sat, 18 Jul 2026 14:33:40 +0200 Subject: [PATCH 4/4] fix: make storage retries idempotent --- routstr/core/db.py | 35 +++++++++++++--- .../test_cashu_transaction_storage_retry.py | 42 +++++++++++++++++++ 2 files changed, 71 insertions(+), 6 deletions(-) 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")