From 5c8e722df9cf67d506a38694d6386a0bd8ac06e8 Mon Sep 17 00:00:00 2001 From: FujitsuPolycom <87842395+FujitsuPolycom@users.noreply.github.com> Date: Sun, 30 Aug 2026 06:38:14 -0500 Subject: [PATCH 1/2] Distribute verified image archives through direct fabric Plan one four-rank direct-link chain without remote work, verify SHA-256 on every rank, resume bounded partial transfers, and optionally import only the expected image ID. Conflicting final files remain unchanged and interrupted operations retain resumable evidence. This adds no runtime-profile, cache-identity, or serving behavior. --- docs/DIRECT_FABRIC_IMAGE_ARCHIVE_FANOUT.md | 170 +++++ scripts/config/README.md | 8 + scripts/fanout_image_archive.py | 782 +++++++++++++++++++++ scripts/test_fanout_image_archive.py | 321 +++++++++ 4 files changed, 1281 insertions(+) create mode 100644 docs/DIRECT_FABRIC_IMAGE_ARCHIVE_FANOUT.md create mode 100644 scripts/fanout_image_archive.py create mode 100644 scripts/test_fanout_image_archive.py 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 From 75809e4026a7fd7633882dec5d93d210bf597d14 Mon Sep 17 00:00:00 2001 From: FujitsuPolycom <87842395+FujitsuPolycom@users.noreply.github.com> Date: Sun, 30 Aug 2026 06:41:02 -0500 Subject: [PATCH 2/2] Bound GLM-5.3 research summary to valid observations State that C4 and C8 were capacity-limited and remain omitted from the research-only throughput row. The FP8 KV and NVMe capacity descriptions remain unchanged. --- README.md | 2 ++ 1 file changed, 2 insertions(+) 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 | |---|---:|---:|---:|---:|---:|