diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index ae922e5..63f5d47 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -244,6 +244,81 @@ def _coverage_map( return coverage +def _node_health(node: "_NodeEntry", heartbeat_timeout: float) -> dict: + """Per-node health detail for the availability map.""" + age = time.monotonic() - node.last_heartbeat + alive = age <= heartbeat_timeout + hb_expected = max(1, round(node.uptime_seconds / 20.0)) # assume ~20s interval + hb_rate = round(min(1.0, node.heartbeats_received / hb_expected), 4) if node.heartbeats_received else 0.0 + total = node.total_requests + inf_rate = round((total - node.failed_requests) / total, 4) if total > 0 else 1.0 + return { + "node_id": node.node_id, + "endpoint": node.endpoint, + "alive": alive, + "last_seen_seconds_ago": round(age, 1), + "status": node.status, + "queue_depth": node.queue_depth, + "total_requests": node.total_requests, + "heartbeat_success_rate": hb_rate, + "inference_success_rate": inf_rate, + } + + +def _coverage_map_detailed( + nodes: list["_NodeEntry"], + required_start: int, + required_end: int, + heartbeat_timeout: float, +) -> list[dict]: + """Like _coverage_map but with per-node identity and health in each band. + + Includes all nodes (alive and stale) so operators can see both coverage + holes and which dead nodes used to fill them. + """ + now = time.monotonic() + + def covers(node: "_NodeEntry", layer: int) -> bool: + return ( + node.shard_start is not None + and node.shard_end is not None + and node.shard_start <= layer <= node.shard_end + ) + + # Build per-layer list of nodes (with alive flag) + layer_nodes: list[list[tuple["_NodeEntry", bool]]] = [] + for layer in range(required_start, required_end + 1): + ns = [ + (n, (now - n.last_heartbeat) <= heartbeat_timeout) + for n in nodes + if covers(n, layer) + ] + layer_nodes.append(ns) + + # Merge consecutive layers with identical node-set (same node_ids, same alive) + coverage: list[dict] = [] + for i, layer_idx in enumerate(range(required_start, required_end + 1)): + ns = layer_nodes[i] + node_ids = [n.node_id for n, _ in ns] + alive_flags = [a for _, a in ns] + # Compare with last band + if ( + coverage + and coverage[-1]["end_layer"] == layer_idx - 1 + and [nd["node_id"] for nd in coverage[-1]["nodes"]] == node_ids + and [nd["alive"] for nd in coverage[-1]["nodes"]] == alive_flags + ): + coverage[-1]["end_layer"] = layer_idx + else: + coverage.append({ + "start_layer": layer_idx, + "end_layer": layer_idx, + "node_count": len(ns), + "nodes": [_node_health(n, heartbeat_timeout) for n, _ in ns], + }) + return coverage + + def _coverage_gaps(coverage: list[dict]) -> list[tuple[int, int]]: return [ (segment["start_layer"], segment["end_layer"]) @@ -624,19 +699,20 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): def _handle_coverage(self, model: str): server: _TrackerHTTPServer = self.server # type: ignore[assignment] + # Do NOT purge before coverage — dead nodes are included with alive=false + # so operators can see what was covering each layer band before failure. with server.lock: - self._purge_expired_nodes() resolved = _nodes_and_bounds_for_model(server, model) if resolved is None: self._send_json(404, {"error": f"no nodes registered for model {model!r}"}) return - alive, required_start, required_end = resolved + all_nodes, required_start, required_end = resolved if server.contracts is not None: - alive = [ - node for node in alive + all_nodes = [ + node for node in all_nodes if not node.wallet_address or not server.contracts.registry.get_wallet(node.wallet_address).banned ] - coverage = _coverage_map(alive, required_start, required_end) + coverage = _coverage_map_detailed(all_nodes, required_start, required_end, server.heartbeat_timeout) self._send_json(200, {"model": model, "coverage": coverage}) def _handle_tracker_nodes(self, model: str): diff --git a/tests/test_tracker_routing.py b/tests/test_tracker_routing.py index 4acf18f..ea71639 100644 --- a/tests/test_tracker_routing.py +++ b/tests/test_tracker_routing.py @@ -243,10 +243,24 @@ def test_tracker_coverage_endpoint_reports_uncovered_ranges(): ) coverage_resp = _get_json(f"http://127.0.0.1:{tracker_port}/v1/coverage/tiny-model") - assert coverage_resp["coverage"] == [ - {"start_layer": 0, "end_layer": 2, "node_count": 1}, - {"start_layer": 3, "end_layer": 5, "node_count": 0}, - ] + bands = coverage_resp["coverage"] + assert len(bands) == 2 + # Covered band: layers 0-2, one node + assert bands[0]["start_layer"] == 0 + assert bands[0]["end_layer"] == 2 + assert bands[0]["node_count"] == 1 + assert len(bands[0]["nodes"]) == 1 + node_info = bands[0]["nodes"][0] + assert node_info["endpoint"] == "http://127.0.0.1:9001" + assert node_info["alive"] is True + assert "last_seen_seconds_ago" in node_info + assert "heartbeat_success_rate" in node_info + assert "inference_success_rate" in node_info + # Gap band: layers 3-5, no nodes + assert bands[1]["start_layer"] == 3 + assert bands[1]["end_layer"] == 5 + assert bands[1]["node_count"] == 0 + assert bands[1]["nodes"] == [] try: _get_json(f"http://127.0.0.1:{tracker_port}/v1/route?model=tiny-model") @@ -283,11 +297,12 @@ def test_tracker_coverage_endpoint_accepts_registered_hf_repo_or_short_name(): ) assert by_repo["model"] == "Qwen/Qwen2.5-0.5B-Instruct" - assert by_repo["coverage"] == [ - {"start_layer": 0, "end_layer": 19, "node_count": 1}, - {"start_layer": 20, "end_layer": 21, "node_count": 2}, - {"start_layer": 22, "end_layer": 23, "node_count": 1}, - ] + bands = by_repo["coverage"] + assert len(bands) == 3 + assert bands[0]["start_layer"] == 0 and bands[0]["end_layer"] == 19 and bands[0]["node_count"] == 1 + assert bands[1]["start_layer"] == 20 and bands[1]["end_layer"] == 21 and bands[1]["node_count"] == 2 + assert bands[2]["start_layer"] == 22 and bands[2]["end_layer"] == 23 and bands[2]["node_count"] == 1 + assert all("nodes" in b for b in bands) assert by_short_name["coverage"] == by_repo["coverage"] finally: tracker.stop()