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 <vikhyathmondreti@Vikhyaths-Air.attlocal.net>
This commit is contained in:
Vikhyath Mondreti
2025-06-27 15:21:46 -07:00
committed by GitHub
co-authored by Vikhyath Mondreti
parent e4523961ea
commit bf9b3fdc99
7 changed files with 109 additions and 64 deletions
@@ -333,7 +333,7 @@ export const WorkspaceHeader = React.memo<WorkspaceHeaderProps>(
}, [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<WorkspaceHeaderProps>(
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<WorkspaceHeaderProps>(
// 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<WorkspaceHeaderProps>(
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)
+5 -5
View File
@@ -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
}
+25 -8
View File
@@ -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<typeof schema> | undefined
+46 -2
View File
@@ -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
+18 -2
View File
@@ -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)
})
@@ -267,41 +267,6 @@ export const useWorkflowRegistry = create<WorkflowRegistry>()(
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
+1 -2
View File
@@ -32,9 +32,8 @@ export interface WorkflowRegistryState {
export interface WorkflowRegistryActions {
setLoading: (loading: boolean) => void
setActiveWorkflow: (id: string) => Promise<void>
switchToWorkspace: (id: string) => void
switchToWorkspace: (id: string) => Promise<void>
loadWorkflows: (workspaceId?: string) => Promise<void>
handleWorkspaceDeletion: (newWorkspaceId: string) => void
removeWorkflow: (id: string) => Promise<void>
updateWorkflow: (id: string, metadata: Partial<WorkflowMetadata>) => Promise<void>
createWorkflow: (options?: {