diff --git a/README.md b/README.md index 485c7fdf..d2f9d655 100644 --- a/README.md +++ b/README.md @@ -67,6 +67,8 @@ See the [profile registry](docs/profiles/README.md) for recipe identities and ev **Research-only — 16K context, single observation.** The SparkCache-enabled profile recorded 2,371 tok/s prefill and 36.06 tok/s sustained C1 decode on random tokens. No A/B baseline has been completed. +C4 and C8 were capacity-limited and are omitted rather than reported as +throughput results. | Profile | Prefill | C1 decode | C8 decode | Highest valid decode | Coding peak | |---|---:|---:|---:|---:|---:| diff --git a/docs/DIRECT_FABRIC_IMAGE_ARCHIVE_FANOUT.md b/docs/DIRECT_FABRIC_IMAGE_ARCHIVE_FANOUT.md new file mode 100644 index 00000000..a77bab6d --- /dev/null +++ b/docs/DIRECT_FABRIC_IMAGE_ARCHIVE_FANOUT.md @@ -0,0 +1,170 @@ +# Distribute an image archive through the direct fabric + +Status: **implemented** with GPU-free command and state-transition coverage. +Direct-fabric throughput is **research-only**. Live four-rank image import is +**unsupported** by retained repository evidence. + +`scripts/fanout_image_archive.py` downloads one immutable image archive on a +configured seed rank. It then forwards the verified file through three direct +links so the management network carries one download instead of four. + +The tool reads ranks, cycle edges, direct addresses, and SSH targets from a +validated SparkRing site file. It does not contain deployment hostnames or +assume that rank numbers map to particular addresses. + +## Interface and evidence + +| Mode | Remote effect | Output schema | +|---|---|---| +| default plan | none | `sparkring-image-archive-fabric-plan/v1` | +| `--verify` | reads final-file type, SHA-256, and byte count on every rank | `sparkring-image-archive-fabric-verification/v1` | +| `--execute --create-only` | creates, resumes, and verifies archive files | `sparkring-image-archive-fabric-receipt/v1` | +| `--execute` | creates and verifies archive files, then imports and inspects the named image on every rank | `sparkring-image-archive-fabric-receipt/v1` | + +`--execute` requires `--confirmation FANOUT_IMAGE_ARCHIVE`. Every remote +operation has an overall `--timeout`, and every SSH or rsync connection uses +`--connect-timeout`. The defaults are 7,200 and 45 seconds respectively. +`--output` writes the same JSON document printed to standard output. + +Inspect the complete command interface without contacting a host: + +```bash +python scripts/fanout_image_archive.py --help +``` + +## Requirements + +- A four-rank SparkRing site file whose direct edges form one cycle. +- Passwordless SSH from the operator to each rank's management target. +- Passwordless SSH between adjacent ranks on their direct-link addresses. +- `curl`, `sha256sum`, `rsync`, and Docker on every applicable host. +- One HTTP or HTTPS archive URL without embedded credentials or query tokens. +- An explicit absolute target directory dedicated to image archives. + +The target directory must contain at least three components below `/` so a +broad path such as `/var/tmp` is rejected. +The archive name must be one filename without separators. Final and partial +files are constructed beneath that directory. + +## Plan + +Planning is the default and performs no remote operation: + +```bash +python scripts/fanout_image_archive.py \ + --site /secure/site.yaml \ + --source-url https://images.example/runtime-arm64.tar.zst \ + --archive-name runtime-arm64.tar.zst \ + --expected-sha256 <64-lowercase-hex> \ + --target-directory /var/lib/sparkring/images \ + --seed-rank 0 \ + --create-only +``` + +The JSON plan names every management SSH command and every fabric hop. It also +records the expected archive SHA-256, target directory, selected path through +the cycle, MTU, link speed, optional image reference, and expected image ID. The +default first hop is the seed's lowest-numbered direct neighbour. Use +`--first-hop-rank` to select the other direction around the cycle. + +## Verify existing files + +Verification is read-only. It requires the final archive to exist with the +expected digest on every rank: + +```bash +python scripts/fanout_image_archive.py \ + --site /secure/site.yaml \ + --archive-name runtime-arm64.tar.zst \ + --expected-sha256 <64-lowercase-hex> \ + --target-directory /var/lib/sparkring/images \ + --verify \ + --output ./evidence/image-archive-verification.json +``` + +## Create files without importing the image + +Execution requires an explicit confirmation token. `--create-only` downloads, +forwards, and verifies the archive without invoking Docker image import: + +```bash +python scripts/fanout_image_archive.py \ + --site /secure/site.yaml \ + --source-url https://images.example/runtime-arm64.tar.zst \ + --archive-name runtime-arm64.tar.zst \ + --expected-sha256 <64-lowercase-hex> \ + --target-directory /var/lib/sparkring/images \ + --seed-rank 0 \ + --create-only \ + --execute \ + --confirmation FANOUT_IMAGE_ARCHIVE \ + --output ./evidence/image-archive-fanout.json +``` + +## Import one image on every rank + +Omit `--create-only`, provide the image reference stored in the archive, and +require its local image ID: + +```bash +python scripts/fanout_image_archive.py \ + --site /secure/site.yaml \ + --source-url https://images.example/runtime-arm64.tar.zst \ + --archive-name runtime-arm64.tar.zst \ + --expected-sha256 <64-lowercase-hex> \ + --target-directory /var/lib/sparkring/images \ + --image registry.example/runtime@sha256: \ + --expected-image-id sha256: \ + --execute \ + --confirmation FANOUT_IMAGE_ARCHIVE \ + --output ./evidence/image-import.json +``` + +An archive created with `docker image save sha256:` may contain no +repository tag. After loading such an archive, the tool verifies that the +expected image ID exists locally before applying the requested tag. An +existing tag that points at another image remains a conflict. + +## File and interruption behavior + +- An exact final file is reused without downloading or transferring it. +- A final path with another digest, or a non-regular file at that path, + rejects the operation. The tool never overwrites it. +- Downloads and fabric transfers write `..partial` beneath the + target directory. +- `curl --continue-at -` resumes the seed download when the server supports + byte ranges. +- `rsync --partial --append-verify` preserves and verifies an interrupted hop. +- A verified partial file is hard-linked to the final name. Hard-link creation + is atomic and refuses an existing final path. +- Every destination computes SHA-256 after its hop. The evidence receipt names + whether a rank reused an existing file or received it from another rank. +- A command error or timeout stops subsequent actions. Exact final files and + bounded partial files already created on earlier ranks remain in place; the + utility performs no distributed rollback. A retry reuses exact finals and + resumes eligible partials. +- Image imports run in rank order after all four archives verify. If an import + stops partway through, already imported images remain. A retry accepts only + the requested tag mapped to the expected image ID. + +The transfer command binds SSH to the source rank's address on the selected +cycle edge and connects to the peer address from the same edge. Management SSH +is used only to start and inspect operations. + +## Limitations + +The utility supports exactly four ranks whose configured direct edges form one +cycle. It chooses one three-hop path around that cycle and executes hops +sequentially; it does not broadcast, stripe one archive across links, or use +both directions concurrently. + +The utility verifies archive and optional image identity. It does not validate +model serving, collective transport, SparkCache behavior, available disk +capacity, source-server throughput, or achieved fabric throughput. Image +archives are large, and download, local storage, hashing, Docker import, or one +slow hop can dominate elapsed time even when every direct link is 200 Gb/s. + +Host-key enrollment, SSH authorization, remote tool installation, archive +creation, and cleanup policy remain operator responsibilities. The source URL +must use HTTP or HTTPS and cannot contain credentials, query parameters, or a +fragment. The tool does not publish an image or upload evidence. diff --git a/scripts/config/README.md b/scripts/config/README.md index 76291c21..57f63653 100644 --- a/scripts/config/README.md +++ b/scripts/config/README.md @@ -173,6 +173,14 @@ commands require the same six read-only, SHA-bound runtime mounts for the vLLM capture overlays and SIRCL implementation. The control leaves their activation variables unset. +## Direct-fabric image archive distribution + +The rank and edge data in a validated four-rank site file can drive +`scripts/fanout_image_archive.py`. The utility downloads one checksum-bound +archive on a seed rank and forwards it through adjacent direct links. See +[`docs/DIRECT_FABRIC_IMAGE_ARCHIVE_FANOUT.md`](../../docs/DIRECT_FABRIC_IMAGE_ARCHIVE_FANOUT.md) +for planning, verification, create-only, and image-import behavior. + ## Safety Copying or validating a template is **OFFLINE**. Contacting configured ranks diff --git a/scripts/fanout_image_archive.py b/scripts/fanout_image_archive.py new file mode 100644 index 00000000..49edf547 --- /dev/null +++ b/scripts/fanout_image_archive.py @@ -0,0 +1,782 @@ +#!/usr/bin/env python3 +"""Distribute one verified image archive through the configured direct fabric. + +Planning is the default and performs no remote work. ``--verify`` reads the +archive state on every rank. ``--execute`` seeds one rank, forwards the exact +archive through a direct-link chain, verifies every hop, and imports the image +unless ``--create-only`` is set. +""" + +from __future__ import annotations + +import argparse +import json +import re +import shlex +import subprocess +from dataclasses import asdict, dataclass +from pathlib import Path, PurePosixPath +from typing import Any, Callable, Sequence +from urllib.parse import urlsplit + +from sparkring_site import SiteConfigError, load_site + + +PLAN_SCHEMA = "sparkring-image-archive-fabric-plan/v1" +RECEIPT_SCHEMA = "sparkring-image-archive-fabric-receipt/v1" +VERIFY_SCHEMA = "sparkring-image-archive-fabric-verification/v1" +CONFIRMATION = "FANOUT_IMAGE_ARCHIVE" +_SHA256 = re.compile(r"[0-9a-f]{64}\Z") +_IMAGE_ID = re.compile(r"sha256:[0-9a-f]{64}\Z") +_ARCHIVE_NAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9._+-]{0,190}\Z") +_IMAGE_REFERENCE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:/@+-]{0,510}\Z") + + +class FanoutError(RuntimeError): + """The archive cannot be distributed or verified safely.""" + + +@dataclass(frozen=True) +class ArchivePaths: + directory: str + final: str + partial: str + + +@dataclass(frozen=True) +class Hop: + index: int + edge: str + source_rank: int + destination_rank: int + source_address: str + destination_address: str + + +@dataclass(frozen=True) +class Probe: + state: str + sha256: str | None = None + bytes: int | None = None + detail: str | None = None + + +Runner = Callable[..., subprocess.CompletedProcess[str]] + + +def archive_paths(target_directory: str, archive_name: str) -> ArchivePaths: + if _ARCHIVE_NAME.fullmatch(archive_name) is None: + raise FanoutError( + "archive name must be one safe filename without path separators" + ) + directory = PurePosixPath(target_directory) + if not directory.is_absolute() or ".." in directory.parts: + raise FanoutError("target directory must be an absolute normalized path") + if directory == PurePosixPath("/") or len(directory.parts) < 4: + raise FanoutError("target directory is too broad") + final = directory / archive_name + partial = directory / f".{archive_name}.partial" + if final.parent != directory or partial.parent != directory: + raise FanoutError("archive paths must remain inside the target directory") + return ArchivePaths(str(directory), str(final), str(partial)) + + +def source_url(value: str) -> str: + parsed = urlsplit(value) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise FanoutError("source URL must use HTTP or HTTPS") + if parsed.username or parsed.password or parsed.query or parsed.fragment: + raise FanoutError( + "source URL must not contain credentials, a query, or a fragment" + ) + return value + + +def _rank_map(site: Any) -> dict[int, Any]: + ranks = {rank.id: rank for rank in site.ranks} + if len(ranks) != 4 or sorted(ranks) != list(range(4)): + raise FanoutError("direct archive fan-out requires ranks 0, 1, 2, and 3") + return ranks + + +def _port_to(rank: Any, destination_rank: int) -> Any: + matches = [ + port for port in rank.ring_ports if port.peer_rank == destination_rank + ] + if len(matches) != 1 or matches[0].peer_address is None: + raise FanoutError( + f"rank {rank.id} has no unique direct port to rank {destination_rank}" + ) + return matches[0] + + +def fabric_chain( + site: Any, + seed_rank: int, + first_hop_rank: int | None = None, +) -> tuple[Hop, ...]: + ranks = _rank_map(site) + if seed_rank not in ranks: + raise FanoutError(f"seed rank {seed_rank} is not configured") + neighbours = tuple(sorted(ranks[seed_rank].neighbour_ranks)) + if len(neighbours) != 2: + raise FanoutError("seed rank must have exactly two direct neighbours") + if first_hop_rank is None: + first_hop_rank = neighbours[0] + if first_hop_rank not in neighbours: + raise FanoutError("first hop must be a direct neighbour of the seed rank") + + order = [seed_rank, first_hop_rank] + while len(order) < len(ranks): + candidates = [ + rank_id + for rank_id in sorted(ranks[order[-1]].neighbour_ranks) + if rank_id not in order + ] + if len(candidates) != 1: + raise FanoutError("configured direct links do not form one chainable cycle") + order.append(candidates[0]) + + hops = [] + for index, (source_rank, destination_rank) in enumerate( + zip(order, order[1:]), start=1 + ): + port = _port_to(ranks[source_rank], destination_rank) + hops.append( + Hop( + index=index, + edge=port.edge, + source_rank=source_rank, + destination_rank=destination_rank, + source_address=str(port.address), + destination_address=str(port.peer_address), + ) + ) + return tuple(hops) + + +def _remote_argv(rank: Any, script: str, connect_timeout: int) -> tuple[str, ...]: + return ( + "ssh", + "-o", + "BatchMode=yes", + "-o", + "StrictHostKeyChecking=yes", + "-o", + f"ConnectTimeout={connect_timeout}", + rank.ssh_target, + shlex.join(("sh", "-lc", script)), + ) + + +def _probe_script(paths: ArchivePaths, expected_sha256: str) -> str: + final = shlex.quote(paths.final) + expected = shlex.quote(expected_sha256) + return ( + f"if [ ! -e {final} ]; then printf 'MISSING\\n'; exit 0; fi; " + f"if [ ! -f {final} ]; then printf 'CONFLICT non-regular\\n'; exit 0; fi; " + f"actual=$(sha256sum -- {final} | awk '{{print $1}}'); " + f"bytes=$(wc -c < {final}); " + f"if [ \"$actual\" = {expected} ]; then " + "printf 'EXACT %s %s\\n' \"$actual\" \"$bytes\"; " + "else printf 'CONFLICT %s %s\\n' \"$actual\" \"$bytes\"; fi" + ) + + +def _prepare_script(paths: ArchivePaths) -> str: + return f"install -d -m 0750 -- {shlex.quote(paths.directory)}" + + +def _download_script( + paths: ArchivePaths, + url: str, + expected_sha256: str, +) -> str: + directory = shlex.quote(paths.directory) + final = shlex.quote(paths.final) + partial = shlex.quote(paths.partial) + url_arg = shlex.quote(url) + expected = shlex.quote(expected_sha256) + return ( + "set -eu; " + f"install -d -m 0750 -- {directory}; " + f"test ! -e {final}; " + f"curl --fail --location --retry 3 --continue-at - --output {partial}" + f" -- {url_arg}; " + f"actual=$(sha256sum -- {partial} | awk '{{print $1}}'); " + f"test \"$actual\" = {expected}; " + f"sync -f {partial}; ln -- {partial} {final}; rm -- {partial}; " + f"sync -f {directory}; sha256sum -- {final}" + ) + + +def _transfer_script( + paths: ArchivePaths, + hop: Hop, + destination_user: str, + connect_timeout: int, +) -> str: + source = shlex.quote(paths.final) + destination = shlex.quote( + f"{destination_user}@{hop.destination_address}:{paths.partial}" + ) + remote_shell = shlex.quote( + "ssh -o BatchMode=yes" + " -o StrictHostKeyChecking=yes" + f" -o ConnectTimeout={connect_timeout}" + f" -b {hop.source_address}" + ) + return ( + "rsync --archive --partial --append-verify --protect-args" + f" --timeout={connect_timeout} -e {remote_shell} -- {source} {destination}" + ) + + +def _promote_script(paths: ArchivePaths, expected_sha256: str) -> str: + directory = shlex.quote(paths.directory) + final = shlex.quote(paths.final) + partial = shlex.quote(paths.partial) + expected = shlex.quote(expected_sha256) + return ( + "set -eu; " + f"test ! -e {final}; test -f {partial}; " + f"actual=$(sha256sum -- {partial} | awk '{{print $1}}'); " + f"test \"$actual\" = {expected}; " + f"sync -f {partial}; ln -- {partial} {final}; rm -- {partial}; " + f"sync -f {directory}; sha256sum -- {final}" + ) + + +def _load_script( + paths: ArchivePaths, + image: str, + expected_image_id: str, +) -> str: + final = shlex.quote(paths.final) + image_arg = shlex.quote(image) + expected = shlex.quote(expected_image_id) + return ( + f"if docker image inspect {image_arg} >/dev/null 2>&1; then " + f"existing=$(docker image inspect {image_arg} --format '{{{{.Id}}}}'); " + f"test \"$existing\" = {expected}; " + "else " + f"docker image load --input {final} >/dev/null; " + f"if ! docker image inspect {image_arg} >/dev/null 2>&1; then " + f"loaded=$(docker image inspect {expected} --format '{{{{.Id}}}}'); " + f"test \"$loaded\" = {expected}; " + f"docker image tag {expected} {image_arg}; fi; fi; " + f"docker image inspect {image_arg}" + " --format '{{.Id}} {{.Os}}/{{.Architecture}}'; " + f"image_id=$(docker image inspect {image_arg} --format '{{{{.Id}}}}')" + f"; test \"$image_id\" = {expected}" + ) + + +def _destination_user(rank: Any) -> str: + if "@" not in rank.ssh_target: + raise FanoutError(f"rank {rank.id} SSH target has no user") + return rank.ssh_target.split("@", 1)[0] + + +def plan_document( + site: Any, + *, + url: str, + expected_sha256: str, + paths: ArchivePaths, + seed_rank: int, + first_hop_rank: int | None, + create_only: bool, + image: str | None, + expected_image_id: str | None, + connect_timeout: int, +) -> dict[str, Any]: + if _SHA256.fullmatch(expected_sha256) is None: + raise FanoutError("expected SHA-256 must be 64 lowercase hex characters") + if expected_image_id and _IMAGE_ID.fullmatch(expected_image_id) is None: + raise FanoutError("expected image ID must be sha256:<64 lowercase hex>") + if not create_only: + if not image or _IMAGE_REFERENCE.fullmatch(image) is None: + raise FanoutError( + "a safe image reference is required unless --create-only is set" + ) + if not expected_image_id: + raise FanoutError( + "expected image ID is required unless --create-only is set" + ) + ranks = _rank_map(site) + hops = fabric_chain(site, seed_rank, first_hop_rank) + actions: list[dict[str, Any]] = [] + + seed = ranks[seed_rank] + actions.append( + { + "id": "seed-probe", + "kind": "probe", + "rank": seed_rank, + "command": list( + _remote_argv(seed, _probe_script(paths, expected_sha256), connect_timeout) + ), + } + ) + actions.append( + { + "id": "seed-download", + "kind": "download-if-missing", + "rank": seed_rank, + "command": list( + _remote_argv( + seed, + _download_script(paths, url, expected_sha256), + connect_timeout, + ) + ), + } + ) + for hop in hops: + source = ranks[hop.source_rank] + destination = ranks[hop.destination_rank] + actions.extend( + ( + { + "id": f"hop-{hop.index}-probe", + "kind": "probe", + "rank": hop.destination_rank, + "command": list( + _remote_argv( + destination, + _probe_script(paths, expected_sha256), + connect_timeout, + ) + ), + }, + { + "id": f"hop-{hop.index}-prepare", + "kind": "prepare-if-missing", + "rank": hop.destination_rank, + "command": list( + _remote_argv( + destination, + _prepare_script(paths), + connect_timeout, + ) + ), + }, + { + "id": f"hop-{hop.index}-transfer", + "kind": "direct-rsync-if-missing", + "rank": hop.source_rank, + "hop": asdict(hop), + "command": list( + _remote_argv( + source, + _transfer_script( + paths, + hop, + _destination_user(destination), + connect_timeout, + ), + connect_timeout, + ) + ), + }, + { + "id": f"hop-{hop.index}-verify", + "kind": "verify-and-promote-if-missing", + "rank": hop.destination_rank, + "command": list( + _remote_argv( + destination, + _promote_script(paths, expected_sha256), + connect_timeout, + ) + ), + }, + ) + ) + if not create_only: + assert image is not None + for rank_id in sorted(ranks): + rank = ranks[rank_id] + actions.append( + { + "id": f"rank-{rank_id}-load", + "kind": "load-image", + "rank": rank_id, + "command": list( + _remote_argv( + rank, + _load_script(paths, image, expected_image_id), + connect_timeout, + ) + ), + } + ) + return { + "schema": PLAN_SCHEMA, + "status": "implemented", + "safety": ["MUTATES HOST"], + "source_url": url, + "archive": { + "name": PurePosixPath(paths.final).name, + "sha256": expected_sha256, + "target_directory": paths.directory, + }, + "seed_rank": seed_rank, + "topology": { + "link_speed_mbps": int(site.topology.link_speed_mbps), + "mtu": int(site.topology.mtu), + }, + "chain": [asdict(hop) for hop in hops], + "create_only": create_only, + "image": image, + "expected_image_id": expected_image_id, + "actions": actions, + } + + +def parse_probe(output: str) -> Probe: + tokens = output.strip().split() + if tokens == ["MISSING"]: + return Probe("missing") + if len(tokens) >= 2 and tokens[0] == "CONFLICT": + if len(tokens) == 3 and _SHA256.fullmatch(tokens[1]): + return Probe("conflict", tokens[1], int(tokens[2])) + return Probe("conflict", detail=" ".join(tokens[1:])) + if ( + len(tokens) == 3 + and tokens[0] == "EXACT" + and _SHA256.fullmatch(tokens[1]) + ): + return Probe("exact", tokens[1], int(tokens[2])) + raise FanoutError(f"archive probe returned malformed evidence: {output!r}") + + +def _run( + argv: Sequence[str], + *, + timeout: int, + runner: Runner, + action: str, +) -> subprocess.CompletedProcess[str]: + try: + completed = runner( + tuple(argv), + check=False, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=timeout, + ) + except subprocess.TimeoutExpired as error: + raise FanoutError( + f"{action} was interrupted; the bounded partial file is resumable" + ) from error + if completed.returncode: + detail = completed.stderr.strip() or completed.stdout.strip() + raise FanoutError(f"{action} did not complete: {detail}") + return completed + + +def _probe_rank( + rank: Any, + paths: ArchivePaths, + expected_sha256: str, + *, + timeout: int, + connect_timeout: int, + runner: Runner, +) -> Probe: + completed = _run( + _remote_argv(rank, _probe_script(paths, expected_sha256), connect_timeout), + timeout=timeout, + runner=runner, + action=f"rank {rank.id} archive probe", + ) + return parse_probe(completed.stdout) + + +def verify_cluster( + site: Any, + paths: ArchivePaths, + expected_sha256: str, + *, + timeout: int, + connect_timeout: int, + runner: Runner = subprocess.run, +) -> dict[str, Any]: + ranks = _rank_map(site) + evidence = [] + for rank_id in sorted(ranks): + probe = _probe_rank( + ranks[rank_id], + paths, + expected_sha256, + timeout=timeout, + connect_timeout=connect_timeout, + runner=runner, + ) + evidence.append({"rank": rank_id, **asdict(probe)}) + if any(item["state"] != "exact" for item in evidence): + raise FanoutError( + "archive verification did not prove the expected file on every rank: " + f"{evidence}" + ) + return { + "schema": VERIFY_SCHEMA, + "status": "verified", + "archive": paths.final, + "sha256": expected_sha256, + "topology": { + "link_speed_mbps": int(site.topology.link_speed_mbps), + "mtu": int(site.topology.mtu), + }, + "ranks": evidence, + } + + +def execute_fanout( + site: Any, + plan: dict[str, Any], + paths: ArchivePaths, + *, + timeout: int, + connect_timeout: int, + runner: Runner = subprocess.run, +) -> dict[str, Any]: + ranks = _rank_map(site) + expected_sha256 = plan["archive"]["sha256"] + url = plan["source_url"] + seed_rank = int(plan["seed_rank"]) + rank_receipts: dict[int, dict[str, Any]] = {} + hop_receipts = [] + + seed_probe = _probe_rank( + ranks[seed_rank], + paths, + expected_sha256, + timeout=timeout, + connect_timeout=connect_timeout, + runner=runner, + ) + if seed_probe.state == "conflict": + raise FanoutError(f"seed rank {seed_rank} has a conflicting archive") + seed_reused = seed_probe.state == "exact" + if not seed_reused: + _run( + _remote_argv( + ranks[seed_rank], + _download_script(paths, url, expected_sha256), + connect_timeout, + ), + timeout=timeout, + runner=runner, + action=f"rank {seed_rank} archive download", + ) + seed_probe = _probe_rank( + ranks[seed_rank], + paths, + expected_sha256, + timeout=timeout, + connect_timeout=connect_timeout, + runner=runner, + ) + if seed_probe.state != "exact": + raise FanoutError("seed archive was not verified after download") + rank_receipts[seed_rank] = { + "rank": seed_rank, + **asdict(seed_probe), + "source": "existing" if seed_reused else "download", + } + + for hop_record in plan["chain"]: + hop = Hop(**hop_record) + destination = ranks[hop.destination_rank] + probe = _probe_rank( + destination, + paths, + expected_sha256, + timeout=timeout, + connect_timeout=connect_timeout, + runner=runner, + ) + if probe.state == "conflict": + raise FanoutError( + f"rank {hop.destination_rank} has a conflicting archive" + ) + reused = probe.state == "exact" + if not reused: + _run( + _remote_argv( + destination, _prepare_script(paths), connect_timeout + ), + timeout=timeout, + runner=runner, + action=f"hop {hop.index} destination preparation", + ) + _run( + _remote_argv( + ranks[hop.source_rank], + _transfer_script( + paths, + hop, + _destination_user(destination), + connect_timeout, + ), + connect_timeout, + ), + timeout=timeout, + runner=runner, + action=f"hop {hop.index} direct rsync", + ) + _run( + _remote_argv( + destination, + _promote_script(paths, expected_sha256), + connect_timeout, + ), + timeout=timeout, + runner=runner, + action=f"hop {hop.index} checksum and promotion", + ) + probe = _probe_rank( + destination, + paths, + expected_sha256, + timeout=timeout, + connect_timeout=connect_timeout, + runner=runner, + ) + if probe.state != "exact": + raise FanoutError(f"hop {hop.index} did not produce an exact archive") + rank_receipts[hop.destination_rank] = { + "rank": hop.destination_rank, + **asdict(probe), + "source": "existing" if reused else f"rank-{hop.source_rank}", + } + hop_receipts.append( + { + **asdict(hop), + "status": "reused" if reused else "transferred-and-verified", + "sha256": probe.sha256, + "bytes": probe.bytes, + } + ) + + loaded = [] + if not plan["create_only"]: + image = plan["image"] + assert image + for rank_id in sorted(ranks): + completed = _run( + _remote_argv( + ranks[rank_id], + _load_script(paths, image, plan["expected_image_id"]), + connect_timeout, + ), + timeout=timeout, + runner=runner, + action=f"rank {rank_id} image import", + ) + loaded.append({"rank": rank_id, "inspection": completed.stdout.strip()}) + + return { + "schema": RECEIPT_SCHEMA, + "status": "implemented", + "archive": { + "path": paths.final, + "sha256": expected_sha256, + }, + "seed_rank": seed_rank, + "topology": plan["topology"], + "ranks": [rank_receipts[index] for index in sorted(rank_receipts)], + "hops": hop_receipts, + "create_only": plan["create_only"], + "loaded_images": loaded, + "limitation": ( + "This receipt proves archive identity and optional image import. It" + " does not qualify model serving or direct-fabric throughput." + ), + } + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--site", required=True) + parser.add_argument("--source-url") + parser.add_argument("--archive-name", required=True) + parser.add_argument("--expected-sha256", required=True) + parser.add_argument("--target-directory", required=True) + parser.add_argument("--seed-rank", type=int, default=0) + parser.add_argument("--first-hop-rank", type=int) + parser.add_argument("--image") + parser.add_argument("--expected-image-id") + parser.add_argument("--create-only", action="store_true") + parser.add_argument("--verify", action="store_true") + parser.add_argument("--execute", action="store_true") + parser.add_argument("--confirmation") + parser.add_argument("--timeout", type=int, default=7200) + parser.add_argument("--connect-timeout", type=int, default=45) + parser.add_argument("--output", type=Path) + args = parser.parse_args() + + try: + site = load_site(args.site) + paths = archive_paths(args.target_directory, args.archive_name) + if _SHA256.fullmatch(args.expected_sha256) is None: + raise FanoutError( + "expected SHA-256 must be 64 lowercase hex characters" + ) + if args.verify: + if args.execute: + parser.error("--verify and --execute are mutually exclusive") + document = verify_cluster( + site, + paths, + args.expected_sha256, + timeout=args.timeout, + connect_timeout=args.connect_timeout, + ) + else: + if not args.source_url: + parser.error("--source-url is required for planning or execution") + url = source_url(args.source_url) + plan = plan_document( + site, + url=url, + expected_sha256=args.expected_sha256, + paths=paths, + seed_rank=args.seed_rank, + first_hop_rank=args.first_hop_rank, + create_only=args.create_only, + image=args.image, + expected_image_id=args.expected_image_id, + connect_timeout=args.connect_timeout, + ) + if not args.execute: + document = plan + else: + if args.confirmation != CONFIRMATION: + parser.error( + f"execute requires --confirmation {CONFIRMATION}" + ) + document = execute_fanout( + site, + plan, + paths, + timeout=args.timeout, + connect_timeout=args.connect_timeout, + ) + except (FanoutError, OSError, SiteConfigError) as error: + parser.error(str(error)) + + rendered = json.dumps(document, indent=2, sort_keys=True) + "\n" + if args.output: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(rendered, encoding="utf-8", newline="\n") + print(rendered, end="") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/test_fanout_image_archive.py b/scripts/test_fanout_image_archive.py new file mode 100644 index 00000000..1ccccf0b --- /dev/null +++ b/scripts/test_fanout_image_archive.py @@ -0,0 +1,321 @@ +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +import fanout_image_archive as fanout + + +DIGEST = "a" * 64 +IMAGE_ID = "sha256:" + "b" * 64 +ROOT = Path(__file__).resolve().parents[1] + + +def _site() -> SimpleNamespace: + edges = ( + (0, 1, "r0-r1", "10.0.1.10", "10.0.1.11"), + (1, 2, "r1-r2", "10.0.2.11", "10.0.2.12"), + (2, 3, "r2-r3", "10.0.3.12", "10.0.3.13"), + (3, 0, "r3-r0", "10.0.4.13", "10.0.4.10"), + ) + ports: dict[int, list[SimpleNamespace]] = {rank: [] for rank in range(4)} + for first, second, edge, first_address, second_address in edges: + ports[first].append( + SimpleNamespace( + edge=edge, + address=first_address, + peer_rank=second, + peer_address=second_address, + ) + ) + ports[second].append( + SimpleNamespace( + edge=edge, + address=second_address, + peer_rank=first, + peer_address=first_address, + ) + ) + return SimpleNamespace( + topology=SimpleNamespace(link_speed_mbps=200000, mtu=9000), + ranks=tuple( + SimpleNamespace( + id=rank, + ssh_target=f"operator@management-rank-{rank}.example", + ring_ports=tuple(ports[rank]), + neighbour_ranks=tuple( + sorted(port.peer_rank for port in ports[rank]) + ), + ) + for rank in range(4) + ) + ) + + +def _plan() -> tuple[SimpleNamespace, fanout.ArchivePaths, dict]: + site = _site() + paths = fanout.archive_paths("/var/lib/sparkring/images", "runtime.tar.zst") + plan = fanout.plan_document( + site, + url="https://images.example/runtime.tar.zst", + expected_sha256=DIGEST, + paths=paths, + seed_rank=0, + first_hop_rank=1, + create_only=True, + image=None, + expected_image_id=None, + connect_timeout=15, + ) + return site, paths, plan + + +def _completed(argv: tuple[str, ...], output: str = "") -> subprocess.CompletedProcess: + return subprocess.CompletedProcess(argv, 0, output, "") + + +def test_topology_builds_one_direct_chain_without_management_addresses() -> None: + hops = fanout.fabric_chain(_site(), seed_rank=0, first_hop_rank=1) + assert [(hop.source_rank, hop.destination_rank) for hop in hops] == [ + (0, 1), + (1, 2), + (2, 3), + ] + assert [hop.edge for hop in hops] == ["r0-r1", "r1-r2", "r2-r3"] + assert all(hop.destination_address.startswith("10.0.") for hop in hops) + + +def test_command_plan_uses_resumable_rsync_and_binds_direct_source() -> None: + _site_config, paths, plan = _plan() + transfers = [ + action for action in plan["actions"] if action["kind"] == "direct-rsync-if-missing" + ] + assert len(transfers) == 3 + first_command = " ".join(transfers[0]["command"]) + assert "--partial" in first_command + assert "--append-verify" in first_command + assert "10.0.1.10" in first_command + assert "10.0.1.11" in first_command + assert paths.partial in first_command + assert "management-rank-1.example" not in first_command + assert not any(action["kind"] == "load-image" for action in plan["actions"]) + assert plan["topology"] == {"link_speed_mbps": 200000, "mtu": 9000} + + +def test_verify_mode_reports_exact_evidence_for_every_rank() -> None: + site, paths, _plan_document = _plan() + + def runner(argv, **_kwargs): + return _completed(argv, f"EXACT {DIGEST} 8192\n") + + receipt = fanout.verify_cluster( + site, + paths, + DIGEST, + timeout=60, + connect_timeout=15, + runner=runner, + ) + assert receipt["schema"] == fanout.VERIFY_SCHEMA + assert receipt["status"] == "verified" + assert [rank["rank"] for rank in receipt["ranks"]] == [0, 1, 2, 3] + + +def test_verify_mode_rejects_one_wrong_digest() -> None: + site, paths, _plan_document = _plan() + calls = 0 + + def runner(argv, **_kwargs): + nonlocal calls + calls += 1 + digest = DIGEST if calls != 3 else "c" * 64 + state = "EXACT" if calls != 3 else "CONFLICT" + return _completed(argv, f"{state} {digest} 8192\n") + + with pytest.raises(fanout.FanoutError, match="did not prove"): + fanout.verify_cluster( + site, + paths, + DIGEST, + timeout=60, + connect_timeout=15, + runner=runner, + ) + + +def test_wrong_digest_probe_is_conflicting_evidence() -> None: + probe = fanout.parse_probe(f"CONFLICT {'c' * 64} 12345\n") + assert probe.state == "conflict" + assert probe.sha256 == "c" * 64 + assert probe.bytes == 12345 + + +def test_existing_exact_archive_skips_download_and_every_hop() -> None: + site, paths, plan = _plan() + calls = [] + + def runner(argv, **_kwargs): + calls.append(argv) + return _completed(argv, f"EXACT {DIGEST} 8192\n") + + receipt = fanout.execute_fanout( + site, + plan, + paths, + timeout=60, + connect_timeout=15, + runner=runner, + ) + assert len(calls) == 4 + assert all(rank["source"] == "existing" for rank in receipt["ranks"]) + assert all(hop["status"] == "reused" for hop in receipt["hops"]) + + +def test_existing_conflicting_archive_is_never_overwritten() -> None: + site, paths, plan = _plan() + + def runner(argv, **_kwargs): + return _completed(argv, f"CONFLICT {'d' * 64} 8192\n") + + with pytest.raises(fanout.FanoutError, match="conflicting archive"): + fanout.execute_fanout( + site, + plan, + paths, + timeout=60, + connect_timeout=15, + runner=runner, + ) + + +def test_interrupted_hop_reports_resumable_partial() -> None: + site, paths, plan = _plan() + probe_count = 0 + + def runner(argv, **_kwargs): + nonlocal probe_count + command = " ".join(argv) + if "rsync" in command: + raise subprocess.TimeoutExpired(argv, 60) + if "MISSING" in command or "sha256sum" in command: + probe_count += 1 + if probe_count == 1: + return _completed(argv, f"EXACT {DIGEST} 8192\n") + return _completed(argv, "MISSING\n") + return _completed(argv) + + with pytest.raises(fanout.FanoutError, match="partial file is resumable"): + fanout.execute_fanout( + site, + plan, + paths, + timeout=60, + connect_timeout=15, + runner=runner, + ) + + +def test_nonzero_remote_command_names_the_stopped_action() -> None: + def runner(argv, **_kwargs): + return subprocess.CompletedProcess(argv, 23, "", "fabric unavailable") + + with pytest.raises( + fanout.FanoutError, + match="hop 2 direct rsync did not complete: fabric unavailable", + ): + fanout._run( + ("ssh", "rank-1", "rsync"), + timeout=60, + runner=runner, + action="hop 2 direct rsync", + ) + + +def test_archive_commands_verify_sha_before_immutable_promotion() -> None: + paths = fanout.archive_paths("/var/lib/sparkring/images", "runtime.tar.zst") + download = fanout._download_script( + paths, + "https://images.example/runtime.tar.zst", + DIGEST, + ) + promote = fanout._promote_script(paths, DIGEST) + + for command in (download, promote): + assert "sha256sum" in command + assert f'test "$actual" = {DIGEST}' in command + assert f"ln -- {paths.partial} {paths.final}" in command + + +def test_source_url_rejects_embedded_authority_and_request_state() -> None: + for url in ( + "https://user:secret@images.example/runtime.tar.zst", + "https://images.example/runtime.tar.zst?token=secret", + "https://images.example/runtime.tar.zst#fragment", + ): + with pytest.raises(fanout.FanoutError, match="must not contain"): + fanout.source_url(url) + + +def test_path_and_image_identity_inputs_are_bounded() -> None: + with pytest.raises(fanout.FanoutError, match="too broad"): + fanout.archive_paths("/", "runtime.tar.zst") + with pytest.raises(fanout.FanoutError, match="safe filename"): + fanout.archive_paths("/var/lib/images", "../runtime.tar.zst") + site, paths, _document = _plan() + plan = fanout.plan_document( + site, + url="https://images.example/runtime.tar.zst", + expected_sha256=DIGEST, + paths=paths, + seed_rank=0, + first_hop_rank=1, + create_only=False, + image="registry.example/runtime@sha256:" + "e" * 64, + expected_image_id=IMAGE_ID, + connect_timeout=15, + ) + assert plan["expected_image_id"] == IMAGE_ID + + +def test_image_import_tags_an_archive_saved_by_image_id() -> None: + paths = fanout.archive_paths("/var/lib/sparkring/images", "runtime.tar.zst") + image = "registry.example/runtime:verified" + script = fanout._load_script(paths, image, IMAGE_ID) + + assert f"docker image inspect {IMAGE_ID}" in script + assert f"test \"$loaded\" = {IMAGE_ID}" in script + assert f"docker image tag {IMAGE_ID} {image}" in script + + +def test_operator_documentation_link_resolves() -> None: + document = ROOT / "docs/DIRECT_FABRIC_IMAGE_ARCHIVE_FANOUT.md" + assert document.is_file() + config_readme = (ROOT / "scripts/config/README.md").read_text( + encoding="utf-8" + ) + assert "../../docs/DIRECT_FABRIC_IMAGE_ARCHIVE_FANOUT.md" in config_readme + + +def test_cli_help_names_all_modes_and_confirmation_inputs() -> None: + completed = subprocess.run( + [sys.executable, str(ROOT / "scripts/fanout_image_archive.py"), "--help"], + cwd=ROOT, + check=False, + text=True, + capture_output=True, + ) + assert completed.returncode == 0, completed.stderr + for option in ( + "--verify", + "--execute", + "--create-only", + "--confirmation", + "--timeout", + "--connect-timeout", + "--output", + ): + assert option in completed.stdout