medplum / medplum / 30312328185

92%

main: 92%

push

github

feat(server): add `bullmq` globalConcurrency setting (#9826)

* feat(server): add bullmq globalConcurrency setting

Add a `globalConcurrency` option to `MedplumBullmqConfig` that sets a
cluster-wide concurrency limit on a queue (enforced globally via Redis,
unlike per-worker `concurrency`).

Because global concurrency is a queue-level property persisted in Redis,
it must be set imperatively rather than through constructor options. A new
`applyGlobalConcurrency` helper is called for each queue in `initWorkers`:
it sets the limit when configured, and removes any previously-set limit
when the config is omitted so that clearing the setting takes effect on the
next startup.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Derrick Farris <derrick@medplum.com>

* Pass error directly to logger for stack traces

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Derrick Farris <derrick@medplum.com>

* Keep error prop name, pass err object for stack traces

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Derrick Farris <derrick@medplum.com>

* Add global concurrency methods to bullmq mock

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Derrick Farris <derrick@medplum.com>

* Throw on global concurrency failure instead of logging

It is unsafe to run the workers without the configured global concurrency
applied, so let the underlying Redis command error propagate and fail
startup rather than catching and logging it. Makes initWorkers async and
awaits the applied limit before continuing.

Signed-off-by: Derrick Farris <derrick@medplum.com>

* fix(server): adapt globalConcurrency wiring to new getWorkerBullmqConfig signature

`getWorkerBullmqConfig` now takes the queue options and returns `WorkerOptions`,
so `initWorkers` can no longer use it to resolve the merged BullMQ config for a
queue. Extract the config merge into `get... (continued)

22597 of 25672 branches covered (88.02%)

Branch coverage included in aggregate %.

8 of 8 new or added lines in 3 files covered. (100.0%)

40107 of 42501 relevant lines covered (94.37%)

12269.43 hits per line