Messages barrier max time

README.md

@@ -31,6 +31,7 @@ They implement the following public methods:
         * `messageTypeField` - which field in the message describes the type of a message. This field needs to be defined as `z.literal` in the schema and is used for resolving the correct schema for validation
         * `locatorConfig` - configuration for resolving existing queue and/or topic. Should not be specified together with the `creationConfig`.
         * `creationConfig` - configuration for queue and/or topic to create, if one does not exist. Should not be specified together with the `locatorConfig`.
+        * `messageTimestampField` - which field in the message contain the message creation date. This field needs to be ISO-8601 date string and by default it is `timestamp`, if your messages doesn't contain it the library will add one automatically to avoid infinite loops on consumer;
 * `init()`, prepare publisher for use (e. g. establish all necessary connections);
 * `close()`, stop publisher use (e. g. disconnect);
 * `publish()`, send a message to a queue or topic. It accepts the following parameters:
@@ -54,6 +55,8 @@ Multi-schema consumers support multiple message types via handler configs. They
     * `options`, composed by
         * `handlers` – configuration for handling each of the supported message types. See "Multi-schema handler definition" for more details;
         * `messageTypeField` - which field in the message describes the type of a message. This field needs to be defined as `z.literal` in the schema and is used for routing the message to the correct handler;
+        * `messageTimestampField` - which field in the message contain the message creation date. This field needs to be ISO-8601 date string and by default it is `timestamp`;
+        * `maxRetryDuration` - how long (in seconds) the message should be retried due to the `retryLater` result before marking it as consumed (avoid infinite loops). Default is 4 days;
         * `queueName`; (for SNS publishers this is a misnomer which actually refers to a topic name)
         * `locatorConfig` - configuration for resolving existing queue and/or topic. Should not be specified together with the `creationConfig`.
         * `creationConfig` - configuration for queue and/or topic to create, if one does not exist. Should not be specified together with the `locatorConfig`.

packages/amqp/lib/AbstractAmqpConsumer.ts

@@ -2,6 +2,7 @@ import type { Either, ErrorResolver } from '@lokalise/node-core'
 import type {
   BarrierResult,
   DeadLetterQueueOptions,
+  ParseMessageResult,
   Prehandler,
   PreHandlingOutputs,
   QueueConsumer,
@@ -25,6 +26,7 @@ import { AbstractAmqpService } from './AbstractAmqpService'
 import { readAmqpMessage } from './amqpMessageReader'
 
 const ABORT_EARLY_EITHER: Either<'abort', never> = { error: 'abort' }
+const DEFAULT_MAX_RETRY_DURATION = 4 * 24 * 60 * 60
 
 export type AMQPConsumerOptions<
   MessagePayloadType extends object,
@@ -60,6 +62,7 @@ export abstract class AbstractAmqpConsumer<
     AMQPLocator,
     NonNullable<unknown>
   >
+  private readonly maxRetryDuration: number
 
   private readonly messageSchemaContainer: MessageSchemaContainer<MessagePayloadType>
   private readonly handlerContainer: HandlerContainer<
@@ -78,6 +81,7 @@ export abstract class AbstractAmqpConsumer<
     this.transactionObservabilityManager = dependencies.transactionObservabilityManager
     this.errorResolver = dependencies.consumerErrorResolver
     this.deadLetterQueueOptions = options.deadLetterQueue
+    this.maxRetryDuration = options.maxRetryDuration ?? DEFAULT_MAX_RETRY_DURATION
 
     const messageSchemas = options.handlers.map((entry) => entry.schema)
     this.messageSchemaContainer = new MessageSchemaContainer<MessagePayloadType>({
@@ -128,37 +132,48 @@ export abstract class AbstractAmqpConsumer<
         this.handleMessageProcessed(null, 'invalid_message', messageId.result)
         return
       }
+      const { originalMessage, parsedMessage } = deserializedMessage.result
+
       // @ts-ignore
-      const messageType = deserializedMessage.result[this.messageTypeField]
+      const messageType = parsedMessage[this.messageTypeField]
       const transactionSpanId = `queue_${this.queueName}:${
         // @ts-ignore
         // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
-        deserializedMessage.result[this.messageTypeField]
+        parsedMessage[this.messageTypeField]
       }`
 
       // @ts-ignore
-      const uniqueTransactionKey = deserializedMessage.result[this.messageIdField]
+      const uniqueTransactionKey = parsedMessage[this.messageIdField]
       this.transactionObservabilityManager?.start(transactionSpanId, uniqueTransactionKey)
       if (this.logMessages) {
-        const resolvedLogMessage = this.resolveMessageLog(deserializedMessage.result, messageType)
+        const resolvedLogMessage = this.resolveMessageLog(parsedMessage, messageType)
         this.logMessage(resolvedLogMessage)
       }
-      this.internalProcessMessage(deserializedMessage.result, messageType)
+      this.internalProcessMessage(parsedMessage, messageType)
         .then((result) => {
-          if (result.error === 'retryLater') {
-            this.channel.nack(message, false, true)
-            this.handleMessageProcessed(deserializedMessage.result, 'retryLater')
-          }
           if (result.result === 'success') {
             this.channel.ack(message)
-            this.handleMessageProcessed(deserializedMessage.result, 'consumed')
+            this.handleMessageProcessed(originalMessage, 'consumed')
+            return
+          }
+
+          // retryLater
+          const timestamp = this.tryToExtractTimestamp(originalMessage) ?? new Date()
+          const lastRetryDate = new Date(timestamp.getTime() + this.maxRetryDuration * 1000)
+          // requeue the message if maxRetryDuration is not exceeded, else ack it to avoid infinite loop
+          if (lastRetryDate > new Date()) {
+            this.channel.nack(message, false, true)
+            this.handleMessageProcessed(originalMessage, 'retryLater')
+          } else {
+            this.channel.ack(message)
+            this.handleMessageProcessed(originalMessage, '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(deserializedMessage.result, 'retryLater')
+          this.handleMessageProcessed(originalMessage, 'retryLater')
           this.handleError(err)
         })
         .finally(() => {
@@ -180,6 +195,7 @@ export abstract class AbstractAmqpConsumer<
         barrierOutput: barrierResult.output,
       })
     }
+
     return { error: 'retryLater' }
   }
 
@@ -245,7 +261,9 @@ export abstract class AbstractAmqpConsumer<
     )
   }
 
-  private deserializeMessage(message: Message): Either<'abort', MessagePayloadType> {
+  private deserializeMessage(
+    message: Message,
+  ): Either<'abort', ParseMessageResult<MessagePayloadType>> {
     const resolveMessageResult = this.resolveMessage(message)
     if (isMessageError(resolveMessageResult.error)) {
       this.handleError(resolveMessageResult.error)

packages/amqp/lib/AbstractAmqpPublisher.ts

@@ -42,6 +42,16 @@ export abstract class AbstractAmqpPublisher<MessagePayloadType extends object>
     }
     resolveSchemaResult.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`)
+    }
+
     if (this.logMessages) {
       // @ts-ignore
       const resolvedLogMessage = this.resolveMessageLog(message, message[this.messageTypeField])

packages/amqp/lib/amqpMessageDeserializer.spec.ts

@@ -8,11 +8,12 @@ import { AmqpConsumerErrorResolver } from './errors/AmqpConsumerErrorResolver'
 
 describe('messageDeserializer', () => {
   it('deserializes valid JSON', () => {
-    const messagePayload: PERMISSIONS_MESSAGE_TYPE = {
+    const messagePayload = {
       id: '1',
       messageType: 'add',
       userIds: [1],
       permissions: ['perm'],
+      nonSchemaField: 'nonSchemaField',
     }
     const message: Message = {
       content: Buffer.from(JSON.stringify(messagePayload)),
@@ -26,7 +27,15 @@ describe('messageDeserializer', () => {
       errorProcessor,
     )
 
-    expect(deserializedPayload.result).toMatchObject(messagePayload)
+    expect(deserializedPayload.result).toMatchObject({
+      originalMessage: messagePayload,
+      parsedMessage: {
+        id: '1',
+        messageType: 'add',
+        userIds: [1],
+        permissions: ['perm'],
+      },
+    })
   })
 
   it('throws an error on invalid JSON', () => {

packages/amqp/lib/amqpMessageDeserializer.ts

@@ -1,5 +1,10 @@
 import type { Either } from '@lokalise/node-core'
-import type { MessageInvalidFormatError, MessageValidationError } from '@message-queue-toolkit/core'
+import type {
+  MessageInvalidFormatError,
+  MessageValidationError,
+  ParseMessageResult,
+} from '@message-queue-toolkit/core'
+import { parseMessage } from '@message-queue-toolkit/core'
 import type { Message } from 'amqplib'
 import type { ZodType } from 'zod'
 
@@ -9,11 +14,9 @@ export const deserializeAmqpMessage = <T extends object>(
   message: Message,
   type: ZodType<T>,
   errorProcessor: AmqpConsumerErrorResolver,
-): Either<MessageInvalidFormatError | MessageValidationError, T> => {
+): Either<MessageInvalidFormatError | MessageValidationError, ParseMessageResult<T>> => {
   try {
-    return {
-      result: type.parse(JSON.parse(message.content.toString())),
-    }
+    return parseMessage(JSON.parse(message.content.toString()), type, errorProcessor)
   } catch (exception) {
     return {
       error: errorProcessor.processError(exception),

packages/amqp/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@message-queue-toolkit/amqp",
-  "version": "14.0.0",
+  "version": "14.1.0",
   "private": false,
   "license": "MIT",
   "description": "AMQP adapter for message-queue-toolkit",
@@ -29,7 +29,7 @@
     "zod": "^3.23.6"
   },
   "peerDependencies": {
-    "@message-queue-toolkit/core": "^12.0.0",
+    "@message-queue-toolkit/core": "^12.1.0",
     "amqplib": "^0.10.3"
   },
   "devDependencies": {

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

@@ -13,6 +13,7 @@ import type { Dependencies } from '../utils/testContext'
 import { registerDependencies, SINGLETON_CONFIG } from '../utils/testContext'
 
 import { AmqpPermissionConsumer } from './AmqpPermissionConsumer'
+import type { PERMISSIONS_MESSAGE_TYPE } from './userConsumerSchemas'
 
 describe('AmqpPermissionConsumer', () => {
   describe('init', () => {
@@ -86,24 +87,24 @@ describe('AmqpPermissionConsumer', () => {
 
       await newConsumer.handlerSpy.waitForMessageWithId('1', 'consumed')
 
-      expect(logger.loggedMessages.length).toBe(4)
-      expect(logger.loggedMessages).toMatchInlineSnapshot(`
-        [
-          "Propagating new connection across 0 receivers",
-          {
-            "id": "1",
-            "messageType": "add",
-          },
-          {
-            "id": "1",
-            "messageType": "add",
-          },
-          {
-            "messageId": "1",
-            "processingResult": "consumed",
-          },
-        ]
-      `)
+      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),
+        },
+        {
+          id: '1',
+          messageType: 'add',
+        },
+        {
+          messageId: '1',
+          processingResult: 'consumed',
+        },
+      ])
     })
   })
 
@@ -380,4 +381,69 @@ describe('AmqpPermissionConsumer', () => {
       await consumer.close()
     })
   })
+
+  describe('messages stuck on retryLater', () => {
+    let diContainer: AwilixContainer<Dependencies>
+    let publisher: AmqpPermissionPublisher
+    beforeEach(async () => {
+      diContainer = await registerDependencies(TEST_AMQP_CONFIG, undefined, false)
+      publisher = diContainer.cradle.permissionPublisher
+      await publisher.init()
+    })
+
+    afterEach(async () => {
+      await diContainer.cradle.awilixManager.executeDispose()
+      await diContainer.dispose()
+    })
+
+    it('stuck on barrier', async () => {
+      let counter = 0
+      const consumer = new AmqpPermissionConsumer(diContainer.cradle, {
+        addPreHandlerBarrier: async () => {
+          counter++
+          return { isPassing: false }
+        },
+        maxRetryDuration: 3,
+      })
+      await consumer.start()
+
+      const message: PERMISSIONS_MESSAGE_TYPE = {
+        id: '1',
+        messageType: 'add',
+        userIds: [1],
+        permissions: ['100'],
+        timestamp: new Date(new Date().getTime() - 2 * 1000).toISOString(),
+      }
+      publisher.publish(message)
+
+      const jobSpy = await consumer.handlerSpy.waitForMessageWithId('1', 'error')
+      expect(jobSpy.message).toEqual(message)
+      expect(counter).toBeGreaterThan(2)
+    })
+
+    it('stuck on handler', async () => {
+      let counter = 0
+      const consumer = new AmqpPermissionConsumer(diContainer.cradle, {
+        removeHandlerOverride: async () => {
+          counter++
+          return { error: 'retryLater' }
+        },
+        maxRetryDuration: 3,
+      })
+      await consumer.start()
+
+      const message: PERMISSIONS_MESSAGE_TYPE = {
+        id: '1',
+        messageType: 'remove',
+        userIds: [1],
+        permissions: ['100'],
+        timestamp: new Date(new Date().getTime() - 2 * 1000).toISOString(),
+      }
+      publisher.publish(message)
+
+      const jobSpy = await consumer.handlerSpy.waitForMessageWithId('1', 'error')
+      expect(jobSpy.message).toEqual(message)
+      expect(counter).toBeGreaterThan(2)
+    })
+  })
 })

packages/amqp/test/consumers/AmqpPermissionConsumer.ts

@@ -25,7 +25,7 @@ type PrehandlerOutput = {
 
 type AmqpPermissionConsumerOptions = Pick<
   AMQPConsumerOptions<SupportedEvents, ExecutionContext, PrehandlerOutput>,
-  'creationConfig' | 'locatorConfig' | 'logMessages' | 'deadLetterQueue'
+  'creationConfig' | 'locatorConfig' | 'logMessages' | 'deadLetterQueue' | 'maxRetryDuration'
 > & {
   addPreHandlerBarrier?: (message: SupportedEvents) => Promise<BarrierResult<number>>
   removeHandlerOverride?: (
@@ -76,9 +76,8 @@ export class AmqpPermissionConsumer extends AbstractAmqpConsumer<
         logMessages: options?.logMessages,
         handlerSpy: true,
         messageTypeField: 'messageType',
-        deletionConfig: {
-          deleteIfExists: true,
-        },
+        deletionConfig: { deleteIfExists: true },
+        maxRetryDuration: options?.maxRetryDuration,
         handlers: new MessageHandlerConfigBuilder<
           SupportedEvents,
           ExecutionContext,

packages/amqp/test/consumers/userConsumerSchemas.ts

@@ -5,6 +5,7 @@ export const PERMISSIONS_MESSAGE_SCHEMA = z.object({
   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(),
 })
 
 export const PERMISSIONS_ADD_MESSAGE_SCHEMA = z.object({