From ad3368f7eaded00dfd1d56c63d232f77a4fa3706 Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Tue, 30 Jun 2026 12:22:21 +0200 Subject: [PATCH 1/4] docs --- QUICKSTART.md | 26 +++++++++++++++++++++++++- 1 file changed, 25 insertions(+), 1 deletion(-) diff --git a/QUICKSTART.md b/QUICKSTART.md index e1d2c9b..08c9a92 100644 --- a/QUICKSTART.md +++ b/QUICKSTART.md @@ -11,8 +11,14 @@ Tested on: AMD Ryzen AI Max (Strix Halo APU), 124 GB RAM, Linux, CPU inference. # Clone and enter repo cd /run/media/popov/d/DEV/repos/d-popov.com/AI +# Create the virtualenv if it does not exist yet +python3 -m venv .venv + +# Keep packaging tools current enough for editable installs +.venv/bin/python -m pip install --upgrade pip setuptools wheel + # Install Python packages (editable — picks up code changes immediately) -.venv/bin/pip install -e packages/tracker packages/node packages/p2p packages/relay +.venv/bin/pip install -e packages/tracker -e packages/node -e packages/p2p -e packages/gateway # CPU-only PyTorch (skip if you have CUDA/ROCm already) .venv/bin/pip install torch --index-url https://download.pytorch.org/whl/cpu @@ -24,6 +30,24 @@ cd /run/media/popov/d/DEV/repos/d-popov.com/AI > **NVIDIA GPU (CUDA):** replace the torch line with `pip install torch` (default index). > **AMD GPU (ROCm):** `pip install torch --index-url https://download.pytorch.org/whl/rocm6.2` +### Windows / WSL2 + +Run the Linux commands from WSL, not Git Bash. From the repo opened in Git Bash: + +```bash +wsl +cd /mnt/d/DEV/workspace/REPOS/git.d-popov.com/neuron-tai +python3 -m venv .venv +.venv/bin/python -m pip install --upgrade pip setuptools wheel +.venv/bin/pip install -e packages/tracker -e packages/node -e packages/p2p -e packages/gateway +.venv/bin/pip install torch --index-url https://download.pytorch.org/whl/cpu +.venv/bin/pip install transformers accelerate +.venv/bin/meshnet-node --help +``` + +If `.venv/bin/meshnet-node` is missing, the editable install step did not finish +successfully. Re-run the `.venv/bin/pip install -e ...` command above inside WSL. + --- ## Step 1 — Start the tracker (Terminal 1) From 9ca198ee1e7f564a0f0d3a11113e8a5e7ea76474 Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Tue, 30 Jun 2026 15:01:49 +0200 Subject: [PATCH 2/4] Add relay-backed public node registration --- QUICKSTART.md | 32 ++++- packages/node/meshnet_node/relay_bridge.py | 148 +++++++++++++++++++++ packages/node/meshnet_node/startup.py | 79 +++++++++++ packages/node/pyproject.toml | 1 + packages/relay/meshnet_relay/server.py | 57 ++++++++ packages/tracker/meshnet_tracker/cli.py | 6 + packages/tracker/meshnet_tracker/server.py | 101 ++++++++++++++ packages/tracker/pyproject.toml | 4 + tests/test_gossip_and_relay.py | 90 +++++++++++++ 9 files changed, 516 insertions(+), 2 deletions(-) create mode 100644 packages/node/meshnet_node/relay_bridge.py diff --git a/QUICKSTART.md b/QUICKSTART.md index 08c9a92..1182538 100644 --- a/QUICKSTART.md +++ b/QUICKSTART.md @@ -18,7 +18,7 @@ python3 -m venv .venv .venv/bin/python -m pip install --upgrade pip setuptools wheel # Install Python packages (editable — picks up code changes immediately) -.venv/bin/pip install -e packages/tracker -e packages/node -e packages/p2p -e packages/gateway +.venv/bin/pip install -e packages/tracker -e packages/node -e packages/p2p -e packages/gateway -e packages/relay # CPU-only PyTorch (skip if you have CUDA/ROCm already) .venv/bin/pip install torch --index-url https://download.pytorch.org/whl/cpu @@ -39,7 +39,7 @@ wsl cd /mnt/d/DEV/workspace/REPOS/git.d-popov.com/neuron-tai python3 -m venv .venv .venv/bin/python -m pip install --upgrade pip setuptools wheel -.venv/bin/pip install -e packages/tracker -e packages/node -e packages/p2p -e packages/gateway +.venv/bin/pip install -e packages/tracker -e packages/node -e packages/p2p -e packages/gateway -e packages/relay .venv/bin/pip install torch --index-url https://download.pytorch.org/whl/cpu .venv/bin/pip install transformers accelerate .venv/bin/meshnet-node --help @@ -48,6 +48,34 @@ python3 -m venv .venv If `.venv/bin/meshnet-node` is missing, the editable install step did not finish successfully. Re-run the `.venv/bin/pip install -e ...` command above inside WSL. +### Public tracker + WSS relay + +For internet nodes, expose one public HTTPS host and proxy these paths: + +```text +/v1/* -> meshnet-tracker, for registration, heartbeats, routing, and OpenAI requests +/ws -> meshnet-relay, for outbound node gossip/bridge connections +/rpc/* -> meshnet-relay, for tracker-to-node relay requests +``` + +Start the tracker with the public relay URL it should advertise: + +```bash +.venv/bin/meshnet-relay --host 0.0.0.0 --port 8765 +.venv/bin/meshnet-tracker start \ + --host 0.0.0.0 \ + --port 8081 \ + --relay-url wss://ai.neuron.d-popov.com/ws +``` + +Then a node only needs the public tracker address: + +```bash +.venv/bin/meshnet-node start \ + --tracker https://ai.neuron.d-popov.com \ + --model-id Qwen/Qwen2.5-0.5B-Instruct +``` + --- ## Step 1 — Start the tracker (Terminal 1) diff --git a/packages/node/meshnet_node/relay_bridge.py b/packages/node/meshnet_node/relay_bridge.py new file mode 100644 index 0000000..d7b63ea --- /dev/null +++ b/packages/node/meshnet_node/relay_bridge.py @@ -0,0 +1,148 @@ +"""Outbound relay bridge for NAT-safe node HTTP requests.""" + +from __future__ import annotations + +import json +import logging +import threading +import time +import urllib.error +import urllib.request +from dataclasses import dataclass + +log = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class RelayBridgeInfo: + peer_id: str + relay_addr: str + + +def _make_envelope(topic: str, payload: dict, peer_id: str) -> dict: + return { + "topic": topic, + "version": 1, + "from_peer": peer_id, + "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "msg_id": f"{peer_id}-{time.time_ns():x}", + "ttl": 1, + "payload": payload, + } + + +class RelayHttpBridge: + """Connect outbound to a relay and proxy relay HTTP requests to localhost.""" + + def __init__( + self, + relay_url: str, + peer_id: str, + local_base_url: str, + advertised_addr: str, + reconnect_interval: float = 3.0, + ) -> None: + self.relay_url = relay_url.rstrip("/") + self.peer_id = peer_id + self.local_base_url = local_base_url.rstrip("/") + self.advertised_addr = advertised_addr + self.reconnect_interval = reconnect_interval + self._running = False + self._thread: threading.Thread | None = None + self._connected = threading.Event() + + @property + def relay_addr(self) -> str: + base = self.relay_url + if base.endswith("/ws"): + base = base[:-3] + return f"{base}/rpc/{self.peer_id}" + + def start(self) -> RelayBridgeInfo: + self._running = True + self._thread = threading.Thread(target=self._run, daemon=True, name="relay-http-bridge") + self._thread.start() + return RelayBridgeInfo(peer_id=self.peer_id, relay_addr=self.relay_addr) + + def wait_connected(self, timeout: float = 5.0) -> bool: + return self._connected.wait(timeout) + + def stop(self) -> None: + self._running = False + if self._thread: + self._thread.join(timeout=3.0) + + def _run(self) -> None: + import websockets.sync.client as wsc # type: ignore[import] + + while self._running: + try: + with wsc.connect(self.relay_url, open_timeout=5) as ws: + self._connected.set() + ws.send(json.dumps(_make_envelope( + "peer-register", + {"peer_id": self.peer_id, "addr": self.advertised_addr}, + self.peer_id, + ))) + while self._running: + try: + raw = ws.recv(timeout=1) + except TimeoutError: + continue + try: + envelope = json.loads(raw) + except (TypeError, json.JSONDecodeError): + continue + if envelope.get("topic") != "relay-http-request": + continue + payload = envelope.get("payload", {}) + if payload.get("target_peer") not in {None, self.peer_id}: + continue + response = self._handle_request(payload) + ws.send(json.dumps(_make_envelope( + "relay-http-response", + response, + self.peer_id, + ))) + except Exception as exc: + self._connected.clear() + if self._running: + log.debug("relay bridge disconnected: %s", exc) + time.sleep(self.reconnect_interval) + + def _handle_request(self, payload: dict) -> dict: + request_id = str(payload.get("request_id") or "") + method = str(payload.get("method") or "POST").upper() + path = str(payload.get("path") or "/") + body_text = payload.get("body") or "" + headers = payload.get("headers") if isinstance(payload.get("headers"), dict) else {} + + url = f"{self.local_base_url}{path}" + data = body_text.encode() if isinstance(body_text, str) else bytes(body_text) + req = urllib.request.Request(url, data=data, headers=headers, method=method) + try: + with urllib.request.urlopen(req, timeout=300.0) as resp: + return { + "request_id": request_id, + "status": resp.status, + "headers": {"Content-Type": resp.headers.get("Content-Type", "application/json")}, + "body": resp.read().decode(errors="replace"), + } + except urllib.error.HTTPError as exc: + return { + "request_id": request_id, + "status": exc.code, + "headers": {"Content-Type": exc.headers.get("Content-Type", "application/json")}, + "body": exc.read().decode(errors="replace"), + } + except Exception as exc: + return { + "request_id": request_id, + "status": 503, + "headers": {"Content-Type": "application/json"}, + "body": json.dumps({"error": f"relay bridge local request failed: {exc}"}), + } + + +def peer_id_from_wallet(wallet_address: str) -> str: + return wallet_address[:16] if len(wallet_address) >= 16 else wallet_address diff --git a/packages/node/meshnet_node/startup.py b/packages/node/meshnet_node/startup.py index 14bdfac..afe05b9 100644 --- a/packages/node/meshnet_node/startup.py +++ b/packages/node/meshnet_node/startup.py @@ -15,6 +15,7 @@ from typing import Any from .downloader import compute_shard_checksum, download_shard from .hardware import detect_hardware +from .relay_bridge import RelayHttpBridge, peer_id_from_wallet from .server import StubNodeServer from .torch_server import TorchNodeServer from .wallet import load_or_create_wallet @@ -34,6 +35,57 @@ def _get_json(url: str, timeout: float = 10.0) -> dict: return json.loads(r.read()) +def _discover_relay_url(tracker_url: str) -> str | None: + try: + network_map = _get_json(f"{tracker_url}/v1/network/map", timeout=5.0) + except Exception: + return None + relay_url = network_map.get("relay_url") + return relay_url if isinstance(relay_url, str) and relay_url else None + + +def _start_relay_bridge_if_available( + tracker_url: str, + wallet_address: str, + local_base_url: str, + advertised_endpoint: str, +) -> tuple[RelayHttpBridge | None, dict]: + relay_url = _discover_relay_url(tracker_url) + if not relay_url: + return None, {} + peer_id = peer_id_from_wallet(wallet_address) + bridge = RelayHttpBridge( + relay_url=relay_url, + peer_id=peer_id, + local_base_url=local_base_url, + advertised_addr=advertised_endpoint, + ) + info = bridge.start() + if bridge.wait_connected(timeout=5.0): + print(f" Relay connected — {info.relay_addr}", flush=True) + else: + print(f" Relay configured but not connected yet — {info.relay_addr}", flush=True) + return bridge, { + "relay_addr": info.relay_addr, + "peer_id": info.peer_id, + } + + +def _attach_relay_bridge(node: StubNodeServer | TorchNodeServer, bridge: RelayHttpBridge | None) -> None: + setattr(node, "relay_bridge", bridge) + if bridge is None: + return + original_stop = node.stop + + def _stop_with_bridge() -> None: + try: + bridge.stop() + finally: + original_stop() + + node.stop = _stop_with_bridge # type: ignore[method-assign] + + def _start_heartbeat( tracker_url: str, node_id: str, @@ -212,6 +264,14 @@ def run_startup( shard_label = f"layers {shard_start}–{shard_end}" public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host) endpoint = f"http://{public_host}:{actual_port}" + local_base_url = f"http://127.0.0.1:{actual_port}" + relay_bridge, relay_fields = _start_relay_bridge_if_available( + tracker_url, + address, + local_base_url, + endpoint, + ) + _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) reg_payload = { @@ -226,6 +286,7 @@ def run_startup( "quantization": quantization, "score": 1.0, "tracker_mode": (shard_start == 0), + **relay_fields, } tracker_node_id: str | None = None try: @@ -289,6 +350,14 @@ def run_startup( 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}" + local_base_url = f"http://127.0.0.1:{actual_port}" + relay_bridge, relay_fields = _start_relay_bridge_if_available( + tracker_url, + address, + local_base_url, + endpoint, + ) + _attach_relay_bridge(node, relay_bridge) auto_reg_payload = { "endpoint": endpoint, "model": assigned_hf_repo.split("/")[-1], @@ -301,6 +370,7 @@ def run_startup( "quantization": quantization, "score": 1.0, "tracker_mode": (assigned_shard_start == 0), + **relay_fields, } tracker_node_id = None try: @@ -376,6 +446,14 @@ def run_startup( 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}" + local_base_url = f"http://127.0.0.1:{actual_port}" + relay_bridge, relay_fields = _start_relay_bridge_if_available( + tracker_url, + address, + local_base_url, + endpoint, + ) + _attach_relay_bridge(node, relay_bridge) # 6. Register with tracker print("Registering with tracker...", flush=True) @@ -391,6 +469,7 @@ def run_startup( "hardware_profile": hw, "wallet_address": address, "score": 1.0, + **relay_fields, }, ) node_id = str(reg_resp["node_id"]) diff --git a/packages/node/pyproject.toml b/packages/node/pyproject.toml index 6acbc09..657ea57 100644 --- a/packages/node/pyproject.toml +++ b/packages/node/pyproject.toml @@ -17,6 +17,7 @@ dependencies = [ "safetensors>=0.4", "torch>=2.1", "transformers>=4.39", + "websockets>=13", "zstandard>=0.22", ] diff --git a/packages/relay/meshnet_relay/server.py b/packages/relay/meshnet_relay/server.py index bca3d86..fd73f76 100644 --- a/packages/relay/meshnet_relay/server.py +++ b/packages/relay/meshnet_relay/server.py @@ -16,6 +16,7 @@ import asyncio import json import logging import threading +import uuid from pathlib import Path from .peer_registry import PeerRegistry @@ -56,6 +57,7 @@ class RelayServer: self._ready = threading.Event() self._running = False self._stop_event: asyncio.Event | None = None + self._pending_rpc: dict[str, asyncio.Future] = {} @property def registry(self) -> PeerRegistry: @@ -121,6 +123,9 @@ class RelayServer: if path.startswith("/relay/"): peer_id = path[len("/relay/"):] await self._handle_circuit_relay(ws, peer_id) + elif path.startswith("/rpc/"): + peer_id = path[len("/rpc/"):] + await self._handle_rpc(ws, peer_id) elif path == "/health": await ws.send(json.dumps({"status": "ok", "peers": len(self._registry)})) await ws.close() @@ -164,6 +169,14 @@ class RelayServer: })) continue + if topic == "relay-http-response": + payload = envelope.get("payload", {}) + request_id = payload.get("request_id") + fut = self._pending_rpc.pop(request_id, None) + if fut is not None and not fut.done(): + fut.set_result(payload) + continue + # Fan out to all other registered peers if peer_id: self._registry.touch(peer_id) @@ -205,6 +218,50 @@ class RelayServer: return_exceptions=True, ) + async def _handle_rpc(self, ws_requester, target_peer_id: str) -> None: + """Send one HTTP-shaped request to a connected peer and relay its response.""" + target = self._registry.get(target_peer_id) + if target is None: + await ws_requester.send(json.dumps({ + "status": 503, + "headers": {"Content-Type": "application/json"}, + "body": json.dumps({"error": f"peer {target_peer_id!r} not connected to relay"}), + })) + await ws_requester.close() + return + + try: + raw = await asyncio.wait_for(ws_requester.recv(), timeout=30.0) + payload = json.loads(raw) + except Exception: + await ws_requester.close(1003, "invalid relay rpc request") + return + + request_id = str(payload.get("request_id") or uuid.uuid4()) + payload["request_id"] = request_id + payload["target_peer"] = target_peer_id + fut = self._loop.create_future() if self._loop is not None else asyncio.get_running_loop().create_future() + self._pending_rpc[request_id] = fut + try: + await target.ws.send(json.dumps({ + "topic": "relay-http-request", + "version": 1, + "from_peer": "relay", + "payload": payload, + })) + response = await asyncio.wait_for(fut, timeout=310.0) + await ws_requester.send(json.dumps(response)) + except asyncio.TimeoutError: + await ws_requester.send(json.dumps({ + "request_id": request_id, + "status": 504, + "headers": {"Content-Type": "application/json"}, + "body": json.dumps({"error": "relay rpc timed out"}), + })) + finally: + self._pending_rpc.pop(request_id, None) + await ws_requester.close() + async def _broadcast(raw: str | bytes, peers: list) -> None: """Send raw message to all peers; ignore individual send failures.""" diff --git a/packages/tracker/meshnet_tracker/cli.py b/packages/tracker/meshnet_tracker/cli.py index b9b05b0..fe432a1 100644 --- a/packages/tracker/meshnet_tracker/cli.py +++ b/packages/tracker/meshnet_tracker/cli.py @@ -33,6 +33,11 @@ def main() -> None: default=None, help="This tracker's own URL as seen by peers (auto-derived from --host/--port if omitted)", ) + start_cmd.add_argument( + "--relay-url", + default=None, + help="Public ws(s):// relay URL advertised to nodes, for example wss://ai.neuron.d-popov.com/ws", + ) args = parser.parse_args() @@ -44,6 +49,7 @@ def main() -> None: heartbeat_timeout=args.heartbeat_timeout, cluster_peers=cluster_peers or None, cluster_self_url=args.self_url, + relay_url=args.relay_url, ) port = server.start() print(f"meshnet-tracker listening on http://{args.host}:{port}", flush=True) diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index d79af49..a8113da 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -28,6 +28,8 @@ import socketserver import threading import time import urllib.parse +import urllib.request +import uuid from typing import Any from .gossip import NodeGossip @@ -238,6 +240,35 @@ def _coverage_gaps(coverage: list[dict]) -> list[tuple[int, int]]: ] +def _relay_http_request( + relay_addr: str, + path: str, + body: bytes, + headers: dict[str, str], + timeout: float = 310.0, +) -> dict | None: + """Send an HTTP-shaped request through a relay RPC WebSocket.""" + try: + import websockets.sync.client as wsc # type: ignore[import] + + request_id = str(uuid.uuid4()) + with wsc.connect(relay_addr, open_timeout=10, close_timeout=5) as ws: + ws.send(json.dumps({ + "request_id": request_id, + "method": "POST", + "path": path, + "headers": headers, + "body": body.decode(errors="replace"), + })) + raw = ws.recv(timeout=timeout) + response = json.loads(raw) + if response.get("request_id") not in {None, request_id}: + return None + return response + except Exception: + return None + + def _nodes_and_bounds_for_model( server: "_TrackerHTTPServer", model: str, @@ -432,6 +463,7 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): model_presets: dict, contracts: Any | None, minimum_stake: int, + relay_url: str | None = None, raft: "RaftNode | None" = None, gossip: "NodeGossip | None" = None, ) -> None: @@ -442,6 +474,7 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): self.model_presets = model_presets self.contracts = contracts self.minimum_stake = minimum_stake + self.relay_url = relay_url.rstrip("/") if relay_url else None self.raft = raft self.gossip = gossip @@ -511,6 +544,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): self._handle_assign(parsed) elif parsed.path == "/v1/network/assign": self._handle_network_assign(parsed) + elif parsed.path == "/v1/network/map": + self._handle_network_map() elif parsed.path == "/v1/models": self._handle_models() elif parsed.path.startswith("/v1/coverage/"): @@ -653,12 +688,38 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): { "node_id": node.node_id, "endpoint": node.endpoint, + "relay_addr": node.relay_addr, + "peer_id": node.peer_id, "benchmark_tokens_per_sec": node.benchmark_tokens_per_sec, } for node in tracker_nodes ], }) + def _handle_network_map(self) -> None: + server: _TrackerHTTPServer = self.server # type: ignore[assignment] + with server.lock: + self._purge_expired_nodes() + nodes = list(server.registry.values()) + self._send_json(200, { + "relay_url": server.relay_url, + "nodes": [ + { + "node_id": node.node_id, + "endpoint": node.endpoint, + "relay_addr": node.relay_addr, + "peer_id": node.peer_id, + "model": node.model, + "hf_repo": node.hf_repo, + "shard_start": node.shard_start, + "shard_end": node.shard_end, + "tracker_mode": node.tracker_mode, + "last_heartbeat": node.last_heartbeat, + } + for node in nodes + ], + }) + # ---------------------------------------------------------------- OpenAI proxy def _handle_proxy_chat(self) -> None: @@ -771,6 +832,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"trying relay {node.relay_addr}", + flush=True, + ) + relayed = _relay_http_request( + node.relay_addr, + path="/v1/chat/completions", + body=raw_body, + headers={ + "Content-Type": "application/json", + "X-Meshnet-Route": downstream_urls, + **({"Authorization": auth} if auth else {}), + }, + ) + if relayed is not None: + self._send_relayed_response(relayed) + return print(f"[tracker] proxy failed {request_id}: {target_url}: {exc}", flush=True) self._send_json(503, {"error": { "message": f"upstream node unreachable: {exc}", @@ -812,6 +892,20 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): except BrokenPipeError: pass + def _send_relayed_response(self, response: dict) -> None: + status = int(response.get("status", 503)) + headers = response.get("headers") if isinstance(response.get("headers"), dict) else {} + body_text = response.get("body") or "" + body = body_text.encode() if isinstance(body_text, str) else bytes(body_text) + self.send_response(status) + self.send_header("Content-Type", headers.get("Content-Type", "application/json")) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + try: + self.wfile.write(body) + except BrokenPipeError: + pass + def _handle_register(self): server: _TrackerHTTPServer = self.server # type: ignore[assignment] body = self._read_json_body() @@ -1340,6 +1434,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): { "node_id": e.node_id, "endpoint": e.endpoint, + "relay_addr": e.relay_addr, + "peer_id": e.peer_id, "wallet_address": e.wallet_address, "shard_start": e.shard_start, "shard_end": e.shard_end, @@ -1400,6 +1496,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): { "node_id": e.node_id, "endpoint": e.endpoint, + "relay_addr": e.relay_addr, + "peer_id": e.peer_id, "wallet_address": e.wallet_address, "shard_start": e.shard_start, "shard_end": e.shard_end, @@ -1433,6 +1531,7 @@ class TrackerServer: minimum_stake: int = 0, cluster_peers: list[str] | None = None, cluster_self_url: str | None = None, + relay_url: str | None = None, ) -> None: self._host = host self._requested_port = port @@ -1445,6 +1544,7 @@ class TrackerServer: self._minimum_stake = minimum_stake self._cluster_peers: list[str] = list(cluster_peers) if cluster_peers else [] self._cluster_self_url: str | None = cluster_self_url + self._relay_url = relay_url self._registry: dict[str, _NodeEntry] = {} self._lock = threading.Lock() self._server: _TrackerHTTPServer | None = None @@ -1469,6 +1569,7 @@ class TrackerServer: self._model_presets, self._contracts, self._minimum_stake, + self._relay_url, ) self.port = self._server.server_address[1] diff --git a/packages/tracker/pyproject.toml b/packages/tracker/pyproject.toml index 24738d7..1737ec4 100644 --- a/packages/tracker/pyproject.toml +++ b/packages/tracker/pyproject.toml @@ -8,6 +8,10 @@ version = "0.1.0" description = "Distributed Inference Network node registry and route selection" requires-python = ">=3.10" +dependencies = [ + "websockets>=13", +] + [project.scripts] meshnet-tracker = "meshnet_tracker.cli:main" diff --git a/tests/test_gossip_and_relay.py b/tests/test_gossip_and_relay.py index 7b89287..7344444 100644 --- a/tests/test_gossip_and_relay.py +++ b/tests/test_gossip_and_relay.py @@ -295,6 +295,61 @@ def test_relay_circuit_relay_proxies_message(): assert received_via_relay, "NAT'd peer did not receive message via circuit relay" +def test_relay_rpc_round_trips_http_request_to_peer(): + """Relay /rpc/ sends one HTTP-shaped request to a connected peer.""" + import websockets.sync.client as wsc # type: ignore[import] + from meshnet_relay.server import RelayServer + + relay = RelayServer(host="127.0.0.1", port=0) + port = relay.start() + ready = threading.Event() + + def run_peer(): + with wsc.connect(f"ws://127.0.0.1:{port}/ws") as ws: + ws.send(json.dumps({ + "topic": "peer-register", + "version": 1, + "from_peer": "rpc_peer", + "msg_id": "rpc-reg-001", + "payload": {"peer_id": "rpc_peer", "addr": ""}, + })) + ws.recv() + ready.set() + envelope = json.loads(ws.recv(timeout=3)) + payload = envelope["payload"] + ws.send(json.dumps({ + "topic": "relay-http-response", + "version": 1, + "from_peer": "rpc_peer", + "payload": { + "request_id": payload["request_id"], + "status": 200, + "headers": {"Content-Type": "application/json"}, + "body": json.dumps({"ok": True, "path": payload["path"]}), + }, + })) + + peer_thread = threading.Thread(target=run_peer, daemon=True) + peer_thread.start() + assert ready.wait(timeout=5) + + with wsc.connect(f"ws://127.0.0.1:{port}/rpc/rpc_peer") as ws: + ws.send(json.dumps({ + "request_id": "req-1", + "method": "POST", + "path": "/v1/chat/completions", + "headers": {"Content-Type": "application/json"}, + "body": "{}", + })) + response = json.loads(ws.recv(timeout=5)) + + relay.stop() + peer_thread.join(timeout=3) + + assert response["status"] == 200 + assert json.loads(response["body"]) == {"ok": True, "path": "/v1/chat/completions"} + + # --------------------------------------------------------------------------- # Tracker gossip fields tests # --------------------------------------------------------------------------- @@ -349,6 +404,41 @@ def test_tracker_accepts_registration_without_gossip_fields(): assert "node_id" in resp +def test_tracker_network_map_exposes_relay_and_registered_peer(): + import json as _json + import urllib.request + + from meshnet_tracker.server import TrackerServer + + tracker = TrackerServer(host="127.0.0.1", port=0, relay_url="wss://ai.neuron.d-popov.com/ws") + port = tracker.start() + url = f"http://127.0.0.1:{port}" + try: + req = urllib.request.Request( + f"{url}/v1/nodes/register", + data=_json.dumps({ + "endpoint": "http://192.0.2.10:7000", + "model": "stub-model", + "shard_start": 0, + "shard_end": 31, + "relay_addr": "wss://ai.neuron.d-popov.com/rpc/peer123", + "peer_id": "peer123", + }).encode(), + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=5): + pass + with urllib.request.urlopen(f"{url}/v1/network/map", timeout=5) as resp: + body = _json.loads(resp.read()) + finally: + tracker.stop() + + assert body["relay_url"] == "wss://ai.neuron.d-popov.com/ws" + assert body["nodes"][0]["relay_addr"] == "wss://ai.neuron.d-popov.com/rpc/peer123" + assert body["nodes"][0]["peer_id"] == "peer123" + + # --------------------------------------------------------------------------- # mDNS (no-op without zeroconf installed) # --------------------------------------------------------------------------- From df473ef278bd48fd431908863773c00dc0d9b6af Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Tue, 30 Jun 2026 16:16:20 +0200 Subject: [PATCH 3/4] fix LAN discovery/connection --- ...ary-data-plane-and-peer-weight-transfer.md | 68 +++++++++++++++++ _DEV_NOTES.md | 4 + packages/node/meshnet_node/cli.py | 10 ++- packages/tracker/meshnet_tracker/cli.py | 31 ++++---- tests/test_mining_cli.py | 73 +++++++++++++++++++ 5 files changed, 171 insertions(+), 15 deletions(-) create mode 100644 .scratch/distributed-inference-network/issues/19-binary-data-plane-and-peer-weight-transfer.md create mode 100644 _DEV_NOTES.md diff --git a/.scratch/distributed-inference-network/issues/19-binary-data-plane-and-peer-weight-transfer.md b/.scratch/distributed-inference-network/issues/19-binary-data-plane-and-peer-weight-transfer.md new file mode 100644 index 0000000..f768a72 --- /dev/null +++ b/.scratch/distributed-inference-network/issues/19-binary-data-plane-and-peer-weight-transfer.md @@ -0,0 +1,68 @@ +# US-019 — Binary data plane and optional peer weight transfer + +Status: needs-triage +Priority: Low +Stage: Design parking lot + +## Context + +The current project focus is inference democratization: let small GPU owners contribute useful compute and let users run inference on models larger than one host can serve alone. Weight distribution is useful, but it is secondary to low-latency distributed inference. + +Recent findings: + +- HuggingFace already handles initial model origin distribution well enough for the first working version. +- The inference-critical path is activation transfer between shard nodes, not torrenting model files. +- Torrent/content-addressed transfer is a good future fit for model weights, shard cache replication, fine-tuned models, and offline/local swarm behavior. +- Torrenting is not a good fit for activation traffic because activations are latency-sensitive, ordered, session-specific binary streams. + +## Design note + +Keep the tracker as the control plane: + +- node registration +- heartbeats +- route selection +- model/shard manifests +- peer/checksum metadata + +Keep binary payloads on the data plane: + +- direct node-to-node activation transfer where reachable +- relay/QUIC/WSS fallback where direct transport is unavailable +- future peer weight transfer as content-addressed blobs or pieces + +## Potential future direction + +For inference traffic: + +- Prefer direct binary transport with backpressure. +- Use raw binary activation frames rather than JSON/base64. +- Preserve tensor metadata out-of-band: shape, dtype, session, chunk index, encoding. +- Consider QUIC or a mature NAT-friendly transport for direct-when-possible, relay-when-needed behavior. + +For model weights: + +- Keep HuggingFace as default origin and fallback. +- Add peer cache transfer only as an optimization. +- Consider a real content-addressed/torrent-like library or sidecar for weight blobs. +- Store manifests and checksums in tracker state so peers can verify exact shard contents. +- Prefer piece/chunk transfer with resume and hash verification over one giant tarball. + +## Not in scope now + +- Replacing HuggingFace as the primary model origin. +- Building a full BitTorrent/IPFS/libp2p subsystem. +- Routing activation traffic through a torrent protocol. +- Making peer weight transfer mandatory for node startup. + +## Acceptance criteria for a future implementation issue + +- [ ] Activation transfer remains binary end-to-end and avoids JSON/base64 payloads. +- [ ] Tracker does not proxy large binary payloads except as an explicit fallback path. +- [ ] Weight transfer, if added, is optional and falls back to HuggingFace. +- [ ] Weight pieces are content-addressed and checksum-verified. +- [ ] The design preserves low-latency inference as the primary objective. + +## Comments + +Created as a low-priority design parking-lot item after discussing inference democratization versus weight distribution. Do not pick up for implementation until the core public tracker, relay, and binary activation path are stable. diff --git a/_DEV_NOTES.md b/_DEV_NOTES.md new file mode 100644 index 0000000..a9585f7 --- /dev/null +++ b/_DEV_NOTES.md @@ -0,0 +1,4 @@ + win + .venv/bin/meshnet-node start --tracker http://192.168.0.179:8081 --model Qwen/Qwen2.5-0.5B-Instruct --shard-start 20 + + diff --git a/packages/node/meshnet_node/cli.py b/packages/node/meshnet_node/cli.py index da3e864..a874176 100644 --- a/packages/node/meshnet_node/cli.py +++ b/packages/node/meshnet_node/cli.py @@ -26,6 +26,7 @@ def _run_node(cfg: dict) -> None: wallet_path=Path(cfg["wallet_path"]) if cfg.get("wallet_path") else None, cache_dir=Path(cfg["download_dir"]) if cfg.get("download_dir") else None, host=cfg.get("host", "0.0.0.0"), + advertise_host=cfg.get("advertise_host"), ) except Exception as exc: print(f"\nERROR: {exc}", file=sys.stderr, flush=True) @@ -86,6 +87,8 @@ def _cmd_default(args) -> int: overrides["port"] = args.port if args.host: overrides["host"] = args.host + if args.advertise_host: + overrides["advertise_host"] = args.advertise_host if overrides: cfg = merge_cli_overrides(cfg, **overrides) @@ -143,7 +146,11 @@ def _cmd_start(args) -> int: cfg = dict(DEFAULTS) cfg["tracker_url"] = args.tracker cfg["port"] = args.port - cfg["model_name"] = args.model + if args.model_id is None and "/" in args.model: + cfg["model_hf_repo"] = args.model + cfg["model_name"] = args.model.split("/")[-1] + else: + cfg["model_name"] = args.model cfg["quantization"] = args.quantization cfg["host"] = args.host if args.model_id: @@ -212,6 +219,7 @@ def main() -> None: parser.add_argument("--shard-end", type=int, metavar="N", help="Pin shard end layer") parser.add_argument("--port", type=int, metavar="N", help="Port to listen on") parser.add_argument("--host", metavar="ADDR", help="Interface to bind (default 0.0.0.0)") + parser.add_argument("--advertise-host", metavar="ADDR", help="Host/IP advertised to the tracker") parser.add_argument("--no-tui", action="store_true", help="Plain-text output (no rich dashboard)") parser.add_argument("--compact", action="store_true", help="Single-line status output") parser.add_argument("--reset-config", action="store_true", help="Re-run wizard even if config exists") diff --git a/packages/tracker/meshnet_tracker/cli.py b/packages/tracker/meshnet_tracker/cli.py index fe432a1..68fb851 100644 --- a/packages/tracker/meshnet_tracker/cli.py +++ b/packages/tracker/meshnet_tracker/cli.py @@ -8,40 +8,43 @@ from .server import TrackerServer def main() -> None: - parser = argparse.ArgumentParser( - prog="meshnet-tracker", - description="Distributed Inference Network node registry and route selection", - ) - subparsers = parser.add_subparsers(dest="command") - - start_cmd = subparsers.add_parser("start", help="Start the tracker server") - start_cmd.add_argument("--host", default="127.0.0.1", help="Host interface to listen on") - start_cmd.add_argument("--port", type=int, default=8080, help="Port to listen on") - start_cmd.add_argument( + common = argparse.ArgumentParser(add_help=False) + common.add_argument("--host", default="0.0.0.0", help="Host interface to listen on") + common.add_argument("--port", type=int, default=8080, help="Port to listen on") + common.add_argument( "--heartbeat-timeout", type=float, default=30.0, help="Seconds before a node is removed from the registry after missed heartbeat", ) - start_cmd.add_argument( + common.add_argument( "--cluster-peers", default="", help="Comma-separated URLs of peer tracker nodes (enables Raft cluster mode)", ) - start_cmd.add_argument( + common.add_argument( "--self-url", default=None, help="This tracker's own URL as seen by peers (auto-derived from --host/--port if omitted)", ) - start_cmd.add_argument( + common.add_argument( "--relay-url", default=None, help="Public ws(s):// relay URL advertised to nodes, for example wss://ai.neuron.d-popov.com/ws", ) + parser = argparse.ArgumentParser( + prog="meshnet-tracker", + description="Distributed Inference Network node registry and route selection", + parents=[common], + ) + subparsers = parser.add_subparsers(dest="command") + + subparsers.add_parser("start", help="Start the tracker server", parents=[common]) + args = parser.parse_args() - if args.command == "start": + if args.command in {None, "start"}: cluster_peers = [u.strip() for u in args.cluster_peers.split(",") if u.strip()] server = TrackerServer( host=args.host, diff --git a/tests/test_mining_cli.py b/tests/test_mining_cli.py index 211ea06..cc553c2 100644 --- a/tests/test_mining_cli.py +++ b/tests/test_mining_cli.py @@ -354,3 +354,76 @@ def test_legacy_start_subcommand_accepted(monkeypatch): # Exited (either 0 or via KeyboardInterrupt caught in _cmd_start) # The important thing is no unhandled exception from arg parsing + + +def test_legacy_start_treats_repo_model_as_model_id(monkeypatch): + """`meshnet-node start --model org/repo` enters the HF model startup path.""" + from meshnet_node.cli import main + + captured = {} + + def fake_run_startup(*args, **kwargs): + captured.update(kwargs) + class _FakeNode: + chat_completion_count = 0 + def stop(self): pass + return _FakeNode() + + monkeypatch.setattr(sys, "argv", [ + "meshnet-node", "start", + "--tracker", "http://192.168.0.179:8081", + "--model", "Qwen/Qwen2.5-0.5B-Instruct", + "--port", "0", + ]) + + with patch("meshnet_node.startup.run_startup", side_effect=fake_run_startup): + with patch("time.sleep", side_effect=KeyboardInterrupt): + try: + main() + except SystemExit as exc: + assert exc.code == 0 + + assert captured["model"] == "Qwen2.5-0.5B-Instruct" + assert captured["model_id"] == "Qwen/Qwen2.5-0.5B-Instruct" + + +def test_default_cli_passes_advertise_host(monkeypatch): + """The documented no-subcommand LAN flag reaches startup.""" + from meshnet_node.cli import main + + saved = { + "model_hf_repo": "Qwen/Qwen2.5-0.5B-Instruct", + "model_name": "Qwen2.5-0.5B-Instruct", + "quantization": "nf4", + "tracker_url": "http://localhost:8080", + "wallet_path": "", + "download_dir": "", + "port": 7000, + "host": "0.0.0.0", + } + captured = {} + + def fake_run_startup(*args, **kwargs): + captured.update(kwargs) + class _FakeNode: + chat_completion_count = 0 + def stop(self): pass + return _FakeNode() + + monkeypatch.setattr(sys, "argv", [ + "meshnet-node", + "--tracker", "http://192.168.0.179:8081", + "--advertise-host", "192.168.0.42", + "--no-tui", + ]) + + with patch("meshnet_node.config.load_config", return_value=saved): + with patch("meshnet_node.startup.run_startup", side_effect=fake_run_startup): + with patch("meshnet_node.dashboard.run_dashboard", side_effect=KeyboardInterrupt): + try: + main() + except SystemExit as exc: + assert exc.code == 0 + + assert captured["tracker_url"] == "http://192.168.0.179:8081" + assert captured["advertise_host"] == "192.168.0.42" From c587e02c2c6512f2eb64f718471fdf43aac42fd4 Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Tue, 30 Jun 2026 16:21:17 +0200 Subject: [PATCH 4/4] auto find next available local port --- packages/node/meshnet_node/cli.py | 18 +++++++++++++-- tests/test_mining_cli.py | 38 +++++++++++++++++++++++++++++++ 2 files changed, 54 insertions(+), 2 deletions(-) diff --git a/packages/node/meshnet_node/cli.py b/packages/node/meshnet_node/cli.py index a874176..cd35ca3 100644 --- a/packages/node/meshnet_node/cli.py +++ b/packages/node/meshnet_node/cli.py @@ -3,6 +3,7 @@ from __future__ import annotations import argparse +import socket import sys import time from pathlib import Path @@ -49,6 +50,19 @@ def _run_node(cfg: dict) -> None: ) +def _first_available_port(host: str, start: int = 7000, attempts: int = 100) -> int: + """Return the first TCP port bindable on host, starting at start.""" + bind_host = "" if host == "0.0.0.0" else host + for port in range(start, start + attempts): + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + try: + sock.bind((bind_host, port)) + except OSError: + continue + return port + raise OSError(f"no available TCP port in range {start}-{start + attempts - 1}") + + def _cmd_default(args) -> int: """No subcommand: wizard if no config, else start with saved config.""" from .config import load_config, save_config, merge_cli_overrides @@ -145,7 +159,7 @@ def _cmd_start(args) -> int: # Build a transient config from flags (don't write to disk) cfg = dict(DEFAULTS) cfg["tracker_url"] = args.tracker - cfg["port"] = args.port + cfg["port"] = args.port if args.port is not None else _first_available_port(args.host) if args.model_id is None and "/" in args.model: cfg["model_hf_repo"] = args.model cfg["model_name"] = args.model.split("/")[-1] @@ -236,7 +250,7 @@ def main() -> None: # start subcommand (legacy / backward-compat) start_cmd = subparsers.add_parser("start", help="Start node (legacy flags)") start_cmd.add_argument("--tracker", default="http://localhost:8080") - start_cmd.add_argument("--port", type=int, default=7000) + start_cmd.add_argument("--port", type=int) start_cmd.add_argument("--model", default="stub-model") start_cmd.add_argument("--model-id", help="HuggingFace repo ID") start_cmd.add_argument("--shard-start", type=int) diff --git a/tests/test_mining_cli.py b/tests/test_mining_cli.py index cc553c2..e26331f 100644 --- a/tests/test_mining_cli.py +++ b/tests/test_mining_cli.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import socket import sys import types from pathlib import Path @@ -387,6 +388,43 @@ def test_legacy_start_treats_repo_model_as_model_id(monkeypatch): assert captured["model_id"] == "Qwen/Qwen2.5-0.5B-Instruct" +def test_legacy_start_without_port_uses_next_available_port(monkeypatch): + """Omitting --port skips an occupied default port before startup loads the model.""" + from meshnet_node.cli import main + + captured = {} + + def fake_run_startup(*args, **kwargs): + captured.update(kwargs) + class _FakeNode: + chat_completion_count = 0 + def stop(self): pass + return _FakeNode() + + occupied = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + occupied.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + occupied.bind(("127.0.0.1", 7000)) + occupied.listen(1) + try: + monkeypatch.setattr(sys, "argv", [ + "meshnet-node", "start", + "--tracker", "http://192.168.0.179:8081", + "--model", "Qwen/Qwen2.5-0.5B-Instruct", + "--host", "127.0.0.1", + ]) + + with patch("meshnet_node.startup.run_startup", side_effect=fake_run_startup): + with patch("time.sleep", side_effect=KeyboardInterrupt): + try: + main() + except SystemExit as exc: + assert exc.code == 0 + finally: + occupied.close() + + assert captured["port"] == 7001 + + def test_default_cli_passes_advertise_host(monkeypatch): """The documented no-subcommand LAN flag reaches startup.""" from meshnet_node.cli import main