Revert "Retry Cashu transaction storage on transient DB failures"

migrations/versions/d7e8f9a0b1c2_unique_token_type_cashu_transactions.py

@@ -1,106 +0,0 @@
-"""Add unique token/type index to Cashu transactions.
-
-Revision ID: d7e8f9a0b1c2
-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"
-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()
-    _merge_duplicate_transactions(connection)
-    op.create_index(
-        _INDEX_NAME,
-        "cashu_transactions",
-        ["token", "type"],
-        unique=True,
-    )
-
-
-def downgrade() -> None:
-    op.drop_index(_INDEX_NAME, table_name="cashu_transactions")

routstr/core/db.py

@@ -1,4 +1,3 @@
-import asyncio
 import os
 import pathlib
 import sqlite3
@@ -11,7 +10,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 IntegrityError, OperationalError
+from sqlalchemy.exc import 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
@@ -106,7 +105,8 @@ 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,7 +154,9 @@ 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)
     )
@@ -244,9 +246,6 @@ 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,
@@ -277,23 +276,6 @@ 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,
@@ -305,80 +287,28 @@ async def store_cashu_transaction(
     created_at: int | None = None,
     source: str = "x-cashu",
     api_key_hashed_key: str | None = None,
-    max_attempts: int = 3,
+) -> None:
-) -> bool:
+    try:
-    """Store a Cashu transaction, retrying transient database failures."""
+        async with create_session() as session:
-    transaction = CashuTransaction(
+            tx = CashuTransaction(
-        token=token,
+                token=token,
-        amount=amount,
+                amount=amount,
-        unit=unit,
+                unit=unit,
-        mint_url=mint_url,
+                mint_url=mint_url,
-        type=typ,
+                type=typ,
-        request_id=request_id,
+                request_id=request_id,
-        collected=collected,
+                collected=collected,
-        created_at=created_at or int(time.time()),
+                created_at=created_at or int(time.time()),
-        source=source,
+                source=source,
-        api_key_hashed_key=api_key_hashed_key,
+                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 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_type": last_error_type,
-                    "type": typ,
-                    "request_id": request_id,
-                    "attempt": attempt,
-                    "max_attempts": max_attempts,
-                    "retry_delay_seconds": delay,
-                },
             )
-            await asyncio.sleep(delay)
+            session.add(tx)
-
+            await session.commit()
-    logger.critical(
+    except Exception as e:
-        "Cashu transaction could not be stored",
+        logger.warning(
-        extra={
+            f"Failed to store cashu transaction: {e} (type={typ})",
-            "error_type": last_error_type,
+            extra={"error": str(e), "type": typ},
-            "type": typ,
+        )
-            "request_id": request_id,
-            "amount": amount,
-            "unit": unit,
-            "mint_url": mint_url,
-            "attempts_performed": attempts_performed,
-            "max_attempts": max_attempts,
-        },
-    )
-    return False
 
 
 class UpstreamProviderRow(SQLModel, table=True):  # type: ignore
@@ -428,7 +358,9 @@ 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()))

routstr/wallet.py

@@ -733,9 +733,8 @@ async def credit_balance(
             extra={"new_balance": key.balance},
         )
 
-        transaction_stored = False
         try:
-            transaction_stored = await store_cashu_transaction(
+            await store_cashu_transaction(
                 token=cashu_token,
                 amount=original_amount,
                 unit=original_unit,
@@ -748,13 +747,8 @@ async def credit_balance(
             pass
 
         logger.debug(
-            "Cashu token successfully redeemed",
+            "Cashu token successfully redeemed and stored",
-            extra={
+            extra={"amount": amount, "unit": unit, "mint_url": mint_url},
-                "amount": amount,
-                "unit": unit,
-                "mint_url": mint_url,
-                "transaction_stored": transaction_stored,
-            },
         )
         return amount
     except Exception as e:

tests/unit/test_cashu_transaction_storage.py

@@ -1,181 +0,0 @@
-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_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
-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")
-
-
-@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,
-        },
-    )

tests/unit/test_cashu_transaction_uniqueness_migration.py

@@ -1,110 +0,0 @@
-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",
-        }
-    ]

tests/unit/test_wallet.py

@@ -1059,72 +1059,32 @@ async def test_credit_balance_msat_unit_not_converted() -> None:
     assert mock_session.commit.called
 
 
-@pytest.mark.asyncio
-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"]),
-        patch(
-            "routstr.wallet.recieve_token",
-            return_value=(1000, "sat", "http://mint:3338"),
-        ),
-        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."""
+    """A failure writing the CashuTransaction history record must not undo the
-    mock_key = Mock(balance=0, hashed_key="test_hash")
+    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"
     mock_session = AsyncMock()
-    debug = Mock()
 
     from routstr.core.settings import settings
 
-    with (
+    with patch.object(settings, "cashu_mints", ["http://mint:3338"]):
-        patch.object(settings, "cashu_mints", ["http://mint:3338"]),
+        with patch(
-        patch(
             "routstr.wallet.recieve_token",
             return_value=(1000, "sat", "http://mint:3338"),
-        ),
+        ):
-        patch(
+            with patch(
-            "routstr.wallet.store_cashu_transaction",
+                "routstr.wallet.store_cashu_transaction",
-            side_effect=Exception("history table locked"),
+                side_effect=Exception("history table locked"),
-        ),
+            ):
-        patch("routstr.wallet.logger.debug", debug),
+                amount = await credit_balance("cashuAtest", mock_key, mock_session)
-    ):
-        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