diff --git a/CLAUDE.md b/CLAUDE.md index 057c518f..b82e30de 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -159,6 +159,8 @@ Unified table using TanStack Table v8 (headless) + CSS Grid rows. Key files: `Da Separate Bun package that runs as a sidecar container alongside Docker hosts. Provides a REST/SSE API for Docker management operations (deploy, logs, stats streaming). Uses raw `Bun.serve()` with manual route matching and timing-safe auth middleware (zero framework dependencies beyond Dockerode). The agent replaces direct Docker API calls from the worker; the main app communicates with agents rather than Docker hosts directly. +Every agent SSE route (`stats`, `logs`, `containers-events`, `zfs`) builds its `Response` through `agent/src/lib/sse-stream.ts`, which owns the headers, initial flush, 5s comment heartbeat, frame grammar, and abort/enqueue-failure teardown. It is a deliberate copy of the web app's `src/lib/sse/create-sse-stream.ts`: the agent is not a workspace member and cannot import web code, so the seam exists once on each side of the split. Routes that own a subprocess or subscription return a cleanup from `onStart` (run exactly once at teardown) and must register it before any long-running loop, otherwise a teardown mid-loop never reaches it. + The agent is intentionally NOT a workspace member of the homelab-manager `package.json`: its `agent/bun.lock` is the only lockfile the docker build (`context: ./agent`) sees, and workspace membership would mask drift by routing local `bun install` to the homelab-manager lockfile. Web/worker import only types from the agent via the TS path alias `@homelab-manager/agent/*` (resolved at compile time, no runtime dependency). Run `bun run setup` for a full install. ### Authentication (`src/lib/auth/`) diff --git a/agent/src/__tests__/containers-events.test.ts b/agent/src/__tests__/containers-events.test.ts index dc2f8430..d878818c 100644 --- a/agent/src/__tests__/containers-events.test.ts +++ b/agent/src/__tests__/containers-events.test.ts @@ -3,6 +3,7 @@ import { EventEmitter } from 'node:events'; import { handleContainerEvents } from '../routes/containers-events'; import { _resetBroadcasterForTesting } from '../lib/docker-events-broadcaster'; import { zInventorySnapshotContainer } from '../types/protocol'; +import { readUntil, parseDataFrames } from '../lib/test/sse-test-utils'; const originalConsoleError = console.error; @@ -18,37 +19,23 @@ beforeEach(() => { _resetBroadcasterForTesting(); }); -/** Read chunks from the stream until predicate is satisfied or timeout. */ -async function readUntil( - response: Response, - predicate: (accumulated: string) => boolean, - timeoutMs = 3000, -): Promise { - let text = ''; - const reader = response.body!.getReader(); - const decoder = new TextDecoder(); - try { - while (true) { - let timeoutId: ReturnType | undefined; - const timeoutPromise = new Promise((_, reject) => { - timeoutId = setTimeout( - () => reject(new Error(`readUntil timed out after ${timeoutMs}ms`)), - timeoutMs, - ); - }); - try { - const { done, value } = await Promise.race([reader.read(), timeoutPromise]); - if (done) break; - text += decoder.decode(value, { stream: true }); - if (predicate(text)) break; - } finally { - if (timeoutId !== undefined) clearTimeout(timeoutId); - } - } - } finally { - reader.cancel(); - } - return text; +function countDataFrames(text: string): number { + return text.split('\n\n').filter((frame) => frame.startsWith('data: ')).length; +} + +/** + * A docker events stream whose teardown is awaitable. The broadcaster destroys + * it once its last subscriber unsubscribes, which is how a test observes that a + * disconnected client released its subscription. + */ +function makeDestroyableEventsStream(): { stream: EventEmitter; destroyed: Promise } { + const stream = new EventEmitter(); + let markDestroyed = () => {}; + const destroyed = new Promise((resolve) => { + markDestroyed = resolve; + }); + Object.assign(stream, { destroy: mock(() => markDestroyed()) }); + return { stream, destroyed }; } function makeContainer(id: string, name: string, state = 'running', image = 'nginx:latest') { @@ -118,7 +105,7 @@ describe('handleContainerEvents: init snapshot', () => { const text = await readUntil(response, (s) => s.includes('"op":"init"')); ac.abort(); - const event = JSON.parse(text.split('\n\n').find(Boolean)!.replace(/^data: /, '')); + const event = parseDataFrames(text)[0]; expect(event.op).toBe('init'); expect(event.containers).toHaveLength(2); expect(event.containers.map((c: { id: string }) => c.id).sort()).toEqual(['c1', 'c2'].sort((a, b) => a.localeCompare(b))); @@ -138,7 +125,7 @@ describe('handleContainerEvents: init snapshot', () => { const text = await readUntil(response, (s) => s.includes('"op":"init"')); ac.abort(); - const event = JSON.parse(text.split('\n\n').find(Boolean)!.replace(/^data: /, '')); + const event = parseDataFrames(text)[0]; expect(event.containers).toHaveLength(2); }); @@ -153,7 +140,7 @@ describe('handleContainerEvents: init snapshot', () => { const text = await readUntil(response, (s) => s.includes('"op":"init"')); ac.abort(); - const event = JSON.parse(text.split('\n\n').find(Boolean)!.replace(/^data: /, '')); + const event = parseDataFrames(text)[0]; const c = event.containers[0]; expect(c.id).toBe('abc123'); expect(c.name).toBe('my-app'); @@ -176,7 +163,7 @@ describe('handleContainerEvents: init snapshot', () => { const text = await readUntil(response, (s) => s.includes('"op":"init"')); ac.abort(); - const event = JSON.parse(text.split('\n\n').find(Boolean)!.replace(/^data: /, '')); + const event = parseDataFrames(text)[0]; expect(event.containers[0].name).toBe('test-app'); }); @@ -191,7 +178,7 @@ describe('handleContainerEvents: init snapshot', () => { const text = await readUntil(response, (s) => s.includes('"op":"init"')); ac.abort(); - const event = JSON.parse(text.split('\n\n').find(Boolean)!.replace(/^data: /, '')); + const event = parseDataFrames(text)[0]; expect(event.containers[0].state).toBe('unknown'); }); @@ -205,7 +192,7 @@ describe('handleContainerEvents: init snapshot', () => { const text = await readUntil(response, (s) => s.includes('"op":"init"')); ac.abort(); - const event = JSON.parse(text.split('\n\n').find(Boolean)!.replace(/^data: /, '')); + const event = parseDataFrames(text)[0]; expect(event.containers).toHaveLength(0); }); }); @@ -231,12 +218,11 @@ describe('handleContainerEvents: start event produces upsert', () => { }) + '\n')); const text = await readUntil(response, (s) => { - const events = s.split('\n\n').filter(Boolean); - return events.length >= 2; + return countDataFrames(s) >= 2; }); ac.abort(); - const events = text.split('\n\n').filter(Boolean).map((line) => JSON.parse(line.replace(/^data: /, ''))); + const events = parseDataFrames(text); const upsert = events.find((e) => e.op === 'upsert'); expect(upsert).toBeDefined(); expect(upsert.container.id).toBe('c1'); @@ -275,12 +261,11 @@ describe('handleContainerEvents: die event produces upsert with exited state', ( }) + '\n')); const text = await readUntil(response, (s) => { - const events = s.split('\n\n').filter(Boolean); - return events.length >= 2; + return countDataFrames(s) >= 2; }); ac.abort(); - const events = text.split('\n\n').filter(Boolean).map((line) => JSON.parse(line.replace(/^data: /, ''))); + const events = parseDataFrames(text); const upsert = events.find((e) => e.op === 'upsert'); expect(upsert).toBeDefined(); expect(upsert.container.id).toBe('c1'); @@ -307,12 +292,11 @@ describe('handleContainerEvents: destroy event', () => { }) + '\n')); const text = await readUntil(response, (s) => { - const events = s.split('\n\n').filter(Boolean); - return events.length >= 2; + return countDataFrames(s) >= 2; }); ac.abort(); - const events = text.split('\n\n').filter(Boolean).map((line) => JSON.parse(line.replace(/^data: /, ''))); + const events = parseDataFrames(text); const destroy = events.find((e) => e.op === 'destroy'); expect(destroy).toBeDefined(); expect(destroy.containerId).toBe('c1'); @@ -420,27 +404,32 @@ describe('handleContainerEvents: request abort cleanup', () => { expect(done).toBe(true); }); - test('already-aborted request does not register a broadcaster subscriber', async () => { - const docker = makeDocker([]); + test('already-aborted request unsubscribes the broadcaster once subscribe resolves', async () => { + const { stream, destroyed } = makeDestroyableEventsStream(); + const docker = makeDocker([], stream); const ac = new AbortController(); ac.abort(); const request = new Request('http://localhost/containers/events', { signal: ac.signal }); const response = await handleContainerEvents(docker as any, request); const reader = response.body!.getReader(); - const result = await reader.read(); - expect(result.done).toBe(true); - expect(docker.getEvents).not.toHaveBeenCalled(); + let done = false; + while (!done) { + done = (await reader.read()).done; + } + + await destroyed; }); test('abort during broadcasterSubscribe tears down the late-arriving subscriber', async () => { // Slow listContainers so subscribe()'s await is in-flight when we abort. const listHolder: { resolve: ((v: unknown[]) => void) | null } = { resolve: null }; + const { stream, destroyed } = makeDestroyableEventsStream(); const docker = { listContainers: mock(() => new Promise((resolve) => { listHolder.resolve = resolve; })), - getEvents: mock(() => Promise.resolve(new EventEmitter())), + getEvents: mock(() => Promise.resolve(stream)), getContainer: mock(() => ({ inspect: mock(() => Promise.reject(Object.assign(new Error('n/a'), { statusCode: 404 }))), })), @@ -455,8 +444,11 @@ describe('handleContainerEvents: request abort cleanup', () => { ac.abort(); listHolder.resolve?.([]); - const result = await reader.read(); - expect(result.done).toBe(true); + let done = false; + while (!done) { + done = (await reader.read()).done; + } + await destroyed; }); }); @@ -512,7 +504,7 @@ describe('handleContainerEvents: ports and mounts pass-through', () => { const text = await readUntil(response, (s) => s.includes('"op":"init"')); ac.abort(); - const event = JSON.parse(text.split('\n\n').find(Boolean)!.replace(/^data: /, '')); + const event = parseDataFrames(text)[0]; const c = event.containers[0]; expect(c.ports).toEqual([{ containerPort: 80, protocol: 'tcp', hostIp: '0.0.0.0', hostPort: 8080 }]); expect(c.mounts).toEqual([{ type: 'bind', source: '/host', destination: '/data', rw: true }]); @@ -529,7 +521,7 @@ describe('handleContainerEvents: ports and mounts pass-through', () => { const text = await readUntil(response, (s) => s.includes('"op":"init"')); ac.abort(); - const event = JSON.parse(text.split('\n\n').find(Boolean)!.replace(/^data: /, '')); + const event = parseDataFrames(text)[0]; const c = event.containers[0]; expect(c.ports).toEqual([]); expect(c.mounts).toEqual([]); @@ -569,12 +561,11 @@ describe('handleContainerEvents: ports and mounts pass-through', () => { }) + '\n')); const text = await readUntil(response, (s) => { - const events = s.split('\n\n').filter(Boolean); - return events.length >= 2; + return countDataFrames(s) >= 2; }); ac.abort(); - const events = text.split('\n\n').filter(Boolean).map((line) => JSON.parse(line.replace(/^data: /, ''))); + const events = parseDataFrames(text); const upsert = events.find((e) => e.op === 'upsert'); expect(upsert.container.ports).toEqual([{ containerPort: 53, protocol: 'udp', hostIp: '::', hostPort: 5353 }]); expect(upsert.container.mounts).toEqual([{ type: 'volume', source: 'vol1', destination: '/var/data', rw: true }]); @@ -612,20 +603,18 @@ describe('zInventorySnapshotContainer: ports/mounts schema round-trip', () => { describe('handleContainerEvents: idle heartbeat', () => { const realSetInterval = globalThis.setInterval; - const realClearInterval = globalThis.clearInterval; afterEach(() => { globalThis.setInterval = realSetInterval; - globalThis.clearInterval = realClearInterval; }); - test('enqueues a comment heartbeat to keep the idle socket alive', async () => { - // A quiet host emits no container events, so without this the socket sits - // silent past Bun's 10s HTTP idleTimeout and the worker reconnects in a loop. - let captured: (() => void) | null = null; - globalThis.setInterval = mock((cb: () => void) => { - captured = cb; - return 1 as unknown as ReturnType; + test('a quiet host still gets a comment heartbeat on the 5s cadence', async () => { + // Bun's HTTP idleTimeout defaults to 10s, so a host with no container + // activity would otherwise go silent long enough to drop the socket. + const ticks: Array<{ cb: () => void; ms: number }> = []; + globalThis.setInterval = mock((cb: () => void, ms: number) => { + ticks.push({ cb, ms }); + return ticks.length as unknown as ReturnType; }) as unknown as typeof setInterval; const docker = makeDocker([]); @@ -633,34 +622,13 @@ describe('handleContainerEvents: idle heartbeat', () => { const request = new Request('http://localhost/containers/events', { signal: ac.signal }); const response = await handleContainerEvents(docker as any, request); - // Let start() finish broadcaster setup and register the interval. - await new Promise((r) => setTimeout(r, 50)); - expect(captured).not.toBeNull(); + expect(ticks).toHaveLength(1); + expect(ticks[0].ms).toBe(5000); - captured!(); - const text = await readUntil(response, (s) => s.includes(':\n\n')); + ticks[0].cb(); + const text = await readUntil(response, (s) => s.split('\n\n').includes(':')); ac.abort(); - // Init frame plus a bare ':' comment heartbeat. expect(text.split('\n\n')).toContain(':'); }); - - test('clears the heartbeat interval when the request aborts', async () => { - const fakeId = Symbol('hb') as unknown as ReturnType; - globalThis.setInterval = mock(() => fakeId) as unknown as typeof setInterval; - const clearSpy = mock(() => {}); - globalThis.clearInterval = clearSpy as unknown as typeof clearInterval; - - const docker = makeDocker([]); - const ac = new AbortController(); - const request = new Request('http://localhost/containers/events', { signal: ac.signal }); - const response = await handleContainerEvents(docker as any, request); - - await new Promise((r) => setTimeout(r, 50)); // let the interval register - ac.abort(); - await new Promise((r) => setTimeout(r, 10)); - - expect(clearSpy).toHaveBeenCalledWith(fakeId); - void response; - }); }); diff --git a/agent/src/__tests__/logs.test.ts b/agent/src/__tests__/logs.test.ts index a855e774..640ae12b 100644 --- a/agent/src/__tests__/logs.test.ts +++ b/agent/src/__tests__/logs.test.ts @@ -381,9 +381,11 @@ describe('handleLogStream', () => { const originalSetInterval = globalThis.setInterval; const originalClearInterval = globalThis.clearInterval; let heartbeatCb: (() => void) | null = null; + const heartbeatDelays: number[] = []; const fakeTimerId = 999; - globalThis.setInterval = ((cb: () => void) => { + globalThis.setInterval = ((cb: () => void, ms: number) => { heartbeatCb = cb; + heartbeatDelays.push(ms); return fakeTimerId as unknown as ReturnType; }) as typeof setInterval; globalThis.clearInterval = mock(() => {}) as typeof clearInterval; @@ -399,6 +401,7 @@ describe('handleLogStream', () => { // Fire the heartbeat callback expect(heartbeatCb).not.toBeNull(); + expect(heartbeatDelays).toEqual([5000]); heartbeatCb!(); liveEmitter.emit('end'); diff --git a/agent/src/__tests__/stats.test.ts b/agent/src/__tests__/stats.test.ts index 3e895e03..7eb87137 100644 --- a/agent/src/__tests__/stats.test.ts +++ b/agent/src/__tests__/stats.test.ts @@ -1,36 +1,12 @@ import { describe, expect, test, mock, beforeAll } from 'bun:test'; import { EventEmitter } from 'node:events'; import { handleStatsStream, computeMetrics, type StatsStreamOptions, type ComputedStats } from '../routes/stats'; +import { readUntil } from '../lib/test/sse-test-utils'; beforeAll(() => { console.error = mock(() => {}); }); -/** Read chunks from the stream until predicate is satisfied or timeout. */ -async function readUntil( - response: Response, - predicate: (accumulated: string) => boolean, - timeoutMs = 5000, -): Promise { - let text = ''; - const reader = response.body!.getReader(); - const decoder = new TextDecoder(); - try { - while (true) { - const timeoutPromise = new Promise((_, reject) => - setTimeout(() => reject(new Error(`readUntil timed out after ${timeoutMs}ms`)), timeoutMs), - ); - const { done, value } = await Promise.race([reader.read(), timeoutPromise]); - if (done) break; - text += decoder.decode(value, { stream: true }); - if (predicate(text)) break; - } - } finally { - reader.cancel(); - } - return text; -} - function makeStatsJson(overrides?: { cpuTotal?: number; preCpuTotal?: number; @@ -623,4 +599,33 @@ describe('handleStatsStream: container refresh', () => { expect(text).toContain('"type":"refresh_failed"'); expect(text).toContain('"error":"Docker daemon down"'); }); + + test('emits a comment heartbeat when the host runs no containers', async () => { + // With nothing running there are no stats frames at all, so the stream would + // otherwise sit silent past Bun's 10s HTTP idleTimeout. + const realSetInterval = globalThis.setInterval; + const ticks: Array<{ cb: () => void; ms: number }> = []; + globalThis.setInterval = mock((cb: () => void, ms: number) => { + ticks.push({ cb, ms }); + return ticks.length as unknown as ReturnType; + }) as unknown as typeof setInterval; + + try { + const mockDocker = { listContainers: mock(() => Promise.resolve([])) }; + const ac = new AbortController(); + const request = new Request('http://localhost/stats/stream', { signal: ac.signal }); + const response = handleStatsStream(mockDocker as any, request, fastOptions); + + expect(ticks).toHaveLength(1); + expect(ticks[0].ms).toBe(5000); + + ticks[0].cb(); + const text = await readUntil(response, (s) => s.split('\n\n').includes(':')); + ac.abort(); + + expect(text.split('\n\n')).toContain(':'); + } finally { + globalThis.setInterval = realSetInterval; + } + }); }); diff --git a/agent/src/__tests__/zfs.test.ts b/agent/src/__tests__/zfs.test.ts index 17ac18a5..d630533b 100644 --- a/agent/src/__tests__/zfs.test.ts +++ b/agent/src/__tests__/zfs.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test, mock, beforeEach, afterEach } from 'bun:test'; import type { ZfsCapabilities } from '../lib/zfs-capabilities'; import { handleZfsStatsStream, handleZfsPools } from '../routes/zfs'; +import { readUntil, parseDataFrames } from '../lib/test/sse-test-utils'; const originalConsoleError = console.error; @@ -31,30 +32,6 @@ afterEach(() => { Bun.spawn = originalSpawn; }); -/** Read SSE chunks from a response until a predicate is satisfied or timeout. */ -async function readUntil( - response: Response, - predicate: (accumulated: string) => boolean, - timeoutMs = 5000, -): Promise { - const deadline = Date.now() + timeoutMs; - let text = ''; - const reader = response.body!.getReader(); - const decoder = new TextDecoder(); - try { - while (true) { - if (Date.now() > deadline) break; - const { done, value } = await reader.read(); - if (done) break; - text += decoder.decode(value, { stream: true }); - if (predicate(text)) break; - } - } finally { - reader.cancel(); - } - return text; -} - describe('handleZfsStatsStream', () => { test('returns 503 when ZFS is not available', () => { const request = new Request('http://localhost/zfs/stats/stream'); @@ -94,6 +71,44 @@ describe('handleZfsStatsStream', () => { expect(response.headers.get('Connection')).toBe('keep-alive'); }); + test('pipes subprocess stderr and drains it into the log', async () => { + let spawnOptions: unknown; + let resolveLogged!: (line: string) => void; + const logged = new Promise((resolve) => { + resolveLogged = resolve; + }); + console.error = mock((...args: unknown[]) => { + if (args[0] === 'zpool iostat:') resolveLogged(String(args[1])); + }); + + Bun.spawn = mock((_cmd: string[], options: unknown) => { + spawnOptions = options; + return { + stdout: new ReadableStream({ start() {} }), + stderr: new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode("cannot open 'tank': no such pool\n"), + ); + }, + }), + kill: mock(() => {}), + exited: new Promise(() => {}), + }; + }) as any; + + const ac = new AbortController(); + const request = new Request('http://localhost/zfs/stats/stream', { + signal: ac.signal, + }); + handleZfsStatsStream(request, zfsAvailable); + + expect(await logged).toBe("cannot open 'tank': no such pool"); + ac.abort(); + + expect(spawnOptions).toMatchObject({ stderr: 'pipe' }); + }); + test('streams parsed lines as SSE events', async () => { const killMock = mock(() => {}); const iostatOutput = [ @@ -126,13 +141,7 @@ describe('handleZfsStatsStream', () => { expect(text).toContain('data:'); // Each non-empty line should be a separate SSE event - const events = text - .split('\n\n') - .filter(Boolean) - .map((e) => { - const json = e.replace(/^data:\s*/, ''); - return JSON.parse(json); - }); + const events = parseDataFrames(text); expect(events.length).toBeGreaterThanOrEqual(1); const tankEvent = events.find((e: { line: string }) => e.line.includes('tank')); @@ -245,6 +254,37 @@ describe('handleZfsStatsStream', () => { expect(console.error).toHaveBeenCalled(); }); + test('logs error when stderr reader throws unexpectedly', async () => { + let resolveLogged!: (err: unknown) => void; + const logged = new Promise((resolve) => { + resolveLogged = resolve; + }); + console.error = mock((...args: unknown[]) => { + if (args[0] === 'ZFS stats stderr read error:') resolveLogged(args[1]); + }); + + Bun.spawn = mock(() => ({ + stdout: new ReadableStream({ start() {} }), + stderr: new ReadableStream({ + start() {}, + pull() { + throw new Error('stderr read failure'); + }, + }), + kill: mock(() => {}), + exited: new Promise(() => {}), + })) as any; + + const ac = new AbortController(); + const request = new Request('http://localhost/zfs/stats/stream', { + signal: ac.signal, + }); + handleZfsStatsStream(request, zfsAvailable); + + expect(await logged).toBeInstanceOf(Error); + ac.abort(); + }); + test('skips empty lines in output', async () => { const killMock = mock(() => {}); const output = '\n\n \nactual data line\n\n'; @@ -270,10 +310,44 @@ describe('handleZfsStatsStream', () => { ac.abort(); // Only the non-empty line should appear - const events = text.split('\n\n').filter(Boolean); + const events = parseDataFrames(text); expect(events.length).toBe(1); - const parsed = JSON.parse(events[0].replace(/^data:\s*/, '')); - expect(parsed.line).toBe('actual data line'); + expect(events[0].line).toBe('actual data line'); + }); + + test('emits a comment heartbeat while zpool produces no output', async () => { + // `zpool iostat -v 1` normally ticks every second, but a degraded pool with + // a hung disk can stall it well past Bun's 10s HTTP idleTimeout. + const realSetInterval = globalThis.setInterval; + const ticks: Array<{ cb: () => void; ms: number }> = []; + globalThis.setInterval = mock((cb: () => void, ms: number) => { + ticks.push({ cb, ms }); + return ticks.length as unknown as ReturnType; + }) as unknown as typeof setInterval; + + Bun.spawn = mock(() => ({ + stdout: new ReadableStream({ start() {} }), + stderr: new ReadableStream(), + kill: mock(() => {}), + exited: new Promise(() => {}), + })) as any; + + try { + const ac = new AbortController(); + const request = new Request('http://localhost/zfs/stats/stream', { signal: ac.signal }); + const response = handleZfsStatsStream(request, zfsAvailable); + + expect(ticks).toHaveLength(1); + expect(ticks[0].ms).toBe(5000); + + ticks[0].cb(); + const text = await readUntil(response, (s) => s.split('\n\n').includes(':')); + ac.abort(); + + expect(text.split('\n\n')).toContain(':'); + } finally { + globalThis.setInterval = realSetInterval; + } }); }); diff --git a/agent/src/lib/__tests__/sse-stream.test.ts b/agent/src/lib/__tests__/sse-stream.test.ts new file mode 100644 index 00000000..1e0fa20f --- /dev/null +++ b/agent/src/lib/__tests__/sse-stream.test.ts @@ -0,0 +1,337 @@ +import { describe, it, expect, mock, spyOn, beforeEach, afterEach } from 'bun:test'; +import { createSseStream, isCloseRelatedError, type SseEmitter } from '../sse-stream'; + +interface IntervalHandle { + cb: () => void; + ms: number; + cleared: boolean; +} + +/** Replaces setInterval/clearInterval so the heartbeat is a callback tests can fire deterministically. */ +function createIntervalHarness() { + const intervals: IntervalHandle[] = []; + const setSpy = spyOn(globalThis, 'setInterval').mockImplementation(((cb: () => void, ms: number) => { + const handle: IntervalHandle = { cb, ms, cleared: false }; + intervals.push(handle); + return handle as unknown as ReturnType; + }) as typeof setInterval); + const clearSpy = spyOn(globalThis, 'clearInterval').mockImplementation(((h: unknown) => { + const handle = h as IntervalHandle; + if (handle) handle.cleared = true; + }) as typeof clearInterval); + return { + intervals, + restore() { + setSpy.mockRestore(); + clearSpy.mockRestore(); + }, + }; +} + +function makeRequest(ac: AbortController): Request { + return new Request('http://localhost/', { signal: ac.signal }); +} + +function readerOf(res: Response): ReadableStreamDefaultReader { + if (!res.body) throw new Error('Response had no body'); + return res.body.getReader(); +} + +async function readFrame(reader: ReadableStreamDefaultReader): Promise { + const { value, done } = await reader.read(); + if (done) return ''; + return new TextDecoder().decode(value); +} + + +async function readAll(reader: ReadableStreamDefaultReader): Promise { + const decoder = new TextDecoder(); + let out = ''; + for (;;) { + const { value, done } = await reader.read(); + if (done) break; + out += decoder.decode(value); + } + return out; +} + +/** Sets up a stream whose onStart hands the test direct access to `emit`/`signal`. */ +function setup(opts: { heartbeatMs?: number; cleanup?: () => void } = {}) { + let capturedEmit: SseEmitter | null = null; + let capturedSignal: AbortSignal | null = null; + const cleanup = opts.cleanup ?? mock(() => {}); + + const ac = new AbortController(); + const res = createSseStream(makeRequest(ac), { + heartbeatMs: opts.heartbeatMs, + onStart: (emit, signal) => { + capturedEmit = emit; + capturedSignal = signal; + return cleanup; + }, + }); + + return { + res, + ac, + cleanup, + getEmit: () => capturedEmit!, + getSignal: () => capturedSignal!, + }; +} + +describe('createSseStream', () => { + let harness: ReturnType; + let errorSpy: ReturnType; + + beforeEach(() => { + harness = createIntervalHarness(); + errorSpy = spyOn(console, 'error').mockImplementation(() => {}); + }); + + afterEach(() => { + errorSpy.mockRestore(); + harness.restore(); + }); + + it('returns the standard SSE response headers', () => { + const { res, ac } = setup(); + + expect(res.status).toBe(200); + expect(res.headers.get('Content-Type')).toBe('text/event-stream'); + expect(res.headers.get('Cache-Control')).toBe('no-cache'); + expect(res.headers.get('Connection')).toBe('keep-alive'); + + ac.abort(); + }); + + it('flushes ": ok\\n\\n" as the first frame', async () => { + const { res, ac } = setup(); + const reader = readerOf(res); + + expect(await readFrame(reader)).toBe(': ok\n\n'); + + ac.abort(); + reader.cancel(); + }); + + it('arms exactly one heartbeat timer at the default 5000ms cadence', () => { + const { ac } = setup(); + + expect(harness.intervals).toHaveLength(1); + expect(harness.intervals[0].ms).toBe(5000); + + ac.abort(); + }); + + it('respects a caller-supplied heartbeatMs', () => { + const { ac } = setup({ heartbeatMs: 1234 }); + + expect(harness.intervals[0].ms).toBe(1234); + + ac.abort(); + }); + + it('writes a bare ":\\n\\n" comment on each heartbeat tick', async () => { + const { res, ac } = setup(); + const reader = readerOf(res); + await readFrame(reader); // ": ok\n\n" + + harness.intervals[0].cb(); + expect(await readFrame(reader)).toBe(':\n\n'); + + harness.intervals[0].cb(); + expect(await readFrame(reader)).toBe(':\n\n'); + + ac.abort(); + reader.cancel(); + }); + + it('clears the heartbeat timer on abort so no further ticks are written', async () => { + const { ac } = setup(); + + ac.abort(); + + expect(harness.intervals[0].cleared).toBe(true); + }); + + it('emit.data writes a "data: \\n\\n" frame', async () => { + const { res, ac, getEmit } = setup(); + const reader = readerOf(res); + await readFrame(reader); // flush comment + + getEmit().data({ n: 1 }); + expect(await readFrame(reader)).toBe('data: {"n":1}\n\n'); + + ac.abort(); + reader.cancel(); + }); + + it('emit.event writes an "event: \\ndata: \\n\\n" frame', async () => { + const { res, ac, getEmit } = setup(); + const reader = readerOf(res); + await readFrame(reader); + + getEmit().event('stats_error', { message: 'db down' }); + expect(await readFrame(reader)).toBe('event: stats_error\ndata: {"message":"db down"}\n\n'); + + ac.abort(); + reader.cancel(); + }); + + it('emit.raw writes a pre-formatted string frame verbatim', async () => { + const { res, ac, getEmit } = setup(); + const reader = readerOf(res); + await readFrame(reader); + + getEmit().raw('event: custom\ndata: 42\n\n'); + expect(await readFrame(reader)).toBe('event: custom\ndata: 42\n\n'); + + ac.abort(); + reader.cancel(); + }); + + it('emit.raw writes bytes verbatim without re-encoding', async () => { + const { res, ac, getEmit } = setup(); + const reader = readerOf(res); + await readFrame(reader); + + const bytes = new TextEncoder().encode('data: raw-bytes\n\n'); + getEmit().raw(bytes); + const frame = await reader.read(); + expect(new TextDecoder().decode(frame.value)).toBe('data: raw-bytes\n\n'); + + ac.abort(); + reader.cancel(); + }); + + it('runs the onStart-returned cleanup exactly once on abort', async () => { + let markCleanup = () => {}; + const cleanupDone = new Promise((resolve) => { markCleanup = resolve; }); + const cleanup = mock(() => markCleanup()); + const { ac } = setup({ cleanup }); + + ac.abort(); + ac.abort(); // idempotent; guards the assertion anyway + await cleanupDone; + + expect(cleanup).toHaveBeenCalledTimes(1); + }); + + it('runs cleanup once even when onStart resolves after the abort already tore the stream down', async () => { + let markCleanup = () => {}; + const cleanupDone = new Promise((resolve) => { markCleanup = resolve; }); + const cleanup = mock(() => markCleanup()); + const ac = new AbortController(); + let releaseOnStart: (() => void) | null = null; + + const res = createSseStream(makeRequest(ac), { + onStart: async () => { + // Simulates slow setup that hasn't returned its cleanup by the time the client disconnects. + await new Promise((resolve) => { + releaseOnStart = resolve; + }); + return cleanup; + }, + }); + const reader = readerOf(res); + await readFrame(reader); // ": ok\n\n" + + ac.abort(); + expect(cleanup).not.toHaveBeenCalled(); + + releaseOnStart!(); + await cleanupDone; + + expect(cleanup).toHaveBeenCalledTimes(1); + + reader.cancel(); + }); + + it('emit.close() ends the stream immediately: clears the heartbeat and runs cleanup', async () => { + const cleanup = mock(() => {}); + const { res, ac, getEmit } = setup({ cleanup }); + const reader = readerOf(res); + await readFrame(reader); + + getEmit().close(); + + expect(cleanup).toHaveBeenCalledTimes(1); + expect(harness.intervals[0].cleared).toBe(true); + + const { done } = await reader.read(); + expect(done).toBe(true); + + ac.abort(); + }); + + it('drops writes issued after teardown without throwing', async () => { + const { ac, getEmit } = setup(); + ac.abort(); + + expect(() => getEmit().data({ n: 1 })).not.toThrow(); + }); + + it('classifies a close-related write failure without logging, but still tears down', async () => { + const cleanup = mock(() => {}); + const { res, ac, getEmit } = setup({ cleanup }); + const reader = readerOf(res); + await readFrame(reader); + + // Serialization throws a close-shaped TypeError, exercising write()'s isCloseRelatedError branch. + const payload = { + toJSON() { + throw new TypeError('The stream is closed'); + }, + }; + + getEmit().data(payload); + + expect(errorSpy).not.toHaveBeenCalled(); + expect(cleanup).toHaveBeenCalledTimes(1); + + ac.abort(); + reader.cancel(); + }); + + it('isCloseRelatedError classifies TypeErrors mentioning "closed" and nothing else', () => { + expect(isCloseRelatedError(new TypeError('The stream is closed'))).toBe(true); + expect(isCloseRelatedError(new TypeError('Invalid state: Controller is already closed'))).toBe(true); + expect(isCloseRelatedError(new TypeError('something unrelated'))).toBe(false); + expect(isCloseRelatedError(new Error('closed'))).toBe(false); + expect(isCloseRelatedError('not an error')).toBe(false); + }); + + it('logs and tears down on an unexpected (non-close) write failure', async () => { + const cleanup = mock(() => {}); + const { res, ac, getEmit } = setup({ cleanup }); + const reader = readerOf(res); + await readFrame(reader); + + const circular: Record = {}; + circular.self = circular; + + getEmit().data(circular); + + expect(errorSpy).toHaveBeenCalledWith('Unexpected error during SSE enqueue:', expect.any(Error)); + expect(cleanup).toHaveBeenCalledTimes(1); + + ac.abort(); + reader.cancel(); + }); + + it('logs but does not crash the stream when onStart itself rejects', async () => { + const ac = new AbortController(); + const res = createSseStream(makeRequest(ac), { + onStart: async () => { + throw new Error('setup failed'); + }, + }); + const reader = readerOf(res); + + const body = await readAll(reader); + + expect(body).toBe(': ok\n\n'); + expect(errorSpy).toHaveBeenCalledWith('Unexpected error in SSE onStart handler:', expect.any(Error)); + }); +}); diff --git a/agent/src/lib/__tests__/sse-utils.test.ts b/agent/src/lib/__tests__/sse-utils.test.ts deleted file mode 100644 index afda027d..00000000 --- a/agent/src/lib/__tests__/sse-utils.test.ts +++ /dev/null @@ -1,103 +0,0 @@ -import { describe, expect, test, mock, beforeAll, beforeEach, afterAll } from 'bun:test'; -import { sendSSE } from '../sse-utils'; - -const originalConsoleError = console.error; - -beforeAll(() => { - console.error = mock(() => {}); -}); - -beforeEach(() => { - (console.error as ReturnType).mockClear(); -}); - -afterAll(() => { - console.error = originalConsoleError; -}); - -function makeEncoder(): TextEncoder { - return new TextEncoder(); -} - -function makeController(): { - controller: ReadableStreamDefaultController; - enqueued: string[]; - makeEnqueueThrow: (err: unknown) => void; -} { - const enqueued: string[] = []; - const decoder = new TextDecoder(); - let thrower: (() => never) | null = null; - const controller = { - enqueue: (chunk: Uint8Array) => { - if (thrower) thrower(); - enqueued.push(decoder.decode(chunk)); - }, - close: () => {}, - error: () => {}, - desiredSize: 1, - } as unknown as ReadableStreamDefaultController; - - return { - controller, - enqueued, - makeEnqueueThrow: (err) => { - thrower = () => { - throw err; - }; - }, - }; -} - -describe('sendSSE', () => { - test('enqueues data: frame when not closed', () => { - const { controller, enqueued } = makeController(); - sendSSE(controller, makeEncoder(), { value: false }, 'hello'); - expect(enqueued).toEqual(['data: hello\n\n']); - }); - - test('does nothing when closed flag is true', () => { - const { controller, enqueued } = makeController(); - sendSSE(controller, makeEncoder(), { value: true }, 'hello'); - expect(enqueued).toHaveLength(0); - }); - - test('swallows TypeError from enqueue (enqueue-after-close)', () => { - const { controller, makeEnqueueThrow } = makeController(); - makeEnqueueThrow(new TypeError('Controller is closed')); - expect(() => { - sendSSE(controller, makeEncoder(), { value: false }, 'hello'); - }).not.toThrow(); - expect(console.error).not.toHaveBeenCalledWith( - 'Unexpected error during SSE enqueue:', - expect.anything(), - ); - }); - - test('sets closed.value to true when enqueue fails with close-matching TypeError', () => { - const { controller, makeEnqueueThrow } = makeController(); - makeEnqueueThrow(new TypeError('Controller is closed')); - const closed = { value: false }; - sendSSE(controller, makeEncoder(), closed, 'hello'); - expect(closed.value).toBe(true); - }); - - test('logs unexpected non-TypeError errors', () => { - const { controller, makeEnqueueThrow } = makeController(); - const err = new Error('something weird'); - makeEnqueueThrow(err); - sendSSE(controller, makeEncoder(), { value: false }, 'hello'); - expect(console.error).toHaveBeenCalledWith( - 'Unexpected error during SSE enqueue:', - err, - ); - }); - - test('propagates non-close TypeError (genuine bug)', () => { - const { controller, makeEnqueueThrow } = makeController(); - const err = new TypeError('argument must be a string'); - makeEnqueueThrow(err); - expect(() => { - sendSSE(controller, makeEncoder(), { value: false }, 'hello'); - }).toThrow(err); - }); -}); diff --git a/agent/src/lib/sse-stream.ts b/agent/src/lib/sse-stream.ts new file mode 100644 index 00000000..0f735ad4 --- /dev/null +++ b/agent/src/lib/sse-stream.ts @@ -0,0 +1,121 @@ +/** + * Single owner of the SSE wire protocol for the agent: headers, initial flush, + * heartbeat, `data:`/`event:` frame grammar, and abort/enqueue-failure teardown. + * Every SSE route builds its `Response` through this so those facts live in one + * place instead of being re-derived per route. + * + * Deliberate duplicate of the web app's `src/lib/sse/create-sse-stream.ts`: the + * agent ships as its own package with its own lockfile and cannot import web + * code, so the seam exists once on each side of that split. + */ + +// Quiet streams go silent past idle timeouts (Bun's HTTP idleTimeout defaults +// to 10s), causing consumer reconnect churn; a periodic comment keeps them warm. +const DEFAULT_HEARTBEAT_MS = 5000; + +/** Frame-writing surface handed to `onStart`; no caller writes `\n\n` or `data: ` by hand. */ +export interface SseEmitter { + /** Writes a `data: \n\n` frame. */ + data(payload: unknown): void; + /** Writes an `event: \ndata: \n\n` frame. */ + event(name: string, payload: unknown): void; + /** Writes a chunk verbatim (a pre-formatted frame string, or raw bytes piped from an upstream SSE source). */ + raw(chunk: string | Uint8Array): void; + /** Ends the stream now (clears heartbeat, runs `onStart` cleanup, closes controller). */ + close(): void; +} + +export type SseCleanup = () => void; + +export type SseOnStart = ( + emit: SseEmitter, + signal: AbortSignal, +) => void | SseCleanup | Promise; + +export interface CreateSseStreamOptions { + /** Called once per request after the flush and heartbeat are armed. May return a cleanup, run once at teardown. */ + onStart: SseOnStart; + /** Heartbeat cadence in ms. Defaults to 5000. */ + heartbeatMs?: number; +} + +/** True for the TypeError Web Streams throws on enqueue/close after the controller already closed. */ +export function isCloseRelatedError(err: unknown): boolean { + return err instanceof TypeError && /closed/i.test(err.message); +} + +export function createSseStream(request: Request, opts: CreateSseStreamOptions): Response { + const { onStart, heartbeatMs = DEFAULT_HEARTBEAT_MS } = opts; + const encoder = new TextEncoder(); + let closed = false; + let cleanup: SseCleanup | undefined; + + const stream = new ReadableStream({ + async start(controller) { + let heartbeatTimer: ReturnType | undefined; + + const teardown = () => { + if (closed) return; + closed = true; + if (heartbeatTimer) clearInterval(heartbeatTimer); + cleanup?.(); + try { + controller.close(); + } catch {} + }; + + // Deferred build lets a JSON.stringify failure (circular ref, BigInt) hit the same catch as an enqueue failure. + const write = (build: () => string | Uint8Array) => { + if (closed) return; + try { + const value = build(); + controller.enqueue(typeof value === 'string' ? encoder.encode(value) : value); + } catch (err) { + if (!isCloseRelatedError(err)) { + console.error('Unexpected error during SSE enqueue:', err); + } + teardown(); + } + }; + + const emit: SseEmitter = { + data: (payload) => write(() => `data: ${JSON.stringify(payload)}\n\n`), + event: (name, payload) => write(() => `event: ${name}\ndata: ${JSON.stringify(payload)}\n\n`), + raw: (chunk) => write(() => chunk), + close: teardown, + }; + + // Flushes response headers immediately so clients don't stall on the first byte. + write(() => ': ok\n\n'); + heartbeatTimer = setInterval(() => write(() => ':\n\n'), heartbeatMs); + + // Registered before onStart so an abort mid-setup tears down right away. + request.signal.addEventListener('abort', teardown); + + let result: void | SseCleanup = undefined; + try { + result = await onStart(emit, request.signal); + } catch (err) { + console.error('Unexpected error in SSE onStart handler:', err); + teardown(); + } + + if (closed) { + // Teardown already ran mid-setup (abort, or onStart called emit.close()); run its cleanup now. + result?.(); + } else { + cleanup = result ?? undefined; + // onStart can resolve after an abort mid-setup; recheck so a subscription doesn't leak past disconnect. + if (request.signal.aborted) teardown(); + } + }, + }); + + return new Response(stream, { + headers: { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + }, + }); +} diff --git a/agent/src/lib/sse-utils.ts b/agent/src/lib/sse-utils.ts deleted file mode 100644 index 944ea4bd..00000000 --- a/agent/src/lib/sse-utils.ts +++ /dev/null @@ -1,21 +0,0 @@ -/** Enqueue an SSE data event, silently swallowing enqueue-after-close TypeError. */ -export function sendSSE( - controller: ReadableStreamDefaultController, - encoder: TextEncoder, - closed: { value: boolean }, - data: string, -): void { - if (closed.value) return; - try { - controller.enqueue(encoder.encode(`data: ${data}\n\n`)); - } catch (err) { - if (err instanceof TypeError && /closed/i.test(err.message)) { - // Controller was closed between our check and enqueue. Mark closed so - // subsequent sends short-circuit without another enqueue attempt. - closed.value = true; - return; - } - if (err instanceof TypeError) throw err; - console.error('Unexpected error during SSE enqueue:', err); - } -} diff --git a/agent/src/lib/test/sse-test-utils.ts b/agent/src/lib/test/sse-test-utils.ts new file mode 100644 index 00000000..fac63203 --- /dev/null +++ b/agent/src/lib/test/sse-test-utils.ts @@ -0,0 +1,44 @@ +/** Read chunks from an SSE response until the predicate is satisfied or the timeout elapses. */ +export async function readUntil( + response: Response, + predicate: (accumulated: string) => boolean, + timeoutMs = 3000, +): Promise { + let text = ''; + const reader = response.body!.getReader(); + const decoder = new TextDecoder(); + // One deadline for the whole read, not one per chunk: a per-chunk timer is reset + // by every heartbeat frame, so a predicate that never matches would never fire it. + let timeoutId: ReturnType | undefined; + const deadline = new Promise((_, reject) => { + timeoutId = setTimeout( + () => reject(new Error(`readUntil timed out after ${timeoutMs}ms`)), + timeoutMs, + ); + }); + + try { + while (true) { + const { done, value } = await Promise.race([reader.read(), deadline]); + if (done) break; + text += decoder.decode(value, { stream: true }); + if (predicate(text)) break; + } + } finally { + clearTimeout(timeoutId); + try { + await reader.cancel(); + } catch { + // cancel() rejects when the stream already errored, which several tests induce + } + } + return text; +} + +/** Parse the `data:` frames out of SSE text, ignoring the seam's flush and heartbeat comments. */ +export function parseDataFrames(text: string): any[] { + return text + .split('\n\n') + .filter((frame) => frame.startsWith('data: ')) + .map((frame) => JSON.parse(frame.slice('data: '.length))); +} diff --git a/agent/src/routes/containers-events.ts b/agent/src/routes/containers-events.ts index bae3ab2c..ae968940 100644 --- a/agent/src/routes/containers-events.ts +++ b/agent/src/routes/containers-events.ts @@ -1,7 +1,7 @@ import type Dockerode from 'dockerode'; import { subscribe as broadcasterSubscribe } from '../lib/docker-events-broadcaster'; import type { MinimalContainerInfo, BroadcasterEvent } from '../lib/docker-events-broadcaster'; -import { sendSSE } from '../lib/sse-utils'; +import { createSseStream } from '../lib/sse-stream'; import type { AgentContainerEvent, ContainerState, @@ -68,32 +68,10 @@ function toUpdateContainer(c: MinimalContainerInfo): InventoryUpdateContainer { * @param docker - Dockerode client used to interact with the Docker daemon * @param request - The HTTP request; its abort signal triggers cleanup */ -export async function handleContainerEvents(docker: Dockerode, request: Request): Promise { - const encoder = new TextEncoder(); - const closed = { value: false }; - let unsubscribe: (() => void) | null = null; - let heartbeat: ReturnType | null = null; - - const stream = new ReadableStream({ - async start(controller) { - request.signal.addEventListener('abort', () => { - closed.value = true; - if (heartbeat) clearInterval(heartbeat); - unsubscribe?.(); - try { - controller.close(); - } catch {} - }); - - if (request.signal.aborted) { - closed.value = true; - try { - controller.close(); - } catch {} - return; - } - - const sub = await broadcasterSubscribe(docker, (event: BroadcasterEvent) => { +export function handleContainerEvents(docker: Dockerode, request: Request): Response { + return createSseStream(request, { + onStart: async (emit) => + broadcasterSubscribe(docker, (event: BroadcasterEvent) => { let message: AgentContainerEvent; if (event.op === 'init') { @@ -110,43 +88,7 @@ export async function handleContainerEvents(docker: Dockerode, request: Request) message = { op: 'destroy', containerId: event.containerId }; } - sendSSE(controller, encoder, closed, JSON.stringify(message)); - }); - - // If abort fired while subscribe was pending, the abort handler saw a - // null unsubscribe. Tear down the late subscriber ourselves. - if (closed.value) { - sub(); - return; - } - - unsubscribe = sub; - - // The inventory stream only emits on container state changes, so a quiet - // host stays silent past the agent's HTTP idleTimeout (Bun default 10s), - // which drops the socket and forces the worker into a reconnect loop. A - // comment heartbeat keeps it warm; mirrors the logs route. 5s stays under - // typical Bun/proxy idle defaults. - const hb = setInterval(() => { - if (closed.value) { - clearInterval(hb); - return; - } - try { - controller.enqueue(encoder.encode(':\n\n')); - } catch { - clearInterval(hb); - } - }, 5_000); - heartbeat = hb; - }, - }); - - return new Response(stream, { - headers: { - 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', - Connection: 'keep-alive', - }, + emit.data(message); + }), }); } diff --git a/agent/src/routes/logs.ts b/agent/src/routes/logs.ts index e13c0622..396316a9 100644 --- a/agent/src/routes/logs.ts +++ b/agent/src/routes/logs.ts @@ -1,5 +1,6 @@ import type { Readable } from 'node:stream'; import type Dockerode from 'dockerode'; +import { createSseStream } from '../lib/sse-stream'; import { isContainerGone } from './stats'; /** @@ -24,23 +25,14 @@ export function handleLogStream( containerId: string, request: Request ): Response { - let closed = false; - let activeStream: Readable | null = null; - const encoder = new TextEncoder(); - - const stream = new ReadableStream({ - async start(controller) { - request.signal.addEventListener('abort', () => { - closed = true; + return createSseStream(request, { + onStart: async (emit, signal) => { + let activeStream: Readable | null = null; + const destroyActiveStream = () => { if (typeof activeStream?.destroy === 'function') { activeStream.destroy(); } - try { - controller.close(); - } catch { - // controller already closed - } - }); + }; try { const container = docker.getContainer(containerId); @@ -50,7 +42,7 @@ export function handleLogStream( let muxedRemainder: Buffer = Buffer.alloc(0); let lastTimestamp: string | null = null; - /** Process a chunk and enqueue parsed lines. Returns parsed lines for timestamp tracking. */ + /** Process a chunk and emit parsed lines. Returns parsed lines for timestamp tracking. */ function processChunk(chunk: Buffer): LogLine[] { let lines: LogLine[]; if (isTty) { @@ -64,16 +56,8 @@ export function handleLogStream( muxedRemainder = result.remainder; } - try { - for (const line of lines) { - if (!closed) { - controller.enqueue( - encoder.encode(`data: ${JSON.stringify(line)}\n\n`) - ); - } - } - } catch (err) { - if (!(err instanceof TypeError)) console.error('Unexpected error enqueuing log line:', err); + for (const line of lines) { + emit.data(line); } return lines; @@ -97,18 +81,14 @@ export function handleLogStream( ? backlogResult : Buffer.from(String(backlogResult)); - if (!closed) { - const lines = processChunk(backlogBuffer); - for (const line of lines) { - const ts = extractTimestamp(line.text); - if (ts) lastTimestamp = ts; - } + for (const line of processChunk(backlogBuffer)) { + const ts = extractTimestamp(line.text); + if (ts) lastTimestamp = ts; } - if (closed) return; + if (signal.aborted) return destroyActiveStream; - // Emit backlog_done separator - controller.enqueue(encoder.encode('event: backlog_done\ndata: {}\n\n')); + emit.event('backlog_done', {}); // Reset muxed remainder for the live phase muxedRemainder = Buffer.alloc(0); @@ -121,16 +101,8 @@ export function handleLogStream( : fallbackSinceSeconds; /** Live phase: follow new logs from the last backlog timestamp. */ - let liveStream: Readable | null = null; - const onAbortDuringAwait = () => { - if (liveStream && typeof liveStream.destroy === 'function') { - liveStream.destroy(); - } - }; - request.signal.addEventListener('abort', onAbortDuringAwait); - // @types/dockerode 4.0.1 types logs() stream result as any; cast required to use Readable API - liveStream = (await container.logs({ + const liveStream = (await container.logs({ follow: true, stdout: true, stderr: true, @@ -138,93 +110,43 @@ export function handleLogStream( timestamps: true, })) as unknown as Readable; - request.signal.removeEventListener('abort', onAbortDuringAwait); activeStream = liveStream; - if (request.signal.aborted) { + if (signal.aborted) { liveStream.destroy(); - return; + return destroyActiveStream; } - // Send periodic heartbeat to prevent idle socket timeouts from killing - // the connection. Must be shorter than any intermediary timeout (Bun - // fetch, Nitro, reverse proxies); 5 s is safe for typical defaults. - controller.enqueue(encoder.encode(':\n\n')); - const heartbeatInterval = setInterval(() => { - if (closed) { clearInterval(heartbeatInterval); return; } - try { - controller.enqueue(encoder.encode(':\n\n')); - } catch { - clearInterval(heartbeatInterval); - } - }, 5_000); - liveStream.on('data', (chunk: Buffer) => { - if (closed) return; processChunk(chunk); }); liveStream.on('end', () => { - clearInterval(heartbeatInterval); - if (!closed) { - // Signal to the client that the stream ended cleanly (container stopped) - // so it can suppress the reconnect loop. - controller.enqueue(encoder.encode('event: stream_end\ndata: {}\n\n')); - closed = true; - controller.close(); - } + // Signal to the client that the stream ended cleanly (container stopped) + // so it can suppress the reconnect loop. + emit.event('stream_end', {}); + emit.close(); }); liveStream.on('error', (error: Error) => { - clearInterval(heartbeatInterval); console.error(`Log stream error for container ${containerId}:`, error); - if (!closed) { - controller.enqueue( - encoder.encode( - `event: error\ndata: ${JSON.stringify({ error: error.message })}\n\n` - ) - ); - closed = true; - controller.close(); - } + emit.event('error', { error: error.message }); + emit.close(); }); + + return destroyActiveStream; } catch (error) { if (error instanceof Error && isContainerGone(error)) { - try { - controller.enqueue( - encoder.encode( - `event: error\ndata: ${JSON.stringify({ error: 'Container not found', gone: true })}\n\n` - ) - ); - closed = true; - controller.close(); - } catch { /* controller already closed */ } - return; - } - console.error(`Failed to start log stream for container ${containerId}:`, error); - const msg = error instanceof Error ? error.message : String(error); - try { - controller.enqueue( - encoder.encode( - `event: error\ndata: ${JSON.stringify({ error: msg })}\n\n` - ) - ); - closed = true; - controller.close(); - } catch { - // controller already closed + emit.event('error', { error: 'Container not found', gone: true }); + } else { + console.error(`Failed to start log stream for container ${containerId}:`, error); + emit.event('error', { error: error instanceof Error ? error.message : String(error) }); } + emit.close(); + return destroyActiveStream; } }, }); - - return new Response(stream, { - headers: { - 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', - Connection: 'keep-alive', - }, - }); } /** diff --git a/agent/src/routes/stats.ts b/agent/src/routes/stats.ts index b3199152..b467e975 100644 --- a/agent/src/routes/stats.ts +++ b/agent/src/routes/stats.ts @@ -1,5 +1,6 @@ import type { Readable } from 'node:stream'; import type Dockerode from 'dockerode'; +import { createSseStream, type SseEmitter } from '../lib/sse-stream'; const DEFAULT_REFRESH_INTERVAL_MS = 60_000; const DEFAULT_POLL_INTERVAL_MS = 5_000; @@ -149,26 +150,15 @@ export function computeMetrics( /** Shared mutable state for a single SSE stats session. */ interface StreamContext { - closed: boolean; - readonly encoder: TextEncoder; + /** Flipped by the stream's cleanup, so the poll loop stops once the consumer is gone. */ + stopped: boolean; + readonly emit: SseEmitter; readonly containerStreams: Map; - readonly controller: ReadableStreamDefaultController; -} - -/** Enqueue an SSE message, silently swallowing enqueue-after-close TypeError. */ -function sendSSE(ctx: StreamContext, data: string, event?: string): void { - if (ctx.closed) return; - try { - const prefix = event ? `event: ${event}\n` : ''; - ctx.controller.enqueue(ctx.encoder.encode(`${prefix}data: ${data}\n\n`)); - } catch (err) { - if (!(err instanceof TypeError)) console.error('Unexpected error during SSE enqueue:', err); - } } /** Enqueue a JSON error payload as an SSE event. */ function sendErrorSSE(ctx: StreamContext, payload: Record, event = 'container-error'): void { - sendSSE(ctx, JSON.stringify(payload), event); + ctx.emit.event(event, payload); } /** Destroy all tracked container streams and clear the map. */ @@ -224,7 +214,7 @@ function openContainerStream( container.stats({ stream: true }).then((statsStream) => { // @types/dockerode 4.0.1 types stats() stream result as any; cast required to use Readable API const readable = statsStream as unknown as Readable; - if (ctx.closed) { + if (ctx.stopped) { if (typeof readable.destroy === 'function') readable.destroy(); return; } @@ -232,9 +222,9 @@ function openContainerStream( let buffer = ''; readable.on('data', (chunk: Buffer) => { - if (ctx.closed) return; + if (ctx.stopped) return; buffer = parseStatsChunks(buffer, chunk, (stats) => { - sendSSE(ctx, JSON.stringify(computeMetrics(id, name, image, stats as Record, prevFrames))); + ctx.emit.data(computeMetrics(id, name, image, stats as Record, prevFrames)); }, () => { console.error(`Malformed stats JSON from container ${id}, skipping frame`); }); @@ -290,16 +280,7 @@ function reconcileContainers( if (!previousIds.has(c.Id)) openContainerStream(ctx, docker, c, prevFrames); } - sendSSE(ctx, JSON.stringify({ ids: [...currentIds] }), 'containers'); -} - -/** Try to close a ReadableStream controller, ignoring errors if already closed. */ -function tryCloseController(controller: ReadableStreamDefaultController): void { - try { - controller.close(); - } catch (err) { - if (!(err instanceof TypeError)) console.error('Unexpected error closing controller:', err); - } + ctx.emit.event('containers', { ids: [...currentIds] }); } /** @@ -316,7 +297,7 @@ async function tryRefreshContainers( ): Promise<{ containers: Dockerode.ContainerInfo[]; shouldBreak: boolean } | null> { try { const current = await docker.listContainers({ all: false }); - if (ctx.closed) return { containers: current, shouldBreak: true }; + if (ctx.stopped) return { containers: current, shouldBreak: true }; reconcileContainers(ctx, docker, previous, current, prevFrames); return { containers: current, shouldBreak: false }; } catch (error) { @@ -327,7 +308,7 @@ async function tryRefreshContainers( if (count >= maxFailures) { console.error('Max consecutive refresh failures reached, closing stats stream'); - sendSSE(ctx, JSON.stringify({ error: 'Docker daemon unreachable, stream closed' }), 'error'); + ctx.emit.event('error', { error: 'Docker daemon unreachable, stream closed' }); } return null; } @@ -343,7 +324,7 @@ async function runStatsLoop( ): Promise { const prevFrames = new Map(); let containers = await docker.listContainers({ all: false }); - if (ctx.closed) return; + if (ctx.stopped) return; let lastRefresh = Date.now(); let consecutiveFailures = 0; @@ -351,11 +332,11 @@ async function runStatsLoop( openContainerStream(ctx, docker, c, prevFrames); } - sendSSE(ctx, JSON.stringify({ ids: containers.map(c => c.Id) }), 'containers'); + ctx.emit.event('containers', { ids: containers.map(c => c.Id) }); - while (!ctx.closed) { + while (!ctx.stopped) { await new Promise((resolve) => setTimeout(resolve, pollIntervalMs)); - if (ctx.closed) break; + if (ctx.stopped) break; const now = Date.now(); if (now - lastRefresh < refreshIntervalMs) continue; @@ -373,8 +354,7 @@ async function runStatsLoop( } destroyAllStreams(ctx.containerStreams); - ctx.closed = true; - tryCloseController(ctx.controller); + ctx.emit.close(); } /** @@ -411,37 +391,27 @@ export function handleStatsStream( const pollIntervalMs = options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS; const maxConsecutiveFailures = options.maxConsecutiveFailures ?? DEFAULT_MAX_CONSECUTIVE_FAILURES; - const stream = new ReadableStream({ - async start(controller) { + return createSseStream(request, { + onStart: (emit) => { const ctx: StreamContext = { - closed: false, - encoder: new TextEncoder(), + stopped: false, + emit, containerStreams: new Map(), - controller, }; - request.signal.addEventListener('abort', () => { - ctx.closed = true; - destroyAllStreams(ctx.containerStreams); - tryCloseController(controller); - }); - - try { - await runStatsLoop(ctx, docker, pollIntervalMs, refreshIntervalMs, maxConsecutiveFailures); - } catch (error) { + // Not awaited: the cleanup below has to be registered before the poll + // loop blocks, or a teardown mid-loop would never stop it. + void runStatsLoop(ctx, docker, pollIntervalMs, refreshIntervalMs, maxConsecutiveFailures).catch((error) => { console.error('Failed to start stats stream:', error); const msg = error instanceof Error ? error.message : String(error); - sendSSE(ctx, JSON.stringify({ error: msg }), 'error'); - tryCloseController(controller); - } - }, - }); + emit.event('error', { error: msg }); + emit.close(); + }); - return new Response(stream, { - headers: { - 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', - Connection: 'keep-alive', + return () => { + ctx.stopped = true; + destroyAllStreams(ctx.containerStreams); + }; }, }); } diff --git a/agent/src/routes/zfs.ts b/agent/src/routes/zfs.ts index 491a492f..c59d2059 100644 --- a/agent/src/routes/zfs.ts +++ b/agent/src/routes/zfs.ts @@ -1,5 +1,68 @@ +import { createSseStream, type SseEmitter } from '../lib/sse-stream'; import type { ZfsCapabilities } from '../lib/zfs-capabilities'; +async function readLines( + stream: ReadableStream, + isStopped: () => boolean, + onLine: (line: string) => void, +): Promise { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + + while (!isStopped()) { + const { done, value } = await reader.read(); + if (done) break; + + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split('\n'); + buffer = lines.pop() ?? ''; + + for (const line of lines) { + if (isStopped()) return; + if (line.trim()) onLine(line); + } + } +} + +async function pumpZpoolOutput( + emit: SseEmitter, + stdout: ReadableStream, + isStopped: () => boolean, +): Promise { + try { + await readLines(stdout, isStopped, (line) => { + emit.data({ line, timestamp: Date.now() }); + }); + } catch (err) { + if (!isStopped()) { + console.error('ZFS stats stream error:', err); + } + } finally { + emit.close(); + } +} + +/** + * Drain the subprocess's stderr into the agent log. Reading it is not optional: + * this process lives for the whole SSE session, so an unread pipe fills its OS + * buffer and blocks zpool's stdout, stalling the stream it is meant to feed. + */ +async function drainZpoolErrors( + stderr: ReadableStream, + isStopped: () => boolean, +): Promise { + try { + await readLines(stderr, isStopped, (line) => { + console.error('zpool iostat:', line); + }); + } catch (err) { + if (!isStopped()) { + console.error('ZFS stats stderr read error:', err); + } + } +} + /** * GET /zfs/stats/stream: SSE endpoint that streams `zpool iostat -v 1` output. * @@ -17,80 +80,25 @@ export function handleZfsStatsStream( ); } - const encoder = new TextEncoder(); - - return new Response( - new ReadableStream({ - async start(controller) { - const proc = Bun.spawn(['zpool', 'iostat', '-v', '1'], { - stdout: 'pipe', - stderr: 'pipe', - }); - - let closed = false; - - request.signal.addEventListener('abort', () => { - closed = true; - proc.kill(); - try { - controller.close(); - } catch { - // controller already closed - } - }); - - try { - const reader = proc.stdout.getReader(); - const decoder = new TextDecoder(); - let buffer = ''; - - while (!closed) { - const { done, value } = await reader.read(); - if (done) break; - - buffer += decoder.decode(value, { stream: true }); - const lines = buffer.split('\n'); - buffer = lines.pop() ?? ''; - - for (const line of lines) { - if (closed) break; - if (!line.trim()) continue; - - try { - controller.enqueue( - encoder.encode(`data: ${JSON.stringify({ line, timestamp: Date.now() })}\n\n`), - ); - } catch { - closed = true; - break; - } - } - } - } catch (err) { - if (!closed) { - console.error('ZFS stats stream error:', err); - } - } finally { - proc.kill(); - if (!closed) { - closed = true; - try { - controller.close(); - } catch { - // controller already closed - } - } - } - }, - }), - { - headers: { - 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', - Connection: 'keep-alive', - }, + return createSseStream(request, { + onStart: (emit) => { + const proc = Bun.spawn(['zpool', 'iostat', '-v', '1'], { + stdout: 'pipe', + stderr: 'pipe', + }); + + let stopped = false; + // Not awaited: the cleanup below has to be registered before the read + // loops block, or a teardown mid-loop would never kill the subprocess. + void pumpZpoolOutput(emit, proc.stdout, () => stopped); + void drainZpoolErrors(proc.stderr, () => stopped); + + return () => { + stopped = true; + proc.kill(); + }; }, - ); + }); } /**