fix: recover stale Cashu reservations safely

routstr/wallet.py

@@ -219,8 +219,10 @@ async def send(amount: int, unit: str, mint_url: str | None = None) -> tuple[int
     """Internal send function - returns amount and serialized token"""
     effective_mint_url = mint_url or settings.primary_mint
     wallet: Wallet = await get_wallet(effective_mint_url, unit)
-    proofs = get_proofs_per_mint_and_unit(wallet, effective_mint_url, unit)
+    all_proofs = get_proofs_per_mint_and_unit(wallet, effective_mint_url, unit)
+    proofs = [proof for proof in all_proofs if not proof.reserved]
     proofs_for_mint = sum(p.amount for p in proofs)
+    reserved_for_mint = sum(p.amount for p in all_proofs if p.reserved)
 
     # Fallback: proofs from untrusted source mints are swapped to primary_mint
     # during receive, so the user's preferred refund_mint_url may have no proofs
@@ -232,8 +234,10 @@ async def send(amount: int, unit: str, mint_url: str | None = None) -> tuple[int
         )
         effective_mint_url = settings.primary_mint
         wallet = await get_wallet(effective_mint_url, unit)
-        proofs = get_proofs_per_mint_and_unit(wallet, effective_mint_url, unit)
+        all_proofs = get_proofs_per_mint_and_unit(wallet, effective_mint_url, unit)
+        proofs = [proof for proof in all_proofs if not proof.reserved]
         proofs_for_mint = sum(p.amount for p in proofs)
+        reserved_for_mint = sum(p.amount for p in all_proofs if p.reserved)
 
     all_mint_urls = list({k.mint_url for k in wallet.keysets.values()})
     proof_summary = {
@@ -247,7 +251,8 @@ async def send(amount: int, unit: str, mint_url: str | None = None) -> tuple[int
         raw_proofs_by_keyset[p.id] = raw_proofs_by_keyset.get(p.id, 0) + p.amount
     logger.info(
         f"send: proof inventory | mint={effective_mint_url} unit={unit} amount={amount} "
-        f"primary_mint={settings.primary_mint} proofs_for_mint={proofs_for_mint} "
+        f"primary_mint={settings.primary_mint} liquid_proofs_for_mint={proofs_for_mint} "
+        f"reserved_proofs_for_mint={reserved_for_mint} "
         f"all_mints={all_mint_urls} by_keyset={proof_summary} "
         f"raw_proofs_by_keyset_id={raw_proofs_by_keyset} "
         f"total_wallet_proofs={sum(p.amount for p in wallet.proofs)}"
@@ -848,7 +853,7 @@ async def fetch_all_balances(
                 "unit": unit,
                 "wallet_balance": proofs_balance,
                 "user_balance": user_balance,
-                "owner_balance": proofs_balance - user_balance if proofs_balance != 0 else 0,
+                "owner_balance": proofs_balance - user_balance,
             }
             return result
         except Exception as e:
@@ -904,9 +909,7 @@ async def fetch_all_balances(
             total_wallet_balance_sats += proofs_balance_sats
             total_user_balance_sats += user_balance_sats
 
-    owner_balance = 0
-    if total_wallet_balance_sats != 0:
-        owner_balance = total_wallet_balance_sats - total_user_balance_sats
+    owner_balance = total_wallet_balance_sats - total_user_balance_sats
 
     return (
         balance_details,

scripts/reconcile_reserved_proofs.py

@@ -0,0 +1,233 @@
+#!/usr/bin/env python3
+"""Reconcile Routstr's reserved Cashu proofs against their mints.
+
+Safe default is dry-run. --apply mutates wallet.sqlite3 and keys.db.
+Run only while no process is using these databases and after verified backups.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import sqlite3
+import time
+from collections import defaultdict
+from pathlib import Path
+
+import httpx
+from cashu.core.base import Proof
+from cashu.wallet.helpers import deserialize_token_from_string
+
+
+def proof_from_row(row: sqlite3.Row) -> Proof:
+    return Proof(amount=row["amount"], C=row["C"], secret=row["secret"], id=row["id"])
+
+
+async def fetch_states(
+    client: httpx.AsyncClient,
+    mint_url: str,
+    proofs: list[Proof],
+    batch_size: int,
+) -> dict[str, str]:
+    states: dict[str, str] = {}
+    endpoint = mint_url.rstrip("/") + "/v1/checkstate"
+    for offset in range(0, len(proofs), batch_size):
+        batch = proofs[offset : offset + batch_size]
+        last_error: Exception | None = None
+        for attempt in range(4):
+            try:
+                response = await client.post(endpoint, json={"Ys": [proof.Y for proof in batch]})
+                response.raise_for_status()
+                for item in response.json().get("states", []):
+                    states[item["Y"]] = item["state"]
+                last_error = None
+                break
+            except (httpx.TimeoutException, httpx.TransportError) as exc:
+                last_error = exc
+                await asyncio.sleep(2**attempt)
+        if last_error is not None:
+            raise last_error
+    return states
+
+
+def load_pending_refund_secrets(keys: sqlite3.Connection) -> tuple[set[str], dict[str, set[str]]]:
+    pending: set[str] = set()
+    transaction_secrets: dict[str, set[str]] = {}
+    rows = keys.execute(
+        """
+        SELECT id, token
+        FROM cashu_transactions
+        WHERE type = 'out' AND collected = 0 AND swept = 0
+        """
+    ).fetchall()
+    for row in rows:
+        try:
+            token = deserialize_token_from_string(row["token"])
+        except Exception:
+            continue
+        secrets = {proof.secret for proof in token.proofs}
+        transaction_secrets[row["id"]] = secrets
+        pending.update(secrets)
+    return pending, transaction_secrets
+
+
+async def run(args: argparse.Namespace) -> int:
+    root = Path(args.root).resolve()
+    keys_path = root / "keys.db"
+    wallet_path = root / ".wallet" / "wallet.sqlite3"
+    if not keys_path.exists() or not wallet_path.exists():
+        raise SystemExit("keys.db or .wallet/wallet.sqlite3 not found")
+
+    keys = sqlite3.connect(keys_path)
+    wallet = sqlite3.connect(wallet_path)
+    keys.row_factory = sqlite3.Row
+    wallet.row_factory = sqlite3.Row
+    keys.execute("PRAGMA foreign_keys=ON")
+    wallet.execute("PRAGMA foreign_keys=ON")
+
+    if keys.execute("PRAGMA integrity_check").fetchone()[0] != "ok":
+        raise SystemExit("keys.db integrity check failed")
+    if wallet.execute("PRAGMA integrity_check").fetchone()[0] != "ok":
+        raise SystemExit("wallet.sqlite3 integrity check failed")
+
+    pending_secrets, transaction_secrets = load_pending_refund_secrets(keys)
+    rows = wallet.execute(
+        """
+        SELECT p.rowid AS proof_rowid, p.*, k.mint_url, k.unit
+        FROM proofs p
+        JOIN keysets k ON k.id = p.id
+        WHERE COALESCE(p.reserved, 0) != 0
+        ORDER BY k.mint_url, k.unit, p.time_reserved
+        """
+    ).fetchall()
+    grouped: dict[tuple[str, str], list[sqlite3.Row]] = defaultdict(list)
+    for row in rows:
+        grouped[(row["mint_url"], row["unit"])].append(row)
+
+    report: dict[str, object] = {
+        "mode": "apply" if args.apply else "dry-run",
+        "root": str(root),
+        "started_at": int(time.time()),
+        "pending_refund_secrets": len(pending_secrets),
+        "mints": {},
+        "errors": {},
+    }
+    state_by_secret: dict[str, str] = {}
+
+    async with httpx.AsyncClient(timeout=args.timeout) as client:
+        for (mint_url, unit), mint_rows in grouped.items():
+            proofs = [proof_from_row(row) for row in mint_rows]
+            try:
+                states = await fetch_states(client, mint_url, proofs, args.batch_size)
+            except Exception as exc:
+                report["errors"][f"{mint_url}|{unit}"] = f"{type(exc).__name__}: {exc}"
+                continue
+
+            summary: dict[str, dict[str, int]] = defaultdict(lambda: {"proofs": 0, "amount": 0})
+            actions = {"delete_spent": 0, "release_untracked_unspent": 0, "preserve_pending": 0, "preserve_unknown": 0}
+            for row, proof in zip(mint_rows, proofs):
+                state = states.get(proof.Y, "MISSING")
+                state_by_secret[row["secret"]] = state
+                summary[state]["proofs"] += 1
+                summary[state]["amount"] += row["amount"]
+
+                if state == "SPENT":
+                    actions["delete_spent"] += 1
+                    if args.apply:
+                        wallet.execute(
+                            """
+                            INSERT OR IGNORE INTO proofs_used
+                                (amount, C, secret, time_used, id, derivation_path, mint_id, melt_id)
+                            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+                            """,
+                            (
+                                row["amount"], row["C"], row["secret"],
+                                row["time_reserved"] or int(time.time()), row["id"],
+                                row["derivation_path"], row["mint_id"], row["melt_id"],
+                            ),
+                        )
+                        wallet.execute("DELETE FROM proofs WHERE rowid = ?", (row["proof_rowid"],))
+                elif state == "UNSPENT" and row["secret"] not in pending_secrets:
+                    actions["release_untracked_unspent"] += 1
+                    if args.apply:
+                        wallet.execute(
+                            "UPDATE proofs SET reserved = 0, send_id = NULL, time_reserved = NULL WHERE rowid = ?",
+                            (row["proof_rowid"],),
+                        )
+                elif state == "UNSPENT":
+                    actions["preserve_pending"] += 1
+                else:
+                    actions["preserve_unknown"] += 1
+
+            report["mints"][f"{mint_url}|{unit}"] = {
+                "states": dict(summary),
+                "actions": actions,
+            }
+
+    # Mark pending outgoing tokens collected only when every proof the mint reported is SPENT.
+    collected_transactions: list[str] = []
+    for transaction_id, secrets in transaction_secrets.items():
+        known = [state_by_secret.get(secret) for secret in secrets]
+        if known and all(state == "SPENT" for state in known):
+            collected_transactions.append(transaction_id)
+            if args.apply:
+                keys.execute(
+                    "UPDATE cashu_transactions SET collected = 1 WHERE id = ? AND collected = 0 AND swept = 0",
+                    (transaction_id,),
+                )
+    report["mark_collected_transactions"] = len(collected_transactions)
+
+    malformed = keys.execute(
+        "SELECT COUNT(*), COALESCE(SUM(balance), 0) FROM api_keys WHERE refund_mint_url = ?",
+        ("https://mint.minibits.cash/Bi",),
+    ).fetchone()
+    report["canonicalize_minibits_url"] = {"keys": malformed[0], "balance_msat": malformed[1]}
+    if args.apply:
+        keys.execute(
+            "UPDATE api_keys SET refund_mint_url = ? WHERE refund_mint_url = ?",
+            ("https://mint.minibits.cash/Bitcoin", "https://mint.minibits.cash/Bi"),
+        )
+
+    negative_rows = keys.execute(
+        "SELECT hashed_key, balance FROM api_keys WHERE balance < 0 ORDER BY balance"
+    ).fetchall()
+    report["negative_balances"] = {
+        "keys": len(negative_rows),
+        "amount_msat": sum(row["balance"] for row in negative_rows),
+        "key_prefixes": [row["hashed_key"][:12] for row in negative_rows],
+    }
+    if args.apply:
+        keys.execute("UPDATE api_keys SET balance = 0 WHERE balance < 0")
+
+    if args.apply:
+        wallet.commit()
+        keys.commit()
+    else:
+        wallet.rollback()
+        keys.rollback()
+
+    report["finished_at"] = int(time.time())
+    output = root / f"reconciliation-{'applied' if args.apply else 'dry-run'}-{report['finished_at']}.json"
+    output.write_text(json.dumps(report, indent=2, sort_keys=True))
+    os.chmod(output, 0o600)
+    print(json.dumps(report, indent=2, sort_keys=True))
+    print(f"report={output}")
+
+    wallet.close()
+    keys.close()
+    return 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--root", default=".")
+    parser.add_argument("--apply", action="store_true")
+    parser.add_argument("--batch-size", type=int, default=300)
+    parser.add_argument("--timeout", type=float, default=45.0)
+    return asyncio.run(run(parser.parse_args()))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/retry_minibits_reconciliation.sh

@@ -0,0 +1,62 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT="${HOME}/proxy"
+MARKER="${ROOT}/.minibits-reconciliation-complete"
+LOCK="${ROOT}/.minibits-reconciliation.lock"
+LOG="${ROOT}/logs/minibits-reconciliation.log"
+TAG="ROUTSTR-MINIBITS-RECONCILE"
+
+cd "$ROOT"
+[[ -e "$MARKER" ]] && exit 0
+exec 9>"$LOCK"
+flock -n 9 || exit 0
+
+printf '%s starting reconciliation retry\n' "$(date -Is)" >>"$LOG"
+if ! .venv/bin/python scripts/reconcile_reserved_proofs.py --root . --timeout 90 --apply >>"$LOG" 2>&1; then
+  printf '%s reconciliation command failed\n' "$(date -Is)" >>"$LOG"
+  exit 0
+fi
+
+latest=$(ls -1t reconciliation-applied-*.json 2>/dev/null | head -1 || true)
+[[ -n "$latest" ]] || exit 0
+if .venv/bin/python - "$latest" <<'PY'
+import json, sys
+report = json.load(open(sys.argv[1]))
+error_keys = report.get("errors", {})
+raise SystemExit(any(key.startswith("https://mint.minibits.cash/Bitcoin|") for key in error_keys))
+PY
+then
+  if .venv/bin/python - <<'PY' >>"$LOG" 2>&1
+import sqlite3
+keys = sqlite3.connect("keys.db")
+wallet = sqlite3.connect(".wallet/wallet.sqlite3")
+positive_balances = keys.execute(
+    "SELECT COALESCE(SUM(CASE WHEN balance > 0 THEN balance ELSE 0 END), 0) FROM api_keys"
+).fetchone()[0]
+pending_refunds = keys.execute(
+    "SELECT COALESCE(SUM(CASE WHEN unit = 'sat' THEN amount * 1000 ELSE amount END), 0) "
+    "FROM cashu_transactions WHERE type = 'out' AND collected = 0 AND swept = 0"
+).fetchone()[0]
+fees = keys.execute("SELECT COALESCE(SUM(accumulated_msats), 0) FROM routstr_fees").fetchone()[0]
+liquid_msats = wallet.execute(
+    "SELECT COALESCE(SUM(CASE WHEN k.unit = 'msat' THEN p.amount ELSE p.amount * 1000 END), 0) "
+    "FROM proofs p JOIN keysets k ON k.id = p.id WHERE COALESCE(p.reserved, 0) = 0"
+).fetchone()[0]
+obligations = positive_balances + pending_refunds + fees
+print(f"liquid_msats={liquid_msats} obligations_msats={obligations} surplus_msats={liquid_msats-obligations}")
+keys.close(); wallet.close()
+raise SystemExit(0 if liquid_msats >= obligations else 1)
+PY
+  then
+    touch "$MARKER"
+    chmod 600 "$MARKER"
+    printf '%s Minibits reconciliation completed and solvency verified; starting Routstr\n' "$(date -Is)" >>"$LOG"
+    docker compose up -d routstr >>"$LOG" 2>&1
+    (crontab -l 2>/dev/null | grep -v "$TAG" || true) | crontab -
+  else
+    printf '%s reconciliation completed but liquid assets remain below obligations; node stays stopped\n' "$(date -Is)" >>"$LOG"
+  fi
+else
+  printf '%s Minibits remains unavailable; retry retained\n' "$(date -Is)" >>"$LOG"
+fi

tests/unit/test_fetch_all_balances.py

@@ -11,7 +11,9 @@ async def _fake_session():  # type: ignore[no-untyped-def]
     yield MagicMock()
 
 
-def _patches(proof_amount: int = 1000):  # type: ignore[no-untyped-def]
+def _patches(  # type: ignore[no-untyped-def]
+    proof_amount: int = 1000, user_balance_msats: int = 0
+):
     proof = MagicMock(amount=proof_amount)
     return [
         patch("routstr.wallet.get_wallet", AsyncMock(return_value=MagicMock())),
@@ -25,7 +27,7 @@ def _patches(proof_amount: int = 1000):  # type: ignore[no-untyped-def]
         ),
         patch(
             "routstr.wallet.db.balances_for_mint_and_unit",
-            AsyncMock(return_value=0),
+            AsyncMock(return_value=user_balance_msats),
         ),
         patch("routstr.wallet.db.create_session", _fake_session),
     ]
@@ -52,6 +54,31 @@ async def test_fetch_all_balances_falls_back_to_primary_mint() -> None:
     assert total_wallet == 1000
 
 
+@pytest.mark.asyncio
+async def test_fetch_all_balances_reports_liability_when_wallet_is_empty() -> None:
+    """An empty wallet must not hide outstanding user liabilities."""
+    from routstr.core.settings import settings
+
+    with patch.object(settings, "cashu_mints", []), patch.object(
+        settings, "primary_mint", "http://primary:3338"
+    ):
+        for p in _patches(proof_amount=0, user_balance_msats=5000):
+            p.start()
+        try:
+            details, total_wallet, total_user, owner = await fetch_all_balances(
+                units=["sat"]
+            )
+        finally:
+            patch.stopall()
+
+    assert details[0]["wallet_balance"] == 0
+    assert details[0]["user_balance"] == 5
+    assert details[0]["owner_balance"] == -5
+    assert total_wallet == 0
+    assert total_user == 5
+    assert owner == -5
+
+
 @pytest.mark.asyncio
 async def test_fetch_all_balances_no_duplicate_primary_mint() -> None:
     """primary_mint already in cashu_mints is not inspected twice."""