fix: isolate monitor polling lifecycles

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;
   }
-}
+}

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<typeof setInterval> | null} */
     this._tailscaleIntervalId = null;
     this._running = false;
-    /** @type {Record<string, boolean>} in-flight domain guards */
+    this._runGeneration = 0;
+    /** @type {Record<string, boolean | symbol>} 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 {
     };
   }
 }
-