diff --git a/packages/node/meshnet_node/startup.py b/packages/node/meshnet_node/startup.py index 91ecda2..4b841ce 100644 --- a/packages/node/meshnet_node/startup.py +++ b/packages/node/meshnet_node/startup.py @@ -5,6 +5,8 @@ from __future__ import annotations import json import socket import sys +import threading +import time import urllib.error import urllib.parse import urllib.request @@ -32,6 +34,23 @@ def _get_json(url: str, timeout: float = 10.0) -> dict: return json.loads(r.read()) +def _start_heartbeat(tracker_url: str, node_id: str, interval: float = 30.0) -> threading.Thread: + """Daemon thread that sends periodic heartbeats to the tracker.""" + def _loop() -> None: + hb_url = f"{tracker_url}/v1/nodes/{node_id}/heartbeat" + while True: + time.sleep(interval) + try: + _post_json(hb_url, {}) + print(f" [node] heartbeat sent → tracker (node {node_id[:8]})", flush=True) + except Exception as exc: + print(f" [node] WARNING: heartbeat failed: {exc}", flush=True) + + t = threading.Thread(target=_loop, daemon=True, name="heartbeat") + t.start() + return t + + def run_startup( tracker_url: str, port: int = 0, @@ -119,7 +138,7 @@ def run_startup( # Register with tracker so other nodes can auto-join this model. total_layers = getattr(node.backend, "total_layers", None) try: - _post_json( + reg_resp = _post_json( f"{tracker_url}/v1/nodes/register", { "endpoint": endpoint, @@ -135,6 +154,9 @@ def run_startup( "tracker_mode": (shard_start == 0), }, ) + node_id = reg_resp.get("node_id", "?") + print(f" Registered with tracker — node ID: {node_id}", flush=True) + _start_heartbeat(tracker_url, node_id) except Exception as exc: print(f" Warning: tracker registration failed: {exc}", flush=True) @@ -187,7 +209,7 @@ def run_startup( public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host) endpoint = f"http://{public_host}:{actual_port}" try: - _post_json( + reg_resp = _post_json( f"{tracker_url}/v1/nodes/register", { "endpoint": endpoint, @@ -203,6 +225,9 @@ def run_startup( "tracker_mode": (assigned_shard_start == 0), }, ) + node_id = reg_resp.get("node_id", "?") + print(f" Registered with tracker — node ID: {node_id}", flush=True) + _start_heartbeat(tracker_url, node_id) except Exception as exc: print(f" Warning: tracker registration failed: {exc}", flush=True) shard_count = assigned_shard_end - assigned_shard_start + 1 diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index cd42d1d..e43687b 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -656,6 +656,13 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): if assignment_directive is not None: entry.pending_directives.clear() + 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 "" + print( + f"[tracker] node registered: {node_id[:8]} {endpoint} {model}{repo_info} {shard_info}", + flush=True, + ) + payload = {"node_id": node_id} if assignment_directive is not None: payload["directive"] = assignment_directive @@ -673,6 +680,10 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): _rebalance_model_locked(server, entry.model or "stub-model") directives = list(entry.pending_directives) entry.pending_directives.clear() + print( + f"[tracker] heartbeat: {node_id[:8]} {entry.endpoint}", + flush=True, + ) if directives: self._send_json(200, {"directives": directives}) else: