diff --git a/server/collectors/SystemCollector.js b/server/collectors/SystemCollector.js index ecba788d..1383dbf8 100644 --- a/server/collectors/SystemCollector.js +++ b/server/collectors/SystemCollector.js @@ -4,6 +4,21 @@ import { HOST_PATHS, GPU_MEMORY_JSON_PATH, DGX_SPARK, HARDWARE_DEFAULTS } from " import { normalizeMac, WOL_INTERFACE } from "../wol.js"; import { sshExec } from "./ssh.js"; +export const COLLECTION_SUCCESS = Symbol("sparkdash.collectionSuccess"); + +export function collectionWasSuccessful(result) { + return result?.[COLLECTION_SUCCESS] === true; +} + +function tagCollectionResult(result, successful) { + Object.defineProperty(result, COLLECTION_SUCCESS, { + value: successful === true, + enumerable: false, + configurable: true, + }); + return result; +} + /** * SystemCollector — collects hardware metrics for a Spark. * In Phase 2, this is the LOCAL path only (no SSH). @@ -17,6 +32,7 @@ export class SystemCollector { // Rate-tracking baselines this.lastNetworkStats = new Map(); this.lastCpuStat = null; + this._cpuCollectionSequence = 0; /** Last computed CPU usage percentage (0-100) — used by GPU system-draw estimate. */ this.lastCpuUsagePct = 0; this.lastRaplReading = null; @@ -36,44 +52,94 @@ export class SystemCollector { /** Collect GPU metrics (temperature, usage, power, VRAM). */ async collectGpu() { - if (!this.spark.isLocal) return this._getRemoteGpu(); try { - const gpuData = await this._getGPUAll(); - return gpuData; + const gpuData = this.spark.isLocal + ? await this._getGPUAll() + : await this._getRemoteGpu(); + return tagCollectionResult(gpuData, this._isSuccessfulGpuCollection(gpuData)); } catch (err) { console.error(`[SystemCollector] GPU error for ${this.spark.id}:`, err.message); - return this._defaultGpu(); + return tagCollectionResult(this._defaultGpu(), false); } } /** Collect CPU metrics (usage, temperature, power). */ async collectCpu() { - if (!this.spark.isLocal) return this._getRemoteCpu(); + const collectionSequence = ++this._cpuCollectionSequence; try { + if (!this.spark.isLocal) { + const cpuData = await this._getRemoteCpu(collectionSequence); + return tagCollectionResult(cpuData, this._isSuccessfulCpuCollection(cpuData)); + } + // Read /proc/stat once and compute usage BEFORE estimating power. // Previously _getCPUPower re-read /proc/stat in parallel with _getCPUUsage, // racing on lastCpuStat and producing 0% (idle power) on the first poll. const usage = await this._getCPUUsage(); + if (!this._isValidCpuStat(usage)) { + throw new Error("invalid /proc/stat CPU counters"); + } const totalDiff = usage.total - (this.lastCpuStat?.total || usage.total); const usedDiff = usage.used - (this.lastCpuStat?.used || usage.used); const cpuPercentage = totalDiff > 0 ? Math.round((usedDiff / totalDiff) * 100) : 0; const usageFraction = totalDiff > 0 ? usedDiff / totalDiff : 0; - this.lastCpuStat = usage; - this.lastCpuUsagePct = cpuPercentage; - // Temperature and power can run in parallel — power is now a pure // function of the usage fraction (no extra /proc/stat read). const [temp, power] = await Promise.all([ this._getCPUTemperature(), this._getCPUPower(usageFraction), ]); - return { usage: cpuPercentage, temperature: temp, ...power }; + if (collectionSequence === this._cpuCollectionSequence) { + this.lastCpuStat = usage; + this.lastCpuUsagePct = cpuPercentage; + } + const cpuData = { usage: cpuPercentage, temperature: temp, ...power }; + return tagCollectionResult(cpuData, this._isSuccessfulCpuCollection(cpuData)); } catch (err) { console.error(`[SystemCollector] CPU error for ${this.spark.id}:`, err.message); - return this._defaultCpu(); + return tagCollectionResult(this._defaultCpu(), false); } } + _isSuccessfulGpuCollection(gpu) { + return ( + Number.isFinite(gpu?.temperature) && + gpu.temperature > 0 && + Number.isFinite(gpu?.usage) && + Number.isFinite(gpu?.power?.draw) && + gpu.power.draw >= 0 && + Number.isFinite(gpu?.power?.limit) && + gpu.power.limit > 0 + ); + } + + _isSuccessfulCpuCollection(cpu) { + return ( + Number.isFinite(cpu?.usage) && + cpu.usage >= 0 && + cpu.usage <= 100 && + Number.isFinite(cpu?.draw) && + cpu.draw > 0 && + Number.isFinite(cpu?.tdp) && + cpu.tdp > 0 + ); + } + + _isValidCpuStat(cpuStat) { + return ( + Number.isFinite(cpuStat?.total) && + cpuStat.total > 0 && + Number.isFinite(cpuStat?.used) && + cpuStat.used >= 0 && + cpuStat.used <= cpuStat.total + ); + } + + /** Prevent an earlier monitor lifecycle from updating shared CPU baselines. */ + invalidatePendingCollections() { + this._cpuCollectionSequence += 1; + } + /** Collect RAM metrics. */ async collectRam() { if (!this.spark.isLocal) return this._getRemoteRam(); @@ -1001,7 +1067,10 @@ export class SystemCollector { } } - async _getRemoteCpu() { + async _getRemoteCpu(collectionSequence = null) { + const attemptSequence = Number.isInteger(collectionSequence) + ? collectionSequence + : ++this._cpuCollectionSequence; try { const cmd = [ "cat /proc/stat | head -1", @@ -1025,10 +1094,16 @@ export class SystemCollector { const tempOut = this.spark.kind === "host" ? sections[2] || "" : ""; const cpuStat = this._parseCPUUsage(statOut); + if (!this._isValidCpuStat(cpuStat)) { + throw new Error("invalid remote /proc/stat CPU counters"); + } const totalDiff = cpuStat.total - (this.lastCpuStat?.total || cpuStat.total); const usedDiff = cpuStat.used - (this.lastCpuStat?.used || cpuStat.used); const usage = totalDiff > 0 ? Math.round((usedDiff / totalDiff) * 100) : 0; - this.lastCpuStat = cpuStat; + if (attemptSequence === this._cpuCollectionSequence) { + this.lastCpuStat = cpuStat; + this.lastCpuUsagePct = usage; + } // ARM/Neoverse power estimation const isArm = /CPU architecture:\s*[89]|aarch64|ARMv[89]|armv[89]/i.test(cpuinfoOut); @@ -1545,4 +1620,4 @@ export class SystemCollector { this._nvidiaSmiPath = "nvidia-smi"; return this._nvidiaSmiPath; } -} \ No newline at end of file +} diff --git a/server/sparks/SparkMonitor.js b/server/sparks/SparkMonitor.js index b89b3be0..25756136 100644 --- a/server/sparks/SparkMonitor.js +++ b/server/sparks/SparkMonitor.js @@ -1,6 +1,9 @@ import fs from "fs"; import path from "path"; -import { SystemCollector } from "../collectors/SystemCollector.js"; +import { + SystemCollector, + collectionWasSuccessful, +} from "../collectors/SystemCollector.js"; import { LlmProbe } from "../collectors/LlmProbe.js"; import { ComfyProbe } from "../collectors/ComfyProbe.js"; import { HermesProbe } from "../collectors/HermesProbe.js"; @@ -99,6 +102,7 @@ export class SparkMonitor { tailscale: null, }; this._lastUpdate = {}; + this._metricCollectionSuccessful = { gpu: false, cpu: false }; // Hardware summary: kind "spark" uses the static DGX Spark specs; kind // "host" (dedicated GPU Linux box) detects real hardware once in the @@ -126,7 +130,8 @@ export class SparkMonitor { /** @type {ReturnType | null} */ this._tailscaleIntervalId = null; this._running = false; - /** @type {Record} in-flight domain guards */ + this._runGeneration = 0; + /** @type {Record} in-flight domain guards */ this._inflight = {}; } @@ -137,6 +142,10 @@ export class SparkMonitor { const prevComfyPort = this._comfyPort(this.spark); const wasHermes = this._hermesMonitoringEnabled(this.spark); const wasTailscale = this._tailscaleMonitoringEnabled(this.spark); + this.collector.invalidatePendingCollections(); + this._runGeneration += 1; + this._inflight = {}; + this._metricCollectionSuccessful = { gpu: false, cpu: false }; this.spark = spark; this.collector.spark = spark; @@ -341,6 +350,7 @@ export class SparkMonitor { /** Start background polling. */ start() { if (this._running) return; + this._runGeneration += 1; this._running = true; this._stopped = false; this._poll(); @@ -361,6 +371,9 @@ export class SparkMonitor { /** Stop background polling. */ stop() { + this.collector.invalidatePendingCollections(); + this._runGeneration += 1; + this._metricCollectionSuccessful = { gpu: false, cpu: false }; this._running = false; this._stopped = true; for (const id of this._intervals) clearInterval(id); @@ -452,36 +465,42 @@ export class SparkMonitor { // ─── Liveness ───────────────────────────────────────────── async _checkOnline() { if (!this._running || this._inflight.online) return; - this._inflight.online = true; + const runGeneration = this._runGeneration; + const checkToken = Symbol("online"); + this._inflight.online = checkToken; + const isCurrentRun = () => + this._running && this._runGeneration === runGeneration; try { if (this.spark.isLocal) { await this.collector.pingHost(); } else { const result = await sshTest(this.spark); - // Re-check after the (up to 10s) SSH await — `stop()` may have fired - // mid-flight (removeSpark / updateSpark). Bail before mutating state or - // running into a stopped registry entry. - if (!this._running) return; + if (!isCurrentRun()) return; if (!result.ok) throw new Error(result.message); } - if (!this._running) return; - this.online = true; - this.lastOnlineOk = Date.now(); + if (!isCurrentRun()) return; // Collect system uptime + let uptimeSeconds = this._uptimeSeconds; try { - this._uptimeSeconds = await this._readUptime(); + uptimeSeconds = await this._readUptime(); } catch { // Non-fatal — uptime stays at previous value or null } + if (!isCurrentRun()) return; + this.online = true; + this.lastOnlineOk = Date.now(); + this._uptimeSeconds = uptimeSeconds; } catch { - if (!this._running) return; + if (!isCurrentRun()) return; if (!this.lastOnlineOk || Date.now() - this.lastOnlineOk > ONLINE_GRACE_MS) { this.online = false; this._uptimeSeconds = null; } } finally { - this._inflight.online = false; + if (this._inflight.online === checkToken) { + this._inflight.online = false; + } } } @@ -512,7 +531,9 @@ export class SparkMonitor { if (domain === "comfy" && !this._comfyMonitoringEnabled()) return; if (domain === "hermes" && !this._hermesMonitoringEnabled()) return; if (domain === "tailscale" && !this._tailscaleMonitoringEnabled()) return; - this._inflight[domain] = true; + const runGeneration = this._runGeneration; + const pollToken = Symbol(domain); + this._inflight[domain] = pollToken; try { let result; switch (domain) { @@ -555,13 +576,15 @@ export class SparkMonitor { // isn't user-visible (monitors.delete already happened) but it's a // latent class of bug worth killing, and a replaced monitor could // otherwise race the tail-end await onto the wrong object. - if (!this._running) return; + if (!this._running || this._runGeneration !== runGeneration) return; switch (domain) { case "gpu": this._metrics.gpu = result; + this._metricCollectionSuccessful.gpu = collectionWasSuccessful(result); break; case "cpu": this._metrics.cpu = result; + this._metricCollectionSuccessful.cpu = collectionWasSuccessful(result); break; case "ram": this._metrics.ram = result; @@ -604,34 +627,39 @@ export class SparkMonitor { } this._lastUpdate[domain] = Date.now(); } catch (err) { + if ( + this._running && + this._runGeneration === runGeneration && + (domain === "gpu" || domain === "cpu") + ) { + this._metricCollectionSuccessful[domain] = false; + } console.error(`[SparkMonitor] ${this.spark.id} ${domain} poll error:`, err.message); } finally { - this._inflight[domain] = false; + if (this._inflight[domain] === pollToken) { + this._inflight[domain] = false; + } } } /** Manually refresh a single domain, bypassing auto-poll guards. */ async refreshDomain(domain) { - if (this._inflight[domain]) return; - this._inflight[domain] = true; + if (domain !== "storage") return this._pollDomain(domain); + if (!this._running || this._inflight[domain]) return; + const runGeneration = this._runGeneration; + const refreshToken = Symbol(domain); + this._inflight[domain] = refreshToken; try { - let result; - switch (domain) { - case "storage": - result = await this.collector.collectStorage(); - break; - default: - // Fall back to _pollDomain for other domains - this._inflight[domain] = false; - return this._pollDomain(domain); - } - if (!this._running) return; + const result = await this.collector.collectStorage(); + if (!this._running || this._runGeneration !== runGeneration) return; this._metrics.storage = result; this._lastUpdate[domain] = Date.now(); } catch (err) { console.error(`[SparkMonitor] ${this.spark.id} ${domain} refresh error:`, err.message); } finally { - this._inflight[domain] = false; + if (this._inflight[domain] === refreshToken) { + this._inflight[domain] = false; + } } } @@ -777,4 +805,3 @@ export class SparkMonitor { }; } } - diff --git a/server/sparks/__tests__/monitor-lifecycle.test.js b/server/sparks/__tests__/monitor-lifecycle.test.js new file mode 100644 index 00000000..91d8536f --- /dev/null +++ b/server/sparks/__tests__/monitor-lifecycle.test.js @@ -0,0 +1,239 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { SparkMonitor } from "../SparkMonitor.js"; +import { + COLLECTION_SUCCESS, + SystemCollector, + collectionWasSuccessful, +} from "../../collectors/SystemCollector.js"; + +function spark() { + return { + id: "spark-test", + name: "Spark Test", + lanIp: "127.0.0.1", + isLocal: true, + llmMonitoring: false, + comfyMonitoring: false, + }; +} + +function validGpu(temperature = 42) { + return { + temperature, + usage: 10, + power: { draw: 18.5, limit: 120, systemDraw: 39 }, + vram: { used: 100, total: 128_000, percentage: 0, available: 120_000 }, + processes: [], + throttle: {}, + }; +} + +function tagged(result, successful = true) { + Object.defineProperty(result, COLLECTION_SUCCESS, { + value: successful, + enumerable: false, + }); + return result; +} + +test("collector provenance rejects partial GPU and malformed CPU samples", async (t) => { + t.mock.method(console, "error", () => {}); + const remote = new SystemCollector({ id: "remote", isLocal: false }); + remote._getRemoteGpu = async () => ({ + ...validGpu(), + power: { draw: Number.NaN, limit: 120, systemDraw: 39 }, + }); + assert.equal(collectionWasSuccessful(await remote.collectGpu()), false); + + const local = new SystemCollector({ id: "local", isLocal: true }); + local._getCPUUsage = async () => ({ total: 0, used: 0 }); + assert.equal(collectionWasSuccessful(await local.collectCpu()), false); +}); + +test("monitor publishes per-domain collection provenance", async () => { + const monitor = new SparkMonitor(spark()); + monitor._running = true; + monitor.collector.collectGpu = async () => tagged(validGpu()); + + await monitor._pollDomain("gpu"); + assert.equal(monitor._metricCollectionSuccessful.gpu, true); + + monitor.collector.collectGpu = async () => validGpu(43); + await monitor._pollDomain("gpu"); + assert.equal(monitor._metricCollectionSuccessful.gpu, false); + + monitor.updateConfig({ ...spark(), lanIp: "127.0.0.2" }); + assert.deepEqual(monitor._metricCollectionSuccessful, { gpu: false, cpu: false }); +}); + +test("a rejected prior-run poll cannot clear current collection provenance", async (t) => { + t.mock.method(console, "error", () => {}); + const monitor = new SparkMonitor(spark()); + monitor._running = true; + let rejectPrior; + monitor.collector.collectGpu = () => + new Promise((_resolve, reject) => { + rejectPrior = reject; + }); + + const priorPoll = monitor._pollDomain("gpu"); + await Promise.resolve(); + monitor.updateConfig({ ...spark(), lanIp: "127.0.0.2" }); + monitor.collector.collectGpu = async () => tagged(validGpu(43)); + await monitor._pollDomain("gpu"); + assert.equal(monitor._metricCollectionSuccessful.gpu, true); + + rejectPrior(new Error("prior target failed late")); + await priorPoll; + assert.equal(monitor._metricCollectionSuccessful.gpu, true); +}); + +test("a poll from an earlier monitor run cannot commit or clear a restarted poll", async (t) => { + t.mock.method(console, "log", () => {}); + t.mock.method(globalThis, "setInterval", () => Symbol("interval")); + t.mock.method(globalThis, "clearInterval", () => {}); + const monitor = new SparkMonitor(spark()); + monitor._poll = async () => {}; + const pendingResolvers = []; + monitor.collector.collectGpu = () => + new Promise((resolve) => pendingResolvers.push(resolve)); + + monitor.start(); + const priorRunPoll = monitor._pollDomain("gpu"); + await Promise.resolve(); + monitor.stop(); + monitor.start(); + const currentRunPoll = monitor._pollDomain("gpu"); + await Promise.resolve(); + + pendingResolvers[0](validGpu(41)); + await priorRunPoll; + assert.equal(monitor.snapshot().metrics.gpu.temperature, 0); + assert.equal(monitor._lastUpdate.gpu, undefined); + assert.ok(monitor._inflight.gpu, "the earlier poll must not clear the current guard"); + + pendingResolvers[1](validGpu(42)); + await currentRunPoll; + assert.equal(monitor.snapshot().metrics.gpu.temperature, 42); + assert.ok(Number.isFinite(monitor._lastUpdate.gpu)); + monitor.stop(); +}); + +test("a poll from before updateConfig cannot commit against the new target", async () => { + const monitor = new SparkMonitor(spark()); + monitor._running = true; + const pendingResolvers = []; + monitor.collector.collectGpu = () => + new Promise((resolve) => pendingResolvers.push(resolve)); + + const priorTargetPoll = monitor._pollDomain("gpu"); + await Promise.resolve(); + monitor.updateConfig({ ...spark(), lanIp: "127.0.0.2" }); + const currentTargetPoll = monitor._pollDomain("gpu"); + await Promise.resolve(); + + pendingResolvers[0](validGpu(41)); + await priorTargetPoll; + assert.equal(monitor.snapshot().metrics.gpu.temperature, 0); + assert.ok(monitor._inflight.gpu, "the prior target poll must not clear the current guard"); + + pendingResolvers[1](validGpu(42)); + await currentTargetPoll; + assert.equal(monitor.snapshot().metrics.gpu.temperature, 42); +}); + +test("a rejected older CPU poll cannot rewind the accepted generation baseline", async () => { + const monitor = new SparkMonitor(spark()); + monitor._running = true; + monitor.collector.lastCpuStat = { total: 100, used: 20 }; + let resolveOlder; + let resolveCurrent; + let callCount = 0; + monitor.collector._getCPUUsage = () => { + callCount += 1; + if (callCount === 1) return new Promise((resolve) => (resolveOlder = resolve)); + if (callCount === 2) return new Promise((resolve) => (resolveCurrent = resolve)); + return Promise.resolve({ total: 250, used: 120 }); + }; + monitor.collector._getCPUTemperature = async () => 0; + monitor.collector._getCPUPower = async () => ({ draw: 5.2, tdp: 65 }); + + const olderPoll = monitor._pollDomain("cpu"); + await Promise.resolve(); + monitor.updateConfig({ ...spark(), lanIp: "127.0.0.2" }); + const currentPoll = monitor._pollDomain("cpu"); + await Promise.resolve(); + + resolveCurrent({ total: 200, used: 100 }); + await currentPoll; + assert.equal(monitor.snapshot().metrics.cpu.usage, 80); + + resolveOlder({ total: 150, used: 70 }); + await olderPoll; + await monitor._pollDomain("cpu"); + assert.equal(monitor.snapshot().metrics.cpu.usage, 40); +}); + +test("a liveness check from an earlier run cannot commit or clear a restarted check", async (t) => { + t.mock.method(console, "log", () => {}); + t.mock.method(globalThis, "setInterval", () => Symbol("interval")); + t.mock.method(globalThis, "clearInterval", () => {}); + const monitor = new SparkMonitor(spark()); + monitor._poll = async () => {}; + monitor._readUptime = async () => 123; + const pendingResolvers = []; + monitor.collector.pingHost = () => + new Promise((resolve) => pendingResolvers.push(resolve)); + + monitor.start(); + const priorRunCheck = monitor._checkOnline(); + await Promise.resolve(); + monitor.stop(); + monitor.start(); + const currentRunCheck = monitor._checkOnline(); + await Promise.resolve(); + + pendingResolvers[0](); + await priorRunCheck; + assert.equal(monitor.online, false); + assert.ok(monitor._inflight.online, "the earlier check must not clear the current guard"); + + pendingResolvers[1](); + await currentRunCheck; + assert.equal(monitor.online, true); + assert.equal(monitor._uptimeSeconds, 123); + assert.equal(monitor._inflight.online, false); + monitor.stop(); +}); + +test("a storage refresh from an earlier run cannot commit or clear a restarted refresh", async (t) => { + t.mock.method(console, "log", () => {}); + t.mock.method(globalThis, "setInterval", () => Symbol("interval")); + t.mock.method(globalThis, "clearInterval", () => {}); + const monitor = new SparkMonitor(spark()); + monitor._poll = async () => {}; + const pendingResolvers = []; + monitor.collector.collectStorage = () => + new Promise((resolve) => pendingResolvers.push(resolve)); + + monitor.start(); + const priorRunRefresh = monitor.refreshDomain("storage"); + await Promise.resolve(); + monitor.stop(); + monitor.start(); + const currentRunRefresh = monitor.refreshDomain("storage"); + await Promise.resolve(); + + pendingResolvers[0]([{ mount: "/old" }]); + await priorRunRefresh; + assert.deepEqual(monitor.snapshot().metrics.storage, []); + assert.ok(monitor._inflight.storage, "the earlier refresh must not clear the current guard"); + + pendingResolvers[1]([{ mount: "/current" }]); + await currentRunRefresh; + assert.deepEqual(monitor.snapshot().metrics.storage, [{ mount: "/current" }]); + assert.equal(monitor._inflight.storage, false); + monitor.stop(); +});