Fix false stale Fleet reports caused by clock skew
This commit is contained in:
@@ -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}
|
||||
<span class="err">{lastError}</span>
|
||||
{:else if lastSeenAt}
|
||||
<span>Last seen {ago(lastSeenAt)}</span>
|
||||
<span>Last seen {ago(lastSeenAt, $uiNow)}</span>
|
||||
{:else if lastRestartedAt}
|
||||
<span>Restart {ago(lastRestartedAt)}</span>
|
||||
<span>Restart {ago(lastRestartedAt, $uiNow)}</span>
|
||||
{:else}
|
||||
<span class="dim">No data</span>
|
||||
{/if}
|
||||
|
||||
@@ -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<void> {
|
||||
]);
|
||||
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 });
|
||||
|
||||
@@ -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<typeof setInterval> | 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<string, GameTelemetry>() : 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<string, ServerStatus>();
|
||||
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; };
|
||||
}
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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'}
|
||||
<div class="metric-cell">
|
||||
<Stat label="Last seen" value={ago(effectiveSeen)} size="md" tween={false} />
|
||||
<Stat label="Last seen" value={ago(effectiveSeen, $uiNow)} size="md" tween={false} />
|
||||
</div>
|
||||
{/if}
|
||||
</div>
|
||||
@@ -517,7 +517,7 @@
|
||||
{/if}
|
||||
</span>
|
||||
<span class="hist-files">{mine.files_changed} file{mine.files_changed === 1 ? '' : 's'}</span>
|
||||
<span class="hist-when">{ago(d.started_at)}</span>
|
||||
<span class="hist-when">{ago(d.started_at, $uiNow)}</span>
|
||||
{#if mine.error}<span class="hist-err">{mine.error}</span>{/if}
|
||||
</a>
|
||||
{/each}
|
||||
|
||||
@@ -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');
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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> = {}): 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> = {}): 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);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user