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

feat(us-023): heartbeat stats payload, request counters, dynamic reassignment response

Browse Source 
Node now sends cumulative stats in heartbeat body:
  total_requests, failed_requests, queue_depth, uptime_seconds, status
Stats are tracked thread-safely in _TorchHTTPServer; buffered locally during
outage streak and flushed on next successful heartbeat.

Tracker stores stats on _NodeEntry (total_requests, failed_requests,
queue_depth, uptime_seconds, status, heartbeats_expected/received) and
returns new_assignment in heartbeat response when pending_new_assignment is set.

Node logs received new_assignment (hot-reload wired in US-026).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Dobromir Popov
2026-06-30 13:20:52 +03:00
parent 96af82892f
commit 34fb1ec65d
3 changed files with 102 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -39,8 +39,29 @@ def _start_heartbeat(
node_id: str,
register_payload: dict,
interval: float = 20.0,
node_ref: Any | None = None,
start_time: float | None = None,
) -> threading.Thread:
"""Daemon thread: sends heartbeats and re-registers automatically after tracker restarts."""
"""Daemon thread: sends heartbeats and re-registers automatically after tracker restarts.
Heartbeat body carries cumulative stats (total_requests, failed_requests,
queue_depth, uptime_seconds, status). Stats are buffered locally during
outage and flushed on next successful heartbeat.
Heartbeat response may include new_assignment: {model, shard_start, shard_end}
which is logged for now (hot-reload implemented in US-026).
"""
_start_time = start_time or time.monotonic()
def _get_stats() -> dict:
uptime = time.monotonic() - _start_time
stats: dict = {"uptime_seconds": round(uptime, 1), "status": "ready"}
if node_ref is not None:
stats["total_requests"] = getattr(node_ref, "total_requests", 0)
stats["failed_requests"] = getattr(node_ref, "failed_requests", 0)
stats["queue_depth"] = getattr(node_ref, "queue_depth", 0)
return stats
def _reregister() -> bool:
nonlocal node_id
try:
@@ -76,7 +97,15 @@ def _start_heartbeat(
continue
try:
_post_json(hb_url, {})
resp = _post_json(hb_url, _get_stats())
new_asgn = resp.get("new_assignment")
if new_asgn:
print(
f" [node] tracker reassignment received: "
f"model={new_asgn.get('model')!r} "
f"shards={new_asgn.get('shard_start')}-{new_asgn.get('shard_end')}",
flush=True,
)
except urllib.error.HTTPError as exc:
if exc.code == 404:
# Node was purged (e.g. long gap before restart noticed) — re-register now.
@@ -205,6 +234,7 @@ def run_startup(
tracker_url=tracker_url,
route_timeout=route_timeout,
)
_node_start_time = time.monotonic()
actual_port = node.start()
total_layers = getattr(getattr(node, "backend", None), "total_layers", None)
if isinstance(total_layers, int) and total_layers > 0:
@@ -235,7 +265,7 @@ def run_startup(
tracker_node_id = str(reg_resp.get("node_id") or "?")
setattr(node, "tracker_node_id", tracker_node_id)
print(f" Registered with tracker — node ID: {tracker_node_id}", flush=True)
_start_heartbeat(tracker_url, tracker_node_id, reg_payload)
_start_heartbeat(tracker_url, tracker_node_id, reg_payload, node_ref=node, start_time=_node_start_time)
except Exception as exc:
setattr(node, "tracker_node_id", None)
print(f" Warning: tracker registration failed: {exc}", flush=True)
@@ -289,6 +319,7 @@ def run_startup(
tracker_url=tracker_url,
route_timeout=route_timeout,
)
_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}"
@@ -311,7 +342,7 @@ def run_startup(
tracker_node_id = str(reg_resp.get("node_id") or "?")
setattr(node, "tracker_node_id", tracker_node_id)
print(f" Registered with tracker — node ID: {tracker_node_id}", flush=True)
_start_heartbeat(tracker_url, tracker_node_id, auto_reg_payload)
_start_heartbeat(tracker_url, tracker_node_id, auto_reg_payload, node_ref=node, start_time=_node_start_time)
except Exception as exc:
setattr(node, "tracker_node_id", None)
print(f" Warning: tracker registration failed: {exc}", flush=True)