fix(sockets): added debouncing for sub-block values to prevent overloading socket server, fixed persistence issue during streaming back from LLM response format, removed unused events

Author: waleedlatif1

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/code.tsx

@@ -73,8 +73,21 @@ export function Code({
     }
   }, [generationType])
 
-  // State management
-  const [storeValue, setStoreValue] = useSubBlockValue(blockId, subBlockId)
+  const [storeValue, setStoreValue, { isStreaming }] = useSubBlockValue(
+    blockId,
+    subBlockId,
+    false,
+    {
+      debounceMs: 150,
+      onStreamingStart: () => {
+        logger.debug('Streaming started for code editing', { blockId, subBlockId })
+      },
+      onStreamingEnd: () => {
+        logger.debug('Streaming ended for code editing', { blockId, subBlockId })
+      },
+    }
+  )
+
   const [code, setCode] = useState<string>('')
   const [_lineCount, setLineCount] = useState(1)
   const [showTags, setShowTags] = useState(false)
@@ -104,8 +117,7 @@ export function Code({
   // AI Code Generation Hook
   const handleStreamStart = () => {
     setCode('')
-    // Optionally clear the store value too, though handleStreamChunk will update it
-    // setStoreValue('')
+    // No need to manually manage streaming - it's automatic now
   }
 
   const handleGeneratedContent = (generatedCode: string) => {
@@ -120,6 +132,7 @@ export function Code({
     setCode((currentCode) => {
       const newCode = currentCode + chunk
       if (!isPreview && !disabled) {
+        // Just update the value - streaming detection is automatic
         setStoreValue(newCode)
       }
       return newCode

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/components/response/response-format.tsx

@@ -50,7 +50,16 @@ export function ResponseFormat({
   isPreview = false,
   previewValue,
 }: ResponseFormatProps) {
-  const [storeValue, setStoreValue] = useSubBlockValue<JSONProperty[]>(blockId, subBlockId)
+  const [storeValue, setStoreValue] = useSubBlockValue<JSONProperty[]>(blockId, subBlockId, false, {
+    debounceMs: 200, // Slightly longer debounce for complex structures
+    onStreamingStart: () => {
+      console.debug('Starting bulk property operation', { blockId, subBlockId })
+    },
+    onStreamingEnd: () => {
+      console.debug('Ending bulk property operation', { blockId, subBlockId })
+    },
+  })
+
   const [showPreview, setShowPreview] = useState(false)
 
   const value = isPreview ? previewValue : storeValue

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/hooks/use-sub-block-value.ts

@@ -1,11 +1,15 @@
-import { useCallback, useEffect, useRef } from 'react'
+import { useCallback, useEffect, useRef, useState } from 'react'
 import { isEqual } from 'lodash'
+import { createLogger } from '@/lib/logs/console-logger'
 import { useCollaborativeWorkflow } from '@/hooks/use-collaborative-workflow'
 import { getProviderFromModel } from '@/providers/utils'
 import { useGeneralStore } from '@/stores/settings/general/store'
+import { useWorkflowRegistry } from '@/stores/workflows/registry/store'
 import { useSubBlockStore } from '@/stores/workflows/subblock/store'
 import { useWorkflowStore } from '@/stores/workflows/workflow/store'
 
+const logger = createLogger('SubBlockValue')
+
 // Helper function to dispatch collaborative subblock updates
 const dispatchSubblockUpdate = (blockId: string, subBlockId: string, value: any) => {
   const event = new CustomEvent('update-subblock-value', {
@@ -154,20 +158,52 @@ function storeApiKeyValue(
   }
 }
 
+interface UseSubBlockValueOptions {
+  debounceMs?: number
+  streamingThresholdMs?: number
+  onStreamingStart?: () => void
+  onStreamingEnd?: () => void
+}
+
 /**
  * Custom hook to get and set values for a sub-block in a workflow.
  * Handles complex object values properly by using deep equality comparison.
+ * Includes automatic debouncing and streaming detection for large text operations.
  *
  * @param blockId The ID of the block containing the sub-block
  * @param subBlockId The ID of the sub-block
  * @param triggerWorkflowUpdate Whether to trigger a workflow update when the value changes
- * @returns A tuple containing the current value and a setter function
+ * @param options Configuration for debouncing and streaming behavior
+ * @returns A tuple containing the current value, setter function, and optionally streaming state
  */
 export function useSubBlockValue<T = any>(
   blockId: string,
   subBlockId: string,
-  triggerWorkflowUpdate = false
-): readonly [T | null, (value: T) => void] {
+  triggerWorkflowUpdate?: boolean
+): readonly [T | null, (value: T) => void]
+
+export function useSubBlockValue<T = any>(
+  blockId: string,
+  subBlockId: string,
+  triggerWorkflowUpdate: boolean,
+  options: UseSubBlockValueOptions
+): readonly [T | null, (value: T) => void, { isStreaming: boolean }]
+
+export function useSubBlockValue<T = any>(
+  blockId: string,
+  subBlockId: string,
+  triggerWorkflowUpdate = false,
+  options?: UseSubBlockValueOptions
+):
+  | readonly [T | null, (value: T) => void]
+  | readonly [T | null, (value: T) => void, { isStreaming: boolean }] {
+  const {
+    debounceMs = 100,
+    streamingThresholdMs = 50,
+    onStreamingStart,
+    onStreamingEnd,
+  } = options || {}
+
   const { collaborativeSetSubblockValue } = useCollaborativeWorkflow()
 
   const blockType = useWorkflowStore(
@@ -187,6 +223,14 @@ export function useSubBlockValue<T = any>(
   // Previous model reference for detecting model changes
   const prevModelRef = useRef<string | null>(null)
 
+  // Debouncing and streaming detection refs
+  const debounceTimerRef = useRef<NodeJS.Timeout | null>(null)
+  const lastUpdateTimeRef = useRef<number>(0)
+  const updateCountRef = useRef<number>(0)
+  const streamingTimeoutRef = useRef<NodeJS.Timeout | null>(null)
+  const lastEmittedValueRef = useRef<T | null>(null)
+  const [isStreaming, setIsStreaming] = useState(false)
+
   // Get value from subblock store - always call this hook unconditionally
   const storeValue = useSubBlockStore(
     useCallback((state) => state.getValue(blockId, subBlockId), [blockId, subBlockId])
@@ -211,13 +255,104 @@ export function useSubBlockValue<T = any>(
   // Compute the modelValue based on block type
   const modelValue = isProviderBasedBlock ? (modelSubBlockValue as string) : null
 
+  // Cleanup timers on unmount
+  useEffect(() => {
+    return () => {
+      if (debounceTimerRef.current) {
+        clearTimeout(debounceTimerRef.current)
+      }
+      if (streamingTimeoutRef.current) {
+        clearTimeout(streamingTimeoutRef.current)
+      }
+    }
+  }, [])
+
+  // Emit the value to socket/DB
+  const emitValue = useCallback(
+    (value: T) => {
+      collaborativeSetSubblockValue(blockId, subBlockId, value)
+      lastEmittedValueRef.current = value
+    },
+    [blockId, subBlockId, collaborativeSetSubblockValue]
+  )
+
+  // Detect and handle streaming
+  const detectStreaming = useCallback(() => {
+    const now = Date.now()
+    const timeSinceLastUpdate = now - lastUpdateTimeRef.current
+
+    // If updates are coming in rapidly, we're likely streaming
+    if (timeSinceLastUpdate < streamingThresholdMs) {
+      updateCountRef.current++
+
+      // Start streaming mode after 3 rapid updates
+      if (updateCountRef.current >= 3 && !isStreaming) {
+        logger.debug('Streaming detected', {
+          blockId,
+          subBlockId,
+          updateCount: updateCountRef.current,
+        })
+        setIsStreaming(true)
+        onStreamingStart?.()
+      }
+    } else {
+      // Reset counter if updates slow down
+      updateCountRef.current = 1
+    }
+
+    lastUpdateTimeRef.current = now
+
+    // Set up timeout to end streaming
+    if (streamingTimeoutRef.current) {
+      clearTimeout(streamingTimeoutRef.current)
+    }
+
+    if (isStreaming) {
+      streamingTimeoutRef.current = setTimeout(() => {
+        logger.debug('Ending streaming mode', { blockId, subBlockId })
+        setIsStreaming(false)
+        updateCountRef.current = 0
+        onStreamingEnd?.()
+
+        // Emit the final value when streaming ends
+        if (valueRef.current !== null && valueRef.current !== lastEmittedValueRef.current) {
+          emitValue(valueRef.current)
+        }
+      }, 300) // End streaming 300ms after last update
+    }
+  }, [
+    blockId,
+    subBlockId,
+    isStreaming,
+    streamingThresholdMs,
+    onStreamingStart,
+    onStreamingEnd,
+    emitValue,
+  ])
+
   // Hook to set a value in the subblock store
   const setValue = useCallback(
     (newValue: T) => {
       // Use deep comparison to avoid unnecessary updates for complex objects
       if (!isEqual(valueRef.current, newValue)) {
         valueRef.current = newValue
 
+        // Always update local store immediately for UI responsiveness
+        useSubBlockStore.setState((state) => ({
+          workflowValues: {
+            ...state.workflowValues,
+            [useWorkflowRegistry.getState().activeWorkflowId || '']: {
+              ...state.workflowValues[useWorkflowRegistry.getState().activeWorkflowId || ''],
+              [blockId]: {
+                ...state.workflowValues[useWorkflowRegistry.getState().activeWorkflowId || '']?.[
+                  blockId
+                ],
+                [subBlockId]: newValue,
+              },
+            },
+          },
+        }))
+
         // Ensure we're passing the actual value, not a reference that might change
         const valueCopy =
           newValue === null
@@ -231,8 +366,27 @@ export function useSubBlockValue<T = any>(
           storeApiKeyValue(blockId, blockType, modelValue, newValue, storeValue)
         }
 
-        // Use collaborative function which handles both local store update and socket emission
-        collaborativeSetSubblockValue(blockId, subBlockId, valueCopy)
+        // Detect if we're in a streaming scenario
+        detectStreaming()
+
+        // Clear any existing debounce timer
+        if (debounceTimerRef.current) {
+          clearTimeout(debounceTimerRef.current)
+        }
+
+        // If streaming, don't emit immediately - wait for streaming to end
+        if (!isStreaming) {
+          // Detect large changes for automatic bulk operation mode
+          const isLargeChange = detectLargeChange(lastEmittedValueRef.current, valueCopy)
+          const effectiveDebounceMs = isLargeChange ? debounceMs * 2 : debounceMs
+
+          // Debounce the socket emission
+          debounceTimerRef.current = setTimeout(() => {
+            if (valueRef.current !== null && valueRef.current !== lastEmittedValueRef.current) {
+              emitValue(valueCopy)
+            }
+          }, effectiveDebounceMs)
+        }
 
         if (triggerWorkflowUpdate) {
           useWorkflowStore.getState().triggerUpdate()
@@ -247,7 +401,10 @@ export function useSubBlockValue<T = any>(
       storeValue,
       triggerWorkflowUpdate,
       modelValue,
-      collaborativeSetSubblockValue,
+      isStreaming,
+      debounceMs,
+      detectStreaming,
+      emitValue,
     ]
   )
 
@@ -320,5 +477,36 @@ export function useSubBlockValue<T = any>(
     }
   }, [storeValue, initialValue])
 
+  // Return appropriate tuple based on whether options were provided
+  if (options) {
+    return [
+      storeValue !== undefined ? storeValue : initialValue,
+      setValue,
+      { isStreaming },
+    ] as const
+  }
   return [storeValue !== undefined ? storeValue : initialValue, setValue] as const
 }
+
+// Helper function to detect large changes
+function detectLargeChange(oldValue: any, newValue: any): boolean {
+  // Handle null/undefined
+  if (oldValue == null && newValue == null) return false
+  if (oldValue == null || newValue == null) return true
+
+  // For strings, check if it's a large paste or deletion
+  if (typeof oldValue === 'string' && typeof newValue === 'string') {
+    const sizeDiff = Math.abs(newValue.length - oldValue.length)
+    // Consider it a large change if more than 50 characters changed at once
+    return sizeDiff > 50
+  }
+
+  // For arrays, check length difference
+  if (Array.isArray(oldValue) && Array.isArray(newValue)) {
+    const sizeDiff = Math.abs(newValue.length - oldValue.length)
+    return sizeDiff > 5
+  }
+
+  // For other types, always treat as small change
+  return false
+}

apps/sim/socket-server/handlers/connection.ts

@@ -28,7 +28,5 @@ export function setupConnectionHandlers(
       roomManager.cleanupUserFromRoom(socket.id, workflowId)
       roomManager.broadcastPresenceUpdate(workflowId)
     }
-
-    roomManager.clearPendingOperations(socket.id)
   })
 }

apps/sim/socket-server/rooms/manager.ts

@@ -75,11 +75,6 @@ export class RoomManager {
     this.userSessions.delete(socketId)
   }
 
-  // This would be used if we implement operation queuing
-  clearPendingOperations(socketId: string) {
-    logger.debug(`Cleared pending operations for socket ${socketId}`)
-  }
-
   handleWorkflowDeletion(workflowId: string) {
     logger.info(`Handling workflow deletion notification for ${workflowId}`)