fix(sse): bound workspace SSE connection lifetime

Teardown ran only from the request abort listener and the stream cancel callback, both of which fire only when the runtime reports a client disconnect. Nothing else bounded the connection, so a missed report left the pub/sub handler, the heartbeat timer, and the stream's undrained queue held for the life of the process.

Add a jittered lifetime ceiling checked on the existing heartbeat tick, tighten reclaim for a vanished consumer via desiredSize, remove the abort listener on every teardown path, and run full teardown when a heartbeat enqueue fails. Log the close reason so opens minus closes is observable.

Author: waleedlatif1

apps/sim/lib/events/sse-endpoint.test.ts

@@ -0,0 +1,111 @@
+/**
+ * @vitest-environment node
+ */
+
+import { authMockFns, permissionsMock, permissionsMockFns } from '@sim/testing'
+import { NextRequest } from 'next/server'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import {
+  createWorkspaceSSE,
+  HEARTBEAT_INTERVAL_MS,
+  MAX_CONNECTION_JITTER_MS,
+  MAX_CONNECTION_MS,
+  MAX_UNDRAINED_CHUNKS,
+} from '@/lib/events/sse-endpoint'
+
+vi.mock('@/lib/workspaces/permissions/utils', () => permissionsMock)
+
+const PAST_CEILING_MS = MAX_CONNECTION_MS + MAX_CONNECTION_JITTER_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) {
+  const unsubscribe = vi.fn()
+  const handler = createWorkspaceSSE({
+    label: 'test',
+    subscriptions: [{ subscribe: () => unsubscribe }],
+  })
+  const request = new NextRequest(new URL('https://sim.test/api/test/events?workspaceId=ws-1'), {
+    signal,
+  })
+  const response = await handler(request)
+
+  return { body: response.body as ReadableStream<Uint8Array>, unsubscribe }
+}
+
+/** Resolves once the stream closes. */
+async function drain(body: ReadableStream<Uint8Array>): Promise<void> {
+  const reader = body.getReader()
+  while (true) {
+    const { done } = await reader.read()
+    if (done) return
+  }
+}
+
+describe('createWorkspaceSSE', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    vi.useFakeTimers()
+    authMockFns.mockGetSession.mockResolvedValue({ user: { id: 'user-1' } })
+    permissionsMockFns.mockGetUserEntityPermissions.mockResolvedValue('admin')
+  })
+
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  it('releases subscriptions and closes the stream when the connection reaches its ceiling', async () => {
+    const { body, unsubscribe } = await openConnection()
+    const drained = drain(body)
+
+    await vi.advanceTimersByTimeAsync(PAST_CEILING_MS)
+
+    await drained
+    expect(unsubscribe).toHaveBeenCalledTimes(1)
+  })
+
+  it('releases subscriptions when the consumer stops draining the stream', async () => {
+    const { unsubscribe } = await openConnection()
+
+    await vi.advanceTimersByTimeAsync(PAST_UNREAD_MS)
+
+    expect(unsubscribe).toHaveBeenCalledTimes(1)
+  })
+
+  it('keeps a drained connection alive past the unread threshold', async () => {
+    const { body, unsubscribe } = await openConnection()
+    void drain(body)
+
+    await vi.advanceTimersByTimeAsync(PAST_UNREAD_MS)
+
+    expect(unsubscribe).not.toHaveBeenCalled()
+  })
+
+  it('releases subscriptions when the request aborts', async () => {
+    const controller = new AbortController()
+    const { unsubscribe } = await openConnection(controller.signal)
+
+    controller.abort()
+
+    expect(unsubscribe).toHaveBeenCalledTimes(1)
+  })
+
+  it('releases subscriptions when the consumer cancels the stream', async () => {
+    const { body, unsubscribe } = await openConnection()
+
+    await body.cancel()
+
+    expect(unsubscribe).toHaveBeenCalledTimes(1)
+  })
+
+  it('releases subscriptions once when abort and the ceiling both elapse', async () => {
+    const controller = new AbortController()
+    const { unsubscribe } = await openConnection(controller.signal)
+
+    controller.abort()
+    await vi.advanceTimersByTimeAsync(PAST_CEILING_MS)
+
+    expect(unsubscribe).toHaveBeenCalledTimes(1)
+  })
+})

apps/sim/lib/events/sse-endpoint.ts

@@ -6,6 +6,7 @@
  */
 
 import { createLogger } from '@sim/logger'
+import { randomFloat } from '@sim/utils/random'
 import type { NextRequest } from 'next/server'
 import { getSession } from '@/lib/auth'
 import { SSE_HEADERS } from '@/lib/core/utils/sse'
@@ -23,7 +24,34 @@ interface WorkspaceSSEConfig {
   subscriptions: SSESubscription[]
 }
 
-const HEARTBEAT_INTERVAL_MS = 30_000
+const encoder = new TextEncoder()
+
+export const HEARTBEAT_INTERVAL_MS = 30_000
+
+/**
+ * Hard ceiling on one connection's lifetime.
+ *
+ * `request.signal` abort and stream `cancel()` are the primary teardown paths,
+ * but both fire only when the runtime reports the client disconnect. This
+ * ceiling releases the connection without depending on that report, so a
+ * missed disconnect costs one connection rather than accumulating for the life
+ * of the process. `EventSource` reconnects on its own, so delivery continues
+ * across the boundary.
+ */
+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
+
+/**
+ * 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.
+ */
+export const MAX_UNDRAINED_CHUNKS = 16
 
 export function createWorkspaceSSE(config: WorkspaceSSEConfig) {
   const logger = createLogger(`${config.label}-SSE`)
@@ -45,21 +73,30 @@ export function createWorkspaceSSE(config: WorkspaceSSEConfig) {
       return new Response('Access denied to workspace', { status: 403 })
     }
 
-    const encoder = new TextEncoder()
-    const unsubscribers: Array<() => void> = []
+    const teardowns: Array<() => void> = []
     let cleaned = false
 
-    const cleanup = () => {
+    const cleanup = (reason: string) => {
       if (cleaned) return
       cleaned = true
-      for (const unsub of unsubscribers) {
-        unsub()
+      for (const teardown of teardowns) {
+        teardown()
       }
-      logger.info(`SSE connection closed for workspace ${workspaceId}`)
+      teardowns.length = 0
+      logger.info(`SSE connection closed for workspace ${workspaceId}`, { reason })
     }
 
     const stream = new ReadableStream({
       start(controller) {
+        const close = (reason: string) => {
+          cleanup(reason)
+          try {
+            controller.close()
+          } catch {
+            // Already closed
+          }
+        }
+
         const send = (eventName: string, data: Record<string, unknown>) => {
           if (cleaned) return
           try {
@@ -72,40 +109,47 @@ export function createWorkspaceSSE(config: WorkspaceSSEConfig) {
         }
 
         for (const subscription of config.subscriptions) {
-          const unsub = subscription.subscribe(workspaceId, send)
-          unsubscribers.push(unsub)
+          teardowns.push(subscription.subscribe(workspaceId, send))
         }
 
+        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 {
             controller.enqueue(encoder.encode(': heartbeat\n\n'))
           } catch {
-            clearInterval(heartbeat)
+            close('errored')
           }
         }, HEARTBEAT_INTERVAL_MS)
-        unsubscribers.push(() => clearInterval(heartbeat))
-
-        request.signal.addEventListener(
-          'abort',
-          () => {
-            cleanup()
-            try {
-              controller.close()
-            } catch {
-              // Already closed
-            }
-          },
-          { once: true }
-        )
+        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}`)
       },
       cancel() {
-        cleanup()
+        cleanup('cancelled')
       },
     })