diff --git a/apps/sim/hooks/queries/mcp.ts b/apps/sim/hooks/queries/mcp.ts index cbfb00cb34c..8cb44396489 100644 --- a/apps/sim/hooks/queries/mcp.ts +++ b/apps/sim/hooks/queries/mcp.ts @@ -27,6 +27,10 @@ import { testMcpServerConnectionContract, updateMcpServerContract, } from '@/lib/api/contracts/mcp' +import { + createRotatingEventSource, + type RotatingEventSourceConnection, +} from '@/lib/events/rotating-event-source' import { sanitizeForHttp, sanitizeHeaders } from '@/lib/mcp/shared' import type { McpAuthType, @@ -541,7 +545,7 @@ export function useStoredMcpTools(workspaceId: string, options?: { enabled?: boo */ const SSE_KEY = '__mcp_sse_connections' as const -type SseEntry = { source: EventSource; refs: number } +type SseEntry = { source: RotatingEventSourceConnection; refs: number } const sseConnections: Map = ((globalThis as Record)[SSE_KEY] as Map) ?? @@ -576,38 +580,27 @@ export function useMcpToolsEvents(workspaceId: string) { let entry = sseConnections.get(workspaceId) if (!entry) { - const source = new EventSource(`/api/mcp/events?workspaceId=${workspaceId}`) - - source.addEventListener('tools_changed', (e) => { - let serverId: string | undefined - try { - const parsed = JSON.parse((e as MessageEvent).data) as { serverId?: string } - serverId = parsed.serverId - } catch { - // Non-JSON payload → workspace-wide fallback. - } - invalidate(serverId) - }) - - // EventSource fires `onopen` on the initial connect and on every auto-reconnect. Re-sync - // the workspace whenever we could have missed a `tools_changed` event: on any reconnect, - // on the first open of a RE-subscription (leaving the tab tears the connection down), and - // on a first open that only succeeded after an earlier connection error (the initial tools - // query may have failed during that gap and won't retry itself). Skip only a clean first - // subscription — the queries fetch fresh on their own initial mount. const isResubscribe = sseEverSubscribed.has(workspaceId) sseEverSubscribed.add(workspaceId) - let opened = false - let erroredBeforeOpen = false - source.onopen = () => { - if (opened || isResubscribe || erroredBeforeOpen) invalidate() - opened = true - } - - source.onerror = () => { - if (!opened) erroredBeforeOpen = true - logger.warn(`SSE connection error for workspace ${workspaceId}`) - } + const source = createRotatingEventSource({ + url: `/api/mcp/events?workspaceId=${encodeURIComponent(workspaceId)}`, + events: { + tools_changed: (event) => { + let serverId: string | undefined + try { + const parsed = JSON.parse((event as MessageEvent).data) as { serverId?: string } + serverId = parsed.serverId + } catch {} + invalidate(serverId) + }, + }, + onOpen: (reason) => { + if (reason === 'reconnect' || (reason === 'initial' && isResubscribe)) invalidate() + }, + onError: () => { + logger.warn(`SSE connection error for workspace ${workspaceId}`) + }, + }) entry = { source, refs: 0 } sseConnections.set(workspaceId, entry) diff --git a/apps/sim/hooks/use-mothership-chat-events.ts b/apps/sim/hooks/use-mothership-chat-events.ts index 175891a4dbd..02079c3a3a9 100644 --- a/apps/sim/hooks/use-mothership-chat-events.ts +++ b/apps/sim/hooks/use-mothership-chat-events.ts @@ -5,6 +5,7 @@ import { useQueryClient } from '@tanstack/react-query' import { getLiveAssistantMessageId } from '@/lib/copilot/chat/effective-transcript' import { isChatEnabled } from '@/lib/core/config/env-flags' import { suspendDesktopChatScopes } from '@/lib/desktop/chat-scope' +import { createRotatingEventSource } from '@/lib/events/rotating-event-source' import { type MothershipChatHistory, mothershipChatKeys } from '@/hooks/queries/mothership-chats' const logger = createLogger('MothershipChatEvents') @@ -167,45 +168,31 @@ export function useMothershipChatEvents(workspaceId: string | undefined) { useEffect(() => { if (!workspaceId || !isChatEnabled) return - const eventSource = new EventSource( - `/api/mothership/events?workspaceId=${encodeURIComponent(workspaceId)}` - ) - - eventSource.addEventListener('task_status', (event) => { - handleMothershipChatStatusEvent( - queryClient, - workspaceId, - event instanceof MessageEvent ? event.data : undefined - ) - }) - - // `onopen` fires on the initial connect and on every auto-reconnect. Re-sync - // whenever a gap could have swallowed an event: on any reconnect, on the - // first open of a RE-subscription (switching workspace away and back tears - // the connection down, and the list/detail queries remount inside their - // stale times so they do not refetch on their own), and on a first open that - // only succeeded after an error (the initial queries may have failed during - // that gap and will not retry themselves). Skip only a clean first - // subscription — those queries fetch fresh on their own initial mount. const isResubscribe = everSubscribed.has(workspaceId) everSubscribed.add(workspaceId) - let opened = false - let erroredBeforeOpen = false - - eventSource.onopen = () => { - if (opened || isResubscribe || erroredBeforeOpen) { - resyncMothershipChatCaches(queryClient, workspaceId) - } - opened = true - } - - eventSource.onerror = () => { - if (!opened) erroredBeforeOpen = true - logger.warn(`SSE connection error for workspace ${workspaceId}`) - } + const connection = createRotatingEventSource({ + url: `/api/mothership/events?workspaceId=${encodeURIComponent(workspaceId)}`, + events: { + task_status: (event) => { + handleMothershipChatStatusEvent( + queryClient, + workspaceId, + event instanceof MessageEvent ? event.data : undefined + ) + }, + }, + onOpen: (reason) => { + if (reason === 'reconnect' || (reason === 'initial' && isResubscribe)) { + resyncMothershipChatCaches(queryClient, workspaceId) + } + }, + onError: () => { + logger.warn(`SSE connection error for workspace ${workspaceId}`) + }, + }) return () => { - eventSource.close() + connection.close() } }, [workspaceId, queryClient]) } diff --git a/apps/sim/lib/events/rotating-event-source.test.ts b/apps/sim/lib/events/rotating-event-source.test.ts new file mode 100644 index 00000000000..a3212838eb4 --- /dev/null +++ b/apps/sim/lib/events/rotating-event-source.test.ts @@ -0,0 +1,162 @@ +/** + * @vitest-environment node + */ + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { + createRotatingEventSource, + type EventSourceOpenReason, +} from '@/lib/events/rotating-event-source' + +class MockEventSource { + static readonly CONNECTING = 0 + static readonly OPEN = 1 + static readonly CLOSED = 2 + static instances: MockEventSource[] = [] + static failNextConstruction = false + + readonly listeners = new Map void>>() + readyState = MockEventSource.CONNECTING + onopen: (() => void) | null = null + onerror: (() => void) | null = null + + constructor(readonly url: string) { + if (MockEventSource.failNextConstruction) { + MockEventSource.failNextConstruction = false + throw new Error('construction failed') + } + MockEventSource.instances.push(this) + } + + addEventListener(eventName: string, listener: (event: Event) => void): void { + const listeners = this.listeners.get(eventName) ?? [] + listeners.push(listener) + this.listeners.set(eventName, listeners) + } + + close(): void { + this.readyState = MockEventSource.CLOSED + } + + open(): void { + this.readyState = MockEventSource.OPEN + this.onopen?.() + } + + error(): void { + this.readyState = MockEventSource.CONNECTING + this.onerror?.() + } + + emit(eventName: string): void { + for (const listener of this.listeners.get(eventName) ?? []) { + listener(new Event(eventName)) + } + } +} + +describe('createRotatingEventSource', () => { + beforeEach(() => { + MockEventSource.instances = [] + MockEventSource.failNextConstruction = false + vi.stubGlobal('EventSource', MockEventSource) + }) + + afterEach(() => { + vi.unstubAllGlobals() + }) + + it('keeps the current source open until its replacement connects', () => { + const reasons: EventSourceOpenReason[] = [] + const onMessage = vi.fn() + const connection = createRotatingEventSource({ + url: '/api/events', + events: { message: onMessage }, + onOpen: (reason) => reasons.push(reason), + }) + const first = MockEventSource.instances[0] + first.open() + first.emit('message') + + first.emit('rotate') + + expect(MockEventSource.instances).toHaveLength(2) + expect(first.readyState).toBe(MockEventSource.OPEN) + + const second = MockEventSource.instances[1] + second.open() + second.emit('message') + + expect(first.readyState).toBe(MockEventSource.CLOSED) + expect(second.readyState).toBe(MockEventSource.OPEN) + expect(reasons).toEqual(['initial', 'rotation']) + expect(onMessage).toHaveBeenCalledTimes(2) + connection.close() + }) + + it('classifies a replacement as reconnecting when the old source dropped first', () => { + const reasons: EventSourceOpenReason[] = [] + const onMessage = vi.fn() + const connection = createRotatingEventSource({ + url: '/api/events', + events: { message: onMessage }, + onOpen: (reason) => reasons.push(reason), + }) + const first = MockEventSource.instances[0] + first.open() + first.emit('rotate') + first.error() + + const second = MockEventSource.instances[1] + second.open() + second.emit('message') + + expect(reasons).toEqual(['initial', 'reconnect']) + expect(onMessage).toHaveBeenCalledTimes(1) + connection.close() + }) + + it('classifies an automatic EventSource recovery as a reconnect', () => { + const reasons: EventSourceOpenReason[] = [] + const connection = createRotatingEventSource({ + url: '/api/events', + events: {}, + onOpen: (reason) => reasons.push(reason), + }) + const source = MockEventSource.instances[0] + source.open() + source.error() + source.open() + + expect(reasons).toEqual(['initial', 'reconnect']) + connection.close() + }) + + it('closes both sources when disposed during rotation', () => { + const connection = createRotatingEventSource({ url: '/api/events', events: {} }) + const first = MockEventSource.instances[0] + first.open() + first.emit('rotate') + const second = MockEventSource.instances[1] + + connection.close() + + expect(first.readyState).toBe(MockEventSource.CLOSED) + expect(second.readyState).toBe(MockEventSource.CLOSED) + }) + + it('keeps the current source when opening its replacement fails', () => { + const onError = vi.fn() + const connection = createRotatingEventSource({ url: '/api/events', events: {}, onError }) + const first = MockEventSource.instances[0] + first.open() + MockEventSource.failNextConstruction = true + + first.emit('rotate') + + expect(first.readyState).toBe(MockEventSource.OPEN) + expect(MockEventSource.instances).toHaveLength(1) + expect(onError).toHaveBeenCalledTimes(1) + connection.close() + }) +}) diff --git a/apps/sim/lib/events/rotating-event-source.ts b/apps/sim/lib/events/rotating-event-source.ts new file mode 100644 index 00000000000..2bc4e4a58ba --- /dev/null +++ b/apps/sim/lib/events/rotating-event-source.ts @@ -0,0 +1,111 @@ +export type EventSourceOpenReason = 'initial' | 'reconnect' | 'rotation' + +interface RotatingEventSourceOptions { + url: string + events: Record void> + onOpen?: (reason: EventSourceOpenReason) => void + onError?: () => void +} + +export interface RotatingEventSourceConnection { + close(): void +} + +interface SourceState { + opened: boolean + erroredBeforeOpen: boolean +} + +/** + * Maintains one live EventSource and performs make-before-break rotation when + * the server emits `rotate`. The old source remains open until its replacement + * connects, so planned lifetime bounds do not create an event-delivery gap. + */ +export function createRotatingEventSource( + options: RotatingEventSourceOptions +): RotatingEventSourceConnection { + const sources = new Set() + const states = new WeakMap() + let current: EventSource | null = null + let replacement: EventSource | null = null + let closed = false + + const openSource = (isReplacement: boolean): EventSource => { + const source = new EventSource(options.url) + const state: SourceState = { opened: false, erroredBeforeOpen: false } + states.set(source, state) + sources.add(source) + + for (const [eventName, listener] of Object.entries(options.events)) { + source.addEventListener(eventName, listener) + } + + source.addEventListener('rotate', () => { + if (closed || source !== current || replacement) return + try { + replacement = openSource(true) + } catch { + replacement = null + options.onError?.() + } + }) + + source.onopen = () => { + if (closed) { + source.close() + sources.delete(source) + return + } + + if (isReplacement) { + if (source !== replacement) { + source.close() + sources.delete(source) + return + } + + const previous = current + const seamless = previous?.readyState === EventSource.OPEN + current = source + replacement = null + if (previous && previous !== source) { + previous.close() + sources.delete(previous) + } + state.opened = true + state.erroredBeforeOpen = false + options.onOpen?.(seamless ? 'rotation' : 'reconnect') + return + } + + const reason: EventSourceOpenReason = + state.opened || state.erroredBeforeOpen ? 'reconnect' : 'initial' + state.opened = true + state.erroredBeforeOpen = false + options.onOpen?.(reason) + } + + source.onerror = () => { + if (closed || !sources.has(source)) return + if (!state.opened) state.erroredBeforeOpen = true + options.onError?.() + } + + return source + } + + current = openSource(false) + + return { + close() { + if (closed) return + closed = true + for (const source of sources) { + source.close() + } + sources.clear() + current = null + replacement = null + }, + } +} diff --git a/apps/sim/lib/events/sse-endpoint.test.ts b/apps/sim/lib/events/sse-endpoint.test.ts index 242a9be1b7c..1d99d7cf4b8 100644 --- a/apps/sim/lib/events/sse-endpoint.test.ts +++ b/apps/sim/lib/events/sse-endpoint.test.ts @@ -8,23 +8,28 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { createWorkspaceSSE, HEARTBEAT_INTERVAL_MS, - MAX_CONNECTION_JITTER_MS, MAX_CONNECTION_MS, MAX_UNDRAINED_CHUNKS, + ROTATION_GRACE_MS, } from '@/lib/events/sse-endpoint' vi.mock('@/lib/workspaces/permissions/utils', () => permissionsMock) +vi.mock('@sim/utils/random', () => ({ randomFloat: () => 0 })) -const PAST_CEILING_MS = MAX_CONNECTION_MS + MAX_CONNECTION_JITTER_MS + HEARTBEAT_INTERVAL_MS +const PAST_ROTATION_MS = MAX_CONNECTION_MS +const PAST_ROTATION_CLOSE_MS = PAST_ROTATION_MS + ROTATION_GRACE_MS + HEARTBEAT_INTERVAL_MS /** Enough undrained heartbeats to trip the unread check, and no more. */ const PAST_UNREAD_MS = (MAX_UNDRAINED_CHUNKS + 2) * HEARTBEAT_INTERVAL_MS -async function openConnection(signal: AbortSignal = new AbortController().signal) { +async function openConnection( + signal: AbortSignal = new AbortController().signal, + subscriptions?: Array<{ subscribe: () => () => void }> +) { const unsubscribe = vi.fn() const handler = createWorkspaceSSE({ label: 'test', - subscriptions: [{ subscribe: () => unsubscribe }], + subscriptions: subscriptions ?? [{ subscribe: () => unsubscribe }], }) const request = new NextRequest(new URL('https://sim.test/api/test/events?workspaceId=ws-1'), { signal, @@ -43,6 +48,16 @@ async function drain(body: ReadableStream): Promise { } } +async function collect(body: ReadableStream, chunks: string[]): Promise { + const decoder = new TextDecoder() + const reader = body.getReader() + while (true) { + const { done, value } = await reader.read() + if (done) return + chunks.push(decoder.decode(value)) + } +} + describe('createWorkspaceSSE', () => { beforeEach(() => { vi.clearAllMocks() @@ -55,11 +70,27 @@ describe('createWorkspaceSSE', () => { vi.useRealTimers() }) - it('releases subscriptions and closes the stream when the connection reaches its ceiling', async () => { + it('announces rotation before releasing the old connection', async () => { + const { body, unsubscribe } = await openConnection() + const chunks: string[] = [] + const collected = collect(body, chunks) + + await vi.advanceTimersByTimeAsync(PAST_ROTATION_MS) + + expect(chunks).toContain('event: rotate\ndata: {}\n\n') + expect(unsubscribe).not.toHaveBeenCalled() + + await vi.advanceTimersByTimeAsync(ROTATION_GRACE_MS + HEARTBEAT_INTERVAL_MS) + + await collected + expect(unsubscribe).toHaveBeenCalledTimes(1) + }) + + it('closes an orphaned connection after the rotation grace period', async () => { const { body, unsubscribe } = await openConnection() const drained = drain(body) - await vi.advanceTimersByTimeAsync(PAST_CEILING_MS) + await vi.advanceTimersByTimeAsync(PAST_ROTATION_CLOSE_MS) await drained expect(unsubscribe).toHaveBeenCalledTimes(1) @@ -104,8 +135,46 @@ describe('createWorkspaceSSE', () => { const { unsubscribe } = await openConnection(controller.signal) controller.abort() - await vi.advanceTimersByTimeAsync(PAST_CEILING_MS) + await vi.advanceTimersByTimeAsync(PAST_ROTATION_CLOSE_MS) + + expect(unsubscribe).toHaveBeenCalledTimes(1) + }) + + it('runs every teardown when one unsubscribe throws', async () => { + const first = vi.fn(() => { + throw new Error('unsubscribe failed') + }) + const second = vi.fn() + const controller = new AbortController() + await openConnection(controller.signal, [ + { subscribe: () => first }, + { subscribe: () => second }, + ]) + + controller.abort() + + expect(first).toHaveBeenCalledTimes(1) + expect(second).toHaveBeenCalledTimes(1) + }) + it('releases earlier subscriptions when a later subscription fails to initialize', async () => { + const unsubscribe = vi.fn() + const handler = createWorkspaceSSE({ + label: 'test', + subscriptions: [ + { subscribe: () => unsubscribe }, + { + subscribe: () => { + throw new Error('subscribe failed') + }, + }, + ], + }) + const request = new NextRequest(new URL('https://sim.test/api/test/events?workspaceId=ws-1')) + + const response = await handler(request) + + await expect(response.body?.getReader().read()).rejects.toThrow('subscribe failed') expect(unsubscribe).toHaveBeenCalledTimes(1) }) }) diff --git a/apps/sim/lib/events/sse-endpoint.ts b/apps/sim/lib/events/sse-endpoint.ts index 8e5df184e1d..63a0111c8d6 100644 --- a/apps/sim/lib/events/sse-endpoint.ts +++ b/apps/sim/lib/events/sse-endpoint.ts @@ -6,6 +6,7 @@ */ import { createLogger } from '@sim/logger' +import { getErrorMessage } from '@sim/utils/errors' import { randomFloat } from '@sim/utils/random' import type { NextRequest } from 'next/server' import { getSession } from '@/lib/auth' @@ -29,33 +30,30 @@ const encoder = new TextEncoder() export const HEARTBEAT_INTERVAL_MS = 30_000 /** - * Defensive ceiling on one connection's lifetime; `EventSource` reconnects past - * this, so delivery continues across the boundary. + * Starts a make-before-break rotation for one connection. Healthy clients open + * a replacement before this stream closes; orphaned streams are released after + * the grace period without relying on runtime disconnect propagation. Because + * checks run on the heartbeat interval, the upper bound is the lifetime, jitter, + * grace period, and up to one heartbeat of scheduling delay. * * `request.signal` abort and stream `cancel()` are the primary teardown paths, * but both fire only when the runtime reports the client disconnect, and the - * unread check below only catches a consumer that has stopped draining. This - * releases whatever both miss, so retention is bounded by the ceiling instead - * of by process uptime. - * - * Matches the ceiling `lib/realtime/event-stream-route.ts` already uses for the - * same purpose. It is deliberately far longer than the unread window: a healthy - * client is drained and therefore never unread, so a short ceiling would only - * force reconnects on the connections that are working, and every reconnect is - * a window in which a transient event can be missed. + * unread check below only catches queues the HTTP adapter leaves undrained. The + * production adapter may keep pulling after the socket disappears, so this + * deadline is the primary bound rather than a fallback. */ -export const MAX_CONNECTION_MS = 4 * 60 * 60 * 1000 +export const MAX_CONNECTION_MS = 15 * 60 * 1000 /** Spreads reconnects so connections opened together do not expire together. */ export const MAX_CONNECTION_JITTER_MS = 60_000 +/** Time for a healthy client to connect its replacement before the old stream closes. */ +export const ROTATION_GRACE_MS = 30_000 + /** - * Undrained chunk count that marks a connection unread, which shortens the - * reclaim window for a vanished consumer that the ceiling above would - * otherwise hold for its full duration. The default queuing strategy reports - * `desiredSize` as `1 - queued`, so this trips only once the consumer has - * pulled nothing for several minutes — well beyond the transient backpressure - * of a slow but live client. + * Best-effort queued-chunk limit for adapters that propagate backpressure into + * the Web Stream. This is not the lifecycle guarantee: adapters may keep + * pulling after a socket disappears, so the rotation deadline remains required. */ export const MAX_UNDRAINED_CHUNKS = 16 @@ -85,10 +83,16 @@ export function createWorkspaceSSE(config: WorkspaceSSEConfig) { const cleanup = (reason: string) => { if (cleaned) return cleaned = true - for (const teardown of teardowns) { - teardown() + for (const teardown of teardowns.splice(0)) { + try { + teardown() + } catch (error) { + logger.warn(`SSE teardown failed for workspace ${workspaceId}`, { + reason, + error: getErrorMessage(error), + }) + } } - teardowns.length = 0 logger.info(`SSE connection closed for workspace ${workspaceId}`, { reason }) } @@ -103,56 +107,74 @@ export function createWorkspaceSSE(config: WorkspaceSSEConfig) { } } - const send = (eventName: string, data: Record) => { - if (cleaned) return + const enqueue = (payload: string): boolean => { + if (cleaned) return false try { - controller.enqueue( - encoder.encode(`event: ${eventName}\ndata: ${JSON.stringify(data)}\n\n`) - ) + controller.enqueue(encoder.encode(payload)) + return true } catch { - // Stream already closed + close('errored') + return false } } - for (const subscription of config.subscriptions) { - teardowns.push(subscription.subscribe(workspaceId, send)) + const send = (eventName: string, data: Record) => { + enqueue(`event: ${eventName}\ndata: ${JSON.stringify(data)}\n\n`) } - const deadline = Date.now() + MAX_CONNECTION_MS + randomFloat() * MAX_CONNECTION_JITTER_MS - - const heartbeat = setInterval(() => { - if (cleaned) { - clearInterval(heartbeat) - return - } - if (Date.now() >= deadline) { - close('expired') - return - } - const desiredSize = controller.desiredSize - if (desiredSize !== null && desiredSize <= -MAX_UNDRAINED_CHUNKS) { - close('unread') - return + try { + for (const subscription of config.subscriptions) { + teardowns.push(subscription.subscribe(workspaceId, send)) } + + const rotationDeadline = + Date.now() + MAX_CONNECTION_MS + randomFloat() * MAX_CONNECTION_JITTER_MS + let rotationStartedAt: number | null = null + + const heartbeat = setInterval(() => { + if (cleaned) { + clearInterval(heartbeat) + return + } + + const now = Date.now() + if (rotationStartedAt !== null && now - rotationStartedAt >= ROTATION_GRACE_MS) { + close('rotated') + return + } + if (rotationStartedAt === null && now >= rotationDeadline) { + if (enqueue('event: rotate\ndata: {}\n\n')) { + rotationStartedAt = now + } + return + } + + const desiredSize = controller.desiredSize + if (desiredSize !== null && desiredSize <= -MAX_UNDRAINED_CHUNKS) { + close('unread') + return + } + enqueue(': heartbeat\n\n') + }, HEARTBEAT_INTERVAL_MS) + teardowns.push(() => clearInterval(heartbeat)) + + const listenerScope = new AbortController() + request.signal.addEventListener('abort', () => close('aborted'), { + once: true, + signal: listenerScope.signal, + }) + teardowns.push(() => listenerScope.abort()) + + logger.info(`SSE connection opened for workspace ${workspaceId}`) + } catch (error) { + cleanup('setup_failed') + logger.error(`Failed to open SSE connection for workspace ${workspaceId}`, { + error: getErrorMessage(error), + }) try { - controller.enqueue(encoder.encode(': heartbeat\n\n')) - } catch { - close('errored') - } - }, HEARTBEAT_INTERVAL_MS) - teardowns.push(() => clearInterval(heartbeat)) - - // `once` only self-removes if abort fires; the expiry and unread paths - // close the connection while the signal is still live, so the listener - // needs its own removal or it retains this whole scope. - const listenerScope = new AbortController() - request.signal.addEventListener('abort', () => close('aborted'), { - once: true, - signal: listenerScope.signal, - }) - teardowns.push(() => listenerScope.abort()) - - logger.info(`SSE connection opened for workspace ${workspaceId}`) + controller.error(error) + } catch {} + } }, cancel() { cleanup('cancelled')