fix(prefix): propagate key prefix to Worker, QueueEvents, and FlowProducer - #25
Merged
Conversation
…ts, 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.
There was a problem hiding this comment.
Pull request overview
This PR fixes a keyspace mismatch in @bymax-one/nest-queue when a non-default BullMQ prefix is configured, ensuring producers and consumers operate on the same Redis key prefix so jobs, events, and flows are processed correctly.
Changes:
- Propagates the resolved
prefixinto BullMQWorkerconstruction viaWorkerRegistry.buildBullOptions. - Injects resolved options into
QueueEventsRegistryand passesprefixintoQueueEventscreation. - Extends
FlowServiceto accept aprefixargument and forwards it intoFlowProducer, updating module wiring and adding regression tests.
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated no comments.
Show a summary per file
| File | Description |
|---|---|
| src/server/services/worker-registry.service.ts | Adds prefix to worker options so workers poll the same prefixed keyspace as the producer queue. |
| src/server/services/worker-registry.service.spec.ts | Adds a regression test asserting Worker is constructed with the configured prefix. |
| src/server/services/queue-events-registry.service.ts | Injects resolved options and passes prefix into QueueEvents so event listeners watch the correct keyspace. |
| src/server/services/queue-events-registry.service.spec.ts | Updates setup for new constructor dependency and adds a regression test for QueueEvents prefix. |
| src/server/services/flow.service.ts | Adds prefix argument and passes it to FlowProducer so flow jobs land in the correct keyspace. |
| src/server/services/flow.service.spec.ts | Updates constructor calls and adds a regression test for FlowProducer prefix. |
| src/server/bymax-queue.module.ts | Wires resolved prefix into FlowService for both sync and async module registration paths. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
Only the producer
Queue(QueueService) receivedoptions.prefix. TheWorker(WorkerRegistry.buildBullOptions),QueueEvents(QueueEventsRegistry), andFlowProducer(FlowService) were constructed without it and fell back to BullMQ's defaultbullprefix.With any non-default
QUEUE_PREFIX, producers and consumers diverge onto separate Redis keyspaces: workers never consume jobs,@OnQueueEventlisteners never fire, and flow jobs are enqueued where no worker polls. The existing E2E missed this because it runs with the default prefix.Surfaced while verifying
nest-queue-example(which setsQUEUE_PREFIX=nqex): the app booted but the smoke job sat unconsumed innqex:audit:waitwhile the worker blocked onbzpopminagainstbull:audit:marker.Fix
Pass the resolved prefix into all three:
WorkerRegistry.buildBullOptions→prefix: this.options.prefix.QueueEventsRegistrynow injectsBYMAX_QUEUE_RESOLVED_OPTIONSand passesprefixtonew QueueEvents.FlowServicetakes a newprefixconstructor arg (wired from both module factories) and passes it tonew FlowProducer.Verification
build/typecheck/lintOK; unit suite 261 tests, 100% coverage.nest-queue-examplewithQUEUE_PREFIX=nqex: the smoke job now enqueues and completes (audit trail populated within ~1s).