feat(core): Separate concurrency limits for production and evaluation executions (no-changelog)

packages/cli/src/concurrency/concurrency-control.service.ts

@@ -1,4 +1,5 @@
 import { Service } from '@n8n/di';
+import { capitalize } from 'lodash';
 import { Logger } from 'n8n-core';
 import type { WorkflowExecuteMode as ExecutionMode } from 'n8n-workflow';
 
@@ -15,13 +16,15 @@ import { ConcurrencyQueue } from './concurrency-queue';
 export const CLOUD_TEMP_PRODUCTION_LIMIT = 999;
 export const CLOUD_TEMP_REPORTABLE_THRESHOLDS = [5, 10, 20, 50, 100, 200];
 
+export type ConcurrencyType = 'production' | 'evaluation';
+
 @Service()
 export class ConcurrencyControlService {
 	private isEnabled: boolean;
 
-	private readonly productionLimit: number;
+	private readonly limits: Map<ConcurrencyType, number>;
 
-	private readonly productionQueue: ConcurrencyQueue;
+	private readonly queues: Map<ConcurrencyType, ConcurrencyQueue>;
 
 	private readonly limitsToReport = CLOUD_TEMP_REPORTABLE_THRESHOLDS.map(
 		(t) => CLOUD_TEMP_PRODUCTION_LIMIT - t,
@@ -35,52 +38,68 @@ export class ConcurrencyControlService {
 	) {
 		this.logger = this.logger.scoped('concurrency');
 
-		this.productionLimit = config.getEnv('executions.concurrency.productionLimit');
+		this.limits = new Map([
+			['production', config.getEnv('executions.concurrency.productionLimit')],
+			['evaluation', config.getEnv('executions.concurrency.evaluationLimit')],
+		]);
 
-		if (this.productionLimit === 0) {
-			throw new InvalidConcurrencyLimitError(this.productionLimit);
-		}
+		this.limits.forEach((limit, type) => {
+			if (limit === 0) {
+				throw new InvalidConcurrencyLimitError(limit);
+			}
 
-		if (this.productionLimit < -1) {
-			this.productionLimit = -1;
-		}
+			if (limit < -1) {
+				this.limits.set(type, -1);
+			}
+		});
 
-		if (this.productionLimit === -1 || config.getEnv('executions.mode') === 'queue') {
+		if (
+			Array.from(this.limits.values()).every((limit) => limit === -1) ||
+			config.getEnv('executions.mode') === 'queue'
+		) {
 			this.isEnabled = false;
 			return;
 		}
 
-		this.productionQueue = new ConcurrencyQueue(this.productionLimit);
+		this.queues = new Map();
+		this.limits.forEach((limit, type) => {
+			if (limit > 0) {
+				this.queues.set(type, new ConcurrencyQueue(limit));
+			}
+		});
 
 		this.logInit();
 
 		this.isEnabled = true;
 
-		this.productionQueue.on('concurrency-check', ({ capacity }) => {
-			if (this.shouldReport(capacity)) {
-				this.telemetry.track('User hit concurrency limit', {
-					threshold: CLOUD_TEMP_PRODUCTION_LIMIT - capacity,
-				});
-			}
-		});
-
-		this.productionQueue.on('execution-throttled', ({ executionId }) => {
-			this.logger.debug('Execution throttled', { executionId });
-			this.eventService.emit('execution-throttled', { executionId });
-		});
-
-		this.productionQueue.on('execution-released', async (executionId) => {
-			this.logger.debug('Execution released', { executionId });
+		this.queues.forEach((queue, type) => {
+			queue.on('concurrency-check', ({ capacity }) => {
+				if (this.shouldReport(capacity)) {
+					this.telemetry.track('User hit concurrency limit', {
+						threshold: CLOUD_TEMP_PRODUCTION_LIMIT - capacity,
+						concurrencyType: type,
+					});
+				}
+			});
+
+			queue.on('execution-throttled', ({ executionId }) => {
+				this.logger.debug('Execution throttled', { executionId, type });
+				this.eventService.emit('execution-throttled', { executionId, type });
+			});
+
+			queue.on('execution-released', async (executionId) => {
+				this.logger.debug('Execution released', { executionId, type });
+			});
 		});
 	}
 
 	/**
-	 * Check whether an execution is in the production queue.
+	 * Check whether an execution is in any of the queues.
 	 */
 	has(executionId: string) {
 		if (!this.isEnabled) return false;
 
-		return this.productionQueue.getAll().has(executionId);
+		return Array.from(this.queues.values()).some((queue) => queue.getAll().has(executionId));
 	}
 
 	/**
@@ -89,16 +108,24 @@ export class ConcurrencyControlService {
 	async throttle({ mode, executionId }: { mode: ExecutionMode; executionId: string }) {
 		if (!this.isEnabled || this.isUnlimited(mode)) return;
 
-		await this.productionQueue.enqueue(executionId);
+		const queue = this.getQueue(mode);
+
+		if (queue) {
+			await queue.enqueue(executionId);
+		}
 	}
 
 	/**
-	 * Release capacity back so the next execution in the production queue can proceed.
+	 * Release capacity back so the next execution in the queue can proceed.
 	 */
 	release({ mode }: { mode: ExecutionMode }) {
 		if (!this.isEnabled || this.isUnlimited(mode)) return;
 
-		this.productionQueue.dequeue();
+		const queue = this.getQueue(mode);
+
+		if (queue) {
+			queue.dequeue();
+		}
 	}
 
 	/**
@@ -107,7 +134,11 @@ export class ConcurrencyControlService {
 	remove({ mode, executionId }: { mode: ExecutionMode; executionId: string }) {
 		if (!this.isEnabled || this.isUnlimited(mode)) return;
 
-		this.productionQueue.remove(executionId);
+		const queue = this.getQueue(mode);
+
+		if (queue) {
+			queue.remove(executionId);
+		}
 	}
 
 	/**
@@ -118,11 +149,13 @@ export class ConcurrencyControlService {
 	async removeAll(activeExecutions: { [executionId: string]: IExecutingWorkflowData }) {
 		if (!this.isEnabled) return;
 
-		const enqueuedProductionIds = this.productionQueue.getAll();
+		this.queues.forEach((queue) => {
+			const enqueuedExecutionIds = queue.getAll();
 
-		for (const id of enqueuedProductionIds) {
-			this.productionQueue.remove(id);
-		}
+			for (const id of enqueuedExecutionIds) {
+				queue.remove(id);
+			}
+		});
 
 		const executionIds = Object.entries(activeExecutions)
 			.filter(([_, execution]) => execution.status === 'new' && execution.responsePromise)
@@ -146,15 +179,30 @@ export class ConcurrencyControlService {
 	private logInit() {
 		this.logger.debug('Enabled');
 
-		this.logger.debug(
-			[
-				'Production execution concurrency is',
-				this.productionLimit === -1 ? 'unlimited' : 'limited to ' + this.productionLimit.toString(),
-			].join(' '),
-		);
+		this.limits.forEach((limit, type) => {
+			this.logger.debug(
+				[
+					`${capitalize(type)} execution concurrency is`,
+					limit === -1 ? 'unlimited' : 'limited to ' + limit.toString(),
+				].join(' '),
+			);
+		});
 	}
 
 	private isUnlimited(mode: ExecutionMode) {
+		const queue = this.getQueue(mode);
+
+		return queue === undefined;
+	}
+
+	private shouldReport(capacity: number) {
+		return config.getEnv('deployment.type') === 'cloud' && this.limitsToReport.includes(capacity);
+	}
+
+	/**
+	 * Get the concurrency queue based on the execution mode.
+	 */
+	private getQueue(mode: ExecutionMode) {
 		if (
 			mode === 'error' ||
 			mode === 'integrated' ||
@@ -163,15 +211,13 @@ export class ConcurrencyControlService {
 			mode === 'manual' ||
 			mode === 'retry'
 		) {
-			return true;
+			return undefined;
 		}
 
-		if (mode === 'webhook' || mode === 'trigger') return this.productionLimit === -1;
+		if (mode === 'webhook' || mode === 'trigger') return this.queues.get('production');
 
-		throw new UnknownExecutionModeError(mode);
-	}
+		if (mode === 'evaluation') return this.queues.get('evaluation');
 
-	private shouldReport(capacity: number) {
-		return config.getEnv('deployment.type') === 'cloud' && this.limitsToReport.includes(capacity);
+		throw new UnknownExecutionModeError(mode);
 	}
 }

packages/cli/src/config/schema.ts

@@ -35,6 +35,12 @@ export const schema = {
 				default: -1,
 				env: 'N8N_CONCURRENCY_PRODUCTION_LIMIT',
 			},
+			evaluationLimit: {
+				doc: 'Max evaluation executions allowed to run concurrently.',
+				format: Number,
+				default: -1,
+				env: 'N8N_CONCURRENCY_EVALUATION_LIMIT',
+			},
 		},
 
 		// A Workflow times out and gets canceled after this time (seconds).

packages/cli/src/events/__tests__/log-streaming-event-relay.test.ts

@@ -934,6 +934,7 @@ describe('LogStreamingEventRelay', () => {
 		it('should log on `execution-throttled` event', () => {
 			const event: RelayEventMap['execution-throttled'] = {
 				executionId: 'exec123456',
+				type: 'production',
 			};
 
 			eventService.emit('execution-throttled', event);
@@ -942,6 +943,7 @@ describe('LogStreamingEventRelay', () => {
 				eventName: 'n8n.execution.throttled',
 				payload: {
 					executionId: 'exec123456',
+					type: 'production',
 				},
 			});
 		});

packages/cli/src/events/maps/relay.event-map.ts

@@ -6,6 +6,7 @@ import type {
 	IWorkflowExecutionDataProcess,
 } from 'n8n-workflow';
 
+import type { ConcurrencyType } from '@/concurrency/concurrency-control.service';
 import type { AuthProviderType } from '@/databases/entities/auth-identity';
 import type { ProjectRole } from '@/databases/entities/project-relation';
 import type { GlobalRole, User } from '@/databases/entities/user';
@@ -338,6 +339,7 @@ export type RelayEventMap = {
 
 	'execution-throttled': {
 		executionId: string;
+		type: ConcurrencyType;
 	};
 
 	'execution-started-during-bootup': {

packages/cli/src/events/relays/log-streaming.event-relay.ts

@@ -385,10 +385,10 @@ export class LogStreamingEventRelay extends EventRelay {
 
 	// #region Execution
 
-	private executionThrottled({ executionId }: RelayEventMap['execution-throttled']) {
+	private executionThrottled({ executionId, type }: RelayEventMap['execution-throttled']) {
 		void this.eventBus.sendExecutionEvent({
 			eventName: 'n8n.execution.throttled',
-			payload: { executionId },
+			payload: { executionId, type },
 		});
 	}
 

packages/editor-ui/src/components/executions/ConcurrentExecutionsHeader.vue

@@ -1,6 +1,8 @@
 <script lang="ts" setup>
 import { computed } from 'vue';
 import { useI18n } from '@/composables/useI18n';
+import { WORKFLOW_EVALUATION_EXPERIMENT } from '@/constants';
+import { usePostHog } from '@/stores/posthog.store';
 
 const props = defineProps<{
 	runningExecutionsCount: number;
@@ -14,14 +16,22 @@ const emit = defineEmits<{
 
 const i18n = useI18n();
 
-const tooltipText = computed(() =>
-	i18n.baseText('executionsList.activeExecutions.tooltip', {
+const posthogStore = usePostHog();
+
+const tooltipText = computed(() => {
+	let text = i18n.baseText('executionsList.activeExecutions.tooltip', {
 		interpolate: {
 			running: props.runningExecutionsCount,
 			cap: props.concurrencyCap,
 		},
-	}),
-);
+	});
+
+	if (posthogStore.isFeatureEnabled(WORKFLOW_EVALUATION_EXPERIMENT)) {
+		text += '\n' + i18n.baseText('executionsList.activeExecutions.evaluationNote');
+	}
+
+	return text;
+});
 
 const headerText = computed(() => {
 	if (props.runningExecutionsCount === 0) {
@@ -38,6 +48,7 @@ const headerText = computed(() => {
 
 <template>
 	<div data-test-id="concurrent-executions-header">
+		<n8n-text>{{ headerText }}</n8n-text>
 		<n8n-tooltip>
 			<template #content>
 				<div :class="$style.tooltip">
@@ -60,9 +71,8 @@ const headerText = computed(() => {
 					>
 				</div>
 			</template>
-			<font-awesome-icon icon="info-circle" class="mr-2xs" />
+			<font-awesome-icon icon="info-circle" class="ml-2xs" />
 		</n8n-tooltip>
-		<n8n-text>{{ headerText }}</n8n-text>
 	</div>
 </template>
 

packages/editor-ui/src/components/executions/global/GlobalExecutionsList.vue

@@ -74,7 +74,10 @@ const isAnnotationEnabled = computed(
 );
 
 const runningExecutionsCount = computed(() => {
-	return props.executions.filter((execution) => execution.status === 'running').length;
+	return props.executions.filter(
+		(execution) =>
+			execution.status === 'running' && ['webhook', 'trigger'].includes(execution.mode),
+	).length;
 });
 
 watch(

packages/editor-ui/src/components/executions/workflow/WorkflowExecutionsSidebar.vue

@@ -55,7 +55,10 @@ const executionListRef = ref<HTMLElement | null>(null);
 const workflowPermissions = computed(() => getResourcePermissions(props.workflow?.scopes).workflow);
 
 const runningExecutionsCount = computed(() => {
-	return props.executions.filter((execution) => execution.status === 'running').length;
+	return props.executions.filter(
+		(execution) =>
+			execution.status === 'running' && ['webhook', 'trigger'].includes(execution.mode),
+	).length;
 });
 
 watch(

packages/editor-ui/src/plugins/i18n/locales/en.json

@@ -712,6 +712,7 @@
 	"executionsList.activeExecutions.none": "No active executions",
 	"executionsList.activeExecutions.header": "{running}/{cap} active executions",
 	"executionsList.activeExecutions.tooltip": "Current active executions: {running} out of {cap}. This instance is limited to {cap} concurrent production executions.",
+	"executionsList.activeExecutions.evaluationNote": "Evaluation runs appear in the list of executions but do not count towards your execution concurrency.",
 	"executionsList.allWorkflows": "All Workflows",
 	"executionsList.anyStatus": "Any Status",
 	"executionsList.autoRefresh": "Auto refresh",