From bf9b3fdc997ede3c70b38e23c550ba19247a717e Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Fri, 27 Jun 2025 15:21:46 -0700 Subject: [PATCH] fix(conn pool separation): conn pool separation test (#568) * add logs to see freestyle payload * fix lint * db connection config changes * fix lint * db client fix + workspace switching * fix lint * remove unecessary logging * revert route file --------- Co-authored-by: Vikhyath Mondreti --- .../workspace-header/workspace-header.tsx | 24 ++++++---- apps/sim/contexts/socket-context.tsx | 10 ++-- apps/sim/db/index.ts | 33 +++++++++---- apps/sim/socket-server/database/operations.ts | 48 ++++++++++++++++++- apps/sim/socket-server/rooms/manager.ts | 20 +++++++- apps/sim/stores/workflows/registry/store.ts | 35 -------------- apps/sim/stores/workflows/registry/types.ts | 3 +- 7 files changed, 109 insertions(+), 64 deletions(-) diff --git a/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/components/workspace-header/workspace-header.tsx b/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/components/workspace-header/workspace-header.tsx index 4ffc0ba6de..0630c63f58 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/components/workspace-header/workspace-header.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/components/workspace-header/workspace-header.tsx @@ -333,7 +333,7 @@ export const WorkspaceHeader = React.memo( }, [sessionData?.user?.id, fetchSubscriptionStatus, fetchWorkspaces]) const switchWorkspace = useCallback( - (workspace: Workspace) => { + async (workspace: Workspace) => { // If already on this workspace, close dropdown and do nothing else if (activeWorkspace?.id === workspace.id) { setWorkspaceDropdownOpen(false) @@ -344,9 +344,9 @@ export const WorkspaceHeader = React.memo( setWorkspaceDropdownOpen(false) // Use full workspace switch which now handles localStorage automatically - switchToWorkspace(workspace.id) + await switchToWorkspace(workspace.id) - // Update URL to include workspace ID + // Update URL to include workspace ID - only after workspace switch completes router.push(`/workspace/${workspace.id}/w`) }, [activeWorkspace?.id, switchToWorkspace, router, setWorkspaceDropdownOpen] @@ -374,9 +374,9 @@ export const WorkspaceHeader = React.memo( // Use switchToWorkspace to properly load workflows for the new workspace // This will clear existing workflows, set loading state, and fetch workflows from DB - switchToWorkspace(newWorkspace.id) + await switchToWorkspace(newWorkspace.id) - // Update URL to include new workspace ID + // Update URL to include new workspace ID - only after workspace switch completes router.push(`/workspace/${newWorkspace.id}/w`) } } catch (err) { @@ -464,11 +464,15 @@ export const WorkspaceHeader = React.memo( setWorkspaces(updatedWorkspaces) // If deleted workspace was active, switch to another workspace - if (activeWorkspace?.id === id && updatedWorkspaces.length > 0) { - // Use the specialized method for handling workspace deletion - const newWorkspaceId = updatedWorkspaces[0].id - useWorkflowRegistry.getState().handleWorkspaceDeletion(newWorkspaceId) - setActiveWorkspace(updatedWorkspaces[0]) + if (activeWorkspace?.id === id) { + const newWorkspace = updatedWorkspaces[0] + setActiveWorkspace(newWorkspace) + + // Switch to the new workspace (this handles all workflow state management) + await switchToWorkspace(newWorkspace.id) + + // Navigate to the new workspace - only after workspace switch completes + router.push(`/workspace/${newWorkspace.id}/w`) } setWorkspaceDropdownOpen(false) diff --git a/apps/sim/contexts/socket-context.tsx b/apps/sim/contexts/socket-context.tsx index 02f3aff85d..c270e602cd 100644 --- a/apps/sim/contexts/socket-context.tsx +++ b/apps/sim/contexts/socket-context.tsx @@ -427,7 +427,7 @@ export function SocketProvider({ children, user }: SocketProviderProps) { // Check if we already have a pending timeout for this block if (!positionUpdateTimeouts.current.has(blockId)) { - // Schedule emission with light throttling (120fps = ~8ms) + // Schedule emission with optimized throttling (30fps = ~33ms) to reduce DB load const timeoutId = window.setTimeout(() => { const latestUpdate = pendingPositionUpdates.current.get(blockId) if (latestUpdate) { @@ -435,7 +435,7 @@ export function SocketProvider({ children, user }: SocketProviderProps) { pendingPositionUpdates.current.delete(blockId) } positionUpdateTimeouts.current.delete(blockId) - }, 8) // 120fps for smooth movement + }, 33) // 30fps - good balance between smoothness and DB performance positionUpdateTimeouts.current.set(blockId, timeoutId) } @@ -475,14 +475,14 @@ export function SocketProvider({ children, user }: SocketProviderProps) { [socket, currentWorkflowId] ) - // Minimal cursor throttling (reduced from 30fps to 120fps) + // Cursor throttling optimized for database connection health const lastCursorEmit = useRef(0) const emitCursorUpdate = useCallback( (cursor: { x: number; y: number }) => { if (socket && currentWorkflowId) { const now = performance.now() - // Very light throttling at 120fps (8ms) to prevent excessive spam - if (now - lastCursorEmit.current >= 8) { + // Reduced to 30fps (33ms) to reduce database load while maintaining smooth UX + if (now - lastCursorEmit.current >= 33) { socket.emit('cursor-update', { cursor }) lastCursorEmit.current = now } diff --git a/apps/sim/db/index.ts b/apps/sim/db/index.ts index a725fed90e..cdedbc0a09 100644 --- a/apps/sim/db/index.ts +++ b/apps/sim/db/index.ts @@ -7,14 +7,31 @@ import * as schema from './schema' // In development, use the direct DATABASE_URL const connectionString = env.POSTGRES_URL ?? env.DATABASE_URL -const drizzleClient = drizzle( - postgres(connectionString, { - prepare: false, // Disable prefetch as it is not supported for "Transaction" pool mode - idle_timeout: 30, // Keep connections alive for 30 seconds when idle - connect_timeout: 30, // Timeout after 30 seconds when connecting - }), - { schema } -) +/** + * Connection Pool Allocation Strategy + * + * Main App (this file): 3 connections per instance + * Socket Server Operations: 2 connections + * Socket Server Room Manager: 1 connection + * + * With ~3-4 Vercel serverless instances typically active: + * - Main app: 3 × 4 = 12 connections + * - Socket server: 2 + 1 = 3 connections + * - Buffer: 5 connections for spikes/other services + * - Total: ~20 connections (at capacity limit) + * + * This conservative allocation prevents pool exhaustion while maintaining performance. + */ + +const postgresClient = postgres(connectionString, { + prepare: false, // Disable prefetch as it is not supported for "Transaction" pool mode + idle_timeout: 20, // Reduce idle timeout to 20 seconds to free up connections faster + connect_timeout: 10, // Reduce connect timeout to 10 seconds + max: 3, // Conservative limit - with multiple serverless functions, this prevents pool exhaustion + onnotice: () => {}, // Disable notices to reduce noise +}) + +const drizzleClient = drizzle(postgresClient, { schema }) declare global { var database: PostgresJsDatabase | undefined diff --git a/apps/sim/socket-server/database/operations.ts b/apps/sim/socket-server/database/operations.ts index 44092bdeb1..caf84c46cf 100644 --- a/apps/sim/socket-server/database/operations.ts +++ b/apps/sim/socket-server/database/operations.ts @@ -1,11 +1,31 @@ import { and, eq, or } from 'drizzle-orm' -import { db } from '../../db' +import { drizzle } from 'drizzle-orm/postgres-js' +import postgres from 'postgres' +import * as schema from '../../db/schema' import { workflow, workflowBlocks, workflowEdges, workflowSubflows } from '../../db/schema' +import { env } from '../../lib/env' import { createLogger } from '../../lib/logs/console-logger' import { loadWorkflowFromNormalizedTables } from '../../lib/workflows/db-helpers' const logger = createLogger('SocketDatabase') +// Create dedicated database connection for socket server with optimized settings +const connectionString = env.POSTGRES_URL ?? env.DATABASE_URL +const socketDb = drizzle( + postgres(connectionString, { + prepare: false, + idle_timeout: 10, // Shorter idle timeout for socket operations + connect_timeout: 5, // Faster connection timeout + max: 2, // Very small pool for socket server to avoid exhausting Supabase limit + onnotice: () => {}, // Disable notices + debug: false, // Disable debug for socket operations + }), + { schema } +) + +// Use dedicated connection for socket operations, fallback to shared db for compatibility +const db = socketDb + // Constants const DEFAULT_LOOP_ITERATIONS = 5 @@ -115,9 +135,20 @@ export async function getWorkflowState(workflowId: string) { // Persist workflow operation export async function persistWorkflowOperation(workflowId: string, operation: any) { + const startTime = Date.now() try { const { operation: op, target, payload, timestamp, userId } = operation + // Log high-frequency operations for monitoring + if (op === 'update-position' && Math.random() < 0.01) { + // Log 1% of position updates + logger.debug('Socket DB operation sample:', { + operation: op, + target, + workflowId: `${workflowId.substring(0, 8)}...`, + }) + } + await db.transaction(async (tx) => { // Update the workflow's last modified timestamp first await tx @@ -140,9 +171,22 @@ export async function persistWorkflowOperation(workflowId: string, operation: an throw new Error(`Unknown operation target: ${target}`) } }) + + // Log slow operations for monitoring + const duration = Date.now() - startTime + if (duration > 100) { + // Log operations taking more than 100ms + logger.warn('Slow socket DB operation:', { + operation: operation.operation, + target: operation.target, + duration: `${duration}ms`, + workflowId: `${workflowId.substring(0, 8)}...`, + }) + } } catch (error) { + const duration = Date.now() - startTime logger.error( - `❌ Error persisting workflow operation (${operation.operation} on ${operation.target}):`, + `❌ Error persisting workflow operation (${operation.operation} on ${operation.target}) after ${duration}ms:`, error ) throw error diff --git a/apps/sim/socket-server/rooms/manager.ts b/apps/sim/socket-server/rooms/manager.ts index 1875226a03..3159fe0d4f 100644 --- a/apps/sim/socket-server/rooms/manager.ts +++ b/apps/sim/socket-server/rooms/manager.ts @@ -1,9 +1,25 @@ import { and, eq, isNull } from 'drizzle-orm' +import { drizzle } from 'drizzle-orm/postgres-js' +import postgres from 'postgres' import type { Server } from 'socket.io' -import { db } from '../../db' +import * as schema from '../../db/schema' import { workflowBlocks, workflowEdges } from '../../db/schema' +import { env } from '../../lib/env' import { createLogger } from '../../lib/logs/console-logger' +// Create dedicated database connection for room manager +const connectionString = env.POSTGRES_URL ?? env.DATABASE_URL +const db = drizzle( + postgres(connectionString, { + prepare: false, + idle_timeout: 15, + connect_timeout: 5, + max: 1, // Minimal pool for room operations to conserve connections + onnotice: () => {}, + }), + { schema } +) + const logger = createLogger('RoomManager') export interface UserPresence { @@ -80,7 +96,7 @@ export class RoomManager { }) const socketsToDisconnect: string[] = [] - room.users.forEach((presence, socketId) => { + room.users.forEach((_presence, socketId) => { socketsToDisconnect.push(socketId) }) diff --git a/apps/sim/stores/workflows/registry/store.ts b/apps/sim/stores/workflows/registry/store.ts index 4968c2dac2..27ffabe4f0 100644 --- a/apps/sim/stores/workflows/registry/store.ts +++ b/apps/sim/stores/workflows/registry/store.ts @@ -267,41 +267,6 @@ export const useWorkflowRegistry = create()( await fetchWorkflowsFromDB(workspaceId) }, - // Handle cleanup on workspace deletion - handleWorkspaceDeletion: async (newWorkspaceId: string) => { - // Set transition state - setWorkspaceTransitioning(true) - - try { - logger.info(`Switching to new workspace after deletion: ${newWorkspaceId}`) - - // Reset all workflow state - resetWorkflowStores() - - // Set loading state while we fetch workflows - set({ - isLoading: true, - workflows: {}, - activeWorkflowId: null, - }) - - // Properly await workflow fetching to prevent race conditions - await fetchWorkflowsFromDB(newWorkspaceId) - - set({ isLoading: false }) - logger.info(`Successfully switched to workspace after deletion: ${newWorkspaceId}`) - } catch (error) { - logger.error('Error fetching workflows after workspace deletion:', { - error, - workspaceId: newWorkspaceId, - }) - set({ isLoading: false, error: 'Failed to load workspace data' }) - } finally { - // End transition state - setWorkspaceTransitioning(false) - } - }, - // Switch to workspace with comprehensive error handling and loading states switchToWorkspace: async (workspaceId: string) => { // Prevent multiple simultaneous transitions diff --git a/apps/sim/stores/workflows/registry/types.ts b/apps/sim/stores/workflows/registry/types.ts index 5e9234b28b..a540598fb0 100644 --- a/apps/sim/stores/workflows/registry/types.ts +++ b/apps/sim/stores/workflows/registry/types.ts @@ -32,9 +32,8 @@ export interface WorkflowRegistryState { export interface WorkflowRegistryActions { setLoading: (loading: boolean) => void setActiveWorkflow: (id: string) => Promise - switchToWorkspace: (id: string) => void + switchToWorkspace: (id: string) => Promise loadWorkflows: (workspaceId?: string) => Promise - handleWorkspaceDeletion: (newWorkspaceId: string) => void removeWorkflow: (id: string) => Promise updateWorkflow: (id: string, metadata: Partial) => Promise createWorkflow: (options?: {