admin/n8n-enterprise-unlocked
1
0
Fork 0
mirror of https://github.com/Abdulazizzn/n8n-enterprise-unlocked.git synced 2025-12-16 17:46:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(core): Port concurrency config (#18324)

Browse Source
This commit is contained in:
Iván Ovejero
2025-08-15 10:47:11 +02:00
committed by GitHub
parent 2cbea86fec
commit 9cc1b11f7f
7 changed files with 71 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
packages/@n8n/config/src/configs/executions.config.ts
View File

@@ -11,6 +11,21 @@ class PruningIntervalsConfig {
softDelete: number = 60;
}
@Config
class ConcurrencyConfig {
/**
* Max production executions allowed to run concurrently. `-1` means unlimited.
*
* Default for scaling mode is taken from the worker's `--concurrency` flag.
*/
@Env('N8N_CONCURRENCY_PRODUCTION_LIMIT')
productionLimit: number = -1;
/** Max evaluation executions allowed to run concurrently. `-1` means unlimited. */
@Env('N8N_CONCURRENCY_EVALUATION_LIMIT')
evaluationLimit: number = -1;
}
@Config
export class ExecutionsConfig {
/** Whether to delete past executions on a rolling basis. */
@@ -38,4 +53,7 @@ export class ExecutionsConfig {
@Nested
pruneDataIntervals: PruningIntervalsConfig;
@Nested
concurrency: ConcurrencyConfig;
}

4
packages/@n8n/config/test/config.test.ts
View File

@@ -310,6 +310,10 @@ describe('GlobalConfig', () => {
hardDelete: 15,
softDelete: 60,
},
concurrency: {
productionLimit: -1,
evaluationLimit: -1,
},
},
diagnostics: {
enabled: true,

2
packages/cli/src/commands/worker.ts
View File

@@ -136,7 +136,7 @@ export class Worker extends BaseCommand<z.infer<typeof flagsSchema>> {
async setConcurrency() {
const { flags } = this;
const envConcurrency = config.getEnv('executions.concurrency.productionLimit');
const envConcurrency = this.globalConfig.executions.concurrency.productionLimit;
this.concurrency = envConcurrency !== -1 ? envConcurrency : flags.concurrency;

75
packages/cli/src/concurrency/__tests__/concurrency-control.service.test.ts
View File

@@ -22,11 +22,18 @@ describe('ConcurrencyControlService', () => {
const executionRepository = mock<ExecutionRepository>();
const telemetry = mock<Telemetry>();
const eventService = mock<EventService>();
const globalConfig = mock<GlobalConfig>();
const globalConfig = mock<GlobalConfig>({
executions: {
concurrency: {
productionLimit: -1,
evaluationLimit: -1,
},
},
});
afterEach(() => {
config.set('executions.concurrency.productionLimit', -1);
config.set('executions.concurrency.evaluationLimit', -1);
globalConfig.executions.concurrency.productionLimit = -1;
globalConfig.executions.concurrency.evaluationLimit = -1;
config.set('executions.mode', 'integrated');
jest.clearAllMocks();
@@ -39,7 +46,8 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set(`executions.concurrency.${type}Limit`, 1);
// @ts-expect-error Testing
globalConfig.executions.concurrency[type + 'Limit'] = 1;
/**
* Act
@@ -70,7 +78,8 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set(`executions.concurrency.${type}Limit`, 0);
// @ts-expect-error Testing
globalConfig.executions.concurrency[type + 'Limit'] = 0;
try {
/**
@@ -96,8 +105,8 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.productionLimit', -1);
config.set('executions.concurrency.evaluationLimit', -1);
globalConfig.executions.concurrency.productionLimit = -1;
globalConfig.executions.concurrency.evaluationLimit = -1;
/**
* Act
@@ -123,7 +132,8 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set(`executions.concurrency.${type}Limit`, -2);
// @ts-expect-error Testing
globalConfig.executions.concurrency[type + 'Limit'] = -2;
/**
* Act
@@ -149,7 +159,7 @@ describe('ConcurrencyControlService', () => {
* Arrange
*/
config.set('executions.mode', 'queue');
config.set('executions.concurrency.productionLimit', 2);
globalConfig.executions.concurrency.productionLimit = 2;
/**
* Act
@@ -182,7 +192,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.productionLimit', 1);
globalConfig.executions.concurrency.productionLimit = 1;
const service = new ConcurrencyControlService(
logger,
@@ -209,7 +219,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.productionLimit', 1);
globalConfig.executions.concurrency.productionLimit = 1;
const service = new ConcurrencyControlService(
logger,
@@ -235,7 +245,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.evaluationLimit', 1);
globalConfig.executions.concurrency.evaluationLimit = 1;
const service = new ConcurrencyControlService(
logger,
@@ -265,7 +275,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.productionLimit', 1);
globalConfig.executions.concurrency.evaluationLimit = 1;
const service = new ConcurrencyControlService(
logger,
@@ -292,7 +302,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.productionLimit', 1);
globalConfig.executions.concurrency.productionLimit = 1;
const service = new ConcurrencyControlService(
logger,
@@ -318,7 +328,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.evaluationLimit', 1);
globalConfig.executions.concurrency.evaluationLimit = 1;
const service = new ConcurrencyControlService(
logger,
@@ -348,7 +358,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.productionLimit', 1);
globalConfig.executions.concurrency.productionLimit = 1;
const service = new ConcurrencyControlService(
logger,
@@ -377,7 +387,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.productionLimit', 1);
globalConfig.executions.concurrency.productionLimit = 1;
const service = new ConcurrencyControlService(
logger,
@@ -404,7 +414,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.evaluationLimit', 1);
globalConfig.executions.concurrency.evaluationLimit = 1;
const service = new ConcurrencyControlService(
logger,
@@ -434,7 +444,8 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set(`executions.concurrency.${type}Limit`, 2);
// @ts-expect-error Testing
globalConfig.executions.concurrency[type + 'Limit'] = 2;
const service = new ConcurrencyControlService(
logger,
@@ -470,8 +481,8 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.productionLimit', 2);
config.set('executions.concurrency.evaluationLimit', 2);
globalConfig.executions.concurrency.productionLimit = 2;
globalConfig.executions.concurrency.evaluationLimit = 2;
/**
* Act
@@ -497,8 +508,8 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.productionLimit', 2);
config.set('executions.concurrency.evaluationLimit', 2);
globalConfig.executions.concurrency.productionLimit = 2;
globalConfig.executions.concurrency.evaluationLimit = 2;
/**
* Act
@@ -532,7 +543,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.productionLimit', -1);
globalConfig.executions.concurrency.productionLimit = -1;
const service = new ConcurrencyControlService(
logger,
@@ -559,7 +570,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.evaluationLimit', -1);
globalConfig.executions.concurrency.evaluationLimit = -1;
const service = new ConcurrencyControlService(
logger,
@@ -588,7 +599,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.productionLimit', -1);
globalConfig.executions.concurrency.evaluationLimit = -1;
const service = new ConcurrencyControlService(
logger,
@@ -614,7 +625,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.evaluationLimit', -1);
globalConfig.executions.concurrency.evaluationLimit = -1;
const service = new ConcurrencyControlService(
logger,
@@ -642,7 +653,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.productionLimit', -1);
globalConfig.executions.concurrency.productionLimit = -1;
const service = new ConcurrencyControlService(
logger,
@@ -668,7 +679,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.evaluationLimit', -1);
globalConfig.executions.concurrency.evaluationLimit = -1;
const service = new ConcurrencyControlService(
logger,
@@ -704,7 +715,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.productionLimit', CLOUD_TEMP_PRODUCTION_LIMIT);
globalConfig.executions.concurrency.productionLimit = CLOUD_TEMP_PRODUCTION_LIMIT;
globalConfig.deployment.type = 'cloud';
const service = new ConcurrencyControlService(
logger,
@@ -738,7 +749,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.productionLimit', CLOUD_TEMP_PRODUCTION_LIMIT);
globalConfig.executions.concurrency.productionLimit = CLOUD_TEMP_PRODUCTION_LIMIT;
globalConfig.deployment.type = 'cloud';
const service = new ConcurrencyControlService(
logger,
@@ -771,7 +782,7 @@ describe('ConcurrencyControlService', () => {
/**
* Arrange
*/
config.set('executions.concurrency.productionLimit', CLOUD_TEMP_PRODUCTION_LIMIT);
globalConfig.executions.concurrency.productionLimit = CLOUD_TEMP_PRODUCTION_LIMIT;
globalConfig.deployment.type = 'cloud';
const service = new ConcurrencyControlService(
logger,

6
packages/cli/src/concurrency/concurrency-control.service.ts
View File

@@ -39,9 +39,11 @@ export class ConcurrencyControlService {
) {
this.logger = this.logger.scoped('concurrency');
const { productionLimit, evaluationLimit } = this.globalConfig.executions.concurrency;
this.limits = new Map([
['production', config.getEnv('executions.concurrency.productionLimit')],
['evaluation', config.getEnv('executions.concurrency.evaluationLimit')],
['production', productionLimit],
['evaluation', evaluationLimit],
]);
this.limits.forEach((limit, type) => {

15
packages/cli/src/config/schema.ts
View File

@@ -10,21 +10,6 @@ export const schema = {
env: 'EXECUTIONS_MODE',
},
concurrency: {
productionLimit: {
doc: "Max production executions allowed to run concurrently, in main process for regular mode and in worker for queue mode. Default for main mode is `-1` (disabled). Default for queue mode is taken from the worker's `--concurrency` flag.",
format: Number,
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).
// If the workflow is executed in the main process a soft timeout
// is executed (takes effect after the current node finishes).

2
packages/cli/src/services/frontend.service.ts
View File

@@ -129,7 +129,7 @@ export class FrontendService {
binaryDataMode: this.binaryDataConfig.mode,
nodeJsVersion: process.version.replace(/^v/, ''),
versionCli: N8N_VERSION,
concurrency: config.getEnv('executions.concurrency.productionLimit'),
concurrency: this.globalConfig.executions.concurrency.productionLimit,
authCookie: {
secure: this.globalConfig.auth.cookie.secure,
},