vdm/tests/integration/nightly_run.py

#!/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/<pid>/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())