diff --git a/packages/twenty-server/src/engine/core-modules/message-queue/message-queue.explorer.ts b/packages/twenty-server/src/engine/core-modules/message-queue/message-queue.explorer.ts index d6bf9eb222..6030ccf8e6 100644 --- a/packages/twenty-server/src/engine/core-modules/message-queue/message-queue.explorer.ts +++ b/packages/twenty-server/src/engine/core-modules/message-queue/message-queue.explorer.ts @@ -18,9 +18,11 @@ import { type MessageQueueWorkerOptions } from 'src/engine/core-modules/message- import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; import { MessageQueueMetadataAccessor } from 'src/engine/core-modules/message-queue/message-queue-metadata.accessor'; import { type MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { type MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; +import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { QUEUE_WORKER_OPTIONS } from 'src/engine/core-modules/message-queue/message-queue-worker-options.constant'; import { getQueueToken } from 'src/engine/core-modules/message-queue/utils/get-queue-token.util'; +import { shouldCreateWorkerForQueue } from 'src/engine/core-modules/message-queue/utils/should-create-worker-for-queue.util'; +import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; import { shouldCaptureException } from 'src/engine/utils/global-exception-handler.util'; interface ProcessorGroup { @@ -41,6 +43,7 @@ export class MessageQueueExplorer implements OnModuleInit { private readonly metadataAccessor: MessageQueueMetadataAccessor, private readonly metadataScanner: MetadataScanner, private readonly exceptionHandlerService: ExceptionHandlerService, + private readonly twentyConfigService: TwentyConfigService, ) {} onModuleInit() { @@ -60,20 +63,63 @@ export class MessageQueueExplorer implements OnModuleInit { const groupedProcessors = this.groupProcessorsByQueueName(processors); - for (const [queueName, processorGroupCollection] of Object.entries( - groupedProcessors, - )) { + // Filter out empty entries: an explicit empty env value is parsed as [''] + // by the shared ARRAY transformer, which would otherwise turn an empty + // allowlist (meaning "all queues") into "no queues" + const enabledQueues = this.twentyConfigService + .get('WORKER_ENABLED_QUEUES') + .filter((queueName) => queueName.length > 0); + const excludedQueues = this.twentyConfigService + .get('WORKER_EXCLUDED_QUEUES') + .filter((queueName) => queueName.length > 0); + + this.warnAboutUnknownQueueNames([...enabledQueues, ...excludedQueues]); + + const groupedProcessorEntries = Object.entries(groupedProcessors) as [ + MessageQueue, + ProcessorGroup[], + ][]; + + for (const [ + queueName, + processorGroupCollection, + ] of groupedProcessorEntries) { + if ( + !shouldCreateWorkerForQueue({ + queueName, + enabledQueues, + excludedQueues, + }) + ) { + this.logger.log( + `Skipping worker creation for queue ${queueName} (filtered out by WORKER_ENABLED_QUEUES/WORKER_EXCLUDED_QUEUES)`, + ); + continue; + } + const queueToken = getQueueToken(queueName); const messageQueueService = this.getQueueService(queueToken); this.handleProcessorGroupCollection( processorGroupCollection, messageQueueService, - QUEUE_WORKER_OPTIONS[queueName as MessageQueue], + QUEUE_WORKER_OPTIONS[queueName], ); } } + private warnAboutUnknownQueueNames(queueNames: string[]) { + const knownQueueNames = Object.values(MessageQueue) as string[]; + + for (const queueName of queueNames) { + if (!knownQueueNames.includes(queueName)) { + this.logger.warn( + `Unknown queue name "${queueName}" in WORKER_ENABLED_QUEUES/WORKER_EXCLUDED_QUEUES, expected one of: ${knownQueueNames.join(', ')}`, + ); + } + } + } + private groupProcessorsByQueueName(processors: InstanceWrapper[]) { return processors.reduce( (acc, wrapper) => { diff --git a/packages/twenty-server/src/engine/core-modules/message-queue/utils/__tests__/should-create-worker-for-queue.util.spec.ts b/packages/twenty-server/src/engine/core-modules/message-queue/utils/__tests__/should-create-worker-for-queue.util.spec.ts new file mode 100644 index 0000000000..d9853d7a5e --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/message-queue/utils/__tests__/should-create-worker-for-queue.util.spec.ts @@ -0,0 +1,65 @@ +import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; +import { shouldCreateWorkerForQueue } from 'src/engine/core-modules/message-queue/utils/should-create-worker-for-queue.util'; + +describe('shouldCreateWorkerForQueue', () => { + it('should create a worker for every queue when no filters are set', () => { + expect( + shouldCreateWorkerForQueue({ + queueName: MessageQueue.workspaceQueue, + enabledQueues: [], + excludedQueues: [], + }), + ).toBe(true); + }); + + it('should only create a worker for queues in the enabled list', () => { + expect( + shouldCreateWorkerForQueue({ + queueName: MessageQueue.workspaceQueue, + enabledQueues: [MessageQueue.workspaceQueue], + excludedQueues: [], + }), + ).toBe(true); + + expect( + shouldCreateWorkerForQueue({ + queueName: MessageQueue.workflowQueue, + enabledQueues: [MessageQueue.workspaceQueue], + excludedQueues: [], + }), + ).toBe(false); + }); + + it('should not create a worker for excluded queues', () => { + expect( + shouldCreateWorkerForQueue({ + queueName: MessageQueue.workspaceQueue, + enabledQueues: [], + excludedQueues: [MessageQueue.workspaceQueue], + }), + ).toBe(false); + }); + + it('should create a worker when the queue passes the allowlist and is not in the denylist', () => { + expect( + shouldCreateWorkerForQueue({ + queueName: MessageQueue.workspaceQueue, + enabledQueues: [ + MessageQueue.workspaceQueue, + MessageQueue.workflowQueue, + ], + excludedQueues: [MessageQueue.aiQueue], + }), + ).toBe(true); + }); + + it('should apply the excluded list after the enabled list', () => { + expect( + shouldCreateWorkerForQueue({ + queueName: MessageQueue.workspaceQueue, + enabledQueues: [MessageQueue.workspaceQueue], + excludedQueues: [MessageQueue.workspaceQueue], + }), + ).toBe(false); + }); +}); diff --git a/packages/twenty-server/src/engine/core-modules/message-queue/utils/should-create-worker-for-queue.util.ts b/packages/twenty-server/src/engine/core-modules/message-queue/utils/should-create-worker-for-queue.util.ts new file mode 100644 index 0000000000..8f1e0015d1 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/message-queue/utils/should-create-worker-for-queue.util.ts @@ -0,0 +1,17 @@ +import { type MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; + +export const shouldCreateWorkerForQueue = ({ + queueName, + enabledQueues, + excludedQueues, +}: { + queueName: MessageQueue; + enabledQueues: string[]; + excludedQueues: string[]; +}): boolean => { + if (enabledQueues.length > 0 && !enabledQueues.includes(queueName)) { + return false; + } + + return !excludedQueues.includes(queueName); +}; diff --git a/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts b/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts index c282133799..78e7fb66f4 100644 --- a/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts +++ b/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts @@ -1261,6 +1261,26 @@ export class ConfigVariables { }) REDIS_QUEUE_URL: string; + @ConfigVariablesMetadata({ + group: ConfigVariablesGroup.ADVANCED_SETTINGS, + description: + 'Comma-separated list of queues this worker processes (e.g. workspace-queue). Empty means all queues. Used to dedicate worker pods to specific queues.', + isEnvOnly: true, + type: ConfigVariableType.ARRAY, + }) + @IsOptional() + WORKER_ENABLED_QUEUES: string[] = []; + + @ConfigVariablesMetadata({ + group: ConfigVariablesGroup.ADVANCED_SETTINGS, + description: + 'Comma-separated list of queues this worker does not process (e.g. workspace-queue). Applied after WORKER_ENABLED_QUEUES. Used to keep long-running queues off general-purpose worker pods.', + isEnvOnly: true, + type: ConfigVariableType.ARRAY, + }) + @IsOptional() + WORKER_EXCLUDED_QUEUES: string[] = []; + @ConfigVariablesMetadata({ group: ConfigVariablesGroup.SERVER_CONFIG, description: 'Node environment (development, production, etc.)',