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: reconcile fee payouts safely

Browse Source
This commit is contained in:
9qeklajc
2026-08-11 01:32:41 +02:00
parent 2e04cf5925
commit 816e07332d
7 changed files with 613 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
migrations/versions/b4f7a1c9d2e3_add_fee_payout_reconciliation_metadata.py
+37
View File
@@ -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")
routstr/core/db.py
+60 -8
View File
@@ -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()),
)
routstr/payment/lnurl.py
+7 -2
View File
@@ -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
routstr/wallet.py
+115 -32
View File
@@ -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
tests/unit/test_fee_payout_crash_safety.py
+351 -26
View File
@@ -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
tests/unit/test_fee_payout_migration.py
+9 -5
View File
@@ -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(
tests/unit/test_lnurl_melt_timeout.py
+34 -15
View File
@@ -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()