Merge pull request #92 from autonomys/fix-blocking-batches

fix: blocking batches

Author: clostao

services/file-retriever/src/services/batchOptimizer.ts

@@ -1,6 +1,11 @@
 import { config } from '../config.js'
 import { ObjectMapping } from './objectMappingIndexer.js'
 
+const maxObjectsPerBatch = Math.min(
+  config.maxObjectsPerFetch,
+  config.maxSimultaneousFetches,
+)
+
 export const optimizeBatchFetch = (
   objects: ObjectMapping[],
 ): ObjectMapping[][] => {
@@ -21,7 +26,7 @@ export const optimizeBatchFetch = (
       pieceIndex === safePieceIndex ||
       pieceIndex === safePieceIndex + 1 ||
       lastPieceIndex === null
-    const isFull = currentBatch.length === config.maxObjectsPerFetch
+    const isFull = currentBatch.length === maxObjectsPerBatch
 
     // if pieces are not consecutive, they're not sharing the same piece
     if (isSameOrConsecutive && !isFull) {
@@ -43,7 +48,7 @@ export const optimizeBatchFetch = (
   return optimizedObjects.reduce(
     (acc, curr) => {
       const lastBatch = acc[acc.length - 1]
-      if (lastBatch.length + curr.length <= config.maxObjectsPerFetch) {
+      if (lastBatch.length + curr.length <= maxObjectsPerBatch) {
         acc[acc.length - 1] = lastBatch.concat(curr)
       } else {
         acc.push(curr)

services/file-retriever/src/services/dsnFetcher.ts

@@ -25,6 +25,7 @@ import { fromEntries, promiseAll } from '../utils/array.js'
 import { weightedRequestConcurrencyController } from '@autonomys/asynchronous'
 import { optimizeBatchFetch } from './batchOptimizer.js'
 import { readableToStream } from '../utils/stream.js'
+import { withRetries } from '../utils/retries.js'
 
 const fetchNodesSchema = z.object({
   jsonrpc: z.string(),
@@ -75,41 +76,60 @@ const fetchObjects = async (objects: ObjectMapping[]) => {
     id: requestId,
   }
 
-  return concurrencyController(async () => {
-    logger.debug(
-      `Fetching nodes (requestId=${requestId}): ${objects.map((e) => e[0]).join(', ')}`,
-    )
-    const fetchStart = performance.now()
-    const response = await axios.post(gatewayUrl, body, {
-      timeout: 3600_000,
-      responseType: 'json',
-    })
-    if (response.status !== 200) {
-      console.error('Failed to fetch nodes', response.status, response.data)
-      throw new HttpError(500, 'Internal server error: Failed to fetch nodes')
-    }
+  return concurrencyController(
+    async () =>
+      withRetries(
+        async () => {
+          logger.debug(
+            `Fetching nodes (requestId=${requestId}): ${objects.map((e) => e[0]).join(', ')}`,
+          )
+          const fetchStart = performance.now()
+          const response = await axios.post(gatewayUrl, body, {
+            timeout: 3600_000,
+            responseType: 'json',
+          })
+          if (response.status !== 200) {
+            console.error(
+              'Failed to fetch nodes',
+              response.status,
+              response.data,
+            )
+            throw new HttpError(
+              500,
+              'Internal server error: Failed to fetch nodes',
+            )
+          }
 
-    const validatedResponseData = fetchNodesSchema.safeParse(response.data)
-    if (!validatedResponseData.success) {
-      console.error(
-        'Failed to parse fetch nodes response',
-        validatedResponseData.error,
-      )
-      throw new HttpError(
-        500,
-        'Internal server error: Failed to parse fetch nodes response',
-      )
-    }
+          const validatedResponseData = fetchNodesSchema.safeParse(
+            response.data,
+          )
+          if (!validatedResponseData.success) {
+            console.error(
+              'Failed to parse fetch nodes response',
+              validatedResponseData.error,
+            )
+            throw new HttpError(
+              500,
+              'Internal server error: Failed to parse fetch nodes response',
+            )
+          }
 
-    const end = performance.now()
-    logger.debug(
-      `Fetched ${objects.length} nodes in total=${end - now}ms fetch=${end - fetchStart}ms (requestId=${requestId})`,
-    )
+          const end = performance.now()
+          logger.debug(
+            `Fetched ${objects.length} nodes in total=${end - now}ms fetch=${end - fetchStart}ms (requestId=${requestId})`,
+          )
 
-    return validatedResponseData.data.result.map((hex) =>
-      decodeNode(Buffer.from(hex, 'hex')),
-    )
-  }, objects.length)
+          return validatedResponseData.data.result.map((hex) =>
+            decodeNode(Buffer.from(hex, 'hex')),
+          )
+        },
+        {
+          maxRetries: 3,
+          delay: 500,
+        },
+      ),
+    objects.length,
+  )
 }
 
 /**

services/file-retriever/src/utils/retries.ts

@@ -0,0 +1,26 @@
+type RetryOptions = {
+  maxRetries?: number
+  delay?: number
+}
+
+const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms))
+
+export const withRetries = async <T>(
+  fn: () => Promise<T>,
+  { maxRetries = 5, delay = 1000 }: RetryOptions = {},
+): Promise<T> => {
+  let retries = 0
+  let lastError: Error
+
+  while (retries < maxRetries) {
+    try {
+      return await fn()
+    } catch (error) {
+      lastError = error as Error
+      retries++
+      await sleep(delay)
+    }
+  }
+
+  throw lastError!
+}