diff --git a/packages/node/meshnet_node/torch_server.py b/packages/node/meshnet_node/torch_server.py index e58cf6d..7bbabac 100644 --- a/packages/node/meshnet_node/torch_server.py +++ b/packages/node/meshnet_node/torch_server.py @@ -1545,6 +1545,26 @@ class TorchNodeServer: def apply_tracker_directives(self, directives: list[dict]) -> dict | None: """Apply tracker shard directives (LOAD_SHARD replace, ADD_SHARD load-more).""" + drop_directive = next( + (directive for directive in reversed(directives) if directive.get("action") == "DROP_SHARD"), + None, + ) + if drop_directive is not None: + model_id = str(drop_directive.get("model") or "") + removed = self._backends.pop(model_id, None) + if removed is None: + return None + if self._backends: + self._backend = next(iter(self._backends.values())) + self._tracker_mode = self._backend.shard_start == 0 + else: + self._backend = None + self._tracker_mode = False + if self._server is not None: + self._server.backends = dict(self._backends) + self._server.backend = self._backend + self._server.tracker_mode = self._tracker_mode + return {"action": "DROP_SHARD", "model": model_id} add_directive = next( (directive for directive in reversed(directives) if directive.get("action") == "ADD_SHARD"), None, @@ -1574,6 +1594,8 @@ class TorchNodeServer: flush=True, ) try: + if replacing: + self._backends.clear() new_backend = _load_backend(model_id, shard_start, shard_end, quantization, self._cache_dir) except TypeError: new_backend = _load_backend(model_id, shard_start, shard_end, quantization) diff --git a/packages/tracker/meshnet_tracker/dashboard.html b/packages/tracker/meshnet_tracker/dashboard.html index 03fd932..b3b0c79 100644 --- a/packages/tracker/meshnet_tracker/dashboard.html +++ b/packages/tracker/meshnet_tracker/dashboard.html @@ -288,6 +288,7 @@

Node pending payouts

admin login required

Settlement history

admin login required

Tracker hive

loading…
+

Model placement

admin login required

All accounts (admin)

Strikes / bans / forfeitures

admin login required

Client balances

admin login required
@@ -1791,6 +1792,34 @@ async function requestSelectedModelLoad() { $("chat-status").textContent = `load queued on ${short(assignment.node_id || "node")} for layers ${assignment.shard_start}-${assignment.shard_end}`; } +async function requestAdminModelLoad(model) { + const result = await apiCall("/v1/models/load", "POST", { model, force: true }); + if (!result.ok) return alert(result.data.error || "model load request failed"); + await refreshActiveTab(true); +} + +async function releaseAdminModel(model) { + if (!confirm(`Release ${model} from every serving node?`)) return; + const result = await apiCall("/v1/models/release", "POST", { model }); + if (!result.ok) return alert(result.data.error || "model release request failed"); + await refreshActiveTab(true); +} + +function renderAdminModelPlacement(models, map) { + const nodes = (map && map.nodes) || []; + const rows = ((models && models.data) || []).map(model => { + const aliases = new Set([model.id, model.hf_repo, ...(model.aliases || [])].filter(Boolean)); + const serving = nodes.filter(node => aliases.has(node.model) || aliases.has(node.hf_repo)).length; + const downloaded = nodes.filter(node => (node.downloaded_models || []).some(item => aliases.has(item.model) || aliases.has(item.hf_repo))).length; + const actions = ` ` + + ``; + return [esc(model.name || model.id), String(serving), String(downloaded), actions]; + }); + $("admin-model-placement").innerHTML = rows.length + ? table(["model", "serving nodes", "downloaded on nodes", "admin action"], rows) + : '
no model presets configured
'; +} + function chatAuthToken() { if (accountApiKeys.length) return accountApiKeys[0]; return null; @@ -2427,14 +2456,17 @@ async function fetchAdminTab() { fetchJson("/v1/console"), fetchJson("/v1/billing/summary"), fetchJson("/v1/registry/wallets"), + fetchJson("/v1/models"), + fetchJson("/v1/network/map"), ]; if (isAdmin) fetches.push(apiCall("/v1/admin/accounts")); const results = await Promise.all(fetches); - const [raft, consoleData, summary, wallets, adminResp] = results; + const [raft, consoleData, summary, wallets, models, map, adminResp] = results; renderIfChanged("hive", raft, renderHive); renderIfChanged("console", consoleData, renderConsole); renderIfChanged("billing-summary", summary, data => renderBilling(data)); renderIfChanged("fraud", { wallets, summary }, data => renderFraud(data.wallets, data.summary)); + renderIfChanged("admin-model-placement", { models, map }, data => renderAdminModelPlacement(data.models, data.map)); if (adminResp && adminResp.ok) { renderIfChanged("admin", adminResp.data.accounts || [], accounts => { const rows = accounts.map(a => { diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 0a428ef..32c7239 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -1528,6 +1528,49 @@ def _request_model_load_locked(server: "_TrackerHTTPServer", model_key: str) -> return None +def _force_model_load_locked(server: "_TrackerHTTPServer", model_key: str) -> dict | None: + """Replace the fastest ready assignment after an explicit admin eviction.""" + resolved_name, preset = _resolve_model_preset(server.model_presets, model_key) + if preset is None or not preset.get("hf_repo"): + return None + start, end = _preset_layer_bounds(preset) + candidates = [node for node in server.registry.values() + if node.status == "ready" and node.pending_new_assignment is None + and _has_usable_quantization(node)] + if not candidates: + return None + node = max(candidates, key=lambda item: item.benchmark_tokens_per_sec) + shard_end = min(end, start + min(_node_layer_capacity(node, preset), end - start + 1) - 1) + if shard_end < start: + return None + quantization = _node_quantization(node, preset) + directive = _load_directive(node, str(preset["hf_repo"]), start, shard_end, quantization) + replaced = node.hf_repo or node.model + node.model, node.hf_repo = resolved_name, str(preset["hf_repo"]) + node.shard_start, node.shard_end, node.quantization = start, shard_end, quantization + node.managed_assignment, node.pending_new_assignment = True, directive + node.pending_directives.append(directive) + _tracker_log(server, "warn", "model load forced", node_id=node.node_id, + model=resolved_name, replaced_model=replaced, shard=f"{start}-{shard_end}") + return {"node_id": node.node_id, "model": resolved_name, "hf_repo": preset["hf_repo"], + "shard_start": start, "shard_end": shard_end, "replaced_model": replaced} + + +def _release_model_locked(server: "_TrackerHTTPServer", model_key: str) -> int: + """Queue DROP_SHARD for every served shard and remove it from routing immediately.""" + resolved_name, preset = _resolve_model_preset(server.model_presets, model_key) + if preset is None: + return 0 + released = 0 + for node in server.registry.values(): + if not _node_matches_preset(node, resolved_name, preset) or node.shard_start is None or node.shard_end is None: + continue + node.pending_directives.append(_drop_directive(node, str(preset.get("hf_repo") or resolved_name), node.shard_start, node.shard_end, node.quantization or "bfloat16")) + node.status = "loading" + released += 1 + return released + + def _preferred_node_quantization( node: _NodeEntry, preset: dict, @@ -3043,6 +3086,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): if self.path == "/v1/models/load": self._handle_model_load_request() return + if self.path == "/v1/models/release": + self._handle_model_release_request() + return if self.path == "/v1/models/vote": self._handle_model_coverage_vote() return @@ -3170,8 +3216,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): seen_ids: set[str] = set() for name, preset in server.model_presets.items(): model_nodes = [node for node in alive if _node_matches_preset(node, name, preset)] - if not model_nodes and not preset.get("recommended"): - continue required_start, required_end = _preset_layer_bounds(preset) coverage = _coverage_percentage( model_nodes, @@ -4834,11 +4878,32 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): with server.lock: self._purge_expired_nodes() assignment = _request_model_load_locked(server, model) + if assignment is None and body.get("force") is True: + assignment = _force_model_load_locked(server, model) if assignment is None: self._send_json(409, {"error": "no ready joined node has an available model slot and sufficient capacity"}) return self._send_json(202, {"status": "queued", "assignment": assignment}) + def _handle_model_release_request(self): + server: _TrackerHTTPServer = self.server # type: ignore[assignment] + if not self._require_role("admin", "validator"): + return + body = self._read_json_body() + if body is None: + return + model = body.get("model") + if not isinstance(model, str) or not model.strip(): + self._send_json(400, {"error": "model is required"}) + return + with server.lock: + self._purge_expired_nodes() + released = _release_model_locked(server, model) + if not released: + self._send_json(404, {"error": "no served shards found for model"}) + return + self._send_json(202, {"status": "release_queued", "released": released}) + def _handle_model_coverage_vote(self): """Record a rolling wish-list signal for an unavailable precision.""" server: _TrackerHTTPServer = self.server # type: ignore[assignment] diff --git a/tests/test_dashboard.py b/tests/test_dashboard.py index dbe7283..0f2a545 100644 --- a/tests/test_dashboard.py +++ b/tests/test_dashboard.py @@ -100,6 +100,17 @@ def test_dashboard_allows_admin_to_request_selected_model_load(): assert '$("request-model-load").style.display = enabled ? "" : "none"' in html +def test_dashboard_exposes_admin_model_inventory_and_release_controls(): + "Admin placement controls show the full model inventory and can release capacity." + html = _dashboard_html() + + assert 'id="admin-model-placement"' in html + assert "renderAdminModelPlacement" in html + assert '"/v1/models/release"' in html + assert "requestAdminModelLoad" in html + assert "releaseAdminModel" in html + + def test_network_map_includes_node_friendly_name(): "Network map includes node friendly name\n\nTags: dashboard, http" tracker = TrackerServer() diff --git a/tests/test_dynamic_routing.py b/tests/test_dynamic_routing.py index 83a9d2f..85c7ad6 100644 --- a/tests/test_dynamic_routing.py +++ b/tests/test_dynamic_routing.py @@ -355,6 +355,39 @@ def test_admin_model_load_request_queues_directive_on_joined_node(): assert heartbeat["directives"][0]["model"] == "Qwen/Qwen2.5-0.5B-Instruct" +def test_admin_can_replace_a_served_model_and_release_it(): + "Forced admin placement replaces a served shard; release queues DROP_SHARD." + tracker = TrackerServer(enable_billing=False, validator_service_token="test-admin") + port = tracker.start() + try: + node = _post_json( + f"http://127.0.0.1:{port}/v1/nodes/register", + {"endpoint": "http://127.0.0.1:9912", "model": "stub-model", + "shard_start": 0, "shard_end": 3, "managed_assignment": True, + "max_loaded_shards": 1, "memory_mb": 1, + "hardware_profile": {"host_id": "full-host"}}, + ) + headers = {"Content-Type": "application/json", "Authorization": "Bearer test-admin"} + load = urllib.request.Request( + f"http://127.0.0.1:{port}/v1/models/load", + data=json.dumps({"model": "qwen2.5-0.5b-instruct", "force": True}).encode(), + headers=headers, method="POST") + with urllib.request.urlopen(load) as response: + assert json.loads(response.read())["assignment"]["node_id"] == node["node_id"] + heartbeat = _post_json(f"http://127.0.0.1:{port}/v1/nodes/{node['node_id']}/heartbeat", {}) + assert heartbeat["directives"][0]["action"] == "LOAD_SHARD" + release = urllib.request.Request( + f"http://127.0.0.1:{port}/v1/models/release", + data=json.dumps({"model": "qwen2.5-0.5b-instruct"}).encode(), headers=headers, method="POST") + with urllib.request.urlopen(release) as response: + assert json.loads(response.read())["released"] == 1 + heartbeat = _post_json(f"http://127.0.0.1:{port}/v1/nodes/{node['node_id']}/heartbeat", {}) + finally: + tracker.stop() + + assert heartbeat["directives"][0]["action"] == "DROP_SHARD" + + def test_endpoint_key_distinguishes_same_port_different_hosts(): "Endpoint key distinguishes same port different hosts\n\nTags: http, performance, routing, tracker" from meshnet_node.torch_server import _clamp_downstream_hops, _endpoint_key