Reuse serialized server events across SSE connections

src/routes/api/events/servers/+server.ts

@@ -1,11 +1,14 @@
 import { observationReader } from '$lib/server/fleet/status.js';
 import { listServers } from '$lib/server/fleet/servers.js';
 import { statusEntry } from '$lib/server/pipeline/worker.js';
-import { subscribe } from '$lib/server/pipeline/bus.js';
+import { subscribe, type PipelineEvent } from '$lib/server/pipeline/bus.js';
 import { listTelemetry } from '$lib/server/fleet/telemetry.js';
 
 const encoder = new TextEncoder();
 const HEARTBEAT_MS = 20_000;
+// Bus subscribers receive the same event object. Reuse its complete frame
+// across connections without retaining events after delivery.
+const frames = new WeakMap<PipelineEvent, Uint8Array>();
 
 export async function GET() {
     let heartbeat: NodeJS.Timeout | null = null;
@@ -15,7 +18,12 @@ 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, serverNow: Date.now() })}\n\n`));
+                    let frame = frames.get(event);
+                    if (!frame) {
+                        frame = encoder.encode(`event: ${event.type}\ndata: ${JSON.stringify({ ...event, serverNow: Date.now() })}\n\n`);
+                        frames.set(event, frame);
+                    }
+                    controller.enqueue(frame);
                 }
             });
             const observation = observationReader();

tests/unit/telemetry.test.ts

@@ -90,6 +90,45 @@ it('timestamps initial telemetry and subsequent status and telemetry events with
     }
 });
 
+it('serializes each live server event once across connections and releases cancelled subscribers', async () => {
+    const { GET } = await import('../../src/routes/api/events/servers/+server.js');
+    const { emit } = await import('../../src/lib/server/pipeline/bus.js');
+    const readers = await Promise.all([GET(), GET()].map(async response => (await response).body!.getReader()));
+    const stringify = vi.spyOn(JSON, 'stringify');
+    try {
+        for (const reader of readers) {
+            await reader.read(); // Initial comment and per-connection snapshots.
+            await reader.read();
+            await reader.read();
+        }
+        for (const event of [
+            { type: 'server-status' as const, delta: true, entries: [] },
+            { type: 'server-telemetry' as const, entries: [{ ...payload, receivedAt: new Date().toISOString() }] }
+        ]) {
+            stringify.mockClear();
+            emit(event);
+            expect(stringify).toHaveBeenCalledTimes(1);
+            const [first, second] = await Promise.all(readers.map(reader => reader.read()));
+            expect(first.value).toBe(second.value);
+            const frame = new TextDecoder().decode(first.value);
+            expect(frame.startsWith(`event: ${event.type}\ndata: `)).toBe(true);
+            expect(JSON.parse(frame.split('\n')[1].slice(6))).toEqual({ ...event, serverNow: expect.any(Number) });
+        }
+        await readers[0].cancel();
+        stringify.mockClear();
+        emit({ type: 'server-status', entries: [] });
+        expect(stringify).toHaveBeenCalledTimes(1);
+        expect((await readers[1].read()).done).toBe(false);
+        await readers[1].cancel();
+        stringify.mockClear();
+        emit({ type: 'server-telemetry', entries: [] });
+        expect(stringify).not.toHaveBeenCalled();
+    } finally {
+        stringify.mockRestore();
+        await Promise.all(readers.map(reader => reader.cancel()));
+    }
+});
+
 it.each(['round_change', 'heartbeat'] as const)('persists %s stats, broadcasts updates, and leaves RCON safety data untouched', async (event) => {
     const payload = gameEvent(event, 17);
     const { subscribe } = await import('../../src/lib/server/pipeline/bus.js');