From aea24d23dae15ef5ee28ca697f4ce9a4d6e38fac Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sun, 12 Jul 2026 13:29:09 +0200 Subject: [PATCH 1/2] fix: retry Cashu transaction storage --- ...c2_unique_token_type_cashu_transactions.py | 45 +++++++ routstr/core/db.py | 110 +++++++++++++----- tests/unit/test_cashu_transaction_storage.py | 73 ++++++++++++ 3 files changed, 198 insertions(+), 30 deletions(-) create mode 100644 migrations/versions/d7e8f9a0b1c2_unique_token_type_cashu_transactions.py create mode 100644 tests/unit/test_cashu_transaction_storage.py diff --git a/migrations/versions/d7e8f9a0b1c2_unique_token_type_cashu_transactions.py b/migrations/versions/d7e8f9a0b1c2_unique_token_type_cashu_transactions.py new file mode 100644 index 00000000..2241c766 --- /dev/null +++ b/migrations/versions/d7e8f9a0b1c2_unique_token_type_cashu_transactions.py @@ -0,0 +1,45 @@ +"""Add unique token/type index to Cashu transactions. + +Revision ID: d7e8f9a0b1c2 +Revises: c6d7e8f9a0b1 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision = "d7e8f9a0b1c2" +down_revision = "c6d7e8f9a0b1" +branch_labels = None +depends_on = None + +_INDEX_NAME = "uq_cashu_transactions_token_type" + + +def upgrade() -> None: + connection = op.get_bind() + connection.execute( + sa.text( + "DELETE FROM cashu_transactions " + "WHERE id IN (" + " SELECT id FROM (" + " SELECT id, ROW_NUMBER() OVER (" + " PARTITION BY token, type " + " ORDER BY created_at ASC, id ASC" + " ) AS row_number " + " FROM cashu_transactions" + " ) WHERE row_number > 1" + ")" + ) + ) + op.create_index( + _INDEX_NAME, + "cashu_transactions", + ["token", "type"], + unique=True, + ) + + +def downgrade() -> None: + op.drop_index(_INDEX_NAME, table_name="cashu_transactions") diff --git a/routstr/core/db.py b/routstr/core/db.py index 586f467e..ba7a8e44 100644 --- a/routstr/core/db.py +++ b/routstr/core/db.py @@ -1,3 +1,4 @@ +import asyncio import os import pathlib import sqlite3 @@ -10,7 +11,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 @@ -105,8 +106,7 @@ async def reset_all_reserved_balances(session: AsyncSession) -> None: async def release_stale_reservations( session: AsyncSession, max_age_seconds: int ) -> int: - """Release reservations whose last reserve is older than max_age_seconds. - """ + """Release reservations whose last reserve is older than max_age_seconds.""" cutoff = int(time.time()) - max_age_seconds stmt = ( update(ApiKey) @@ -154,9 +154,7 @@ async def prune_dead_api_keys(session: AsyncSession, min_age_seconds: int) -> in .where(col(ApiKey.total_spent) == 0) .where(col(ApiKey.total_requests) == 0) .where(col(ApiKey.parent_key_hash).is_(None)) - .where( - (col(ApiKey.created_at).is_(None)) | (col(ApiKey.created_at) < cutoff) - ) + .where((col(ApiKey.created_at).is_(None)) | (col(ApiKey.created_at) < cutoff)) .where(~pending_invoice) .where(~has_children) ) @@ -246,6 +244,9 @@ class LightningInvoice(SQLModel, table=True): # type: ignore class CashuTransaction(SQLModel, table=True): # type: ignore __tablename__ = "cashu_transactions" + __table_args__ = ( + UniqueConstraint("token", "type", name="uq_cashu_transactions_token_type"), + ) id: str = Field( primary_key=True, @@ -276,6 +277,23 @@ class CashuTransaction(SQLModel, table=True): # type: ignore ) +async def _insert_cashu_transaction(transaction: CashuTransaction) -> None: + async with create_session() as session: + session.add(transaction) + await session.commit() + + +async def _cashu_transaction_exists(token: str, typ: str) -> bool: + async with create_session() as session: + result = await session.exec( + select(CashuTransaction).where( + CashuTransaction.token == token, + CashuTransaction.type == typ, + ) + ) + return result.first() is not None + + async def store_cashu_transaction( token: str, amount: int, @@ -287,28 +305,62 @@ async def store_cashu_transaction( created_at: int | None = None, source: str = "x-cashu", api_key_hashed_key: str | None = None, -) -> None: - 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, + max_attempts: int = 3, +) -> bool: + """Store a Cashu transaction, retrying transient database failures.""" + transaction = 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, + ) + + for attempt in range(1, max_attempts + 1): + try: + await _insert_cashu_transaction(transaction) + return True + except IntegrityError: + if await _cashu_transaction_exists(token, typ): + return True + break + except OperationalError as error: + if attempt == max_attempts: + break + delay = 0.25 * (2 ** (attempt - 1)) + logger.warning( + "Transient database failure storing Cashu transaction; retrying", + extra={ + "error": str(error), + "type": typ, + "request_id": request_id, + "attempt": attempt, + "max_attempts": max_attempts, + "retry_delay_seconds": delay, + }, ) - session.add(tx) - await session.commit() - except Exception as e: - logger.warning( - f"Failed to store cashu transaction: {e} (type={typ})", - extra={"error": str(e), "type": typ}, - ) + await asyncio.sleep(delay) + except Exception: + break + + logger.critical( + "Cashu transaction could not be stored after bounded retries", + extra={ + "type": typ, + "request_id": request_id, + "amount": amount, + "unit": unit, + "mint_url": mint_url, + "token": token, + "max_attempts": max_attempts, + }, + ) + return False class UpstreamProviderRow(SQLModel, table=True): # type: ignore @@ -358,9 +410,7 @@ class CliToken(SQLModel, table=True): # type: ignore """Long-lived authorization token for CLI/agent use against admin endpoints.""" __tablename__ = "cli_tokens" - id: str = Field( - primary_key=True, default_factory=lambda: uuid.uuid4().hex - ) + id: str = Field(primary_key=True, default_factory=lambda: uuid.uuid4().hex) token: str = Field(unique=True, index=True, description="Bearer token value") name: str = Field(description="Human-readable label for this token") created_at: int = Field(default_factory=lambda: int(time.time())) diff --git a/tests/unit/test_cashu_transaction_storage.py b/tests/unit/test_cashu_transaction_storage.py new file mode 100644 index 00000000..5e7be756 --- /dev/null +++ b/tests/unit/test_cashu_transaction_storage.py @@ -0,0 +1,73 @@ +from unittest.mock import AsyncMock + +import pytest +from sqlalchemy.exc import IntegrityError, OperationalError + +from routstr.core import db + + +@pytest.mark.asyncio +async def test_store_cashu_transaction_retries_transient_database_failures( + monkeypatch: pytest.MonkeyPatch, +) -> None: + insert = AsyncMock( + side_effect=[ + OperationalError("insert", {}, Exception("database is locked")), + None, + ] + ) + sleep = AsyncMock() + monkeypatch.setattr(db, "_insert_cashu_transaction", insert) + monkeypatch.setattr(db.asyncio, "sleep", sleep) + + stored = await db.store_cashu_transaction( + token="cashuAretry", amount=100, unit="sat", typ="out" + ) + + assert stored is True + assert insert.await_count == 2 + sleep.assert_awaited_once_with(0.25) + + +@pytest.mark.asyncio +async def test_store_cashu_transaction_stops_after_bounded_retries( + monkeypatch: pytest.MonkeyPatch, +) -> None: + insert = AsyncMock( + side_effect=OperationalError("insert", {}, Exception("database is locked")) + ) + sleep = AsyncMock() + critical = AsyncMock() + monkeypatch.setattr(db, "_insert_cashu_transaction", insert) + monkeypatch.setattr(db.asyncio, "sleep", sleep) + monkeypatch.setattr(db.logger, "critical", critical) + + stored = await db.store_cashu_transaction( + token="cashuAfailed", amount=100, unit="sat", typ="out" + ) + + assert stored is False + assert insert.await_count == 3 + assert sleep.await_count == 2 + critical.assert_called_once() + assert critical.call_args.kwargs["extra"]["token"] == "cashuAfailed" + + +@pytest.mark.asyncio +async def test_store_cashu_transaction_treats_duplicate_as_success( + monkeypatch: pytest.MonkeyPatch, +) -> None: + insert = AsyncMock( + side_effect=IntegrityError("insert", {}, Exception("unique constraint")) + ) + exists = AsyncMock(return_value=True) + monkeypatch.setattr(db, "_insert_cashu_transaction", insert) + monkeypatch.setattr(db, "_cashu_transaction_exists", exists) + + stored = await db.store_cashu_transaction( + token="cashuAduplicate", amount=100, unit="sat", typ="out" + ) + + assert stored is True + insert.assert_awaited_once() + exists.assert_awaited_once_with("cashuAduplicate", "out") From 07c15de106e7868f87bd8e074190b39baa20c595 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sun, 12 Jul 2026 14:10:44 +0200 Subject: [PATCH 2/2] better retry logic --- ...c2_unique_token_type_cashu_transactions.py | 89 +++++++++++--- routstr/core/db.py | 36 ++++-- routstr/wallet.py | 12 +- tests/unit/test_cashu_transaction_storage.py | 112 +++++++++++++++++- ..._cashu_transaction_uniqueness_migration.py | 110 +++++++++++++++++ tests/unit/test_wallet.py | 72 ++++++++--- 6 files changed, 387 insertions(+), 44 deletions(-) create mode 100644 tests/unit/test_cashu_transaction_uniqueness_migration.py diff --git a/migrations/versions/d7e8f9a0b1c2_unique_token_type_cashu_transactions.py b/migrations/versions/d7e8f9a0b1c2_unique_token_type_cashu_transactions.py index 2241c766..b6bf42b8 100644 --- a/migrations/versions/d7e8f9a0b1c2_unique_token_type_cashu_transactions.py +++ b/migrations/versions/d7e8f9a0b1c2_unique_token_type_cashu_transactions.py @@ -6,8 +6,11 @@ Revises: c6d7e8f9a0b1 from __future__ import annotations +from itertools import groupby + import sqlalchemy as sa from alembic import op +from sqlalchemy.engine import Connection, RowMapping revision = "d7e8f9a0b1c2" down_revision = "c6d7e8f9a0b1" @@ -15,24 +18,82 @@ branch_labels = None depends_on = None _INDEX_NAME = "uq_cashu_transactions_token_type" +_NULLABLE_METADATA = ("request_id", "mint_url", "api_key_hashed_key") + + +def _merge_duplicate_transactions(connection: Connection) -> None: + transactions = sa.Table( + "cashu_transactions", + sa.MetaData(), + autoload_with=connection, + ) + duplicate_keys = ( + sa.select(transactions.c.token, transactions.c.type) + .group_by(transactions.c.token, transactions.c.type) + .having(sa.func.count() > 1) + .subquery() + ) + rows = ( + connection.execute( + sa.select(transactions) + .join( + duplicate_keys, + sa.and_( + transactions.c.token == duplicate_keys.c.token, + transactions.c.type == duplicate_keys.c.type, + ), + ) + .order_by( + transactions.c.token, + transactions.c.type, + transactions.c.created_at, + transactions.c.id, + ) + ) + .mappings() + .all() + ) + + def transaction_key(row: RowMapping) -> tuple[str, str]: + return row["token"], row["type"] + + for _, grouped_rows in groupby(rows, key=transaction_key): + duplicates = list(grouped_rows) + if len(duplicates) < 2: + continue + + keeper = duplicates[0] + updates: dict[str, object] = { + "collected": any(row["collected"] for row in duplicates), + "swept": any(row["swept"] for row in duplicates), + } + for column in _NULLABLE_METADATA: + if not keeper[column]: + updates[column] = next( + (row[column] for row in duplicates if row[column]), + keeper[column], + ) + if not keeper["source"]: + updates["source"] = next( + (row["source"] for row in duplicates if row["source"]), + keeper["source"], + ) + + connection.execute( + transactions.update() + .where(transactions.c.id == keeper["id"]) + .values(**updates) + ) + connection.execute( + transactions.delete().where( + transactions.c.id.in_(row["id"] for row in duplicates[1:]) + ) + ) def upgrade() -> None: connection = op.get_bind() - connection.execute( - sa.text( - "DELETE FROM cashu_transactions " - "WHERE id IN (" - " SELECT id FROM (" - " SELECT id, ROW_NUMBER() OVER (" - " PARTITION BY token, type " - " ORDER BY created_at ASC, id ASC" - " ) AS row_number " - " FROM cashu_transactions" - " ) WHERE row_number > 1" - ")" - ) - ) + _merge_duplicate_transactions(connection) op.create_index( _INDEX_NAME, "cashu_transactions", diff --git a/routstr/core/db.py b/routstr/core/db.py index ba7a8e44..d5352994 100644 --- a/routstr/core/db.py +++ b/routstr/core/db.py @@ -321,22 +321,41 @@ async def store_cashu_transaction( api_key_hashed_key=api_key_hashed_key, ) + last_error_type: str | None = None + attempts_performed = 0 for attempt in range(1, max_attempts + 1): + attempts_performed = attempt + retry_error: OperationalError | None = None try: await _insert_cashu_transaction(transaction) return True - except IntegrityError: - if await _cashu_transaction_exists(token, typ): - return True - break + except IntegrityError as error: + last_error_type = type(error).__name__ + try: + if await _cashu_transaction_exists(token, typ): + return True + except OperationalError as lookup_error: + retry_error = lookup_error + except Exception as lookup_error: + last_error_type = type(lookup_error).__name__ + break + else: + break except OperationalError as error: + retry_error = error + except Exception as error: + last_error_type = type(error).__name__ + break + + if retry_error is not None: + last_error_type = type(retry_error.orig).__name__ if attempt == max_attempts: break delay = 0.25 * (2 ** (attempt - 1)) logger.warning( "Transient database failure storing Cashu transaction; retrying", extra={ - "error": str(error), + "error_type": last_error_type, "type": typ, "request_id": request_id, "attempt": attempt, @@ -345,18 +364,17 @@ async def store_cashu_transaction( }, ) await asyncio.sleep(delay) - except Exception: - break logger.critical( - "Cashu transaction could not be stored after bounded retries", + "Cashu transaction could not be stored", extra={ + "error_type": last_error_type, "type": typ, "request_id": request_id, "amount": amount, "unit": unit, "mint_url": mint_url, - "token": token, + "attempts_performed": attempts_performed, "max_attempts": max_attempts, }, ) diff --git a/routstr/wallet.py b/routstr/wallet.py index f1902919..5b78c8ec 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -733,8 +733,9 @@ async def credit_balance( extra={"new_balance": key.balance}, ) + transaction_stored = False try: - await store_cashu_transaction( + transaction_stored = await store_cashu_transaction( token=cashu_token, amount=original_amount, unit=original_unit, @@ -747,8 +748,13 @@ async def credit_balance( pass logger.debug( - "Cashu token successfully redeemed and stored", - extra={"amount": amount, "unit": unit, "mint_url": mint_url}, + "Cashu token successfully redeemed", + extra={ + "amount": amount, + "unit": unit, + "mint_url": mint_url, + "transaction_stored": transaction_stored, + }, ) return amount except Exception as e: diff --git a/tests/unit/test_cashu_transaction_storage.py b/tests/unit/test_cashu_transaction_storage.py index 5e7be756..8cd3b53b 100644 --- a/tests/unit/test_cashu_transaction_storage.py +++ b/tests/unit/test_cashu_transaction_storage.py @@ -49,8 +49,20 @@ async def test_store_cashu_transaction_stops_after_bounded_retries( assert stored is False assert insert.await_count == 3 assert sleep.await_count == 2 - critical.assert_called_once() - assert critical.call_args.kwargs["extra"]["token"] == "cashuAfailed" + critical.assert_called_once_with( + "Cashu transaction could not be stored", + extra={ + "error_type": "Exception", + "type": "out", + "request_id": None, + "amount": 100, + "unit": "sat", + "mint_url": None, + "attempts_performed": 3, + "max_attempts": 3, + }, + ) + assert "cashuAfailed" not in repr(critical.call_args) @pytest.mark.asyncio @@ -71,3 +83,99 @@ async def test_store_cashu_transaction_treats_duplicate_as_success( assert stored is True insert.assert_awaited_once() exists.assert_awaited_once_with("cashuAduplicate", "out") + + +@pytest.mark.asyncio +async def test_store_cashu_transaction_retries_duplicate_lookup_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + insert = AsyncMock( + side_effect=IntegrityError("insert", {}, Exception("unique constraint")) + ) + exists = AsyncMock( + side_effect=[ + OperationalError("select", {}, Exception("database is locked")), + True, + ] + ) + sleep = AsyncMock() + warning = AsyncMock() + monkeypatch.setattr(db, "_insert_cashu_transaction", insert) + monkeypatch.setattr(db, "_cashu_transaction_exists", exists) + monkeypatch.setattr(db.asyncio, "sleep", sleep) + monkeypatch.setattr(db.logger, "warning", warning) + + stored = await db.store_cashu_transaction( + token="cashuAlookup-retry", amount=100, unit="sat", typ="out" + ) + + assert stored is True + assert insert.await_count == 2 + assert exists.await_count == 2 + sleep.assert_awaited_once_with(0.25) + assert "cashuAlookup-retry" not in repr(warning.call_args) + assert warning.call_args.kwargs["extra"]["error_type"] == "Exception" + + +@pytest.mark.asyncio +async def test_store_cashu_transaction_bounds_duplicate_lookup_failures( + monkeypatch: pytest.MonkeyPatch, +) -> None: + insert = AsyncMock( + side_effect=IntegrityError("insert", {}, Exception("unique constraint")) + ) + exists = AsyncMock( + side_effect=OperationalError("select", {}, Exception("database is locked")) + ) + sleep = AsyncMock() + critical = AsyncMock() + monkeypatch.setattr(db, "_insert_cashu_transaction", insert) + monkeypatch.setattr(db, "_cashu_transaction_exists", exists) + monkeypatch.setattr(db.asyncio, "sleep", sleep) + monkeypatch.setattr(db.logger, "critical", critical) + + stored = await db.store_cashu_transaction( + token="cashuAlookup-failed", amount=100, unit="sat", typ="out" + ) + + assert stored is False + assert insert.await_count == 3 + assert exists.await_count == 3 + assert sleep.await_count == 2 + critical.assert_called_once() + assert "cashuAlookup-failed" not in repr(critical.call_args) + + +@pytest.mark.asyncio +async def test_store_cashu_transaction_contains_non_transient_lookup_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + insert = AsyncMock( + side_effect=IntegrityError("insert", {}, Exception("unique constraint")) + ) + exists = AsyncMock(side_effect=RuntimeError("lookup failed")) + critical = AsyncMock() + monkeypatch.setattr(db, "_insert_cashu_transaction", insert) + monkeypatch.setattr(db, "_cashu_transaction_exists", exists) + monkeypatch.setattr(db.logger, "critical", critical) + + stored = await db.store_cashu_transaction( + token="cashuAlookup-error", amount=100, unit="sat", typ="out" + ) + + assert stored is False + insert.assert_awaited_once() + exists.assert_awaited_once() + critical.assert_called_once_with( + "Cashu transaction could not be stored", + extra={ + "error_type": "RuntimeError", + "type": "out", + "request_id": None, + "amount": 100, + "unit": "sat", + "mint_url": None, + "attempts_performed": 1, + "max_attempts": 3, + }, + ) diff --git a/tests/unit/test_cashu_transaction_uniqueness_migration.py b/tests/unit/test_cashu_transaction_uniqueness_migration.py new file mode 100644 index 00000000..a2827ed2 --- /dev/null +++ b/tests/unit/test_cashu_transaction_uniqueness_migration.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +import importlib.util +from pathlib import Path + +import sqlalchemy as sa + +_MIGRATION_PATH = ( + Path(__file__).resolve().parents[2] + / "migrations" + / "versions" + / "d7e8f9a0b1c2_unique_token_type_cashu_transactions.py" +) +_spec = importlib.util.spec_from_file_location( + "cashu_transaction_uniqueness_migration", _MIGRATION_PATH +) +assert _spec is not None and _spec.loader is not None +migration = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(migration) + + +def test_duplicate_merge_preserves_state_and_missing_linkage() -> None: + engine = sa.create_engine("sqlite:///:memory:") + metadata = sa.MetaData() + transactions = sa.Table( + "cashu_transactions", + metadata, + sa.Column("id", sa.String, primary_key=True), + sa.Column("token", sa.String, nullable=False), + sa.Column("amount", sa.Integer, nullable=False), + sa.Column("unit", sa.String, nullable=False), + sa.Column("mint_url", sa.String), + sa.Column("type", sa.String, nullable=False), + sa.Column("request_id", sa.String), + sa.Column("created_at", sa.Integer, nullable=False), + sa.Column("collected", sa.Boolean, nullable=False), + sa.Column("swept", sa.Boolean, nullable=False), + sa.Column("source", sa.String, nullable=False), + sa.Column("api_key_hashed_key", sa.String), + ) + metadata.create_all(engine) + + with engine.begin() as connection: + connection.execute( + transactions.insert(), + [ + { + "id": "oldest", + "token": "cashuAduplicate", + "amount": 100, + "unit": "sat", + "mint_url": "", + "type": "out", + "request_id": "", + "created_at": 1, + "collected": False, + "swept": False, + "source": "", + "api_key_hashed_key": "", + }, + { + "id": "newer", + "token": "cashuAduplicate", + "amount": 100, + "unit": "sat", + "mint_url": "https://mint.example", + "type": "out", + "request_id": "request-newer", + "created_at": 2, + "collected": True, + "swept": False, + "source": "apikey", + "api_key_hashed_key": "hashed-key", + }, + { + "id": "newest", + "token": "cashuAduplicate", + "amount": 100, + "unit": "sat", + "mint_url": "https://other.example", + "type": "out", + "request_id": "request-newest", + "created_at": 3, + "collected": False, + "swept": True, + "source": "x-cashu", + "api_key_hashed_key": "other-key", + }, + ], + ) + + migration._merge_duplicate_transactions(connection) + rows = connection.execute(sa.select(transactions)).mappings().all() + + assert rows == [ + { + "id": "oldest", + "token": "cashuAduplicate", + "amount": 100, + "unit": "sat", + "mint_url": "https://mint.example", + "type": "out", + "request_id": "request-newer", + "created_at": 1, + "collected": True, + "swept": True, + "source": "apikey", + "api_key_hashed_key": "hashed-key", + } + ] diff --git a/tests/unit/test_wallet.py b/tests/unit/test_wallet.py index 3bb36a28..9f6bf51f 100644 --- a/tests/unit/test_wallet.py +++ b/tests/unit/test_wallet.py @@ -1060,31 +1060,71 @@ async def test_credit_balance_msat_unit_not_converted() -> None: @pytest.mark.asyncio -async def test_credit_balance_survives_audit_store_failure() -> None: - """A failure writing the CashuTransaction history record must not undo the - already-committed balance credit. (The silent swallow is a known - audit-trail gap slated for its own fix — this test pins the financial - invariant that the user keeps their credit, not the swallow itself.)""" - mock_key = Mock() - mock_key.balance = 0 - mock_key.hashed_key = "test_hash" +async def test_credit_balance_reports_audit_store_false() -> None: + mock_key = Mock(balance=0, hashed_key="test_hash") mock_session = AsyncMock() + debug = Mock() from routstr.core.settings import settings - with patch.object(settings, "cashu_mints", ["http://mint:3338"]): - with patch( + with ( + patch.object(settings, "cashu_mints", ["http://mint:3338"]), + patch( "routstr.wallet.recieve_token", return_value=(1000, "sat", "http://mint:3338"), - ): - with patch( - "routstr.wallet.store_cashu_transaction", - side_effect=Exception("history table locked"), - ): - amount = await credit_balance("cashuAtest", mock_key, mock_session) + ), + patch("routstr.wallet.store_cashu_transaction", return_value=False), + patch("routstr.wallet.logger.debug", debug), + ): + amount = await credit_balance("cashuAtest", mock_key, mock_session) assert amount == 1_000_000 assert mock_session.commit.called + debug.assert_called_once_with( + "Cashu token successfully redeemed", + extra={ + "amount": 1_000_000, + "unit": "sat", + "mint_url": "http://mint:3338", + "transaction_stored": False, + }, + ) + + +@pytest.mark.asyncio +async def test_credit_balance_survives_audit_store_failure() -> None: + """A failed CashuTransaction history write must not undo balance credit.""" + mock_key = Mock(balance=0, hashed_key="test_hash") + mock_session = AsyncMock() + debug = Mock() + + from routstr.core.settings import settings + + with ( + patch.object(settings, "cashu_mints", ["http://mint:3338"]), + patch( + "routstr.wallet.recieve_token", + return_value=(1000, "sat", "http://mint:3338"), + ), + patch( + "routstr.wallet.store_cashu_transaction", + side_effect=Exception("history table locked"), + ), + patch("routstr.wallet.logger.debug", debug), + ): + amount = await credit_balance("cashuAtest", mock_key, mock_session) + + assert amount == 1_000_000 + assert mock_session.commit.called + debug.assert_called_once_with( + "Cashu token successfully redeemed", + extra={ + "amount": 1_000_000, + "unit": "sat", + "mint_url": "http://mint:3338", + "transaction_stored": False, + }, + ) @pytest.mark.asyncio