AP-3648 SQS exponential delay between retries

packages/amqp/lib/AbstractAmqpConsumer.ts

@@ -10,7 +10,7 @@
   TransactionObservabilityManager,
 } from '@message-queue-toolkit/core'
 import {
   isRetryDateExceeded,
   isMessageError,
   parseMessage,
   HandlerContainer,
@@ -154,27 +154,26 @@
         .then((result) => {
           if (result.result === 'success') {
             this.channel.ack(message)
-            this.handleMessageProcessed(originalMessage, 'consumed')
+            this.handleMessageProcessed(parsedMessage, 'consumed')
             return
           }
 
-          // retryLater
-          const timestamp = this.tryToExtractTimestamp(originalMessage) ?? new Date()
           // requeue the message if maxRetryDuration is not exceeded, else ack it to avoid infinite loop
-          if (!isRetryDateExceeded(timestamp, this.maxRetryDuration)) {
+          if (this.shouldBeRetried(originalMessage, this.maxRetryDuration)) {
+            // TODO: Add retry delay + republish message updating internal properties
             this.channel.nack(message, false, true)
-            this.handleMessageProcessed(originalMessage, 'retryLater')
+            this.handleMessageProcessed(parsedMessage, 'retryLater')
           } else {
             // ToDo move message to DLQ once it is implemented
             this.channel.ack(message)
-            this.handleMessageProcessed(originalMessage, 'error')
+            this.handleMessageProcessed(parsedMessage, 'error')
           }
         })
         .catch((err) => {
           // ToDo we need sanity check to stop trying at some point, perhaps some kind of Redis counter
           // If we fail due to unknown reason, let's retry
           this.channel.nack(message, false, true)
-          this.handleMessageProcessed(originalMessage, 'retryLater')
+          this.handleMessageProcessed(parsedMessage, 'retryLater')
           this.handleError(err)
         })
         .finally(() => {

packages/amqp/lib/AbstractAmqpPublisher.ts

@@ -69,22 +69,14 @@ export abstract class AbstractAmqpPublisher<MessagePayloadType extends object, M
       return
     }
 
-    /**
-     * If the message doesn't have a timestamp field -> add it
-     * will be used on the consumer to prevent infinite retries on the same message
-     */
-    if (!this.tryToExtractTimestamp(message)) {
-      // @ts-ignore
-      message[this.messageTimestampField] = new Date().toISOString()
-      this.logger.warn(`${this.messageTimestampField} not defined, adding it automatically`)
-    }
-
     if (this.logMessages) {
       // @ts-ignore
       const resolvedLogMessage = this.resolveMessageLog(message, message[this.messageTypeField])
       this.logMessage(resolvedLogMessage)
     }
 
+    message = this.updateInternalProperties(message)
+
     try {
       this.publishInternal(objectToBuffer(message), options)
     } catch (err) {

packages/amqp/test/consumers/AmqpPermissionConsumer.spec.ts

@@ -13,7 +13,11 @@
 import { registerDependencies, SINGLETON_CONFIG } from '../utils/testContext'
 
 import { AmqpPermissionConsumer } from './AmqpPermissionConsumer'
-import type { PERMISSIONS_MESSAGE_TYPE } from './userConsumerSchemas'
+import type {
+  PERMISSIONS_ADD_MESSAGE_TYPE,
+  PERMISSIONS_MESSAGE_TYPE,
+  PERMISSIONS_REMOVE_MESSAGE_TYPE,
+} from './userConsumerSchemas'
 
 describe('AmqpPermissionConsumer', () => {
   describe('init', () => {
@@ -90,15 +94,15 @@
       expect(logger.loggedMessages.length).toBe(5)
       expect(logger.loggedMessages).toEqual([
         'Propagating new connection across 0 receivers',
-        'timestamp not defined, adding it automatically',
         {
           id: '1',
           messageType: 'add',
-          timestamp: expect.any(String),
         },
+        'timestamp not defined, adding it automatically',
         {
           id: '1',
           messageType: 'add',
+          timestamp: expect.any(String),
         },
         {
           messageId: '1',
@@ -407,11 +411,9 @@
       })
       await consumer.start()
 
-      const message: PERMISSIONS_MESSAGE_TYPE = {
+      const message: PERMISSIONS_ADD_MESSAGE_TYPE = {
         id: '1',
         messageType: 'add',
-        userIds: [1],
-        permissions: ['100'],
         timestamp: new Date(new Date().getTime() - 2 * 1000).toISOString(),
       }
       publisher.publish(message)
@@ -432,11 +434,9 @@
       })
       await consumer.start()
 
-      const message: PERMISSIONS_MESSAGE_TYPE = {
+      const message: PERMISSIONS_REMOVE_MESSAGE_TYPE = {
         id: '1',
         messageType: 'remove',
-        userIds: [1],
-        permissions: ['100'],
         timestamp: new Date(new Date().getTime() - 2 * 1000).toISOString(),
       }
       publisher.publish(message)

packages/amqp/test/consumers/userConsumerSchemas.ts

@@ -11,11 +11,13 @@ export const PERMISSIONS_MESSAGE_SCHEMA = z.object({
 export const PERMISSIONS_ADD_MESSAGE_SCHEMA = z.object({
   id: z.string(),
   messageType: z.literal('add'),
+  timestamp: z.string().or(z.date()).optional(),
 })
 
 export const PERMISSIONS_REMOVE_MESSAGE_SCHEMA = z.object({
   id: z.string(),
   messageType: z.literal('remove'),
+  timestamp: z.string().or(z.date()).optional(),
 })
 
 export type PERMISSIONS_MESSAGE_TYPE = z.infer<typeof PERMISSIONS_MESSAGE_SCHEMA>

packages/amqp/test/publishers/AmqpPermissionPublisher.spec.ts

@@ -51,10 +51,9 @@ describe('PermissionPublisher', () => {
         return logger.loggedMessages.length === 2
       })
 
-      expect(logger.loggedMessages[2]).toEqual({
+      expect(logger.loggedMessages[1]).toEqual({
         id: '1',
         messageType: 'add',
-        timestamp: expect.any(String),
       })
     })
   })
@@ -245,6 +244,7 @@ describe('PermissionPublisher', () => {
           userIds: [1],
           permissions: ['100'],
           timestamp: message.timestamp.toISOString(),
+          _internalNumberOfRetries: 0,
         },
       })
     })
@@ -292,11 +292,12 @@ describe('PermissionPublisher', () => {
           userIds: [1],
           permissions: ['100'],
           timestamp: message.timestamp.toISOString(),
+          _internalNumberOfRetries: 0,
         },
       })
     })
 
-    it('publishes a message auto-filling timestamp', async () => {
+    it('publishes a message auto-filling internal properties', async () => {
       await permissionConsumer.close()
 
       const message = {
@@ -330,6 +331,7 @@ describe('PermissionPublisher', () => {
           id: '2',
           messageType: 'add',
           timestamp: expect.any(String),
+          _internalNumberOfRetries: 0,
         },
       })
     })

packages/core/lib/queues/AbstractQueueService.ts

@@ -7,6 +7,7 @@ import type { ZodSchema, ZodType } from 'zod'
 import type { MessageInvalidFormatError, MessageValidationError } from '../errors/Errors'
 import type { Logger, MessageProcessingResult } from '../types/MessageQueueTypes'
 import type { DeletionConfig, QueueDependencies, QueueOptions } from '../types/queueOptionsTypes'
+import { isRetryDateExceeded } from '../utils/dateUtils'
 import { toDatePreprocessor } from '../utils/toDateProcessor'
 
 import type {
@@ -42,11 +43,15 @@ export abstract class AbstractQueueService<
   ExecutionContext = undefined,
   PrehandlerOutput = undefined,
 > {
+  // Used to keep track of the number of retries performed on consumer
+  private readonly messageNumberOfRetriesField = '_internalNumberOfRetries'
+  // Used to avoid infinite retries on the same message
+  protected readonly messageTimestampField: string
+
   protected readonly errorReporter: ErrorReporter
   public readonly logger: Logger
   protected readonly messageIdField: string
   protected readonly messageTypeField: string
-  protected readonly messageTimestampField: string
   protected readonly logMessages: boolean
   protected readonly creationConfig?: QueueConfiguration
   protected readonly locatorConfig?: QueueLocatorType
@@ -198,7 +203,45 @@ export abstract class AbstractQueueService<
     return await barrier(message, executionContext, preHandlerOutput)
   }
 
-  protected tryToExtractTimestamp(message: MessagePayloadSchemas): Date | undefined {
+  shouldBeRetried(message: MessagePayloadSchemas, maxRetryDuration: number): boolean {
+    const timestamp = this.tryToExtractTimestamp(message) ?? new Date()
+    return !isRetryDateExceeded(timestamp, maxRetryDuration)
+  }
+
+  protected getMessageRetryDelayInSeconds(message: MessagePayloadSchemas): number {
+    // if not defined, this is the first attempt
+    const retries = this.tryToExtractNumberOfRetries(message) ?? 0
+
+    // exponential backoff -> (2 ^ (attempts)) * delay
+    // delay = 1 second
+    return Math.pow(2, retries)
+  }
+
+  protected updateInternalProperties(message: MessagePayloadSchemas): MessagePayloadSchemas {
+    const messageCopy = { ...message } // clone the message to avoid mutation
+
+    /**
+     * If the message doesn't have a timestamp field -> add it
+     * will be used to prevent infinite retries on the same message
+     */
+    if (!this.tryToExtractTimestamp(message)) {
+      // @ts-ignore
+      messageCopy[this.messageTimestampField] = new Date().toISOString()
+      this.logger.warn(`${this.messageTimestampField} not defined, adding it automatically`)
+    }
+
+    /**
+     * add/increment the number of retries performed to exponential message delay
+     */
+    const numberOfRetries = this.tryToExtractNumberOfRetries(message)
+    // @ts-ignore
+    messageCopy[this.messageNumberOfRetriesField] =
+      numberOfRetries !== undefined ? numberOfRetries + 1 : 0
+
+    return messageCopy
+  }
+
+  private tryToExtractTimestamp(message: MessagePayloadSchemas): Date | undefined {
     // @ts-ignore
     if (this.messageTimestampField in message) {
       // @ts-ignore
@@ -213,6 +256,18 @@ export abstract class AbstractQueueService<
     return undefined
   }
 
+  private tryToExtractNumberOfRetries(message: MessagePayloadSchemas): number | undefined {
+    if (
+      this.messageNumberOfRetriesField in message &&
+      typeof message[this.messageNumberOfRetriesField] === 'number'
+    ) {
+      // @ts-ignore
+      return message[this.messageNumberOfRetriesField]
+    }
+
+    return undefined
+  }
+
   protected abstract resolveNextFunction(
     preHandlers: Prehandler<MessagePayloadSchemas, ExecutionContext, PrehandlerOutput>[],
     message: MessagePayloadSchemas,

packages/sns/lib/sns/AbstractSnsPublisher.ts

@@ -60,32 +60,24 @@ export abstract class AbstractSnsPublisher<MessagePayloadType extends object>
     }
 
     try {
-      messageSchemaResult.result.parse(message)
-
-      /**
-       * If the message doesn't have a timestamp field -> add it
-       * will be used on the consumer to prevent infinite retries on the same message
-       */
-      if (!this.tryToExtractTimestamp(message)) {
-        // @ts-ignore
-        message[this.messageTimestampField] = new Date().toISOString()
-        this.logger.warn(`${this.messageTimestampField} not defined, adding it automatically`)
-      }
+      const parsedMessage = messageSchemaResult.result.parse(message)
 
       if (this.logMessages) {
         // @ts-ignore
         const resolvedLogMessage = this.resolveMessageLog(message, message[this.messageTypeField])
         this.logMessage(resolvedLogMessage)
       }
 
+      message = this.updateInternalProperties(message)
+
       const input = {
         Message: JSON.stringify(message),
         TopicArn: this.topicArn,
         ...options,
       } satisfies PublishCommandInput
       const command = new PublishCommand(input)
       await this.snsClient.send(command)
-      this.handleMessageProcessed(message, 'published')
+      this.handleMessageProcessed(parsedMessage, 'published')
     } catch (error) {
       const err = error as Error
       this.handleError(err)

packages/sns/test/consumers/SnsSqsPermissionConsumer.deadLetterQueue.spec.ts

@@ -17,7 +17,10 @@
 import type { Dependencies } from '../utils/testContext'
 
 import { SnsSqsPermissionConsumer } from './SnsSqsPermissionConsumer'
-import type { PERMISSIONS_MESSAGE_TYPE } from './userConsumerSchemas'
+import type {
+  PERMISSIONS_MESSAGE_TYPE,
+  PERMISSIONS_REMOVE_MESSAGE_TYPE,
+} from './userConsumerSchemas'
 
 // Note that dead letter queue are fully tested by sqs library - only including a few tests here to make sure the integration works
 describe('SnsSqsPermissionConsumer - dead letter queue', () => {
@@ -156,11 +159,9 @@
       })
       dlqConsumer.start()
 
-      const message: PERMISSIONS_MESSAGE_TYPE = {
+      const message: PERMISSIONS_REMOVE_MESSAGE_TYPE = {
         id: '1',
         messageType: 'remove',
-        userIds: [1],
-        permissions: ['100'],
         timestamp: new Date(new Date().getTime() - 2 * 1000).toISOString(),
       }
       await publisher.publish(message)
@@ -170,7 +171,15 @@
       expect(counter).toBeGreaterThan(2)
 
       await waitAndRetry(async () => dlqMessage)
-      expect(JSON.parse(dlqMessage.Body)).toMatchObject({ id: '1', messageType: 'remove' })
+
+      const messageBody = JSON.parse(dlqMessage.Body)
+      expect(messageBody).toEqual({
+        id: '1',
+        messageType: 'remove',
+        timestamp: message.timestamp,
+        _internalNumberOfRetries: expect.any(Number),
+      })
+      expect(messageBody._internalNumberOfRetries).toBeGreaterThan(2)
 
       dlqConsumer.stop()
     })

packages/sns/test/consumers/userConsumerSchemas.ts

@@ -3,7 +3,6 @@ import z from 'zod'
 export const PERMISSIONS_MESSAGE_SCHEMA = z.object({
   id: z.string(),
   messageType: z.enum(['add', 'remove']),
-  userIds: z.array(z.number()).describe('User IDs'),
   permissions: z.array(z.string()).nonempty().describe('List of user permissions'),
   timestamp: z.string().optional(),
 })
@@ -12,12 +11,14 @@ export const PERMISSIONS_ADD_MESSAGE_SCHEMA = z.object({
   id: z.string(),
   preHandlerIncrement: z.optional(z.number()),
   messageType: z.literal('add'),
+  timestamp: z.string().optional(),
 })
 
 export const PERMISSIONS_REMOVE_MESSAGE_SCHEMA = z.object({
   id: z.string(),
   preHandlerIncrement: z.optional(z.number()),
   messageType: z.literal('remove'),
+  timestamp: z.string().optional(),
 })
 
 export type PERMISSIONS_MESSAGE_TYPE = z.infer<typeof PERMISSIONS_MESSAGE_SCHEMA>