clusterflux-public/scripts/self-hosted-coordinator-smoke.js

#!/usr/bin/env node

const assert = require("assert");
const cp = require("child_process");
const fs = require("fs");
const net = require("net");
const path = require("path");

const repo = path.resolve(__dirname, "..");
const teamArtifactDigest = `sha256:${"a".repeat(64)}`;
const releaseManifestPath =
  process.env.DISASMER_PUBLIC_RELEASE_MANIFEST ||
  path.join(repo, "target/public-release-dryrun/public-release-manifest.json");

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(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 linuxNodeCapabilities() {
  return {
    os: "Linux",
    arch: "x86_64",
    capabilities: ["Command", "RootlessPodman", "VfsArtifacts", "QuicDirect"],
    environment_backends: ["Container"],
    source_providers: ["filesystem", "git"]
  };
}

function commandOutput(command, args) {
  try {
    return cp
      .execFileSync(command, args, {
        cwd: repo,
        encoding: "utf8",
        stdio: ["ignore", "pipe", "ignore"],
      })
      .trim();
  } catch (_) {
    return null;
  }
}

function expectedSourceCommit() {
  return (
    process.env.DISASMER_ACCEPTANCE_COMMIT ||
    commandOutput("git", ["rev-parse", "HEAD"])
  );
}

function readReleaseManifest() {
  if (!fs.existsSync(releaseManifestPath)) return null;
  const manifest = JSON.parse(fs.readFileSync(releaseManifestPath, "utf8"));
  assert.strictEqual(manifest.kind, "disasmer-public-release-dryrun");
  const expectedCommit = expectedSourceCommit();
  if (expectedCommit) {
    assert.strictEqual(
      manifest.source_commit,
      expectedCommit,
      "self-hosted coordinator smoke must use the manifest for the current acceptance commit"
    );
  }
  return manifest;
}

function releaseIdentity() {
  const manifest = readReleaseManifest();
  return {
    sourceCommit: manifest ? manifest.source_commit : expectedSourceCommit(),
    releaseName:
      (manifest && manifest.release_name) ||
      process.env.DISASMER_PUBLIC_RELEASE_NAME ||
      null,
  };
}

async function attachTrustedNode(addr, node) {
  const attached = await send(addr, {
    type: "attach_node",
    tenant: "team",
    project: "self-hosted",
    node,
    public_key: `${node}-public-key`
  });
  assert.strictEqual(attached.type, "node_attached");

  const heartbeat = await send(addr, {
    type: "node_heartbeat",
    node
  });
  assert.strictEqual(heartbeat.type, "node_heartbeat");

  const reported = await send(addr, {
    type: "report_node_capabilities",
    tenant: "team",
    project: "self-hosted",
    node,
    capabilities: linuxNodeCapabilities(),
    cached_environment_digests: ["sha256:env-team-linux"],
    dependency_cache_digests: ["sha256:cargo-cache"],
    source_snapshots: ["sha256:source-team"],
    artifact_locations: [],
    direct_connectivity: true,
    online: true
  });
  assert.strictEqual(reported.type, "node_capabilities_recorded");
}

(async () => {
  const release = releaseIdentity();
  const coordinator = cp.spawn(
    "cargo",
    [
      "run",
      "-q",
      "-p",
      "disasmer-coordinator",
      "--bin",
      "disasmer-coordinator",
      "--",
      "--listen",
      "127.0.0.1:0"
    ],
    { cwd: repo }
  );

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

    await attachTrustedNode(addr, "team-linux-a");
    await attachTrustedNode(addr, "team-linux-b");

    const placement = await send(addr, {
      type: "schedule_task",
      tenant: "team",
      project: "self-hosted",
      environment: {
        os: "Linux",
        arch: "x86_64",
        capabilities: ["Command", "QuicDirect"]
      },
      environment_digest: "sha256:env-team-linux",
      required_capabilities: ["VfsArtifacts"],
      source_snapshot: "sha256:source-team",
      required_artifacts: [],
      prefer_node: "team-linux-a"
    });
    assert.strictEqual(placement.type, "task_placement");
    assert.strictEqual(placement.placement.node, "team-linux-a");
    assert(placement.placement.reasons.includes("preferred node"));
    assert(placement.placement.reasons.includes("warm environment cache"));
    assert(placement.placement.reasons.includes("source snapshot already local"));

    const started = await send(addr, {
      type: "start_process",
      tenant: "team",
      project: "self-hosted",
      process: "vp-team-build"
    });
    assert.strictEqual(started.type, "process_started");

    const reconnected = await send(addr, {
      type: "reconnect_node",
      node: "team-linux-a",
      process: "vp-team-build",
      epoch: started.epoch
    });
    assert.strictEqual(reconnected.type, "node_reconnected");

    const completed = await send(addr, {
      type: "task_completed",
      tenant: "team",
      project: "self-hosted",
      process: "vp-team-build",
      node: "team-linux-a",
      task: "compile-linux",
      status_code: 0,
      stdout_bytes: 18,
      stderr_bytes: 0,
      artifact_path: "/vfs/artifacts/team-output.txt",
      artifact_digest: teamArtifactDigest,
      artifact_size_bytes: 18
    });
    assert.strictEqual(completed.type, "task_recorded");

    const events = await send(addr, {
      type: "list_task_events",
      tenant: "team",
      project: "self-hosted",
      actor_user: "developer",
      process: "vp-team-build"
    });
    assert.strictEqual(events.type, "task_events");
    assert.strictEqual(events.events.length, 1);
    assert.strictEqual(events.events[0].node, "team-linux-a");
    assert.strictEqual(events.events[0].artifact_path, "/vfs/artifacts/team-output.txt");

    const link = await send(addr, {
      type: "create_artifact_download_link",
      tenant: "team",
      project: "self-hosted",
      actor_user: "developer",
      artifact: "team-output.txt",
      max_bytes: 1024 * 1024,
      token_nonce: "team-download",
      now_epoch_seconds: 10,
      ttl_seconds: 60
    });
    assert.strictEqual(link.type, "artifact_download_link");
    assert.deepStrictEqual(link.link.source, { RetainedNode: "team-linux-a" });

    const exportPlan = await send(addr, {
      type: "export_artifact_to_node",
      tenant: "team",
      project: "self-hosted",
      actor_user: "developer",
      artifact: "team-output.txt",
      receiver_node: "team-linux-b",
      direct_connectivity: true,
      failure_reason: ""
    });
    assert.strictEqual(exportPlan.type, "artifact_export_plan");
    assert.strictEqual(exportPlan.source_node, "team-linux-a");
    assert.strictEqual(exportPlan.receiver_node, "team-linux-b");
    assert.strictEqual(exportPlan.plan.transport, "NativeQuic");
    assert.strictEqual(exportPlan.plan.coordinator_assisted_rendezvous, true);
    assert.strictEqual(exportPlan.plan.coordinator_bulk_relay_allowed, false);

    const reportPath = path.join(repo, "target/acceptance/public-coordinator-compat.json");
    fs.mkdirSync(path.dirname(reportPath), { recursive: true });
    fs.writeFileSync(
      reportPath,
      `${JSON.stringify(
        {
          kind: "disasmer-public-coordinator-compatibility",
          source_commit: release.sourceCommit,
          release_name: release.releaseName,
          operator_implementation: "standalone-public-coordinator",
          coordinator_addr: ready.listen,
          ping: "pong",
          nodes: ["team-linux-a", "team-linux-b"],
          task_placement: placement.type,
          process_started: started.type,
          task_completion: completed.type,
          task_events: events.events.length,
          artifact_download_link: link.type,
          artifact_export_plan: exportPlan.type,
        },
        null,
        2
      )}\n`
    );
  } finally {
    coordinator.kill("SIGTERM");
  }

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