asset/arena/quantus-arena-exporter.py

@@ -19,6 +19,7 @@ import hashlib
 import http.server
 import json
 import os
+import pathlib
 import logging
 import socketserver
 import threading
@@ -372,6 +373,8 @@ ATTRIBUTION_WINDOW = 20
 # Startup backfill of the rolling window from chain history is capped by
 # wall clock so a slow node cannot hold the exporter's first poll forever.
 BACKFILL_BUDGET_S = 180.0
+# Attribution state is written at most this often (and on shutdown).
+STATE_SAVE_INTERVAL_S = 10.0
 
 GAP_BUCKETS = [0.5, 1, 2, 3, 5, 8, 12, 16, 24, 32, 48, 64, 128]
 # Block timestamp minus the moment this exporter first saw the block, seconds.
@@ -479,7 +482,7 @@ def u512_le(hex_str: str) -> int:
 
 class Arena:
     def __init__(self, rpc, self_preimage, window_blocks, target_block_time, top_n,
-                 reward_address="", telemetry=None):
+                 reward_address="", telemetry=None, state_file=""):
         self.rpc = rpc
         self.telemetry = telemetry
         self.reward_address = reward_address or None
@@ -516,6 +519,10 @@ class Arena:
         # reporter did not lead by ATTRIBUTION_LEAD_MS count as attempts only.
         self.attr = {}
         self.pending_attr = deque()   # (block hash, preimage, seen at)
+        self.state_file = state_file
+        self._state_dirty = False
+        self._state_saved_at = 0.0
+        self.load_state()
         self.drift_hist = {}      # preimage -> Hist(DRIFT_BUCKETS)
         self.consecutive = {}     # preimage -> blocks authored directly after its own
         # Tip observations only — used for the block interval, which is real
@@ -606,6 +613,55 @@ class Arena:
             self.drift_hist.setdefault(pre, Hist(DRIFT_BUCKETS)).observe(ts / 1000.0 - seen)
             self.last_tip = (height, ts, pre, seen)
 
+    def load_state(self):
+        """Restore attribution votes, held names and the peer-id -> name map
+        from the state file, so a restart does not strip every author of its
+        name for a few blocks (that churn split the per-author series)."""
+        if not self.state_file:
+            return
+        try:
+            st = json.loads(pathlib.Path(self.state_file).read_text())
+        except FileNotFoundError:
+            return
+        except Exception as e:  # noqa: BLE001 - a bad file is ignored, not fatal
+            LOG.warning("attribution state %s unreadable, starting fresh: %s", self.state_file, e)
+            return
+        key = lambda k: tuple(k) if k else None
+        with self.lock:
+            for pre, a in st.get("attr", {}).items():
+                self.attr[pre] = {"attempts": int(a.get("attempts", 0)), "attributed": int(a.get("attributed", 0)),
+                                  "votes": deque((key(k) for k in a.get("votes", [])), maxlen=ATTRIBUTION_WINDOW),
+                                  "held": key(a.get("held"))}
+        if self.telemetry is not None:
+            with self.telemetry.lock:
+                for peer, name in st.get("peer_names", {}).items():
+                    self.telemetry.peer_names.setdefault(peer, name)
+        LOG.info("attribution state restored: %s authors, %s of them named", len(st.get("attr", {})),
+                 sum(1 for a in st.get("attr", {}).values() if a.get("held")))
+
+    def save_state(self, force=False):
+        if not self.state_file or not self._state_dirty:
+            return
+        if not force and time.time() - self._state_saved_at < STATE_SAVE_INTERVAL_S:
+            return
+        with self.lock:
+            attr = {pre: {"attempts": a["attempts"], "attributed": a["attributed"],
+                          "votes": [list(k) if k else None for k in a["votes"]],
+                          "held": list(a["held"]) if a.get("held") else None}
+                    for pre, a in self.attr.items()}
+        peer_names = {}
+        if self.telemetry is not None:
+            with self.telemetry.lock:
+                peer_names = dict(self.telemetry.peer_names)
+        tmp = self.state_file + ".tmp"
+        try:
+            pathlib.Path(tmp).write_text(json.dumps({"attr": attr, "peer_names": peer_names}))
+            os.replace(tmp, self.state_file)
+            self._state_dirty = False
+            self._state_saved_at = time.time()
+        except Exception as e:  # noqa: BLE001 - persistence is best effort
+            LOG.warning("attribution state not saved to %s: %s", self.state_file, e)
+
     def resolve_attributions(self):
         """Join tip blocks with the feed's first-import reports once the
         reports have had time to arrive."""
@@ -625,7 +681,7 @@ class Arena:
                 LOG.debug("attribution: %s by %s… first import %r lead %s ms", block_hash[:10], pre[:10], name, fi[1])
             with self.lock:
                 a = self.attr.setdefault(pre, {"attempts": 0, "attributed": 0,
-                                               "votes": deque(maxlen=ATTRIBUTION_WINDOW)})
+                                               "votes": deque(maxlen=ATTRIBUTION_WINDOW), "held": None})
                 a["attempts"] += 1
                 key = None
                 if fi is not None:
@@ -634,6 +690,9 @@ class Arena:
                         key = self.telemetry.node_key(nid)
                         a["attributed"] += 1
                 a["votes"].append(key)
+                self._state_dirty = True
+        if due:
+            self.save_state()
 
     def author_display(self, pre):
         """(display, node_name, peer_id, confidence). Node name when the
@@ -641,13 +700,35 @@ class Arena:
         short = f"{pre[:8]}…{pre[-4:]}" if len(pre) > 14 else pre
         with self.lock:
             a = self.attr.get(pre)
-            votes = [k for k in a["votes"] if k is not None] if a else []
-            if not a or len(a["votes"]) < ATTRIBUTION_MIN_ATTEMPTS or not votes:
+            if not a:
                 return short, "", "", 0.0
-            key, n = max(((k, votes.count(k)) for k in set(votes)), key=lambda kv: kv[1])
-            conf = n / len(a["votes"])
-        name, peer = self.telemetry.key_name(key) if self.telemetry else ("", "")
-        if conf < ATTRIBUTION_MIN_CONFIDENCE or not name:
+            counts = {}
+            for k in a["votes"]:
+                if k is not None:
+                    counts[k] = counts.get(k, 0) + 1
+            total = len(a["votes"])
+            held = a.get("held")
+            # Hysteresis: a name, once held, stays until its node has no vote
+            # left in the window or another node out-votes it outright. A dip
+            # in confidence alone never drops back to the preimage, which
+            # would split the author's series on every wobble.
+            if held is not None and counts.get(held, 0) == 0:
+                held = None
+            if counts:
+                best, n = max(counts.items(), key=lambda kv: kv[1])
+                if held is None:
+                    if total >= ATTRIBUTION_MIN_ATTEMPTS and n / total >= ATTRIBUTION_MIN_CONFIDENCE:
+                        held = best
+                elif best != held and n > counts[held]:
+                    held = best
+            if held != a.get("held"):
+                a["held"] = held
+                self._state_dirty = True
+            conf = (counts[held] / total) if held else ((max(counts.values()) / total) if counts and total else 0.0)
+        if held is None:
+            return short, "", "", conf
+        name, peer = self.telemetry.key_name(held) if self.telemetry else ("", "")
+        if not name:
             return short, "", "", conf
         return name, name, peer, conf
 
@@ -805,16 +886,22 @@ class Arena:
         # `display` is what the leaderboard shows in place of the preimage.
         out.append("# HELP quantus_author_node_info Reward preimage to telemetry node attribution, by first block import. Value is the confidence; display is the node name when confident, else the abbreviated preimage.")
         out.append("# TYPE quantus_author_node_info gauge")
-        names_used = {}
+        # A name used by more than one node on the feed always carries the
+        # peer id suffix, whether or not the others are attributed to anyone:
+        # deciding it from the attributed set alone re-labelled the first
+        # QuantusMinerGUI the moment a second one was named.
+        peers_by_name = {}
+        if self.telemetry is not None:
+            with self.telemetry.lock:
+                for peer, name in self.telemetry.peer_names.items():
+                    peers_by_name.setdefault(name, set()).add(peer)
         rows = []
         for pre in keep:
             display, name, peer, conf = self.author_display(pre)
             rows.append([pre, display, name, peer, conf])
-            if name:
-                names_used.setdefault(name, set()).add(peer)
         for row in rows:
             pre, display, name, peer, conf = row
-            if name and len(names_used.get(name, ())) > 1:
+            if name and len(peers_by_name.get(name, ())) > 1:
                 display = f"{name} ({peer[-6:]})" if peer else name
             is_self = "true" if pre == self.self_preimage else "false"
             esc = lambda s: s.replace('\\', '\\\\').replace('"', '\\"')
@@ -948,6 +1035,8 @@ def main():
                    help="substrate-telemetry feed websocket; empty disables telemetry")
     p.add_argument("--reward-address", default="",
                    help="our wormhole SS58 address, to track accrued rewards")
+    p.add_argument("--state-file", default="",
+                   help="JSON file for attribution votes and held names, so restarts keep author names")
     p.add_argument("--self-preimage", default="",
                    help="our own --rewards-inner-hash, to label and score our share")
     args = p.parse_args()
@@ -963,7 +1052,8 @@ def main():
         threading.Thread(target=telemetry.run_forever, daemon=True).start()
 
     arena = Arena(rpc, args.self_preimage, args.window_blocks,
-                  args.target_block_time, args.top_n, args.reward_address, telemetry)
+                  args.target_block_time, args.top_n, args.reward_address, telemetry,
+                  state_file=args.state_file)
     Handler.arena = arena
 
     def loop():

asset/grafana/quantus.json

@@ -9,7 +9,7 @@
   ],
   "timezone": "browser",
   "schemaVersion": 39,
-  "version": 23,
+  "version": 24,
   "refresh": "30s",
   "time": {
     "from": "now-6h",
@@ -2323,7 +2323,7 @@
         "x": 0,
         "y": 72
       },
-      "description": "Each author's estimated hashrate over time: its share of the blocks authored in the trailing window times the network hashrate averaged over the same window, in MH/s. Authors below 10 MH/s are left out. The legend uses the telemetry node name where the arena exporter has attributed the preimage, else the abbreviated preimage; an author gets a new line when its label changes. Pools grow by headcount, so a steadily rising line with no step is a pool filling up; a step is hardware arriving or leaving. The window follows the dashboard range on auto (about a quarter of it, never under 2 h) and can be pinned with the 'author window' selector; 6 h is about 1800 blocks, at which a 3% author's line carries roughly \u00b115% sampling noise, 2 h roughly \u00b125%.",
+      "description": "Each author's estimated hashrate over time: its share of the blocks authored in the trailing window times the network hashrate averaged over the same window, in MH/s. Authors below 10 MH/s are left out. The legend uses the telemetry node name where the arena exporter has attributed the preimage, else the abbreviated preimage; a label change shows as a gap, not a slope; the exporter keeps names across restarts and confidence dips so that is rare. Pools grow by headcount, so a steadily rising line with no step is a pool filling up; a step is hardware arriving or leaving. The window follows the dashboard range on auto (about a quarter of it, never under 2 h) and can be pinned with the 'author window' selector; 6 h is about 1800 blocks, at which a 3% author's line carries roughly \u00b115% sampling noise, 2 h roughly \u00b125%.",
       "targets": [
         {
           "datasource": {
@@ -2346,7 +2346,7 @@
             "lineWidth": 2,
             "fillOpacity": 0,
             "showPoints": "never",
-            "spanNulls": 3600000
+            "spanNulls": false
           }
         },
         "overrides": [

asset/systemd/quantus-arena.service

@@ -35,6 +35,7 @@ ExecStart=/usr/bin/python3 /usr/local/bin/quantus-arena-exporter.py \
     --rpc-url http://127.0.0.1:9944 \
     --port 25033 \
     --window-blocks 3600 \
+    --state-file /var/lib/quantus-arena/attribution.json \
     --target-block-time 6 \
     --self-preimage ${QUANTUS_INNER_HASH} \
     --reward-address ${QUANTUS_REWARD_ADDRESS} \