diff --git a/packages/cli/src/__tests__/active-executions.test.ts b/packages/cli/src/__tests__/active-executions.test.ts index d8ee28f3160..c12772f66fd 100644 --- a/packages/cli/src/__tests__/active-executions.test.ts +++ b/packages/cli/src/__tests__/active-executions.test.ts @@ -573,6 +573,52 @@ describe('ActiveExecutions', () => { expect(activeExecutions.getRunningExecutionIds()).toEqual([runningExecutionId]); }); + test('Should not list a running execution attached as enqueued', async () => { + const inProcessExecutionId = await addExecutionWithStatus('running'); + activeExecutions.attachWorkflowExecution(inProcessExecutionId, workflowExecution); + + const enqueuedExecutionId = await addExecutionWithStatus('running'); + activeExecutions.attachWorkflowExecution(enqueuedExecutionId, workflowExecution, { + isQueueJob: true, + }); + + expect(activeExecutions.getRunningExecutionIds()).toEqual([inProcessExecutionId]); + }); + + test('Should cancel only the in-process execution when one is enqueued', async () => { + const inProcessExecutionId = await addExecutionWithStatus('running'); + activeExecutions.attachWorkflowExecution(inProcessExecutionId, workflowExecution); + const inProcessPostExecutePromise = + activeExecutions.getPostExecutePromise(inProcessExecutionId); + + const enqueuedExecutionId = await addExecutionWithStatus('running'); + const enqueuedWorkflowExecution = new PCancelable((resolve) => resolve()); + enqueuedWorkflowExecution.cancel = vi.fn(); + activeExecutions.attachWorkflowExecution(enqueuedExecutionId, enqueuedWorkflowExecution, { + isQueueJob: true, + }); + const enqueuedPostExecutePromise = + activeExecutions.getPostExecutePromise(enqueuedExecutionId); + + await expect(activeExecutions.cancelRunningExecutions()).resolves.toEqual([ + inProcessExecutionId, + ]); + + await expect(inProcessPostExecutePromise).rejects.toThrow( + SystemShutdownExecutionCancelledError, + ); + expect(workflowExecution.cancel).toHaveBeenCalled(); + expect(enqueuedWorkflowExecution.cancel).not.toHaveBeenCalled(); + expect(executionRepository.cancelManyRunning).toHaveBeenCalledWith([inProcessExecutionId]); + expect(activeExecutions.has(enqueuedExecutionId)).toBe(true); + + const outcome = await Promise.race([ + enqueuedPostExecutePromise.then(() => 'settled').catch(() => 'settled'), + new Promise((resolve) => setTimeout(() => resolve('pending'), 0)), + ]); + expect(outcome).toBe('pending'); + }); + test('Should cancel only the executions with a running status', async () => { const runningExecutionId = await addExecutionWithStatus('running'); activeExecutions.attachWorkflowExecution(runningExecutionId, workflowExecution); diff --git a/packages/cli/src/__tests__/workflow-runner.test.ts b/packages/cli/src/__tests__/workflow-runner.test.ts index 9d0005e191b..58c1a93bc19 100644 --- a/packages/cli/src/__tests__/workflow-runner.test.ts +++ b/packages/cli/src/__tests__/workflow-runner.test.ts @@ -1240,6 +1240,25 @@ describe('enqueueExecution', () => { expect(addJob).toHaveBeenCalledWith(expect.objectContaining(processData), expect.any(Object)); }); + + it('excludes the execution from getRunningExecutionIds once enqueued', async () => { + const workflow = await createWorkflow({}, owner); + const activeExecutions = Container.get(ActiveExecutions); + const data: IWorkflowExecutionDataProcess = { + executionMode: 'trigger', + workflowData: workflow, + }; + const executionId = await activeExecutions.add(data); + + const job = mock({ id: '1', data: { executionId, workflowId: workflow.id } }); + addJob.mockResolvedValueOnce(job); + waitForJob.mockReturnValueOnce(new Promise(() => {})); + + // @ts-expect-error Private method + await runner.enqueueExecution(executionId, workflow.id, data); + + expect(activeExecutions.getRunningExecutionIds()).not.toContain(executionId); + }); }); describe('workflow timeout with startedAt', () => { diff --git a/packages/cli/src/active-executions.ts b/packages/cli/src/active-executions.ts index a80a36214d9..6ba924d8ef4 100644 --- a/packages/cli/src/active-executions.ts +++ b/packages/cli/src/active-executions.ts @@ -196,8 +196,14 @@ export class ActiveExecutions { * Attaches an execution */ - attachWorkflowExecution(executionId: string, workflowExecution: PCancelable) { - this.getExecutionOrFail(executionId).workflowExecution = workflowExecution; + attachWorkflowExecution( + executionId: string, + workflowExecution: PCancelable, + options?: { isQueueJob?: boolean }, + ) { + const execution = this.getExecutionOrFail(executionId); + execution.workflowExecution = workflowExecution; + execution.isQueueJob = options?.isQueueJob; } attachResponsePromise( @@ -320,8 +326,11 @@ export class ActiveExecutions { } getRunningExecutionIds(): string[] { + // An enqueued execution runs as a Bull job, which the queue drain tracks, so this only returns executions this process runs outside the queue. return Object.keys(this.activeExecutions).filter( - (executionId) => this.activeExecutions[executionId].status === 'running', + (executionId) => + this.activeExecutions[executionId].status === 'running' && + !this.activeExecutions[executionId].isQueueJob, ); } diff --git a/packages/cli/src/interfaces.ts b/packages/cli/src/interfaces.ts index bc55a544e55..840e16a0dfd 100644 --- a/packages/cli/src/interfaces.ts +++ b/packages/cli/src/interfaces.ts @@ -153,6 +153,7 @@ export interface IExecutingWorkflowData { responsePromise?: IDeferredPromise; workflowExecution?: PCancelable; status: ExecutionStatus; + isQueueJob?: boolean; } export interface IActiveDirectorySettings { diff --git a/packages/cli/src/workflow-runner.ts b/packages/cli/src/workflow-runner.ts index c757d650b71..0701c58e436 100644 --- a/packages/cli/src/workflow-runner.ts +++ b/packages/cli/src/workflow-runner.ts @@ -952,7 +952,9 @@ export class WorkflowRunner { // So we're just preventing crashes here. }); - this.activeExecutions.attachWorkflowExecution(executionId, workflowExecution); + this.activeExecutions.attachWorkflowExecution(executionId, workflowExecution, { + isQueueJob: true, + }); } /**