#!/usr/bin/env python3 """np-stratum-probe — times how long a pool takes to hand you WORK, not to greet you. Connects to one or more Stratum V1 destinations you choose, from YOUR network, and times the session by milestone. Then it does NOT hang up: it stays connected and records when fresh work arrives after each network tip change, which is the only thing that decides whether your rig hashes on live work or dead work. No dependencies: Python 3.8+ standard library only. python3 np-stratum-probe.py --addr nexuspool.io:3350 python3 np-stratum-probe.py --addr --tips 3 a.pool:3333 b.pool:3333 With two or more destinations it connects to all of them at once from this same machine, so the delta between them over the SAME block is exact without your clock being right. License: MIT, same as the rest of NexusPool. """ import argparse import json import socket import ssl import statistics import sys import threading import time CLOCK = time.perf_counter # monotonic: immune to NTP stepping the clock PROTO_TIMEOUT_S = 12.0 # window for one application round trip CONNECT_TIMEOUT_S = 8.0 # window for the TCP establishment READ_CHUNK = 8192 def parse_target(spec): """'host:port' -> (host, port). IPv6 goes in brackets: '[::1]:3350'.""" if spec.startswith("["): host, _, rest = spec[1:].partition("]") return host, int(rest.lstrip(":")) host, _, port = spec.rpartition(":") if not host: raise ValueError(f"destination has no port: {spec!r} (use host:port)") return host, int(port) def block_hash_from_prevhash(prevhash_hex): """A stratum prevhash arrives with its eight 4-byte groups swapped among themselves AND each group little-endian. Reverse both and you get the hash the way any block explorer shows it, so you can check for yourself which chain you are being served.""" try: raw = bytes.fromhex(prevhash_hex) except ValueError: return None if len(raw) != 32: return None groups = [raw[i:i + 4][::-1] for i in range(0, 32, 4)] return b"".join(reversed(groups)).hex() class Probe: """One session against one destination. Times milestones, then listens for tip changes.""" def __init__(self, label, host, port, addr, addr2, worker, use_tls, tips_wanted): self.label = label self.host = host self.port = port self.addr = addr self.addr2 = addr2 self.worker = worker self.use_tls = use_tls self.tips_wanted = tips_wanted self.milestones = {} # name -> ms since t0 self.tips = [] # every prevhash change observed self.error = None self.ip = None self.dns_ms = None self._current_prevhash = None # -- send / receive ---------------------------------------------------- def _send(self, sock, obj): sock.sendall((json.dumps(obj) + "\n").encode()) def _user(self): """A merged-mining chain (Litecoin+Dogecoin) needs BOTH addresses in the username, because each leg pays out on its own chain. With --addr2 the dual username is built; without it, the plain one goes out.""" base = f"{self.addr}.{self.addr2}" if self.addr2 else self.addr return f"{base}.{self.worker}" def _done(self): """Without --tips the first job is enough; with --tips we wait for that many real tip changes. Careful: 'len(tips) >= 0' is true on entry and would cut the session before the very first notify.""" if self.tips_wanted <= 0: return "first_notify" in self.milestones return len(self.tips) >= self.tips_wanted def run(self, t0_global, deadline): try: self._run(t0_global, deadline) except Exception as e: # noqa: BLE001 self.error = f"{type(e).__name__}: {e}" def _run(self, t0_global, deadline): # DNS OUTSIDE the network stopwatch: resolving it inside mixes your resolver's own # cost into the pool's latency and ruins the comparison between destinations. t_dns = CLOCK() info = socket.getaddrinfo(self.host, self.port, proto=socket.IPPROTO_TCP) self.dns_ms = (CLOCK() - t_dns) * 1000 family, kind, proto, _, sockaddr = info[0] self.ip = sockaddr[0] t0 = CLOCK() sock = socket.socket(family, kind, proto) sock.settimeout(CONNECT_TIMEOUT_S) sock.connect(sockaddr) self.milestones["connect"] = (CLOCK() - t0) * 1000 if self.use_tls: ctx = ssl.create_default_context() sock = ctx.wrap_socket(sock, server_hostname=self.host) self.milestones["tls"] = (CLOCK() - t0) * 1000 sock.settimeout(PROTO_TIMEOUT_S) self._send(sock, {"id": 1, "method": "mining.subscribe", "params": ["np-stratum-probe/1.0"]}) self._send(sock, {"id": 2, "method": "mining.authorize", "params": [self._user(), "x"]}) buf = b"" saw_subscribe = False saw_authorize = False while CLOCK() < deadline: try: chunk = sock.recv(READ_CHUNK) except socket.timeout: if self._done(): break continue if not chunk: self.error = "the pool closed the connection" break buf += chunk while b"\n" in buf: line, buf = buf.split(b"\n", 1) line = line.strip() if not line: continue try: msg = json.loads(line) except ValueError: continue # junk or a fragment: ignore it # Dispatch by message TYPE, never by arrival order: a pool may interleave # a notify between the acks, and they do. if msg.get("id") == 1 and not saw_subscribe: saw_subscribe = True self.milestones["handshake"] = (CLOCK() - t0) * 1000 elif msg.get("id") == 2 and not saw_authorize: saw_authorize = True self.milestones["authorize"] = (CLOCK() - t0) * 1000 if msg.get("result") is False: # A decent pool says WHY it refused. Stratum's shape is # error:[code, message, data]; that message is more useful than any # generic text we could invent, so it is shown verbatim. err = msg.get("error") detail = err[1] if isinstance(err, list) and len(err) > 1 else None self.error = (f"authorize refused: {detail}" if detail else "authorize refused (the pool did not say why)") elif msg.get("method") == "mining.notify": p = msg.get("params") or [] if len(p) < 9: continue prevhash, clean = p[1], bool(p[8]) now = (CLOCK() - t0_global) * 1000 if "first_notify" not in self.milestones: self.milestones["first_notify"] = (CLOCK() - t0) * 1000 self._current_prevhash = prevhash elif prevhash != self._current_prevhash: # A real tip change: the network found a block. self._current_prevhash = prevhash self.tips.append({ "t_global_ms": now, "prevhash": prevhash, "block_hash": block_hash_from_prevhash(prevhash), "jobid": p[0], "clean_jobs": clean, }) if self._done(): break try: sock.close() except OSError: pass def main(): ap = argparse.ArgumentParser( description="Time how long a stratum pool takes to hand you WORK.", epilog="Run this from the SAME network your miners are on: measuring from another " "connection gives you a result that is not yours.") ap.add_argument("targets", nargs="+", metavar="HOST:PORT") ap.add_argument("--addr", required=True, help="your payout address (it is the username in stratum)") ap.add_argument("--addr2", default=None, help="second address, for merged-mining chains where the username is " "ADDR1.ADDR2.worker (e.g. Litecoin+Dogecoin)") ap.add_argument("--worker", default="probe") ap.add_argument("--tips", type=int, default=0, help="how many tip changes to wait for (0 = do not wait, handshake " "only). Each tip takes about 10 min on Bitcoin.") ap.add_argument("--max-min", type=float, default=40.0, help="minutes to wait before giving up (default 40)") ap.add_argument("--tls", action="store_true", help="use TLS against every destination") ap.add_argument("--json", action="store_true", help="raw JSON output") args = ap.parse_args() probes = [] for spec in args.targets: host, port = parse_target(spec) probes.append(Probe(spec, host, port, args.addr, args.addr2, args.worker, args.tls, args.tips)) t0_global = CLOCK() deadline = t0_global + args.max_min * 60 if args.tips else t0_global + 30 threads = [threading.Thread(target=p.run, args=(t0_global, deadline), daemon=True) for p in probes] for t in threads: t.start() for t in threads: t.join(timeout=args.max_min * 60 + 30) out = { "measured_at_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "contract": { "t0": "immediately before the TCP connect; DNS is resolved BEFORE and " "reported separately", "first_notify": "first usable work after connecting", "tip_*": "arrival of fresh work after a REAL network tip change, on an " "already-open connection", "note": "deltas between destinations are exact because they share a clock; " "the absolute numbers depend on yours", }, "targets": [], } for p in probes: out["targets"].append({ "target": p.label, "ip": p.ip, "dns_ms": p.dns_ms, "milestones_ms": p.milestones, "tips": p.tips, "error": p.error, }) if args.json: print(json.dumps(out, indent=2)) return print(f"\nnp-stratum-probe — {out['measured_at_utc']}") print(f"{'target':<26} {'DNS':>8} {'Connect':>9} {'Handshake':>10} {'WORK':>9} status") print("-" * 82) for d in out["targets"]: m = d["milestones_ms"] f = lambda k: f"{m[k]:.1f}ms" if k in m else "—" # noqa: E731 dns = f"{d['dns_ms']:.1f}ms" if d["dns_ms"] is not None else "—" status = d["error"] or ("ok" if "first_notify" in m else "no work") print(f"{d['target']:<26} {dns:>8} {f('connect'):>9} {f('handshake'):>10} " f"{f('first_notify'):>9} {status}") # The race only means anything with 2+ destinations over the SAME block. live = [d for d in out["targets"] if d["tips"]] if len(live) >= 2: print("\nTip race — who handed you work first on each block:") by_block = {} for d in live: for t in d["tips"]: by_block.setdefault(t["block_hash"] or t["prevhash"], []).append( (d["target"], t["t_global_ms"])) for bh, entries in by_block.items(): if len(entries) < 2: continue entries.sort(key=lambda x: x[1]) base = entries[0][1] print(f"\n block {bh[:20]}…") for name, t in entries: delta = t - base print(f" {name:<26} {'FIRST' if delta == 0 else f'+{delta:.0f}ms'}") elif args.tips: for d in out["targets"]: if len(d["tips"]) >= 2: gaps = [d["tips"][i]["t_global_ms"] - d["tips"][i - 1]["t_global_ms"] for i in range(1, len(d["tips"]))] print(f"\n{d['target']}: {len(d['tips'])} tips, " f"median between blocks {statistics.median(gaps) / 1000:.0f}s") print("\nWhat these numbers mean:") print(" DNS, Connect and Handshake are paid ONCE, when you connect. They do not") print(" repeat while your rig mines, so they do not affect what it earns.") print(" What does matter is the WORK column and, above all, the tip race: that is") print(" where you see how long your rig hashes on work that is already dead.") print(" A distant pool that spots the block quickly serves you work BEFORE a close") print(" one that hears about it late.") if __name__ == "__main__": try: main() except KeyboardInterrupt: sys.exit(130)