feat: publishing tasks for nodes failing to be published into error queues (#433)

* feat: publishing failing nodes task

* refactor: update eventRouter to include frontendErrorPublishedQueue and improve import structure

* chore: update yarn.lock to reflect dependency version changes and add new bun packages for cross-platform support

* fix: enhance error logging in onchainPublisher to include maxRetries from config

* refactor: improve import structure in eventRouter by adding downloadErrorPublishedQueue

Author: clostao

backend/src/infrastructure/eventRouter/index.ts

@@ -1,6 +1,12 @@
 import { Rabbit } from '../drivers/rabbit.js'
-import { processDownloadTask } from './processors/download.js'
-import { processFrontendTask } from './processors/frontend.js'
+import {
+  downloadErrorPublishedQueue,
+  processDownloadTask,
+} from './processors/download.js'
+import {
+  frontendErrorPublishedQueue,
+  processFrontendTask,
+} from './processors/frontend.js'
 import { Task } from './tasks.js'
 
 export const EventRouter = {
@@ -19,6 +25,9 @@ export const EventRouter = {
       Rabbit.publish(getTargetQueueByTask(tasks), tasks)
     }
   },
+  publishFailedTask: (task: Task) => {
+    Rabbit.publish(getFailedTaskQueue(task), task)
+  },
 }
 
 const getTargetQueueByTask = (task: Task) => {
@@ -30,3 +39,14 @@ const getTargetQueueByTask = (task: Task) => {
       return 'task-manager'
   }
 }
+
+const getFailedTaskQueue = (task: Task) => {
+  switch (getTargetQueueByTask(task)) {
+    case 'task-manager':
+      return frontendErrorPublishedQueue
+    case 'download-manager':
+      return downloadErrorPublishedQueue
+    default:
+      throw new Error(`Unknown task: ${task.id}`)
+  }
+}

backend/src/infrastructure/eventRouter/processors/download.ts

@@ -6,6 +6,8 @@ import { createHandlerWithRetries } from '../utils.js'
 
 const logger = createLogger('eventRouter:processor:download')
 
+export const downloadErrorPublishedQueue = 'download-errors'
+
 export const processDownloadTask = createHandlerWithRetries(
   ({ id, params }: Task) => {
     if (id === 'async-download-created') {

backend/src/infrastructure/eventRouter/processors/frontend.ts

@@ -4,6 +4,8 @@ import { UploadsUseCases } from '../../../core/uploads/uploads.js'
 import { Task } from '../tasks.js'
 import { createHandlerWithRetries } from '../utils.js'
 
+export const frontendErrorPublishedQueue = 'frontend-errors'
+
 export const processFrontendTask = createHandlerWithRetries(
   ({ id, params, retriesLeft }: Task) => {
     if (id === 'migrate-upload-nodes') {
@@ -19,6 +21,6 @@ export const processFrontendTask = createHandlerWithRetries(
     }
   },
   {
-    errorPublishQueue: 'frontend-errors',
+    errorPublishQueue: frontendErrorPublishedQueue,
   },
 )

backend/src/infrastructure/services/upload/onchainPublisher/index.ts

@@ -5,6 +5,8 @@ import { createTransactionManager } from './transactionManager.js'
 import { compactAddLength } from '@polkadot/util'
 import { nodesRepository } from '../../../repositories/objects/nodes.js'
 import { EventRouter } from '../../../eventRouter/index.js'
+import { MAX_RETRIES } from '../../../eventRouter/tasks.js'
+import { config } from '../../../../config.js'
 
 const logger = createLogger('upload:onchainPublisher')
 
@@ -72,6 +74,20 @@ const republishNodes = safeCallback(
           nodes,
         },
       })
+    } else {
+      logger.error(
+        'Failed to publish nodes after %d retries',
+        config.services.taskManager.maxRetries,
+      )
+      logger.info('Publishing failed task to event router')
+      logger.debug('Nodes: %s', nodes.join(', '))
+      EventRouter.publishFailedTask({
+        id: 'publish-nodes',
+        retriesLeft: MAX_RETRIES,
+        params: {
+          nodes,
+        },
+      })
     }
   },
 )