don't block shutdown on any waiting executions

n8n-io · Feb 6, 2025 · 37bb649 · 37bb649
1 parent bb8c22d
commit 37bb649
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 26 deletions.
diff --git a/packages/cli/src/__tests__/active-executions.test.ts b/packages/cli/src/__tests__/active-executions.test.ts
@@ -245,7 +245,7 @@ describe('ActiveExecutions', () => {
 			activeExecutions.setStatus(waitingExecutionId2, 'waiting');
 		});
 
-		test('Should cancel only new and waiting executions with response-promises by default', async () => {
+		test('Should cancel only executions with response-promises by default', async () => {
 			const stopExecutionSpy = jest.spyOn(activeExecutions, 'stopExecution');
 
 			expect(activeExecutions.getActiveExecutions()).toHaveLength(4);
@@ -263,6 +263,10 @@ describe('ActiveExecutions', () => {
 			expect(stopExecutionSpy).toHaveBeenCalledWith(waitingExecutionId1);
 			expect(stopExecutionSpy).not.toHaveBeenCalledWith(newExecutionId2);
 			expect(stopExecutionSpy).not.toHaveBeenCalledWith(waitingExecutionId2);
+
+			await new Promise(setImmediate);
+			// the other two executions aren't cancelled, but still removed from memory
+			expect(activeExecutions.getActiveExecutions()).toHaveLength(0);
 		});
 
 		test('Should cancel all executions when cancelAll is true', async () => {

diff --git a/packages/cli/src/active-executions.ts b/packages/cli/src/active-executions.ts
@@ -214,24 +214,31 @@ export class ActiveExecutions {
 
 	/** Wait for all active executions to finish */
 	async shutdown(cancelAll = false) {
-		const { withResponsePromise, others } = this.groupExecutionIds();
-		const executionIdsToCancel = [...withResponsePromise];
-
-		if (config.getEnv('executions.mode') === 'regular') {
+		const isRegularMode = config.getEnv('executions.mode') === 'regular';
+		if (isRegularMode) {
 			// removal of active executions will no longer release capacity back,
 			// so that throttled executions cannot resume during shutdown
 			this.concurrencyControl.disable();
+		}
 
-			if (cancelAll) {
-				executionIdsToCancel.push(...others);
+		let executionIds = Object.keys(this.activeExecutions);
+		const toCancel: string[] = [];
+		for (const executionId of executionIds) {
+			const { responsePromise, status } = this.activeExecutions[executionId];
+			if (!!responsePromise || (isRegularMode && cancelAll)) {
+				// Cancel all exectutions that have a response promise, because these promises can't be retained between restarts
+				this.stopExecution(executionId);
+				toCancel.push(executionId);
+			} else if (status === 'waiting' || status === 'new') {
+				// Remove waiting and new executions to not block shutdown
+				delete this.activeExecutions[executionId];
 			}
 		}
 
-		executionIdsToCancel.forEach((executionId) => this.stopExecution(executionId));
-		await this.concurrencyControl.removeAll(executionIdsToCancel);
+		await this.concurrencyControl.removeAll(toCancel);
 
 		let count = 0;
-		let executionIds = Object.keys(this.activeExecutions);
+		executionIds = Object.keys(this.activeExecutions);
 		while (executionIds.length !== 0) {
 			if (count++ % 4 === 0) {
 				this.logger.info(`Waiting for ${executionIds.length} active executions to finish...`);
@@ -242,22 +249,6 @@ export class ActiveExecutions {
 		}
 	}
 
-	private groupExecutionIds() {
-		const groups: Record<'withResponsePromise' | 'others', string[]> = {
-			withResponsePromise: [],
-			others: [],
-		};
-		return Object.entries(this.activeExecutions).reduce((acc, [executionId, execution]) => {
-			const { status, responsePromise } = execution;
-			const group =
-				responsePromise && (status === 'new' || status === 'waiting')
-					? acc.withResponsePromise
-					: acc.others;
-			group.push(executionId);
-			return acc;
-		}, groups);
-	}
-
 	getExecutionOrFail(executionId: string): IExecutingWorkflowData {
 		const execution = this.activeExecutions[executionId];
 		if (!execution) {