Batch live server updates once per animation frame

src/lib/ui/stores/servers.ts

@@ -1,4 +1,4 @@
-import { writable, readonly } from 'svelte/store';
+import { writable, readonly, get } from 'svelte/store';
 import { browser } from '$app/environment';
 import type { GameTelemetry } from '$lib/telemetry.js';
 
@@ -56,13 +56,17 @@ export function serverTimeNow(): number {
     return serverClock ? serverClock.time + performance.now() - serverClock.receivedAt : Date.now();
 }
 
-export function syncServerTime(serverNow?: number): void {
+function updateServerClock(serverNow?: number): void {
     if (typeof serverNow === 'number' && Number.isFinite(serverNow)) {
         // A delayed HTTP snapshot must not rewind time behind a newer stream
         // update and make that update look like it came from the future.
         const time = serverClock ? Math.max(serverNow, serverTimeNow()) : serverNow;
         serverClock = { time, receivedAt: performance.now() };
     }
+}
+
+export function syncServerTime(serverNow?: number): void {
+    updateServerClock(serverNow);
     // Refresh immediately: a suspended tab's interval may be well behind.
     _uiNow.set(serverTimeNow());
 }
@@ -74,35 +78,67 @@ let source: EventSource | null = null;
 export function startServerStream(): (() => void) | undefined {
     if (!browser || source) return;
     syncServerTime();
-    source = new EventSource('/api/events/servers');
+    const stream = source = new EventSource('/api/events/servers');
-    source.addEventListener('server-telemetry', (evt) => {
+    let stopped = false;
+    let frame: number | null = null;
+    let pendingStatus: Map<string, ServerStatus> | null = null;
+    let pendingTelemetry: Map<string, GameTelemetry> | null = null;
+    let pendingMatches: Map<string, GameTelemetry> | null = null;
+    let receivedAt = 0;
+    const flush = () => {
+        frame = null;
+        if (stopped) return;
+        const statuses = pendingStatus, telemetry = pendingTelemetry, matches = pendingMatches;
+        const statusAt = receivedAt;
+        pendingStatus = pendingTelemetry = pendingMatches = null;
+        // Publish the latest clock before exposing reports, including after a
+        // suspended frame. Every changed store is published at most once per frame.
+        syncServerTime();
+        if (statuses) { statusByServer.set(statuses); statusReceivedAt.set(statusAt); }
+        if (matches) matchEventsByServer.set(matches);
+        if (telemetry) telemetryByServer.set(telemetry);
+    };
+    const schedule = () => { if (frame === null) frame = requestAnimationFrame(flush); };
+    stream.addEventListener('server-telemetry', (evt) => {
+        if (stopped) return;
         const data = JSON.parse((evt as MessageEvent).data) as { entries: GameTelemetry[]; snapshot?: boolean; serverNow?: number };
-        syncServerTime(data.serverNow);
+        updateServerClock(data.serverNow);
-        matchEventsByServer.update((previous) => {
+        // Merge into one private map per frame. Snapshots replace all earlier
-            const next = data.snapshot ? new Map<string, GameTelemetry>() : new Map(previous);
+        // evidence; later deltas still apply in arrival order.
-            for (const entry of data.entries) {
+        if (data.snapshot) {
-                if (entry.event === 'game_start' || entry.event === 'game_end') next.set(entry.server, entry);
+            pendingTelemetry = new Map();
+            pendingMatches = new Map();
         }
-            return next;
+        pendingTelemetry ??= new Map(get(telemetryByServer));
+        for (const entry of data.entries) {
+            pendingTelemetry.set(entry.server, entry);
+            if (entry.event === 'game_start' || entry.event === 'game_end') {
+                pendingMatches ??= new Map(get(matchEventsByServer));
+                pendingMatches.set(entry.server, entry);
+            }
+        }
+        schedule();
     });
-        telemetryByServer.update((previous) => {
+    stream.addEventListener('server-status', (evt) => {
-            const next = data.snapshot ? new Map<string, GameTelemetry>() : new Map(previous);
+        if (stopped) return;
-            for (const entry of data.entries) next.set(entry.server, entry);
-            return next;
-        });
-    });
-    source.addEventListener('server-status', (evt) => {
         const data = JSON.parse((evt as MessageEvent).data) as { entries: ServerStatus[]; serverNow?: number; delta?: boolean };
-        syncServerTime(data.serverNow);
+        updateServerClock(data.serverNow);
-        statusByServer.update(previous => {
+        receivedAt = Date.now();
-            const m = data.delta ? new Map(previous) : new Map<string, ServerStatus>();
+        if (!data.delta) pendingStatus = new Map();
-            for (const e of data.entries) m.set(e.server, { ...e, receivedAt: Date.now() });
+        pendingStatus ??= new Map(get(statusByServer));
-            return m;
+        for (const entry of data.entries) pendingStatus.set(entry.server, { ...entry, receivedAt });
-        });
+        schedule();
-        statusReceivedAt.set(Date.now());
     });
     if (tickInterval === null) {
         tickInterval = setInterval(() => { if (typeof document === 'undefined' || document.visibilityState !== 'hidden') _uiNow.set(serverTimeNow()); }, 1000);
     }
-    return () => { source?.close(); source = null; if (tickInterval) clearInterval(tickInterval); tickInterval = null; };
+    return () => {
+        if (stopped) return;
+        stopped = true;
+        if (frame !== null) cancelAnimationFrame(frame);
+        pendingStatus = pendingTelemetry = pendingMatches = null;
+        stream.close(); source = null;
+        if (tickInterval) clearInterval(tickInterval);
+        tickInterval = null;
+    };
 }

tests/unit/ui-server-clock.test.ts

@@ -8,9 +8,16 @@ import { gameEvent } from '../helpers/telemetry.js';
 vi.mock('$app/environment', () => ({ browser: true }));
 
 const epoch = Date.parse('2026-09-19T12:00:00Z');
-let source: EventTarget;
+let source: EventTarget & { close: () => void };
 let stop: (() => void) | undefined;
 let snapshot: { serverNow: number; servers: FleetServer[]; integrations: object };
+let frames: Map<number, FrameRequestCallback>;
+
+function flushFrame() {
+    const callbacks = [...frames.values()];
+    frames.clear();
+    for (const callback of callbacks) callback(performance.now());
+}
 
 function server(timestamp = epoch, changes: Partial<FleetServer> = {}): FleetServer {
     return {
@@ -35,12 +42,20 @@ function status(timestamp: number, changes: Partial<ServerStatus> = {}): ServerS
 
 function emit(serverNow: number, entries: ServerStatus[]) {
     source.dispatchEvent(new MessageEvent('server-status', { data: JSON.stringify({ serverNow, entries }) }));
+    flushFrame();
 }
 
 beforeEach(() => {
     vi.resetModules();
     vi.useFakeTimers({ toFake: ['Date', 'performance', 'setTimeout', 'clearTimeout', 'setInterval', 'clearInterval'] });
     vi.setSystemTime(epoch);
+    frames = new Map();
+    let frameId = 0;
+    vi.stubGlobal('requestAnimationFrame', vi.fn((callback: FrameRequestCallback) => {
+        frames.set(++frameId, callback);
+        return frameId;
+    }));
+    vi.stubGlobal('cancelAnimationFrame', vi.fn((id: number) => { frames.delete(id); }));
     snapshot = { serverNow: epoch, servers: [server()], integrations: {} };
     vi.stubGlobal('fetch', vi.fn(async (url: string) => new Response(JSON.stringify(
         url === '/api/servers' ? snapshot : { operations: [] }
@@ -66,6 +81,130 @@ async function setup() {
     return { ...clock, ...fleet };
 }
 
+function queueEvent(type: string, data: object) {
+    source.dispatchEvent(new MessageEvent(type, { data: JSON.stringify(data) }));
+}
+
+describe('animation-frame stream batching', () => {
+    it('publishes each changed store once for a burst and preserves lifecycle evidence through heartbeats', async () => {
+        const stores = await setup();
+        emit(epoch, [status(epoch), status(epoch, { server: 'two' })]);
+        const before = get(stores.statusByServer);
+        await vi.advanceTimersByTimeAsync(1000);
+        const watched = [stores.statusByServer, stores.statusReceivedAt, stores.telemetryByServer, stores.matchEventsByServer, stores.uiNow];
+        const changes = watched.map(() => vi.fn());
+        const unsubscribe = watched.map((store, index) => store.subscribe(changes[index]));
+        changes.forEach(change => change.mockClear());
+        vi.mocked(requestAnimationFrame).mockClear();
+        try {
+            for (let i = 0; i < 100; i++) {
+                queueEvent('server-status', { delta: true, serverNow: epoch + 1000 + i,
+                    entries: [status(epoch + 1000, { players: i })] });
+                queueEvent('server-telemetry', { serverNow: epoch + 1000 + i,
+                    entries: [{ ...gameEvent(i === 0 ? 'game_start' : 'heartbeat', i, 1), receivedAt: new Date(epoch + 1000 + i).toISOString() }] });
+            }
+            expect(requestAnimationFrame).toHaveBeenCalledTimes(1);
+            changes.forEach(change => expect(change).not.toHaveBeenCalled());
+            expect(get(stores.statusByServer)).toBe(before);
+            expect(before.get('t5_zombie_theater_1')?.players).toBe(0);
+            flushFrame();
+            changes.forEach(change => expect(change).toHaveBeenCalledTimes(1));
+            expect(get(stores.statusByServer).size).toBe(2);
+            expect(get(stores.statusByServer).get('t5_zombie_theater_1')?.players).toBe(99);
+            expect(get(stores.telemetryByServer).get('t5_zombie_theater_1')).toMatchObject({ event: 'heartbeat', round: 99 });
+            expect(get(stores.matchEventsByServer).get('t5_zombie_theater_1')?.event).toBe('game_start');
+
+            queueEvent('server-telemetry', { entries: [{ ...gameEvent('heartbeat', 100, 1), receivedAt: new Date(epoch + 1100).toISOString() }] });
+            flushFrame();
+            expect(requestAnimationFrame).toHaveBeenCalledTimes(2);
+            expect(changes[2]).toHaveBeenCalledTimes(2);
+            expect(changes[3]).toHaveBeenCalledTimes(1);
+        } finally { unsubscribe.forEach(unsubscribe => unsubscribe()); }
+    });
+
+    it('applies snapshots and subsequent deltas in order, including empty snapshots', async () => {
+        const { statusByServer, telemetryByServer, matchEventsByServer } = await setup();
+        const telemetry = (event: GameEvent['event'], name: string) => ({ ...gameEvent(event, 1, 1, name), receivedAt: new Date(epoch).toISOString() });
+        emit(epoch, [status(epoch, { server: 'old' })]);
+        queueEvent('server-telemetry', { entries: [telemetry('game_start', 'old')] });
+        flushFrame();
+
+        queueEvent('server-status', { delta: true, entries: [status(epoch, { server: 'discarded' })] });
+        queueEvent('server-status', { entries: [status(epoch, { server: 'two' })] });
+        queueEvent('server-status', { delta: true, entries: [status(epoch, { server: 'three' })] });
+        queueEvent('server-telemetry', { entries: [telemetry('game_end', 'discarded')] });
+        queueEvent('server-telemetry', { snapshot: true, entries: [telemetry('heartbeat', 'two')] });
+        queueEvent('server-telemetry', { entries: [telemetry('game_end', 'three')] });
+        queueEvent('server-telemetry', { entries: [telemetry('heartbeat', 'three')] });
+        flushFrame();
+        expect([...get(statusByServer).keys()]).toEqual(['two', 'three']);
+        expect([...get(telemetryByServer).keys()]).toEqual(['two', 'three']);
+        expect([...get(matchEventsByServer).keys()]).toEqual(['three']);
+        expect(get(matchEventsByServer).get('three')?.event).toBe('game_end');
+        expect(get(telemetryByServer).get('three')?.event).toBe('heartbeat');
+
+        queueEvent('server-status', { entries: [] });
+        queueEvent('server-telemetry', { snapshot: true, entries: [] });
+        flushFrame();
+        expect(get(statusByServer).size).toBe(0);
+        expect(get(telemetryByServer).size).toBe(0);
+        expect(get(matchEventsByServer).size).toBe(0);
+    });
+
+    it('retains event arrival times so buffered status cannot overwrite a newer HTTP snapshot', async () => {
+        const { fleetServers, refreshFleet, statusByServer, statusReceivedAt } = await setup();
+        await vi.advanceTimersByTimeAsync(1000);
+        queueEvent('server-status', { delta: true, serverNow: epoch + 1000, entries: [status(epoch + 1000, { players: 1 })] });
+        await vi.advanceTimersByTimeAsync(1000);
+        snapshot = { ...snapshot, serverNow: epoch + 2000, servers: [server(epoch + 2000, { last_players: 4 })] };
+        await refreshFleet();
+        queueEvent('server-status', { delta: true, serverNow: epoch + 2000, entries: [status(epoch + 2000, { server: 'two' })] });
+        flushFrame();
+        expect(get(statusByServer).get('t5_zombie_theater_1')?.receivedAt).toBe(epoch + 1000);
+        expect(get(statusReceivedAt)).toBe(epoch + 2000);
+        expect(get(fleetServers)[0].last_players).toBe(4);
+    });
+
+    it('accounts for frame delay without rewinding the server clock', async () => {
+        const { uiNow, statusByServer } = await setup();
+        vi.stubGlobal('document', { visibilityState: 'hidden' });
+        queueEvent('server-status', { serverNow: epoch + 5000, entries: [status(epoch + 5000)] });
+        await vi.advanceTimersByTimeAsync(2000);
+        queueEvent('server-telemetry', { serverNow: epoch + 4000, entries: [] });
+        await vi.advanceTimersByTimeAsync(1000);
+        expect(get(uiNow)).toBe(epoch);
+        expect(get(statusByServer).size).toBe(0);
+        expect(frames.size).toBe(1);
+        flushFrame();
+        expect(get(uiNow)).toBe(epoch + 8000);
+        expect(get(statusByServer).size).toBe(1);
+    });
+
+    it('cancels queued frames and discards buffered events when stopping and restarting', async () => {
+        const { statusByServer, telemetryByServer, startServerStream } = await setup();
+        const oldSource = source;
+        const close = vi.spyOn(oldSource, 'close');
+        queueEvent('server-status', { entries: [status(epoch)] });
+        queueEvent('server-telemetry', { entries: [{ ...gameEvent('game_start'), receivedAt: new Date(epoch).toISOString() }] });
+        const callback = [...frames.values()][0];
+        const oldStop = stop!;
+        oldStop();
+        expect(close).toHaveBeenCalledTimes(1);
+        expect(cancelAnimationFrame).toHaveBeenCalledTimes(1);
+        expect(frames.size).toBe(0);
+        callback(0);
+        expect(get(statusByServer).size).toBe(0);
+        expect(get(telemetryByServer).size).toBe(0);
+        stop = startServerStream();
+        oldStop();
+        oldSource.dispatchEvent(new MessageEvent('server-status', { data: JSON.stringify({ entries: [status(epoch)] }) }));
+        queueEvent('server-status', { delta: true, entries: [status(epoch, { server: 'new' })] });
+        flushFrame();
+        expect([...get(statusByServer).keys()]).toEqual(['new']);
+        expect(get(telemetryByServer).size).toBe(0);
+    });
+});
+
 describe('Fleet report clock', () => {
     it('accepts newer navigation snapshots while keeping stream evidence newer than the snapshot', async () => {
         const { fleetServers, hydrateFleet } = await setup();
@@ -86,6 +225,7 @@ describe('Fleet report clock', () => {
         source.dispatchEvent(new MessageEvent('server-status', { data: JSON.stringify({
             delta: true, serverNow: epoch + 1000, entries: [status(epoch + 1000, { players: 2 })]
         }) }));
+        flushFrame();
         expect(get(statusByServer).size).toBe(2);
         expect(get(fleetServers).find(server => server.name === 't5_zombie_theater_1')?.last_players).toBe(2);
         const changes = vi.fn();
@@ -103,6 +243,7 @@ describe('Fleet report clock', () => {
             source.dispatchEvent(new MessageEvent('server-telemetry', { data: JSON.stringify({ serverNow: epoch + seconds * 1000,
                 entries: [{ ...gameEvent(event, 2, players), receivedAt: at(seconds),
                     timing: { matchId: 1, players: {}, server: timing } }] }) }));
+            flushFrame();
             return get(fleetServers)[0];
         };
         const active = { activeSince: at(1), inactiveSince: null, gameStartedAt: at(1), gameEndedAt: null };
@@ -132,15 +273,17 @@ describe('Fleet report clock', () => {
             entries: [{ ...gameEvent('heartbeat', 2, 1), receivedAt: new Date(epoch).toISOString(),
                 timing: { matchId: 1, players: {}, server: { activeSince: startedAt, inactiveSince: null,
                     gameStartedAt: startedAt, gameEndedAt: null } } }] }) }));
+        flushFrame();
         expect(get(fleetServers)[0]).toMatchObject({ active_since: startedAt, game_started_at: startedAt });
     });
 
-    it('updates count and roster immediately from telemetry and expires them together', async () => {
+    it('updates count and roster on the next frame and expires them together', async () => {
         snapshot.servers = [server(epoch, { container_state: 'running', container_checked_at: new Date(epoch).toISOString(), last_players: 4 })];
         const { fleetServers, uiNow } = await setup();
         await vi.advanceTimersByTimeAsync(1000);
         source.dispatchEvent(new MessageEvent('server-telemetry', { data: JSON.stringify({ serverNow: epoch + 1000,
             entries: [{ ...gameEvent('heartbeat', 15, 3), receivedAt: new Date(epoch + 1000).toISOString() }] }) }));
+        flushFrame();
         expect(get(fleetServers)[0]).toMatchObject({ health: 'online', last_players: 3, status_source: 'telemetry' });
         expect(get(fleetServers)[0].roster).toHaveLength(3);
         expect(matchesFleet(get(fleetServers)[0], 'Player 3', 'occupied', '', [], [], get(uiNow))).toBe(true);
@@ -149,12 +292,15 @@ describe('Fleet report clock', () => {
         expect(matchesFleet(get(fleetServers)[0], '', 'occupied', '', [], [], get(uiNow))).toBe(false);
     });
 
-    it('applies game lifecycle events immediately and clears them after an empty report', async () => {
+    it('applies game lifecycle events on the next frame and clears them after an empty report', async () => {
         const { fleetServers } = await setup();
-        const report = (event: string, time: number) => source.dispatchEvent(new MessageEvent('server-telemetry', {
+        const report = (event: string, time: number) => {
+            source.dispatchEvent(new MessageEvent('server-telemetry', {
                 data: JSON.stringify({ serverNow: time, entries: [{ server: 't5_zombie_theater_1', event,
                     receivedAt: new Date(time).toISOString(), players: [] }] })
             }));
+            flushFrame();
+        };
         emit(epoch, [status(epoch, { players: 1 })]);
         report('game_end', epoch + 1000);
         expect(get(fleetServers)[0].game_ended_at).toBe(new Date(epoch + 1000).toISOString());
@@ -249,6 +395,7 @@ describe('Fleet report clock', () => {
         source.dispatchEvent(new MessageEvent('server-telemetry', {
             data: JSON.stringify({ serverNow: epoch, entries: [], snapshot: true })
         }));
+        flushFrame();
         expect(get(uiNow)).toBe(epoch);
         await vi.advanceTimersByTimeAsync(1000);
         expect(get(uiNow)).toBe(epoch + 1000);