From 5ea5024608d0f5e4ec8d1c02179d9fb69a68134c Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Wed, 29 Jul 2026 22:50:33 +0200 Subject: [PATCH] resolve review comments --- routstr/core/db.py | 3 +- routstr/lightning.py | 60 +++++++++++----- routstr/wallet.py | 40 +++++------ .../test_lightning_invoice_constraints.py | 59 +++++++++++++-- tests/unit/test_fee_payout_crash_safety.py | 39 ++++++++++ tests/unit/test_refund_sweep.py | 72 ++++++++++++++++--- 6 files changed, 217 insertions(+), 56 deletions(-) diff --git a/routstr/core/db.py b/routstr/core/db.py index 9e0c5973..12dcc659 100644 --- a/routstr/core/db.py +++ b/routstr/core/db.py @@ -382,7 +382,8 @@ class LightningInvoice(SQLModel, table=True): # type: ignore description: str = Field(description="Invoice description") payment_hash: str = Field(description="Payment hash for tracking", unique=True) status: str = Field( - default="pending", description="pending, paid, expired, cancelled" + default="pending", + description="pending, paid, expired, cancelled, reconciliation_required", ) api_key_hash: str | None = Field( default=None, description="Associated API key hash for topup operations" diff --git a/routstr/lightning.py b/routstr/lightning.py index 554925f2..d696403a 100644 --- a/routstr/lightning.py +++ b/routstr/lightning.py @@ -5,6 +5,7 @@ import time from fastapi import APIRouter, Depends, Header, HTTPException from pydantic import BaseModel, Field +from sqlalchemy.orm.attributes import set_committed_value from sqlmodel import col, select, update from sqlmodel.ext.asyncio.session import AsyncSession @@ -233,25 +234,46 @@ async def check_invoice_payment( # potentially slow mint request so it cannot pin a pool connection. await session.commit() + wallet = await get_wallet(settings.primary_mint, "sat") + mint_status = await wallet.get_mint_quote(invoice_payment_hash) + if not mint_status.paid: + return + # Do not redeem a paid top-up quote if its target has already been # pruned. This validation owns a short-lived session and releases its - # connection before any mint I/O starts. + # connection before mint redemption starts. if invoice_purpose == "topup": if not finalized_api_key_hash: raise ValueError("No API key associated with topup invoice") async with create_session() as validation_session: target = await validation_session.get(ApiKey, finalized_api_key_hash) - if target is None: - logger.error( - "Topup invoice target API key was not found; skipping mint", - extra={"invoice_id": invoice_id}, - ) - return - - wallet = await get_wallet(settings.primary_mint, "sat") - mint_status = await wallet.get_mint_quote(invoice_payment_hash) - if not mint_status.paid: - return + if target is None: + terminal = await validation_session.exec( # type: ignore[call-overload] + update(LightningInvoice) + .where( + col(LightningInvoice.id) == invoice_id, + col(LightningInvoice.status) == "pending", + ) + .values(status="reconciliation_required") + ) + await validation_session.commit() + if terminal.rowcount == 1: + set_committed_value( + invoice, "status", "reconciliation_required" + ) + else: + committed_invoice = await validation_session.get( + LightningInvoice, invoice_id + ) + if committed_invoice is not None: + set_committed_value( + invoice, "status", committed_invoice.status + ) + logger.critical( + "Paid topup invoice target API key was not found; reconciliation required", + extra={"invoice_id": invoice_id}, + ) + return # The mint enforces single-use quotes, so a concurrent checker that # races us here fails inside wallet.mint rather than double-minting. @@ -294,17 +316,19 @@ async def check_invoice_payment( # A concurrent finalizer won the CAS. Publish only the # state observed from the database after ending the owned # read transaction; never refresh the caller's session. - invoice.api_key_hash = committed_invoice.api_key_hash - invoice.status = committed_invoice.status - invoice.paid_at = committed_invoice.paid_at + set_committed_value( + invoice, "api_key_hash", committed_invoice.api_key_hash + ) + set_committed_value(invoice, "status", committed_invoice.status) + set_committed_value(invoice, "paid_at", committed_invoice.paid_at) return await finalization_session.commit() # Only publish finalized values to the caller-owned object after the # owned transaction has committed successfully. - invoice.api_key_hash = finalized_api_key_hash - invoice.status = "paid" - invoice.paid_at = paid_at + set_committed_value(invoice, "api_key_hash", finalized_api_key_hash) + set_committed_value(invoice, "status", "paid") + set_committed_value(invoice, "paid_at", paid_at) logger.info( "Lightning invoice paid", diff --git a/routstr/wallet.py b/routstr/wallet.py index e2a20a7c..aed99f92 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -1116,12 +1116,12 @@ async def _refund_sweep_once(cutoff: int) -> None: extra={"id": refund.id}, ) except BaseException as e: - if redeemed: - # The token is already in the node wallet. Retain the claim so - # a stale retry classifies "already spent" as swept, never as a - # client collection. + if redeemed or isinstance(e, TokenConsumedError): + # The token was spent, or the redemption outcome is known to be + # post-spend. Retain the claim so a stale retry classifies + # "already spent" as swept, never as a client collection. logger.critical( - "Refund token swept but checkpoint was not completed; manual reconciliation required", + "Refund token spent but sweep checkpoint was not completed; manual reconciliation required", extra={"id": refund.id}, exc_info=isinstance(e, Exception), ) @@ -1163,25 +1163,17 @@ async def _refund_sweep_once(cutoff: int) -> None: extra={"id": refund.id}, ) else: - # Cancellation or a transient pre-redemption failure is a - # known-no-send outcome, so release the lease for a later retry. - released = await asyncio.shield( - _set_refund_sweep_state( - refund.id, - predicates=(claim_owned,), - sweep_started_at=None, - ) + # Once redemption starts, an exception cannot prove the token + # was not spent (for example, a melt may land before the + # response is lost). Retain the claim so a stale retry treats + # an "already spent" result as a completed sweep. + logger.critical( + "Refund token redemption outcome is unknown; retaining sweep claim for reconciliation", + extra={"id": refund.id, "error": str(e)}, + exc_info=isinstance(e, Exception), ) if not isinstance(e, Exception): raise - logger.warning( - "Failed to sweep refund", - extra={ - "id": refund.id, - "error": str(e), - "claim_released": released == 1, - }, - ) async def refund_sweep_once() -> None: @@ -1254,12 +1246,14 @@ async def periodic_routstr_fee_payout() -> None: "sat", amount=accumulated_sats, ) - except Exception: + except BaseException as e: logger.critical( "Routstr fee payout outcome is unknown; manual reconciliation required", extra={"payout_in_progress_msats": paid_msats}, - exc_info=True, + exc_info=isinstance(e, Exception), ) + if not isinstance(e, Exception): + raise continue try: diff --git a/tests/integration/test_lightning_invoice_constraints.py b/tests/integration/test_lightning_invoice_constraints.py index b4e12a7f..079954ce 100644 --- a/tests/integration/test_lightning_invoice_constraints.py +++ b/tests/integration/test_lightning_invoice_constraints.py @@ -254,6 +254,40 @@ async def test_failed_mint_keeps_invoice_pending_for_retry( assert stored.status == "pending" +@pytest.mark.asyncio +async def test_unpaid_topup_does_not_query_target_key( + integration_engine: AsyncEngine, + patched_db_engine: None, +) -> None: + invoice = _make_invoice( + id="inv_unpaid_topup", + status="pending", + paid_at=None, + purpose="topup", + api_key_hash="target-key", + ) + async with AsyncSession(integration_engine, expire_on_commit=False) as setup: + setup.add(invoice) + await setup.commit() + + wallet = MagicMock() + wallet.get_mint_quote = AsyncMock(return_value=MagicMock(paid=False)) + create_session = MagicMock(side_effect=RuntimeError("target lookup should not run")) + async with AsyncSession(integration_engine, expire_on_commit=False) as session: + stored = await session.get(LightningInvoice, invoice.id) + assert stored is not None + with ( + patch("routstr.lightning.get_wallet", AsyncMock(return_value=wallet)), + patch("routstr.lightning.create_session", create_session), + ): + from routstr.lightning import check_invoice_payment + + await check_invoice_payment(stored, session) + + wallet.get_mint_quote.assert_awaited_once_with(invoice.payment_hash) + create_session.assert_not_called() + + @pytest.mark.asyncio async def test_missing_topup_target_is_rejected_before_mint( integration_engine: AsyncEngine, @@ -265,25 +299,36 @@ async def test_missing_topup_target_is_rejected_before_mint( paid_at=None, purpose="topup", api_key_hash="pruned-key", + expires_at=int(time.time()) - 1, ) async with AsyncSession(integration_engine, expire_on_commit=False) as setup: setup.add(invoice) await setup.commit() - get_wallet = AsyncMock() + wallet = MagicMock() + wallet.get_mint_quote = AsyncMock(return_value=MagicMock(paid=True)) + wallet.mint = AsyncMock() async with AsyncSession(integration_engine, expire_on_commit=False) as session: stored = await session.get(LightningInvoice, invoice.id) assert stored is not None - with patch("routstr.lightning.get_wallet", get_wallet): - from routstr.lightning import check_invoice_payment + with ( + patch("routstr.lightning.get_wallet", AsyncMock(return_value=wallet)), + patch("routstr.lightning.logger.critical") as critical, + ): + from routstr.lightning import get_invoice_status - await check_invoice_payment(stored, session) + response = await get_invoice_status(invoice.id, session) - get_wallet.assert_not_awaited() + assert response.status == "reconciliation_required" + assert stored.status == "reconciliation_required" + assert stored not in session.dirty + critical.assert_called_once() + + wallet.mint.assert_not_awaited() async with AsyncSession(integration_engine) as verify: stored = await verify.get(LightningInvoice, invoice.id) assert stored is not None - assert stored.status == "pending" + assert stored.status == "reconciliation_required" @pytest.mark.asyncio @@ -424,6 +469,8 @@ async def test_db_guard_credits_once_when_both_mints_succeed( assert second_invoice.id == invoice.id assert first_invoice.status == "paid" assert second_invoice.status == "paid" + assert first_invoice not in first.dirty + assert second_invoice not in second.dirty assert wallet.mint.await_count == 2 async with AsyncSession(integration_engine, expire_on_commit=False) as verify: diff --git a/tests/unit/test_fee_payout_crash_safety.py b/tests/unit/test_fee_payout_crash_safety.py index 0768db16..ae7d0788 100644 --- a/tests/unit/test_fee_payout_crash_safety.py +++ b/tests/unit/test_fee_payout_crash_safety.py @@ -252,6 +252,45 @@ async def test_fee_payout_keeps_checkpoint_when_send_outcome_is_unknown() -> Non critical.assert_called_once() +@pytest.mark.asyncio +async def test_fee_payout_cancellation_during_send_alerts_and_propagates() -> None: + session = Mock() + fee = SimpleNamespace( + accumulated_msats=5_000, + payout_in_progress_msats=0, + payout_started_at=None, + ) + complete = AsyncMock() + + with ( + patch("routstr.auth.ROUTSTR_FEE_DEFAULT_PAYOUT", 1), + patch("routstr.auth.ROUTSTR_FEE_PAYOUT_INTERVAL_SECONDS", 1), + patch("routstr.auth.ROUTSTR_LN_ADDRESS", "fees@example.com"), + patch("routstr.wallet.asyncio.sleep", AsyncMock(return_value=None)), + patch( + "routstr.wallet.db.create_session", return_value=_session_context(session) + ), + patch("routstr.wallet.db.get_routstr_fee", AsyncMock(return_value=fee)), + patch("routstr.wallet.db.reset_routstr_fee", AsyncMock(return_value=True)), + patch("routstr.wallet.db.complete_routstr_fee_payout", complete), + patch("routstr.wallet.get_wallet", AsyncMock(return_value=Mock())), + patch("routstr.wallet.get_proofs_per_mint_and_unit", return_value=[]), + patch( + "routstr.wallet.raw_send_to_lnurl", + AsyncMock(side_effect=asyncio.CancelledError()), + ), + patch("routstr.wallet.logger.critical") as critical, + ): + with pytest.raises(asyncio.CancelledError): + await wallet.periodic_routstr_fee_payout() + + complete.assert_not_awaited() + critical.assert_called_once() + assert critical.call_args.args[0] == ( + "Routstr fee payout outcome is unknown; manual reconciliation required" + ) + + @pytest.mark.asyncio @pytest.mark.parametrize("failure_site", ["session", "completion"]) async def test_fee_payout_completion_failures_use_sent_checkpoint_alert( diff --git a/tests/unit/test_refund_sweep.py b/tests/unit/test_refund_sweep.py index b43c3152..f2937936 100644 --- a/tests/unit/test_refund_sweep.py +++ b/tests/unit/test_refund_sweep.py @@ -130,14 +130,17 @@ async def test_refund_sweep_only_processes_expired_eligible_outgoing_tokens( @pytest.mark.asyncio @pytest.mark.parametrize( - ("error", "collected"), + ("error", "collected", "claim_started_at"), [ - (RuntimeError("token already spent"), True), - (RuntimeError("mint unavailable"), False), + (RuntimeError("token already spent"), True, None), + (RuntimeError("mint unavailable"), False, 1000), ], ) -async def test_refund_sweep_records_terminal_but_not_transient_failures( - session_factory: async_sessionmaker[AsyncSession], error: Exception, collected: bool +async def test_refund_sweep_records_spent_and_unknown_outcomes_safely( + session_factory: async_sessionmaker[AsyncSession], + error: Exception, + collected: bool, + claim_started_at: int | None, ) -> None: await _insert( session_factory, @@ -156,11 +159,64 @@ async def test_refund_sweep_records_terminal_but_not_transient_failures( refund = (await _load(session_factory))["refund"] assert refund.collected is collected assert refund.swept is False - assert refund.sweep_started_at is None + assert refund.sweep_started_at == claim_started_at @pytest.mark.asyncio -async def test_refund_sweep_releases_claim_on_cancellation( +async def test_post_spend_failure_retains_claim_and_stale_retry_records_sweep( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _insert( + session_factory, + CashuTransaction( + token="post-spend-failure", + amount=1, + unit="sat", + type="out", + created_at=800, + ), + ) + with ( + patch("routstr.wallet.db.create_session", side_effect=session_factory), + patch("routstr.wallet.settings.refund_sweep_ttl_seconds", 100), + patch("routstr.wallet.settings.refund_sweep_claim_timeout_seconds", 200), + patch("routstr.wallet.time.time", return_value=1000), + patch( + "routstr.wallet.recieve_token", + AsyncMock( + side_effect=wallet.TokenConsumedError( + "Mint on primary failed after successful melt" + ) + ), + ), + ): + await refund_sweep_once() + + retained = (await _load(session_factory))["post-spend-failure"] + assert retained.swept is False + assert retained.collected is False + assert retained.sweep_started_at == 1000 + + with ( + patch("routstr.wallet.db.create_session", side_effect=session_factory), + patch("routstr.wallet.settings.refund_sweep_ttl_seconds", 100), + patch("routstr.wallet.settings.refund_sweep_claim_timeout_seconds", 200), + patch("routstr.wallet.time.time", return_value=1300), + patch( + "routstr.wallet.recieve_token", + AsyncMock(side_effect=RuntimeError("token already spent")), + ), + ): + await refund_sweep_once() + + recovered = (await _load(session_factory))["post-spend-failure"] + assert recovered.swept is True + assert recovered.collected is False + assert recovered.sweep_started_at is None + + +@pytest.mark.asyncio +async def test_refund_sweep_retains_claim_on_cancellation_during_redemption( session_factory: async_sessionmaker[AsyncSession], ) -> None: await _insert( @@ -183,7 +239,7 @@ async def test_refund_sweep_releases_claim_on_cancellation( refund = (await _load(session_factory))["cancelled"] assert refund.swept is False - assert refund.sweep_started_at is None + assert refund.sweep_started_at == 1000 @pytest.mark.asyncio