fix(cjson): santize bullmq payloads accurately

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

@@ -4,8 +4,9 @@ import { and, eq, isNull, lt, lte, ne, not, or, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { v4 as uuidv4 } from 'uuid'
 import { verifyCronAuth } from '@/lib/auth/internal'
-import { getJobQueue, shouldExecuteInline } from '@/lib/core/async-jobs'
+import { getJobQueue } from '@/lib/core/async-jobs'
 import { createBullMQJobData, isBullMQEnabled } from '@/lib/core/bullmq'
+import { isTriggerDevEnabled } from '@/lib/core/config/feature-flags'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { enqueueWorkspaceDispatch } from '@/lib/core/workspace-dispatch'
 import {
@@ -157,7 +158,7 @@ export async function GET(request: NextRequest) {
           `[${requestId}] Queued schedule execution task ${jobId} for workflow ${schedule.workflowId}`
         )
 
-        if (shouldExecuteInline()) {
+        if (!isBullMQEnabled() && !isTriggerDevEnabled) {
           try {
             await jobQueue.startJob(jobId)
             const output = await executeScheduleJob(payload)

apps/sim/app/api/workflows/[id]/execute/route.async.test.ts

@@ -46,15 +46,24 @@ vi.mock('@/lib/core/async-jobs', () => ({
     completeJob: vi.fn(),
     markJobFailed: vi.fn(),
   }),
-  shouldExecuteInline: vi.fn().mockReturnValue(false),
-  shouldUseBullMQ: vi.fn().mockReturnValue(true),
 }))
 
 vi.mock('@/lib/core/bullmq', () => ({
-  createBullMQJobData: vi.fn((payload: unknown, metadata?: unknown) => ({ payload, metadata })),
+  createBullMQJobData: vi.fn((payload: unknown, metadata?: unknown) => ({
+    payload,
+    metadata: metadata ?? {},
+  })),
+  isBullMQEnabled: vi.fn().mockReturnValue(true),
 }))
 
 vi.mock('@/lib/core/workspace-dispatch', () => ({
+  DispatchQueueFullError: class DispatchQueueFullError extends Error {
+    statusCode = 503
+    constructor(scope: string, depth: number, limit: number) {
+      super(`${scope} queue at capacity (${depth}/${limit})`)
+      this.name = 'DispatchQueueFullError'
+    }
+  },
   enqueueWorkspaceDispatch: mockEnqueueWorkspaceDispatch,
   waitForDispatchJob: vi.fn(),
 }))

apps/sim/app/api/workflows/[id]/execute/route.ts

@@ -4,8 +4,9 @@ import { validate as uuidValidate, v4 as uuidv4 } from 'uuid'
 import { z } from 'zod'
 import { AuthType, checkHybridAuth, hasExternalApiCredentials } from '@/lib/auth/hybrid'
 import { admissionRejectedResponse, tryAdmit } from '@/lib/core/admission/gate'
-import { getJobQueue, shouldExecuteInline, shouldUseBullMQ } from '@/lib/core/async-jobs'
-import { createBullMQJobData } from '@/lib/core/bullmq'
+import { getJobQueue } from '@/lib/core/async-jobs'
+import { createBullMQJobData, isBullMQEnabled } from '@/lib/core/bullmq'
+import { isTriggerDevEnabled } from '@/lib/core/config/feature-flags'
 import {
   createTimeoutAbortController,
   getTimeoutErrorMessage,
@@ -216,7 +217,7 @@ async function handleAsyncExecution(params: AsyncExecutionParams): Promise<NextR
   }
 
   try {
-    const useBullMQ = shouldUseBullMQ()
+    const useBullMQ = isBullMQEnabled()
     const jobQueue = useBullMQ ? null : await getJobQueue()
     const jobId = useBullMQ
       ? await enqueueWorkspaceDispatch({
@@ -242,7 +243,7 @@ async function handleAsyncExecution(params: AsyncExecutionParams): Promise<NextR
 
     asyncLogger.info('Queued async workflow execution', { jobId })
 
-    if (shouldExecuteInline() && jobQueue) {
+    if (!isBullMQEnabled() && !isTriggerDevEnabled && jobQueue) {
       const inlineJobQueue = jobQueue
       void (async () => {
         try {
@@ -793,7 +794,7 @@ async function handleExecutePost(
 
       const executionVariables = cachedWorkflowData?.variables ?? workflow.variables ?? {}
 
-      if (shouldUseBullMQ() && !INLINE_TRIGGER_TYPES.has(triggerType)) {
+      if (isBullMQEnabled() && !INLINE_TRIGGER_TYPES.has(triggerType)) {
         try {
           const dispatchJobId = await enqueueDirectWorkflowExecution(
             {
@@ -993,7 +994,7 @@ async function handleExecutePost(
     }
 
     if (shouldUseDraftState) {
-      const shouldDispatchViaQueue = shouldUseBullMQ() && !INLINE_TRIGGER_TYPES.has(triggerType)
+      const shouldDispatchViaQueue = isBullMQEnabled() && !INLINE_TRIGGER_TYPES.has(triggerType)
       if (shouldDispatchViaQueue) {
         const metadata: ExecutionMetadata = {
           requestId,

apps/sim/background/workflow-execution.ts

@@ -3,6 +3,7 @@ import { task } from '@trigger.dev/sdk'
 import { v4 as uuidv4 } from 'uuid'
 import type { AsyncExecutionCorrelation } from '@/lib/core/async-jobs/types'
 import { createTimeoutAbortController, getTimeoutErrorMessage } from '@/lib/core/execution-limits'
+import { ensureArray } from '@/lib/core/utils/json-sanitize'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
@@ -110,7 +111,7 @@ export async function executeWorkflowJob(payload: WorkflowExecutionPayload) {
       useDraftState: false,
       startTime: new Date().toISOString(),
       isClientSession: false,
-      callChain: payload.callChain,
+      callChain: ensureArray(payload.callChain),
       correlation,
     }
 

apps/sim/executor/execution/block-executor.ts

@@ -148,7 +148,7 @@ export class BlockExecutor {
             block,
             streamingExec,
             resolvedInputs,
-            ctx.selectedOutputs ?? []
+            Array.isArray(ctx.selectedOutputs) ? ctx.selectedOutputs : []
           )
         }
 

apps/sim/executor/execution/executor.ts

@@ -325,7 +325,9 @@ export class DAGExecutor {
         : new Set(),
       workflow: this.workflow,
       stream: this.contextExtensions.stream ?? false,
-      selectedOutputs: this.contextExtensions.selectedOutputs ?? [],
+      selectedOutputs: Array.isArray(this.contextExtensions.selectedOutputs)
+        ? this.contextExtensions.selectedOutputs
+        : [],
       edges: this.contextExtensions.edges ?? [],
       onStream: this.contextExtensions.onStream,
       onBlockStart: this.contextExtensions.onBlockStart,

apps/sim/executor/execution/snapshot-serializer.ts

@@ -119,7 +119,7 @@ export function serializePauseSnapshot(
     context.workflow,
     {},
     context.workflowVariables ?? {},
-    context.selectedOutputs ?? [],
+    Array.isArray(context.selectedOutputs) ? context.selectedOutputs : [],
     state
   )
 

apps/sim/executor/execution/snapshot.ts

@@ -28,7 +28,7 @@ export class ExecutionSnapshot {
       data.workflow,
       data.input,
       data.workflowVariables,
-      data.selectedOutputs,
+      Array.isArray(data.selectedOutputs) ? data.selectedOutputs : [],
       data.state
     )
   }

apps/sim/lib/core/async-jobs/config.ts

@@ -95,18 +95,6 @@ export async function getInlineJobQueue(): Promise<JobQueueBackend> {
   return cachedInlineBackend
 }
 
-/**
- * Checks if jobs should be executed inline in-process.
- * Database fallback is the only mode that still relies on inline execution.
- */
-export function shouldExecuteInline(): boolean {
-  return getAsyncBackendType() === 'database'
-}
-
-export function shouldUseBullMQ(): boolean {
-  return isBullMQEnabled()
-}
-
 /**
  * Resets the cached backend (useful for testing)
  */

apps/sim/lib/core/async-jobs/index.ts

@@ -4,8 +4,6 @@ export {
   getInlineJobQueue,
   getJobQueue,
   resetJobQueueCache,
-  shouldExecuteInline,
-  shouldUseBullMQ,
 } from './config'
 export type {
   AsyncBackendType,

apps/sim/lib/core/bullmq/connection.ts

@@ -1,8 +1,8 @@
 import type { ConnectionOptions } from 'bullmq'
-import { env, isTruthy } from '@/lib/core/config/env'
+import { env } from '@/lib/core/config/env'
 
 export function isBullMQEnabled(): boolean {
-  return isTruthy(env.CONCURRENCY_CONTROL_ENABLED) && Boolean(env.REDIS_URL)
+  return Boolean(env.REDIS_URL)
 }
 
 export function getBullMQConnectionOptions(): ConnectionOptions {

apps/sim/lib/core/bullmq/queues.ts

@@ -28,22 +28,22 @@ function getQueueDefaultOptions(type: JobType) {
       return {
         attempts: 3,
         backoff: { type: 'exponential' as const, delay: 1000 },
-        removeOnComplete: { age: 24 * 60 * 60 },
-        removeOnFail: { age: 7 * 24 * 60 * 60 },
+        removeOnComplete: { age: 2 * 60 * 60, count: 1000 },
+        removeOnFail: { age: 2 * 60 * 60, count: 500 },
       }
     case 'webhook-execution':
       return {
         attempts: 2,
         backoff: { type: 'exponential' as const, delay: 2000 },
-        removeOnComplete: { age: 24 * 60 * 60 },
-        removeOnFail: { age: 3 * 24 * 60 * 60 },
+        removeOnComplete: { age: 2 * 60 * 60, count: 1000 },
+        removeOnFail: { age: 2 * 60 * 60, count: 500 },
       }
     case 'schedule-execution':
       return {
         attempts: 2,
         backoff: { type: 'exponential' as const, delay: 5000 },
-        removeOnComplete: { age: 24 * 60 * 60 },
-        removeOnFail: { age: 3 * 24 * 60 * 60 },
+        removeOnComplete: { age: 2 * 60 * 60, count: 1000 },
+        removeOnFail: { age: 2 * 60 * 60, count: 500 },
       }
   }
 }
@@ -69,8 +69,8 @@ function createNamedQueue(
         defaultJobOptions: {
           attempts: 3,
           backoff: { type: 'exponential', delay: 5000 },
-          removeOnComplete: { age: 24 * 60 * 60 },
-          removeOnFail: { age: 7 * 24 * 60 * 60 },
+          removeOnComplete: { age: 2 * 60 * 60, count: 500 },
+          removeOnFail: { age: 2 * 60 * 60, count: 200 },
         },
       })
     case KNOWLEDGE_DOCUMENT_PROCESSING_QUEUE:
@@ -79,26 +79,26 @@ function createNamedQueue(
         defaultJobOptions: {
           attempts: 3,
           backoff: { type: 'exponential', delay: 1000 },
-          removeOnComplete: { age: 24 * 60 * 60 },
-          removeOnFail: { age: 7 * 24 * 60 * 60 },
+          removeOnComplete: { age: 2 * 60 * 60, count: 500 },
+          removeOnFail: { age: 2 * 60 * 60, count: 200 },
         },
       })
     case MOTHERSHIP_JOB_EXECUTION_QUEUE:
       return new Queue(name, {
         connection: getBullMQConnectionOptions(),
         defaultJobOptions: {
           attempts: 1,
-          removeOnComplete: { age: 24 * 60 * 60 },
-          removeOnFail: { age: 7 * 24 * 60 * 60 },
+          removeOnComplete: { age: 2 * 60 * 60, count: 500 },
+          removeOnFail: { age: 2 * 60 * 60, count: 200 },
         },
       })
     case WORKSPACE_NOTIFICATION_DELIVERY_QUEUE:
       return new Queue(name, {
         connection: getBullMQConnectionOptions(),
         defaultJobOptions: {
           attempts: 1,
-          removeOnComplete: { age: 24 * 60 * 60 },
-          removeOnFail: { age: 7 * 24 * 60 * 60 },
+          removeOnComplete: { age: 2 * 60 * 60, count: 500 },
+          removeOnFail: { age: 2 * 60 * 60, count: 200 },
         },
       })
   }

apps/sim/lib/core/config/env.ts

@@ -187,7 +187,6 @@ export const env = createEnv({
     FREE_PLAN_LOG_RETENTION_DAYS:          z.string().optional(),                  // Log retention days for free plan users
 
     // Admission & Burst Protection
-    CONCURRENCY_CONTROL_ENABLED:           z.string().optional().default('false'),  // Set to 'true' to enable BullMQ-based concurrency control (default: inline execution)
     ADMISSION_GATE_MAX_INFLIGHT:           z.string().optional().default('500'),   // Max concurrent in-flight execution requests per pod
     DISPATCH_MAX_QUEUE_PER_WORKSPACE:      z.string().optional().default('1000'),  // Max queued dispatch jobs per workspace
     DISPATCH_MAX_QUEUE_GLOBAL:             z.string().optional().default('50000'), // Max queued dispatch jobs globally

apps/sim/lib/core/utils/json-sanitize.ts

@@ -0,0 +1,68 @@
+/**
+ * Sanitization for JSON data round-tripped through Redis Lua cjson.
+ *
+ * Lua's cjson library cannot distinguish between empty arrays `[]` and empty objects `{}`.
+ * Both serialize to `{}` in Lua tables. When BullMQ's internal Lua scripts touch job data,
+ * any empty array in the payload silently becomes `{}`.
+ *
+ * Applied once at the worker boundary before data enters the execution engine.
+ */
+
+/**
+ * Returns `value` if it's an array, otherwise `[]`.
+ */
+export function ensureArray<T>(value: unknown): T[] {
+  return Array.isArray(value) ? value : []
+}
+
+const EXECUTION_STATE_ARRAY_FIELDS = [
+  'executedBlocks',
+  'blockLogs',
+  'completedLoops',
+  'activeExecutionPath',
+  'pendingQueue',
+  'remainingEdges',
+  'completedPauseContexts',
+]
+
+/**
+ * Normalizes all known array fields on a BullMQ-deserialized workflow execution payload.
+ * Mutates in place — call once before passing into the execution engine.
+ */
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+export function sanitizeBullMQPayload(payload: any): void {
+  if (!payload) return
+
+  payload.selectedOutputs = ensureArray(payload.selectedOutputs)
+
+  if (payload.metadata) {
+    payload.metadata.callChain = ensureArray(payload.metadata.callChain)
+
+    if (payload.metadata.pendingBlocks !== undefined) {
+      payload.metadata.pendingBlocks = ensureArray(payload.metadata.pendingBlocks)
+    }
+
+    if (payload.metadata.workflowStateOverride?.edges !== undefined) {
+      payload.metadata.workflowStateOverride.edges = ensureArray(
+        payload.metadata.workflowStateOverride.edges
+      )
+    }
+  }
+
+  if (payload.runFromBlock?.sourceSnapshot) {
+    const state = payload.runFromBlock.sourceSnapshot
+    for (const field of EXECUTION_STATE_ARRAY_FIELDS) {
+      if (field in state && !Array.isArray(state[field])) {
+        state[field] = []
+      }
+    }
+
+    if (state.dagIncomingEdges && typeof state.dagIncomingEdges === 'object') {
+      for (const key of Object.keys(state.dagIncomingEdges)) {
+        if (!Array.isArray(state.dagIncomingEdges[key])) {
+          state.dagIncomingEdges[key] = []
+        }
+      }
+    }
+  }
+}

apps/sim/lib/core/workspace-dispatch/memory-store.ts

@@ -473,6 +473,7 @@ export class MemoryWorkspaceDispatchStorage implements WorkspaceDispatchStorageA
       status: 'completed',
       completedAt: Date.now(),
       output,
+      bullmqPayload: undefined,
     }))
   }
 
@@ -482,6 +483,7 @@ export class MemoryWorkspaceDispatchStorage implements WorkspaceDispatchStorageA
       status: 'failed',
       completedAt: Date.now(),
       error,
+      bullmqPayload: undefined,
     }))
   }