Termix/plugins/host-metrics/tests/backend/state.test.ts

import { describe, expect, it, vi } from "vitest";
import {
  AuthFailureTracker,
  canStartInitialMetrics,
  ConcurrentLimiter,
  createMetricsState,
  HostPollCache,
  metricsConcurrencyFor,
} from "../../src/backend/state.js";

const authFailureTracker = new AuthFailureTracker();

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("needs a viewer and a host core does not see as offline", () => {
    expect(canStartInitialMetrics("online", true)).toBe(true);
    expect(canStartInitialMetrics("offline", true)).toBe(false);
    expect(canStartInitialMetrics("online", false)).toBe(false);
    // Status checks off: core has no status, so the sample is tried.
    expect(canStartInitialMetrics(null, true)).toBe(true);
  });
});

describe("createMetricsState", () => {
  it("gives every activation its own trackers", () => {
    const first = createMetricsState();
    const second = createMetricsState();
    first.authFailures.recordFailure(1, "AUTH", true);
    expect(first.authFailures.shouldSkip(1)).toBe(true);
    expect(second.authFailures.shouldSkip(1)).toBe(false);
  });
});

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();
  });
});