resolve review comments

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"

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",

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:

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:

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(

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