fix: high cpu from status probes and full database saves on every sample (#1300)

plugins/alerts/src/backend/repository.ts

@@ -304,7 +304,7 @@ export async function createAlertsRepository(
         dedupeKey: item.dedupeKey ?? null,
         createdAt,
       });
-      await db.persist();
+      await db.persist({ lazy: true });
       return toItem({
         id,
         ...item,
@@ -322,7 +322,7 @@ export async function createAlertsRepository(
         .update(items)
         .set({ deliveries: JSON.stringify(results) })
         .where(eq(items.id, id));
-      await db.persist();
+      await db.persist({ lazy: true });
     },
 
     async hasUnreadWithKey(userId: string, key: string): Promise<boolean> {

plugins/automations/src/backend/repository.ts

@@ -528,7 +528,7 @@ export async function createAutomationRepository(
           .set({ lastRunAt: run[0].startedAt, lastRunStatus: input.status })
           .where(eq(automations.id, run[0].automationId));
       }
-      await db.persist();
+      await db.persist({ lazy: true });
     },
 
     async listRuns(
@@ -630,7 +630,7 @@ export async function createAutomationRepository(
         .where(lt(runs.startedAt, cutoff));
       if (old.length === 0) return 0;
       await drizzle.delete(runs).where(lt(runs.startedAt, cutoff));
-      await db.persist();
+      await db.persist({ lazy: true });
       return old.length;
     },
 

plugins/host-metrics/src/backend/repository.ts

@@ -159,7 +159,7 @@ export async function createHostMetricsRepository(db: PluginDatabase) {
             )
           : scope,
       );
-      await db.persist();
+      await db.persist({ lazy: true });
     },
 
     async listHealth(

plugins/proxmox/src/backend/proxmox-node-history-repository.ts

@@ -40,7 +40,7 @@ export function createProxmoxNodeHistoryRepository(
     async create(input: ProxmoxNodeHistoryCreateInput): Promise<void> {
       const drizzle = await client();
       await drizzle.insert(table).values(input);
-      await db.persist();
+      await db.persist({ lazy: true });
     },
 
     async listRange(
@@ -73,7 +73,7 @@ export function createProxmoxNodeHistoryRepository(
       await drizzle
         .delete(table)
         .where(and(eq(table.hostId, hostId), lt(table.ts, cutoff)));
-      await db.persist();
+      await db.persist({ lazy: true });
     },
   };
 }

src/backend/hosts/status/host-status-service.ts

@@ -7,8 +7,10 @@
  * failed (a session, a ctx.ssh connection or a plugin's
  * ctx.hosts.status.reportLogin said so). It clears on the next working login,
  * when the host is edited, or when a changed host key is accepted. Probing
- * never logs in: on RADIUS or Duo backed devices a login fires a real 2FA
- * push, every interval.
+ * never logs in to the host itself: on RADIUS or Duo backed devices a login
+ * fires a real 2FA push, every interval. A host behind jump hosts is probed
+ * through a chain, which does log in to the hops, so hosts that share a hop
+ * list share one chain (jump-probe-pool.ts).
  *
  * Polling is demand driven. A user asking for statuses starts polling their
  * own hosts; nothing is probed for a user who never opened the app.
@@ -28,7 +30,8 @@ import {
   type HostStatus,
 } from "./host-status.js";
 import { ConcurrentLimiter } from "./limiter.js";
-import { tcpPing, tcpPingThroughJumpHost } from "./tcp-ping.js";
+import { tcpPing } from "./tcp-ping.js";
+import { JumpProbePool } from "./jump-probe-pool.js";
 
 export const GLOBAL_STATUS_INTERVAL_KEY = "global_status_check_interval";
 export const DEFAULT_STATUS_INTERVAL = 60;
@@ -111,6 +114,11 @@ export function toStatusTarget(row: HostStatusTargetRow): StatusTarget {
   };
 }
 
+const jumpProbes = new JumpProbePool(async (jumpHosts, userId) => {
+  const { createJumpHostChain } = await import("../jump-host-chain.js");
+  return createJumpHostChain(jumpHosts, userId);
+});
+
 const defaultDeps: HostStatusDeps = {
   loadTargets: async (filter) =>
     (
@@ -129,16 +137,8 @@ const defaultDeps: HostStatusDeps = {
   },
   hasActiveSession: (hostId) => hostSessionStatus.hasActiveSession(hostId),
   ping: (host, port) => tcpPing(host, port, 5000),
-  pingThroughJumpHosts: async (target, port) => {
-    const { createJumpHostChain } = await import("../jump-host-chain.js");
-    const client = await createJumpHostChain(
-      target.jumpHosts,
-      target.userId,
-    ).catch(() => null);
-    return client
-      ? tcpPingThroughJumpHost(client, target.ip, port, 5000)
-      : false;
-  },
+  pingThroughJumpHosts: (target, port) =>
+    jumpProbes.ping(target.jumpHosts, target.userId, target.ip, port, 5000),
   globalInterval: () => {
     const value = Number(getCurrentSettingValue(GLOBAL_STATUS_INTERVAL_KEY));
     return Number.isInteger(value) && value >= 5

src/backend/hosts/status/jump-probe-pool.ts

@@ -0,0 +1,87 @@
+import type { Client } from "ssh2";
+import { tcpPingThroughJumpHost } from "./tcp-ping.js";
+
+type OpenChain = (
+  jumpHosts: Array<{ hostId: number }>,
+  userId: string,
+) => Promise<Client | null>;
+
+interface Entry {
+  client: Promise<Client | null>;
+  idle?: NodeJS.Timeout;
+}
+
+/**
+ * Status probes for hosts behind jump hosts share one chain per hop list.
+ * Opening a chain logs in to every hop, so doing it for every host on every
+ * interval made an import of jump-hosted hosts open dozens of SSH sessions
+ * each minute. A chain closes after it sits idle.
+ */
+export class JumpProbePool {
+  private readonly entries = new Map<string, Entry>();
+
+  constructor(
+    private readonly open: OpenChain,
+    private readonly idleMs = 120_000,
+  ) {}
+
+  async ping(
+    jumpHosts: Array<{ hostId: number }>,
+    userId: string,
+    host: string,
+    port: number,
+    timeoutMs = 5000,
+  ): Promise<boolean> {
+    const key = `${userId}:${jumpHosts.map((hop) => hop.hostId).join(">")}`;
+    let entry = this.entries.get(key);
+    if (!entry) {
+      entry = { client: this.connect(key, jumpHosts, userId) };
+      this.entries.set(key, entry);
+    }
+    const client = await entry.client;
+    if (!client) {
+      this.drop(key, entry);
+      return false;
+    }
+    this.touch(key, entry, client);
+    return tcpPingThroughJumpHost(client, host, port, timeoutMs, true);
+  }
+
+  closeAll(): void {
+    for (const [key, entry] of [...this.entries]) {
+      this.drop(key, entry);
+      void entry.client.then((client) => client?.end()).catch(() => {});
+    }
+  }
+
+  private async connect(
+    key: string,
+    jumpHosts: Array<{ hostId: number }>,
+    userId: string,
+  ): Promise<Client | null> {
+    const client = await this.open(jumpHosts, userId).catch(() => null);
+    if (client) {
+      const forget = () => {
+        const current = this.entries.get(key);
+        if (current) this.drop(key, current);
+      };
+      client.once("close", forget);
+      client.once("error", forget);
+    }
+    return client;
+  }
+
+  private touch(key: string, entry: Entry, client: Client): void {
+    if (entry.idle) clearTimeout(entry.idle);
+    entry.idle = setTimeout(() => {
+      this.drop(key, entry);
+      client.end();
+    }, this.idleMs);
+    entry.idle.unref?.();
+  }
+
+  private drop(key: string, entry: Entry): void {
+    if (entry.idle) clearTimeout(entry.idle);
+    if (this.entries.get(key) === entry) this.entries.delete(key);
+  }
+}

src/backend/hosts/status/tcp-ping.ts

@@ -70,6 +70,7 @@ export function tcpPingThroughJumpHost(
   host: string,
   port: number,
   timeoutMs = 5000,
+  keepOpen = false,
 ): Promise<boolean> {
   return new Promise((resolve) => {
     let settled = false;
@@ -78,7 +79,7 @@ export function tcpPingThroughJumpHost(
       if (settled) return;
       settled = true;
       clearTimeout(timeout);
-      jumpClient.end();
+      if (!keepOpen) jumpClient.end();
       resolve(result);
     };
 

src/backend/tests/hosts/status/jump-probe-pool.test.ts

@@ -0,0 +1,104 @@
+import { EventEmitter } from "node:events";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { JumpProbePool } from "../../../hosts/status/jump-probe-pool.js";
+
+function fakeClient(reachable = true) {
+  const client = Object.assign(new EventEmitter(), {
+    end: vi.fn(),
+    forwardOut: vi.fn(
+      (
+        _srcIp: string,
+        _srcPort: number,
+        _host: string,
+        _port: number,
+        callback: (error: Error | null, stream?: { destroy(): void }) => void,
+      ) => {
+        callback(
+          reachable ? null : new Error("refused"),
+          reachable ? { destroy: () => {} } : undefined,
+        );
+      },
+    ),
+  });
+  return client;
+}
+
+afterEach(() => {
+  vi.useRealTimers();
+});
+
+describe("JumpProbePool", () => {
+  it("shares one chain across hosts behind the same hops", async () => {
+    const client = fakeClient();
+    const open = vi.fn(async () => client as never);
+    const pool = new JumpProbePool(open);
+    const hops = [{ hostId: 1 }, { hostId: 2 }];
+
+    const results = await Promise.all([
+      pool.ping(hops, "u1", "10.0.0.3", 22),
+      pool.ping(hops, "u1", "10.0.0.4", 22),
+      pool.ping(hops, "u1", "10.0.0.5", 22),
+    ]);
+
+    expect(results).toEqual([true, true, true]);
+    expect(open).toHaveBeenCalledTimes(1);
+    expect(client.forwardOut).toHaveBeenCalledTimes(3);
+    expect(client.end).not.toHaveBeenCalled();
+    pool.closeAll();
+  });
+
+  it("keeps chains apart per user and hop list", async () => {
+    const open = vi.fn(async () => fakeClient() as never);
+    const pool = new JumpProbePool(open);
+    await pool.ping([{ hostId: 1 }], "u1", "a", 22);
+    await pool.ping([{ hostId: 1 }], "u2", "a", 22);
+    await pool.ping([{ hostId: 2 }], "u1", "a", 22);
+    expect(open).toHaveBeenCalledTimes(3);
+    pool.closeAll();
+  });
+
+  it("reports an unreachable target without dropping the chain", async () => {
+    const client = fakeClient(false);
+    const open = vi.fn(async () => client as never);
+    const pool = new JumpProbePool(open);
+    expect(await pool.ping([{ hostId: 1 }], "u1", "a", 22)).toBe(false);
+    expect(await pool.ping([{ hostId: 1 }], "u1", "b", 22)).toBe(false);
+    expect(open).toHaveBeenCalledTimes(1);
+    pool.closeAll();
+  });
+
+  it("retries a chain that failed to open", async () => {
+    const open = vi
+      .fn()
+      .mockResolvedValueOnce(null)
+      .mockResolvedValueOnce(fakeClient());
+    const pool = new JumpProbePool(open);
+    expect(await pool.ping([{ hostId: 1 }], "u1", "a", 22)).toBe(false);
+    expect(await pool.ping([{ hostId: 1 }], "u1", "a", 22)).toBe(true);
+    pool.closeAll();
+  });
+
+  it("opens a new chain after the old one closed or sat idle", async () => {
+    vi.useFakeTimers();
+    const first = fakeClient();
+    const second = fakeClient();
+    const third = fakeClient();
+    const open = vi
+      .fn()
+      .mockResolvedValueOnce(first)
+      .mockResolvedValueOnce(second)
+      .mockResolvedValueOnce(third);
+    const pool = new JumpProbePool(open, 1000);
+
+    await pool.ping([{ hostId: 1 }], "u1", "a", 22);
+    first.emit("close");
+    await pool.ping([{ hostId: 1 }], "u1", "a", 22);
+    expect(open).toHaveBeenCalledTimes(2);
+
+    await vi.advanceTimersByTimeAsync(1500);
+    expect(second.end).toHaveBeenCalled();
+    await pool.ping([{ hostId: 1 }], "u1", "a", 22);
+    expect(open).toHaveBeenCalledTimes(3);
+    pool.closeAll();
+  });
+});

src/backend/tests/utils/database-save-trigger.test.ts

@@ -71,6 +71,17 @@ describe("DatabaseSaveTrigger", () => {
     expect(DatabaseSaveTrigger.getStatus().pendingSave).toBe(false);
   });
 
+  it("saves within the max wait even while writes keep coming", async () => {
+    vi.useFakeTimers();
+    const save = vi.fn().mockResolvedValue(undefined);
+    DatabaseSaveTrigger.initialize(save);
+    for (let second = 0; second < 40; second++) {
+      void DatabaseSaveTrigger.triggerSave("sample");
+      await vi.advanceTimersByTimeAsync(1000);
+    }
+    expect(save).toHaveBeenCalledTimes(1);
+  });
+
   it("queues a force save behind an in-flight save", async () => {
     let finishFirstSave: (() => void) | undefined;
     const firstSave = new Promise<void>((resolve) => {

src/backend/utils/database-save-trigger.ts

@@ -2,12 +2,16 @@ import { AsyncLocalStorage } from "node:async_hooks";
 import { getErrorMessage } from "./error-message.js";
 import { databaseLogger } from "./logger.js";
 
+const DEBOUNCE_MS = 2000;
+const MAX_DEBOUNCE_WAIT_MS = 30_000;
+
 export class DatabaseSaveTrigger {
   private static saveFunction: (() => Promise<void>) | null = null;
   private static isInitialized = false;
   private static pendingSave = false;
   private static activeSave: Promise<void> | null = null;
   private static saveTimeout: NodeJS.Timeout | null = null;
+  private static firstTriggerAt: number | null = null;
   private static _dirty = false;
   private static batch = new AsyncLocalStorage<{
     open: boolean;
@@ -49,8 +53,18 @@ export class DatabaseSaveTrigger {
       clearTimeout(this.saveTimeout);
     }
 
+    // Each call pushes the save back, so steady writes (samples every few
+    // seconds) would keep it from ever running. Cap how long it can wait.
+    const now = Date.now();
+    this.firstTriggerAt ??= now;
+    const delay = Math.max(
+      0,
+      Math.min(DEBOUNCE_MS, this.firstTriggerAt + MAX_DEBOUNCE_WAIT_MS - now),
+    );
+
     this.saveTimeout = setTimeout(async () => {
       this.saveTimeout = null;
+      this.firstTriggerAt = null;
 
       try {
         await this.runSave();
@@ -62,7 +76,7 @@ export class DatabaseSaveTrigger {
           error: getErrorMessage(error),
         });
       }
-    }, 2000);
+    }, delay);
   }
 
   /**
@@ -109,6 +123,7 @@ export class DatabaseSaveTrigger {
       clearTimeout(this.saveTimeout);
       this.saveTimeout = null;
     }
+    this.firstTriggerAt = null;
 
     try {
       await this.runSave();
@@ -169,6 +184,7 @@ export class DatabaseSaveTrigger {
       clearTimeout(this.saveTimeout);
       this.saveTimeout = null;
     }
+    this.firstTriggerAt = null;
 
     this.pendingSave = false;
     this.activeSave = null;