refactor(core): prepare Worker Pools for rollout behind a flag (#32974)
Some checks are pending
CI: Python / Checks (push) Waiting to run

This commit is contained in:
Mike Repeć 2026-06-25 08:57:16 +02:00 committed by GitHub
parent 3eb2984659
commit 5be3ebe127
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
59 changed files with 530 additions and 1642 deletions

View File

@ -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

View File

@ -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

View File

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

View File

@ -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

View File

@ -20,7 +20,7 @@ CREATE TABLE "project" ("id" varchar(36) PRIMARY KEY NOT NULL, "name" varchar(25
| customTelemetryTags | TEXT | '[]' | false | | | |
| description | varchar(512) | | true | | | |
| icon | TEXT | | true | | | |
| id | varchar(36) | | false | [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

View File

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

View File

@ -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';

View File

@ -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(),
}) {}

View File

@ -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(),
}) {}

View File

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

View File

@ -70,7 +70,6 @@ export type ClusterCheckSummary = z.infer<typeof ClusterCheckSummarySchema>;
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

View File

@ -1,17 +1,6 @@
export type ExecutionCategory = 'production' | 'manual' | 'evaluation';
export type PoolAssignment = Partial<Record<ExecutionCategory, string>>;
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;
};

View File

@ -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);

View File

@ -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.

View File

@ -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';

View File

@ -369,6 +369,7 @@ describe('GlobalConfig', () => {
},
},
workerPool: {
enabled: false,
name: '',
},
},

View File

@ -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;
}

View File

@ -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',

View File

@ -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,
];

View File

@ -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 };

View File

@ -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<string, keyof ProjectPoolSettings>;
@Service()
export class ProjectPoolSettingsRepository extends Repository<ProjectPoolSettings> {
constructor(dataSource: DataSource) {
super(ProjectPoolSettings, dataSource.manager);
}
async getPoolForCategory(
projectId: string,
category: 'production' | 'manual' | 'evaluation',
): Promise<string | undefined> {
const column = CATEGORY_COLUMN[category];
const row = await this.findOne({
where: { projectId },
select: [column],
});
return row?.[column] ?? undefined;
async getDefaultPool(projectId: string): Promise<string | null> {
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<void> {
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<void> {
await this.upsert({ projectId, defaultPool }, ['projectId']);
}
}

View File

@ -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,

View File

@ -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<IWorkflowExecutionDataProcess>({
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', () => {

View File

@ -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<z.infer<typeof flagsSchema>> {
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');

View File

@ -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<License>();
const workerStatusService = mock<WorkerStatusService>();
const workerPoolsService = mock<WorkerPoolsService>();
const poolConfigService = mock<PoolConfigService>();
const controller = new OrchestrationController(
license,
workerStatusService,
workerPoolsService,
poolConfigService,
);
const controller = new OrchestrationController(license, workerStatusService);
const req = mock<AuthenticatedRequest>();
const res = mock<Response>();
const req = mock<AuthenticatedRequest>({ 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();
});
});
});

View File

@ -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'] });
});
});
});

View File

@ -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<WorkerPoolsResponse> {
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<PoolAssignment> {
return await this.poolConfigService.setPoolAssignment(dto);
}
}

View File

@ -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<InstanceRegistryService>;
let checkService: jest.Mocked<CheckService>;
let poolConfigService: jest.Mocked<PoolConfigService>;
beforeEach(() => {
service = mock<InstanceRegistryService>();
checkService = mock<CheckService>();
poolConfigService = mock<PoolConfigService>();
checkService.runChecks.mockResolvedValue({ currentState: new Map(), results: [] });
poolConfigService.getPoolAssignment.mockResolvedValue({});
controller = new InstanceRegistryController(service, checkService, poolConfigService);
controller = new InstanceRegistryController(service, checkService);
});
describe('getClusterInfo', () => {

View File

@ -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<InstanceSettings> = {}) =>
const makeExecutionsConfig = (mode: 'regular' | 'queue' = 'regular') =>
mock<ExecutionsConfig>({ mode });
const makeScalingModeConfig = (poolName: string = '') =>
mock<ScalingModeConfig>({ workerPool: { name: poolName } });
const makeScalingModeConfig = (poolName: string = '', enabled: boolean = true) => {
const workerPool = Object.assign(new WorkerPoolConfig(), { name: poolName, enabled });
return mock<ScalingModeConfig>({ workerPool });
};
const makeLogger = () => {
const logger = mock<Logger>();
@ -103,11 +106,12 @@ describe('InstanceRegistryService', () => {
settingsOverrides: Partial<InstanceSettings> = {},
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');

View File

@ -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<ClusterInfoResponse> {
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,
};
}
}

View File

@ -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;

View File

@ -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

View File

@ -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<SettingsRepository>();
const projectPoolSettingsRepository = mock<ProjectPoolSettingsRepository>();
const cacheService = mock<CacheService>();
const instanceRegistryService = mock<InstanceRegistryService>();
const workerPoolsService = mock<WorkerPoolsService>();
const globalConfig = mock<GlobalConfig>();
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);
});
});
});

View File

@ -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('');
});
});

View File

@ -1,5 +1,5 @@
import { mockLogger, mockInstance } from '@n8n/backend-test-utils';
import { GlobalConfig } from '@n8n/config';
import { GlobalConfig, WorkerPoolConfig } from '@n8n/config';
import type { ExecutionRepository } from '@n8n/db';
import { Container } from '@n8n/di';
import * as BullModule from 'bull';
@ -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';

View File

@ -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> = {}): InstanceRegistration => ({
@ -18,9 +17,8 @@ const buildWorker = (overrides: Partial<InstanceRegistration> = {}): InstanceReg
});
describe('WorkerPoolsService', () => {
const poolConfigService = mock<PoolConfigService>();
const instanceRegistryService = mock<InstanceRegistryService>();
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);
});
});
});

View File

@ -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;
}
}

View File

@ -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<string[]> {
const instances = await this.instanceRegistryService.getAllInstances();
const names = new Set<string>();
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<IWorkflowExecutionDataProcess, 'projectId'>,
): Promise<{ queueName: string; poolName: string | undefined }> {
const systemDefault = { queueName: DEFAULT_QUEUE_NAME, poolName: undefined };
async getQueueNameForCategory(category: ExecutionCategory): Promise<string> {
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<PoolAssignment> {
const cached = await this.cacheService.get<string>(SETTINGS_KEY);
if (cached !== undefined) {
return jsonParse<PoolAssignment>(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<PoolAssignment>(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<PoolAssignment> {
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<string | undefined> {
const cacheKey = `projectPool:${projectId}:${category}`;
const cached = await this.cacheService.get<string>(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<ProjectPoolSettingsResponse> {
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<string>(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<ProjectPoolSettingsResponse> {
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<void> {
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<string | undefined> {
const cacheKey = this.cacheKey(projectId);
const cached = await this.cacheService.get<string>(cacheKey);
if (cached !== undefined) return cached || undefined;
const pool = await this.projectPoolSettingsRepository.getDefaultPool(projectId);
void this.cacheService.set(cacheKey, pool ?? '');
return pool ?? undefined;
}
}

View File

@ -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}`;

View File

@ -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);

View File

@ -15,10 +15,6 @@ export type Job = Bull.Job<JobData>;
export type JobId = Job['id'];
export type ExecutionCategory = 'production' | 'manual' | 'evaluation';
export type PoolAssignment = Partial<Record<ExecutionCategory, string>>;
export type JobData = {
workflowId: string;
executionId: string;

View File

@ -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<string[]> {
const instances = await this.instanceRegistryService.getAllInstances();
@ -23,13 +18,4 @@ export class WorkerPoolsService {
return Array.from(poolNames).sort();
}
async getPoolsState(): Promise<WorkerPoolsResponse> {
const [pools, assignment] = await Promise.all([
this.getAvailablePools(),
this.poolConfigService.getPoolAssignment(),
]);
return { pools, assignment };
}
}

View File

@ -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),
),
};
}

View File

@ -77,6 +77,7 @@ describe('FrontendService', () => {
userManagement: {
password: { minLength: 8 },
},
queue: { workerPool: { enabled: false } },
});
const instanceSettings = mock<InstanceSettings>({

View File

@ -410,6 +410,9 @@ export class FrontendService {
folders: {
enabled: false,
},
workerPools: {
enabled: this.globalConfig.queue.workerPool.enabled,
},
evaluation: {
quota: this.licenseState.getMaxWorkflowsWithEvaluations(),
},

View File

@ -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;
}

View File

@ -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<typeof getExecutionCategory>,
): 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.
*

View File

@ -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.",

View File

@ -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<void> => {
await makeRestApiRequest(context, 'POST', GET_STATUS_ENDPOINT);
};
export const getWorkerPools = async (context: IRestApiContext): Promise<WorkerPoolsResponse> => {
return await makeRestApiRequest(context, 'GET', WORKER_POOLS_ENDPOINT);
};
export const updateWorkerPoolAssignment = async (
context: IRestApiContext,
dto: UpdateWorkerPoolAssignmentDto,
): Promise<PoolAssignment> => {
return await makeRestApiRequest(context, 'PATCH', WORKER_POOL_ASSIGNMENT_ENDPOINT, dto);
};
export const getProjectPoolSettings = async (
context: IRestApiContext,
projectId: string,

View File

@ -431,9 +431,6 @@ export interface IWorkflowSettings extends IWorkflowSettingsWorkflow {
executionOrder: NonNullable<IWorkflowSettingsWorkflow['executionOrder']>;
binaryMode?: WorkflowSettingsBinaryMode;
availableInMCP?: boolean;
workerPoolOverrideProduction?: string;
workerPoolOverrideManual?: string;
workerPoolOverrideEvaluation?: string;
}
export interface ITimeoutHMS {

View File

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

View File

@ -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<ITimeoutHMS>({ hours: 0, minutes: 0, seconds: 0 });
type WorkerPoolCategory = 'production' | 'manual' | 'evaluation';
type WorkerPoolSettingKey =
| 'workerPoolOverrideProduction'
| 'workerPoolOverrideManual'
| 'workerPoolOverrideEvaluation';
const workerPoolNames = ref<string[]>([]);
const globalPoolAssignment = ref<Partial<Record<WorkerPoolCategory, string>>>({});
const projectPoolAssignment = ref<Partial<Record<WorkerPoolCategory, string>>>({});
const projectAllowedPools = ref<string[]>([]);
const workerPoolSettingKeys = {
production: 'workerPoolOverrideProduction',
manual: 'workerPoolOverrideManual',
evaluation: 'workerPoolOverrideEvaluation',
} satisfies Record<WorkerPoolCategory, WorkerPoolSettingKey>;
const instanceRegistryStore = useInstanceRegistryStore();
const loadWorkerPoolOptions = async () => {
await instanceRegistryStore.fetchClusterInfo();
const clusterInfo = instanceRegistryStore.clusterInfo;
const instances = clusterInfo?.instances ?? [];
const poolNames = new Set<string>();
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(() => {
</div>
</ElCol>
</ElRow>
<template v-if="settingsStore.isQueueModeEnabled && workerPoolNames.length > 0">
<ElRow>
<ElCol :span="10" :class="$style['setting-name']">
{{ i18n.baseText('workflowSettings.workerPool.production') }}
<N8nTooltip placement="top">
<template #content>
<div
v-text="i18n.baseText('workflowSettings.workerPool.production.tooltip')"
></div>
</template>
<N8nIcon icon="circle-help" />
</N8nTooltip>
</ElCol>
<ElCol :span="14" class="ignore-key-press-canvas">
<N8nSelect
:model-value="selectedWorkerPool('production')"
:disabled="readOnlyEnv || !workflowPermissions.update"
:limit-popper-width="true"
data-test-id="workflow-settings-worker-pool-production"
@update:model-value="(val: string) => onWorkerPoolChange('production', val)"
>
<N8nOption
v-for="option of workerPoolOptionsForCategory('production')"
:key="option.key"
:label="option.value"
:value="option.key"
/>
</N8nSelect>
</ElCol>
</ElRow>
<ElRow>
<ElCol :span="10" :class="$style['setting-name']">
{{ i18n.baseText('workflowSettings.workerPool.manual') }}
<N8nTooltip placement="top">
<template #content>
<div v-text="i18n.baseText('workflowSettings.workerPool.manual.tooltip')"></div>
</template>
<N8nIcon icon="circle-help" />
</N8nTooltip>
</ElCol>
<ElCol :span="14" class="ignore-key-press-canvas">
<N8nSelect
:model-value="selectedWorkerPool('manual')"
:disabled="readOnlyEnv || !workflowPermissions.update"
:limit-popper-width="true"
data-test-id="workflow-settings-worker-pool-manual"
@update:model-value="(val: string) => onWorkerPoolChange('manual', val)"
>
<N8nOption
v-for="option of workerPoolOptionsForCategory('manual')"
:key="option.key"
:label="option.value"
:value="option.key"
/>
</N8nSelect>
</ElCol>
</ElRow>
<ElRow>
<ElCol :span="10" :class="$style['setting-name']">
{{ i18n.baseText('workflowSettings.workerPool.evaluation') }}
<N8nTooltip placement="top">
<template #content>
<div
v-text="i18n.baseText('workflowSettings.workerPool.evaluation.tooltip')"
></div>
</template>
<N8nIcon icon="circle-help" />
</N8nTooltip>
</ElCol>
<ElCol :span="14" class="ignore-key-press-canvas">
<N8nSelect
:model-value="selectedWorkerPool('evaluation')"
:disabled="readOnlyEnv || !workflowPermissions.update"
:limit-popper-width="true"
data-test-id="workflow-settings-worker-pool-evaluation"
@update:model-value="(val: string) => onWorkerPoolChange('evaluation', val)"
>
<N8nOption
v-for="option of workerPoolOptionsForCategory('evaluation')"
:key="option.key"
:label="option.value"
:value="option.key"
/>
</N8nSelect>
</ElCol>
</ElRow>
</template>
<ElRow data-test-id="workflow-settings-priority">
<ElCol :span="10" :class="$style['setting-name']">
<label for="priority">
{{ i18n.baseText('workflowSettings.priority') }}
<N8nTooltip placement="top">
<template #content>
{{ i18n.baseText('workflowSettings.priority.tooltip') }}
</template>
<N8nIcon icon="circle-help" />
</N8nTooltip>
</label>
</ElCol>
<ElCol :span="14">
<div class="ignore-key-press-canvas">
<N8nSelect
v-model="workflowSettings.priority"
:disabled="readOnlyEnv || !workflowPermissions.update"
data-test-id="workflow-settings-priority-select"
size="medium"
>
<N8nOption
:label="i18n.baseText('workflowSettings.priority.options.default')"
value="default"
/>
<N8nOption
:label="i18n.baseText('workflowSettings.priority.options.high')"
value="high"
/>
</N8nSelect>
</div>
</ElCol>
</ElRow>
<ElRow>
<ElCol :span="10" :class="$style['setting-name']">
<label for="timeSavedPerExecution">

View File

@ -183,6 +183,14 @@ export const useSettingsStore = defineStore(STORES.SETTINGS, () => {
return isOtelCustomSpanAttributesLicensed && isOtelModuleActive;
});
const isWorkerPoolsEnabled = computed(() => settings.value.workerPools?.enabled === true);
// Opt-in flag: the `node-tools-searcher` token must be listed in the backend
// `N8N_AGENTS_MODULES` env var for this to evaluate true.
const isAgentsNodeToolsFeatureEnabled = computed(() =>
isAgentModuleActive('node-tools-searcher'),
);
// Opt-in flag: requires `N8N_AGENTS_AI_SANDBOX_ENABLED=true` and
// `N8N_AGENTS_AI_SANDBOX_PROVIDER=daytona` on the backend.
const isAgentsKnowledgeBaseFeatureEnabled = computed(
@ -486,6 +494,8 @@ export const useSettingsStore = defineStore(STORES.SETTINGS, () => {
isDataTableFeatureEnabled,
isChatFeatureEnabled,
isOtelCustomSpanAttributesEnabled,
isWorkerPoolsEnabled,
isAgentsNodeToolsFeatureEnabled,
isAgentsKnowledgeBaseFeatureEnabled,
isPublicChatTriggerDisabled,
};

View File

@ -1,181 +1,50 @@
<script setup lang="ts">
import { computed, watch } from 'vue';
import type { ExecutionCategory, PoolAssignment } from '@n8n/api-types';
import { N8nOption, N8nSelect, N8nText } from '@n8n/design-system';
import { useI18n } from '@n8n/i18n';
const i18n = useI18n();
const props = defineProps<{
assignment: PoolAssignment;
allowedPools: string[];
defaultPool: string | null;
availablePools: string[];
instanceDefaults: PoolAssignment;
disabled?: boolean;
}>();
const emit = defineEmits<{
'update:assignment': [value: PoolAssignment];
'update:allowedPools': [value: string[]];
'update:isValid': [value: boolean];
'update:defaultPool': [value: string | null];
}>();
const CATEGORY_KEYS: ExecutionCategory[] = ['production', 'manual', 'evaluation'];
const onUpdateAssignment = (category: ExecutionCategory, value: string) => {
const next: PoolAssignment = { ...props.assignment };
if (value === '') {
delete next[category];
} else {
next[category] = value;
}
emit('update:assignment', next);
const onUpdate = (value: string) => {
emit('update:defaultPool', value === '' ? '' : value);
};
const onUpdateAllowedPools = (value: string[]) => {
// De-dup and trim whitespace from any new chips the user typed
const seen = new Set<string>();
const cleaned: string[] = [];
for (const raw of value) {
const trimmed = raw.trim();
if (!trimmed || seen.has(trimmed)) continue;
seen.add(trimmed);
cleaned.push(trimmed);
}
emit('update:allowedPools', cleaned);
};
const valueForDropdown = (category: ExecutionCategory): string => props.assignment[category] ?? '';
const inheritLabel = (category: ExecutionCategory): string => {
const instanceValue = props.instanceDefaults[category];
const base = i18n.baseText('projects.settings.workerPools.defaults.inherit');
return instanceValue ? `${base} (${instanceValue})` : base;
};
// Union of cluster-available pools + any pools the project has already opted in to.
// Sorted, deduped — used to populate the Allowed pools dropdown options so the admin
// sees both what's actually online and what they had previously configured.
const allowedPoolOptions = computed(() => {
const set = new Set<string>([...props.availablePools, ...props.allowedPools]);
return Array.from(set).sort();
});
// When the project has not set an allowlist, the per-category default dropdown should expose
// all cluster pools so the admin can pick anything. Otherwise restrict to the explicit allowlist.
const defaultRoutingOptions = computed(() =>
props.allowedPools.length > 0 ? [...props.allowedPools].sort() : [...props.availablePools].sort(),
);
const invalidCategories = computed(() => {
// Empty allowedPools = no restriction; defaults can be any pool.
if (props.allowedPools.length === 0) return [];
return CATEGORY_KEYS.filter((category) => {
const pool = props.assignment[category];
return pool !== undefined && pool !== '' && !props.allowedPools.includes(pool);
});
});
const isValid = computed(() => invalidCategories.value.length === 0);
const errorMessage = computed(() => {
if (isValid.value) return '';
const category = invalidCategories.value[0];
const pool = props.assignment[category] ?? '';
return i18n.baseText('projects.settings.workerPools.error.notInAllowedPools', {
interpolate: { category, pool },
});
});
watch(isValid, (value) => emit('update:isValid', value), { immediate: true });
</script>
<template>
<fieldset class="project-worker-pools" data-test-id="project-worker-pools-section">
<h3>
<label>{{ i18n.baseText('projects.settings.workerPools.title') }}</label>
<label for="projectDefaultPool">
{{ i18n.baseText('projects.settings.workerPools.title') }}
</label>
</h3>
<div :class="$style.subsection">
<label for="projectAllowedPools">
{{ i18n.baseText('projects.settings.workerPools.allowedPools.label') }}
</label>
<N8nSelect
id="projectAllowedPools"
:model-value="props.allowedPools"
multiple
filterable
allow-create
default-first-option
:disabled="props.disabled"
:placeholder="i18n.baseText('projects.settings.workerPools.allowedPools.placeholder')"
:class="$style.input"
data-test-id="project-allowed-pools-input"
@update:model-value="onUpdateAllowedPools"
>
<N8nOption v-for="pool in allowedPoolOptions" :key="pool" :label="pool" :value="pool" />
</N8nSelect>
<N8nText color="text-light" size="small" :class="$style.help">
{{ i18n.baseText('projects.settings.workerPools.allowedPools.help') }}
</N8nText>
</div>
<div :class="$style.subsection">
<N8nText tag="div" bold :class="$style.subheading">
{{ i18n.baseText('projects.settings.workerPools.defaults.title') }}
</N8nText>
<N8nText tag="div" color="text-light" size="small" :class="$style.subheadingHelp">
{{ i18n.baseText('projects.settings.workerPools.defaults.description') }}
</N8nText>
<div
v-for="category in CATEGORY_KEYS"
:key="category"
:class="$style.field"
:data-test-id="`project-pool-default-${category}`"
>
<label :for="`projectPoolDefault-${category}`">
{{ i18n.baseText(`projects.settings.workerPools.defaults.${category}`) }}
</label>
<N8nSelect
:id="`projectPoolDefault-${category}`"
:model-value="valueForDropdown(category)"
:disabled="props.disabled"
:class="$style.input"
@update:model-value="(v: string) => onUpdateAssignment(category, v)"
>
<N8nOption :label="inheritLabel(category)" :value="''" />
<N8nOption
v-for="pool in defaultRoutingOptions"
:key="pool"
:label="pool"
:value="pool"
/>
</N8nSelect>
</div>
<N8nText
v-if="!isValid"
color="danger"
size="small"
:class="$style.error"
data-test-id="project-worker-pools-error"
>
{{ errorMessage }}
</N8nText>
</div>
<N8nSelect
id="projectDefaultPool"
:model-value="props.defaultPool ?? ''"
:disabled="props.disabled"
:class="$style.input"
data-test-id="project-default-pool-input"
@update:model-value="onUpdate"
>
<N8nOption :label="i18n.baseText('projects.settings.workerPools.defaultQueue')" :value="''" />
<N8nOption v-for="pool in props.availablePools" :key="pool" :label="pool" :value="pool" />
</N8nSelect>
<N8nText color="text-light" size="small" :class="$style.help">
{{ i18n.baseText('projects.settings.workerPools.help') }}
</N8nText>
</fieldset>
</template>
<style module lang="scss">
.subsection {
margin-top: var(--spacing--md);
&:first-of-type {
margin-top: 0;
}
}
.input {
max-width: var(--project-field--width);
}
@ -184,60 +53,4 @@ watch(isValid, (value) => emit('update:isValid', value), { immediate: true });
display: block;
margin-top: var(--spacing--3xs);
}
.subheading {
font-size: var(--font-size--sm);
margin-bottom: var(--spacing--3xs);
}
.subheadingHelp {
margin-bottom: var(--spacing--sm);
}
.field {
margin-bottom: var(--spacing--sm);
&:last-of-type {
margin-bottom: 0;
}
}
.error {
display: block;
margin-top: var(--spacing--xs);
}
</style>
<style lang="scss">
// Apply n8n's tag tokens to Element Plus's internal `.el-tag` markup so the
// chips rendered for the Allowed pools multi-select match the rest of the app
// (e.g. Execution tags). Non-module + parent-class scoped so the overrides
// don't leak globally.
.project-worker-pools {
.el-tag,
.el-tag.el-tag--info {
display: inline-flex;
align-items: center;
gap: var(--spacing--4xs);
height: var(--tag--height);
padding: var(--tag--padding);
line-height: var(--tag--line-height);
color: var(--tag--color--text);
background-color: var(--tag--color--background);
border: 1px solid var(--tag--border-color);
border-radius: var(--tag--radius);
font-size: var(--tag--font-size);
.el-tag__close {
max-height: 14px;
max-width: 14px;
line-height: 14px;
&:hover {
background-color: transparent !important;
color: var(--color--primary--shade-1);
}
}
}
}
</style>

View File

@ -16,7 +16,6 @@ import ProjectRoleUpgradeDialog from '../components/ProjectRoleUpgradeDialog.vue
import ProjectMembersTable from '../components/ProjectMembersTable.vue';
import { useRolesStore } from '@/app/stores/roles.store';
import { ROLE } from '@n8n/api-types';
import type { PoolAssignment } from '@n8n/api-types';
import { useCloudPlanStore } from '@/app/stores/cloudPlan.store';
import { useSettingsStore } from '@/app/stores/settings.store';
import { useTelemetry } from '@/app/composables/useTelemetry';
@ -88,18 +87,15 @@ const resourceCounts = ref<ResourceCounts>({
const formData = ref<
Pick<Project, 'name' | 'description' | 'relations'> & {
customTelemetryTags: NonNullable<Project['customTelemetryTags']>;
assignment: PoolAssignment;
allowedPools: string[];
defaultPool: string | null;
}
>({
name: '',
description: '',
relations: [],
customTelemetryTags: [],
assignment: {},
allowedPools: [],
defaultPool: null,
});
const isWorkerPoolsValid = ref(true);
// Used to skip one watcher sync after targeted server updates (e.g., immediate removal)
const suppressNextSync = ref(false);
@ -288,26 +284,14 @@ const resetFormData = () => {
? deepCopy(projectsStore.currentProject.customTelemetryTags)
: [];
telemetryTagsRef.value?.resetTouched();
formData.value.assignment = projectsStore.currentProjectPoolSettings
? { ...projectsStore.currentProjectPoolSettings.assignment }
: {};
formData.value.allowedPools = projectsStore.currentProjectPoolSettings
? [...projectsStore.currentProjectPoolSettings.allowedPools]
: [];
formData.value.defaultPool = projectsStore.currentProjectPoolSettings?.defaultPool ?? null;
};
const hasWorkerPoolsChanges = (): boolean => {
const stored = projectsStore.currentProjectPoolSettings;
if (!stored) {
// No row exists; consider any non-default formData as a change
return (
Object.keys(formData.value.assignment).length > 0 || formData.value.allowedPools.length > 0
);
}
return (
JSON.stringify(formData.value.assignment) !== JSON.stringify(stored.assignment) ||
JSON.stringify(formData.value.allowedPools) !== JSON.stringify(stored.allowedPools)
);
const stored = projectsStore.currentProjectPoolSettings?.defaultPool || null;
// Treat empty string and null as equivalent (both mean "default queue")
const current = formData.value.defaultPool || null;
return current !== stored;
};
const onCancel = () => {
@ -395,11 +379,10 @@ const updateProject = async () => {
}),
);
if (hasWorkerPoolsChanges()) {
if (isWorkerPoolsEnabled.value && hasWorkerPoolsChanges()) {
tasks.push(
projectsStore.updateCurrentProjectPoolSettings(projectId, {
assignment: formData.value.assignment,
allowedPools: formData.value.allowedPools,
defaultPool: formData.value.defaultPool ?? '',
}),
);
}
@ -413,7 +396,7 @@ const updateProject = async () => {
};
const onSubmit = async () => {
if (!isDirty.value || !isWorkerPoolsValid.value) {
if (!isDirty.value) {
return;
}
try {
@ -481,7 +464,15 @@ const onIconUpdated = async () => {
}
};
// Skip one sync after targeted updates (e.g. removal) to preserve unsaved edits
const isWorkerPoolsEnabled = computed(() => settingsStore.isWorkerPoolsEnabled);
const resetPoolSettingsFormData = () => {
formData.value.defaultPool = projectsStore.currentProjectPoolSettings?.defaultPool ?? null;
};
// Skip one sync after targeted updates (e.g. removal) to preserve unsaved edits.
// Also the single hook reacting to the project becoming available (it loads asynchronously on a
// direct page load), so project-scoped data like the pool settings is fetched here too.
watch(
() => projectsStore.currentProject,
async () => {
@ -495,6 +486,16 @@ watch(
if (projectsStore.currentProject?.icon && isIconOrEmoji(projectsStore.currentProject.icon)) {
projectIcon.value = projectsStore.currentProject.icon;
}
const projectId = projectsStore.currentProjectId;
if (canUpdateProject.value && projectId && isWorkerPoolsEnabled.value) {
await projectsStore
.fetchProjectPoolSettings(projectId)
.catch(() => {
// Non-fatal; store has been cleared, fall back to defaults below.
})
.finally(resetPoolSettingsFormData);
}
},
{ immediate: true },
);
@ -599,30 +600,9 @@ const searchUsers = async (query: string) => {
const debouncedUserSearch = useDebounceFn(searchUsers, getDebounceTime(DEBOUNCE_TIME.INPUT.SEARCH));
const resetPoolSettingsFormData = () => {
formData.value.assignment = projectsStore.currentProjectPoolSettings
? { ...projectsStore.currentProjectPoolSettings.assignment }
: {};
formData.value.allowedPools = projectsStore.currentProjectPoolSettings
? [...projectsStore.currentProjectPoolSettings.allowedPools]
: [];
};
onBeforeMount(async () => {
if (!canUpdateProject.value) return;
const projectId = projectsStore.currentProjectId;
const tasks: Array<Promise<unknown>> = [searchUsers('')];
if (projectId) {
tasks.push(
projectsStore
.fetchProjectPoolSettings(projectId)
.catch(() => {
// Non-fatal; store has been cleared, fall back to defaults below.
})
.finally(resetPoolSettingsFormData),
);
}
await Promise.all(tasks);
await searchUsers('');
});
const isProjectRoleProvisioningEnabled = computed(
@ -662,7 +642,7 @@ onMounted(async () => {
>{{ i18n.baseText('projects.settings.button.cancel') }}</N8nButton
>
<N8nButton
:disabled="!isValid || !isDirty || !isWorkerPoolsValid"
:disabled="!isValid || !isDirty"
variant="solid"
data-test-id="project-settings-save-button"
@click.stop.prevent="onSubmit"
@ -807,23 +787,15 @@ onMounted(async () => {
/>
</fieldset>
<ProjectWorkerPoolsSection
:assignment="formData.assignment"
:allowed-pools="formData.allowedPools"
v-if="isWorkerPoolsEnabled"
:default-pool="formData.defaultPool"
:available-pools="projectsStore.currentProjectPoolSettings?.availablePools ?? []"
:instance-defaults="projectsStore.currentProjectPoolSettings?.instanceDefaults ?? {}"
@update:assignment="
@update:default-pool="
(v) => {
formData.assignment = v;
formData.defaultPool = v;
onTextInput();
}
"
@update:allowed-pools="
(v) => {
formData.allowedPools = v;
onTextInput();
}
"
@update:is-valid="isWorkerPoolsValid = $event"
/>
<fieldset>

View File

@ -10,7 +10,6 @@ import { usePushConnectionStore } from '@/app/stores/pushConnection.store';
import { useRootStore } from '@n8n/stores/useRootStore';
import { useTelemetry } from '@/app/composables/useTelemetry';
import WorkerCard from './WorkerCard.vue';
import WorkerPoolsCard from './WorkerPoolsCard.vue';
import { N8nHeading, N8nSpinner } from '@n8n/design-system';
import ConnectionTracker from '@/app/components/ConnectionTracker.vue';
@ -51,12 +50,10 @@ onBeforeMount(() => {
pushConnection.initialize();
pushStore.pushConnect();
orchestrationManagerStore.startWorkerStatusPolling();
orchestrationManagerStore.startWorkerPoolsPolling();
});
onBeforeUnmount(() => {
orchestrationManagerStore.stopWorkerStatusPolling();
orchestrationManagerStore.stopWorkerPoolsPolling();
pushStore.pushDisconnect();
pushConnection.terminate();
});
@ -78,9 +75,6 @@ onBeforeUnmount(() => {
<WorkerCard :worker-id="workerId" data-test-id="worker-card" />
</div>
</div>
<div :class="$style.card">
<WorkerPoolsCard />
</div>
</div>
</div>
</template>

View File

@ -1,232 +0,0 @@
<script setup lang="ts">
import { computed, onMounted, ref } from 'vue';
import type { ExecutionCategory, PoolAssignment } from '@n8n/api-types';
import {
N8nBadge,
N8nButton,
N8nCard,
N8nHeading,
N8nOption,
N8nSelect,
N8nText,
} from '@n8n/design-system';
import { useI18n } from '@n8n/i18n';
import { useToast } from '@/app/composables/useToast';
import { useOrchestrationStore } from '../orchestration.store';
const DEFAULT_POOL_VALUE = '';
const i18n = useI18n();
const toast = useToast();
const orchestrationStore = useOrchestrationStore();
const executionCategories: ExecutionCategory[] = ['production', 'manual', 'evaluation'];
const formState = ref<Record<ExecutionCategory, string>>({
production: DEFAULT_POOL_VALUE,
manual: DEFAULT_POOL_VALUE,
evaluation: DEFAULT_POOL_VALUE,
});
const isLoading = ref(true);
const isSaving = ref(false);
const availablePools = computed(() => orchestrationStore.availablePools);
const poolOptions = computed(() => [
{ label: i18n.baseText('workerList.pools.defaults.unassigned'), value: DEFAULT_POOL_VALUE },
...availablePools.value.map((pool) => ({ label: pool, value: pool })),
]);
const persistedFormState = computed<Record<ExecutionCategory, string>>(() => {
const assignment = orchestrationStore.poolAssignment ?? {};
return {
production: assignment.production ?? DEFAULT_POOL_VALUE,
manual: assignment.manual ?? DEFAULT_POOL_VALUE,
evaluation: assignment.evaluation ?? DEFAULT_POOL_VALUE,
};
});
const hasChanges = computed(() => {
const persisted = persistedFormState.value;
return executionCategories.some((category) => formState.value[category] !== persisted[category]);
});
const saveDisabled = computed(() => !hasChanges.value || isSaving.value);
onMounted(async () => {
try {
await orchestrationStore.fetchWorkerPools();
formState.value = { ...persistedFormState.value };
} catch (error) {
toast.showError(error, i18n.baseText('workerList.pools.loading.error'));
} finally {
isLoading.value = false;
}
});
async function onSave() {
const persisted = persistedFormState.value;
const dto: PoolAssignment = {};
for (const category of executionCategories) {
if (formState.value[category] !== persisted[category]) {
dto[category] = formState.value[category];
}
}
isSaving.value = true;
try {
await orchestrationStore.updateWorkerPoolAssignment(dto);
formState.value = { ...persistedFormState.value };
toast.showMessage({
title: i18n.baseText('workerList.pools.defaults.saved.success'),
type: 'success',
});
} catch (error) {
toast.showError(error, i18n.baseText('workerList.pools.defaults.saved.error'));
} finally {
isSaving.value = false;
}
}
</script>
<template>
<N8nCard v-if="!isLoading" data-test-id="worker-pools-card" :class="$style.card">
<template #header>
<N8nHeading tag="h2" size="medium" bold>
{{ i18n.baseText('workerList.pools.title') }}
</N8nHeading>
</template>
<div :class="$style.section">
<N8nText tag="div" bold size="small" :class="$style.sectionTitle">
{{ i18n.baseText('workerList.pools.available.title') }}
</N8nText>
<div v-if="availablePools.length === 0" :class="$style.empty">
<N8nText color="text-light" size="small">
{{ i18n.baseText('workerList.pools.available.empty') }}
</N8nText>
</div>
<div v-else :class="$style.badges">
<N8nBadge
v-for="pool in availablePools"
:key="pool"
theme="tertiary"
data-test-id="worker-pool-badge"
>
{{ pool }}
</N8nBadge>
</div>
</div>
<div :class="$style.section">
<N8nText tag="div" bold size="small" :class="$style.sectionTitle">
{{ i18n.baseText('workerList.pools.defaults.title') }}
</N8nText>
<N8nText color="text-light" size="small" :class="$style.sectionDescription">
{{ i18n.baseText('workerList.pools.defaults.description') }}
</N8nText>
<div :class="$style.field" data-test-id="worker-pool-default-production">
<label>{{ i18n.baseText('workerList.pools.defaults.production') }}</label>
<N8nSelect v-model="formState.production" size="medium">
<N8nOption
v-for="option in poolOptions"
:key="option.value"
:label="option.label"
:value="option.value"
/>
</N8nSelect>
</div>
<div :class="$style.field" data-test-id="worker-pool-default-manual">
<label>{{ i18n.baseText('workerList.pools.defaults.manual') }}</label>
<N8nSelect v-model="formState.manual" size="medium">
<N8nOption
v-for="option in poolOptions"
:key="option.value"
:label="option.label"
:value="option.value"
/>
</N8nSelect>
</div>
<div :class="$style.field" data-test-id="worker-pool-default-evaluation">
<label>{{ i18n.baseText('workerList.pools.defaults.evaluation') }}</label>
<N8nSelect v-model="formState.evaluation" size="medium">
<N8nOption
v-for="option in poolOptions"
:key="option.value"
:label="option.label"
:value="option.value"
/>
</N8nSelect>
</div>
<div :class="$style.actions">
<N8nButton
:disabled="saveDisabled"
:loading="isSaving"
data-test-id="worker-pools-save"
@click="onSave"
>
{{ i18n.baseText('workerList.pools.defaults.save') }}
</N8nButton>
</div>
</div>
</N8nCard>
</template>
<style module lang="scss">
.card {
padding: var(--spacing--sm);
}
.section {
display: flex;
flex-direction: column;
gap: var(--spacing--3xs);
padding: var(--spacing--sm) 0;
&:not(:last-child) {
border-bottom: var(--border--base);
}
}
.sectionTitle {
margin-bottom: var(--spacing--3xs);
}
.sectionDescription {
margin-bottom: var(--spacing--xs);
}
.badges {
display: flex;
flex-wrap: wrap;
gap: var(--spacing--2xs);
}
.empty {
padding: var(--spacing--2xs) 0;
}
.field {
display: flex;
flex-direction: column;
gap: var(--spacing--3xs);
margin-bottom: var(--spacing--xs);
label {
font-size: var(--font-size--xs);
color: var(--color--text--shade-1);
}
}
.actions {
display: flex;
justify-content: flex-end;
margin-top: var(--spacing--xs);
}
</style>

View File

@ -1,16 +1,11 @@
import { defineStore } from 'pinia';
import type { PoolAssignment, UpdateWorkerPoolAssignmentDto, WorkerStatus } from '@n8n/api-types';
import type { WorkerStatus } from '@n8n/api-types';
import { useRootStore } from '@n8n/stores/useRootStore';
import {
getWorkerPools,
sendGetWorkerStatus,
updateWorkerPoolAssignment,
} from '@n8n/rest-api-client/api/orchestration';
import { sendGetWorkerStatus } from '@n8n/rest-api-client/api/orchestration';
export const WORKER_HISTORY_LENGTH = 100;
const STALE_SECONDS = 120 * 1000;
const WORKER_POOLS_POLL_INTERVAL_MS = 10_000;
export interface IOrchestrationStoreState {
initialStatusReceived: boolean;
@ -20,9 +15,6 @@ export interface IOrchestrationStoreState {
};
workersLastUpdated: { [id: string]: number };
statusInterval: NodeJS.Timeout | null;
availablePools: string[];
poolAssignment: PoolAssignment | null;
workerPoolsInterval: NodeJS.Timeout | null;
}
export interface IWorkerHistoryItem {
@ -37,9 +29,6 @@ export const useOrchestrationStore = defineStore('orchestrationManager', {
workersHistory: {},
workersLastUpdated: {},
statusInterval: null,
availablePools: [],
poolAssignment: null,
workerPoolsInterval: null,
}),
actions: {
updateWorkerStatus(data: WorkerStatus) {
@ -88,32 +77,5 @@ export const useOrchestrationStore = defineStore('orchestrationManager', {
getWorkerStatusHistory(workerId: string): IWorkerHistoryItem[] {
return this.workersHistory[workerId] ?? [];
},
async fetchWorkerPools() {
const rootStore = useRootStore();
const response = await getWorkerPools(rootStore.restApiContext);
this.availablePools = response.pools;
this.poolAssignment = response.assignment;
},
startWorkerPoolsPolling() {
if (this.workerPoolsInterval) return;
this.workerPoolsInterval = setInterval(async () => {
try {
await this.fetchWorkerPools();
} catch {
// Swallow errors during polling — surfacing a toast every 10s would be noisy.
// The initial fetch on mount surfaces errors to the user.
}
}, WORKER_POOLS_POLL_INTERVAL_MS);
},
stopWorkerPoolsPolling() {
if (this.workerPoolsInterval) {
clearInterval(this.workerPoolsInterval);
this.workerPoolsInterval = null;
}
},
async updateWorkerPoolAssignment(dto: UpdateWorkerPoolAssignmentDto) {
const rootStore = useRootStore();
this.poolAssignment = await updateWorkerPoolAssignment(rootStore.restApiContext, dto);
},
},
});

View File

@ -3464,10 +3464,6 @@ export interface IWorkflowSettings {
credentialResolverId?: string;
redactionPolicy?: WorkflowSettings.RedactionPolicy;
customTelemetryTags?: ICustomTelemetryTag[];
workerPoolOverrideProduction?: string;
workerPoolOverrideManual?: string;
workerPoolOverrideEvaluation?: string;
priority?: 'default' | 'high';
}
export interface WorkflowFEMeta {