From dae0719a32f1c2c40c665fa9d4218ab121c511dd Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Tue, 7 Jul 2026 22:00:54 +0300 Subject: [PATCH 1/3] logging, routing --- packages/tracker/meshnet_tracker/cli.py | 61 ++++++++++-- .../tracker/meshnet_tracker/logging_setup.py | 99 +++++++++++++++++++ packages/tracker/meshnet_tracker/server.py | 56 +++++++---- tests/test_tracker_logging.py | 67 +++++++++++++ tests/test_tracker_routing.py | 69 +++++++++++++ 5 files changed, 325 insertions(+), 27 deletions(-) create mode 100644 packages/tracker/meshnet_tracker/logging_setup.py create mode 100644 tests/test_tracker_logging.py diff --git a/packages/tracker/meshnet_tracker/cli.py b/packages/tracker/meshnet_tracker/cli.py index e6d7bdd..642ccd5 100644 --- a/packages/tracker/meshnet_tracker/cli.py +++ b/packages/tracker/meshnet_tracker/cli.py @@ -8,8 +8,14 @@ from pathlib import Path from .accounts import DEFAULT_ACCOUNTS_DB_PATH from .billing import DEFAULT_BILLING_DB_PATH -from .hf_pricing import DEFAULT_HF_PRICING_LOG_DB_PATH -from .server import ( +from .hf_pricing import DEFAULT_HF_PRICING_LOG_DB_PATH +from .logging_setup import ( + DEFAULT_LOG_BACKUP_COUNT, + DEFAULT_LOG_DIR, + DEFAULT_LOG_MAX_BYTES, + configure_tracker_file_logging, +) +from .server import ( DEFAULT_CALLER_CREDIT_USDT, DEFAULT_DEVNET_TOPUP_USDT, TrackerServer, @@ -255,12 +261,40 @@ def main() -> None: default=86400.0, help="Seconds between dynamic pricing refresh passes (default: daily)", ) - common.add_argument( - "--models-dir", - default=None, - metavar="PATH", - help="Local HuggingFace snapshot root advertised as tracker model-file source (default: MESHNET_MODELS_DIR)", - ) + common.add_argument( + "--models-dir", + default=None, + metavar="PATH", + help="Local HuggingFace snapshot root advertised as tracker model-file source (default: MESHNET_MODELS_DIR)", + ) + common.add_argument( + "--log-dir", + default=DEFAULT_LOG_DIR, + metavar="PATH", + help=( + "Directory for rotating tracker logs " + f"(default: {DEFAULT_LOG_DIR}; files: info.log, warning.log, error.log)" + ), + ) + common.add_argument( + "--log-max-bytes", + type=int, + default=DEFAULT_LOG_MAX_BYTES, + metavar="BYTES", + help=f"Rotate each tracker log file after this many bytes (default: {DEFAULT_LOG_MAX_BYTES})", + ) + common.add_argument( + "--log-backup-count", + type=int, + default=DEFAULT_LOG_BACKUP_COUNT, + metavar="N", + help=f"Number of rotated tracker log files to keep per level (default: {DEFAULT_LOG_BACKUP_COUNT})", + ) + common.add_argument( + "--no-file-logs", + action="store_true", + help="Disable rotating tracker log files and only write to the terminal", + ) parser = argparse.ArgumentParser( prog="meshnet-tracker", @@ -273,8 +307,15 @@ def main() -> None: args = parser.parse_args() - if args.command in {None, "start"}: - cluster_peers = [u.strip() for u in args.cluster_peers.split(",") if u.strip()] + if args.command in {None, "start"}: + if not args.no_file_logs: + log_dir = configure_tracker_file_logging( + args.log_dir, + max_bytes=args.log_max_bytes, + backup_count=args.log_backup_count, + ) + print(f"meshnet-tracker logs: {log_dir}", flush=True) + cluster_peers = [u.strip() for u in args.cluster_peers.split(",") if u.strip()] relay_url = args.relay_url or derive_relay_url_from_public_tracker_url(args.self_url) treasury = None if args.solana_rpc_url and args.usdt_mint and args.treasury_keypair: diff --git a/packages/tracker/meshnet_tracker/logging_setup.py b/packages/tracker/meshnet_tracker/logging_setup.py new file mode 100644 index 0000000..7d8ec73 --- /dev/null +++ b/packages/tracker/meshnet_tracker/logging_setup.py @@ -0,0 +1,99 @@ +"""Rotating file logging for the tracker CLI.""" + +from __future__ import annotations + +import logging +import sys +from logging.handlers import RotatingFileHandler +from pathlib import Path +from typing import TextIO + + +DEFAULT_LOG_DIR = "logs/tracker" +DEFAULT_LOG_MAX_BYTES = 10 * 1024 * 1024 +DEFAULT_LOG_BACKUP_COUNT = 5 +TRACKER_LOGGER_NAME = "meshnet.tracker" + + +class _ExactLevelFilter(logging.Filter): + def __init__(self, level: int) -> None: + super().__init__() + self._level = level + + def filter(self, record: logging.LogRecord) -> bool: + return record.levelno == self._level + + +class _TeeStream: + def __init__(self, stream: TextIO, logger: logging.Logger, level: int) -> None: + self._stream = stream + self._logger = logger + self._level = level + self._buffer = "" + + def write(self, text: str) -> int: + self._stream.write(text) + self._stream.flush() + self._buffer += text + while "\n" in self._buffer: + line, self._buffer = self._buffer.split("\n", 1) + line = line.rstrip() + if line: + self._logger.log(self._level, line) + return len(text) + + def flush(self) -> None: + self._stream.flush() + line = self._buffer.rstrip() + if line: + self._logger.log(self._level, line) + self._buffer = "" + + def isatty(self) -> bool: + return self._stream.isatty() + + +def _make_handler(path: Path, level: int, *, max_bytes: int, backup_count: int) -> RotatingFileHandler: + handler = RotatingFileHandler( + path, + maxBytes=max_bytes, + backupCount=backup_count, + encoding="utf-8", + ) + handler.setLevel(level) + handler.addFilter(_ExactLevelFilter(level)) + handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s")) + return handler + + +def configure_tracker_file_logging( + log_dir: str | Path = DEFAULT_LOG_DIR, + *, + max_bytes: int = DEFAULT_LOG_MAX_BYTES, + backup_count: int = DEFAULT_LOG_BACKUP_COUNT, + tee_stdio: bool = True, +) -> Path: + """Configure rotatable info/warning/error log files and return the directory.""" + + path = Path(log_dir).expanduser() + path.mkdir(parents=True, exist_ok=True) + + logger = logging.getLogger(TRACKER_LOGGER_NAME) + logger.setLevel(logging.INFO) + logger.propagate = False + logger.handlers.clear() + logger.addHandler(_make_handler(path / "info.log", logging.INFO, max_bytes=max_bytes, backup_count=backup_count)) + logger.addHandler(_make_handler(path / "warning.log", logging.WARNING, max_bytes=max_bytes, backup_count=backup_count)) + logger.addHandler(_make_handler(path / "error.log", logging.ERROR, max_bytes=max_bytes, backup_count=backup_count)) + + if tee_stdio: + if not isinstance(sys.stdout, _TeeStream): + sys.stdout = _TeeStream(sys.stdout, logger, logging.INFO) # type: ignore[assignment] + if not isinstance(sys.stderr, _TeeStream): + sys.stderr = _TeeStream(sys.stderr, logger, logging.ERROR) # type: ignore[assignment] + + return path + + +def tracker_logger() -> logging.Logger: + return logging.getLogger(TRACKER_LOGGER_NAME) diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index d28ed8a..5fb2b06 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -26,6 +26,7 @@ import hashlib import itertools import json import os +import select import socketserver import sqlite3 import tarfile @@ -46,8 +47,9 @@ from .wallet_proof import binding_message, verify_wallet_signature from .billing import DEFAULT_BILLING_DB_PATH, BillingLedger from .calibration import DEFAULT_CALIBRATION_DB_PATH, ToplocCalibrationStore from .hf_pricing import DEFAULT_HF_PRICING_LOG_DB_PATH, HfPricingLog, refresh_preset_price -from .gossip import NodeGossip -from .model_files import files_for_layer_range, snapshot_dir_for_repo +from .gossip import NodeGossip +from .logging_setup import tracker_logger +from .model_files import files_for_layer_range, snapshot_dir_for_repo from .raft import RaftNode @@ -2092,7 +2094,7 @@ def _registration_ban_error(contracts: Any | None, wallet_address: str | None) - return None -def _tracker_log( +def _tracker_log( server: "_TrackerHTTPServer", level: str, message: str, @@ -2100,11 +2102,18 @@ def _tracker_log( stdout: bool = True, update_console_key: str | None = None, **fields: Any, -) -> None: - event = { - "ts": time.time(), - "level": level, - "message": message, +) -> None: + log_level = { + "debug": 10, + "info": 20, + "warn": 30, + "warning": 30, + "error": 40, + }.get(level.lower(), 20) + event = { + "ts": time.time(), + "level": level, + "message": message, "fields": { key: value for key, value in fields.items() @@ -2125,12 +2134,13 @@ def _tracker_log( break if not updated: server.console_events.append(event) - else: - server.console_events.append(event) - if stdout: - extras = " ".join(f"{key}={value}" for key, value in event["fields"].items()) - suffix = f" {extras}" if extras else "" - print(f"[tracker] {level}: {message}{suffix}", flush=True) + else: + server.console_events.append(event) + extras = " ".join(f"{key}={value}" for key, value in event["fields"].items()) + suffix = f" {extras}" if extras else "" + tracker_logger().log(log_level, f"{message}{suffix}") + if stdout: + print(f"[tracker] {level}: {message}{suffix}", flush=True) @dataclass @@ -2182,10 +2192,14 @@ def _request_proxy_cancel(server: "_TrackerHTTPServer", request_id: str) -> bool return True -def _set_upstream_read_timeout(upstream: Any, timeout: float) -> None: +def _upstream_socket(upstream: Any) -> Any | None: fp = getattr(upstream, "fp", None) raw = getattr(fp, "raw", None) if fp is not None else None - sock = getattr(raw, "_sock", None) if raw is not None else None + return getattr(raw, "_sock", None) if raw is not None else None + + +def _set_upstream_read_timeout(upstream: Any, timeout: float | None) -> None: + sock = _upstream_socket(upstream) if sock is not None: sock.settimeout(timeout) @@ -3206,7 +3220,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): upstream = upstream_result[0] with proxy_ctx.upstream_lock: proxy_ctx.upstream = upstream - _set_upstream_read_timeout(upstream, 0.5) + upstream_sock = _upstream_socket(upstream) + if upstream_sock is not None: + _set_upstream_read_timeout(upstream, None) + else: + _set_upstream_read_timeout(upstream, 0.5) _tracker_log(server, "info", "proxy connected", request_id=request_id, target_url=target_url) except urllib.error.HTTPError as exc: # Relay error status + body from node @@ -3266,6 +3284,10 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): while True: if proxy_ctx.cancel_event.is_set(): break + if upstream_sock is not None: + readable, _, _ = select.select([upstream_sock], [], [], 0.5) + if not readable: + continue try: line = upstream.readline() except TimeoutError: diff --git a/tests/test_tracker_logging.py b/tests/test_tracker_logging.py new file mode 100644 index 0000000..6d54e7b --- /dev/null +++ b/tests/test_tracker_logging.py @@ -0,0 +1,67 @@ +import logging +import sys + +from meshnet_tracker.logging_setup import configure_tracker_file_logging, tracker_logger + + +def test_tracker_file_logging_writes_separate_level_files(tmp_path): + original_stdout = sys.stdout + original_stderr = sys.stderr + try: + log_dir = configure_tracker_file_logging(tmp_path, tee_stdio=False) + logger = tracker_logger() + + logger.info("info-event") + logger.warning("warning-event") + logger.error("error-event") + for handler in logger.handlers: + handler.flush() + + assert (log_dir / "info.log").read_text().count("info-event") == 1 + assert "warning-event" not in (log_dir / "info.log").read_text() + assert "error-event" not in (log_dir / "info.log").read_text() + + assert "warning-event" in (log_dir / "warning.log").read_text() + assert "info-event" not in (log_dir / "warning.log").read_text() + assert "error-event" not in (log_dir / "warning.log").read_text() + + assert "error-event" in (log_dir / "error.log").read_text() + assert "info-event" not in (log_dir / "error.log").read_text() + assert "warning-event" not in (log_dir / "error.log").read_text() + finally: + sys.stdout = original_stdout + sys.stderr = original_stderr + + +def test_tracker_file_logging_tees_stdio_and_rotates(tmp_path): + original_stdout = sys.stdout + original_stderr = sys.stderr + try: + log_dir = configure_tracker_file_logging( + tmp_path, + max_bytes=120, + backup_count=1, + ) + + print("stdout goes to info", flush=True) + print("stderr goes to error", file=sys.stderr, flush=True) + for handler in tracker_logger().handlers: + handler.flush() + + assert "stdout goes to info" in (log_dir / "info.log").read_text() + assert "stderr goes to error" in (log_dir / "error.log").read_text() + + for index in range(12): + tracker_logger().info("rotating-info-line-%02d", index) + for handler in tracker_logger().handlers: + handler.flush() + + assert (log_dir / "info.log.1").exists() + finally: + sys.stdout = original_stdout + sys.stderr = original_stderr + logger = tracker_logger() + for handler in logger.handlers: + handler.close() + logger.handlers.clear() + logger.setLevel(logging.NOTSET) diff --git a/tests/test_tracker_routing.py b/tests/test_tracker_routing.py index 7e17cae..a6e60dc 100644 --- a/tests/test_tracker_routing.py +++ b/tests/test_tracker_routing.py @@ -502,6 +502,75 @@ def test_tracker_logs_stream_progress_before_request_completes(): node_thread.join(timeout=1.0) +def test_tracker_stream_survives_idle_gap_between_sse_chunks(): + first_chunk_sent = threading.Event() + + class IdleStreamingChatHandler(http.server.BaseHTTPRequestHandler): + def log_message(self, format, *args): + pass + + def do_POST(self): + if self.path != "/v1/chat/completions": + self.send_response(404) + self.end_headers() + return + self.rfile.read(int(self.headers.get("Content-Length", 0))) + self.send_response(200) + self.send_header("Content-Type", "text/event-stream; charset=utf-8") + self.end_headers() + first = json.dumps({ + "choices": [{"delta": {"content": "hello"}}], + }).encode() + second = json.dumps({ + "choices": [{"delta": {"content": " world"}}], + }).encode() + self.wfile.write(b"data: " + first + b"\n\n") + self.wfile.flush() + first_chunk_sent.set() + time.sleep(1.0) + self.wfile.write(b"data: " + second + b"\n\n") + self.wfile.write(b"data: [DONE]\n\n") + self.wfile.flush() + + node = http.server.HTTPServer(("127.0.0.1", 0), IdleStreamingChatHandler) + node_thread = threading.Thread(target=node.serve_forever, daemon=True) + node_thread.start() + tracker = TrackerServer(heartbeat_timeout=60.0) + tracker_port = tracker.start() + response = None + try: + _post_json( + f"http://127.0.0.1:{tracker_port}/v1/nodes/register", + {"endpoint": f"http://127.0.0.1:{node.server_address[1]}", + "model": "idle-stream-model", "num_layers": 1, + "shard_start": 0, "shard_end": 0, + "hardware_profile": {}, "score": 1.0}, + ) + req = urllib.request.Request( + f"http://127.0.0.1:{tracker_port}/v1/chat/completions", + data=json.dumps({ + "model": "idle-stream-model", + "stream": True, + "messages": [{"role": "user", "content": "hi"}], + }).encode(), + headers={"Content-Type": "application/json"}, + method="POST", + ) + response = urllib.request.urlopen(req, timeout=3.0) + assert response.readline().startswith(b"data:") + assert first_chunk_sent.wait(timeout=1.0) + + remaining = response.read().splitlines() + assert b"data: [DONE]" in remaining + finally: + if response is not None: + response.close() + tracker.stop() + node.shutdown() + node.server_close() + node_thread.join(timeout=1.0) + + def test_tracker_dashboard_can_cancel_inflight_proxy(): chunk_sent = threading.Event() release = threading.Event() From a0b37ad1b9f850bf00bc0ddd8c42f85f943d04a4 Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Tue, 7 Jul 2026 22:13:12 +0300 Subject: [PATCH 2/3] store sessions in the DB --- .gitignore | 3 + packages/tracker/meshnet_tracker/accounts.py | 32 +++++- .../tracker/meshnet_tracker/dashboard.html | 6 +- packages/tracker/meshnet_tracker/server.py | 104 +++++++++++++----- tests/test_accounts.py | 65 +++++++++++ 5 files changed, 176 insertions(+), 34 deletions(-) diff --git a/.gitignore b/.gitignore index 26860cd..1820726 100644 --- a/.gitignore +++ b/.gitignore @@ -23,3 +23,6 @@ dist/ # Local tracker/node sqlite databases (never commit runtime state) *.sqlite *.sqlite3 +logs/tracker/error.log +logs/tracker/info.log +logs/tracker/warning.log diff --git a/packages/tracker/meshnet_tracker/accounts.py b/packages/tracker/meshnet_tracker/accounts.py index 626ce6f..68099c5 100644 --- a/packages/tracker/meshnet_tracker/accounts.py +++ b/packages/tracker/meshnet_tracker/accounts.py @@ -8,7 +8,8 @@ regular user. Mutations are append-only events with unique ids — the same replication model as ``BillingLedger`` — so accounts and API keys converge across the tracker hive via gossip, and every dashboard can serve registration/login. -Sessions are deliberately local to each tracker (bearer tokens in memory). +Sessions are local to each tracker and persisted so dashboard cookies survive +tracker restarts. """ from __future__ import annotations @@ -115,6 +116,8 @@ class AccountStore: "account_id": account_id, "expires": time.time() + SESSION_TTL, } + self._dirty = True + self.save_to_db() return token def session_account(self, token: str | None) -> dict | None: @@ -134,7 +137,9 @@ class AccountStore: if not token: return with self._lock: - self._sessions.pop(token, None) + if self._sessions.pop(token, None) is not None: + self._dirty = True + self.save_to_db() # ---- API keys ---- @@ -271,6 +276,10 @@ class AccountStore: "CREATE TABLE IF NOT EXISTS account_events " "(event_id TEXT PRIMARY KEY, payload TEXT NOT NULL, ts REAL NOT NULL)" ) + con.execute( + "CREATE TABLE IF NOT EXISTS account_sessions " + "(token TEXT PRIMARY KEY, account_id TEXT NOT NULL, expires REAL NOT NULL)" + ) con.commit() con.close() @@ -279,6 +288,10 @@ class AccountStore: rows = con.execute( "SELECT payload FROM account_events ORDER BY ts, event_id" ).fetchall() + session_rows = con.execute( + "SELECT token, account_id, expires FROM account_sessions WHERE expires >= ?", + (time.time(),), + ).fetchall() con.close() with self._lock: for (payload,) in rows: @@ -288,6 +301,11 @@ class AccountStore: continue if event.get("id") not in self._seen_event_ids: self._apply_locked(event) + self._sessions = { + token: {"account_id": account_id, "expires": float(expires)} + for token, account_id, expires in session_rows + if account_id in self._accounts + } self._dirty = False def save_to_db(self) -> None: @@ -297,11 +315,21 @@ class AccountStore: if not self._dirty: return events = list(self._event_log) + sessions = [ + (token, session["account_id"], float(session["expires"])) + for token, session in self._sessions.items() + if session["expires"] >= time.time() + ] self._dirty = False con = sqlite3.connect(self._db_path) # type: ignore[arg-type] con.executemany( "INSERT OR IGNORE INTO account_events (event_id, payload, ts) VALUES (?, ?, ?)", [(e["id"], json.dumps(e), float(e.get("ts", 0.0))) for e in events], ) + con.execute("DELETE FROM account_sessions") + con.executemany( + "INSERT INTO account_sessions (token, account_id, expires) VALUES (?, ?, ?)", + sessions, + ) con.commit() con.close() diff --git a/packages/tracker/meshnet_tracker/dashboard.html b/packages/tracker/meshnet_tracker/dashboard.html index faa07fb..b22cfec 100644 --- a/packages/tracker/meshnet_tracker/dashboard.html +++ b/packages/tracker/meshnet_tracker/dashboard.html @@ -224,6 +224,7 @@

Model usage (RPM)

loading…

Call wall

loading...
+

Chat / inference