mirror of
https://github.com/Routstr/routstr-core.git
synced 2026-10-05 12:28:22 +00:00
fix: harden Lightning payout history against public endpoints, DB failures and stale rows
This commit is contained in:
@@ -1050,6 +1050,10 @@ async def settle_lightning_payout(
|
||||
)
|
||||
payout = result.first()
|
||||
if payout is None:
|
||||
logger.warning(
|
||||
"No Lightning payout history row for quote",
|
||||
extra={"quote_id": quote_id, "status": status},
|
||||
)
|
||||
return
|
||||
payout.status = status
|
||||
if status == "paid":
|
||||
@@ -1060,6 +1064,24 @@ async def settle_lightning_payout(
|
||||
await session.commit()
|
||||
|
||||
|
||||
UNSETTLED_PAYOUT_STATUSES = ("pending", "reconciliation_required")
|
||||
|
||||
|
||||
async def list_unsettled_lightning_payouts(
|
||||
session: AsyncSession, mint_url: str, *, created_before: int
|
||||
) -> list[LightningInvoice]:
|
||||
"""Payout rows whose mint outcome was never written back to history."""
|
||||
result = await session.exec(
|
||||
select(LightningInvoice)
|
||||
.where(col(LightningInvoice.direction) == "out")
|
||||
.where(col(LightningInvoice.mint_url) == mint_url)
|
||||
.where(col(LightningInvoice.status).in_(UNSETTLED_PAYOUT_STATUSES))
|
||||
.where(col(LightningInvoice.created_at) < created_before)
|
||||
.order_by(col(LightningInvoice.created_at))
|
||||
)
|
||||
return list(result.all())
|
||||
|
||||
|
||||
async def total_user_liability(db_session: AsyncSession) -> int:
|
||||
"""Return all outstanding user funds in millisatoshis.
|
||||
|
||||
|
||||
@@ -443,7 +443,8 @@ async def get_invoice_status(
|
||||
structured_errors: bool = Depends(_uses_v2_errors),
|
||||
) -> InvoiceStatusResponse:
|
||||
invoice = await session.get(LightningInvoice, invoice_id)
|
||||
if not invoice:
|
||||
# Payout rows (direction="out") are operator history, never user invoices.
|
||||
if not invoice or invoice.direction != "in":
|
||||
raise _invoice_error(
|
||||
404,
|
||||
"Invoice not found",
|
||||
@@ -486,7 +487,9 @@ async def recover_invoice(
|
||||
structured_errors: bool = Depends(_uses_v2_errors),
|
||||
) -> InvoiceStatusResponse:
|
||||
result = await session.exec(
|
||||
select(LightningInvoice).where(LightningInvoice.bolt11 == request.bolt11)
|
||||
select(LightningInvoice)
|
||||
.where(LightningInvoice.bolt11 == request.bolt11)
|
||||
.where(col(LightningInvoice.direction) == "in")
|
||||
)
|
||||
invoice = result.first()
|
||||
|
||||
|
||||
+97
-28
@@ -1540,6 +1540,81 @@ async def fetch_all_balances(
|
||||
)
|
||||
|
||||
|
||||
PAYOUT_HISTORY_STALE_SECONDS = 600
|
||||
|
||||
|
||||
async def _record_payout_history(
|
||||
*,
|
||||
quote_id: str,
|
||||
bolt11: str,
|
||||
amount_sats: int,
|
||||
mint_url: str,
|
||||
destination: str,
|
||||
) -> None:
|
||||
"""Best-effort history insert; a history failure must never block a payout."""
|
||||
try:
|
||||
async with db.create_session() as session:
|
||||
await db.record_lightning_payout(
|
||||
session,
|
||||
quote_id=quote_id,
|
||||
bolt11=bolt11,
|
||||
amount_sats=amount_sats,
|
||||
mint_url=mint_url,
|
||||
destination=destination,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"Failed to record Lightning payout history",
|
||||
extra={
|
||||
"error": str(e),
|
||||
"error_type": type(e).__name__,
|
||||
"quote_id": quote_id,
|
||||
"mint_url": mint_url,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
async def _reconcile_stale_payout_history(mint_url: str, unit: str) -> None:
|
||||
"""Resolve payout rows left pending by a crash or an ambiguous melt.
|
||||
|
||||
Runs under ``wallet_operation_guard``. Only writes what the mint asserts
|
||||
(paid/unpaid); quotes still pending or unreachable are left for later.
|
||||
"""
|
||||
try:
|
||||
cutoff = int(time.time()) - PAYOUT_HISTORY_STALE_SECONDS
|
||||
async with db.create_session() as session:
|
||||
stale = await db.list_unsettled_lightning_payouts(
|
||||
session, mint_url, created_before=cutoff
|
||||
)
|
||||
for payout in stale:
|
||||
quote_state = await _check_bolt11_payment_status_locked(
|
||||
mint_url, unit, payout.payment_hash
|
||||
)
|
||||
if quote_state == "paid":
|
||||
await _settle_payout_history(payout.payment_hash, status="paid")
|
||||
elif quote_state == "unpaid":
|
||||
await _settle_payout_history(payout.payment_hash, status="failed")
|
||||
else:
|
||||
continue
|
||||
logger.info(
|
||||
"Reconciled stale Lightning payout history",
|
||||
extra={
|
||||
"quote_id": payout.payment_hash,
|
||||
"mint_url": mint_url,
|
||||
"quote_state": quote_state,
|
||||
},
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"Failed to reconcile Lightning payout history",
|
||||
extra={
|
||||
"error": str(e),
|
||||
"error_type": type(e).__name__,
|
||||
"mint_url": mint_url,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
async def _settle_payout_history(
|
||||
quote_id: str, *, status: str, amount_sats: int | None = None
|
||||
) -> None:
|
||||
@@ -1614,19 +1689,17 @@ async def _payout_mint_and_unit(mint_url: str, unit: str) -> None:
|
||||
async def record_payout(quote_id: str, bolt11: str) -> None:
|
||||
nonlocal payout_quote_id
|
||||
payout_quote_id = quote_id
|
||||
async with db.create_session() as session:
|
||||
await db.record_lightning_payout(
|
||||
session,
|
||||
quote_id=quote_id,
|
||||
bolt11=bolt11,
|
||||
amount_sats=(
|
||||
available_balance
|
||||
if unit == "sat"
|
||||
else _msats_to_sats(available_balance)
|
||||
),
|
||||
mint_url=mint_url,
|
||||
destination=settings.receive_ln_address,
|
||||
)
|
||||
await _record_payout_history(
|
||||
quote_id=quote_id,
|
||||
bolt11=bolt11,
|
||||
amount_sats=(
|
||||
available_balance
|
||||
if unit == "sat"
|
||||
else _msats_to_sats(available_balance)
|
||||
),
|
||||
mint_url=mint_url,
|
||||
destination=settings.receive_ln_address,
|
||||
)
|
||||
|
||||
try:
|
||||
amount_received = await raw_send_to_lnurl(
|
||||
@@ -1699,6 +1772,7 @@ async def periodic_payout() -> None:
|
||||
# Proof mutation, liability observation, and sending are one
|
||||
# cross-process critical section. Credits take the same lock.
|
||||
async with wallet_operation_guard():
|
||||
await _reconcile_stale_payout_history(mint_url, unit)
|
||||
await _payout_mint_and_unit(mint_url, unit)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
@@ -1986,15 +2060,13 @@ async def periodic_routstr_fee_payout() -> None:
|
||||
if not checkpointed:
|
||||
raise _RoutstrFeePayoutAlreadyClaimed
|
||||
attempt_quote_id = quote_id
|
||||
async with db.create_session() as session:
|
||||
await db.record_lightning_payout(
|
||||
session,
|
||||
quote_id=quote_id,
|
||||
bolt11=bolt11,
|
||||
amount_sats=accumulated_sats,
|
||||
mint_url=settings.primary_mint,
|
||||
destination=ROUTSTR_LN_ADDRESS,
|
||||
)
|
||||
await _record_payout_history(
|
||||
quote_id=quote_id,
|
||||
bolt11=bolt11,
|
||||
amount_sats=accumulated_sats,
|
||||
mint_url=settings.primary_mint,
|
||||
destination=ROUTSTR_LN_ADDRESS,
|
||||
)
|
||||
|
||||
try:
|
||||
amount_received = await raw_send_to_lnurl(
|
||||
@@ -2021,12 +2093,9 @@ async def periodic_routstr_fee_payout() -> None:
|
||||
extra={"payout_in_progress_msats": paid_msats},
|
||||
exc_info=isinstance(e, Exception),
|
||||
)
|
||||
async with db.create_session() as session:
|
||||
await db.settle_lightning_payout(
|
||||
session,
|
||||
attempt_quote_id,
|
||||
status="reconciliation_required",
|
||||
)
|
||||
await _settle_payout_history(
|
||||
attempt_quote_id, status="reconciliation_required"
|
||||
)
|
||||
if not isinstance(e, Exception):
|
||||
raise
|
||||
continue
|
||||
|
||||
@@ -5,6 +5,7 @@ from unittest.mock import AsyncMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
from cashu.core.base import Proof
|
||||
from fastapi import HTTPException
|
||||
from sqlalchemy.ext.asyncio import AsyncEngine
|
||||
from sqlmodel import col, update
|
||||
from sqlmodel.ext.asyncio.session import AsyncSession
|
||||
@@ -13,12 +14,15 @@ from routstr.core.db import ApiKey, LightningInvoice
|
||||
from routstr.lightning import (
|
||||
INVOICE_EXPIRY_GRACE_SECONDS,
|
||||
INVOICE_WATCH_BATCH_LIMIT,
|
||||
InvoiceRecoverRequest,
|
||||
_expire_invoice_if_authoritatively_unpaid,
|
||||
_expire_overdue_invoices,
|
||||
_finalize_invoice_settlement,
|
||||
_InvoiceSettlement,
|
||||
_process_invoice_watch_batch,
|
||||
check_invoice_payment,
|
||||
get_invoice_status,
|
||||
recover_invoice,
|
||||
)
|
||||
|
||||
|
||||
@@ -621,3 +625,34 @@ async def test_recovery_tail_cannot_starve_owed_or_live_invoices(
|
||||
assert len(polled) == INVOICE_WATCH_BATCH_LIMIT
|
||||
assert {inv.id for inv in settling} <= set(polled)
|
||||
assert {inv.id for inv in fresh} <= set(polled)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_public_invoice_endpoints_ignore_payout_rows(
|
||||
integration_engine: AsyncEngine,
|
||||
patched_db_engine: None,
|
||||
) -> None:
|
||||
"""A payout's bolt11/id must not let /recover or /status touch the row."""
|
||||
payout = _lightning_invoice(
|
||||
direction="out",
|
||||
purpose="payout",
|
||||
expires_at=int(time.time()) - 1,
|
||||
)
|
||||
async with AsyncSession(integration_engine, expire_on_commit=False) as seed:
|
||||
seed.add(payout)
|
||||
await seed.commit()
|
||||
|
||||
async with AsyncSession(integration_engine, expire_on_commit=False) as session:
|
||||
with pytest.raises(HTTPException) as recover_error:
|
||||
await recover_invoice(
|
||||
InvoiceRecoverRequest(bolt11=payout.bolt11), session, False
|
||||
)
|
||||
with pytest.raises(HTTPException) as status_error:
|
||||
await get_invoice_status(payout.id, session, False)
|
||||
assert recover_error.value.status_code == 404
|
||||
assert status_error.value.status_code == 404
|
||||
|
||||
async with AsyncSession(integration_engine, expire_on_commit=False) as verify:
|
||||
stored = await verify.get(LightningInvoice, payout.id)
|
||||
assert stored is not None
|
||||
assert stored.status == "pending"
|
||||
|
||||
@@ -37,6 +37,7 @@ def _invoice(**overrides: object) -> SimpleNamespace:
|
||||
"payment_hash": "quote-1",
|
||||
"amount_sats": 100,
|
||||
"purpose": "create",
|
||||
"direction": "in",
|
||||
"status": "pending",
|
||||
"paid_at": None,
|
||||
"api_key_hash": None,
|
||||
|
||||
@@ -17,7 +17,11 @@ from unittest.mock import ANY, AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from routstr.wallet import _payout_units, periodic_payout
|
||||
from routstr.wallet import (
|
||||
_payout_units,
|
||||
_reconcile_stale_payout_history,
|
||||
periodic_payout,
|
||||
)
|
||||
|
||||
# Sentinel interval used to break the otherwise-infinite payout loop after
|
||||
# exactly one full cycle.
|
||||
@@ -262,11 +266,12 @@ async def test_periodic_payout_handles_session_creation_failure() -> None:
|
||||
with pytest.raises(_LoopBreak):
|
||||
await periodic_payout()
|
||||
|
||||
# The liability session is opened per mint/unit (sat + msat), and each
|
||||
# DB failure retains the cycle-specific alert wording while remaining
|
||||
# isolated to its own iteration.
|
||||
assert create_session.call_count == 2
|
||||
assert logger.error.call_count == 2
|
||||
# Per mint/unit (sat + msat) a session is opened twice: once by the stale
|
||||
# payout-history sweep and once for the liability read. Each DB failure is
|
||||
# logged and isolated to its own step; the liability error keeps the
|
||||
# cycle-specific alert wording.
|
||||
assert create_session.call_count == 4
|
||||
assert logger.error.call_count == 4
|
||||
message = logger.error.call_args.args[0]
|
||||
extra = logger.error.call_args.kwargs["extra"]
|
||||
assert message == "Error in periodic payout cycle: RuntimeError"
|
||||
@@ -280,3 +285,101 @@ async def test_payout_units_excludes_units_the_sender_cannot_pay() -> None:
|
||||
AsyncMock(return_value=["usd", "sat", "eur", "msat"]),
|
||||
):
|
||||
assert await _payout_units("http://mint:3338") == ["sat", "msat"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_payout_history_write_failure_does_not_block_payout() -> None:
|
||||
"""A failing history insert is logged; the melt and settlement still run."""
|
||||
from routstr.core.settings import settings
|
||||
|
||||
get_wallet = AsyncMock(return_value=MagicMock())
|
||||
record_payout = AsyncMock(side_effect=RuntimeError("database is locked"))
|
||||
settle_payout = AsyncMock()
|
||||
logger = MagicMock()
|
||||
|
||||
async def send(*args: object, **kwargs: object) -> int:
|
||||
await kwargs["on_melt_quote"]( # type: ignore[index,operator]
|
||||
"quote-1", "lnbc1payout"
|
||||
)
|
||||
return 1_000_000
|
||||
|
||||
raw_send = AsyncMock(side_effect=send)
|
||||
|
||||
with (
|
||||
patch.object(settings, "cashu_mints", []),
|
||||
patch.object(settings, "primary_mint", "http://primary:3338"),
|
||||
patch.object(settings, "receive_ln_address", "owner@ln.tld"),
|
||||
patch.object(settings, "payout_interval_seconds", _INTERVAL),
|
||||
patch.object(settings, "min_payout_sat", 10),
|
||||
patch("routstr.wallet.asyncio.sleep", _one_cycle_sleep()),
|
||||
patch("routstr.wallet.db.create_session", _fake_session),
|
||||
patch(
|
||||
"routstr.wallet._get_supported_mint_units",
|
||||
AsyncMock(return_value=["sat"]),
|
||||
),
|
||||
patch("routstr.wallet.get_wallet", get_wallet),
|
||||
patch(
|
||||
"routstr.wallet.get_proofs_per_mint_and_unit",
|
||||
MagicMock(return_value=[MagicMock(amount=100_000)]),
|
||||
),
|
||||
patch(
|
||||
"routstr.wallet.slow_filter_spend_proofs",
|
||||
AsyncMock(side_effect=lambda proofs, wallet: proofs),
|
||||
),
|
||||
patch("routstr.wallet.db.total_user_liability", AsyncMock(return_value=0)),
|
||||
patch(
|
||||
"routstr.wallet.db.list_unsettled_lightning_payouts",
|
||||
AsyncMock(return_value=[]),
|
||||
),
|
||||
patch("routstr.wallet.db.record_lightning_payout", record_payout),
|
||||
patch("routstr.wallet.db.settle_lightning_payout", settle_payout),
|
||||
patch("routstr.wallet.raw_send_to_lnurl", raw_send),
|
||||
patch("routstr.wallet.logger", logger),
|
||||
):
|
||||
with pytest.raises(_LoopBreak):
|
||||
await periodic_payout()
|
||||
|
||||
record_payout.assert_awaited_once()
|
||||
assert raw_send.await_count == 1
|
||||
settle_payout.assert_awaited_once_with(
|
||||
ANY, "quote-1", status="paid", amount_sats=1_000
|
||||
)
|
||||
messages = [call.args[0] for call in logger.error.call_args_list]
|
||||
assert "Failed to record Lightning payout history" in messages
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stale_payout_history_is_reconciled_from_mint_state() -> None:
|
||||
"""Stale out-rows follow the mint's verdict; pending/unknown are left alone."""
|
||||
stale = [
|
||||
MagicMock(payment_hash="q-paid"),
|
||||
MagicMock(payment_hash="q-unpaid"),
|
||||
MagicMock(payment_hash="q-pending"),
|
||||
MagicMock(payment_hash="q-unknown"),
|
||||
]
|
||||
states = {
|
||||
"q-paid": "paid",
|
||||
"q-unpaid": "unpaid",
|
||||
"q-pending": "pending",
|
||||
"q-unknown": "unknown",
|
||||
}
|
||||
settle_payout = AsyncMock()
|
||||
|
||||
async def _state(_mint: str, _unit: str, quote_id: str) -> str:
|
||||
return states[quote_id]
|
||||
|
||||
with (
|
||||
patch("routstr.wallet.db.create_session", _fake_session),
|
||||
patch(
|
||||
"routstr.wallet.db.list_unsettled_lightning_payouts",
|
||||
AsyncMock(return_value=stale),
|
||||
),
|
||||
patch("routstr.wallet._check_bolt11_payment_status_locked", _state),
|
||||
patch("routstr.wallet.db.settle_lightning_payout", settle_payout),
|
||||
):
|
||||
await _reconcile_stale_payout_history("http://mint:3338", "sat")
|
||||
|
||||
assert settle_payout.await_args_list == [
|
||||
((ANY, "q-paid"), {"status": "paid", "amount_sats": None}),
|
||||
((ANY, "q-unpaid"), {"status": "failed", "amount_sats": None}),
|
||||
]
|
||||
|
||||
@@ -757,7 +757,7 @@ export default function TransactionsPage() {
|
||||
<SelectContent>
|
||||
<SelectItem value='all'>All Types</SelectItem>
|
||||
<SelectItem value='in'>Incoming (Payments)</SelectItem>
|
||||
<SelectItem value='out'>Outgoing (Refunds)</SelectItem>
|
||||
<SelectItem value='out'>Outgoing (Refunds & Payouts)</SelectItem>
|
||||
</SelectContent>
|
||||
</Select>
|
||||
</div>
|
||||
|
||||
Reference in New Issue
Block a user