fix(concurrency): cleanup worker code

Author: icecrasher321

README.md

@@ -74,10 +74,6 @@ docker compose -f docker-compose.prod.yml up -d
 
 Open [http://localhost:3000](http://localhost:3000)
 
-#### Background worker note
-
-The Docker Compose stack starts a dedicated worker container by default. If `REDIS_URL` is not configured, the worker will start, log that it is idle, and do no queue processing. This is expected. Queue-backed API, webhook, and schedule execution requires Redis; installs without Redis continue to use the inline execution path.
-
 Sim also supports local models via [Ollama](https://ollama.ai) and [vLLM](https://docs.vllm.ai/) — see the [Docker self-hosting docs](https://docs.sim.ai/self-hosting/docker) for setup details.
 
 ### Self-hosted: Manual Setup
@@ -117,12 +113,10 @@ cd packages/db && bunx drizzle-kit migrate --config=./drizzle.config.ts
 5. Start development servers:
 
 ```bash
-bun run dev:full  # Starts Next.js app, realtime socket server, and the BullMQ worker
+bun run dev:full  # Starts Next.js app and realtime socket server
 ```
 
-If `REDIS_URL` is not configured, the worker will remain idle and execution continues inline.
-
-Or run separately: `bun run dev` (Next.js), `cd apps/sim && bun run dev:sockets` (realtime), and `cd apps/sim && bun run worker` (BullMQ worker).
+Or run separately: `bun run dev` (Next.js) and `cd apps/sim && bun run dev:sockets` (realtime).
 
 ## Copilot API Keys
 

apps/docs/content/docs/en/execution/costs.mdx

@@ -195,17 +195,6 @@ 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 |

apps/sim/app/api/jobs/[jobId]/route.test.ts

@@ -4,19 +4,13 @@
 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(),
-}))
+const { mockCheckHybridAuth, mockGetJobQueue, mockVerifyWorkflowAccess, mockGetWorkflowById } =
+  vi.hoisted(() => ({
+    mockCheckHybridAuth: vi.fn(),
+    mockGetJobQueue: vi.fn(),
+    mockVerifyWorkflowAccess: vi.fn(),
+    mockGetWorkflowById: vi.fn(),
+  }))
 
 vi.mock('@sim/logger', () => ({
   createLogger: () => ({
@@ -32,19 +26,9 @@ vi.mock('@/lib/auth/hybrid', () => ({
 }))
 
 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'),
 }))
@@ -89,72 +73,78 @@ describe('GET /api/jobs/[jobId]', () => {
     })
   })
 
-  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,
+  it('returns pending status for a queued job', async () => {
+    mockGetJobQueue.mockResolvedValue({
+      getJob: vi.fn().mockResolvedValue({
+        id: 'job-1',
+        type: 'workflow-execution',
+        payload: {},
+        status: 'pending',
+        createdAt: new Date('2025-01-01T00:00:00Z'),
+        attempts: 0,
+        maxAttempts: 1,
+        metadata: {
+          workflowId: 'workflow-1',
+        },
+      }),
     })
 
     const response = await GET(createMockRequest(), {
-      params: Promise.resolve({ jobId: 'dispatch-1' }),
+      params: Promise.resolve({ jobId: 'job-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')
+    expect(body.status).toBe('pending')
   })
 
-  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 },
+  it('returns completed output from job', async () => {
+    mockGetJobQueue.mockResolvedValue({
+      getJob: vi.fn().mockResolvedValue({
+        id: 'job-2',
+        type: 'workflow-execution',
+        payload: {},
+        status: 'completed',
+        createdAt: new Date('2025-01-01T00:00:00Z'),
+        startedAt: new Date('2025-01-01T00:00:01Z'),
+        completedAt: new Date('2025-01-01T00:00:06Z'),
+        attempts: 1,
+        maxAttempts: 1,
+        output: { success: true },
+        metadata: {
+          workflowId: 'workflow-1',
+        },
+      }),
     })
 
     const response = await GET(createMockRequest(), {
-      params: Promise.resolve({ jobId: 'dispatch-2' }),
+      params: Promise.resolve({ jobId: 'job-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)
-
+  it('returns 404 when job does not exist', async () => {
     const response = await GET(createMockRequest(), {
       params: Promise.resolve({ jobId: 'missing-job' }),
     })
 
     expect(response.status).toBe(404)
   })
+
+  it('returns 401 for unauthenticated requests', async () => {
+    mockCheckHybridAuth.mockResolvedValue({
+      success: false,
+      error: 'Not authenticated',
+    })
+
+    const response = await GET(createMockRequest(), {
+      params: Promise.resolve({ jobId: 'job-1' }),
+    })
+
+    expect(response.status).toBe(401)
+  })
 })

apps/sim/app/api/jobs/[jobId]/route.ts

@@ -2,13 +2,27 @@ import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
 import { checkHybridAuth } from '@/lib/auth/hybrid'
 import { getJobQueue } from '@/lib/core/async-jobs'
+import type { Job } from '@/lib/core/async-jobs/types'
 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')
 
+function presentJobStatus(job: Job) {
+  return {
+    status: job.status,
+    metadata: {
+      createdAt: job.createdAt.toISOString(),
+      startedAt: job.startedAt?.toISOString(),
+      completedAt: job.completedAt?.toISOString(),
+      attempts: job.attempts,
+      maxAttempts: job.maxAttempts,
+    },
+    output: job.output,
+    error: job.error,
+  }
+}
+
 export async function GET(
   request: NextRequest,
   { params }: { params: Promise<{ jobId: string }> }
@@ -25,15 +39,14 @@ export async function GET(
 
     const authenticatedUserId = authResult.userId
 
-    const dispatchJob = await getDispatchJobRecord(taskId)
     const jobQueue = await getJobQueue()
-    const job = dispatchJob ? null : await jobQueue.getJob(taskId)
+    const job = await jobQueue.getJob(taskId)
 
-    if (!job && !dispatchJob) {
+    if (!job) {
       return createErrorResponse('Task not found', 404)
     }
 
-    const metadataToCheck = dispatchJob?.metadata ?? job?.metadata
+    const metadataToCheck = job.metadata
 
     if (metadataToCheck?.workflowId) {
       const { verifyWorkflowAccess } = await import('@/socket/middleware/permissions')
@@ -61,7 +74,7 @@ export async function GET(
       return createErrorResponse('Access denied', 403)
     }
 
-    const presented = presentDispatchOrJobStatus(dispatchJob, job)
+    const presented = presentJobStatus(job)
     const response: any = {
       success: true,
       taskId,
@@ -71,9 +84,6 @@ export async function GET(
 
     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)
   } catch (error: any) {

apps/sim/app/api/schedules/execute/route.test.ts

@@ -14,7 +14,6 @@ const {
   mockDbReturning,
   mockDbUpdate,
   mockEnqueue,
-  mockEnqueueWorkspaceDispatch,
   mockStartJob,
   mockCompleteJob,
   mockMarkJobFailed,
@@ -24,7 +23,6 @@ const {
   const mockDbSet = vi.fn().mockReturnValue({ where: mockDbWhere })
   const mockDbUpdate = vi.fn().mockReturnValue({ set: mockDbSet })
   const mockEnqueue = vi.fn().mockResolvedValue('job-id-1')
-  const mockEnqueueWorkspaceDispatch = vi.fn().mockResolvedValue('job-id-1')
   const mockStartJob = vi.fn().mockResolvedValue(undefined)
   const mockCompleteJob = vi.fn().mockResolvedValue(undefined)
   const mockMarkJobFailed = vi.fn().mockResolvedValue(undefined)
@@ -42,7 +40,6 @@ const {
     mockDbReturning,
     mockDbUpdate,
     mockEnqueue,
-    mockEnqueueWorkspaceDispatch,
     mockStartJob,
     mockCompleteJob,
     mockMarkJobFailed,
@@ -75,15 +72,6 @@ vi.mock('@/lib/core/async-jobs', () => ({
   shouldExecuteInline: vi.fn().mockReturnValue(false),
 }))
 
-vi.mock('@/lib/core/bullmq', () => ({
-  isBullMQEnabled: vi.fn().mockReturnValue(true),
-  createBullMQJobData: vi.fn((payload: unknown) => ({ payload })),
-}))
-
-vi.mock('@/lib/core/workspace-dispatch', () => ({
-  enqueueWorkspaceDispatch: mockEnqueueWorkspaceDispatch,
-}))
-
 vi.mock('@/lib/workflows/utils', () => ({
   getWorkflowById: vi.fn().mockResolvedValue({
     id: 'workflow-1',
@@ -246,29 +234,19 @@ describe('Scheduled Workflow Execution API Route', () => {
     expect(data).toHaveProperty('executedCount', 2)
   })
 
-  it('should queue mothership jobs to BullMQ when available', async () => {
+  it('should execute mothership jobs inline', async () => {
     mockDbReturning.mockReturnValueOnce([]).mockReturnValueOnce(SINGLE_JOB)
 
     const response = await GET(createMockRequest())
 
     expect(response.status).toBe(200)
-    expect(mockEnqueueWorkspaceDispatch).toHaveBeenCalledWith(
+    expect(mockExecuteJobInline).toHaveBeenCalledWith(
       expect.objectContaining({
-        workspaceId: 'workspace-1',
-        lane: 'runtime',
-        queueName: 'mothership-job-execution',
-        bullmqJobName: 'mothership-job-execution',
-        bullmqPayload: {
-          payload: {
-            scheduleId: 'job-1',
-            cronExpression: '0 * * * *',
-            failedCount: 0,
-            now: expect.any(String),
-          },
-        },
+        scheduleId: 'job-1',
+        cronExpression: '0 * * * *',
+        failedCount: 0,
       })
     )
-    expect(mockExecuteJobInline).not.toHaveBeenCalled()
   })
 
   it('should enqueue preassigned correlation metadata for schedules', async () => {
@@ -277,25 +255,23 @@ describe('Scheduled Workflow Execution API Route', () => {
     const response = await GET(createMockRequest())
 
     expect(response.status).toBe(200)
-    expect(mockEnqueueWorkspaceDispatch).toHaveBeenCalledWith(
+    expect(mockEnqueue).toHaveBeenCalledWith(
+      'schedule-execution',
+      expect.objectContaining({
+        scheduleId: 'schedule-1',
+        workflowId: 'workflow-1',
+        executionId: 'schedule-execution-1',
+      }),
       expect.objectContaining({
-        id: 'schedule-execution-1',
-        workspaceId: 'workspace-1',
-        lane: 'runtime',
-        queueName: 'schedule-execution',
-        bullmqJobName: 'schedule-execution',
-        metadata: {
+        metadata: expect.objectContaining({
           workflowId: 'workflow-1',
-          correlation: {
+          correlation: expect.objectContaining({
             executionId: 'schedule-execution-1',
             requestId: 'test-request-id',
             source: 'schedule',
             workflowId: 'workflow-1',
-            scheduleId: 'schedule-1',
-            triggerType: 'schedule',
-            scheduledFor: '2025-01-01T00:00:00.000Z',
-          },
-        },
+          }),
+        }),
       })
     )
   })