feat(watcher): migrate from Agenda to BullMQ for job queuing (#35)

* feat(watcher): migrate from Agenda to BullMQ for job queuing

- Replace Agenda with BullMQ to resolve MongoDB version conflicts
- Agenda 5.0.0 only supports MongoDB 4.x, causing compatibility issues with Mongoose 8.16.0
- Implement BullMQ loader with Redis backend for job processing
- Migrate syncWatcher and replayer controllers to use BullMQ
- Add proper connection management and cleanup functions
- Remove deprecated Agenda dependencies and loader
- Maintain same job scheduling functionality with improved reliability

* fix: add job name in worker logs

* fix: close bullmq connections to redis before stopping the process on critical errors

* fix: remove locktime extension

Co-authored-by: pjt <[email protected]>

* refactor(watcher): remove unused parameter

---------

Co-authored-by: Pierre Jeanjacquot <[email protected]>

Author: SeddikBellamine

watcher/package-lock.json

@@ -35,7 +35,7 @@
   "dependencies": {
     "@iexec/poco": "^5.5.0",
     "@socket.io/redis-adapter": "^8.3.0",
-    "agenda": "^5.0.0",
+    "bullmq": "^5.58.4",
     "debug": "^4.4.0",
     "dotenv": "^16.5.0",
     "ethers": "^6.13.5",

watcher/package.json

@@ -19,6 +19,7 @@ import { startReplayer, stopReplayer } from './controllers/replayer.js';
 import { getNextBlockToProcess, setLastBlock } from './services/counter.js';
 import { getLogger, APP_NAMESPACE } from './utils/logger.js';
 import { errorHandler } from './utils/error.js';
+import { clearAllQueues } from './loaders/bullmq.js';
 
 const logger = getLogger(APP_NAMESPACE);
 
@@ -92,6 +93,10 @@ const stop = async () => {
     logger.log('STOPPING...');
     await Promise.all([stopSyncWatcher(), stopReplayer()]);
     unsubscribeAllEvents();
+
+    // Clear BullMQ queues instead of closing connections
+    await clearAllQueues();
+
     logger.log('STOPPED');
   } catch (error) {
     errorHandler(error, {

watcher/src/app.js

@@ -1,5 +1,5 @@
 import * as config from '../config.js';
-import { getAgenda } from '../loaders/agenda.js';
+import { getQueue, getWorker } from '../loaders/bullmq.js';
 import * as ethereum from '../loaders/ethereum.js';
 import { replayPastEvents } from './ethEventsWatcher.js';
 import {
@@ -12,7 +12,6 @@ import { getBlockNumber } from '../utils/eth-utils.js';
 import { errorHandler } from '../utils/error.js';
 import { traceAll } from '../utils/trace.js';
 
-const { chainId } = config.chain;
 const { replayInterval } = config.runtime;
 
 const logger = getLogger('controllers:replayer');
@@ -59,34 +58,43 @@ const _replayPastOnly = async ({
 };
 
 const startReplayer = async () => {
-  const agenda = await getAgenda(chainId);
-  agenda.define(
+  const queue = getQueue(EVENT_REPLAY_JOB);
+
+  // Create worker to process jobs
+  getWorker(EVENT_REPLAY_JOB, async () => {
+    try {
+      await _replayPastOnly({
+        handleIndexedBlock: async (blockNumber) => {
+          await setCheckpointBlock(blockNumber);
+        },
+      });
+    } catch (error) {
+      errorHandler(error, { type: 'replay-job' });
+      throw error;
+    }
+  });
+
+  // Schedule recurring job
+  await queue.add(
     EVENT_REPLAY_JOB,
-    { lockLifetime: 10 * 60 * 1000 },
-    async (job) => {
-      try {
-        await _replayPastOnly({
-          handleIndexedBlock: async (blockNumber) => {
-            await setCheckpointBlock(blockNumber);
-            // reset job lock after every iteration
-            await job.touch();
-          },
-        });
-      } catch (error) {
-        errorHandler(error, { type: 'replay-job' });
-        throw error;
-      }
+    {},
+    {
+      repeat: {
+        every: replayInterval * 1000, // Convert seconds to milliseconds
+      },
     },
   );
-  await agenda.every(`${replayInterval} seconds`, EVENT_REPLAY_JOB);
+
   logger.log(
     `${EVENT_REPLAY_JOB} jobs added (run every ${replayInterval} seconds)`,
   );
 };
 
 const stopReplayer = async () => {
-  const agenda = await getAgenda(chainId);
-  await agenda.cancel({ name: EVENT_REPLAY_JOB });
+  // Clear the specific queue instead of obliterating
+  const queue = getQueue(EVENT_REPLAY_JOB);
+  await queue.obliterate({ force: true });
+  logger.log(`Stopped ${EVENT_REPLAY_JOB} jobs`);
 };
 
 const replayPastOnly = traceAll(_replayPastOnly, { logger });

watcher/src/controllers/replayer.js

@@ -1,12 +1,11 @@
 import * as config from '../config.js';
-import { getAgenda } from '../loaders/agenda.js';
+import { getQueue, getWorker } from '../loaders/bullmq.js';
 import { getProvider, getRpcProvider } from '../loaders/ethereum.js';
 import { getLogger } from '../utils/logger.js';
 import { sleep } from '../utils/utils.js';
 import { errorHandler } from '../utils/error.js';
 import { getBlockNumber } from '../utils/eth-utils.js';
 
-const { chainId } = config.chain;
 const { checkSyncInterval } = config.runtime;
 
 const logger = getLogger('controllers:syncWatcher');
@@ -15,7 +14,7 @@ const SYNC_WATCHER_JOB = 'watch-eth-node';
 
 const MAX_ERROR_COUNT = 3;
 
-const checkSync = () => async () => {
+const checkSync = async () => {
   let errorCount = 0;
   let isSync = false;
   await sleep(config.runtime.syncWatcherInterval);
@@ -57,17 +56,32 @@ const checkSync = () => async () => {
 };
 
 const startSyncWatcher = async () => {
-  const agenda = await getAgenda(chainId);
-  agenda.define(SYNC_WATCHER_JOB, { lockLifetime: 16000 }, checkSync());
-  await agenda.every(`${checkSyncInterval} seconds`, SYNC_WATCHER_JOB);
+  const queue = getQueue(SYNC_WATCHER_JOB);
+
+  // Create worker to process jobs
+  getWorker(SYNC_WATCHER_JOB, checkSync);
+
+  // Schedule recurring job
+  await queue.add(
+    SYNC_WATCHER_JOB,
+    {},
+    {
+      repeat: {
+        every: checkSyncInterval * 1000, // Convert seconds to milliseconds
+      },
+    },
+  );
+
   logger.log(
     `${SYNC_WATCHER_JOB} jobs added (run every ${checkSyncInterval} seconds)`,
   );
 };
 
 const stopSyncWatcher = async () => {
-  const agenda = await getAgenda(chainId);
-  await agenda.cancel({ name: SYNC_WATCHER_JOB });
+  // Clear the specific queue instead of obliterating
+  const queue = getQueue(SYNC_WATCHER_JOB);
+  await queue.obliterate({ force: true });
+  logger.log(`Stopped ${SYNC_WATCHER_JOB} jobs`);
 };
 
 export { startSyncWatcher, stopSyncWatcher };

watcher/src/controllers/syncWatcher.js

@@ -0,0 +1,115 @@
+import { Queue, Worker } from 'bullmq';
+import { redis as redisConfig } from '../config.js';
+import { getLogger } from '../utils/logger.js';
+
+const logger = getLogger('bullmq');
+
+// Cache for queues and workers
+const queues = new Map();
+const workers = new Map();
+
+/**
+ * Creates or returns a cached BullMQ Queue instance.
+ * Uses Redis for job queuing instead of MongoDB.
+ */
+const getQueue = (name, options = {}) => {
+  if (queues.has(name)) {
+    return queues.get(name);
+  }
+
+  const queue = new Queue(name, {
+    connection: redisConfig,
+    defaultJobOptions: {
+      removeOnComplete: 100, // Keep last 100 completed jobs
+      removeOnFail: 50, // Keep last 50 failed jobs
+      attempts: 3, // Retry failed jobs up to 3 times
+      backoff: {
+        type: 'exponential',
+        delay: 2000,
+      },
+    },
+    ...options,
+  });
+
+  queues.set(name, queue);
+  logger.log(`Created queue: ${name}`);
+  return queue;
+};
+
+/**
+ * Creates or returns a cached BullMQ Worker instance.
+ */
+const getWorker = (name, processor, options = {}) => {
+  if (workers.has(name)) {
+    return workers.get(name);
+  }
+
+  const worker = new Worker(name, processor, {
+    connection: redisConfig,
+    concurrency: 1, // Process one job at a time
+    ...options,
+  });
+
+  // Handle worker events
+  worker.on('completed', (job) => {
+    logger.debug(`Job ${job.name} ${job.id} completed successfully`);
+  });
+
+  worker.on('failed', (job, err) => {
+    logger.warn(`Job ${job.name} ${job.id} failed:`, err.message);
+  });
+
+  worker.on('error', (err) => {
+    logger.error(`Worker error:`, err);
+  });
+
+  workers.set(name, worker);
+  logger.log(`Created worker: ${name}`);
+  return worker;
+};
+
+/**
+ * Gracefully closes all queues and workers.
+ */
+const closeAll = async ({ force = false } = {}) => {
+  logger.log('Closing all BullMQ connections...');
+
+  const closePromises = [];
+
+  // Close all workers
+  for (const worker of workers.values()) {
+    closePromises.push(worker.close(force));
+  }
+
+  // Close all queues
+  for (const queue of queues.values()) {
+    closePromises.push(queue.close(force));
+  }
+
+  await Promise.all(closePromises);
+
+  // Clear the caches
+  queues.clear();
+  workers.clear();
+
+  logger.log('All BullMQ connections closed');
+};
+
+/**
+ * Clears all jobs from all queues without closing connections.
+ * This is useful for testing when you want to reset the state.
+ */
+const clearAllQueues = async () => {
+  logger.log('Clearing all queues...');
+
+  const clearPromises = [];
+
+  for (const queue of queues.values()) {
+    clearPromises.push(queue.obliterate({ force: true }));
+  }
+
+  await Promise.all(clearPromises);
+  logger.log('All queues cleared');
+};
+
+export { getQueue, getWorker, closeAll, clearAllQueues };

watcher/src/loaders/agenda.js

@@ -1,3 +1,4 @@
+import { closeAll } from '../loaders/bullmq.js';
 import { getLogger, APP_NAMESPACE } from './logger.js';
 import { sleep } from './utils.js';
 
@@ -61,7 +62,8 @@ const recoverOnCriticalError = async () => {
     logError(
       `A critical error has occurred - Stopping process in ${GRACE_PERIOD}ms`,
     );
-    sleep(GRACE_PERIOD);
+    closeAll({ force: true });
+    await sleep(GRACE_PERIOD);
     logError('A critical error has occurred - Stopping process now');
     process.exit(1);
   }