F 594 fix stuck event handler queue

src/auditLog/feathersElasticSearch.js

@@ -67,6 +67,7 @@ const configureAuditLog = app => {
   setAuditLogToFeathersService({ app, serviceName: 'donations' });
   setAuditLogToFeathersService({ app, serviceName: 'pledgeAdmins' });
   setAuditLogToFeathersService({ app, serviceName: 'events' });
+  setAuditLogToFeathersService({ app, serviceName: 'emails' });
 };
 
 module.exports = {

src/blockchain/lib/eventHandlerQueue.js

@@ -10,6 +10,7 @@ const { EventStatus } = require('../../models/events.model');
 
 const handleEventQueue = new Queue('eventHandler', { redis: config.get('redis') });
 const pendingEventQueue = new Queue('NewEventQueue', { redis: config.get('redis') });
+const TWO_MINUTES = 1000 * 60 * 2;
 
 setInterval(async () => {
   const eventHandlerQueueCount = await handleEventQueue.count();
@@ -18,7 +19,7 @@ setInterval(async () => {
     eventHandlerQueueCount,
     NewEventQueueCount,
   });
-}, 1000 * 60 * 2);
+}, TWO_MINUTES);
 
 const removeEvent = async (app, event) => {
   const { id, transactionHash } = event;
@@ -155,16 +156,36 @@ const initEventHandlerQueue = app => {
 
   handleEventQueue.process(1, async (job, done) => {
     const { event } = job.data;
+    const callDoneTimeout = setTimeout(() => {
+      logger.error('The event handler didnt respond, call done() to prevent stocking queue');
+      done();
+    }, TWO_MINUTES);
+
     try {
+      const remainingEventsInQueue = await handleEventQueue.count();
       const handler = handlers[event.event];
+
       logger.info('Handling Event: ', {
+        remainingEventsInQueue,
         event: event.event,
         transactionHash: event.transactionHash,
         status: event.status,
         transactionIndex: event.transactionIndex,
         logIndex: event.logIndex,
         _id: event._id,
       });
+      const eventInDb = await eventService.get(event._id);
+      if (eventInDb === EventStatus.PROCESSED) {
+        logger.info('Event is already processed, so dont need to handle again', {
+          event: event.event,
+          _id: event._id,
+          transactionHash: event.transactionHash,
+          transactionIndex: event.transactionIndex,
+        });
+        clearTimeout(callDoneTimeout);
+        done();
+        return;
+      }
       if (typeof handler === 'function') {
         await handler(event);
       } else {
@@ -181,6 +202,7 @@ const initEventHandlerQueue = app => {
         processingError: e.toString(),
       });
     } finally {
+      clearTimeout(callDoneTimeout);
       done();
     }
   });

src/services/serviceCommons.js

@@ -1,5 +1,5 @@
 const defaultFeatherMongooseOptions = {
-  multi: ['patch'],
+  multi: ['patch', 'remove'],
   whitelist: [
     '$exists',
     '$and',