clusterflux-public/scripts/quic-smoke.js

#!/usr/bin/env node

const assert = require("assert");
const cp = require("child_process");
const net = require("net");
const path = require("path");
const { coordinatorWireRequest } = require("./coordinator-wire");

const repo = path.resolve(__dirname, "..");

function waitForJsonLine(child) {
  return new Promise((resolve, reject) => {
    let buffer = "";
    child.stdout.on("data", (chunk) => {
      buffer += chunk.toString();
      const newline = buffer.indexOf("\n");
      if (newline < 0) return;
      try {
        resolve(JSON.parse(buffer.slice(0, newline).trim()));
      } catch (error) {
        reject(error);
      }
    });
    child.once("exit", (code) => {
      reject(new Error(`process exited before JSON line with code ${code}`));
    });
  });
}

function send(addr, message) {
  return new Promise((resolve, reject) => {
    const socket = net.connect(addr.port, addr.host, () => {
      socket.write(`${JSON.stringify(coordinatorWireRequest(message))}\n`);
    });
    let buffer = "";
    socket.on("data", (chunk) => {
      buffer += chunk.toString();
      const newline = buffer.indexOf("\n");
      if (newline < 0) return;
      socket.end();
      try {
        resolve(JSON.parse(buffer.slice(0, newline)));
      } catch (error) {
        reject(error);
      }
    });
    socket.on("error", reject);
  });
}

function rendezvousRequest(overrides = {}) {
  return {
    type: "request_rendezvous",
    scope: {
      tenant: "tenant",
      project: "project",
      process: "vp-quic",
      object: { Artifact: "quic-artifact" },
      authorization_subject: "node-a-to-node-b",
    },
    source: {
      node: "node-a",
      advertised_addr: "node-a.mesh.invalid:4433",
      public_key_fingerprint: "sha256:node-a-public-key",
    },
    destination: {
      node: "node-b",
      advertised_addr: "node-b.mesh.invalid:4433",
      public_key_fingerprint: "sha256:node-b-public-key",
    },
    direct_connectivity: true,
    failure_reason: "",
    ...overrides,
  };
}

(async () => {
  const output = cp.execFileSync(
    "cargo",
    ["run", "-q", "-p", "clusterflux-node", "--bin", "clusterflux-quic-smoke"],
    { cwd: repo, encoding: "utf8" }
  );
  const report = JSON.parse(output.trim().split("\n").at(-1));

  assert.strictEqual(report.kind, "clusterflux_quic_smoke");
  assert.strictEqual(report.transport, "NativeQuic");
  assert.strictEqual(report.rust_native_quic, true);
  assert.strictEqual(report.authenticated_direct_connection, true);
  assert.strictEqual(report.coordinator_assisted_rendezvous, true);
  assert.strictEqual(report.coordinator_bulk_relay_allowed, false);
  assert.strictEqual(report.source_node, "node-a");
  assert.strictEqual(report.destination_node, "node-b");
  assert.strictEqual(report.scope.tenant, "tenant");
  assert.strictEqual(report.scope.project, "project");
  assert.strictEqual(report.scope.process, "vp-quic");
  assert.deepStrictEqual(report.scope.object, { Artifact: "quic-artifact" });
  assert.ok(report.authorization_digest.startsWith("sha256:"));
  assert.ok(report.request_bytes > 0);
  assert.strictEqual(report.server_received_request_bytes, report.request_bytes);
  assert.ok(report.payload_bytes > 0);

  const coordinator = cp.spawn(
    "cargo",
    [
      "run",
      "-q",
      "-p",
      "clusterflux-coordinator",
      "--bin",
      "clusterflux-coordinator",
      "--",
      "--listen",
      "127.0.0.1:0",
      "--allow-local-trusted-loopback",
    ],
    { cwd: repo }
  );

  try {
    const ready = await waitForJsonLine(coordinator);
    const [host, portText] = ready.listen.split(":");
    const addr = { host, port: Number(portText) };

    const plan = await send(addr, rendezvousRequest());
    assert.strictEqual(plan.type, "rendezvous_plan");
    assert.strictEqual(plan.charged_rendezvous_attempts, 1);
    assert.strictEqual(plan.plan.transport, "NativeQuic");
    assert.strictEqual(plan.plan.scope.tenant, "tenant");
    assert.strictEqual(plan.plan.scope.project, "project");
    assert.strictEqual(plan.plan.source.node, "node-a");
    assert.strictEqual(plan.plan.destination.node, "node-b");
    assert.strictEqual(plan.plan.coordinator_assisted_rendezvous, true);
    assert.strictEqual(plan.plan.coordinator_bulk_relay_allowed, false);
    assert.ok(plan.plan.authorization_digest.startsWith("sha256:"));

    const failed = await send(
      addr,
      rendezvousRequest({
        direct_connectivity: false,
        failure_reason: "nat traversal failed",
      })
    );
    assert.strictEqual(failed.type, "error");
    assert.match(failed.message, /nat traversal failed/);
    assert.match(failed.message, /coordinator bulk relay is disabled/);
  } finally {
    coordinator.kill("SIGTERM");
  }

  console.log("QUIC smoke passed");
})().catch((error) => {
  console.error(error.stack || error.message);
  process.exit(1);
});