From 640ef78711b0a35930277cf7b0f08a7dd4bb49ee Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Tue, 7 Jul 2026 15:51:27 +0200 Subject: [PATCH] better dash and inference api QOL --- .claude/memory/project-status.md | 1 + QUICKSTART.md | 1 + packages/tracker/meshnet_tracker/server.py | 205 ++++++++++------ tests/test_tracker_routing.py | 266 +++++++++++++++++++++ 4 files changed, 404 insertions(+), 69 deletions(-) diff --git a/.claude/memory/project-status.md b/.claude/memory/project-status.md index 1c40d96..b261e02 100644 --- a/.claude/memory/project-status.md +++ b/.claude/memory/project-status.md @@ -44,3 +44,4 @@ Historical handoff note: `/mnt/c/Users/popov/Downloads/neuron-tai-alpha-handoff- - Follow-up fix: preset-model startup now starts the heartbeat thread after registration; without this, the node appeared briefly on the dashboard and was purged on first inference/route after heartbeat expiry. Tracker dashboard now has a "Console output" panel backed by `/v1/console` for node register/expiry, routing failures, and proxy events. - Qwen3.6-35B-A3B reserve-based split is expected: an 79 GB CPU node may be assigned layers 0-36, and a second node fills 37-39. Do not "fix" this by bypassing the 20% assignment reserve unless the shard-planning policy changes. - Route hardening: tracker chat proxy and `/v1/route` diagnostics now use alias-aware preset node matching for split Qwen3.6 routes; dashboard derives grouped inference history from proxy route/complete console events and shows observed TPS after completion. +- Live proxy hardening: model lookup trims outer whitespace before alias matching (`qwen3.6-35b-a3b ` resolves), and tracker route logs/dashboard queue depth combine heartbeat queue with tracker-local proxy in-flight counts so Postman-style bursts no longer show every selected route as queue `0`. diff --git a/QUICKSTART.md b/QUICKSTART.md index 0c5beb3..ec3821b 100644 --- a/QUICKSTART.md +++ b/QUICKSTART.md @@ -98,6 +98,7 @@ Nodes can then join with either the LAN tracker URL or the public URL: ```bash .venv/bin/meshnet-node start --tracker http://192.168.0.179:8080 --model Qwen/Qwen2.5-0.5B-Instruct .venv/bin/meshnet-node start --tracker https://ai.neuron.d-popov.com --model Qwen/Qwen2.5-0.5B-Instruct +.venv/bin/meshnet-node start --tracker https://ai.d-popov.com --model Qwen/Qwen2.5-0.5B-Instruct ``` ### Windows / WSL2 diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 127cd7b..5d42ba9 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -142,15 +142,18 @@ DEFAULT_CALLER_CREDIT_USDT = 1.0 DEFAULT_DEVNET_TOPUP_USDT = 1.0 -def _model_aliases(model: str | None) -> set[str]: - """Return stable lookup aliases for a model repo or display name.""" - if not model: - return set() - aliases = {model} - short = model.rsplit("/", 1)[-1] - aliases.add(short) - lowered = short.lower() - aliases.add(lowered) +def _model_aliases(model: str | None) -> set[str]: + """Return stable lookup aliases for a model repo or display name.""" + if not model: + return set() + normalized = model.strip() + if not normalized: + return set() + aliases = {normalized} + short = normalized.rsplit("/", 1)[-1] + aliases.add(short) + lowered = short.lower() + aliases.add(lowered) if lowered.endswith("-instruct"): aliases.add(lowered.removesuffix("-instruct")) return aliases @@ -539,8 +542,8 @@ class _NodeEntry: "model_tokens_per_sec", "pending_directives", "last_heartbeat", "tracker_mode", "relay_addr", "cert_fingerprint", "peer_id", - # heartbeat stats (reported by node, cumulative) - "total_requests", "failed_requests", "queue_depth", "uptime_seconds", + # heartbeat stats (reported by node, cumulative) + "total_requests", "failed_requests", "queue_depth", "proxy_inflight", "uptime_seconds", "status", # "ready" | "loading" "heartbeats_expected", "heartbeats_received", # dynamic reassignment queued by the tracker @@ -601,17 +604,18 @@ class _NodeEntry: self.peer_id = peer_id self.pending_directives: list[dict] = [] self.last_heartbeat: float = time.monotonic() - self.total_requests: int = 0 - self.failed_requests: int = 0 - self.queue_depth: int = 0 - self.uptime_seconds: float = 0.0 + self.total_requests: int = 0 + self.failed_requests: int = 0 + self.queue_depth: int = 0 + self.proxy_inflight: int = 0 + self.uptime_seconds: float = 0.0 self.status: str = "ready" self.heartbeats_expected: int = 0 self.heartbeats_received: int = 0 self.pending_new_assignment: dict | None = None -def _effective_throughput(node: "_NodeEntry", model: str | None = None) -> float: +def _effective_throughput(node: "_NodeEntry", model: str | None = None) -> float: """Effective tokens/s accounting for current queue depth.""" observed = None if model: @@ -622,7 +626,22 @@ def _effective_throughput(node: "_NodeEntry", model: str | None = None) -> float if observed is not None: break base = observed if observed is not None and observed > 0 else node.benchmark_tokens_per_sec - return base / (node.queue_depth + 1) + return base / (_effective_queue_depth(node) + 1) + + +def _effective_queue_depth(node: "_NodeEntry") -> int: + """Best current load estimate: heartbeat queue or tracker-routed in-flight.""" + return max(node.queue_depth, node.proxy_inflight) + + +def _record_proxy_inflight( + server: "_TrackerHTTPServer", + nodes: list["_NodeEntry"], + delta: int, +) -> None: + with server.lock: + for node in nodes: + node.proxy_inflight = max(0, node.proxy_inflight + delta) def _reputation_multiplier(node: "_NodeEntry", contracts: Any | None) -> float: @@ -688,10 +707,12 @@ def _node_route_summary(nodes: list["_NodeEntry"]) -> list[dict]: "model": node.model, "hf_repo": node.hf_repo, "endpoint": node.endpoint, - "shard": f"{node.shard_start}-{node.shard_end}", - "num_layers": node.num_layers, - "queue_depth": node.queue_depth, - } + "shard": f"{node.shard_start}-{node.shard_end}", + "num_layers": node.num_layers, + "queue_depth": _effective_queue_depth(node), + "heartbeat_queue_depth": node.queue_depth, + "proxy_inflight": node.proxy_inflight, + } for node in nodes ] @@ -965,11 +986,13 @@ def _node_health(node: "_NodeEntry", heartbeat_timeout: float) -> dict: 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, + "alive": alive, + "last_seen_seconds_ago": round(age, 1), + "status": node.status, + "queue_depth": _effective_queue_depth(node), + "heartbeat_queue_depth": node.queue_depth, + "proxy_inflight": node.proxy_inflight, + "total_requests": node.total_requests, "heartbeat_success_rate": hb_rate, "inference_success_rate": inf_rate, "capacity": _node_capacity_summary(node), @@ -1722,7 +1745,7 @@ def _registration_ban_error(contracts: Any | None, wallet_address: str | None) - return None -def _tracker_log(server: "_TrackerHTTPServer", level: str, message: str, **fields: Any) -> None: +def _tracker_log(server: "_TrackerHTTPServer", level: str, message: str, **fields: Any) -> None: event = { "ts": time.time(), "level": level, @@ -1736,11 +1759,40 @@ def _tracker_log(server: "_TrackerHTTPServer", level: str, message: str, **field 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( + suffix = f" {extras}" if extras else "" + print(f"[tracker] {level}: {message}{suffix}", flush=True) + + +def _tracker_log_proxy_progress( + server: "_TrackerHTTPServer", + *, + request_id: str, + model: str, + route_model: str, + tokens: int, + started: float, + route_nodes: list["_NodeEntry"], + stream: bool = True, + relay: bool = False, +) -> None: + elapsed = time.monotonic() - started + _tracker_log( + server, + "info", + "proxy progress", + request_id=request_id, + model=model, + route_model=route_model, + stream=stream, + relay=relay or None, + tokens=tokens, + elapsed_seconds=round(elapsed, 4), + tokens_per_sec=round(tokens / elapsed, 4) if elapsed > 0 else 0.0, + route=_node_route_summary(route_nodes), + ) + + +def _node_id_for_registration( endpoint: str, model: str, wallet_address: str | None, @@ -2485,13 +2537,23 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): # Strip the first-shard node we're about to proxy to — it's already handling the request. downstream_hops = [h for h in route_hops if h["endpoint"].rstrip("/") != node.endpoint.rstrip("/")] downstream_urls = json.dumps(downstream_hops) - route_debug = " -> ".join( - f"{n.node_id}@{n.endpoint}[{n.shard_start}-{n.shard_end}]" - for n in route_nodes - ) - _tracker_log( - server, - "info", + route_debug = " -> ".join( + f"{n.node_id}@{n.endpoint}[{n.shard_start}-{n.shard_end}]" + for n in route_nodes + ) + inflight_nodes = route_nodes or [node] + inflight_recorded = True + _record_proxy_inflight(server, inflight_nodes, 1) + + def finish_proxy_inflight() -> None: + nonlocal inflight_recorded + if inflight_recorded: + _record_proxy_inflight(server, inflight_nodes, -1) + inflight_recorded = False + + _tracker_log( + server, + "info", "proxy route selected", request_id=request_id, model=model, @@ -2542,14 +2604,15 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): if first is not None and first.get("stream"): # Streamed response (US-036): forward SSE chunks as they arrive # and run the same token accounting as the direct stream path. - self._stream_relayed_frames( - first, frames, started, - model, route_model, route_nodes, api_key, node_work, - request_body=body, - ) - return - if first is not None: - elapsed = time.monotonic() - started + self._stream_relayed_frames( + first, frames, started, + model, route_model, route_nodes, api_key, node_work, + request_body=body, + ) + finish_proxy_inflight() + return + if first is not None: + elapsed = time.monotonic() - started self._send_relayed_response(first) if int(first.get("status", 503)) < 400: body_text = first.get("body") or "" @@ -2572,11 +2635,12 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): tokens_per_sec=round(tokens / elapsed, 4) if elapsed > 0 else 0.0, route=_node_route_summary(route_nodes), ) - self._bill_completed( - api_key, model, tokens, node_work, - input_tokens=in_tokens, output_tokens=out_tokens, - ) - return + self._bill_completed( + api_key, model, tokens, node_work, + input_tokens=in_tokens, output_tokens=out_tokens, + ) + finish_proxy_inflight() + return _tracker_log( server, "warn", @@ -2597,11 +2661,12 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(err_body))) self.end_headers() - try: - self.wfile.write(err_body) - except BrokenPipeError: - pass - return + try: + self.wfile.write(err_body) + except BrokenPipeError: + pass + finish_proxy_inflight() + return except Exception as exc: if node.relay_addr: _tracker_log( @@ -2622,12 +2687,13 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): target_url=target_url, error=repr(exc), ) - self._send_json(503, {"error": { - "message": f"upstream node unreachable: {exc}", - "type": "service_unavailable", - "code": "upstream_error", - }}) - return + self._send_json(503, {"error": { + "message": f"upstream node unreachable: {exc}", + "type": "service_unavailable", + "code": "upstream_error", + }}) + finish_proxy_inflight() + return with upstream: content_type = upstream.headers.get("Content-Type", "application/json") @@ -2718,12 +2784,13 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(resp_body))) self.end_headers() - try: - self.wfile.write(resp_body) - except BrokenPipeError: - pass - - def _record_observed_throughput( + try: + self.wfile.write(resp_body) + except BrokenPipeError: + pass + finish_proxy_inflight() + + def _record_observed_throughput( self, requested_model: str, route_model: str, diff --git a/tests/test_tracker_routing.py b/tests/test_tracker_routing.py index 43e26dc..dbc7f9b 100644 --- a/tests/test_tracker_routing.py +++ b/tests/test_tracker_routing.py @@ -324,6 +324,177 @@ def test_tracker_serves_health_while_proxy_request_is_in_flight(): slow_thread.join(timeout=1.0) +def test_tracker_route_log_counts_proxy_inflight_requests(): + entered = threading.Event() + release = threading.Event() + + class SlowChatHandler(http.server.BaseHTTPRequestHandler): + def log_message(self, fmt, *args): + pass + + def do_POST(self): + if self.path != "/v1/chat/completions": + self.send_response(404) + self.end_headers() + return + length = int(self.headers.get("Content-Length", 0)) + self.rfile.read(length) + entered.set() + release.wait(timeout=3.0) + body = json.dumps({"choices": [{"message": {"content": "ok"}}]}).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + slow_node = http.server.HTTPServer(("127.0.0.1", 0), SlowChatHandler) + slow_thread = threading.Thread(target=slow_node.serve_forever, daemon=True) + slow_thread.start() + tracker = TrackerServer(heartbeat_timeout=60.0) + tracker_port = tracker.start() + errors = [] + try: + _post_json( + f"http://127.0.0.1:{tracker_port}/v1/nodes/register", + {"endpoint": f"http://127.0.0.1:{slow_node.server_address[1]}", + "model": "burst-model", "num_layers": 1, + "shard_start": 0, "shard_end": 0, + "hardware_profile": {}, "score": 1.0}, + ) + + def call_proxy(): + try: + _post_json( + f"http://127.0.0.1:{tracker_port}/v1/chat/completions", + {"model": "burst-model", "messages": [{"role": "user", "content": "hi"}]}, + ) + except Exception as exc: + errors.append(exc) + + first = threading.Thread(target=call_proxy) + second = threading.Thread(target=call_proxy) + first.start() + assert entered.wait(timeout=2.0) + second.start() + + selected_events = [] + deadline = time.time() + 2.0 + while time.time() < deadline: + console = _get_json(f"http://127.0.0.1:{tracker_port}/v1/console") + selected_events = [ + event for event in console["events"] + if event["message"] == "proxy route selected" + ] + if len(selected_events) >= 2: + break + time.sleep(0.05) + + assert len(selected_events) >= 2 + second_nodes = selected_events[-1]["fields"]["nodes"] + assert second_nodes[0]["queue_depth"] == 2 + assert second_nodes[0]["proxy_inflight"] == 2 + finally: + release.set() + first.join(timeout=3.0) + second.join(timeout=3.0) + tracker.stop() + slow_node.shutdown() + slow_node.server_close() + slow_thread.join(timeout=1.0) + + assert not first.is_alive() + assert not second.is_alive() + assert not errors + + +def test_tracker_logs_stream_progress_before_request_completes(): + chunk_sent = threading.Event() + release = threading.Event() + + class StreamingChatHandler(http.server.BaseHTTPRequestHandler): + def log_message(self, fmt, *args): + pass + + def do_POST(self): + if self.path != "/v1/chat/completions": + self.send_response(404) + self.end_headers() + return + self.rfile.read(int(self.headers.get("Content-Length", 0))) + self.send_response(200) + self.send_header("Content-Type", "text/event-stream; charset=utf-8") + self.end_headers() + payload = json.dumps({ + "choices": [{"delta": {"content": "hello world"}}], + }).encode() + self.wfile.write(b"data: " + payload + b"\n\n") + self.wfile.flush() + chunk_sent.set() + release.wait(timeout=3.0) + self.wfile.write(b"data: [DONE]\n\n") + self.wfile.flush() + + node = http.server.HTTPServer(("127.0.0.1", 0), StreamingChatHandler) + node_thread = threading.Thread(target=node.serve_forever, daemon=True) + node_thread.start() + tracker = TrackerServer(heartbeat_timeout=60.0) + tracker_port = tracker.start() + response = None + try: + _post_json( + f"http://127.0.0.1:{tracker_port}/v1/nodes/register", + {"endpoint": f"http://127.0.0.1:{node.server_address[1]}", + "model": "stream-progress-model", "num_layers": 1, + "shard_start": 0, "shard_end": 0, + "hardware_profile": {}, "score": 1.0}, + ) + req = urllib.request.Request( + f"http://127.0.0.1:{tracker_port}/v1/chat/completions", + data=json.dumps({ + "model": "stream-progress-model", + "stream": True, + "messages": [{"role": "user", "content": "hi"}], + }).encode(), + headers={"Content-Type": "application/json"}, + method="POST", + ) + response = urllib.request.urlopen(req, timeout=3.0) + first_line = response.readline() + assert first_line.startswith(b"data:") + assert chunk_sent.wait(timeout=1.0) + + progress_events = [] + deadline = time.time() + 2.0 + while time.time() < deadline: + console = _get_json(f"http://127.0.0.1:{tracker_port}/v1/console") + progress_events = [ + event for event in console["events"] + if event["message"] == "proxy progress" + ] + if progress_events: + break + time.sleep(0.05) + + assert progress_events + fields = progress_events[-1]["fields"] + assert fields["tokens"] == 2 + assert fields["tokens_per_sec"] > 0 + active = [ + event for event in console["events"] + if event["message"] == "proxy route selected" + ] + assert active + finally: + release.set() + if response is not None: + response.close() + tracker.stop() + node.shutdown() + node.server_close() + node_thread.join(timeout=1.0) + + def test_tracker_routes_hf_model_alias_from_quickstart(): """The documented qwen2.5-0.5b alias resolves a full HF repo registration.""" tracker = TrackerServer() @@ -536,6 +707,101 @@ def test_tracker_route_endpoint_routes_split_preset_nodes_by_alias(): assert [node["start_layer"] for node in response["nodes"]] == [0, 22] +def test_tracker_route_endpoint_ignores_model_case_and_outer_whitespace(): + tracker = TrackerServer(model_presets={ + "qwen3.6-35b-a3b": { + "layers_start": 0, + "layers_end": 39, + "hf_repo": "unsloth/Qwen3.6-35B-A3B", + "aliases": ["Qwen3.6-35B-A3B"], + } + }) + tracker_port = tracker.start() + try: + _post_json( + f"http://127.0.0.1:{tracker_port}/v1/nodes/register", + {"endpoint": "http://127.0.0.1:9101", + "model": "qwen3.6-35b-a3b", + "hf_repo": "unsloth/Qwen3.6-35B-A3B", + "num_layers": 40, + "shard_start": 0, + "shard_end": 39, + "tracker_mode": True, + "hardware_profile": {}, + "score": 1.0}, + ) + + response = _get_json( + f"http://127.0.0.1:{tracker_port}/v1/route?model=%20Qwen3.6-35B-A3B%20" + ) + finally: + tracker.stop() + + assert response["route"] == ["http://127.0.0.1:9101"] + + +def test_tracker_proxy_ignores_model_case_and_outer_whitespace(): + class ChatHandler(http.server.BaseHTTPRequestHandler): + def log_message(self, fmt, *args): + pass + + def do_POST(self): + if self.path != "/v1/chat/completions": + self.send_response(404) + self.end_headers() + return + length = int(self.headers.get("Content-Length", 0)) + request_body = json.loads(self.rfile.read(length) or b"{}") + body = json.dumps({ + "model": request_body["model"], + "choices": [{"message": {"content": "ok"}}], + }).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + node = http.server.HTTPServer(("127.0.0.1", 0), ChatHandler) + node_thread = threading.Thread(target=node.serve_forever, daemon=True) + node_thread.start() + tracker = TrackerServer(model_presets={ + "qwen3.6-35b-a3b": { + "layers_start": 0, + "layers_end": 39, + "hf_repo": "unsloth/Qwen3.6-35B-A3B", + "aliases": ["Qwen3.6-35B-A3B"], + } + }) + tracker_port = tracker.start() + try: + _post_json( + f"http://127.0.0.1:{tracker_port}/v1/nodes/register", + {"endpoint": f"http://127.0.0.1:{node.server_address[1]}", + "model": "qwen3.6-35b-a3b", + "hf_repo": "unsloth/Qwen3.6-35B-A3B", + "num_layers": 40, + "shard_start": 0, + "shard_end": 39, + "tracker_mode": True, + "hardware_profile": {}, + "score": 1.0}, + ) + + response = _post_json( + f"http://127.0.0.1:{tracker_port}/v1/chat/completions", + {"model": "Qwen3.6-35B-A3B ", + "messages": [{"role": "user", "content": "hi"}]}, + ) + finally: + tracker.stop() + node.shutdown() + node.server_close() + node_thread.join(timeout=1.0) + + assert response["choices"][0]["message"]["content"] == "ok" + + def test_tracker_registration_node_id_includes_wallet_prefix_and_stable_suffix(): tracker = TrackerServer() tracker_port = tracker.start()