Retries with backoff algorithm

[zaheershaik]

I'm trying to implement a basic multiple retry mechanism in case of processing failures in message consumer. Basically, requeue the message with some backoff time period (preferably exponential backoff) and exhaust after certain number of attempts.

For retry: I was trying to set 'x-retry-count' message header before channel.reject without success. The message header is getting reset and not propagating in the next delivery attempt.

For backoff: I was setting expiration property, which was getting reset too.

Could you please suggest some mechanism to do retries with some kind of backoff, if this is not the right way?

channel.consume(queue, function (msg) { //initiate consume
  work(msg, function (ok) { //process message
    try {
      if (ok)
        channel.ack(msg); //ack message
      else {
        var retryCount = msg.properties.headers["x-retry-count"] || 0;
        msg.properties.expiration = ++retryCount * 2000 * (Math.pow(3, (retryCount - 1)));
        msg.properties.headers["x-retry-count"] = retryCount;
        console.debug("Set retry count on msg", retryCount);
        channel.reject(msg, true); //nack message
      }
    } catch (e) {
      console.error("Message processing error", e);
    }
  });
}

[zaheershaik]

Got it working with DLQ and the following code.

Ref: https://www.cloudamqp.com/docs/delayed-messages.html

channel.consume(queue, function (msg) { //initiate consume
  work(msg, function (ok) { //process message
    try {
      if (ok) {
        channel.ack(msg); //ack message
      }
      else {
        channel.reject(msg, false); //ack message and publish a new message

        var retryCount = msg.properties.headers["x-retry-count"] || 0;
        if(++retryCount > 4) {
          console.error("Exhausted max retries on the message:", msg.content.toString());
        } else {
          var options = {persistent: true, headers: {"x-retry-count": retryCount}};
          var ttl = 3000 * (Math.pow(5, (retryCount - 1)));
          options.expiration = ttl.toString();

          console.debug(util.format("Set retry count: %s on msg with expiry: %s ms " +
              "and republish", retryCount, ttl));
          _publish("delayedQ", msg.content, channel, options, _.noop);
        }
      }
    } catch (e) {
      console.error("Message processing error", e);
      return close(channel, conn, false, function (err) {
        console.error("Error in consuming messaging from AMQP; Consumption error", err);
      })
    }
  });
}

[ismriv]

@zaheershaik you may want to check amqp-retry.node out.

[zaheershaik]

@ismriv thanks for the reference. will try that out.