From b5fa7245df360ab2266168e9294f12418b74c320 Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Mon, 13 Jul 2026 10:27:45 +0300 Subject: [PATCH] [verified] fix: preserve tracker precision eligibility --- .scratch/proxy-stream-cancellation/PRD.md | 27 ++ .../issues/01-direct-sse-cancellation-race.md | 34 +++ .../routing-compatibility-regression/PRD.md | 27 ++ ...-legacy-registration-precision-fallback.md | 38 +++ .../routing-compatibility-regression/prd.json | 29 ++ packages/tracker/meshnet_tracker/server.py | 104 +++++++- tests/test_hf_pricing_dispatch.py | 2 +- tests/test_meshnet_sdk.py | 8 +- tests/test_mining_cli.py | 52 ++-- tests/test_node_startup.py | 2 + tests/test_real_model_backend.py | 31 ++- tests/test_tracker_routing.py | 251 ++++++++++++++++++ 12 files changed, 557 insertions(+), 48 deletions(-) create mode 100644 .scratch/proxy-stream-cancellation/PRD.md create mode 100644 .scratch/proxy-stream-cancellation/issues/01-direct-sse-cancellation-race.md create mode 100644 .scratch/routing-compatibility-regression/PRD.md create mode 100644 .scratch/routing-compatibility-regression/issues/01-legacy-registration-precision-fallback.md create mode 100644 .scratch/routing-compatibility-regression/prd.json diff --git a/.scratch/proxy-stream-cancellation/PRD.md b/.scratch/proxy-stream-cancellation/PRD.md new file mode 100644 index 0000000..944f5e0 --- /dev/null +++ b/.scratch/proxy-stream-cancellation/PRD.md @@ -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. diff --git a/.scratch/proxy-stream-cancellation/issues/01-direct-sse-cancellation-race.md b/.scratch/proxy-stream-cancellation/issues/01-direct-sse-cancellation-race.md new file mode 100644 index 0000000..e379c25 --- /dev/null +++ b/.scratch/proxy-stream-cancellation/issues/01-direct-sse-cancellation-race.md @@ -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 tracker’s 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//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 3953–4019). 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. diff --git a/.scratch/routing-compatibility-regression/PRD.md b/.scratch/routing-compatibility-regression/PRD.md new file mode 100644 index 0000000..5c3d2e6 --- /dev/null +++ b/.scratch/routing-compatibility-regression/PRD.md @@ -0,0 +1,27 @@ +# PRD: Legacy routing compatibility regression + +## Overview + +The tracker’s 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. diff --git a/.scratch/routing-compatibility-regression/issues/01-legacy-registration-precision-fallback.md b/.scratch/routing-compatibility-regression/issues/01-legacy-registration-precision-fallback.md new file mode 100644 index 0000000..ded858e --- /dev/null +++ b/.scratch/routing-compatibility-regression/issues/01-legacy-registration-precision-fallback.md @@ -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. diff --git a/.scratch/routing-compatibility-regression/prd.json b/.scratch/routing-compatibility-regression/prd.json new file mode 100644 index 0000000..c576c16 --- /dev/null +++ b/.scratch/routing-compatibility-regression/prd.json @@ -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" + } +} \ No newline at end of file diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 824c45a..2683036 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -972,7 +972,7 @@ def _coverage_percentage( intervals = sorted( ( (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 and node.shard_end is not None if node.shard_end >= required_start and node.shard_start <= required_end @@ -1004,7 +1004,7 @@ def _served_model_copies( layer_counts = [] for layer in range(required_start, required_end + 1): count = 0 - for node in nodes: + for node in _serviceable_nodes(nodes): if node.shard_start is None or node.shard_end is None: continue 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 +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: """Return whether a shard meets a request's minimum precision.""" available_normalized = _normalize_quantization(available) @@ -1406,7 +1457,10 @@ def _scale_demanded_models_locked(server: "_TrackerHTTPServer") -> None: continue if any((n.hf_repo or n.model) == hf_repo for n in host_nodes): 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: continue 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: continue 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 - 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: continue capacity = min(_node_layer_capacity(anchor, preset), total_layers) @@ -1479,6 +1534,8 @@ def _preferred_node_quantization( requested: str, ) -> str | None: """Choose the highest supported precision that satisfies a request.""" + if not _has_usable_quantization(node): + return None supported = { normalized for value in ([node.quantization] if node.quantization else []) + list(node.quantizations) @@ -1631,7 +1688,7 @@ def _coverage_map( layer_counts = [] for layer in range(required_start, required_end + 1): count = 0 - for node in nodes: + for node in _serviceable_nodes(nodes): if node.shard_start is None or node.shard_end is None: continue if node.shard_start <= layer <= node.shard_end: @@ -1680,8 +1737,10 @@ def _coverage_map_detailed( ) -> list[dict]: """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 - holes and which dead nodes used to fill them. + Includes all nodes (alive and stale, serviceable or not) so operators can see + 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() @@ -1793,6 +1852,8 @@ def _assign_redundant_managed_nodes( ): if node.shard_start is not None and node.shard_end is not None: continue + if not _has_usable_quantization(node): + continue capacity = min(_node_layer_capacity(node, preset), total_layers) if capacity <= 0: continue @@ -2225,7 +2286,8 @@ def _rebalance_model_locked(server: "_TrackerHTTPServer", model: str) -> None: eligible_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 for gap_start, gap_end in gaps: @@ -2315,7 +2377,8 @@ def _rebalance_hf_model_locked(server: "_TrackerHTTPServer", hf_repo: str) -> No eligible_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 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"}) return quantizations = list(quantizations_body) - quantization = body.get("quantization") - if quantization is not None and not isinstance(quantization, str): + if body.get("quantization") is not None and not isinstance(body["quantization"], str): self._send_json(400, {"error": "quantization must be a string"}) return + quantization = _registration_quantization(body, quantizations) wallet_address = body.get("wallet_address") if wallet_address is not None and not isinstance(wallet_address, str): 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")) except ValueError: 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( node_id=node_id, endpoint=endpoint.rstrip("/"), @@ -6885,6 +6963,8 @@ class TrackerServer: hardware_profile=payload.get("hardware_profile", {}), wallet_address=payload.get("wallet_address"), score=float(payload.get("score", 1.0)), + quantizations=quantizations, + quantization=_registration_quantization(payload, quantizations), tracker_mode=bool(payload.get("tracker_mode", False)), hf_repo=payload.get("hf_repo"), num_layers=int(payload["num_layers"]) if payload.get("num_layers") is not None else None, diff --git a/tests/test_hf_pricing_dispatch.py b/tests/test_hf_pricing_dispatch.py index 9d14817..1c6a6bd 100644 --- a/tests/test_hf_pricing_dispatch.py +++ b/tests/test_hf_pricing_dispatch.py @@ -95,7 +95,7 @@ def test_refresh_loop_repriced_model_with_curated_alias(pricing_tracker): tracker_url, ledger, tracker = pricing_tracker new_price = _wait_for_price_change(ledger, PRICED_MODEL) 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] assert preset["hf_last_price_per_1k"] == expected diff --git a/tests/test_meshnet_sdk.py b/tests/test_meshnet_sdk.py index beac1cb..21a55b5 100644 --- a/tests/test_meshnet_sdk.py +++ b/tests/test_meshnet_sdk.py @@ -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" 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 = [ - _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), + _NodeEntry("a", "http://node-a", 0, 15, "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 diff --git a/tests/test_mining_cli.py b/tests/test_mining_cli.py index 0f3971d..22b9b61 100644 --- a/tests/test_mining_cli.py +++ b/tests/test_mining_cli.py @@ -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" -def test_legacy_start_without_port_uses_next_available_port(monkeypatch): - "Omitting --port skips an occupied default port before startup loads the model.\n\nTags: general" - from meshnet_node.cli import main +def test_first_available_port_skips_an_occupied_custom_port(): + "Port search skips an occupied custom base port.\n\nTags: general" + 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 = {} @@ -566,26 +580,20 @@ def test_legacy_start_without_port_uses_next_available_port(monkeypatch): 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", - ]) + monkeypatch.setattr(cli_mod, "_first_available_port", lambda *_args, **_kwargs: 7001) + 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() + with patch("meshnet_node.startup.run_startup", side_effect=fake_run_startup): + with patch("time.sleep", side_effect=KeyboardInterrupt): + try: + cli_mod.main() + except SystemExit as exc: + assert exc.code == 0 assert captured["port"] == 7001 diff --git a/tests/test_node_startup.py b/tests/test_node_startup.py index 04b2c0e..1c302a7 100644 --- a/tests/test_node_startup.py +++ b/tests/test_node_startup.py @@ -2454,7 +2454,9 @@ def test_cli_loads_local_env_before_config_defaults(tmp_path, monkeypatch): from meshnet_node import config as config_mod monkeypatch.delenv("MESHNET_DOWNLOAD_DIR", raising=False) + monkeypatch.delenv("HF_TOKEN", raising=False) monkeypatch.chdir(tmp_path) + cli_mod = importlib.reload(cli_mod) (tmp_path / ".env").write_text( "MESHNET_DOWNLOAD_DIR=/run/media/popov/DATA/llm/safetensor/models\n" "HF_TOKEN=hf_test_token\n" diff --git a/tests/test_real_model_backend.py b/tests/test_real_model_backend.py index 92b0901..6f533b3 100644 --- a/tests/test_real_model_backend.py +++ b/tests/test_real_model_backend.py @@ -32,6 +32,15 @@ from meshnet_node.model_backend import ( 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: model_id = "fake-model" total_layers = 12 @@ -526,13 +535,14 @@ def test_distributed_generating_log_includes_tps(capsys): out = capsys.readouterr().out assert "generating step=1/1" 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 def test_int_tensor_header_serializes_torch_tensors(): "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)) @@ -544,7 +554,7 @@ def test_bfloat16_wire_decode_views_owned_bytes_without_float32_round_trip(): 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() 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(): "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) 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" return hidden_states - hidden, cache_state = _call_layer( + hidden = _call_layer( NeedsPositionEmbeddings(), "hidden", attention_mask=None, @@ -582,7 +592,6 @@ def test_call_layer_passes_rotary_position_embeddings(): ) assert hidden == "hidden" - assert cache_state is None 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(): "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) 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(): "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) with pytest.raises(KVCacheMiss): @@ -674,7 +683,7 @@ def test_shard_cache_decode_miss_is_explicit(): def test_shard_cache_lru_bounds_sessions(): "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) for session in ("old", "new"): @@ -1154,7 +1163,7 @@ def test_torch_model_shard_prefers_partial_loader_for_local_snapshot(tmp_path, m "torch", types.SimpleNamespace( cuda=types.SimpleNamespace(is_available=lambda: False), - device=lambda value: value, + device=lambda value: types.SimpleNamespace(type=value), 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" if os.environ.get("CI"): pytest.skip("GPT-2 integration test is skipped in CI") - torch = pytest.importorskip("torch") + torch = _require_functional_torch() pytest.importorskip("transformers") pytest.importorskip("safetensors") pytest.importorskip("accelerate") diff --git a/tests/test_tracker_routing.py b/tests/test_tracker_routing.py index e58f501..853ef6e 100644 --- a/tests/test_tracker_routing.py +++ b/tests/test_tracker_routing.py @@ -14,11 +14,13 @@ from meshnet_gateway.server import GatewayServer, _banned_route_wallet from meshnet_node.server import StubNodeServer from meshnet_contracts import LocalSolanaContracts from meshnet_tracker.auth import sign_hive_request +from meshnet_tracker.capability import POLICY_COMPAT, POLICY_ENFORCE from meshnet_tracker.server import ( TrackerServer, _NodeEntry, _available_quantizations, _memory_pool_map, + _rebalance_all_locked, _registration_ban_error, _scale_demanded_models_locked, ) @@ -2906,3 +2908,252 @@ def test_scale_demanded_models_queues_add_shard_on_spare_host(): assert assignment["model"] == "org/ModelA" finally: 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()