refactor: improve retry logic

Author: maxceem

config/test.js

@@ -6,5 +6,7 @@ module.exports = {
   zapier: {
     ZAPIER_SWITCH: process.env.ZAPIER_SWITCH || 'ON',
     ZAPIER_JOB_CANDIDATE_SWITCH: process.env.ZAPIER_JOB_CANDIDATE_SWITCH || 'ON'
-  }
+  },
+  // don't retry actions during tests because tests for now don't expect it and should be updated first
+  MAX_RETRY: 0,
 }

src/common/helper.js

@@ -225,7 +225,18 @@ async function postEvent (topic, payload) {
   await client.postEvent(message)
 }
 
+/**
+ * Sleep for a given number of milliseconds.
+ *
+ * @param {Number} milliseconds the sleep time
+ * @returns {undefined}
+ */
+async function sleep (milliseconds) {
+  return new Promise((resolve) => setTimeout(resolve, milliseconds))
+}
+
 module.exports = {
+  sleep,
   getKafkaOptions,
   getESClient,
   checkEsMutexRelease,

src/common/logger.js

@@ -56,6 +56,24 @@ logger.logFullError = (err, context = {}) => {
   err.logged = true
 }
 
+/**
+ * Log warning details
+ * @param {Object} err the error
+ * @param {Object} context contains extra info about errors
+ */
+logger.logFullWarning = (err, context = {}) => {
+  if (!err) {
+    return
+  }
+  if (err.logged) {
+    return
+  }
+  const signature = context.signature ? `${context.signature} : ` : ''
+  const errMessage = err.message || util.inspect(err).split('\n')[0]
+  logger.warn({ ..._.pick(context, ['component', 'context']), message: `${signature}${errMessage}` })
+  err.logged = true
+}
+
 /**
  * Remove invalid properties from the object and hide long arrays
  * @param {Object} obj the object

src/services/ActionProcessorService.js

@@ -48,33 +48,36 @@ processRetry.schema = {
   * @param {String} originalTopic the failed topic name
   * @param {Object} originalPayload the payload
   * @param {Number} retry how many times has it been retried
+  *
+  * @returns {Promise|null} returns Promise which would be resolved when retry event sent to Kafka,
+  *                         or `null` if it would not be scheduled
   */
-async function processCreate (originalTopic, originalPayload, retry) {
+function scheduleRetry (originalTopic, originalPayload, retry) {
   retry = retry + 1
   if (retry > config.MAX_RETRY) {
-    localLogger.debug({ context: 'processCreate', message: `retry: ${retry} for ${originalPayload.id} exceeds the max retry: ${config.MAX_RETRY} - ignored` })
+    localLogger.debug({ context: 'scheduleRetry', message: `retry: ${retry} for topic: ${originalTopic} id: ${originalPayload.id} exceeds the max retry: ${config.MAX_RETRY} - ignored` })
     return
   }
-  localLogger.debug({ context: 'processCreate', message: `retry: ${retry} for ${originalPayload.id}` })
+
+  localLogger.debug({ context: 'scheduleRetry', message: `retry: ${retry} for topic: ${originalTopic} id: ${originalPayload.id}` })
+
   const payload = {
     originalTopic,
     originalPayload,
     retry
   }
-  setTimeout(async () => {
-    await helper.postEvent(config.topics.TAAS_ACTION_RETRY_TOPIC, payload)
-  }, 2 ** retry * config.BASE_RETRY_DELAY)
-}
 
-processCreate.schema = {
-  originalTopic: Joi.string().required(),
-  originalPayload: Joi.object().required(),
-  retry: Joi.number().integer().min(0).required()
+  return helper.sleep(2 ** retry * config.BASE_RETRY_DELAY).then(() =>
+    helper.postEvent(config.topics.TAAS_ACTION_RETRY_TOPIC, payload)
+  )
 }
 
 module.exports = {
-  processRetry,
-  processCreate
+  processRetry
 }
 
 logger.buildService(module.exports, 'ActionProcessorService')
+
+// we don't want to wrap this method into service wrappers
+// because it would transform this method to `async` while we want to keep it sync
+module.exports.scheduleRetry = scheduleRetry

src/services/WorkPeriodProcessorService.js

@@ -31,8 +31,14 @@ async function processCreate (message, transactionId, options) {
     // if resource booking was not found, it may be because
     // it has not yet been created. We should send a retry request.
     if (err.httpStatus === 404) {
-      logger.logFullError(err, { component: 'WorkPeriodProcessorService', context: 'processCreate' })
-      await ActionProcessorService.processCreate(message.topic, workPeriod, options.retry)
+      const schedulePromise = ActionProcessorService.scheduleRetry(message.topic, workPeriod, options.retry)
+      if (schedulePromise) {
+        // as retry was scheduled, log this error as warning
+        logger.logFullWarning(err, { component: 'WorkPeriodProcessorService', context: 'processCreate' })
+      } else {
+        // as retry was not scheduled, then log this error as error
+        logger.logFullError(err, { component: 'WorkPeriodProcessorService', context: 'processCreate' })
+      }
       return
     } else {
       throw err