227 lines
8.9 KiB
TypeScript
227 lines
8.9 KiB
TypeScript
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 });
|
|
}
|
|
);
|
|
});
|