mesh-lab/test/integration/bus-kept.ts

/**
 * Whether genesis kept its bus through the broker step (novox/hq ADR 0333 §5, issue 590).
 *
 * Pure, so the judgement is tested without a lab (test/bus-kept.test.ts) and the bed only gathers what it
 * judges: the containers the anchor runs, the streams its bus holds, and what the mesh says of the node,
 * once before the broker step and once after it.
 */

/** One running container, as `docker ps --format '{{.Names}}\t{{.Image}}\t{{.Ports}}'` says it. */
export interface Running {
  name: string;
  image: string;
  ports: string;
}

/** One stream on the bus, as the server's monitor says it (`/jsz?accounts=true&streams=true`). */
export interface Stream {
  account: string;
  name: string;
  created: string;
  messages: number;
  lastSeq: number;
}

/** What the anchor showed at one moment. */
export interface Seen {
  running: Running[];
  streams: Stream[];
}

/** The container the bundle raises the bus as, and the nats module keeps it as (ADR 0333 §1). */
export const BUS = "mesh-broker-nats";

/** `docker ps --format '{{.Names}}\t{{.Image}}\t{{.Ports}}'`, read. */
export function runningFrom(ps: string): Running[] {
  return ps.split("\n").map((l) => l.trim()).filter(Boolean).map((l) => {
    const [name = "", image = "", ports = ""] = l.split("\t");
    return { name, image, ports };
  });
}

/** The server's `/jsz?accounts=true&streams=true`, read: every stream of every account. */
export function streamsFrom(jsz: string): Stream[] {
  const parsed = JSON.parse(jsz) as {
    account_details?: Array<{
      name?: string;
      stream_detail?: Array<{ name?: string; created?: string; state?: { messages?: number; last_seq?: number } }>;
    }>;
  };
  const out: Stream[] = [];
  for (const account of parsed.account_details ?? []) {
    for (const s of account.stream_detail ?? []) {
      out.push({
        account: account.name ?? "", name: s.name ?? "", created: s.created ?? "",
        messages: s.state?.messages ?? 0, lastSeq: s.state?.last_seq ?? 0,
      });
    }
  }
  return out;
}

/**
 * A bus container: one that serves the bus's port inside it, or carries a bus's name. Both, because a
 * second bus raised by a module under another name is exactly what this looks for, and so is one raised on
 * another port under the bus's name.
 */
export function busContainers(running: Running[]): Running[] {
  return running.filter((r) => /->4222\/tcp/.test(r.ports) || r.name.startsWith("mesh-broker"));
}

/** The streams written before the step: the ones holding anything. */
export function written(seen: Seen): Stream[] {
  return seen.streams.filter((s) => s.messages > 0 || s.lastSeq > 0);
}

/**
 * Every way the bus was not kept, in sentences; empty when it was. `node` is the mesh's line for the first
 * node from `node list`; `said` is what the installer printed on the run that crossed the step.
 */
export function busNotKept(before: Seen, after: Seen, node: string, said: string): string[] {
  const wrong: string[] = [];
  const buses = busContainers(after.running);
  if (buses.length !== 1) {
    wrong.push(`${buses.length} bus containers run after the broker step, not one: ` +
      (buses.map((b) => `${b.name} (${b.image}, ${b.ports || "nothing published"})`).join("; ") || "none"));
  }
  const bus = after.running.find((r) => r.name === BUS);
  if (!bus) {
    wrong.push(`${BUS} does not run after the broker step`);
  }
  const was = before.running.find((r) => r.name === BUS);
  if (bus && was && bus.image === was.image) {
    wrong.push(`${BUS} still runs ${bus.image}, the image the bundle raised it from: the module's image never ` +
      `took its place (ADR 0333 §3)`);
  }
  if (!/restarting\s+mesh-broker-nats/.test(said)) {
    wrong.push("the installer did not say the bus restarts onto the module's image (ADR 0333 §3)");
  }
  for (const s of written(before)) {
    const now = after.streams.find((a) => a.account === s.account && a.name === s.name);
    if (!now) {
      wrong.push(`the stream ${s.name} (${s.account}), holding ${s.messages} message(s) before the step, is gone`);
    } else if (now.created !== s.created || now.lastSeq < s.lastSeq) {
      wrong.push(`the stream ${s.name} (${s.account}) was made again: created ${s.created}, last ${s.lastSeq} ` +
        `before the step; created ${now.created}, last ${now.lastSeq} after it`);
    }
  }
  if (!/^\S+\s+here\b/.test(node.trim())) {
    wrong.push(`the mesh has not heard from the first node since the step: ${node.trim() || "it has no record of it"}`);
  }
  return wrong;
}