From 816e07332daca5b6dc1b63ea7b0879185526c511 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Tue, 11 Aug 2026 01:32:41 +0200 Subject: [PATCH] fix: reconcile fee payouts safely --- ..._add_fee_payout_reconciliation_metadata.py | 37 ++ routstr/core/db.py | 68 +++- routstr/payment/lnurl.py | 9 +- routstr/wallet.py | 147 +++++-- tests/unit/test_fee_payout_crash_safety.py | 377 ++++++++++++++++-- tests/unit/test_fee_payout_migration.py | 14 +- tests/unit/test_lnurl_melt_timeout.py | 49 ++- 7 files changed, 613 insertions(+), 88 deletions(-) create mode 100644 migrations/versions/b4f7a1c9d2e3_add_fee_payout_reconciliation_metadata.py diff --git a/migrations/versions/b4f7a1c9d2e3_add_fee_payout_reconciliation_metadata.py b/migrations/versions/b4f7a1c9d2e3_add_fee_payout_reconciliation_metadata.py new file mode 100644 index 00000000..edbd4b53 --- /dev/null +++ b/migrations/versions/b4f7a1c9d2e3_add_fee_payout_reconciliation_metadata.py @@ -0,0 +1,37 @@ +"""add fee payout reconciliation metadata + +Revision ID: b4f7a1c9d2e3 +Revises: f2a7c9d4e8b1 +Create Date: 2026-08-11 00:00:00.000000 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision = "b4f7a1c9d2e3" +down_revision = "f2a7c9d4e8b1" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "routstr_fees", + sa.Column("payout_quote_id", sa.String(), nullable=True), + ) + op.add_column( + "routstr_fees", + sa.Column("payout_mint_url", sa.String(), nullable=True), + ) + op.add_column( + "routstr_fees", + sa.Column("payout_unit", sa.String(), nullable=True), + ) + + +def downgrade() -> None: + op.drop_column("routstr_fees", "payout_unit") + op.drop_column("routstr_fees", "payout_mint_url") + op.drop_column("routstr_fees", "payout_quote_id") diff --git a/routstr/core/db.py b/routstr/core/db.py index 0a820b92..d7679933 100644 --- a/routstr/core/db.py +++ b/routstr/core/db.py @@ -307,9 +307,7 @@ async def prune_dead_api_keys(session: AsyncSession, min_age_seconds: int) -> in pending_invoice = ( select(LightningInvoice.id) .where(col(LightningInvoice.api_key_hash) == col(ApiKey.hashed_key)) - .where( - col(LightningInvoice.status).in_(("pending", "settlement_pending")) - ) + .where(col(LightningInvoice.status).in_(("pending", "settlement_pending"))) ).exists() eligible_hashes = ( @@ -447,7 +445,8 @@ class LightningInvoice(SQLModel, table=True): # type: ignore ) purpose: str = Field(description="create or topup") mint_url: str | None = Field( - default=None, description="Mint URL where the quote was created (fallback tracking)" + default=None, + description="Mint URL where the quote was created (fallback tracking)", ) created_at: int = Field( default_factory=lambda: int(time.time()), description="Unix timestamp" @@ -682,6 +681,9 @@ class RoutstrFee(SQLModel, table=True): # type: ignore last_paid_at: int | None = Field(default=None) payout_in_progress_msats: int = Field(default=0) payout_started_at: int | None = Field(default=None) + payout_quote_id: str | None = Field(default=None) + payout_mint_url: str | None = Field(default=None) + payout_unit: str | None = Field(default=None) class NsecState(str, Enum): @@ -805,8 +807,14 @@ async def set_nsec(session: AsyncSession, nsec: str) -> None: await session.commit() -async def reset_routstr_fee(session: AsyncSession, paid_msats: int) -> bool: - """Checkpoint a fee payout before making the external payment.""" +async def reset_routstr_fee( + session: AsyncSession, + paid_msats: int, + quote_id: str, + mint_url: str, + unit: str, +) -> bool: + """Checkpoint a fee payout and its reconciliation metadata before dispatch.""" stmt = ( update(RoutstrFee) .where(col(RoutstrFee.id) == 1) @@ -816,6 +824,9 @@ async def reset_routstr_fee(session: AsyncSession, paid_msats: int) -> bool: accumulated_msats=RoutstrFee.accumulated_msats - paid_msats, payout_in_progress_msats=paid_msats, payout_started_at=int(time.time()), + payout_quote_id=quote_id, + payout_mint_url=mint_url, + payout_unit=unit, ) ) result = await session.exec(stmt) # type: ignore[call-overload] @@ -823,15 +834,56 @@ async def reset_routstr_fee(session: AsyncSession, paid_msats: int) -> bool: return result.rowcount == 1 -async def complete_routstr_fee_payout(session: AsyncSession, paid_msats: int) -> bool: - """Mark a checkpointed payout complete after the external payment succeeds.""" +async def restore_routstr_fee_payout( + session: AsyncSession, + paid_msats: int, + quote_id: str, + mint_url: str, + unit: str, +) -> bool: + """Return the matching unresolved payout to the accumulated fee balance.""" stmt = ( update(RoutstrFee) .where(col(RoutstrFee.id) == 1) .where(col(RoutstrFee.payout_in_progress_msats) == paid_msats) + .where(col(RoutstrFee.payout_quote_id) == quote_id) + .where(col(RoutstrFee.payout_mint_url) == mint_url) + .where(col(RoutstrFee.payout_unit) == unit) + .values( + accumulated_msats=RoutstrFee.accumulated_msats + paid_msats, + payout_in_progress_msats=0, + payout_started_at=None, + payout_quote_id=None, + payout_mint_url=None, + payout_unit=None, + ) + ) + result = await session.exec(stmt) # type: ignore[call-overload] + await session.commit() + return result.rowcount == 1 + + +async def complete_routstr_fee_payout( + session: AsyncSession, + paid_msats: int, + quote_id: str, + mint_url: str, + unit: str, +) -> bool: + """Mark the matching checkpoint complete after external payment succeeds.""" + stmt = ( + update(RoutstrFee) + .where(col(RoutstrFee.id) == 1) + .where(col(RoutstrFee.payout_in_progress_msats) == paid_msats) + .where(col(RoutstrFee.payout_quote_id) == quote_id) + .where(col(RoutstrFee.payout_mint_url) == mint_url) + .where(col(RoutstrFee.payout_unit) == unit) .values( payout_in_progress_msats=0, payout_started_at=None, + payout_quote_id=None, + payout_mint_url=None, + payout_unit=None, total_paid_msats=RoutstrFee.total_paid_msats + paid_msats, last_paid_at=int(time.time()), ) diff --git a/routstr/payment/lnurl.py b/routstr/payment/lnurl.py index f03d1ef5..37ecb1cf 100644 --- a/routstr/payment/lnurl.py +++ b/routstr/payment/lnurl.py @@ -1,6 +1,7 @@ from __future__ import annotations import math +from collections.abc import Awaitable, Callable from typing import TypedDict import httpx @@ -175,6 +176,8 @@ async def raw_send_to_lnurl( lnurl: str, unit: str, amount: int | None = None, + *, + on_melt_quote: Callable[[str], Awaitable[None]] | None = None, ) -> int: """Send funds to an LNURL address. @@ -237,6 +240,9 @@ async def raw_send_to_lnurl( mint_url=str(wallet.url), ) + if on_melt_quote is not None: + await on_melt_quote(melt_quote_resp.quote) + if amount: proofs, _ = await wallet.select_to_send(proofs, amount, set_reserved=True) @@ -287,6 +293,5 @@ async def raw_send_to_lnurl( state = getattr(getattr(quote, "state", None), "value", "unknown") raise MeltOutcomeAmbiguousError( - "Melt outcome is ambiguous; proofs must not be retried " - f"(quote_state={state})" + f"Melt outcome is ambiguous; proofs must not be retried (quote_state={state})" ) from melt_error diff --git a/routstr/wallet.py b/routstr/wallet.py index bc92f028..3ab20c78 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -748,10 +748,17 @@ async def check_bolt11_payment_status(mint_url: str, unit: str, quote_id: str) - Runs under ``wallet_operation_guard`` because of that side effect: it mutates proof state and must not race other processes' wallet operations. """ + async with wallet_operation_guard(): + return await _check_bolt11_payment_status_locked(mint_url, unit, quote_id) + + +async def _check_bolt11_payment_status_locked( + mint_url: str, unit: str, quote_id: str +) -> str: + """Check a melt quote while the caller holds ``wallet_operation_guard``.""" try: - async with wallet_operation_guard(): - wallet = await get_wallet(mint_url, unit, force_reload=True) - quote = await wallet.get_melt_quote(quote_id) + wallet = await get_wallet(mint_url, unit, force_reload=True) + quote = await wallet.get_melt_quote(quote_id) except Exception as e: logger.warning( "Could not query the mint for a melt quote's status", @@ -2346,6 +2353,10 @@ async def periodic_refund_sweep() -> None: ) +class _RoutstrFeePayoutAlreadyClaimed(Exception): + """Another worker claimed the fee balance before melt dispatch.""" + + async def periodic_routstr_fee_payout() -> None: from .auth import ( ROUTSTR_FEE_DEFAULT_PAYOUT, @@ -2361,27 +2372,82 @@ async def periodic_routstr_fee_payout() -> None: try: async with db.create_session() as session: fee = await db.get_routstr_fee(session) - if fee.payout_in_progress_msats: - logger.critical( - "Routstr fee payout requires manual reconciliation", - extra={ - "payout_in_progress_msats": fee.payout_in_progress_msats, - "payout_started_at": fee.payout_started_at, - }, - ) - continue - + payout_in_progress_msats = fee.payout_in_progress_msats accumulated_sats = _msats_to_sats(fee.accumulated_msats) - if accumulated_sats < ROUTSTR_FEE_DEFAULT_PAYOUT: - continue - paid_msats = _sats_to_msats(accumulated_sats) - # Serialize proof refresh, reservation, sending, and checkpoint - # finalization with every other wallet mutation across workers. + if payout_in_progress_msats: + # Dispatch holds the same guard from before checkpoint creation + # through melt completion. Re-read after taking it so a second + # worker cannot reconcile the quote between checkpoint and melt. + async with wallet_operation_guard(): + async with db.create_session() as session: + fee = await db.get_routstr_fee(session) + payout_in_progress_msats = fee.payout_in_progress_msats + payout_started_at = fee.payout_started_at + payout_quote_id = getattr(fee, "payout_quote_id", None) + payout_mint_url = getattr(fee, "payout_mint_url", None) + payout_unit = getattr(fee, "payout_unit", None) + + if not payout_in_progress_msats: + continue + if not (payout_quote_id and payout_mint_url and payout_unit): + logger.critical( + "Routstr fee payout lacks reconciliation metadata", + extra={ + "payout_in_progress_msats": payout_in_progress_msats, + "payout_started_at": payout_started_at, + }, + ) + continue + + quote_state = await _check_bolt11_payment_status_locked( + payout_mint_url, payout_unit, payout_quote_id + ) + if quote_state == "paid": + async with db.create_session() as session: + completed = await db.complete_routstr_fee_payout( + session, + payout_in_progress_msats, + payout_quote_id, + payout_mint_url, + payout_unit, + ) + if completed: + logger.info( + "Routstr fee payout reconciled as paid", + extra={"payout_quote_id": payout_quote_id}, + ) + elif quote_state == "unpaid": + async with db.create_session() as session: + restored = await db.restore_routstr_fee_payout( + session, + payout_in_progress_msats, + payout_quote_id, + payout_mint_url, + payout_unit, + ) + if restored: + logger.warning( + "Routstr fee payout reconciled as unpaid and restored for retry", + extra={"payout_quote_id": payout_quote_id}, + ) + else: + logger.warning( + "Routstr fee payout is still awaiting reconciliation", + extra={ + "payout_quote_id": payout_quote_id, + "quote_state": quote_state, + }, + ) + continue + + if accumulated_sats < ROUTSTR_FEE_DEFAULT_PAYOUT: + continue + paid_msats = _sats_to_msats(accumulated_sats) + + # Serialize proof refresh, quote creation, checkpointing, sending, + # and finalization with every other wallet mutation across workers. async with wallet_operation_guard(): - # Wallet/proof preparation cannot send funds, so do it before - # the durable checkpoint. Force a DB reload after taking the - # guard so another worker's reservations are visible. wallet = await get_wallet( settings.primary_mint, "sat", force_reload=True ) @@ -2389,13 +2455,21 @@ async def periodic_routstr_fee_payout() -> None: wallet, settings.primary_mint, "sat", not_reserved=True ) - async with db.create_session() as session: - payout_checkpointed = await db.reset_routstr_fee( - session, paid_msats - ) - if not payout_checkpointed: - logger.warning("Routstr fee payout was already claimed") - continue + attempt_quote_id: str | None = None + + async def checkpoint_quote(quote_id: str) -> None: + nonlocal attempt_quote_id + async with db.create_session() as session: + checkpointed = await db.reset_routstr_fee( + session, + paid_msats, + quote_id, + settings.primary_mint, + "sat", + ) + if not checkpointed: + raise _RoutstrFeePayoutAlreadyClaimed + attempt_quote_id = quote_id try: amount_received = await raw_send_to_lnurl( @@ -2404,10 +2478,14 @@ async def periodic_routstr_fee_payout() -> None: ROUTSTR_LN_ADDRESS, "sat", amount=accumulated_sats, + on_melt_quote=checkpoint_quote, ) + except _RoutstrFeePayoutAlreadyClaimed: + logger.warning("Routstr fee payout was already claimed") + continue except BaseException as e: logger.critical( - "Routstr fee payout outcome is unknown; manual reconciliation required", + "Routstr fee payout outcome is unknown; awaiting quote reconciliation", extra={"payout_in_progress_msats": paid_msats}, exc_info=isinstance(e, Exception), ) @@ -2415,14 +2493,19 @@ async def periodic_routstr_fee_payout() -> None: raise continue + assert attempt_quote_id is not None try: async with db.create_session() as session: payout_completed = await db.complete_routstr_fee_payout( - session, paid_msats + session, + paid_msats, + attempt_quote_id, + settings.primary_mint, + "sat", ) except BaseException as e: logger.critical( - "Routstr fee payout sent but checkpoint was not completed", + "Routstr fee payout sent but checkpoint was not completed; awaiting quote reconciliation", extra={"payout_in_progress_msats": paid_msats}, exc_info=isinstance(e, Exception), ) @@ -2431,7 +2514,7 @@ async def periodic_routstr_fee_payout() -> None: continue if not payout_completed: logger.critical( - "Routstr fee payout sent but checkpoint was not completed", + "Routstr fee payout sent but checkpoint was not completed; awaiting quote reconciliation", extra={"payout_in_progress_msats": paid_msats}, ) continue diff --git a/tests/unit/test_fee_payout_crash_safety.py b/tests/unit/test_fee_payout_crash_safety.py index 98cbe362..9e1dfaf2 100644 --- a/tests/unit/test_fee_payout_crash_safety.py +++ b/tests/unit/test_fee_payout_crash_safety.py @@ -38,24 +38,124 @@ async def test_fee_payout_checkpoint_is_atomic_and_durable() -> None: session.add(db.RoutstrFee(id=1, accumulated_msats=5_000)) await session.commit() - assert await db.reset_routstr_fee(session, 5_000) is True - assert await db.reset_routstr_fee(session, 5_000) is False + assert ( + await db.reset_routstr_fee( + session, 5_000, "quote-1", "https://mint.test", "sat" + ) + is True + ) + assert ( + await db.reset_routstr_fee( + session, 5_000, "quote-2", "https://mint.test", "sat" + ) + is False + ) fee = await db.get_routstr_fee(session) await session.refresh(fee) assert fee.accumulated_msats == 0 assert fee.payout_in_progress_msats == 5_000 + assert fee.payout_quote_id == "quote-1" + assert fee.payout_mint_url == "https://mint.test" + assert fee.payout_unit == "sat" assert fee.total_paid_msats == 0 - assert await db.complete_routstr_fee_payout(session, 5_000) is True + assert ( + await db.complete_routstr_fee_payout( + session, 5_000, "quote-1", "https://mint.test", "sat" + ) + is True + ) await session.refresh(fee) assert fee.payout_in_progress_msats == 0 + assert fee.payout_quote_id is None + assert fee.payout_mint_url is None + assert fee.payout_unit is None assert fee.total_paid_msats == 5_000 assert fee.last_paid_at is not None await engine.dispose() +@pytest.mark.asyncio +async def test_fee_payout_checkpoint_can_be_restored_for_retry() -> None: + engine = create_async_engine("sqlite+aiosqlite://") + async with engine.begin() as connection: + await connection.run_sync(SQLModel.metadata.create_all) + + async with AsyncSession(engine) as session: + session.add(db.RoutstrFee(id=1, accumulated_msats=7_000)) + await session.commit() + + assert ( + await db.reset_routstr_fee( + session, 5_000, "quote-1", "https://mint.test", "sat" + ) + is True + ) + assert ( + await db.restore_routstr_fee_payout( + session, 5_000, "quote-1", "https://mint.test", "sat" + ) + is True + ) + assert ( + await db.restore_routstr_fee_payout( + session, 5_000, "quote-1", "https://mint.test", "sat" + ) + is False + ) + + fee = await db.get_routstr_fee(session) + await session.refresh(fee) + assert fee.accumulated_msats == 7_000 + assert fee.payout_in_progress_msats == 0 + assert fee.payout_started_at is None + assert fee.payout_quote_id is None + assert fee.payout_mint_url is None + assert fee.payout_unit is None + assert fee.total_paid_msats == 0 + + await engine.dispose() + + +@pytest.mark.asyncio +async def test_stale_reconciliation_cannot_mutate_replacement_quote() -> None: + engine = create_async_engine("sqlite+aiosqlite://") + async with engine.begin() as connection: + await connection.run_sync(SQLModel.metadata.create_all) + + async with AsyncSession(engine) as session: + session.add(db.RoutstrFee(id=1, accumulated_msats=10_000)) + await session.commit() + + assert await db.reset_routstr_fee( + session, 5_000, "quote-1", "https://mint.test", "sat" + ) + assert await db.restore_routstr_fee_payout( + session, 5_000, "quote-1", "https://mint.test", "sat" + ) + assert await db.reset_routstr_fee( + session, 5_000, "quote-2", "https://mint.test", "sat" + ) + + assert not await db.restore_routstr_fee_payout( + session, 5_000, "quote-1", "https://mint.test", "sat" + ) + assert not await db.complete_routstr_fee_payout( + session, 5_000, "quote-1", "https://mint.test", "sat" + ) + + fee = await db.get_routstr_fee(session) + await session.refresh(fee) + assert fee.accumulated_msats == 5_000 + assert fee.payout_in_progress_msats == 5_000 + assert fee.payout_quote_id == "quote-2" + assert fee.total_paid_msats == 0 + + await engine.dispose() + + @pytest.mark.asyncio async def test_fee_payout_prepares_wallet_then_checkpoints_before_sending() -> None: session = Mock() @@ -75,7 +175,9 @@ async def test_fee_payout_prepares_wallet_then_checkpoints_before_sending() -> N events.append("checkpoint") return True - async def send(*_args: object, **_kwargs: object) -> int: + async def send(*_args: object, **kwargs: object) -> int: + checkpoint_quote = kwargs["on_melt_quote"] + await checkpoint_quote("quote-1") # type: ignore[operator] events.append("send") return 5 @@ -149,7 +251,13 @@ async def test_fee_payout_lost_checkpoint_race_does_not_send() -> None: payout_in_progress_msats=0, payout_started_at=None, ) - send = AsyncMock() + dispatched = AsyncMock() + + async def send(*_args: object, **kwargs: object) -> int: + checkpoint_quote = kwargs["on_melt_quote"] + await checkpoint_quote("quote-1") # type: ignore[operator] + await dispatched() + return 5 with ( patch("routstr.auth.ROUTSTR_FEE_DEFAULT_PAYOUT", 1), @@ -169,24 +277,30 @@ async def test_fee_payout_lost_checkpoint_race_does_not_send() -> None: ), 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", send), + patch("routstr.wallet.raw_send_to_lnurl", side_effect=send), patch("routstr.wallet.logger.warning") as warning, ): with pytest.raises(asyncio.CancelledError): await wallet.periodic_routstr_fee_payout() - send.assert_not_awaited() + dispatched.assert_not_awaited() warning.assert_called_once_with("Routstr fee payout was already claimed") @pytest.mark.asyncio -async def test_fee_payout_does_not_retry_an_unresolved_checkpoint() -> None: +async def test_fee_payout_finalizes_a_paid_unresolved_quote_without_resending() -> None: session = Mock() fee = SimpleNamespace( accumulated_msats=10_000, payout_in_progress_msats=5_000, payout_started_at=123, + payout_quote_id="quote-1", + payout_mint_url="https://mint.test", + payout_unit="sat", ) + complete = AsyncMock(return_value=True) + restore = AsyncMock() + send = AsyncMock() with ( patch("routstr.auth.ROUTSTR_FEE_PAYOUT_INTERVAL_SECONDS", 1), @@ -199,17 +313,215 @@ async def test_fee_payout_does_not_retry_an_unresolved_checkpoint() -> None: "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()) as checkpoint, - patch("routstr.wallet.get_wallet", AsyncMock()) as get_wallet, - patch("routstr.wallet.raw_send_to_lnurl", AsyncMock()) as send, + patch("routstr.wallet.db.complete_routstr_fee_payout", complete), + patch("routstr.wallet.db.restore_routstr_fee_payout", restore), + patch( + "routstr.wallet._check_bolt11_payment_status_locked", + AsyncMock(return_value="paid"), + ) as status, + patch("routstr.wallet.raw_send_to_lnurl", send), + ): + with pytest.raises(asyncio.CancelledError): + await wallet.periodic_routstr_fee_payout() + + status.assert_awaited_once_with("https://mint.test", "sat", "quote-1") + complete.assert_awaited_once_with( + session, 5_000, "quote-1", "https://mint.test", "sat" + ) + restore.assert_not_awaited() + send.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_fee_payout_restores_only_an_unpaid_quote_and_retries() -> None: + session = Mock() + unresolved_fee = SimpleNamespace( + accumulated_msats=10_000, + payout_in_progress_msats=5_000, + payout_started_at=123, + payout_quote_id="quote-1", + payout_mint_url="https://mint.test", + payout_unit="sat", + ) + restored_fee = SimpleNamespace( + accumulated_msats=15_000, + payout_in_progress_msats=0, + payout_started_at=None, + ) + + async def send(*_args: object, **kwargs: object) -> int: + await kwargs["on_melt_quote"]("quote-2") # type: ignore[index,operator] + return 15 + + 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(side_effect=[None, None, asyncio.CancelledError()]), + ), + patch( + "routstr.wallet.db.create_session", return_value=_session_context(session) + ), + patch( + "routstr.wallet.db.get_routstr_fee", + AsyncMock(side_effect=[unresolved_fee, unresolved_fee, restored_fee]), + ), + patch( + "routstr.wallet._check_bolt11_payment_status_locked", + AsyncMock(return_value="unpaid"), + ), + patch( + "routstr.wallet.db.restore_routstr_fee_payout", + AsyncMock(return_value=True), + ) as restore, + patch( + "routstr.wallet.db.reset_routstr_fee", AsyncMock(return_value=True) + ) as reset, + patch( + "routstr.wallet.db.complete_routstr_fee_payout", + AsyncMock(return_value=True), + ), + 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", side_effect=send) as raw_send, + ): + with pytest.raises(asyncio.CancelledError): + await wallet.periodic_routstr_fee_payout() + + restore.assert_awaited_once_with( + session, 5_000, "quote-1", "https://mint.test", "sat" + ) + reset.assert_awaited_once_with( + session, 15_000, "quote-2", wallet.settings.primary_mint, "sat" + ) + raw_send.assert_awaited_once() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("quote_state", ["pending", "unknown"]) +async def test_fee_payout_keeps_nonfinal_quote_locked(quote_state: str) -> None: + session = Mock() + fee = SimpleNamespace( + accumulated_msats=10_000, + payout_in_progress_msats=5_000, + payout_started_at=123, + payout_quote_id="quote-1", + payout_mint_url="https://mint.test", + payout_unit="sat", + ) + complete = AsyncMock() + restore = AsyncMock() + + with ( + patch("routstr.auth.ROUTSTR_FEE_PAYOUT_INTERVAL_SECONDS", 1), + patch("routstr.auth.ROUTSTR_LN_ADDRESS", "fees@example.com"), + patch( + "routstr.wallet.asyncio.sleep", + AsyncMock(side_effect=[None, asyncio.CancelledError()]), + ), + 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.complete_routstr_fee_payout", complete), + patch("routstr.wallet.db.restore_routstr_fee_payout", restore), + patch( + "routstr.wallet._check_bolt11_payment_status_locked", + AsyncMock(return_value=quote_state), + ), + ): + with pytest.raises(asyncio.CancelledError): + await wallet.periodic_routstr_fee_payout() + + complete.assert_not_awaited() + restore.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_fee_payout_reconciliation_rechecks_state_under_wallet_guard() -> None: + session = Mock() + fee = SimpleNamespace( + accumulated_msats=10_000, + payout_in_progress_msats=5_000, + payout_started_at=123, + payout_quote_id="quote-1", + payout_mint_url="https://mint.test", + payout_unit="sat", + ) + guard_held = False + + @asynccontextmanager + async def guard() -> AsyncGenerator[None, None]: + nonlocal guard_held + assert not guard_held + guard_held = True + try: + yield + finally: + guard_held = False + + async def status(*_args: object) -> str: + assert guard_held + return "pending" + + get_fee = AsyncMock(return_value=fee) + with ( + patch("routstr.auth.ROUTSTR_FEE_PAYOUT_INTERVAL_SECONDS", 1), + patch("routstr.auth.ROUTSTR_LN_ADDRESS", "fees@example.com"), + patch( + "routstr.wallet.asyncio.sleep", + AsyncMock(side_effect=[None, asyncio.CancelledError()]), + ), + patch( + "routstr.wallet.db.create_session", return_value=_session_context(session) + ), + patch("routstr.wallet.db.get_routstr_fee", get_fee), + patch("routstr.wallet.wallet_operation_guard", guard), + patch( + "routstr.wallet._check_bolt11_payment_status_locked", + side_effect=status, + ), + ): + with pytest.raises(asyncio.CancelledError): + await wallet.periodic_routstr_fee_payout() + + assert get_fee.await_count == 2 + assert not guard_held + + +@pytest.mark.asyncio +async def test_fee_payout_keeps_legacy_checkpoint_without_quote_locked() -> None: + session = Mock() + fee = SimpleNamespace( + accumulated_msats=10_000, + payout_in_progress_msats=5_000, + payout_started_at=123, + payout_quote_id=None, + payout_mint_url=None, + payout_unit=None, + ) + restore = AsyncMock() + + with ( + patch("routstr.auth.ROUTSTR_FEE_PAYOUT_INTERVAL_SECONDS", 1), + patch("routstr.auth.ROUTSTR_LN_ADDRESS", "fees@example.com"), + patch( + "routstr.wallet.asyncio.sleep", + AsyncMock(side_effect=[None, asyncio.CancelledError()]), + ), + 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.restore_routstr_fee_payout", restore), patch("routstr.wallet.logger.critical") as critical, ): with pytest.raises(asyncio.CancelledError): await wallet.periodic_routstr_fee_payout() - checkpoint.assert_not_awaited() - get_wallet.assert_not_awaited() - send.assert_not_awaited() + restore.assert_not_awaited() critical.assert_called_once() @@ -223,6 +535,11 @@ async def test_fee_payout_keeps_checkpoint_when_send_outcome_is_unknown() -> Non ) complete = AsyncMock() + async def send(*_args: object, **kwargs: object) -> int: + checkpoint_quote = kwargs["on_melt_quote"] + await checkpoint_quote("quote-1") # type: ignore[operator] + raise TimeoutError("unknown outcome") + with ( patch("routstr.auth.ROUTSTR_FEE_DEFAULT_PAYOUT", 1), patch("routstr.auth.ROUTSTR_FEE_PAYOUT_INTERVAL_SECONDS", 1), @@ -239,10 +556,7 @@ async def test_fee_payout_keeps_checkpoint_when_send_outcome_is_unknown() -> Non 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=TimeoutError("unknown outcome")), - ), + patch("routstr.wallet.raw_send_to_lnurl", side_effect=send), patch("routstr.wallet.logger.critical") as critical, ): with pytest.raises(asyncio.CancelledError): @@ -262,6 +576,11 @@ async def test_fee_payout_cancellation_during_send_alerts_and_propagates() -> No ) complete = AsyncMock() + async def cancel_send(*_args: object, **kwargs: object) -> int: + checkpoint_quote = kwargs["on_melt_quote"] + await checkpoint_quote("quote-1") # type: ignore[operator] + raise asyncio.CancelledError + with ( patch("routstr.auth.ROUTSTR_FEE_DEFAULT_PAYOUT", 1), patch("routstr.auth.ROUTSTR_FEE_PAYOUT_INTERVAL_SECONDS", 1), @@ -275,10 +594,7 @@ async def test_fee_payout_cancellation_during_send_alerts_and_propagates() -> No 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.raw_send_to_lnurl", side_effect=cancel_send), patch("routstr.wallet.logger.critical") as critical, ): with pytest.raises(asyncio.CancelledError): @@ -287,7 +603,7 @@ async def test_fee_payout_cancellation_during_send_alerts_and_propagates() -> No complete.assert_not_awaited() critical.assert_called_once() assert critical.call_args.args[0] == ( - "Routstr fee payout outcome is unknown; manual reconciliation required" + "Routstr fee payout outcome is unknown; awaiting quote reconciliation" ) @@ -315,6 +631,11 @@ async def test_fee_payout_completion_failures_use_sent_checkpoint_alert( create_session = Mock(return_value=_session_context(session)) completion.side_effect = RuntimeError("checkpoint unavailable") + async def send(*_args: object, **kwargs: object) -> int: + checkpoint_quote = kwargs["on_melt_quote"] + await checkpoint_quote("quote-1") # type: ignore[operator] + return 5 + with ( patch("routstr.auth.ROUTSTR_FEE_DEFAULT_PAYOUT", 1), patch("routstr.auth.ROUTSTR_FEE_PAYOUT_INTERVAL_SECONDS", 1), @@ -329,7 +650,7 @@ async def test_fee_payout_completion_failures_use_sent_checkpoint_alert( patch("routstr.wallet.db.complete_routstr_fee_payout", completion), 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(return_value=5)), + patch("routstr.wallet.raw_send_to_lnurl", side_effect=send), patch("routstr.wallet.logger.critical") as critical, ): with pytest.raises(asyncio.CancelledError): @@ -337,7 +658,8 @@ async def test_fee_payout_completion_failures_use_sent_checkpoint_alert( critical.assert_called_once() assert critical.call_args.args[0] == ( - "Routstr fee payout sent but checkpoint was not completed" + "Routstr fee payout sent but checkpoint was not completed; " + "awaiting quote reconciliation" ) @@ -359,7 +681,10 @@ async def test_fee_payout_releases_db_connection_during_send(tmp_path: object) - async with AsyncSession(engine, expire_on_commit=False) as session: yield session - async def send(*_args: object, **_kwargs: object) -> int: + async def send(*_args: object, **kwargs: object) -> int: + assert engine.pool.checkedout() == 0 # type: ignore[attr-defined] + checkpoint_quote = kwargs["on_melt_quote"] + await checkpoint_quote("quote-1") # type: ignore[operator] assert engine.pool.checkedout() == 0 # type: ignore[attr-defined] return 5 diff --git a/tests/unit/test_fee_payout_migration.py b/tests/unit/test_fee_payout_migration.py index 17be72ec..08d33c40 100644 --- a/tests/unit/test_fee_payout_migration.py +++ b/tests/unit/test_fee_payout_migration.py @@ -37,7 +37,8 @@ def test_fresh_node_migrates_fee_payout_schema_to_head(tmp_path: Path) -> None: } fee = connection.execute( "SELECT id, accumulated_msats, total_paid_msats, last_paid_at, " - "payout_in_progress_msats, payout_started_at FROM routstr_fees" + "payout_in_progress_msats, payout_started_at, payout_quote_id, " + "payout_mint_url, payout_unit FROM routstr_fees" ).fetchone() migration_config = Config(str(root / "alembic.ini")) @@ -51,8 +52,11 @@ def test_fresh_node_migrates_fee_payout_schema_to_head(tmp_path: Path) -> None: "last_paid_at", "payout_in_progress_msats", "payout_started_at", + "payout_quote_id", + "payout_mint_url", + "payout_unit", } <= columns - assert fee == (1, 0, 0, None, 0, None) + assert fee == (1, 0, 0, None, 0, None, None, None, None) def test_fee_payout_checkpoint_migration_preserves_existing_row( @@ -76,11 +80,11 @@ def test_fee_payout_checkpoint_migration_preserves_existing_row( with sqlite3.connect(database_path) as connection: row = connection.execute( "SELECT accumulated_msats, total_paid_msats, last_paid_at, " - "payout_in_progress_msats, payout_started_at " - "FROM routstr_fees WHERE id = 1" + "payout_in_progress_msats, payout_started_at, payout_quote_id, " + "payout_mint_url, payout_unit FROM routstr_fees WHERE id = 1" ).fetchone() - assert row == (5000, 1000, 123, 0, None) + assert row == (5000, 1000, 123, 0, None, None, None, None) def test_fee_payout_checkpoint_repair_restores_columns_missing_at_old_head( diff --git a/tests/unit/test_lnurl_melt_timeout.py b/tests/unit/test_lnurl_melt_timeout.py index b5567570..a370a7a8 100644 --- a/tests/unit/test_lnurl_melt_timeout.py +++ b/tests/unit/test_lnurl_melt_timeout.py @@ -77,9 +77,7 @@ async def test_raw_send_to_lnurl_timeout_reconciled_paid_is_success() -> None: await asyncio.sleep(5) wallet.melt = AsyncMock(side_effect=_hang) - wallet.get_melt_quote = AsyncMock( - return_value=MagicMock(state=MeltQuoteState.paid) - ) + wallet.get_melt_quote = AsyncMock(return_value=MagicMock(state=MeltQuoteState.paid)) data_patch, invoice_patch = _lnurl_patches() with ( @@ -146,14 +144,10 @@ async def test_raw_send_to_lnurl_rate_rejection_unreserves_proofs( ), pytest.raises((MintCooldownError, httpx.HTTPStatusError)), ): - await raw_send_to_lnurl( - wallet, proofs, "owner@ln.tld", "sat", amount=1000 - ) + await raw_send_to_lnurl(wallet, proofs, "owner@ln.tld", "sat", amount=1000) wallet.melt.assert_not_awaited() - wallet.set_reserved_for_send.assert_awaited_once_with( - proofs, reserved=False - ) + wallet.set_reserved_for_send.assert_awaited_once_with(proofs, reserved=False) @pytest.mark.asyncio @@ -175,17 +169,42 @@ async def test_real_mint_wrapper_http_429_unreserves_proofs() -> None: invoice_patch, pytest.raises(httpx.HTTPStatusError), ): - await raw_send_to_lnurl( - wallet, proofs, "owner@ln.tld", "sat", amount=1000 - ) + await raw_send_to_lnurl(wallet, proofs, "owner@ln.tld", "sat", amount=1000) wallet.melt.assert_awaited_once() - wallet.set_reserved_for_send.assert_awaited_once_with( - proofs, reserved=False - ) + wallet.set_reserved_for_send.assert_awaited_once_with(proofs, reserved=False) MintRateGuard._guards.pop(str(wallet.url), None) +@pytest.mark.asyncio +async def test_raw_send_to_lnurl_checkpoints_quote_before_melt_dispatch() -> None: + wallet, proofs = _wallet() + events: list[str] = [] + + async def checkpoint(quote_id: str) -> None: + assert quote_id == "q" + events.append("checkpoint") + + async def melt(**_kwargs: object) -> MagicMock: + events.append("melt") + return MagicMock(state=MeltQuoteState.paid) + + wallet.melt = AsyncMock(side_effect=melt) + data_patch, invoice_patch = _lnurl_patches() + + with data_patch, invoice_patch: + await raw_send_to_lnurl( + wallet, + proofs, + "owner@ln.tld", + "sat", + amount=1000, + on_melt_quote=checkpoint, + ) + + assert events == ["checkpoint", "melt"] + + @pytest.mark.asyncio async def test_raw_send_to_lnurl_succeeds_on_explicit_paid_response() -> None: wallet, proofs = _wallet()