Limit status checks to two per host

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
+// Match each agent's two health lanes without making unrelated hosts wait.
-// host all at once.
+const STATUS_CONCURRENCY_PER_HOST = 2;
-const STATUS_CONCURRENCY = 4;
 
 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<string, Promise<number | null>>();
-const statusQueue: Array<() => Promise<void>> = [];
+interface StatusQueue { active: number; queued: Array<() => Promise<void>> }
-let activeStatus = 0;
+const statusQueues = new Map<string, StatusQueue>();
 const changedStatus = new Map<string, ServerStatusEntry>();
 function publishStatus(entry: ServerStatusEntry) {
     const scheduled = changedStatus.size > 0;
@@ -88,21 +87,27 @@ function publishStatus(entry: ServerStatusEntry) {
     });
 }
 
-function drainStatusQueue(): void {
+function drainStatusQueue(host: string, queue: StatusQueue): void {
-    while (activeStatus < STATUS_CONCURRENCY && statusQueue.length) {
+    while (queue.active < STATUS_CONCURRENCY_PER_HOST && queue.queued.length) {
-        const next = statusQueue.shift()!;
+        const next = queue.queued.shift()!;
-        activeStatus++;
+        queue.active++;
-        void next().finally(() => { activeStatus--; drainStatusQueue(); });
+        void next().finally(() => { queue.active--; drainStatusQueue(host, queue); });
     }
+    if (!queue.active && !queue.queued.length) statusQueues.delete(host);
 }
 
-function queueStatus(server: string): Promise<number | null> {
+function queueStatus(srv: ServerRow): Promise<number | null> {
+    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<number | null>(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<number | null> {
         } 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<void> {
     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

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<void>(resolve => { releaseSlow = resolve; });
+        const fast = new Promise<void>(resolve => { releaseFast = resolve; });
+        const active = new Map<string, number>();
+        const peak = new Map<string, number>();
+        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'] });