diff --git a/routstr/lightning.py b/routstr/lightning.py index 6ef85f3c..6b269847 100644 --- a/routstr/lightning.py +++ b/routstr/lightning.py @@ -783,17 +783,60 @@ async def _credit_topup_record( # quote, so polling faster just burns the global request budget for nothing. INVOICE_WATCH_INTERVAL_SECONDS = 10 INVOICE_WATCH_BATCH_LIMIT = 100 +INVOICE_WATCH_CANDIDATE_LIMIT = 500 +INVOICE_POLL_MAX_INTERVAL_SECONDS = 600 +SETTLEMENT_POLL_MAX_INTERVAL_SECONDS = 60 -async def _process_invoice_watch_batch(session: AsyncSession) -> None: - result = await session.exec( +def _invoice_poll_interval(age_seconds: int) -> int: + """Older quotes rarely settle, and the mint request budget is per IP.""" + if age_seconds < 60: + return 10 + if age_seconds < 300: + return 30 + if age_seconds < 1800: + return 120 + return INVOICE_POLL_MAX_INTERVAL_SECONDS + + +def _invoice_poll_due( + invoice: LightningInvoice, now: int, prev_now: int, max_interval: int +) -> bool: + """Whether the invoice's backoff interval elapsed between the two cycles.""" + if invoice.created_at > prev_now: + return True + age = now - invoice.created_at + prev_age = prev_now - invoice.created_at + interval = min(_invoice_poll_interval(age), max_interval) + return age // interval != prev_age // interval + + +async def _process_invoice_watch_batch(session: AsyncSession, prev_now: int) -> int: + now = int(time.time()) + # Already-paid rows still owe a credit, so they poll far more eagerly. + settling = await session.exec( select(LightningInvoice) - .where( - col(LightningInvoice.status).in_(_RETRYABLE_INVOICE_STATUSES) - ) - .limit(INVOICE_WATCH_BATCH_LIMIT) + .where(col(LightningInvoice.status) == "settlement_pending") + .order_by(col(LightningInvoice.created_at)) + .limit(INVOICE_WATCH_BATCH_LIMIT // 2) ) - for invoice in result.all(): + unpaid = await session.exec( + select(LightningInvoice) + .where(col(LightningInvoice.status) == "pending") + .order_by(col(LightningInvoice.created_at).desc()) + .limit(INVOICE_WATCH_CANDIDATE_LIMIT) + ) + due = [ + inv + for inv in settling.all() + if _invoice_poll_due(inv, now, prev_now, SETTLEMENT_POLL_MAX_INTERVAL_SECONDS) + ] + due += [ + inv + for inv in unpaid.all() + if _invoice_poll_due(inv, now, prev_now, INVOICE_POLL_MAX_INTERVAL_SECONDS) + ] + for invoice in due[:INVOICE_WATCH_BATCH_LIMIT]: try: definitively_unpaid = await check_invoice_payment(invoice, session) await _expire_invoice_if_authoritatively_unpaid( @@ -804,14 +847,16 @@ async def _process_invoice_watch_batch(session: AsyncSession) -> None: "Invoice watcher failed for invoice", extra={"invoice_id": invoice.id, "error": str(e)}, ) + return now async def periodic_invoice_watcher() -> None: """Background task: detect paid Lightning invoices and credit balances.""" + prev_now = int(time.time()) - INVOICE_WATCH_INTERVAL_SECONDS while True: try: async with create_session() as session: - await _process_invoice_watch_batch(session) + prev_now = await _process_invoice_watch_batch(session, prev_now) except asyncio.CancelledError: raise except Exception as e: diff --git a/routstr/wallet.py b/routstr/wallet.py index 383c79c8..f83ff8a6 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -2119,7 +2119,7 @@ async def fetch_all_balances( _balance_fetch_failures.pop(key, None) user_balance = user_balances.get((mint_url, unit), 0) if unit == "sat": - user_balance = _msats_to_sats(user_balance) + user_balance = _msats_to_sats_ceil(user_balance) proofs_balance = sum(proof.amount for proof in proofs) return { "mint_url": mint_url, @@ -2151,7 +2151,7 @@ async def fetch_all_balances( total_user_balance_sats += ( detail["user_balance"] if unit == "sat" - else _msats_to_sats(detail["user_balance"]) + else _msats_to_sats_ceil(detail["user_balance"]) ) if liabilities_error is None: @@ -2179,6 +2179,10 @@ async def _payout_mint_and_unit(mint_url: str, unit: str) -> None: # cross-process lock is only safe with a fresh reload. wallet = await get_wallet(mint_url, unit, force_reload=True) proofs = get_proofs_per_mint_and_unit(wallet, mint_url, unit, not_reserved=True) + if not proofs: + # Nothing to pay out, so skip the settle delay rather than hold the + # cross-process guard (and block credits) for a wallet with no funds. + return proofs = await slow_filter_spend_proofs(proofs, wallet) await asyncio.sleep(5) except Exception as e: @@ -2206,7 +2210,7 @@ async def _payout_mint_and_unit(mint_url: str, unit: str) -> None: try: if unit == "sat": - user_balance = _msats_to_sats(user_balance) + user_balance = _msats_to_sats_ceil(user_balance) proofs_balance = sum(proof.amount for proof in proofs) available_balance = proofs_balance - user_balance min_amount = ( @@ -2238,6 +2242,23 @@ async def _payout_mint_and_unit(mint_url: str, unit: str) -> None: ) +async def _payout_units(mint_url: str) -> list[str]: + """Only sat and msat are payable: raw_send_to_lnurl rejects other units.""" + try: + units = await _get_supported_mint_units(mint_url) + except Exception as e: + logger.warning( + "Unable to discover payout units, falling back to primary unit", + extra={ + "mint_url": mint_url, + "error": str(e), + "error_type": type(e).__name__, + }, + ) + units = [settings.primary_mint_unit] + return [unit for unit in units if unit in ("sat", "msat")] + + async def periodic_payout() -> None: while True: await asyncio.sleep(settings.payout_interval_seconds) @@ -2246,7 +2267,7 @@ async def periodic_payout() -> None: continue for mint_url in _mints_to_inspect(): - for unit in ["sat", "msat"]: + for unit in await _payout_units(mint_url): # Proof mutation, liability observation, and sending are one # cross-process critical section. Credits take the same lock. async with wallet_operation_guard(): diff --git a/tests/unit/test_invoice_watch_backoff.py b/tests/unit/test_invoice_watch_backoff.py new file mode 100644 index 00000000..ba1e2dc8 --- /dev/null +++ b/tests/unit/test_invoice_watch_backoff.py @@ -0,0 +1,96 @@ +"""Tests for invoice watcher poll backoff.""" + +from itertools import cycle, islice +from typing import cast +from unittest.mock import MagicMock + +from routstr.core.db import LightningInvoice +from routstr.lightning import ( + INVOICE_POLL_MAX_INTERVAL_SECONDS, + INVOICE_WATCH_INTERVAL_SECONDS, + SETTLEMENT_POLL_MAX_INTERVAL_SECONDS, + _invoice_poll_due, + _invoice_poll_interval, +) + + +def _invoice(created_at: int) -> LightningInvoice: + invoice = MagicMock() + invoice.created_at = created_at + return cast(LightningInvoice, invoice) + + +def test_poll_interval_backs_off_at_each_threshold() -> None: + assert _invoice_poll_interval(59) == 10 + assert _invoice_poll_interval(60) == 30 + assert _invoice_poll_interval(299) == 30 + assert _invoice_poll_interval(300) == 120 + assert _invoice_poll_interval(1799) == 120 + assert _invoice_poll_interval(1800) == 600 + + +def test_fresh_invoice_is_polled_every_cycle() -> None: + now = 1_000_000 + for age in range(0, 60, INVOICE_WATCH_INTERVAL_SECONDS): + prev_now = now - INVOICE_WATCH_INTERVAL_SECONDS + assert ( + _invoice_poll_due( + _invoice(now - age), now, prev_now, INVOICE_POLL_MAX_INTERVAL_SECONDS + ) + is True + ) + + +def test_future_created_at_is_polled_immediately() -> None: + now = 1_000_000 + assert ( + _invoice_poll_due( + _invoice(now + 500), now, now - 10, INVOICE_POLL_MAX_INTERVAL_SECONDS + ) + is True + ) + + +def test_settlement_pending_polls_at_its_capped_interval() -> None: + # Confirmed-paid rows owe a credit, so their backoff stops at 60s instead of + # decaying to the 600s bucket a day-old unpaid invoice would reach. + now = 100_000 + invoices = [_invoice(now - 86_400 - offset) for offset in range(600)] + due = sum( + 1 + for inv in invoices + if _invoice_poll_due(inv, now, now - 10, SETTLEMENT_POLL_MAX_INTERVAL_SECONDS) + ) + assert due == 100 + + +def test_aged_invoices_do_not_stampede_together() -> None: + now = 100_000 + invoices = [_invoice(now - 86_400 - offset) for offset in range(600)] + due = sum( + 1 + for inv in invoices + if _invoice_poll_due(inv, now, now - 10, INVOICE_POLL_MAX_INTERVAL_SECONDS) + ) + # One 600s bucket boundary falls inside each 10s cycle, so 600 day-old + # invoices cost 10 mint calls per cycle instead of 600. + assert due == 10 + + +def test_jittered_cycle_times_still_poll_every_invoice() -> None: + # A cycle is 10s of sleep plus however long the batch took, so `now` never + # advances on a clean 10s grid. + invoices = [_invoice(offset) for offset in range(600)] + polls = [0] * len(invoices) + start = 100_000 + now = start + for step in islice(cycle((13, 27)), 400): + prev_now, now = now, now + step + for index, invoice in enumerate(invoices): + if _invoice_poll_due( + invoice, now, prev_now, INVOICE_POLL_MAX_INTERVAL_SECONDS + ): + polls[index] += 1 + + elapsed = now - start + assert min(polls) >= elapsed // 600 diff --git a/tests/unit/test_periodic_payout.py b/tests/unit/test_periodic_payout.py index c869d7ca..fb105cd5 100644 --- a/tests/unit/test_periodic_payout.py +++ b/tests/unit/test_periodic_payout.py @@ -17,7 +17,7 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest -from routstr.wallet import periodic_payout +from routstr.wallet import _payout_units, periodic_payout # Sentinel interval used to break the otherwise-infinite payout loop after # exactly one full cycle. @@ -67,6 +67,10 @@ async def test_periodic_payout_includes_primary_mint_not_in_cashu_mints() -> Non 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", @@ -121,6 +125,10 @@ async def test_periodic_payout_releases_session_before_slow_mint_send() -> None: patch.object(settings, "min_payout_sat", 10), patch("routstr.wallet.asyncio.sleep", _one_cycle_sleep()), patch("routstr.wallet.db.create_session", tracked_session), + patch( + "routstr.wallet._get_supported_mint_units", + AsyncMock(return_value=["sat"]), + ), patch("routstr.wallet.get_wallet", AsyncMock(return_value=MagicMock())), patch( "routstr.wallet.get_proofs_per_mint_and_unit", @@ -139,7 +147,7 @@ async def test_periodic_payout_releases_session_before_slow_mint_send() -> None: with pytest.raises(_LoopBreak): await periodic_payout() - assert sends_completed == 2 + assert sends_completed == 1 @pytest.mark.asyncio @@ -165,6 +173,10 @@ async def test_periodic_payout_isolates_failing_mint() -> None: 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", "msat"]), + ), patch("routstr.wallet.get_wallet", get_wallet), patch( "routstr.wallet.get_proofs_per_mint_and_unit", @@ -207,6 +219,10 @@ async def test_periodic_payout_handles_session_creation_failure() -> None: patch.object(settings, "payout_interval_seconds", _INTERVAL), patch("routstr.wallet.asyncio.sleep", _one_cycle_sleep()), patch("routstr.wallet.db.create_session", create_session), + patch( + "routstr.wallet._get_supported_mint_units", + AsyncMock(return_value=["sat", "msat"]), + ), patch("routstr.wallet.get_wallet", AsyncMock(return_value=MagicMock())), patch( "routstr.wallet.get_proofs_per_mint_and_unit", @@ -230,3 +246,12 @@ async def test_periodic_payout_handles_session_creation_failure() -> None: extra = logger.error.call_args.kwargs["extra"] assert message == "Error in periodic payout cycle: RuntimeError" assert extra["error"] == "db unavailable" + + +@pytest.mark.asyncio +async def test_payout_units_excludes_units_the_sender_cannot_pay() -> None: + with patch( + "routstr.wallet._get_supported_mint_units", + AsyncMock(return_value=["usd", "sat", "eur", "msat"]), + ): + assert await _payout_units("http://mint:3338") == ["sat", "msat"]