Merge pull request #763 from Routstr/fix/payout-proof-batching-761

fix: make payouts resilient to mint proof limits

docs/provider/configuration.md

@@ -213,6 +213,7 @@ Use environment variables for:
 | `MINT_RETRY_MAX_ATTEMPTS` | Retries after a timeout or HTTP 429 (`0` disables retries) | `3` |
 | `RECEIVE_LN_ADDRESS` | Lightning address for withdrawals | —                                    |
 | `MIN_PAYOUT_SAT`     | Min payout balance in sats (applies to all mints) | `210`                |
+| `MAX_PAYOUT_SAT`     | Maximum gross budget per periodic payout in sats, including fees (all mints) | `250000`             |
 | `PAYOUT_INTERVAL_SECONDS` | Payout loop interval (seconds) | `900`                            |
 | `TOR_PROXY_URL`      | SOCKS5 proxy for Tor              | `socks5://127.0.0.1:9050`            |
 | `CORS_ORIGINS`       | Allowed CORS origins              | `*`                                  |
@@ -225,6 +226,26 @@ Routstr's wallet mutation lock fail fast during that cooldown instead of waiting
 while blocking every other wallet mutation. Callers receive an error and may retry
 later; the current response does not include the cooldown duration.
 
+Read-only `/v1/checkstate` requests start at the SDK request-model limit
+(currently 1,000 proofs) and adapt downward on HTTP 413 or 500, down to one
+proof. A 500 is a size hypothesis, not a confirmed limit. Successful reduced
+sizes are cached per mint within each worker for 24 hours (and refreshed while
+in use). HTTP 429 never reduces the batch
+size. Scan deadline expiry opens a transport cooldown without shortening any
+existing rate-limit cooldown. Invalid, incomplete, or failed scans do
+not produce a partial spendable balance. Only explicit UNSPENT proofs qualify;
+PENDING proofs are retained but excluded from payouts.
+
+Each scan is bounded by a fixed 60-second deadline and a 128-request budget;
+exhausting either aborts that scan safely. Automatic splitting applies only to
+state checks, **not swaps or melts**. Their limits are independent, and ambiguous
+mutation outcomes must be reconciled rather than retried with different inputs.
+Periodic payouts reload local proofs without forcing a keyset refresh, skip
+state checks at/below `MIN_PAYOUT_SAT`, and cap each gross payout budget at
+`MAX_PAYOUT_SAT`. Oversized inputs receive enough change outputs to return the
+excess; they are not automatically swapped. The cap is not a proof-count limit
+or a guarantee of Lightning payment success.
+
 ### Priority
 
 Environment variables are read on startup. Dashboard settings override them and persist in the database. Once you change a setting in the dashboard, the env var is ignored for that setting.

routstr/checkstate.py

@@ -0,0 +1,124 @@
+"""Bounded adaptive batching for read-only NUT-07 requests, never mutations."""
+
+import asyncio
+import time
+
+import httpx
+from cashu.core.base import Proof, ProofSpentState, ProofState
+from cashu.core.models import PostCheckStateRequest
+from cashu.wallet.wallet import Wallet
+
+from .core.logging import get_logger
+from .mint import MINT_TRANSPORT_COOLDOWN_SECONDS, MintRateGuard, run_mint_operation
+
+logger = get_logger(__name__)
+_SDK_BATCH_LIMIT = PostCheckStateRequest.model_json_schema()["properties"]["Ys"][
+    "maxItems"
+]
+_LEARNED_TTL = 24 * 60 * 60
+# Fixed scan bounds: adaptive halving makes the start size near irrelevant, and
+# the deadline/request budget are safety limits, not tuning knobs.
+_DEFAULT_BATCH_SIZE = _SDK_BATCH_LIMIT
+_SCAN_TIMEOUT_SECONDS = 60
+_MAX_REQUESTS = 128
+_learned_sizes: dict[str, tuple[int, float]] = {}
+
+
+async def filter_unspent_proofs(
+    proofs: list[Proof], wallet: Wallet, *, retry_on_rate_limit: bool = True
+) -> list[Proof]:
+    if not proofs:
+        return []
+    mint_url = str(wallet.url)
+    key = mint_url.rstrip("/")
+    configured = _DEFAULT_BATCH_SIZE
+    learned, expires = _learned_sizes.get(key, (configured, 0.0))
+    batch_size = min(configured, learned) if expires > time.monotonic() else configured
+    unspent: list[Proof] = []
+    spent: list[Proof] = []
+    offset = 0
+    requests = 0
+
+    async def check_batch() -> tuple[list[Proof], list[ProofState]]:
+        nonlocal batch_size, requests
+        # Size fallback stays inside the rate guard's operation. A recoverable
+        # 500 during a cooldown probe must not open another cooldown first.
+        while True:
+            batch = proofs[offset : offset + batch_size]
+            if requests >= _MAX_REQUESTS:
+                raise ValueError("Proof-state request budget exhausted")
+            requests += 1
+            try:
+                response = await wallet.check_proof_state(batch)
+            except httpx.HTTPStatusError as error:
+                # A proxy 500 can mean a body limit (#761), but is not proof of
+                # one. Diagnostic retries are safe here because this is a read.
+                if error.response.status_code not in {413, 500} or len(batch) == 1:
+                    logger.warning(
+                        "Proof-state request failed; scan aborted",
+                        extra={
+                            "mint_url": mint_url,
+                            "endpoint": "/v1/checkstate",
+                            "status": error.response.status_code,
+                            "content_type": error.response.headers.get("content-type"),
+                            "request_bytes": error.request.headers.get(
+                                "content-length"
+                            ),
+                            "proof_count": len(batch),
+                            "requests": requests,
+                        },
+                    )
+                    raise
+                batch_size = max(1, len(batch) // 2)
+                logger.warning(
+                    "Retrying proof-state check with a smaller batch",
+                    extra={
+                        "mint_url": mint_url,
+                        "endpoint": "/v1/checkstate",
+                        "status": error.response.status_code,
+                        "content_type": error.response.headers.get("content-type"),
+                        "request_bytes": error.request.headers.get("content-length"),
+                        "proof_count": len(batch),
+                        "next_batch_size": batch_size,
+                        "requests": requests,
+                    },
+                )
+                continue
+            states = response.states
+            if len(states) != len(batch) or any(
+                state.Y != proof.Y for proof, state in zip(batch, states)
+            ):
+                raise ValueError("Invalid proof-state response: count or Y mismatch")
+            if any(state.state not in set(ProofSpentState) for state in states):
+                raise ValueError("Invalid proof-state response: unknown state")
+            return batch, states
+
+    # Bound the entire scan, including retries and cooldown waits.
+    deadline = asyncio.timeout(_SCAN_TIMEOUT_SECONDS)
+    try:
+        async with deadline:
+            while offset < len(proofs):
+                batch, states = await run_mint_operation(
+                    check_batch,
+                    op_name="check_proof_state",
+                    mint_url=mint_url,
+                    retry_on_rate_limit=retry_on_rate_limit,
+                )
+                if batch_size < configured:
+                    _learned_sizes[key] = (batch_size, time.monotonic() + _LEARNED_TTL)
+                for proof, state in zip(batch, states):
+                    if state.state == ProofSpentState.unspent:
+                        unspent.append(proof)
+                    elif state.state == ProofSpentState.spent:
+                        spent.append(proof)
+                    # Retain PENDING proofs without making them spendable.
+                offset += len(batch)
+            if spent:
+                await wallet.set_reserved_for_send(spent, reserved=True)
+    except TimeoutError:
+        if deadline.expired():
+            MintRateGuard.get(mint_url).apply_cooldown(
+                MINT_TRANSPORT_COOLDOWN_SECONDS, reason="transport"
+            )
+        raise
+    return unspent

routstr/core/settings.py

@@ -62,6 +62,8 @@ class Settings(BaseSettings):
     # Minimum available balance (in satoshis) before profit is paid out over
     # Lightning
     min_payout_sat: int = Field(default=210, gt=0, env="MIN_PAYOUT_SAT")
+    # Gross payout budget in sats, including fees.
+    max_payout_sat: int = Field(default=250_000, gt=0, env="MAX_PAYOUT_SAT")
     # Interval (seconds) between periodic payout attempts. Must be positive.
     payout_interval_seconds: int = Field(
         default=900, gt=0, env="PAYOUT_INTERVAL_SECONDS"

routstr/payment/lnurl.py

@@ -417,6 +417,8 @@ async def raw_send_to_lnurl(
 
     assert selected_proofs is not None
     proofs = selected_proofs
+    # Cashu uses this argument only to size blank outputs, not set mint fees.
+    change_budget = sum(proof.amount for proof in proofs) - quoted_amount
     await wallet.set_reserved_for_send(proofs, reserved=True)
 
     try:
@@ -424,7 +426,7 @@ async def raw_send_to_lnurl(
             lambda: wallet.melt(
                 proofs=proofs,
                 invoice=bolt11_invoice,
-                fee_reserve_sat=melt_quote_resp.fee_reserve,
+                fee_reserve_sat=change_budget,
                 quote_id=melt_quote_resp.quote,
             ),
             op_name="lnurl_melt",

routstr/wallet.py

@@ -21,6 +21,7 @@ from pydantic_core import PydanticUndefined
 from sqlmodel import col, select, update
 
 from .cashu_compat import install_cashu_httpx_shim
+from .checkstate import filter_unspent_proofs
 from .core import db, get_logger
 from .core.db import store_cashu_transaction_with_retry as store_cashu_transaction
 from .core.settings import settings
@@ -143,6 +144,12 @@ class Wallet(_CashuWallet):
                 request=resp.request,
                 response=resp,
             )
+        if resp.status_code in {413, 500} and resp.request.url.path.endswith(
+            "/v1/checkstate"
+        ):
+            # Preserve size/HTTP diagnostics even when a proxy or mint returns
+            # JSON with a detail field. Mutation error handling stays unchanged.
+            resp.raise_for_status()
         try:
             response_data = resp.json()
         except json.JSONDecodeError:
@@ -1165,6 +1172,7 @@ async def get_wallet(
     retry_on_rate_limit: bool = True,
     force_reload: bool = False,
     load_proofs: bool = True,
+    force_reload_proofs: bool = False,
 ) -> Wallet:
     global _wallets, _wallet_last_load, _wallet_last_mint_load, _wallet_load_locks
     id = f"{mint_url}_{unit}"
@@ -1197,6 +1205,7 @@ async def get_wallet(
                 last_proof_load = _wallet_last_load.get(id)
                 if (
                     force_reload
+                    or force_reload_proofs
                     or last_proof_load is None
                     or now - last_proof_load
                     >= _WALLET_PROOF_RELOAD_MIN_INTERVAL_SECONDS
@@ -1231,29 +1240,9 @@ async def slow_filter_spend_proofs(
     *,
     retry_on_rate_limit: bool = True,
 ) -> list[Proof]:
-    if not proofs:
-        return []
-    _proofs = []
-    _spent_proofs = []
-    # Keep proof-state checks in large batches. Mint quotas count HTTP requests,
-    # so smaller batches make balance reads slower and more likely to hit 429s.
-    batch_size = 1000
-    for i in range(0, len(proofs), batch_size):
-        pb = proofs[i : i + batch_size]
-        proof_states = await run_mint_operation(
-            lambda: wallet.check_proof_state(pb),
-            op_name="check_proof_state",
-            mint_url=str(wallet.url),
-            retry_on_rate_limit=retry_on_rate_limit,
-        )
-        for proof, state in zip(pb, proof_states.states):
-            if str(state.state) != "spent":
-                _proofs.append(proof)
-            else:
-                _spent_proofs.append(proof)
-    if _spent_proofs:
-        await wallet.set_reserved_for_send(_spent_proofs, reserved=True)
-    return _proofs
+    return await filter_unspent_proofs(
+        proofs, wallet, retry_on_rate_limit=retry_on_rate_limit
+    )
 
 
 class BalanceDetail(TypedDict, total=False):
@@ -1545,12 +1534,16 @@ async def _payout_mint_and_unit(mint_url: str, unit: str) -> None:
     try:
         # Runs under wallet_operation_guard; a cached wallet may carry a proof
         # snapshot up to 30s stale from another process's reservation, so the
-        # cross-process lock is only safe with a fresh reload.
-        wallet = await get_wallet(mint_url, unit, force_reload=True)
+        # cross-process lock is only safe with fresh local proofs, not a
+        # forced network refresh of every keyset.
+        wallet = await get_wallet(mint_url, unit, force_reload_proofs=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.
+        min_amount = (
+            settings.min_payout_sat
+            if unit == "sat"
+            else _sats_to_msats(settings.min_payout_sat)
+        )
+        if sum(proof.amount for proof in proofs) <= min_amount:
             return
         proofs = await slow_filter_spend_proofs(proofs, wallet)
         await asyncio.sleep(5)
@@ -1582,18 +1575,19 @@ async def _payout_mint_and_unit(mint_url: str, unit: str) -> None:
             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 = (
-            settings.min_payout_sat
+        max_amount = (
+            settings.max_payout_sat
             if unit == "sat"
-            else _sats_to_msats(settings.min_payout_sat)
+            else _sats_to_msats(settings.max_payout_sat)
         )
         if available_balance > min_amount:
+            payout_amount = min(available_balance, max_amount)
             amount_received = await raw_send_to_lnurl(
                 wallet,
                 proofs,
                 settings.receive_ln_address,
                 unit,
-                amount=available_balance,
+                amount=payout_amount,
             )
             logger.info(
                 "Payout sent successfully",
@@ -1601,6 +1595,7 @@ async def _payout_mint_and_unit(mint_url: str, unit: str) -> None:
                     "mint_url": mint_url,
                     "unit": unit,
                     "balance": available_balance,
+                    "amount": payout_amount,
                     "amount_received": amount_received,
                 },
             )

tests/unit/test_checkstate.py

@@ -0,0 +1,314 @@
+import asyncio
+from collections.abc import Callable, Iterator
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock, patch
+
+import httpx
+import pytest
+from cashu.core.base import Proof, ProofSpentState
+
+from routstr import checkstate
+from routstr.checkstate import _learned_sizes, filter_unspent_proofs
+from routstr.mint import MintRateGuard, fail_fast_mint_operations
+
+
+@pytest.fixture(autouse=True)
+def isolate() -> Iterator[None]:
+    _learned_sizes.clear()
+    MintRateGuard._guards.clear()
+    yield
+    _learned_sizes.clear()
+    MintRateGuard._guards.clear()
+
+
+def proofs(count: int) -> list[Proof]:
+    return [Mock(Y=str(i)) for i in range(count)]
+
+
+def response(batch: list[Proof]) -> SimpleNamespace:
+    return SimpleNamespace(
+        states=[SimpleNamespace(Y=p.Y, state=ProofSpentState.unspent) for p in batch]
+    )
+
+
+def rejection(status: int) -> httpx.HTTPStatusError:
+    request = httpx.Request("POST", "https://mint.test/v1/checkstate")
+    return httpx.HTTPStatusError(
+        "rejected",
+        request=request,
+        response=httpx.Response(
+            status,
+            request=request,
+            headers={"content-type": "text/html", "retry-after": "120"},
+        ),
+    )
+
+
+def wallet(check: Callable[[list[Proof]], object]) -> Mock:
+    return Mock(
+        url="https://mint.test",
+        check_proof_state=AsyncMock(side_effect=check),
+        set_reserved_for_send=AsyncMock(),
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status", [413, 500])
+async def test_adapts_and_reuses_size_without_skipping_proofs(status: int) -> None:
+    checked: list[Proof] = []
+
+    async def check(batch: list[Proof]) -> SimpleNamespace:
+        if len(batch) > 120:
+            raise rejection(status)
+        checked.extend(batch)
+        return response(batch)
+
+    w = wallet(check)
+    ps = proofs(1001)
+    assert await filter_unspent_proofs(ps, w) == ps
+    assert checked == ps
+    sizes = [len(c.args[0]) for c in w.check_proof_state.await_args_list]
+    assert sizes[:5] == [1000, 500, 250, 125, 62]
+    w.check_proof_state.reset_mock()
+    assert await filter_unspent_proofs(ps, w) == ps
+    assert max(len(c.args[0]) for c in w.check_proof_state.await_args_list) == 62
+
+
+@pytest.mark.asyncio
+async def test_size_fallback_works_inside_cooldown_probe_under_wallet_guard() -> None:
+    async def check(batch: list[Proof]) -> SimpleNamespace:
+        if len(batch) > 2:
+            raise rejection(500)
+        return response(batch)
+
+    w = wallet(check)
+    MintRateGuard.get(w.url).apply_cooldown(0, reason="transport")
+    async with fail_fast_mint_operations():
+        ps = proofs(8)
+        assert await filter_unspent_proofs(ps, w) == ps
+    assert MintRateGuard.get(w.url).cooldown_remaining() == 0
+
+
+@pytest.mark.asyncio
+async def test_429_is_not_a_size_signal() -> None:
+    w = wallet(Mock(side_effect=rejection(429)))
+    with pytest.raises(httpx.HTTPStatusError):
+        await filter_unspent_proofs(proofs(1000), w, retry_on_rate_limit=False)
+    assert w.check_proof_state.await_count == 1
+    assert not _learned_sizes
+    assert MintRateGuard.get(w.url).cooldown_remaining() > 100
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status", [400, 401, 422, 503])
+async def test_other_http_errors_are_not_split(status: int) -> None:
+    w = wallet(Mock(side_effect=rejection(status)))
+    with pytest.raises(httpx.HTTPStatusError):
+        await filter_unspent_proofs(proofs(10), w)
+    assert w.check_proof_state.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_singleton_failure_is_bounded_and_does_not_poison_cache() -> None:
+    w = wallet(Mock(side_effect=rejection(500)))
+    with pytest.raises(httpx.HTTPStatusError):
+        await filter_unspent_proofs(proofs(1000), w)
+    assert [len(c.args[0]) for c in w.check_proof_state.await_args_list] == [
+        1000,
+        500,
+        250,
+        125,
+        62,
+        31,
+        15,
+        7,
+        3,
+        1,
+    ]
+    assert not _learned_sizes
+    w.set_reserved_for_send.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_request_budget_counts_successes_and_failures() -> None:
+    w = wallet(response)
+    with (
+        patch.object(checkstate, "_DEFAULT_BATCH_SIZE", 1),
+        patch.object(checkstate, "_MAX_REQUESTS", 2),
+        pytest.raises(ValueError, match="budget"),
+    ):
+        await filter_unspent_proofs(proofs(3), w)
+    assert w.check_proof_state.await_count == 2
+    w.set_reserved_for_send.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_total_deadline_cancels_slow_check() -> None:
+    async def check(batch: list[Proof]) -> None:
+        await asyncio.Event().wait()
+
+    w = wallet(check)
+    with (
+        patch.object(checkstate, "_SCAN_TIMEOUT_SECONDS", 0.01),
+        pytest.raises(TimeoutError),
+    ):
+        await filter_unspent_proofs(proofs(1), w)
+    assert w.check_proof_state.await_count == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("malformation", ["missing", "reordered", "unknown"])
+async def test_invalid_response_fails_closed(malformation: str) -> None:
+    def check(batch: list[Proof]) -> SimpleNamespace:
+        result = response(batch)
+        if malformation == "missing":
+            result.states.pop()
+        elif malformation == "reordered":
+            result.states.reverse()
+        else:
+            result.states[0].state = "UNKNOWN"
+        return result
+
+    w = wallet(check)
+    with pytest.raises(ValueError, match="Invalid proof-state"):
+        await filter_unspent_proofs(proofs(3), w)
+    w.set_reserved_for_send.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_only_unspent_proofs_are_spendable() -> None:
+    ps = proofs(3)
+    states = [ProofSpentState.unspent, ProofSpentState.pending, ProofSpentState.spent]
+    w = wallet(
+        lambda batch: SimpleNamespace(
+            states=[SimpleNamespace(Y=p.Y, state=s) for p, s in zip(batch, states)]
+        )
+    )
+    assert await filter_unspent_proofs(ps, w) == ps[:1]
+    w.set_reserved_for_send.assert_awaited_once_with(ps[2:], reserved=True)
+
+
+@pytest.mark.asyncio
+async def test_learned_size_is_per_mint_and_expires() -> None:
+    w = wallet(response)
+    ps = proofs(5)
+    _learned_sizes[w.url] = (1, 0)
+    other = wallet(response)
+    other.url = "https://other.test"
+    _learned_sizes[other.url] = (1, float("inf"))
+    with patch.object(checkstate, "_DEFAULT_BATCH_SIZE", 2):
+        assert await filter_unspent_proofs(ps, w) == ps
+        assert [len(c.args[0]) for c in w.check_proof_state.await_args_list] == [
+            2,
+            2,
+            1,
+        ]
+        assert await filter_unspent_proofs(ps, other) == ps
+        assert [len(c.args[0]) for c in other.check_proof_state.await_args_list] == [
+            1
+        ] * 5
+
+
+@pytest.mark.asyncio
+async def test_smaller_later_batch_failure_does_not_skip_or_return_partial() -> None:
+    ps = proofs(9)
+    checked: list[Proof] = []
+
+    def check(batch: list[Proof]) -> SimpleNamespace:
+        if batch[0] is not ps[0] and len(batch) > 1:
+            raise rejection(500)
+        checked.extend(batch)
+        return response(batch)
+
+    w = wallet(check)
+    with patch.object(checkstate, "_DEFAULT_BATCH_SIZE", 4):
+        assert await filter_unspent_proofs(ps, w) == ps
+    assert checked == ps
+
+
+@pytest.mark.parametrize("status", [413, 500])
+@pytest.mark.parametrize("body", [{"detail": "too big"}, "<html>error</html>"])
+def test_wallet_adapter_preserves_checkstate_http_status(
+    status: int, body: dict[str, str] | str
+) -> None:
+    from routstr.wallet import Wallet
+
+    request = httpx.Request("POST", "https://mint.test/v1/checkstate")
+    reply = (
+        httpx.Response(status, request=request, json=body)
+        if isinstance(body, dict)
+        else httpx.Response(status, request=request, text=body)
+    )
+    with pytest.raises(httpx.HTTPStatusError) as error:
+        Wallet.raise_on_error_request(reply)
+    assert error.value.response is reply
+
+
+@pytest.mark.asyncio
+async def test_default_batch_fits_real_sdk_model() -> None:
+    from cashu.core.models import PostCheckStateRequest
+
+    limit = PostCheckStateRequest.model_json_schema()["properties"]["Ys"]["maxItems"]
+    ps = [
+        Proof(id="00", amount=1, secret=f"sdk-{i}", C="02" + "00" * 32)
+        for i in range(limit + 1)
+    ]
+    sizes: list[int] = []
+
+    def check(batch: list[Proof]) -> SimpleNamespace:
+        payload = PostCheckStateRequest(Ys=[p.Y for p in batch])
+        sizes.append(len(payload.Ys))
+        return response(batch)
+
+    w = wallet(check)
+    assert await filter_unspent_proofs(ps, w) == ps
+    assert sizes == [limit, 1]
+
+
+@pytest.mark.asyncio
+async def test_scan_deadline_opens_cooldown_for_next_guarded_scan() -> None:
+    from routstr.mint import MintCooldownError
+
+    async def check(batch: list[Proof]) -> None:
+        await asyncio.Event().wait()
+
+    w = wallet(check)
+    with patch.object(checkstate, "_SCAN_TIMEOUT_SECONDS", 0.01):
+        async with fail_fast_mint_operations():
+            with pytest.raises(TimeoutError):
+                await filter_unspent_proofs(proofs(1), w)
+            with pytest.raises(MintCooldownError):
+                await filter_unspent_proofs(proofs(1), w)
+    assert w.check_proof_state.await_count == 1
+    assert MintRateGuard.get(w.url).cooldown_reason() == "transport"
+
+
+@pytest.mark.asyncio
+async def test_external_cancellation_does_not_open_cooldown() -> None:
+    started = asyncio.Event()
+
+    async def check(batch: list[Proof]) -> None:
+        started.set()
+        await asyncio.Event().wait()
+
+    w = wallet(check)
+    task = asyncio.create_task(filter_unspent_proofs(proofs(1), w))
+    await started.wait()
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    assert MintRateGuard.get(w.url).cooldown_remaining() == 0
+
+
+@pytest.mark.asyncio
+async def test_scan_deadline_preserves_longer_rate_limit_cooldown() -> None:
+    w = wallet(response)
+    guard = MintRateGuard.get(w.url)
+    guard.apply_rate_limit_cooldown(120)
+    until = guard._cooldown_until
+    with patch.object(checkstate, "_SCAN_TIMEOUT_SECONDS", 0.01):
+        with pytest.raises(TimeoutError):
+            await filter_unspent_proofs(proofs(1), w)
+    assert guard._cooldown_until == until
+    assert guard.cooldown_reason() == "rate_limited"
+    w.check_proof_state.assert_not_awaited()

tests/unit/test_lnurl_change.py

@@ -0,0 +1,167 @@
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock, patch
+
+import pytest
+from cashu.core.base import BlindedMessage, BlindedSignature, Proof, Unit
+from cashu.core.crypto import b_dhke
+from cashu.core.models import PostMeltQuoteResponse
+from cashu.wallet.v1_api import LedgerAPI
+from cashu.wallet.wallet import Wallet as CashuWallet
+
+from routstr.core.settings import settings
+from routstr.mint import MintRateGuard
+from routstr.wallet import _payout_mint_and_unit
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("unit,scale", [("sat", 1), ("msat", 1000)])
+@pytest.mark.parametrize(
+    "liability,input_fee,reserve,actual_fee",
+    [(0, 0, 0, 0), (0, 7, 10, 3), (300000, 7, 10, 3)],
+)
+async def test_capped_payout_recovers_all_change_with_real_cashu_sdk(
+    unit: str,
+    scale: int,
+    liability: int,
+    input_fee: int,
+    reserve: int,
+    actual_fee: int,
+) -> None:
+    MintRateGuard._guards.clear()
+    private_key = b_dhke.PrivateKey()
+    proof = Proof(
+        id="00",
+        amount=524288 * scale,
+        secret="input-proof",
+        C=private_key.public_key.format().hex(),
+    )
+    w = CashuWallet.__new__(CashuWallet)
+    w.url = "https://mint.test"
+    w.unit = Unit[unit]
+    w.keyset_id = "00"
+    w.keysets = {
+        "00": SimpleNamespace(
+            public_keys={2**i: private_key.public_key for i in range(40)}
+        )
+    }
+    w.proofs = [proof]
+    w.db = Mock()
+    w.get_fees_for_proofs = Mock(return_value=input_fee * scale)
+    w.set_reserved_for_send = AsyncMock()
+    w.set_reserved_for_melt = AsyncMock()
+    w.sign_proofs_inplace_melt = Mock(side_effect=lambda ps, outputs, quote: ps)
+    w._store_proofs = AsyncMock()
+
+    async def invalidate(ps: list[Proof]) -> None:
+        w.proofs = [p for p in w.proofs if p not in ps]
+
+    w.invalidate = AsyncMock(side_effect=invalidate)
+    w.generate_n_secrets = AsyncMock(
+        side_effect=lambda n: (
+            [f"change-{i}" for i in range(n)],
+            [],
+            [f"path-{i}" for i in range(n)],
+        )
+    )
+    quotes: dict[str, PostMeltQuoteResponse] = {}
+
+    async def quote(invoice: str) -> PostMeltQuoteResponse:
+        amount_msat = int(invoice)
+        amount = amount_msat // 1000 if unit == "sat" else amount_msat
+        q = PostMeltQuoteResponse(
+            quote=str(amount),
+            amount=amount,
+            unit=unit,
+            request=invoice,
+            fee_reserve=reserve * scale,
+            state="UNPAID",
+            expiry=None,
+        )
+        quotes[q.quote] = q
+        return q
+
+    w.melt_quote = AsyncMock(side_effect=quote)
+    selected_total = 0
+    returned_change = 0
+    blank_count = 0
+    paid_amount = 0
+
+    async def mint_melt(
+        quote_id: str, inputs: list[Proof], outputs: list[BlindedMessage]
+    ) -> PostMeltQuoteResponse:
+        nonlocal selected_total, returned_change, blank_count, paid_amount
+        q = quotes[quote_id]
+        selected_total = sum(p.amount for p in inputs)
+        paid_amount = q.amount
+        blank_count = len(outputs)
+        assert q.fee_reserve == reserve * scale
+        change = selected_total - q.amount - (input_fee + actual_fee) * scale
+        amounts = [2**i for i in range(change.bit_length()) if change & (2**i)]
+        signatures = []
+        for amount, output in zip(amounts, outputs):
+            blinded, _, _ = b_dhke.step2_bob(
+                b_dhke.PublicKey(bytes.fromhex(output.B_)), private_key
+            )
+            signatures.append(
+                BlindedSignature(id="00", amount=amount, C_=blinded.format().hex())
+            )
+        returned_change = sum(s.amount for s in signatures)
+        assert returned_change == change
+        return q.model_copy(update={"state": "PAID", "change": signatures})
+
+    @asynccontextmanager
+    async def session() -> AsyncIterator[Mock]:
+        yield Mock()
+
+    with (
+        patch.object(settings, "max_payout_sat", 250000),
+        patch.object(settings, "min_payout_sat", 210),
+        patch("routstr.wallet.get_wallet", AsyncMock(return_value=w)),
+        patch("routstr.wallet.get_proofs_per_mint_and_unit", return_value=[proof]),
+        patch(
+            "routstr.wallet.slow_filter_spend_proofs", AsyncMock(return_value=[proof])
+        ),
+        patch("routstr.wallet.asyncio.sleep", AsyncMock()),
+        patch("routstr.wallet.db.create_session", session),
+        patch(
+            "routstr.wallet.db.total_user_liability",
+            AsyncMock(return_value=liability * 1000),
+        ),
+        patch(
+            "routstr.payment.lnurl.get_lnurl_data",
+            AsyncMock(
+                return_value={
+                    "callback_url": "https://ln.test/cb",
+                    "min_sendable": 1000,
+                    "max_sendable": 10**12,
+                }
+            ),
+        ),
+        patch(
+            "routstr.payment.lnurl.get_lnurl_invoice",
+            AsyncMock(side_effect=lambda callback, amount: (str(amount), {})),
+        ),
+        patch.object(LedgerAPI, "melt", AsyncMock(side_effect=mint_melt)) as transport,
+        patch("cashu.wallet.wallet.update_bolt11_melt_quote", AsyncMock()),
+    ):
+        await _payout_mint_and_unit(w.url, unit)
+
+    transport.assert_awaited_once()
+    assert selected_total == 524288 * scale
+    assert blank_count > 0
+    assert sum(p.amount for p in w.proofs) == returned_change
+    assert all(
+        b_dhke.verify(private_key, b_dhke.PublicKey(bytes.fromhex(p.C)), p.secret)
+        for p in w.proofs
+    )
+    net_debit = selected_total - returned_change
+    assert net_debit == paid_amount + (input_fee + actual_fee) * scale
+    assert net_debit <= min(250000, 524288 - liability) * scale
+    assert returned_change >= liability * scale
+    if liability == input_fee == reserve == actual_fee == 0:
+        assert returned_change == 274288 * scale
+        assert net_debit == 250000 * scale
+    w._store_proofs.assert_awaited_once()
+    MintRateGuard._guards.clear()

tests/unit/test_lnurl_melt_timeout.py

@@ -65,6 +65,33 @@ def _lnurl_patches() -> tuple[Any, Any]:
     )
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("outcome", ["timeout", "pending"])
+async def test_oversized_proof_change_budget_preserves_ambiguous_melt(
+    outcome: str,
+) -> None:
+    wallet, proofs = _wallet()
+    proofs[0].amount = 524288
+    if outcome == "timeout":
+        wallet.melt.side_effect = httpx.ReadTimeout("response lost")
+    else:
+        wallet.melt.return_value = MagicMock(state=MeltQuoteState.pending)
+    wallet.get_melt_quote = AsyncMock(
+        return_value=MagicMock(state=MeltQuoteState.pending)
+    )
+    data_patch, invoice_patch = _lnurl_patches()
+    with data_patch, invoice_patch, pytest.raises(MeltOutcomeAmbiguousError):
+        await raw_send_to_lnurl(wallet, proofs, "owner@ln.tld", "sat", amount=1000)
+    wallet.melt.assert_awaited_once()
+    assert wallet.melt.await_args.kwargs["fee_reserve_sat"] == 524288 - QUOTE_AMOUNT_SAT
+    wallet.set_reserved_for_send.assert_awaited_once_with(proofs, reserved=True)
+    if outcome == "timeout":
+        wallet.set_reserved_for_melt.assert_awaited_once_with(
+            proofs, reserved=True, quote_id="q"
+        )
+    wallet.get_melt_quote.assert_awaited_once_with("q")
+
+
 @pytest.mark.asyncio
 async def test_raw_send_to_lnurl_direct_unpaid_is_retry_safe() -> None:
     wallet, proofs = _wallet()

tests/unit/test_payout_limits.py

@@ -0,0 +1,75 @@
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from unittest.mock import AsyncMock, Mock, patch
+
+import pytest
+
+from routstr.core.settings import settings
+from routstr.wallet import _payout_mint_and_unit
+
+
+@asynccontextmanager
+async def session() -> AsyncIterator[Mock]:
+    yield Mock()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("unit,scale", [("sat", 1), ("msat", 1000)])
+@pytest.mark.parametrize(
+    "balance,liability,expected",
+    [(1000, 0, 100), (80, 30000, 50), (20, 20000, None), (0, 0, None), (10, 0, None)],
+)
+async def test_payout_limits_and_proof_refresh(
+    unit: str, scale: int, balance: int, liability: int, expected: int | None
+) -> None:
+    send = AsyncMock()
+    get_wallet = AsyncMock()
+    check = AsyncMock(side_effect=lambda ps, w: ps)
+    sleep = AsyncMock()
+    with (
+        patch.object(settings, "min_payout_sat", 10),
+        patch.object(settings, "max_payout_sat", 100),
+        patch("routstr.wallet.get_wallet", get_wallet),
+        patch(
+            "routstr.wallet.get_proofs_per_mint_and_unit",
+            return_value=[Mock(amount=balance * scale)],
+        ),
+        patch("routstr.wallet.slow_filter_spend_proofs", check),
+        patch("routstr.wallet.db.create_session", session),
+        patch(
+            "routstr.wallet.db.total_user_liability", AsyncMock(return_value=liability)
+        ),
+        patch("routstr.wallet.asyncio.sleep", sleep),
+        patch("routstr.wallet.raw_send_to_lnurl", send),
+    ):
+        await _payout_mint_and_unit("https://mint.test", unit)
+    get_wallet.assert_awaited_once_with(
+        "https://mint.test", unit, force_reload_proofs=True
+    )
+    if expected is None:
+        send.assert_not_awaited()
+    else:
+        assert send.await_args is not None
+        assert send.await_args.kwargs["amount"] == expected * scale
+    if balance <= 10:
+        check.assert_not_awaited()
+        sleep.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_failed_proof_check_never_pays_partial_balance() -> None:
+    send = AsyncMock()
+    with (
+        patch("routstr.wallet.get_wallet", AsyncMock()),
+        patch(
+            "routstr.wallet.get_proofs_per_mint_and_unit",
+            return_value=[Mock(amount=1_000_000)],
+        ),
+        patch(
+            "routstr.wallet.slow_filter_spend_proofs",
+            AsyncMock(side_effect=ValueError("Invalid proof-state response")),
+        ),
+        patch("routstr.wallet.raw_send_to_lnurl", send),
+    ):
+        await _payout_mint_and_unit("https://mint.test", "sat")
+    send.assert_not_awaited()

tests/unit/test_periodic_payout.py

@@ -12,6 +12,7 @@ Covers two regressions from the auto-payout / primary-mint audit
 
 from collections.abc import Callable, Coroutine
 from contextlib import asynccontextmanager
+from pathlib import Path
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -19,6 +20,14 @@ import pytest
 
 from routstr.wallet import _payout_units, periodic_payout
 
+
+@pytest.fixture(autouse=True)
+def isolate_wallet_lock(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        "routstr.wallet._WALLET_OPERATION_LOCK", tmp_path / "wallet.lock"
+    )
+
+
 # Sentinel interval used to break the otherwise-infinite payout loop after
 # exactly one full cycle.
 _INTERVAL = 987
@@ -156,7 +165,7 @@ async def test_periodic_payout_isolates_failing_mint() -> None:
     from routstr.core.settings import settings
 
     async def _get_wallet(
-        mint_url: str, unit: str, force_reload: bool = False
+        mint_url: str, unit: str, force_reload_proofs: bool = False
     ) -> MagicMock:
         if mint_url == "http://bad:3338":
             raise RuntimeError("mint unreachable")
@@ -219,6 +228,7 @@ 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.object(settings, "min_payout_sat", 10),
         patch(
             "routstr.wallet._get_supported_mint_units",
             AsyncMock(return_value=["sat", "msat"]),
@@ -255,3 +265,45 @@ 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_periodic_payout_caps_amount_at_max_payout_sat() -> None:
+    """Available balance above max_payout_sat is capped for a single payout."""
+    from routstr.core.settings import settings
+
+    raw_send = AsyncMock(return_value=1000)
+
+    with (
+        patch.object(settings, "cashu_mints", ["http://mint:3338"]),
+        patch.object(settings, "primary_mint", "http://mint: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.object(settings, "max_payout_sat", 250_000),
+        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", AsyncMock(return_value=MagicMock())),
+        patch(
+            "routstr.wallet.get_proofs_per_mint_and_unit",
+            MagicMock(return_value=[MagicMock(amount=1_000_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.raw_send_to_lnurl", raw_send),
+    ):
+        with pytest.raises(_LoopBreak):
+            await periodic_payout()
+
+    assert raw_send.await_count >= 1
+    assert raw_send.await_args_list[0].kwargs["amount"] == 250_000

tests/unit/test_settings.py

@@ -207,9 +207,7 @@ async def test_settings_initialize_discards_unknown_keys() -> None:
 
         # Simulate older persisted key name and an unknown key.
         await session.exec(  # type: ignore
-            text(
-                "UPDATE settings SET data = :data WHERE id = 1"
-            ).bindparams(
+            text("UPDATE settings SET data = :data WHERE id = 1").bindparams(
                 data='{"name":"LegacyNode","nostr_analytics_enabled":false,"unknown_key":123}'
             )
         )
@@ -279,7 +277,9 @@ async def test_upstream_api_key_survives_persistence(
         await SettingsService.initialize(session)
         await session.exec(  # type: ignore
             text("UPDATE settings SET data = :d WHERE id = 1").bindparams(
-                d=json.dumps({"name": "LegacyNode", "upstream_api_key": "sk-only-in-db"})
+                d=json.dumps(
+                    {"name": "LegacyNode", "upstream_api_key": "sk-only-in-db"}
+                )
             )
         )
         await session.commit()

tests/unit/test_wallet.py

@@ -4,6 +4,7 @@ import json
 import socket
 from collections.abc import AsyncIterator, Generator
 from contextlib import asynccontextmanager
+from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 import httpx
@@ -31,11 +32,16 @@ from routstr.wallet import (
 
 
 @pytest.fixture(autouse=True)
-def isolate_wallet_runtime_state() -> Generator[None, None, None]:
+def isolate_wallet_runtime_state(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> Generator[None, None, None]:
     """Keep production limiter/wallet caches from leaking across unit tests."""
     from routstr import wallet as wallet_module
     from routstr.core.settings import settings
 
+    monkeypatch.setattr(
+        wallet_module, "_WALLET_OPERATION_LOCK", tmp_path / "wallet.lock"
+    )
     original_concurrency = settings.mint_max_concurrency
     settings.mint_max_concurrency = 0
     wallet_module._MintRateGuard._guards.clear()
@@ -143,6 +149,19 @@ async def test_get_wallet_force_reload_bypasses_reload_interval() -> None:
     assert mock_wallet.load_proofs.await_count == 2
 
 
+@pytest.mark.asyncio
+async def test_get_wallet_force_reload_proofs_keeps_cached_keysets() -> None:
+    from routstr.wallet import get_wallet
+
+    mock_wallet = Mock(load_mint=AsyncMock(), load_proofs=AsyncMock())
+    with patch("routstr.wallet.Wallet.with_db", AsyncMock(return_value=mock_wallet)):
+        await get_wallet("http://mint:3338", "sat")
+        await get_wallet("http://mint:3338", "sat", force_reload_proofs=True)
+
+    assert mock_wallet.load_mint.await_count == 1
+    assert mock_wallet.load_proofs.await_count == 2
+
+
 @pytest.mark.asyncio
 async def test_public_recieve_token_holds_wallet_operation_guard() -> None:
     inside_guard = False
@@ -1301,10 +1320,12 @@ async def test_execute_bolt11_payment_rereserves_when_cancelled() -> None:
 @pytest.mark.asyncio
 async def test_balance_proof_check_uses_large_batches_to_avoid_rate_limit() -> None:
     """Balance reads must not turn a few hundred proofs into many mint requests."""
+    from cashu.core.base import ProofSpentState
+
     from routstr.wallet import slow_filter_spend_proofs
 
-    proofs = [Mock() for _ in range(250)]
-    states = [Mock(state="UNSPENT") for _ in proofs]
+    proofs = [Mock(Y=str(i)) for i in range(250)]
+    states = [Mock(Y=proof.Y, state=ProofSpentState.unspent) for proof in proofs]
     wallet = Mock()
     wallet.url = "http://mint:3338"
     wallet.check_proof_state = AsyncMock(return_value=Mock(states=states))
@@ -2017,7 +2038,7 @@ async def test_payout_reloads_wallet_snapshot_under_guard() -> None:
         await _payout_mint_and_unit("https://mint.example.com", "sat")
 
     mock_get_wallet.assert_awaited_once_with(
-        "https://mint.example.com", "sat", force_reload=True
+        "https://mint.example.com", "sat", force_reload_proofs=True
     )