/** * SSE liveness heartbeat. * * `cleanupDeadClients()` runs every SSE_HEARTBEAT_INTERVAL (15s) and does two * jobs: evict clients whose socket died, and write a liveness frame to the * ones that are still up. * * The regression these guard: that frame used to be an SSE `:keepalive` * COMMENT, and comments are invisible to `EventSource` by spec. A stream that * stopped delivering without erroring was therefore undetectable to the * client: `onerror` never fired, the header dot stayed green, and every * SSE-driven surface froze until the user reloaded. A named `sse:heartbeat` * event reaches a listener, which is what lets the client's staleness * watchdog notice the silence (see test/sse-staleness.test.ts). * * No port needed (the manager is driven directly with fake replies). */ import type { FastifyReply } from 'fastify'; import { describe, it, expect } from 'vitest'; import { SSE_PADDING_SIZE } from '../src/config/server-timing.js'; import { SseEvent } from '../src/web/sse-events.js'; import { SseStreamManager } from '../src/web/sse-stream-manager.js'; import { CleanupManager } from '../src/utils/index.js'; /** A FastifyReply stand-in that records every raw write. */ function fakeClient(opts: { destroyed?: boolean; writable?: boolean; throwOnAccess?: boolean } = {}) { const writes: string[] = []; const socket = { destroyed: opts.destroyed ?? false, writable: opts.writable ?? true }; const raw = { get socket() { if (opts.throwOnAccess) throw new Error('socket gone'); return socket; }, write(chunk: string) { writes.push(chunk); return true; }, }; return { reply: { raw } as unknown as FastifyReply, writes }; } function makeManager() { const cleanup = new CleanupManager(); const manager = new SseStreamManager({ getSessionStateWithRespawn: () => null }, cleanup); return { manager, cleanup }; } describe('SSE liveness heartbeat', () => { it('writes a NAMED sse:heartbeat event, not an invisible comment', () => { const { manager, cleanup } = makeManager(); const client = fakeClient(); manager.addClient(client.reply, null, false); manager.cleanupDeadClients(); expect(client.writes).toHaveLength(1); const frame = client.writes[0]; // A comment (`:keepalive`) never reaches an EventSource listener, and that is // the entire bug. The frame must be a dispatchable named event. expect(frame.startsWith(':')).toBe(false); expect(frame).toMatch(/^event: sse:heartbeat\n/); expect(frame.endsWith('\n\n')).toBe(true); expect(SseEvent.Heartbeat).toBe('sse:heartbeat'); cleanup.dispose(); }); it('carries a parseable epoch-ms payload', () => { const { manager, cleanup } = makeManager(); const client = fakeClient(); manager.addClient(client.reply, null, false); const before = Date.now(); manager.cleanupDeadClients(); const dataLine = client.writes[0].split('\n').find((l) => l.startsWith('data: ')); expect(dataLine).toBeDefined(); const payload = JSON.parse(dataLine!.slice('data: '.length)) as { t: number }; expect(payload.t).toBeGreaterThanOrEqual(before); expect(payload.t).toBeLessThanOrEqual(Date.now()); cleanup.dispose(); }); it('still appends Cloudflare tunnel padding when a tunnel is active', () => { const { manager, cleanup } = makeManager(); const client = fakeClient(); manager.addClient(client.reply, null, false); manager.setTunnelActive(true); manager.cleanupDeadClients(); const frame = client.writes[0]; expect(frame).toMatch(/^event: sse:heartbeat\n/); // Padding rides AFTER the terminating blank line, so the event still parses. const [event, padding] = frame.split('\n\n'); expect(event).toMatch(/^event: sse:heartbeat\ndata: \{/); expect(padding.startsWith(':')).toBe(true); expect(padding.length).toBeGreaterThanOrEqual(SSE_PADDING_SIZE); cleanup.dispose(); }); it('sends no padding without a tunnel', () => { const { manager, cleanup } = makeManager(); const client = fakeClient(); manager.addClient(client.reply, null, false); manager.cleanupDeadClients(); expect(client.writes[0].length).toBeLessThan(200); cleanup.dispose(); }); it('still evicts dead clients instead of heartbeating them', () => { const { manager, cleanup } = makeManager(); const alive = fakeClient(); const destroyed = fakeClient({ destroyed: true }); const unwritable = fakeClient({ writable: false }); const exploding = fakeClient({ throwOnAccess: true }); for (const c of [alive, destroyed, unwritable, exploding]) manager.addClient(c.reply, null, false); expect(manager.clientCount).toBe(4); manager.cleanupDeadClients(); expect(manager.clientCount).toBe(1); expect(alive.writes).toHaveLength(1); for (const c of [destroyed, unwritable, exploding]) expect(c.writes).toHaveLength(0); cleanup.dispose(); }); it('heartbeats every client on each pass', () => { const { manager, cleanup } = makeManager(); const a = fakeClient(); const b = fakeClient(); manager.addClient(a.reply, null, false); manager.addClient(b.reply, null, false); manager.cleanupDeadClients(); manager.cleanupDeadClients(); // The frame carries no session data, so it needs no owner routing and is // written per-client rather than through the scoped broadcast() path. expect(a.writes).toHaveLength(2); expect(b.writes).toHaveLength(2); cleanup.dispose(); }); });