From a86499a0f9986396cfa2bfd824c258550f1350d4 Mon Sep 17 00:00:00 2001 From: Josh Wright Date: Mon, 5 Oct 2026 15:43:23 -0400 Subject: [PATCH] Batch live server updates once per animation frame --- src/lib/ui/stores/servers.ts | 86 ++++++++++----- tests/unit/ui-server-clock.test.ts | 161 +++++++++++++++++++++++++++-- 2 files changed, 215 insertions(+), 32 deletions(-) diff --git a/src/lib/ui/stores/servers.ts b/src/lib/ui/stores/servers.ts index b264133..a82afb8 100644 --- a/src/lib/ui/stores/servers.ts +++ b/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'); - source.addEventListener('server-telemetry', (evt) => { + const stream = source = new EventSource('/api/events/servers'); + let stopped = false; + let frame: number | null = null; + let pendingStatus: Map | null = null; + let pendingTelemetry: Map | null = null; + let pendingMatches: Map | 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); - matchEventsByServer.update((previous) => { - const next = data.snapshot ? new Map() : new Map(previous); - for (const entry of data.entries) { - if (entry.event === 'game_start' || entry.event === 'game_end') next.set(entry.server, entry); + updateServerClock(data.serverNow); + // Merge into one private map per frame. Snapshots replace all earlier + // evidence; later deltas still apply in arrival order. + if (data.snapshot) { + pendingTelemetry = new Map(); + pendingMatches = new Map(); + } + 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); } - return next; - }); - telemetryByServer.update((previous) => { - const next = data.snapshot ? new Map() : new Map(previous); - for (const entry of data.entries) next.set(entry.server, entry); - return next; - }); + } + schedule(); }); - source.addEventListener('server-status', (evt) => { + stream.addEventListener('server-status', (evt) => { + if (stopped) return; const data = JSON.parse((evt as MessageEvent).data) as { entries: ServerStatus[]; serverNow?: number; delta?: boolean }; - syncServerTime(data.serverNow); - statusByServer.update(previous => { - const m = data.delta ? new Map(previous) : new Map(); - for (const e of data.entries) m.set(e.server, { ...e, receivedAt: Date.now() }); - return m; - }); - statusReceivedAt.set(Date.now()); + updateServerClock(data.serverNow); + receivedAt = Date.now(); + if (!data.delta) pendingStatus = new Map(); + pendingStatus ??= new Map(get(statusByServer)); + for (const entry of data.entries) pendingStatus.set(entry.server, { ...entry, receivedAt }); + schedule(); }); 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; + }; } diff --git a/tests/unit/ui-server-clock.test.ts b/tests/unit/ui-server-clock.test.ts index 04f89bb..923a1cc 100644 --- a/tests/unit/ui-server-clock.test.ts +++ b/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; + +function flushFrame() { + const callbacks = [...frames.values()]; + frames.clear(); + for (const callback of callbacks) callback(performance.now()); +} function server(timestamp = epoch, changes: Partial = {}): FleetServer { return { @@ -35,12 +42,20 @@ function status(timestamp: number, changes: Partial = {}): 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', { - data: JSON.stringify({ serverNow: time, entries: [{ server: 't5_zombie_theater_1', event, - receivedAt: new Date(time).toISOString(), players: [] }] }) - })); + 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);