Coalesce fleet refreshes and suppress unchanged sweep events

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;
 };

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<SweepResult> | null = null;
@@ -39,7 +40,9 @@ async function runSweep(): Promise<SweepResult> {
             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());

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<typeof setTimeout> | 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;
+        if (document.visibilityState === 'hidden') return;
-        refreshTimer = setTimeout(() => { refreshTimer = undefined; void refreshFleet(true); }, 250);
+        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;

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<void>(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<void>(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<void>(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);
+    });
+});

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');