#!/usr/bin/env python3 """tests/integration/nightly_run.py — the nightly integration gate. Real veloxd + tools/testserver, 50 concurrent downloads mixing hostile testserver modes, every completed file's SHA-256 checked against testserver's own /sha256/ route, and veloxd's open-FD count checked flat across the run. See tests/integration/README.md#nightly-integration-run for what this proves and how each failure mode was forced once to prove the check is not vacuous. Standard library only (matches tools/testserver's own constraint) — this runs on whatever CI image happens to have Python and does not need its own dependency install step. Usage: python3 tests/integration/nightly_run.py --veloxd /path/to/build/dev/bin/veloxd Exit codes: 0 all good, 1 a task failed or a checksum/FD assertion tripped, 2 usage/environment error (daemon or testserver would not start). """ from __future__ import annotations import argparse import ctypes import hashlib import json import os import re import shutil import signal import socket import sqlite3 import subprocess import sys import tempfile import time import urllib.request from dataclasses import dataclass from typing import NoReturn REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) TESTSERVER = os.path.join(REPO_ROOT, "tools", "testserver", "testserver.py") _PR_SET_PDEATHSIG = 1 def _die_with_parent() -> None: """preexec_fn for every child we spawn: ask the kernel to SIGKILL this child the moment its parent (this harness) goes away, by any means — including this process's own SIGKILL, which no `finally:` block ever runs for. Verified live: a prior run left a stranded veloxd running under systemd --user for 4h40m after the harness was killed -9'd, because cleanup lived only in `finally:`. Proven fixed by SIGKILL-ing a running harness mid-download and confirming both children exit with it — see tests/integration/README.md. ctypes, not python-prctl: this is stdlib-only by the same rule as testserver.py, and it's one syscall. """ libc = ctypes.CDLL("libc.so.6", use_errno=True) libc.prctl(_PR_SET_PDEATHSIG, signal.SIGKILL, 0, 0, 0) # --- self-test hooks -------------------------------------------------------------- # Each of these deliberately breaks one assertion so the gate can be proven red on # demand, without hand-editing the script. `tests/integration/README.md` records one # transcript per flag from the session that added this file; re-run any of them # whenever the corresponding assertion is touched, to re-prove it still catches the # failure it claims to. FORCE_BAD_HASH = os.environ.get("VDM_NIGHTLY_FORCE_BAD_HASH") == "1" FORCE_FD_LEAK = os.environ.get("VDM_NIGHTLY_FORCE_FD_LEAK") == "1" FORCE_HOSTILE_STALL = os.environ.get("VDM_NIGHTLY_FORCE_HOSTILE_STALL") == "1" @dataclass class TaskPlan: idx: int mode: str size: str task_id: str | None = None url_override: str | None = None # self-test hook only; see FORCE_HOSTILE_STALL @dataclass class RpcClient: sock: socket.socket _next_id: int = 1 _buf: bytes = b"" def call(self, method: str, params: dict, timeout: float = 10.0) -> dict: req = {"jsonrpc": "2.0", "id": self._next_id, "method": method, "params": params} self._next_id += 1 self.sock.sendall((json.dumps(req) + "\n").encode()) self.sock.settimeout(timeout) while b"\n" not in self._buf: chunk = self.sock.recv(65536) if not chunk: raise ConnectionError(f"veloxd closed the connection mid-{method}") self._buf += chunk line, _, self._buf = self._buf.partition(b"\n") return json.loads(line.decode()) def die(msg: str, code: int = 2) -> NoReturn: print(f"nightly_run: {msg}", file=sys.stderr) sys.exit(code) def wait_for(pred, timeout: float, interval: float = 0.1, what: str = "condition"): deadline = time.monotonic() + timeout while time.monotonic() < deadline: if pred(): return time.sleep(interval) raise TimeoutError(f"timed out waiting for {what}") def start_testserver(log_path: str) -> tuple[subprocess.Popen, int]: proc = subprocess.Popen( [sys.executable, TESTSERVER, "--port", "0"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, preexec_fn=_die_with_parent, start_new_session=True, ) log = open(log_path, "w") port = None deadline = time.monotonic() + 10 while time.monotonic() < deadline: line = proc.stdout.readline() if not line: break log.write(line) log.flush() m = re.search(r"\b(\d{2,5})\b", line) if m: port = int(m.group(1)) break if port is None: proc.terminate() die("testserver never printed its ephemeral port") # Drain the rest of stdout into the log in the background so the pipe never fills. def _drain(): for line in proc.stdout: log.write(line) log.flush() import threading threading.Thread(target=_drain, daemon=True).start() return proc, port def start_veloxd(veloxd_bin: str, env: dict, sock_path: str, log_path: str) -> subprocess.Popen: proc = subprocess.Popen( [veloxd_bin], env=env, stdout=open(log_path, "a"), stderr=subprocess.STDOUT, preexec_fn=_die_with_parent, start_new_session=True, ) try: wait_for(lambda: os.path.exists(sock_path), timeout=10, what="veloxd's UDS socket") except TimeoutError: proc.terminate() die( "veloxd never created its socket — check the log at " + log_path, ) return proc def seed_settings(db_path: str, allowed_root: str, max_concurrent: int) -> None: """Pre-seed saveTo.allowedRoots (and raise the concurrency cap) directly in the DB. settings.set is not implemented in this build yet (verified live: it returns -32603 "not implemented in this build"), so this is the only way in — not a workaround for a wrong contract, just the one entry point that currently exists. """ conn = sqlite3.connect(db_path) try: for key, value in ( ("saveTo.allowedRoots", json.dumps([allowed_root])), ("saveTo.defaultDir", json.dumps(allowed_root)), ("connection.maxConcurrentDownloads", json.dumps(max_concurrent)), ("connection.maxActiveSegments", json.dumps(max_concurrent * 2)), ): conn.execute( "INSERT INTO settings(key, value) VALUES(?, ?) " "ON CONFLICT(key) DO UPDATE SET value = excluded.value", (key, value), ) conn.commit() finally: conn.close() def open_fd_count(pid: int) -> int: return len(os.listdir(f"/proc/{pid}/fd")) def build_plan(n: int, hostile_stall: bool) -> list[TaskPlan]: # 60% plain, mixed in with the three named hostile modes from the brief. modes = ( ["plain"] * 30 + ["flaky-reset"] * 8 + ["throttled"] * 6 + ["no-range"] * 6 ) assert len(modes) == n plan = [] for i, mode in enumerate(modes): # Small enough that even `throttled` (1 MiB/s default) finishes well inside # the budget. Sizes repeat across tasks (only 8 distinct values) — that's # fine for testserver, whose content is a function of (seed, path, size), # but every task must still get its own `filename` override below: two # tasks that share both mode and size share a URL, and without an override # they'd share a destination filename too and race each other's rename — # verified live, see tests/integration/README.md. size = f"{256 + (i % 8) * 16}K" t = TaskPlan(idx=i, mode=mode, size=size) if hostile_stall and i == 0: # Self-test hook, verified live: the first attempt at this used # testserver's 416-always mode, expecting it to never yield a 2xx to a # Range probe — but veloxd correctly falls back to a plain full GET on a # 416, so that task completed fine and the hook proved nothing. # TEST-NET-1 (RFC 5737) is unassigned and unroutable, so a connection # attempt to it hangs until veloxd's own connection.timeoutSec — also # verified live: it lands the task in 'failed' (retryable timeout) well # inside this script's own --timeout, so this hook actually exercises the # "reached a terminal state that isn't complete" branch below, not the # "never reached one at all" branch — both are real ways a hostile source # must not be allowed to read as a silent pass. See # VDM_NIGHTLY_FORCE_HOSTILE_STALL in the README. t.url_override = "http://192.0.2.1:1/black-hole" plan.append(t) return plan def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--veloxd", default=os.path.join(REPO_ROOT, "build", "dev", "bin", "veloxd")) ap.add_argument("--tasks", type=int, default=50) ap.add_argument("--timeout", type=float, default=120.0, help="seconds to wait for all tasks") args = ap.parse_args() if not os.path.isfile(args.veloxd): die(f"veloxd binary not found at {args.veloxd} — build it first (cmake --build ...)") # $TMPDIR, not the caller's scratch dir: AF_UNIX paths are capped at ~108 bytes # (sun_path) and a deeply-nested scratch/session path blows that budget — verified # live ("File name too long") before switching to mkdtemp() here. base = tempfile.mkdtemp(prefix="vdm-nightly-") print(f"nightly_run: workdir {base}") ok = True daemon = None ts = None try: for d in ("runtime", "data", "config", "downloads"): os.makedirs(os.path.join(base, d), exist_ok=True) os.chmod(os.path.join(base, "runtime"), 0o700) downloads_dir = os.path.join(base, "downloads") env = os.environ.copy() env["XDG_RUNTIME_DIR"] = os.path.join(base, "runtime") env["XDG_DATA_HOME"] = os.path.join(base, "data") env["XDG_CONFIG_HOME"] = os.path.join(base, "config") sock_path = os.path.join(base, "runtime", "velox", "velox.sock") veloxd_log = os.path.join(base, "veloxd.log") # Warm-up start: apply migrations and create velox.db, nothing else. warm = start_veloxd(args.veloxd, env, sock_path, veloxd_log) warm.terminate() warm.wait(timeout=5) db_path = os.path.join(base, "data", "velox", "velox.db") if not os.path.isfile(db_path): die(f"veloxd's warm-up start never created {db_path}") seed_settings(db_path, downloads_dir, max_concurrent=args.tasks) ts, port = start_testserver(os.path.join(base, "testserver.log")) daemon = start_veloxd(args.veloxd, env, sock_path, veloxd_log) s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) s.connect(sock_path) rpc = RpcClient(sock=s) hello = rpc.call( "session.hello", {"clientType": "cli", "clientName": "nightly-run", "protocolVersion": "1.0.0"}, ) if "error" in hello: die(f"session.hello failed: {hello['error']}") baseline_fds = open_fd_count(daemon.pid) print(f"nightly_run: veloxd baseline FDs = {baseline_fds}") plan = build_plan(args.tasks, hostile_stall=FORCE_HOSTILE_STALL) for t in plan: url = t.url_override or f"http://127.0.0.1:{port}/{t.mode}/file/{t.size}" # Every task gets its own filename: several tasks share a (mode, size) and # therefore a URL, and without this override they'd race each other for the # same destination path — verified live (see README). filename = f"task-{t.idx:02d}-{t.mode}-{t.size}" add = rpc.call( "download.add", {"url": url, "saveDir": downloads_dir, "segments": 4, "filename": filename}, ) if "error" in add: ok = False print(f"task {t.idx} ({t.mode}, {t.size}): download.add failed: {add['error']}") continue t.task_id = add["result"]["taskId"] # --- poll every task to a terminal state ------------------------------------- terminal = {"complete", "failed", "cancelled"} by_id = {t.task_id: t for t in plan if t.task_id} results: dict[str, dict] = {} deadline = time.monotonic() + args.timeout while time.monotonic() < deadline and len(results) < len(by_id): lst = rpc.call("download.list", {"limit": 5000}) if "error" in lst: die(f"download.list failed: {lst['error']}") for item in lst["result"]["items"]: if item["taskId"] in by_id and item["state"] in terminal: results[item["taskId"]] = item time.sleep(0.5) missing = set(by_id) - set(results) if missing: ok = False print(f"FAIL: {len(missing)}/{len(by_id)} task(s) never reached a terminal state " f"within {args.timeout}s: {sorted(missing)}") # --- checksum: every completed file against testserver's own /sha256/ route - for task_id, item in results.items(): t = by_id[task_id] if item["state"] != "complete": ok = False err = item.get("error") print(f"FAIL: task {t.idx} ({t.mode}, {t.size}) ended in state " f"'{item['state']}', error={err}") continue path = os.path.join(item["saveDir"], item["filename"]) if not os.path.isfile(path): ok = False print(f"FAIL: task {t.idx} ({t.mode}, {t.size}) reported complete but " f"{path} does not exist") continue h = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(1 << 20), b""): h.update(chunk) actual = h.hexdigest() expected_url = f"http://127.0.0.1:{port}/{t.mode}/sha256/{t.size}" with urllib.request.urlopen(expected_url, timeout=5) as resp: expected = json.loads(resp.read())["sha256"] if FORCE_BAD_HASH and t.idx == 0: # Self-test hook: corrupt the comparison itself, not the file, so a # clean re-run without the env var proves nothing was left broken. expected = "0" * 64 if actual != expected: ok = False print(f"FAIL: task {t.idx} ({t.mode}, {t.size}) checksum mismatch: " f"got {actual}, testserver says {expected}") # --- FD leak: veloxd's own /proc//fd count returns to baseline ---------- # One settle window for the last connections/segment threads to actually # close, then compare. A real leak grows roughly with the number of tasks # that touched a socket, so a fixed small tolerance (not "not more than # double") is the point of this check. time.sleep(2.0) final_fds = open_fd_count(daemon.pid) if FORCE_FD_LEAK: final_fds += 25 # self-test hook — see VDM_NIGHTLY_FORCE_FD_LEAK in the README # Verified live across several 50-task runs: a clean run settles at baseline+3 # to baseline+5 (thread-pool and libcurl connection-cache bookkeeping that # doesn't unwind instantly), so 4 alone was noisy enough to fail a clean run by # chance. 10 still catches a real leak by a wide margin — a run with tasks stuck # retrying showed baseline+7 before those tasks even reached their timeout. tolerance = 10 print(f"nightly_run: veloxd FDs after run = {final_fds} (baseline {baseline_fds}, " f"tolerance {tolerance})") if final_fds > baseline_fds + tolerance: ok = False print(f"FAIL: veloxd leaked file descriptors: {baseline_fds} -> {final_fds} " f"(tolerance {tolerance})") print(f"nightly_run: {len(results)}/{len(by_id)} tasks reached a terminal state, " f"{'OK' if ok else 'FAILURES ABOVE'}") return 0 if ok else 1 finally: # Belt-and-suspenders on top of PR_SET_PDEATHSIG: a graceful path here still # prefers SIGTERM before SIGKILL. killpg (not kill) because start_new_session # made each child its own process group leader, so this also reaps anything # it spawned (testserver's --drain thread aside, which is ours, not a child). for proc in (daemon, ts): if proc is not None and proc.poll() is None: try: os.killpg(proc.pid, signal.SIGTERM) except ProcessLookupError: continue try: proc.wait(timeout=5) except subprocess.TimeoutExpired: try: os.killpg(proc.pid, signal.SIGKILL) except ProcessLookupError: pass shutil.rmtree(base, ignore_errors=True) if __name__ == "__main__": sys.exit(main())