From 5be3ebe127e222e7cf8e19b6a5ec117d93169154 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mike=20Repe=C4=87?= Date: Thu, 25 Jun 2026 08:57:16 +0200 Subject: [PATCH] refactor(core): prepare Worker Pools for rollout behind a flag (#32974) --- 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 ++++ packages/@n8n/api-types/src/dto/index.ts | 1 - .../update-project-pool-settings.dto.ts | 11 +- .../update-worker-pool-assignment.dto.ts | 9 - .../@n8n/api-types/src/frontend-settings.ts | 3 + .../api-types/src/instance-registry-types.ts | 1 - packages/@n8n/api-types/src/worker-pools.ts | 17 +- .../__tests__/scaling-mode.config.test.ts | 16 +- .../config/src/configs/scaling-mode.config.ts | 10 +- packages/@n8n/config/src/index.ts | 2 +- packages/@n8n/config/test/config.test.ts | 1 + .../db/src/entities/project-pool-settings.ts | 18 +- ...784000000038-CreateProjectPoolSettings.ts} | 7 +- .../db/src/migrations/postgresdb/index.ts | 4 +- .../@n8n/db/src/migrations/sqlite/index.ts | 4 +- .../project-pool-settings.repository.ts | 57 +--- .../src/__tests__/workflow-helpers.test.ts | 23 -- .../cli/src/__tests__/workflow-runner.test.ts | 44 --- packages/cli/src/commands/worker.ts | 6 +- .../orchestration.controller.test.ts | 46 +-- .../project-pool-settings.controller.test.ts | 27 +- .../controllers/orchestration.controller.ts | 28 +- .../instance-registry.controller.test.ts | 7 +- .../instance-registry.service.test.ts | 24 +- .../instance-registry.controller.ts | 7 +- .../instance-registry.service.ts | 5 +- .../spec/schemas/workflowSettings.yml | 12 - .../__tests__/pool-config.service.test.ts | 317 ++++++------------ .../src/scaling/__tests__/queue-name.test.ts | 14 +- .../scaling/__tests__/scaling.service.test.ts | 16 +- .../__tests__/worker-pools.service.test.ts | 19 +- .../cli/src/scaling/execution-category.ts | 20 -- .../cli/src/scaling/pool-config.service.ts | 209 +++--------- packages/cli/src/scaling/queue-name.ts | 6 + packages/cli/src/scaling/scaling.service.ts | 4 +- packages/cli/src/scaling/scaling.types.ts | 4 - .../cli/src/scaling/worker-pools.service.ts | 18 +- .../src/scaling/worker-status.service.ee.ts | 6 +- .../__tests__/frontend.service.test.ts | 1 + packages/cli/src/services/frontend.service.ts | 3 + packages/cli/src/workflow-helpers.ts | 12 - packages/cli/src/workflow-runner.ts | 60 +--- .../frontend/@n8n/i18n/src/locales/en.json | 45 +-- .../rest-api-client/src/api/orchestration.ts | 21 +- packages/frontend/editor-ui/src/Interface.ts | 3 - .../editor-ui/src/__tests__/defaults.ts | 3 + .../WorkflowSettings/WorkflowSettings.vue | 225 ------------- .../src/app/stores/settings.store.ts | 10 + .../components/ProjectWorkerPoolsSection.vue | 229 ++----------- .../projects/views/ProjectSettings.vue | 98 ++---- .../components/WorkerList.vue | 6 - .../components/WorkerPoolsCard.vue | 232 ------------- .../orchestration.ee/orchestration.store.ts | 42 +-- packages/workflow/src/interfaces.ts | 4 - 59 files changed, 530 insertions(+), 1642 deletions(-) create mode 100644 docs/generated/postgres-schema/public.project_pool_settings.md create mode 100644 docs/generated/sqlite-schema/project_pool_settings.md delete mode 100644 packages/@n8n/api-types/src/dto/worker-pools/update-worker-pool-assignment.dto.ts rename packages/@n8n/db/src/migrations/common/{1785000000000-CreateProjectPoolSettings.ts => 1784000000038-CreateProjectPoolSettings.ts} (69%) delete mode 100644 packages/cli/src/scaling/execution-category.ts delete mode 100644 packages/frontend/editor-ui/src/features/settings/orchestration.ee/components/WorkerPoolsCard.vue diff --git a/docs/generated/postgres-schema/README.md b/docs/generated/postgres-schema/README.md index 8c1eb78c974..6853241dc6b 100644 --- a/docs/generated/postgres-schema/README.md +++ b/docs/generated/postgres-schema/README.md @@ -86,6 +86,7 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand. | [public.oauth_user_consents](public.oauth_user_consents.md) | 4 | | 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 | @@ -251,6 +252,7 @@ erDiagram "public.oauth_user_consents" }o--|| "public.oauth_clients" : "FOREIGN KEY (#quot;clientId#quot;) REFERENCES oauth_clients(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)" @@ -1048,6 +1050,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 bb1ccd7615f..f6a83e420b8 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.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.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.variables](public.variables.md) | | | +| id | varchar(36) | | false | [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.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.variables](public.variables.md) | | | | name | varchar(255) | | false | | | | | type | varchar(36) | | false | | | | | updatedAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | | @@ -45,6 +45,7 @@ erDiagram "public.folder" }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" @@ -143,6 +144,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 c638fc73745..6065aa2f908 100644 --- a/docs/generated/sqlite-schema/README.md +++ b/docs/generated/sqlite-schema/README.md @@ -86,6 +86,7 @@ Auto-generated from the SQLite migrations in @n8n/db. Do not edit by hand. | [oauth_user_consents](oauth_user_consents.md) | 4 | | 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 | @@ -235,6 +236,7 @@ erDiagram "oauth_user_consents" }o--|| "oauth_clients" : "FOREIGN KEY (clientId) REFERENCES oauth_clients (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" @@ -1036,6 +1038,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 b481e0d33ca..d7086d1b040 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 | [agent_execution_threads](agent_execution_threads.md) [agents](agents.md) [data_table](data_table.md) [folder](folder.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) [variables](variables.md) | | | +| id | varchar(36) | | false | [agent_execution_threads](agent_execution_threads.md) [agents](agents.md) [data_table](data_table.md) [folder](folder.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) [variables](variables.md) | | | | name | varchar(255) | | false | | | | | type | varchar(36) | | false | | | | | updatedAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | | @@ -51,6 +51,7 @@ erDiagram "folder" }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" @@ -149,6 +150,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/dto/index.ts b/packages/@n8n/api-types/src/dto/index.ts index 10eeab22a79..29e0a549fa6 100644 --- a/packages/@n8n/api-types/src/dto/index.ts +++ b/packages/@n8n/api-types/src/dto/index.ts @@ -239,7 +239,6 @@ export { UpdateSecuritySettingsDto, } from './security-settings/security-settings.dto'; -export { UpdateWorkerPoolAssignmentDto } from './worker-pools/update-worker-pool-assignment.dto'; export { UpdateProjectPoolSettingsDto } from './worker-pools/update-project-pool-settings.dto'; export { WorkflowHistoryVersionsByIdsDto } from './workflow-history/workflow-history-versions-by-ids.dto'; 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 index 9fa2ad1ad2c..3050f38f64a 100644 --- 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 @@ -2,16 +2,9 @@ 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})?$/); -const nonEmptyPoolNameSchema = z.string().regex(/^[a-z0-9][a-z0-9-]{0,62}$/); export class UpdateProjectPoolSettingsDto extends Z.class({ - assignment: z - .object({ - production: poolNameOrEmptySchema.optional(), - manual: poolNameOrEmptySchema.optional(), - evaluation: poolNameOrEmptySchema.optional(), - }) - .optional(), - allowedPools: z.array(nonEmptyPoolNameSchema).optional(), + defaultPool: poolNameOrEmptySchema.nullable(), }) {} diff --git a/packages/@n8n/api-types/src/dto/worker-pools/update-worker-pool-assignment.dto.ts b/packages/@n8n/api-types/src/dto/worker-pools/update-worker-pool-assignment.dto.ts deleted file mode 100644 index 6fdd83e80f9..00000000000 --- a/packages/@n8n/api-types/src/dto/worker-pools/update-worker-pool-assignment.dto.ts +++ /dev/null @@ -1,9 +0,0 @@ -import { z } from 'zod'; - -import { Z } from '../../zod-class'; - -export class UpdateWorkerPoolAssignmentDto extends Z.class({ - production: z.string().optional(), - manual: z.string().optional(), - evaluation: z.string().optional(), -}) {} diff --git a/packages/@n8n/api-types/src/frontend-settings.ts b/packages/@n8n/api-types/src/frontend-settings.ts index a87a2ad1552..e271bb0ae6c 100644 --- a/packages/@n8n/api-types/src/frontend-settings.ts +++ b/packages/@n8n/api-types/src/frontend-settings.ts @@ -216,6 +216,9 @@ export interface FrontendSettings { folders: { enabled: boolean; }; + workerPools: { + enabled: boolean; + }; collaboration: { crdt: 'off' | 'local' | 'server'; }; diff --git a/packages/@n8n/api-types/src/instance-registry-types.ts b/packages/@n8n/api-types/src/instance-registry-types.ts index 41c2d1ddd88..9eeab861ddc 100644 --- a/packages/@n8n/api-types/src/instance-registry-types.ts +++ b/packages/@n8n/api-types/src/instance-registry-types.ts @@ -70,7 +70,6 @@ export type ClusterCheckSummary = z.infer; const ClusterInfoResponseSchema = z.object({ instances: z.array(instanceRegistrationSchema), checks: ClusterCheckSummarySchema, - poolAssignment: z.record(z.enum(['production', 'manual', 'evaluation']), z.string()).optional(), }); // REST API response type diff --git a/packages/@n8n/api-types/src/worker-pools.ts b/packages/@n8n/api-types/src/worker-pools.ts index f850d000c78..3c2b5dd6061 100644 --- a/packages/@n8n/api-types/src/worker-pools.ts +++ b/packages/@n8n/api-types/src/worker-pools.ts @@ -1,17 +1,6 @@ -export type ExecutionCategory = 'production' | 'manual' | 'evaluation'; - -export type PoolAssignment = Partial>; - -export type WorkerPoolsResponse = { - pools: string[]; - assignment: PoolAssignment; -}; - export type ProjectPoolSettingsResponse = { - assignment: PoolAssignment; - allowedPools: string[]; - /** Pool names registered by workers in the cluster (read-only context). */ + /** 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[]; - /** Instance-level pool assignment, shown in the UI as the "Inherit from instance" fallback (read-only context). */ - instanceDefaults: PoolAssignment; }; 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 index 25f553953a8..fdb9c57ee82 100644 --- a/packages/@n8n/config/src/configs/__tests__/scaling-mode.config.test.ts +++ b/packages/@n8n/config/src/configs/__tests__/scaling-mode.config.test.ts @@ -7,7 +7,7 @@ describe('ScalingModeConfig.workerPool', () => { beforeEach(() => { Container.reset(); - jest.clearAllMocks(); + vi.clearAllMocks(); }); afterEach(() => { @@ -20,6 +20,18 @@ describe('ScalingModeConfig.workerPool', () => { 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); @@ -41,7 +53,7 @@ describe('ScalingModeConfig.workerPool', () => { ['dot', 'g.pu'], ['too long', 'a'.repeat(64)], ])('falls back to default and warns on invalid label (%s)', (_label, value) => { - const consoleWarnSpy = jest.spyOn(console, 'warn').mockImplementation(); + const consoleWarnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); process.env = { N8N_WORKER_POOL_NAME: value }; const config = Container.get(ScalingModeConfig); diff --git a/packages/@n8n/config/src/configs/scaling-mode.config.ts b/packages/@n8n/config/src/configs/scaling-mode.config.ts index 0b9f8b84fd9..956658eb333 100644 --- a/packages/@n8n/config/src/configs/scaling-mode.config.ts +++ b/packages/@n8n/config/src/configs/scaling-mode.config.ts @@ -11,7 +11,15 @@ const workerPoolNameSchema = z ); @Config -class WorkerPoolConfig { +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. diff --git a/packages/@n8n/config/src/index.ts b/packages/@n8n/config/src/index.ts index 1dd669df4ce..7bb05f7773d 100644 --- a/packages/@n8n/config/src/index.ts +++ b/packages/@n8n/config/src/index.ts @@ -65,7 +65,7 @@ export { } from './configs/ssrf-protection.config'; export { EngineConfig } from './configs/engine.config'; export { ExecutionsConfig } from './configs/executions.config'; -export { ScalingModeConfig } from './configs/scaling-mode.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 f627b1f1d45..eb2eb22c179 100644 --- a/packages/@n8n/config/test/config.test.ts +++ b/packages/@n8n/config/test/config.test.ts @@ -369,6 +369,7 @@ describe('GlobalConfig', () => { }, }, workerPool: { + enabled: false, name: '', }, }, diff --git a/packages/@n8n/db/src/entities/project-pool-settings.ts b/packages/@n8n/db/src/entities/project-pool-settings.ts index ee8ab134dac..cd06c8cdff3 100644 --- a/packages/@n8n/db/src/entities/project-pool-settings.ts +++ b/packages/@n8n/db/src/entities/project-pool-settings.ts @@ -1,6 +1,6 @@ import { Column, Entity, JoinColumn, OneToOne, PrimaryColumn } from '@n8n/typeorm'; -import { JsonColumn, WithTimestamps } from './abstract-entity'; +import { WithTimestamps } from './abstract-entity'; import { Project } from './project'; @Entity() @@ -12,19 +12,7 @@ export class ProjectPoolSettings extends WithTimestamps { @JoinColumn({ name: 'projectId' }) project: Project; - /** Pool for production executions (webhook/trigger/chat). `null` means inherit instance default. */ + /** Pool all executions in this project route to. `null` means the system default queue. */ @Column({ type: 'varchar', length: 63, nullable: true }) - productionPool: string | null; - - /** Pool for manual/retry executions. `null` means inherit instance default. */ - @Column({ type: 'varchar', length: 63, nullable: true }) - manualPool: string | null; - - /** Pool for evaluation executions. `null` means inherit instance default. */ - @Column({ type: 'varchar', length: 63, nullable: true }) - evaluationPool: string | null; - - /** Pool labels that workflow-level overrides may pick from. Empty array = no overrides allowed. */ - @JsonColumn({ default: () => "'[]'" }) - allowedPools: string[]; + defaultPool: string | null; } diff --git a/packages/@n8n/db/src/migrations/common/1785000000000-CreateProjectPoolSettings.ts b/packages/@n8n/db/src/migrations/common/1784000000038-CreateProjectPoolSettings.ts similarity index 69% rename from packages/@n8n/db/src/migrations/common/1785000000000-CreateProjectPoolSettings.ts rename to packages/@n8n/db/src/migrations/common/1784000000038-CreateProjectPoolSettings.ts index 073bea0c6c7..7a853b42720 100644 --- a/packages/@n8n/db/src/migrations/common/1785000000000-CreateProjectPoolSettings.ts +++ b/packages/@n8n/db/src/migrations/common/1784000000038-CreateProjectPoolSettings.ts @@ -2,15 +2,12 @@ import type { MigrationContext, ReversibleMigration } from '../migration-types'; const tableName = 'project_pool_settings'; -export class CreateProjectPoolSettings1785000000000 implements ReversibleMigration { +export class CreateProjectPoolSettings1784000000038 implements ReversibleMigration { async up({ schemaBuilder: { createTable, column } }: MigrationContext) { await createTable(tableName) .withColumns( column('projectId').varchar(36).primary.notNull, - column('productionPool').varchar(63), - column('manualPool').varchar(63), - column('evaluationPool').varchar(63), - column('allowedPools').json.notNull.default("'[]'"), + column('defaultPool').varchar(63), ) .withForeignKey('projectId', { tableName: 'project', diff --git a/packages/@n8n/db/src/migrations/postgresdb/index.ts b/packages/@n8n/db/src/migrations/postgresdb/index.ts index e9012c83d02..2c67de119d3 100644 --- a/packages/@n8n/db/src/migrations/postgresdb/index.ts +++ b/packages/@n8n/db/src/migrations/postgresdb/index.ts @@ -211,7 +211,7 @@ import { AllowAzureStoredAt1784000000034 } from '../common/1784000000034-AllowAz import { AddUniqueAgentFileNames1784000000035 } from '../common/1784000000035-AddUniqueAgentFileNames'; import { CreateInstanceAiThreadGrantTable1784000000036 } from '../common/1784000000036-CreateInstanceAiThreadGrantTable'; import { DropAgentDescriptionFromAgents1784000000037 } from '../common/1784000000037-DropAgentDescriptionFromAgents'; -import { CreateProjectPoolSettings1785000000000 } from '../common/1785000000000-CreateProjectPoolSettings'; +import { CreateProjectPoolSettings1784000000038 } from '../common/1784000000038-CreateProjectPoolSettings'; import type { Migration } from '../migration-types'; export const postgresMigrations: Migration[] = [ @@ -428,5 +428,5 @@ export const postgresMigrations: Migration[] = [ AddUniqueAgentFileNames1784000000035, CreateInstanceAiThreadGrantTable1784000000036, DropAgentDescriptionFromAgents1784000000037, - CreateProjectPoolSettings1785000000000, + CreateProjectPoolSettings1784000000038, ]; diff --git a/packages/@n8n/db/src/migrations/sqlite/index.ts b/packages/@n8n/db/src/migrations/sqlite/index.ts index 2774a857671..c5cb2628017 100644 --- a/packages/@n8n/db/src/migrations/sqlite/index.ts +++ b/packages/@n8n/db/src/migrations/sqlite/index.ts @@ -197,7 +197,7 @@ import { CreateInstanceAiMcpRegistryConnectionTable1784000000023 } from '../comm import { AddResourceToOAuthAuthorizationCodes1784000000024 } from '../common/1784000000024-AddResourceToOAuthAuthorizationCodes'; import { CreateWorkflowPublicationOutboxTable1784000000027 } from '../common/1784000000027-CreateWorkflowPublicationOutboxTable'; import { CreateAgentChatSubscriptions1784000000030 } from '../common/1784000000030-CreateAgentChatSubscriptions'; -import { CreateProjectPoolSettings1785000000000 } from '../common/1785000000000-CreateProjectPoolSettings'; +import { CreateProjectPoolSettings1784000000038 } from '../common/1784000000038-CreateProjectPoolSettings'; import type { Migration } from '../migration-types'; import { CreateAgentTaskDefinitionTable1784000000021 } from './1784000000021-CreateAgentTaskDefinitionTable'; import { MigrateRedactionEnforcementToFloor1784000000025 } from '../common/1784000000025-MigrateRedactionEnforcementToFloor'; @@ -412,7 +412,7 @@ const sqliteMigrations: Migration[] = [ AddUniqueAgentFileNames1784000000035, CreateInstanceAiThreadGrantTable1784000000036, DropAgentDescriptionFromAgents1784000000037, - CreateProjectPoolSettings1785000000000, + CreateProjectPoolSettings1784000000038, ]; export { sqliteMigrations }; diff --git a/packages/@n8n/db/src/repositories/project-pool-settings.repository.ts b/packages/@n8n/db/src/repositories/project-pool-settings.repository.ts index 8310a03eba1..a94bb89a8bd 100644 --- a/packages/@n8n/db/src/repositories/project-pool-settings.repository.ts +++ b/packages/@n8n/db/src/repositories/project-pool-settings.repository.ts @@ -1,67 +1,20 @@ -import type { PoolAssignment } from '@n8n/api-types'; import { Service } from '@n8n/di'; import { DataSource, Repository } from '@n8n/typeorm'; import { ProjectPoolSettings } from '../entities'; -const CATEGORY_COLUMN = { - production: 'productionPool', - manual: 'manualPool', - evaluation: 'evaluationPool', -} as const satisfies Record; - @Service() export class ProjectPoolSettingsRepository extends Repository { constructor(dataSource: DataSource) { super(ProjectPoolSettings, dataSource.manager); } - async getPoolForCategory( - projectId: string, - category: 'production' | 'manual' | 'evaluation', - ): Promise { - const column = CATEGORY_COLUMN[category]; - const row = await this.findOne({ - where: { projectId }, - select: [column], - }); - return row?.[column] ?? undefined; + async getDefaultPool(projectId: string): Promise { + const row = await this.findOne({ where: { projectId }, select: ['defaultPool'] }); + return row?.defaultPool ?? null; } - async getSettings( - projectId: string, - ): Promise<{ assignment: PoolAssignment; allowedPools: string[] } | undefined> { - const row = await this.findOneBy({ projectId }); - if (!row) return undefined; - - const assignment: PoolAssignment = {}; - if (row.productionPool) assignment.production = row.productionPool; - if (row.manualPool) assignment.manual = row.manualPool; - if (row.evaluationPool) assignment.evaluation = row.evaluationPool; - - return { assignment, allowedPools: row.allowedPools }; - } - - async setSettings( - projectId: string, - next: { assignment: PoolAssignment; allowedPools: string[] }, - ): Promise { - const existing = await this.findOneBy({ projectId }); - const row = - existing ?? - this.create({ - projectId, - productionPool: null, - manualPool: null, - evaluationPool: null, - allowedPools: [], - }); - - row.productionPool = next.assignment.production ?? null; - row.manualPool = next.assignment.manual ?? null; - row.evaluationPool = next.assignment.evaluation ?? null; - row.allowedPools = next.allowedPools; - - await this.save(row); + async setDefaultPool(projectId: string, defaultPool: string | null): Promise { + await this.upsert({ projectId, defaultPool }, ['projectId']); } } diff --git a/packages/cli/src/__tests__/workflow-helpers.test.ts b/packages/cli/src/__tests__/workflow-helpers.test.ts index 338eda5a957..eb9de5e5904 100644 --- a/packages/cli/src/__tests__/workflow-helpers.test.ts +++ b/packages/cli/src/__tests__/workflow-helpers.test.ts @@ -399,29 +399,6 @@ describe('removeDefaultValues', () => { }); }); - it('should remove worker pool overrides when cleared (empty string)', () => { - const settings: IWorkflowSettings = { - workerPoolOverrideProduction: '', - workerPoolOverrideManual: '', - workerPoolOverrideEvaluation: '', - timezone: 'America/New_York', - }; - const result = removeDefaultValues(settings, DEFAULT_EXECUTION_TIMEOUT); - expect(result).toEqual({ timezone: 'America/New_York' }); - }); - - it('should keep worker pool overrides when set to a pool name', () => { - const settings: IWorkflowSettings = { - workerPoolOverrideProduction: 'gpu', - workerPoolOverrideManual: 'cpu', - }; - const result = removeDefaultValues(settings, DEFAULT_EXECUTION_TIMEOUT); - expect(result).toEqual({ - workerPoolOverrideProduction: 'gpu', - workerPoolOverrideManual: 'cpu', - }); - }); - it('should not mutate the original settings object', () => { const settings = { errorWorkflow: DEFAULT, diff --git a/packages/cli/src/__tests__/workflow-runner.test.ts b/packages/cli/src/__tests__/workflow-runner.test.ts index 6fba603ed0e..6d5dcb16314 100644 --- a/packages/cli/src/__tests__/workflow-runner.test.ts +++ b/packages/cli/src/__tests__/workflow-runner.test.ts @@ -413,50 +413,6 @@ describe('enqueueExecution', () => { expect.any(Object), ); }); - - describe('job priority', () => { - it.each([ - { name: 'manual/test (realtime)', realtime: true, settingsPriority: 'high', expected: 50 }, - { name: 'high-priority production', realtime: false, settingsPriority: 'high', expected: 25 }, - { - name: 'default production', - realtime: false, - settingsPriority: 'default', - expected: 100, - }, - { - name: 'production without explicit priority', - realtime: false, - settingsPriority: undefined, - expected: 100, - }, - ] as const)( - 'enqueues $name with priority $expected', - async ({ realtime, settingsPriority, expected }) => { - const activeExecutions = Container.get(ActiveExecutions); - jest.spyOn(activeExecutions, 'attachWorkflowExecution').mockReturnValue(); - jest.spyOn(runner, 'processError').mockResolvedValue(); - const data = mock({ - workflowData: { - nodes: [], - settings: settingsPriority ? { priority: settingsPriority } : {}, - }, - executionData: undefined, - }); - addJob.mockRejectedValueOnce(new Error('stop for test purposes')); - - await expect( - // @ts-expect-error Private method - runner.enqueueExecution('1', 'workflow-xyz', data, false, realtime), - ).rejects.toThrow(); - - expect(addJob).toHaveBeenCalledWith( - expect.any(Object), - expect.objectContaining({ priority: expected }), - ); - }, - ); - }); }); describe('workflow timeout with startedAt', () => { diff --git a/packages/cli/src/commands/worker.ts b/packages/cli/src/commands/worker.ts index fa6bfd6f382..2e01a6db5c7 100644 --- a/packages/cli/src/commands/worker.ts +++ b/packages/cli/src/commands/worker.ts @@ -13,13 +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 } from '@/scaling/queue-name'; +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'; @@ -206,7 +206,7 @@ export class Worker extends BaseCommand> { workerServer?.markAsReady(); - const poolName = this.globalConfig.queue.workerPool.name; + const poolName = resolveWorkerPoolName(this.globalConfig.queue.workerPool); const queueName = resolveQueueName('worker', poolName); this.logger.info('\nn8n worker is now ready'); diff --git a/packages/cli/src/controllers/__tests__/orchestration.controller.test.ts b/packages/cli/src/controllers/__tests__/orchestration.controller.test.ts index 6f9d3cf151a..cc62f8272d4 100644 --- a/packages/cli/src/controllers/__tests__/orchestration.controller.test.ts +++ b/packages/cli/src/controllers/__tests__/orchestration.controller.test.ts @@ -1,60 +1,36 @@ -import { UpdateWorkerPoolAssignmentDto } from '@n8n/api-types'; import type { AuthenticatedRequest } from '@n8n/db'; -import type { Response } from 'express'; import { mock } from 'jest-mock-extended'; import { OrchestrationController } from '@/controllers/orchestration.controller'; import type { License } from '@/license'; -import type { PoolConfigService } from '@/scaling/pool-config.service'; -import type { WorkerPoolsService } from '@/scaling/worker-pools.service'; import type { WorkerStatusService } from '@/scaling/worker-status.service.ee'; describe('OrchestrationController', () => { const license = mock(); const workerStatusService = mock(); - const workerPoolsService = mock(); - const poolConfigService = mock(); - const controller = new OrchestrationController( - license, - workerStatusService, - workerPoolsService, - poolConfigService, - ); + const controller = new OrchestrationController(license, workerStatusService); - const req = mock(); - const res = mock(); + const req = mock({ user: { id: 'user-1' } }); beforeEach(() => { jest.clearAllMocks(); }); - describe('getWorkerPools', () => { - it('should return pools and assignment from WorkerPoolsService', async () => { - workerPoolsService.getPoolsState.mockResolvedValue({ - pools: ['default'], - assignment: { production: 'gpu' }, - }); + describe('getWorkersStatusAll', () => { + it('requests worker status when licensed', async () => { + license.isWorkerViewLicensed.mockReturnValue(true); - const result = await controller.getWorkerPools(); + await controller.getWorkersStatusAll(req); - expect(result).toEqual({ - pools: ['default'], - assignment: { production: 'gpu' }, - }); - expect(workerPoolsService.getPoolsState).toHaveBeenCalledTimes(1); + expect(workerStatusService.requestWorkerStatus).toHaveBeenCalledWith('user-1'); }); - }); - describe('updateWorkerPoolAssignment', () => { - it('should delegate to PoolConfigService.setPoolAssignment with the DTO', async () => { - const dto = new UpdateWorkerPoolAssignmentDto({ production: 'gpu' }); + it('does nothing when not licensed', async () => { + license.isWorkerViewLicensed.mockReturnValue(false); - poolConfigService.setPoolAssignment.mockResolvedValue({ production: 'gpu' }); + await controller.getWorkersStatusAll(req); - const result = await controller.updateWorkerPoolAssignment(req, res, dto); - - expect(poolConfigService.setPoolAssignment).toHaveBeenCalledWith(dto); - expect(result).toEqual({ production: 'gpu' }); + expect(workerStatusService.requestWorkerStatus).not.toHaveBeenCalled(); }); }); }); diff --git a/packages/cli/src/controllers/__tests__/project-pool-settings.controller.test.ts b/packages/cli/src/controllers/__tests__/project-pool-settings.controller.test.ts index b47c374be02..ad3282b81c1 100644 --- a/packages/cli/src/controllers/__tests__/project-pool-settings.controller.test.ts +++ b/packages/cli/src/controllers/__tests__/project-pool-settings.controller.test.ts @@ -20,46 +20,29 @@ describe('ProjectPoolSettingsController', () => { describe('getPoolSettings', () => { it('delegates to PoolConfigService.getProjectPoolSettings', async () => { poolConfigService.getProjectPoolSettings.mockResolvedValue({ - assignment: { production: 'gpu' }, - allowedPools: ['gpu', 'cpu'], + defaultPool: 'gpu', availablePools: ['gpu', 'cpu'], - instanceDefaults: {}, }); const result = await controller.getPoolSettings(req, res, projectId); expect(poolConfigService.getProjectPoolSettings).toHaveBeenCalledWith(projectId); - expect(result).toEqual({ - assignment: { production: 'gpu' }, - allowedPools: ['gpu', 'cpu'], - availablePools: ['gpu', 'cpu'], - instanceDefaults: {}, - }); + expect(result).toEqual({ defaultPool: 'gpu', availablePools: ['gpu', 'cpu'] }); }); }); describe('updatePoolSettings', () => { it('delegates to PoolConfigService.setProjectPoolSettings with the DTO', async () => { - const dto = new UpdateProjectPoolSettingsDto({ - assignment: { production: 'gpu' }, - allowedPools: ['gpu'], - }); + const dto = new UpdateProjectPoolSettingsDto({ defaultPool: 'gpu' }); poolConfigService.setProjectPoolSettings.mockResolvedValue({ - assignment: { production: 'gpu' }, - allowedPools: ['gpu'], + defaultPool: 'gpu', availablePools: ['gpu'], - instanceDefaults: {}, }); const result = await controller.updatePoolSettings(req, res, dto, projectId); expect(poolConfigService.setProjectPoolSettings).toHaveBeenCalledWith(projectId, dto); - expect(result).toEqual({ - assignment: { production: 'gpu' }, - allowedPools: ['gpu'], - availablePools: ['gpu'], - instanceDefaults: {}, - }); + expect(result).toEqual({ defaultPool: 'gpu', availablePools: ['gpu'] }); }); }); }); diff --git a/packages/cli/src/controllers/orchestration.controller.ts b/packages/cli/src/controllers/orchestration.controller.ts index b2ad06b6137..bd03b57c8e3 100644 --- a/packages/cli/src/controllers/orchestration.controller.ts +++ b/packages/cli/src/controllers/orchestration.controller.ts @@ -1,13 +1,7 @@ -import type { PoolAssignment, WorkerPoolsResponse } from '@n8n/api-types'; -import { UpdateWorkerPoolAssignmentDto } from '@n8n/api-types'; -import { LICENSE_FEATURES } from '@n8n/constants'; import type { AuthenticatedRequest } from '@n8n/db'; -import { Body, Get, GlobalScope, Licensed, Patch, Post, RestController } from '@n8n/decorators'; -import type { Response } from 'express'; +import { GlobalScope, Post, RestController } from '@n8n/decorators'; import { License } from '@/license'; -import { PoolConfigService } from '@/scaling/pool-config.service'; -import { WorkerPoolsService } from '@/scaling/worker-pools.service'; import { WorkerStatusService } from '@/scaling/worker-status.service.ee'; @RestController('/orchestration') @@ -15,8 +9,6 @@ export class OrchestrationController { constructor( private readonly licenseService: License, private readonly workerStatusService: WorkerStatusService, - private readonly workerPoolsService: WorkerPoolsService, - private readonly poolConfigService: PoolConfigService, ) {} /** @@ -30,22 +22,4 @@ export class OrchestrationController { return await this.workerStatusService.requestWorkerStatus(req.user.id); } - - @Licensed(LICENSE_FEATURES.WORKER_VIEW) - @GlobalScope('orchestration:read') - @Get('/worker/pools') - async getWorkerPools(): Promise { - return await this.workerPoolsService.getPoolsState(); - } - - @Licensed(LICENSE_FEATURES.WORKER_VIEW) - @GlobalScope('orchestration:manage') - @Patch('/worker/pools/assignment') - async updateWorkerPoolAssignment( - _req: AuthenticatedRequest, - _res: Response, - @Body dto: UpdateWorkerPoolAssignmentDto, - ): Promise { - return await this.poolConfigService.setPoolAssignment(dto); - } } diff --git a/packages/cli/src/modules/instance-registry/__tests__/instance-registry.controller.test.ts b/packages/cli/src/modules/instance-registry/__tests__/instance-registry.controller.test.ts index 6304ba74485..2a38cf20a3a 100644 --- a/packages/cli/src/modules/instance-registry/__tests__/instance-registry.controller.test.ts +++ b/packages/cli/src/modules/instance-registry/__tests__/instance-registry.controller.test.ts @@ -1,8 +1,6 @@ import type { InstanceRegistration } from '@n8n/api-types'; import { mock } from 'jest-mock-extended'; -import type { PoolConfigService } from '@/scaling/pool-config.service'; - import type { CheckService } from '../checks/check.service'; import { InstanceRegistryController } from '../instance-registry.controller'; import type { InstanceRegistryService } from '../instance-registry.service'; @@ -23,15 +21,12 @@ describe('InstanceRegistryController', () => { let controller: InstanceRegistryController; let service: jest.Mocked; let checkService: jest.Mocked; - let poolConfigService: jest.Mocked; beforeEach(() => { service = mock(); checkService = mock(); - poolConfigService = mock(); checkService.runChecks.mockResolvedValue({ currentState: new Map(), results: [] }); - poolConfigService.getPoolAssignment.mockResolvedValue({}); - controller = new InstanceRegistryController(service, checkService, poolConfigService); + controller = new InstanceRegistryController(service, checkService); }); describe('getClusterInfo', () => { 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 b99f6ed6fe5..e4f71185418 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, ScalingModeConfig } from '@n8n/config'; +import { WorkerPoolConfig } from '@n8n/config'; import { Container } from '@n8n/di'; import type { InstanceSettings } from 'n8n-core'; import { mock } from 'jest-mock-extended'; @@ -71,8 +72,10 @@ const makeInstanceSettings = (overrides: Partial = {}) => const makeExecutionsConfig = (mode: 'regular' | 'queue' = 'regular') => mock({ mode }); -const makeScalingModeConfig = (poolName: string = '') => - mock({ workerPool: { name: poolName } }); +const makeScalingModeConfig = (poolName: string = '', enabled: boolean = true) => { + const workerPool = Object.assign(new WorkerPoolConfig(), { name: poolName, enabled }); + return mock({ workerPool }); +}; const makeLogger = () => { const logger = mock(); @@ -103,11 +106,12 @@ describe('InstanceRegistryService', () => { settingsOverrides: Partial = {}, executionMode: 'regular' | 'queue' = 'regular', poolName: string = '', + poolsEnabled: boolean = true, ) => new InstanceRegistryService( makeInstanceSettings(settingsOverrides), makeExecutionsConfig(executionMode), - makeScalingModeConfig(poolName), + makeScalingModeConfig(poolName, poolsEnabled), logger, ); @@ -446,6 +450,20 @@ describe('InstanceRegistryService', () => { 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'); diff --git a/packages/cli/src/modules/instance-registry/instance-registry.controller.ts b/packages/cli/src/modules/instance-registry/instance-registry.controller.ts index 8e478ffad88..7f6c86cadb3 100644 --- a/packages/cli/src/modules/instance-registry/instance-registry.controller.ts +++ b/packages/cli/src/modules/instance-registry/instance-registry.controller.ts @@ -1,8 +1,6 @@ import type { ClusterCheckSummary, ClusterInfoResponse } from '@n8n/api-types'; import { Get, RestController } from '@n8n/decorators'; -import { PoolConfigService } from '@/scaling/pool-config.service'; - import { CheckService } from './checks/check.service'; import { InstanceRegistryService } from './instance-registry.service'; @@ -11,15 +9,13 @@ export class InstanceRegistryController { constructor( private readonly instanceRegistryService: InstanceRegistryService, private readonly checkService: CheckService, - private readonly poolConfigService: PoolConfigService, ) {} @Get('/') async getClusterInfo(): Promise { - const [instances, { results }, poolAssignment] = await Promise.all([ + const [instances, { results }] = await Promise.all([ this.instanceRegistryService.getAllInstances(), this.checkService.runChecks(), - this.poolConfigService.getPoolAssignment(), ]); const now = Date.now(); @@ -65,7 +61,6 @@ export class InstanceRegistryController { return { instances, checks, - poolAssignment, }; } } 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 72ffb905281..8fc673c8b37 100644 --- a/packages/cli/src/modules/instance-registry/instance-registry.service.ts +++ b/packages/cli/src/modules/instance-registry/instance-registry.service.ts @@ -2,10 +2,11 @@ import type { InstanceRegistration } from '@n8n/api-types'; import { Logger } from '@n8n/backend-common'; import { ExecutionsConfig, ScalingModeConfig } from '@n8n/config'; import { Service } from '@n8n/di'; -import { randomUUID } from 'node:crypto'; 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'; @@ -107,7 +108,7 @@ export class InstanceRegistryService { }; if (this.instanceSettings.instanceType === 'worker') { - return { ...base, poolName: this.scalingModeConfig.workerPool.name }; + return { ...base, poolName: resolveWorkerPoolName(this.scalingModeConfig.workerPool) }; } return base; diff --git a/packages/cli/src/public-api/v1/handlers/workflows/spec/schemas/workflowSettings.yml b/packages/cli/src/public-api/v1/handlers/workflows/spec/schemas/workflowSettings.yml index 2c7f07dd4e2..b633a086f0a 100644 --- a/packages/cli/src/public-api/v1/handlers/workflows/spec/schemas/workflowSettings.yml +++ b/packages/cli/src/public-api/v1/handlers/workflows/spec/schemas/workflowSettings.yml @@ -99,15 +99,3 @@ properties: type: string value: type: string - workerPoolOverrideProduction: - type: string - description: Worker pool override for production executions. - example: gpu - workerPoolOverrideManual: - type: string - description: Worker pool override for manual executions. - example: gpu - workerPoolOverrideEvaluation: - type: string - description: Worker pool override for evaluation executions. - example: gpu diff --git a/packages/cli/src/scaling/__tests__/pool-config.service.test.ts b/packages/cli/src/scaling/__tests__/pool-config.service.test.ts index 306ad2c8ec4..3e1a1a3c84f 100644 --- a/packages/cli/src/scaling/__tests__/pool-config.service.test.ts +++ b/packages/cli/src/scaling/__tests__/pool-config.service.test.ts @@ -1,267 +1,140 @@ -import type { ProjectPoolSettingsRepository, SettingsRepository } from '@n8n/db'; +import type { GlobalConfig } from '@n8n/config'; +import type { ProjectPoolSettingsRepository } from '@n8n/db'; import { mock } from 'jest-mock-extended'; -import type { InstanceRegistryService } from '@/modules/instance-registry/instance-registry.service'; import { PoolConfigService } from '@/scaling/pool-config.service'; +import type { WorkerPoolsService } from '@/scaling/worker-pools.service'; import type { CacheService } from '@/services/cache/cache.service'; describe('PoolConfigService', () => { - const settingsRepository = mock(); const projectPoolSettingsRepository = mock(); const cacheService = mock(); - const instanceRegistryService = mock(); + const workerPoolsService = mock(); + const globalConfig = mock(); + + const setEnabled = (enabled: boolean) => { + globalConfig.queue = { workerPool: { enabled } } as GlobalConfig['queue']; + }; + const service = new PoolConfigService( - settingsRepository, projectPoolSettingsRepository, cacheService, - instanceRegistryService, + workerPoolsService, + globalConfig, ); beforeEach(() => { jest.clearAllMocks(); - // Default: no workers registered in the cluster - instanceRegistryService.getAllInstances.mockResolvedValue([]); + setEnabled(true); + // Default: cache miss, no workers registered + cacheService.get.mockResolvedValue(undefined); + workerPoolsService.getAvailablePools.mockResolvedValue([]); }); - describe('setPoolAssignment', () => { - it('should merge partial updates with the existing assignment', async () => { - cacheService.get.mockResolvedValue(undefined); - settingsRepository.findByKey.mockResolvedValue({ - key: 'workerPools.assignment', - value: JSON.stringify({ production: 'gpu' }), - loadOnStartup: true, - }); - - const result = await service.setPoolAssignment({ manual: 'cpu' }); - - expect(result).toEqual({ production: 'gpu', manual: 'cpu' }); - expect(settingsRepository.upsert).toHaveBeenCalledWith( - { - key: 'workerPools.assignment', - value: JSON.stringify({ production: 'gpu', manual: 'cpu' }), - loadOnStartup: true, - }, - ['key'], - ); - expect(cacheService.set).toHaveBeenCalledWith( - 'workerPools.assignment', - JSON.stringify({ production: 'gpu', manual: 'cpu' }), - ); - }); - - it('should remove a category when its value is an empty string', async () => { - cacheService.get.mockResolvedValue(JSON.stringify({ production: 'gpu', manual: 'cpu' })); - - const result = await service.setPoolAssignment({ manual: '' }); - - expect(result).toEqual({ production: 'gpu' }); - expect(settingsRepository.upsert).toHaveBeenCalledWith( - { - key: 'workerPools.assignment', - value: JSON.stringify({ production: 'gpu' }), - loadOnStartup: true, - }, - ['key'], - ); - }); - - it('should write to settings repo and update cache', async () => { - cacheService.get.mockResolvedValue(JSON.stringify({})); - - await service.setPoolAssignment({ evaluation: 'eval-pool' }); - - expect(settingsRepository.upsert).toHaveBeenCalledTimes(1); - expect(cacheService.set).toHaveBeenCalledTimes(1); - }); - }); - - describe('getProjectPool', () => { + describe('resolvePoolForExecution', () => { const projectId = 'project-123'; - it('should return pool from DB on cache miss', async () => { - cacheService.get.mockResolvedValue(undefined); - projectPoolSettingsRepository.getPoolForCategory.mockResolvedValue('gpu'); + it("resolves to the project's default pool when it is live", async () => { + projectPoolSettingsRepository.getDefaultPool.mockResolvedValue('gpu'); + workerPoolsService.getAvailablePools.mockResolvedValue(['gpu', 'cpu']); - const result = await service.getProjectPool(projectId, 'production'); + const result = await service.resolvePoolForExecution({ projectId }); - expect(result).toBe('gpu'); - expect(projectPoolSettingsRepository.getPoolForCategory).toHaveBeenCalledWith( - projectId, - 'production', - ); - expect(cacheService.set).toHaveBeenCalledWith(`projectPool:${projectId}:production`, 'gpu'); + expect(result).toEqual({ queueName: 'jobs-gpu', poolName: 'gpu' }); }); - it('should return pool from cache on cache hit', async () => { + 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 the configured pool is not registered by any worker', async () => { + projectPoolSettingsRepository.getDefaultPool.mockResolvedValue('gpu'); + workerPoolsService.getAvailablePools.mockResolvedValue(['cpu']); + + 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.getProjectPool(projectId, 'production'); + const result = await service.resolvePoolForExecution({ projectId }); - expect(result).toBe('gpu'); - expect(projectPoolSettingsRepository.getPoolForCategory).not.toHaveBeenCalled(); + 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('should return undefined when no settings row exists', async () => { - cacheService.get.mockResolvedValue(undefined); - projectPoolSettingsRepository.getPoolForCategory.mockResolvedValue(undefined); + it('returns null defaultPool when none is stored', async () => { + projectPoolSettingsRepository.getDefaultPool.mockResolvedValue(null); - const result = await service.getProjectPool(projectId, 'production'); + const result = await service.getProjectPoolSettings(projectId); - expect(result).toBeUndefined(); - expect(cacheService.set).toHaveBeenCalledWith(`projectPool:${projectId}:production`, ''); - }); - - it('should return undefined when cached value is empty string', async () => { - cacheService.get.mockResolvedValue(''); - - const result = await service.getProjectPool(projectId, 'production'); - - expect(result).toBeUndefined(); - expect(projectPoolSettingsRepository.getPoolForCategory).not.toHaveBeenCalled(); - }); - - it('should cache per category', async () => { - cacheService.get.mockResolvedValue(undefined); - projectPoolSettingsRepository.getPoolForCategory - .mockResolvedValueOnce('gpu') - .mockResolvedValueOnce('cpu'); - - await service.getProjectPool(projectId, 'production'); - await service.getProjectPool(projectId, 'manual'); - - expect(cacheService.set).toHaveBeenCalledWith(`projectPool:${projectId}:production`, 'gpu'); - expect(cacheService.set).toHaveBeenCalledWith(`projectPool:${projectId}:manual`, 'cpu'); + expect(result).toEqual({ defaultPool: null, availablePools: [] }); }); }); describe('setProjectPoolSettings', () => { const projectId = 'project-123'; - beforeEach(() => { - // By default, no existing row, no cluster context, no instance defaults - cacheService.get.mockResolvedValue(undefined); - projectPoolSettingsRepository.getSettings.mockResolvedValue(undefined); - settingsRepository.findByKey.mockResolvedValue(null); + 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('writes assignment and allowedPools when both are provided', async () => { - const result = await service.setProjectPoolSettings(projectId, { - assignment: { production: 'gpu' }, - allowedPools: ['gpu', 'cpu'], - }); + it('clears the default pool when given an empty string', async () => { + const result = await service.setProjectPoolSettings(projectId, { defaultPool: '' }); - expect(result.assignment).toEqual({ production: 'gpu' }); - expect(result.allowedPools).toEqual(['gpu', 'cpu']); - expect(projectPoolSettingsRepository.setSettings).toHaveBeenCalledWith(projectId, { - assignment: { production: 'gpu' }, - allowedPools: ['gpu', 'cpu'], - }); + expect(result.defaultPool).toBeNull(); + expect(projectPoolSettingsRepository.setDefaultPool).toHaveBeenCalledWith(projectId, null); }); - it('merges assignment with existing values; keeps unspecified categories', async () => { - projectPoolSettingsRepository.getSettings.mockResolvedValueOnce({ - assignment: { production: 'gpu', manual: 'cpu' }, - allowedPools: ['gpu', 'cpu'], - }); + it('clears the default pool when given null', async () => { + const result = await service.setProjectPoolSettings(projectId, { defaultPool: null }); - const result = await service.setProjectPoolSettings(projectId, { - assignment: { evaluation: 'cpu' }, - }); - - expect(result.assignment).toEqual({ production: 'gpu', manual: 'cpu', evaluation: 'cpu' }); - expect(result.allowedPools).toEqual(['gpu', 'cpu']); - }); - - it('clears a category when assignment value is an empty string', async () => { - projectPoolSettingsRepository.getSettings.mockResolvedValueOnce({ - assignment: { production: 'gpu', manual: 'cpu' }, - allowedPools: ['gpu', 'cpu'], - }); - - const result = await service.setProjectPoolSettings(projectId, { - assignment: { production: '' }, - }); - - expect(result.assignment).toEqual({ manual: 'cpu' }); - }); - - it('keeps existing allowedPools when not specified in patch', async () => { - projectPoolSettingsRepository.getSettings.mockResolvedValueOnce({ - assignment: { production: 'gpu' }, - allowedPools: ['gpu', 'cpu'], - }); - - const result = await service.setProjectPoolSettings(projectId, { - assignment: { manual: 'cpu' }, - }); - - expect(result.allowedPools).toEqual(['gpu', 'cpu']); - }); - - it('rejects when a per-category default is not in allowedPools (new write)', async () => { - await expect( - service.setProjectPoolSettings(projectId, { - assignment: { production: 'gpu' }, - allowedPools: ['cpu'], - }), - ).rejects.toThrow('production pool "gpu" must be one of allowedPools'); - - expect(projectPoolSettingsRepository.setSettings).not.toHaveBeenCalled(); - }); - - it('rejects when shrinking allowedPools below an existing default', async () => { - projectPoolSettingsRepository.getSettings.mockResolvedValueOnce({ - assignment: { production: 'gpu' }, - allowedPools: ['gpu', 'cpu'], - }); - - await expect( - service.setProjectPoolSettings(projectId, { - allowedPools: ['cpu'], - }), - ).rejects.toThrow('production pool "gpu" must be one of allowedPools'); - }); - - it('accepts any pool as a default when allowedPools is empty (no restriction)', async () => { - const result = await service.setProjectPoolSettings(projectId, { - assignment: { production: 'gpu' }, - allowedPools: [], - }); - - expect(result.assignment).toEqual({ production: 'gpu' }); - expect(result.allowedPools).toEqual([]); - expect(projectPoolSettingsRepository.setSettings).toHaveBeenCalledWith(projectId, { - assignment: { production: 'gpu' }, - allowedPools: [], - }); - }); - - it('accepts clearing allowedPools while keeping defaults that were previously constrained', async () => { - projectPoolSettingsRepository.getSettings.mockResolvedValueOnce({ - assignment: { production: 'gpu' }, - allowedPools: ['gpu', 'cpu'], - }); - - const result = await service.setProjectPoolSettings(projectId, { - allowedPools: [], - }); - - expect(result.assignment).toEqual({ production: 'gpu' }); - expect(result.allowedPools).toEqual([]); - }); - - it('invalidates the project caches after a successful write', async () => { - await service.setProjectPoolSettings(projectId, { - assignment: { production: 'gpu' }, - allowedPools: ['gpu'], - }); - - expect(cacheService.deleteMany).toHaveBeenCalledWith([ - `projectPoolSettings:${projectId}`, - `projectPool:${projectId}:production`, - `projectPool:${projectId}:manual`, - `projectPool:${projectId}:evaluation`, - ]); + 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 index cc878178a7a..b0df8992558 100644 --- a/packages/cli/src/scaling/__tests__/queue-name.test.ts +++ b/packages/cli/src/scaling/__tests__/queue-name.test.ts @@ -1,4 +1,6 @@ -import { resolveQueueName } from '../queue-name'; +import type { WorkerPoolConfig } from '@n8n/config'; + +import { resolveQueueName, resolveWorkerPoolName } from '../queue-name'; describe('resolveQueueName', () => { test('returns "jobs" on main regardless of pool', () => { @@ -20,3 +22,13 @@ describe('resolveQueueName', () => { 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 51925c2f573..6ad42f878ef 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'; @@ -39,7 +39,7 @@ describe('ScalingService', () => { tls: false, }, }, - workerPool: { name: '' }, + workerPool: Object.assign(new WorkerPoolConfig(), { enabled: true, name: '' }), }, endpoints: { metrics: { @@ -176,6 +176,7 @@ describe('ScalingService', () => { describe('queue name resolution', () => { afterEach(() => { globalConfig.queue.workerPool.name = ''; + globalConfig.queue.workerPool.enabled = true; }); it('uses "jobs" on worker when pool is empty', async () => { @@ -197,6 +198,17 @@ describe('ScalingService', () => { 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'; diff --git a/packages/cli/src/scaling/__tests__/worker-pools.service.test.ts b/packages/cli/src/scaling/__tests__/worker-pools.service.test.ts index a531ca4a9dd..afe44092cdd 100644 --- a/packages/cli/src/scaling/__tests__/worker-pools.service.test.ts +++ b/packages/cli/src/scaling/__tests__/worker-pools.service.test.ts @@ -2,7 +2,6 @@ import type { InstanceRegistration } from '@n8n/api-types'; import { mock } from 'jest-mock-extended'; import type { InstanceRegistryService } from '@/modules/instance-registry/instance-registry.service'; -import type { PoolConfigService } from '@/scaling/pool-config.service'; import { WorkerPoolsService } from '@/scaling/worker-pools.service'; const buildWorker = (overrides: Partial = {}): InstanceRegistration => ({ @@ -18,9 +17,8 @@ const buildWorker = (overrides: Partial = {}): InstanceReg }); describe('WorkerPoolsService', () => { - const poolConfigService = mock(); const instanceRegistryService = mock(); - const service = new WorkerPoolsService(poolConfigService, instanceRegistryService); + const service = new WorkerPoolsService(instanceRegistryService); beforeEach(() => { jest.clearAllMocks(); @@ -65,19 +63,4 @@ describe('WorkerPoolsService', () => { await expect(service.getAvailablePools()).resolves.toEqual([]); }); }); - - describe('getPoolsState', () => { - it('should return pools and assignment together', async () => { - instanceRegistryService.getAllInstances.mockResolvedValue([buildWorker({ poolName: 'gpu' })]); - poolConfigService.getPoolAssignment.mockResolvedValue({ production: 'gpu' }); - - const result = await service.getPoolsState(); - - expect(result).toEqual({ - pools: ['gpu'], - assignment: { production: 'gpu' }, - }); - expect(poolConfigService.getPoolAssignment).toHaveBeenCalledTimes(1); - }); - }); }); diff --git a/packages/cli/src/scaling/execution-category.ts b/packages/cli/src/scaling/execution-category.ts deleted file mode 100644 index 8dbc794864a..00000000000 --- a/packages/cli/src/scaling/execution-category.ts +++ /dev/null @@ -1,20 +0,0 @@ -import type { WorkflowExecuteMode } from 'n8n-workflow'; - -import type { ExecutionCategory } from './scaling.types'; - -/** Map an execution mode to its pool category. Returns `undefined` for modes that run locally. */ -export function getExecutionCategory(mode: WorkflowExecuteMode): ExecutionCategory | undefined { - switch (mode) { - case 'webhook': - case 'trigger': - case 'chat': - return 'production'; - case 'manual': - case 'retry': - return 'manual'; - case 'evaluation': - return 'evaluation'; - default: - return undefined; - } -} diff --git a/packages/cli/src/scaling/pool-config.service.ts b/packages/cli/src/scaling/pool-config.service.ts index 9d7200e000d..f2f50f077c6 100644 --- a/packages/cli/src/scaling/pool-config.service.ts +++ b/packages/cli/src/scaling/pool-config.service.ts @@ -1,190 +1,87 @@ import type { ProjectPoolSettingsResponse, UpdateProjectPoolSettingsDto } from '@n8n/api-types'; -import { ProjectPoolSettingsRepository, SettingsRepository } from '@n8n/db'; +import { GlobalConfig } from '@n8n/config'; +import { ProjectPoolSettingsRepository } from '@n8n/db'; import { Service } from '@n8n/di'; -import { jsonParse } from 'n8n-workflow'; +import type { IWorkflowExecutionDataProcess } from 'n8n-workflow'; -import { BadRequestError } from '@/errors/response-errors/bad-request.error'; -import { InstanceRegistryService } from '@/modules/instance-registry/instance-registry.service'; import { CacheService } from '@/services/cache/cache.service'; -import { poolQueueName } from './queue-name'; -import type { ExecutionCategory, PoolAssignment } from './scaling.types'; +import { DEFAULT_QUEUE_NAME, poolQueueName } from './queue-name'; +import { WorkerPoolsService } from './worker-pools.service'; -const EXECUTION_CATEGORIES: ExecutionCategory[] = ['production', 'manual', 'evaluation']; - -const SETTINGS_KEY = 'workerPools.assignment'; - -/** Resolves execution categories to worker pool queue names using DB-backed settings with cache. */ +/** + * 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 settingsRepository: SettingsRepository, private readonly projectPoolSettingsRepository: ProjectPoolSettingsRepository, private readonly cacheService: CacheService, - private readonly instanceRegistryService: InstanceRegistryService, + private readonly workerPoolsService: WorkerPoolsService, + private readonly globalConfig: GlobalConfig, ) {} - private async getAvailablePoolsFromCluster(): Promise { - const instances = await this.instanceRegistryService.getAllInstances(); - const names = new Set(); - for (const instance of instances) { - if (instance.instanceType !== 'worker') continue; - if (instance.poolName) names.add(instance.poolName); - } - return Array.from(names).sort(); - } + /** + * Resolve the queue an execution should run on. Returns the project's default pool when worker + * pools are enabled and the pool is live; otherwise the system default queue. Never routes to a + * pool that isn't registered by a worker. + */ + async resolvePoolForExecution( + data: Pick, + ): Promise<{ queueName: string; poolName: string | undefined }> { + const systemDefault = { queueName: DEFAULT_QUEUE_NAME, poolName: undefined }; - async getQueueNameForCategory(category: ExecutionCategory): Promise { - const assignment = await this.getPoolAssignment(); - return poolQueueName(assignment[category]); - } + if (!this.globalConfig.queue.workerPool.enabled) return systemDefault; + if (!data.projectId) return systemDefault; - async getPoolAssignment(): Promise { - const cached = await this.cacheService.get(SETTINGS_KEY); - if (cached !== undefined) { - return jsonParse(cached, { fallbackValue: {} }); - } + const pool = await this.getProjectDefaultPool(data.projectId); + if (!pool) return systemDefault; - const row = await this.settingsRepository.findByKey(SETTINGS_KEY); - const assignment = row?.value - ? jsonParse(row.value, { fallbackValue: {} }) - : {}; + const availablePools = await this.workerPoolsService.getAvailablePools(); + if (!availablePools.includes(pool)) return systemDefault; - await this.cacheService.set(SETTINGS_KEY, JSON.stringify(assignment)); - return assignment; - } - - async setPoolAssignment(partial: PoolAssignment): Promise { - const current = await this.getPoolAssignment(); - const next: PoolAssignment = { ...current }; - for (const [category, pool] of Object.entries(partial) as Array< - [ExecutionCategory, string | undefined] - >) { - if (pool === undefined || pool === '') { - delete next[category]; - } else { - next[category] = pool; - } - } - - const serialized = JSON.stringify(next); - await this.settingsRepository.upsert( - { key: SETTINGS_KEY, value: serialized, loadOnStartup: true }, - ['key'], - ); - await this.cacheService.set(SETTINGS_KEY, serialized); - - return next; - } - - async getProjectPool( - projectId: string, - category: ExecutionCategory, - ): Promise { - const cacheKey = `projectPool:${projectId}:${category}`; - - const cached = await this.cacheService.get(cacheKey); - if (cached !== undefined) return cached || undefined; - - const poolName = await this.projectPoolSettingsRepository.getPoolForCategory( - projectId, - category, - ); - - void this.cacheService.set(cacheKey, poolName ?? ''); - return poolName; + return { queueName: poolQueueName(pool), poolName: pool }; } async getProjectPoolSettings(projectId: string): Promise { - const [stored, availablePools, instanceDefaults] = await Promise.all([ - this.getStoredProjectPoolSettings(projectId), - this.getAvailablePoolsFromCluster(), - this.getPoolAssignment(), + const [defaultPool, availablePools] = await Promise.all([ + this.getProjectDefaultPool(projectId), + this.workerPoolsService.getAvailablePools(), ]); - return { - ...stored, - availablePools, - instanceDefaults, - }; - } - - private async getStoredProjectPoolSettings( - projectId: string, - ): Promise<{ assignment: PoolAssignment; allowedPools: string[] }> { - const cacheKey = `projectPoolSettings:${projectId}`; - - const cached = await this.cacheService.get(cacheKey); - if (cached !== undefined) { - return jsonParse<{ assignment: PoolAssignment; allowedPools: string[] }>(cached, { - fallbackValue: { assignment: {}, allowedPools: [] }, - }); - } - - const settings = await this.projectPoolSettingsRepository.getSettings(projectId); - const result = settings ?? { assignment: {}, allowedPools: [] }; - - void this.cacheService.set(cacheKey, JSON.stringify(result)); - return result; + return { defaultPool: defaultPool ?? null, availablePools }; } async setProjectPoolSettings( projectId: string, patch: UpdateProjectPoolSettingsDto, ): Promise { - const current = await this.getStoredProjectPoolSettings(projectId); + const defaultPool = patch.defaultPool ? patch.defaultPool : null; - const nextAssignment: PoolAssignment = { ...current.assignment }; - if (patch.assignment) { - for (const [category, pool] of Object.entries(patch.assignment) as Array< - [ExecutionCategory, string | undefined] - >) { - if (pool === undefined || pool === '') { - delete nextAssignment[category]; - } else { - nextAssignment[category] = pool; - } - } - } + await this.projectPoolSettingsRepository.setDefaultPool(projectId, defaultPool); + await this.cacheService.delete(this.cacheKey(projectId)); - const nextAllowedPools = patch.allowedPools ?? current.allowedPools; - - // `allowedPools` is an optional allowlist that restricts which pools workflows in this - // project may route to. An empty list means "no restriction" — any pool can be used. - // We only enforce coherence between defaults and allowedPools when the allowlist is in use. - if (nextAllowedPools.length > 0) { - for (const category of EXECUTION_CATEGORIES) { - const pool = nextAssignment[category]; - if (pool && !nextAllowedPools.includes(pool)) { - throw new BadRequestError(`${category} pool "${pool}" must be one of allowedPools`); - } - } - } - - await this.projectPoolSettingsRepository.setSettings(projectId, { - assignment: nextAssignment, - allowedPools: nextAllowedPools, - }); - await this.invalidateProjectCaches(projectId); - - const [availablePools, instanceDefaults] = await Promise.all([ - this.getAvailablePoolsFromCluster(), - this.getPoolAssignment(), - ]); - - return { - assignment: nextAssignment, - allowedPools: nextAllowedPools, - availablePools, - instanceDefaults, - }; + const availablePools = await this.workerPoolsService.getAvailablePools(); + return { defaultPool, availablePools }; } - private async invalidateProjectCaches(projectId: string): Promise { - const keys = [ - `projectPoolSettings:${projectId}`, - ...EXECUTION_CATEGORIES.map((category) => `projectPool:${projectId}:${category}`), - ]; - await this.cacheService.deleteMany(keys); + 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 index e50cfb695ca..e36f440389c 100644 --- a/packages/cli/src/scaling/queue-name.ts +++ b/packages/cli/src/scaling/queue-name.ts @@ -1,9 +1,15 @@ +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}`; diff --git a/packages/cli/src/scaling/scaling.service.ts b/packages/cli/src/scaling/scaling.service.ts index 9b8f8ce03f0..b803bf220d0 100644 --- a/packages/cli/src/scaling/scaling.service.ts +++ b/packages/cli/src/scaling/scaling.service.ts @@ -17,7 +17,7 @@ import { assertNever } from '@/utils'; import { JOB_TYPE_NAME } from './constants'; import { JobProcessor } from './job-processor'; -import { DEFAULT_QUEUE_NAME, resolveQueueName } from './queue-name'; +import { DEFAULT_QUEUE_NAME, resolveQueueName, resolveWorkerPoolName } from './queue-name'; import type { JobQueue, Job, @@ -99,7 +99,7 @@ export class ScalingService { createClient: (type) => service.createClient({ type: `${type}(bull)` }), }); - const poolName = this.globalConfig.queue.workerPool.name; + const poolName = resolveWorkerPoolName(this.globalConfig.queue.workerPool); const queueName = resolveQueueName(this.instanceSettings.instanceType, poolName); const queue = this.createBullQueue(queueName); diff --git a/packages/cli/src/scaling/scaling.types.ts b/packages/cli/src/scaling/scaling.types.ts index bb6707c6121..d9faeafa2a7 100644 --- a/packages/cli/src/scaling/scaling.types.ts +++ b/packages/cli/src/scaling/scaling.types.ts @@ -15,10 +15,6 @@ export type Job = Bull.Job; export type JobId = Job['id']; -export type ExecutionCategory = 'production' | 'manual' | 'evaluation'; - -export type PoolAssignment = Partial>; - export type JobData = { workflowId: string; executionId: string; diff --git a/packages/cli/src/scaling/worker-pools.service.ts b/packages/cli/src/scaling/worker-pools.service.ts index 63e0591166e..c26be814d01 100644 --- a/packages/cli/src/scaling/worker-pools.service.ts +++ b/packages/cli/src/scaling/worker-pools.service.ts @@ -1,16 +1,11 @@ -import type { WorkerPoolsResponse } from '@n8n/api-types'; import { Service } from '@n8n/di'; import { InstanceRegistryService } from '@/modules/instance-registry/instance-registry.service'; -import { PoolConfigService } from './pool-config.service'; - +/** Discovers worker pool names registered by workers in the cluster. */ @Service() export class WorkerPoolsService { - constructor( - private readonly poolConfigService: PoolConfigService, - private readonly instanceRegistryService: InstanceRegistryService, - ) {} + constructor(private readonly instanceRegistryService: InstanceRegistryService) {} async getAvailablePools(): Promise { const instances = await this.instanceRegistryService.getAllInstances(); @@ -23,13 +18,4 @@ export class WorkerPoolsService { return Array.from(poolNames).sort(); } - - async getPoolsState(): Promise { - const [pools, assignment] = await Promise.all([ - this.getAvailablePools(), - this.poolConfigService.getPoolAssignment(), - ]); - - return { pools, assignment }; - } } diff --git a/packages/cli/src/scaling/worker-status.service.ee.ts b/packages/cli/src/scaling/worker-status.service.ee.ts index a526d99b349..4edf873795a 100644 --- a/packages/cli/src/scaling/worker-status.service.ee.ts +++ b/packages/cli/src/scaling/worker-status.service.ee.ts @@ -11,7 +11,7 @@ import { Push } from '@/push'; import { JobProcessor } from './job-processor'; import { Publisher } from './pubsub/publisher.service'; -import { resolveQueueName } from './queue-name'; +import { resolveQueueName, resolveWorkerPoolName } from './queue-name'; @Service() export class WorkerStatusService { @@ -99,10 +99,10 @@ export class WorkerStatusService { })), ), version: N8N_VERSION, - poolName: this.globalConfig.queue.workerPool.name, + poolName: resolveWorkerPoolName(this.globalConfig.queue.workerPool), queueName: resolveQueueName( this.instanceSettings.instanceType, - this.globalConfig.queue.workerPool.name, + resolveWorkerPoolName(this.globalConfig.queue.workerPool), ), }; } diff --git a/packages/cli/src/services/__tests__/frontend.service.test.ts b/packages/cli/src/services/__tests__/frontend.service.test.ts index 7a574260ee1..6227d31b8e7 100644 --- a/packages/cli/src/services/__tests__/frontend.service.test.ts +++ b/packages/cli/src/services/__tests__/frontend.service.test.ts @@ -77,6 +77,7 @@ describe('FrontendService', () => { userManagement: { password: { minLength: 8 }, }, + queue: { workerPool: { enabled: false } }, }); const instanceSettings = mock({ diff --git a/packages/cli/src/services/frontend.service.ts b/packages/cli/src/services/frontend.service.ts index b6afa7ba5f0..671df9779ab 100644 --- a/packages/cli/src/services/frontend.service.ts +++ b/packages/cli/src/services/frontend.service.ts @@ -410,6 +410,9 @@ export class FrontendService { folders: { enabled: false, }, + workerPools: { + enabled: this.globalConfig.queue.workerPool.enabled, + }, evaluation: { quota: this.licenseState.getMaxWorkflowsWithEvaluations(), }, diff --git a/packages/cli/src/workflow-helpers.ts b/packages/cli/src/workflow-helpers.ts index 0e789b0bdc5..cb3be073a5d 100644 --- a/packages/cli/src/workflow-helpers.ts +++ b/packages/cli/src/workflow-helpers.ts @@ -358,18 +358,6 @@ export function removeDefaultValues( delete cleanedSettings.credentialResolverId; } - // Remove worker pool overrides if they were cleared (empty string / falsy values from UI clear action). - const poolOverrideKeys = [ - 'workerPoolOverrideProduction', - 'workerPoolOverrideManual', - 'workerPoolOverrideEvaluation', - ] as const; - for (const key of poolOverrideKeys) { - if (!cleanedSettings[key]) { - delete cleanedSettings[key]; - } - } - return cleanedSettings; } diff --git a/packages/cli/src/workflow-runner.ts b/packages/cli/src/workflow-runner.ts index 6e09516e635..d2735d0b2d7 100644 --- a/packages/cli/src/workflow-runner.ts +++ b/packages/cli/src/workflow-runner.ts @@ -26,6 +26,7 @@ import type { } from 'n8n-workflow'; import { createRunExecutionData, + ensureError, ExecutionCancelledError, ManualExecutionCancelledError, TimeoutExecutionCancelledError, @@ -48,9 +49,8 @@ import { CredentialsPermissionChecker } from '@/executions/pre-execution-checks' import { ExternalHooks } from '@/external-hooks'; import { ManualExecutionService } from '@/manual-execution.service'; import { NodeTypes } from '@/node-types'; -import { poolQueueName } from '@/scaling/queue-name'; -import { getExecutionCategory } from '@/scaling/execution-category'; import type { PoolConfigService } from '@/scaling/pool-config.service'; +import { DEFAULT_QUEUE_NAME } from '@/scaling/queue-name'; import type { ScalingService } from '@/scaling/scaling.service'; import type { Job, JobData } from '@/scaling/scaling.types'; import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data'; @@ -517,7 +517,18 @@ export class WorkflowRunner { this.poolConfigService = Container.get(PoolConfigService); } - const { queueName, poolName } = await this.getQueueForWorkflow(data); + // A pool-resolution failure must not abort the execution: fall back to the default queue. + let queueName: string; + let poolName: string | undefined; + try { + ({ queueName, poolName } = await this.poolConfigService.resolvePoolForExecution(data)); + } catch (error) { + this.logger.warn( + `Failed to resolve worker pool for execution ${executionId}, falling back to default queue: ${ensureError(error).message}`, + ); + queueName = DEFAULT_QUEUE_NAME; + poolName = undefined; + } const jobData: JobData = { workflowId, @@ -542,9 +553,7 @@ export class WorkflowRunner { let job: Job; let lifecycleHooks: ExecutionLifecycleHooks; try { - const isHighPriorityProduction = !realtime && data.workflowData.settings?.priority === 'high'; - const jobPriority = realtime ? 50 : isHighPriorityProduction ? 25 : 100; - job = await this.scalingService.addJob(jobData, { priority: jobPriority, queueName }); + job = await this.scalingService.addJob(jobData, { priority: realtime ? 50 : 100, queueName }); lifecycleHooks = getLifecycleHooksForScalingMain(data, executionId); @@ -684,45 +693,6 @@ export class WorkflowRunner { this.activeExecutions.attachWorkflowExecution(executionId, workflowExecution); } - private async getQueueForWorkflow( - data: IWorkflowExecutionDataProcess, - ): Promise<{ queueName: string; poolName: string | undefined }> { - const category = getExecutionCategory(data.executionMode); - - const workflowPool = this.getWorkflowPool(data, category); - if (workflowPool) { - return { queueName: poolQueueName(workflowPool), poolName: workflowPool }; - } - - if (category && data.projectId) { - const projectPool = await this.poolConfigService.getProjectPool(data.projectId, category); - if (projectPool) { - return { queueName: poolQueueName(projectPool), poolName: projectPool }; - } - } - - const poolAssignment = await this.poolConfigService.getPoolAssignment(); - const poolName = category ? poolAssignment[category] : undefined; - - return { queueName: poolQueueName(poolName), poolName }; - } - - private getWorkflowPool( - data: IWorkflowExecutionDataProcess, - category: ReturnType, - ): string | undefined { - switch (category) { - case 'production': - return data.workflowData.settings?.workerPoolOverrideProduction; - case 'manual': - return data.workflowData.settings?.workerPoolOverrideManual; - case 'evaluation': - return data.workflowData.settings?.workerPoolOverrideEvaluation; - default: - return undefined; - } - } - /** * Whether main must retrieve full execution data from the DB on job completion. * diff --git a/packages/frontend/@n8n/i18n/src/locales/en.json b/packages/frontend/@n8n/i18n/src/locales/en.json index 8bb6165fe49..e60fa8cf79e 100644 --- a/packages/frontend/@n8n/i18n/src/locales/en.json +++ b/packages/frontend/@n8n/i18n/src/locales/en.json @@ -1537,10 +1537,10 @@ "fromAiParametersModal.cancel": "Cancel", "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.lastUpdated": "Last updated", "workerList.item.jobList.empty": "No current jobs", "workerList.item.jobListTitle": "Current Jobs", "workerList.item.netListTitle": "Network Interfaces", @@ -1552,19 +1552,6 @@ "workerList.actionBox.description.link": "More info", "workerList.actionBox.buttonText": "See plans", "workerList.docs.url": "https://docs.n8n.io/hosting/scaling/queue-mode/#view-running-workers", - "workerList.pools.title": "Worker pools", - "workerList.pools.available.title": "Available pools", - "workerList.pools.available.empty": "No pools available", - "workerList.pools.defaults.title": "Default pool routing", - "workerList.pools.defaults.description": "Choose which pool handles each type of execution. Unassigned executions run on the default queue.", - "workerList.pools.defaults.production": "Production executions", - "workerList.pools.defaults.manual": "Manual executions", - "workerList.pools.defaults.evaluation": "Evaluation executions", - "workerList.pools.defaults.unassigned": "Unassigned (default queue)", - "workerList.pools.defaults.save": "Save", - "workerList.pools.defaults.saved.success": "Default pool routing updated", - "workerList.pools.defaults.saved.error": "Could not update default pool routing", - "workerList.pools.loading.error": "Could not load worker pools", "executionSidebar.executionName": "Execution {id}", "executionSidebar.searchPlaceholder": "Search executions...", "executionView.onPaste.title": "Cannot paste here", @@ -4246,22 +4233,6 @@ "workflowSettings.timeSavedPerExecution.fixedTabWarning.link": "time saved nodes", "workflowSettings.availableInMCP": "Available in MCP", "workflowSettings.availableInMCP.tooltip": "Make this workflow visible to AI Agents through n8n MCP", - "workflowSettings.workerPoolOverrideProduction": "Worker Pool (Production)", - "workflowSettings.workerPoolOverrideManual": "Worker Pool (Manual)", - "workflowSettings.workerPoolOverrideEvaluation": "Worker Pool (Evaluation)", - "workflowSettings.workerPool.default": "Default", - "workflowSettings.workerPool.defaultWithPool": "Default - '{poolName}' pool", - "workflowSettings.workerPool.production": "Worker Pool (Production)", - "workflowSettings.workerPool.production.tooltip": "Override which worker pool handles production executions (webhook, trigger, chat) for this workflow.", - "workflowSettings.workerPool.manual": "Worker Pool (Manual)", - "workflowSettings.workerPool.manual.tooltip": "Override which worker pool handles manual and retry executions for this workflow.", - "workflowSettings.workerPool.evaluation": "Worker Pool (Evaluation)", - "workflowSettings.workerPool.evaluation.tooltip": "Override which worker pool handles evaluation executions for this workflow.", - "workflowSettings.workerPool.notAllowed": "not in allowed pools", - "workflowSettings.priority": "Execution priority", - "workflowSettings.priority.tooltip": "Production executions of high-priority workflows are picked up before default-priority ones when workers are busy. Manual and test executions are unaffected.", - "workflowSettings.priority.options.default": "Default", - "workflowSettings.priority.options.high": "High", "workflowSettings.toggleMCP.error.title": "Error updating MCP settings", "workflowSettings.toggleMCP.notFoundError": "Workflow not found", "workflowSettings.toggleMCP.updateSkippedError": "Workflow {workflowId} could not be updated. It may be archived or you may no longer have permission to edit it.", @@ -5167,17 +5138,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 pools", - "projects.settings.workerPools.allowedPools.label": "Allowed pools", - "projects.settings.workerPools.allowedPools.placeholder": "Type and press Enter to add a pool", - "projects.settings.workerPools.allowedPools.help": "Optional. Restrict workflows in this project to a subset of pools. Leave empty to allow any pool.", - "projects.settings.workerPools.defaults.title": "Default pool routing", - "projects.settings.workerPools.defaults.description": "Choose which pool handles each type of execution. \"Inherit from instance\" falls back to the instance-wide default.", - "projects.settings.workerPools.defaults.production": "Production executions", - "projects.settings.workerPools.defaults.manual": "Manual executions", - "projects.settings.workerPools.defaults.evaluation": "Evaluation executions", - "projects.settings.workerPools.defaults.inherit": "Inherit from instance", - "projects.settings.workerPools.error.notInAllowedPools": "{category} pool \"{pool}\" must be in the allowed pools list", + "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 a417d2ddd52..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,33 +1,14 @@ -import type { - PoolAssignment, - ProjectPoolSettingsResponse, - UpdateProjectPoolSettingsDto, - UpdateWorkerPoolAssignmentDto, - WorkerPoolsResponse, -} from '@n8n/api-types'; +import type { ProjectPoolSettingsResponse, UpdateProjectPoolSettingsDto } from '@n8n/api-types'; import type { IRestApiContext } from '../types'; import { makeRestApiRequest } from '../utils'; const GET_STATUS_ENDPOINT = '/orchestration/worker/status'; -const WORKER_POOLS_ENDPOINT = '/orchestration/worker/pools'; -const WORKER_POOL_ASSIGNMENT_ENDPOINT = '/orchestration/worker/pools/assignment'; export const sendGetWorkerStatus = async (context: IRestApiContext): Promise => { await makeRestApiRequest(context, 'POST', GET_STATUS_ENDPOINT); }; -export const getWorkerPools = async (context: IRestApiContext): Promise => { - return await makeRestApiRequest(context, 'GET', WORKER_POOLS_ENDPOINT); -}; - -export const updateWorkerPoolAssignment = async ( - context: IRestApiContext, - dto: UpdateWorkerPoolAssignmentDto, -): Promise => { - return await makeRestApiRequest(context, 'PATCH', WORKER_POOL_ASSIGNMENT_ENDPOINT, dto); -}; - export const getProjectPoolSettings = async ( context: IRestApiContext, projectId: string, diff --git a/packages/frontend/editor-ui/src/Interface.ts b/packages/frontend/editor-ui/src/Interface.ts index 5e1de11f47c..81a589d83d2 100644 --- a/packages/frontend/editor-ui/src/Interface.ts +++ b/packages/frontend/editor-ui/src/Interface.ts @@ -431,9 +431,6 @@ export interface IWorkflowSettings extends IWorkflowSettingsWorkflow { executionOrder: NonNullable; binaryMode?: WorkflowSettingsBinaryMode; availableInMCP?: boolean; - workerPoolOverrideProduction?: string; - workerPoolOverrideManual?: string; - workerPoolOverrideEvaluation?: string; } export interface ITimeoutHMS { diff --git a/packages/frontend/editor-ui/src/__tests__/defaults.ts b/packages/frontend/editor-ui/src/__tests__/defaults.ts index 9f663a6158d..a885dcc18b7 100644 --- a/packages/frontend/editor-ui/src/__tests__/defaults.ts +++ b/packages/frontend/editor-ui/src/__tests__/defaults.ts @@ -176,6 +176,9 @@ export const defaultSettings: FrontendSettings = { folders: { enabled: false, }, + workerPools: { + enabled: false, + }, collaboration: { crdt: 'off', }, diff --git a/packages/frontend/editor-ui/src/app/components/WorkflowSettings/WorkflowSettings.vue b/packages/frontend/editor-ui/src/app/components/WorkflowSettings/WorkflowSettings.vue index 61d81c618de..0349f8be51b 100644 --- a/packages/frontend/editor-ui/src/app/components/WorkflowSettings/WorkflowSettings.vue +++ b/packages/frontend/editor-ui/src/app/components/WorkflowSettings/WorkflowSettings.vue @@ -66,8 +66,6 @@ import * as securitySettingsApi from '@n8n/rest-api-client/api/security-settings import type { RedactionFloor } from '@n8n/api-types'; import { hasPermission } from '@/app/utils/rbac/permissions'; import WorkflowCustomTelemetryTags from '@/app/components/WorkflowSettings/WorkflowCustomTelemetryTags.vue'; -import { useInstanceRegistryStore } from '@/features/instanceRegistry/stores/instanceRegistry.store'; -import { getProjectPoolSettings } from '@n8n/rest-api-client/api/orchestration'; import { ElCol, ElRow, ElSwitch } from 'element-plus'; @@ -155,103 +153,6 @@ const { const executionTimeout = ref(0); const maxExecutionTimeout = ref(0); const timeoutHMS = ref({ hours: 0, minutes: 0, seconds: 0 }); -type WorkerPoolCategory = 'production' | 'manual' | 'evaluation'; -type WorkerPoolSettingKey = - | 'workerPoolOverrideProduction' - | 'workerPoolOverrideManual' - | 'workerPoolOverrideEvaluation'; - -const workerPoolNames = ref([]); -const globalPoolAssignment = ref>>({}); -const projectPoolAssignment = ref>>({}); -const projectAllowedPools = ref([]); - -const workerPoolSettingKeys = { - production: 'workerPoolOverrideProduction', - manual: 'workerPoolOverrideManual', - evaluation: 'workerPoolOverrideEvaluation', -} satisfies Record; - -const instanceRegistryStore = useInstanceRegistryStore(); - -const loadWorkerPoolOptions = async () => { - await instanceRegistryStore.fetchClusterInfo(); - - const clusterInfo = instanceRegistryStore.clusterInfo; - const instances = clusterInfo?.instances ?? []; - const poolNames = new Set(); - - for (const instance of instances) { - if ('poolName' in instance && typeof instance.poolName === 'string' && instance.poolName) { - poolNames.add(instance.poolName); - } - } - - workerPoolNames.value = Array.from(poolNames).sort(); - - if (clusterInfo?.poolAssignment) { - globalPoolAssignment.value = clusterInfo.poolAssignment; - } - - const projectId = workflowDocumentStore.value?.homeProject?.id; - if (projectId) { - try { - const projectSettings = await getProjectPoolSettings(rootStore.restApiContext, projectId); - projectPoolAssignment.value = projectSettings.assignment; - projectAllowedPools.value = projectSettings.allowedPools; - } catch { - // Fall back to instance defaults if project settings unavailable - } - } -}; - -const workerPoolOptionsForCategory = (category: WorkerPoolCategory) => { - const effectiveDefault = - projectPoolAssignment.value[category] ?? globalPoolAssignment.value[category]; - const poolName = - effectiveDefault || i18n.baseText('workflowSettings.workerPool.default').toLowerCase(); - const defaultLabel = i18n.baseText('workflowSettings.workerPool.defaultWithPool', { - interpolate: { poolName }, - }); - - let availablePools = workerPoolNames.value; - if (projectAllowedPools.value.length > 0) { - const allowed = new Set(projectAllowedPools.value); - availablePools = workerPoolNames.value.filter((name) => allowed.has(name)); - } - - const options = [ - { key: 'DEFAULT', value: defaultLabel }, - ...availablePools.map((name) => ({ key: name, value: name })), - ]; - - const currentOverride = workflowSettings.value[workerPoolSettingKeys[category]]; - if ( - currentOverride && - currentOverride !== 'DEFAULT' && - !availablePools.includes(currentOverride) - ) { - options.push({ - key: currentOverride, - value: `${currentOverride} (${i18n.baseText('workflowSettings.workerPool.notAllowed')})`, - }); - } - - return options; -}; - -const selectedWorkerPool = (category: WorkerPoolCategory) => { - return workflowSettings.value[workerPoolSettingKeys[category]] || 'DEFAULT'; -}; - -const onWorkerPoolChange = (category: WorkerPoolCategory, value: string) => { - const settingKey = workerPoolSettingKeys[category]; - - // Send empty string (not `delete`) so the backend's PATCH-merge of `settings` - // actually overwrites any previously-stored override. `removeDefaultValues` - // strips empty-string overrides before persisting to keep the JSON tidy. - workflowSettings.value[settingKey] = value === 'DEFAULT' ? '' : value; -}; const isSelectedResolverEditable = computed(() => { const resolverId = workflowSettings.value.credentialResolverId; @@ -977,10 +878,6 @@ onMounted(async () => { ); } - if (settingsStore.isQueueModeEnabled) { - promises.push(loadWorkerPoolOptions()); - } - await Promise.all(promises); } catch (error) { toast.showError(error, 'Problem loading settings', { @@ -1031,9 +928,6 @@ onMounted(async () => { if (workflowSettingsData.availableInMCP === undefined) { workflowSettingsData.availableInMCP = defaultValues.value.availableInMCP; } - if (workflowSettingsData.priority === undefined) { - workflowSettingsData.priority = 'default'; - } originalBinaryMode.value = workflowSettingsData.binaryMode; workflowSettings.value = workflowSettingsData; @@ -1753,125 +1647,6 @@ onBeforeUnmount(() => { - - - - - - -
- - - - -
- -