#!/usr/bin/env node const assert = require("assert"); const cp = require("child_process"); const net = require("net"); const path = require("path"); 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(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 linuxCapabilities() { return { os: "Linux", arch: "x86_64", capabilities: [ "Command", "Containers", "RootlessPodman", "VfsArtifacts" ], environment_backends: ["Container"], source_providers: ["filesystem"] }; } async function attachNode(addr, node) { const attached = await send(addr, { type: "attach_node", tenant: "tenant", project: "project", node, public_key: `${node}-public-key` }); assert.strictEqual(attached.type, "node_attached"); assert.strictEqual(attached.node, node); } async function reportNode(addr, node, locality) { const recorded = await send(addr, { type: "report_node_capabilities", tenant: "tenant", project: "project", node, capabilities: linuxCapabilities(), cached_environment_digests: locality.cached_environment_digests, dependency_cache_digests: locality.dependency_cache_digests, source_snapshots: locality.source_snapshots, artifact_locations: locality.artifact_locations, direct_connectivity: locality.direct_connectivity !== false, online: true }); assert.strictEqual(recorded.type, "node_capabilities_recorded"); assert.strictEqual(recorded.node, node); return recorded; } (async () => { const coordinator = cp.spawn( "cargo", [ "run", "-q", "-p", "disasmer-coordinator", "--bin", "disasmer-coordinator", "--", "--listen", "127.0.0.1:0" ], { cwd: repo } ); let coordinatorStderr = ""; coordinator.stderr.on("data", (chunk) => { coordinatorStderr += chunk.toString(); }); 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 attachNode(addr, "cold-node"); await attachNode(addr, "warm-node"); const cold = await reportNode(addr, "cold-node", { cached_environment_digests: [], dependency_cache_digests: [], source_snapshots: [], artifact_locations: [], direct_connectivity: false }); assert.strictEqual(cold.node_descriptors, 1); const warm = await reportNode(addr, "warm-node", { cached_environment_digests: ["sha256:env-linux-container"], dependency_cache_digests: ["sha256:deps-toolchain"], source_snapshots: ["sha256:source-tree"], artifact_locations: ["toolchain-cache"] }); assert.strictEqual(warm.node_descriptors, 2); const reportedNodes = new Set([cold.node, warm.node]); assert.strictEqual(reportedNodes.size, 2); assert(reportedNodes.has("cold-node")); assert(reportedNodes.has("warm-node")); const inspected = await send(addr, { type: "list_node_descriptors", tenant: "tenant", project: "project", actor_user: "operator" }); assert.strictEqual(inspected.type, "node_descriptors"); assert.strictEqual(inspected.actor, "operator"); assert.strictEqual(inspected.descriptors.length, 2); const warmDescriptor = inspected.descriptors.find( (descriptor) => descriptor.id === "warm-node" ); assert(warmDescriptor, "warm node descriptor must be visible to inspector state"); assert(warmDescriptor.capabilities.capabilities.includes("Command")); assert(warmDescriptor.capabilities.capabilities.includes("RootlessPodman")); assert(warmDescriptor.cached_environments.includes("sha256:env-linux-container")); assert(warmDescriptor.dependency_caches.includes("sha256:deps-toolchain")); assert(warmDescriptor.source_snapshots.includes("sha256:source-tree")); assert(warmDescriptor.artifact_locations.includes("toolchain-cache")); const crossScopeInspection = await send(addr, { type: "list_node_descriptors", tenant: "other-tenant", project: "project", actor_user: "operator" }); assert.strictEqual(crossScopeInspection.type, "node_descriptors"); assert.strictEqual(crossScopeInspection.descriptors.length, 0); const crossTenantReport = await send(addr, { type: "report_node_capabilities", tenant: "other-tenant", project: "project", node: "warm-node", capabilities: linuxCapabilities(), cached_environment_digests: [], dependency_cache_digests: [], source_snapshots: [], artifact_locations: [], direct_connectivity: true, online: true }); assert.strictEqual(crossTenantReport.type, "error"); assert.match(crossTenantReport.message, /tenant\/project scope/); const placement = await send(addr, { type: "schedule_task", tenant: "tenant", project: "project", environment: { os: "Linux", arch: null, capabilities: ["Containers", "RootlessPodman"] }, environment_digest: "sha256:env-linux-container", required_capabilities: ["Command"], dependency_cache: "sha256:deps-toolchain", source_snapshot: "sha256:source-tree", required_artifacts: ["toolchain-cache"], prefer_node: null }); assert.strictEqual(placement.type, "task_placement"); assert.strictEqual(placement.placement.node, "warm-node"); assert.ok(placement.placement.score > 0); assert.ok(placement.placement.reasons.includes("warm environment cache")); assert.ok(placement.placement.reasons.includes("warm dependency cache")); assert.ok(placement.placement.reasons.includes("source snapshot already local")); assert.ok( placement.placement.reasons.includes("1 required artifact(s) already local") ); const impossible = await send(addr, { type: "schedule_task", tenant: "tenant", project: "project", environment: null, environment_digest: null, required_capabilities: ["WindowsCommandDev"], dependency_cache: null, source_snapshot: null, required_artifacts: [], prefer_node: null }); assert.strictEqual(impossible.type, "error"); assert.match(impossible.message, /WindowsCommandDev/); const quotaDenied = await send(addr, { type: "schedule_task", tenant: "tenant", project: "project", environment: null, environment_digest: null, required_capabilities: ["Command"], dependency_cache: null, source_snapshot: null, required_artifacts: [], quota_available: false, policy_allowed: true, prefer_node: null }); assert.strictEqual(quotaDenied.type, "error"); assert.match(quotaDenied.message, /quota unavailable for placement/); const policyDenied = await send(addr, { type: "schedule_task", tenant: "tenant", project: "project", environment: null, environment_digest: null, required_capabilities: ["Command"], dependency_cache: null, source_snapshot: null, required_artifacts: [], quota_available: true, policy_allowed: false, prefer_node: null }); assert.strictEqual(policyDenied.type, "error"); assert.match(policyDenied.message, /policy denied placement/); await reportNode(addr, "warm-node", { cached_environment_digests: [], dependency_cache_digests: [], source_snapshots: [], artifact_locations: [], direct_connectivity: false }); const disconnectedTransfer = await send(addr, { type: "schedule_task", tenant: "tenant", project: "project", environment: null, environment_digest: null, required_capabilities: ["Command"], dependency_cache: null, source_snapshot: "sha256:source-tree", required_artifacts: ["toolchain-cache"], prefer_node: null }); assert.strictEqual(disconnectedTransfer.type, "error"); assert.match(disconnectedTransfer.message, /source snapshot unavailable/); assert.match(disconnectedTransfer.message, /required artifact\(s\) unavailable/); assert.match(disconnectedTransfer.message, /direct connectivity unavailable/); } catch (error) { if (coordinatorStderr) { error.message = `${error.message}\ncoordinator stderr:\n${coordinatorStderr}`; } throw error; } finally { coordinator.kill("SIGTERM"); } console.log("Scheduler placement smoke passed"); })().catch((error) => { console.error(error.stack || error.message); process.exit(1); });