Preserve server status when batch probes miss the deadline

agent/ec_agent.py

@@ -32,7 +32,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.4.12"
+VERSION = "1.4.13"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -1368,7 +1368,7 @@ class Agent:
         def probe(identity):
             instance_id, generation = identity
             target = {"instanceId": instance_id, "generation": generation}
-            reply = {"ok": False, "reason": "timeout", "text": ""}
+            reply = {"ok": False, "reason": "deadline", "text": ""}
             # A batch does not hold every instance lock while one server waits.
             # Bound concurrency across overlapping batches as well as within one.
             lock = self.instance_lock(target)
@@ -1382,7 +1382,7 @@ class Agent:
                             if current.get("generation", generation) != generation or current.get("removed"):
                                 reply = {"ok": False, "reason": "superseded", "text": ""}
                             elif time.monotonic() < deadline:
-                                reply = self.rcon(instance_id, generation, "status", deadline=deadline)
+                                reply = self.rcon(instance_id, generation, "status", deadline=deadline, status_attempts=1)
                         except (AgentError, OSError, ValueError, KeyError):
                             reply = {"ok": False, "reason": "unreachable", "text": ""}
                         finally:
@@ -1395,7 +1395,7 @@ class Agent:
             results = list(executor.map(probe, identities))
         return {"ok": all(row["reply"].get("ok") for row in results), "results": results}
 
-    def rcon(self, instance_id, generation, command, deadline=None):
+    def rcon(self, instance_id, generation, command, deadline=None, status_attempts=3):
         if not isinstance(command, str) or not command or len(command) > 2048 or any(ch in command for ch in ("\x00", "\r", "\n")):
             raise AgentError("Invalid RCON command")
         saved = self.spec(instance_id, generation)
@@ -1407,11 +1407,12 @@ class Agent:
         family = socket.AF_INET6 if ":" in host else socket.AF_INET
         # Retry only the read-only occupancy query. A missing UDP reply must
         # never cause commands with effects to be replayed.
-        for attempt in range(3 if command == "status" else 1):
+        attempts = status_attempts if command == "status" else 1
+        for attempt in range(attempts):
             if deadline is not None and time.monotonic() >= deadline:
-                return {"ok": False, "reason": "timeout", "players": None, "text": ""}
+                return {"ok": False, "reason": "deadline" if attempt == 0 else "timeout", "players": None, "text": ""}
             result = self.rcon_attempt(family, host, int(saved["port"]), password, command, **({"deadline": deadline} if deadline is not None else {}))
-            if result["ok"] or result.get("reason") not in ("timeout", "unreachable") or attempt == 2 or command != "status":
+            if result["ok"] or result.get("reason") not in ("timeout", "unreachable") or attempt == attempts - 1:
                 return result
             time.sleep(min(0.25 * 2 ** attempt, max(0, deadline - time.monotonic())) if deadline is not None else 0.25 * 2 ** attempt)
 

agent/tests/test_agent.py

@@ -613,7 +613,7 @@ class AgentTests(unittest.TestCase):
         lock = threading.Lock()
         active, peak = 0, 0
 
-        def rcon(instance_id, generation, command, deadline):
+        def rcon(instance_id, generation, command, deadline, status_attempts):
             nonlocal active, peak
             with lock:
                 active += 1
@@ -622,6 +622,7 @@ class AgentTests(unittest.TestCase):
                     ready.set()
             try:
                 self.assertEqual(command, "status")
+                self.assertEqual(status_attempts, 1)
                 self.assertTrue(release.wait(2))
                 if instance_id == "00000000-0000-4000-8000-000000000002":
                     raise AgentError("local failure")
@@ -647,6 +648,43 @@ class AgentTests(unittest.TestCase):
         self.assertEqual(results[0]["results"][2]["reply"]["reason"], "unreachable")
         self.assertTrue(results[0]["results"][4]["reply"]["ok"])
 
+    def test_batch_uses_one_attempt_but_individual_status_still_retries(self):
+        saved = {"rconPassword": "secret", "port": 28960}
+        timeout = {"ok": False, "reason": "timeout", "text": ""}
+        operation = {"kind": "rcon.status-batch", "payload": {"instances": [{"instanceId": INSTANCE, "generation": 1}]}}
+        with patch.object(self.agent, "spec", return_value=saved), patch.object(self.agent, "rcon_attempt", return_value=timeout) as attempt, patch.object(module.time, "sleep") as sleep:
+            self.assertEqual(self.agent.dispatch(operation)["results"][0]["reply"], timeout)
+            attempt.assert_called_once()
+            sleep.assert_not_called()
+            attempt.reset_mock()
+            self.assertEqual(self.agent.rcon(INSTANCE, 1, "status"), timeout)
+            self.assertEqual(attempt.call_count, 3)
+            self.assertEqual(sleep.call_count, 2)
+            attempt.reset_mock()
+            self.assertEqual(self.agent.rcon(INSTANCE, 1, "status", deadline=0, status_attempts=1)["reason"], "deadline")
+            attempt.assert_not_called()
+
+    def test_batch_reaches_healthy_servers_after_four_unresponsive_servers(self):
+        clock = [0]
+        checked = []
+        targets = [{"instanceId": f"00000000-0000-4000-8000-{i:012d}", "generation": 1} for i in range(16)]
+
+        def attempt(family, host, port, password, command, deadline):
+            checked.append(port)
+            if port < 28964:
+                clock[0] += 3
+                return {"ok": False, "reason": "timeout", "text": ""}
+            return {"ok": True, "text": "status"}
+
+        # Serial execution is a conservative budget test; concurrency is tested separately.
+        with patch.object(module, "ThreadPoolExecutor") as pool, patch.object(module.time, "monotonic", side_effect=lambda: clock[0]), patch.object(self.agent, "spec", side_effect=lambda instance, generation: {"port": 28960 + int(instance[-12:]), "rconPassword": "secret"}), patch.object(self.agent, "rcon_attempt", side_effect=attempt), patch.object(module.time, "sleep") as sleep:
+            pool.return_value.__enter__.return_value.map.side_effect = map
+            result = self.agent.rcon_status_batch({"payload": {"instances": targets}})
+        self.assertEqual(len(checked), 16)
+        self.assertEqual(sum(row["reply"]["ok"] for row in result["results"]), 12)
+        self.assertEqual(clock[0], 12)
+        sleep.assert_not_called()
+
     def test_status_batch_rejects_old_generations_and_bounds_expired_work(self):
         marker = self.agent.instance_dir(INSTANCE, 1).parent / "current.json"
         module.atomic_json(marker, {"generation": 2})
@@ -658,7 +696,7 @@ class AgentTests(unittest.TestCase):
             marker.unlink()
             with patch.object(module.time, "monotonic", side_effect=[0] + [21] * 20):
                 result = self.agent.dispatch(operation)
-            self.assertEqual(result["results"][0]["reply"]["reason"], "timeout")
+            self.assertEqual(result["results"][0]["reply"]["reason"], "deadline")
             probe.assert_not_called()
         for targets in ([], [None], [{"instanceId": INSTANCE, "generation": 1}] * 2):
             with self.assertRaises(AgentError):

src/lib/server/fleet/health-observations.ts

@@ -23,6 +23,7 @@ interface Entry<T> {
 }
 
 export class MonitoringChangedError extends Error {}
+export class MonitoringSkippedError extends Error {}
 
 function identity(server: string, probe: Probe): Identity {
     const row = getServer(server);
@@ -199,6 +200,7 @@ export function prepareMonitoringRcon(servers: string[]): Map<string, () => Prom
                     const operation = withOperationSource('Status monitor', () => queueHostOperation(instance.host_id, 'rcon.status-batch', {
                         instances: instances.map(value => ({ instanceId: value.id, generation: value.generation }))
                     }));
+                    // Allow delivery and receipt overhead beyond the agent's 20-second budget.
                     const receipt = await waitHostOperationStatus(operation.id, 25_000);
                     if (receipt.state !== 'completed') throw new Error(receipt.error ?? 'Host status batch did not complete.');
                     return receipt.result as { results: Array<{ instanceId: string; generation: number; reply: AgentRconReply }> };
@@ -211,8 +213,12 @@ export function prepareMonitoringRcon(servers: string[]): Map<string, () => Prom
                 const receipt = await batch.result;
                 const replies = receipt.results.filter(value => value.instanceId === instance.id && value.generation === instance.generation);
                 if (replies.length !== 1 || typeof replies[0].reply?.text !== 'string') throw new Error('Host status batch omitted a valid instance result.');
+                if (replies[0].reply.reason === 'deadline') throw new MonitoringSkippedError('Host status batch budget expired before this server was checked.');
                 return agentRconStatusResult(replies[0].reply, endpoint);
-            } catch (error) { return { ok: false as const, endpoint, reason: 'timeout' as const, detail: (error as Error).message }; }
+            } catch (error) {
+                if (error instanceof MonitoringSkippedError) throw error;
+                return { ok: false as const, endpoint, reason: 'timeout' as const, detail: (error as Error).message };
+            }
         };
         // Observe before dispatch, including servers still waiting for their
         // queue slot. Consume failures too, so a failed batch is never retried

src/lib/server/pipeline/worker.ts

@@ -6,7 +6,7 @@ import { emit } from './bus.js';
 import type { ServerStatusEntry } from './bus.js';
 import type { RconFailure } from '../integrations/rcon.js';
 import { getServer, listServers, setServerFields, type ServerRow } from '../fleet/servers.js';
-import { inspectForMonitoring, rconForMonitoring, prepareMonitoringRcon, monitoringStatusWriter, MonitoringChangedError } from '../fleet/health-observations.js';
+import { inspectForMonitoring, rconForMonitoring, prepareMonitoringRcon, monitoringStatus, monitoringStatusWriter, MonitoringChangedError, MonitoringSkippedError } from '../fleet/health-observations.js';
 import { restartSettling } from '../provisioning/index.js';
 import { serverObservation, observationReader } from '../fleet/status.js';
 import { telemetryEvidence, latestGameSubmission } from '../fleet/status-evidence.js';
@@ -149,6 +149,7 @@ async function refreshServerStatus(server: string, probeStatus?: () => ReturnTyp
     let updates: Partial<Record<keyof ServerRow, string | number | null>> = {};
     const flush = () => { setServerFields(srv.name, updates); updates = {}; };
     if (!isProvisioning(srv.name)) {
+        const previousGame = monitoringStatus(srv.name)?.game;
         const publishObservation = monitoringStatusWriter(srv.name);
         let gameObservation: Awaited<ReturnType<typeof rconForMonitoring>> | undefined;
         const settling = restartSettling(srv.name);
@@ -180,8 +181,15 @@ async function refreshServerStatus(server: string, probeStatus?: () => ReturnTyp
             } else if (unavailable || !telemetryEvidence(srv.name)) {
                 flush();
                 const probeStartedAt = new Date().toISOString();
-                const probe = await (probeStatus ? probeStatus() : rconForMonitoring(srv.name)).catch(err => err instanceof MonitoringChangedError ? null
+                const probe = await (probeStatus ? probeStatus() : rconForMonitoring(srv.name)).catch(err => {
-                    : { result: { ok: false as const, endpoint: null, reason: 'send-error' as RconFailure, detail: (err as Error).message, rawReply: undefined }, observedAt: probeStartedAt });
+                    if (err instanceof MonitoringSkippedError) {
+                        // The container was checked, but game evidence retains its original age.
+                        if (unchanged()) publishObservation(inspected, previousGame);
+                        return null;
+                    }
+                    if (err instanceof MonitoringChangedError) return null;
+                    return { result: { ok: false as const, endpoint: null, reason: 'send-error' as RconFailure, detail: (err as Error).message, rawReply: undefined }, observedAt: probeStartedAt };
+                });
                 if (!probe) return null;
                 const result = probe.result;
                 if (unchanged() && (getServer(srv.name)?.last_checked_at ?? '') <= probe.observedAt) {

static/agent/ec-agent.py

@@ -32,7 +32,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.4.12"
+VERSION = "1.4.13"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -1368,7 +1368,7 @@ class Agent:
         def probe(identity):
             instance_id, generation = identity
             target = {"instanceId": instance_id, "generation": generation}
-            reply = {"ok": False, "reason": "timeout", "text": ""}
+            reply = {"ok": False, "reason": "deadline", "text": ""}
             # A batch does not hold every instance lock while one server waits.
             # Bound concurrency across overlapping batches as well as within one.
             lock = self.instance_lock(target)
@@ -1382,7 +1382,7 @@ class Agent:
                             if current.get("generation", generation) != generation or current.get("removed"):
                                 reply = {"ok": False, "reason": "superseded", "text": ""}
                             elif time.monotonic() < deadline:
-                                reply = self.rcon(instance_id, generation, "status", deadline=deadline)
+                                reply = self.rcon(instance_id, generation, "status", deadline=deadline, status_attempts=1)
                         except (AgentError, OSError, ValueError, KeyError):
                             reply = {"ok": False, "reason": "unreachable", "text": ""}
                         finally:
@@ -1395,7 +1395,7 @@ class Agent:
             results = list(executor.map(probe, identities))
         return {"ok": all(row["reply"].get("ok") for row in results), "results": results}
 
-    def rcon(self, instance_id, generation, command, deadline=None):
+    def rcon(self, instance_id, generation, command, deadline=None, status_attempts=3):
         if not isinstance(command, str) or not command or len(command) > 2048 or any(ch in command for ch in ("\x00", "\r", "\n")):
             raise AgentError("Invalid RCON command")
         saved = self.spec(instance_id, generation)
@@ -1407,11 +1407,12 @@ class Agent:
         family = socket.AF_INET6 if ":" in host else socket.AF_INET
         # Retry only the read-only occupancy query. A missing UDP reply must
         # never cause commands with effects to be replayed.
-        for attempt in range(3 if command == "status" else 1):
+        attempts = status_attempts if command == "status" else 1
+        for attempt in range(attempts):
             if deadline is not None and time.monotonic() >= deadline:
-                return {"ok": False, "reason": "timeout", "players": None, "text": ""}
+                return {"ok": False, "reason": "deadline" if attempt == 0 else "timeout", "players": None, "text": ""}
             result = self.rcon_attempt(family, host, int(saved["port"]), password, command, **({"deadline": deadline} if deadline is not None else {}))
-            if result["ok"] or result.get("reason") not in ("timeout", "unreachable") or attempt == 2 or command != "status":
+            if result["ok"] or result.get("reason") not in ("timeout", "unreachable") or attempt == attempts - 1:
                 return result
             time.sleep(min(0.25 * 2 ** attempt, max(0, deadline - time.monotonic())) if deadline is not None else 0.25 * 2 ** attempt)
 

static/agent/manifest.json

@@ -1,8 +1,8 @@
 {
-  "version": "1.4.12",
+  "version": "1.4.13",
   "protocolVersion": 1,
   "platform": "debian-amd64",
-  "sha256": "1e1bf2f04c02f0c8d0db8e4dbf39fadddc39c8bd04db5b8db753b6fe9e04c315",
+  "sha256": "2f3449058cd4b3ab9e3ff8f541fa3c65ad2d94aabe20694cdec3a6e59c8a4fd6",
   "updaterProtocolVersion": 1,
   "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8"
 }

tests/unit/host-status-batch.test.ts

@@ -47,11 +47,14 @@ async function setup(count = 5, hostCount = 1) {
 describe('host status observations', () => {
     it('uses one batch per host for a fifteen-server pass and shares overlapping scans', async () => {
         const api = await setup(5, 3);
+        const hosts = await import('../../src/lib/server/hosts/index.js');
+        const wait = vi.spyOn(hosts, 'waitHostOperationStatus');
         const passes = [api.fanoutStatus(), api.fanoutStatus()];
         try {
             await vi.waitFor(() => expect(api.operations()).toHaveLength(3));
             for (const operation of api.operations()) {
                 expect(operation.kind).toBe('rcon.status-batch');
+                expect(wait).toHaveBeenCalledWith(operation.id, 25_000);
                 expect(operation.payload.instances).toHaveLength(5);
                 expect(api.pendingHostOperations(operation.hostId, 'health').map(op => op.id)).toEqual([operation.id]);
                 expect(api.pendingHostOperations(operation.hostId, 'write')).toEqual([]);
@@ -115,6 +118,32 @@ describe('host status observations', () => {
         expect(api.operations()).toHaveLength(1);
     });
 
+    it('preserves previous game evidence for deadline skips while recording actual timeouts', async () => {
+        const api = await setup(3);
+        const first = api.fanoutStatus();
+        await vi.waitFor(() => expect(api.operations()).toHaveLength(1));
+        api.complete(); await first;
+        const previous = api.getServer('server-0-0')!;
+        const previousGame = api.monitoringStatus('server-0-0')?.game;
+        vi.setSystemTime(at + 31_000);
+        const pass = api.fanoutStatus();
+        try {
+            await vi.waitFor(() => expect(api.operations()).toHaveLength(2));
+            const operation = api.operations().find(op => op.state !== 'completed')!;
+            api.reportHostOperation(operation.hostId, operation.id, { state: 'completed', result: { ok: false, results: [
+                { instanceId: 'instance-0-0', generation: 1, reply: { ok: false, reason: 'deadline', text: '' } },
+                { instanceId: 'instance-0-1', generation: 1, reply: { ok: false, reason: 'timeout', text: '' } },
+                { instanceId: 'instance-0-2', generation: 1, reply: { ok: true, text: 'map: mp_kino\nnum score ping guid name\n--- ----- ---- ---- ----\n' } }
+            ] } });
+        } finally { api.complete(); await pass; }
+        expect(api.getServer('server-0-0')).toMatchObject({ last_checked_at: previous.last_checked_at,
+            last_seen_at: previous.last_seen_at, last_error: previous.last_error, last_players: previous.last_players,
+            last_reply_snippet: previous.last_reply_snippet });
+        expect(api.monitoringStatus('server-0-0')?.game).toEqual(previousGame);
+        expect(api.getServer('server-0-1')?.last_error).toContain('timeout:');
+        expect(Date.parse(api.getServer('server-0-2')!.last_seen_at!)).toBeGreaterThanOrEqual(at + 31_000);
+    });
+
     it('records each failed or omitted result without retrying the batch per server', async () => {
         const api = await setup();
         const pass = api.fanoutStatus();