From 1d073779fe070606a45fde8898913b528b6a50c0 Mon Sep 17 00:00:00 2001 From: josh Date: Sat, 19 Sep 2026 20:17:43 -0400 Subject: [PATCH] Fix false stale Fleet reports caused by clock skew --- src/lib/ui/components/ServerCard.svelte | 8 +- src/lib/ui/stores/fleet.ts | 5 +- src/lib/ui/stores/servers.ts | 28 ++++- src/routes/api/events/servers/+server.ts | 4 +- src/routes/api/servers/+server.ts | 1 + src/routes/servers/[name]/+page.svelte | 8 +- tests/unit/server-status.test.ts | 9 ++ tests/unit/telemetry.test.ts | 25 ++++ tests/unit/ui-server-clock.test.ts | 151 +++++++++++++++++++++++ 9 files changed, 224 insertions(+), 15 deletions(-) create mode 100644 tests/unit/ui-server-clock.test.ts diff --git a/src/lib/ui/components/ServerCard.svelte b/src/lib/ui/components/ServerCard.svelte index 8210d02..b14d5e5 100644 --- a/src/lib/ui/components/ServerCard.svelte +++ b/src/lib/ui/components/ServerCard.svelte @@ -54,9 +54,9 @@ mapLabelOverride }: Props = $props(); - function ago(iso: string | null): string { + function ago(iso: string | null, now: number): string { if (!iso) return 'never'; - const d = Date.now() - new Date(iso).getTime(); + const d = now - new Date(iso).getTime(); if (d < 60_000) return `${Math.max(1, Math.floor(d / 1000))}s ago`; if (d < 3_600_000) return `${Math.floor(d / 60_000)}m ago`; if (d < 86_400_000) return `${Math.floor(d / 3_600_000)}h ago`; @@ -175,9 +175,9 @@ {#if health === 'offline' && lastError} {lastError} {:else if lastSeenAt} - Last seen {ago(lastSeenAt)} + Last seen {ago(lastSeenAt, $uiNow)} {:else if lastRestartedAt} - Restart {ago(lastRestartedAt)} + Restart {ago(lastRestartedAt, $uiNow)} {:else} No data {/if} diff --git a/src/lib/ui/stores/fleet.ts b/src/lib/ui/stores/fleet.ts index ab86092..5d6f425 100644 --- a/src/lib/ui/stores/fleet.ts +++ b/src/lib/ui/stores/fleet.ts @@ -1,6 +1,6 @@ import { derived, get, writable } from 'svelte/store'; import { browser } from '$app/environment'; -import { statusByServer, statusReceivedAt } from './servers.js'; +import { statusByServer, statusReceivedAt, serverTimeNow, syncServerTime } from './servers.js'; import { watchSatisfied, type FleetServer, type FleetOperation, type LobbyWatch } from '$lib/fleet-overview.js'; import { serverTitle } from '../server-labels.js'; import { toast } from '../toast.js'; @@ -39,6 +39,7 @@ export function refreshFleet(): Promise { ]); if (responses.some(response => !response.ok)) throw new Error('Could not refresh Fleet. Retrying automatically.'); const [servers, operations] = await Promise.all(responses.map(response => response.json())); + syncServerTime(servers.serverNow); fleetError.set(''); fleetIntegrations.set(servers.integrations); fleet.set({ servers: servers.servers, operations: operations.operations, receivedAt: requestedAt }); @@ -93,7 +94,7 @@ export function startFleetMonitor(): () => void { if (!snapshot.receivedAt || Date.now() - snapshot.receivedAt > 30_000) return; const servers = get(fleetServers); for (const watch of get(lobbyWatches)) { - if (!watchSatisfied(watch, servers.find(server => server.name === watch.server), snapshot.operations, Date.now())) continue; + if (!watchSatisfied(watch, servers.find(server => server.name === watch.server), snapshot.operations, serverTimeNow())) continue; const message = `${serverTitle(watch.server)} ${watch.kind === 'empty' ? 'is empty' : watch.kind === 'online' ? 'is back online' : 'has finished its operation'}.`; cancelWatch(watch.id); toast(message, { variant: 'success', ttl: 0, href: watch.href }); diff --git a/src/lib/ui/stores/servers.ts b/src/lib/ui/stores/servers.ts index 8143e01..6525a9c 100644 --- a/src/lib/ui/stores/servers.ts +++ b/src/lib/ui/stores/servers.ts @@ -46,15 +46,36 @@ export const statusReceivedAt = writable(0); // startServerStream() call in the browser. const _uiNow = writable(Date.now()); export const uiNow = readonly(_uiNow); +let serverClock: { time: number; receivedAt: number } | null = null; + +// Report timestamps belong to the backend clock, which can differ from the +// browser's wall clock. Advance that clock locally even if updates stop. +export function serverTimeNow(): number { + return serverClock ? serverClock.time + performance.now() - serverClock.receivedAt : Date.now(); +} + +export function syncServerTime(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() }; + } + // Refresh immediately: a suspended tab's interval may be well behind. + _uiNow.set(serverTimeNow()); +} + let tickInterval: ReturnType | null = null; 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 data = JSON.parse((evt as MessageEvent).data) as { entries: GameTelemetry[]; snapshot?: boolean }; + const data = JSON.parse((evt as MessageEvent).data) as { entries: GameTelemetry[]; snapshot?: boolean; serverNow?: number }; + syncServerTime(data.serverNow); telemetryByServer.update((previous) => { const next = data.snapshot ? new Map() : new Map(previous); for (const entry of data.entries) next.set(entry.server, entry); @@ -62,7 +83,8 @@ export function startServerStream(): (() => void) | undefined { }); }); source.addEventListener('server-status', (evt) => { - const data = JSON.parse((evt as MessageEvent).data) as { entries: ServerStatus[] }; + const data = JSON.parse((evt as MessageEvent).data) as { entries: ServerStatus[]; serverNow?: number }; + syncServerTime(data.serverNow); statusByServer.update(() => { const m = new Map(); for (const e of data.entries) m.set(e.server, e); @@ -71,7 +93,7 @@ export function startServerStream(): (() => void) | undefined { statusReceivedAt.set(Date.now()); }); if (tickInterval === null) { - tickInterval = setInterval(() => _uiNow.set(Date.now()), 1000); + tickInterval = setInterval(() => _uiNow.set(serverTimeNow()), 1000); } return () => { source?.close(); source = null; if (tickInterval) clearInterval(tickInterval); tickInterval = null; }; } diff --git a/src/routes/api/events/servers/+server.ts b/src/routes/api/events/servers/+server.ts index ef176ab..0da1282 100644 --- a/src/routes/api/events/servers/+server.ts +++ b/src/routes/api/events/servers/+server.ts @@ -12,11 +12,11 @@ export async function GET() { start(controller) { unsubscribe = subscribe((event) => { if (event.type === 'server-status' || event.type === 'server-telemetry') { - controller.enqueue(encoder.encode(`event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`)); + controller.enqueue(encoder.encode(`event: ${event.type}\ndata: ${JSON.stringify({ ...event, serverNow: Date.now() })}\n\n`)); } }); controller.enqueue(encoder.encode(': hello\n\n')); - controller.enqueue(encoder.encode(`event: server-telemetry\ndata: ${JSON.stringify({ entries: listTelemetry(), snapshot: true })}\n\n`)); + controller.enqueue(encoder.encode(`event: server-telemetry\ndata: ${JSON.stringify({ entries: listTelemetry(), snapshot: true, serverNow: Date.now() })}\n\n`)); heartbeat = setInterval(() => controller.enqueue(encoder.encode(': ping\n\n')), HEARTBEAT_MS); }, cancel() { diff --git a/src/routes/api/servers/+server.ts b/src/routes/api/servers/+server.ts index f747f60..d89d322 100644 --- a/src/routes/api/servers/+server.ts +++ b/src/routes/api/servers/+server.ts @@ -40,6 +40,7 @@ export async function GET() { }); return json({ servers, + serverNow: Date.now(), reservedNames: (db().prepare("SELECT j.server FROM server_creation_jobs j JOIN server_deployments d ON d.id = j.deployment_id WHERE d.state IN ('queued', 'running', 'awaiting-health')").all() as { server: string }[]).map(row => row.server), integrations: { rcon: rconConfigured(), diff --git a/src/routes/servers/[name]/+page.svelte b/src/routes/servers/[name]/+page.svelte index 9e7e5ca..0b828e3 100644 --- a/src/routes/servers/[name]/+page.svelte +++ b/src/routes/servers/[name]/+page.svelte @@ -154,9 +154,9 @@ let hasDiagnosticIssue = $derived(effectiveHealth !== 'online' && effectiveHealth !== 'starting'); - function ago(iso: string | null): string { + function ago(iso: string | null, now: number): string { if (!iso) return 'never'; - const d = Date.now() - new Date(iso).getTime(); + const d = now - new Date(iso).getTime(); if (d < 60_000) return `${Math.max(1, Math.floor(d / 1000))}s ago`; if (d < 3_600_000) return `${Math.floor(d / 60_000)}m ago`; if (d < 86_400_000) return `${Math.floor(d / 3_600_000)}h ago`; @@ -270,7 +270,7 @@ {/if} {#if effectiveHealth !== 'online'}
- +
{/if} @@ -517,7 +517,7 @@ {/if} {mine.files_changed} file{mine.files_changed === 1 ? '' : 's'} - {ago(d.started_at)} + {ago(d.started_at, $uiNow)} {#if mine.error}{mine.error}{/if} {/each} diff --git a/tests/unit/server-status.test.ts b/tests/unit/server-status.test.ts index f920132..693c628 100644 --- a/tests/unit/server-status.test.ts +++ b/tests/unit/server-status.test.ts @@ -27,6 +27,15 @@ async function setup() { } describe('database fleet and container readiness', () => { + it('includes the server clock with the health snapshot', async () => { + await setup(); + const { GET } = await import('../../src/routes/api/servers/+server.js'); + const requestedAt = Date.now(); + const snapshot = await (await GET()).json(); + expect(snapshot.serverNow).toBeGreaterThanOrEqual(requestedAt); + expect(snapshot.serverNow).toBeLessThanOrEqual(Date.now()); + }); + it('lists registered servers without folders and ignores unregistered folders and tombstones', async () => { const { db } = await setup(); const { allServers, listServers } = await import('../../src/lib/server/fleet/servers.js'); diff --git a/tests/unit/telemetry.test.ts b/tests/unit/telemetry.test.ts index 179bd30..3ac1add 100644 --- a/tests/unit/telemetry.test.ts +++ b/tests/unit/telemetry.test.ts @@ -13,6 +13,31 @@ afterEach(async () => fixture.cleanup()); const payload = gameEvent('round_change', 17); +it('timestamps initial telemetry and subsequent status and telemetry events with the server clock', async () => { + const { GET } = await import('../../src/routes/api/events/servers/+server.js'); + const { emit } = await import('../../src/lib/server/pipeline/bus.js'); + const startedAt = Date.parse('2026-09-19T12:00:00Z'); + const clock = vi.spyOn(Date, 'now').mockReturnValue(startedAt); + const reader = (await GET()).body!.getReader(); + const nextEvent = async () => { + const frame = new TextDecoder().decode((await reader.read()).value); + return JSON.parse(frame.split('\n').find(line => line.startsWith('data: '))!.slice(6)); + }; + try { + await reader.read(); // Initial SSE comment. + expect(await nextEvent()).toMatchObject({ snapshot: true, entries: [], serverNow: startedAt }); + clock.mockReturnValue(startedAt + 30_000); + emit({ type: 'server-status', entries: [] }); + expect(await nextEvent()).toEqual({ type: 'server-status', entries: [], serverNow: startedAt + 30_000 }); + clock.mockReturnValue(startedAt + 31_000); + emit({ type: 'server-telemetry', entries: [] }); + expect(await nextEvent()).toEqual({ type: 'server-telemetry', entries: [], serverNow: startedAt + 31_000 }); + } finally { + await reader.cancel(); + clock.mockRestore(); + } +}); + it('persists full player stats, broadcasts updates, and leaves RCON safety data untouched', async () => { const { subscribe } = await import('../../src/lib/server/pipeline/bus.js'); const listener = vi.fn(); diff --git a/tests/unit/ui-server-clock.test.ts b/tests/unit/ui-server-clock.test.ts new file mode 100644 index 0000000..513c7a8 --- /dev/null +++ b/tests/unit/ui-server-clock.test.ts @@ -0,0 +1,151 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { get } from 'svelte/store'; +import { freshServer, matchesFleet, type FleetServer } from '../../src/lib/fleet-overview.js'; +import type { ServerStatus } from '../../src/lib/ui/stores/servers.js'; + +vi.mock('$app/environment', () => ({ browser: true })); + +const epoch = Date.parse('2026-09-19T12:00:00Z'); +let source: EventTarget; +let stop: (() => void) | undefined; +let snapshot: { serverNow: number; servers: FleetServer[]; integrations: object }; + +function server(timestamp = epoch, changes: Partial = {}): FleetServer { + return { + name: 't5_zombie_theater_1', game: 't5', map: 'zombie_theater', + last_players: 0, last_map: 'zombie_theater', last_gametype: 'zom', + last_seen_at: new Date(timestamp).toISOString(), last_error: null, + roster: [], health: 'online', held: [], active_since: null, + inactive_since: new Date(epoch).toISOString(), game_started_at: null, + game_ended_at: null, ...changes + }; +} + +function status(timestamp: number, changes: Partial = {}): ServerStatus { + return { + server: 't5_zombie_theater_1', players: 0, map: 'zombie_theater', gametype: 'zom', + roster: [], health: 'online', lastSeenAt: new Date(timestamp).toISOString(), + lastError: null, endpoint: null, reason: null, replySnippet: null, + activeSince: null, inactiveSince: new Date(epoch).toISOString(), + gameStartedAt: null, gameEndedAt: null, ...changes + }; +} + +function emit(serverNow: number, entries: ServerStatus[]) { + source.dispatchEvent(new MessageEvent('server-status', { data: JSON.stringify({ serverNow, entries }) })); +} + +beforeEach(() => { + vi.resetModules(); + vi.useFakeTimers({ toFake: ['Date', 'performance', 'setTimeout', 'clearTimeout', 'setInterval', 'clearInterval'] }); + vi.setSystemTime(epoch); + snapshot = { serverNow: epoch, servers: [server()], integrations: {} }; + vi.stubGlobal('fetch', vi.fn(async (url: string) => new Response(JSON.stringify( + url === '/api/servers' ? snapshot : { operations: [] } + )))); + vi.stubGlobal('EventSource', class extends EventTarget { + constructor() { super(); source = this; } + close() {} + }); +}); + +afterEach(() => { + stop?.(); stop = undefined; + vi.useRealTimers(); + vi.unstubAllGlobals(); + vi.restoreAllMocks(); +}); + +async function setup() { + const clock = await import('../../src/lib/ui/stores/servers.js'); + const fleet = await import('../../src/lib/ui/stores/fleet.js'); + stop = clock.startServerStream(); + await fleet.refreshFleet(); + return { ...clock, ...fleet }; +} + +describe('Fleet report clock', () => { + it.each([-2500, 60_000])('keeps fresh HTTP and streamed reports healthy with a browser clock offset of %i ms', async (offset) => { + vi.setSystemTime(epoch + offset); + const { fleetServers, uiNow, refreshFleet } = await setup(); + const expectHealthy = () => { + const current = get(fleetServers)[0]; + expect(freshServer(current, get(uiNow))).toBe(true); + expect(matchesFleet(current, '', 'empty', '', [], [], get(uiNow))).toBe(true); + expect(matchesFleet(current, '', 'attention', '', [], [], get(uiNow))).toBe(false); + }; + expect(get(uiNow)).toBe(epoch); + expectHealthy(); + for (let cycle = 1; cycle <= 3; cycle++) { + for (let second = 0; second < 30; second++) { + await vi.advanceTimersByTimeAsync(1000); + expectHealthy(); + } + const now = epoch + cycle * 30_000; + emit(now, [status(now)]); + expectHealthy(); + snapshot = { ...snapshot, serverNow: now, servers: [server(now)] }; + await refreshFleet(); + expectHealthy(); + } + }); + + it('updates the clock before exposing a new status after a suspended UI ticker', async () => { + const { fleetServers, uiNow } = await setup(); + vi.spyOn(performance, 'now').mockReturnValue(30_000); + vi.setSystemTime(epoch + 30_000); + expect(get(uiNow)).toBe(epoch); + emit(epoch + 30_000, [status(epoch + 30_000)]); + expect(get(uiNow)).toBe(epoch + 30_000); + expect(freshServer(get(fleetServers)[0], get(uiNow))).toBe(true); + }); + + it('does not rewind the clock when an older HTTP snapshot follows a newer stream update', async () => { + const { fleetServers, uiNow, refreshFleet } = await setup(); + let resolveSnapshot!: (response: Response) => void; + vi.mocked(fetch).mockImplementationOnce(() => new Promise(resolve => { resolveSnapshot = resolve; })); + const refreshing = refreshFleet(); + await vi.advanceTimersByTimeAsync(4000); + emit(epoch + 4000, [status(epoch + 4000)]); + resolveSnapshot(new Response(JSON.stringify(snapshot))); + await refreshing; + expect(get(uiNow)).toBe(epoch + 4000); + expect(get(fleetServers)[0].last_seen_at).toBe(new Date(epoch + 4000).toISOString()); + expect(freshServer(get(fleetServers)[0], get(uiNow))).toBe(true); + }); + + it('keeps advancing through wall-clock changes and expires a report if updates stop', async () => { + const { fleetServers, uiNow } = await setup(); + vi.setSystemTime(epoch - 120_000); + await vi.advanceTimersByTimeAsync(45_000); + expect(get(uiNow)).toBe(epoch + 45_000); + expect(freshServer(get(fleetServers)[0], get(uiNow))).toBe(true); + await vi.advanceTimersByTimeAsync(1000); + expect(get(uiNow)).toBe(epoch + 46_000); + expect(freshServer(get(fleetServers)[0], get(uiNow))).toBe(false); + expect(matchesFleet(get(fleetServers)[0], '', 'attention', '', [], [], get(uiNow))).toBe(true); + }); + + it('preserves actual RCON failures and stopped-container health', async () => { + const { fleetServers, uiNow } = await setup(); + await vi.advanceTimersByTimeAsync(1000); + emit(epoch + 1000, [status(epoch, { health: 'stale', lastError: 'timeout:no reply', reason: 'timeout' })]); + expect(freshServer(get(fleetServers)[0], get(uiNow))).toBe(false); + expect(matchesFleet(get(fleetServers)[0], '', 'attention', '', [], [], get(uiNow))).toBe(true); + emit(epoch + 1000, [status(epoch, { health: 'offline' })]); + expect(get(fleetServers)[0].health).toBe('offline'); + expect(freshServer(get(fleetServers)[0], get(uiNow))).toBe(false); + }); + + it('anchors the initial telemetry snapshot even before Fleet finishes loading', async () => { + vi.setSystemTime(epoch - 60_000); + const { startServerStream, uiNow } = await import('../../src/lib/ui/stores/servers.js'); + stop = startServerStream(); + source.dispatchEvent(new MessageEvent('server-telemetry', { + data: JSON.stringify({ serverNow: epoch, entries: [], snapshot: true }) + })); + expect(get(uiNow)).toBe(epoch); + await vi.advanceTimersByTimeAsync(1000); + expect(get(uiNow)).toBe(epoch + 1000); + }); +});