feat(concurrency): bullmq based concurrency control system (#3605)

* feat(concurrency): bullmq based queueing system * fix bun lock * remove manual execs off queues * address comments * fix legacy team limits * cleanup enterprise typing code * inline child triggers * fix status check * address more comments * optimize reconciler scan * remove dead code * add to landing page * Add load testing framework * update bullmq * fix * fix headless path --------- Co-authored-by: Theodore Li <teddy@zenobiapay.com>
simstudioai · waleedlatif1 · Mar 30, 2026 · Mar 27, 2026 · Mar 27, 2026 · Mar 27, 2026
commit dda012eae9f1ca55b388b6f703046528280a3033
diff --git a/apps/docs/content/docs/en/execution/costs.mdx b/apps/docs/content/docs/en/execution/costs.mdx
@@ -195,6 +195,17 @@ By default, your usage is capped at the credits included in your plan. To allow
 
 Max (individual) shares the same rate limits as team plans. Team plans (Pro or Max for Teams) use the Max-tier rate limits.
 
+### Concurrent Execution Limits
+
+| Plan | Concurrent Executions |
+|------|----------------------|
+| **Free** | 5 |
+| **Pro** | 50 |
+| **Max / Team** | 200 |
+| **Enterprise** | 200 (customizable) |
+
+Concurrent execution limits control how many workflow executions can run simultaneously within a workspace. When the limit is reached, new executions are queued and admitted as running executions complete. Manual runs from the editor are not subject to these limits.
+
 ### File Storage
 
 | Plan | Storage |

diff --git a/apps/sim/app/(home)/components/pricing/pricing.tsx b/apps/sim/app/(home)/components/pricing/pricing.tsx
@@ -25,6 +25,7 @@ const PRICING_TIERS: PricingTier[] = [
       '5GB file storage',
       '3 tables · 1,000 rows each',
       '5 min execution limit',
+      '5 concurrent/workspace',
       '7-day log retention',
       'CLI/SDK/MCP Access',
     ],
@@ -42,6 +43,7 @@ const PRICING_TIERS: PricingTier[] = [
       '50GB file storage',
       '25 tables · 5,000 rows each',
       '50 min execution · 150 runs/min',
+      '50 concurrent/workspace',
       'Unlimited log retention',
       'CLI/SDK/MCP Access',
     ],
@@ -59,6 +61,7 @@ const PRICING_TIERS: PricingTier[] = [
       '500GB file storage',
       '25 tables · 5,000 rows each',
       '50 min execution · 300 runs/min',
+      '200 concurrent/workspace',
       'Unlimited log retention',
       'CLI/SDK/MCP Access',
     ],
@@ -75,6 +78,7 @@ const PRICING_TIERS: PricingTier[] = [
       'Custom file storage',
       '10,000 tables · 1M rows each',
       'Custom execution limits',
+      'Custom concurrency limits',
       'Unlimited log retention',
       'SSO & SCIM · SOC2 & HIPAA',
       'Self hosting · Dedicated support',

diff --git a/apps/sim/app/api/copilot/chat/route.ts b/apps/sim/app/api/copilot/chat/route.ts
@@ -5,6 +5,7 @@ import { and, desc, eq, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { z } from 'zod'
 import { getSession } from '@/lib/auth'
+import { createRunSegment } from '@/lib/copilot/async-runs/repository'
 import { getAccessibleCopilotChat, resolveOrCreateChat } from '@/lib/copilot/chat-lifecycle'
 import { buildCopilotRequestPayload } from '@/lib/copilot/chat-payload'
 import {
@@ -539,10 +540,26 @@ export async function POST(req: NextRequest) {
       return new Response(sseStream, { headers: SSE_RESPONSE_HEADERS })
     }
 
+    const nsExecutionId = crypto.randomUUID()
+    const nsRunId = crypto.randomUUID()
+
+    if (actualChatId) {
+      await createRunSegment({
+        id: nsRunId,
+        executionId: nsExecutionId,
+        chatId: actualChatId,
+        userId: authenticatedUserId,
+        workflowId,
+        streamId: userMessageIdToUse,
+      }).catch(() => {})
+    }
+
     const nonStreamingResult = await orchestrateCopilotStream(requestPayload, {
       userId: authenticatedUserId,
       workflowId,
       chatId: actualChatId,
+      executionId: nsExecutionId,
+      runId: nsRunId,
       goRoute: '/api/copilot',
       autoExecuteTools: true,
       interactive: true,

diff --git a/apps/sim/app/api/jobs/[jobId]/route.test.ts b/apps/sim/app/api/jobs/[jobId]/route.test.ts
@@ -0,0 +1,160 @@
+/**
+ * @vitest-environment node
+ */
+import type { NextRequest } from 'next/server'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const {
+  mockCheckHybridAuth,
+  mockGetDispatchJobRecord,
+  mockGetJobQueue,
+  mockVerifyWorkflowAccess,
+  mockGetWorkflowById,
+} = vi.hoisted(() => ({
+  mockCheckHybridAuth: vi.fn(),
+  mockGetDispatchJobRecord: vi.fn(),
+  mockGetJobQueue: vi.fn(),
+  mockVerifyWorkflowAccess: vi.fn(),
+  mockGetWorkflowById: vi.fn(),
+}))
+
+vi.mock('@sim/logger', () => ({
+  createLogger: () => ({
+    info: vi.fn(),
+    warn: vi.fn(),
+    error: vi.fn(),
+    debug: vi.fn(),
+  }),
+}))
+
+vi.mock('@/lib/auth/hybrid', () => ({
+  checkHybridAuth: mockCheckHybridAuth,
+}))
+
+vi.mock('@/lib/core/async-jobs', () => ({
+  JOB_STATUS: {
+    PENDING: 'pending',
+    PROCESSING: 'processing',
+    COMPLETED: 'completed',
+    FAILED: 'failed',
+  },
+  getJobQueue: mockGetJobQueue,
+}))
+
+vi.mock('@/lib/core/workspace-dispatch/store', () => ({
+  getDispatchJobRecord: mockGetDispatchJobRecord,
+}))
+
+vi.mock('@/lib/core/utils/request', () => ({
+  generateRequestId: vi.fn().mockReturnValue('request-1'),
+}))
+
+vi.mock('@/socket/middleware/permissions', () => ({
+  verifyWorkflowAccess: mockVerifyWorkflowAccess,
+}))
+
+vi.mock('@/lib/workflows/utils', () => ({
+  getWorkflowById: mockGetWorkflowById,
+}))
+
+import { GET } from './route'
+
+function createMockRequest(): NextRequest {
+  return {
+    headers: {
+      get: () => null,
+    },
+  } as NextRequest
+}
+
+describe('GET /api/jobs/[jobId]', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+
+    mockCheckHybridAuth.mockResolvedValue({
+      success: true,
+      userId: 'user-1',
+      apiKeyType: undefined,
+      workspaceId: undefined,
+    })
+
+    mockVerifyWorkflowAccess.mockResolvedValue({ hasAccess: true })
+    mockGetWorkflowById.mockResolvedValue({
+      id: 'workflow-1',
+      workspaceId: 'workspace-1',
+    })
+
+    mockGetJobQueue.mockResolvedValue({
+      getJob: vi.fn().mockResolvedValue(null),
+    })
+  })
+
+  it('returns dispatcher-aware waiting status with metadata', async () => {
+    mockGetDispatchJobRecord.mockResolvedValue({
+      id: 'dispatch-1',
+      workspaceId: 'workspace-1',
+      lane: 'runtime',
+      queueName: 'workflow-execution',
+      bullmqJobName: 'workflow-execution',
+      bullmqPayload: {},
+      metadata: {
+        workflowId: 'workflow-1',
+      },
+      priority: 10,
+      status: 'waiting',
+      createdAt: 1000,
+      admittedAt: 2000,
+    })
+
+    const response = await GET(createMockRequest(), {
+      params: Promise.resolve({ jobId: 'dispatch-1' }),
+    })
+    const body = await response.json()
+
+    expect(response.status).toBe(200)
+    expect(body.status).toBe('waiting')
+    expect(body.metadata.queueName).toBe('workflow-execution')
+    expect(body.metadata.lane).toBe('runtime')
+    expect(body.metadata.workspaceId).toBe('workspace-1')
+  })
+
+  it('returns completed output from dispatch state', async () => {
+    mockGetDispatchJobRecord.mockResolvedValue({
+      id: 'dispatch-2',
+      workspaceId: 'workspace-1',
+      lane: 'interactive',
+      queueName: 'workflow-execution',
+      bullmqJobName: 'direct-workflow-execution',
+      bullmqPayload: {},
+      metadata: {
+        workflowId: 'workflow-1',
+      },
+      priority: 1,
+      status: 'completed',
+      createdAt: 1000,
+      startedAt: 2000,
+      completedAt: 7000,
+      output: { success: true },
+    })
+
+    const response = await GET(createMockRequest(), {
+      params: Promise.resolve({ jobId: 'dispatch-2' }),
+    })
+    const body = await response.json()
+
+    expect(response.status).toBe(200)
+    expect(body.status).toBe('completed')
+    expect(body.output).toEqual({ success: true })
+    expect(body.metadata.duration).toBe(5000)
+  })
+
+  it('returns 404 when neither dispatch nor BullMQ job exists', async () => {
+    mockGetDispatchJobRecord.mockResolvedValue(null)
+
+    const response = await GET(createMockRequest(), {
+      params: Promise.resolve({ jobId: 'missing-job' }),
+    })
+
+    expect(response.status).toBe(404)
+  })
+})
diff --git a/apps/sim/app/api/jobs/[jobId]/route.ts b/apps/sim/app/api/jobs/[jobId]/route.ts
@@ -1,8 +1,10 @@
 import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
 import { checkHybridAuth } from '@/lib/auth/hybrid'
-import { getJobQueue, JOB_STATUS } from '@/lib/core/async-jobs'
+import { getJobQueue } from '@/lib/core/async-jobs'
 import { generateRequestId } from '@/lib/core/utils/request'
+import { presentDispatchOrJobStatus } from '@/lib/core/workspace-dispatch/status'
+import { getDispatchJobRecord } from '@/lib/core/workspace-dispatch/store'
 import { createErrorResponse } from '@/app/api/workflows/utils'
 
 const logger = createLogger('TaskStatusAPI')
@@ -23,68 +25,54 @@ export async function GET(
 
     const authenticatedUserId = authResult.userId
 
+    const dispatchJob = await getDispatchJobRecord(taskId)
     const jobQueue = await getJobQueue()
-    const job = await jobQueue.getJob(taskId)
+    const job = dispatchJob ? null : await jobQueue.getJob(taskId)
 
-    if (!job) {
+    if (!job && !dispatchJob) {
       return createErrorResponse('Task not found', 404)
     }
 
-    if (job.metadata?.workflowId) {
+    const metadataToCheck = dispatchJob?.metadata ?? job?.metadata
+
+    if (metadataToCheck?.workflowId) {
       const { verifyWorkflowAccess } = await import('@/socket/middleware/permissions')
       const accessCheck = await verifyWorkflowAccess(
         authenticatedUserId,
-        job.metadata.workflowId as string
+        metadataToCheck.workflowId as string
       )
       if (!accessCheck.hasAccess) {
-        logger.warn(`[${requestId}] Access denied to workflow ${job.metadata.workflowId}`)
+        logger.warn(`[${requestId}] Access denied to workflow ${metadataToCheck.workflowId}`)
         return createErrorResponse('Access denied', 403)
       }
 
       if (authResult.apiKeyType === 'workspace' && authResult.workspaceId) {
         const { getWorkflowById } = await import('@/lib/workflows/utils')
-        const workflow = await getWorkflowById(job.metadata.workflowId as string)
+        const workflow = await getWorkflowById(metadataToCheck.workflowId as string)
         if (!workflow?.workspaceId || workflow.workspaceId !== authResult.workspaceId) {
           return createErrorResponse('API key is not authorized for this workspace', 403)
         }
       }
-    } else if (job.metadata?.userId && job.metadata.userId !== authenticatedUserId) {
-      logger.warn(`[${requestId}] Access denied to user ${job.metadata.userId}`)
+    } else if (metadataToCheck?.userId && metadataToCheck.userId !== authenticatedUserId) {
+      logger.warn(`[${requestId}] Access denied to user ${metadataToCheck.userId}`)
       return createErrorResponse('Access denied', 403)
-    } else if (!job.metadata?.userId && !job.metadata?.workflowId) {
+    } else if (!metadataToCheck?.userId && !metadataToCheck?.workflowId) {
       logger.warn(`[${requestId}] Access denied to job ${taskId}`)
       return createErrorResponse('Access denied', 403)
     }
 
-    const mappedStatus = job.status === JOB_STATUS.PENDING ? 'queued' : job.status
-
+    const presented = presentDispatchOrJobStatus(dispatchJob, job)
     const response: any = {
       success: true,
       taskId,
-      status: mappedStatus,
-      metadata: {
-        startedAt: job.startedAt,
-      },
-    }
-
-    if (job.status === JOB_STATUS.COMPLETED) {
-      response.output = job.output
-      response.metadata.completedAt = job.completedAt
-      if (job.startedAt && job.completedAt) {
-        response.metadata.duration = job.completedAt.getTime() - job.startedAt.getTime()
-      }
-    }
-
-    if (job.status === JOB_STATUS.FAILED) {
-      response.error = job.error
-      response.metadata.completedAt = job.completedAt
-      if (job.startedAt && job.completedAt) {
-        response.metadata.duration = job.completedAt.getTime() - job.startedAt.getTime()
-      }
+      status: presented.status,
+      metadata: presented.metadata,
     }
 
-    if (job.status === JOB_STATUS.PROCESSING || job.status === JOB_STATUS.PENDING) {
-      response.estimatedDuration = 300000
+    if (presented.output !== undefined) response.output = presented.output
+    if (presented.error !== undefined) response.error = presented.error
+    if (presented.estimatedDuration !== undefined) {
+      response.estimatedDuration = presented.estimatedDuration
     }
 
     return NextResponse.json(response)

diff --git a/apps/sim/app/api/mcp/copilot/route.ts b/apps/sim/app/api/mcp/copilot/route.ts
@@ -18,6 +18,7 @@ import { eq, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { validateOAuthAccessToken } from '@/lib/auth/oauth-token'
 import { getHighestPrioritySubscription } from '@/lib/billing/core/subscription'
+import { createRunSegment } from '@/lib/copilot/async-runs/repository'
 import { ORCHESTRATION_TIMEOUT_MS, SIM_AGENT_API_URL } from '@/lib/copilot/constants'
 import { orchestrateCopilotStream } from '@/lib/copilot/orchestrator'
 import { orchestrateSubagentStream } from '@/lib/copilot/orchestrator/subagent'
@@ -727,10 +728,25 @@ async function handleBuildToolCall(
       chatId,
     }
 
+    const executionId = crypto.randomUUID()
+    const runId = crypto.randomUUID()
+    const messageId = requestPayload.messageId as string
+
+    await createRunSegment({
+      id: runId,
+      executionId,
+      chatId,
+      userId,
+      workflowId: resolved.workflowId,
+      streamId: messageId,
+    }).catch(() => {})
+
     const result = await orchestrateCopilotStream(requestPayload, {
       userId,
       workflowId: resolved.workflowId,
       chatId,
+      executionId,
+      runId,
       goRoute: '/api/mcp',
       autoExecuteTools: true,
       timeout: ORCHESTRATION_TIMEOUT_MS,