Enqueue a new job every 20 step executions (#15068)

To avoid huge workflows to block the worker, we will enqueue a new job
every 20 steps.
This could be more than 20 if there are branches but I think this is
fine, the goal is only to have a limit set.

Also cleaning a bit the code to mark running steps as failed when
workflow fails.

I tested it on a huge workflow:


https://github.com/user-attachments/assets/d7b8e345-d1a1-4467-96fd-92117b500120

packages/twenty-server/src/modules/workflow/workflow-executor/types/workflow-executor-input.ts

@@ -3,6 +3,7 @@ export type WorkflowExecutorInput = {
   workflowRunId: string;
   workspaceId: string;
   shouldComputeWorkflowRunStatus?: boolean;
+  executedStepsCount?: number;
 };
 
 export type WorkflowBranchExecutorInput = {
@@ -10,4 +11,5 @@ export type WorkflowBranchExecutorInput = {
   attemptCount?: number;
   workflowRunId: string;
   workspaceId: string;
+  executedStepsCount?: number;
 };

packages/twenty-server/src/modules/workflow/workflow-executor/utils/__tests__/should-execute-child-step.util.spec.ts

@@ -0,0 +1,492 @@
+import { StepStatus } from 'twenty-shared/workflow';
+
+import { shouldExecuteChildStep } from 'src/modules/workflow/workflow-executor/utils/should-execute-child-step.util';
+import {
+  type WorkflowAction,
+  WorkflowActionType,
+} from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type';
+
+describe('shouldExecuteChildStep', () => {
+  const parentSteps = [
+    {
+      id: 'parent-1',
+      type: WorkflowActionType.CODE,
+      settings: {
+        errorHandlingOptions: {
+          continueOnFailure: { value: false },
+          retryOnFailure: { value: false },
+        },
+        outputSchema: {},
+      },
+      nextStepIds: [],
+    } as unknown as WorkflowAction,
+    {
+      id: 'parent-2',
+      type: WorkflowActionType.SEND_EMAIL,
+      settings: {
+        errorHandlingOptions: {
+          continueOnFailure: { value: false },
+          retryOnFailure: { value: false },
+        },
+        outputSchema: {},
+      },
+      nextStepIds: [],
+    } as unknown as WorkflowAction,
+  ];
+
+  it('should return true when there are no parent steps', () => {
+    const result = shouldExecuteChildStep({
+      parentSteps: [],
+      stepInfos: {},
+    });
+
+    expect(result).toBe(true);
+  });
+
+  it('should return true when at least one parent succeeded and all parents are completed', () => {
+    const stepInfos = {
+      'parent-1': {
+        status: StepStatus.SUCCESS,
+      },
+      'parent-2': {
+        status: StepStatus.SUCCESS,
+      },
+    };
+
+    const result = shouldExecuteChildStep({
+      parentSteps,
+      stepInfos,
+    });
+
+    expect(result).toBe(true);
+  });
+
+  it('should return true when one parent succeeded and others are stopped', () => {
+    const stepInfos = {
+      'parent-1': {
+        status: StepStatus.SUCCESS,
+      },
+      'parent-2': {
+        status: StepStatus.STOPPED,
+      },
+    };
+
+    const result = shouldExecuteChildStep({
+      parentSteps,
+      stepInfos,
+    });
+
+    expect(result).toBe(true);
+  });
+
+  it('should return true when one parent succeeded and others are skipped', () => {
+    const stepInfos = {
+      'parent-1': {
+        status: StepStatus.SUCCESS,
+      },
+      'parent-2': {
+        status: StepStatus.SKIPPED,
+      },
+    };
+
+    const result = shouldExecuteChildStep({
+      parentSteps,
+      stepInfos,
+    });
+
+    expect(result).toBe(true);
+  });
+
+  it('should return true when one parent succeeded and others are mix of stopped and skipped', () => {
+    const multiParentSteps = [
+      {
+        id: 'parent-1',
+        type: WorkflowActionType.CODE,
+        settings: {
+          errorHandlingOptions: {
+            continueOnFailure: { value: false },
+            retryOnFailure: { value: false },
+          },
+          outputSchema: {},
+        },
+        nextStepIds: [],
+      } as unknown as WorkflowAction,
+      {
+        id: 'parent-2',
+        type: WorkflowActionType.SEND_EMAIL,
+        settings: {
+          errorHandlingOptions: {
+            continueOnFailure: { value: false },
+            retryOnFailure: { value: false },
+          },
+          outputSchema: {},
+        },
+        nextStepIds: [],
+      } as unknown as WorkflowAction,
+      {
+        id: 'parent-3',
+        type: WorkflowActionType.CODE,
+        settings: {
+          errorHandlingOptions: {
+            continueOnFailure: { value: false },
+            retryOnFailure: { value: false },
+          },
+          outputSchema: {},
+        },
+        nextStepIds: [],
+      } as unknown as WorkflowAction,
+    ];
+
+    const stepInfos = {
+      'parent-1': {
+        status: StepStatus.SUCCESS,
+      },
+      'parent-2': {
+        status: StepStatus.STOPPED,
+      },
+      'parent-3': {
+        status: StepStatus.SKIPPED,
+      },
+    };
+
+    const result = shouldExecuteChildStep({
+      parentSteps: multiParentSteps,
+      stepInfos,
+    });
+
+    expect(result).toBe(true);
+  });
+
+  it('should return false when all parents are completed but none succeeded', () => {
+    const stepInfos = {
+      'parent-1': {
+        status: StepStatus.STOPPED,
+      },
+      'parent-2': {
+        status: StepStatus.SKIPPED,
+      },
+    };
+
+    const result = shouldExecuteChildStep({
+      parentSteps,
+      stepInfos,
+    });
+
+    expect(result).toBe(false);
+  });
+
+  it('should return false when one parent succeeded but another is still running', () => {
+    const stepInfos = {
+      'parent-1': {
+        status: StepStatus.SUCCESS,
+      },
+      'parent-2': {
+        status: StepStatus.RUNNING,
+      },
+    };
+
+    const result = shouldExecuteChildStep({
+      parentSteps,
+      stepInfos,
+    });
+
+    expect(result).toBe(false);
+  });
+
+  it('should return false when one parent succeeded but another has not started', () => {
+    const stepInfos = {
+      'parent-1': {
+        status: StepStatus.SUCCESS,
+      },
+      'parent-2': {
+        status: StepStatus.NOT_STARTED,
+      },
+    };
+
+    const result = shouldExecuteChildStep({
+      parentSteps,
+      stepInfos,
+    });
+
+    expect(result).toBe(false);
+  });
+
+  it('should return false when one parent succeeded but another is pending', () => {
+    const stepInfos = {
+      'parent-1': {
+        status: StepStatus.SUCCESS,
+      },
+      'parent-2': {
+        status: StepStatus.PENDING,
+      },
+    };
+
+    const result = shouldExecuteChildStep({
+      parentSteps,
+      stepInfos,
+    });
+
+    expect(result).toBe(false);
+  });
+
+  it('should return false when one parent succeeded but another has failed', () => {
+    const stepInfos = {
+      'parent-1': {
+        status: StepStatus.SUCCESS,
+      },
+      'parent-2': {
+        status: StepStatus.FAILED,
+      },
+    };
+
+    const result = shouldExecuteChildStep({
+      parentSteps,
+      stepInfos,
+    });
+
+    expect(result).toBe(false);
+  });
+
+  it('should return false when all parents are still running', () => {
+    const stepInfos = {
+      'parent-1': {
+        status: StepStatus.RUNNING,
+      },
+      'parent-2': {
+        status: StepStatus.RUNNING,
+      },
+    };
+
+    const result = shouldExecuteChildStep({
+      parentSteps,
+      stepInfos,
+    });
+
+    expect(result).toBe(false);
+  });
+
+  it('should return false when all parents have not started', () => {
+    const stepInfos = {
+      'parent-1': {
+        status: StepStatus.NOT_STARTED,
+      },
+      'parent-2': {
+        status: StepStatus.NOT_STARTED,
+      },
+    };
+
+    const result = shouldExecuteChildStep({
+      parentSteps,
+      stepInfos,
+    });
+
+    expect(result).toBe(false);
+  });
+
+  it('should return false when no parent has succeeded even though all are completed', () => {
+    const stepInfos = {
+      'parent-1': {
+        status: StepStatus.SKIPPED,
+      },
+      'parent-2': {
+        status: StepStatus.STOPPED,
+      },
+    };
+
+    const result = shouldExecuteChildStep({
+      parentSteps,
+      stepInfos,
+    });
+
+    expect(result).toBe(false);
+  });
+
+  it('should handle single parent step successfully', () => {
+    const singleParent = [
+      {
+        id: 'parent-1',
+        type: WorkflowActionType.CODE,
+        settings: {
+          errorHandlingOptions: {
+            continueOnFailure: { value: false },
+            retryOnFailure: { value: false },
+          },
+          outputSchema: {},
+        },
+        nextStepIds: [],
+      } as unknown as WorkflowAction,
+    ];
+
+    const stepInfos = {
+      'parent-1': {
+        status: StepStatus.SUCCESS,
+      },
+    };
+
+    const result = shouldExecuteChildStep({
+      parentSteps: singleParent,
+      stepInfos,
+    });
+
+    expect(result).toBe(true);
+  });
+
+  it('should return false when single parent has stopped without success', () => {
+    const singleParent = [
+      {
+        id: 'parent-1',
+        type: WorkflowActionType.CODE,
+        settings: {
+          errorHandlingOptions: {
+            continueOnFailure: { value: false },
+            retryOnFailure: { value: false },
+          },
+          outputSchema: {},
+        },
+        nextStepIds: [],
+      } as unknown as WorkflowAction,
+    ];
+
+    const stepInfos = {
+      'parent-1': {
+        status: StepStatus.STOPPED,
+      },
+    };
+
+    const result = shouldExecuteChildStep({
+      parentSteps: singleParent,
+      stepInfos,
+    });
+
+    expect(result).toBe(false);
+  });
+
+  it('should handle missing step info gracefully', () => {
+    const stepInfos = {
+      'parent-1': {
+        status: StepStatus.SUCCESS,
+      },
+    };
+
+    const result = shouldExecuteChildStep({
+      parentSteps,
+      stepInfos,
+    });
+
+    expect(result).toBe(false);
+  });
+
+  it('should work correctly with multiple successful parents', () => {
+    const multiParentSteps = [
+      {
+        id: 'parent-1',
+        type: WorkflowActionType.CODE,
+        settings: {
+          errorHandlingOptions: {
+            continueOnFailure: { value: false },
+            retryOnFailure: { value: false },
+          },
+          outputSchema: {},
+        },
+        nextStepIds: [],
+      } as unknown as WorkflowAction,
+      {
+        id: 'parent-2',
+        type: WorkflowActionType.SEND_EMAIL,
+        settings: {
+          errorHandlingOptions: {
+            continueOnFailure: { value: false },
+            retryOnFailure: { value: false },
+          },
+          outputSchema: {},
+        },
+        nextStepIds: [],
+      } as unknown as WorkflowAction,
+      {
+        id: 'parent-3',
+        type: WorkflowActionType.CODE,
+        settings: {
+          errorHandlingOptions: {
+            continueOnFailure: { value: false },
+            retryOnFailure: { value: false },
+          },
+          outputSchema: {},
+        },
+        nextStepIds: [],
+      } as unknown as WorkflowAction,
+    ];
+
+    const stepInfos = {
+      'parent-1': { status: StepStatus.SUCCESS },
+      'parent-2': { status: StepStatus.SUCCESS },
+      'parent-3': { status: StepStatus.SUCCESS },
+    };
+
+    const result = shouldExecuteChildStep({
+      parentSteps: multiParentSteps,
+      stepInfos,
+    });
+
+    expect(result).toBe(true);
+  });
+
+  it('should return false with large number of parents when none succeeded', () => {
+    const manyParentSteps = Array.from({ length: 25 }, (_, i) => ({
+      id: `parent-${i + 1}`,
+      type: WorkflowActionType.CODE,
+      settings: {
+        errorHandlingOptions: {
+          continueOnFailure: { value: false },
+          retryOnFailure: { value: false },
+        },
+        outputSchema: {},
+      },
+      nextStepIds: [],
+    })) as unknown as WorkflowAction[];
+
+    // All parents stopped or skipped, none succeeded
+    const stepInfos = Object.fromEntries(
+      manyParentSteps.map((step, i) => [
+        step.id,
+        { status: i % 2 === 0 ? StepStatus.STOPPED : StepStatus.SKIPPED },
+      ]),
+    );
+
+    const result = shouldExecuteChildStep({
+      parentSteps: manyParentSteps,
+      stepInfos,
+    });
+
+    expect(result).toBe(false);
+  });
+
+  it('should return false with large number of parents when not all completed', () => {
+    const manyParentSteps = Array.from({ length: 25 }, (_, i) => ({
+      id: `parent-${i + 1}`,
+      type: WorkflowActionType.CODE,
+      settings: {
+        errorHandlingOptions: {
+          continueOnFailure: { value: false },
+          retryOnFailure: { value: false },
+        },
+        outputSchema: {},
+      },
+      nextStepIds: [],
+    })) as unknown as WorkflowAction[];
+
+    // First parent succeeded, rest are still running
+    const stepInfos = Object.fromEntries(
+      manyParentSteps.map((step, i) => [
+        step.id,
+        { status: i === 0 ? StepStatus.SUCCESS : StepStatus.RUNNING },
+      ]),
+    );
+
+    const result = shouldExecuteChildStep({
+      parentSteps: manyParentSteps,
+      stepInfos,
+    });
+
+    expect(result).toBe(false);
+  });
+});

packages/twenty-server/src/modules/workflow/workflow-executor/utils/should-execute-child-step.util.ts

@@ -0,0 +1,27 @@
+import { StepStatus, type WorkflowRunStepInfos } from 'twenty-shared/workflow';
+
+import { type WorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type';
+
+export const shouldExecuteChildStep = ({
+  parentSteps,
+  stepInfos,
+}: {
+  parentSteps: WorkflowAction[];
+  stepInfos: WorkflowRunStepInfos;
+}) => {
+  if (parentSteps.length === 0) {
+    return true;
+  }
+  const hasSuccessfulParentStep = parentSteps.some(
+    (parentStep) => stepInfos[parentStep.id]?.status === StepStatus.SUCCESS,
+  );
+
+  const areAllParentsCompleted = parentSteps.every(
+    (parentStep) =>
+      stepInfos[parentStep.id]?.status === StepStatus.SUCCESS ||
+      stepInfos[parentStep.id]?.status === StepStatus.STOPPED ||
+      stepInfos[parentStep.id]?.status === StepStatus.SKIPPED,
+  );
+
+  return hasSuccessfulParentStep && areAllParentsCompleted;
+};

packages/twenty-server/src/modules/workflow/workflow-executor/utils/should-execute-step.util.ts

@@ -1,7 +1,8 @@
 import { isDefined } from 'twenty-shared/utils';
-import { StepStatus, type WorkflowRunStepInfos } from 'twenty-shared/workflow';
+import { type WorkflowRunStepInfos } from 'twenty-shared/workflow';
 
 import { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity';
+import { shouldExecuteChildStep } from 'src/modules/workflow/workflow-executor/utils/should-execute-child-step.util';
 import { stepHasBeenStarted } from 'src/modules/workflow/workflow-executor/utils/step-has-been-started.util';
 import { isWorkflowIteratorAction } from 'src/modules/workflow/workflow-executor/workflow-actions/iterator/guards/is-workflow-iterator-action.guard';
 import { shouldExecuteIteratorStep } from 'src/modules/workflow/workflow-executor/workflow-actions/iterator/utils/should-execute-iterator-step.util';
@@ -39,20 +40,8 @@ export const shouldExecuteStep = ({
       isDefined(parentStep) && parentStep.nextStepIds?.includes(step.id),
   );
 
-  if (parentSteps.length === 0) {
-    return true;
-  }
-
-  const hasSuccessfulParentStep = parentSteps.some(
-    (parentStep) => stepInfos[parentStep.id]?.status === StepStatus.SUCCESS,
-  );
-
-  const areAllParentsCompleted = parentSteps.every(
-    (parentStep) =>
-      stepInfos[parentStep.id]?.status === StepStatus.SUCCESS ||
-      stepInfos[parentStep.id]?.status === StepStatus.STOPPED ||
-      stepInfos[parentStep.id]?.status === StepStatus.SKIPPED,
-  );
-
-  return hasSuccessfulParentStep && areAllParentsCompleted;
+  return shouldExecuteChildStep({
+    parentSteps,
+    stepInfos,
+  });
 };

packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/utils/should-execute-iterator-step.util.ts

@@ -1,6 +1,7 @@
 import { isDefined } from 'twenty-shared/utils';
-import { StepStatus, type WorkflowRunStepInfos } from 'twenty-shared/workflow';
+import { type WorkflowRunStepInfos } from 'twenty-shared/workflow';
 
+import { shouldExecuteChildStep } from 'src/modules/workflow/workflow-executor/utils/should-execute-child-step.util';
 import { stepHasBeenStarted } from 'src/modules/workflow/workflow-executor/utils/step-has-been-started.util';
 import { getAllStepIdsInLoop } from 'src/modules/workflow/workflow-executor/workflow-actions/iterator/utils/get-all-step-ids-in-loop.util';
 import {
@@ -40,20 +41,8 @@ export const shouldExecuteIteratorStep = ({
     ? stepsTargetingIterator
     : parentSteps;
 
-  if (stepsToCheck.length === 0) {
-    return true;
-  }
-
-  const hasSuccessfulParentStep = stepsToCheck.some(
-    (parentStep) => stepInfos[parentStep.id]?.status === StepStatus.SUCCESS,
-  );
-
-  const hasFailedNorNotStartedOrRunningParentStep = stepsToCheck.some(
-    (parentStep) =>
-      stepInfos[parentStep.id]?.status === StepStatus.FAILED ||
-      stepInfos[parentStep.id]?.status === StepStatus.NOT_STARTED ||
-      stepInfos[parentStep.id]?.status === StepStatus.RUNNING,
-  );
-
-  return hasSuccessfulParentStep && !hasFailedNorNotStartedOrRunningParentStep;
+  return shouldExecuteChildStep({
+    parentSteps: stepsToCheck,
+    stepInfos,
+  });
 };

packages/twenty-server/src/modules/workflow/workflow-executor/workflow-executor.module.ts

@@ -15,6 +15,7 @@ import { IteratorActionModule } from 'src/modules/workflow/workflow-executor/wor
 import { RecordCRUDActionModule } from 'src/modules/workflow/workflow-executor/workflow-actions/record-crud/record-crud-action.module';
 import { ToolExecutorWorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/tool-executor-workflow-action';
 import { WorkflowExecutorWorkspaceService } from 'src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service';
+import { WorkflowRunQueueModule } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module';
 import { WorkflowRunModule } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.module';
 
 @Module({
@@ -30,6 +31,7 @@ import { WorkflowRunModule } from 'src/modules/workflow/workflow-runner/workflow
     AiAgentActionModule,
     EmptyActionModule,
     FeatureFlagModule,
+    WorkflowRunQueueModule,
     AiModule,
   ],
   providers: [

packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/__tests__/workflow-executor.workspace-service.spec.ts

@@ -6,6 +6,7 @@ import { BILLING_FEATURE_USED } from 'src/engine/core-modules/billing/constants/
 import { BILLING_WORKFLOW_EXECUTION_ERROR_MESSAGE } from 'src/engine/core-modules/billing/constants/billing-workflow-execution-error-message.constant';
 import { BillingMeterEventName } from 'src/engine/core-modules/billing/enums/billing-meter-event-names';
 import { BillingService } from 'src/engine/core-modules/billing/services/billing.service';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
 import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter';
 import { WorkflowActionFactory } from 'src/modules/workflow/workflow-executor/factories/workflow-action.factory';
 import { shouldExecuteStep } from 'src/modules/workflow/workflow-executor/utils/should-execute-step.util';
@@ -14,6 +15,7 @@ import {
   WorkflowActionType,
 } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type';
 import { WorkflowExecutorWorkspaceService } from 'src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service';
+import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service';
 import { WorkflowRunWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service';
 
 jest.mock(
@@ -55,6 +57,14 @@ describe('WorkflowExecutorWorkspaceService', () => {
     canBillMeteredProduct: jest.fn().mockReturnValue(true),
   };
 
+  const mockMessageQueueService = {
+    add: jest.fn(),
+  };
+
+  const mockWorkflowRunQueueWorkspaceService = {
+    increaseWorkflowRunQueuedCount: jest.fn(),
+  };
+
   beforeEach(async () => {
     jest.clearAllMocks();
 
@@ -79,6 +89,14 @@ describe('WorkflowExecutorWorkspaceService', () => {
           provide: BillingService,
           useValue: mockBillingService,
         },
+        {
+          provide: `MESSAGE_QUEUE_${MessageQueue.workflowQueue}`,
+          useValue: mockMessageQueueService,
+        },
+        {
+          provide: WorkflowRunQueueWorkspaceService,
+          useValue: mockWorkflowRunQueueWorkspaceService,
+        },
       ],
     }).compile();
 
@@ -335,6 +353,40 @@ describe('WorkflowExecutorWorkspaceService', () => {
 
       expect(workflowActionFactory.get).not.toHaveBeenCalled();
     });
+
+    it('should queue another job when max executed step count is reached', async () => {
+      const mockStepResult = {
+        result: { stepOutput: 'success' },
+      };
+
+      mockWorkflowExecutor.execute.mockResolvedValueOnce(mockStepResult);
+
+      await service.executeFromSteps({
+        workflowRunId: mockWorkflowRunId,
+        stepIds: ['step-1'],
+        workspaceId: mockWorkspaceId,
+        executedStepsCount: 21, // exceeds MAX_EXECUTED_STEPS_COUNT (20)
+      });
+
+      expect(mockMessageQueueService.add).toHaveBeenCalledWith(
+        'RunWorkflowJob',
+        {
+          workspaceId: mockWorkspaceId,
+          workflowRunId: mockWorkflowRunId,
+          lastExecutedStepId: 'step-1',
+        },
+      );
+
+      expect(
+        mockWorkflowRunQueueWorkspaceService.increaseWorkflowRunQueuedCount,
+      ).toHaveBeenCalledWith(mockWorkspaceId);
+
+      // Should not execute the next step (step-2) in the same job
+      expect(workflowActionFactory.get).toHaveBeenCalledTimes(1);
+      expect(workflowActionFactory.get).toHaveBeenCalledWith(
+        WorkflowActionType.CODE,
+      );
+    });
   });
 
   describe('sendWorkflowNodeRunEvent', () => {

packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service.ts

@@ -15,6 +15,9 @@ import { BillingMeterEventName } from 'src/engine/core-modules/billing/enums/bil
 import { BillingProductKey } from 'src/engine/core-modules/billing/enums/billing-product-key.enum';
 import { BillingService } from 'src/engine/core-modules/billing/services/billing.service';
 import { type BillingUsageEvent } from 'src/engine/core-modules/billing/types/billing-usage-event.type';
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
 import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter';
 import { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity';
 import { WorkflowActionFactory } from 'src/modules/workflow/workflow-executor/factories/workflow-action.factory';
@@ -30,8 +33,13 @@ import { workflowShouldKeepRunning } from 'src/modules/workflow/workflow-executo
 import { isWorkflowIteratorAction } from 'src/modules/workflow/workflow-executor/workflow-actions/iterator/guards/is-workflow-iterator-action.guard';
 import { WorkflowIteratorResult } from 'src/modules/workflow/workflow-executor/workflow-actions/iterator/types/workflow-iterator-result.type';
 import { WorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type';
+import { RUN_WORKFLOW_JOB_NAME } from 'src/modules/workflow/workflow-runner/constants/run-workflow-job-name';
+import { type RunWorkflowJobData } from 'src/modules/workflow/workflow-runner/types/run-workflow-job-data.type';
+import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service';
 import { WorkflowRunWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service';
 
+const MAX_EXECUTED_STEPS_COUNT = 20;
+
 @Injectable()
 export class WorkflowExecutorWorkspaceService {
   constructor(
@@ -39,6 +47,9 @@ export class WorkflowExecutorWorkspaceService {
     private readonly workspaceEventEmitter: WorkspaceEventEmitter,
     private readonly workflowRunWorkspaceService: WorkflowRunWorkspaceService,
     private readonly billingService: BillingService,
+    @InjectMessageQueue(MessageQueue.workflowQueue)
+    private readonly messageQueueService: MessageQueueService,
+    private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService,
   ) {}
 
   async executeFromSteps({
@@ -46,6 +57,7 @@ export class WorkflowExecutorWorkspaceService {
     workflowRunId,
     workspaceId,
     shouldComputeWorkflowRunStatus = true,
+    executedStepsCount = 0,
   }: WorkflowExecutorInput) {
     await Promise.all(
       stepIds.map(async (stepIdToExecute) => {
@@ -53,6 +65,7 @@ export class WorkflowExecutorWorkspaceService {
           stepId: stepIdToExecute,
           workflowRunId,
           workspaceId,
+          executedStepsCount,
         });
       }),
     );
@@ -69,6 +82,7 @@ export class WorkflowExecutorWorkspaceService {
     stepId,
     workflowRunId,
     workspaceId,
+    executedStepsCount,
   }: WorkflowBranchExecutorInput) {
     const workflowRun =
       await this.workflowRunWorkspaceService.getWorkflowRunOrFail({
@@ -137,21 +151,60 @@ export class WorkflowExecutorWorkspaceService {
       workspaceId,
     });
 
-    if (shouldProcessNextSteps) {
-      const nextStepIdsToExecute = await this.getNextStepIdsToExecute({
-        executedStep: stepToExecute,
-        executedStepResult: actionOutput,
+    if (!shouldProcessNextSteps) {
+      return;
+    }
+
+    const shouldRunAnotherJob =
+      executedStepsCount && executedStepsCount > MAX_EXECUTED_STEPS_COUNT;
+
+    if (shouldRunAnotherJob) {
+      await this.continueExecutionFromStepInAnotherJob({
+        lastExecutedStepId: stepId,
+        workflowRunId,
+        workspaceId,
       });
 
-      if (isDefined(nextStepIdsToExecute) && nextStepIdsToExecute.length > 0) {
-        await this.executeFromSteps({
-          stepIds: nextStepIdsToExecute,
-          workflowRunId,
-          workspaceId,
-          shouldComputeWorkflowRunStatus: false,
-        });
+      return;
+    }
+
+    const nextStepIdsToExecute = await this.getNextStepIdsToExecute({
+      executedStep: stepToExecute,
+      executedStepResult: actionOutput,
+    });
+
+    if (isDefined(nextStepIdsToExecute) && nextStepIdsToExecute.length > 0) {
+      await this.executeFromSteps({
+        stepIds: nextStepIdsToExecute,
+        workflowRunId,
+        workspaceId,
+        shouldComputeWorkflowRunStatus: false,
+        executedStepsCount: (executedStepsCount ?? 0) + 1,
+      });
+    }
+  }
+
+  async getNextStepIdsToExecute({
+    executedStep,
+    executedStepResult,
+  }: {
+    executedStep: WorkflowAction;
+    executedStepResult: WorkflowActionOutput;
+  }): Promise<string[] | undefined> {
+    const isIteratorStep = isWorkflowIteratorAction(executedStep);
+
+    if (isIteratorStep) {
+      const iteratorStepResult =
+        executedStepResult.result as WorkflowIteratorResult;
+
+      if (!iteratorStepResult.hasProcessedAllItems) {
+        return isString(executedStep.settings.input.initialLoopStepIds)
+          ? JSON.parse(executedStep.settings.input.initialLoopStepIds)
+          : executedStep.settings.input.initialLoopStepIds;
       }
     }
+
+    return executedStep.nextStepIds;
   }
 
   private async computeWorkflowRunStatus({
@@ -216,29 +269,6 @@ export class WorkflowExecutorWorkspaceService {
     );
   }
 
-  private async getNextStepIdsToExecute({
-    executedStep,
-    executedStepResult,
-  }: {
-    executedStep: WorkflowAction;
-    executedStepResult: WorkflowActionOutput;
-  }) {
-    const isIteratorStep = isWorkflowIteratorAction(executedStep);
-
-    if (isIteratorStep) {
-      const iteratorStepResult =
-        executedStepResult.result as WorkflowIteratorResult;
-
-      if (!iteratorStepResult.hasProcessedAllItems) {
-        return isString(executedStep.settings.input.initialLoopStepIds)
-          ? JSON.parse(executedStep.settings.input.initialLoopStepIds)
-          : executedStep.settings.input.initialLoopStepIds;
-      }
-    }
-
-    return executedStep.nextStepIds;
-  }
-
   private async processStepExecutionResult({
     actionOutput,
     stepId,
@@ -351,4 +381,26 @@ export class WorkflowExecutorWorkspaceService {
       };
     }
   }
+
+  private async continueExecutionFromStepInAnotherJob({
+    lastExecutedStepId,
+    workflowRunId,
+    workspaceId,
+  }: {
+    lastExecutedStepId: string;
+    workflowRunId: string;
+    workspaceId: string;
+  }) {
+    await this.messageQueueService.add<RunWorkflowJobData>(
+      RUN_WORKFLOW_JOB_NAME,
+      {
+        workspaceId,
+        workflowRunId,
+        lastExecutedStepId,
+      },
+    );
+    await this.workflowRunQueueWorkspaceService.increaseWorkflowRunQueuedCount(
+      workspaceId,
+    );
+  }
 }