fix(sse): fix memory leaks in SSE stream cleanup and add memory telemetry

Author: waleedlatif1

apps/sim/app/api/a2a/serve/[agentId]/route.ts

@@ -19,6 +19,7 @@ import { validateUrlWithDNS } from '@/lib/core/security/input-validation.server'
 import { SSE_HEADERS } from '@/lib/core/utils/sse'
 import { getBaseUrl } from '@/lib/core/utils/urls'
 import { markExecutionCancelled } from '@/lib/execution/cancellation'
+import { decrementSSEConnections, incrementSSEConnections } from '@/lib/monitoring/sse-connections'
 import { checkWorkspaceAccess } from '@/lib/workspaces/permissions/utils'
 import { getWorkspaceBilledAccountUserId } from '@/lib/workspaces/utils'
 import {
@@ -1016,16 +1017,22 @@ async function handleTaskResubscribe(
   let pollTimeoutId: ReturnType<typeof setTimeout> | null = null
 
   const abortSignal = request.signal
-  abortSignal.addEventListener('abort', () => {
-    isCancelled = true
-    if (pollTimeoutId) {
-      clearTimeout(pollTimeoutId)
-      pollTimeoutId = null
-    }
-  })
+  abortSignal.addEventListener(
+    'abort',
+    () => {
+      isCancelled = true
+      if (pollTimeoutId) {
+        clearTimeout(pollTimeoutId)
+        pollTimeoutId = null
+      }
+    },
+    { once: true }
+  )
 
+  let sseDecremented = false
   const stream = new ReadableStream({
     async start(controller) {
+      incrementSSEConnections('a2a-resubscribe')
       const sendEvent = (event: string, data: unknown): boolean => {
         if (isCancelled || abortSignal.aborted) return false
         try {
@@ -1047,6 +1054,10 @@ async function handleTaskResubscribe(
           clearTimeout(pollTimeoutId)
           pollTimeoutId = null
         }
+        if (!sseDecremented) {
+          sseDecremented = true
+          decrementSSEConnections('a2a-resubscribe')
+        }
       }
 
       if (
@@ -1165,6 +1176,10 @@ async function handleTaskResubscribe(
         clearTimeout(pollTimeoutId)
         pollTimeoutId = null
       }
+      if (!sseDecremented) {
+        sseDecremented = true
+        decrementSSEConnections('a2a-resubscribe')
+      }
     },
   })
 

apps/sim/app/api/mcp/events/route.ts

@@ -14,6 +14,7 @@ import { getSession } from '@/lib/auth'
 import { SSE_HEADERS } from '@/lib/core/utils/sse'
 import { mcpConnectionManager } from '@/lib/mcp/connection-manager'
 import { mcpPubSub } from '@/lib/mcp/pubsub'
+import { decrementSSEConnections, incrementSSEConnections } from '@/lib/monitoring/sse-connections'
 import { getUserEntityPermissions } from '@/lib/workspaces/permissions/utils'
 
 const logger = createLogger('McpEventsSSE')
@@ -41,10 +42,24 @@ export async function GET(request: NextRequest) {
 
   const encoder = new TextEncoder()
   const unsubscribers: Array<() => void> = []
+  let cleaned = false
+
+  const cleanup = () => {
+    if (cleaned) return
+    cleaned = true
+    for (const unsub of unsubscribers) {
+      unsub()
+    }
+    decrementSSEConnections('mcp-events')
+    logger.info(`SSE connection closed for workspace ${workspaceId}`)
+  }
 
   const stream = new ReadableStream({
     start(controller) {
+      incrementSSEConnections('mcp-events')
+
       const send = (eventName: string, data: Record<string, unknown>) => {
+        if (cleaned) return
         try {
           controller.enqueue(
             encoder.encode(`event: ${eventName}\ndata: ${JSON.stringify(data)}\n\n`)
@@ -82,6 +97,10 @@ export async function GET(request: NextRequest) {
 
       // Heartbeat to keep the connection alive
       const heartbeat = setInterval(() => {
+        if (cleaned) {
+          clearInterval(heartbeat)
+          return
+        }
         try {
           controller.enqueue(encoder.encode(': heartbeat\n\n'))
         } catch {
@@ -91,20 +110,24 @@ export async function GET(request: NextRequest) {
       unsubscribers.push(() => clearInterval(heartbeat))
 
       // Cleanup when client disconnects
-      request.signal.addEventListener('abort', () => {
-        for (const unsub of unsubscribers) {
-          unsub()
-        }
-        try {
-          controller.close()
-        } catch {
-          // Already closed
-        }
-        logger.info(`SSE connection closed for workspace ${workspaceId}`)
-      })
+      request.signal.addEventListener(
+        'abort',
+        () => {
+          cleanup()
+          try {
+            controller.close()
+          } catch {
+            // Already closed
+          }
+        },
+        { once: true }
+      )
 
       logger.info(`SSE connection opened for workspace ${workspaceId}`)
     },
+    cancel() {
+      cleanup()
+    },
   })
 
   return new Response(stream, { headers: SSE_HEADERS })

apps/sim/app/api/workflows/[id]/execute/route.ts

@@ -22,6 +22,7 @@ import { createExecutionEventWriter, setExecutionMeta } from '@/lib/execution/ev
 import { processInputFileFields } from '@/lib/execution/files'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
+import { decrementSSEConnections, incrementSSEConnections } from '@/lib/monitoring/sse-connections'
 import {
   cleanupExecutionBase64Cache,
   hydrateUserFilesWithBase64,
@@ -763,6 +764,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
     const encoder = new TextEncoder()
     const timeoutController = createTimeoutAbortController(preprocessResult.executionTimeout?.sync)
     let isStreamClosed = false
+    let sseDecremented = false
 
     const eventWriter = createExecutionEventWriter(executionId)
     setExecutionMeta(executionId, {
@@ -773,6 +775,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
 
     const stream = new ReadableStream<Uint8Array>({
       async start(controller) {
+        incrementSSEConnections('workflow-execute')
         let finalMetaStatus: 'complete' | 'error' | 'cancelled' | null = null
 
         const sendEvent = (event: ExecutionEvent) => {
@@ -1147,6 +1150,10 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
           if (executionId) {
             await cleanupExecutionBase64Cache(executionId)
           }
+          if (!sseDecremented) {
+            sseDecremented = true
+            decrementSSEConnections('workflow-execute')
+          }
           if (!isStreamClosed) {
             try {
               controller.enqueue(encoder.encode('data: [DONE]\n\n'))
@@ -1155,9 +1162,16 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
           }
         }
       },
-      cancel() {
+      async cancel() {
         isStreamClosed = true
         logger.info(`[${requestId}] Client disconnected from SSE stream`)
+        timeoutController.abort()
+        timeoutController.cleanup()
+        await eventWriter.close().catch(() => {})
+        if (!sseDecremented) {
+          sseDecremented = true
+          decrementSSEConnections('workflow-execute')
+        }
       },
     })
 

apps/sim/instrumentation-node.ts

@@ -160,4 +160,7 @@ async function initializeOpenTelemetry() {
 
 export async function register() {
   await initializeOpenTelemetry()
+
+  const { startMemoryTelemetry } = await import('./lib/monitoring/memory-telemetry')
+  startMemoryTelemetry()
 }

apps/sim/lib/execution/event-buffer.ts

@@ -10,6 +10,7 @@ const EVENT_LIMIT = 1000
 const RESERVE_BATCH = 100
 const FLUSH_INTERVAL_MS = 15
 const FLUSH_MAX_BATCH = 200
+const MAX_PENDING_EVENTS = 1000
 
 function getEventsKey(executionId: string) {
   return `${REDIS_PREFIX}${executionId}:events`
@@ -184,6 +185,15 @@ export function createExecutionEventWriter(executionId: string): ExecutionEventW
         stack: error instanceof Error ? error.stack : undefined,
       })
       pending = batch.concat(pending)
+      if (pending.length > MAX_PENDING_EVENTS) {
+        const dropped = pending.length - MAX_PENDING_EVENTS
+        pending = pending.slice(-MAX_PENDING_EVENTS)
+        logger.warn('Dropped oldest pending events due to sustained Redis failure', {
+          executionId,
+          dropped,
+          remaining: pending.length,
+        })
+      }
     }
   }
 

apps/sim/lib/execution/isolated-vm.ts

@@ -716,9 +716,15 @@ function spawnWorker(): Promise<WorkerInfo> {
 
         proc.on('message', (message: unknown) => handleWorkerMessage(workerId, message))
 
+        const MAX_STDERR_SIZE = 64 * 1024
         let stderrData = ''
         proc.stderr?.on('data', (data: Buffer) => {
-          stderrData += data.toString()
+          if (stderrData.length < MAX_STDERR_SIZE) {
+            stderrData += data.toString()
+            if (stderrData.length > MAX_STDERR_SIZE) {
+              stderrData = stderrData.slice(0, MAX_STDERR_SIZE)
+            }
+          }
         })
 
         const startTimeout = setTimeout(() => {

apps/sim/lib/monitoring/memory-telemetry.ts

@@ -0,0 +1,36 @@
+/**
+ * Periodic memory telemetry for diagnosing heap growth in production.
+ * Logs process.memoryUsage() every 60s alongside active SSE connection
+ * counts, enabling correlation between connection leaks and memory spikes.
+ */
+
+import { createLogger } from '@sim/logger'
+import {
+  getActiveSSEConnectionCount,
+  getActiveSSEConnectionsByRoute,
+} from '@/lib/monitoring/sse-connections'
+
+const logger = createLogger('MemoryTelemetry')
+
+const MB = 1024 * 1024
+
+let started = false
+
+export function startMemoryTelemetry(intervalMs = 60_000) {
+  if (started) return
+  started = true
+
+  const timer = setInterval(() => {
+    const mem = process.memoryUsage()
+    logger.info('Memory snapshot', {
+      heapUsedMB: Math.round(mem.heapUsed / MB),
+      heapTotalMB: Math.round(mem.heapTotal / MB),
+      rssMB: Math.round(mem.rss / MB),
+      externalMB: Math.round(mem.external / MB),
+      uptimeMin: Math.round(process.uptime() / 60),
+      activeSSEConnections: getActiveSSEConnectionCount(),
+      sseByRoute: getActiveSSEConnectionsByRoute(),
+    })
+  }, intervalMs)
+  timer.unref()
+}

apps/sim/lib/monitoring/sse-connections.ts

@@ -0,0 +1,27 @@
+/**
+ * Tracks active SSE connections by route for memory leak diagnostics.
+ * Logged alongside periodic memory telemetry to correlate connection
+ * counts with heap growth.
+ */
+
+const connections = new Map<string, number>()
+
+export function incrementSSEConnections(route: string) {
+  connections.set(route, (connections.get(route) ?? 0) + 1)
+}
+
+export function decrementSSEConnections(route: string) {
+  const count = (connections.get(route) ?? 0) - 1
+  if (count <= 0) connections.delete(route)
+  else connections.set(route, count)
+}
+
+export function getActiveSSEConnectionCount(): number {
+  let total = 0
+  for (const count of connections.values()) total += count
+  return total
+}
+
+export function getActiveSSEConnectionsByRoute(): Record<string, number> {
+  return Object.fromEntries(connections)
+}