Unify task retry handling (#479)

* feat: add ensureObjectPublished function to validate and publish nodes by root CID; update TaskSchema to include new task type

* update: unify how errors in tasks are managed

* feat: add handler for ensureObjectPublished task in frontend processor

* remove maxUploadNodesPerBatch parameter from config

* refactor: remove safeCallback utility from publishNodes function for improved clarity

Author: clostao

apps/backend/src/config.ts

@@ -61,7 +61,6 @@ export const config = {
   },
   params: {
     maxConcurrentUploads: Number(env('MAX_CONCURRENT_UPLOADS', '40')),
-    maxUploadNodesPerBatch: Number(env('MAX_UPLOAD_NODES_PER_BATCH', '20')),
     maxAnonymousDownloadSize: Number(
       env('MAX_ANONYMOUS_DOWNLOAD_SIZE', ONE_HUNDRED_MiB.toString()),
     ),

apps/backend/src/core/objects/nodes.ts

@@ -27,6 +27,7 @@ import { createLogger } from '../../infrastructure/drivers/logger.js'
 import { Node } from '../../infrastructure/repositories/objects/nodes.js'
 import { EventRouter } from '../../infrastructure/eventRouter/index.js'
 import { createTask, Task } from '../../infrastructure/eventRouter/tasks.js'
+import { OnchainPublisher } from '../../infrastructure/services/upload/onchainPublisher/index.js'
 
 const logger = createLogger('useCases:objects:nodes')
 
@@ -259,6 +260,15 @@ const setPublishedOn = async (
   return
 }
 
+const ensureObjectPublished = async (cid: string): Promise<void> => {
+  const nodes = await nodesRepository.getNodesByRootCid(cid)
+  if (nodes.length === 0) {
+    throw new Error(`Nodes not found for ${cid}`)
+  }
+
+  await OnchainPublisher.publishNodes(nodes.map((node) => node.cid))
+}
+
 export const NodesUseCases = {
   getNode,
   saveNode,
@@ -270,4 +280,5 @@ export const NodesUseCases = {
   getNodesByCids,
   scheduleNodeArchiving,
   setPublishedOn,
+  ensureObjectPublished,
 }

apps/backend/src/core/uploads/uploads.ts

@@ -20,7 +20,6 @@ import { ObjectUseCases } from '../objects/object.js'
 import { cidToString, FileUploadOptions } from '@autonomys/auto-dag-data'
 import { EventRouter } from '../../infrastructure/eventRouter/index.js'
 import { createTask, Task } from '../../infrastructure/eventRouter/tasks.js'
-import { chunkArray } from '../../shared/utils/misc.js'
 import { config } from '../../config.js'
 import { blockstoreRepository } from '../../infrastructure/repositories/uploads/blockstore.js'
 import { BlockstoreUseCases } from './blockstore.js'
@@ -310,19 +309,14 @@ const removeUploadArtifacts = async (uploadId: string): Promise<void> => {
 const scheduleNodesPublish = async (cid: string): Promise<void> => {
   const nodes = await NodesUseCases.getCidsByRootCid(cid)
 
-  const tasks: Task[] = chunkArray(
-    nodes,
-    config.params.maxUploadNodesPerBatch,
-  ).map((nodes) =>
-    createTask({
-      id: 'publish-nodes',
-      params: {
-        nodes,
-      },
-    }),
-  )
-
-  EventRouter.publish(tasks)
+  nodes.forEach((node) => {
+    EventRouter.publish(
+      createTask({
+        id: 'publish-nodes',
+        params: { nodes: [node] },
+      }),
+    )
+  })
 }
 
 const scheduleUploadTagging = async (cid: string): Promise<void> => {

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

@@ -7,15 +7,17 @@ import { createHandlerWithRetries } from '../utils.js'
 export const frontendErrorPublishedQueue = 'frontend-errors'
 
 export const processFrontendTask = createHandlerWithRetries(
-  ({ id, params, retriesLeft }: Task) => {
+  ({ id, params }: Task) => {
     if (id === 'migrate-upload-nodes') {
       return UploadsUseCases.processMigration(params.uploadId)
     } else if (id === 'archive-objects') {
       return NodesUseCases.processNodeArchived(params.objects)
     } else if (id === 'publish-nodes') {
-      return OnchainPublisher.publishNodes(params.nodes, retriesLeft)
+      return OnchainPublisher.publishNodes(params.nodes)
     } else if (id === 'tag-upload') {
       return UploadsUseCases.tagUpload(params.cid)
+    } else if (id === 'ensure-object-published') {
+      return NodesUseCases.ensureObjectPublished(params.cid)
     } else {
       throw new Error(`Received task ${id} but no handler found.`)
     }

apps/backend/src/infrastructure/eventRouter/tasks.ts

@@ -48,6 +48,13 @@ export const TaskSchema = z.discriminatedUnion('id', [
       cid: z.string(),
     }),
   }),
+  z.object({
+    id: z.literal('ensure-object-published'),
+    retriesLeft: z.number().default(MAX_RETRIES),
+    params: z.object({
+      cid: z.string(),
+    }),
+  }),
 ])
 
 export type MigrateUploadTask = z.infer<typeof TaskSchema>
@@ -90,6 +97,12 @@ type TaskCreateParams =
         cid: string
       }
     }
+  | {
+      id: 'ensure-object-published'
+      params: {
+        cid: string
+      }
+    }
 
 export const createTask = (task: TaskCreateParams): Task => {
   switch (task.id) {
@@ -129,6 +142,12 @@ export const createTask = (task: TaskCreateParams): Task => {
         params: task.params,
         retriesLeft: MAX_RETRIES,
       }
+    case 'ensure-object-published':
+      return {
+        id: task.id,
+        params: task.params,
+        retriesLeft: MAX_RETRIES,
+      }
     default:
       return exhaustiveCheck(task)
   }

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

@@ -1,96 +1,54 @@
 import { createLogger } from '../../../drivers/logger.js'
 import { NodesUseCases } from '../../../../core/objects/nodes.js'
-import { safeCallback } from '../../../../shared/utils/safe.js'
 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')
 
 export const transactionManager = createTransactionManager()
 
-const REPUBLISH_DELAY = 10_000
+const publishNodes = async (cids: string[]) => {
+  logger.info('Uploading %d nodes', cids.length)
 
-const publishNodes = safeCallback(
-  async (cids: string[], retriesLeft: number) => {
-    logger.info('Uploading %d nodes', cids.length)
-
-    const nodes = await nodesRepository.getNodesByCids(cids)
-    const nodeCounts = await Promise.all(
-      nodes.map((e) =>
-        nodesRepository.getNodeCount({
-          cid: e.cid,
-        }),
-      ),
-    )
-
-    const filteredNodes = nodes.filter(
-      (_, index) =>
-        nodeCounts[index] && (nodeCounts[index].publishedCount ?? 0) === 0,
-    )
-
-    const transactions = filteredNodes.map((node) => {
-      const buffer = Buffer.from(node.encoded_node, 'base64')
-
-      return {
-        module: 'system',
-        method: 'remark',
-        params: [compactAddLength(buffer)],
-      }
-    })
+  const nodes = await nodesRepository.getNodesByCids(cids)
+  const nodeCounts = await Promise.all(
+    nodes.map((e) =>
+      nodesRepository.getNodeCount({
+        cid: e.cid,
+      }),
+    ),
+  )
 
-    const results = await transactionManager.submit(transactions)
-    const someNodeFailed = results.some((result) => !result.success)
-    if (someNodeFailed) {
-      republishNodes(
-        filteredNodes
-          .filter((_, index) => !results[index].success)
-          .map((node) => node.cid),
-        retriesLeft,
-      )
-    }
+  const filteredNodes = nodes.filter(
+    (_, index) =>
+      nodeCounts[index] && (nodeCounts[index].publishedCount ?? 0) === 0,
+  )
 
-    await Promise.all(
-      filteredNodes.map((node, index) => {
-        const isSuccess = results[index].success
-        if (!isSuccess) return null
-        return NodesUseCases.setPublishedOn(node.cid, results[index])
-      }),
-    )
-  },
-)
+  const transactions = filteredNodes.map((node) => {
+    const buffer = Buffer.from(node.encoded_node, 'base64')
 
-const republishNodes = safeCallback(
-  async (nodes: string[], retriesLeft: number) => {
-    await new Promise((resolve) => setTimeout(resolve, REPUBLISH_DELAY))
-    if (retriesLeft > 0) {
-      EventRouter.publish({
-        id: 'publish-nodes',
-        retriesLeft: retriesLeft - 1,
-        params: {
-          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,
-        },
-      })
+    return {
+      module: 'system',
+      method: 'remark',
+      params: [compactAddLength(buffer)],
     }
-  },
-)
+  })
+
+  const results = await transactionManager.submit(transactions)
+  const someNodeFailed = results.some((result) => !result.success)
+  if (someNodeFailed) {
+    throw new Error('Failed to publish nodes')
+  }
+
+  await Promise.all(
+    filteredNodes.map((node, index) => {
+      const isSuccess = results[index].success
+      if (!isSuccess) return null
+      return NodesUseCases.setPublishedOn(node.cid, results[index])
+    }),
+  )
+}
 
 export const OnchainPublisher = {
   publishNodes,