fix(cjson): santize bullmq payloads accurately (#3892)

icecrasher321 · web-flow · commit 7c68061e9809 · 2026-04-01T18:19:40.000-07:00
* improvement(queue): remove indirection for flag

* improvement(lua): return job id instedad of serializing within lua

* memory improvements

* cleanup code

* fix types

* fix trigger dev, bullmq, inline prioritization

* address comments

* fix tests

* remove unused lua code

* fix(cjson): santization bullmq payloads accurately
diff --git a/apps/sim/app/api/schedules/execute/route.ts b/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)
diff --git a/apps/sim/app/api/workflows/[id]/execute/route.async.test.ts b/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(),
 }))
diff --git a/apps/sim/app/api/workflows/[id]/execute/route.ts b/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,
diff --git a/apps/sim/background/workflow-execution.ts b/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,
     }
 
diff --git a/apps/sim/executor/execution/block-executor.ts b/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 : []
           )
         }
 
diff --git a/apps/sim/executor/execution/executor.ts b/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,
diff --git a/apps/sim/executor/execution/snapshot-serializer.ts b/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
   )
 
diff --git a/apps/sim/executor/execution/snapshot.ts b/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
     )
   }
diff --git a/apps/sim/lib/core/async-jobs/config.ts b/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)
  */
diff --git a/apps/sim/lib/core/async-jobs/index.ts b/apps/sim/lib/core/async-jobs/index.ts
@@ -4,8 +4,6 @@ export {
   getInlineJobQueue,
   getJobQueue,
   resetJobQueueCache,
-  shouldExecuteInline,
-  shouldUseBullMQ,
 } from './config'
 export type {
   AsyncBackendType,
diff --git a/apps/sim/lib/core/bullmq/queues.ts b/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 },
         },
       })
   }
diff --git a/apps/sim/lib/core/utils/json-sanitize.ts b/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] = []
+        }
+      }
+    }
+  }
+}
diff --git a/apps/sim/lib/core/workspace-dispatch/memory-store.ts b/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,
     }))
   }
 
diff --git a/apps/sim/lib/core/workspace-dispatch/redis-store.ts b/apps/sim/lib/core/workspace-dispatch/redis-store.ts
diff --git a/apps/sim/lib/webhooks/processor.test.ts b/apps/sim/lib/webhooks/processor.test.ts
diff --git a/apps/sim/lib/webhooks/processor.ts b/apps/sim/lib/webhooks/processor.ts
diff --git a/apps/sim/lib/workflows/executor/execute-workflow.ts b/apps/sim/lib/workflows/executor/execute-workflow.ts
diff --git a/apps/sim/lib/workflows/executor/queued-workflow-execution.ts b/apps/sim/lib/workflows/executor/queued-workflow-execution.ts
diff --git a/bun.lock b/bun.lock

Original file line number	Diff line number	Diff line change
`@@ -148,7 +148,7 @@ export class BlockExecutor {`
`148`	`148`	`block,`
`149`	`149`	`streamingExec,`
`150`	`150`	`resolvedInputs,`
`151`		`- ctx.selectedOutputs ?? []`
	`151`	`+ Array.isArray(ctx.selectedOutputs) ? ctx.selectedOutputs : []`
`152`	`152`	`)`
`153`	`153`	`}`
`154`	`154`
Original file line number	Diff line number	Diff line change
`@@ -119,7 +119,7 @@ export function serializePauseSnapshot(`
`119`	`119`	`context.workflow,`
`120`	`120`	`{},`
`121`	`121`	`context.workflowVariables ?? {},`
`122`		`- context.selectedOutputs ?? [],`
	`122`	`+ Array.isArray(context.selectedOutputs) ? context.selectedOutputs : [],`
`123`	`123`	`state`
`124`	`124`	`)`
`125`	`125`
Original file line number	Diff line number	Diff line change
`@@ -28,7 +28,7 @@ export class ExecutionSnapshot {`
`28`	`28`	`data.workflow,`
`29`	`29`	`data.input,`
`30`	`30`	`data.workflowVariables,`
`31`		`- data.selectedOutputs,`
	`31`	`+ Array.isArray(data.selectedOutputs) ? data.selectedOutputs : [],`
`32`	`32`	`data.state`
`33`	`33`	`)`
`34`	`34`	`}`