diff --git a/src/hooks.server.ts b/src/hooks.server.ts index 7d7deaf..4717552 100644 --- a/src/hooks.server.ts +++ b/src/hooks.server.ts @@ -3,7 +3,7 @@ import { migrate } from '$lib/server/db/migrate.js'; import { startWorker } from '$lib/server/pipeline/worker.js'; import { interruptDeployments } from '$lib/server/provisioning/history.js'; import { authenticationHandle } from '$lib/server/auth/guard.js'; -import { emit } from '$lib/server/pipeline/bus.js'; +import { invalidateFleet } from '$lib/server/pipeline/bus.js'; import { startPerformanceMonitor } from '$lib/server/performance.js'; import { withOperationSource } from '$lib/server/hosts/operation-context.js'; @@ -30,6 +30,6 @@ export const handle: Handle = async (input) => { if (process.env.EC_PERF_LOG === 'true' && elapsed > 250 && !input.event.route.id?.startsWith('/api/agent/')) { console.info('[request]', { route: input.event.route.id, method: input.event.request.method, status: response.status, elapsedMs: Math.round(elapsed) }); } - if (input.event.locals.user && !['GET','HEAD','OPTIONS'].includes(input.event.request.method) && response.status < 400) emit({ type: 'fleet-invalidated' }); + if (input.event.locals.user && !['GET','HEAD','OPTIONS'].includes(input.event.request.method) && response.status < 400) invalidateFleet(); return response; }; diff --git a/src/lib/server/pipeline/sweep.ts b/src/lib/server/pipeline/sweep.ts index 525f7b6..a84a3af 100644 --- a/src/lib/server/pipeline/sweep.ts +++ b/src/lib/server/pipeline/sweep.ts @@ -7,6 +7,7 @@ import { pruneHostObservations } from '../hosts/index.js'; interface PendingRow { deployment_id: string; server: string; restart_state: string; restarted_at: string | null; files_changed: number; + players: number | null; error: string | null; } type SweepResult = { checked: number; restarted: number; healthy: number; held: number; failed: number }; let sweepInFlight: Promise | null = null; @@ -39,7 +40,9 @@ async function runSweep(): Promise { attempts = attempts + CASE WHEN ? IS NOT NULL AND checked_at IS NOT ? THEN 1 ELSE 0 END, restarted_at = COALESCE(restarted_at, ?) WHERE deployment_id = ? AND server = ?`) .run(state, operation.checked_at, operation.players, error, operation.checked_at, operation.checked_at, restartedAt, row.deployment_id, row.server); - emit({ type: 'server-state', deploymentId: row.deployment_id, server: row.server, state, players: operation.players, error }); + if (state !== row.restart_state || operation.players !== row.players || error !== row.error) { + emit({ type: 'server-state', deploymentId: row.deployment_id, server: row.server, state, players: operation.players, error }); + } } db().prepare(`UPDATE deployments SET finished_at = ? WHERE finished_at IS NULL AND NOT EXISTS ( SELECT 1 FROM deployment_servers WHERE deployment_id = deployments.id AND server_deleted_at IS NULL AND restart_state IN ('pending','failed','held'))`).run(nowIso()); diff --git a/src/lib/ui/stores/fleet.ts b/src/lib/ui/stores/fleet.ts index f68832e..baae9df 100644 --- a/src/lib/ui/stores/fleet.ts +++ b/src/lib/ui/stores/fleet.ts @@ -124,12 +124,37 @@ export function startFleetMonitor(): () => void { if (!browser || stop) return () => {}; const stopPolling = pollVisible(() => refreshFleet(), 60_000, !get(fleet).receivedAt || Date.now() - get(fleet).receivedAt > 30_000); let refreshTimer: ReturnType | undefined; + let stopped = false; + let pending = false; + let refreshing = false; + let lastEventRefresh = -Infinity; + const schedule = () => { + if (stopped || refreshing || refreshTimer !== undefined || !pending || document.visibilityState === 'hidden') return; + refreshTimer = setTimeout(async () => { + refreshTimer = undefined; + refreshing = true; + try { + // Wait for any manual/poll refresh, then fetch evidence newer than + // the event. Do not use refreshFleet(true)'s immediate follow-up. + if (inFlight) await inFlight; + if (stopped || document.visibilityState === 'hidden') return; + pending = false; + lastEventRefresh = Date.now(); + await refreshFleet(); + } finally { + refreshing = false; + schedule(); + } + }, Math.max(250, lastEventRefresh + 2000 - Date.now())); + }; const changed = () => { - if (refreshTimer || document.visibilityState === 'hidden') return; - refreshTimer = setTimeout(() => { refreshTimer = undefined; void refreshFleet(true); }, 250); + if (document.visibilityState === 'hidden') return; + pending = true; + schedule(); }; window.addEventListener('ec:fleet-changed', changed); stop = () => { + stopped = true; stopPolling(); clearTimeout(refreshTimer); window.removeEventListener('ec:fleet-changed', changed); stop = null; }; return stop; diff --git a/tests/unit/fleet-monitor.test.ts b/tests/unit/fleet-monitor.test.ts new file mode 100644 index 0000000..6226ef2 --- /dev/null +++ b/tests/unit/fleet-monitor.test.ts @@ -0,0 +1,118 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +vi.mock('$app/environment', () => ({ browser: true })); + +let documentStub: EventTarget & { visibilityState: string }; +let stop: (() => void) | undefined; +const changed = () => window.dispatchEvent(new Event('ec:fleet-changed')); +const visibility = (state: string) => { + documentStub.visibilityState = state; + documentStub.dispatchEvent(new Event('visibilitychange')); +}; +const response = (url: string) => Response.json(url === '/api/servers' + ? { serverNow: Date.now(), servers: [], integrations: {} } : { operations: [] }); + +beforeEach(() => { + vi.resetModules(); + vi.useFakeTimers(); + vi.setSystemTime(new Date('2026-10-05T12:00:00Z')); + documentStub = Object.assign(new EventTarget(), { visibilityState: 'visible' }); + vi.stubGlobal('document', documentStub); + vi.stubGlobal('window', new EventTarget()); + vi.stubGlobal('fetch', vi.fn(async (url: string) => response(url))); +}); +afterEach(() => { + stop?.(); stop = undefined; + vi.useRealTimers(); vi.unstubAllGlobals(); vi.restoreAllMocks(); +}); +async function setup() { + const store = await import('../../src/lib/ui/stores/fleet.js'); + store.fleet.set({ servers: [], operations: [], receivedAt: Date.now() }); + stop = store.startFleetMonitor(); + return store; +} + +describe('fleet event refreshes', () => { + it('coalesces bursts and refreshes at most once every two seconds with a trailing refresh', async () => { + await setup(); + changed(); changed(); changed(); + await vi.advanceTimersByTimeAsync(249); + expect(fetch).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + expect(fetch).toHaveBeenCalledTimes(2); + for (let i = 0; i < 7; i++) { + changed(); + await vi.advanceTimersByTimeAsync(250); + expect(fetch).toHaveBeenCalledTimes(2); + } + await vi.advanceTimersByTimeAsync(250); + expect(fetch).toHaveBeenCalledTimes(4); + changed(); + await vi.advanceTimersByTimeAsync(2000); + expect(fetch).toHaveBeenCalledTimes(6); + await vi.advanceTimersByTimeAsync(5000); + expect(fetch).toHaveBeenCalledTimes(6); + }); + + it('keeps events during a request without an immediate follow-up when it finishes', async () => { + await setup(); + let finish!: () => void; + const gate = new Promise(resolve => { finish = resolve; }); + vi.mocked(fetch).mockImplementation(async url => { await gate; return response(String(url)); }); + changed(); + await vi.advanceTimersByTimeAsync(250); + expect(fetch).toHaveBeenCalledTimes(2); + changed(); changed(); + await vi.advanceTimersByTimeAsync(500); + finish(); + await vi.advanceTimersByTimeAsync(1499); + expect(fetch).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(1); + expect(fetch).toHaveBeenCalledTimes(4); + }); + + it('waits for a slow manual refresh and coalesces events arriving during it', async () => { + const { refreshFleet } = await setup(); + let finish!: () => void; + const gate = new Promise(resolve => { finish = resolve; }); + vi.mocked(fetch).mockImplementation(async url => { await gate; return response(String(url)); }); + const manual = refreshFleet(); + changed(); + await vi.advanceTimersByTimeAsync(3000); + changed(); changed(); + expect(fetch).toHaveBeenCalledTimes(2); + finish(); await manual; + await vi.advanceTimersByTimeAsync(0); + expect(fetch).toHaveBeenCalledTimes(4); + await vi.advanceTimersByTimeAsync(3000); + expect(fetch).toHaveBeenCalledTimes(4); + }); + + it('skips queued refreshes while hidden, refreshes on return, and cancels on stop', async () => { + await setup(); + changed(); visibility('hidden'); + await vi.advanceTimersByTimeAsync(250); + changed(); + await vi.advanceTimersByTimeAsync(3000); + expect(fetch).not.toHaveBeenCalled(); + visibility('visible'); + await vi.advanceTimersByTimeAsync(0); + expect(fetch).toHaveBeenCalledTimes(2); + changed(); stop!(); + await vi.advanceTimersByTimeAsync(3000); + expect(fetch).toHaveBeenCalledTimes(2); + }); + + it('cancels a follow-up waiting for an active request when stopped', async () => { + const { refreshFleet } = await setup(); + let finish!: () => void; + const gate = new Promise(resolve => { finish = resolve; }); + vi.mocked(fetch).mockImplementation(async url => { await gate; return response(String(url)); }); + const manual = refreshFleet(); + changed(); + await vi.advanceTimersByTimeAsync(250); + stop!(); finish(); await manual; + await vi.advanceTimersByTimeAsync(3000); + expect(fetch).toHaveBeenCalledTimes(2); + }); +}); diff --git a/tests/unit/sweep.test.ts b/tests/unit/sweep.test.ts index 0ebd6bc..d52c4c8 100644 --- a/tests/unit/sweep.test.ts +++ b/tests/unit/sweep.test.ts @@ -16,6 +16,53 @@ afterEach(async () => { }); describe('sweep', () => { + it('emits server state only when state, players, or error changes', async () => { + const { sweep } = await import('../../src/lib/server/pipeline/sweep.js'); + const { seedServerRegistry } = await import('../../src/lib/server/fleet/servers.js'); + const { db } = await import('../../src/lib/server/db/index.js'); + const { subscribe } = await import('../../src/lib/server/pipeline/bus.js'); + await seedServerRegistry(); + db().prepare("INSERT INTO deployments (id, action, started_at, triggered_by) VALUES ('changes', 'test', ?, 'ui')") + .run(new Date().toISOString()); + db().prepare(`INSERT INTO deployment_servers (deployment_id, server, files_changed, restart_state) + VALUES ('changes', 't5_zombie_theater_1', 1, 'pending')`).run(); + const events = vi.fn(); + const unsubscribe = subscribe(event => { if (event.type === 'server-state') events(event); }); + try { + await sweep(); + expect(events).toHaveBeenCalledTimes(1); + await sweep(); + expect(events).toHaveBeenCalledTimes(1); + + // Fresh check timestamps still update the campaign, without an SSE event. + const checkedAt = new Date().toISOString(); + db().prepare('UPDATE server_deployments SET checked_at = ?').run(checkedAt); + await sweep(); + expect(events).toHaveBeenCalledTimes(1); + expect(db().prepare("SELECT checked_at, attempts FROM deployment_servers WHERE deployment_id = 'changes'").get()) + .toEqual({ checked_at: checkedAt, attempts: 1 }); + + db().prepare('UPDATE server_deployments SET players = 2').run(); + await sweep(); + expect(events).toHaveBeenCalledTimes(2); + expect(events).toHaveBeenLastCalledWith(expect.objectContaining({ state: 'pending', players: 2 })); + db().prepare("UPDATE server_deployments SET error = 'Waiting for players'").run(); + await sweep(); + expect(events).toHaveBeenCalledTimes(3); + expect(events).toHaveBeenLastCalledWith(expect.objectContaining({ error: 'Waiting for players' })); + db().prepare("UPDATE server_deployments SET state = 'failed'").run(); + await sweep(); + expect(events).toHaveBeenCalledTimes(4); + expect(events).toHaveBeenLastCalledWith(expect.objectContaining({ state: 'failed' })); + await sweep(); + expect(events).toHaveBeenCalledTimes(4); + db().prepare("UPDATE server_deployments SET state = 'done'").run(); + await sweep(); + expect(events).toHaveBeenCalledTimes(5); + expect(events).toHaveBeenLastCalledWith(expect.objectContaining({ state: 'done', error: null })); + } finally { unsubscribe(); } + }); + it('queues pending rows in Provisioning even when integrations are unset', async () => { const { sweep } = await import('../../src/lib/server/pipeline/sweep.js'); const { seedServerRegistry } = await import('../../src/lib/server/fleet/servers.js');