diff --git a/docs/issues/47-model-source-download-visibility.md b/docs/issues/47-model-source-download-visibility.md new file mode 100644 index 0000000..8ca1e9b --- /dev/null +++ b/docs/issues/47-model-source-download-visibility.md @@ -0,0 +1,58 @@ +# US-047 — Model-source race: visibility, sane timeouts, quiet tracker aborts + +Status: in progress +Priority: High (follow-up to US-044/US-046; blocks usable LAN downloads) + +## Context + +Reported 2026-07-06 (Windows CPU node, 79.2 GB RAM, `--tracker +http://192.168.0.179:8080 --model Qwen3.6-35B-A3B`): + +1. Startup prints `(auto-join unavailable: HTTP Error 503)` even though the + user explicitly named a model. The auto-join query (`/v1/network/assign`) + never sends the requested model, so a fresh tracker + a caller too small + for the *recommended* preset 503s (expected per US-046) — but the whole + auto-join step is pointless when the user already picked a model: the + `/v1/nodes/assign?model=…` call right after it succeeds (assigned layers + 0–2 with tracker `model_sources`). +2. The tracker-vs-HuggingFace race then starts, but only HuggingFace shows + progress (hf tqdm bars). The tracker tar download prints nothing and + swallows every failure (`except Exception: return None`), so the node + *appears* to download only from slow HF; the user killed it. Tracker-side + log showed the tar stream reset mid-`archive.add` — with no way to tell + whether the client timed out or the user aborted. +3. `_download_model_source` inherits `peer_timeout` (2.0 s) as its urlopen + socket timeout. Any 2 s read stall during a multi-GB tar stream silently + kills the tracker source and leaves HF as the only contender. +4. Every client abort spams the tracker console with a full + `BrokenPipeError`/`ConnectionResetError` traceback from `socketserver`. + +## Fix + +1. `startup.py`: skip the network auto-join query entirely when a model was + explicitly requested (`model` set and not `"stub-model"`); path 3b + (`/v1/nodes/assign?model=…`) is the authoritative one there. +2. `downloader.py`: model-source downloads get their own timeout constant + (30 s socket timeout) instead of the 2 s peer-probe timeout. Peer shard + downloads keep 2 s — they run sequentially before the race, and a dead + peer must not hang startup for 30 s; the race is concurrent so a slow + source costs nothing. +3. `downloader.py`: progress + failure visibility for the race — + `_download_model_source` prints received bytes every 512 MB and prints + the exception when a source fails, so "downloads only from HF" can never + happen silently again. +4. Tracker `_handle_model_files_download`: catch + `BrokenPipeError`/`ConnectionResetError` around the tar stream and log a + single line instead of a traceback. + +## Acceptance criteria + +- [ ] Node started with an explicit `--model` never queries + `/v1/network/assign` and never prints `auto-join unavailable`. +- [ ] During the race, tracker-source progress lines appear alongside HF + tqdm output; a failing tracker source prints its exception. +- [ ] A ≥2 s read stall no longer aborts a tracker model-source download + (30 s socket timeout). +- [ ] Client disconnect during `/v1/model-files/download` logs one line on + the tracker, no traceback. +- [ ] `python -m pytest` passes from repo root. diff --git a/packages/node/meshnet_node/downloader.py b/packages/node/meshnet_node/downloader.py index c57958d..15c4fac 100644 --- a/packages/node/meshnet_node/downloader.py +++ b/packages/node/meshnet_node/downloader.py @@ -19,6 +19,11 @@ from typing import Any _DEFAULT_CACHE = Path.home() / ".cache" / "meshnet" / "shards" _PEER_TIMEOUT_SECONDS = 2.0 +# Model-source tar streams are multi-GB; a short socket timeout must not kill +# them on a transient read stall. Peer probes keep the short timeout because +# they run sequentially before the race and may hit dead endpoints. +_MODEL_SOURCE_TIMEOUT_SECONDS = 30.0 +_PROGRESS_INTERVAL_BYTES = 512 * 1024 * 1024 def compute_shard_checksum(shard_dir: Path) -> str: @@ -111,6 +116,8 @@ def _download_model_source( source: dict, shard_dir: Path, timeout: float, + progress: bool = False, + label: str = "model-source", ) -> Path | None: url = source.get("url") if not isinstance(url, str) or not url: @@ -125,16 +132,26 @@ def _download_model_source( extract_dir = tmp_root / "extract" extract_dir.mkdir() try: + received = 0 + next_report = _PROGRESS_INTERVAL_BYTES with urllib.request.urlopen(url, timeout=timeout) as resp, archive_path.open("wb") as out: while True: chunk = resp.read(1024 * 1024) if not chunk: break out.write(chunk) + received += len(chunk) + if progress and received >= next_report: + print(f" {label}: {received / 1e9:.1f} GB received ...", flush=True) + next_report += _PROGRESS_INTERVAL_BYTES + if progress: + print(f" {label}: transfer complete ({received / 1e9:.2f} GB), extracting ...", flush=True) _safe_extract_shard(archive_path, extract_dir) shutil.move(str(extract_dir), str(shard_dir)) return shard_dir - except Exception: + except Exception as exc: + if progress: + print(f" {label}: download failed: {exc!r}", flush=True) return None @@ -178,7 +195,9 @@ def _download_from_fastest_source( for index, source in enumerate(model_sources): label = str(source.get("type") or "model-source") candidate = tmp_root / f"source-{index}" - jobs[pool.submit(_download_model_source, source, candidate, timeout)] = (label, candidate) + jobs[pool.submit( + _download_model_source, source, candidate, timeout, progress, label, + )] = (label, candidate) allow_patterns = _allow_patterns_from_sources(model_sources) hf_candidate = tmp_root / "huggingface" jobs[pool.submit(_download_huggingface_subset, hf_repo, cache_dir, hf_candidate, allow_patterns)] = ( @@ -189,7 +208,9 @@ def _download_from_fastest_source( label, candidate = jobs[future] try: result = future.result() - except Exception: + except Exception as exc: + if progress: + print(f" {label}: download failed: {exc!r}", flush=True) continue if result is None: continue @@ -324,7 +345,7 @@ def download_shard( cache_dir=cache_dir, shard_dir=shard_dir, progress=progress, - timeout=peer_timeout, + timeout=max(peer_timeout, _MODEL_SOURCE_TIMEOUT_SECONDS), ) if raced is not None: return raced[1] diff --git a/packages/node/meshnet_node/startup.py b/packages/node/meshnet_node/startup.py index 5700cf3..a3365f5 100644 --- a/packages/node/meshnet_node/startup.py +++ b/packages/node/meshnet_node/startup.py @@ -561,13 +561,19 @@ def run_startup( raise ValueError("--shard-start / --shard-end require --model-id") # 3a. Auto-join: query tracker for network-wide HF model assignment. - print("Querying tracker for network assignment...", flush=True) - assign_qs = urllib.parse.urlencode({"device": device, "vram_mb": assignment_vram_mb, "ram_mb": ram_mb}) + # Skipped when the user explicitly requested a model — the shard-assignment + # query below (/v1/nodes/assign?model=…) is authoritative there, and a fresh + # tracker would otherwise print a scary 503 for the model-less auto-join. net_assignment: dict = {} - try: - net_assignment = _get_json(f"{tracker_url}/v1/network/assign?{assign_qs}") - except Exception as exc: - print(f" (auto-join unavailable: {exc})", flush=True) + if model and model != "stub-model": + print(f"Model {model!r} requested explicitly — skipping network auto-join.", flush=True) + else: + print("Querying tracker for network assignment...", flush=True) + assign_qs = urllib.parse.urlencode({"device": device, "vram_mb": assignment_vram_mb, "ram_mb": ram_mb}) + try: + net_assignment = _get_json(f"{tracker_url}/v1/network/assign?{assign_qs}") + except Exception as exc: + print(f" (auto-join unavailable: {exc})", flush=True) assigned_hf_repo: str | None = net_assignment.get("hf_repo") _gap_found: bool = bool(net_assignment.get("gap_found", False)) diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index fbd724a..8ae8c80 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -3879,10 +3879,17 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): self.send_header("Content-Type", "application/x-tar") self.send_header("X-Meshnet-Model-Source", "tracker") self.end_headers() - # dereference: HF cache snapshots are symlinks into blobs/ — ship contents. - with tarfile.open(fileobj=self.wfile, mode="w|", dereference=True) as archive: - for rel in rel_files: - archive.add(snapshot_dir / rel, arcname=rel) + try: + # dereference: HF cache snapshots are symlinks into blobs/ — ship contents. + with tarfile.open(fileobj=self.wfile, mode="w|", dereference=True) as archive: + for rel in rel_files: + archive.add(snapshot_dir / rel, arcname=rel) + except (BrokenPipeError, ConnectionResetError): + print( + f"model-file download aborted by client " + f"({self.client_address[0]}, model={resolved_name})", + flush=True, + ) def _handle_network_assign(self, parsed: urllib.parse.ParseResult): """Assign a new node to fill the biggest uncovered shard gap across HF-model nodes.