mirror of
https://github.com/ZmnSCPxj/clboss.git
synced 2026-08-15 12:50:42 +02:00
610 lines
19 KiB
Python
Executable file
610 lines
19 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
import argparse
|
|
import bisect
|
|
import json
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
|
|
COMMON_FIELDS = [
|
|
"set_base",
|
|
"set_ppm",
|
|
"baseline_base",
|
|
"baseline_ppm",
|
|
"size_mult",
|
|
"size_total_peers",
|
|
"size_less_peers",
|
|
"balance_mult",
|
|
"balance_our_msat",
|
|
"balance_total_msat",
|
|
"price_level",
|
|
"price_mult",
|
|
"mult_product",
|
|
"est_base",
|
|
"est_ppm",
|
|
]
|
|
FLOAT_FIELDS = {
|
|
"size_mult",
|
|
"balance_mult",
|
|
"price_mult",
|
|
"mult_product",
|
|
}
|
|
REL_INT_FIELDS = {
|
|
"est_base",
|
|
"est_ppm",
|
|
}
|
|
|
|
|
|
def normalize_dt(dt):
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=datetime.now().astimezone().tzinfo)
|
|
return dt.astimezone(timezone.utc)
|
|
|
|
|
|
def parse_time_arg(value):
|
|
if value is None:
|
|
return None
|
|
if value.endswith("Z"):
|
|
value = value[:-1] + "+00:00"
|
|
if re.match(r"^[+-]?\d+(?:\.\d+)?$", value):
|
|
return datetime.fromtimestamp(float(value), timezone.utc)
|
|
if re.match(r"^[0-9]{4}-[0-9]{2}$", value):
|
|
return normalize_dt(datetime.fromisoformat(f"{value}-01T00:00:00"))
|
|
if re.match(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}$", value):
|
|
return normalize_dt(datetime.fromisoformat(f"{value}T00:00:00"))
|
|
rel = re.match(r"^([+-]?)(\d+)([smhdw])$", value)
|
|
if rel:
|
|
sign, num_s, unit = rel.groups()
|
|
num = int(num_s)
|
|
delta = {
|
|
"s": timedelta(seconds=num),
|
|
"m": timedelta(minutes=num),
|
|
"h": timedelta(hours=num),
|
|
"d": timedelta(days=num),
|
|
"w": timedelta(weeks=num),
|
|
}[unit]
|
|
if sign == "-":
|
|
delta = -delta
|
|
return normalize_dt(datetime.now().astimezone() + delta)
|
|
if value.count(":") in (1, 2) and "T" not in value and "-" not in value:
|
|
today = datetime.now().astimezone().date().isoformat()
|
|
return normalize_dt(datetime.fromisoformat(f"{today}T{value}"))
|
|
return normalize_dt(datetime.fromisoformat(value))
|
|
|
|
|
|
def epoch_arg(value):
|
|
text = f"{value:.6f}"
|
|
text = text.rstrip("0").rstrip(".")
|
|
if text == "":
|
|
return "0"
|
|
return text
|
|
|
|
|
|
def add_lightning_args(parser):
|
|
parser.add_argument("--mainnet", action="store_true", help="Run on mainnet.")
|
|
parser.add_argument("--testnet", action="store_true", help="Run on testnet.")
|
|
parser.add_argument("--signet", action="store_true", help="Run on signet.")
|
|
parser.add_argument("--regtest", action="store_true", help="Run on regtest.")
|
|
parser.add_argument("--network", help="Set network explicitly.")
|
|
parser.add_argument("--lightning-dir", help="Path to lightning data directory.")
|
|
|
|
|
|
def resolve_network_option(args):
|
|
if args.network:
|
|
return f"--network={args.network}"
|
|
if args.testnet:
|
|
return "--network=testnet"
|
|
if args.signet:
|
|
return "--network=signet"
|
|
if args.regtest:
|
|
return "--network=regtest"
|
|
return "--network=bitcoin"
|
|
|
|
|
|
def resolve_lightning_dir(args):
|
|
if args.lightning_dir:
|
|
if not os.path.isdir(args.lightning_dir):
|
|
raise ValueError(f'"{args.lightning_dir}" is not a valid directory')
|
|
return args.lightning_dir
|
|
return None
|
|
|
|
|
|
def run_lightning_cli_command(lightning_dir, network_option, subcommand, *args):
|
|
command = ["lightning-cli", network_option]
|
|
if lightning_dir:
|
|
command.append(f"--lightning-dir={lightning_dir}")
|
|
command.append(subcommand)
|
|
command.extend(args)
|
|
try:
|
|
result = subprocess.run(command, capture_output=True, text=True, check=True)
|
|
except FileNotFoundError:
|
|
raise RuntimeError("lightning-cli not found in PATH")
|
|
except subprocess.CalledProcessError as e:
|
|
stderr = e.stderr.strip()
|
|
if stderr:
|
|
raise RuntimeError(f"command failed: {' '.join(command)}: {stderr}")
|
|
raise RuntimeError(f"command failed: {' '.join(command)}")
|
|
try:
|
|
return json.loads(result.stdout)
|
|
except json.JSONDecodeError as e:
|
|
raise RuntimeError(f"invalid JSON from lightning-cli: {e}")
|
|
|
|
|
|
def fetch_external(conn, since_ts, before_ts):
|
|
conn.row_factory = sqlite3.Row
|
|
where = ["e.ts >= :since"]
|
|
binds = {"since": since_ts}
|
|
if before_ts is not None:
|
|
where.append("e.ts <= :before")
|
|
binds["before"] = before_ts
|
|
query = f"""
|
|
SELECT
|
|
p.node_id AS nodeid,
|
|
e.id,
|
|
e.ts,
|
|
e.peer_id,
|
|
e.set_base,
|
|
e.set_ppm,
|
|
e.baseline_base,
|
|
e.baseline_ppm,
|
|
e.size_mult,
|
|
e.size_total_peers,
|
|
e.size_less_peers,
|
|
e.balance_mult,
|
|
e.balance_our_msat,
|
|
e.balance_total_msat,
|
|
e.price_level,
|
|
e.price_mult,
|
|
e.mult_product,
|
|
e.est_base,
|
|
e.est_ppm,
|
|
e.dedupe_hash
|
|
FROM fee_change_events e
|
|
JOIN peers p ON p.id = e.peer_id
|
|
WHERE {" AND ".join(where)}
|
|
ORDER BY p.node_id ASC, e.ts ASC, e.id ASC
|
|
"""
|
|
out = {}
|
|
for row in conn.execute(query, binds):
|
|
raw = dict(row)
|
|
nodeid = raw["nodeid"]
|
|
out.setdefault(nodeid, []).append(make_norm_record(raw))
|
|
return out
|
|
|
|
|
|
def fetch_internal(lightning_dir, network_option, nodeid, since_ts, before_ts):
|
|
args = [nodeid, epoch_arg(since_ts)]
|
|
if before_ts is not None:
|
|
args.append(epoch_arg(before_ts))
|
|
rsp = run_lightning_cli_command(
|
|
lightning_dir, network_option, "clboss-feemon-history", *args
|
|
)
|
|
history = rsp.get("history")
|
|
if not isinstance(history, list):
|
|
raise RuntimeError("unexpected clboss-feemon-history response (no history)")
|
|
out = []
|
|
for item in history:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
out.append(make_norm_record(item))
|
|
return out
|
|
|
|
|
|
def make_norm_record(raw):
|
|
if "ts" not in raw:
|
|
raise RuntimeError(f"record missing ts: {raw}")
|
|
norm = {"ts": float(raw["ts"]), "fields": {}, "raw": raw}
|
|
for field in COMMON_FIELDS:
|
|
value = raw.get(field)
|
|
if value is None:
|
|
norm["fields"][field] = None
|
|
elif field in FLOAT_FIELDS:
|
|
norm["fields"][field] = float(value)
|
|
else:
|
|
norm["fields"][field] = int(value)
|
|
return norm
|
|
|
|
|
|
def field_equal(name, a, b, float_tol, int_rel_tol):
|
|
if a is None or b is None:
|
|
return a is None and b is None
|
|
if name in FLOAT_FIELDS:
|
|
# Handle JSON precision reduction robustly across different magnitudes.
|
|
# For values near zero this is absolute tolerance, otherwise relative.
|
|
scale = max(1.0, abs(float(a)), abs(float(b)))
|
|
return abs(float(a) - float(b)) <= (float_tol * scale)
|
|
if name in REL_INT_FIELDS:
|
|
ai = int(a)
|
|
bi = int(b)
|
|
scale = max(1.0, abs(float(ai)), abs(float(bi)))
|
|
return abs(ai - bi) <= (int_rel_tol * scale)
|
|
return int(a) == int(b)
|
|
|
|
|
|
def fields_equal(ext_rec, int_rec, float_tol, int_rel_tol):
|
|
for name in COMMON_FIELDS:
|
|
if not field_equal(
|
|
name,
|
|
ext_rec["fields"][name],
|
|
int_rec["fields"][name],
|
|
float_tol,
|
|
int_rel_tol,
|
|
):
|
|
return False
|
|
return True
|
|
|
|
|
|
def record_match(ext_rec, int_rec, ts_tol, float_tol, int_rel_tol):
|
|
if abs(ext_rec["ts"] - int_rec["ts"]) > ts_tol:
|
|
return False
|
|
return fields_equal(ext_rec, int_rec, float_tol, int_rel_tol)
|
|
|
|
|
|
def find_internal_match(ext_rec, internal, start, lookahead, ts_tol, float_tol, int_rel_tol):
|
|
end = min(len(internal), start + lookahead + 1)
|
|
for j in range(start + 1, end):
|
|
if record_match(ext_rec, internal[j], ts_tol, float_tol, int_rel_tol):
|
|
return j
|
|
return None
|
|
|
|
|
|
def find_external_match(int_rec, external, start, lookahead, ts_tol, float_tol, int_rel_tol):
|
|
end = min(len(external), start + lookahead + 1)
|
|
for i in range(start + 1, end):
|
|
if record_match(external[i], int_rec, ts_tol, float_tol, int_rel_tol):
|
|
return i
|
|
return None
|
|
|
|
|
|
def field_diffs(ext_rec, int_rec, float_tol, int_rel_tol):
|
|
diffs = []
|
|
for name in COMMON_FIELDS:
|
|
a = ext_rec["fields"][name]
|
|
b = int_rec["fields"][name]
|
|
if not field_equal(name, a, b, float_tol, int_rel_tol):
|
|
diffs.append({"field": name, "external": a, "internal": b})
|
|
return diffs
|
|
|
|
|
|
def compare_peer(external, internal, ts_tol, float_tol, int_rel_tol, lookahead):
|
|
matched = 0
|
|
discrepancies = []
|
|
i = 0
|
|
j = 0
|
|
while i < len(external) and j < len(internal):
|
|
ext_rec = external[i]
|
|
int_rec = internal[j]
|
|
if record_match(ext_rec, int_rec, ts_tol, float_tol, int_rel_tol):
|
|
matched += 1
|
|
i += 1
|
|
j += 1
|
|
continue
|
|
|
|
j_match = find_internal_match(
|
|
ext_rec, internal, j, lookahead, ts_tol, float_tol, int_rel_tol
|
|
)
|
|
i_match = find_external_match(
|
|
int_rec, external, i, lookahead, ts_tol, float_tol, int_rel_tol
|
|
)
|
|
|
|
if j_match is not None and (i_match is None or (j_match - j) <= (i_match - i)):
|
|
for missing_j in range(j, j_match):
|
|
discrepancies.append({
|
|
"kind": "extra_internal",
|
|
"internal": internal[missing_j]["raw"],
|
|
})
|
|
j = j_match
|
|
continue
|
|
if i_match is not None:
|
|
for missing_i in range(i, i_match):
|
|
discrepancies.append({
|
|
"kind": "missing_internal",
|
|
"external": external[missing_i]["raw"],
|
|
})
|
|
i = i_match
|
|
continue
|
|
|
|
ts_delta = ext_rec["ts"] - int_rec["ts"]
|
|
if abs(ts_delta) <= ts_tol:
|
|
discrepancies.append({
|
|
"kind": "field_mismatch",
|
|
"ts_delta": ts_delta,
|
|
"field_diffs": field_diffs(ext_rec, int_rec, float_tol, int_rel_tol),
|
|
"external": ext_rec["raw"],
|
|
"internal": int_rec["raw"],
|
|
})
|
|
i += 1
|
|
j += 1
|
|
elif ext_rec["ts"] < int_rec["ts"]:
|
|
discrepancies.append({
|
|
"kind": "missing_internal",
|
|
"external": ext_rec["raw"],
|
|
})
|
|
i += 1
|
|
else:
|
|
discrepancies.append({
|
|
"kind": "extra_internal",
|
|
"internal": int_rec["raw"],
|
|
})
|
|
j += 1
|
|
|
|
while i < len(external):
|
|
discrepancies.append({
|
|
"kind": "missing_internal",
|
|
"external": external[i]["raw"],
|
|
})
|
|
i += 1
|
|
while j < len(internal):
|
|
discrepancies.append({
|
|
"kind": "extra_internal",
|
|
"internal": internal[j]["raw"],
|
|
})
|
|
j += 1
|
|
|
|
return matched, discrepancies
|
|
|
|
|
|
def fmt_ts(ts):
|
|
if ts is None:
|
|
return "n/a"
|
|
return f"{float(ts):.6f}"
|
|
|
|
|
|
def fmt_ts_local(ts):
|
|
if ts is None:
|
|
return "n/a"
|
|
ts_f = float(ts)
|
|
local_dt = datetime.fromtimestamp(ts_f).astimezone()
|
|
return f"{ts_f:.6f} ({local_dt.strftime('%Y-%m-%d %H:%M:%S.%f %z')})"
|
|
|
|
|
|
def neighbor_summary(target_ts, candidates):
|
|
if target_ts is None:
|
|
return {
|
|
"prev": None,
|
|
"next": None,
|
|
"nearest": None,
|
|
"nearest_delta": None,
|
|
}
|
|
if not candidates:
|
|
return {
|
|
"prev": None,
|
|
"next": None,
|
|
"nearest": None,
|
|
"nearest_delta": None,
|
|
}
|
|
idx = bisect.bisect_left(candidates, target_ts)
|
|
prev_ts = candidates[idx - 1] if idx > 0 else None
|
|
next_ts = candidates[idx] if idx < len(candidates) else None
|
|
nearest = prev_ts
|
|
if nearest is None or (
|
|
next_ts is not None and abs(next_ts - target_ts) < abs(nearest - target_ts)
|
|
):
|
|
nearest = next_ts
|
|
nearest_delta = None if nearest is None else (target_ts - nearest)
|
|
return {
|
|
"prev": prev_ts,
|
|
"next": next_ts,
|
|
"nearest": nearest,
|
|
"nearest_delta": nearest_delta,
|
|
}
|
|
|
|
|
|
def print_discrepancy(nodeid, index, discrepancy, external_ts, internal_ts):
|
|
kind = discrepancy["kind"]
|
|
print(f" discrepancy {index} ({kind}) for {nodeid}")
|
|
|
|
if kind == "missing_internal":
|
|
ext_ts = float(discrepancy["external"]["ts"])
|
|
summary = neighbor_summary(ext_ts, internal_ts)
|
|
print(
|
|
" external_ts={ext} internal_prev_ts={prev} internal_next_ts={nxt} "
|
|
"nearest_internal_delta={delta}".format(
|
|
ext=fmt_ts_local(ext_ts),
|
|
prev=fmt_ts_local(summary["prev"]),
|
|
nxt=fmt_ts_local(summary["next"]),
|
|
delta=fmt_ts(summary["nearest_delta"]),
|
|
)
|
|
)
|
|
return
|
|
if kind == "extra_internal":
|
|
int_ts = float(discrepancy["internal"]["ts"])
|
|
summary = neighbor_summary(int_ts, external_ts)
|
|
print(
|
|
" internal_ts={it} external_prev_ts={prev} external_next_ts={nxt} "
|
|
"nearest_external_delta={delta}".format(
|
|
it=fmt_ts_local(int_ts),
|
|
prev=fmt_ts_local(summary["prev"]),
|
|
nxt=fmt_ts_local(summary["next"]),
|
|
delta=fmt_ts(summary["nearest_delta"]),
|
|
)
|
|
)
|
|
return
|
|
|
|
if "ts_delta" in discrepancy:
|
|
print(f" ts_delta={discrepancy['ts_delta']:.6f}")
|
|
if "field_diffs" in discrepancy:
|
|
for diff in discrepancy["field_diffs"]:
|
|
print(
|
|
" field {field}: external={external} internal={internal}".format(
|
|
field=diff["field"],
|
|
external=diff["external"],
|
|
internal=diff["internal"],
|
|
)
|
|
)
|
|
if "external" in discrepancy:
|
|
print(" external:", json.dumps(discrepancy["external"], sort_keys=True))
|
|
if "internal" in discrepancy:
|
|
print(" internal:", json.dumps(discrepancy["internal"], sort_keys=True))
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description=(
|
|
"Validate external fee-log-parser sqlite data against "
|
|
"clboss-feemon-history."
|
|
)
|
|
)
|
|
parser.add_argument(
|
|
"--db",
|
|
default=os.path.expanduser("./clboss-fee-info.sqlite3"),
|
|
help="Path to external sqlite database (default: ./clboss-fee-info.sqlite3).",
|
|
)
|
|
parser.add_argument(
|
|
"--since",
|
|
required=True,
|
|
help=(
|
|
"Start time (Unix epoch seconds, ISO-8601, HH:MM[:SS], "
|
|
"YYYY-MM, YYYY-MM-DD, or relative +/-Nd). Naive/non-offset "
|
|
"timestamps are interpreted in local time."
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--before",
|
|
default=None,
|
|
help=(
|
|
"Optional end time (same formats as --since). "
|
|
"Naive/non-offset timestamps are interpreted in local time."
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--ts-tolerance",
|
|
type=float,
|
|
default=60.0,
|
|
help="Allowed timestamp delta in seconds (default: 60).",
|
|
)
|
|
parser.add_argument(
|
|
"--float-tolerance",
|
|
type=float,
|
|
default=1e-5,
|
|
help=(
|
|
"Float tolerance for comparison (default: 1e-5). "
|
|
"Effective limit is tolerance*max(1, |a|, |b|)."
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--int-rel-tolerance",
|
|
type=float,
|
|
default=1e-3,
|
|
help=(
|
|
"Relative tolerance for integer derived fields "
|
|
"(est_base/est_ppm), default: 1e-3. "
|
|
"Effective limit is tolerance*max(1, |a|, |b|)."
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--lookahead",
|
|
type=int,
|
|
default=3,
|
|
help="Lookahead window used to re-sync around missing/extra records (default: 3).",
|
|
)
|
|
add_lightning_args(parser)
|
|
args = parser.parse_args()
|
|
|
|
since_dt = parse_time_arg(args.since)
|
|
before_dt = parse_time_arg(args.before) if args.before is not None else None
|
|
since_ts = since_dt.timestamp()
|
|
before_ts = before_dt.timestamp() if before_dt is not None else None
|
|
if before_ts is not None and since_ts > before_ts:
|
|
raise SystemExit("Error: --since must be less than or equal to --before.")
|
|
if args.ts_tolerance < 0:
|
|
raise SystemExit("Error: --ts-tolerance must be non-negative.")
|
|
if args.float_tolerance < 0:
|
|
raise SystemExit("Error: --float-tolerance must be non-negative.")
|
|
if args.int_rel_tolerance < 0:
|
|
raise SystemExit("Error: --int-rel-tolerance must be non-negative.")
|
|
if args.lookahead < 0:
|
|
raise SystemExit("Error: --lookahead must be non-negative.")
|
|
|
|
network_option = resolve_network_option(args)
|
|
try:
|
|
lightning_dir = resolve_lightning_dir(args)
|
|
except ValueError as e:
|
|
raise SystemExit(f"Error: {e}")
|
|
|
|
try:
|
|
conn = sqlite3.connect(args.db)
|
|
except sqlite3.Error as e:
|
|
raise SystemExit(f"Error opening sqlite DB {args.db}: {e}")
|
|
|
|
try:
|
|
external_by_peer = fetch_external(conn, since_ts, before_ts)
|
|
except sqlite3.Error as e:
|
|
raise SystemExit(f"Error reading external DB schema/data: {e}")
|
|
finally:
|
|
conn.close()
|
|
|
|
peers = sorted(external_by_peer.keys())
|
|
print(
|
|
f"Loaded external data for {len(peers)} peers "
|
|
f"from {args.db} in [{since_ts}, {before_ts if before_ts is not None else 'inf'}]."
|
|
)
|
|
if not peers:
|
|
print("No external records in the selected window.")
|
|
return 0
|
|
|
|
total_matched = 0
|
|
total_external = 0
|
|
total_internal = 0
|
|
total_discrepancies = 0
|
|
peers_with_discrepancies = 0
|
|
|
|
for index, nodeid in enumerate(peers, start=1):
|
|
external = external_by_peer[nodeid]
|
|
total_external += len(external)
|
|
|
|
try:
|
|
internal = fetch_internal(
|
|
lightning_dir,
|
|
network_option,
|
|
nodeid,
|
|
since_ts,
|
|
before_ts,
|
|
)
|
|
except RuntimeError as e:
|
|
print(f"[{index}/{len(peers)}] {nodeid}: RPC error: {e}", file=sys.stderr)
|
|
return 2
|
|
|
|
total_internal += len(internal)
|
|
matched, discrepancies = compare_peer(
|
|
external,
|
|
internal,
|
|
args.ts_tolerance,
|
|
args.float_tolerance,
|
|
args.int_rel_tolerance,
|
|
args.lookahead,
|
|
)
|
|
total_matched += matched
|
|
total_discrepancies += len(discrepancies)
|
|
if discrepancies:
|
|
peers_with_discrepancies += 1
|
|
|
|
print(
|
|
f"[{index}/{len(peers)}] {nodeid}: "
|
|
f"external={len(external)} internal={len(internal)} "
|
|
f"matched={matched} discrepancies={len(discrepancies)}"
|
|
)
|
|
external_ts = [record["ts"] for record in external]
|
|
internal_ts = [record["ts"] for record in internal]
|
|
for d_idx, discrepancy in enumerate(discrepancies, start=1):
|
|
print_discrepancy(nodeid, d_idx, discrepancy, external_ts, internal_ts)
|
|
|
|
print("Summary:")
|
|
print(f" peers={len(peers)}")
|
|
print(f" peers_with_discrepancies={peers_with_discrepancies}")
|
|
print(f" external_records={total_external}")
|
|
print(f" internal_records={total_internal}")
|
|
print(f" matched_records={total_matched}")
|
|
print(f" discrepancies={total_discrepancies}")
|
|
|
|
if total_discrepancies > 0:
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|