Bound live event stream queues (Refs #71)

src/lib/server/pipeline/event-stream.ts

@@ -0,0 +1,60 @@
+import { subscribe, type PipelineEvent } from './bus.js';
+
+const encoder = new TextEncoder();
+const hello = encoder.encode(': hello\n\n');
+const ping = encoder.encode(': ping\n\n');
+const HEARTBEAT_MS = 20_000;
+// Allow snapshots and short bursts, but never retain an unbounded event history.
+const MAX_QUEUED_FRAMES = 64;
+
+interface EventStreamOptions {
+    accepts: (event: PipelineEvent) => boolean;
+    encode: (event: PipelineEvent) => Uint8Array;
+    initialFrames?: Uint8Array[];
+}
+
+export function eventStreamResponse({ accepts, encode, initialFrames = [] }: EventStreamOptions): Response {
+    let heartbeat: ReturnType<typeof setInterval> | undefined;
+    let unsubscribe: (() => void) | undefined;
+    let closed = false;
+    const cleanup = () => {
+        closed = true;
+        if (heartbeat) clearInterval(heartbeat);
+        unsubscribe?.();
+        heartbeat = undefined;
+        unsubscribe = undefined;
+    };
+
+    const stream = new ReadableStream<Uint8Array>({
+        start(controller) {
+            const send = (frame: () => Uint8Array) => {
+                if (closed) return;
+                if ((controller.desiredSize ?? 0) <= 0) {
+                    // Reconnect for fresh state instead of silently losing deltas.
+                    // close() does not invoke cancel(), so release resources here.
+                    cleanup();
+                    controller.close();
+                    return;
+                }
+                try { controller.enqueue(frame()); }
+                catch (cause) { cleanup(); controller.error(cause); }
+            };
+            send(() => hello);
+            for (const frame of initialFrames) send(() => frame);
+            if (closed) return;
+            unsubscribe = subscribe(event => {
+                if (accepts(event)) send(() => encode(event));
+            });
+            heartbeat = setInterval(() => send(() => ping), HEARTBEAT_MS);
+        },
+        cancel: cleanup
+    }, { highWaterMark: MAX_QUEUED_FRAMES });
+
+    return new Response(stream, {
+        headers: {
+            'Content-Type': 'text/event-stream',
+            'Cache-Control': 'no-cache, no-transform',
+            Connection: 'keep-alive'
+        }
+    });
+}

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

@@ -1,45 +1,14 @@
-import { subscribe, type PipelineEvent } from '$lib/server/pipeline/bus.js';
+import { eventStreamResponse } from '$lib/server/pipeline/event-stream.js';
 
-// Multiplexed SSE. Every event with a type field goes out. The client filters
+// Multiplexed SSE for deployment progress and fleet invalidations.
-// by type (server-state | deployment-opened | sweep-tick | server-changed).
 
 const encoder = new TextEncoder();
-const HEARTBEAT_MS = 20_000;
-
-function send(controller: ReadableStreamDefaultController<Uint8Array>, event: PipelineEvent): void {
-    controller.enqueue(encoder.encode(`event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`));
-}
 
 export async function GET() {
-    let heartbeat: NodeJS.Timeout | null = null;
+    return eventStreamResponse({
-    let unsubscribe: (() => void) | null = null;
+        accepts: event => event.type === 'server-state' || event.type === 'deployment-opened' ||
-
+            event.type === 'server-changed' || event.type === 'sweep-tick' ||
-    const stream = new ReadableStream<Uint8Array>({
+            event.type === 'fleet-invalidated' || event.type === 'beta-changed',
-        start(controller) {
+        encode: event => encoder.encode(`event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`)
-            unsubscribe = subscribe((event) => {
-                if (
-                    event.type === 'server-state' ||
-                    event.type === 'deployment-opened' ||
-                    event.type === 'server-changed' ||
-                    event.type === 'sweep-tick' || event.type === 'fleet-invalidated' || event.type === 'beta-changed'
-                ) {
-                    send(controller, event);
-                }
-            });
-            controller.enqueue(encoder.encode(': hello\n\n'));
-            heartbeat = setInterval(() => controller.enqueue(encoder.encode(': ping\n\n')), HEARTBEAT_MS);
-        },
-        cancel() {
-            if (heartbeat) clearInterval(heartbeat);
-            if (unsubscribe) unsubscribe();
-        }
-    });
-
-    return new Response(stream, {
-        headers: {
-            'Content-Type': 'text/event-stream',
-            'Cache-Control': 'no-cache, no-transform',
-            Connection: 'keep-alive'
-        }
     });
 }

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

@@ -1,51 +1,34 @@
 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, type PipelineEvent } from '$lib/server/pipeline/bus.js';
+import type { PipelineEvent } from '$lib/server/pipeline/bus.js';
+import { eventStreamResponse } from '$lib/server/pipeline/event-stream.js';
 import { listTelemetry } from '$lib/server/fleet/telemetry.js';
 import { playerCapacityReader } from '$lib/server/fleet/player-capacity.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;
+    const observation = observationReader();
-    let unsubscribe: (() => void) | null = null;
+    const capacity = playerCapacityReader();
+    const observe = (server: ReturnType<typeof listServers>[number]) => statusEntry(server, observation, capacity(server.name));
 
-    const stream = new ReadableStream<Uint8Array>({
+    return eventStreamResponse({
-        start(controller) {
+        accepts: event => event.type === 'server-status' || event.type === 'server-telemetry',
-            unsubscribe = subscribe((event) => {
+        encode(event) {
-                if (event.type === 'server-status' || event.type === 'server-telemetry') {
+            let frame = frames.get(event);
-                    let frame = frames.get(event);
+            if (!frame) {
-                    if (!frame) {
+                frame = encoder.encode(`event: ${event.type}\ndata: ${JSON.stringify({ ...event, serverNow: Date.now() })}\n\n`);
-                        frame = encoder.encode(`event: ${event.type}\ndata: ${JSON.stringify({ ...event, serverNow: Date.now() })}\n\n`);
+                frames.set(event, frame);
-                        frames.set(event, frame);
+            }
-                    }
+            return frame;
-                    controller.enqueue(frame);
-                }
-            });
-            const observation = observationReader();
-            const capacity = playerCapacityReader();
-            const observe = (server: ReturnType<typeof listServers>[number]) => statusEntry(server, observation, capacity(server.name));
-            controller.enqueue(encoder.encode(': hello\n\n'));
-            controller.enqueue(encoder.encode(`event: server-status\ndata: ${JSON.stringify({ entries: listServers().map(server => observe(server)), snapshot: true, serverNow: Date.now() })}\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() {
+        initialFrames: [
-            if (heartbeat) clearInterval(heartbeat);
+            encoder.encode(`event: server-status\ndata: ${JSON.stringify({ entries: listServers().map(server => observe(server)), snapshot: true, serverNow: Date.now() })}\n\n`),
-            if (unsubscribe) unsubscribe();
+            encoder.encode(`event: server-telemetry\ndata: ${JSON.stringify({ entries: listTelemetry(), snapshot: true, serverNow: Date.now() })}\n\n`)
-        }
+        ]
-    });
-
-    return new Response(stream, {
-        headers: {
-            'Content-Type': 'text/event-stream',
-            'Cache-Control': 'no-cache, no-transform',
-            Connection: 'keep-alive'
-        }
     });
 }

tests/unit/event-streams.test.ts

@@ -0,0 +1,159 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { emit, subscribe, type PipelineEvent } from '../../src/lib/server/pipeline/bus.js';
+import { GET as serverEvents } from '../../src/routes/api/events/servers/+server.js';
+import { GET as pipelineEvents } from '../../src/routes/api/events/pipeline/+server.js';
+
+vi.mock('../../src/lib/server/pipeline/bus.js', async importOriginal => {
+    const bus = await importOriginal<typeof import('../../src/lib/server/pipeline/bus.js')>();
+    return { ...bus, subscribe: vi.fn((listener: Parameters<typeof bus.subscribe>[0]) => vi.fn(bus.subscribe(listener))) };
+});
+vi.mock('../../src/lib/server/fleet/status.js', () => ({ observationReader: vi.fn(() => ({})) }));
+vi.mock('../../src/lib/server/fleet/player-capacity.js', () => ({ playerCapacityReader: vi.fn(() => () => 4) }));
+vi.mock('../../src/lib/server/fleet/servers.js', () => ({ listServers: vi.fn(() => [{ name: 'server-1' }]) }));
+vi.mock('../../src/lib/server/pipeline/worker.js', () => ({
+    statusEntry: vi.fn((server: { name: string }) => ({ server: server.name, players: 0, health: 'online' }))
+}));
+vi.mock('../../src/lib/server/fleet/telemetry.js', () => ({
+    listTelemetry: vi.fn(() => [{ server: 'server-1', event: 'heartbeat' }])
+}));
+
+const decoder = new TextDecoder();
+const readers: ReadableStreamDefaultReader<Uint8Array>[] = [];
+beforeEach(() => { vi.clearAllMocks(); vi.useFakeTimers(); });
+afterEach(async () => {
+    for (const reader of readers.splice(0)) await reader.cancel();
+    vi.useRealTimers();
+});
+
+async function open(get: () => Promise<Response>) {
+    const response = await get();
+    const reader = response.body!.getReader();
+    readers.push(reader);
+    return { response, reader, unsubscribe: vi.mocked(subscribe).mock.results.at(-1)!.value as ReturnType<typeof vi.fn> };
+}
+
+async function readFrame(reader: ReadableStreamDefaultReader<Uint8Array>) {
+    const next = await reader.read();
+    expect(next.done).toBe(false);
+    return decoder.decode(next.value);
+}
+
+async function drain(reader: ReadableStreamDefaultReader<Uint8Array>) {
+    const frames: string[] = [];
+    for (;;) {
+        const next = await reader.read();
+        if (next.done) return frames;
+        frames.push(decoder.decode(next.value));
+    }
+}
+
+const status: PipelineEvent = { type: 'server-status', entries: [], delta: true };
+const telemetry: PipelineEvent = { type: 'server-telemetry', entries: [] };
+const sweep: PipelineEvent = { type: 'sweep-tick', frozen: false };
+const routes = [
+    { name: 'servers', get: serverEvents, initialCount: 3, event: status, ignored: sweep, accepted: [status, telemetry] },
+    { name: 'pipeline', get: pipelineEvents, initialCount: 1, event: sweep, ignored: status, accepted: [
+        sweep,
+        { type: 'deployment-opened', deploymentId: 'deploy-1', action: 'push', servers: ['server-1'] },
+        { type: 'server-changed', deploymentId: 'deploy-1', server: 'server-1', files: 2 },
+        { type: 'server-state', deploymentId: 'deploy-1', server: 'server-1', state: 'done', players: 0, error: null },
+        { type: 'fleet-invalidated' },
+        { type: 'beta-changed', betaId: 1 }
+    ] as PipelineEvent[] }
+];
+
+describe.each(routes)('$name live events', route => {
+    it('preserves headers, initial frames, event filtering and heartbeats', async () => {
+        const { response, reader, unsubscribe } = await open(route.get);
+        expect(response.headers.get('content-type')).toBe('text/event-stream');
+        expect(response.headers.get('cache-control')).toBe('no-cache, no-transform');
+        expect(response.headers.get('connection')).toBe('keep-alive');
+        expect(await readFrame(reader)).toBe(': hello\n\n');
+        if (route.name === 'servers') {
+            expect(await readFrame(reader)).toBe(`event: server-status\ndata: ${JSON.stringify({
+                entries: [{ server: 'server-1', players: 0, health: 'online' }], snapshot: true, serverNow: Date.now()
+            })}\n\n`);
+            expect(await readFrame(reader)).toBe(`event: server-telemetry\ndata: ${JSON.stringify({
+                entries: [{ server: 'server-1', event: 'heartbeat' }], snapshot: true, serverNow: Date.now()
+            })}\n\n`);
+        }
+        emit(route.ignored);
+        for (const event of route.accepted) {
+            emit(event);
+            const data = route.name === 'servers' ? { ...event, serverNow: Date.now() } : event;
+            expect(await readFrame(reader)).toBe(`event: ${event.type}\ndata: ${JSON.stringify(data)}\n\n`);
+        }
+        vi.advanceTimersByTime(20_000);
+        expect(await readFrame(reader)).toBe(': ping\n\n');
+        expect(unsubscribe).not.toHaveBeenCalled();
+    });
+
+    it('bounds an unread connection and releases its subscription and heartbeat on overflow', async () => {
+        const { reader, unsubscribe } = await open(route.get);
+        for (let i = route.initialCount; i < 64; i++) emit({ ...route.event });
+        // An unrelated event must not close a full connection or use queue space.
+        for (let i = 0; i < 100; i++) emit(route.ignored);
+        expect(unsubscribe).not.toHaveBeenCalled();
+        for (let i = 0; i < 1000; i++) emit({ ...route.event });
+        expect(unsubscribe).toHaveBeenCalledOnce();
+        expect(vi.getTimerCount()).toBe(0);
+        const frames = await drain(reader);
+        expect(frames).toHaveLength(64);
+        expect(frames[0]).toBe(': hello\n\n');
+        expect(frames.at(-1)).toContain(`event: ${route.event.type}\n`);
+        await reader.cancel();
+        expect(unsubscribe).toHaveBeenCalledOnce();
+
+        // A reconnect gets the greeting and current snapshots again.
+        vi.advanceTimersByTime(1000);
+        const reconnect = await open(route.get);
+        expect(await readFrame(reconnect.reader)).toBe(': hello\n\n');
+        for (let i = 1; i < route.initialCount; i++) {
+            expect(await readFrame(reconnect.reader)).toContain(`"snapshot":true,"serverNow":${Date.now()}`);
+        }
+        expect(reconnect.unsubscribe).not.toHaveBeenCalled();
+    });
+
+    it('keeps a reading client live when another client stalls', async () => {
+        const slow = await open(route.get);
+        const fast = await open(route.get);
+        for (let i = 0; i < route.initialCount; i++) await readFrame(fast.reader);
+        for (let i = 0; i < 200; i++) {
+            const next = fast.reader.read();
+            emit({ ...route.event });
+            const frame = await next;
+            expect(frame.done).toBe(false);
+            expect(decoder.decode(frame.value)).toContain(`event: ${route.event.type}\n`);
+        }
+        expect(slow.unsubscribe).toHaveBeenCalledOnce();
+        expect(fast.unsubscribe).not.toHaveBeenCalled();
+        expect(vi.getTimerCount()).toBe(1);
+        expect(await drain(slow.reader)).toHaveLength(64);
+        await fast.reader.cancel();
+        expect(fast.unsubscribe).toHaveBeenCalledOnce();
+        expect(vi.getTimerCount()).toBe(0);
+    });
+
+    it('also bounds heartbeats when no events arrive', async () => {
+        const { reader, unsubscribe } = await open(route.get);
+        vi.advanceTimersByTime(20_000 * 65);
+        expect(unsubscribe).toHaveBeenCalledOnce();
+        expect(vi.getTimerCount()).toBe(0);
+        const frames = await drain(reader);
+        expect(frames).toHaveLength(64);
+        expect(frames.slice(route.initialCount)).toEqual(Array(64 - route.initialCount).fill(': ping\n\n'));
+    });
+
+    it.each([false, true])('cleans up cancellation with initial frames consumed: %s', async consumed => {
+        const { reader, unsubscribe } = await open(route.get);
+        const listener = vi.mocked(subscribe).mock.calls.at(-1)![0];
+        if (consumed) for (let i = 0; i < route.initialCount; i++) await readFrame(reader);
+        await reader.cancel();
+        await reader.cancel();
+        expect(unsubscribe).toHaveBeenCalledOnce();
+        expect(vi.getTimerCount()).toBe(0);
+        // A callback already captured by the bus must be harmless after cleanup.
+        expect(() => { listener(route.event); emit(route.event); vi.advanceTimersByTime(60_000); }).not.toThrow();
+        expect(await reader.read()).toEqual({ done: true, value: undefined });
+    });
+});