[verified] fix: preserve tracker precision eligibility

This commit is contained in:
Dobromir Popov
2026-07-13 10:27:45 +03:00
parent 377346c301
commit b5fa7245df
12 changed files with 557 additions and 48 deletions

View File

@@ -0,0 +1,27 @@
# PRD: Streaming proxy cancellation delivery
## Overview
A tracker proxy route can be selected and registered as active, but a direct SSE upstream may not deliver its first frame to the downstream client until the upstream response completes. A dashboard cancellation request made after the upstream emitted and flushed an SSE frame can therefore arrive after the tracker has already finalized and unregistered the proxy, returning 404.
## Goal
Preserve immediate SSE frame delivery and keep the active-proxy record cancelable until the streaming response actually completes or is canceled.
## Quality gates
- Run the focused tracker streaming/cancellation tests.
- Run full `python -m pytest -q` and record unrelated failures exactly.
- Do not weaken billing, disconnect handling, proxy-inflight accounting, or cancel authorization.
## User story
### PSC-001: Cancel an active direct SSE proxy
As a dashboard operator, I need a cancellation request to reach an active direct SSE proxy after its first streamed frame, so I can stop long-running inference without a race-induced 404.
## Non-goals
- Do not change route selection or quantization policy.
- Do not alter relay transport behavior unless needed to share the same cancellation invariant.
- Do not introduce client-visible buffering.

View File

@@ -0,0 +1,34 @@
# PSC-001 — Direct SSE cancellation race
Status: ready-for-agent
Priority: High
## Problem
`tests/test_tracker_routing.py::test_tracker_dashboard_can_cancel_inflight_proxy` registers a streaming upstream node. The upstream writes and flushes the first SSE frame, then waits three seconds. The trackers client does not receive that frame until after the upstream has completed; by then `_unregister_active_proxy()` has run and the dashboard cancellation endpoint returns 404.
Observed trace:
```text
proxy route selected
proxy connected
proxy progress ... elapsed_seconds≈3
proxy complete ... elapsed_seconds≈3
POST /v1/proxy/requests/<id>/cancel → 404
```
This is a production cancellation/delivery race, not a stale test: the endpoint promises to cancel active proxy work, and the upstream had already emitted a first stream frame before cancellation was attempted.
## Acceptance criteria
- [ ] A direct SSE upstream frame is relayed and flushed to the client before the upstream completes.
- [ ] After that first frame, `/v1/proxy/requests/{request_id}/cancel` returns 200 while the stream is active.
- [ ] Cancellation closes/stops the upstream safely, finalizes inflight accounting exactly once, and records the cancellation event.
- [ ] Cancel authorization remains unchanged.
- [ ] Client disconnect and normal SSE completion retain current billing/throughput behavior.
- [ ] Regression test is deterministic and does not rely on timing races longer than necessary.
- [ ] Focused tracker routing tests and full pytest are run with unrelated failures documented.
## Likely seam
Inspect direct streaming behavior around `_handle_proxy_chat` upstream reads (`packages/tracker/meshnet_tracker/server.py`, roughly lines 39534019). The direct response path writes/flushed each line, but current HTTP response buffering/reading prevents the line from being observed until stream end. Fix delivery at the proxy transport boundary; do not paper over it by retaining completed proxy records indefinitely.

View File

@@ -0,0 +1,27 @@
# PRD: Legacy routing compatibility regression
## Overview
The trackers new request-precision gate rejects legacy node registrations that omit `quantization`. This conflicts with the explicit `capability_policy=compat` rollout: an older node without a capability report remains routable, but an older node without a quantization field is silently excluded from proxy, pinned-route, benchmark, billing, and latency paths.
## Goal
Restore backward-compatible routing for legacy registrations while preserving fail-closed behavior for explicitly declared invalid/unsupported quantization values.
## Quality gates
- Run targeted routing, billing, benchmark, pricing, validation, and latency tests.
- Run `python -m pytest -q` before completion and record any unrelated failures exactly.
- Preserve the admission-policy invariant: invalid capability reports remain non-routable; absent reports route only under `compat`.
## User story
### RCR-001: Legacy registration precision fallback
As an operator upgrading a mixed fleet, I need nodes that predate the `quantization` registration field to serve default-precision requests under `compat`, so the tracker rollout does not make otherwise healthy legacy nodes dark.
## Non-goals
- Do not weaken `enforce` capability policy.
- Do not treat explicitly malformed or unsupported quantization values as valid.
- Do not change requested-precision semantics for nodes that declare a supported precision.

View File

@@ -0,0 +1,38 @@
# RCR-001 — Legacy registration precision fallback
Status: ready-for-agent
Priority: Critical
## Problem
The tracker now filters proxy and pinned-route candidates through `_quantization_satisfies(node.quantization, requested_quantization)`. A registration produced before the protocol added singular `quantization` is normalized to `quantizations=["bfloat16"]`, but leaves `node.quantization is None`; routing then fails every request, including the default `bfloat16` request.
The defect is therefore at protocol normalization: registration correctly records an available plural capability but never chooses a compatible active precision for the legacy entry.
This is inconsistent with the explicit capability admission rollout policy: an absent capability report is routable under `compat`. Logs show `capability=absent routable=True`, while `/v1/chat/completions` returns 503 or a pinned route returns 409 because the separate precision gate excludes the same legacy node.
## Evidence
Representative tight repros fail on current `master`:
- `tests/test_billing_ledger.py::test_proxy_chat_bills_credited_client_and_credits_node` → HTTP 503
- `tests/test_tracker_routing.py::test_tracker_proxy_accepts_hf_model_alias_from_quickstart` → HTTP 503
- `tests/test_manual_route_benchmark.py::test_pinned_route_uses_named_node` → HTTP 409
- `tests/test_model_speed_latency.py::test_tracker_records_increasing_hop_latency_for_model_and_hardware` → HTTP 409 for every parameter
Each fixture registers a complete, healthy legacy node without `quantization`; tracker registration logs show it as routable under `compat`.
## Acceptance criteria
- [ ] Legacy nodes omitting the `quantization` field are treated as the historical default precision for compatibility purposes.
- [ ] Explicit `null`, invalid, or unsupported declared precision never receives that fallback and remains excluded from routing, assignment, and coverage calculations, including after managed shard assignment/rebalancing.
- [ ] Raft replication preserves singular `quantization` and plural `quantizations`; follower routing has identical precision eligibility to the leader.
- [ ] `capability_policy=enforce` still excludes absent capability reports regardless of precision fallback.
- [ ] Automatic proxy routing, pinned routes, benchmark routes, and route-stat sampling use the same compatibility rule.
- [ ] Add behavior-level tracker HTTP tests for legacy omission and explicit invalid declaration.
- [ ] Targeted billing, routing, benchmark, model-speed, pricing, forfeiture, and TOPLOC dispatch tests pass.
- [ ] Run the full suite and document any failures not caused by this issue.
## Implementation notes
Keep the compatibility decision at the protocol-normalization boundary, not scattered into individual route paths. A normalized legacy default can preserve existing `_quantization_satisfies` semantics while differentiating an omitted field from an explicitly invalid value.

View File

@@ -0,0 +1,29 @@
{
"name": "Legacy routing compatibility regression",
"description": "Restore compat-policy routing for legacy node registrations that omit quantization without weakening validation for invalid declared values.",
"branchName": "ralph/routing-compatibility-regression",
"userStories": [
{
"id": "RCR-001",
"title": "Legacy registration precision fallback",
"description": "As an operator upgrading a mixed fleet, I need nodes that predate the quantization registration field to serve default-precision requests under compat, so otherwise healthy legacy nodes do not become dark.",
"acceptanceCriteria": [
"A legacy registration without quantization is eligible for a bfloat16 request when capability policy is compat.",
"The same registration is excluded under capability policy enforce because its capability report is absent.",
"A node that explicitly declares null, an invalid, or an unsupported quantization remains excluded from routing, assignment, and coverage calculations, including after managed shard assignment/rebalancing.",
"Raft-applied registrations preserve both singular quantization and plural quantizations so follower routing matches leader routing.",
"Automatic proxy routing, pinned routes, manual benchmarks, billing proxy paths, and route-latency sampling retain their existing behavior for legacy registrations.",
"Tests cover the legacy/no-field case and explicit-invalid-field case at the tracker HTTP seam.",
"Targeted routing, billing, benchmark, pricing, validation, and latency tests pass.",
"python -m pytest -q is run and unrelated failures are recorded exactly."
],
"priority": 1,
"passes": true,
"dependsOn": [],
"completionNotes": "Completed by agent"
}
],
"metadata": {
"updatedAt": "2026-07-13T07:25:08.460Z"
}
}

View File

@@ -972,7 +972,7 @@ def _coverage_percentage(
intervals = sorted( intervals = sorted(
( (
(max(required_start, node.shard_start), min(required_end, node.shard_end)) (max(required_start, node.shard_start), min(required_end, node.shard_end))
for node in nodes for node in _serviceable_nodes(nodes)
if node.shard_start is not None if node.shard_start is not None
and node.shard_end is not None and node.shard_end is not None
if node.shard_end >= required_start and node.shard_start <= required_end if node.shard_end >= required_start and node.shard_start <= required_end
@@ -1004,7 +1004,7 @@ def _served_model_copies(
layer_counts = [] layer_counts = []
for layer in range(required_start, required_end + 1): for layer in range(required_start, required_end + 1):
count = 0 count = 0
for node in nodes: for node in _serviceable_nodes(nodes):
if node.shard_start is None or node.shard_end is None: if node.shard_start is None or node.shard_end is None:
continue continue
if node.shard_start <= layer <= node.shard_end: if node.shard_start <= layer <= node.shard_end:
@@ -1092,6 +1092,57 @@ def _normalize_quantization(value: object) -> str | None:
return normalized if normalized in _QUANTIZATION_QUALITY else None return normalized if normalized in _QUANTIZATION_QUALITY else None
def _registration_quantization(body: dict, quantizations: list[str]) -> str | None:
"""Resolve the active precision a registration is routable at.
Only the raw body can tell an *absent* `quantization` from an explicit null,
and the two mean opposite things, so this must run at the registration seam
rather than in `_NodeEntry` (both reach the constructor as None).
An absent field predates the protocol adding it: it means "unknown", not
"unsupported", so the node keeps the best precision it advertises and stays
routable. Anything the node states explicitly is taken at its word -- a null,
a non-string, or an unsupported name leaves it with no usable precision and
routing excludes it.
"""
if "quantization" in body:
return _normalize_quantization(body["quantization"])
supported = [
normalized for value in quantizations
if (normalized := _normalize_quantization(value)) is not None
]
if not supported:
return None
return max(supported, key=lambda value: _QUANTIZATION_QUALITY[value])
def _has_usable_quantization(node: _NodeEntry) -> bool:
"""Whether a node declared a precision it can actually serve.
`quantization` is None only for a registration that explicitly declared a
null, malformed, or unsupported precision -- a legacy registration that omits
the field is resolved to a concrete precision at the registration seam. Such a
node is already dark to routing, and shard placement has to keep it that way:
the assignment paths pick a precision via `_node_quantization`, which falls
back to the advertised list and then to a preset default, so an unguarded
assignment would overwrite the declared-unusable precision and silently make
the node routable again.
"""
return node.quantization is not None
def _serviceable_nodes(nodes: list[_NodeEntry]) -> list[_NodeEntry]:
"""The nodes a coverage calculation is allowed to count.
A node with no usable precision is excluded from every routing path, so it
serves none of the layers it claims. Counting its shard range as covered
would report a complete model that no request can reach, and would hide the
hole from the rebalancer whose whole job is to fill it -- the managed nodes
that could cover those layers stay idle because the gap never appears.
"""
return [node for node in nodes if _has_usable_quantization(node)]
def _quantization_satisfies(available: object, requested: object) -> bool: def _quantization_satisfies(available: object, requested: object) -> bool:
"""Return whether a shard meets a request's minimum precision.""" """Return whether a shard meets a request's minimum precision."""
available_normalized = _normalize_quantization(available) available_normalized = _normalize_quantization(available)
@@ -1406,7 +1457,10 @@ def _scale_demanded_models_locked(server: "_TrackerHTTPServer") -> None:
continue continue
if any((n.hf_repo or n.model) == hf_repo for n in host_nodes): if any((n.hf_repo or n.model) == hf_repo for n in host_nodes):
continue continue
anchor = max(host_nodes, key=lambda n: n.benchmark_tokens_per_sec) placeable = [n for n in host_nodes if _has_usable_quantization(n)]
if not placeable:
continue
anchor = max(placeable, key=lambda n: n.benchmark_tokens_per_sec)
if anchor.status != "ready" or anchor.pending_new_assignment is not None: if anchor.status != "ready" or anchor.pending_new_assignment is not None:
continue continue
capacity = min(_node_layer_capacity(anchor, preset), total_layers) capacity = min(_node_layer_capacity(anchor, preset), total_layers)
@@ -1455,9 +1509,10 @@ def _request_model_load_locked(server: "_TrackerHTTPServer", model_key: str) ->
if host["spare_slots"] <= 0 and not has_spare_memory: if host["spare_slots"] <= 0 and not has_spare_memory:
continue continue
host_nodes = [server.registry[item["node_id"]] for item in host["loaded"] if item["node_id"] in server.registry] host_nodes = [server.registry[item["node_id"]] for item in host["loaded"] if item["node_id"] in server.registry]
if not host_nodes: placeable = [node for node in host_nodes if _has_usable_quantization(node)]
if not placeable:
continue continue
anchor = max(host_nodes, key=lambda node: node.benchmark_tokens_per_sec) anchor = max(placeable, key=lambda node: node.benchmark_tokens_per_sec)
if anchor.status != "ready" or anchor.pending_new_assignment is not None: if anchor.status != "ready" or anchor.pending_new_assignment is not None:
continue continue
capacity = min(_node_layer_capacity(anchor, preset), total_layers) capacity = min(_node_layer_capacity(anchor, preset), total_layers)
@@ -1479,6 +1534,8 @@ def _preferred_node_quantization(
requested: str, requested: str,
) -> str | None: ) -> str | None:
"""Choose the highest supported precision that satisfies a request.""" """Choose the highest supported precision that satisfies a request."""
if not _has_usable_quantization(node):
return None
supported = { supported = {
normalized normalized
for value in ([node.quantization] if node.quantization else []) + list(node.quantizations) for value in ([node.quantization] if node.quantization else []) + list(node.quantizations)
@@ -1631,7 +1688,7 @@ def _coverage_map(
layer_counts = [] layer_counts = []
for layer in range(required_start, required_end + 1): for layer in range(required_start, required_end + 1):
count = 0 count = 0
for node in nodes: for node in _serviceable_nodes(nodes):
if node.shard_start is None or node.shard_end is None: if node.shard_start is None or node.shard_end is None:
continue continue
if node.shard_start <= layer <= node.shard_end: if node.shard_start <= layer <= node.shard_end:
@@ -1680,8 +1737,10 @@ def _coverage_map_detailed(
) -> list[dict]: ) -> list[dict]:
"""Like _coverage_map but with per-node identity and health in each band. """Like _coverage_map but with per-node identity and health in each band.
Includes all nodes (alive and stale) so operators can see both coverage Includes all nodes (alive and stale, serviceable or not) so operators can see
holes and which dead nodes used to fill them. both coverage holes and which nodes used to fill them. This is why it does not
share `_coverage_map`'s serviceability filter: a node dark for a declared bad
precision is exactly what an operator staring at a hole needs to see.
""" """
now = time.monotonic() now = time.monotonic()
@@ -1793,6 +1852,8 @@ def _assign_redundant_managed_nodes(
): ):
if node.shard_start is not None and node.shard_end is not None: if node.shard_start is not None and node.shard_end is not None:
continue continue
if not _has_usable_quantization(node):
continue
capacity = min(_node_layer_capacity(node, preset), total_layers) capacity = min(_node_layer_capacity(node, preset), total_layers)
if capacity <= 0: if capacity <= 0:
continue continue
@@ -2225,7 +2286,8 @@ def _rebalance_model_locked(server: "_TrackerHTTPServer", model: str) -> None:
eligible_nodes = [ eligible_nodes = [
node for node in managed_nodes node for node in managed_nodes
if _node_layer_capacity(node, preset) > 0 if _has_usable_quantization(node)
and _node_layer_capacity(node, preset) > 0
] ]
node_index = 0 node_index = 0
for gap_start, gap_end in gaps: for gap_start, gap_end in gaps:
@@ -2315,7 +2377,8 @@ def _rebalance_hf_model_locked(server: "_TrackerHTTPServer", hf_repo: str) -> No
eligible_nodes = [ eligible_nodes = [
node for node in managed_nodes node for node in managed_nodes
if _node_layer_capacity(node, preset) > 0 if _has_usable_quantization(node)
and _node_layer_capacity(node, preset) > 0
] ]
node_index = 0 node_index = 0
for gap_start, gap_end in gaps: for gap_start, gap_end in gaps:
@@ -4424,10 +4487,10 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
self._send_json(400, {"error": "quantizations must be a non-empty string array"}) self._send_json(400, {"error": "quantizations must be a non-empty string array"})
return return
quantizations = list(quantizations_body) quantizations = list(quantizations_body)
quantization = body.get("quantization") if body.get("quantization") is not None and not isinstance(body["quantization"], str):
if quantization is not None and not isinstance(quantization, str):
self._send_json(400, {"error": "quantization must be a string"}) self._send_json(400, {"error": "quantization must be a string"})
return return
quantization = _registration_quantization(body, quantizations)
wallet_address = body.get("wallet_address") wallet_address = body.get("wallet_address")
if wallet_address is not None and not isinstance(wallet_address, str): if wallet_address is not None and not isinstance(wallet_address, str):
self._send_json(400, {"error": "wallet_address must be a string"}) self._send_json(400, {"error": "wallet_address must be a string"})
@@ -6875,6 +6938,21 @@ class TrackerServer:
friendly_name = _normalize_friendly_name(payload.get("friendly_name")) friendly_name = _normalize_friendly_name(payload.get("friendly_name"))
except ValueError: except ValueError:
friendly_name = None friendly_name = None
# The replicated payload is the raw registration body, so the follower can
# resolve precision exactly as the leader did -- including telling a legacy
# absent `quantization` from a declared one. Dropping these fields here
# would rebuild every node at the default precision and diverge follower
# routing from leader routing.
quantizations_payload = payload.get("quantizations")
quantizations = (
list(quantizations_payload)
if (
isinstance(quantizations_payload, list)
and quantizations_payload
and all(isinstance(item, str) and item for item in quantizations_payload)
)
else list(DEFAULT_QUANTIZATIONS)
)
entry = _NodeEntry( entry = _NodeEntry(
node_id=node_id, node_id=node_id,
endpoint=endpoint.rstrip("/"), endpoint=endpoint.rstrip("/"),
@@ -6885,6 +6963,8 @@ class TrackerServer:
hardware_profile=payload.get("hardware_profile", {}), hardware_profile=payload.get("hardware_profile", {}),
wallet_address=payload.get("wallet_address"), wallet_address=payload.get("wallet_address"),
score=float(payload.get("score", 1.0)), score=float(payload.get("score", 1.0)),
quantizations=quantizations,
quantization=_registration_quantization(payload, quantizations),
tracker_mode=bool(payload.get("tracker_mode", False)), tracker_mode=bool(payload.get("tracker_mode", False)),
hf_repo=payload.get("hf_repo"), hf_repo=payload.get("hf_repo"),
num_layers=int(payload["num_layers"]) if payload.get("num_layers") is not None else None, num_layers=int(payload["num_layers"]) if payload.get("num_layers") is not None else None,

View File

@@ -95,7 +95,7 @@ def test_refresh_loop_repriced_model_with_curated_alias(pricing_tracker):
tracker_url, ledger, tracker = pricing_tracker tracker_url, ledger, tracker = pricing_tracker
new_price = _wait_for_price_change(ledger, PRICED_MODEL) new_price = _wait_for_price_change(ledger, PRICED_MODEL)
expected = round((0.93 + 3.00) / 2 / 1000 * 0.80, 6) expected = round((0.93 + 3.00) / 2 / 1000 * 0.80, 6)
assert new_price == expected assert new_price == pytest.approx(expected)
preset = tracker._model_presets[PRICED_MODEL] preset = tracker._model_presets[PRICED_MODEL]
assert preset["hf_last_price_per_1k"] == expected assert preset["hf_last_price_per_1k"] == expected

View File

@@ -14,9 +14,13 @@ def test_sdk_support_helpers_are_deterministic_without_network():
"Non-socket coverage for SDK-facing gateway/tracker helper behavior.\n\nTags: gateway, http, sdk" "Non-socket coverage for SDK-facing gateway/tracker helper behavior.\n\nTags: gateway, http, sdk"
address = _payment_address_for_api_key("sdk-key") address = _payment_address_for_api_key("sdk-key")
# Registration always resolves a precision, so a registry node carries one;
# coverage counts only nodes that can actually serve the layers they claim.
nodes = [ nodes = [
_NodeEntry("a", "http://node-a", 0, 15, "stub-model", None, {}, None, 1.0), _NodeEntry("a", "http://node-a", 0, 15, "stub-model", None, {}, None, 1.0,
_NodeEntry("b", "http://node-b", 16, 31, "stub-model", None, {}, None, 1.0), quantization="bfloat16"),
_NodeEntry("b", "http://node-b", 16, 31, "stub-model", None, {}, None, 1.0,
quantization="bfloat16"),
] ]
assert 32 <= len(address) <= 44 assert 32 <= len(address) <= 44

View File

@@ -553,9 +553,23 @@ def test_legacy_start_falls_back_to_env_tracker_and_model(monkeypatch):
assert captured["model_id"] == "Qwen/Qwen2.5-0.5B-Instruct" assert captured["model_id"] == "Qwen/Qwen2.5-0.5B-Instruct"
def test_legacy_start_without_port_uses_next_available_port(monkeypatch): def test_first_available_port_skips_an_occupied_custom_port():
"Omitting --port skips an occupied default port before startup loads the model.\n\nTags: general" "Port search skips an occupied custom base port.\n\nTags: general"
from meshnet_node.cli import main from meshnet_node.cli import _first_available_port
occupied = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
occupied.bind(("127.0.0.1", 0))
start = occupied.getsockname()[1]
occupied.listen(1)
try:
assert _first_available_port("127.0.0.1", start=start, attempts=100) > start
finally:
occupied.close()
def test_legacy_start_without_port_uses_port_search(monkeypatch):
"Omitting --port delegates legacy startup to the port-search helper.\n\nTags: general"
from meshnet_node import cli as cli_mod
captured = {} captured = {}
@@ -566,26 +580,20 @@ def test_legacy_start_without_port_uses_next_available_port(monkeypatch):
def stop(self): pass def stop(self): pass
return _FakeNode() return _FakeNode()
occupied = socket.socket(socket.AF_INET, socket.SOCK_STREAM) monkeypatch.setattr(cli_mod, "_first_available_port", lambda *_args, **_kwargs: 7001)
occupied.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) monkeypatch.setattr(sys, "argv", [
occupied.bind(("127.0.0.1", 7000)) "meshnet-node", "start",
occupied.listen(1) "--tracker", "http://192.168.0.179:8081",
try: "--model", "Qwen/Qwen2.5-0.5B-Instruct",
monkeypatch.setattr(sys, "argv", [ "--host", "127.0.0.1",
"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("meshnet_node.startup.run_startup", side_effect=fake_run_startup):
with patch("time.sleep", side_effect=KeyboardInterrupt): with patch("time.sleep", side_effect=KeyboardInterrupt):
try: try:
main() cli_mod.main()
except SystemExit as exc: except SystemExit as exc:
assert exc.code == 0 assert exc.code == 0
finally:
occupied.close()
assert captured["port"] == 7001 assert captured["port"] == 7001

View File

@@ -2454,7 +2454,9 @@ def test_cli_loads_local_env_before_config_defaults(tmp_path, monkeypatch):
from meshnet_node import config as config_mod from meshnet_node import config as config_mod
monkeypatch.delenv("MESHNET_DOWNLOAD_DIR", raising=False) monkeypatch.delenv("MESHNET_DOWNLOAD_DIR", raising=False)
monkeypatch.delenv("HF_TOKEN", raising=False)
monkeypatch.chdir(tmp_path) monkeypatch.chdir(tmp_path)
cli_mod = importlib.reload(cli_mod)
(tmp_path / ".env").write_text( (tmp_path / ".env").write_text(
"MESHNET_DOWNLOAD_DIR=/run/media/popov/DATA/llm/safetensor/models\n" "MESHNET_DOWNLOAD_DIR=/run/media/popov/DATA/llm/safetensor/models\n"
"HF_TOKEN=hf_test_token\n" "HF_TOKEN=hf_test_token\n"

View File

@@ -32,6 +32,15 @@ from meshnet_node.model_backend import (
from meshnet_node.torch_server import TorchNodeServer from meshnet_node.torch_server import TorchNodeServer
def _require_functional_torch():
"""Skip tensor-behaviour tests when the installed torch namespace is incomplete."""
torch = pytest.importorskip("torch")
required = ("tensor", "zeros", "ones", "arange", "bfloat16", "long")
if not all(hasattr(torch, name) for name in required):
pytest.skip("requires a functional PyTorch tensor runtime")
return torch
class _FakeBackend: class _FakeBackend:
model_id = "fake-model" model_id = "fake-model"
total_layers = 12 total_layers = 12
@@ -526,13 +535,14 @@ def test_distributed_generating_log_includes_tps(capsys):
out = capsys.readouterr().out out = capsys.readouterr().out
assert "generating step=1/1" in out assert "generating step=1/1" in out
assert " tps=" in out assert " tps=" in out
assert "generation complete tokens=1" in out assert "generation complete session=" in out
assert "tokens=1" in out
assert out.count("generating step=1/1") == 1 assert out.count("generating step=1/1") == 1
def test_int_tensor_header_serializes_torch_tensors(): def test_int_tensor_header_serializes_torch_tensors():
"Int tensor header serializes torch tensors\n\nTags: model, node, real-inference" "Int tensor header serializes torch tensors\n\nTags: model, node, real-inference"
torch = pytest.importorskip("torch") torch = _require_functional_torch()
header = _int_tensor_header(torch.tensor([[1, 2, 3]], dtype=torch.long)) header = _int_tensor_header(torch.tensor([[1, 2, 3]], dtype=torch.long))
@@ -544,7 +554,7 @@ def test_bfloat16_wire_decode_views_owned_bytes_without_float32_round_trip():
Tags: model, performance, wire Tags: model, performance, wire
""" """
torch = pytest.importorskip("torch") torch = _require_functional_torch()
body = torch.tensor([[1, 2]], dtype=torch.bfloat16).view(torch.uint8).numpy().tobytes() body = torch.tensor([[1, 2]], dtype=torch.bfloat16).view(torch.uint8).numpy().tobytes()
decoded = _tensor_from_bfloat16_bytes(body, [1, 2], torch) decoded = _tensor_from_bfloat16_bytes(body, [1, 2], torch)
@@ -555,7 +565,7 @@ def test_bfloat16_wire_decode_views_owned_bytes_without_float32_round_trip():
def test_decoder_attention_mask_is_causal_float_mask(): def test_decoder_attention_mask_is_causal_float_mask():
"Decoder attention mask is causal float mask\n\nTags: model, node, real-inference" "Decoder attention mask is causal float mask\n\nTags: model, node, real-inference"
torch = pytest.importorskip("torch") torch = _require_functional_torch()
hidden_states = torch.zeros((1, 3, 8), dtype=torch.bfloat16) hidden_states = torch.zeros((1, 3, 8), dtype=torch.bfloat16)
mask = _decoder_attention_mask(torch.ones((1, 3), dtype=torch.long), hidden_states, torch) mask = _decoder_attention_mask(torch.ones((1, 3), dtype=torch.long), hidden_states, torch)
@@ -573,7 +583,7 @@ def test_call_layer_passes_rotary_position_embeddings():
assert kwargs["position_embeddings"] == "rotary" assert kwargs["position_embeddings"] == "rotary"
return hidden_states return hidden_states
hidden, cache_state = _call_layer( hidden = _call_layer(
NeedsPositionEmbeddings(), NeedsPositionEmbeddings(),
"hidden", "hidden",
attention_mask=None, attention_mask=None,
@@ -582,7 +592,6 @@ def test_call_layer_passes_rotary_position_embeddings():
) )
assert hidden == "hidden" assert hidden == "hidden"
assert cache_state is None
def _fake_cache_shard(torch, *, max_sessions=16, ttl=600.0): def _fake_cache_shard(torch, *, max_sessions=16, ttl=600.0):
@@ -618,7 +627,7 @@ def _fake_cache_shard(torch, *, max_sessions=16, ttl=600.0):
def test_shard_cache_prefill_then_decode_reuses_opaque_layer_state(): def test_shard_cache_prefill_then_decode_reuses_opaque_layer_state():
"Shard cache prefill then decode reuses opaque layer state\n\nTags: cache, model, node, real-inference" "Shard cache prefill then decode reuses opaque layer state\n\nTags: cache, model, node, real-inference"
torch = pytest.importorskip("torch") torch = _require_functional_torch()
shard = _fake_cache_shard(torch) shard = _fake_cache_shard(torch)
prefill_hidden = torch.zeros((1, 4, 2), dtype=torch.bfloat16) prefill_hidden = torch.zeros((1, 4, 2), dtype=torch.bfloat16)
@@ -658,7 +667,7 @@ def test_shard_cache_prefill_then_decode_reuses_opaque_layer_state():
def test_shard_cache_decode_miss_is_explicit(): def test_shard_cache_decode_miss_is_explicit():
"Shard cache decode miss is explicit\n\nTags: cache, model, node, real-inference" "Shard cache decode miss is explicit\n\nTags: cache, model, node, real-inference"
torch = pytest.importorskip("torch") torch = _require_functional_torch()
shard = _fake_cache_shard(torch) shard = _fake_cache_shard(torch)
with pytest.raises(KVCacheMiss): with pytest.raises(KVCacheMiss):
@@ -674,7 +683,7 @@ def test_shard_cache_decode_miss_is_explicit():
def test_shard_cache_lru_bounds_sessions(): def test_shard_cache_lru_bounds_sessions():
"Shard cache lru bounds sessions\n\nTags: cache, model, node, real-inference" "Shard cache lru bounds sessions\n\nTags: cache, model, node, real-inference"
torch = pytest.importorskip("torch") torch = _require_functional_torch()
shard = _fake_cache_shard(torch, max_sessions=1) shard = _fake_cache_shard(torch, max_sessions=1)
for session in ("old", "new"): for session in ("old", "new"):
@@ -1154,7 +1163,7 @@ def test_torch_model_shard_prefers_partial_loader_for_local_snapshot(tmp_path, m
"torch", "torch",
types.SimpleNamespace( types.SimpleNamespace(
cuda=types.SimpleNamespace(is_available=lambda: False), cuda=types.SimpleNamespace(is_available=lambda: False),
device=lambda value: value, device=lambda value: types.SimpleNamespace(type=value),
bfloat16="bf16", bfloat16="bf16",
), ),
) )
@@ -1194,7 +1203,7 @@ def test_two_node_gpt2_completion_is_deterministic():
"Two node gpt2 completion is deterministic\n\nTags: model, node, real-inference" "Two node gpt2 completion is deterministic\n\nTags: model, node, real-inference"
if os.environ.get("CI"): if os.environ.get("CI"):
pytest.skip("GPT-2 integration test is skipped in CI") pytest.skip("GPT-2 integration test is skipped in CI")
torch = pytest.importorskip("torch") torch = _require_functional_torch()
pytest.importorskip("transformers") pytest.importorskip("transformers")
pytest.importorskip("safetensors") pytest.importorskip("safetensors")
pytest.importorskip("accelerate") pytest.importorskip("accelerate")

View File

@@ -14,11 +14,13 @@ from meshnet_gateway.server import GatewayServer, _banned_route_wallet
from meshnet_node.server import StubNodeServer from meshnet_node.server import StubNodeServer
from meshnet_contracts import LocalSolanaContracts from meshnet_contracts import LocalSolanaContracts
from meshnet_tracker.auth import sign_hive_request from meshnet_tracker.auth import sign_hive_request
from meshnet_tracker.capability import POLICY_COMPAT, POLICY_ENFORCE
from meshnet_tracker.server import ( from meshnet_tracker.server import (
TrackerServer, TrackerServer,
_NodeEntry, _NodeEntry,
_available_quantizations, _available_quantizations,
_memory_pool_map, _memory_pool_map,
_rebalance_all_locked,
_registration_ban_error, _registration_ban_error,
_scale_demanded_models_locked, _scale_demanded_models_locked,
) )
@@ -2906,3 +2908,252 @@ def test_scale_demanded_models_queues_add_shard_on_spare_host():
assert assignment["model"] == "org/ModelA" assert assignment["model"] == "org/ModelA"
finally: finally:
tracker.stop() tracker.stop()
# ------------------------------------------------- RCR-001: legacy precision fallback
class _EchoChatHandler(http.server.BaseHTTPRequestHandler):
"""A tracker-mode node that answers the proxied chat request."""
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)
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)
def _legacy_registration(endpoint: str, quantization: str | None = None) -> dict:
"""A registration as a pre-quantization-field node sends it: no `quantization`."""
payload = {
"endpoint": endpoint,
"model": "Qwen2.5-0.5B-Instruct",
"hf_repo": "Qwen/Qwen2.5-0.5B-Instruct",
"num_layers": 24,
"shard_start": 0,
"shard_end": 23,
"tracker_mode": True,
"hardware_profile": {},
"score": 1.0,
}
if quantization is not None:
payload["quantization"] = quantization
return payload
def _proxy_chat_status(policy: str, quantization: str | None) -> tuple[int, dict | None]:
"""Register one legacy node under `policy` and proxy a default-precision chat."""
node = http.server.HTTPServer(("127.0.0.1", 0), _EchoChatHandler)
node_thread = threading.Thread(target=node.serve_forever, daemon=True)
node_thread.start()
tracker = TrackerServer(capability_policy=policy)
tracker_port = tracker.start()
try:
endpoint = f"http://127.0.0.1:{node.server_address[1]}"
_post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
_legacy_registration(endpoint, quantization),
)
try:
response = _post_json(
f"http://127.0.0.1:{tracker_port}/v1/chat/completions",
{"model": "qwen2.5-0.5b",
"messages": [{"role": "user", "content": "hi"}]},
)
except urllib.error.HTTPError as exc:
return exc.code, None
return 200, response
finally:
tracker.stop()
node.shutdown()
node.server_close()
node_thread.join(timeout=1.0)
def test_a_legacy_registration_without_quantization_serves_a_default_precision_request():
"An omitted field predates the protocol: it must not make a healthy node dark.\n\nTags: http, routing, tracker"
status, response = _proxy_chat_status(POLICY_COMPAT, quantization=None)
assert status == 200
assert response["choices"][0]["message"]["content"] == "ok"
def test_a_legacy_registration_without_quantization_stays_dark_under_enforce():
"The precision fallback must not smuggle an absent capability report past `enforce`.\n\nTags: http, routing, tracker"
status, _ = _proxy_chat_status(POLICY_ENFORCE, quantization=None)
assert status == 503
def test_a_node_declaring_an_unsupported_quantization_is_never_routed():
"Silence is grandfathered; an explicit unsupported precision is not.\n\nTags: http, routing, tracker"
status, _ = _proxy_chat_status(POLICY_COMPAT, quantization="fp4-experimental")
assert status == 503
def test_a_node_declaring_a_null_quantization_is_never_routed():
"An explicit null states 'no usable precision' -- only an absent field is legacy.\n\nTags: http, routing, tracker"
node = http.server.HTTPServer(("127.0.0.1", 0), _EchoChatHandler)
node_thread = threading.Thread(target=node.serve_forever, daemon=True)
node_thread.start()
tracker = TrackerServer(capability_policy=POLICY_COMPAT)
tracker_port = tracker.start()
try:
endpoint = f"http://127.0.0.1:{node.server_address[1]}"
registration = _legacy_registration(endpoint)
registration["quantization"] = None # declared, not omitted
_post_json(f"http://127.0.0.1:{tracker_port}/v1/nodes/register", registration)
with pytest.raises(urllib.error.HTTPError) as excinfo:
_post_json(
f"http://127.0.0.1:{tracker_port}/v1/chat/completions",
{"model": "qwen2.5-0.5b",
"messages": [{"role": "user", "content": "hi"}]},
)
assert excinfo.value.code == 503
finally:
tracker.stop()
node.shutdown()
node.server_close()
node_thread.join(timeout=1.0)
def test_raft_apply_preserves_a_declared_precision_for_follower_routing():
"A follower that drops the declared precision would route unlike the leader.\n\nTags: raft, routing, tracker"
tracker = TrackerServer()
tracker.start()
try:
tracker._raft_apply("register", {
"node_id": "declared-node",
"endpoint": "http://127.0.0.1:1/",
"model": "Qwen2.5-0.5B-Instruct",
"hf_repo": "Qwen/Qwen2.5-0.5B-Instruct",
"shard_start": 0,
"shard_end": 23,
"num_layers": 24,
"quantization": "nf4",
"quantizations": ["nf4", "int8"],
})
entry = tracker._registry["declared-node"]
assert entry.quantization == "nf4"
assert entry.quantizations == ["nf4", "int8"]
finally:
tracker.stop()
def test_raft_apply_grandfathers_a_legacy_registration_that_omits_quantization():
"The follower must reach the same legacy fallback the leader did, not a bare None.\n\nTags: raft, routing, tracker"
tracker = TrackerServer()
tracker.start()
try:
tracker._raft_apply("register", {
"node_id": "legacy-node",
"endpoint": "http://127.0.0.1:1/",
"model": "Qwen2.5-0.5B-Instruct",
"hf_repo": "Qwen/Qwen2.5-0.5B-Instruct",
"shard_start": 0,
"shard_end": 23,
"num_layers": 24,
})
entry = tracker._registry["legacy-node"]
assert entry.quantization == "bfloat16"
finally:
tracker.stop()
def test_rebalancing_does_not_resurrect_a_node_that_declared_an_unusable_precision():
"Managed assignment picks a node's precision, so it must not overwrite an explicit unusable one.\n\nTags: http, routing, tracker"
tracker = TrackerServer(heartbeat_timeout=0.15, rebalance_interval=10.0)
tracker_port = tracker.start()
try:
managed = _post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": "http://127.0.0.1:9111", "model": "Qwen2.5-0.5B-Instruct",
"hf_repo": "Qwen/Qwen2.5-0.5B-Instruct", "num_layers": 24,
"shard_start": 0, "shard_end": 21, "managed_assignment": True,
"quantization": "fp4-experimental",
"vram_bytes": 1_000_000_000, "hardware_profile": {}, "score": 1.0},
)
# Letting this peer expire is what drives the rebalance that used to hand
# the managed node a fresh shard range -- and a default precision with it.
_post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": "http://127.0.0.1:9112", "model": "Qwen2.5-0.5B-Instruct",
"hf_repo": "Qwen/Qwen2.5-0.5B-Instruct", "num_layers": 24,
"shard_start": 22, "shard_end": 23,
"vram_bytes": 1_000_000_000, "hardware_profile": {}, "score": 1.0},
)
time.sleep(0.10)
_post_json(f"http://127.0.0.1:{tracker_port}/v1/nodes/{managed['node_id']}/heartbeat", {})
time.sleep(0.10)
hb = _post_json(f"http://127.0.0.1:{tracker_port}/v1/nodes/{managed['node_id']}/heartbeat", {})
entry = tracker._registry[managed["node_id"]]
assert entry.quantization is None
assert not [d for d in hb.get("directives", []) if d["action"] == "LOAD_SHARD"]
finally:
tracker.stop()
def _register_shard(tracker_port: int, port: int, **overrides) -> dict:
payload = {
"endpoint": f"http://127.0.0.1:{port}",
"model": "Qwen2.5-0.5B-Instruct",
"hf_repo": "Qwen/Qwen2.5-0.5B-Instruct",
"num_layers": 24,
"vram_bytes": 40_000_000_000,
"hardware_profile": {},
"score": 1.0,
}
payload.update(overrides)
return _post_json(f"http://127.0.0.1:{tracker_port}/v1/nodes/register", payload)
def test_a_node_with_an_unusable_precision_covers_no_layers():
"A node no request can reach must not report its shards as served coverage.\n\nTags: http, routing, tracker"
tracker = TrackerServer(rebalance_interval=10.0)
tracker_port = tracker.start()
try:
_register_shard(
tracker_port, 9121,
shard_start=0, shard_end=23, quantization="fp4-experimental",
)
models = _get_json(f"http://127.0.0.1:{tracker_port}/v1/models")
served = [m for m in models["data"] if m["id"] == "Qwen/Qwen2.5-0.5B-Instruct"]
assert [m["shard_coverage_percentage"] for m in served] == [0.0]
finally:
tracker.stop()
def test_rebalancing_fills_the_layers_an_unusable_node_only_appears_to_cover():
"Counting a dark node as coverage hides the gap, so the rebalancer never fills it.\n\nTags: http, routing, tracker"
tracker = TrackerServer(rebalance_interval=10.0)
tracker_port = tracker.start()
try:
# The dark node claims the model's tail; the managed node holds the head.
# Both ends look covered, so the rebalancer used to see nothing to do.
dark = _register_shard(
tracker_port, 9131,
shard_start=12, shard_end=23, quantization="fp4-experimental",
)
managed = _register_shard(
tracker_port, 9132,
shard_start=0, shard_end=11, managed_assignment=True,
)
with tracker._lock:
_rebalance_all_locked(tracker._server) # type: ignore[arg-type]
assert tracker._registry[dark["node_id"]].quantization is None
entry = tracker._registry[managed["node_id"]]
assert (entry.shard_start, entry.shard_end) == (0, 23)
finally:
tracker.stop()