From 85ccf143becb651994c9e939b63ed294e2df4a5e Mon Sep 17 00:00:00 2001 From: Maximiliano Salvatti <40447063+msalvatti@users.noreply.github.com> Date: Thu, 9 Jul 2026 18:42:01 -0300 Subject: [PATCH] fix(prefix): propagate the configured key prefix to Worker, QueueEvents, and FlowProducer Only the producer Queue received options.prefix; the Worker (WorkerRegistry), QueueEvents (QueueEventsRegistry), and FlowProducer (FlowService) were created without it and fell back to BullMQ's default "bull" prefix. With any non-default QUEUE_PREFIX, producers and consumers diverged onto separate keyspaces, so workers never consumed jobs, event listeners never fired, and flow jobs were enqueued where no worker polled. Pass the resolved prefix into all three: WorkerRegistry.buildBullOptions, QueueEventsRegistry (now injects BYMAX_QUEUE_RESOLVED_OPTIONS), and FlowService (new prefix constructor arg, wired from both module factories). Add regression tests asserting each object is constructed with the configured prefix. --- src/server/bymax-queue.module.ts | 4 +- src/server/services/flow.service.spec.ts | 35 ++++++++++------- src/server/services/flow.service.ts | 10 ++++- .../queue-events-registry.service.spec.ts | 38 ++++++++++++++----- .../services/queue-events-registry.service.ts | 9 ++++- .../services/worker-registry.service.spec.ts | 9 +++++ .../services/worker-registry.service.ts | 3 ++ 7 files changed, 80 insertions(+), 28 deletions(-) diff --git a/src/server/bymax-queue.module.ts b/src/server/bymax-queue.module.ts index 929690b..ff2e2e6 100644 --- a/src/server/bymax-queue.module.ts +++ b/src/server/bymax-queue.module.ts @@ -83,7 +83,7 @@ export class BymaxQueueModule extends ConfigurableModuleClass { { provide: FlowService, useFactory: (conn: ConnectionResolver): FlowService => - new FlowService(conn, resolved.flows.enabled, resolved.telemetry), + new FlowService(conn, resolved.flows.enabled, resolved.prefix, resolved.telemetry), inject: [ConnectionResolver], }, { @@ -128,7 +128,7 @@ export class BymaxQueueModule extends ConfigurableModuleClass { { provide: FlowService, useFactory: (conn: ConnectionResolver, resolved: ResolvedQueueOptions): FlowService => - new FlowService(conn, resolved.flows.enabled, resolved.telemetry), + new FlowService(conn, resolved.flows.enabled, resolved.prefix, resolved.telemetry), inject: [ConnectionResolver, BYMAX_QUEUE_RESOLVED_OPTIONS], }, { diff --git a/src/server/services/flow.service.spec.ts b/src/server/services/flow.service.spec.ts index 8ca3e75..3bcf710 100644 --- a/src/server/services/flow.service.spec.ts +++ b/src/server/services/flow.service.spec.ts @@ -53,7 +53,7 @@ beforeEach(() => { describe('FlowService — enabled', () => { it('constructs a FlowProducer on the main connection', () => { // When enabled, the producer is built with the resolver's Queue-role client. - new FlowService(makeConnection(), true) + new FlowService(makeConnection(), true, 'bull') expect(producerConstructorArgs).toHaveLength(1) const [opts] = producerConstructorArgs[0] as [{ connection: unknown }] @@ -62,7 +62,7 @@ describe('FlowService — enabled', () => { it('delegates add to the underlying producer', async () => { // add forwards the flow definition unchanged and returns the producer result. - const service = new FlowService(makeConnection(), true) + const service = new FlowService(makeConnection(), true, 'bull') const flow = { name: 'root', queueName: 'pdf', data: {} } as FlowJob const node = { job: { id: '1' } } as unknown as JobNode producerInstances[0]?.add.mockResolvedValue(node) @@ -73,7 +73,7 @@ describe('FlowService — enabled', () => { it('delegates addBulk to the underlying producer as a mutable array', async () => { // addBulk copies the readonly input into the array BullMQ expects. - const service = new FlowService(makeConnection(), true) + const service = new FlowService(makeConnection(), true, 'bull') const flows = [{ name: 'a', queueName: 'q', data: {} }] as FlowJob[] const nodes = [{ job: { id: '1' } }] as unknown as JobNode[] producerInstances[0]?.addBulk.mockResolvedValue(nodes) @@ -84,7 +84,7 @@ describe('FlowService — enabled', () => { it('returns the underlying producer from getProducer', () => { // getProducer exposes the constructed FlowProducer instance. - const service = new FlowService(makeConnection(), true) + const service = new FlowService(makeConnection(), true, 'bull') expect(service.getProducer()).toBe(producerInstances[0]) }) @@ -92,15 +92,24 @@ describe('FlowService — enabled', () => { it('passes the configured telemetry into the FlowProducer constructor', () => { // A configured telemetry instance reaches the producer so spans propagate to children. const telemetry = { name: 'sentinel-telemetry' } as unknown as Telemetry - new FlowService(makeConnection(), true, telemetry) + new FlowService(makeConnection(), true, 'bull', telemetry) const [opts] = producerConstructorArgs[0] as [{ telemetry?: unknown }] expect(opts.telemetry).toBe(telemetry) }) + it('passes the configured prefix into the FlowProducer constructor', () => { + // The producer must use the same key prefix as the Queue so flow jobs land + // in the keyspace the workers poll; a non-default prefix breaks otherwise. + new FlowService(makeConnection(), true, 'tenant:flows') + + const [opts] = producerConstructorArgs[0] as [{ prefix?: unknown }] + expect(opts.prefix).toBe('tenant:flows') + }) + it('omits the telemetry key when telemetry is not configured', () => { // Without telemetry, the producer options never carry the key. - new FlowService(makeConnection(), true) + new FlowService(makeConnection(), true, 'bull') const [opts] = producerConstructorArgs[0] as [Record] expect('telemetry' in opts).toBe(false) @@ -110,14 +119,14 @@ describe('FlowService — enabled', () => { describe('FlowService — disabled', () => { it('does not construct a FlowProducer', () => { // When disabled, no producer is built (no Redis connection is opened). - new FlowService(makeConnection(), false) + new FlowService(makeConnection(), false, 'bull') expect(producerConstructorArgs).toHaveLength(0) }) it('throws FLOW_DISABLED (503) from add', async () => { // add is guarded and rejects with the typed disabled error. - const service = new FlowService(makeConnection(), false) + const service = new FlowService(makeConnection(), false, 'bull') await expect(service.add({} as FlowJob)).rejects.toBeInstanceOf(QueueException) await service.add({} as FlowJob).catch((err: unknown) => { @@ -128,14 +137,14 @@ describe('FlowService — disabled', () => { it('throws FLOW_DISABLED from addBulk', async () => { // addBulk is guarded the same way as add. - const service = new FlowService(makeConnection(), false) + const service = new FlowService(makeConnection(), false, 'bull') await expect(service.addBulk([])).rejects.toBeInstanceOf(QueueException) }) it('throws FLOW_DISABLED from getProducer', () => { // getProducer cannot return a producer that was never created. - const service = new FlowService(makeConnection(), false) + const service = new FlowService(makeConnection(), false, 'bull') expect(() => service.getProducer()).toThrow(QueueException) }) @@ -144,7 +153,7 @@ describe('FlowService — disabled', () => { describe('FlowService — onModuleDestroy', () => { it('closes the producer when active', async () => { // Shutdown closes the active producer a single time. - const service = new FlowService(makeConnection(), true) + const service = new FlowService(makeConnection(), true, 'bull') await service.onModuleDestroy() @@ -153,7 +162,7 @@ describe('FlowService — onModuleDestroy', () => { it('swallows a close rejection', async () => { // A failing close must not abort the shutdown sequence. - const service = new FlowService(makeConnection(), true) + const service = new FlowService(makeConnection(), true, 'bull') producerInstances[0]?.close.mockRejectedValue(new Error('close failed')) await expect(service.onModuleDestroy()).resolves.toBeUndefined() @@ -161,7 +170,7 @@ describe('FlowService — onModuleDestroy', () => { it('is a no-op when inactive', async () => { // With no producer there is nothing to close. - const service = new FlowService(makeConnection(), false) + const service = new FlowService(makeConnection(), false, 'bull') await expect(service.onModuleDestroy()).resolves.toBeUndefined() expect(producerInstances).toHaveLength(0) diff --git a/src/server/services/flow.service.ts b/src/server/services/flow.service.ts index 357595a..ecbb3eb 100644 --- a/src/server/services/flow.service.ts +++ b/src/server/services/flow.service.ts @@ -33,10 +33,18 @@ import { QUEUE_ERROR_CODES } from '../constants/error-codes' export class FlowService implements OnModuleDestroy { private readonly producer?: FlowProducer - constructor(connection: ConnectionResolver, enabled: boolean, telemetry?: Telemetry) { + constructor( + connection: ConnectionResolver, + enabled: boolean, + prefix: string, + telemetry?: Telemetry, + ) { if (enabled) { this.producer = new FlowProducer({ connection: connection.getClient(), + // Match the producer Queue's key prefix so flow jobs land in the same + // keyspace the workers poll; a non-default prefix breaks flows otherwise. + prefix, ...(telemetry ? { telemetry } : {}), }) } diff --git a/src/server/services/queue-events-registry.service.spec.ts b/src/server/services/queue-events-registry.service.spec.ts index df33094..3300742 100644 --- a/src/server/services/queue-events-registry.service.spec.ts +++ b/src/server/services/queue-events-registry.service.spec.ts @@ -6,6 +6,7 @@ import { QueueEventsRegistry } from './queue-events-registry.service' import type { ConnectionResolver } from './connection-resolver.service' +import type { ResolvedQueueOptions } from '../config/resolved-options' import type { Redis } from 'ioredis' /** Track instances created by the mocked constructor. */ @@ -37,6 +38,11 @@ function fakeConnection(redis?: Redis): ConnectionResolver { } as unknown as ConnectionResolver } +/** Minimal resolved-options stub carrying just the key prefix. */ +function fakeOptions(prefix = 'bull'): ResolvedQueueOptions { + return { prefix } as unknown as ResolvedQueueOptions +} + beforeEach(() => { createdQueueEvents.length = 0 }) @@ -44,7 +50,7 @@ beforeEach(() => { describe('QueueEventsRegistry.getOrCreate', () => { it('creates a QueueEvents on first call and returns it', () => { // First call must instantiate and cache a QueueEvents for the queue. - const registry = new QueueEventsRegistry(fakeConnection()) + const registry = new QueueEventsRegistry(fakeConnection(), fakeOptions()) const qe = registry.getOrCreate('email') expect(qe).toBeDefined() expect(createdQueueEvents).toHaveLength(1) @@ -55,10 +61,22 @@ describe('QueueEventsRegistry.getOrCreate', () => { // SUBSCRIBE needs maxRetriesPerRequest: null), not an empty/default options bag. const { QueueEvents: MockQueueEvents } = jest.requireMock<{ QueueEvents: jest.Mock }>('bullmq') const redis = fakeRedis() - const registry = new QueueEventsRegistry(fakeConnection(redis)) + const registry = new QueueEventsRegistry(fakeConnection(redis), fakeOptions()) registry.getOrCreate('email') const dup = (redis.duplicate as jest.Mock).mock.results[0]!.value as unknown - expect(MockQueueEvents).toHaveBeenCalledWith('email', { connection: dup }) + expect(MockQueueEvents).toHaveBeenCalledWith('email', { connection: dup, prefix: 'bull' }) + }) + + it('constructs QueueEvents with the configured key prefix', () => { + // QueueEvents must watch the same prefixed keyspace the producer writes to; + // a non-default prefix would otherwise silently drop every event. + const { QueueEvents: MockQueueEvents } = jest.requireMock<{ QueueEvents: jest.Mock }>('bullmq') + const registry = new QueueEventsRegistry(fakeConnection(), fakeOptions('tenant:events')) + registry.getOrCreate('email') + expect(MockQueueEvents).toHaveBeenCalledWith( + 'email', + expect.objectContaining({ prefix: 'tenant:events' }), + ) }) it('disconnects the duplicated connection when QueueEvents constructor throws (no leak)', () => { @@ -70,7 +88,7 @@ describe('QueueEventsRegistry.getOrCreate', () => { throw new Error('subscribe failed') }) const redis = fakeRedis() - const registry = new QueueEventsRegistry(fakeConnection(redis)) + const registry = new QueueEventsRegistry(fakeConnection(redis), fakeOptions()) expect(() => registry.getOrCreate('email')).toThrow('subscribe failed') const dupResult = (redis.duplicate as jest.Mock).mock.results[0]!.value as { disconnect: jest.Mock @@ -80,7 +98,7 @@ describe('QueueEventsRegistry.getOrCreate', () => { it('returns the same instance on subsequent calls (idempotent)', () => { // Idempotency is critical — no second Redis connection must be opened. - const registry = new QueueEventsRegistry(fakeConnection()) + const registry = new QueueEventsRegistry(fakeConnection(), fakeOptions()) const first = registry.getOrCreate('email') const second = registry.getOrCreate('email') expect(first).toBe(second) @@ -89,7 +107,7 @@ describe('QueueEventsRegistry.getOrCreate', () => { it('creates separate instances for different queues', () => { // Each queue requires its own QueueEvents (and its own Redis connection). - const registry = new QueueEventsRegistry(fakeConnection()) + const registry = new QueueEventsRegistry(fakeConnection(), fakeOptions()) const emailQe = registry.getOrCreate('email') const smsQe = registry.getOrCreate('sms') expect(emailQe).not.toBe(smsQe) @@ -100,7 +118,7 @@ describe('QueueEventsRegistry.getOrCreate', () => { describe('QueueEventsRegistry.list / getAll', () => { it('list returns the names of queues with an open QueueEvents', () => { // list() surfaces which queues currently have a QueueEvents connection. - const registry = new QueueEventsRegistry(fakeConnection()) + const registry = new QueueEventsRegistry(fakeConnection(), fakeOptions()) registry.getOrCreate('email') registry.getOrCreate('sms') expect(registry.list()).toEqual(expect.arrayContaining(['email', 'sms'])) @@ -108,7 +126,7 @@ describe('QueueEventsRegistry.list / getAll', () => { it('getAll returns a ReadonlyMap of all QueueEvents', () => { // getAll() is used by the shutdown orchestrator. - const registry = new QueueEventsRegistry(fakeConnection()) + const registry = new QueueEventsRegistry(fakeConnection(), fakeOptions()) registry.getOrCreate('email') const map = registry.getAll() expect(map.size).toBe(1) @@ -119,7 +137,7 @@ describe('QueueEventsRegistry.list / getAll', () => { describe('QueueEventsRegistry.getConnections', () => { it('exposes the duplicated connection opened per queue', () => { // The shutdown orchestrator closes exactly these library-created connections. - const registry = new QueueEventsRegistry(fakeConnection()) + const registry = new QueueEventsRegistry(fakeConnection(), fakeOptions()) registry.getOrCreate('email') registry.getOrCreate('sms') const connections = registry.getConnections() @@ -130,7 +148,7 @@ describe('QueueEventsRegistry.getConnections', () => { it('returns an empty map before any QueueEvents are created', () => { // With nothing observed there are no library-created connections to close. - const registry = new QueueEventsRegistry(fakeConnection()) + const registry = new QueueEventsRegistry(fakeConnection(), fakeOptions()) expect(registry.getConnections().size).toBe(0) }) }) diff --git a/src/server/services/queue-events-registry.service.ts b/src/server/services/queue-events-registry.service.ts index e0cd0a6..3dc7cf6 100644 --- a/src/server/services/queue-events-registry.service.ts +++ b/src/server/services/queue-events-registry.service.ts @@ -11,6 +11,8 @@ import { Inject, Injectable } from '@nestjs/common' import { QueueEvents } from 'bullmq' import type { Redis } from 'ioredis' import { ConnectionResolver } from './connection-resolver.service' +import { BYMAX_QUEUE_RESOLVED_OPTIONS } from '../bymax-queue.constants' +import type { ResolvedQueueOptions } from '../config/resolved-options' import { duplicateConnection } from '../utils/duplicate-connection' /** @@ -34,7 +36,10 @@ export class QueueEventsRegistry { private readonly events = new Map() private readonly connections = new Map() - constructor(@Inject(ConnectionResolver) private readonly connection: ConnectionResolver) {} + constructor( + @Inject(ConnectionResolver) private readonly connection: ConnectionResolver, + @Inject(BYMAX_QUEUE_RESOLVED_OPTIONS) private readonly options: ResolvedQueueOptions, + ) {} /** * Returns the cached `QueueEvents` for `queueName`, creating one on first @@ -50,7 +55,7 @@ export class QueueEventsRegistry { const conn = duplicateConnection(this.connection.getClient()) let qe: QueueEvents try { - qe = new QueueEvents(queueName, { connection: conn }) + qe = new QueueEvents(queueName, { connection: conn, prefix: this.options.prefix }) } catch (err) { // Disconnect the duplicated connection to prevent a Redis resource leak. conn.disconnect() diff --git a/src/server/services/worker-registry.service.spec.ts b/src/server/services/worker-registry.service.spec.ts index 42234c9..c330831 100644 --- a/src/server/services/worker-registry.service.spec.ts +++ b/src/server/services/worker-registry.service.spec.ts @@ -91,6 +91,15 @@ describe('WorkerRegistry.register', () => { expect(registry.list()).toContain('email') }) + it('constructs the Worker with the configured key prefix', () => { + // The worker must poll the same prefixed keyspace the producer Queue writes + // to; a non-default prefix would otherwise leave every job unconsumed. + const registry = new WorkerRegistry(fakeConnection(), { ...makeOptions(), prefix: 'tenant:wk' }) + registry.register({ queueName: 'email', handler: noopHandler }) + const [, , workerOpts] = workerConstructorArgs[0] as [unknown, unknown, { prefix?: unknown }] + expect(workerOpts.prefix).toBe('tenant:wk') + }) + it('throws DUPLICATE_PROCESSOR when the same queue is registered twice', () => { // Two processors targeting the same queue is an application-level error. const registry = new WorkerRegistry(fakeConnection(), makeOptions()) diff --git a/src/server/services/worker-registry.service.ts b/src/server/services/worker-registry.service.ts index ff9a77b..ea9ebe4 100644 --- a/src/server/services/worker-registry.service.ts +++ b/src/server/services/worker-registry.service.ts @@ -360,6 +360,9 @@ export class WorkerRegistry { connection: conn, concurrency: opts?.concurrency ?? DEFAULT_WORKER_CONCURRENCY, autorun: opts?.autorun ?? true, + // Match the producer Queue's key prefix so the worker polls the same + // keyspace; without this a non-default prefix leaves jobs unconsumed. + prefix: this.options.prefix, } if (opts?.limiter !== undefined) result.limiter = opts.limiter if (opts?.lockDuration !== undefined) result.lockDuration = opts.lockDuration