From 79bd10a1b0aead742cafe837aac26eac308d78bc Mon Sep 17 00:00:00 2001 From: LukeGus Date: Sat, 3 Oct 2026 15:37:11 -0500 Subject: [PATCH] fix: high cpu from status probes and full database saves on every sample (#1300) --- plugins/alerts/src/backend/repository.ts | 4 +- plugins/automations/src/backend/repository.ts | 4 +- .../host-metrics/src/backend/repository.ts | 2 +- .../proxmox-node-history-repository.ts | 4 +- .../hosts/status/host-status-service.ts | 26 ++--- src/backend/hosts/status/jump-probe-pool.ts | 87 +++++++++++++++ src/backend/hosts/status/tcp-ping.ts | 3 +- .../hosts/status/jump-probe-pool.test.ts | 104 ++++++++++++++++++ .../tests/utils/database-save-trigger.test.ts | 11 ++ src/backend/utils/database-save-trigger.ts | 18 ++- 10 files changed, 241 insertions(+), 22 deletions(-) create mode 100644 src/backend/hosts/status/jump-probe-pool.ts create mode 100644 src/backend/tests/hosts/status/jump-probe-pool.test.ts diff --git a/plugins/alerts/src/backend/repository.ts b/plugins/alerts/src/backend/repository.ts index 76bc04671..41358df4f 100644 --- a/plugins/alerts/src/backend/repository.ts +++ b/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 { diff --git a/plugins/automations/src/backend/repository.ts b/plugins/automations/src/backend/repository.ts index 5592a8b94..d200e6dc5 100644 --- a/plugins/automations/src/backend/repository.ts +++ b/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; }, diff --git a/plugins/host-metrics/src/backend/repository.ts b/plugins/host-metrics/src/backend/repository.ts index abbbd975c..e835a6278 100644 --- a/plugins/host-metrics/src/backend/repository.ts +++ b/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( diff --git a/plugins/proxmox/src/backend/proxmox-node-history-repository.ts b/plugins/proxmox/src/backend/proxmox-node-history-repository.ts index 4272994ff..e5b66dab2 100644 --- a/plugins/proxmox/src/backend/proxmox-node-history-repository.ts +++ b/plugins/proxmox/src/backend/proxmox-node-history-repository.ts @@ -40,7 +40,7 @@ export function createProxmoxNodeHistoryRepository( async create(input: ProxmoxNodeHistoryCreateInput): Promise { 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 }); }, }; } diff --git a/src/backend/hosts/status/host-status-service.ts b/src/backend/hosts/status/host-status-service.ts index ddea258a2..c238dfd95 100644 --- a/src/backend/hosts/status/host-status-service.ts +++ b/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 diff --git a/src/backend/hosts/status/jump-probe-pool.ts b/src/backend/hosts/status/jump-probe-pool.ts new file mode 100644 index 000000000..a9e2459ef --- /dev/null +++ b/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; + +interface Entry { + client: Promise; + 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(); + + 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 { + 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 { + 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); + } +} diff --git a/src/backend/hosts/status/tcp-ping.ts b/src/backend/hosts/status/tcp-ping.ts index 807663606..69ea46144 100644 --- a/src/backend/hosts/status/tcp-ping.ts +++ b/src/backend/hosts/status/tcp-ping.ts @@ -70,6 +70,7 @@ export function tcpPingThroughJumpHost( host: string, port: number, timeoutMs = 5000, + keepOpen = false, ): Promise { 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); }; diff --git a/src/backend/tests/hosts/status/jump-probe-pool.test.ts b/src/backend/tests/hosts/status/jump-probe-pool.test.ts new file mode 100644 index 000000000..e788a12e1 --- /dev/null +++ b/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(); + }); +}); diff --git a/src/backend/tests/utils/database-save-trigger.test.ts b/src/backend/tests/utils/database-save-trigger.test.ts index e2d659b64..64791d5c2 100644 --- a/src/backend/tests/utils/database-save-trigger.test.ts +++ b/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((resolve) => { diff --git a/src/backend/utils/database-save-trigger.ts b/src/backend/utils/database-save-trigger.ts index c2d8d0ddf..1125e78dc 100644 --- a/src/backend/utils/database-save-trigger.ts +++ b/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) | null = null; private static isInitialized = false; private static pendingSave = false; private static activeSave: Promise | 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;