Supplement telemetry rosters with fresh RCON pings

docs/beta-test-plans.md

@@ -33,6 +33,17 @@ there are no players. Names and GUIDs are strings; keep GUIDs as strings to
 preserve their precision. Round and stat values are nonnegative JSON integers.
 Each GUID must appear only once within a report's player array.
 
+Every event also accepts an optional per-player `ping`: the game server's current
+latency measurement in milliseconds, as a nonnegative JSON integer (for example,
+`"ping": 42`). Send `null` or omit the field when unavailable; do not substitute
+zero for a missing measurement. Older reporters without this field remain valid.
+Fleet prefers telemetry ping and fills missing readings from fresh RCON status,
+matching players by GUID. Occupied lobbies continue RCON polling every 30 seconds.
+RCON supplements only ping; telemetry still supplies roster membership and scores.
+RCON pings expire after 90 seconds and must match the current map and server
+lifecycle. Missing or ambiguous matches display `—`. Telemetry ping is retained
+and streamed with the latest report; it does not contribute to leaderboard statistics.
+
 ### Game start
 
 Send when the game starts. Round and stats are not required.

src/lib/server-observation.ts

@@ -50,15 +50,27 @@ export function freshTelemetry(report: GameTelemetry | undefined, now: number, a
 export function selectServerObservation<T extends ServerObservation>(server: T, report: GameTelemetry | undefined, now: number): T {
     if (server.telemetry && (!report || Date.parse(server.telemetry.receivedAt) > Date.parse(report.receivedAt))) report = server.telemetry;
     const raw = server.rcon_status ?? server;
-    const result: T = { ...server, ...raw, last_error: rconHealthError(raw), telemetry: report, status_source: raw.last_seen_at ? 'rcon' : null };
+    const result: T = { ...server, ...raw, rcon_status: raw, last_error: rconHealthError(raw), telemetry: report, status_source: raw.last_seen_at ? 'rcon' : null };
     const containerUnknown = server.container_state === 'unknown';
     if (['provisioning','starting'].includes(server.health) || (server.container_state && server.container_state !== 'running' && !containerUnknown) || (!server.container_state && server.health === 'offline')) {
         return { ...result, health: server.health, last_error: null, last_players: null, roster: null, status_source: null };
     }
     const checked = server.container_checked_at ?? server.last_checked_at ?? raw.last_seen_at;
     const containerFresh = server.container_state === 'running' && freshStatusTimestamp(checked ?? null, now);
+    const after = !server.telemetry_after || Date.parse(raw.last_seen_at ?? '') > Date.parse(server.telemetry_after);
+    const fresh = after && freshStatusTimestamp(raw.last_seen_at, now) && !result.last_error;
     if (containerFresh && report?.server === server.name && freshTelemetry(report, now, server.telemetry_after)) {
         const players = report.players.length;
+        // Telemetry owns roster membership. Supplement missing pings only with
+        // unambiguous RCON identities from the current map and server lifecycle.
+        const pings = new Map<string, number | null>();
+        if (fresh && raw.last_map === report.map) {
+            for (const player of raw.roster ?? []) {
+                if (!player.guid) continue;
+                const ping = player.ping;
+                pings.set(player.guid, pings.has(player.guid) || ping === null || !Number.isSafeInteger(ping) || ping < 0 ? null : ping);
+            }
+        }
         // New reports carry persisted anchors through both SSE and HTTP. Keep
         // the fallback for telemetry saved before server timing was included.
         const timing = report.timing?.server ?? {
@@ -70,14 +82,12 @@ export function selectServerObservation<T extends ServerObservation>(server: T,
         return { ...result, health: 'online', status_source: 'telemetry', last_players: players,
             last_map: report.map, last_seen_at: report.receivedAt, last_error: null,
             roster: report.players.map(player => ({ num: null, score: 'score' in player ? player.score : null,
-                ping: null, guid: player.guid, name: player.name })),
+                ping: player.ping ?? pings.get(player.guid) ?? null, guid: player.guid, name: player.name })),
             active_since: players > 0 ? timing.activeSince : null,
             inactive_since: players === 0 ? timing.inactiveSince : null,
             game_started_at: players === 0 ? null : timing.gameStartedAt,
             game_ended_at: players === 0 ? null : timing.gameEndedAt };
     }
-    const after = !server.telemetry_after || Date.parse(raw.last_seen_at ?? '') > Date.parse(server.telemetry_after);
-    const fresh = after && freshStatusTimestamp(raw.last_seen_at, now) && !result.last_error;
     // A current authenticated RCON reply can establish game liveness when
     // inspection is unavailable, without claiming the container was inspected.
     const containerExpired = !containerUnknown && !!server.container_checked_at && !containerFresh;

src/lib/server/pipeline/worker.ts

@@ -138,7 +138,7 @@ async function statusPass(): Promise<void> {
     const servers = snapshot.servers();
     if (!servers.length) return;
     const probes = prepareMonitoringRcon(servers.filter(server => !pendingStatus.has(server.name)
-        && !snapshot.provisioning(server.name) && !snapshot.settling(server.name) && !snapshot.telemetryEvidence(server.name)).map(server => server.name), snapshot);
+        && !snapshot.provisioning(server.name) && !snapshot.settling(server.name) && snapshot.telemetryEvidence(server.name)?.players.length !== 0).map(server => server.name), snapshot);
     const results = await Promise.all(servers.map(server => queueStatus(server, snapshot, probes.get(server.name))));
     const capacities = new Map(servers.map((server, i) => [server.name, results[i]]));
     try {
@@ -197,7 +197,8 @@ async function refreshServerStatus(server: string, snapshot: MonitoringSnapshot,
                 updates.last_error = null;
                 updates.rcon_failures = 0;
                 updates.last_reply_snippet = null;
-            } else if (unavailable || !snapshot.telemetryEvidence(srv.name)) {
+            } else if (unavailable || snapshot.telemetryEvidence(srv.name)?.players.length !== 0) {
+                // Occupied telemetry rosters still need RCON for player pings.
                 flush();
                 const probeStartedAt = new Date().toISOString();
                 const probe = await (probeStatus ? probeStatus() : rconForMonitoring(srv.name, false, snapshot)).catch(err => {

src/lib/telemetry.ts

@@ -1,6 +1,8 @@
 export interface TelemetryPlayer {
     name: string;
     guid: string;
+    /** Server-measured latency in milliseconds; absent/null when unavailable. */
+    ping?: number | null;
 }
 
 export interface TelemetryPlayerStats extends TelemetryPlayer {
@@ -51,7 +53,13 @@ export function parseGameEvent(value: unknown): GameEvent {
         for (const key of ['name', 'guid'] as const) {
             if (typeof player[key] !== 'string' || !player[key].trim()) throw new Error(`players[${index}].${key} must be a nonempty string`);
         }
-        const identity = { name: player.name as string, guid: player.guid as string };
+        const identity: TelemetryPlayer = { name: player.name as string, guid: player.guid as string };
+        if (player.ping !== undefined) {
+            if (player.ping !== null && (!Number.isSafeInteger(player.ping) || (player.ping as number) < 0)) {
+                throw new Error(`players[${index}].ping must be a nonnegative integer or null`);
+            }
+            identity.ping = player.ping as number | null;
+        }
         if (guids.has(identity.guid)) throw new Error(`players[${index}].guid must be unique within the report`);
         guids.add(identity.guid);
         if (body.event === 'game_start') return identity;

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

@@ -1,5 +1,6 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import { newFixture, type Fixture } from '../helpers/fixture.js';
+import { gameEvent, postTelemetry } from '../helpers/telemetry.js';
 
 let fixture: Fixture;
 let at: number;
@@ -45,6 +46,19 @@ async function setup(count = 5, hostCount = 1) {
 }
 
 describe('host status observations', () => {
+    it('batches occupied telemetry lobbies for pings while skipping empty lobbies', async () => {
+        const api = await setup(2);
+        await postTelemetry(gameEvent('heartbeat', 15, 1, 'server-0-0'));
+        await postTelemetry(gameEvent('heartbeat', 15, 0, 'server-0-1'));
+        const pass = api.fanoutStatus();
+        try {
+            await vi.waitFor(() => expect(api.operations()).toHaveLength(1));
+            expect(api.operations()[0]).toMatchObject({ kind: 'rcon.status-batch',
+                payload: { instances: [{ instanceId: 'instance-0-0', generation: 1 }] } });
+        } finally { api.complete(); await pass; }
+        expect(api.getServer('server-0-1')?.last_seen_at).toBeNull();
+    });
+
     it('bounds monitoring reads below five per server across thirty instances', async () => {
         const api = await setup(10, 3);
         const hosts = await import('../../src/lib/server/hosts/index.js');

tests/unit/server-observation.test.ts

@@ -47,6 +47,61 @@ describe('shared server observation', () => {
         expect(selected.rcon_status).toBe(raw);
     });
 
+    it('supplements telemetry members with GUID-matched pings without changing scores or names', () => {
+        const telemetry = report();
+        const raw = server({ roster: [
+            { ...telemetry.players[1], num: 2, score: 10, ping: 0, name: 'Old name' },
+            { ...telemetry.players[0], num: 3, score: 20, ping: 999 },
+            { ...telemetry.players[2], num: 4, score: 30, ping: 45, guid: 'different-player' }
+        ] });
+        const selected = selectServerObservation(server({ rcon_status: raw }), telemetry, now);
+        expect(selected.roster).toEqual(telemetry.players.map((player, i) => ({
+            name: player.name, guid: player.guid, num: null, score: 5000, ping: [999, 0, null][i]
+        })));
+        expect(selected.last_players).toBe(3);
+        expect(selected.rcon_status).toBe(raw);
+        // A new telemetry report must not renew the RCON sample's age.
+        const expired = selectServerObservation({ ...selected, container_checked_at: at(90_001) }, report(90_001), now + 90_001);
+        expect(expired.health).toBe('online');
+        expect(expired.roster?.map(player => player.ping)).toEqual([null, null, null]);
+    });
+
+    it('prefers telemetry ping, including zero, and fills absent or null readings from RCON', () => {
+        const telemetry = report();
+        telemetry.players[0].ping = 0;
+        telemetry.players[1].ping = null;
+        const raw = server({ roster: telemetry.players.map(player => ({ ...player, num: 0, score: 10, ping: 42 })) });
+        expect(selectServerObservation(server({ rcon_status: raw }), telemetry, now).roster?.map(player => player.ping)).toEqual([0, 42, 42]);
+    });
+
+    it.each([
+        { last_seen_at: at(-90_001) }, { last_seen_at: at(2000) }, { last_seen_at: null },
+        { last_error: 'timeout', rcon_failures: 2 }, { last_map: 'other-map' }
+    ])('does not supplement pings from unusable RCON evidence: %j', changes => {
+        const telemetry = report();
+        const raw = server({ ...changes, roster: [{ ...telemetry.players[0], num: 0, score: 100, ping: 42 }] });
+        expect(selectServerObservation(server({ rcon_status: raw }), telemetry, now).roster?.[0].ping).toBeNull();
+    });
+
+    it('requires RCON samples after the latest lifecycle and tolerates one transient failure', () => {
+        const telemetry = report();
+        const raw = server({ last_seen_at: at(-90_000), roster: [{ ...telemetry.players[0], num: 0, score: 100, ping: 42 }] });
+        expect(selectServerObservation(server({ rcon_status: raw }), telemetry, now).roster?.[0].ping).toBe(42);
+        expect(selectServerObservation(server({ rcon_status: raw, telemetry_after: at(-90_000) }), telemetry, now).roster?.[0].ping).toBeNull();
+        expect(selectServerObservation(server({ rcon_status: { ...raw, last_error: 'timeout', rcon_failures: 1 } }), telemetry, now).roster?.[0].ping).toBe(42);
+    });
+
+    it.each(['duplicate', 'missing-guid', 'null', 'negative', 'invalid'])('rejects %s ping matches', mode => {
+        const telemetry = report();
+        const player = { ...telemetry.players[0], num: 0, score: 100, ping: 42 as number | null, guid: telemetry.players[0].guid as string | null };
+        if (mode === 'missing-guid') player.guid = null;
+        if (mode === 'null') player.ping = null;
+        if (mode === 'negative') player.ping = -1;
+        if (mode === 'invalid') player.ping = NaN;
+        const raw = server({ roster: mode === 'duplicate' ? [player, { ...player, ping: 50 }] : [player] });
+        expect(selectServerObservation(server({ rcon_status: raw }), telemetry, now).roster?.[0].ping).toBeNull();
+    });
+
     it('uses the latest HTTP or streamed telemetry and falls back without extending freshness', () => {
         const original = server();
         const selected = selectServerObservation(server({ rcon_status: original, telemetry: report(0, 2) }), report(-1000), now);
@@ -56,6 +111,17 @@ describe('shared server observation', () => {
         expect(fallback).toMatchObject({ status_source: 'rcon', health: 'online', last_players: 4, last_seen_at: at(40_000) });
     });
 
+    it('displays telemetry pings and clears missing readings on subsequent reports', () => {
+        const telemetry = report(0, 4);
+        telemetry.players = telemetry.players.map((player, i) => ({ ...player, ping: [0, 42, 999, null][i] }));
+        const selected = selectServerObservation(server(), telemetry, now);
+        expect(selected.roster?.map(player => player.ping)).toEqual([0, 42, 999, null]);
+        expect(selectServerObservation(selected, report(1, 4), now + 1).roster?.map(player => player.ping)).toEqual([null, null, null, null]);
+        const start = { ...gameEvent('game_start', 0, 1, 'managed'), event: 'game_start' as const, receivedAt: at(),
+            players: [{ name: 'Player 1', guid: '123', ping: 25 }] };
+        expect(selectServerObservation(server(), start, now).roster?.[0]).toMatchObject({ score: null, ping: 25 });
+    });
+
     it.each(['starting', 'restarting', 'unhealthy', 'exited'])('telemetry cannot override container state %s', state => {
         const selected = selectServerObservation(server({ health: state === 'starting' || state === 'restarting' ? 'starting' : 'offline', container_state: state }), report(), now);
         expect(selected.health).not.toBe('online');

tests/unit/server-status.test.ts

@@ -372,6 +372,8 @@ describe('database fleet and container readiness', () => {
         vi.useFakeTimers({ toFake: ['Date'] });
         vi.setSystemTime(startedAt);
         inspect.mockResolvedValue({ ok: true, state: 'running' });
+        rcon.mockResolvedValue({ ok: true, endpoint: { host: 'localhost', port: 28960 }, rawReply: 'status',
+            status: { players: 1, map: 'zombie_theater', gametype: 'zom', roster: [] } });
         expect((await postTelemetry(gameEvent('game_start', 0, 1, 'managed'))).status).toBe(200);
         expect(getServer('managed')).toMatchObject({ active_since: at(0), inactive_since: null, game_started_at: at(0) });
 
@@ -384,7 +386,7 @@ describe('database fleet and container readiness', () => {
         const events: unknown[] = [];
         const unsubscribe = subscribe(event => events.push(event));
         try { await fanoutStatus(); } finally { unsubscribe(); }
-        expect(rcon).not.toHaveBeenCalled();
+        expect(rcon).toHaveBeenCalledOnce();
         expect((await (await GET()).json()).servers[0]).toMatchObject({
             active_since: at(0), game_started_at: at(0), last_seen_at: at(19)
         });
@@ -488,28 +490,57 @@ describe('database fleet and container readiness', () => {
         expect(snapshot.serverNow).toBeLessThanOrEqual(Date.now());
     });
 
-    it('uses telemetry consistently in REST, status events, and metrics without polling RCON', async () => {
+    it('supplements telemetry with RCON pings consistently in REST and status events', async () => {
         const { db, inspect, rcon, fanoutStatus } = await setup();
         const { GET } = await import('../../src/routes/api/servers/+server.js');
         const { subscribe } = await import('../../src/lib/server/pipeline/bus.js');
         db.prepare("UPDATE servers SET last_players = 4, last_seen_at = ? WHERE name = 'managed'").run(new Date().toISOString());
         inspect.mockResolvedValue({ ok: true, state: 'running' });
-        await postTelemetry(gameEvent('heartbeat', 15, 3, 'managed'));
+        const telemetry = gameEvent('heartbeat', 15, 3, 'managed');
+        telemetry.players[0].ping = 42;
+        rcon.mockResolvedValue({ ok: true, endpoint: { host: 'localhost', port: 28960 }, rawReply: 'status',
+            status: { players: 4, map: 'zombie_theater', gametype: 'zom', roster: [
+                { num: 0, guid: telemetry.players[0].guid, name: 'Old name', score: 50, ping: 100 },
+                { num: 1, guid: telemetry.players[1].guid, name: 'Another old name', score: 100, ping: 30 }
+            ] } });
+        await postTelemetry(telemetry);
         const events: unknown[] = [];
         const unsubscribe = subscribe(event => events.push(event));
         try { await fanoutStatus(); } finally { unsubscribe(); }
-        expect(rcon).not.toHaveBeenCalled();
+        expect(rcon).toHaveBeenCalledOnce();
         const server = (await (await GET()).json()).servers[0];
         expect(server).toMatchObject({ health: 'online', last_players: 3, status_source: 'telemetry', last_error: null,
-            rcon_status: { last_players: 4, last_error: 'timeout:old process' } });
+            rcon_status: { last_players: 4, last_error: null } });
         expect(server.roster).toHaveLength(3);
-        expect(server.roster[0]).toMatchObject({ ping: null, score: 5000 });
+        expect(server.roster[0]).toMatchObject({ ping: 42, score: 5000 });
+        expect(server.roster[1]).toMatchObject({ ping: 30, score: 5000, name: 'Player 2' });
+        expect(server.roster[2].ping).toBeNull();
         expect(events).toContainEqual(expect.objectContaining({ type: 'server-status', entries: [expect.objectContaining({
             health: 'online', players: 3, roster: server.roster, lastSeenAt: server.last_seen_at
         })] }));
         expect(db.prepare("SELECT players FROM metric_samples WHERE server = 'managed'").get()).toEqual({ players: 3 });
     });
 
+    it('keeps occupied telemetry online when supplementary RCON fails', async () => {
+        const { inspect, rcon, fanoutStatus } = await setup();
+        const { GET } = await import('../../src/routes/api/servers/+server.js');
+        inspect.mockResolvedValue({ ok: true, state: 'running' });
+        rcon.mockResolvedValue({ ok: false, endpoint: null, reason: 'timeout', detail: 'no reply' });
+        await postTelemetry(gameEvent('heartbeat', 15, 1, 'managed'));
+        await fanoutStatus();
+        expect(rcon).toHaveBeenCalledOnce();
+        expect((await (await GET()).json()).servers[0]).toMatchObject({ health: 'online', status_source: 'telemetry',
+            last_players: 1, last_error: null, roster: [{ name: 'Player 1', score: 5000, ping: null }] });
+    });
+
+    it('skips supplementary RCON for empty telemetry rosters', async () => {
+        const { inspect, rcon, fanoutStatus } = await setup();
+        inspect.mockResolvedValue({ ok: true, state: 'running' });
+        await postTelemetry(gameEvent('heartbeat', 15, 0, 'managed'));
+        await fanoutStatus();
+        expect(rcon).not.toHaveBeenCalled();
+    });
+
     it('falls back to RCON when telemetry expires and does not renew old reports', async () => {
         const { db, inspect, rcon, fanoutStatus } = await setup();
         const { GET } = await import('../../src/routes/api/servers/+server.js');

tests/unit/telemetry.test.ts

@@ -180,6 +180,33 @@ it('accepts all four events without an active beta and updates match lifecycle',
     expect((await post(gameEvent('game_end', 0, 0))).status).toBe(200);
 });
 
+it.each(['game_start', 'round_change', 'heartbeat', 'game_end'] as const)('preserves %s player pings through ingestion, streaming and reload', async event => {
+    const payload = gameEvent(event, 17);
+    payload.players = payload.players.map((player, i) => ({ ...player, ping: [0, 42, 999, null][i] }));
+    const { subscribe } = await import('../../src/lib/server/pipeline/bus.js');
+    const listener = vi.fn();
+    const unsubscribe = subscribe(listener);
+    try {
+        const response = await post(payload);
+        expect(response.status).toBe(200);
+        expect((await response.json()).telemetry.players).toEqual(payload.players);
+        expect(listener).toHaveBeenCalledWith({ type: 'server-telemetry', entries: [expect.objectContaining({ players: payload.players })] });
+        const { closeDb } = await import('../../src/lib/server/db/index.js');
+        closeDb();
+        const { listTelemetry } = await import('../../src/lib/server/fleet/telemetry.js');
+        const { telemetryReport } = await import('../../src/lib/server/fleet/status-evidence.js');
+        expect(listTelemetry()[0].players).toEqual(payload.players);
+        expect(telemetryReport(payload.server)?.players).toEqual(payload.players);
+    } finally { unsubscribe(); }
+});
+
+it.each([-1, 1.5, '42', false, {}, [], Number.MAX_SAFE_INTEGER + 1])('rejects invalid player ping: %j', async ping => {
+    for (const event of ['game_start', 'heartbeat'] as const) {
+        const body = gameEvent(event, 17, 1);
+        expect((await post({ ...body, players: [{ ...body.players[0], ping }] })).status).toBe(400);
+    }
+});
+
 it.each([null, [], {}, ...['boot', 'session_start', 'match_end', 'feature', 'other'].map(event => ({ ...payload, event })),
     { ...payload, map: '' }, { ...payload, map: undefined }, { ...payload, round: -1 },
     { ...payload, round: 1.5 }, { ...payload, round: '17' }, { ...payload, players: 4 }, { ...payload, players: null },