diff --git a/tools/fleet/lib/standing.py b/tools/fleet/lib/standing.py index fc2ae192e..745027462 100644 --- a/tools/fleet/lib/standing.py +++ b/tools/fleet/lib/standing.py @@ -48,11 +48,20 @@ def install(label, role=None, prover=None, mine=None): rc, out, err = b_.run(f"cd {F} && chmod +x in/box-standing.sh in/box-exec-snapshot.sh; pgrep -f '^bash in/box-standing.sh' >/dev/null && echo supervisor-already || {{ ROLE={role} LABEL={label} WALLET={b.get('wallet')} PROVER={prover} MINE={mine} HUB_PEER={hub.get('hub_peer','')} SEED={seed} setsid nohup bash in/box-standing.sh /dev/null 2>&1 & sleep 2; echo supervisor-started; }}", 60) Registry.patch(iid, standing=True, role=role, standing_since=b.get("standing_since") or now(), doing=f"standing {role}: node, {'miner, ' if mine == '1' else ''}{'prover' if prover == '1' else 'no prover'} (supervised)") return out.strip() +DIGEST_ALERTS = os.path.join(ROOT, "digest-alerts.json") +def digest_alert(label, got, expected): + """One #incidents line per box per hour when a standing box's handshake digest is not the live one (the seed's 7bd98cc4 at 15:3xZ froze a home miner's tip for 53 minutes).""" + st = json.load(open(DIGEST_ALERTS)) if os.path.exists(DIGEST_ALERTS) else {} + if time.time() - st.get(label, 0) < 3600: return + what = f"Fleet box {label} answers handshakes with consensus digest {got[:8]}, the live object is {expected[:8]}: peers on the live object refuse it and a node whose only peer it is sits on a frozen tip" + subprocess.run(["node", "/Users/joshm/Projects/igneum/tools/community/discord-hooks.mjs", "incident", "open", "--what", what, "--affected", "nodes that peer only with that box (a home miner reads synced on a frozen tip)", "--doing", "the fleet agent restarts the box's node on the live object; the supervisor's five-minute check found it", "--id", f"digest-{label}-{int(time.time())}", "--live"], capture_output=True, text=True, timeout=60) + st[label] = time.time(); json.dump(st, open(DIGEST_ALERTS, "w")); print(now(), "digest alert", label, got[:8], "expected", expected[:8], flush=True) def check_one(r): b = Box.from_registry(r["label"]); row = dict(r); row["t"] = now() - rc, out, err = b.run("echo alive; pgrep -fc '^bash in/box-standing.sh'; tail -1 /root/fleet/out/standing.log 2>/dev/null | cut -c1-400", 45) + rc, out, err = b.run("echo alive; pgrep -fc '^bash in/box-standing.sh'; tail -1 /root/fleet/out/standing.log 2>/dev/null | cut -c1-400; echo DIGEST=$(grep -o 'digest: [0-9a-f]*' /root/fleet/node.log | tail -1 | awk '{print substr($2,1,16)}')", 45) if rc != 0 or "alive" not in out: row.update(alive=False); return row - lines = out.strip().split("\n"); row.update(alive=True, supervisor=int(lines[1] or 0) if len(lines) > 1 else 0, last=lines[2] if len(lines) > 2 else "") + lines = [l for l in out.strip().split("\n") if not l.startswith("DIGEST=")]; row["digest"] = next((l.split("=", 1)[1] for l in out.strip().split("\n") if l.startswith("DIGEST=")), "") + row.update(alive=True, supervisor=int(lines[1] or 0) if len(lines) > 1 else 0, last=lines[2] if len(lines) > 2 else "") kv = dict(x.split("=", 1) for x in row["last"].split() if "=" in x and not x.startswith("miner=")) row.update(blocks=int(kv.get("blocks", 0) or 0), synced=kv.get("synced") == "true", exec_tip=int(kv.get("exec", 0) or 0), version=kv.get("version", ""), node_pid=kv.get("node", ""), prover=int(kv.get("prover", 0) or 0), bin=kv.get("bin", "")) row["bin_sha16"] = os.path.basename(row["bin"]).replace("igneumd-", "") if "igneumd-" in row["bin"] and len(os.path.basename(row["bin"])) == 24 else "" @@ -61,8 +70,10 @@ def check(write=True): from concurrent.futures import ThreadPoolExecutor rows = roster(); m = manifest(); want = m.get("version", "") with ThreadPoolExecutor(16) as ex: res = list(ex.map(check_one, rows)) - reg = Registry.load() + reg = Registry.load(); expected = open(os.path.join(ROOT, "expected-digest")).read().strip()[:16] if os.path.exists(os.path.join(ROOT, "expected-digest")) else "" for r in res: + r["expected_digest"] = expected; r["digest_ok"] = (not expected) or (not r.get("digest")) or r["digest"][:16] == expected + if expected and r.get("digest") and not r["digest_ok"]: digest_alert(r["label"], r["digest"], expected) r["manifest_version"] = want; wanted = (reg.get(r["iid"]) or {}).get("node_sha16_wanted") r["behind"] = bool(wanted and r.get("bin_sha16") and r["bin_sha16"] != wanted) # only an explicit publish sets the want; the package version string is not a node version if write: @@ -136,7 +147,7 @@ def loop(every=600): else: dead.pop(r["label"], None) if r["behind"]: print(now(), "behind:", r["label"], r.get("bin_sha16"), "wanted", (Registry.load().get(r["iid"]) or {}).get("node_sha16_wanted"), "(a publish script moves it; the loop only reports)", flush=True) - print(now(), "standing check:", len(res), "boxes,", sum(1 for r in res if r["alive"]), "alive,", sum(1 for r in res if r.get("synced")), "synced,", sum(1 for r in res if r["behind"]), "behind", flush=True) + print(now(), "standing check:", len(res), "boxes,", sum(1 for r in res if r["alive"]), "alive,", sum(1 for r in res if r.get("synced")), "synced,", sum(1 for r in res if r["behind"]), "behind,", sum(1 for r in res if not r.get("digest_ok", True)), "off the live digest", flush=True) except Exception as e: print(now(), "loop error", str(e)[:200], flush=True) time.sleep(every) if __name__ == "__main__":