lair/quantus
2
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

arena: name authors from telemetry first-import; leaderboard shows author, drops preimage and ours #17

Merged
grenade merged 1 commits from arena/author-names into main 2026-09-04 06:42:51 +00:00
Conversation 0 Commits 1 Files Changed 2 +186 -68
2 changed files with 186 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

155
asset/arena/quantus-arena-exporter.py
View File

@@ -18,6 +18,7 @@ import argparse
import hashlib
import http.server
import json
import os
import logging
import socketserver
import threading
@@ -149,6 +150,18 @@ class Telemetry:
self.chain_node_count = None # authoritative count from action 11
self.connected = False
self.errors = 0
# node_id -> (name, peer id). Kept across reconnects: attributions made
# while connected must still render a name after a feed hiccup.
self.names = {}
# Block imports near the tip, by hash: who reported it first and how far
# ahead of the next reporter (ms). The feed stamps the first reporter of
# a hash with propagation_time 0 and later reporters with their delay.
# A node that authored a block imports it before announcing it, so a
# large lead over the second reporter means the first reporter is the
# author; a tiny lead means the author is not on telemetry and the first
# reporter merely heard about it first (lair/quantus#11).
self.imports = {} # hash -> {"first": node_id, "ts": ms, "second_ms": ms|None, "reports": n}
self.import_order = deque(maxlen=4000)
def _ingest(self, arr):
for i in range(0, len(arr) - 1, 2):
@@ -165,9 +178,31 @@ class Telemetry:
"cores": hw.get("core_count") or 0,
"vm": bool(hw.get("is_virtual_machine")),
}
with self.lock:
self.names[nid] = (str(d[0] or "") if len(d) > 0 else "",
str(d[4] or "") if len(d) > 4 else "")
elif code == 4: # RemovedNode
with self.lock:
self.nodes.pop(payload, None)
elif code == 6: # ImportedBlock: [node_id, [height, hash, block_time, ts, propagation]]
nid, b = payload[0], payload[1]
h, ts, prop = b[1], b[3], b[4]
with self.lock:
e = self.imports.get(h)
if e is None:
if len(self.import_order) == self.import_order.maxlen:
self.imports.pop(self.import_order[0], None)
self.import_order.append(h)
e = self.imports[h] = {"first": None, "ts": None, "second_ms": None, "reports": 0}
e["reports"] += 1
if prop == 0 or e["first"] is None:
if e["first"] is None:
e["first"], e["ts"] = nid, ts
else:
# A second reporter also stamped 0: a tie, nobody leads.
e["second_ms"] = 0
elif prop is not None and prop > 0:
e["second_ms"] = prop if e["second_ms"] is None else min(e["second_ms"], prop)
elif code == 11: # AddedChain: [name, genesis, count]
if isinstance(payload, list) and len(payload) >= 3:
with self.lock:
@@ -205,6 +240,19 @@ class Telemetry:
self.nodes.clear()
time.sleep(15)
def first_import(self, block_hash):
"""(node_id, lead_ms) for the first reporter of a block hash, or None.
lead_ms is None until a second reporter has been seen."""
with self.lock:
e = self.imports.get(block_hash)
if e is None or e["first"] is None:
return None
return e["first"], e["second_ms"]
def node_name(self, nid):
with self.lock:
return self.names.get(nid, ("", ""))
def render(self, g, out):
with self.lock:
nodes = dict(self.nodes)
@@ -284,6 +332,20 @@ def decode_compact(data: bytes):
# lowest buckets right after other people's blocks is releasing withheld blocks
# (selfish mining); the network's own distribution is exponential around the
# target.
# Telemetry attribution: the author's own node reports a block's import before
# anyone else, but on a well-connected network its lead over the second
# reporter is small (measured 2026-09-04: 40-100 ms for authors on the feed,
# up to 400 ms for slow-peered ones), so the lead only has to be positive by
# this much; identity does the work instead. An author on the feed is first
# on nearly every block, an author off the feed is first-reported by a
# different peer each time, so a name is shown once this many blocks have
# been attempted and this fraction of them point at the same node, after
# waiting this long for the reports to arrive.
ATTRIBUTION_LEAD_MS = 20
ATTRIBUTION_SETTLE_S = 8.0
ATTRIBUTION_MIN_ATTEMPTS = 3
ATTRIBUTION_MIN_CONFIDENCE = 0.6
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.
# Negative is ordinary propagation plus poll lag; positive means the author
@@ -422,6 +484,11 @@ class Arena:
# in arrival timing. Resolution is the poll interval until telemetry
# first-seen replaces it (lair/quantus#11).
self.arrival_hist = {} # preimage -> Hist(GAP_BUCKETS)
# Author attribution via telemetry first import (lair/quantus#11):
# preimage -> {"attempts": n, "nodes": {node_id: n}}; blocks whose first
# reporter did not lead by ATTRIBUTION_LEAD_MS count as attempts only.
self.attr = {}
self.pending_attr = deque() # (block hash, preimage, seen at)
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
@@ -457,22 +524,25 @@ class Arena:
if at_tip:
self.observe_tip(height, pre)
def block_timestamp_ms(self, height):
"""Author timestamp of the block at `height`, from its inherent."""
def block_meta(self, height):
"""(author timestamp ms, block hash) for the block at `height`, or (None, None)."""
try:
h = self.rpc.call("chain_getBlockHash", [height])
blk = self.rpc.call("chain_getBlock", [h]) if h else None
xts = (blk or {}).get("block", {}).get("extrinsics") or []
return timestamp_inherent_ms(xts[0]) if xts else None
return (timestamp_inherent_ms(xts[0]) if xts else None), h
except Exception as e: # noqa: BLE001 - one bad block must not stop the loop
LOG.warning("block timestamp at %s unavailable: %s", height, e)
return None
LOG.warning("block metadata at %s unavailable: %s", height, e)
return None, None
def observe_tip(self, height, pre):
ts = self.block_timestamp_ms(height)
ts, block_hash = self.block_meta(height)
if ts is None:
return
seen = time.time()
if block_hash and self.telemetry is not None:
with self.lock:
self.pending_attr.append((block_hash, pre, seen))
with self.lock:
last = self.last_tip
if last is not None and height == last[0] + 1:
@@ -483,6 +553,46 @@ class Arena:
self.drift_hist.setdefault(pre, Hist(DRIFT_BUCKETS)).observe(ts / 1000.0 - seen)
self.last_tip = (height, ts, pre, seen)
def resolve_attributions(self):
"""Join tip blocks with the feed's first-import reports once the
reports have had time to arrive."""
if self.telemetry is None:
return
now = time.time()
with self.lock:
due = []
while self.pending_attr and now - self.pending_attr[0][2] >= ATTRIBUTION_SETTLE_S:
due.append(self.pending_attr.popleft())
for block_hash, pre, _ in due:
fi = self.telemetry.first_import(block_hash)
if fi is None:
LOG.debug("attribution: %s by %s… not on the feed", block_hash[:10], pre[:10])
else:
name, _ = self.telemetry.node_name(fi[0])
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, "nodes": {}})
a["attempts"] += 1
if fi is not None:
nid, lead = fi
if lead is not None and lead >= ATTRIBUTION_LEAD_MS:
a["nodes"][nid] = a["nodes"].get(nid, 0) + 1
def author_display(self, pre):
"""(display, node_name, peer_id, confidence). Node name when the
attribution is confident, else the abbreviated preimage."""
short = f"{pre[:8]}…{pre[-4:]}" if len(pre) > 14 else pre
with self.lock:
a = self.attr.get(pre)
if not a or a["attempts"] < ATTRIBUTION_MIN_ATTEMPTS or not a["nodes"]:
return short, "", "", 0.0
nid, n = max(a["nodes"].items(), key=lambda kv: kv[1])
conf = n / a["attempts"]
name, peer = self.telemetry.node_name(nid) if self.telemetry else ("", "")
if conf < ATTRIBUTION_MIN_CONFIDENCE or not name:
return short, "", "", conf
return name, name, peer, conf
def poll_once(self):
health = self.rpc.call("system_health", [])
self.syncing = bool(health.get("isSyncing", False))
@@ -500,6 +610,7 @@ class Arena:
if self.last_height is None or height > self.last_height:
self.record(head, at_tip=not self.syncing)
self.last_height = height
self.resolve_attributions()
self.difficulty = u512_le(self.rpc.call("state_call", ["QPoWApi_get_difficulty", "0x"]))
@@ -629,6 +740,36 @@ class Arena:
is_self = "true" if pre == self.self_preimage else "false"
out.append(f'quantus_blocks_authored_total{{preimage="{pre}",self="{is_self}"}} {n}')
# Who is behind a preimage, when the telemetry feed can tell. Value is
# the fraction of attempted attributions pointing at the named node;
# `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 = {}
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:
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('"', '\\"')
out.append(f'quantus_author_node_info{{preimage="{pre}",display="{esc(display)}",node_name="{esc(name)}",peer_id="{peer}",self="{is_self}"}} {conf:.3f}')
with self.lock:
attempted = {pre: a["attempts"] for pre, a in self.attr.items()}
attributed = {pre: sum(a["nodes"].values()) for pre, a in self.attr.items()}
out.append("# HELP quantus_author_attribution_attempts_total Tip blocks checked against telemetry first imports, by author.")
out.append("# TYPE quantus_author_attribution_attempts_total counter")
out.append("# HELP quantus_author_attribution_attributed_total Tip blocks whose first reporter led by enough to be called the author, by author.")
out.append("# TYPE quantus_author_attribution_attributed_total counter")
for pre in keep:
out.append(f'quantus_author_attribution_attempts_total{{preimage="{pre}"}} {attempted.get(pre, 0)}')
out.append(f'quantus_author_attribution_attributed_total{{preimage="{pre}"}} {attributed.get(pre, 0)}')
# Per-author timing, same top-N cap as the leaderboard; everyone else
# is merged into preimage="other" and the whole network into "all".
with self.lock:
@@ -751,7 +892,7 @@ def main():
help="our own --rewards-inner-hash, to label and score our share")
args = p.parse_args()
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
logging.basicConfig(level=logging.DEBUG if os.environ.get("ARENA_DEBUG") else logging.INFO, format="%(levelname)s %(message)s")
rpc = Rpc(args.rpc_url)
telemetry = None
if args.telemetry_url:

99
asset/grafana/quantus.json
View File

@@ -9,7 +9,7 @@
],
"timezone": "browser",
"schemaVersion": 39,
"version": 18,
"version": 19,
"refresh": "30s",
"time": {
"from": "now-6h",
@@ -1705,7 +1705,7 @@
},
{
"type": "table",
"title": "Authorship leaderboard (dashboard time range)",
"title": "Authorship leaderboard",
"datasource": {
"type": "prometheus",
"uid": "prometheus"
@@ -1716,7 +1716,7 @@
"x": 0,
"y": 62
},
"description": "Blocks observed authored within the selected time range, by reward preimage (cumulative counter, exists from the exporter's restart onward). est. MH/s is the author's share of blocks times the network hashrate estimate over the range: a statistical figure, \u00b110% or so at a few hundred blocks. quick-follow \u00d7: the share of the author's blocks that arrived (as seen by the exporter, 1 s poll) within 2 s of the previous block, divided by the network's share; about 1 is normal, 2 or more means blocks released right after others' (withheld). Needs 10 samples. drift median: the author's block timestamp minus when we first saw it; a few seconds negative is normal, positive is future-dated. consecutive: blocks authored directly after the author's own previous block. Timing columns fill in from when the exporter started recording them. burst \u00d7: consecutive self-blocks divided by the count expected from the author's share (blocks \u00d7 share); about 1 is normal, sustained values above 2 mean blocks held and released together. Shown only where the expected count exceeds one. Cell colours: quick-follow amber above 2 and red above 3; drift green in the normal negative band, amber once future-dated, red past 5 s, blue when more than 20 s in the past; burst amber above 2, red above 4.",
"description": "author: the telemetry node name when the arena exporter can attribute the reward preimage to a node with confidence (the same telemetry node is the first to report the import of at least 60% of 3 or more of its blocks), otherwise the abbreviated preimage. Authors whose node is not on telemetry keep the preimage; duplicate names carry the last six characters of the peer id. Blocks observed authored within the selected time range, by reward preimage (cumulative counter, exists from the exporter's restart onward). est. MH/s is the author's share of blocks times the network hashrate estimate over the range: a statistical figure, \u00b110% or so at a few hundred blocks. quick-follow \u00d7: the share of the author's blocks that arrived (as seen by the exporter, 1 s poll) within 2 s of the previous block, divided by the network's share; about 1 is normal, 2 or more means blocks released right after others' (withheld). Needs 10 samples. drift median: the author's block timestamp minus when we first saw it; a few seconds negative is normal, positive is future-dated. consecutive: blocks authored directly after the author's own previous block. Timing columns fill in from when the exporter started recording them. burst \u00d7: consecutive self-blocks divided by the count expected from the author's share (blocks \u00d7 share); about 1 is normal, sustained values above 2 mean blocks held and released together. Shown only where the expected count exceeds one. Cell colours: quick-follow amber above 2 and red above 3; drift green in the normal negative band, amber once future-dated, red past 5 s, blue when more than 20 s in the past; burst amber above 2, red above 4.",
"targets": [
{
"datasource": {
@@ -1789,6 +1789,18 @@
"range": false,
"format": "table",
"expr": "sum by (preimage) (increase(quantus_consecutive_self_blocks_total{preimage!=\"other\"}[$__range])) / on(preimage) group_left() ((increase(quantus_blocks_authored_total[$__range]) * increase(quantus_blocks_authored_total[$__range]) / scalar(sum(increase(quantus_blocks_authored_total[$__range])))) > 1)"
},
{
"datasource": {
"type": "prometheus",
"uid": "prometheus"
},
"editorMode": "code",
"refId": "G",
"instant": true,
"range": false,
"format": "table",
"expr": "quantus_author_node_info"
}
],
"transformations": [
@@ -1837,29 +1849,31 @@
"component": true,
"component 1": true,
"component 2": true,
"self 2": true
"self 2": true,
"preimage": true,
"self": true,
"self 1": true,
"node_name": true,
"peer_id": true,
"Value #G": true
},
"renameByName": {
"preimage": "reward preimage",
"self": "ours",
"self 1": "ours",
"Value #A": "blocks",
"Value #B": "est. MH/s",
"Value #C": "quick-follow \u00d7",
"Value #D": "drift median (s)",
"Value #E": "consecutive",
"Value #F": "burst \u00d7"
"Value #F": "burst \u00d7",
"display": "author"
},
"indexByName": {
"preimage": 0,
"self": 1,
"self 1": 1,
"Value #A": 2,
"Value #B": 3,
"Value #C": 4,
"Value #D": 5,
"Value #E": 6,
"Value #F": 7
"Value #F": 7,
"display": 0
}
}
},
@@ -1919,55 +1933,6 @@
}
]
},
{
"matcher": {
"id": "byName",
"options": "ours"
},
"properties": [
{
"id": "custom.width",
"value": 70
},
{
"id": "mappings",
"value": [
{
"type": "value",
"options": {
"true": {
"text": "\u25c0 us",
"color": "green",
"index": 0
},
"false": {
"text": "",
"index": 1
}
}
}
]
},
{
"id": "custom.cellOptions",
"value": {
"type": "color-text"
}
}
]
},
{
"matcher": {
"id": "byName",
"options": "reward preimage"
},
"properties": [
{
"id": "custom.width",
"value": 620
}
]
},
{
"matcher": {
"id": "byName",
@@ -2187,6 +2152,18 @@
}
}
]
},
{
"matcher": {
"id": "byName",
"options": "author"
},
"properties": [
{
"id": "custom.width",
"value": 260
}
]
}
]
},