From 2b4f70442af28093526a8e07ae50d8be6ce39eaf Mon Sep 17 00:00:00 2001 From: thefux Date: Wed, 22 Jul 2026 00:06:33 +0000 Subject: [PATCH] fix: recover stale Cashu reservations safely --- routstr/wallet.py | 17 +- scripts/reconcile_reserved_proofs.py | 233 +++++++++++++++++++++++ scripts/retry_minibits_reconciliation.sh | 62 ++++++ tests/unit/test_fetch_all_balances.py | 31 ++- 4 files changed, 334 insertions(+), 9 deletions(-) create mode 100755 scripts/reconcile_reserved_proofs.py create mode 100755 scripts/retry_minibits_reconciliation.sh diff --git a/routstr/wallet.py b/routstr/wallet.py index 03d45fb8..5ea4025b 100644 --- a/routstr/wallet.py +++ b/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, diff --git a/scripts/reconcile_reserved_proofs.py b/scripts/reconcile_reserved_proofs.py new file mode 100755 index 00000000..4ae3fa48 --- /dev/null +++ b/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()) diff --git a/scripts/retry_minibits_reconciliation.sh b/scripts/retry_minibits_reconciliation.sh new file mode 100755 index 00000000..8851ea14 --- /dev/null +++ b/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 diff --git a/tests/unit/test_fetch_all_balances.py b/tests/unit/test_fetch_all_balances.py index dcd99107..433e84d4 100644 --- a/tests/unit/test_fetch_all_balances.py +++ b/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."""