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

fix model registration and anouncement. added console panel

Browse Source
This commit is contained in:
Dobromir Popov
2026-07-07 14:24:37 +02:00
parent 0e8acf5d59
commit 339577a26c
9 changed files with 732 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -32,9 +32,10 @@ import tarfile
import threading
import time
import urllib.parse
import urllib.request
import uuid
from importlib.resources import files
import urllib.request
import uuid
from collections import deque
from importlib.resources import files
from pathlib import Path
from typing import Any
@@ -46,7 +47,10 @@ from .calibration import DEFAULT_CALIBRATION_DB_PATH, ToplocCalibrationStore
from .hf_pricing import DEFAULT_HF_PRICING_LOG_DB_PATH, HfPricingLog, refresh_preset_price
from .gossip import NodeGossip
from .model_files import files_for_layer_range, snapshot_dir_for_repo
from .raft import RaftNode
from .raft import RaftNode
_CONSOLE_LIMIT = 300
def _preset_price_keys(name: str, preset: dict) -> set[str]:
@@ -529,7 +533,7 @@ class _StatsCollector:
class _NodeEntry:
__slots__ = (
"node_id", "endpoint", "shard_start", "shard_end",
"model", "hf_repo", "num_layers", "model_metadata", "shard_checksum", "hardware_profile", "wallet_address",
"model", "hf_repo", "num_layers", "model_metadata", "shard_checksum", "downloaded_models", "hardware_profile", "wallet_address",
"score", "vram_bytes", "ram_bytes", "quantizations", "max_loaded_shards",
"benchmark_tokens_per_sec", "quantization", "managed_assignment",
"model_tokens_per_sec",
@@ -562,10 +566,11 @@ class _NodeEntry:
quantization: str | None = None,
managed_assignment: bool = False,
tracker_mode: bool = False,
hf_repo: str | None = None,
num_layers: int | None = None,
model_metadata: dict | None = None,
relay_addr: str | None = None,
hf_repo: str | None = None,
num_layers: int | None = None,
model_metadata: dict | None = None,
downloaded_models: list[dict] | None = None,
relay_addr: str | None = None,
cert_fingerprint: str | None = None,
peer_id: str | None = None,
) -> None:
@@ -588,9 +593,10 @@ class _NodeEntry:
self.model_tokens_per_sec: dict[str, float] = {}
self.tracker_mode = tracker_mode
self.hf_repo = hf_repo
self.num_layers = num_layers
self.model_metadata = dict(model_metadata or {})
self.relay_addr = relay_addr
self.num_layers = num_layers
self.model_metadata = dict(model_metadata or {})
self.downloaded_models = [dict(item) for item in (downloaded_models or []) if isinstance(item, dict)]
self.relay_addr = relay_addr
self.cert_fingerprint = cert_fingerprint
self.peer_id = peer_id
self.pending_directives: list[dict] = []
@@ -1245,13 +1251,18 @@ def _purge_expired_nodes_locked(server: "_TrackerHTTPServer") -> list[str]:
node_id for node_id, entry in server.registry.items()
if (now - entry.last_heartbeat) > server.heartbeat_timeout
]
for node_id in expired_ids:
entry = server.registry.pop(node_id)
print(
f"[tracker] node expired: {node_id} {entry.endpoint} "
f"(no heartbeat for >{server.heartbeat_timeout:.0f}s)",
flush=True,
)
for node_id in expired_ids:
entry = server.registry.pop(node_id)
_tracker_log(
server,
"warn",
"node expired",
node_id=node_id,
endpoint=entry.endpoint,
model=entry.model,
shard=f"{entry.shard_start}-{entry.shard_end}",
heartbeat_timeout_seconds=server.heartbeat_timeout,
)
if expired_ids:
_rebalance_all_locked(server)
return expired_ids
@@ -1609,15 +1620,33 @@ def _billable_stream_tokens(observed_tokens: int, reported_tokens: int | None) -
return min(observed, max(0, reported_tokens))
def _registration_ban_error(contracts: Any | None, wallet_address: str | None) -> str | None:
def _registration_ban_error(contracts: Any | None, wallet_address: str | None) -> str | None:
if contracts is None or not wallet_address:
return None
if contracts.registry.get_wallet(wallet_address).banned:
return "wallet is banned"
return None
def _node_id_for_registration(
return None
def _tracker_log(server: "_TrackerHTTPServer", level: str, message: str, **fields: Any) -> None:
event = {
"ts": time.time(),
"level": level,
"message": message,
"fields": {
key: value
for key, value in fields.items()
if value is not None
},
}
with server.console_lock:
server.console_events.append(event)
extras = " ".join(f"{key}={value}" for key, value in event["fields"].items())
suffix = f" {extras}" if extras else ""
print(f"[tracker] {level}: {message}{suffix}", flush=True)
def _node_id_for_registration(
endpoint: str,
model: str,
wallet_address: str | None,
@@ -1698,8 +1727,10 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
)
self.toploc_calibration_gate_min_hardware_profiles = toploc_calibration_gate_min_hardware_profiles
self.toploc_backend = toploc_backend
self.hf_pricing_log: HfPricingLog | None = hf_pricing_log
self.models_dir = models_dir
self.hf_pricing_log: HfPricingLog | None = hf_pricing_log
self.models_dir = models_dir
self.console_events = deque(maxlen=_CONSOLE_LIMIT)
self.console_lock = threading.Lock()
class _TrackerHandler(http.server.BaseHTTPRequestHandler):
@@ -1885,10 +1916,12 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
self._handle_tracker_nodes(model)
elif parsed.path == "/v1/raft/status":
self._handle_raft_status()
elif parsed.path == "/v1/stats":
self._handle_stats()
elif parsed.path == "/v1/billing/summary":
self._handle_billing_summary()
elif parsed.path == "/v1/stats":
self._handle_stats()
elif parsed.path == "/v1/console":
self._handle_console()
elif parsed.path == "/v1/billing/summary":
self._handle_billing_summary()
elif parsed.path == "/v1/billing/settlements":
self._handle_billing_settlements()
elif parsed.path == "/v1/account":
@@ -2116,9 +2149,10 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
"peer_id": node.peer_id,
"model": node.model,
"hf_repo": node.hf_repo,
"num_layers": node.num_layers,
"model_metadata": dict(node.model_metadata),
"shard_start": node.shard_start,
"num_layers": node.num_layers,
"model_metadata": dict(node.model_metadata),
"downloaded_models": [dict(item) for item in node.downloaded_models],
"shard_start": node.shard_start,
"shard_end": node.shard_end,
"tracker_mode": node.tracker_mode,
"last_heartbeat": node.last_heartbeat,
@@ -2263,9 +2297,27 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
if n.shard_start == 0 and _node_matches_model(n, model)
]
if not candidates:
self._send_json(503, {"error": {
"message": f"no nodes available for model {model!r}",
if not candidates:
with server.lock:
registered = [
{
"node_id": n.node_id,
"model": n.model,
"hf_repo": n.hf_repo,
"shard": f"{n.shard_start}-{n.shard_end}",
"tracker_mode": n.tracker_mode,
}
for n in server.registry.values()
]
_tracker_log(
server,
"warn",
"no nodes available for model",
model=model,
registered_nodes=registered,
)
self._send_json(503, {"error": {
"message": f"no nodes available for model {model!r}",
"type": "service_unavailable",
"code": "model_not_available",
}})
@@ -2294,9 +2346,18 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
if pinned_nodes is not None:
route_nodes = pinned_nodes
else:
route_nodes, route_error = _select_route(all_nodes, rs, re, model=route_model, contracts=server.contracts)
if route_error:
self._send_json(503, {"error": {
route_nodes, route_error = _select_route(all_nodes, rs, re, model=route_model, contracts=server.contracts)
if route_error:
_tracker_log(
server,
"warn",
"route unavailable",
model=model,
route_model=route_model,
error=route_error,
candidate_count=len(all_nodes),
)
self._send_json(503, {"error": {
"message": route_error,
"type": "service_unavailable",
"code": "route_not_available",
@@ -2322,12 +2383,17 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
f"{n.node_id}@{n.endpoint}[{n.shard_start}-{n.shard_end}]"
for n in route_nodes
)
print(
f"[tracker] proxy route {request_id}: model={model!r} "
f"head={node.node_id}@{node.endpoint} downstream={downstream_urls} "
f"route={route_debug or '<empty>'}",
flush=True,
)
_tracker_log(
server,
"info",
"proxy route selected",
request_id=request_id,
model=model,
head_node_id=node.node_id,
head_endpoint=node.endpoint,
downstream=downstream_urls,
route=route_debug or "<empty>",
)
req = urllib.request.Request(
target_url,
@@ -2350,11 +2416,14 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
}
if node.relay_addr:
print(
f"[tracker] proxy via relay {request_id}: {node.relay_addr} "
f"(direct endpoint {target_url})",
flush=True,
)
_tracker_log(
server,
"info",
"proxy via relay",
request_id=request_id,
relay_addr=node.relay_addr,
direct_endpoint=target_url,
)
started = time.monotonic()
frames = _relay_http_request_frames(
node.relay_addr,
@@ -2388,16 +2457,19 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
input_tokens=in_tokens, output_tokens=out_tokens,
)
return
print(
f"[tracker] relay proxy failed {request_id}: {node.relay_addr}; "
f"trying direct {target_url}",
flush=True,
)
_tracker_log(
server,
"warn",
"relay proxy failed, trying direct",
request_id=request_id,
relay_addr=node.relay_addr,
direct_endpoint=target_url,
)
try:
started = time.monotonic()
upstream = urllib.request.urlopen(req, timeout=300.0)
print(f"[tracker] proxy connected {request_id}: {target_url}", flush=True)
started = time.monotonic()
upstream = urllib.request.urlopen(req, timeout=300.0)
_tracker_log(server, "info", "proxy connected", request_id=request_id, target_url=target_url)
except urllib.error.HTTPError as exc:
# Relay error status + body from node
err_body = exc.read()
@@ -2411,14 +2483,25 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
pass
return
except Exception as exc:
if node.relay_addr:
print(
f"[tracker] direct proxy failed {request_id}: {target_url}: {exc}; "
f"relay already attempted for {node.relay_addr}",
flush=True,
)
else:
print(f"[tracker] proxy failed {request_id}: {target_url}: {exc}", flush=True)
if node.relay_addr:
_tracker_log(
server,
"error",
"direct proxy failed after relay",
request_id=request_id,
target_url=target_url,
relay_addr=node.relay_addr,
error=repr(exc),
)
else:
_tracker_log(
server,
"error",
"proxy failed",
request_id=request_id,
target_url=target_url,
error=repr(exc),
)
self._send_json(503, {"error": {
"message": f"upstream node unreachable: {exc}",
"type": "service_unavailable",
@@ -2464,13 +2547,17 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
input_tokens=in_tokens, output_tokens=out_tokens,
)
else:
# Non-streaming: buffer and relay
resp_body = upstream.read()
elapsed = time.monotonic() - started
print(
f"[tracker] proxy complete {request_id}: {target_url} bytes={len(resp_body)}",
flush=True,
)
# Non-streaming: buffer and relay
resp_body = upstream.read()
elapsed = time.monotonic() - started
_tracker_log(
server,
"info",
"proxy complete",
request_id=request_id,
target_url=target_url,
bytes=len(resp_body),
)
observed_output = ""
try:
response_payload = json.loads(resp_body)
@@ -2809,10 +2896,19 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
model_metadata = body.get("model_metadata", {})
if model_metadata is None:
model_metadata = {}
if not isinstance(model_metadata, dict):
self._send_json(400, {"error": "model_metadata must be an object"})
return
relay_addr = body.get("relay_addr") or None
if not isinstance(model_metadata, dict):
self._send_json(400, {"error": "model_metadata must be an object"})
return
downloaded_models = body.get("downloaded_models", [])
if downloaded_models is None:
downloaded_models = []
if not (
isinstance(downloaded_models, list)
and all(isinstance(item, dict) for item in downloaded_models)
):
self._send_json(400, {"error": "downloaded_models must be an array of objects"})
return
relay_addr = body.get("relay_addr") or None
cert_fingerprint = body.get("cert_fingerprint") or None
peer_id = body.get("peer_id") or None
@@ -2842,10 +2938,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
quantization=quantization,
managed_assignment=managed_assignment or not explicit_shard,
tracker_mode=tracker_mode,
hf_repo=hf_repo,
num_layers=num_layers,
model_metadata=model_metadata,
relay_addr=relay_addr,
hf_repo=hf_repo,
num_layers=num_layers,
model_metadata=model_metadata,
downloaded_models=downloaded_models,
relay_addr=relay_addr,
cert_fingerprint=cert_fingerprint,
peer_id=peer_id,
)
@@ -2856,22 +2953,37 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
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} with {node_id}"
f" {old.endpoint}",
flush=True,
)
server.registry[node_id] = entry
for eid in stale_ids:
old = server.registry.pop(eid)
_tracker_log(
server,
"info",
"node re-registered",
old_node_id=eid,
node_id=node_id,
endpoint=old.endpoint,
)
server.registry[node_id] = entry
if entry.managed_assignment:
if entry.hf_repo:
_rebalance_hf_model_locked(server, entry.hf_repo)
else:
_rebalance_model_locked(server, model)
assignment_directive = entry.pending_directives[-1] if entry.pending_directives else None
if assignment_directive is not None:
entry.pending_directives.clear()
if assignment_directive is not None:
entry.pending_directives.clear()
_tracker_log(
server,
"info",
"node registered",
node_id=node_id,
endpoint=entry.endpoint,
model=entry.model,
hf_repo=entry.hf_repo,
shard=f"{entry.shard_start}-{entry.shard_end}",
tracker_mode=entry.tracker_mode,
)
shard_info = f"layers {shard_start}-{shard_end}" if shard_start is not None else "unsharded"
repo_info = f" [{hf_repo}]" if hf_repo else ""
@@ -2910,11 +3022,12 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
except Exception:
pass
with server.lock:
self._purge_expired_nodes()
entry = server.registry.get(node_id)
if entry is None:
self._send_json(404, {"error": "node not found"})
return
self._purge_expired_nodes()
entry = server.registry.get(node_id)
if entry is None:
_tracker_log(server, "warn", "heartbeat for unknown node", node_id=node_id)
self._send_json(404, {"error": "node not found"})
return
entry.last_heartbeat = time.monotonic()
entry.heartbeats_received += 1
# P2P metadata
@@ -3023,7 +3136,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
return
self._send_json(200, server.billing.snapshot())
def _handle_dashboard(self):
def _handle_dashboard(self):
"""Serve the read-only web dashboard (US-035). Any tracker in the
mesh — leader or follower — serves it from its replicated state."""
try:
@@ -3038,10 +3151,16 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
self.end_headers()
try:
self.wfile.write(body)
except BrokenPipeError:
pass
def _handle_registry_wallets(self):
except BrokenPipeError:
pass
def _handle_console(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
with server.console_lock:
events = [dict(event) for event in server.console_events]
self._send_json(200, {"events": events})
def _handle_registry_wallets(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if not self._require_role("admin"):
return
@@ -4194,8 +4313,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
"model": e.model,
"hf_repo": e.hf_repo,
"num_layers": e.num_layers,
"model_metadata": dict(e.model_metadata),
"shard_checksum": e.shard_checksum,
"model_metadata": dict(e.model_metadata),
"downloaded_models": [dict(item) for item in e.downloaded_models],
"shard_checksum": e.shard_checksum,
"score": e.score,
}
for e, start in route_with_start
@@ -4259,8 +4379,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
"wallet_address": e.wallet_address,
"shard_start": e.shard_start,
"shard_end": e.shard_end,
"model": e.model,
"shard_checksum": e.shard_checksum,
"model": e.model,
"downloaded_models": [dict(item) for item in e.downloaded_models],
"shard_checksum": e.shard_checksum,
"score": e.score,
}
for e in route
@@ -4678,10 +4799,15 @@ class TrackerServer:
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,
model_metadata=payload.get("model_metadata") if isinstance(payload.get("model_metadata"), dict) else None,
)
hf_repo=payload.get("hf_repo"),
num_layers=int(payload["num_layers"]) if payload.get("num_layers") is not None else None,
model_metadata=payload.get("model_metadata") if isinstance(payload.get("model_metadata"), dict) else None,
downloaded_models=(
payload.get("downloaded_models")
if isinstance(payload.get("downloaded_models"), list)
else None
),
)
with self._lock:
self._registry[node_id] = entry