github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-17 05:29:44 +02:00
Code Issues Actions 14 Packages Projects Releases Wiki Activity Graph

feat(core): Add project-level worker pools (no-changelog) (#32975)

Browse Source 
Co-authored-by: mfsiega <93014743+mfsiega@users.noreply.github.com>
Co-authored-by: Tomi Turtiainen <10324676+tomi@users.noreply.github.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: Michael Siega <michael.siega@n8n.io>
This commit is contained in:
Mike Repeć 2026-09-04 10:19:28 +00:00 committed by GitHub
parent 8399317ccb
commit 8272a8bc33
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
63 changed files with 1474 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
docs/generated/postgres-schema/README.md generated
View File

@ -99,6 +99,7 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand.
| [public.poller_state](public.poller_state.md) | 7 | | BASE TABLE |
| [public.processed_data](public.processed_data.md) | 5 | | BASE TABLE |
| [public.project](public.project.md) | 9 | | BASE TABLE |
| [public.project_pool_settings](public.project_pool_settings.md) | 4 | | BASE TABLE |
| [public.project_relation](public.project_relation.md) | 5 | | BASE TABLE |
| [public.project_secrets_provider_access](public.project_secrets_provider_access.md) | 5 | | BASE TABLE |
| [public.role](public.role.md) | 7 | | BASE TABLE |
@ -296,6 +297,7 @@ erDiagram
"public.poller_state" }o--|| "public.workflow_entity" : "FOREIGN KEY (#quot;workflowId#quot;) REFERENCES workflow_entity(id) ON DELETE CASCADE"
"public.processed_data" }o--|| "public.workflow_entity" : "FOREIGN KEY (#quot;workflowId#quot;) REFERENCES workflow_entity(id) ON DELETE CASCADE"
"public.project" }o--o| "public.user" : "FOREIGN KEY (#quot;creatorId#quot;) REFERENCES #quot;user#quot;(id) ON DELETE SET NULL"
"public.project_pool_settings" |o--|| "public.project" : "FOREIGN KEY (#quot;projectId#quot;) REFERENCES project(id) ON DELETE CASCADE"
"public.project_relation" }o--|| "public.user" : "FOREIGN KEY (#quot;userId#quot;) REFERENCES #quot;user#quot;(id) ON DELETE CASCADE"
"public.project_relation" }o--|| "public.project" : "FOREIGN KEY (#quot;projectId#quot;) REFERENCES project(id) ON DELETE CASCADE"
"public.project_relation" }o--|| "public.role" : "FOREIGN KEY (role) REFERENCES role(slug)"
@ -1279,6 +1281,12 @@ erDiagram
varchar_36_ type
timestamp_3__with_time_zone updatedAt
}
"public.project_pool_settings" {
timestamp_3__with_time_zone createdAt
varchar_63_ defaultPool
varchar_36_ projectId FK
timestamp_3__with_time_zone updatedAt
}
"public.project_relation" {
timestamp_3__with_time_zone createdAt
varchar_36_ projectId FK

9
docs/generated/postgres-schema/public.project.md generated
View File

@ -9,7 +9,7 @@
| customTelemetryTags | json | '[]'::json | false | | | |
| description | varchar(512) | | true | | | |
| icon | json | | true | | | |
| id | varchar(36) | | false | [public.activity_event](public.activity_event.md) [public.agent_chat_attachments](public.agent_chat_attachments.md) [public.agent_execution_threads](public.agent_execution_threads.md) [public.agents](public.agents.md) [public.data_table](public.data_table.md) [public.folder](public.folder.md) [public.git_connection_project](public.git_connection_project.md) [public.insights_metadata](public.insights_metadata.md) [public.instance_ai_threads](public.instance_ai_threads.md) [public.project_relation](public.project_relation.md) [public.project_secrets_provider_access](public.project_secrets_provider_access.md) [public.role_mapping_rule_project](public.role_mapping_rule_project.md) [public.shared_credentials](public.shared_credentials.md) [public.shared_workflow](public.shared_workflow.md) [public.type_availability_policy_scope](public.type_availability_policy_scope.md) [public.variables](public.variables.md) [public.workflow_review_request](public.workflow_review_request.md) | | |
| id | varchar(36) | | false | [public.activity_event](public.activity_event.md) [public.agent_chat_attachments](public.agent_chat_attachments.md) [public.agent_execution_threads](public.agent_execution_threads.md) [public.agents](public.agents.md) [public.data_table](public.data_table.md) [public.folder](public.folder.md) [public.git_connection_project](public.git_connection_project.md) [public.insights_metadata](public.insights_metadata.md) [public.instance_ai_threads](public.instance_ai_threads.md) [public.project_pool_settings](public.project_pool_settings.md) [public.project_relation](public.project_relation.md) [public.project_secrets_provider_access](public.project_secrets_provider_access.md) [public.role_mapping_rule_project](public.role_mapping_rule_project.md) [public.shared_credentials](public.shared_credentials.md) [public.shared_workflow](public.shared_workflow.md) [public.type_availability_policy_scope](public.type_availability_policy_scope.md) [public.variables](public.variables.md) [public.workflow_review_request](public.workflow_review_request.md) | | |
| name | varchar(255) | | false | | | |
| type | varchar(36) | | false | | | |
| updatedAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
@ -48,6 +48,7 @@ erDiagram
"public.git_connection_project" |o--|| "public.project" : "FOREIGN KEY (#quot;projectId#quot;) REFERENCES project(id) ON DELETE CASCADE"
"public.insights_metadata" }o--o| "public.project" : "FOREIGN KEY (#quot;projectId#quot;) REFERENCES project(id) ON DELETE SET NULL"
"public.instance_ai_threads" }o--|| "public.project" : "FOREIGN KEY (#quot;projectId#quot;) REFERENCES project(id) ON DELETE CASCADE"
"public.project_pool_settings" |o--|| "public.project" : "FOREIGN KEY (#quot;projectId#quot;) REFERENCES project(id) ON DELETE CASCADE"
"public.project_relation" }o--|| "public.project" : "FOREIGN KEY (#quot;projectId#quot;) REFERENCES project(id) ON DELETE CASCADE"
"public.project_secrets_provider_access" }o--|| "public.project" : "FOREIGN KEY (#quot;projectId#quot;) REFERENCES project(id) ON DELETE CASCADE"
"public.role_mapping_rule_project" }o--|| "public.project" : "FOREIGN KEY (#quot;projectId#quot;) REFERENCES project(id) ON DELETE CASCADE"
@ -184,6 +185,12 @@ erDiagram
text title
timestamp_3__with_time_zone updatedAt
}
"public.project_pool_settings" {
timestamp_3__with_time_zone createdAt
varchar_63_ defaultPool
varchar_36_ projectId FK
timestamp_3__with_time_zone updatedAt
}
"public.project_relation" {
timestamp_3__with_time_zone createdAt
varchar_36_ projectId FK

56
docs/generated/postgres-schema/public.project_pool_settings.md generated Normal file
View File

@ -0,0 +1,56 @@
# public.project_pool_settings
## Columns
| Name | Type | Default | Nullable | Children | Parents | Comment |
| ---- | ---- | ------- | -------- | -------- | ------- | ------- |
| createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
| defaultPool | varchar(63) | | true | | | |
| projectId | varchar(36) | | false | | [public.project](public.project.md) | |
| updatedAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
## Constraints
| Name | Type | Definition |
| ---- | ---- | ---------- |
| FK_6f5617b5fb0db43f92e39f5b626 | FOREIGN KEY | FOREIGN KEY ("projectId") REFERENCES project(id) ON DELETE CASCADE |
| PK_6f5617b5fb0db43f92e39f5b626 | PRIMARY KEY | PRIMARY KEY ("projectId") |
| project_pool_settings_createdAt_not_null | n | NOT NULL "createdAt" |
| project_pool_settings_projectId_not_null | n | NOT NULL "projectId" |
| project_pool_settings_updatedAt_not_null | n | NOT NULL "updatedAt" |
## Indexes
| Name | Definition |
| ---- | ---------- |
| PK_6f5617b5fb0db43f92e39f5b626 | CREATE UNIQUE INDEX "PK_6f5617b5fb0db43f92e39f5b626" ON public.project_pool_settings USING btree ("projectId") |
## Relations
```mermaid
erDiagram
"public.project_pool_settings" |o--|| "public.project" : "FOREIGN KEY (#quot;projectId#quot;) REFERENCES project(id) ON DELETE CASCADE"
"public.project_pool_settings" {
timestamp_3__with_time_zone createdAt
varchar_63_ defaultPool
varchar_36_ projectId FK
timestamp_3__with_time_zone updatedAt
}
"public.project" {
timestamp_3__with_time_zone createdAt
uuid creatorId FK
json customTelemetryTags
varchar_512_ description
json icon
varchar_36_ id
varchar_255_ name
varchar_36_ type
timestamp_3__with_time_zone updatedAt
}
```
---
> Generated by [tbls](https://github.com/k1LoW/tbls)

8
docs/generated/sqlite-schema/README.md generated
View File

@ -99,6 +99,7 @@ Auto-generated from the SQLite migrations in @n8n/db. Do not edit by hand.
| [poller_state](poller_state.md) | 7 | | table |
| [processed_data](processed_data.md) | 5 | | table |
| [project](project.md) | 9 | | table |
| [project_pool_settings](project_pool_settings.md) | 4 | | table |
| [project_relation](project_relation.md) | 5 | | table |
| [project_secrets_provider_access](project_secrets_provider_access.md) | 5 | | table |
| [role](role.md) | 7 | | table |
@ -279,6 +280,7 @@ erDiagram
"poller_state" |o--|| "workflow_entity" : "FOREIGN KEY (workflowId) REFERENCES workflow_entity (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"processed_data" |o--|| "workflow_entity" : "FOREIGN KEY (workflowId) REFERENCES workflow_entity (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"project" }o--o| "user" : "FOREIGN KEY (creatorId) REFERENCES user (id) ON UPDATE NO ACTION ON DELETE SET NULL MATCH NONE"
"project_pool_settings" |o--|| "project" : "FOREIGN KEY (projectId) REFERENCES project (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"project_relation" }o--|| "role" : "FOREIGN KEY (role) REFERENCES role (slug) ON UPDATE NO ACTION ON DELETE NO ACTION MATCH NONE"
"project_relation" |o--|| "project" : "FOREIGN KEY (projectId) REFERENCES project (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"project_relation" |o--|| "user" : "FOREIGN KEY (userId) REFERENCES user (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
@ -1266,6 +1268,12 @@ erDiagram
varchar_36_ type
datetime_3_ updatedAt
}
"project_pool_settings" {
datetime_3_ createdAt
varchar_63_ defaultPool
varchar_36_ projectId PK
datetime_3_ updatedAt
}
"project_relation" {
datetime_3_ createdAt
varchar_36_ projectId PK

9
docs/generated/sqlite-schema/project.md generated
View File

@ -20,7 +20,7 @@ CREATE TABLE "project" ("id" varchar(36) PRIMARY KEY NOT NULL, "name" varchar(25
| customTelemetryTags | TEXT | '[]' | false | | | |
| description | varchar(512) | | true | | | |
| icon | TEXT | | true | | | |
| id | varchar(36) | | false | [activity_event](activity_event.md) [agent_chat_attachments](agent_chat_attachments.md) [agent_execution_threads](agent_execution_threads.md) [agents](agents.md) [data_table](data_table.md) [folder](folder.md) [git_connection_project](git_connection_project.md) [insights_metadata](insights_metadata.md) [instance_ai_threads](instance_ai_threads.md) [project_relation](project_relation.md) [project_secrets_provider_access](project_secrets_provider_access.md) [role_mapping_rule_project](role_mapping_rule_project.md) [shared_credentials](shared_credentials.md) [shared_workflow](shared_workflow.md) [type_availability_policy_scope](type_availability_policy_scope.md) [variables](variables.md) [workflow_review_request](workflow_review_request.md) | | |
| id | varchar(36) | | false | [activity_event](activity_event.md) [agent_chat_attachments](agent_chat_attachments.md) [agent_execution_threads](agent_execution_threads.md) [agents](agents.md) [data_table](data_table.md) [folder](folder.md) [git_connection_project](git_connection_project.md) [insights_metadata](insights_metadata.md) [instance_ai_threads](instance_ai_threads.md) [project_pool_settings](project_pool_settings.md) [project_relation](project_relation.md) [project_secrets_provider_access](project_secrets_provider_access.md) [role_mapping_rule_project](role_mapping_rule_project.md) [shared_credentials](shared_credentials.md) [shared_workflow](shared_workflow.md) [type_availability_policy_scope](type_availability_policy_scope.md) [variables](variables.md) [workflow_review_request](workflow_review_request.md) | | |
| name | varchar(255) | | false | | | |
| type | varchar(36) | | false | | | |
| updatedAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
@ -54,6 +54,7 @@ erDiagram
"git_connection_project" |o--|| "project" : "FOREIGN KEY (projectId) REFERENCES project (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"insights_metadata" }o--o| "project" : "FOREIGN KEY (projectId) REFERENCES project (id) ON UPDATE NO ACTION ON DELETE SET NULL MATCH NONE"
"instance_ai_threads" }o--|| "project" : "FOREIGN KEY (projectId) REFERENCES project (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"project_pool_settings" |o--|| "project" : "FOREIGN KEY (projectId) REFERENCES project (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"project_relation" |o--|| "project" : "FOREIGN KEY (projectId) REFERENCES project (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"project_secrets_provider_access" |o--|| "project" : "FOREIGN KEY (projectId) REFERENCES project (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"role_mapping_rule_project" |o--|| "project" : "FOREIGN KEY (projectId) REFERENCES project (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
@ -190,6 +191,12 @@ erDiagram
TEXT title
datetime_3_ updatedAt
}
"project_pool_settings" {
datetime_3_ createdAt
varchar_63_ defaultPool
varchar_36_ projectId PK
datetime_3_ updatedAt
}
"project_relation" {
datetime_3_ createdAt
varchar_36_ projectId PK

65
docs/generated/sqlite-schema/project_pool_settings.md generated Normal file
View File

@ -0,0 +1,65 @@
# project_pool_settings
## Description
<details>
<summary><strong>Table Definition</strong></summary>
```sql
CREATE TABLE "project_pool_settings" ("projectId" varchar(36) PRIMARY KEY NOT NULL, "defaultPool" varchar(63), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), CONSTRAINT "FK_6f5617b5fb0db43f92e39f5b626" FOREIGN KEY ("projectId") REFERENCES "project" ("id") ON DELETE CASCADE)
```
</details>
## Columns
| Name | Type | Default | Nullable | Children | Parents | Comment |
| ---- | ---- | ------- | -------- | -------- | ------- | ------- |
| createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
| defaultPool | varchar(63) | | true | | | |
| projectId | varchar(36) | | false | | [project](project.md) | |
| updatedAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
## Constraints
| Name | Type | Definition |
| ---- | ---- | ---------- |
| - (Foreign key ID: 0) | FOREIGN KEY | FOREIGN KEY (projectId) REFERENCES project (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE |
| projectId | PRIMARY KEY | PRIMARY KEY (projectId) |
| sqlite_autoindex_project_pool_settings_1 | PRIMARY KEY | PRIMARY KEY (projectId) |
## Indexes
| Name | Definition |
| ---- | ---------- |
| sqlite_autoindex_project_pool_settings_1 | PRIMARY KEY (projectId) |
## Relations
```mermaid
erDiagram
"project_pool_settings" |o--|| "project" : "FOREIGN KEY (projectId) REFERENCES project (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"project_pool_settings" {
datetime_3_ createdAt
varchar_63_ defaultPool
varchar_36_ projectId PK
datetime_3_ updatedAt
}
"project" {
datetime_3_ createdAt
varchar creatorId FK
TEXT customTelemetryTags
varchar_512_ description
TEXT icon
varchar_36_ id PK
varchar_255_ name
varchar_36_ type
datetime_3_ updatedAt
}
```
---
> Generated by [tbls](https://github.com/k1LoW/tbls)

25
packages/@n8n/api-types/src/__tests__/instance-registry-types.test.ts
View File

@ -129,6 +129,31 @@ describe('instanceRegistrationSchema', () => {
});
});
describe('poolName', () => {
test('accepts payload without poolName', () => {
expect(() => instanceRegistrationSchema.parse(validV1)).not.toThrow();
});
test('accepts payload with empty poolName', () => {
expect(() => instanceRegistrationSchema.parse({ ...validV1, poolName: '' })).not.toThrow();
});
test('accepts payload with a valid poolName', () => {
const parsed = instanceRegistrationSchema.parse({
...validV1,
instanceType: 'worker',
instanceRole: 'unset',
poolName: 'gpu',
}) as InstanceRegistration & { poolName?: string };
expect(parsed.poolName).toBe('gpu');
});
test('rejects non-string poolName', () => {
expect(() => instanceRegistrationSchema.parse({ ...validV1, poolName: 123 })).toThrow();
});
});
describe('serialization roundtrip', () => {
test('JSON roundtrip yields deep-equal payload', () => {
const serialized = JSON.stringify(validV1);

2
packages/@n8n/api-types/src/dto/index.ts
View File

@ -337,6 +337,8 @@ export {
type SecurityPolicyResponse,
} from './security-settings/security-policy.dto';
export { UpdateProjectPoolSettingsDto } from './worker-pools/update-project-pool-settings.dto';
export { WorkflowHistoryVersionsByIdsDto } from './workflow-history/workflow-history-versions-by-ids.dto';
export { UpdateWorkflowHistoryVersionDto } from './workflow-history/update-workflow-history-version.dto';
export {

10
packages/@n8n/api-types/src/dto/worker-pools/update-project-pool-settings.dto.ts Normal file
View File

@ -0,0 +1,10 @@
import { z } from 'zod';
import { Z } from '../../zod-class';
// Empty string or null clears the project's pool (routes to the system default queue).
const poolNameOrEmptySchema = z.string().regex(/^([a-z0-9][a-z0-9-]{0,62})?$/);
export class UpdateProjectPoolSettingsDto extends Z.class({
defaultPool: poolNameOrEmptySchema.nullable(),
}) {}

3
packages/@n8n/api-types/src/frontend-settings.ts
View File

@ -233,6 +233,9 @@ export interface FrontendSettings {
folders: {
enabled: boolean;
};
workerPools: {
enabled: boolean;
};
collaboration: {
crdt: 'off' | 'local' | 'server';
};

1
packages/@n8n/api-types/src/index.ts
View File

@ -11,6 +11,7 @@ export type * from './community-node-types';
export type * from './quick-connect';
export * from './agents/index';
export * from './instance-registry-types';
export type * from './worker-pools';
export * from './redaction-enforcement';
export * from './redaction-enforcement-floor';
export * from './workflow-reviews-policy';

2
packages/@n8n/api-types/src/instance-registry-types.ts
View File

@ -13,6 +13,8 @@ const InstanceRegistrationSchemaV1 = z
version: z.string(),
registeredAt: z.number(),
lastSeen: z.number(),
/** Worker pool label. Present (possibly empty) on worker registrations; absent on main/webhook. */
poolName: z.string().optional(),
})
.passthrough();

2
packages/@n8n/api-types/src/scaling.ts
View File

@ -44,4 +44,6 @@ export type WorkerStatus = {
internal: boolean;
}>;
version: string;
poolName: string;
queueName: string;
};

6
packages/@n8n/api-types/src/worker-pools.ts Normal file
View File

@ -0,0 +1,6 @@
export type ProjectPoolSettingsResponse = {
/** Pool that all executions in this project route to. `null` = the system default queue. */
defaultPool: string | null;
/** Pool names registered by workers in the cluster (read-only context, populates the dropdown). */
availablePools: string[];
};

4
packages/@n8n/backend-common/src/license-state.ts
View File

@ -170,6 +170,10 @@ export class LicenseState {
return this.isLicensed('feat:workerView');
}
isWorkerPoolsLicensed() {
return this.isLicensed('feat:workerPools');
}
isProjectRoleAdminLicensed() {
return this.isLicensed('feat:projectRole:admin');
}

66
packages/@n8n/config/src/configs/__tests__/scaling-mode.config.test.ts Normal file
View File

@ -0,0 +1,66 @@
import { Container } from '@n8n/di';
import { ScalingModeConfig } from '../scaling-mode.config';
describe('ScalingModeConfig.workerPool', () => {
const originalEnv = process.env;
beforeEach(() => {
Container.reset();
vi.clearAllMocks();
});
afterEach(() => {
process.env = originalEnv;
});
test('defaults to empty string when N8N_WORKER_POOL_NAME is unset', () => {
process.env = {};
const config = Container.get(ScalingModeConfig);
expect(config.workerPool.name).toBe('');
});
test('worker pools are disabled by default', () => {
process.env = {};
const config = Container.get(ScalingModeConfig);
expect(config.workerPool.enabled).toBe(false);
});
test('enables worker pools when N8N_WORKER_POOLS_ENABLED is true', () => {
process.env = { N8N_WORKER_POOLS_ENABLED: 'true' };
const config = Container.get(ScalingModeConfig);
expect(config.workerPool.enabled).toBe(true);
});
test('accepts a valid label', () => {
process.env = { N8N_WORKER_POOL_NAME: 'gpu' };
const config = Container.get(ScalingModeConfig);
expect(config.workerPool.name).toBe('gpu');
});
test('accepts lowercase alphanumeric with hyphens', () => {
process.env = { N8N_WORKER_POOL_NAME: 'a1-b2-c3' };
const config = Container.get(ScalingModeConfig);
expect(config.workerPool.name).toBe('a1-b2-c3');
});
test.each([
['uppercase', 'GPU'],
['underscore', 'g_pu'],
['space', 'g pu'],
['leading hyphen', '-gpu'],
['colon', 'g:pu'],
['dot', 'g.pu'],
['too long', 'a'.repeat(64)],
])('falls back to default and warns on invalid label (%s)', (_label, value) => {
const consoleWarnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {});
process.env = { N8N_WORKER_POOL_NAME: value };
const config = Container.get(ScalingModeConfig);
expect(config.workerPool.name).toBe('');
expect(consoleWarnSpy).toHaveBeenCalledWith(expect.stringContaining('N8N_WORKER_POOL_NAME'));
consoleWarnSpy.mockRestore();
});
});

31
packages/@n8n/config/src/configs/scaling-mode.config.ts
View File

@ -1,7 +1,35 @@
import { Time } from '@n8n/constants';
import { z } from 'zod';
import { Config, Env, Nested } from '../decorators';
const workerPoolNameSchema = z
.string()
.regex(
/^([a-z0-9][a-z0-9-]{0,62})?$/,
'N8N_WORKER_POOL_NAME must be empty or 1-63 chars of lowercase alphanumeric and hyphens, starting with alphanumeric',
);
@Config
export class WorkerPoolConfig {
/**
* Whether worker pools are enabled. When off, all executions route to the default queue and the pool UI is hidden.
*
* **Experimental** - Use at your own risk.
*/
@Env('N8N_WORKER_POOLS_ENABLED')
enabled: boolean = false;
/**
* Label identifying the worker pool this worker belongs to.
* Empty (default) means the worker listens to the unlabeled `jobs` queue.
* When set, the worker listens to `jobs-<name>` instead.
* Read only on worker instances; ignored on main and webhook.
*/
@Env('N8N_WORKER_POOL_NAME', workerPoolNameSchema)
name: string = '';
}
@Config
class HealthConfig {
/**
@ -151,4 +179,7 @@ export class ScalingModeConfig {
@Nested
bull: BullConfig;
@Nested
workerPool: WorkerPoolConfig;
}

1
packages/@n8n/config/src/index.ts
View File

@ -81,6 +81,7 @@ export {
} from './configs/ssrf-protection.config';
export { EngineConfig } from './configs/engine.config';
export { ExecutionsConfig } from './configs/executions.config';
export { ScalingModeConfig, WorkerPoolConfig } from './configs/scaling-mode.config';
export { LOG_SCOPES } from './configs/logging.config';
export type { LogScope } from './configs/logging.config';
export { WorkflowsConfig } from './configs/workflows.config';

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

@ -412,6 +412,10 @@ describe('GlobalConfig', () => {
stalledInterval: 30_000,
},
},
workerPool: {
enabled: false,
name: '',
},
},
taskRunners: {
mode: 'internal',

1
packages/@n8n/constants/src/index.ts
View File

@ -56,6 +56,7 @@ export const LICENSE_FEATURES = {
DATA_REDACTION: 'feat:dataRedaction',
OTEL_CUSTOM_SPAN_ATTRIBUTES: 'feat:otel:customSpanAttributes',
WORKFLOW_REVIEWS: 'feat:workflowReviews',
WORKER_POOLS: 'feat:workerPools',
} as const;
export const LICENSE_QUOTAS = {

3
packages/@n8n/db/src/entities/index.ts
View File

@ -36,6 +36,7 @@ import { InvalidAuthToken } from './invalid-auth-token';
import { PollerState } from './poller-state';
import { ProcessedData } from './processed-data';
import { Project } from './project';
import { ProjectPoolSettings } from './project-pool-settings.ee';
import { ProjectRelation } from './project-relation';
import { ProjectSecretsProviderAccess } from './project-secrets-provider-access';
import type { SecretsProviderAccessRole } from './project-secrets-provider-access';
@ -190,6 +191,7 @@ export {
TestRun,
TestCaseExecution,
ExecutionEntity,
ProjectPoolSettings,
ProjectSecretsProviderAccess,
type SecretsProviderAccessRole,
SecretsProviderConnection,
@ -254,6 +256,7 @@ export const entities = {
Role,
ScheduledJob,
ScheduledTask,
ProjectPoolSettings,
ProjectSecretsProviderAccess,
SecretsProviderConnection,
};

18
packages/@n8n/db/src/entities/project-pool-settings.ee.ts Normal file
View File

@ -0,0 +1,18 @@
import { Column, Entity, JoinColumn, OneToOne, PrimaryColumn } from '@n8n/typeorm';
import { WithTimestamps } from './abstract-entity';
import { Project } from './project';
@Entity()
export class ProjectPoolSettings extends WithTimestamps {
@PrimaryColumn({ type: 'varchar', length: 36 })
projectId: string;
@OneToOne('Project', { onDelete: 'CASCADE' })
@JoinColumn({ name: 'projectId' })
project: Project;
/** Pool all executions in this project route to. `null` means the system default queue. */
@Column({ type: 'varchar', length: 63, nullable: true })
defaultPool: string | null;
}

22
packages/@n8n/db/src/migrations/common/1788445119183-CreateProjectPoolSettings.ts Normal file
View File

@ -0,0 +1,22 @@
import type { MigrationContext, ReversibleMigration } from '../migration-types';
const tableName = 'project_pool_settings';
export class CreateProjectPoolSettings1788445119183 implements ReversibleMigration {
async up({ schemaBuilder: { createTable, column } }: MigrationContext) {
await createTable(tableName)
.withColumns(
column('projectId').varchar(36).primary.notNull,
column('defaultPool').varchar(63),
)
.withForeignKey('projectId', {
tableName: 'project',
columnName: 'id',
onDelete: 'CASCADE',
}).withTimestamps;
}
async down({ schemaBuilder: { dropTable } }: MigrationContext) {
await dropTable(tableName);
}
}

1
packages/@n8n/db/src/repositories/index.ts
View File

@ -44,6 +44,7 @@ export { ScopeRepository } from './scope.repository';
export { InvalidAuthTokenRepository } from './invalid-auth-token.repository';
export { InstanceCredentialAssignmentRepository } from './instance-credential-assignment.repository';
export { LicenseMetricsRepository } from './license-metrics.repository';
export { ProjectPoolSettingsRepository } from './project-pool-settings.repository.ee';
export { ProjectRelationRepository } from './project-relation.repository';
export { ProjectRepository, type ProjectListOptions } from './project.repository';
export { RoleRepository } from './role.repository';

20
packages/@n8n/db/src/repositories/project-pool-settings.repository.ee.ts Normal file
View File

@ -0,0 +1,20 @@
import { Service } from '@n8n/di';
import { DataSource, Repository } from '@n8n/typeorm';
import { ProjectPoolSettings } from '../entities';
@Service()
export class ProjectPoolSettingsRepository extends Repository<ProjectPoolSettings> {
constructor(dataSource: DataSource) {
super(ProjectPoolSettings, dataSource.manager);
}
async getDefaultPool(projectId: string): Promise<string | null> {
const row = await this.findOne({ where: { projectId }, select: ['defaultPool'] });
return row?.defaultPool ?? null;
}
async setDefaultPool(projectId: string, defaultPool: string | null): Promise<void> {
await this.upsert({ projectId, defaultPool }, ['projectId']);
}
}

1
packages/@n8n/permissions/src/__tests__/__snapshots__/scope-information.test.ts.snap
View File

@ -57,6 +57,7 @@ exports[`Scope Information > ensure scopes are defined correctly 1`] = `
"nodeTypePolicy:manage",
"orchestration:read",
"orchestration:list",
"orchestration:manage",
"project:create",
"project:read",
"project:update",

2
packages/@n8n/permissions/src/constants.ee.ts
View File

@ -25,7 +25,7 @@ export const RESOURCES = {
license: ['manage'] as const,
logStreaming: ['manage'] as const,
nodeTypePolicy: ['manage'] as const,
orchestration: ['read', 'list'] as const,
orchestration: ['read', 'list', 'manage'] as const,
// `manageMembers` gates changes to a project's membership list: adding a member
// with a role, changing a member's role, and removing a member. Kept separate
// from `update` so a role can edit project details without being able to

1
packages/@n8n/permissions/src/roles/scopes/global-scopes.ee.ts
View File

@ -54,6 +54,7 @@ export const GLOBAL_OWNER_SCOPES: Scope[] = [
'logStreaming:manage',
'nodeTypePolicy:manage',
'orchestration:read',
'orchestration:manage',
'saml:manage',
'securityAudit:generate',
'securitySettings:manage',

27
packages/cli/src/__tests__/workflow-runner.test.ts
View File

@ -516,6 +516,33 @@ describe('enqueueExecution', () => {
expect(setupQueue).toHaveBeenCalledTimes(1);
});
it('should finalize the execution when pool resolution fails', async () => {
const activeExecutions = Container.get(ActiveExecutions);
vi.spyOn(activeExecutions, 'attachWorkflowExecution').mockReturnValue();
const processError = vi.spyOn(runner, 'processError').mockResolvedValueOnce();
const data = mock<IWorkflowExecutionDataProcess>({
workflowData: { nodes: [], staticData: {} },
executionData: undefined,
});
const error = new Error('pool resolution failed');
const { PoolConfigService } = await import('@/scaling/pool-config.service.ee.js');
vi.spyOn(Container.get(PoolConfigService), 'resolvePoolForExecution').mockRejectedValueOnce(
error,
);
// @ts-expect-error Private method
await expect(runner.enqueueExecution('1', 'workflow-xyz', data)).rejects.toThrowError(error);
expect(processError).toHaveBeenCalledWith(
error,
expect.any(Date),
data.executionMode,
'1',
expect.anything(),
);
expect(addJob).not.toHaveBeenCalled();
});
it('should include restartExecutionId in job data when provided', async () => {
const activeExecutions = Container.get(ActiveExecutions);
vi.spyOn(activeExecutions, 'attachWorkflowExecution').mockReturnValue();

1
packages/cli/src/commands/__tests__/worker.test.ts
View File

@ -153,6 +153,7 @@ describe('Worker', () => {
taskRunners: {},
outboundProxy: { mode: 'all' },
expressionEngine: { engine: 'legacy', poolSize: 1, maxCodeCacheSize: 1024 },
queue: { workerPool: { enabled: false, name: '' } },
};
// Stub the init steps that go beyond `super.init()`, as in start.test.ts
worker.setConcurrency = vi.fn().mockResolvedValue(undefined);

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

@ -1,4 +1,4 @@
import { inTest } from '@n8n/backend-common';
import { inTest, LicenseState } from '@n8n/backend-common';
import { DeploymentKeyRepository } from '@n8n/db';
import { Command } from '@n8n/decorators';
import { Container } from '@n8n/di';
@ -13,12 +13,13 @@ import { EventMessageGeneric } from '@/eventbus/event-message-classes/event-mess
import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus';
import { LogStreamingEventRelay } from '@/events/relays/log-streaming.event-relay';
import { LoadNodesAndCredentials } from '@/load-nodes-and-credentials';
import { ExecutionStopService } from '@/scaling/execution-stop.service';
import { Publisher } from '@/scaling/pubsub/publisher.service';
import { PubSubRegistry } from '@/scaling/pubsub/pubsub.registry';
import { Subscriber } from '@/scaling/pubsub/subscriber.service';
import { resolveQueueName, resolveWorkerPoolName } from '@/scaling/queue-name';
import type { ScalingService } from '@/scaling/scaling.service';
import type { WorkerServer, WorkerServerEndpointsConfig } from '@/scaling/worker-server';
import { ExecutionStopService } from '@/scaling/execution-stop.service';
import { WorkerStatusService } from '@/scaling/worker-status.service.ee';
import { JwtService } from '@/services/jwt.service';
@ -108,6 +109,17 @@ export class Worker extends BaseCommand<z.infer<typeof flagsSchema>> {
await this.initLicense();
this.logger.debug('License init complete');
if (
resolveWorkerPoolName(this.globalConfig.queue.workerPool) !== '' &&
!Container.get(LicenseState).isWorkerPoolsLicensed()
) {
this.logger.error(
'A worker pool is configured (`N8N_WORKER_POOL_NAME`), but worker pools are not licensed. Either remove the worker pool configuration, or upgrade to a license that supports this feature.',
);
process.exit(1);
}
await this.initCommunityPackages();
await Container.get(CredentialsOverwrites).init();
this.logger.debug('Credentials overwrites init complete');
@ -208,9 +220,14 @@ export class Worker extends BaseCommand<z.infer<typeof flagsSchema>> {
workerServer?.markAsReady();
const poolName = resolveWorkerPoolName(this.globalConfig.queue.workerPool);
const queueName = resolveQueueName('worker', poolName);
this.logger.info('\nn8n worker is now ready');
this.logger.info(` * Version: ${N8N_VERSION}`);
this.logger.info(` * Concurrency: ${this.concurrency}`);
this.logger.info(` * Pool: ${poolName === '' ? '(default)' : poolName}`);
this.logger.info(` * Queue: ${queueName}`);
this.logger.info('');
if (!inTest && process.stdout.isTTY) {

36
packages/cli/src/controllers/__tests__/orchestration.controller.test.ts Normal file
View File

@ -0,0 +1,36 @@
import type { AuthenticatedRequest } from '@n8n/db';
import { mock } from 'vitest-mock-extended';
import { OrchestrationController } from '@/controllers/orchestration.controller';
import type { License } from '@/license';
import type { WorkerStatusService } from '@/scaling/worker-status.service.ee';
describe('OrchestrationController', () => {
const license = mock<License>();
const workerStatusService = mock<WorkerStatusService>();
const controller = new OrchestrationController(license, workerStatusService);
const req = mock<AuthenticatedRequest>({ user: { id: 'user-1' } });
beforeEach(() => {
vi.clearAllMocks();
});
describe('getWorkersStatusAll', () => {
it('requests worker status when licensed', async () => {
license.isWorkerViewLicensed.mockReturnValue(true);
await controller.getWorkersStatusAll(req);
expect(workerStatusService.requestWorkerStatus).toHaveBeenCalledWith('user-1');
});
it('does nothing when not licensed', async () => {
license.isWorkerViewLicensed.mockReturnValue(false);
await controller.getWorkersStatusAll(req);
expect(workerStatusService.requestWorkerStatus).not.toHaveBeenCalled();
});
});
});

48
packages/cli/src/controllers/__tests__/project-pool-settings.controller.ee.test.ts Normal file
View File

@ -0,0 +1,48 @@
import { UpdateProjectPoolSettingsDto } from '@n8n/api-types';
import type { Request, Response } from 'express';
import { mock } from 'vitest-mock-extended';
import { ProjectPoolSettingsController } from '@/controllers/project-pool-settings.controller.ee';
import type { PoolConfigService } from '@/scaling/pool-config.service.ee';
describe('ProjectPoolSettingsController', () => {
const poolConfigService = mock<PoolConfigService>();
const controller = new ProjectPoolSettingsController(poolConfigService);
const req = mock<Request>();
const res = mock<Response>();
const projectId = 'project-1';
beforeEach(() => {
vi.clearAllMocks();
});
describe('getPoolSettings', () => {
it('delegates to PoolConfigService.getProjectPoolSettings', async () => {
poolConfigService.getProjectPoolSettings.mockResolvedValue({
defaultPool: 'gpu',
availablePools: ['gpu', 'cpu'],
});
const result = await controller.getPoolSettings(req, res, projectId);
expect(poolConfigService.getProjectPoolSettings).toHaveBeenCalledWith(projectId);
expect(result).toEqual({ defaultPool: 'gpu', availablePools: ['gpu', 'cpu'] });
});
});
describe('updatePoolSettings', () => {
it('delegates to PoolConfigService.setProjectPoolSettings with the DTO', async () => {
const dto = new UpdateProjectPoolSettingsDto({ defaultPool: 'gpu' });
poolConfigService.setProjectPoolSettings.mockResolvedValue({
defaultPool: 'gpu',
availablePools: ['gpu'],
});
const result = await controller.updatePoolSettings(req, res, dto, projectId);
expect(poolConfigService.setProjectPoolSettings).toHaveBeenCalledWith(projectId, dto);
expect(result).toEqual({ defaultPool: 'gpu', availablePools: ['gpu'] });
});
});
});

1
packages/cli/src/controllers/e2e.controller.ts
View File

@ -144,6 +144,7 @@ export class E2EController {
[LICENSE_FEATURES.DATA_REDACTION]: false,
[LICENSE_FEATURES.WORKFLOW_REVIEWS]: false,
[LICENSE_FEATURES.OTEL_CUSTOM_SPAN_ATTRIBUTES]: false,
[LICENSE_FEATURES.WORKER_POOLS]: false,
};
private static readonly numericFeaturesDefaults: Record<NumericLicenseFeature, number> = {

35
packages/cli/src/controllers/project-pool-settings.controller.ee.ts Normal file
View File

@ -0,0 +1,35 @@
import type { ProjectPoolSettingsResponse } from '@n8n/api-types';
import { UpdateProjectPoolSettingsDto } from '@n8n/api-types';
import { LICENSE_FEATURES } from '@n8n/constants';
import { Body, Get, Licensed, Param, Patch, ProjectScope, RestController } from '@n8n/decorators';
import type { Request, Response } from 'express';
import { PoolConfigService } from '@/scaling/pool-config.service.ee';
@RestController('/projects/:projectId/pool-settings')
export class ProjectPoolSettingsController {
constructor(private readonly poolConfigService: PoolConfigService) {}
@Licensed(LICENSE_FEATURES.WORKER_POOLS)
@ProjectScope('project:read')
@Get('/')
async getPoolSettings(
_req: Request,
_res: Response,
@Param('projectId') projectId: string,
): Promise<ProjectPoolSettingsResponse> {
return await this.poolConfigService.getProjectPoolSettings(projectId);
}
@Licensed(LICENSE_FEATURES.WORKER_POOLS)
@ProjectScope('project:update')
@Patch('/')
async updatePoolSettings(
_req: Request,
_res: Response,
@Body dto: UpdateProjectPoolSettingsDto,
@Param('projectId') projectId: string,
): Promise<ProjectPoolSettingsResponse> {
return await this.poolConfigService.setProjectPoolSettings(projectId, dto);
}
}

70
packages/cli/src/modules/instance-registry/__tests__/instance-registry.service.test.ts
View File

@ -1,6 +1,7 @@
import type { InstanceRegistration } from '@n8n/api-types';
import type { Logger } from '@n8n/backend-common';
import type { ExecutionsConfig } from '@n8n/config';
import type { ExecutionsConfig, ScalingModeConfig } from '@n8n/config';
import { WorkerPoolConfig } from '@n8n/config';
import { Container } from '@n8n/di';
import type { InstanceSettings } from 'n8n-core';
import { mock } from 'vitest-mock-extended';
@ -71,6 +72,11 @@ const makeInstanceSettings = (overrides: Partial<InstanceSettings> = {}) =>
const makeExecutionsConfig = (mode: 'regular' | 'queue' = 'regular') =>
mock<ExecutionsConfig>({ mode });
const makeScalingModeConfig = (poolName: string = '', enabled: boolean = true) => {
const workerPool = Object.assign(new WorkerPoolConfig(), { name: poolName, enabled });
return mock<ScalingModeConfig>({ workerPool });
};
const makeLogger = () => {
const logger = mock<Logger>();
logger.scoped.mockReturnValue(logger);
@ -99,10 +105,13 @@ describe('InstanceRegistryService', () => {
const createService = (
settingsOverrides: Partial<InstanceSettings> = {},
executionMode: 'regular' | 'queue' = 'regular',
poolName: string = '',
poolsEnabled: boolean = true,
) =>
new InstanceRegistryService(
makeInstanceSettings(settingsOverrides),
makeExecutionsConfig(executionMode),
makeScalingModeConfig(poolName, poolsEnabled),
logger,
);
@ -281,7 +290,12 @@ describe('InstanceRegistryService', () => {
it('should reflect instanceRole changes in heartbeat', async () => {
const settings = makeInstanceSettings({ instanceRole: 'unset' });
service = new InstanceRegistryService(settings, makeExecutionsConfig(), logger);
service = new InstanceRegistryService(
settings,
makeExecutionsConfig(),
makeScalingModeConfig(),
logger,
);
await service.init();
// Simulate role change (e.g., leader election completed)
@ -416,4 +430,56 @@ describe('InstanceRegistryService', () => {
expect(removed).toBe(0);
});
});
describe('poolName', () => {
it('records poolName on worker registration when pool is set', async () => {
service = createService({ instanceType: 'worker', instanceRole: 'unset' }, 'queue', 'gpu');
await service.init();
const [reg] = await service.getAllInstances();
expect(reg.poolName).toBe('gpu');
});
it('records empty poolName on worker registration when pool is unset', async () => {
service = createService({ instanceType: 'worker', instanceRole: 'unset' }, 'queue');
await service.init();
const [reg] = await service.getAllInstances();
expect(reg.poolName).toBe('');
});
it('records empty poolName on worker registration when pool is set but pools are disabled', async () => {
service = createService(
{ instanceType: 'worker', instanceRole: 'unset' },
'queue',
'gpu',
false,
);
await service.init();
const [reg] = await service.getAllInstances();
expect(reg.poolName).toBe('');
});
it('omits poolName on main registration even when config has a pool', async () => {
service = createService({ instanceType: 'main', instanceRole: 'leader' }, 'regular', 'gpu');
await service.init();
const [reg] = await service.getAllInstances();
expect(reg.poolName).toBeUndefined();
});
it('omits poolName on webhook registration even when config has a pool', async () => {
service = createService({ instanceType: 'webhook', instanceRole: 'unset' }, 'regular', 'gpu');
await service.init();
const [reg] = await service.getAllInstances();
expect(reg.poolName).toBeUndefined();
});
});
});

12
packages/cli/src/modules/instance-registry/instance-registry.service.ts
View File

@ -1,11 +1,12 @@
import type { InstanceRegistration } from '@n8n/api-types';
import { Logger } from '@n8n/backend-common';
import { ExecutionsConfig } from '@n8n/config';
import { ExecutionsConfig, ScalingModeConfig } from '@n8n/config';
import { Service } from '@n8n/di';
import { InstanceSettings } from 'n8n-core';
import { randomUUID } from 'node:crypto';
import { N8N_VERSION } from '@/constants';
import { resolveWorkerPoolName } from '@/scaling/queue-name';
import { REGISTRY_CONSTANTS } from './instance-registry.types';
import type { InstanceStorage } from './storage/instance-storage.interface';
@ -29,6 +30,7 @@ export class InstanceRegistryService {
constructor(
private readonly instanceSettings: InstanceSettings,
private readonly executionsConfig: ExecutionsConfig,
private readonly scalingModeConfig: ScalingModeConfig,
private readonly logger: Logger,
) {
this.logger = this.logger.scoped('instance-registry');
@ -94,7 +96,7 @@ export class InstanceRegistryService {
}
private buildRegistration(): InstanceRegistration {
return {
const base: InstanceRegistration = {
schemaVersion: 1 as const,
instanceKey: this.instanceKey,
hostId: this.instanceSettings.hostId,
@ -104,6 +106,12 @@ export class InstanceRegistryService {
registeredAt: this.registeredAt,
lastSeen: Date.now(),
};
if (this.instanceSettings.instanceType === 'worker') {
return { ...base, poolName: resolveWorkerPoolName(this.scalingModeConfig.workerPool) };
}
return base;
}
private async selectStorage(): Promise<InstanceStorage> {

148
packages/cli/src/scaling/__tests__/pool-config.service.ee.test.ts Normal file
View File

@ -0,0 +1,148 @@
import type { LicenseState } from '@n8n/backend-common';
import type { GlobalConfig } from '@n8n/config';
import type { ProjectPoolSettingsRepository } from '@n8n/db';
import { mock } from 'vitest-mock-extended';
import { PoolConfigService } from '@/scaling/pool-config.service.ee';
import type { WorkerPoolsService } from '@/scaling/worker-pools.service.ee';
import type { CacheService } from '@/services/cache/cache.service';
describe('PoolConfigService', () => {
const projectPoolSettingsRepository = mock<ProjectPoolSettingsRepository>();
const cacheService = mock<CacheService>();
const workerPoolsService = mock<WorkerPoolsService>();
const globalConfig = mock<GlobalConfig>();
const licenseState = mock<LicenseState>();
const setEnabled = (enabled: boolean) => {
globalConfig.queue = { workerPool: { enabled } } as GlobalConfig['queue'];
};
const service = new PoolConfigService(
projectPoolSettingsRepository,
cacheService,
workerPoolsService,
globalConfig,
licenseState,
);
beforeEach(() => {
vi.clearAllMocks();
setEnabled(true);
licenseState.isWorkerPoolsLicensed.mockReturnValue(true);
// Default: cache miss, no workers registered
cacheService.get.mockResolvedValue(undefined);
workerPoolsService.getAvailablePools.mockResolvedValue([]);
});
describe('resolvePoolForExecution', () => {
const projectId = 'project-123';
it("resolves to the project's default pool without checking worker registration", async () => {
projectPoolSettingsRepository.getDefaultPool.mockResolvedValue('gpu');
const result = await service.resolvePoolForExecution({ projectId });
// The pool queue is used even when no worker has registered the pool: executions
// wait on the queue, matching plain queue mode with no workers deployed.
expect(result).toEqual({ queueName: 'jobs-gpu', poolName: 'gpu' });
expect(workerPoolsService.getAvailablePools).not.toHaveBeenCalled();
});
it('falls back to the default queue when worker pools are not licensed', async () => {
licenseState.isWorkerPoolsLicensed.mockReturnValue(false);
projectPoolSettingsRepository.getDefaultPool.mockResolvedValue('gpu');
workerPoolsService.getAvailablePools.mockResolvedValue(['gpu']);
const result = await service.resolvePoolForExecution({ projectId });
expect(result).toEqual({ queueName: 'jobs', poolName: undefined });
expect(projectPoolSettingsRepository.getDefaultPool).not.toHaveBeenCalled();
});
it('falls back to the default queue when the project has no pool set', async () => {
projectPoolSettingsRepository.getDefaultPool.mockResolvedValue(null);
const result = await service.resolvePoolForExecution({ projectId });
expect(result).toEqual({ queueName: 'jobs', poolName: undefined });
});
it('falls back to the default queue when there is no project id', async () => {
const result = await service.resolvePoolForExecution({ projectId: undefined });
expect(result).toEqual({ queueName: 'jobs', poolName: undefined });
expect(projectPoolSettingsRepository.getDefaultPool).not.toHaveBeenCalled();
});
it('forces the default queue when worker pools are disabled, ignoring DB rows', async () => {
setEnabled(false);
projectPoolSettingsRepository.getDefaultPool.mockResolvedValue('gpu');
workerPoolsService.getAvailablePools.mockResolvedValue(['gpu']);
const result = await service.resolvePoolForExecution({ projectId });
expect(result).toEqual({ queueName: 'jobs', poolName: undefined });
expect(projectPoolSettingsRepository.getDefaultPool).not.toHaveBeenCalled();
});
it('reads the project pool from cache on a cache hit', async () => {
cacheService.get.mockResolvedValue('gpu');
workerPoolsService.getAvailablePools.mockResolvedValue(['gpu']);
const result = await service.resolvePoolForExecution({ projectId });
expect(result).toEqual({ queueName: 'jobs-gpu', poolName: 'gpu' });
expect(projectPoolSettingsRepository.getDefaultPool).not.toHaveBeenCalled();
});
});
describe('getProjectPoolSettings', () => {
const projectId = 'project-123';
it('returns the stored default pool and the available pools', async () => {
projectPoolSettingsRepository.getDefaultPool.mockResolvedValue('gpu');
workerPoolsService.getAvailablePools.mockResolvedValue(['gpu', 'cpu']);
const result = await service.getProjectPoolSettings(projectId);
expect(result).toEqual({ defaultPool: 'gpu', availablePools: ['gpu', 'cpu'] });
});
it('returns null defaultPool when none is stored', async () => {
projectPoolSettingsRepository.getDefaultPool.mockResolvedValue(null);
const result = await service.getProjectPoolSettings(projectId);
expect(result).toEqual({ defaultPool: null, availablePools: [] });
});
});
describe('setProjectPoolSettings', () => {
const projectId = 'project-123';
it('persists the default pool and invalidates the project cache', async () => {
workerPoolsService.getAvailablePools.mockResolvedValue(['gpu']);
const result = await service.setProjectPoolSettings(projectId, { defaultPool: 'gpu' });
expect(result).toEqual({ defaultPool: 'gpu', availablePools: ['gpu'] });
expect(projectPoolSettingsRepository.setDefaultPool).toHaveBeenCalledWith(projectId, 'gpu');
expect(cacheService.delete).toHaveBeenCalledWith(`projectPool:${projectId}`);
});
it('clears the default pool when given an empty string', async () => {
const result = await service.setProjectPoolSettings(projectId, { defaultPool: '' });
expect(result.defaultPool).toBeNull();
expect(projectPoolSettingsRepository.setDefaultPool).toHaveBeenCalledWith(projectId, null);
});
it('clears the default pool when given null', async () => {
const result = await service.setProjectPoolSettings(projectId, { defaultPool: null });
expect(result.defaultPool).toBeNull();
expect(projectPoolSettingsRepository.setDefaultPool).toHaveBeenCalledWith(projectId, null);
});
});
});

34
packages/cli/src/scaling/__tests__/queue-name.test.ts Normal file
View File

@ -0,0 +1,34 @@
import type { WorkerPoolConfig } from '@n8n/config';
import { resolveQueueName, resolveWorkerPoolName } from '../queue-name';
describe('resolveQueueName', () => {
test('returns "jobs" on main regardless of pool', () => {
expect(resolveQueueName('main', '')).toBe('jobs');
expect(resolveQueueName('main', 'gpu')).toBe('jobs');
});
test('returns "jobs" on webhook regardless of pool', () => {
expect(resolveQueueName('webhook', '')).toBe('jobs');
expect(resolveQueueName('webhook', 'gpu')).toBe('jobs');
});
test('returns "jobs" on worker when pool is empty', () => {
expect(resolveQueueName('worker', '')).toBe('jobs');
});
test('returns "jobs-<pool>" on worker when pool is set', () => {
expect(resolveQueueName('worker', 'gpu')).toBe('jobs-gpu');
expect(resolveQueueName('worker', 'a1-b2')).toBe('jobs-a1-b2');
});
});
describe('resolveWorkerPoolName', () => {
test('returns the pool name when pools are enabled', () => {
expect(resolveWorkerPoolName({ enabled: true, name: 'gpu' } as WorkerPoolConfig)).toBe('gpu');
});
test('returns empty string when pools are disabled, even with a name set', () => {
expect(resolveWorkerPoolName({ enabled: false, name: 'gpu' } as WorkerPoolConfig)).toBe('');
});
});

76
packages/cli/src/scaling/__tests__/scaling.service.test.ts
View File

@ -1,5 +1,5 @@
import { mockLogger, mockInstance } from '@n8n/backend-test-utils';
import { GlobalConfig } from '@n8n/config';
import { GlobalConfig, WorkerPoolConfig } from '@n8n/config';
import type { ExecutionRepository } from '@n8n/db';
import { Container } from '@n8n/di';
import * as BullModule from 'bull';
@ -11,7 +11,7 @@ import { mock } from 'vitest-mock-extended';
import type { ActiveExecutions } from '@/active-executions';
import type { ExecutionPersistence } from '@/executions/execution-persistence';
import { JOB_TYPE_NAME, QUEUE_NAME } from '../constants';
import { JOB_TYPE_NAME } from '../constants';
import type { JobProcessor } from '../job-processor';
import { ScalingService } from '../scaling.service';
import type { Job, JobData, JobId, JobQueue } from '../scaling.types';
@ -66,6 +66,7 @@ describe('ScalingService', () => {
tls: false,
},
},
workerPool: Object.assign(new WorkerPoolConfig(), { enabled: true, name: '' }),
},
endpoints: {
metrics: {
@ -100,8 +101,8 @@ describe('ScalingService', () => {
let stopQueueMetricsSpy: MockInstance;
let getRunningJobsCountSpy: MockInstance;
const bullConstructorArgs = [
QUEUE_NAME,
const expectedBullArgs = (queueName: string) => [
queueName,
{
prefix: globalConfig.queue.bull.prefix,
settings: { ...globalConfig.queue.bull.settings, maxStalledCount: 0 },
@ -109,6 +110,8 @@ describe('ScalingService', () => {
},
];
const defaultBullArgs = expectedBullArgs('jobs');
beforeEach(() => {
vi.clearAllMocks();
// @ts-expect-error readonly property
@ -153,7 +156,7 @@ describe('ScalingService', () => {
it('should set up queue + listeners + queue recovery', async () => {
await scalingService.setupQueue();
expect(Bull).toHaveBeenCalledWith(...bullConstructorArgs);
expect(Bull).toHaveBeenCalledWith(...defaultBullArgs);
expect(registerMainOrWebhookListenersSpy).toHaveBeenCalled();
expect(registerWorkerListenersSpy).not.toHaveBeenCalled();
expect(scheduleQueueRecoverySpy).toHaveBeenCalledWith(0);
@ -166,7 +169,7 @@ describe('ScalingService', () => {
await scalingService.setupQueue();
expect(Bull).toHaveBeenCalledWith(...bullConstructorArgs);
expect(Bull).toHaveBeenCalledWith(...defaultBullArgs);
expect(registerMainOrWebhookListenersSpy).toHaveBeenCalled();
expect(registerWorkerListenersSpy).not.toHaveBeenCalled();
expect(scheduleQueueRecoverySpy).not.toHaveBeenCalled();
@ -180,7 +183,7 @@ describe('ScalingService', () => {
await scalingService.setupQueue();
expect(Bull).toHaveBeenCalledWith(...bullConstructorArgs);
expect(Bull).toHaveBeenCalledWith(...defaultBullArgs);
expect(registerWorkerListenersSpy).toHaveBeenCalled();
expect(registerMainOrWebhookListenersSpy).not.toHaveBeenCalled();
});
@ -193,11 +196,68 @@ describe('ScalingService', () => {
await scalingService.setupQueue();
expect(Bull).toHaveBeenCalledWith(...bullConstructorArgs);
expect(Bull).toHaveBeenCalledWith(...defaultBullArgs);
expect(registerWorkerListenersSpy).not.toHaveBeenCalled();
expect(registerMainOrWebhookListenersSpy).toHaveBeenCalled();
});
});
describe('queue name resolution', () => {
afterEach(() => {
globalConfig.queue.workerPool.name = '';
globalConfig.queue.workerPool.enabled = true;
});
it('uses "jobs" on worker when pool is empty', async () => {
// @ts-expect-error readonly property
instanceSettings.instanceType = 'worker';
await scalingService.setupQueue();
expect(Bull).toHaveBeenCalledWith(...expectedBullArgs('jobs'));
});
it('uses "jobs-<pool>" on worker when pool is set', async () => {
// @ts-expect-error readonly property
instanceSettings.instanceType = 'worker';
globalConfig.queue.workerPool.name = 'gpu';
await scalingService.setupQueue();
expect(Bull).toHaveBeenCalledWith(...expectedBullArgs('jobs-gpu'));
});
it('uses "jobs" on worker when a pool is set but pools are disabled', async () => {
// @ts-expect-error readonly property
instanceSettings.instanceType = 'worker';
globalConfig.queue.workerPool.name = 'gpu';
globalConfig.queue.workerPool.enabled = false;
await scalingService.setupQueue();
expect(Bull).toHaveBeenCalledWith(...expectedBullArgs('jobs'));
});
it('ignores pool name on main and uses "jobs"', async () => {
// @ts-expect-error readonly property
instanceSettings.instanceType = 'main';
globalConfig.queue.workerPool.name = 'gpu';
await scalingService.setupQueue();
expect(Bull).toHaveBeenCalledWith(...expectedBullArgs('jobs'));
});
it('ignores pool name on webhook and uses "jobs"', async () => {
// @ts-expect-error readonly property
instanceSettings.instanceType = 'webhook';
globalConfig.queue.workerPool.name = 'gpu';
await scalingService.setupQueue();
expect(Bull).toHaveBeenCalledWith(...expectedBullArgs('jobs'));
});
});
});
describe('setupWorker', () => {

66
packages/cli/src/scaling/__tests__/worker-pools.service.ee.test.ts Normal file
View File

@ -0,0 +1,66 @@
import type { InstanceRegistration } from '@n8n/api-types';
import { mock } from 'vitest-mock-extended';
import type { InstanceRegistryService } from '@/modules/instance-registry/instance-registry.service';
import { WorkerPoolsService } from '@/scaling/worker-pools.service.ee';
const buildWorker = (overrides: Partial<InstanceRegistration> = {}): InstanceRegistration => ({
schemaVersion: 1 as const,
instanceKey: 'worker-1',
hostId: 'host-1',
instanceType: 'worker',
instanceRole: 'unset',
version: '1.0.0',
registeredAt: 0,
lastSeen: 0,
...overrides,
});
describe('WorkerPoolsService', () => {
const instanceRegistryService = mock<InstanceRegistryService>();
const service = new WorkerPoolsService(instanceRegistryService);
beforeEach(() => {
vi.clearAllMocks();
});
describe('getAvailablePools', () => {
it('should return distinct pool names from worker registrations, sorted', async () => {
instanceRegistryService.getAllInstances.mockResolvedValue([
buildWorker({ instanceKey: 'w1', poolName: 'gpu' }),
buildWorker({ instanceKey: 'w2', poolName: 'cpu' }),
buildWorker({ instanceKey: 'w3', poolName: 'gpu' }),
]);
await expect(service.getAvailablePools()).resolves.toEqual(['cpu', 'gpu']);
});
it('should skip non-worker instances', async () => {
instanceRegistryService.getAllInstances.mockResolvedValue([
buildWorker({ instanceKey: 'm1', instanceType: 'main' }),
buildWorker({ instanceKey: 'wh1', instanceType: 'webhook' }),
buildWorker({ instanceKey: 'w1', poolName: 'gpu' }),
]);
await expect(service.getAvailablePools()).resolves.toEqual(['gpu']);
});
it('should skip workers without a pool name', async () => {
instanceRegistryService.getAllInstances.mockResolvedValue([
buildWorker({ instanceKey: 'w1' }),
buildWorker({ instanceKey: 'w2', poolName: '' }),
buildWorker({ instanceKey: 'w3', poolName: 'gpu' }),
]);
await expect(service.getAvailablePools()).resolves.toEqual(['gpu']);
});
it('should return an empty array when no workers are registered with pools', async () => {
instanceRegistryService.getAllInstances.mockResolvedValue([
buildWorker({ instanceKey: 'm1', instanceType: 'main' }),
]);
await expect(service.getAvailablePools()).resolves.toEqual([]);
});
});
});

3
packages/cli/src/scaling/__tests__/worker-status.service.test.ts
View File

@ -16,6 +16,7 @@ describe('WorkerStatusService', () => {
workerStatusService = new WorkerStatusService(
mock(), // jobProcessor
mock(), // globalConfig
mock(), // instanceSettings
mock(), // publisher
mockPush,
@ -42,6 +43,8 @@ describe('WorkerStatusService', () => {
hostname: 'test-worker',
interfaces: [],
version: '1.0.0',
poolName: '',
queueName: 'jobs',
isInContainer: false,
process: {
memory: {

2
packages/cli/src/scaling/constants.ts
View File

@ -1,7 +1,5 @@
import type { PubSub } from './pubsub/pubsub.types';
export const QUEUE_NAME = 'jobs';
export const JOB_TYPE_NAME = 'job';
/** Pubsub channel for commands sent by a main process to workers or to other main processes. */

88
packages/cli/src/scaling/pool-config.service.ee.ts Normal file
View File

@ -0,0 +1,88 @@
import type { ProjectPoolSettingsResponse, UpdateProjectPoolSettingsDto } from '@n8n/api-types';
import { LicenseState } from '@n8n/backend-common';
import { GlobalConfig } from '@n8n/config';
import { ProjectPoolSettingsRepository } from '@n8n/db';
import { Service } from '@n8n/di';
import type { IWorkflowExecutionDataProcess } from 'n8n-workflow';
import { CacheService } from '@/services/cache/cache.service';
import { DEFAULT_QUEUE_NAME, poolQueueName } from './queue-name';
import { WorkerPoolsService } from './worker-pools.service.ee';
/**
* Single resolution seam for routing executions to worker pools.
*
* The eventual precedence ladder is:
* workflow override > project per-category > project default > instance default > system queue
* The MVP implements exactly one rung (project default). Re-introducing the other rungs later is a
* change inside `resolvePoolForExecution`, not a re-spread across call sites.
*/
@Service()
export class PoolConfigService {
constructor(
private readonly projectPoolSettingsRepository: ProjectPoolSettingsRepository,
private readonly cacheService: CacheService,
private readonly workerPoolsService: WorkerPoolsService,
private readonly globalConfig: GlobalConfig,
private readonly licenseState: LicenseState,
) {}
/**
* Resolve the queue an execution should run on. Returns the project's default pool when worker
* pools are enabled and licensed; otherwise the system default queue. The pool queue is used
* even when no worker has registered the pool: executions then wait on the queue, matching
* plain queue mode with no workers deployed.
*/
async resolvePoolForExecution(
data: Pick<IWorkflowExecutionDataProcess, 'projectId'>,
): Promise<{ queueName: string; poolName: string | undefined }> {
const systemDefault = { queueName: DEFAULT_QUEUE_NAME, poolName: undefined };
if (!this.globalConfig.queue.workerPool.enabled) return systemDefault;
if (!this.licenseState.isWorkerPoolsLicensed()) return systemDefault;
if (!data.projectId) return systemDefault;
const pool = await this.getProjectDefaultPool(data.projectId);
if (!pool) return systemDefault;
return { queueName: poolQueueName(pool), poolName: pool };
}
async getProjectPoolSettings(projectId: string): Promise<ProjectPoolSettingsResponse> {
const [defaultPool, availablePools] = await Promise.all([
this.getProjectDefaultPool(projectId),
this.workerPoolsService.getAvailablePools(),
]);
return { defaultPool: defaultPool ?? null, availablePools };
}
async setProjectPoolSettings(
projectId: string,
patch: UpdateProjectPoolSettingsDto,
): Promise<ProjectPoolSettingsResponse> {
const defaultPool = patch.defaultPool ? patch.defaultPool : null;
await this.projectPoolSettingsRepository.setDefaultPool(projectId, defaultPool);
await this.cacheService.delete(this.cacheKey(projectId));
const availablePools = await this.workerPoolsService.getAvailablePools();
return { defaultPool, availablePools };
}
private cacheKey(projectId: string): string {
return `projectPool:${projectId}`;
}
private async getProjectDefaultPool(projectId: string): Promise<string | undefined> {
const cacheKey = this.cacheKey(projectId);
const cached = await this.cacheService.get<string>(cacheKey);
if (cached !== undefined) return cached || undefined;
const pool = await this.projectPoolSettingsRepository.getDefaultPool(projectId);
void this.cacheService.set(cacheKey, pool ?? '');
return pool ?? undefined;
}
}

21
packages/cli/src/scaling/queue-name.ts Normal file
View File

@ -0,0 +1,21 @@
import type { WorkerPoolConfig } from '@n8n/config';
import type { InstanceSettings } from 'n8n-core';
export const DEFAULT_QUEUE_NAME = 'jobs';
type InstanceType = InstanceSettings['instanceType'];
/** Disabled pools fall back to the default (unlabeled) queue, so a worker with a pool name still listens to `jobs`. */
export function resolveWorkerPoolName(pool: WorkerPoolConfig): string {
return pool.enabled ? pool.name : '';
}
export function resolveQueueName(instanceType: InstanceType, poolName: string): string {
if (instanceType !== 'worker' || poolName === '') return DEFAULT_QUEUE_NAME;
return `${DEFAULT_QUEUE_NAME}-${poolName}`;
}
export function poolQueueName(poolName: string | undefined): string {
if (!poolName) return DEFAULT_QUEUE_NAME;
return `${DEFAULT_QUEUE_NAME}-${poolName}`;
}

148
packages/cli/src/scaling/scaling.service.ts
View File

@ -17,8 +17,9 @@ import { EventService } from '@/events/event.service';
import { ExecutionPersistence } from '@/executions/execution-persistence';
import { assertNever } from '@/utils';
import { JOB_TYPE_NAME, QUEUE_NAME } from './constants';
import { JOB_TYPE_NAME } from './constants';
import { JobProcessor } from './job-processor';
import { DEFAULT_QUEUE_NAME, resolveQueueName, resolveWorkerPoolName } from './queue-name';
import type {
JobQueue,
Job,
@ -35,7 +36,12 @@ import { decodeRelayedWebhookResponse, WebhookResponseRelay } from './webhook-re
@Service()
export class ScalingService {
private queue: JobQueue;
/** Bull queues keyed by queue name. Pool queues are created lazily. */
private readonly queueByName = new Map<string, JobQueue>();
private initialQueueName = DEFAULT_QUEUE_NAME;
private createBullQueue?: (name: string) => JobQueue;
private jobResults = new Map<string, JobFinishedProps>();
@ -54,27 +60,63 @@ export class ScalingService {
this.logger = this.logger.scoped('scaling');
}
private get defaultQueue(): JobQueue {
const queue = this.queueByName.get(this.initialQueueName);
if (!queue) throw new UnexpectedError('This method must be called after `setupQueue`');
return queue;
}
/** Return an existing queue or lazily create a new Bull queue for the given name. */
async getOrCreateQueue(queueName: string): Promise<JobQueue> {
const existing = this.queueByName.get(queueName);
if (existing) return existing;
if (!this.createBullQueue) {
throw new UnexpectedError('This method must be called after `setupQueue`');
}
const queue = this.createBullQueue(queueName);
this.registerListenersForQueue(queue);
this.queueByName.set(queueName, queue);
this.logger.debug(`Created queue "${queueName}"`);
return queue;
}
// #region Lifecycle
async setupQueue() {
if (this.queueByName.size > 0) return;
const { default: BullQueue } = await import('bull');
const { RedisClientService } = await import('@/services/redis-client.service.js');
if (this.queue) return;
const service = Container.get(RedisClientService);
const bullPrefix = this.globalConfig.queue.bull.prefix;
const prefix = service.toValidPrefix(bullPrefix);
const settings = { ...this.globalConfig.queue.bull.settings, maxStalledCount: 0 };
this.queue = new BullQueue(QUEUE_NAME, {
prefix,
settings: { ...this.globalConfig.queue.bull.settings, maxStalledCount: 0 },
createClient: (type) => service.createClient({ type: `${type}(bull)` }),
this.createBullQueue = (name: string) =>
new BullQueue(name, {
prefix,
settings,
createClient: (type) => service.createClient({ type: `${type}(bull)` }),
});
const poolName = resolveWorkerPoolName(this.globalConfig.queue.workerPool);
const queueName = resolveQueueName(this.instanceSettings.instanceType, poolName);
const queue = this.createBullQueue(queueName);
this.registerListenersForQueue(queue);
this.queueByName.set(queueName, queue);
this.initialQueueName = queueName;
this.logger.debug('Queue setup', {
queueName,
poolName,
instanceType: this.instanceSettings.instanceType,
});
this.registerListeners();
if (this.instanceSettings.isLeader) this.scheduleQueueRecovery(0);
this.scheduleQueueMetrics();
@ -111,7 +153,7 @@ export class ScalingService {
this.assertWorker();
this.assertQueue();
void this.queue.process(JOB_TYPE_NAME, concurrency, async (job: Job) => {
void this.defaultQueue.process(JOB_TYPE_NAME, concurrency, async (job: Job) => {
try {
this.eventService.emit('job-dequeued', {
executionId: job.data.executionId,
@ -167,20 +209,22 @@ export class ScalingService {
else if (instanceType === 'worker') await this.stopWorker();
}
private async pauseQueue() {
await this.queue.pause(true, true); // no more jobs will be enqueued or picked up
this.logger.debug('Paused queue');
private async pauseAllQueues() {
await Promise.all(
[...this.queueByName.values()].map(async (queue) => await queue.pause(true, true)),
);
this.logger.debug('Paused all queues');
}
private async stopMain() {
if (this.instanceSettings.isSingleMain) await this.pauseQueue();
if (this.instanceSettings.isSingleMain) await this.pauseAllQueues();
if (this.queueRecoveryContext.timeout) this.stopQueueRecovery();
if (this.isQueueMetricsEnabled) this.stopQueueMetrics();
}
private async stopWorker() {
await this.pauseQueue();
await this.pauseAllQueues();
let count = 0;
@ -199,7 +243,7 @@ export class ScalingService {
}
async pingQueue() {
await this.queue.client.ping();
await this.defaultQueue.client.ping();
}
// #endregion
@ -214,18 +258,21 @@ export class ScalingService {
}
async getPendingJobCounts() {
const { active, waiting } = await this.queue.getJobCounts();
let active = 0;
let waiting = 0;
for (const queue of this.queueByName.values()) {
const counts = await queue.getJobCounts();
active += counts.active;
waiting += counts.waiting;
}
return { active, waiting };
}
/**
* Add a job to the queue.
*
* @param jobData Data of the job to add to the queue.
* @param priority Priority of the job, from `1` (highest) to `MAX_SAFE_INTEGER` (lowest).
*/
async addJob(jobData: JobData, { priority }: { priority: number }) {
/** Add a job to the given queue (or the default queue). Priority: `1` (highest) to `MAX_SAFE_INTEGER` (lowest). */
async addJob(
jobData: JobData,
{ priority, queueName }: { priority: number; queueName?: string },
) {
strict(priority > 0 && priority <= Number.MAX_SAFE_INTEGER);
const { keepLastCompleted, keepLastFailed } = this.globalConfig.executions.queueRetention;
@ -236,12 +283,18 @@ export class ScalingService {
removeOnFail: keepLastFailed,
};
const job = await this.queue.add(JOB_TYPE_NAME, jobData, jobOptions);
const targetQueue = queueName ? await this.getOrCreateQueue(queueName) : this.defaultQueue;
const job = await targetQueue.add(JOB_TYPE_NAME, jobData, jobOptions);
const { executionId } = jobData;
const jobId = job.id;
this.logger.info(`Enqueued execution ${executionId} (job ${jobId})`, { executionId, jobId });
this.logger.info(`Enqueued execution ${executionId} (job ${jobId})`, {
executionId,
jobId,
...(queueName && queueName !== DEFAULT_QUEUE_NAME ? { queueName } : {}),
});
this.eventService.emit('job-enqueued', {
executionId,
workflowId: jobData.workflowId,
@ -253,13 +306,20 @@ export class ScalingService {
}
async getJob(jobId: JobId) {
return await this.queue.getJob(jobId);
for (const queue of this.queueByName.values()) {
const job = await queue.getJob(jobId);
if (job) return job;
}
return null;
}
async findJobsByStatus(statuses: JobStatus[]) {
const jobs = await this.queue.getJobs(statuses);
return jobs.filter((job) => job !== null);
const allJobs: Job[] = [];
for (const queue of this.queueByName.values()) {
const jobs = await queue.getJobs(statuses);
allJobs.push(...jobs.filter((job): job is Job => job !== null));
}
return allJobs;
}
async stopJob(job: Job) {
@ -297,26 +357,26 @@ export class ScalingService {
// #region Listeners
private registerListeners() {
private registerListenersForQueue(queue: JobQueue) {
const { instanceType } = this.instanceSettings;
if (instanceType === 'main' || instanceType === 'webhook') {
this.registerMainOrWebhookListeners();
this.registerMainOrWebhookListeners(queue);
} else if (instanceType === 'worker') {
this.registerWorkerListeners();
this.registerWorkerListeners(queue);
}
}
/**
* Register listeners on a `worker` process for Bull queue events.
*/
private registerWorkerListeners() {
this.queue.on('global:progress', (jobId: JobId, msg: unknown) => {
private registerWorkerListeners(queue: JobQueue) {
queue.on('global:progress', (jobId: JobId, msg: unknown) => {
if (!this.isJobMessage(msg)) return;
if (msg.kind === 'abort-job') this.jobProcessor.stopJob(jobId);
});
this.queue.on('error', (error: Error) => {
queue.on('error', (error: Error) => {
if ('code' in error && error.code === 'ECONNREFUSED') return; // handled by RedisClientService.retryStrategy
/**
@ -338,8 +398,8 @@ export class ScalingService {
/**
* Register listeners on a `main` or `webhook` process for Bull queue events.
*/
private registerMainOrWebhookListeners() {
this.queue.on('error', (error: Error) => {
private registerMainOrWebhookListeners(queue: JobQueue) {
queue.on('error', (error: Error) => {
if ('code' in error && error.code === 'ECONNREFUSED') return; // handled by RedisClientService.retryStrategy
this.logger.error('Queue errored', { error });
@ -347,7 +407,7 @@ export class ScalingService {
throw error;
});
this.queue.on('global:progress', (jobId: JobId, msg: unknown) => {
queue.on('global:progress', (jobId: JobId, msg: unknown) => {
if (!this.isJobMessage(msg)) return;
// completion and failure are reported via `global:progress` to convey more details
@ -436,8 +496,8 @@ export class ScalingService {
});
if (this.isQueueMetricsEnabled) {
this.queue.on('global:completed', () => this.jobCounters.completed++);
this.queue.on('global:failed', () => this.jobCounters.failed++);
queue.on('global:completed', () => this.jobCounters.completed++);
queue.on('global:failed', () => this.jobCounters.failed++);
}
}
@ -519,7 +579,7 @@ export class ScalingService {
// #endregion
private assertQueue() {
if (this.queue) return;
if (this.queueByName.size > 0) return;
throw new UnexpectedError('This method must be called after `setupQueue`');
}

1
packages/cli/src/scaling/scaling.types.ts
View File

@ -25,6 +25,7 @@ export type JobData = {
restartExecutionId?: string;
projectId?: string;
projectName?: string;
poolName?: string;
/** Manual-execution identity, so offloaded manual runs resolve private credentials on the worker. */
encryptedRunnerIdentity?: string;

21
packages/cli/src/scaling/worker-pools.service.ee.ts Normal file
View File

@ -0,0 +1,21 @@
import { Service } from '@n8n/di';
import { InstanceRegistryService } from '@/modules/instance-registry/instance-registry.service';
/** Discovers worker pool names registered by workers in the cluster. */
@Service()
export class WorkerPoolsService {
constructor(private readonly instanceRegistryService: InstanceRegistryService) {}
async getAvailablePools(): Promise<string[]> {
const instances = await this.instanceRegistryService.getAllInstances();
const poolNames = new Set<string>();
for (const instance of instances) {
if (instance.instanceType !== 'worker') continue;
if (instance.poolName) poolNames.add(instance.poolName);
}
return Array.from(poolNames).sort();
}
}

8
packages/cli/src/scaling/worker-status.service.ee.ts
View File

@ -1,4 +1,5 @@
import { WorkerStatus } from '@n8n/api-types';
import { GlobalConfig } from '@n8n/config';
import { OnPubSubEvent } from '@n8n/decorators';
import { Service } from '@n8n/di';
import { InstanceSettings } from 'n8n-core';
@ -10,11 +11,13 @@ import { Push } from '@/push';
import { JobProcessor } from './job-processor';
import { Publisher } from './pubsub/publisher.service';
import { resolveQueueName, resolveWorkerPoolName } from './queue-name';
@Service()
export class WorkerStatusService {
constructor(
private readonly jobProcessor: JobProcessor,
private readonly globalConfig: GlobalConfig,
private readonly instanceSettings: InstanceSettings,
private readonly publisher: Publisher,
private readonly push: Push,
@ -96,6 +99,11 @@ export class WorkerStatusService {
})),
),
version: N8N_VERSION,
poolName: resolveWorkerPoolName(this.globalConfig.queue.workerPool),
queueName: resolveQueueName(
this.instanceSettings.instanceType,
resolveWorkerPoolName(this.globalConfig.queue.workerPool),
),
};
}

1
packages/cli/src/server.ts
View File

@ -47,6 +47,7 @@ import '@/controllers/orchestration.controller';
import '@/controllers/owner.controller';
import '@/controllers/password-reset.controller';
import '@/controllers/project.controller';
import '@/controllers/project-pool-settings.controller.ee';
import '@/controllers/role.controller';
import '@/controllers/tags.controller';
import '@/controllers/translation.controller';

32
packages/cli/src/services/__tests__/frontend.service.test.ts
View File

@ -81,6 +81,7 @@ describe('FrontendService', () => {
},
aiAssistant: { baseUrl: '' },
aiGateway: { enabled: false },
queue: { workerPool: { enabled: false } },
});
const instanceSettings = mock<InstanceSettings>({
@ -168,6 +169,7 @@ describe('FrontendService', () => {
isOidcLicensed: vi.fn().mockReturnValue(false),
isMFAEnforcementLicensed: vi.fn().mockReturnValue(false),
isOtelCustomSpanAttributesLicensed: vi.fn().mockReturnValue(false),
isWorkerPoolsLicensed: vi.fn().mockReturnValue(false),
getMaxWorkflowsWithEvaluations: vi.fn().mockReturnValue(0),
});
@ -530,6 +532,36 @@ describe('FrontendService', () => {
expect(settings.enterprise.otelCustomSpanAttributes).toBe(true);
});
it('should enable worker pools when both the config flag and the license are on', async () => {
globalConfig.queue = { workerPool: { enabled: true } } as GlobalConfig['queue'];
licenseState.isWorkerPoolsLicensed.mockReturnValue(true);
const { service } = createMockService();
const settings = await service.getSettings();
expect(settings.workerPools.enabled).toBe(true);
});
it('should keep worker pools disabled when the config flag is on but the feature is not licensed', async () => {
globalConfig.queue = { workerPool: { enabled: true } } as GlobalConfig['queue'];
licenseState.isWorkerPoolsLicensed.mockReturnValue(false);
const { service } = createMockService();
const settings = await service.getSettings();
expect(settings.workerPools.enabled).toBe(false);
});
it('should keep worker pools disabled when licensed but the config flag is off', async () => {
globalConfig.queue = { workerPool: { enabled: false } } as GlobalConfig['queue'];
licenseState.isWorkerPoolsLicensed.mockReturnValue(true);
const { service } = createMockService();
const settings = await service.getSettings();
expect(settings.workerPools.enabled).toBe(false);
});
it('should surface useWorkflowPublicationService from workflows config', async () => {
globalConfig.workflows = {
...globalConfig.workflows,

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

@ -423,6 +423,9 @@ export class FrontendService {
folders: {
enabled: false,
},
workerPools: {
enabled: false,
},
evaluation: {
quota: this.licenseState.getMaxWorkflowsWithEvaluations(),
collectionsEnabled: this.globalConfig.evaluation.collectionsEnabled,
@ -544,6 +547,9 @@ export class FrontendService {
workflowReviews: this.licenseState.isWorkflowReviewsLicensed(),
});
this.settings.workerPools.enabled =
this.globalConfig.queue.workerPool.enabled && this.licenseState.isWorkerPoolsLicensed();
if (this.license.isLdapEnabled()) {
Object.assign(this.settings.sso.ldap, {
loginLabel: this.globalConfig.sso.ldap.loginLabel,

5
packages/cli/src/services/ownership-transfer/ownership-transfer.manifest.json
View File

@ -34,6 +34,11 @@
"path": "packages/@n8n/db/src/entities/project-secrets-provider-access.ts",
"reason": "Per-project secrets-provider config; scoped to the deleted project."
},
{
"name": "ProjectPoolSettings",
"path": "packages/@n8n/db/src/entities/project-pool-settings.ee.ts",
"reason": "Per-project worker-pool config; scoped to the deleted project and dropped via FK cascade."
},
{
"name": "RoleMappingRule",
"path": "packages/@n8n/db/src/entities/role-mapping-rule.ts",

87
packages/cli/src/workflow-runner.ts
View File

@ -50,6 +50,7 @@ import { ExternalHooks } from '@/external-hooks';
import type { ResumableExecution } from '@/interfaces';
import { ManualExecutionService } from '@/manual-execution.service';
import { NodeTypes } from '@/node-types';
import type { PoolConfigService } from '@/scaling/pool-config.service.ee';
import type { ScalingService } from '@/scaling/scaling.service';
import type { Job, JobData } from '@/scaling/scaling.types';
import { EngineV2Dispatcher } from '@/services/engine-v2-dispatcher.service';
@ -78,6 +79,8 @@ function flushResponse(res: { flush?: () => void }) {
export class WorkflowRunner {
private scalingService: ScalingService;
private poolConfigService: PoolConfigService;
constructor(
private readonly logger: Logger,
private readonly errorReporter: ErrorReporter,
@ -297,17 +300,29 @@ export class WorkflowRunner {
? this.executionsConfig.mode === 'queue'
: this.executionsConfig.mode === 'queue' && data.executionMode !== 'manual';
if (shouldEnqueue) {
await this.enqueueExecution(
executionId,
workflowId,
data,
loadStaticData,
realtime,
existingExecution?.executionId,
);
} else {
await this.runMainProcess(executionId, data, loadStaticData, existingExecution?.executionId);
try {
if (shouldEnqueue) {
await this.enqueueExecution(
executionId,
workflowId,
data,
loadStaticData,
realtime,
existingExecution?.executionId,
);
} else {
await this.runMainProcess(
executionId,
data,
loadStaticData,
existingExecution?.executionId,
);
}
} catch (error) {
// A failed start means the post-execute promise that normally clears the
// heartbeat never settles, so clear it here.
if (heartbeatInterval) clearInterval(heartbeatInterval);
throw error;
}
// only run these when not in queue mode or when the execution is manual,
@ -528,38 +543,46 @@ export class WorkflowRunner {
realtime?: boolean,
restartExecutionId?: string,
): Promise<void> {
const jobData: JobData = {
workflowId,
executionId,
loadStaticData: !!loadStaticData,
pushRef: data.pushRef,
streamingEnabled: data.streamingEnabled,
restartExecutionId,
projectId: data.projectId,
projectName: data.projectName,
// Carry the manual-execution identity for private credential resolution on the worker.
encryptedRunnerIdentity: data.encryptedRunnerIdentity,
// MCP-specific fields for queue mode support
isMcpExecution: data.isMcpExecution,
mcpType: data.mcpType,
mcpSessionId: data.mcpSessionId,
mcpMessageId: data.mcpMessageId,
mcpToolCall: data.mcpToolCall,
mcpToolInput: data.mcpToolInput,
};
if (!this.scalingService) {
const { ScalingService } = await import('@/scaling/scaling.service.js');
this.scalingService = Container.get(ScalingService);
await this.scalingService.setupQueue();
}
if (!this.poolConfigService) {
const { PoolConfigService } = await import('@/scaling/pool-config.service.ee.js');
this.poolConfigService = Container.get(PoolConfigService);
}
// TODO: For realtime jobs should probably also not do retry or not retry if they are older than x seconds.
// Check if they get retried by default and how often.
let job: Job;
let lifecycleHooks: ExecutionLifecycleHooks;
try {
job = await this.scalingService.addJob(jobData, { priority: realtime ? 50 : 100 });
const { queueName, poolName } = await this.poolConfigService.resolvePoolForExecution(data);
const jobData: JobData = {
workflowId,
executionId,
loadStaticData: !!loadStaticData,
pushRef: data.pushRef,
streamingEnabled: data.streamingEnabled,
restartExecutionId,
projectId: data.projectId,
projectName: data.projectName,
// Carry the manual-execution identity for private credential resolution on the worker.
encryptedRunnerIdentity: data.encryptedRunnerIdentity,
poolName,
// MCP-specific fields for queue mode support
isMcpExecution: data.isMcpExecution,
mcpType: data.mcpType,
mcpSessionId: data.mcpSessionId,
mcpMessageId: data.mcpMessageId,
mcpToolCall: data.mcpToolCall,
mcpToolInput: data.mcpToolInput,
};
job = await this.scalingService.addJob(jobData, { priority: realtime ? 50 : 100, queueName });
lifecycleHooks = getLifecycleHooksForScalingMain(data, executionId);

3
packages/frontend/@n8n/frontend-test-utils/src/defaults.ts
View File

@ -180,6 +180,9 @@ export const defaultSettings: FrontendSettings = {
folders: {
enabled: false,
},
workerPools: {
enabled: false,
},
collaboration: {
crdt: 'off',
},

6
packages/frontend/@n8n/i18n/src/locales/en.json
View File

@ -1652,6 +1652,9 @@
"workerList.pageTitle": "Workers",
"workerList.empty": "No workers are responding or available",
"workerList.item.lastUpdated": "Last updated",
"workerList.item.pool": "Pool",
"workerList.item.pool.default": "default",
"workerList.item.queue": "Queue",
"workerList.item.jobList.empty": "No current jobs",
"workerList.item.jobListTitle": "Current Jobs",
"workerList.item.netListTitle": "Network Interfaces",
@ -5950,6 +5953,9 @@
"projects.settings.member.removed.title": "Member removed successfully",
"projects.settings.member.added.title": "Member added successfully",
"projects.settings.users.search.error": "An error occurred while searching for users",
"projects.settings.workerPools.title": "Worker pool",
"projects.settings.workerPools.defaultQueue": "Default queue",
"projects.settings.workerPools.help": "Route all executions in this project to a specific worker pool. Leave on \"Default queue\" to use the system default.",
"projects.settings.externalSecrets": "External secret vaults",
"projects.settings.externalSecrets.emptyState.heading": "No external secrets available yet",
"projects.settings.externalSecrets.emptyState.instanceAdmin.noProjectProviders.description": "Share secrets vault with this project to make external secrets available here. Once shared, secrets will appear automatically and can be referenced through aliases in credentials.",

17
packages/frontend/@n8n/rest-api-client/src/api/orchestration.ts
View File

@ -1,3 +1,5 @@
import type { ProjectPoolSettingsResponse, UpdateProjectPoolSettingsDto } from '@n8n/api-types';
import type { IRestApiContext } from '../types';
import { makeRestApiRequest } from '../utils';
@ -6,3 +8,18 @@ const GET_STATUS_ENDPOINT = '/orchestration/worker/status';
export const sendGetWorkerStatus = async (context: IRestApiContext): Promise<void> => {
await makeRestApiRequest(context, 'POST', GET_STATUS_ENDPOINT);
};
export const getProjectPoolSettings = async (
context: IRestApiContext,
projectId: string,
): Promise<ProjectPoolSettingsResponse> => {
return await makeRestApiRequest(context, 'GET', `/projects/${projectId}/pool-settings`);
};
export const updateProjectPoolSettings = async (
context: IRestApiContext,
projectId: string,
dto: UpdateProjectPoolSettingsDto,
): Promise<ProjectPoolSettingsResponse> => {
return await makeRestApiRequest(context, 'PATCH', `/projects/${projectId}/pool-settings`, dto);
};

3
packages/frontend/@n8n/stores/src/settings.store.ts
View File

@ -213,6 +213,8 @@ export const useSettingsStore = defineStore(STORES.SETTINGS, () => {
return isOtelCustomSpanAttributesLicensed && isOtelModuleActive;
});
const isWorkerPoolsEnabled = computed(() => settings.value.workerPools?.enabled ?? false);
// Opt-in flag controlled by the backend's N8N_AGENTS_AI_SANDBOX_ENABLED setting.
const isAgentsKnowledgeBaseFeatureEnabled = computed(
() => isModuleActive('agents') && moduleSettings.value.agents?.knowledgeBaseEnabled === true,
@ -546,6 +548,7 @@ export const useSettingsStore = defineStore(STORES.SETTINGS, () => {
isDataTableFeatureEnabled,
isChatFeatureEnabled,
isOtelCustomSpanAttributesEnabled,
isWorkerPoolsEnabled,
isAgentsKnowledgeBaseFeatureEnabled,
isPublicChatTriggerDisabled,
isWorkflowPublicationServiceEnabled,

56
packages/frontend/editor-ui/src/features/collaboration/projects/components/ProjectWorkerPoolsSection.vue Normal file
View File

@ -0,0 +1,56 @@
<script setup lang="ts">
import { N8nOption, N8nSelect, N8nText } from '@n8n/design-system';
import { useI18n } from '@n8n/i18n';
const i18n = useI18n();
const props = defineProps<{
defaultPool: string | null;
availablePools: string[];
disabled?: boolean;
}>();
const emit = defineEmits<{
'update:defaultPool': [value: string | null];
}>();
const onUpdate = (value: string) => {
emit('update:defaultPool', value === '' ? '' : value);
};
</script>
<template>
<fieldset class="project-worker-pools" data-test-id="project-worker-pools-section">
<h3>
<label for="projectDefaultPool">
{{ i18n.baseText('projects.settings.workerPools.title') }}
</label>
</h3>
<N8nSelect
id="projectDefaultPool"
:model-value="props.defaultPool ?? ''"
:disabled="props.disabled"
:class="$style.input"
data-test-id="project-default-pool-input"
@update:model-value="onUpdate"
>
<N8nOption :label="i18n.baseText('projects.settings.workerPools.defaultQueue')" :value="''" />
<N8nOption v-for="pool in props.availablePools" :key="pool" :label="pool" :value="pool" />
</N8nSelect>
<N8nText color="text-light" size="small" :class="$style.help">
{{ i18n.baseText('projects.settings.workerPools.help') }}
</N8nText>
</fieldset>
</template>
<style module lang="scss">
.input {
max-width: var(--project-field--width);
}
.help {
display: block;
margin-top: var(--spacing--3xs);
}
</style>

37
packages/frontend/editor-ui/src/features/collaboration/projects/projects.store.ts
View File

@ -8,7 +8,11 @@ import * as workflowsApi from '@/app/api/workflows';
import * as workflowsEEApi from '@/app/api/workflows.ee';
import * as credentialsApi from '@/features/credentials/credentials.api';
import * as credentialsEEApi from '@/features/credentials/credentials.ee.api';
import { getProjectSecretProviderConnectionsByProjectId } from '@n8n/rest-api-client';
import {
getProjectPoolSettings,
getProjectSecretProviderConnectionsByProjectId,
updateProjectPoolSettings,
} from '@n8n/rest-api-client';
import type { Project, ProjectListItem, ProjectsCount } from './projects.types';
import { ProjectTypes } from './projects.types';
import { useSettingsStore } from '@n8n/stores/settings.store';
@ -18,7 +22,13 @@ import { useCredentialsStore } from '@/features/credentials/credentials.store';
import { STORES } from '@n8n/stores';
import { useUsersStore } from '@n8n/stores/users.store';
import { getResourcePermissions } from '@n8n/permissions';
import type { CreateProjectDto, UpdateProjectDto, SecretProviderConnection } from '@n8n/api-types';
import type {
CreateProjectDto,
ProjectPoolSettingsResponse,
SecretProviderConnection,
UpdateProjectDto,
UpdateProjectPoolSettingsDto,
} from '@n8n/api-types';
import { useSourceControlStore } from '@/features/integrations/sourceControl.ee/sourceControl.store';
import { hasRole } from '@/app/utils/rbac/checks';
import { useFavoritesStore } from '@/app/stores/favorites.store';
@ -41,6 +51,7 @@ export const useProjectsStore = defineStore(STORES.PROJECTS, () => {
const myProjects = ref<ProjectListItem[]>([]);
const personalProject = ref<Project | null>(null);
const currentProject = ref<Project | null>(null);
const currentProjectPoolSettings = ref<ProjectPoolSettingsResponse | null>(null);
const projectsCount = ref<ProjectsCount>({
personal: 0,
team: 0,
@ -189,6 +200,25 @@ export const useProjectsStore = defineStore(STORES.PROJECTS, () => {
}
};
const fetchProjectPoolSettings = async (projectId: string): Promise<void> => {
currentProjectPoolSettings.value = null;
currentProjectPoolSettings.value = await getProjectPoolSettings(
rootStore.restApiContext,
projectId,
);
};
const updateCurrentProjectPoolSettings = async (
projectId: string,
dto: UpdateProjectPoolSettingsDto,
): Promise<void> => {
currentProjectPoolSettings.value = await updateProjectPoolSettings(
rootStore.restApiContext,
projectId,
dto,
);
};
const addMember = async (
projectId: string,
{ userId, role }: { userId: string; role: string },
@ -380,6 +410,9 @@ export const useProjectsStore = defineStore(STORES.PROJECTS, () => {
refreshCurrentProject,
createProject,
updateProject,
currentProjectPoolSettings,
fetchProjectPoolSettings,
updateCurrentProjectPoolSettings,
addMember,
updateMemberRole,
removeMember,

1
packages/frontend/editor-ui/src/features/collaboration/projects/views/ProjectSettings.test.ts
View File

@ -173,6 +173,7 @@ describe('ProjectSettings', () => {
projectsStore = mockedStore(useProjectsStore);
projectsStore.searchProjects.mockResolvedValue({ count: projects.length, data: projects });
projectsStore.fetchProjectPoolSettings.mockResolvedValue(undefined);
projectsStore.globalProjectPermissions = { list: true };
usersStore = mockedStore(useUsersStore);
settingsStore = mockedStore(useSettingsStore);

72
packages/frontend/editor-ui/src/features/collaboration/projects/views/ProjectSettings.vue
View File

@ -28,6 +28,7 @@ import type { UserAction } from '@n8n/design-system';
import { isProjectRole } from '@/app/utils/typeGuards';
import ProjectExternalSecrets from '../components/ProjectExternalSecrets.vue';
import ProjectSettingsCustomTelemetryTags from '../components/ProjectSettingsCustomTelemetryTags.vue';
import ProjectWorkerPoolsSection from '../components/ProjectWorkerPoolsSection.vue';
import { getResourcePermissions } from '@n8n/permissions';
import { hasPermission } from '@/app/utils/rbac/permissions';
@ -91,12 +92,14 @@ const resourceCounts = ref<ResourceCounts>({
const formData = ref<
Pick<Project, 'name' | 'description' | 'relations'> & {
customTelemetryTags: NonNullable<Project['customTelemetryTags']>;
defaultPool: string | null;
}
>({
name: '',
description: '',
relations: [],
customTelemetryTags: [],
defaultPool: null,
});
// Used to skip one watcher sync after targeted server updates (e.g., immediate removal)
const suppressNextSync = ref(false);
@ -290,6 +293,14 @@ const resetFormData = () => {
? deepCopy(projectsStore.currentProject.customTelemetryTags)
: [];
telemetryTagsRef.value?.resetTouched();
formData.value.defaultPool = projectsStore.currentProjectPoolSettings?.defaultPool ?? null;
};
const hasWorkerPoolsChanges = (): boolean => {
const stored = projectsStore.currentProjectPoolSettings?.defaultPool || null;
// Treat empty string and null as equivalent (both mean "default queue")
const current = formData.value.defaultPool || null;
return current !== stored;
};
const onCancel = () => {
@ -364,13 +375,28 @@ const updateProject = async () => {
return;
}
try {
await projectsStore.updateProject(projectsStore.currentProject.id, {
name: formData.value.name ?? '',
description: formData.value.description ?? '',
...(settingsStore.isOtelCustomSpanAttributesEnabled
? { customTelemetryTags: formData.value.customTelemetryTags }
: {}),
});
const projectId = projectsStore.currentProject.id;
const tasks: Array<Promise<void>> = [];
tasks.push(
projectsStore.updateProject(projectId, {
name: formData.value.name ?? '',
description: formData.value.description ?? '',
...(settingsStore.isOtelCustomSpanAttributesEnabled
? { customTelemetryTags: formData.value.customTelemetryTags }
: {}),
}),
);
if (isWorkerPoolsEnabled.value && hasWorkerPoolsChanges()) {
tasks.push(
projectsStore.updateCurrentProjectPoolSettings(projectId, {
defaultPool: formData.value.defaultPool ?? '',
}),
);
}
await Promise.all(tasks);
isDirty.value = false;
} catch (error) {
showSaveError(error);
@ -447,7 +473,15 @@ const onIconUpdated = async () => {
}
};
// Skip one sync after targeted updates (e.g. removal) to preserve unsaved edits
const isWorkerPoolsEnabled = computed(() => settingsStore.isWorkerPoolsEnabled);
const resetPoolSettingsFormData = () => {
formData.value.defaultPool = projectsStore.currentProjectPoolSettings?.defaultPool ?? null;
};
// Skip one sync after targeted updates (e.g. removal) to preserve unsaved edits.
// Also the single hook reacting to the project becoming available (it loads asynchronously on a
// direct page load), so project-scoped data like the pool settings is fetched here too.
watch(
() => projectsStore.currentProject,
async () => {
@ -461,6 +495,16 @@ watch(
if (projectsStore.currentProject?.icon && isIconOrEmoji(projectsStore.currentProject.icon)) {
projectIcon.value = projectsStore.currentProject.icon;
}
const projectId = projectsStore.currentProjectId;
if (canUpdateProject.value && projectId && isWorkerPoolsEnabled.value) {
await projectsStore
.fetchProjectPoolSettings(projectId)
.catch(() => {
// Non-fatal; store has been cleared, fall back to defaults below.
})
.finally(resetPoolSettingsFormData);
}
},
{ immediate: true },
);
@ -736,6 +780,18 @@ onMounted(async () => {
@validate="isTelemetryTagsValid = $event"
/>
</fieldset>
<ProjectWorkerPoolsSection
v-if="isWorkerPoolsEnabled"
:default-pool="formData.defaultPool"
:available-pools="projectsStore.currentProjectPoolSettings?.availablePools ?? []"
@update:default-pool="
(v) => {
formData.defaultPool = v;
onTextInput();
}
"
/>
<fieldset>
<h3 class="mb-m">{{ i18n.baseText('projects.settings.danger.title') }}</h3>
<small :class="$style.danger">{{

6
packages/frontend/editor-ui/src/features/settings/orchestration.ee/components/WorkerCard.vue
View File

@ -69,7 +69,11 @@ onBeforeUnmount(() => {
:class="stale ? [$style.cardHeading, $style.stale] : [$style.cardHeading]"
data-test-id="worker-card-name"
>
Name: {{ worker.senderId }} ({{ worker.hostname }}) <br />
Name: {{ worker.senderId }} ({{ worker.hostname }}) |
{{ i18n.baseText('workerList.item.pool') }}:
{{ worker.poolName || i18n.baseText('workerList.item.pool.default') }} |
{{ i18n.baseText('workerList.item.queue') }}: {{ worker.queueName }}
<br />
Average Load: {{ averageWorkerLoadFromLoadsAsString(worker.loadAvg ?? [0]) }} | Free memory:
{{ memAsGb(worker.process.memory.available) }}GB /
{{