From c75e9708aef5667a1e4f5409f1d8d8be935cfab3 Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Tue, 30 Jun 2026 00:59:15 +0300 Subject: [PATCH] feat(us-016): tracker route for HF models, endpoint dedup, purge logging Tracker /v1/route now resolves HF model nodes (by hf_repo or short name) in addition to preset models, using the same greedy interval-cover logic. This allows distributed inference routing across two nodes each holding half the model. Endpoint dedup: re-registering the same endpoint atomically replaces the old entry so stale registrations don't accumulate across node restarts. Purge logging: tracker now prints when a node expires due to missed heartbeats so operators can see dead nodes being removed. Timing fix: heartbeat timeout raised from 30s to 90s (3 missed beats); node heartbeat interval lowered from 30s to 20s to maintain margin. Co-Authored-By: Claude Sonnet 4.6 --- packages/node/meshnet_node/startup.py | 2 +- packages/tracker/meshnet_tracker/server.py | 46 +++++++++-- tests/test_node_startup.py | 90 ++++++++++++++++++++++ 3 files changed, 129 insertions(+), 9 deletions(-) diff --git a/packages/node/meshnet_node/startup.py b/packages/node/meshnet_node/startup.py index b67cefc..9f3bed8 100644 --- a/packages/node/meshnet_node/startup.py +++ b/packages/node/meshnet_node/startup.py @@ -34,7 +34,7 @@ def _get_json(url: str, timeout: float = 10.0) -> dict: return json.loads(r.read()) -def _start_heartbeat(tracker_url: str, node_id: str, interval: float = 30.0) -> threading.Thread: +def _start_heartbeat(tracker_url: str, node_id: str, interval: float = 20.0) -> threading.Thread: """Daemon thread that sends periodic heartbeats to the tracker.""" def _loop() -> None: hb_url = f"{tracker_url}/v1/nodes/{node_id}/heartbeat" diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index ee5b902..898083b 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -258,7 +258,12 @@ def _purge_expired_nodes_locked(server: "_TrackerHTTPServer") -> list[str]: if (now - entry.last_heartbeat) > server.heartbeat_timeout ] for node_id in expired_ids: - del server.registry[node_id] + entry = server.registry.pop(node_id) + print( + f"[tracker] node expired: {node_id[:8]} {entry.endpoint} " + f"(no heartbeat for >{server.heartbeat_timeout:.0f}s)", + flush=True, + ) if expired_ids: _rebalance_all_locked(server) return expired_ids @@ -649,6 +654,18 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): ) with server.lock: self._purge_expired_nodes() + # Dedup: if this endpoint is already registered, remove the old entry first. + stale_ids = [ + eid for eid, e in server.registry.items() + if e.endpoint == entry.endpoint.rstrip("/") + ] + for eid in stale_ids: + old = server.registry.pop(eid) + print( + f"[tracker] node re-registered: replaced {eid[:8]} with {node_id[:8]}" + f" {old.endpoint}", + flush=True, + ) server.registry[node_id] = entry if entry.managed_assignment: _rebalance_model_locked(server, model) @@ -897,15 +914,28 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): model = model_list[0] preset = server.model_presets.get(model) - if preset is None: - self._send_json(404, {"error": f"unknown model preset: {model!r}"}) - return - - required_start, required_end = _preset_layer_bounds(preset) with server.lock: self._purge_expired_nodes() - alive = [node for node in server.registry.values() if node.model == model] + if preset is not None: + # Preset-based routing (stub-model system). + alive = [node for node in server.registry.values() if node.model == model] + required_start, required_end = _preset_layer_bounds(preset) + else: + # HF model routing: match by hf_repo (full) or model short name. + alive = [ + node for node in server.registry.values() + if (node.hf_repo == model or node.model == model) + and node.shard_start is not None + and node.shard_end is not None + and node.num_layers is not None + ] + if not alive: + self._send_json(404, {"error": f"no nodes registered for model {model!r}"}) + return + required_start = 0 + required_end = max(n.num_layers for n in alive) - 1 # type: ignore[type-var] + if server.contracts is not None: alive = [ node for node in alive @@ -1007,7 +1037,7 @@ class TrackerServer: self, host: str = "127.0.0.1", port: int = 0, - heartbeat_timeout: float = 30.0, + heartbeat_timeout: float = 90.0, rebalance_interval: float = 30.0, model_presets: dict | None = None, contracts: Any | None = None, diff --git a/tests/test_node_startup.py b/tests/test_node_startup.py index 2922c6e..b6f3127 100644 --- a/tests/test_node_startup.py +++ b/tests/test_node_startup.py @@ -574,6 +574,96 @@ def test_network_assign_gap_found_field(): tracker.stop() +def test_route_finds_hf_model_across_two_nodes(): + """Tracker /v1/route returns ordered route for HF model even without a preset.""" + import json as _json + import urllib.request as _ur + + tracker = TrackerServer() + port = tracker.start() + try: + def register(endpoint, shard_start, shard_end): + data = _json.dumps({ + "endpoint": endpoint, + "model": "Qwen2.5-0.5B-Instruct", + "hf_repo": "Qwen/Qwen2.5-0.5B-Instruct", + "num_layers": 24, + "shard_start": shard_start, + "shard_end": shard_end, + "hardware_profile": {}, + "score": 1.0, + }).encode() + req = _ur.Request( + f"http://127.0.0.1:{port}/v1/nodes/register", + data=data, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with _ur.urlopen(req) as r: + r.read() + + register("http://127.0.0.1:9300", 0, 11) + register("http://127.0.0.1:9301", 12, 23) + + # Route by hf_repo (full identifier). + resp = _get_json( + f"http://127.0.0.1:{port}/v1/route?model=Qwen/Qwen2.5-0.5B-Instruct" + ) + assert resp["route"] == ["http://127.0.0.1:9300", "http://127.0.0.1:9301"] + + # Route also works by short model name. + resp2 = _get_json( + f"http://127.0.0.1:{port}/v1/route?model=Qwen2.5-0.5B-Instruct" + ) + assert resp2["route"] == ["http://127.0.0.1:9300", "http://127.0.0.1:9301"] + finally: + tracker.stop() + + +def test_register_deduplicates_same_endpoint(): + """Re-registering the same endpoint replaces the old entry, not duplicates it.""" + import json as _json + import urllib.request as _ur + + tracker = TrackerServer() + port = tracker.start() + try: + def register(shard_start, shard_end): + data = _json.dumps({ + "endpoint": "http://127.0.0.1:9400", + "model": "Qwen2.5-0.5B-Instruct", + "hf_repo": "Qwen/Qwen2.5-0.5B-Instruct", + "num_layers": 24, + "shard_start": shard_start, + "shard_end": shard_end, + "hardware_profile": {}, + "score": 1.0, + }).encode() + req = _ur.Request( + f"http://127.0.0.1:{port}/v1/nodes/register", + data=data, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with _ur.urlopen(req) as r: + return _json.loads(r.read()) + + register(0, 23) # initial full-model registration + register(12, 23) # re-register with corrected shard range + + # After re-register, tracker should see only one node at 12-23 for this endpoint. + # If both were still registered, the gap scan would find no gap (0-23 still covers). + # With dedup, the old 0-23 is gone and a real gap 0-11 exists. + assign_resp = _get_json( + f"http://127.0.0.1:{port}/v1/network/assign?device=cpu&vram_mb=0" + "&hf_repo=Qwen/Qwen2.5-0.5B-Instruct" + ) + assert assign_resp["gap_found"] is True + assert assign_resp["shard_start"] == 0, "old 0-23 entry should have been replaced" + finally: + tracker.stop() + + def test_startup_cpu_fallback(tmp_path, monkeypatch): """Node starts with CPU warning when no GPU is detected.""" import meshnet_node.startup as startup_mod