feat: ship host metrics as a first-party in-process plugin

src/backend/tests/hosts/metrics-shared/common-utils.test.ts

@@ -0,0 +1,41 @@
+import { describe, it, expect } from "vitest";
+import {
+  toFixedNum,
+  kibToGiB,
+  encodePowerShellCommand,
+} from "../../../hosts/metrics-shared/common-utils.js";
+
+describe("toFixedNum", () => {
+  it("rounds to the requested digit count", () => {
+    expect(toFixedNum(3.14159, 2)).toBe(3.14);
+    expect(toFixedNum(3.14159, 0)).toBe(3);
+    expect(toFixedNum(2.5, 0)).toBe(3);
+  });
+
+  it("defaults to 2 digits", () => {
+    expect(toFixedNum(1.23456)).toBe(1.23);
+  });
+
+  it("returns null for non-finite or non-number input", () => {
+    expect(toFixedNum(null)).toBeNull();
+    expect(toFixedNum(undefined)).toBeNull();
+    expect(toFixedNum(NaN)).toBeNull();
+    expect(toFixedNum(Infinity)).toBeNull();
+  });
+});
+
+describe("kibToGiB", () => {
+  it("converts kibibytes to gibibytes", () => {
+    expect(kibToGiB(1024 * 1024)).toBe(1);
+    expect(kibToGiB(0)).toBe(0);
+    expect(kibToGiB(2 * 1024 * 1024)).toBe(2);
+  });
+});
+
+describe("encodePowerShellCommand", () => {
+  it("UTF-16LE base64 encodes the script, decodable back to the source", () => {
+    const script = "Write-Output 'hello'";
+    const encoded = encodePowerShellCommand(script);
+    expect(Buffer.from(encoded, "base64").toString("utf16le")).toBe(script);
+  });
+});

src/backend/tests/hosts/metrics-shared/exec-elevated.test.ts

@@ -0,0 +1,111 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+
+const execCommand = vi.fn();
+vi.mock("../../../hosts/metrics-shared/common-utils.js", () => ({
+  execCommand: (...args: unknown[]) => execCommand(...args),
+}));
+
+import {
+  execElevated,
+  ElevationError,
+} from "../../../hosts/metrics-shared/exec-elevated.js";
+import type { Client } from "ssh2";
+
+const fakeClient = {} as Client;
+
+function result(stdout: string, stderr = "", code: number | null = 0) {
+  return { stdout, stderr, code };
+}
+
+beforeEach(() => {
+  execCommand.mockReset();
+});
+
+describe("execElevated (no force)", () => {
+  it("returns the direct result when the command succeeds unprivileged", async () => {
+    execCommand.mockResolvedValueOnce(result("hello", "", 0));
+    const r = await execElevated(fakeClient, "echo hello", "pw");
+    expect(r.usedSudo).toBe(false);
+    expect(r.stdout).toBe("hello");
+    expect(execCommand).toHaveBeenCalledTimes(1);
+  });
+
+  it("does NOT escalate when stdout merely contains scary words", async () => {
+    // A non-zero exit whose OUTPUT contains 'permission denied' but stderr does
+    // not should be surfaced as-is, never retried under sudo.
+    execCommand.mockResolvedValueOnce(
+      result("log line: permission denied for user foo", "", 1),
+    );
+    const r = await execElevated(fakeClient, "grep denied /tmp/app.log", "pw");
+    expect(r.usedSudo).toBe(false);
+    expect(execCommand).toHaveBeenCalledTimes(1);
+  });
+
+  it("escalates when stderr indicates a permission problem", async () => {
+    execCommand
+      .mockResolvedValueOnce(result("", "Permission denied", 1))
+      .mockResolvedValueOnce(result("__TX_SUDO_OK__\nelevated output", "", 0));
+    const r = await execElevated(fakeClient, "cat /etc/shadow", "pw");
+    expect(r.usedSudo).toBe(true);
+    expect(r.stdout).toBe("elevated output");
+    expect(execCommand).toHaveBeenCalledTimes(2);
+  });
+
+  it("throws SUDO_REQUIRED when elevation is needed but no password is set", async () => {
+    execCommand.mockResolvedValueOnce(result("", "Permission denied", 1));
+    await expect(
+      execElevated(fakeClient, "cat /etc/shadow", undefined),
+    ).rejects.toMatchObject({ code: "SUDO_REQUIRED" });
+  });
+});
+
+describe("execElevated (forced)", () => {
+  it("strips the success marker from stdout", async () => {
+    execCommand.mockResolvedValueOnce(
+      result("__TX_SUDO_OK__\nthe real output\n", "", 0),
+    );
+    const r = await execElevated(fakeClient, "id", "pw", { forceSudo: true });
+    expect(r.stdout).toBe("the real output\n");
+    expect(r.usedSudo).toBe(true);
+  });
+
+  it("does NOT throw when command output contains 'incorrect password' but sudo authenticated", async () => {
+    execCommand.mockResolvedValueOnce(
+      result(
+        "__TX_SUDO_OK__\nUser entered an incorrect password earlier",
+        "",
+        0,
+      ),
+    );
+    const r = await execElevated(fakeClient, "tail /var/log/auth.log", "pw", {
+      forceSudo: true,
+    });
+    expect(r.usedSudo).toBe(true);
+    expect(r.stdout).toContain("incorrect password");
+  });
+
+  it("throws SUDO_FAILED on a real wrong-password (no marker, sudo stderr)", async () => {
+    execCommand.mockResolvedValueOnce(
+      result("", "sudo: 1 incorrect password attempt", 1),
+    );
+    await expect(
+      execElevated(fakeClient, "id", "wrong", { forceSudo: true }),
+    ).rejects.toMatchObject({ code: "SUDO_FAILED" });
+  });
+
+  it("throws NOT_SUDOER when the user is not in sudoers", async () => {
+    execCommand.mockResolvedValueOnce(
+      result("", "deploy is not in the sudoers file.", 1),
+    );
+    await expect(
+      execElevated(fakeClient, "id", "pw", { forceSudo: true }),
+    ).rejects.toMatchObject({ code: "NOT_SUDOER" });
+  });
+
+  it("throws SUDO_REQUIRED when forced without a password", async () => {
+    await expect(
+      execElevated(fakeClient, "id", undefined, { forceSudo: true }),
+    ).rejects.toBeInstanceOf(ElevationError);
+    expect(execCommand).not.toHaveBeenCalled();
+  });
+});

src/backend/tests/hosts/metrics-shared/host-status.test.ts

@@ -0,0 +1,40 @@
+import { describe, expect, it } from "vitest";
+import {
+  isHostKeyVerificationError,
+  statusAfterAuthentication,
+  statusAfterReachabilityCheck,
+} from "../../../hosts/metrics-shared/host-status.js";
+
+describe("isHostKeyVerificationError", () => {
+  it.each([
+    "Host denied (verification failed)",
+    "Host key changed - please connect via Terminal to verify the new key",
+  ])("classifies %s", (message) => {
+    expect(isHostKeyVerificationError(new Error(message))).toBe(true);
+  });
+
+  it("does not classify ordinary authentication failures", () => {
+    expect(isHostKeyVerificationError(new Error("Permission denied"))).toBe(
+      false,
+    );
+  });
+});
+
+describe("host availability status", () => {
+  it("does not call a TCP-reachable host online before authentication", () => {
+    expect(statusAfterReachabilityCheck(true)).toBe("reachable");
+  });
+
+  it("keeps a verified host online across later reachability checks", () => {
+    expect(statusAfterReachabilityCheck(true, "online")).toBe("online");
+  });
+
+  it("marks successful authentication online", () => {
+    expect(statusAfterAuthentication(true, "reachable")).toBe("online");
+  });
+
+  it("downgrades failed authentication without hiding reachability", () => {
+    expect(statusAfterAuthentication(false, "online")).toBe("reachable");
+    expect(statusAfterAuthentication(false, "offline")).toBe("offline");
+  });
+});

src/backend/tests/hosts/metrics-shared/state.test.ts

@@ -0,0 +1,267 @@
+import { describe, expect, it, vi } from "vitest";
+import {
+  authFailureTracker,
+  canStartInitialMetrics,
+  ConcurrentLimiter,
+  HostPollCache,
+  metricsConcurrencyFor,
+} from "../../../hosts/metrics-shared/state.js";
+
+describe("auth failure tracking", () => {
+  it("permanently pauses polling after a host key mismatch", () => {
+    const hostId = 991266;
+    authFailureTracker.recordFailure(hostId, "HOST_KEY", true);
+
+    expect(authFailureTracker.shouldSkip(hostId)).toBe(true);
+    expect(authFailureTracker.getSkipReason(hostId)).toContain(
+      "host key changed",
+    );
+
+    authFailureTracker.reset(hostId);
+  });
+});
+
+describe("initial metrics admission", () => {
+  it("requires both an active viewer and a confirmed online status", () => {
+    expect(canStartInitialMetrics("online", true)).toBe(true);
+    expect(canStartInitialMetrics("reachable", true)).toBe(true);
+    expect(canStartInitialMetrics("offline", true)).toBe(false);
+    expect(canStartInitialMetrics(undefined, true)).toBe(false);
+    expect(canStartInitialMetrics("online", false)).toBe(false);
+    expect(canStartInitialMetrics(undefined, true, false)).toBe(true);
+  });
+});
+
+describe("ConcurrentLimiter", () => {
+  it("never exceeds max concurrent runners", async () => {
+    const limiter = new ConcurrentLimiter(2);
+    let peak = 0;
+    let current = 0;
+
+    const job = async () => {
+      current += 1;
+      peak = Math.max(peak, current);
+      await new Promise((r) => setTimeout(r, 30));
+      current -= 1;
+    };
+
+    await Promise.all([
+      limiter.run(job),
+      limiter.run(job),
+      limiter.run(job),
+      limiter.run(job),
+    ]);
+
+    expect(peak).toBeLessThanOrEqual(2);
+    expect(limiter.activeCount).toBe(0);
+    expect(limiter.pendingCount).toBe(0);
+  });
+
+  it("runs waiters in FIFO order after a slot frees", async () => {
+    const limiter = new ConcurrentLimiter(1);
+    const order: number[] = [];
+
+    const first = limiter.run(async () => {
+      order.push(1);
+      await new Promise((r) => setTimeout(r, 20));
+    });
+    const second = limiter.run(async () => {
+      order.push(2);
+    });
+    const third = limiter.run(async () => {
+      order.push(3);
+    });
+
+    await Promise.all([first, second, third]);
+    expect(order).toEqual([1, 2, 3]);
+  });
+
+  it("rejects invalid maxConcurrent", () => {
+    expect(() => new ConcurrentLimiter(0)).toThrow(/maxConcurrent/);
+  });
+
+  describe("setLimit", () => {
+    it("releases queued waiters as soon as the ceiling is raised", async () => {
+      const limiter = new ConcurrentLimiter(1);
+      let running = 0;
+      let peak = 0;
+      const release: Array<() => void> = [];
+
+      const job = () =>
+        limiter.run(async () => {
+          running += 1;
+          peak = Math.max(peak, running);
+          await new Promise<void>((r) => release.push(r));
+          running -= 1;
+        });
+
+      const jobs = [job(), job(), job(), job()];
+      await new Promise((r) => setTimeout(r, 10));
+      expect(peak).toBe(1);
+      expect(limiter.pendingCount).toBe(3);
+
+      // Widening must drain the backlog without waiting for the running job.
+      limiter.setLimit(4);
+      await new Promise((r) => setTimeout(r, 10));
+      expect(peak).toBe(4);
+      expect(limiter.pendingCount).toBe(0);
+
+      release.forEach((fn) => fn());
+      await Promise.all(jobs);
+      expect(limiter.activeCount).toBe(0);
+    });
+
+    it("does not over-release beyond the new ceiling", async () => {
+      const limiter = new ConcurrentLimiter(1);
+      let running = 0;
+      let peak = 0;
+      const release: Array<() => void> = [];
+      // Later waves of woken jobs enqueue their own resolvers, so draining has
+      // to keep going until nothing is left rather than flushing a snapshot.
+      const drain = async () => {
+        while (release.length > 0) {
+          release.splice(0).forEach((fn) => fn());
+          await new Promise((r) => setTimeout(r, 5));
+        }
+      };
+
+      const job = () =>
+        limiter.run(async () => {
+          running += 1;
+          peak = Math.max(peak, running);
+          await new Promise<void>((r) => release.push(r));
+          running -= 1;
+        });
+
+      const jobs = [job(), job(), job(), job(), job()];
+      await new Promise((r) => setTimeout(r, 10));
+
+      limiter.setLimit(3);
+      await new Promise((r) => setTimeout(r, 10));
+      expect(peak).toBe(3);
+      expect(limiter.pendingCount).toBe(2);
+
+      await drain();
+      await Promise.all(jobs);
+      expect(limiter.activeCount).toBe(0);
+      expect(limiter.pendingCount).toBe(0);
+    });
+
+    it("lets running work finish when the ceiling shrinks", async () => {
+      const limiter = new ConcurrentLimiter(4);
+      let running = 0;
+      let peak = 0;
+      const release: Array<() => void> = [];
+      const drain = async () => {
+        while (release.length > 0) {
+          release.splice(0).forEach((fn) => fn());
+          await new Promise((r) => setTimeout(r, 5));
+        }
+      };
+
+      const job = () =>
+        limiter.run(async () => {
+          running += 1;
+          peak = Math.max(peak, running);
+          await new Promise<void>((r) => release.push(r));
+          running -= 1;
+        });
+
+      const jobs = [job(), job(), job(), job(), job(), job()];
+      await new Promise((r) => setTimeout(r, 10));
+      expect(peak).toBe(4);
+
+      // Shrinking never kills in-flight work; it applies to later releases.
+      limiter.setLimit(2);
+      expect(limiter.activeCount).toBe(4);
+
+      await drain();
+      await Promise.all(jobs);
+      expect(limiter.activeCount).toBe(0);
+      expect(limiter.pendingCount).toBe(0);
+      // The two queued jobs ran only after the shrink, so they never pushed
+      // occupancy back up to the old width.
+      expect(peak).toBe(4);
+    });
+
+    it("rejects an invalid new limit", () => {
+      const limiter = new ConcurrentLimiter(2);
+      expect(() => limiter.setLimit(0)).toThrow(/maxConcurrent/);
+      expect(limiter.limit).toBe(2);
+    });
+  });
+});
+
+describe("metricsConcurrencyFor", () => {
+  it("keeps a floor for small installs", () => {
+    expect(metricsConcurrencyFor(0, {})).toBe(5);
+    expect(metricsConcurrencyFor(1, {})).toBe(5);
+    expect(metricsConcurrencyFor(60, {})).toBe(5);
+  });
+
+  it("scales up with the fleet", () => {
+    expect(metricsConcurrencyFor(200, {})).toBe(10);
+    expect(metricsConcurrencyFor(500, {})).toBe(25);
+  });
+
+  it("caps so a huge fleet cannot exhaust the host", () => {
+    expect(metricsConcurrencyFor(100000, {})).toBe(50);
+  });
+
+  it("lets an operator override the sizing", () => {
+    expect(metricsConcurrencyFor(1000, { METRICS_POLL_CONCURRENCY: "8" })).toBe(
+      8,
+    );
+  });
+
+  it("still caps an oversized override", () => {
+    expect(
+      metricsConcurrencyFor(10, { METRICS_POLL_CONCURRENCY: "9999" }),
+    ).toBe(50);
+  });
+
+  it("ignores a nonsense override", () => {
+    expect(
+      metricsConcurrencyFor(500, { METRICS_POLL_CONCURRENCY: "abc" }),
+    ).toBe(25);
+  });
+
+  it("sweeps 500 hosts inside a 30s interval, which the old fixed 5 could not", () => {
+    const POLL_MS = 400;
+    const hosts = 500;
+    const sweepAt = (c: number) => Math.ceil(hosts / c) * POLL_MS;
+
+    expect(sweepAt(5)).toBeGreaterThan(30_000);
+    expect(sweepAt(metricsConcurrencyFor(hosts, {}))).toBeLessThan(30_000);
+  });
+});
+
+describe("HostPollCache", () => {
+  it("returns cached host within TTL for the same user", () => {
+    const cache = new HostPollCache<{ id: number; name: string }>(60_000);
+    cache.set(1, "user-a", { id: 1, name: "alpha" });
+    expect(cache.get(1, "user-a")).toEqual({ id: 1, name: "alpha" });
+    expect(cache.get(1, "user-b")).toBeNull();
+  });
+
+  it("expires entries after TTL", () => {
+    vi.useFakeTimers();
+    const cache = new HostPollCache<{ id: number }>(1_000);
+    cache.set(7, "u", { id: 7 });
+    expect(cache.get(7, "u")).toEqual({ id: 7 });
+    vi.advanceTimersByTime(1_001);
+    expect(cache.get(7, "u")).toBeNull();
+    vi.useRealTimers();
+  });
+
+  it("invalidate drops a host or the whole cache", () => {
+    const cache = new HostPollCache<{ id: number }>(60_000);
+    cache.set(1, "u", { id: 1 });
+    cache.set(2, "u", { id: 2 });
+    cache.invalidate(1);
+    expect(cache.get(1, "u")).toBeNull();
+    expect(cache.get(2, "u")).toEqual({ id: 2 });
+    cache.invalidate();
+    expect(cache.get(2, "u")).toBeNull();
+  });
+});