improvement(sockets): cleanup debounce logic + add flush mechanism to… (#1152)

* improvement(sockets): cleanup debounce logic + add flush mechanism to not lose ops

* fix optimistic update overwritten race condition

* fix

* fix forever stuck in processing

Author: icecrasher321

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

@@ -163,8 +163,7 @@ export function Code({
 
   // State management - useSubBlockValue with explicit streaming control
   const [storeValue, setStoreValue] = useSubBlockValue(blockId, subBlockId, false, {
-    debounceMs: 150,
-    isStreaming: isAiStreaming, // Use AI streaming state directly
+    isStreaming: isAiStreaming,
     onStreamingEnd: () => {
       logger.debug('AI streaming ended, value persisted', { blockId, subBlockId })
     },

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

@@ -13,6 +13,7 @@ import { useSubBlockValue } from '@/app/workspace/[workspaceId]/w/[workflowId]/c
 import { useWand } from '@/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-wand'
 import type { SubBlockConfig } from '@/blocks/types'
 import { useTagSelection } from '@/hooks/use-tag-selection'
+import { useOperationQueueStore } from '@/stores/operation-queue/store'
 
 const logger = createLogger('LongInput')
 
@@ -73,7 +74,6 @@ export function LongInput({
 
   // State management - useSubBlockValue with explicit streaming control
   const [storeValue, setStoreValue] = useSubBlockValue(blockId, subBlockId, false, {
-    debounceMs: 150,
     isStreaming: wandHook?.isStreaming || false, // Use wand streaming state
     onStreamingEnd: () => {
       logger.debug('Wand streaming ended, value persisted', { blockId, subBlockId })
@@ -379,6 +379,11 @@ export function LongInput({
           onScroll={handleScroll}
           onWheel={handleWheel}
           onKeyDown={handleKeyDown}
+          onBlur={() => {
+            try {
+              useOperationQueueStore.getState().flushDebouncedForBlock(blockId)
+            } catch {}
+          }}
           onFocus={() => {
             setShowEnvVars(false)
             setShowTags(false)

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

@@ -9,6 +9,7 @@ import { cn } from '@/lib/utils'
 import { useSubBlockValue } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/workflow-block/components/sub-block/hooks/use-sub-block-value'
 import type { SubBlockConfig } from '@/blocks/types'
 import { useTagSelection } from '@/hooks/use-tag-selection'
+import { useOperationQueueStore } from '@/stores/operation-queue/store'
 
 const logger = createLogger('ShortInput')
 
@@ -329,6 +330,9 @@ export function ShortInput({
         onBlur={() => {
           setIsFocused(false)
           setShowEnvVars(false)
+          try {
+            useOperationQueueStore.getState().flushDebouncedForBlock(blockId)
+          } catch {}
         }}
         onDrop={handleDrop}
         onDragOver={handleDragOver}

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

@@ -11,8 +11,7 @@ import { useWorkflowStore } from '@/stores/workflows/workflow/store'
 const logger = createLogger('SubBlockValue')
 
 interface UseSubBlockValueOptions {
-  debounceMs?: number
-  isStreaming?: boolean // Explicit streaming state
+  isStreaming?: boolean
   onStreamingEnd?: () => void
 }
 
@@ -130,8 +129,21 @@ export function useSubBlockValue<T = any>(
       if (!isEqual(valueRef.current, newValue)) {
         valueRef.current = newValue
 
-        // Update local store immediately for UI responsiveness
-        // The collaborative function will also update it, but that's okay for idempotency
+        // Ensure we're passing the actual value, not a reference that might change
+        const valueCopy =
+          newValue === null
+            ? null
+            : typeof newValue === 'object'
+              ? JSON.parse(JSON.stringify(newValue))
+              : newValue
+
+        // If streaming, hold value locally and do not update global store to avoid render-phase updates
+        if (isStreaming) {
+          streamingValueRef.current = valueCopy
+          return
+        }
+
+        // Update local store immediately for UI responsiveness (non-streaming)
         useSubBlockStore.setState((state) => ({
           workflowValues: {
             ...state.workflowValues,
@@ -145,44 +157,26 @@ export function useSubBlockValue<T = any>(
           },
         }))
 
-        // Handle model changes for provider-based blocks - clear API key when provider changes
+        // Handle model changes for provider-based blocks - clear API key when provider changes (non-streaming)
         if (
           subBlockId === 'model' &&
           isProviderBasedBlock &&
           newValue &&
           typeof newValue === 'string'
         ) {
           const currentApiKeyValue = useSubBlockStore.getState().getValue(blockId, 'apiKey')
-
-          // Only clear if there's currently an API key value
           if (currentApiKeyValue && currentApiKeyValue !== '') {
             const oldModelValue = storeValue as string
             const oldProvider = oldModelValue ? getProviderFromModel(oldModelValue) : null
             const newProvider = getProviderFromModel(newValue)
-
-            // Clear API key if provider changed
             if (oldProvider !== newProvider) {
-              // Use collaborative function to clear the API key
               collaborativeSetSubblockValue(blockId, 'apiKey', '')
             }
           }
         }
 
-        // Ensure we're passing the actual value, not a reference that might change
-        const valueCopy =
-          newValue === null
-            ? null
-            : typeof newValue === 'object'
-              ? JSON.parse(JSON.stringify(newValue))
-              : newValue
-
-        // If streaming, just store the value without emitting
-        if (isStreaming) {
-          streamingValueRef.current = valueCopy
-        } else {
-          // Emit immediately - let the operation queue handle debouncing and deduplication
-          emitValue(valueCopy)
-        }
+        // Emit immediately - let the operation queue handle debouncing and deduplication
+        emitValue(valueCopy)
 
         if (triggerWorkflowUpdate) {
           useWorkflowStore.getState().triggerUpdate()

apps/sim/contexts/socket-context.tsx

@@ -334,6 +334,18 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
           )
 
           if (data.workflowId === urlWorkflowId) {
+            try {
+              const { useOperationQueueStore } = require('@/stores/operation-queue/store')
+              const hasPending = useOperationQueueStore
+                .getState()
+                .operations.some(
+                  (op: any) => op.workflowId === data.workflowId && op.status !== 'confirmed'
+                )
+              if (hasPending) {
+                logger.info('Skipping copilot rehydration due to pending operations in queue')
+                return
+              }
+            } catch {}
             try {
               // Fetch fresh workflow state directly from API
               const response = await fetch(`/api/workflows/${data.workflowId}`)
@@ -364,27 +376,38 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
                         )
                       })
 
-                      // Update workflow store with fresh state from database
-                      const newWorkflowState = {
-                        blocks: workflowState.blocks || {},
-                        edges: workflowState.edges || [],
-                        loops: workflowState.loops || {},
-                        parallels: workflowState.parallels || {},
-                        lastSaved: workflowState.lastSaved || Date.now(),
-                        isDeployed: workflowState.isDeployed || false,
-                        deployedAt: workflowState.deployedAt,
-                        deploymentStatuses: workflowState.deploymentStatuses || {},
-                        hasActiveSchedule: workflowState.hasActiveSchedule || false,
-                        hasActiveWebhook: workflowState.hasActiveWebhook || false,
+                      // Merge workflow store with server state (do not drop optimistic local state)
+                      const existing = useWorkflowStore.getState()
+                      const mergedBlocks = {
+                        ...(existing.blocks || {}),
+                        ...(workflowState.blocks || {}),
                       }
+                      const edgeById = new Map<string, any>()
+                      ;(existing.edges || []).forEach((e: any) => edgeById.set(e.id, e))
+                      ;(workflowState.edges || []).forEach((e: any) => edgeById.set(e.id, e))
+                      const mergedEdges = Array.from(edgeById.values())
+                      useWorkflowStore.setState({
+                        blocks: mergedBlocks,
+                        edges: mergedEdges,
+                        loops: workflowState.loops || existing.loops || {},
+                        parallels: workflowState.parallels || existing.parallels || {},
+                        lastSaved: workflowState.lastSaved || existing.lastSaved || Date.now(),
+                        isDeployed: workflowState.isDeployed ?? existing.isDeployed ?? false,
+                        deployedAt: workflowState.deployedAt || existing.deployedAt,
+                        deploymentStatuses:
+                          workflowState.deploymentStatuses || existing.deploymentStatuses || {},
+                        hasActiveWebhook:
+                          workflowState.hasActiveWebhook ?? existing.hasActiveWebhook ?? false,
+                      })
 
-                      useWorkflowStore.setState(newWorkflowState)
-
-                      // Update subblock store with fresh values
+                      // Merge subblock store values per workflow
                       useSubBlockStore.setState((state: any) => ({
                         workflowValues: {
                           ...state.workflowValues,
-                          [data.workflowId]: subblockValues,
+                          [data.workflowId]: {
+                            ...(state.workflowValues?.[data.workflowId] || {}),
+                            ...subblockValues,
+                          },
                         },
                       }))
 
@@ -461,19 +484,31 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
 
           // Update local stores with the fresh workflow state (same logic as YAML editor)
           if (workflowData?.state && workflowData.id === urlWorkflowId) {
+            try {
+              const { useOperationQueueStore } = require('@/stores/operation-queue/store')
+              const hasPending = useOperationQueueStore
+                .getState()
+                .operations.some(
+                  (op: any) => op.workflowId === workflowData.id && op.status !== 'confirmed'
+                )
+              if (hasPending) {
+                logger.info(
+                  'Skipping workflow-state rehydration due to pending operations in queue'
+                )
+                return
+              }
+            } catch {}
             logger.info('Updating local stores with fresh workflow state from server')
 
             try {
-              // Import stores dynamically to avoid import issues
               Promise.all([
                 import('@/stores/workflows/workflow/store'),
                 import('@/stores/workflows/subblock/store'),
                 import('@/stores/workflows/registry/store'),
               ])
-                .then(([{ useWorkflowStore }, { useSubBlockStore }, { useWorkflowRegistry }]) => {
+                .then(([{ useWorkflowStore }, { useSubBlockStore }]) => {
                   const workflowState = workflowData.state
 
-                  // Extract subblock values from blocks before updating workflow store
                   const subblockValues: Record<string, Record<string, any>> = {}
                   Object.entries(workflowState.blocks || {}).forEach(([blockId, block]) => {
                     const blockState = block as any
@@ -483,36 +518,40 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
                     })
                   })
 
-                  // Update workflow store with new state
-                  const newWorkflowState = {
-                    blocks: workflowState.blocks || {},
-                    edges: workflowState.edges || [],
-                    loops: workflowState.loops || {},
-                    parallels: workflowState.parallels || {},
-                    lastSaved: workflowState.lastSaved || Date.now(),
-                    isDeployed: workflowState.isDeployed || false,
-                    deployedAt: workflowState.deployedAt,
-                    deploymentStatuses: workflowState.deploymentStatuses || {},
-                    hasActiveSchedule: workflowState.hasActiveSchedule || false,
-                    hasActiveWebhook: workflowState.hasActiveWebhook || false,
+                  const existing = useWorkflowStore.getState()
+                  const mergedBlocks = {
+                    ...(existing.blocks || {}),
+                    ...(workflowState.blocks || {}),
                   }
+                  const edgeById = new Map<string, any>()
+                  ;(existing.edges || []).forEach((e: any) => edgeById.set(e.id, e))
+                  ;(workflowState.edges || []).forEach((e: any) => edgeById.set(e.id, e))
+                  const mergedEdges = Array.from(edgeById.values())
+                  useWorkflowStore.setState({
+                    blocks: mergedBlocks,
+                    edges: mergedEdges,
+                    loops: workflowState.loops || existing.loops || {},
+                    parallels: workflowState.parallels || existing.parallels || {},
+                    lastSaved: workflowState.lastSaved || existing.lastSaved || Date.now(),
+                    isDeployed: workflowState.isDeployed ?? existing.isDeployed ?? false,
+                    deployedAt: workflowState.deployedAt || existing.deployedAt,
+                    deploymentStatuses:
+                      workflowState.deploymentStatuses || existing.deploymentStatuses || {},
+                    hasActiveWebhook:
+                      workflowState.hasActiveWebhook ?? existing.hasActiveWebhook ?? false,
+                  })
 
-                  useWorkflowStore.setState(newWorkflowState)
-
-                  // Update subblock store with fresh values
                   useSubBlockStore.setState((state: any) => ({
                     workflowValues: {
                       ...state.workflowValues,
-                      [workflowData.id]: subblockValues,
+                      [workflowData.id]: {
+                        ...(state.workflowValues?.[workflowData.id] || {}),
+                        ...subblockValues,
+                      },
                     },
                   }))
 
-                  // Note: Auto layout is not triggered here because:
-                  // 1. For copilot edits: positions are already optimized by the backend
-                  // 2. For other syncs: the existing positions should be preserved
-                  // This prevents ID conflicts and unnecessary position updates
-
-                  logger.info('Successfully updated local stores with fresh workflow state')
+                  logger.info('Merged fresh workflow state with local state')
                 })
                 .catch((error) => {
                   logger.error('Failed to import stores for workflow state update:', error)
@@ -558,6 +597,16 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
       `URL workflow changed from ${currentWorkflowId} to ${urlWorkflowId}, switching rooms`
     )
 
+    try {
+      const { useOperationQueueStore } = require('@/stores/operation-queue/store')
+      // Flush debounced updates for the old workflow before switching rooms
+      if (currentWorkflowId) {
+        useOperationQueueStore.getState().flushDebouncedForWorkflow(currentWorkflowId)
+      } else {
+        useOperationQueueStore.getState().flushAllDebounced()
+      }
+    } catch {}
+
     // Leave current workflow first if we're in one
     if (currentWorkflowId) {
       logger.info(`Leaving current workflow ${currentWorkflowId} before joining ${urlWorkflowId}`)
@@ -615,6 +664,11 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
   const leaveWorkflow = useCallback(() => {
     if (socket && currentWorkflowId) {
       logger.info(`Leaving workflow: ${currentWorkflowId}`)
+      try {
+        const { useOperationQueueStore } = require('@/stores/operation-queue/store')
+        useOperationQueueStore.getState().flushDebouncedForWorkflow(currentWorkflowId)
+        useOperationQueueStore.getState().cancelOperationsForWorkflow(currentWorkflowId)
+      } catch {}
       socket.emit('leave-workflow')
       setCurrentWorkflowId(null)
       setPresenceUsers([])

apps/sim/hooks/use-collaborative-workflow.ts

@@ -492,7 +492,6 @@ export function useCollaborativeWorkflow() {
     (operation: string, target: string, payload: any, localAction: () => void) => {
       if (isApplyingRemoteChange.current) return
 
-      // Skip socket operations when in diff mode
       if (isShowingDiff) {
         logger.debug('Skipping debounced socket operation in diff mode:', operation)
         return
@@ -673,32 +672,6 @@ export function useCollaborativeWorkflow() {
     (id: string, name: string) => {
       executeQueuedOperation('update-name', 'block', { id, name }, () => {
         workflowStore.updateBlockName(id, name)
-
-        // Handle pending subblock updates
-        const globalWindow = window as any
-        const pendingUpdates = globalWindow.__pendingSubblockUpdates
-        if (pendingUpdates && Array.isArray(pendingUpdates)) {
-          // Queue each subblock update individually
-          for (const update of pendingUpdates) {
-            const { blockId, subBlockId, newValue } = update
-            const operationId = crypto.randomUUID()
-
-            addToQueue({
-              id: operationId,
-              operation: {
-                operation: 'subblock-update',
-                target: 'subblock',
-                payload: { blockId, subblockId: subBlockId, value: newValue },
-              },
-              workflowId: activeWorkflowId || '',
-              userId: session?.user?.id || 'unknown',
-            })
-
-            subBlockStore.setValue(blockId, subBlockId, newValue)
-          }
-          // Clear the pending updates
-          globalWindow.__pendingSubblockUpdates = undefined
-        }
       })
     },
     [

apps/sim/stores/index.ts

@@ -93,7 +93,10 @@ function handleBeforeUnload(event: BeforeUnloadEvent): void {
     }
   }
 
-  // Note: Socket.IO handles real-time sync automatically
+  try {
+    const { useOperationQueueStore } = require('@/stores/operation-queue/store')
+    useOperationQueueStore.getState().flushAllDebounced()
+  } catch {}
 
   // Standard beforeunload pattern
   event.preventDefault()