perf(core): Introduce concurrency control for main mode

jest.config.js

@@ -24,9 +24,11 @@ const config = {
 	// This resolve the path mappings from the tsconfig relative to each jest.config.js
 	moduleNameMapper: Object.entries(paths || {}).reduce((acc, [path, [mapping]]) => {
 		path = `^${path.replace(/\/\*$/, '/(.*)$')}`;
-		mapping = mapping.replace(/^\.\/(?:(.*)\/)?\*$/, '$1');
+		mapping = mapping.replace(/^\.?\.\/(?:(.*)\/)?\*$/, '$1');
 		mapping = mapping ? `/${mapping}` : '';
-		acc[path] = '<rootDir>' + (baseUrl ? `/${baseUrl.replace(/^\.\//, '')}` : '') + mapping + '/$1';
+		acc[path] = mapping.startsWith('/test')
+			? '<rootDir>' + mapping + '/$1'
+			: '<rootDir>' + (baseUrl ? `/${baseUrl.replace(/^\.\//, '')}` : '') + mapping + '/$1';
 		return acc;
 	}, {}),
 	setupFilesAfterEnv: ['jest-expect-message'],

packages/cli/src/ActiveExecutions.ts

@@ -18,6 +18,7 @@ import type {
 import { isWorkflowIdValid } from '@/utils';
 import { ExecutionRepository } from '@db/repositories/execution.repository';
 import { Logger } from '@/Logger';
+import { ConcurrencyControlService } from './concurrency/concurrency-control.service';
 
 @Service()
 export class ActiveExecutions {
@@ -31,19 +32,21 @@ export class ActiveExecutions {
 	constructor(
 		private readonly logger: Logger,
 		private readonly executionRepository: ExecutionRepository,
+		private readonly concurrencyControl: ConcurrencyControlService,
 	) {}
 
 	/**
 	 * Add a new active execution
 	 */
 	async add(executionData: IWorkflowExecutionDataProcess, executionId?: string): Promise<string> {
 		let executionStatus: ExecutionStatus = executionId ? 'running' : 'new';
+		const mode = executionData.executionMode;
 		if (executionId === undefined) {
 			// Is a new execution so save in DB
 
 			const fullExecutionData: ExecutionPayload = {
 				data: executionData.executionData!,
-				mode: executionData.executionMode,
+				mode,
 				finished: false,
 				startedAt: new Date(),
 				workflowData: executionData.workflowData,
@@ -64,10 +67,14 @@ export class ActiveExecutions {
 			if (executionId === undefined) {
 				throw new ApplicationError('There was an issue assigning an execution id to the execution');
 			}
+
+			await this.concurrencyControl.check({ mode, executionId });
 			executionStatus = 'running';
 		} else {
 			// Is an existing execution we want to finish so update in DB
 
+			await this.concurrencyControl.check({ mode, executionId });
+
 			const execution: Pick<IExecutionDb, 'id' | 'data' | 'waitTill' | 'status'> = {
 				id: executionId,
 				data: executionData.executionData!,
@@ -128,6 +135,8 @@ export class ActiveExecutions {
 
 		// Remove from the list of active executions
 		delete this.activeExecutions[executionId];
+
+		this.concurrencyControl.release({ mode: execution.executionData.executionMode });
 	}
 
 	/**
@@ -191,6 +200,9 @@ export class ActiveExecutions {
 		let executionIds = Object.keys(this.activeExecutions);
 
 		if (cancelAll) {
+			this.concurrencyControl.removeAll();
+			await this.executionRepository.cancelAllNewExecutions();
+
 			const stopPromises = executionIds.map(
 				async (executionId) => await this.stopExecution(executionId),
 			);

packages/cli/src/License.ts

@@ -357,6 +357,12 @@ export class License {
 		return this.getFeatureValue(LICENSE_QUOTAS.TEAM_PROJECT_LIMIT) ?? 0;
 	}
 
+	getConcurrencyProductionCap() {
+		return (
+			this.getFeatureValue(LICENSE_QUOTAS.CONCURRENCY_PRODUCTION_CAP) ?? UNLIMITED_LICENSE_QUOTA
+		);
+	}
+
 	getPlanName(): string {
 		return this.getFeatureValue('planName') ?? 'Community';
 	}

packages/cli/src/PublicApi/v1/handlers/executions/executions.handler.ts

@@ -9,6 +9,7 @@ import { getSharedWorkflowIds } from '../workflows/workflows.service';
 import { encodeNextCursor } from '../../shared/services/pagination.service';
 import { InternalHooks } from '@/InternalHooks';
 import { ExecutionRepository } from '@db/repositories/execution.repository';
+import { ConcurrencyControlService } from '@/concurrency/concurrency-control.service';
 
 export = {
 	deleteExecution: [
@@ -37,6 +38,11 @@ export = {
 				executionId: execution.id,
 			});
 
+			Container.get(ConcurrencyControlService).remove({
+				executionId: execution.id,
+				mode: execution.mode,
+			});
+
 			execution.id = id;
 
 			return res.json(replaceCircularReferences(execution));

packages/cli/src/commands/start.ts

@@ -288,6 +288,11 @@ export class Start extends BaseCommand {
 
 		await this.initPruning();
 
+		if (config.getEnv('executions.mode') === 'regular') {
+			// cancel any lingering enqueued executions due to instance crash
+			await Container.get(ExecutionRepository).cancelAllNewExecutions();
+		}
+
 		// Start to get active workflows and run their triggers
 		await this.activeWorkflowManager.init();