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

Merge branch 'master' of https://git.d-popov.com/popov/neuron-tai

Browse Source
This commit is contained in:
Dobromir Popov
2026-07-07 15:27:34 +03:00
parent 5e89bba78f 339577a26c
commit ca49675f50
9 changed files with 732 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
packages/node/meshnet_node/downloader.py
View File

@@ -1,6 +1,6 @@
"""Shard downloader — fetches model shards from peers or HuggingFace Hub.
Cache layout: ~/.cache/meshnet/shards/<model>/layers_<start>-<end>/
Cache layout: ~/.cache/meshnet/shards/<model>/
For "stub-model" (no HF repo), a placeholder JSON file is written so the
test suite never touches the network.
@@ -106,9 +106,7 @@ def _download_shard_from_peer(
_safe_extract_shard(archive_path, extract_dir)
if compute_shard_checksum(extract_dir) != checksum:
return False
if shard_dir.exists():
shutil.rmtree(shard_dir)
shutil.move(str(extract_dir), str(shard_dir))
_merge_tree(extract_dir, shard_dir)
return True
except Exception:
return False
@@ -217,6 +215,38 @@ def _reuse_local_file(expected: int, dest: Path, final: Path) -> bool:
return False
def _valid_source_rel_files(source: dict) -> list[str]:
return [
rel for rel in (source.get("files") or [])
if isinstance(rel, str) and rel and not rel.startswith("/") and ".." not in Path(rel).parts
]
def _source_files_cached(source: dict, shard_dir: Path) -> bool:
rel_files = _valid_source_rel_files(source)
if not rel_files:
return False
sizes = source.get("file_sizes")
if not isinstance(sizes, dict):
return False
for rel in rel_files:
expected = sizes.get(rel)
path = shard_dir / rel
if not isinstance(expected, int) or not path.exists() or path.stat().st_size != expected:
return False
return True
def _merge_tree(src: Path, dest: Path) -> None:
dest.mkdir(parents=True, exist_ok=True)
for path in sorted(p for p in src.rglob("*") if p.is_file()):
target = dest / path.relative_to(src)
target.parent.mkdir(parents=True, exist_ok=True)
if target.exists():
target.unlink()
shutil.move(str(path), str(target))
def _download_source_files(
source: dict,
shard_dir: Path,
@@ -231,10 +261,7 @@ def _download_source_files(
the download is retried or the node is restarted.
"""
url = source.get("url")
rel_files = [
rel for rel in (source.get("files") or [])
if isinstance(rel, str) and rel and not rel.startswith("/") and ".." not in Path(rel).parts
]
rel_files = _valid_source_rel_files(source)
if not isinstance(url, str) or not url or not rel_files:
return None
sizes = source.get("file_sizes")
@@ -283,9 +310,11 @@ def _download_source_files(
time.sleep(1.0 * attempt)
finally:
tracker_bar.close()
if shard_dir.exists():
shutil.rmtree(shard_dir)
shutil.move(str(partial_dir), str(shard_dir))
_merge_tree(partial_dir, shard_dir)
try:
partial_dir.rmdir()
except OSError:
pass
return shard_dir
@@ -331,9 +360,7 @@ def _download_model_source(
if progress:
print(f" {label}: transfer complete ({received / 1e9:.2f} GB), extracting ...", flush=True)
_safe_extract_shard(archive_path, extract_dir)
if shard_dir.exists():
shutil.rmtree(shard_dir)
shutil.move(str(extract_dir), str(shard_dir))
_merge_tree(extract_dir, shard_dir)
return shard_dir
except Exception as exc:
if progress:
@@ -431,10 +458,17 @@ def download_shard(
the test suite hermetic while the real download path is exercised by
passing a non-stub *hf_repo*.
"""
shard_dir = cache_dir / model / f"layers_{shard_start}-{shard_end}"
shard_dir = cache_dir / model
if progress:
print(f" Target location: {shard_dir}", flush=True)
for source in model_sources or []:
label = str(source.get("type") or "model-source")
if _source_files_cached(source, shard_dir):
if progress:
print(f" [{label}] requested files already cached at {shard_dir}", flush=True)
return shard_dir
for peer in peers or []:
if progress:
print(f" Trying peer shard download from {peer.get('endpoint')} ...", flush=True)

17
packages/node/meshnet_node/model_catalog.py
View File

@@ -64,6 +64,17 @@ def _load_model_metadata() -> dict[str, dict]:
_MODEL_METADATA = _load_model_metadata()
def _local_model_path(hf_repo: str, cache_dir: Path | None) -> Path | None:
if cache_dir is None:
return None
if (cache_dir / "config.json").exists():
return cache_dir
candidate = cache_dir / hf_repo.split("/")[-1]
if (candidate / "config.json").exists():
return candidate
return None
CURATED_MODELS: list[ModelPreset] = [
ModelPreset(
name="Qwen2.5-0.5B-Instruct",
@@ -215,9 +226,11 @@ def model_metadata_for(
try:
from transformers import AutoConfig # type: ignore[import]
local_model = _local_model_path(hf_repo, cache_dir)
load_source = str(local_model) if local_model is not None else hf_repo
cfg = AutoConfig.from_pretrained(
hf_repo,
cache_dir=str(cache_dir) if cache_dir is not None else None,
load_source,
cache_dir=str(cache_dir) if cache_dir is not None and local_model is None else None,
)
# Composite configs (VLM/MoE) nest decoder fields in text_config.
text_cfg = getattr(cfg, "text_config", None) or cfg

151
packages/node/meshnet_node/startup.py
View File

@@ -26,6 +26,86 @@ from .wallet import load_or_create_wallet
_DEFAULT_BYTES_PER_LAYER = 30 * 1024 * 1024
def _downloaded_model_inventory(
model: str,
shard_start: int,
shard_end: int,
shard_path: Path,
hf_repo: str | None = None,
model_sources: list[dict] | None = None,
) -> list[dict]:
"""Return a cheap local inventory record without reading model file contents."""
file_count = 0
total_bytes = 0
existing_rel_files: set[str] = set()
if shard_path.exists():
for path in shard_path.rglob("*"):
if not path.is_file():
continue
file_count += 1
try:
existing_rel_files.add(path.relative_to(shard_path).as_posix())
except ValueError:
pass
try:
total_bytes += path.stat().st_size
except OSError:
pass
record = {
"model": model,
"shard_start": shard_start,
"shard_end": shard_end,
"path": str(shard_path),
"file_count": file_count,
"total_bytes": total_bytes,
}
if hf_repo is not None:
record["hf_repo"] = hf_repo
expected_files: set[str] = set()
file_sizes: dict[str, int] = {}
for source in model_sources or []:
for rel in source.get("full_files") or source.get("files") or []:
if isinstance(rel, str) and rel and not rel.startswith("/") and ".." not in Path(rel).parts:
expected_files.add(rel)
sizes = source.get("file_sizes")
if isinstance(sizes, dict):
for rel, size in sizes.items():
if isinstance(rel, str) and isinstance(size, int):
file_sizes[rel] = size
if expected_files:
expected_bytes = sum(file_sizes.get(rel, 0) for rel in expected_files)
local_expected_files = existing_rel_files & expected_files
local_expected_bytes = sum(file_sizes.get(rel, 0) for rel in local_expected_files)
record["expected_file_count"] = len(expected_files)
record["local_expected_file_count"] = len(local_expected_files)
record["expected_bytes"] = expected_bytes
record["local_expected_bytes"] = local_expected_bytes
record["local_model_percentage"] = (
round((local_expected_bytes / expected_bytes) * 100, 4)
if expected_bytes > 0
else 0.0
)
return [record]
def _registration_shard_checksum(model: str, shard_path: Path) -> str | None:
"""Only checksum tiny stub shards; real model folders are too large to hash at startup."""
if model != "stub-model":
return None
return compute_shard_checksum(shard_path)
def _model_cache_path(model_id: str, cache_dir: Path | None) -> Path | None:
if cache_dir is None:
return None
if (cache_dir / "config.json").exists():
return cache_dir
candidate = cache_dir / model_id.split("/")[-1]
if candidate.exists():
return candidate
return None
def _memory_budget(device: str, vram_mb: int, ram_mb: int, shared_vram_mb: int = 0) -> tuple[int, str]:
"""Return the capacity budget in MB and whether it came from VRAM or RAM."""
if device == "cuda" and vram_mb > 0:
@@ -502,6 +582,7 @@ def run_startup(
if model_id: # treat "" the same as None — no explicit model given
user_pinned_shard = shard_start is not None or shard_end is not None
full_sources: list[dict] = []
# Auto-detect shard range from model config if not explicitly provided
if shard_start is None or shard_end is None:
try:
@@ -581,6 +662,7 @@ def run_startup(
_attach_relay_bridge(node, relay_bridge)
# Register with tracker so other nodes can auto-join this model.
total_layers = getattr(getattr(node, "backend", None), "total_layers", None)
model_cache_path = _model_cache_path(model_id, cache_dir)
reg_payload = {
"endpoint": endpoint,
"model": model_id.split("/")[-1],
@@ -595,6 +677,18 @@ def run_startup(
"tracker_mode": (shard_start == 0),
"managed_assignment": not user_pinned_shard,
"model_metadata": model_metadata_for(model_id, total_layers, cache_dir=cache_dir),
"downloaded_models": (
_downloaded_model_inventory(
model_id.split("/")[-1],
shard_start,
shard_end,
model_cache_path,
hf_repo=model_id,
model_sources=full_sources,
)
if model_cache_path is not None
else []
),
**registration_capabilities,
**relay_fields,
}
@@ -693,6 +787,7 @@ def run_startup(
relay_url=relay_url,
)
_attach_relay_bridge(node, relay_bridge)
model_cache_path = _model_cache_path(assigned_hf_repo, cache_dir)
auto_reg_payload = {
"endpoint": endpoint,
"model": assigned_hf_repo.split("/")[-1],
@@ -707,6 +802,18 @@ def run_startup(
"tracker_mode": (assigned_shard_start == 0),
"managed_assignment": True,
"model_metadata": model_metadata_for(assigned_hf_repo, assigned_num_layers, cache_dir=cache_dir),
"downloaded_models": (
_downloaded_model_inventory(
assigned_hf_repo.split("/")[-1],
assigned_shard_start,
assigned_shard_end,
model_cache_path,
hf_repo=assigned_hf_repo,
model_sources=full_sources,
)
if model_cache_path is not None
else []
),
**registration_capabilities,
**relay_fields,
}
@@ -776,7 +883,15 @@ def run_startup(
if model_sources:
dl_kwargs["model_sources"] = model_sources
shard_path = download_shard(assigned_model, shard_start, shard_end, **dl_kwargs)
shard_checksum = compute_shard_checksum(shard_path)
shard_checksum = _registration_shard_checksum(assigned_model, shard_path)
downloaded_models = _downloaded_model_inventory(
assigned_model,
shard_start,
shard_end,
shard_path,
hf_repo=hf_repo,
model_sources=model_sources,
)
print(f" Cached at: {shard_path}", flush=True)
# 5. Start HTTP server
@@ -790,6 +905,7 @@ def run_startup(
model=assigned_model,
shard_path=shard_path,
)
_node_start_time = time.monotonic()
actual_port = node.start()
public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host)
endpoint = f"http://{public_host}:{actual_port}"
@@ -805,24 +921,27 @@ def run_startup(
# 6. Register with tracker
print("Registering with tracker...", flush=True)
reg_payload = {
"endpoint": endpoint,
"model": assigned_model,
"shard_start": shard_start,
"shard_end": shard_end,
"shard_checksum": shard_checksum,
"downloaded_models": downloaded_models,
"hardware_profile": hw,
"wallet_address": address,
"score": 1.0,
**registration_capabilities,
**relay_fields,
}
try:
reg_resp = _post_json(
f"{tracker_url}/v1/nodes/register",
{
"endpoint": endpoint,
"model": assigned_model,
"shard_start": shard_start,
"shard_end": shard_end,
"shard_checksum": shard_checksum,
"hardware_profile": hw,
"wallet_address": address,
"score": 1.0,
**registration_capabilities,
**relay_fields,
},
reg_payload,
)
node_id = str(reg_resp["node_id"])
setattr(node, "tracker_node_id", node_id)
_start_heartbeat(tracker_url, node_id, reg_payload, node_ref=node, start_time=_node_start_time)
except Exception:
node.stop()
raise
@@ -855,9 +974,11 @@ def _detect_num_layers(model_id: str, cache_dir: Path | None = None) -> int | No
from .model_catalog import layers_from_config
local_model = _model_cache_path(model_id, cache_dir)
load_source = str(local_model) if local_model is not None else model_id
cfg = AutoConfig.from_pretrained(
model_id,
cache_dir=str(cache_dir) if cache_dir is not None else None,
load_source,
cache_dir=str(cache_dir) if cache_dir is not None and local_model is None else None,
)
layers = layers_from_config(cfg)
if layers is None:

69
packages/tracker/meshnet_tracker/dashboard.html
View File

@@ -41,10 +41,21 @@
.error-msg { color:var(--bad); font-size:12px; min-height:16px; }
.keybox { word-break:break-all; background:var(--bg); border:1px solid var(--border);
border-radius:6px; padding:4px 8px; margin:4px 0; font-size:11px; }
.tabs { display:flex; gap:10px; margin-bottom:8px; }
.tabs a { color:var(--dim); cursor:pointer; }
.tabs a.active { color:var(--accent); border-bottom:1px solid var(--accent); }
</style>
.tabs { display:flex; gap:10px; margin-bottom:8px; }
.tabs a { color:var(--dim); cursor:pointer; }
.tabs a.active { color:var(--accent); border-bottom:1px solid var(--accent); }
.wide { grid-column:1 / -1; }
.console {
background:var(--bg); border:1px solid var(--border); border-radius:6px;
min-height:160px; max-height:280px; overflow:auto; padding:7px 9px;
white-space:pre-wrap; word-break:break-word; font-size:11px;
}
.console-line { padding:1px 0; border-bottom:1px solid #161b22; }
.console-time { color:var(--dim); }
.console-level-info { color:var(--accent); }
.console-level-warn { color:var(--warn); }
.console-level-error { color:var(--bad); }
</style>
</head>
<body>
<header>
@@ -61,9 +72,10 @@
<section><h2>Node pending payouts</h2><div id="pending" class="empty">loading…</div></section>
<section><h2>Settlement history</h2><div id="settlements" class="empty">loading…</div></section>
<section><h2>Strikes / bans / forfeitures</h2><div id="fraud" class="empty">loading…</div></section>
<section><h2>Model usage (RPM)</h2><div id="stats" class="empty">loading…</div></section>
<section><h2>Node throughput</h2><div id="throughput" class="empty">loading…</div></section>
</main>
<section><h2>Model usage (RPM)</h2><div id="stats" class="empty">loading…</div></section>
<section><h2>Node throughput</h2><div id="throughput" class="empty">loading…</div></section>
<section class="wide"><h2>Console output</h2><div id="console" class="console empty">loading…</div></section>
</main>
<script>
"use strict";
const $ = id => document.getElementById(id);
@@ -196,7 +208,7 @@ function renderStats(stats) {
$("stats").innerHTML = table(["model", "rpm (1h)", "rpm (24h)", "rpm (30d)"], rows);
}
function renderThroughput(stats) {
function renderThroughput(stats) {
const nodes = (stats && stats.nodes) || {};
const rows = [];
for (const [nodeId, nodeStats] of Object.entries(nodes)) {
@@ -209,8 +221,23 @@ function renderThroughput(stats) {
]);
}
}
$("throughput").innerHTML = table(["node", "model", "tps (1h)", "samples"], rows);
}
$("throughput").innerHTML = table(["node", "model", "tps (1h)", "samples"], rows);
}
function renderConsole(data) {
const events = (data && data.events) || [];
if (!events.length) {
$("console").innerHTML = '<div class="empty">no console events</div>';
return;
}
$("console").innerHTML = events.slice(-120).map(e => {
const level = String(e.level || "info");
const cls = level === "error" ? "console-level-error" : level === "warn" ? "console-level-warn" : "console-level-info";
const fields = e.fields && Object.keys(e.fields).length ? " " + JSON.stringify(e.fields) : "";
return `<div class="console-line"><span class="console-time">${new Date((e.ts || 0) * 1000).toLocaleTimeString()}</span> ` +
`<span class="${cls}">${esc(level.toUpperCase())}</span> ${esc(e.message || "")}${esc(fields)}</div>`;
}).join("");
}
// ---- account panel (registration / login / balance / usage / API keys) ----
@@ -366,21 +393,23 @@ async function renderAdminPanel() {
async function refresh() {
$("self-url").textContent = location.host;
const [raft, map, summary, settlements, wallets, stats] = await Promise.all([
fetchJson("/v1/raft/status"),
fetchJson("/v1/network/map"),
fetchJson("/v1/billing/summary"),
fetchJson("/v1/billing/settlements"),
fetchJson("/v1/registry/wallets"),
fetchJson("/v1/stats"),
]);
const [raft, map, summary, settlements, wallets, stats, consoleData] = await Promise.all([
fetchJson("/v1/raft/status"),
fetchJson("/v1/network/map"),
fetchJson("/v1/billing/summary"),
fetchJson("/v1/billing/settlements"),
fetchJson("/v1/registry/wallets"),
fetchJson("/v1/stats"),
fetchJson("/v1/console"),
]);
renderHive(raft);
renderNodes(map);
renderBilling(summary);
renderSettlements(settlements);
renderFraud(wallets, summary);
renderStats(stats);
renderThroughput(stats);
renderStats(stats);
renderThroughput(stats);
renderConsole(consoleData);
$("refreshed").textContent = "refreshed " + new Date().toLocaleTimeString();
}
refresh();

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