diff --git a/server/ai/mcp/editorBridge.ts b/server/ai/mcp/editorBridge.ts index e1bc46edf..d6aafbdfa 100644 --- a/server/ai/mcp/editorBridge.ts +++ b/server/ai/mcp/editorBridge.ts @@ -28,6 +28,7 @@ interface EditorBridgeEntry { } export type EditorBridgeScope = 'site' | 'content' +const STREAM_LEASE_MS = 120_000 const byUser = new Map>() @@ -63,11 +64,13 @@ export function createEditorBridgeStream( let bridgeId = '' let destroyBridge = (): void => {} let heartbeat: ReturnType | null = null + let lease: ReturnType | null = null const cleanup = () => { if (closed) return closed = true if (heartbeat) clearInterval(heartbeat) + if (lease) clearTimeout(lease) signal.removeEventListener('abort', cleanup) destroyBridge() @@ -119,6 +122,9 @@ export function createEditorBridgeStream( cleanup() } }, 25_000) + // Bound orphan lifetime when a proxy fails to propagate a closed + // downstream connection. The client reconnect loop restores the bridge. + lease = setTimeout(cleanup, STREAM_LEASE_MS) if (signal.aborted) cleanup() else signal.addEventListener('abort', cleanup, { once: true }) diff --git a/server/handlers/cms/plugins/__tests__/events.test.ts b/server/handlers/cms/plugins/__tests__/events.test.ts new file mode 100644 index 000000000..ac1962dfb --- /dev/null +++ b/server/handlers/cms/plugins/__tests__/events.test.ts @@ -0,0 +1,45 @@ +import { describe, expect, it } from 'bun:test' +import { __pluginEventSubscriberCountForTesting } from '../../../../plugins/eventBroadcaster' +import { handlePluginEventsStream } from '../events' + +const decoder = new TextDecoder() + +async function connectPluginEventStream(): Promise<{ + requestController: AbortController + reader: ReadableStreamDefaultReader +}> { + const requestController = new AbortController() + const response = handlePluginEventsStream(new Request( + 'http://localhost/admin/api/cms/plugins/events', + { signal: requestController.signal }, + )) + const reader = response.body!.getReader() + const first = await reader.read() + expect(decoder.decode(first.value)).toContain('event: ping') + return { requestController, reader } +} + +describe('plugin event stream', () => { + it('cleans up when the response consumer cancels without aborting the request', async () => { + const { requestController, reader } = await connectPluginEventStream() + + try { + expect(__pluginEventSubscriberCountForTesting()).toBe(1) + await reader.cancel() + expect(requestController.signal.aborted).toBe(false) + expect(__pluginEventSubscriberCountForTesting()).toBe(0) + } finally { + await reader.cancel().catch(() => {}) + requestController.abort() + } + }) + + it('leaves no subscribers across repeated connect and cancel cycles', async () => { + for (let index = 0; index < 100; index += 1) { + const { requestController, reader } = await connectPluginEventStream() + await reader.cancel() + requestController.abort() + expect(__pluginEventSubscriberCountForTesting()).toBe(0) + } + }) +}) diff --git a/server/handlers/cms/plugins/events.ts b/server/handlers/cms/plugins/events.ts index a45a9a96c..918c96a84 100644 --- a/server/handlers/cms/plugins/events.ts +++ b/server/handlers/cms/plugins/events.ts @@ -15,51 +15,75 @@ * - Initial `event: ping` keeps proxies (vite, nginx) from idle-closing * the long-lived connection. Followed by a periodic heartbeat every * 30s for the same reason. - * - On `req.signal` abort (tab closed, EventSource paused), we - * unsubscribe from the broadcaster and stop the heartbeat. No leaks. + * - Request abort and response-stream cancellation both unsubscribe from + * the broadcaster and stop the heartbeat. Bun reports a disconnected + * response consumer through `ReadableStream.cancel()` without necessarily + * aborting the server Request. * - The stream never ends voluntarily — clients reconnect via the * standard EventSource auto-reconnect on transport errors. */ import { subscribePluginEvents } from '../../../plugins/eventBroadcaster' import { methodNotAllowed } from '../../../http' +const STREAM_LEASE_MS = 120_000 + export function handlePluginEventsStream(req: Request): Response { if (req.method !== 'GET') return methodNotAllowed() const encoder = new TextEncoder() + let closeStream: (() => void) | null = null const stream = new ReadableStream({ start(controller) { + let closed = false + let unsubscribe = (): void => {} + let heartbeat: ReturnType | null = null + let lease: ReturnType | null = null + + const cleanup = () => { + if (closed) return + closed = true + if (heartbeat) clearInterval(heartbeat) + if (lease) clearTimeout(lease) + req.signal.removeEventListener('abort', cleanup) + unsubscribe() + try { controller.close() } catch { /* already closed or cancelled */ } + } + closeStream = cleanup + function send(payload: string): void { + if (closed) return try { controller.enqueue(encoder.encode(payload)) } catch { - // Stream already closed (client gone). Listeners + heartbeat - // are torn down via the abort handler below. + cleanup() } } - // Initial ping so the client sees a successful connection immediately, - // even before the first real event arrives. - send(`event: ping\ndata: connected\n\n`) - // Subscribe to the broadcaster — every event becomes one SSE message. // SSE requires `data:` lines + a terminating blank line. - const unsubscribe = subscribePluginEvents((event) => { + unsubscribe = subscribePluginEvents((event) => { send(`event: ${event.kind}\ndata: ${JSON.stringify(event)}\n\n`) }) // Heartbeat keeps proxies + the EventSource itself happy. SSE comments // (`: heartbeat`) are ignored by the client but reset idle timers. - const heartbeat = setInterval(() => { + heartbeat = setInterval(() => { send(`: heartbeat\n\n`) }, 30_000) + // Bound orphan lifetime even when an intermediary fails to propagate a + // downstream disconnect. EventSource reconnects automatically. + lease = setTimeout(cleanup, STREAM_LEASE_MS) - // Tear down when the client disconnects. - req.signal.addEventListener('abort', () => { - unsubscribe() - clearInterval(heartbeat) - try { controller.close() } catch { /* already closed */ } - }) + if (req.signal.aborted) cleanup() + else req.signal.addEventListener('abort', cleanup, { once: true }) + + // Initial ping so the client sees a successful connection immediately, + // even before the first real event arrives. + send(`event: ping\ndata: connected\n\n`) + }, + cancel() { + closeStream?.() + closeStream = null }, }) diff --git a/server/plugins/eventBroadcaster.ts b/server/plugins/eventBroadcaster.ts index adf348a18..06c8815d9 100644 --- a/server/plugins/eventBroadcaster.ts +++ b/server/plugins/eventBroadcaster.ts @@ -56,3 +56,6 @@ export function broadcastPluginEvent(event: PluginEvent): void { } } +export function __pluginEventSubscriberCountForTesting(): number { + return listeners.size +}