"""US-019 integration tests: distributed tracker consensus (Raft + gossip). Three TrackerServer instances form a Raft cluster in-process. Tests verify: - Leader election completes within 1 second - Registration forwarded from follower reaches all nodes - Killing the leader triggers a new election within 5 seconds - Heartbeat gossip propagates across nodes """ from __future__ import annotations import json import time import urllib.request import pytest from meshnet_tracker.server import TrackerServer # ------------------------------------------------------------------ helpers def _get(url: str, timeout: float = 5.0) -> dict: with urllib.request.urlopen(url, timeout=timeout) as r: return json.loads(r.read()) def _post(url: str, payload: dict, timeout: float = 5.0) -> dict: data = json.dumps(payload).encode() req = urllib.request.Request( url, data=data, headers={"Content-Type": "application/json"}, method="POST" ) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read()) def _register_node(tracker_url: str, port_hint: int) -> str: resp = _post(f"{tracker_url}/v1/nodes/register", { "endpoint": f"http://127.0.0.1:{port_hint}", "model": "stub-model", "shard_start": 0, "shard_end": 15, "hardware_profile": {}, "score": 1.0, }) return resp["node_id"] def _raft_status(tracker_url: str) -> dict: return _get(f"{tracker_url}/v1/raft/status") def _wait_for_leader( urls: list[str], timeout: float = 5.0 ) -> tuple[str, list[str]]: """Poll until exactly one leader is elected; return (leader_url, follower_urls).""" deadline = time.monotonic() + timeout while time.monotonic() < deadline: leaders = [] for url in urls: try: s = _raft_status(url) if s.get("role") == "leader": leaders.append(url) except Exception: pass if len(leaders) == 1: followers = [u for u in urls if u != leaders[0]] return leaders[0], followers time.sleep(0.05) raise TimeoutError(f"No leader elected within {timeout}s") # ------------------------------------------------------------------ fixture @pytest.fixture def three_tracker_cluster(): """Start 3 TrackerServer instances as a Raft cluster. Yields list of (TrackerServer, url) tuples — index 0 first to start. """ # Use fixed starting port range for readability in logs; actual ports # assigned by OS (port=0) to avoid conflicts. trackers: list[TrackerServer] = [] ports: list[int] = [] # Phase 1: start three servers without cluster config to get ports for _ in range(3): t = TrackerServer(host="127.0.0.1", port=0) ports.append(t.start()) trackers.append(t) # Stop them — we need to restart with peer URLs now that we know ports for t in trackers: t.stop() trackers = [] urls: list[str] = [f"http://127.0.0.1:{p}" for p in ports] for i, port in enumerate(ports): peers = [u for j, u in enumerate(urls) if j != i] t = TrackerServer( host="127.0.0.1", port=port, cluster_peers=peers, cluster_self_url=urls[i], ) actual_port = t.start() assert actual_port == port, f"port mismatch: wanted {port}, got {actual_port}" trackers.append(t) yield list(zip(trackers, urls)) for t, _ in zip(trackers, urls): try: t.stop() except Exception: pass # ------------------------------------------------------------------ tests def test_leader_elected(three_tracker_cluster): "Exactly one leader is elected within 1 second of cluster start.\n\nTags: routing, tracker" _, urls = zip(*three_tracker_cluster) leader_url, followers = _wait_for_leader(list(urls), timeout=1.0) assert leader_url in urls assert len(followers) == 2 def _wait_until_follower_knows_leader(follower_url: str, timeout: float = 2.0) -> None: """Block until the follower has received a heartbeat and knows the leader URL.""" deadline = time.monotonic() + timeout while time.monotonic() < deadline: try: s = _raft_status(follower_url) if s.get("leader") and s.get("role") == "follower": return except Exception: pass time.sleep(0.05) raise TimeoutError(f"{follower_url} still doesn't know the leader after {timeout}s") def test_registration_on_follower_visible_on_all_nodes(three_tracker_cluster): "Registering via a follower propagates the entry to all tracker nodes.\n\nTags: routing, tracker" trackers_urls = three_tracker_cluster trackers, urls = zip(*trackers_urls) urls = list(urls) leader_url, followers = _wait_for_leader(urls, timeout=1.0) # Wait until the follower has received at least one heartbeat from the leader follower = followers[0] _wait_until_follower_knows_leader(follower, timeout=2.0) # Register via a follower _register_node(follower, port_hint=19999) # Allow replication to propagate (Raft heartbeat interval is 50ms) time.sleep(0.5) # Every tracker node must know about this node for url in urls: # Use route endpoint as a proxy — if the node is replicated, the route # will include it. Alternatively, check /v1/raft/status log_length. status = _raft_status(url) assert status["log_length"] >= 1, ( f"{url} has log_length={status['log_length']}, expected ≥1 after replication" ) def test_follower_leader_status(three_tracker_cluster): "All nodes agree on who the leader is after election.\n\nTags: routing, tracker" _, urls = zip(*three_tracker_cluster) urls = list(urls) _wait_for_leader(urls, timeout=1.0) time.sleep(0.2) # let heartbeats propagate leader_url to followers statuses = [_raft_status(u) for u in urls] leaders_reported = {s["leader"] for s in statuses if s.get("leader")} # All nodes that have a leader opinion should agree on the same one assert len(leaders_reported) <= 1 or len(leaders_reported) == len( {s["leader"] for s in statuses if s["role"] == "leader"} ), f"Nodes disagree on leader: {leaders_reported}" def test_new_leader_elected_after_kill(three_tracker_cluster): "Killing the leader triggers a new election within 5 seconds.\n\nTags: routing, tracker" trackers, urls = zip(*three_tracker_cluster) trackers = list(trackers) urls = list(urls) # Find the leader object leader_url, followers = _wait_for_leader(urls, timeout=1.0) leader_idx = urls.index(leader_url) leader_tracker = trackers[leader_idx] # Register a node before killing the leader (proves it works) _register_node(leader_url, port_hint=19998) # Kill the leader leader_tracker.stop() remaining_urls = [u for u in urls if u != leader_url] # A new leader must be elected among the remaining 2 nodes new_leader_url, _ = _wait_for_leader(remaining_urls, timeout=5.0) assert new_leader_url in remaining_urls, f"New leader {new_leader_url!r} not in remaining {remaining_urls}" # Registration must still work with the new leader node_id = _register_node(new_leader_url, port_hint=19997) assert node_id, "Registration after leader re-election returned no node_id" def test_registration_on_leader_visible_to_all(three_tracker_cluster): "Registering with the leader replicates to all followers synchronously.\n\nTags: routing, tracker" _, urls = zip(*three_tracker_cluster) urls = list(urls) leader_url, followers = _wait_for_leader(urls, timeout=1.0) _register_node(leader_url, port_hint=19996) # Allow Raft heartbeat to replicate the entry time.sleep(0.3) for url in followers: status = _raft_status(url) assert status["log_length"] >= 1, ( f"Follower {url} log_length={status['log_length']}, expected ≥1" ) def test_gossip_propagates_heartbeat(three_tracker_cluster): "Heartbeat recorded on one tracker propagates to others via gossip.\n\nTags: gossip, network, routing, tracker" trackers, urls = zip(*three_tracker_cluster) trackers = list(trackers) urls = list(urls) _wait_for_leader(urls, timeout=1.0) leader_url, _ = _wait_for_leader(urls, timeout=1.0) # Register a node (so heartbeat makes sense) node_id = _register_node(leader_url, port_hint=19995) # Send a heartbeat directly to the leader _post(f"{leader_url}/v1/nodes/{node_id}/heartbeat", {}) # Allow gossip to propagate to other nodes (gossip interval is 3s in prod, # but we just want to verify the gossip table was updated locally) leader_idx = urls.index(leader_url) leader_gossip = trackers[leader_idx]._gossip assert leader_gossip is not None, "Leader should have gossip enabled" assert leader_gossip.last_seen(node_id) is not None, ( "Gossip table on leader should record the heartbeat" )