From 8272a8bc3314d0332cc0b36f981fc2140d4e83d5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mike=20Repe=C4=87?= Date: Fri, 4 Sep 2026 10:19:28 +0000 Subject: [PATCH] feat(core): Add project-level worker pools (no-changelog) (#32975) 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 Co-authored-by: Michael Siega --- docs/generated/postgres-schema/README.md | 8 + .../postgres-schema/public.project.md | 9 +- .../public.project_pool_settings.md | 56 +++++++ docs/generated/sqlite-schema/README.md | 8 + docs/generated/sqlite-schema/project.md | 9 +- .../sqlite-schema/project_pool_settings.md | 65 ++++++++ .../__tests__/instance-registry-types.test.ts | 25 +++ packages/@n8n/api-types/src/dto/index.ts | 2 + .../update-project-pool-settings.dto.ts | 10 ++ .../@n8n/api-types/src/frontend-settings.ts | 3 + packages/@n8n/api-types/src/index.ts | 1 + .../api-types/src/instance-registry-types.ts | 2 + packages/@n8n/api-types/src/scaling.ts | 2 + packages/@n8n/api-types/src/worker-pools.ts | 6 + .../@n8n/backend-common/src/license-state.ts | 4 + .../__tests__/scaling-mode.config.test.ts | 66 ++++++++ .../config/src/configs/scaling-mode.config.ts | 31 ++++ packages/@n8n/config/src/index.ts | 1 + packages/@n8n/config/test/config.test.ts | 4 + packages/@n8n/constants/src/index.ts | 1 + packages/@n8n/db/src/entities/index.ts | 3 + .../src/entities/project-pool-settings.ee.ts | 18 +++ ...1788445119183-CreateProjectPoolSettings.ts | 22 +++ packages/@n8n/db/src/repositories/index.ts | 1 + .../project-pool-settings.repository.ee.ts | 20 +++ .../scope-information.test.ts.snap | 1 + packages/@n8n/permissions/src/constants.ee.ts | 2 +- .../src/roles/scopes/global-scopes.ee.ts | 1 + .../cli/src/__tests__/workflow-runner.test.ts | 27 ++++ .../cli/src/commands/__tests__/worker.test.ts | 1 + packages/cli/src/commands/worker.ts | 21 ++- .../orchestration.controller.test.ts | 36 +++++ ...roject-pool-settings.controller.ee.test.ts | 48 ++++++ .../cli/src/controllers/e2e.controller.ts | 1 + .../project-pool-settings.controller.ee.ts | 35 +++++ .../instance-registry.service.test.ts | 70 ++++++++- .../instance-registry.service.ts | 12 +- .../__tests__/pool-config.service.ee.test.ts | 148 ++++++++++++++++++ .../src/scaling/__tests__/queue-name.test.ts | 34 ++++ .../scaling/__tests__/scaling.service.test.ts | 76 ++++++++- .../__tests__/worker-pools.service.ee.test.ts | 66 ++++++++ .../__tests__/worker-status.service.test.ts | 3 + packages/cli/src/scaling/constants.ts | 2 - .../cli/src/scaling/pool-config.service.ee.ts | 88 +++++++++++ packages/cli/src/scaling/queue-name.ts | 21 +++ packages/cli/src/scaling/scaling.service.ts | 148 ++++++++++++------ packages/cli/src/scaling/scaling.types.ts | 1 + .../src/scaling/worker-pools.service.ee.ts | 21 +++ .../src/scaling/worker-status.service.ee.ts | 8 + packages/cli/src/server.ts | 1 + .../__tests__/frontend.service.test.ts | 32 ++++ packages/cli/src/services/frontend.service.ts | 6 + .../ownership-transfer.manifest.json | 5 + packages/cli/src/workflow-runner.ts | 87 ++++++---- .../@n8n/frontend-test-utils/src/defaults.ts | 3 + .../frontend/@n8n/i18n/src/locales/en.json | 6 + .../rest-api-client/src/api/orchestration.ts | 17 ++ .../@n8n/stores/src/settings.store.ts | 3 + .../components/ProjectWorkerPoolsSection.vue | 56 +++++++ .../collaboration/projects/projects.store.ts | 37 ++++- .../projects/views/ProjectSettings.test.ts | 1 + .../projects/views/ProjectSettings.vue | 72 ++++++++- .../components/WorkerCard.vue | 6 +- 63 files changed, 1474 insertions(+), 106 deletions(-) create mode 100644 docs/generated/postgres-schema/public.project_pool_settings.md create mode 100644 docs/generated/sqlite-schema/project_pool_settings.md create mode 100644 packages/@n8n/api-types/src/dto/worker-pools/update-project-pool-settings.dto.ts create mode 100644 packages/@n8n/api-types/src/worker-pools.ts create mode 100644 packages/@n8n/config/src/configs/__tests__/scaling-mode.config.test.ts create mode 100644 packages/@n8n/db/src/entities/project-pool-settings.ee.ts create mode 100644 packages/@n8n/db/src/migrations/common/1788445119183-CreateProjectPoolSettings.ts create mode 100644 packages/@n8n/db/src/repositories/project-pool-settings.repository.ee.ts create mode 100644 packages/cli/src/controllers/__tests__/orchestration.controller.test.ts create mode 100644 packages/cli/src/controllers/__tests__/project-pool-settings.controller.ee.test.ts create mode 100644 packages/cli/src/controllers/project-pool-settings.controller.ee.ts create mode 100644 packages/cli/src/scaling/__tests__/pool-config.service.ee.test.ts create mode 100644 packages/cli/src/scaling/__tests__/queue-name.test.ts create mode 100644 packages/cli/src/scaling/__tests__/worker-pools.service.ee.test.ts create mode 100644 packages/cli/src/scaling/pool-config.service.ee.ts create mode 100644 packages/cli/src/scaling/queue-name.ts create mode 100644 packages/cli/src/scaling/worker-pools.service.ee.ts create mode 100644 packages/frontend/editor-ui/src/features/collaboration/projects/components/ProjectWorkerPoolsSection.vue diff --git a/docs/generated/postgres-schema/README.md b/docs/generated/postgres-schema/README.md index b15c27ca1ad..9eed0b3a928 100644 --- a/docs/generated/postgres-schema/README.md +++ b/docs/generated/postgres-schema/README.md @@ -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 diff --git a/docs/generated/postgres-schema/public.project.md b/docs/generated/postgres-schema/public.project.md index c8da0df4530..04551b05278 100644 --- a/docs/generated/postgres-schema/public.project.md +++ b/docs/generated/postgres-schema/public.project.md @@ -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 diff --git a/docs/generated/postgres-schema/public.project_pool_settings.md b/docs/generated/postgres-schema/public.project_pool_settings.md new file mode 100644 index 00000000000..5dc9dc80043 --- /dev/null +++ b/docs/generated/postgres-schema/public.project_pool_settings.md @@ -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) diff --git a/docs/generated/sqlite-schema/README.md b/docs/generated/sqlite-schema/README.md index fe013312518..00b3a6970a1 100644 --- a/docs/generated/sqlite-schema/README.md +++ b/docs/generated/sqlite-schema/README.md @@ -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 diff --git a/docs/generated/sqlite-schema/project.md b/docs/generated/sqlite-schema/project.md index 4744a663081..f2adf8c9679 100644 --- a/docs/generated/sqlite-schema/project.md +++ b/docs/generated/sqlite-schema/project.md @@ -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 diff --git a/docs/generated/sqlite-schema/project_pool_settings.md b/docs/generated/sqlite-schema/project_pool_settings.md new file mode 100644 index 00000000000..60ea686c1eb --- /dev/null +++ b/docs/generated/sqlite-schema/project_pool_settings.md @@ -0,0 +1,65 @@ +# project_pool_settings + +## Description + +
+Table Definition + +```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) +``` + +
+ +## 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) diff --git a/packages/@n8n/api-types/src/__tests__/instance-registry-types.test.ts b/packages/@n8n/api-types/src/__tests__/instance-registry-types.test.ts index e08163e49ab..ef6a50ef2df 100644 --- a/packages/@n8n/api-types/src/__tests__/instance-registry-types.test.ts +++ b/packages/@n8n/api-types/src/__tests__/instance-registry-types.test.ts @@ -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); diff --git a/packages/@n8n/api-types/src/dto/index.ts b/packages/@n8n/api-types/src/dto/index.ts index 320f5511db9..8a91f7ce6e3 100644 --- a/packages/@n8n/api-types/src/dto/index.ts +++ b/packages/@n8n/api-types/src/dto/index.ts @@ -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 { diff --git a/packages/@n8n/api-types/src/dto/worker-pools/update-project-pool-settings.dto.ts b/packages/@n8n/api-types/src/dto/worker-pools/update-project-pool-settings.dto.ts new file mode 100644 index 00000000000..3050f38f64a --- /dev/null +++ b/packages/@n8n/api-types/src/dto/worker-pools/update-project-pool-settings.dto.ts @@ -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(), +}) {} diff --git a/packages/@n8n/api-types/src/frontend-settings.ts b/packages/@n8n/api-types/src/frontend-settings.ts index 9583ccb32de..64ecddb54d4 100644 --- a/packages/@n8n/api-types/src/frontend-settings.ts +++ b/packages/@n8n/api-types/src/frontend-settings.ts @@ -233,6 +233,9 @@ export interface FrontendSettings { folders: { enabled: boolean; }; + workerPools: { + enabled: boolean; + }; collaboration: { crdt: 'off' | 'local' | 'server'; }; diff --git a/packages/@n8n/api-types/src/index.ts b/packages/@n8n/api-types/src/index.ts index bf4950ea140..1a23e6bcba6 100644 --- a/packages/@n8n/api-types/src/index.ts +++ b/packages/@n8n/api-types/src/index.ts @@ -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'; diff --git a/packages/@n8n/api-types/src/instance-registry-types.ts b/packages/@n8n/api-types/src/instance-registry-types.ts index 04143527da5..9eeab861ddc 100644 --- a/packages/@n8n/api-types/src/instance-registry-types.ts +++ b/packages/@n8n/api-types/src/instance-registry-types.ts @@ -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(); diff --git a/packages/@n8n/api-types/src/scaling.ts b/packages/@n8n/api-types/src/scaling.ts index f2dcbb52b95..fa54053673c 100644 --- a/packages/@n8n/api-types/src/scaling.ts +++ b/packages/@n8n/api-types/src/scaling.ts @@ -44,4 +44,6 @@ export type WorkerStatus = { internal: boolean; }>; version: string; + poolName: string; + queueName: string; }; diff --git a/packages/@n8n/api-types/src/worker-pools.ts b/packages/@n8n/api-types/src/worker-pools.ts new file mode 100644 index 00000000000..3c2b5dd6061 --- /dev/null +++ b/packages/@n8n/api-types/src/worker-pools.ts @@ -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[]; +}; diff --git a/packages/@n8n/backend-common/src/license-state.ts b/packages/@n8n/backend-common/src/license-state.ts index d9868423884..3379a2fe0ae 100644 --- a/packages/@n8n/backend-common/src/license-state.ts +++ b/packages/@n8n/backend-common/src/license-state.ts @@ -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'); } diff --git a/packages/@n8n/config/src/configs/__tests__/scaling-mode.config.test.ts b/packages/@n8n/config/src/configs/__tests__/scaling-mode.config.test.ts new file mode 100644 index 00000000000..fdb9c57ee82 --- /dev/null +++ b/packages/@n8n/config/src/configs/__tests__/scaling-mode.config.test.ts @@ -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(); + }); +}); diff --git a/packages/@n8n/config/src/configs/scaling-mode.config.ts b/packages/@n8n/config/src/configs/scaling-mode.config.ts index ac81cf280be..b48fb129f5a 100644 --- a/packages/@n8n/config/src/configs/scaling-mode.config.ts +++ b/packages/@n8n/config/src/configs/scaling-mode.config.ts @@ -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-` 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; } diff --git a/packages/@n8n/config/src/index.ts b/packages/@n8n/config/src/index.ts index 094b46a0de1..690e30219ef 100644 --- a/packages/@n8n/config/src/index.ts +++ b/packages/@n8n/config/src/index.ts @@ -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'; diff --git a/packages/@n8n/config/test/config.test.ts b/packages/@n8n/config/test/config.test.ts index 63e08ac7a14..f46db33c585 100644 --- a/packages/@n8n/config/test/config.test.ts +++ b/packages/@n8n/config/test/config.test.ts @@ -412,6 +412,10 @@ describe('GlobalConfig', () => { stalledInterval: 30_000, }, }, + workerPool: { + enabled: false, + name: '', + }, }, taskRunners: { mode: 'internal', diff --git a/packages/@n8n/constants/src/index.ts b/packages/@n8n/constants/src/index.ts index dfa9e930c80..b3c5c30dd72 100644 --- a/packages/@n8n/constants/src/index.ts +++ b/packages/@n8n/constants/src/index.ts @@ -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 = { diff --git a/packages/@n8n/db/src/entities/index.ts b/packages/@n8n/db/src/entities/index.ts index 28aa10cb5a0..4cce8df3138 100644 --- a/packages/@n8n/db/src/entities/index.ts +++ b/packages/@n8n/db/src/entities/index.ts @@ -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, }; diff --git a/packages/@n8n/db/src/entities/project-pool-settings.ee.ts b/packages/@n8n/db/src/entities/project-pool-settings.ee.ts new file mode 100644 index 00000000000..cd06c8cdff3 --- /dev/null +++ b/packages/@n8n/db/src/entities/project-pool-settings.ee.ts @@ -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; +} diff --git a/packages/@n8n/db/src/migrations/common/1788445119183-CreateProjectPoolSettings.ts b/packages/@n8n/db/src/migrations/common/1788445119183-CreateProjectPoolSettings.ts new file mode 100644 index 00000000000..84059d604d6 --- /dev/null +++ b/packages/@n8n/db/src/migrations/common/1788445119183-CreateProjectPoolSettings.ts @@ -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); + } +} diff --git a/packages/@n8n/db/src/repositories/index.ts b/packages/@n8n/db/src/repositories/index.ts index 22c145da702..b4a40eb148c 100644 --- a/packages/@n8n/db/src/repositories/index.ts +++ b/packages/@n8n/db/src/repositories/index.ts @@ -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'; diff --git a/packages/@n8n/db/src/repositories/project-pool-settings.repository.ee.ts b/packages/@n8n/db/src/repositories/project-pool-settings.repository.ee.ts new file mode 100644 index 00000000000..a94bb89a8bd --- /dev/null +++ b/packages/@n8n/db/src/repositories/project-pool-settings.repository.ee.ts @@ -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 { + constructor(dataSource: DataSource) { + super(ProjectPoolSettings, dataSource.manager); + } + + async getDefaultPool(projectId: string): Promise { + const row = await this.findOne({ where: { projectId }, select: ['defaultPool'] }); + return row?.defaultPool ?? null; + } + + async setDefaultPool(projectId: string, defaultPool: string | null): Promise { + await this.upsert({ projectId, defaultPool }, ['projectId']); + } +} diff --git a/packages/@n8n/permissions/src/__tests__/__snapshots__/scope-information.test.ts.snap b/packages/@n8n/permissions/src/__tests__/__snapshots__/scope-information.test.ts.snap index 43bdd65d303..a2fcf63a0ff 100644 --- a/packages/@n8n/permissions/src/__tests__/__snapshots__/scope-information.test.ts.snap +++ b/packages/@n8n/permissions/src/__tests__/__snapshots__/scope-information.test.ts.snap @@ -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", diff --git a/packages/@n8n/permissions/src/constants.ee.ts b/packages/@n8n/permissions/src/constants.ee.ts index 71fe2ffa21e..d2050161de8 100644 --- a/packages/@n8n/permissions/src/constants.ee.ts +++ b/packages/@n8n/permissions/src/constants.ee.ts @@ -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 diff --git a/packages/@n8n/permissions/src/roles/scopes/global-scopes.ee.ts b/packages/@n8n/permissions/src/roles/scopes/global-scopes.ee.ts index 615d842d242..fc136bcdfcc 100644 --- a/packages/@n8n/permissions/src/roles/scopes/global-scopes.ee.ts +++ b/packages/@n8n/permissions/src/roles/scopes/global-scopes.ee.ts @@ -54,6 +54,7 @@ export const GLOBAL_OWNER_SCOPES: Scope[] = [ 'logStreaming:manage', 'nodeTypePolicy:manage', 'orchestration:read', + 'orchestration:manage', 'saml:manage', 'securityAudit:generate', 'securitySettings:manage', diff --git a/packages/cli/src/__tests__/workflow-runner.test.ts b/packages/cli/src/__tests__/workflow-runner.test.ts index 4cca0cbb3fd..4a86d50f88e 100644 --- a/packages/cli/src/__tests__/workflow-runner.test.ts +++ b/packages/cli/src/__tests__/workflow-runner.test.ts @@ -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({ + 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(); diff --git a/packages/cli/src/commands/__tests__/worker.test.ts b/packages/cli/src/commands/__tests__/worker.test.ts index 230262ff472..404d87e6ccf 100644 --- a/packages/cli/src/commands/__tests__/worker.test.ts +++ b/packages/cli/src/commands/__tests__/worker.test.ts @@ -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); diff --git a/packages/cli/src/commands/worker.ts b/packages/cli/src/commands/worker.ts index 05369aaed7a..4340371b3a9 100644 --- a/packages/cli/src/commands/worker.ts +++ b/packages/cli/src/commands/worker.ts @@ -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> { 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> { 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) { diff --git a/packages/cli/src/controllers/__tests__/orchestration.controller.test.ts b/packages/cli/src/controllers/__tests__/orchestration.controller.test.ts new file mode 100644 index 00000000000..130b1915552 --- /dev/null +++ b/packages/cli/src/controllers/__tests__/orchestration.controller.test.ts @@ -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(); + const workerStatusService = mock(); + const controller = new OrchestrationController(license, workerStatusService); + + const req = mock({ 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(); + }); + }); +}); diff --git a/packages/cli/src/controllers/__tests__/project-pool-settings.controller.ee.test.ts b/packages/cli/src/controllers/__tests__/project-pool-settings.controller.ee.test.ts new file mode 100644 index 00000000000..abb1d5dfb6a --- /dev/null +++ b/packages/cli/src/controllers/__tests__/project-pool-settings.controller.ee.test.ts @@ -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(); + const controller = new ProjectPoolSettingsController(poolConfigService); + + const req = mock(); + const res = mock(); + 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'] }); + }); + }); +}); diff --git a/packages/cli/src/controllers/e2e.controller.ts b/packages/cli/src/controllers/e2e.controller.ts index e1b922e2e9a..b926a24d872 100644 --- a/packages/cli/src/controllers/e2e.controller.ts +++ b/packages/cli/src/controllers/e2e.controller.ts @@ -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 = { diff --git a/packages/cli/src/controllers/project-pool-settings.controller.ee.ts b/packages/cli/src/controllers/project-pool-settings.controller.ee.ts new file mode 100644 index 00000000000..8211e8922dd --- /dev/null +++ b/packages/cli/src/controllers/project-pool-settings.controller.ee.ts @@ -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 { + 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 { + return await this.poolConfigService.setProjectPoolSettings(projectId, dto); + } +} diff --git a/packages/cli/src/modules/instance-registry/__tests__/instance-registry.service.test.ts b/packages/cli/src/modules/instance-registry/__tests__/instance-registry.service.test.ts index 9db8599a886..01ade3a9bd3 100644 --- a/packages/cli/src/modules/instance-registry/__tests__/instance-registry.service.test.ts +++ b/packages/cli/src/modules/instance-registry/__tests__/instance-registry.service.test.ts @@ -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 = {}) => const makeExecutionsConfig = (mode: 'regular' | 'queue' = 'regular') => mock({ mode }); +const makeScalingModeConfig = (poolName: string = '', enabled: boolean = true) => { + const workerPool = Object.assign(new WorkerPoolConfig(), { name: poolName, enabled }); + return mock({ workerPool }); +}; + const makeLogger = () => { const logger = mock(); logger.scoped.mockReturnValue(logger); @@ -99,10 +105,13 @@ describe('InstanceRegistryService', () => { const createService = ( settingsOverrides: Partial = {}, 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(); + }); + }); }); diff --git a/packages/cli/src/modules/instance-registry/instance-registry.service.ts b/packages/cli/src/modules/instance-registry/instance-registry.service.ts index 757a1a26fbd..57c991f3daa 100644 --- a/packages/cli/src/modules/instance-registry/instance-registry.service.ts +++ b/packages/cli/src/modules/instance-registry/instance-registry.service.ts @@ -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 { diff --git a/packages/cli/src/scaling/__tests__/pool-config.service.ee.test.ts b/packages/cli/src/scaling/__tests__/pool-config.service.ee.test.ts new file mode 100644 index 00000000000..4d8ee199a99 --- /dev/null +++ b/packages/cli/src/scaling/__tests__/pool-config.service.ee.test.ts @@ -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(); + const cacheService = mock(); + const workerPoolsService = mock(); + const globalConfig = mock(); + const licenseState = mock(); + + 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); + }); + }); +}); diff --git a/packages/cli/src/scaling/__tests__/queue-name.test.ts b/packages/cli/src/scaling/__tests__/queue-name.test.ts new file mode 100644 index 00000000000..b0df8992558 --- /dev/null +++ b/packages/cli/src/scaling/__tests__/queue-name.test.ts @@ -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-" 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(''); + }); +}); diff --git a/packages/cli/src/scaling/__tests__/scaling.service.test.ts b/packages/cli/src/scaling/__tests__/scaling.service.test.ts index 2bdd5bb09f4..a094ff1bd98 100644 --- a/packages/cli/src/scaling/__tests__/scaling.service.test.ts +++ b/packages/cli/src/scaling/__tests__/scaling.service.test.ts @@ -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-" 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', () => { diff --git a/packages/cli/src/scaling/__tests__/worker-pools.service.ee.test.ts b/packages/cli/src/scaling/__tests__/worker-pools.service.ee.test.ts new file mode 100644 index 00000000000..cb0eba0db8e --- /dev/null +++ b/packages/cli/src/scaling/__tests__/worker-pools.service.ee.test.ts @@ -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 => ({ + 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(); + 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([]); + }); + }); +}); diff --git a/packages/cli/src/scaling/__tests__/worker-status.service.test.ts b/packages/cli/src/scaling/__tests__/worker-status.service.test.ts index 5b9c018015e..d5eb6708c2a 100644 --- a/packages/cli/src/scaling/__tests__/worker-status.service.test.ts +++ b/packages/cli/src/scaling/__tests__/worker-status.service.test.ts @@ -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: { diff --git a/packages/cli/src/scaling/constants.ts b/packages/cli/src/scaling/constants.ts index 20226b9eafd..7aa6cd57497 100644 --- a/packages/cli/src/scaling/constants.ts +++ b/packages/cli/src/scaling/constants.ts @@ -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. */ diff --git a/packages/cli/src/scaling/pool-config.service.ee.ts b/packages/cli/src/scaling/pool-config.service.ee.ts new file mode 100644 index 00000000000..f180d14eb8f --- /dev/null +++ b/packages/cli/src/scaling/pool-config.service.ee.ts @@ -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, + ): 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 { + const [defaultPool, availablePools] = await Promise.all([ + this.getProjectDefaultPool(projectId), + this.workerPoolsService.getAvailablePools(), + ]); + + return { defaultPool: defaultPool ?? null, availablePools }; + } + + async setProjectPoolSettings( + projectId: string, + patch: UpdateProjectPoolSettingsDto, + ): Promise { + 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 { + const cacheKey = this.cacheKey(projectId); + + const cached = await this.cacheService.get(cacheKey); + if (cached !== undefined) return cached || undefined; + + const pool = await this.projectPoolSettingsRepository.getDefaultPool(projectId); + void this.cacheService.set(cacheKey, pool ?? ''); + return pool ?? undefined; + } +} diff --git a/packages/cli/src/scaling/queue-name.ts b/packages/cli/src/scaling/queue-name.ts new file mode 100644 index 00000000000..e36f440389c --- /dev/null +++ b/packages/cli/src/scaling/queue-name.ts @@ -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}`; +} diff --git a/packages/cli/src/scaling/scaling.service.ts b/packages/cli/src/scaling/scaling.service.ts index 536398d6ee2..fafe46befda 100644 --- a/packages/cli/src/scaling/scaling.service.ts +++ b/packages/cli/src/scaling/scaling.service.ts @@ -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(); + + private initialQueueName = DEFAULT_QUEUE_NAME; + + private createBullQueue?: (name: string) => JobQueue; private jobResults = new Map(); @@ -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 { + 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`'); } diff --git a/packages/cli/src/scaling/scaling.types.ts b/packages/cli/src/scaling/scaling.types.ts index c9b1a93beae..fa4cdeb204e 100644 --- a/packages/cli/src/scaling/scaling.types.ts +++ b/packages/cli/src/scaling/scaling.types.ts @@ -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; diff --git a/packages/cli/src/scaling/worker-pools.service.ee.ts b/packages/cli/src/scaling/worker-pools.service.ee.ts new file mode 100644 index 00000000000..c26be814d01 --- /dev/null +++ b/packages/cli/src/scaling/worker-pools.service.ee.ts @@ -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 { + const instances = await this.instanceRegistryService.getAllInstances(); + + const poolNames = new Set(); + for (const instance of instances) { + if (instance.instanceType !== 'worker') continue; + if (instance.poolName) poolNames.add(instance.poolName); + } + + return Array.from(poolNames).sort(); + } +} diff --git a/packages/cli/src/scaling/worker-status.service.ee.ts b/packages/cli/src/scaling/worker-status.service.ee.ts index 4e745b0550f..4edf873795a 100644 --- a/packages/cli/src/scaling/worker-status.service.ee.ts +++ b/packages/cli/src/scaling/worker-status.service.ee.ts @@ -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), + ), }; } diff --git a/packages/cli/src/server.ts b/packages/cli/src/server.ts index 597349b10bd..f7a12cc0c28 100644 --- a/packages/cli/src/server.ts +++ b/packages/cli/src/server.ts @@ -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'; diff --git a/packages/cli/src/services/__tests__/frontend.service.test.ts b/packages/cli/src/services/__tests__/frontend.service.test.ts index a92c6d09dcb..021fbd7b014 100644 --- a/packages/cli/src/services/__tests__/frontend.service.test.ts +++ b/packages/cli/src/services/__tests__/frontend.service.test.ts @@ -81,6 +81,7 @@ describe('FrontendService', () => { }, aiAssistant: { baseUrl: '' }, aiGateway: { enabled: false }, + queue: { workerPool: { enabled: false } }, }); const instanceSettings = mock({ @@ -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, diff --git a/packages/cli/src/services/frontend.service.ts b/packages/cli/src/services/frontend.service.ts index d4aada48fe9..505b467a96b 100644 --- a/packages/cli/src/services/frontend.service.ts +++ b/packages/cli/src/services/frontend.service.ts @@ -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, diff --git a/packages/cli/src/services/ownership-transfer/ownership-transfer.manifest.json b/packages/cli/src/services/ownership-transfer/ownership-transfer.manifest.json index 797c87ed052..4897c3054fc 100644 --- a/packages/cli/src/services/ownership-transfer/ownership-transfer.manifest.json +++ b/packages/cli/src/services/ownership-transfer/ownership-transfer.manifest.json @@ -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", diff --git a/packages/cli/src/workflow-runner.ts b/packages/cli/src/workflow-runner.ts index 7580aaf7c42..e38f244790e 100644 --- a/packages/cli/src/workflow-runner.ts +++ b/packages/cli/src/workflow-runner.ts @@ -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 { - 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); diff --git a/packages/frontend/@n8n/frontend-test-utils/src/defaults.ts b/packages/frontend/@n8n/frontend-test-utils/src/defaults.ts index 951fbb3328f..103437f3450 100644 --- a/packages/frontend/@n8n/frontend-test-utils/src/defaults.ts +++ b/packages/frontend/@n8n/frontend-test-utils/src/defaults.ts @@ -180,6 +180,9 @@ export const defaultSettings: FrontendSettings = { folders: { enabled: false, }, + workerPools: { + enabled: false, + }, collaboration: { crdt: 'off', }, diff --git a/packages/frontend/@n8n/i18n/src/locales/en.json b/packages/frontend/@n8n/i18n/src/locales/en.json index 53a5eb77e1c..e839037d331 100644 --- a/packages/frontend/@n8n/i18n/src/locales/en.json +++ b/packages/frontend/@n8n/i18n/src/locales/en.json @@ -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.", diff --git a/packages/frontend/@n8n/rest-api-client/src/api/orchestration.ts b/packages/frontend/@n8n/rest-api-client/src/api/orchestration.ts index bf81fc45bc4..179b28582b2 100644 --- a/packages/frontend/@n8n/rest-api-client/src/api/orchestration.ts +++ b/packages/frontend/@n8n/rest-api-client/src/api/orchestration.ts @@ -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 => { await makeRestApiRequest(context, 'POST', GET_STATUS_ENDPOINT); }; + +export const getProjectPoolSettings = async ( + context: IRestApiContext, + projectId: string, +): Promise => { + return await makeRestApiRequest(context, 'GET', `/projects/${projectId}/pool-settings`); +}; + +export const updateProjectPoolSettings = async ( + context: IRestApiContext, + projectId: string, + dto: UpdateProjectPoolSettingsDto, +): Promise => { + return await makeRestApiRequest(context, 'PATCH', `/projects/${projectId}/pool-settings`, dto); +}; diff --git a/packages/frontend/@n8n/stores/src/settings.store.ts b/packages/frontend/@n8n/stores/src/settings.store.ts index e5521bfea24..2e7a9cb0dd5 100644 --- a/packages/frontend/@n8n/stores/src/settings.store.ts +++ b/packages/frontend/@n8n/stores/src/settings.store.ts @@ -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, diff --git a/packages/frontend/editor-ui/src/features/collaboration/projects/components/ProjectWorkerPoolsSection.vue b/packages/frontend/editor-ui/src/features/collaboration/projects/components/ProjectWorkerPoolsSection.vue new file mode 100644 index 00000000000..63295586258 --- /dev/null +++ b/packages/frontend/editor-ui/src/features/collaboration/projects/components/ProjectWorkerPoolsSection.vue @@ -0,0 +1,56 @@ + + + + + diff --git a/packages/frontend/editor-ui/src/features/collaboration/projects/projects.store.ts b/packages/frontend/editor-ui/src/features/collaboration/projects/projects.store.ts index 1df22b39f1d..e8c748925f4 100644 --- a/packages/frontend/editor-ui/src/features/collaboration/projects/projects.store.ts +++ b/packages/frontend/editor-ui/src/features/collaboration/projects/projects.store.ts @@ -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([]); const personalProject = ref(null); const currentProject = ref(null); + const currentProjectPoolSettings = ref(null); const projectsCount = ref({ personal: 0, team: 0, @@ -189,6 +200,25 @@ export const useProjectsStore = defineStore(STORES.PROJECTS, () => { } }; + const fetchProjectPoolSettings = async (projectId: string): Promise => { + currentProjectPoolSettings.value = null; + currentProjectPoolSettings.value = await getProjectPoolSettings( + rootStore.restApiContext, + projectId, + ); + }; + + const updateCurrentProjectPoolSettings = async ( + projectId: string, + dto: UpdateProjectPoolSettingsDto, + ): Promise => { + 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, diff --git a/packages/frontend/editor-ui/src/features/collaboration/projects/views/ProjectSettings.test.ts b/packages/frontend/editor-ui/src/features/collaboration/projects/views/ProjectSettings.test.ts index d950a7bb95c..5334e9cde44 100644 --- a/packages/frontend/editor-ui/src/features/collaboration/projects/views/ProjectSettings.test.ts +++ b/packages/frontend/editor-ui/src/features/collaboration/projects/views/ProjectSettings.test.ts @@ -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); diff --git a/packages/frontend/editor-ui/src/features/collaboration/projects/views/ProjectSettings.vue b/packages/frontend/editor-ui/src/features/collaboration/projects/views/ProjectSettings.vue index 785eb5eb8df..8a544b496dd 100644 --- a/packages/frontend/editor-ui/src/features/collaboration/projects/views/ProjectSettings.vue +++ b/packages/frontend/editor-ui/src/features/collaboration/projects/views/ProjectSettings.vue @@ -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({ const formData = ref< Pick & { customTelemetryTags: NonNullable; + 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> = []; + + 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" /> + +

{{ i18n.baseText('projects.settings.danger.title') }}

{{ diff --git a/packages/frontend/editor-ui/src/features/settings/orchestration.ee/components/WorkerCard.vue b/packages/frontend/editor-ui/src/features/settings/orchestration.ee/components/WorkerCard.vue index 2d7eaca5376..5217efde345 100644 --- a/packages/frontend/editor-ui/src/features/settings/orchestration.ee/components/WorkerCard.vue +++ b/packages/frontend/editor-ui/src/features/settings/orchestration.ee/components/WorkerCard.vue @@ -69,7 +69,11 @@ onBeforeUnmount(() => { :class="stale ? [$style.cardHeading, $style.stale] : [$style.cardHeading]" data-test-id="worker-card-name" > - Name: {{ worker.senderId }} ({{ worker.hostname }})
+ 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 }} +
Average Load: {{ averageWorkerLoadFromLoadsAsString(worker.loadAvg ?? [0]) }} | Free memory: {{ memAsGb(worker.process.memory.available) }}GB / {{