fix: message visibility timeout (#139)

decentraland · Jul 1, 2024 · 9f1b9d9 · 9f1b9d9
1 parent acc1ecf
commit 9f1b9d9
Show file tree

Hide file tree

Showing 6 changed files with 37 additions and 13 deletions.
diff --git a/consumer-server/src/adapters/memory-queue.ts b/consumer-server/src/adapters/memory-queue.ts
@@ -25,5 +25,5 @@ export function createMemoryQueueAdapter(): QueueComponent {
     queue.delete(receiptHandle)
   }
 
-  return { send, receiveSingleMessage, deleteMessage }
+  return { send, receiveSingleMessage, deleteMessage, increaseMessageVisibility: async () => {} }
 }
diff --git a/consumer-server/src/adapters/sqs.ts b/consumer-server/src/adapters/sqs.ts
@@ -1,4 +1,5 @@
 import {
+  ChangeMessageVisibilityCommand,
   DeleteMessageCommand,
   Message,
   ReceiveMessageCommand,
@@ -49,9 +50,20 @@ export async function createSqsAdapter(endpoint: string): Promise<QueueComponent
     await client.send(deleteCommand)
   }
 
+  async function increaseMessageVisibility(receiptHandle: string): Promise<void> {
+    await client.send(
+      new ChangeMessageVisibilityCommand({
+        QueueUrl: endpoint,
+        ReceiptHandle: receiptHandle,
+        VisibilityTimeout: 120
+      })
+    )
+  }
+
   return {
     send,
     receiveSingleMessage,
-    deleteMessage
+    deleteMessage,
+    increaseMessageVisibility
   }
 }
diff --git a/consumer-server/src/logic/message-processor.ts b/consumer-server/src/logic/message-processor.ts
@@ -19,6 +19,11 @@ export async function createMessageProcesorComponent({
   const logger = logs.getLogger('message-procesor')
   const abServers = (await config.requireString('AB_SERVERS')).split(';')
 
+  function isRelatedToVisibilityTimeout(errorMessage: string | undefined): boolean {
+    const visibilityTimeoutError = 'The receipt handle has expired'
+    return !!errorMessage && errorMessage.includes(visibilityTimeoutError)
+  }
+
   function isRelatedToAssetBundlePublish(errorMessage: string | undefined): boolean {
     return !!errorMessage && abServers.some((abServer) => errorMessage.includes(abServer))
   }
@@ -159,18 +164,22 @@ export async function createMessageProcesorComponent({
         error: error.message
       })
 
-      if (isRelatedToAssetBundlePublish(error?.message)) {
-        await reQueue(message)
+      if(isRelatedToVisibilityTimeout(error?.message)) {
+        await queue.increaseMessageVisibility(receiptMessageHandle)
       } else {
-        if (retry < 3) {
+        if (isRelatedToAssetBundlePublish(error?.message)) {
           await reQueue(message)
         } else {
-          logger.warn('Max attempts reached, message will not be retried', {
-            entityId: message.entity.entityId,
-            base: message.entity.metadata.scene.base,
-            attempt: retry
-          })
-          metrics.increment('lod_generation_count', { status: 'failed' }, 1)
+          if (retry < 3) {
+            await reQueue(message)
+          } else {
+            logger.warn('Max attempts reached, message will not be retried', {
+              entityId: message.entity.entityId,
+              base: message.entity.metadata.scene.base,
+              attempt: retry
+            })
+            metrics.increment('lod_generation_count', { status: 'failed' }, 1)
+          }
         }
       }
 

diff --git a/consumer-server/src/types.ts b/consumer-server/src/types.ts
@@ -77,6 +77,7 @@ export type QueueComponent = {
   send(message: QueueMessage): Promise<void>
   receiveSingleMessage(): Promise<Message[]>
   deleteMessage(receiptHandle: string): Promise<void>
+  increaseMessageVisibility(receiptHandle: string): Promise<void>
 }
 
 export type QueueWorker = IBaseComponent

diff --git a/consumer-server/test/components.ts b/consumer-server/test/components.ts
@@ -44,7 +44,8 @@ async function initComponents(): Promise<TestComponents> {
   const queue: QueueComponent = {
     deleteMessage: jest.fn(),
     receiveSingleMessage: jest.fn(),
-    send: jest.fn()
+    send: jest.fn(),
+    increaseMessageVisibility: jest.fn()
   }
   const messageConsumer: QueueWorker = {
     start: jest.fn(),

diff --git a/consumer-server/test/unit/message-processor.spec.ts b/consumer-server/test/unit/message-processor.spec.ts
@@ -247,7 +247,8 @@ function getMessageProcessorMockComponents() {
     queue: {
       send: jest.fn(),
       receiveSingleMessage: jest.fn(),
-      deleteMessage: jest.fn()
+      deleteMessage: jest.fn(),
+      increaseMessageVisibility: jest.fn()
     },
     lodGenerator: {
       generate: jest.fn()