Merge pull request #671 from simstudioai/fix/queuing

improvement(queuing): queuing with retries for sockets ops

Author: icecrasher321

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/control-bar/components/user-avatar-stack/components/connection-status/connection-status.tsx

@@ -7,17 +7,21 @@ import { useUserPermissionsContext } from '@/app/workspace/[workspaceId]/w/compo
 
 interface ConnectionStatusProps {
   isConnected: boolean
+  hasOperationError?: boolean
 }
 
-export function ConnectionStatus({ isConnected }: ConnectionStatusProps) {
+export function ConnectionStatus({ isConnected, hasOperationError }: ConnectionStatusProps) {
   const userPermissions = useUserPermissionsContext()
 
   const handleRefresh = () => {
     window.location.reload()
   }
 
-  // Don't render anything if not in offline mode
-  if (!userPermissions.isOfflineMode) {
+  // Show error if either offline mode OR operation error
+  const shouldShowError = userPermissions.isOfflineMode || hasOperationError
+
+  // Don't render anything if no errors
+  if (!shouldShowError) {
     return null
   }
 
@@ -32,10 +36,14 @@ export function ConnectionStatus({ isConnected }: ConnectionStatusProps) {
         </div>
         <div className='flex flex-col'>
           <span className='font-medium text-xs leading-tight'>
-            {isConnected ? 'Reconnected' : 'Connection lost - please refresh'}
+            {hasOperationError
+              ? 'Workflow Edit Failed'
+              : isConnected
+                ? 'Reconnected'
+                : 'Connection lost - please refresh'}
           </span>
           <span className='text-red-600 text-xs leading-tight'>
-            {isConnected ? 'Refresh to continue editing' : 'Read-only mode active'}
+            Please refresh to continue editing
           </span>
         </div>
       </div>

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/control-bar/components/user-avatar-stack/user-avatar-stack.tsx

@@ -1,6 +1,7 @@
 'use client'
 
 import { useMemo } from 'react'
+import { useCollaborativeWorkflow } from '@/hooks/use-collaborative-workflow'
 import { usePresence } from '../../../../hooks/use-presence'
 import { ConnectionStatus } from './components/connection-status/connection-status'
 import { UserAvatar } from './components/user-avatar/user-avatar'
@@ -29,6 +30,9 @@ export function UserAvatarStack({
   const { users: presenceUsers, isConnected } = usePresence()
   const users = propUsers || presenceUsers
 
+  // Get operation error state from collaborative workflow
+  const { hasOperationError } = useCollaborativeWorkflow()
+
   // Memoize the processed users to avoid unnecessary re-renders
   const { visibleUsers, overflowCount } = useMemo(() => {
     if (users.length === 0) {
@@ -53,8 +57,8 @@ export function UserAvatarStack({
 
   return (
     <div className={`flex items-center gap-3 ${className}`}>
-      {/* Connection status - always check, shows when offline */}
-      <ConnectionStatus isConnected={isConnected} />
+      {/* Connection status - always check, shows when offline or operation errors */}
+      <ConnectionStatus isConnected={isConnected} hasOperationError={hasOperationError} />
 
       {/* Only show avatar stack when there are multiple users (>1) */}
       {users.length > 1 && (

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/loop-node/components/loop-badges.tsx

@@ -260,7 +260,7 @@ export function LoopBadges({ nodeId, data }: LoopBadgesProps) {
                   <div className='relative min-h-[80px] rounded-md border border-input bg-background px-3 pt-2 pb-3 font-mono text-sm'>
                     {editorValue === '' && (
                       <div className='pointer-events-none absolute top-[8.5px] left-3 select-none text-muted-foreground/50'>
-                        ["item1", "item2", "item3"]
+                        ['item1', 'item2', 'item3']
                       </div>
                     )}
                     <Editor

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

@@ -269,7 +269,8 @@ export function useSubBlockValue<T = any>(
       if (!isEqual(valueRef.current, newValue)) {
         valueRef.current = newValue
 
-        // Always update local store immediately for UI responsiveness
+        // Update local store immediately for UI responsiveness
+        // The collaborative function will also update it, but that's okay for idempotency
         useSubBlockStore.setState((state) => ({
           workflowValues: {
             ...state.workflowValues,

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/workflow.tsx

@@ -22,7 +22,6 @@ import { SkeletonLoading } from '@/app/workspace/[workspaceId]/w/[workflowId]/co
 import { Toolbar } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/toolbar/toolbar'
 import { useUserPermissionsContext } from '@/app/workspace/[workspaceId]/w/components/providers/workspace-permissions-provider'
 import { getBlock } from '@/blocks'
-import { useSocket } from '@/contexts/socket-context'
 import { useCollaborativeWorkflow } from '@/hooks/use-collaborative-workflow'
 import { useWorkspacePermissions } from '@/hooks/use-workspace-permissions'
 import { useExecutionStore } from '@/stores/execution/store'
@@ -120,10 +119,9 @@ const WorkflowContent = React.memo(() => {
     collaborativeRemoveEdge: removeEdge,
     collaborativeUpdateBlockPosition,
     collaborativeUpdateParentId: updateParentId,
-    isConnected,
-    currentWorkflowId,
+    collaborativeSetSubblockValue,
   } = useCollaborativeWorkflow()
-  const { emitSubblockUpdate } = useSocket()
+
   const { markAllAsRead } = useNotificationStore()
   const { resetLoaded: resetVariablesLoaded } = useVariablesStore()
 
@@ -1484,11 +1482,9 @@ const WorkflowContent = React.memo(() => {
     const handleSubBlockValueUpdate = (event: CustomEvent) => {
       const { blockId, subBlockId, value } = event.detail
       if (blockId && subBlockId) {
-        // Only emit the socket update, don't update the store again
-        // The store was already updated in the setValue function
-        if (isConnected && currentWorkflowId && activeWorkflowId === currentWorkflowId) {
-          emitSubblockUpdate(blockId, subBlockId, value)
-        }
+        // Use collaborative function to go through queue system
+        // This ensures 5-second timeout and error detection work
+        collaborativeSetSubblockValue(blockId, subBlockId, value)
       }
     }
 
@@ -1500,7 +1496,7 @@ const WorkflowContent = React.memo(() => {
         handleSubBlockValueUpdate as EventListener
       )
     }
-  }, [emitSubblockUpdate, isConnected, currentWorkflowId, activeWorkflowId])
+  }, [collaborativeSetSubblockValue])
 
   // Show skeleton UI while loading, then smoothly transition to real content
   const showSkeletonUI = !isWorkflowReady

apps/sim/app/workspace/[workspaceId]/w/components/providers/workspace-permissions-provider.tsx

@@ -4,12 +4,12 @@ import type React from 'react'
 import { createContext, useContext, useEffect, useMemo, useState } from 'react'
 import { useParams } from 'next/navigation'
 import { createLogger } from '@/lib/logs/console-logger'
+import { useCollaborativeWorkflow } from '@/hooks/use-collaborative-workflow'
 import { useUserPermissions, type WorkspaceUserPermissions } from '@/hooks/use-user-permissions'
 import {
   useWorkspacePermissions,
   type WorkspacePermissions,
 } from '@/hooks/use-workspace-permissions'
-import { usePresence } from '../../[workflowId]/hooks/use-presence'
 
 const logger = createLogger('WorkspacePermissionsProvider')
 
@@ -57,7 +57,16 @@ export function WorkspacePermissionsProvider({ children }: WorkspacePermissionsP
 
   // Manage offline mode state locally
   const [isOfflineMode, setIsOfflineMode] = useState(false)
-  const [hasBeenConnected, setHasBeenConnected] = useState(false)
+
+  // Get operation error state from collaborative workflow
+  const { hasOperationError } = useCollaborativeWorkflow()
+
+  // Set offline mode when there are operation errors
+  useEffect(() => {
+    if (hasOperationError) {
+      setIsOfflineMode(true)
+    }
+  }, [hasOperationError])
 
   // Fetch workspace permissions and loading state
   const {
@@ -74,26 +83,8 @@ export function WorkspacePermissionsProvider({ children }: WorkspacePermissionsP
     permissionsError
   )
 
-  // Get connection status and update offline mode accordingly
-  const { isConnected } = usePresence()
-
-  useEffect(() => {
-    if (isConnected) {
-      // Mark that we've been connected at least once
-      setHasBeenConnected(true)
-      // On initial connection, allow going online
-      if (!hasBeenConnected) {
-        setIsOfflineMode(false)
-      }
-      // If we were previously connected and this is a reconnection, stay offline (user must refresh)
-    } else if (hasBeenConnected) {
-      const timeoutId = setTimeout(() => {
-        setIsOfflineMode(true)
-      }, 6000)
-      return () => clearTimeout(timeoutId)
-    }
-    // If not connected and never been connected, stay in initial state (not offline mode)
-  }, [isConnected, hasBeenConnected])
+  // Note: Connection-based error detection removed - only rely on operation timeouts
+  // The 5-second operation timeout system will handle all error cases
 
   // Create connection-aware permissions that override user permissions when offline
   const userPermissions = useMemo((): WorkspaceUserPermissions & { isOfflineMode?: boolean } => {

apps/sim/contexts/socket-context.tsx

@@ -38,8 +38,18 @@ interface SocketContextType {
   presenceUsers: PresenceUser[]
   joinWorkflow: (workflowId: string) => void
   leaveWorkflow: () => void
-  emitWorkflowOperation: (operation: string, target: string, payload: any) => void
-  emitSubblockUpdate: (blockId: string, subblockId: string, value: any) => void
+  emitWorkflowOperation: (
+    operation: string,
+    target: string,
+    payload: any,
+    operationId?: string
+  ) => void
+  emitSubblockUpdate: (
+    blockId: string,
+    subblockId: string,
+    value: any,
+    operationId?: string
+  ) => void
   emitCursorUpdate: (cursor: { x: number; y: number }) => void
   emitSelectionUpdate: (selection: { type: 'block' | 'edge' | 'none'; id?: string }) => void
   // Event handlers for receiving real-time updates
@@ -51,6 +61,8 @@ interface SocketContextType {
   onUserLeft: (handler: (data: any) => void) => void
   onWorkflowDeleted: (handler: (data: any) => void) => void
   onWorkflowReverted: (handler: (data: any) => void) => void
+  onOperationConfirmed: (handler: (data: any) => void) => void
+  onOperationFailed: (handler: (data: any) => void) => void
 }
 
 const SocketContext = createContext<SocketContextType>({
@@ -73,6 +85,8 @@ const SocketContext = createContext<SocketContextType>({
   onUserLeft: () => {},
   onWorkflowDeleted: () => {},
   onWorkflowReverted: () => {},
+  onOperationConfirmed: () => {},
+  onOperationFailed: () => {},
 })
 
 export const useSocket = () => useContext(SocketContext)
@@ -103,6 +117,8 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
     userLeft?: (data: any) => void
     workflowDeleted?: (data: any) => void
     workflowReverted?: (data: any) => void
+    operationConfirmed?: (data: any) => void
+    operationFailed?: (data: any) => void
   }>({})
 
   // Helper function to generate a fresh socket token
@@ -290,6 +306,18 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
           eventHandlers.current.workflowReverted?.(data)
         })
 
+        // Operation confirmation events
+        socketInstance.on('operation-confirmed', (data) => {
+          logger.debug('Operation confirmed', { operationId: data.operationId })
+          eventHandlers.current.operationConfirmed?.(data)
+        })
+
+        // Operation failure events
+        socketInstance.on('operation-failed', (data) => {
+          logger.warn('Operation failed', { operationId: data.operationId, error: data.error })
+          eventHandlers.current.operationFailed?.(data)
+        })
+
         // Cursor update events
         socketInstance.on('cursor-update', (data) => {
           setPresenceUsers((prev) =>
@@ -444,8 +472,10 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
 
   // Emit workflow operations (blocks, edges, subflows)
   const emitWorkflowOperation = useCallback(
-    (operation: string, target: string, payload: any) => {
-      if (!socket || !currentWorkflowId) return
+    (operation: string, target: string, payload: any, operationId?: string) => {
+      if (!socket || !currentWorkflowId) {
+        return
+      }
 
       // Apply light throttling only to position updates for smooth collaborative experience
       const isPositionUpdate = operation === 'update-position' && target === 'block'
@@ -459,6 +489,7 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
           target,
           payload,
           timestamp: Date.now(),
+          operationId, // Include operation ID for queue tracking
         })
 
         // Check if we already have a pending timeout for this block
@@ -482,6 +513,7 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
           target,
           payload,
           timestamp: Date.now(),
+          operationId, // Include operation ID for queue tracking
         })
       }
     },
@@ -490,14 +522,15 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
 
   // Emit subblock value updates
   const emitSubblockUpdate = useCallback(
-    (blockId: string, subblockId: string, value: any) => {
+    (blockId: string, subblockId: string, value: any, operationId?: string) => {
       // Only emit if socket is connected and we're in a valid workflow room
       if (socket && currentWorkflowId) {
         socket.emit('subblock-update', {
           blockId,
           subblockId,
           value,
           timestamp: Date.now(),
+          operationId, // Include operation ID for queue tracking
         })
       } else {
         logger.warn('Cannot emit subblock update: no socket connection or workflow room', {
@@ -570,6 +603,14 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
     eventHandlers.current.workflowReverted = handler
   }, [])
 
+  const onOperationConfirmed = useCallback((handler: (data: any) => void) => {
+    eventHandlers.current.operationConfirmed = handler
+  }, [])
+
+  const onOperationFailed = useCallback((handler: (data: any) => void) => {
+    eventHandlers.current.operationFailed = handler
+  }, [])
+
   return (
     <SocketContext.Provider
       value={{
@@ -592,6 +633,8 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
         onUserLeft,
         onWorkflowDeleted,
         onWorkflowReverted,
+        onOperationConfirmed,
+        onOperationFailed,
       }}
     >
       {children}