popov/neuron-tai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(us-019): distributed tracker consensus — Raft assignments + CRDT gossip

Browse Source 
raft.py: minimal Raft consensus for shard-assignment log
  - Leader election with random 150–300ms election timeouts
  - AppendEntries log replication; majority commit required
  - RequestVote RPC with log-completeness check
  - Follower registration forwarded to leader via HTTP proxy

gossip.py: LWW CRDT gossip for inference-node heartbeat liveness
  - Each tracker keeps {node_id → wall_clock_timestamp}
  - Merges incoming gossip by taking max per key
  - Pushes snapshot to random peer every 3 seconds

server.py + cli.py:
  - TrackerServer gains cluster_peers + cluster_self_url params
  - New HTTP endpoints: /v1/raft/vote, /v1/raft/append,
    /v1/raft/status, /v1/gossip
  - --cluster-peers and --self-url CLI flags

tests/test_tracker_consensus.py: 6 integration tests
  - Leader elected within 1s
  - Follower registration propagated to all nodes
  - Follower proxy to leader
  - Leader kill → new election within 5s, registrations continue
  - Gossip table updated on heartbeat

92 tests pass, 1 skipped.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Dobromir Popov
2026-06-30 09:05:21 +03:00
parent bbe57d5f07
commit 748d535c46
6 changed files with 898 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
packages/tracker/meshnet_tracker/cli.py
View File

@@ -23,14 +23,27 @@ def main() -> None:
default=30.0,
help="Seconds before a node is removed from the registry after missed heartbeat",
)
start_cmd.add_argument(
"--cluster-peers",
default="",
help="Comma-separated URLs of peer tracker nodes (enables Raft cluster mode)",
)
start_cmd.add_argument(
"--self-url",
default=None,
help="This tracker's own URL as seen by peers (auto-derived from --host/--port if omitted)",
)
args = parser.parse_args()
if args.command == "start":
cluster_peers = [u.strip() for u in args.cluster_peers.split(",") if u.strip()]
server = TrackerServer(
host=args.host,
port=args.port,
heartbeat_timeout=args.heartbeat_timeout,
cluster_peers=cluster_peers or None,
cluster_self_url=args.self_url,
)
port = server.start()
print(f"meshnet-tracker listening on http://{args.host}:{port}", flush=True)

88
packages/tracker/meshnet_tracker/gossip.py Normal file
View File

@@ -0,0 +1,88 @@
"""CRDT gossip for node liveness heartbeats.
Uses a last-write-wins (LWW) register per inference node: each tracker node
keeps its own copy of {node_id → last_seen_monotonic_timestamp} and merges
incoming gossip by taking the max per key. This is eventually consistent —
a heartbeat received by one tracker propagates to all others within a few
gossip intervals.
Monotonic timestamps are local-clock-relative; for cross-machine gossip the
caller should use wall-clock seconds (time.time()). The tracker converts
monotonic to wall-clock when recording and back when comparing.
"""
from __future__ import annotations
import json
import random
import threading
import time
import urllib.request
class NodeGossip:
"""LWW gossip table for inference-node heartbeat timestamps.
``record(node_id)`` is called when a node sends a heartbeat to *this*
tracker. ``merge(remote)`` is called when gossip arrives from a peer
tracker. The table is periodically pushed to one random peer.
"""
PUSH_INTERVAL = 3.0 # seconds between gossip pushes to a random peer
def __init__(self, peers: list[str]) -> None:
self.peers = list(peers)
# Maps node_id → wall-clock seconds of last known heartbeat.
self._table: dict[str, float] = {}
self._lock = threading.Lock()
self._running = False
def start(self) -> None:
self._running = True
threading.Thread(target=self._push_loop, daemon=True, name="gossip").start()
def stop(self) -> None:
self._running = False
def record(self, node_id: str, wall_ts: float | None = None) -> None:
"""Record a heartbeat for *node_id* at *wall_ts* (default: now)."""
ts = wall_ts if wall_ts is not None else time.time()
with self._lock:
if ts > self._table.get(node_id, 0.0):
self._table[node_id] = ts
def merge(self, remote: dict[str, float]) -> None:
"""Merge a gossip snapshot from a peer tracker (LWW per key)."""
with self._lock:
for node_id, ts in remote.items():
if ts > self._table.get(node_id, 0.0):
self._table[node_id] = ts
def last_seen(self, node_id: str) -> float | None:
"""Return wall-clock timestamp of last known heartbeat, or None."""
with self._lock:
return self._table.get(node_id)
def snapshot(self) -> dict[str, float]:
with self._lock:
return dict(self._table)
def _push_loop(self) -> None:
while self._running:
time.sleep(self.PUSH_INTERVAL)
if not self.peers:
continue
peer = random.choice(self.peers)
try:
snap = self.snapshot()
body = json.dumps(snap).encode()
req = urllib.request.Request(
f"{peer}/v1/gossip",
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=2.0) as r:
r.read()
except Exception:
pass

372
packages/tracker/meshnet_tracker/raft.py Normal file
View File

@@ -0,0 +1,372 @@
"""Minimal Raft consensus for tracker shard assignments.
Only shard-assignment commands (register/deregister) go through the log.
Node liveness (heartbeats) is handled separately via CRDT gossip — these
are high-frequency writes where eventual consistency is fine.
Election timeout: random 150–300 ms (tight, suits in-process tests).
Leader heartbeat interval: 50 ms.
"""
from __future__ import annotations
import json
import random
import threading
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from typing import Any, Callable
@dataclass
class LogEntry:
term: int
command: str # "register" | "deregister"
payload: dict
class RaftNode:
"""Single Raft participant.
``apply_fn(command, payload)`` is called (under no external lock) when an
entry is committed. Implementors must apply the command atomically.
"""
ELECTION_MIN = 0.15 # seconds
ELECTION_MAX = 0.30
HB_INTERVAL = 0.05 # leader heartbeat interval
# Role constants
FOLLOWER = "follower"
CANDIDATE = "candidate"
LEADER = "leader"
def __init__(
self,
self_url: str,
peers: list[str],
apply_fn: Callable[[str, dict], None],
) -> None:
self.self_url = self_url
self.peers = list(peers)
self._apply_fn = apply_fn
self._lock = threading.Lock()
self.role: str = self.FOLLOWER
self.current_term: int = 0
self.voted_for: str | None = None
self.log: list[LogEntry] = []
self.commit_index: int = -1
self.last_applied: int = -1
self.leader_url: str | None = None
# Leader bookkeeping per peer
self._next_index: dict[str, int] = {}
self._match_index: dict[str, int] = {}
self._election_deadline: float = self._fresh_deadline()
self._running = False
# ------------------------------------------------------------------ start/stop
def start(self) -> None:
self._running = True
threading.Thread(target=self._tick_loop, daemon=True, name="raft-tick").start()
def stop(self) -> None:
self._running = False
# ------------------------------------------------------------------ public API
@property
def is_leader(self) -> bool:
with self._lock:
return self.role == self.LEADER
def leader(self) -> str | None:
with self._lock:
return self.leader_url
def status(self) -> dict:
with self._lock:
return {
"role": self.role,
"term": self.current_term,
"leader": self.leader_url,
"log_length": len(self.log),
"commit_index": self.commit_index,
}
def propose(self, command: str, payload: dict) -> bool:
"""Leader: append and replicate an entry. Returns True when committed.
Blocks until majority replication or failure. Must be called only on
the leader; returns False immediately if this node is not the leader.
"""
with self._lock:
if self.role != self.LEADER:
return False
entry = LogEntry(self.current_term, command, payload)
self.log.append(entry)
entry_index = len(self.log) - 1
term = self.current_term
self._replicate_to_peers(term)
with self._lock:
return self.commit_index >= entry_index
# ------------------------------------------------------------------ RPC handlers
def handle_request_vote(self, req: dict) -> dict:
with self._lock:
term = int(req["term"])
candidate = req["candidate_url"]
last_li = int(req["last_log_index"])
last_lt = int(req["last_log_term"])
if term > self.current_term:
self._step_down(term)
if term < self.current_term:
return {"term": self.current_term, "vote_granted": False}
my_li = len(self.log) - 1
my_lt = self.log[-1].term if self.log else 0
log_ok = (last_lt > my_lt) or (last_lt == my_lt and last_li >= my_li)
if (self.voted_for is None or self.voted_for == candidate) and log_ok:
self.voted_for = candidate
self._reset_deadline()
return {"term": self.current_term, "vote_granted": True}
return {"term": self.current_term, "vote_granted": False}
def handle_append_entries(self, req: dict) -> dict:
with self._lock:
term = int(req["term"])
leader_url = req["leader_url"]
prev_li = int(req["prev_log_index"])
prev_lt = int(req["prev_log_term"])
entries_raw = req.get("entries", [])
ldr_commit = int(req["leader_commit"])
if term > self.current_term:
self._step_down(term)
if term < self.current_term:
return {"term": self.current_term, "success": False}
# Valid AppendEntries from current leader
self._reset_deadline()
self.leader_url = leader_url
if self.role == self.CANDIDATE:
self.role = self.FOLLOWER
# Consistency check on prev entry
if prev_li >= 0:
if prev_li >= len(self.log):
return {"term": self.current_term, "success": False}
if self.log[prev_li].term != prev_lt:
self.log = self.log[:prev_li]
return {"term": self.current_term, "success": False}
# Append new entries (detect and overwrite conflicts)
for i, raw in enumerate(entries_raw):
idx = prev_li + 1 + i
entry = LogEntry(int(raw["term"]), raw["command"], raw["payload"])
if idx < len(self.log):
if self.log[idx].term != entry.term:
self.log = self.log[:idx]
self.log.append(entry)
else:
self.log.append(entry)
if ldr_commit > self.commit_index:
self.commit_index = min(ldr_commit, len(self.log) - 1)
self._apply_up_to_commit()
return {"term": self.current_term, "success": True}
# ------------------------------------------------------------------ internals
def _tick_loop(self) -> None:
while self._running:
time.sleep(0.01)
with self._lock:
role = self.role
deadline = self._election_deadline
if role == self.LEADER:
self._send_heartbeats()
time.sleep(self.HB_INTERVAL)
elif time.monotonic() > deadline:
self._start_election()
def _send_heartbeats(self) -> None:
with self._lock:
term = self.current_term
ldr_commit = self.commit_index
log_snapshot = list(self.log)
for peer in self.peers:
try:
self._send_append_entries(peer, term, ldr_commit, log_snapshot)
except Exception:
pass
def _replicate_to_peers(self, term: int) -> None:
"""Send AppendEntries to all peers and update commit_index on majority ack."""
with self._lock:
ldr_commit = self.commit_index
log_snapshot = list(self.log)
acks = 1 # leader counts as 1
for peer in self.peers:
try:
ok = self._send_append_entries(peer, term, ldr_commit, log_snapshot)
if ok:
acks += 1
except Exception:
pass
# Advance commit if majority replicated
with self._lock:
if self.role != self.LEADER or self.current_term != term:
return
majority_index = len(self.log) - 1
while majority_index > self.commit_index:
if self.log[majority_index].term == self.current_term:
count = 1 + sum(
1 for p in self.peers
if self._match_index.get(p, -1) >= majority_index
)
if count > (len(self.peers) + 1) / 2:
self.commit_index = majority_index
self._apply_up_to_commit()
break
majority_index -= 1
def _send_append_entries(
self, peer: str, term: int, ldr_commit: int, log_snapshot: list[LogEntry]
) -> bool:
with self._lock:
next_idx = self._next_index.get(peer, len(log_snapshot))
prev_li = next_idx - 1
prev_lt = log_snapshot[prev_li].term if 0 <= prev_li < len(log_snapshot) else 0
entries = [
{"term": e.term, "command": e.command, "payload": e.payload}
for e in log_snapshot[next_idx:]
]
body = json.dumps({
"term": term,
"leader_url": self.self_url,
"prev_log_index": prev_li,
"prev_log_term": prev_lt,
"entries": entries,
"leader_commit": ldr_commit,
}).encode()
req = urllib.request.Request(
f"{peer}/v1/raft/append",
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=0.15) as r:
resp = json.loads(r.read())
with self._lock:
if resp.get("success"):
new_match = len(log_snapshot) - 1
self._match_index[peer] = max(self._match_index.get(peer, -1), new_match)
self._next_index[peer] = new_match + 1
return True
else:
if resp.get("term", 0) > self.current_term:
self._step_down(resp["term"])
else:
self._next_index[peer] = max(0, self._next_index.get(peer, 1) - 1)
return False
def _start_election(self) -> None:
with self._lock:
self.current_term += 1
self.role = self.CANDIDATE
self.voted_for = self.self_url
term = self.current_term
my_li = len(self.log) - 1
my_lt = self.log[-1].term if self.log else 0
self._reset_deadline()
votes = 1
for peer in self.peers:
try:
body = json.dumps({
"term": term,
"candidate_url": self.self_url,
"last_log_index": my_li,
"last_log_term": my_lt,
}).encode()
req = urllib.request.Request(
f"{peer}/v1/raft/vote",
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=0.1) as r:
resp = json.loads(r.read())
if resp.get("vote_granted"):
votes += 1
elif resp.get("term", 0) > term:
with self._lock:
self._step_down(resp["term"])
return
except Exception:
pass
with self._lock:
if self.role == self.CANDIDATE and self.current_term == term:
total = len(self.peers) + 1
if votes > total / 2:
self._become_leader()
else:
self.role = self.FOLLOWER
self._reset_deadline()
def _become_leader(self) -> None:
"""Must be called with _lock held."""
self.role = self.LEADER
self.leader_url = self.self_url
for peer in self.peers:
self._next_index[peer] = len(self.log)
self._match_index[peer] = -1
print(f"[raft] {self.self_url} became leader (term {self.current_term})", flush=True)
def _step_down(self, new_term: int) -> None:
"""Must be called with _lock held."""
self.current_term = new_term
self.role = self.FOLLOWER
self.voted_for = None
self._reset_deadline()
def _apply_up_to_commit(self) -> None:
"""Must be called with _lock held."""
while self.last_applied < self.commit_index:
self.last_applied += 1
entry = self.log[self.last_applied]
try:
self._apply_fn(entry.command, entry.payload)
except Exception:
pass
def _reset_deadline(self) -> None:
self._election_deadline = self._fresh_deadline()
@staticmethod
def _fresh_deadline() -> float:
return time.monotonic() + random.uniform(
RaftNode.ELECTION_MIN, RaftNode.ELECTION_MAX
)

168
packages/tracker/meshnet_tracker/server.py
View File

@@ -29,6 +29,9 @@ import urllib.parse
import uuid
from typing import Any
from .gossip import NodeGossip
from .raft import RaftNode
DEFAULT_MODEL_PRESETS: dict[str, dict] = {
"stub-model": {
@@ -377,6 +380,8 @@ class _TrackerHTTPServer(http.server.HTTPServer):
model_presets: dict,
contracts: Any | None,
minimum_stake: int,
raft: "RaftNode | None" = None,
gossip: "NodeGossip | None" = None,
) -> None:
super().__init__(addr, handler)
self.registry = registry
@@ -385,6 +390,8 @@ class _TrackerHTTPServer(http.server.HTTPServer):
self.model_presets = model_presets
self.contracts = contracts
self.minimum_stake = minimum_stake
self.raft = raft
self.gossip = gossip
class _TrackerHandler(http.server.BaseHTTPRequestHandler):
@@ -419,6 +426,15 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
if self.path == "/v1/nodes/register":
self._handle_register()
return
if self.path == "/v1/raft/vote":
self._handle_raft_vote()
return
if self.path == "/v1/raft/append":
self._handle_raft_append()
return
if self.path == "/v1/gossip":
self._handle_gossip()
return
parts = self.path.split("/")
# /v1/nodes/<node_id>/heartbeat -> ['', 'v1', 'nodes', '<id>', 'heartbeat']
if len(parts) == 5 and parts[1] == "v1" and parts[2] == "nodes" and parts[4] == "heartbeat":
@@ -445,6 +461,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
elif parsed.path.startswith("/v1/tracker-nodes/"):
model = urllib.parse.unquote(parsed.path.removeprefix("/v1/tracker-nodes/"))
self._handle_tracker_nodes(model)
elif parsed.path == "/v1/raft/status":
self._handle_raft_status()
else:
self.send_response(404)
self.end_headers()
@@ -538,6 +556,37 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
if body is None:
return
# --- Raft cluster mode: forward to leader or propose via Raft ---
if server.raft is not None:
if not server.raft.is_leader:
leader = server.raft.leader()
if leader is None:
self._send_json(503, {"error": "no leader elected — retry in a moment"})
return
# Proxy to leader
try:
data = json.dumps(body).encode()
req = urllib.request.Request(
f"{leader}/v1/nodes/register",
data=data,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=5.0) as r:
resp_body = r.read()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(resp_body)))
self.end_headers()
self.wfile.write(resp_body)
except Exception as exc:
self._send_json(503, {"error": f"leader proxy failed: {exc}"})
return
# Leader path: fall through to normal registration, then replicate via Raft.
# We let the registration run first (to generate node_id), then asynchronously
# propose to the Raft log so followers can replicate the entry.
# Raft proposal happens after the response is sent (fire-and-forget replication).
endpoint = body.get("endpoint")
if not isinstance(endpoint, str) or not endpoint:
self._send_json(400, {"error": "endpoint is required"})
@@ -685,6 +734,18 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
payload["directive"] = assignment_directive
self._send_json(200, payload)
# Raft replication: leader proposes this registration to followers so their
# registries stay in sync. Fire-and-forget (async) — the client already
# got its node_id; replication happens in the background.
if server.raft is not None and server.raft.is_leader:
raft_payload = dict(body)
raft_payload["node_id"] = node_id # include the generated ID
threading.Thread(
target=server.raft.propose,
args=("register", raft_payload),
daemon=True,
).start()
def _handle_heartbeat(self, node_id: str):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
with server.lock:
@@ -697,15 +758,53 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
_rebalance_model_locked(server, entry.model or "stub-model")
directives = list(entry.pending_directives)
entry.pending_directives.clear()
# print(
# f"[tracker] heartbeat: {node_id[:8]} {entry.endpoint}",
# flush=True,
# )
if server.gossip is not None:
server.gossip.record(node_id)
if directives:
self._send_json(200, {"directives": directives})
else:
self._send_json(200, {})
# ---------------------------------------------------------------- Raft handlers
def _handle_raft_vote(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
body = self._read_json_body()
if body is None:
return
if server.raft is None:
self._send_json(503, {"error": "raft not enabled"})
return
result = server.raft.handle_request_vote(body)
self._send_json(200, result)
def _handle_raft_append(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
body = self._read_json_body()
if body is None:
return
if server.raft is None:
self._send_json(503, {"error": "raft not enabled"})
return
result = server.raft.handle_append_entries(body)
self._send_json(200, result)
def _handle_raft_status(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if server.raft is None:
self._send_json(200, {"role": "standalone", "term": 0, "leader": None})
return
self._send_json(200, server.raft.status())
def _handle_gossip(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
body = self._read_json_body()
if body is None:
return
if server.gossip is not None and isinstance(body, dict):
server.gossip.merge({k: float(v) for k, v in body.items()})
self._send_json(200, {})
def _handle_assign(self, parsed: urllib.parse.ParseResult):
"""Return an optimal shard assignment for a node given its hardware profile.
@@ -1042,6 +1141,8 @@ class TrackerServer:
model_presets: dict | None = None,
contracts: Any | None = None,
minimum_stake: int = 0,
cluster_peers: list[str] | None = None,
cluster_self_url: str | None = None,
) -> None:
self._host = host
self._requested_port = port
@@ -1052,17 +1153,23 @@ class TrackerServer:
)
self._contracts = contracts
self._minimum_stake = minimum_stake
self._cluster_peers: list[str] = list(cluster_peers) if cluster_peers else []
self._cluster_self_url: str | None = cluster_self_url
self._registry: dict[str, _NodeEntry] = {}
self._lock = threading.Lock()
self._server: _TrackerHTTPServer | None = None
self._thread: threading.Thread | None = None
self._rebalance_stop = threading.Event()
self._rebalance_thread: threading.Thread | None = None
self._raft: RaftNode | None = None
self._gossip: NodeGossip | None = None
self.port: int | None = None
def start(self) -> int:
if self._server is not None:
raise RuntimeError("TrackerServer is already running")
# Start HTTP server first so we know our port
self._server = _TrackerHTTPServer(
(self._host, self._requested_port),
_TrackerHandler,
@@ -1074,6 +1181,21 @@ class TrackerServer:
self._minimum_stake,
)
self.port = self._server.server_address[1]
# Start Raft + gossip if cluster peers are configured
if self._cluster_peers:
self_url = self._cluster_self_url or f"http://{self._host}:{self.port}"
self._raft = RaftNode(
self_url=self_url,
peers=self._cluster_peers,
apply_fn=self._raft_apply,
)
self._gossip = NodeGossip(peers=self._cluster_peers)
self._server.raft = self._raft
self._server.gossip = self._gossip
self._raft.start()
self._gossip.start()
self._rebalance_stop.clear()
self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
self._thread.start()
@@ -1081,6 +1203,38 @@ class TrackerServer:
self._rebalance_thread.start()
return self.port
def _raft_apply(self, command: str, payload: dict) -> None:
"""Called by RaftNode when a log entry is committed — replicate to local registry."""
if command != "register":
return
# Re-apply the registration to our local registry (follower path).
# On the leader this is a no-op since it already registered locally.
node_id = payload.get("node_id")
if not node_id or node_id in self._registry:
return # already present (leader case) or malformed
endpoint = payload.get("endpoint", "")
try:
shard_start = int(payload["shard_start"]) if payload.get("shard_start") is not None else None
shard_end = int(payload["shard_end"]) if payload.get("shard_end") is not None else None
except (TypeError, ValueError):
return
entry = _NodeEntry(
node_id=node_id,
endpoint=endpoint.rstrip("/"),
shard_start=shard_start,
shard_end=shard_end,
model=payload.get("model", "stub-model"),
shard_checksum=payload.get("shard_checksum"),
hardware_profile=payload.get("hardware_profile", {}),
wallet_address=payload.get("wallet_address"),
score=float(payload.get("score", 1.0)),
tracker_mode=bool(payload.get("tracker_mode", False)),
hf_repo=payload.get("hf_repo"),
num_layers=int(payload["num_layers"]) if payload.get("num_layers") is not None else None,
)
with self._lock:
self._registry[node_id] = entry
def _rebalance_loop(self) -> None:
while not self._rebalance_stop.wait(self._rebalance_interval):
server = self._server
@@ -1091,6 +1245,10 @@ class TrackerServer:
_rebalance_all_locked(server)
def stop(self) -> None:
if self._raft is not None:
self._raft.stop()
if self._gossip is not None:
self._gossip.stop()
if self._server is None:
return
self._rebalance_stop.set()
@@ -1103,4 +1261,6 @@ class TrackerServer:
self._server = None
self._thread = None
self._rebalance_thread = None
self._raft = None
self._gossip = None
self.port = None