diff --git a/src/lib/server/pipeline/event-stream.ts b/src/lib/server/pipeline/event-stream.ts new file mode 100644 index 0000000..33c11d6 --- /dev/null +++ b/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 | 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({ + 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' + } + }); +} diff --git a/src/routes/api/events/pipeline/+server.ts b/src/routes/api/events/pipeline/+server.ts index 6a89845..93a7257 100644 --- a/src/routes/api/events/pipeline/+server.ts +++ b/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 -// by type (server-state | deployment-opened | sweep-tick | server-changed). +// Multiplexed SSE for deployment progress and fleet invalidations. const encoder = new TextEncoder(); -const HEARTBEAT_MS = 20_000; - -function send(controller: ReadableStreamDefaultController, 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; - let unsubscribe: (() => void) | null = null; - - const stream = new ReadableStream({ - start(controller) { - 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' - } + return eventStreamResponse({ + accepts: event => event.type === 'server-state' || event.type === 'deployment-opened' || + event.type === 'server-changed' || event.type === 'sweep-tick' || + event.type === 'fleet-invalidated' || event.type === 'beta-changed', + encode: event => encoder.encode(`event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`) }); } diff --git a/src/routes/api/events/servers/+server.ts b/src/routes/api/events/servers/+server.ts index f7f43a1..dd21c83 100644 --- a/src/routes/api/events/servers/+server.ts +++ b/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(); export async function GET() { - let heartbeat: NodeJS.Timeout | null = null; - let unsubscribe: (() => void) | null = null; + const observation = observationReader(); + const capacity = playerCapacityReader(); + const observe = (server: ReturnType[number]) => statusEntry(server, observation, capacity(server.name)); - const stream = new ReadableStream({ - start(controller) { - unsubscribe = subscribe((event) => { - if (event.type === 'server-status' || event.type === 'server-telemetry') { - 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(); - const capacity = playerCapacityReader(); - const observe = (server: ReturnType[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); + return eventStreamResponse({ + accepts: event => event.type === 'server-status' || event.type === 'server-telemetry', + encode(event) { + 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); + } + return frame; }, - 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' - } + initialFrames: [ + encoder.encode(`event: server-status\ndata: ${JSON.stringify({ entries: listServers().map(server => observe(server)), snapshot: true, serverNow: Date.now() })}\n\n`), + encoder.encode(`event: server-telemetry\ndata: ${JSON.stringify({ entries: listTelemetry(), snapshot: true, serverNow: Date.now() })}\n\n`) + ] }); } diff --git a/tests/unit/event-streams.test.ts b/tests/unit/event-streams.test.ts new file mode 100644 index 0000000..2d25832 --- /dev/null +++ b/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(); + return { ...bus, subscribe: vi.fn((listener: Parameters[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[] = []; +beforeEach(() => { vi.clearAllMocks(); vi.useFakeTimers(); }); +afterEach(async () => { + for (const reader of readers.splice(0)) await reader.cancel(); + vi.useRealTimers(); +}); + +async function open(get: () => Promise) { + 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 }; +} + +async function readFrame(reader: ReadableStreamDefaultReader) { + const next = await reader.read(); + expect(next.done).toBe(false); + return decoder.decode(next.value); +} + +async function drain(reader: ReadableStreamDefaultReader) { + 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 }); + }); +});