Improve indexing and status fetching jobs (#372)

* Better handle MongoDB eventual consistency

* Assign a batch id to bulk index jobs

* Tweak error message

Author: cbovis

src/constants.js

@@ -33,6 +33,7 @@ module.exports = {
     BULK_INDEX_FILLS: 'bulk-index-fills',
     FETCH_FILL_STATUS: 'fetch-fill-status',
     INDEX_FILL: 'index-fill',
+    INDEX_FILL_STATUS: 'index-fill-status',
   },
   QUEUE: {
     BULK_INDEXING: 'bulk-indexing',

src/consumers/bulk-index-fills.js

@@ -34,6 +34,7 @@ const bulkIndexFills = async job => {
   }
 
   const lastFillId = nextBatch[nextBatch.length - 1]._id;
+  const batchId = job.data.batchId || Date.now();
 
   // Get on with processing the next batch whilst this one is being processed
   // to improve batch indexing throughput.
@@ -42,10 +43,11 @@ const bulkIndexFills = async job => {
       QUEUE.BULK_INDEXING,
       JOB.BULK_INDEX_FILLS,
       {
+        batchId,
         batchSize,
         lastFillId,
       },
-      { jobId: `bulk-index-${lastFillId}` },
+      { jobId: `bulk-index-${batchId}-${lastFillId}`, removeOnComplete: false },
     );
     logger.success(
       `scheduled indexing of next ${batchSize} fills after ${lastFillId}`,

src/consumers/fetch-fill-status.js

@@ -1,6 +1,5 @@
 const _ = require('lodash');
 const mongoose = require('mongoose');
-const ms = require('ms');
 const signale = require('signale');
 
 const { FILL_STATUS, JOB, QUEUE } = require('../constants');
@@ -30,19 +29,16 @@ const fetchFillStatus = async job => {
   const status =
     receipt.status === 0 ? FILL_STATUS.FAILED : FILL_STATUS.SUCCESSFUL;
   const statusText = _.findKey(FILL_STATUS, value => value === status);
+  const result = await getModel('Fill').updateOne({ _id: fillId }, { status });
 
-  await getModel('Fill').updateOne({ _id: fillId }, { status });
-  await publishJob(
-    QUEUE.FILL_INDEXING,
-    JOB.INDEX_FILL,
-    {
-      fillId,
-    },
-    {
-      delay: ms('30 seconds'),
-      removeOnComplete: true,
-    },
-  );
+  if (result.modifiedCount === 0) {
+    throw new Error(`No fill found with the id: ${fillId}`);
+  }
+
+  await publishJob(QUEUE.FILL_INDEXING, JOB.INDEX_FILL_STATUS, {
+    fillId,
+    status,
+  });
 
   logger.success(`set status of fill ${fillId} to ${statusText}`);
 };

src/consumers/index-fill-status.js

@@ -0,0 +1,42 @@
+const mongoose = require('mongoose');
+const signale = require('signale');
+
+const { FILL_STATUS, JOB, QUEUE } = require('../constants');
+const elasticsearch = require('../util/elasticsearch');
+
+const logger = signale.scope('index fill status');
+
+const indexFillStatus = async job => {
+  const { fillId, status } = job.data;
+
+  if (!mongoose.Types.ObjectId.isValid(fillId)) {
+    throw new Error(`Invalid fillId: ${fillId}`);
+  }
+
+  if (status !== FILL_STATUS.FAILED && status !== FILL_STATUS.SUCCESSFUL) {
+    throw new Error(`Invalid status: ${status}`);
+  }
+
+  const exists = await elasticsearch
+    .getClient()
+    .exists({ id: fillId, index: 'fills', _source: false });
+  const indexed = exists.body;
+
+  if (!indexed) {
+    throw new Error(`Could not update status of fill: ${fillId}`);
+  }
+
+  await elasticsearch.getClient().update({
+    id: fillId,
+    index: 'fills',
+    body: { doc: { status } },
+  });
+
+  logger.success(`indexed fill status: ${fillId}`);
+};
+
+module.exports = {
+  fn: indexFillStatus,
+  jobName: JOB.INDEX_FILL_STATUS,
+  queueName: QUEUE.FILL_INDEXING,
+};

src/consumers/index.js

@@ -3,8 +3,9 @@ const { getQueues } = require('../queues');
 const bulkIndexFills = require('./bulk-index-fills');
 const fetchFillStatus = require('./fetch-fill-status');
 const indexFill = require('./index-fill');
+const indexFillStatus = require('./index-fill-status');
 
-const consumers = [bulkIndexFills, fetchFillStatus, indexFill];
+const consumers = [bulkIndexFills, fetchFillStatus, indexFill, indexFillStatus];
 
 const initQueueConsumers = config => {
   const queues = getQueues();

src/jobs/create-fills/index.js

@@ -1,4 +1,5 @@
 const bluebird = require('bluebird');
+const ms = require('ms');
 const signale = require('signale');
 
 const {
@@ -53,7 +54,9 @@ const createFills = async ({ batchSize }) => {
             fillId: newFill._id,
             transactionHash: newFill.transactionHash,
           },
-          { removeOnComplete: true },
+          {
+            delay: ms('5 seconds'), // Delay status fetching to ensure MongoDB changes have propagated
+          },
         );
 
         await publishJob(
@@ -62,7 +65,9 @@ const createFills = async ({ batchSize }) => {
           {
             fillId: newFill._id,
           },
-          { removeOnComplete: true },
+          {
+            delay: ms('5 seconds'), // Delay indexing to ensure MongoDB changes have propagated
+          },
         );
       });
 

src/queues.js

@@ -42,6 +42,7 @@ const publishJob = async (queueName, jobName, jobData, options = {}) => {
       delay: ms('10 seconds'),
       type: 'exponential',
     },
+    removeOnComplete: true,
   };
   const queue = getQueue(queueName);