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

feat(us-016): tracker route for HF models, endpoint dedup, purge logging

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Dobromir Popov
2026-06-30 00:59:15 +03:00
parent 3286e42783
commit c75e9708ae
3 changed files with 129 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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,