mesh-lab/src/lifecycle/registry.ts

/**
 * A registry inside the scenario.
 *
 * A sealed machine cannot reach a registry, and an image placed from an archive cannot keep its
 * digest — `docker save` of a digest reference produces an archive with no repo tag, because a
 * repo digest only exists for an image a registry served (novox/hq 04-ISSUES/009). So an image
 * pinned by digest, which is the only kind the host accepts
 * ([ADR 0006](../../02-DECISIONS/0046-the-installer-fetches-what-it-pins.md)), could not be
 * placed at all.
 *
 * The answer is a registry, and it is not a workaround for the lab: ADR 0006 names an OCI
 * registry as substrate, and ADR 0006 says a first node fetches "upstream, wherever the image
 * ordinarily lives". **This is that upstream** — scenery, like the transit router is the
 * internet ([ADR 0016](../../02-DECISIONS/0033-a-router-is-scenery-not-a-node.md)).
 *
 * The digests it serves are its own, not Docker Hub's, and that is correct rather than a
 * compromise. What ADR 0006 requires is a reference that is exact and cannot move. A digest
 * assigned by this registry is both.
 */

import { spawn } from "node:child_process";

import { incus, incusOk, succeeds } from "../incus/client.ts";
import { macFor, networkName } from "./names.ts";
import { addressLink } from "./address.ts";
import { around, log, shorten } from "../log.ts";
import { BASE_IMAGE_ALIAS, placeImage } from "./place.ts";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";

/** The image the registry itself runs from. Placed by tag, which archives keep. */
export const REGISTRY_IMAGE = "registry:2";

/** Where the registry serves, inside its machine. */
export const REGISTRY_PORT = 5000;

export class RegistryError extends Error {
  constructor(message: string) {
    super(message);
    this.name = "RegistryError";
  }
}

export interface StockedImage {
  /** What the scenario asked for, as written. */
  requested: string;
  /** The repository path the registry serves it under. */
  repository: string;
  /** The digest THIS registry assigned. What a declaration pins. */
  digest: string;
}

export interface Stock {
  /** A directory holding the registry's data, ready to be placed in a machine. */
  dataDir: string;
  images: StockedImage[];
}

/**
 * Build a registry's data directory on this workstation, with the given images in it.
 *
 * Runs a throwaway registry here — where there IS a network — pushes into it, and keeps what
 * it wrote. Research 012's reframing again: fetch at build time on a machine that has a
 * network, apply on a target that needs nothing.
 *
 * The caller owns the returned directory and must remove it.
 */
export async function stockRegistry(
  references: string[],
  log: (message: string) => void = () => {},
): Promise<Stock> {
  if (references.length === 0) return { dataDir: "", images: [] };

  const dataDir = await mkdtemp(join(tmpdir(), "mesh-lab-registry-"));
  const container = `mesh-lab-stock-${process.pid}`;
  const port = 5000 + (process.pid % 1000);

  await docker(["rm", "-f", container], 60_000);
  const started = await docker(
    ["run", "-d", "--name", container, "-p", `${port}:5000`, "-v", `${dataDir}:/var/lib/registry`,
      REGISTRY_IMAGE],
    300_000,
  );
  if (!started.ok) {
    await rm(dataDir, { recursive: true, force: true });
    throw new RegistryError(
      `cannot run ${REGISTRY_IMAGE} on this workstation to stock a registry: ${started.stderr.trim()}`,
    );
  }

  try {
    await waitForRegistry(port);
    const images: StockedImage[] = [];

    for (const reference of references) {
      // The repository path a machine will pull from. A tag is dropped: what a declaration
      // pins is the digest, and carrying the tag as well would invite pinning the wrong one.
      const repository = repositoryFor(reference);
      const target = `localhost:${port}/${repository}`;

      const tagged = await docker(["tag", reference, target], 60_000);
      if (!tagged.ok) {
        throw new RegistryError(
          `${reference} is not on this workstation, and the lab does not fetch on a scenario's ` +
            `behalf. Pull it here first.\n  ${tagged.stderr.trim()}`,
        );
      }
      const pushed = await docker(["push", target], 900_000);
      if (!pushed.ok) throw new RegistryError(`cannot push ${reference}: ${pushed.stderr.trim()}`);

      const digest = digestFrom(pushed.stdout + pushed.stderr);
      if (!digest) {
        throw new RegistryError(
          `${reference} was pushed and the registry did not report a digest. Without one there ` +
            `is nothing for a declaration to pin.`,
        );
      }
      images.push({ requested: reference, repository, digest });
      log(`  stocked ${repository}@${digest}`);
    }

    return { dataDir, images };
  } catch (err) {
    await discardStock({ dataDir, images: [] });
    throw err;
  } finally {
    await docker(["rm", "-f", container], 60_000);
  }
}

/**
 * Remove a stocked registry's data.
 *
 * Through a container, because a container wrote it. The registry runs as root inside, so the
 * blobs it writes into a bind mount are owned by root and an ordinary process cannot remove
 * them — `rmdir` fails with EACCES on a directory that looks like ours.
 *
 * Whoever made the files removes them.
 */
export async function discardStock(stock: Stock): Promise<void> {
  if (!stock.dataDir) return;
  await docker(["run", "--rm", "-v", `${stock.dataDir}:/stock`, REGISTRY_IMAGE,
    "sh", "-c", "rm -rf /stock/* /stock/.[!.]* 2>/dev/null || true"], 120_000);
  await rm(stock.dataDir, { recursive: true, force: true }).catch(() => {});
}

/** `alpine:3.20` and `alpine` both serve from `alpine`; `foo/bar:1` from `foo/bar`. */
export function repositoryFor(reference: string): string {
  const withoutDigest = reference.split("@")[0] ?? reference;
  const lastColon = withoutDigest.lastIndexOf(":");
  const lastSlash = withoutDigest.lastIndexOf("/");
  return lastColon > lastSlash ? withoutDigest.slice(0, lastColon) : withoutDigest;
}

/** `docker push` prints `<tag>: digest: sha256:… size: …` on its last useful line. */
export function digestFrom(output: string): string | null {
  const match = output.match(/digest:\s*(sha256:[a-f0-9]{64})/);
  return match?.[1] ?? null;
}

async function waitForRegistry(port: number): Promise<void> {
  for (let i = 0; i < 30; i++) {
    const probe = await docker(["run", "--rm", "--network", "host", REGISTRY_IMAGE,
      "sh", "-c", `wget -q -O- http://localhost:${port}/v2/ >/dev/null 2>&1`], 30_000);
    if (probe.ok) return;
    await new Promise((r) => setTimeout(r, 1_000));
  }
  throw new RegistryError("a registry was started on this workstation and never answered");
}

function docker(
  args: string[],
  timeoutMs: number,
): Promise<{ ok: boolean; stdout: string; stderr: string }> {
  // The second of the three places the lab runs an external program (novox/hq 04-ISSUES/024).
  // `docker push` of a large image is minutes of legitimate silence, which is exactly when a
  // heartbeat earns its keep.
  return around(`docker ${shorten(args)}`, () => runDocker(args, timeoutMs), { heartbeatMs: 15_000 });
}

function runDocker(
  args: string[],
  timeoutMs: number,
): Promise<{ ok: boolean; stdout: string; stderr: string }> {
  return new Promise((resolve) => {
    const child = spawn("docker", args, { stdio: ["ignore", "pipe", "pipe"] });
    let stdout = "";
    let stderr = "";
    const timer = setTimeout(() => child.kill("SIGKILL"), timeoutMs);
    child.stdout.on("data", (d) => (stdout += d));
    child.stderr.on("data", (d) => (stderr += d));
    child.on("error", (err) => {
      clearTimeout(timer);
      resolve({ ok: false, stdout, stderr: err.message });
    });
    child.on("close", (code) => {
      clearTimeout(timer);
      // A docker failure is an answer here rather than an exception, so it would otherwise pass
      // through the log looking exactly like a success.
      if (code !== 0) {
        log.debug(`  exit ${code}: ${shorten([stderr.trim() || "(nothing on stderr)"], 400)}`);
      }
      resolve({ ok: code === 0, stdout, stderr });
    });
  });
}

// --- the registry inside a scenario ------------------------------------------------------------

/**
 * Where the registry sits on its segment.
 *
 * A convention rather than a declaration, like the router's. `.250` is chosen to sit well away
 * from the low addresses scenarios give their machines, so a scenario can be written without
 * thinking about it and a collision is obvious when it happens.
 */
export const REGISTRY_HOST_OCTET = 250;

/** The address the registry answers on, given the segment it is attached to. */
export function registryAddress(cidr: string): string {
  const [network] = cidr.split("/");
  const parts = (network ?? "").split(".");
  if (parts.length !== 4) {
    throw new RegistryError(
      `cannot place a registry on '${cidr}': it is not an IPv4 network, and the registry needs ` +
        `an address a machine can be pointed at.`,
    );
  }
  return `${parts[0]}.${parts[1]}.${parts[2]}.${REGISTRY_HOST_OCTET}`;
}

/** What a declaration should pin, once a scenario is raised. */
export function pinnedReference(address: string, image: StockedImage): string {
  return `${address}:${REGISTRY_PORT}/${image.repository}@${image.digest}`;
}

// --- raising it inside a scenario ---------------------------------------------------------------

/** What a raised registry is, and what a declaration needs from it. */
export interface RaisedRegistry {
  machine: string;
  segment: string;
  address: string;
  /** Each image, as a reference a declaration can pin. */
  pinned: string[];
}

/**
 * Pick the segment the registry sits on.
 *
 * A public segment, because that is what stands in for the outside world — a first node fetches
 * from upstream, and this is upstream. An IPv4 range, because a machine has to be pointed at it
 * by address.
 */
export function registrySegment(
  segments: Record<string, { kind: string; cidr: string[] }>,
): { name: string; cidr: string } | null {
  for (const [name, segment] of Object.entries(segments)) {
    if (segment.kind !== "public") continue;
    const v4 = segment.cidr.find((c) => !c.includes(":"));
    if (v4) return { name, cidr: v4 };
  }
  return null;
}

/**
 * Raise a registry inside the scenario and load the stocked images into it.
 *
 * Scenery, in the same sense the transit router is: nothing under test runs on it, it holds no
 * identity, and no assertion is made about its internals. It exists so that a machine can fetch
 * an image the way a real one does — over the network, from a registry, by digest.
 */
export async function raiseRegistry(
  scenario: { segments: Record<string, { kind: string; cidr: string[] }> },
  instanceId: string,
  stock: Stock,
  log: (message: string) => void = () => {},
  /**
   * The copy-on-write pool the scenario's machines were placed on. The registry goes on it too —
   * NOT the profile's default `dir` pool — so a sized root disk is thin (paid for as it fills)
   * rather than a full allocation on the host's own filesystem. Absent, it falls back to the
   * profile default, which is the historical behaviour for a small registry.
   */
  pool?: string,
): Promise<RaisedRegistry | null> {
  if (stock.images.length === 0) return null;

  const segment = registrySegment(scenario.segments);
  if (!segment) {
    throw new RegistryError(
      `this scenario declares images and has no public IPv4 segment to serve them from.\n` +
        `  The registry stands in for the outside world, so it sits on a public segment.`,
    );
  }

  const address = registryAddress(segment.cidr);
  const name = `mlab-${instanceId}-registry`;
  const prefix = segment.cidr.slice(segment.cidr.lastIndexOf("/"));

  if (!(await succeeds(["config", "show", name], 15_000))) {
    // The registry holds the WHOLE `images:` union on its own root disk, and the base image's
    // default is only ~10GiB. A single-node bed stocks a handful of images and fits; a broad bed
    // — and especially the full segmented mesh, whose union is both server sets at once (~28GiB)
    // — overflows it, and the raise dies "no space left on device" while pushing blobs into the
    // registry. So the registry gets a sized root disk. Thin on a copy-on-write pool, so a small
    // bed pays only for what it actually stocks; MESH_LAB_REGISTRY_DISK overrides for a giant one.
    const registryDisk = process.env["MESH_LAB_REGISTRY_DISK"] ?? "80GiB";
    await incus([
      "init", BASE_IMAGE_ALIAS, name, "--vm",
      "-c", "security.secureboot=false",
      "-c", "limits.memory=1GiB",
      ...(pool ? ["-s", pool] : []),
      "-d", `root,size=${registryDisk}`,
      "-c", `user.mesh-lab.instance=${instanceId}`,
      // Tagged as a machine as well, so `destroy` finds it with one query — a router that
      // carried only its own tag was left behind and held its networks open.
      "-c", "user.mesh-lab.machine=registry",
      "-c", "user.mesh-lab.registry=true",
    ], 300_000);
    await succeeds(["config", "device", "remove", name, "eth0"], 15_000);
    await incus([
      "config", "device", "add", name, "eth0", "nic",
      "nictype=bridged",
      `parent=${networkName(instanceId, segment.name)}`,
      `hwaddr=${macFor(instanceId, "registry", 0)}`,
    ]);
  }
  await succeeds(["start", name], 60_000);
  await waitForAgent(name);

  // Addressed the way every other machine is: a systemd-networkd unit matching the MAC.
  //
  // **This used to be `ip addr add`, and it stalled the lab.** An address set by hand leaves
  // networkd waiting to configure a link it was never told about, so the link sits at
  // `configuring`, `systemd-networkd-wait-online` never returns — its timeout is `infinity` —
  // and `network-online.target` is never reached. Docker is ordered after that target, so
  // `docker load` two lines below blocked on a socket whose daemon was queued behind a target
  // that would never come.
  //
  // Matching on MAC and not on interface name is still the rule: a machine with a container
  // runtime has a `docker0` that sorts before `enp5s0`, and naive selection configures that.
  await addressLink(name, {
    device: "eth0",
    mac: macFor(instanceId, "registry", 0),
    addresses: [`${address}${prefix}`],
    // The registry takes the segment's default. It carried no MTU before this and still does
    // not: what a scenario sets an MTU for is the path under test, and this is scenery.
    mtu: undefined,
  });

  log(`  registry on ${segment.name} at ${address}`);

  // The registry's own image, placed by tag — an archive keeps a tag and cannot keep a digest,
  // which is the whole reason this machine exists.
  // Logged, not silenced. This is the step a stall sat in for thirty-five minutes while the
  // caller had passed it a callback that threw everything away (novox/hq 04-ISSUES/024).
  await placeImage(name, "registry", REGISTRY_IMAGE, log);

  // The destination must EXIST before a recursive push, or incus copies the source's contents
  // rather than the source — the data lands one directory too shallow, the registry finds
  // nothing where it looks, and every pull fails with `not found`.
  await incus(["exec", name, "--", "mkdir", "-p", "/srv/registry"], 60_000);
  await incus(["file", "push", "-r", `${stock.dataDir}/docker`, `${name}/srv/registry/`], 900_000);

  await incus(["exec", name, "--", "docker", "run", "-d",
    "--name", "registry", "--restart", "unless-stopped",
    "-p", `${REGISTRY_PORT}:5000`,
    "-v", "/srv/registry:/var/lib/registry",
    REGISTRY_IMAGE], 300_000);

  // Read back that each image is SERVED, by asking for its manifest by digest — which is
  // exactly what a machine will do.
  //
  // Not that the catalog endpoint answers: `{"repositories":[]}` contains the word
  // `repositories`, so checking for that passed on a registry holding nothing at all, and the
  // failure surfaced much later as a container that could not be pulled.
  let answered = false;
  for (let i = 0; i < 20 && !answered; i++) {
    const ping = await incusOk(["exec", name, "--", "curl", "-s", "-o", "/dev/null",
      "-w", "%{http_code}", "--max-time", "3",
      `http://localhost:${REGISTRY_PORT}/v2/`], 30_000);
    answered = ping?.trim() === "200";
    if (!answered) await new Promise((r) => setTimeout(r, 2_000));
  }
  if (!answered) {
    throw new RegistryError(
      `the registry on ${name} started and never answered. Machines in this scenario cannot ` +
        `fetch an image, so nothing that declares a container will work.`,
    );
  }

  const pinned: string[] = [];
  for (const image of stock.images) {
    const code = await incusOk(["exec", name, "--", "curl", "-s", "-o", "/dev/null",
      "-w", "%{http_code}", "--max-time", "5",
      "-H", "Accept: application/vnd.docker.distribution.manifest.v2+json",
      `http://localhost:${REGISTRY_PORT}/v2/${image.repository}/manifests/${image.digest}`,
    ], 60_000);
    if (code?.trim() !== "200") {
      throw new RegistryError(
        `the registry on ${name} is running and does not serve ${image.repository}@${image.digest} ` +
          `(it answered ${code?.trim() || "nothing"}).\n` +
          `  The images were stocked on this workstation and did not arrive intact, so a ` +
          `machine declaring that image would fail to pull it.`,
      );
    }
    const reference = pinnedReference(address, image);
    pinned.push(reference);
    log(`  serving ${reference}`);
  }
  return { machine: name, segment: segment.name, address, pinned };
}

/**
 * Confirm the registry serves the MACHINES, not just itself.
 *
 * **`raiseRegistry` proves the wrong thing, and the difference cost a whole raise.** It curls
 * `localhost:5000` from inside the registry's own machine — which says the registry process is up
 * and holds the blobs, and says nothing at all about the path every other machine actually uses:
 * across a segment, and for the home nodes through a NAT gateway whose route is applied two steps
 * LATER. So `raise` could return "serving" with the anchor unable to reach the registry at all, the
 * substrate apply's first pull would fail, no node could enrol, and the instance was left a bare
 * shell — VMs and a registry and nothing else.
 *
 * A raise is not finished while that is still possible. This is the check that makes "the registry
 * is serving" mean what the next step needs it to mean: from each machine, on the address it will
 * pin, over the network it will use, once its route and its firewall are in place.
 *
 * **What it proves and what it does not.** It asks for `/v2/` and then for one stocked manifest BY
 * DIGEST — the same two requests a pull begins with, from the same place. It does not fetch layers:
 * every digest was already read back inside the registry machine, so what is in question here is
 * the path, not the content, and a probe per machine keeps the check to seconds rather than the
 * many minutes a full pull of a hundred images would take.
 */
export async function confirmRegistryServes(
  registry: RaisedRegistry,
  machines: string[],
  stock: Stock,
  log: (message: string) => void = () => {},
  waitSeconds = 180,
): Promise<void> {
  const probe = stock.images[0];
  if (!probe) return;
  const base = `http://${registry.address}:${REGISTRY_PORT}`;
  const manifest =
    `-H "Accept: application/vnd.docker.distribution.manifest.v2+json" ` +
    `${base}/v2/${probe.repository}/manifests/${probe.digest}`;

  for (const machine of machines) {
    const deadline = Date.now() + waitSeconds * 1_000;
    let last = "";
    let served = false;
    while (!served && Date.now() < deadline) {
      // One shell, two requests: the machine can reach the registry AND the registry answers for
      // an image by the digest a declaration pins. Either alone passes on a registry serving
      // nothing, which is the failure this whole function exists to stop reporting as success.
      const said = await incusOk(["exec", machine, "--", "sh", "-c",
        `printf '%s %s' ` +
        `"$(curl -s -o /dev/null -w '%{http_code}' --max-time 5 ${base}/v2/)" ` +
        `"$(curl -s -o /dev/null -w '%{http_code}' --max-time 10 ${manifest})"`,
      ], 40_000);
      last = said?.trim() ?? "";
      served = last === "200 200";
      if (!served) await new Promise((r) => setTimeout(r, 3_000));
    }
    if (!served) {
      throw new RegistryError(
        `${machine} cannot pull from the registry at ${registry.address}:${REGISTRY_PORT} after ` +
          `${waitSeconds}s (it got "${last || "nothing"}" for /v2/ and for ` +
          `${probe.repository}@${probe.digest}).\n` +
          `  The registry answers on its own machine, so this is the PATH: this machine's route, ` +
          `its gateway, or its firewall. Every image this scenario declares is unreachable from ` +
          `here, so anything applied to it would fail on its first pull.`,
      );
    }
    log(`  ${machine} can pull from ${registry.address}:${REGISTRY_PORT}`);
  }
}

async function waitForAgent(name: string): Promise<void> {
  for (let i = 0; i < 90; i++) {
    if (await succeeds(["exec", name, "--", "true"], 10_000)) return;
    await new Promise((r) => setTimeout(r, 2_000));
  }
  throw new RegistryError(`${name} started and its agent never answered.`);
}