laantungir/routstr-core
1
0
Fork 0
mirror of https://github.com/Routstr/routstr-core.git synced 2026-10-05 20:28:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: persist outbound Cashu tokens before sending

Browse Source
This commit is contained in:
9qeklajc
2026-07-13 00:07:00 +02:00
parent 110ec5da5d f31929b538
commit 9460e24f21
11 changed files with 437 additions and 586 deletions
Download Patch File Download Diff File Expand all files Collapse all files
migrations/versions/d7e8f9a0b1c2_unique_token_type_cashu_transactions.py
-106
View File
@@ -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/admin.py
+14 -5
View File
@@ -28,6 +28,7 @@ from .db import (
ModelRow,
UpstreamProviderRow,
create_session,
store_cashu_transaction,
)
from .log_manager import log_manager
from .logging import get_logger
@@ -411,12 +412,11 @@ async def withdraw(
# Get wallet and check balance
from .settings import settings as global_settings
wallet = await get_wallet(
withdraw_request.mint_url or global_settings.primary_mint, withdraw_request.unit
)
effective_mint = withdraw_request.mint_url or global_settings.primary_mint
wallet = await get_wallet(effective_mint, withdraw_request.unit)
proofs = get_proofs_per_mint_and_unit(
wallet,
withdraw_request.mint_url or global_settings.primary_mint,
effective_mint,
withdraw_request.unit,
not_reserved=True,
)
@@ -432,7 +432,16 @@ async def withdraw(
raise HTTPException(status_code=400, detail="Insufficient wallet balance")
token = await send_token(
withdraw_request.amount, withdraw_request.unit, withdraw_request.mint_url
withdraw_request.amount, withdraw_request.unit, effective_mint
)
await store_cashu_transaction(
token=token,
amount=withdraw_request.amount,
unit=withdraw_request.unit,
mint_url=effective_mint,
typ="out",
collected=False,
source="admin",
)
return {"token": token}
routstr/core/db.py
+31 -97
View File
@@ -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,30 @@ 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,
) -> 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,
)
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,
},
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,
)
await asyncio.sleep(delay)
logger.critical(
"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,
"attempts_performed": attempts_performed,
"max_attempts": max_attempts,
},
)
return False
session.add(tx)
await session.commit()
return True
except Exception as e:
logger.warning(
f"Failed to store cashu transaction: {e} (type={typ})",
extra={"error": str(e), "type": typ},
)
return False
class UpstreamProviderRow(SQLModel, table=True): # type: ignore
@@ -428,7 +360,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/upstream/auto_topup.py
+42 -2
View File
@@ -4,7 +4,12 @@ import json
from sqlmodel import select
from ..core import get_logger
from ..core.db import UpstreamProviderRow, create_session
from ..core.db import (
CashuTransaction,
UpstreamProviderRow,
create_session,
store_cashu_transaction,
)
from ..wallet import send_token
from .routstr import RoutstrUpstreamProvider
@@ -123,7 +128,6 @@ async def _check_and_topup(row: UpstreamProviderRow) -> None:
},
)
print(amount, mint_url)
try:
token = await send_token(amount, "sat", mint_url)
except Exception as e:
@@ -138,6 +142,22 @@ async def _check_and_topup(row: UpstreamProviderRow) -> None:
)
return
stored = await store_cashu_transaction(
token=token,
amount=amount,
unit="sat",
mint_url=mint_url,
typ="out",
collected=False,
source="auto_topup",
)
if not stored:
logger.critical(
"Aborting auto top-up because its cashu token could not be persisted",
extra={"provider_id": row.id, "mint_url": mint_url},
)
return
result = await provider.topup(token)
if "error" in result:
@@ -149,6 +169,26 @@ async def _check_and_topup(row: UpstreamProviderRow) -> None:
},
)
else:
async with create_session() as session:
transaction = (
await session.exec(
select(CashuTransaction).where(
CashuTransaction.token == token,
CashuTransaction.type == "out",
CashuTransaction.source == "auto_topup",
)
)
).first()
if transaction is None:
logger.critical(
"Completed auto top-up transaction is missing from the database",
extra={"provider_id": row.id, "mint_url": mint_url},
)
else:
transaction.collected = True
session.add(transaction)
await session.commit()
logger.info(
"Auto top-up completed successfully",
extra={
routstr/wallet.py
+15 -30
View File
@@ -237,9 +237,7 @@ async def send(amount: int, unit: str, mint_url: str | None = None) -> tuple[int
all_mint_urls = list({k.mint_url for k in wallet.keysets.values()})
proof_summary = {
f"{k.mint_url}/{k.unit.name}": sum(
p.amount for p in wallet.proofs if p.id == k.id
)
f"{k.mint_url}/{k.unit.name}": sum(p.amount for p in wallet.proofs if p.id == k.id)
for k in wallet.keysets.values()
}
# Show ALL proofs in DB by keyset_id, regardless of whether the loaded wallet
@@ -593,16 +591,11 @@ async def swap_to_primary_mint(
# advance the counter so the next request derives fresh secrets.
logger.warning(
"swap_to_primary_mint: outputs already signed — recovering orphaned proofs",
extra={
"mint_quote_id": mint_quote.quote,
"minted_amount": minted_amount,
},
extra={"mint_quote_id": mint_quote.quote, "minted_amount": minted_amount},
)
try:
for keyset_id in primary_wallet.keysets:
await primary_wallet.restore_tokens_for_keyset(
keyset_id, to=1, batch=25
)
await primary_wallet.restore_tokens_for_keyset(keyset_id, to=1, batch=25)
await primary_wallet.load_proofs(reload=True)
post_recovery_balance = primary_wallet.available_balance.amount
balance_gained = post_recovery_balance - pre_mint_balance
@@ -740,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,
@@ -755,13 +747,8 @@ async def credit_balance(
pass
logger.debug(
"Cashu token successfully redeemed",
extra={
"amount": amount,
"unit": unit,
"mint_url": mint_url,
"transaction_stored": transaction_stored,
},
"Cashu token successfully redeemed and stored",
extra={"amount": amount, "unit": unit, "mint_url": mint_url},
)
return amount
except Exception as e:
@@ -861,9 +848,7 @@ async def fetch_all_balances(
"unit": unit,
"wallet_balance": proofs_balance,
"user_balance": user_balance,
"owner_balance": proofs_balance - user_balance
if proofs_balance != 0
else 0,
"owner_balance": proofs_balance - user_balance if proofs_balance != 0 else 0,
}
return result
except Exception as e:
@@ -1003,9 +988,7 @@ async def periodic_payout() -> None:
)
async def refund_sweep_once() -> None:
"""Sweep eligible uncollected refund tokens once."""
cutoff = int(time.time()) - settings.refund_sweep_ttl_seconds
async def _refund_sweep_once(cutoff: int) -> None:
async with db.create_session() as session:
stmt = select(db.CashuTransaction).where(
db.CashuTransaction.type == "out",
@@ -1051,6 +1034,12 @@ async def refund_sweep_once() -> None:
await session.commit()
async def refund_sweep_once() -> None:
"""Sweep eligible uncollected refund tokens once."""
cutoff = int(time.time()) - settings.refund_sweep_ttl_seconds
await _refund_sweep_once(cutoff)
async def periodic_refund_sweep() -> None:
while True:
await asyncio.sleep(60 * 60) # every hour
@@ -1085,11 +1074,7 @@ async def periodic_routstr_fee_payout() -> None:
wallet, settings.primary_mint, "sat", not_reserved=True
)
amount_received = await raw_send_to_lnurl(
wallet,
proofs,
ROUTSTR_LN_ADDRESS,
"sat",
amount=accumulated_sats,
wallet, proofs, ROUTSTR_LN_ADDRESS, "sat", amount=accumulated_sats
)
paid_msats = accumulated_sats * 1000
await db.reset_routstr_fee(session, paid_msats)
tests/unit/test_admin_withdraw.py
+51
View File
@@ -0,0 +1,51 @@
from types import SimpleNamespace
from unittest.mock import AsyncMock, Mock
import pytest
from routstr.core import admin
@pytest.mark.asyncio
@pytest.mark.parametrize("requested_mint", [None, "https://secondary.example"])
async def test_withdraw_uses_effective_mint_and_records_outgoing_transaction(
monkeypatch: pytest.MonkeyPatch, requested_mint: str | None
) -> None:
primary_mint = "https://primary.example"
effective_mint = requested_mint or primary_mint
wallet = object()
proofs = [SimpleNamespace(amount=40), SimpleNamespace(amount=60)]
token = "cashuBoutgoing"
get_wallet = AsyncMock(return_value=wallet)
get_proofs = Mock(return_value=proofs)
filter_proofs = AsyncMock(return_value=proofs)
send_token = AsyncMock(return_value=token)
store_transaction = AsyncMock(return_value=True)
monkeypatch.setattr(admin, "get_wallet", get_wallet)
monkeypatch.setattr(admin, "get_proofs_per_mint_and_unit", get_proofs)
monkeypatch.setattr(admin, "slow_filter_spend_proofs", filter_proofs)
monkeypatch.setattr(admin, "send_token", send_token)
monkeypatch.setattr(admin, "store_cashu_transaction", store_transaction)
monkeypatch.setattr(admin.settings, "primary_mint", primary_mint)
result = await admin.withdraw(
Mock(),
admin.WithdrawRequest(amount=75, mint_url=requested_mint, unit="sat"),
)
assert result == {"token": token}
get_wallet.assert_awaited_once_with(effective_mint, "sat")
get_proofs.assert_called_once_with(wallet, effective_mint, "sat", not_reserved=True)
filter_proofs.assert_awaited_once_with(proofs, wallet)
send_token.assert_awaited_once_with(75, "sat", effective_mint)
store_transaction.assert_awaited_once_with(
token=token,
amount=75,
unit="sat",
mint_url=effective_mint,
typ="out",
collected=False,
source="admin",
)
tests/unit/test_auto_topup.py
+143
View File
@@ -0,0 +1,143 @@
import json
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from routstr.core.db import CashuTransaction
from routstr.upstream.auto_topup import _check_and_topup
def _row() -> MagicMock:
row = MagicMock()
row.id = "provider-1"
row.base_url = "https://provider.test"
row.api_key = "secret"
row.provider_settings = json.dumps(
{
"auto_topup": True,
"topup_threshold": 100,
"topup_amount_limit": 50,
"topup_mint_url": "https://mint.test",
}
)
return row
class _Session:
def __init__(self, transaction: CashuTransaction) -> None:
self.transaction = transaction
self.commit = AsyncMock()
async def __aenter__(self) -> "_Session":
return self
async def __aexit__(self, *args: object) -> None:
return None
async def exec(self, query: object) -> MagicMock:
result = MagicMock()
result.first.return_value = self.transaction
return result
def add(self, transaction: CashuTransaction) -> None:
self.transaction = transaction
@pytest.mark.asyncio
async def test_auto_topup_persists_before_sending_and_marks_success_collected() -> None:
provider = MagicMock()
provider.get_balance = AsyncMock(return_value=0)
provider.topup = AsyncMock(return_value={"balance": 50})
transaction = CashuTransaction(
token="cashu-token", amount=50, unit="sat", source="auto_topup"
)
session = _Session(transaction)
with (
patch(
"routstr.upstream.auto_topup.RoutstrUpstreamProvider.from_db_row",
return_value=provider,
),
patch(
"routstr.upstream.auto_topup.send_token",
AsyncMock(return_value="cashu-token"),
),
patch(
"routstr.upstream.auto_topup.store_cashu_transaction",
AsyncMock(return_value=True),
) as store,
patch("routstr.upstream.auto_topup.create_session", return_value=session),
):
await _check_and_topup(_row())
store.assert_awaited_once_with(
token="cashu-token",
amount=50,
unit="sat",
mint_url="https://mint.test",
typ="out",
collected=False,
source="auto_topup",
)
provider.topup.assert_awaited_once_with("cashu-token")
assert transaction.collected is True
session.commit.assert_awaited_once()
@pytest.mark.asyncio
@pytest.mark.parametrize("outcome", [{"error": "rejected"}, RuntimeError("network")])
async def test_auto_topup_failure_leaves_persisted_token_uncollected(
outcome: object,
) -> None:
provider = MagicMock()
provider.get_balance = AsyncMock(return_value=0)
provider.topup = AsyncMock(
side_effect=outcome if isinstance(outcome, Exception) else None,
return_value=outcome,
)
with (
patch(
"routstr.upstream.auto_topup.RoutstrUpstreamProvider.from_db_row",
return_value=provider,
),
patch(
"routstr.upstream.auto_topup.send_token",
AsyncMock(return_value="cashu-token"),
),
patch(
"routstr.upstream.auto_topup.store_cashu_transaction",
AsyncMock(return_value=True),
),
patch("routstr.upstream.auto_topup.create_session") as create_session,
):
if isinstance(outcome, Exception):
with pytest.raises(RuntimeError):
await _check_and_topup(_row())
else:
await _check_and_topup(_row())
create_session.assert_not_called()
@pytest.mark.asyncio
async def test_auto_topup_does_not_send_untracked_token() -> None:
provider = MagicMock()
provider.get_balance = AsyncMock(return_value=0)
provider.topup = AsyncMock()
with (
patch(
"routstr.upstream.auto_topup.RoutstrUpstreamProvider.from_db_row",
return_value=provider,
),
patch(
"routstr.upstream.auto_topup.send_token",
AsyncMock(return_value="cashu-token"),
),
patch(
"routstr.upstream.auto_topup.store_cashu_transaction",
AsyncMock(return_value=False),
),
):
await _check_and_topup(_row())
provider.topup.assert_not_awaited()
tests/unit/test_cashu_transaction_storage.py
-181
View File
@@ -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
-110
View File
@@ -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_periodic_refund_sweep.py
+126
View File
@@ -0,0 +1,126 @@
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from unittest.mock import AsyncMock
import pytest
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
from sqlalchemy.pool import StaticPool
from sqlmodel import SQLModel, select
from sqlmodel.ext.asyncio.session import AsyncSession
from routstr import wallet
from routstr.core.db import CashuTransaction
def _make_engine() -> AsyncEngine:
return create_async_engine(
"sqlite+aiosqlite://",
poolclass=StaticPool,
connect_args={"check_same_thread": False},
)
@pytest.fixture
async def refund_db(
monkeypatch: pytest.MonkeyPatch,
) -> AsyncGenerator[AsyncEngine, None]:
engine = _make_engine()
async with engine.begin() as connection:
await connection.run_sync(SQLModel.metadata.create_all)
@asynccontextmanager
async def create_session() -> AsyncGenerator[AsyncSession, None]:
async with AsyncSession(engine, expire_on_commit=False) as session:
yield session
monkeypatch.setattr(wallet.db, "create_session", create_session)
try:
yield engine
finally:
await engine.dispose()
async def _store(
engine: AsyncEngine,
*,
token: str,
created_at: int,
typ: str = "out",
collected: bool = False,
swept: bool = False,
) -> None:
async with AsyncSession(engine, expire_on_commit=False) as session:
session.add(
CashuTransaction(
token=token,
amount=10,
unit="sat",
type=typ,
created_at=created_at,
collected=collected,
swept=swept,
)
)
await session.commit()
async def _transactions(engine: AsyncEngine) -> dict[str, CashuTransaction]:
async with AsyncSession(engine, expire_on_commit=False) as session:
results = await session.exec(select(CashuTransaction))
return {transaction.token: transaction for transaction in results.all()}
@pytest.mark.asyncio
async def test_refund_sweep_only_processes_eligible_outbound_transactions(
refund_db: AsyncEngine, monkeypatch: pytest.MonkeyPatch
) -> None:
cutoff = 1_000
await _store(refund_db, token="eligible", created_at=cutoff - 1)
await _store(refund_db, token="too-new", created_at=cutoff)
await _store(
refund_db, token="collected", created_at=cutoff - 1, collected=True
)
await _store(refund_db, token="swept", created_at=cutoff - 1, swept=True)
await _store(refund_db, token="inbound", created_at=cutoff - 1, typ="in")
receive_token = AsyncMock()
monkeypatch.setattr(wallet, "recieve_token", receive_token)
await wallet._refund_sweep_once(cutoff)
receive_token.assert_awaited_once_with("eligible")
transactions = await _transactions(refund_db)
assert transactions["eligible"].swept is True
assert transactions["too-new"].swept is False
assert transactions["collected"].collected is True
assert transactions["collected"].swept is False
assert transactions["swept"].swept is True
assert transactions["inbound"].swept is False
@pytest.mark.asyncio
async def test_refund_sweep_persists_success_and_isolates_token_failures(
refund_db: AsyncEngine, monkeypatch: pytest.MonkeyPatch
) -> None:
cutoff = 1_000
for token in ("success", "already-spent", "temporary-failure"):
await _store(refund_db, token=token, created_at=cutoff - 1)
async def receive_token(token: str) -> None:
if token == "already-spent":
raise RuntimeError("Token already spent")
if token == "temporary-failure":
raise RuntimeError("mint unavailable")
receive = AsyncMock(side_effect=receive_token)
monkeypatch.setattr(wallet, "recieve_token", receive)
await wallet._refund_sweep_once(cutoff)
assert receive.await_count == 3
transactions = await _transactions(refund_db)
assert transactions["success"].swept is True
assert transactions["success"].collected is False
assert transactions["already-spent"].collected is True
assert transactions["already-spent"].swept is False
assert transactions["temporary-failure"].collected is False
assert transactions["temporary-failure"].swept is False
tests/unit/test_wallet.py
+15 -55
View File
@@ -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."""
mock_key = Mock(balance=0, hashed_key="test_hash")
"""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"
mock_session = AsyncMock()
debug = Mock()
from routstr.core.settings import settings
with (
patch.object(settings, "cashu_mints", ["http://mint:3338"]),
patch(
with patch.object(settings, "cashu_mints", ["http://mint:3338"]):
with 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)
):
with patch(
"routstr.wallet.store_cashu_transaction",
side_effect=Exception("history table locked"),
):
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