From aea24d23dae15ef5ee28ca697f4ce9a4d6e38fac Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sun, 12 Jul 2026 13:29:09 +0200 Subject: [PATCH] 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")