From 057a752b1e908b8b88a734cc6edd7d9d820e954b Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sun, 12 Jul 2026 13:07:05 +0200 Subject: [PATCH 1/3] fix cost breakdown calculation --- routstr/payment/cost_calculation.py | 127 ++++++++++++++------ tests/unit/test_cost_calculation_caching.py | 104 ++++++++++++++++ 2 files changed, 191 insertions(+), 40 deletions(-) diff --git a/routstr/payment/cost_calculation.py b/routstr/payment/cost_calculation.py index 84f3847f..25e94993 100644 --- a/routstr/payment/cost_calculation.py +++ b/routstr/payment/cost_calculation.py @@ -157,11 +157,16 @@ async def calculate_cost( }, ) try: + cost_details = usage_data.get("cost_details", {}) + if not isinstance(cost_details, dict): + cost_details = {} input_usd = _coerce_usd( - usage_data.get("cost_details", {}).get("input_cost", 0) + cost_details.get("input_cost") + or cost_details.get("upstream_inference_prompt_cost") ) output_usd = _coerce_usd( - usage_data.get("cost_details", {}).get("output_cost", 0) + cost_details.get("output_cost") + or cost_details.get("upstream_inference_completions_cost") ) return _calculate_from_usd_cost( usd_cost, @@ -281,54 +286,92 @@ def _resolve_usd_cost(usage_data: dict, response_data: dict) -> float: def _get_pricing_rates( response_data: dict, ) -> tuple[float, float, float, float] | None: - """Get model-based pricing rates or None if using fixed pricing. + """Get configured rates, falling back to LiteLLM's model cost map. - Returns: (input_rate, output_rate, cache_read_rate, cache_write_rate) + Returns: (input_rate, output_rate, cache_read_rate, cache_write_rate). + ``None`` means configured fixed pricing should be used by the caller. """ - if settings.fixed_pricing: + if settings.fixed_pricing and ( + settings.fixed_per_1k_input_tokens + or settings.fixed_per_1k_output_tokens + ): return None from ..proxy import get_model_instance + from .models import litellm_cost_entry response_model = response_data.get("model", "") model_obj = get_model_instance(response_model) - if not model_obj: - logger.error("Invalid model in response", extra={"response_model": response_model}) - raise ValueError(f"Invalid model: {response_model}") + if model_obj and model_obj.sats_pricing: + try: + mspp = float(model_obj.sats_pricing.prompt) + mspc = float(model_obj.sats_pricing.completion) + mscr = float(model_obj.sats_pricing.input_cache_read or 0) + mscw = float(model_obj.sats_pricing.input_cache_write or 0) - if not model_obj.sats_pricing: - logger.error( - "Model pricing not defined", - extra={"model": response_model, "model_id": response_model}, + mspp_1k = mspp * 1_000_000.0 + mspc_1k = mspc * 1_000_000.0 + mscr_1k = mscr * 1_000_000.0 if mscr > 0 else mspp_1k + mscw_1k = mscw * 1_000_000.0 if mscw > 0 else mspp_1k + source = "configured" + except Exception as e: + logger.error("Invalid pricing data", extra={"error": str(e)}) + raise ValueError("Invalid pricing data") from e + else: + pricing_model = ( + model_obj.forwarded_model_id if model_obj else None + ) or response_model + pricing = litellm_cost_entry(pricing_model) + if pricing is None: + logger.error( + "Model pricing not found in configured models or LiteLLM", + extra={ + "response_model": response_model, + "pricing_model": pricing_model, + }, + ) + raise ValueError(f"Pricing not found for model: {response_model}") + + input_usd = _coerce_usd(pricing.get("input_cost_per_token")) + output_usd = _coerce_usd(pricing.get("output_cost_per_token")) + if input_usd <= 0 or output_usd <= 0: + raise ValueError(f"Incomplete LiteLLM pricing for model: {pricing_model}") + + provider_fee = _resolve_provider_fee(response_model) + usd_per_sat = sats_usd_price() + mspp_1k = input_usd * provider_fee * 1_000_000.0 / usd_per_sat + mspc_1k = output_usd * provider_fee * 1_000_000.0 / usd_per_sat + cache_read_usd = _coerce_usd( + pricing.get("cache_read_input_token_cost") ) - raise ValueError("Model pricing not defined") - - try: - mspp = float(model_obj.sats_pricing.prompt) - mspc = float(model_obj.sats_pricing.completion) - mscr = float(model_obj.sats_pricing.input_cache_read or 0) - mscw = float(model_obj.sats_pricing.input_cache_write or 0) - - mspp_1k = mspp * 1_000_000.0 - mspc_1k = mspc * 1_000_000.0 - mscr_1k = mscr * 1_000_000.0 if mscr > 0 else mspp_1k - mscw_1k = mscw * 1_000_000.0 if mscw > 0 else mspp_1k - - logger.info( - "Applied model-specific pricing", - extra={ - "model": response_model, - "input_price_msats_per_1k": mspp_1k, - "output_price_msats_per_1k": mspc_1k, - "cache_read_price_msats_per_1k": mscr_1k, - "cache_write_price_msats_per_1k": mscw_1k, - }, + cache_write_usd = _coerce_usd( + pricing.get("cache_creation_input_token_cost") ) - return mspp_1k, mspc_1k, mscr_1k, mscw_1k - except Exception as e: - logger.error("Invalid pricing data", extra={"error": str(e)}) - raise ValueError("Invalid pricing data") from e + mscr_1k = ( + cache_read_usd * provider_fee * 1_000_000.0 / usd_per_sat + if cache_read_usd > 0 + else mspp_1k + ) + mscw_1k = ( + cache_write_usd * provider_fee * 1_000_000.0 / usd_per_sat + if cache_write_usd > 0 + else mspp_1k + ) + source = "litellm" + + logger.info( + "Applied model-specific pricing", + extra={ + "model": response_model, + "pricing_source": source, + "input_price_msats_per_1k": mspp_1k, + "output_price_msats_per_1k": mspc_1k, + "cache_read_price_msats_per_1k": mscr_1k, + "cache_write_price_msats_per_1k": mscw_1k, + }, + ) + return mspp_1k, mspc_1k, mscr_1k, mscw_1k def _resolve_provider_fee(model_id: str) -> float: @@ -367,8 +410,12 @@ def _calculate_from_usd_cost( cost_in_msats = math.ceil(cost_in_sats * 1000) if input_usd > 0 or output_usd > 0: - input_msats = int((input_usd * sats_per_usd) * 1000) - output_msats = int((output_usd * sats_per_usd) * 1000) + # The total is the authoritative billed amount. Allocating that integer + # total proportionally avoids losing sub-millisatoshi remainders when + # input and output components are each truncated independently. + component_usd = input_usd + output_usd + input_msats = math.floor(cost_in_msats * input_usd / component_usd) + output_msats = cost_in_msats - input_msats else: effective_input_tokens = ( input_tokens + cache_read_tokens + cache_creation_tokens diff --git a/tests/unit/test_cost_calculation_caching.py b/tests/unit/test_cost_calculation_caching.py index 93c68877..6ff7170b 100644 --- a/tests/unit/test_cost_calculation_caching.py +++ b/tests/unit/test_cost_calculation_caching.py @@ -465,9 +465,113 @@ async def test_cache_read_only_usd_cost_response_is_billed( assert result.cache_read_input_tokens == 1000 +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("total_cost", "input_cost", "output_cost", "expected_msats"), + [ + (0.000471, 0.00023451, 0.00023649, 9420), + (0.00000004, 0.00000002, 0.00000002, 1), + ], +) +async def test_small_usd_cost_components_sum_to_rounded_total( + total_cost: float, + input_cost: float, + output_cost: float, + expected_msats: int, +) -> None: + """Small USD component costs must retain every billed millisatoshi.""" + response = { + "model": "gpt-4", + "usage": { + "prompt_tokens": 1, + "completion_tokens": 1, + "cost_details": { + "total_cost": total_cost, + "input_cost": input_cost, + "output_cost": output_cost, + }, + }, + } + + result = await calculate_cost(response, max_cost=100000) + + assert isinstance(result, CostData) + assert result.total_msats == expected_msats + assert result.input_msats + result.output_msats == result.total_msats + + +@pytest.mark.asyncio +async def test_openrouter_upstream_inference_cost_components_are_used() -> None: + """OpenRouter component aliases must determine the input/output split.""" + response = { + "model": "gpt-4", + "usage": { + "prompt_tokens": 375, + "completion_tokens": 158, + "total_tokens": 533, + "cost": 0.00022354, + "is_byok": False, + "prompt_tokens_details": { + "cached_tokens": 286, + "cache_write_tokens": 0, + }, + "cost_details": { + "upstream_inference_cost": 0.00022354, + "upstream_inference_prompt_cost": 0.00004974, + "upstream_inference_completions_cost": 0.0001738, + }, + "completion_tokens_details": {"reasoning_tokens": 17}, + }, + } + + result = await calculate_cost(response, max_cost=100000) + + assert isinstance(result, CostData) + assert result.input_msats == 994 + assert result.output_msats == 3477 + assert result.input_msats + result.output_msats == result.total_msats == 4471 + + # ============================================================================ # Test 13: Missing Usage Block # ============================================================================ +@pytest.mark.asyncio +async def test_missing_upstream_cost_uses_litellm_model_pricing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Token usage without an upstream cost is priced from LiteLLM.""" + monkeypatch.setattr(settings, "fixed_pricing", True) + monkeypatch.setattr(settings, "fixed_per_1k_input_tokens", 0) + monkeypatch.setattr(settings, "fixed_per_1k_output_tokens", 0) + monkeypatch.setattr( + "routstr.payment.models.litellm_cost_entry", + lambda model: { + "input_cost_per_token": 0.000001, + "output_cost_per_token": 0.000002, + }, + ) + response = { + "model": "priced-by-litellm", + "usage": { + "prompt_tokens": 90, + "completion_tokens": 80, + "total_tokens": 170, + "prompt_tokens_details": {"cached_tokens": 0}, + "completion_tokens_details": {"reasoning_tokens": 74}, + "prompt_cache_hit_tokens": 0, + "prompt_cache_miss_tokens": 90, + }, + } + + result = await calculate_cost(response, max_cost=10000) + + assert isinstance(result, CostData) + assert not isinstance(result, MaxCostData) + assert result.input_msats == 1800 + assert result.output_msats == 3200 + assert result.input_msats + result.output_msats == result.total_msats == 5000 + + @pytest.mark.asyncio async def test_missing_usage_block(mock_fixed_pricing: None) -> None: """When usage is missing, return MaxCostData with zero tokens.""" From aea24d23dae15ef5ee28ca697f4ce9a4d6e38fac Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sun, 12 Jul 2026 13:29:09 +0200 Subject: [PATCH 2/3] 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 3/3] 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