events: an e2e test — an emitted event reaches the audit trail over the mesh's broker

Adds test/integration/events.test.ts: raises first-node (which raises the broker as tier-1 substrate), runs the runtime+audit-logger against that broker, emits a module and a node event, and asserts they reach the trail with their metadata read back from ADR 0047 headers (a pure body), plus that the durable per-consumer queue and mesh.events.dead exchange exist on the raised broker — asked of the broker, not assumed. scripts/build-runtime-image.sh builds the self-contained runtime image (mesh-tools + vendored sdk + audit-logger) it runs, saved to a tar for MESH_LAB_RUNTIME. The events path itself is verified; the incus raise is the part a lab run exercises. Claude-Session: https://claude.ai/code/session_01LrgweAeERJYBg88c5cKDzF
2026-09-04 00:56:39 +02:00
parent 01ca02ad04
commit ba7f7998f7
2 changed files with 278 additions and 0 deletions
@@ -0,0 +1,221 @@
+/**
+ * An event a module emits reaches an audit trail, over the broker the mesh raised.
+ *
+ * Everything else here proves the broker carries *commands* — a declaration crosses it, a
+ * credential is delivered over it. This proves the other half of the bus (novox/hq ADR 0046): the
+ * events exchange, where a module emits and any number listen, and the audit logger consumes `#`
+ * and writes down what happened. It runs against the `mesh-broker` this scenario's own host raised
+ * from the substrate bundle — not a broker a test stood up — because "the mesh hosts the broker"
+ * (tier-1 substrate) is the thing being relied on.
+ *
+ * The wire shape it asserts is ADR 0047: metadata rides as headers so the body is only the
+ * payload, a consumer gets a durable per-consumer queue `<node>.<module>.events`, and a
+ * dead-letter exchange `mesh.events.dead` stands behind it. The queue and that exchange existing
+ * on the raised broker is the check that the runtime provisioned the contract, not just that a
+ * message happened to arrive.
+ *
+ * It needs the host binary and the substrate bundle, like the mesh walk, plus a runtime image:
+ *
+ *   MESH_LAB_HOST_BINARY=.../mesh-host
+ *   MESH_LAB_BUNDLE=.../examples/substrate-first-node.lock
+ *   MESH_LAB_RUNTIME=.../mesh-runtime-audit.tar   (docker save of the runtime+audit-logger image;
+ *                                                  built by scripts/build-runtime-image.sh)
+ *
+ * The runtime is run as a container against the broker here, rather than assigned through the
+ * control plane. Assigning it — so the mesh delivers its broker credential the way it does the
+ * builder's — is the next step; this proves the events path itself first.
+ */
+
+import { test, before, after } from "node:test";
+import assert from "node:assert/strict";
+import { existsSync, readFileSync } from "node:fs";
+import { loadScenario } from "../../src/declaration/parse.ts";
+import { raise } from "../../src/lifecycle/raise.ts";
+import { destroy, exec } from "../../src/lifecycle/operate.ts";
+import { hostBinaryPath, HOST_PATH } from "../../src/lifecycle/place.ts";
+import { labIsUsable, destroyAll } from "./harness.ts";
+import { incus } from "../../src/incus/client.ts";
+import { machineName } from "../../src/lifecycle/names.ts";
+
+const capability = await labIsUsable();
+const binary = hostBinaryPath();
+const bundle = process.env["MESH_LAB_BUNDLE"] ?? "";
+const runtime = process.env["MESH_LAB_RUNTIME"] ?? "";
+
+const skip = !capability.usable
+  ? `lab not usable: ${capability.why}`
+  : !binary || !existsSync(binary)
+    ? "MESH_LAB_HOST_BINARY is not set to a built mesh-host"
+    : !bundle || !existsSync(bundle)
+      ? "MESH_LAB_BUNDLE is not set to a substrate bundle (mesh-host examples/)"
+      : !runtime || !existsSync(runtime)
+        ? "MESH_LAB_RUNTIME is not set to a runtime image tar (scripts/build-runtime-image.sh)"
+        : false;
+
+const SCENARIO = "first-node";
+const MACHINE = "anchor";
+/** Where the audit-logger container writes its trail, on the machine — mounted from a host dir. */
+const TRAIL_DIR = "/var/lib/mesh-audit";
+const TRAIL = `${TRAIL_DIR}/audit.log`;
+/** The broker the substrate raised, reachable on the node's loopback (novox/hq ADR 0001). */
+const BROKER = "amqp://guest:guest@127.0.0.1:5672/";
+
+let instanceId = "";
+/** The tag `docker load` reported for the runtime image, so the container names what was loaded. */
+let runtimeImage = "";
+
+function quote(s: string): string {
+  return `'${s.replaceAll("'", `'\\''`)}'`;
+}
+
+/**
+ * A command on the machine, its exit read from a marker on its own line.
+ *
+ * `exec 2>&1` on its own first line and the marker on its own last line, so a command that carries
+ * a heredoc — the substrate bundle is written with one — terminates where it says it does rather
+ * than swallowing the marker (the fault mesh.test.ts documents).
+ */
+async function on(command: string, timeoutMs?: number): Promise<{ out: string; ok: boolean }> {
+  const { stdout } = await exec(instanceId, MACHINE, [
+    "sh", "-c", `exec 2>&1\n${command}\necho "__exit=$?"`,
+  ], timeoutMs);
+  const marker = stdout.lastIndexOf("__exit=");
+  if (marker < 0) return { out: stdout, ok: false };
+  return { out: stdout.slice(0, marker), ok: stdout.slice(marker + 7).trim() === "0" };
+}
+
+async function must(command: string, timeoutMs?: number): Promise<string> {
+  const { out, ok } = await on(command, timeoutMs);
+  if (!ok) throw new Error(`${MACHINE}: ${command}\n${out}`);
+  return out;
+}
+
+/**
+ * The substrate bundle, its image references pointed at this scenario's own registry.
+ *
+ * A digest belongs to whatever registry serves it, so the committed bundle names a registry that
+ * is not this one; matching by repository and rewriting to the digest this registry assigned is
+ * what makes it applicable (the same rewrite mesh.test.ts does).
+ */
+function bundleFor(images: string[]): string {
+  let text = readFileSync(bundle, "utf8");
+  for (const pinned of images) {
+    const repository = pinned.slice(pinned.indexOf("/") + 1, pinned.indexOf("@"));
+    const escaped = repository.replaceAll("/", "\\/").replaceAll(".", "\\.");
+    text = text.replaceAll(
+      new RegExp(`[A-Za-z0-9_.:-]+\\/${escaped}@sha256:[0-9a-f]+`, "g"),
+      pinned,
+    );
+  }
+  return text;
+}
+
+/** Read the trail back as parsed JSON lines. */
+async function trail(): Promise<Record<string, unknown>[]> {
+  const raw = await must(`cat ${TRAIL} 2>/dev/null || true`);
+  return raw
+    .split("\n")
+    .map((l) => l.trim())
+    .filter(Boolean)
+    .map((l) => JSON.parse(l) as Record<string, unknown>);
+}
+
+before(async () => {
+  if (skip) return;
+
+  const raised = await raise(loadScenario(`scenarios/${SCENARIO}.yml`), {
+    onProgress: (m) => console.log(`raise: ${m}`),
+  });
+  instanceId = raised.instanceId;
+
+  // The node raises its substrate — store, broker and the rest — from the bundle, applied from a
+  // file because the digests are this registry's and are not known until it is up.
+  await must(`cat > /tmp/substrate.lock <<'MESHBUNDLE'\n${bundleFor(raised.images)}\nMESHBUNDLE`);
+  await must(`${HOST_PATH} apply /tmp/substrate.lock`, 600_000);
+
+  const running = await must(`docker ps --format '{{.Names}}'`);
+  assert.match(running, /mesh-broker/, `the substrate did not raise a broker:\n${running}`);
+
+  // Bring the runtime+audit-logger image onto the machine. Loaded, not pulled: the machine has no
+  // route out (novox/hq the lab is a closed address space), so the image arrives as a tar the way
+  // the builder binary does, and `docker load` names what it loaded.
+  await incus([
+    "file", "push", runtime, `${machineName(instanceId, MACHINE)}/tmp/runtime.tar`, "--mode", "0644",
+  ], 300_000);
+  const loaded = await must(`docker load < /tmp/runtime.tar`);
+  const named = loaded.match(/Loaded image:\s*(\S+)/)?.[1];
+  assert.ok(named, `docker load did not name the image:\n${loaded}`);
+  runtimeImage = named;
+
+  // Start the audit-logger: the runtime bound to the mesh's broker, consuming `#`, its trail on a
+  // mounted directory so the assertions read what it actually wrote.
+  await must(`mkdir -p ${TRAIL_DIR}`);
+  await must(
+    `docker run -d --name mesh-audit --network host ` +
+      `-e MESH_BROKER_URL=${quote(BROKER)} -e MESH_MODULE=audit-logger -e MESH_NODE=${MACHINE} ` +
+      `-e AUDIT_LOG=/trail/audit.log -v ${TRAIL_DIR}:/trail ` +
+      `${runtimeImage}`,
+  );
+  // Give the subscription a moment to bind before anything is emitted at it.
+  await new Promise((r) => setTimeout(r, 4000));
+}, { timeout: 1_800_000 });
+
+after(async () => {
+  if (instanceId) await destroy(instanceId);
+  await destroyAll(`${SCENARIO}-`);
+}, { timeout: 600_000 });
+
+test("an emitted event reaches the audit trail with its metadata in headers", {
+  skip, timeout: 300_000,
+}, async () => {
+  // Emitted as a different module (source=probe), so the trail's source is the emitter's, read
+  // back from the header — proof it did not come from the body.
+  await must(
+    `docker exec -e MESH_MODULE=probe -e MESH_NODE=${MACHINE} mesh-audit ` +
+      `node dist/main.js emit module.probe.site.created '{"domain":"my-app"}'`,
+  );
+  // A node-origin event too (novox/hq ADR 0047 reserves module.* / mesh.* / node.*), to show the
+  // audit sink takes all of them, not only a module's.
+  await must(
+    `docker exec -e MESH_MODULE=probe -e MESH_NODE=${MACHINE} mesh-audit ` +
+      `node dist/main.js emit node.${MACHINE}.tick '{"n":1}'`,
+  );
+
+  // Poll: the trail is written by a separate container reacting to the broker, so it is not there
+  // the instant emit returns.
+  let lines: Record<string, unknown>[] = [];
+  const until = Date.now() + 30_000;
+  while (Date.now() < until) {
+    lines = await trail();
+    if (lines.length >= 2) break;
+    await new Promise((r) => setTimeout(r, 2000));
+  }
+
+  const types = lines.map((l) => l.type);
+  assert.ok(
+    types.includes("module.probe.site.created") && types.includes(`node.${MACHINE}.tick`),
+    `both events did not reach the trail; it holds ${JSON.stringify(types)}\n` +
+      `${(await on(`docker logs mesh-audit 2>&1 | tail -20`)).out}`,
+  );
+
+  const site = lines.find((l) => l.type === "module.probe.site.created")!;
+  assert.equal(site.source, "probe", "x-source did not survive as the trail's source");
+  assert.equal(site.node, MACHINE, "x-node did not survive");
+  assert.ok(typeof site.id === "string" && site.id.length > 0, "no x-event-id was recorded");
+  assert.deepEqual(site.body, { domain: "my-app" }, "the body was not exactly the payload");
+});
+
+test("the runtime provisioned the ADR 0047 queue and dead-letter on the raised broker", {
+  skip, timeout: 120_000,
+}, async () => {
+  // Asked of the broker itself, so this is the shape that actually exists on the bus, not the
+  // shape the code intends. A durable per-consumer queue and a dead-letter home are what make the
+  // trail survive a restart and set a poison event aside — assert they are there, not assumed.
+  const queues = await must(`docker exec mesh-broker lavinmqctl list_queues name durable`);
+  assert.match(queues, new RegExp(`${MACHINE}\\.audit-logger\\.events`),
+    `the durable per-consumer queue is missing:\n${queues}`);
+
+  const exchanges = await must(`docker exec mesh-broker lavinmqctl list_exchanges name`);
+  assert.match(exchanges, /mesh\.events(\s|$)/m, `the events exchange is missing:\n${exchanges}`);
+  assert.match(exchanges, /mesh\.events\.dead/, `the dead-letter exchange is missing:\n${exchanges}`);
+});