invoice polling backoff

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:

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():

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

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