fix(queue): share one ioredis connection across BullMQ queues and workers (#1009)

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

admin/app/services/queue_service.ts

@@ -1,11 +1,11 @@
 import { Queue } from 'bullmq'
 import queueConfig from '#config/queue'
 
-// Process-wide singleton. Each `Queue` opens two ioredis connections (one for
+// Process-wide singleton. Instantiating a fresh QueueService per dispatch /
-// commands, one blocking). Instantiating a fresh QueueService per dispatch /
+// status lookup leaks connections, and under sustained job churn (e.g.
-// status lookup leaks both, and under sustained job churn (e.g. multi-batch ZIM
+// multi-batch ZIM ingestion enqueueing a continuation every few seconds) it
-// ingestion enqueueing a continuation every few seconds) it saturates Redis's
+// saturates Redis's maxclients within hours. All queues additionally reuse the
-// maxclients within hours.
+// single shared ioredis instance exported from #config/queue (#885).
 export class QueueService {
   private queues: Map<string, Queue> = new Map()
 

admin/config/queue.ts

@@ -1,11 +1,23 @@
 import env from '#start/env'
+import { Redis } from 'ioredis'
+
+// BullMQ treats a plain `{host, port}` connection object as a recipe: every
+// Queue / Worker instantiates its own ioredis client from it, and script
+// commands executed against those clients can spawn further short-lived
+// connections. Under sustained ZIM ingestion this leaked ~1 client/sec until
+// Redis maxclients was exhausted (#885). Passing a single shared ioredis
+// instance instead gives BullMQ a pool to reuse — Workers still duplicate it
+// once for their blocking client, which is expected and bounded.
+// `maxRetriesPerRequest: null` is mandatory for connections shared with BullMQ.
+const sharedConnection = new Redis({
+  host: env.get('REDIS_HOST'),
+  port: env.get('REDIS_PORT') ?? 6379,
+  db: env.get('REDIS_DB') ?? 0,
+  maxRetriesPerRequest: null,
+})
 
 const queueConfig = {
-  connection: {
+  connection: sharedConnection,
-    host: env.get('REDIS_HOST'),
-    port: env.get('REDIS_PORT') ?? 6379,
-    db: env.get('REDIS_DB') ?? 0,
-  },
 }
 
 export default queueConfig

admin/package-lock.json

@@ -47,6 +47,7 @@
         "edge.js": "6.4.0",
         "fast-xml-parser": "5.7.0",
         "fuse.js": "7.1.0",
+        "ioredis": "5.10.1",
         "ipaddr.js": "2.4.0",
         "jszip": "3.10.1",
         "luxon": "3.7.2",

admin/package.json

@@ -100,6 +100,7 @@
     "edge.js": "6.4.0",
     "fast-xml-parser": "5.7.0",
     "fuse.js": "7.1.0",
+    "ioredis": "5.10.1",
     "ipaddr.js": "2.4.0",
     "jszip": "3.10.1",
     "luxon": "3.7.2",