From ae73ce997069ec3bbe6f874a3a97eb8062bc413e Mon Sep 17 00:00:00 2001 From: igneum-labs <337424239+igneum-labs@users.noreply.github.com> Date: Tue, 6 Oct 2026 15:52:59 +0000 Subject: [PATCH] fleet: night.py (the prover night loop: launches on the executed tip, re-runs the exec restart on a reset, 15-minute night rows), the tarball sha check in the swap, the prover script's FLOOR line restored --- tools/fleet/box-node-swap.sh | 3 ++ tools/fleet/box-prover.py | 3 +- tools/fleet/night.py | 67 ++++++++++++++++++++++++++++++++++++ 3 files changed, 72 insertions(+), 1 deletion(-) create mode 100644 tools/fleet/night.py diff --git a/tools/fleet/box-node-swap.sh b/tools/fleet/box-node-swap.sh index 7c26bf07..edecdf77 100755 --- a/tools/fleet/box-node-swap.sh +++ b/tools/fleet/box-node-swap.sh @@ -20,8 +20,11 @@ chmod +x $F/in/igneumd-0313 t0=$(date +%s) bash $F/in/box-kill.sh >/dev/null 2>&1 # every stage process; the node is stopped on the next line pkill -x igneumd; pkill -f igneumd-0313; sleep 4; pkill -9 -x igneumd 2>/dev/null; pkill -9 -f igneumd-0313 2>/dev/null; sleep 1 +TGZ_SHA=e67cc6493cd86dc0a993c3e1005b1df43b661fc4973ef7f0095f602f1536cd85 if [ ! -f $F/node/.full-history ]; then + [ "$(sha256sum $F/observer-datadir.tgz 2>/dev/null | cut -c1-64)" = "$TGZ_SHA" ] || rm -f $F/observer-datadir.tgz # a partial pre-pull is not a tarball [ -s $F/observer-datadir.tgz ] || scp -i $F/in/fleet-internal -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -P "$HUB_PORT" "root@$HUB_SSH:/root/fleet/share/observer-datadir.tgz" $F/observer-datadir.tgz || { echo "RESULT swap_failed pull"; exit 2; } + [ "$(sha256sum $F/observer-datadir.tgz | cut -c1-64)" = "$TGZ_SHA" ] || { echo "RESULT swap_failed tarball sha256 $(sha256sum $F/observer-datadir.tgz | cut -c1-16)"; exit 2; } rm -rf $F/node.proof && mv $F/node $F/node.proof && mkdir -p $F/node && tar -C $F/node -xzf $F/observer-datadir.tgz 2>/dev/null && touch $F/node/.full-history || { echo "RESULT swap_failed untar"; exit 2; } echo "RESULT datadir_in_place $(stamp) s=$(( $(date +%s) - t0 )) $(du -sh $F/node | cut -f1)" fi diff --git a/tools/fleet/box-prover.py b/tools/fleet/box-prover.py index dd426bec..4adcae74 100644 --- a/tools/fleet/box-prover.py +++ b/tools/fleet/box-prover.py @@ -21,7 +21,8 @@ Env: LABEL (the key label, kept for the box's life), WALLET (payout), THRESHOLD MINER (keep|pause), RUN_HOURS (default 9). """ import json, os, sys, time, subprocess, datetime, binascii, urllib.request, signal -F = "/root/fleet"; CARD = os.environ.get("FLEET_CARD"); OUT = f"{F}/card{CARD}/out" if CARD else f"{F}/out" # a rig runs one loop per card under /root/fleet/card/; B = "/opt/igneum/pkg/bin"; FLOOR = "/opt/igneum-floor" +# a rig runs one loop per card under /root/fleet/card/ (FLEET_CARD) +F = "/root/fleet"; CARD = os.environ.get("FLEET_CARD"); OUT = f"{F}/card{CARD}/out" if CARD else f"{F}/out"; B = "/opt/igneum/pkg/bin"; FLOOR = "/opt/igneum-floor" HOST = f"{FLOOR}/bin/igneum-prove-host"; EXPORT = f"{FLOOR}/bin/igneum-prove-export" EVM = "http://127.0.0.1:26790"; GRPC = "grpc://127.0.0.1:26610"; CHAIN = "igneum-devnet" LABEL = os.environ.get("LABEL", "box"); WALLET = os.environ.get("WALLET", "0x" + "19" * 20) diff --git a/tools/fleet/night.py b/tools/fleet/night.py new file mode 100644 index 00000000..d3003a9c --- /dev/null +++ b/tools/fleet/night.py @@ -0,0 +1,67 @@ +#!/usr/bin/env python3 +"""The fleet night loop (phase 2): every 60 s reads each box's swap log and prover state; launches the prover on a box +whose replay reached the tip (once); re-runs the exec restart on a box whose executed tip fell back to 0 for more than +two minutes (the deep-reorg class of 15:43Z, 6 October 2026); every 15 minutes writes a night row (provers up, claims, +submitted, paid segments, shards paid, the chain's segment window from the hub) to ~/Desktop/fleet/night.json and the +page, and appends the hourly row to the plan's fleet night section when the hour turns. Log: ~/Desktop/fleet/night.log.""" +import sys, os, time, json, datetime, subprocess +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))); import fleet +from concurrent.futures import ThreadPoolExecutor +SHA = "d6350586fe837b1f71696546628ec071b6accce2531aef776cf2b1e5487a8cdc"; DIG = "b18ed271f75dd46406d230f4156c37472127415a4c32c558bac662f6f840e61c" +LOG = os.path.join(fleet.ROOT, "night.log"); NIGHT = os.path.join(fleet.ROOT, "night.json") +def now(): return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") +def log(s): open(LOG, "a").write(f"{now()} {s}\n"); print(now(), s, flush=True) +PROBE = r"""grep -E '^RESULT (swap_done|swap_failed)' /root/fleet/out/node-swap.log 2>/dev/null | grep -v step=binary | tail -1 | cut -c1-60 +curl -s -m 6 -X POST -H 'Content-Type: application/json' --data '{"jsonrpc":"2.0","id":1,"method":"igneum_getExecStatus","params":[]}' http://127.0.0.1:26790/ | python3 -c 'import sys,json; r=sys.stdin.read(); d=json.loads(r).get("result",{}) if r.strip() else {}; print("exec", int(d.get("executedTip","0x0"),16), "blocked", "yes" if d.get("blocked") else "no", "from", str(d.get("startedFrom",""))[:30])' 2>/dev/null +curl -s -m 6 -X POST -H 'Content-Type: application/json' --data '{"jsonrpc":"2.0","id":1,"method":"igneum_getProvingStatus","params":[]}' http://127.0.0.1:26790/ | python3 -c 'import sys,json; r=sys.stdin.read(); d=json.loads(r).get("result",{}) if r.strip() else {}; v=d.get("v1",{}); w=v.get("segmentsInWindow",{}); print("tipDaa", int(d.get("tipDaa","0x0"),16), "fresh", v.get("freshRuleActive"), "paidShards", d.get("paidShards"), "pending", w.get("pending"), "proven", w.get("proven"), "unproven", w.get("unproven"), "paidSeg", v.get("paidSegments"))' 2>/dev/null +/opt/igneum/pkg/bin/igneum-miner watch 1 grpc://127.0.0.1:26610 2>/dev/null | grep -o 'daa=[0-9]*' | tail -1 +pgrep -c -f 'box-prover.py' +for f in /root/fleet/out/prover-state.json /root/fleet/card*/out/prover-state.json; do [ -f $f ] && python3 -c 'import json,sys; s=json.load(open(sys.argv[1])); print("ps", s.get("claimed",0), s.get("submitted",0), s.get("paid",0), s.get("shards_accepted",0), s.get("shards_refused",0), s.get("segment_refused",0), s.get("held",0), s.get("miner_mhs",0))' $f; done""" +def probe(b): + r, out, err = fleet.ssh(b, PROBE, timeout=45) + if r not in (0,): return None + l = out.strip().split("\n"); d = {"swap": "", "exec": None, "blocked": "", "from": "", "tipDaa": None, "fresh": "", "paidShards": None, "pending": None, "proven": None, "unproven": None, "paidSeg": None, "daa": None, "provers": 0, "ps": []} + for x in l: + if x.startswith("RESULT swap"): d["swap"] = x + elif x.startswith("exec "): p = x.split(); d["exec"] = int(p[1]); d["blocked"] = p[3]; d["from"] = " ".join(p[5:]) + elif x.startswith("tipDaa"): p = x.split(); d.update(tipDaa=int(p[1]), fresh=p[3], paidShards=p[5], pending=p[7], proven=p[9], unproven=p[11], paidSeg=p[13]) + elif x.startswith("daa="): d["daa"] = int(x[4:]) + elif x.isdigit(): d["provers"] = int(x) + elif x.startswith("ps "): d["ps"].append([float(v) for v in x.split()[1:]]) + return d +zero_since = {}; launched = set(); last_row = 0; last_hour = None +hub = next(b for b in fleet.load().values() if b.get("hub") and b.get("state") != "destroyed" and b.get("hub_peer")) +while True: + reg = fleet.load() + live = [(iid, b) for iid, b in reg.items() if b.get("state") != "destroyed" and b.get("ssh_host") and (b.get("phase") in ("1", "2") or b.get("hub"))] + with ThreadPoolExecutor(14) as ex: res = dict(zip([i for i, _ in live], ex.map(lambda x: probe(x[1]), live))) + tot = {"provers": 0, "claimed": 0, "submitted": 0, "paid": 0, "shards": 0, "refused": 0, "segref": 0, "held": 0, "exec_ok": 0, "boxes": 0} + hubd = None + for iid, b in live: + d = res.get(iid) + if not d: continue + tot["boxes"] += 1 + if b.get("hub"): hubd = d + at_tip = d["exec"] is not None and d["tipDaa"] and d["daa"] and d["tipDaa"] >= d["daa"] - 30 and d["exec"] > 0 + if at_tip: tot["exec_ok"] += 1; zero_since.pop(iid, None) + elif d["exec"] == 0 and "swap_done" in d["swap"]: + zero_since.setdefault(iid, time.time()) + if time.time() - zero_since[iid] > 150: + fleet.ssh(b, f"cd /root/fleet && bash in/box-kill.sh >/dev/null 2>&1; mv out/node-swap.log out/node-swap-reset-$(date +%s).log 2>/dev/null; STEP=file NODE_SHA256={SHA} EXPECT_DIGEST={DIG} HUB_PEER={hub['hub_peer']} HUB_SSH={hub['ssh_host']} HUB_PORT={hub['ssh_port']} setsid nohup in/box-node-swap.sh /dev/null 2>&1 &", timeout=60) + log(f"{b['label']}: exec at 0 for 150 s after a completed replay; exec restart re-run"); zero_since.pop(iid, None); launched.discard(iid) + if at_tip and iid not in launched and d["provers"] == 0: + if (b.get("num_gpus") or 1) > 1: fleet.ssh(b, f"cd /root/fleet && LABEL={b['label']} WALLET={b['wallet']} setsid nohup in/box-rig-prover.sh out/rig-provers.log 2>&1 &", timeout=60) + else: fleet.ssh(b, f"cd /root/fleet && mv out/prover.log out/prover-prev-$(date +%s).log 2>/dev/null; LABEL={b['label']} WALLET={b['wallet']} HUB_PEER={hub['hub_peer']} setsid nohup in/box-prover.sh /dev/null 2>&1 &", timeout=60) + launched.add(iid); fleet.patch(iid, stage="prover", state="running", doing=f"phase 2 prover on the executed tip ({d['exec']})"); log(f"{b['label']}: replay at the tip (exec {d['exec']}), prover started") + tot["provers"] += d["provers"] + for p in d["ps"]: tot["claimed"] += p[0]; tot["submitted"] += p[1]; tot["paid"] += p[2]; tot["shards"] += p[3]; tot["refused"] += p[4]; tot["segref"] += p[5]; tot["held"] += p[6] + fleet.patch(iid, last_line=f"exec {d['exec']} tip {d['tipDaa']} provers {d['provers']} " + (" ".join(f"{int(p[0])}/{int(p[1])}/{int(p[2])}" for p in d["ps"]))) + if time.time() - last_row > 900: + row = {"hour": now()[:13], "at": now(), "provers": tot["provers"], "boxes_executing": tot["exec_ok"], "boxes": tot["boxes"], "claimed": tot["claimed"], "segments_submitted": tot["submitted"], "segments_done": tot["paid"], "shards_paid": tot["shards"], "shards_refused": tot["refused"], "segment_records_refused": tot["segref"], "held": tot["held"], + "chain": ({"tipDaa": hubd["tipDaa"], "fresh": hubd["fresh"], "pending": hubd["pending"], "proven": hubd["proven"], "unproven": hubd["unproven"], "paidSeg": hubd["paidSeg"], "paidShards": hubd["paidShards"]} if hubd else {}), + "coverage_pct": (round(100 * int(hubd["proven"]) / max(1, int(hubd["proven"]) + int(hubd["pending"]) + int(hubd["unproven"])), 1) if hubd and hubd["proven"] not in (None, "None") else None), + "note": f"{tot['exec_ok']} of {tot['boxes']} boxes executing"} + rows = json.load(open(NIGHT)) if os.path.exists(NIGHT) else []; rows.append(row); json.dump(rows, open(NIGHT, "w"), indent=1) + log("row " + json.dumps(row)); last_row = time.time() + subprocess.run([sys.executable, os.path.join(fleet.HERE, "page.py"), "publish"], capture_output=True) + time.sleep(60)