From 3f5cc9d3ba8a23f1bbb00598c234d87021306e9c Mon Sep 17 00:00:00 2001 From: Josh Wright Date: Mon, 5 Oct 2026 15:02:19 -0400 Subject: [PATCH] Limit status checks to two per host --- src/lib/server/pipeline/worker.ts | 33 +++++++++------- tests/unit/server-status.test.ts | 62 ++++++++++++++++++++++++++++++- 2 files changed, 79 insertions(+), 16 deletions(-) diff --git a/src/lib/server/pipeline/worker.ts b/src/lib/server/pipeline/worker.ts index c5cb477..cfe0411 100644 --- a/src/lib/server/pipeline/worker.ts +++ b/src/lib/server/pipeline/worker.ts @@ -20,9 +20,8 @@ import { trackActivity } from './activity.js'; import { withOperationSource } from '../hosts/operation-context.js'; const TICK_MS = 15_000; -// Cap concurrent UDP RCON calls so a 15-server fleet doesn't blast the -// host all at once. -const STATUS_CONCURRENCY = 4; +// Match each agent's two health lanes without making unrelated hosts wait. +const STATUS_CONCURRENCY_PER_HOST = 2; let started = false; let tickTimer: NodeJS.Timeout | null = null; @@ -76,8 +75,8 @@ function encodeError(reason: RconFailure, detail: string): string { // Deduplicate queued/running checks per server. A slow server must not prevent // completed servers from being polled again on the next 30-second tick. const pendingStatus = new Map>(); -const statusQueue: Array<() => Promise> = []; -let activeStatus = 0; +interface StatusQueue { active: number; queued: Array<() => Promise> } +const statusQueues = new Map(); const changedStatus = new Map(); function publishStatus(entry: ServerStatusEntry) { const scheduled = changedStatus.size > 0; @@ -88,21 +87,27 @@ function publishStatus(entry: ServerStatusEntry) { }); } -function drainStatusQueue(): void { - while (activeStatus < STATUS_CONCURRENCY && statusQueue.length) { - const next = statusQueue.shift()!; - activeStatus++; - void next().finally(() => { activeStatus--; drainStatusQueue(); }); +function drainStatusQueue(host: string, queue: StatusQueue): void { + while (queue.active < STATUS_CONCURRENCY_PER_HOST && queue.queued.length) { + const next = queue.queued.shift()!; + queue.active++; + void next().finally(() => { queue.active--; drainStatusQueue(host, queue); }); } + if (!queue.active && !queue.queued.length) statusQueues.delete(host); } -function queueStatus(server: string): Promise { +function queueStatus(srv: ServerRow): Promise { + const server = srv.name; const pending = pendingStatus.get(server); if (pending) return pending; + // Legacy servers without a recorded host share the configured host's pool. + const host = srv.host_id ? `native:${srv.host_id}` : `komodo:${srv.komodo_server || config.komodo.targetServer}`; + let queue = statusQueues.get(host); + if (!queue) { queue = { active: 0, queued: [] }; statusQueues.set(host, queue); } let finish!: (capacity: number | null) => void; const work = new Promise(resolve => { finish = resolve; }); pendingStatus.set(server, work); - statusQueue.push(async () => { + queue.queued.push(async () => { let capacity: number | null = null; try { capacity = await withOperationSource('Status monitor', () => refreshServerStatus(server)); @@ -112,14 +117,14 @@ function queueStatus(server: string): Promise { } catch (err) { console.error(`[worker] status error for ${server}:`, err); } finally { pendingStatus.delete(server); finish(capacity); } }); - drainStatusQueue(); + drainStatusQueue(host, queue); return work; } export async function fanoutStatus(): Promise { const servers = listServers(); if (!servers.length) return; - const results = await Promise.all(servers.map(server => queueStatus(server.name))); + const results = await Promise.all(servers.map(queueStatus)); const capacities = new Map(servers.map((server, i) => [server.name, results[i]])); try { // Metrics remain complete fleet collections, using freshness at capture diff --git a/tests/unit/server-status.test.ts b/tests/unit/server-status.test.ts index 832022e..1734d60 100644 --- a/tests/unit/server-status.test.ts +++ b/tests/unit/server-status.test.ts @@ -75,7 +75,7 @@ describe('database fleet and container readiness', () => { expect(getServer('managed')).toMatchObject({ last_error: 'timeout:manual failure', last_seen_at: null, last_checked_at: new Date(at + 10_000).toISOString() }); }); - it('refreshes a ten-server fleet while one check is slow, without duplicating or exceeding four checks', async () => { + it('refreshes a ten-server host while one check is slow, without duplicating or exceeding two checks', async () => { const { db, inspect, rcon, fanoutStatus, getServer } = await setup(); const { subscribe } = await import('../../src/lib/server/pipeline/bus.js'); vi.useFakeTimers({ toFake: ['Date'] }); @@ -109,7 +109,7 @@ describe('database fleet and container readiness', () => { await vi.waitFor(() => expect(Date.parse(getServer('fast-8')?.last_seen_at ?? '')).toBeGreaterThanOrEqual(now + 30_000)); expect(rcon).toHaveBeenCalledTimes(18); expect(inspect.mock.calls.filter(([name]) => name === 'managed')).toHaveLength(1); - expect(peak).toBe(4); + expect(peak).toBe(2); release({ ok: true, state: 'running' }); await Promise.all([first, second]); expect(rcon).toHaveBeenCalledTimes(19); @@ -120,6 +120,64 @@ describe('database fleet and container readiness', () => { } }); + it.each(['native', 'komodo'])('gives each %s host two slots through inspection and RCON, independently of a stalled host', async backend => { + const { db, inspect, rcon, fanoutStatus, getServer } = await setup(); + db.prepare("DELETE FROM servers WHERE name='managed'").run(); + const hosts = await import('../../src/lib/server/hosts/index.js'); + for (let host = 0; host < 3; host++) { + const registered = hosts.createHost({ name: `Host ${host}` }); + hosts.enrollHost(hosts.createEnrollment(registered.id).token, { protocolVersion: 1 }); + for (let instance = 0; instance < 4; instance++) { + const name = `host-${host}-${instance}`; + db.prepare('INSERT INTO servers (name, komodo_name, komodo_server) VALUES (?, ?, ?)') + .run(name, name, backend === 'komodo' ? registered.id : null); + if (backend === 'native') db.prepare(`INSERT INTO native_instances (id,server,host_id,library_id,image,port,created_at) + VALUES (?, ?, ?, 'library', 'image', ?, ?)`).run(name, name, registered.id, 28960 + instance, new Date().toISOString()); + } + } + let releaseSlow!: () => void; + let releaseFast!: () => void; + const slow = new Promise(resolve => { releaseSlow = resolve; }); + const fast = new Promise(resolve => { releaseFast = resolve; }); + const active = new Map(); + const peak = new Map(); + inspect.mockResolvedValue({ ok: true, state: 'running' }); + rcon.mockImplementation(async name => { + const host = name.slice(0, -2); + active.set(host, (active.get(host) ?? 0) + 1); + peak.set(host, Math.max(peak.get(host) ?? 0, active.get(host)!)); + try { + await (host === 'host-0' ? slow : fast); + if (name === 'host-0-0') throw new Error('simulated failed probe'); + return { ok: true, endpoint: { host: 'localhost', port: 28960 }, rawReply: 'status', + status: { players: 2, map: 'zombie_theater', gametype: 'zom', roster: [] } }; + } finally { active.set(host, active.get(host)! - 1); } + }); + const passes = [fanoutStatus()]; + try { + await vi.waitFor(() => expect(rcon).toHaveBeenCalledTimes(6)); + // A slot covers the whole check: waiting for RCON cannot admit a third inspection. + expect(inspect).toHaveBeenCalledTimes(6); + expect([...active.values()]).toEqual([2, 2, 2]); + passes.push(fanoutStatus()); + releaseFast(); + await vi.waitFor(() => { + expect(getServer('host-1-3')?.last_players).toBe(2); + expect(getServer('host-2-3')?.last_players).toBe(2); + }); + expect(getServer('host-0-2')?.container_checked_at).toBeNull(); + expect(rcon).toHaveBeenCalledTimes(10); + } finally { + releaseSlow(); releaseFast(); + await Promise.all(passes); + } + expect([...peak.values()]).toEqual([2, 2, 2]); + expect(inspect).toHaveBeenCalledTimes(12); + expect(rcon).toHaveBeenCalledTimes(12); + expect(getServer('host-0-0')?.last_error).toBe('send-error:simulated failed probe'); + expect(getServer('host-0-3')?.last_players).toBe(2); + }); + it('expires early results before recording a fleet collection delayed by a slow check', async () => { const { db, inspect, rcon, fanoutStatus } = await setup(); vi.useFakeTimers({ toFake: ['Date'] });