mirror of
https://github.com/Crosstalk-Solutions/project-nomad.git
synced 2026-10-17 21:29:20 +02:00
fe6735fdcb
BullMQ instantiates a fresh ioredis client per Queue/Worker when handed a
plain {host, port} config object, and under sustained ZIM ingestion the
embed pipeline leaked ~1 client/sec until Redis maxclients was exhausted.
Pass a single shared ioredis instance (maxRetriesPerRequest: null, as
required by BullMQ) so all queues and workers reuse one client pool.
Workers still duplicate the connection once for their blocking client,
which is expected and bounded.
Closes #885
40 lines
1.1 KiB
TypeScript
40 lines
1.1 KiB
TypeScript
import { Queue } from 'bullmq'
|
|
import queueConfig from '#config/queue'
|
|
|
|
// Process-wide singleton. Instantiating a fresh QueueService per dispatch /
|
|
// status lookup leaks connections, and under sustained job churn (e.g.
|
|
// multi-batch ZIM ingestion enqueueing a continuation every few seconds) it
|
|
// saturates Redis's maxclients within hours. All queues additionally reuse the
|
|
// single shared ioredis instance exported from #config/queue (#885).
|
|
export class QueueService {
|
|
private queues: Map<string, Queue> = new Map()
|
|
|
|
private static _instance: QueueService | null = null
|
|
|
|
private constructor() {}
|
|
|
|
static getInstance(): QueueService {
|
|
if (!QueueService._instance) {
|
|
QueueService._instance = new QueueService()
|
|
}
|
|
return QueueService._instance
|
|
}
|
|
|
|
getQueue(name: string): Queue {
|
|
if (!this.queues.has(name)) {
|
|
const queue = new Queue(name, {
|
|
connection: queueConfig.connection,
|
|
})
|
|
this.queues.set(name, queue)
|
|
}
|
|
return this.queues.get(name)!
|
|
}
|
|
|
|
async close() {
|
|
for (const queue of this.queues.values()) {
|
|
await queue.close()
|
|
}
|
|
this.queues.clear()
|
|
}
|
|
}
|