Earlier exit when failing to fetch blocks with workers

packages/node-core/CHANGELOG.md

@@ -7,10 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 ### Added
 - Schema Migration support for Enums, Relations, Subscription (#2251)
+
 ### Fixed
 - Fixed non-atomic schema migration execution (#2244)
 - Testing Suites should run with unfinalizedBlocks `false` (#2258)
 
+### Changed
+- Improve error handling when fetching blocks (#2256)
+
 ## [7.2.1] - 2024-02-07
 ### Added
 - Update `ParentProject` to use `untilBlock` as and alias for `block` (#2235)

packages/node-core/src/api.service.ts

@@ -43,18 +43,30 @@ export abstract class ApiService<A = any, SA = any, B extends Array<any> = any[]
   private timeouts: Record<string, NodeJS.Timeout | undefined> = {};
 
   async fetchBlocks(heights: number[], numAttempts = MAX_RECONNECT_ATTEMPTS): Promise<B> {
+    return this.retryFetch(async () => {
+      // Get the latest fetch function from the provider
+      const apiInstance = this.connectionPoolService.api;
+      return apiInstance.fetchBlocks(heights);
+    }, numAttempts);
+  }
+
+  protected async retryFetch(fn: () => Promise<B>, numAttempts = MAX_RECONNECT_ATTEMPTS): Promise<B> {
     let reconnectAttempts = 0;
+    let lastError: Error | null = null;
     while (reconnectAttempts < numAttempts) {
       try {
-        // Get the latest fetch function from the provider
-        const apiInstance = this.connectionPoolService.api;
-        return await apiInstance.fetchBlocks(heights);
+        return await fn();
       } catch (e: any) {
-        logger.error(e, `Failed to fetch blocks ${heights[0]}...${heights[heights.length - 1]}`);
-
+        lastError = e;
         reconnectAttempts++;
       }
     }
+    if (lastError !== null) {
+      logger.error(
+        `Maximum number of retries (${numAttempts}) reached. See the following error for the underlying reason.`
+      );
+      throw lastError;
+    }
     throw new Error(`Maximum number of retries (${numAttempts}) reached.`);
   }
 

packages/node-core/src/indexer/blockDispatcher/block-dispatcher.ts

@@ -206,7 +206,7 @@ export abstract class BlockDispatcher<B, DS>
               // Do nothing, fetching the block was flushed, this could be caused by forked blocks or dynamic datasources
               return;
             }
-            logger.warn(e, 'Failed to enqueue fetched block to process');
+            logger.error(e, 'Failed to enqueue fetched block to process');
             process.exit(1);
           });
 

packages/node-core/src/indexer/worker/worker.service.ts

@@ -69,6 +69,7 @@ export abstract class BaseWorkerService<
         return;
       }
       logger.error(e, `Failed to fetch block ${height}`);
+      throw e;
     }
   }
 

packages/node/src/indexer/api.service.ts

@@ -263,27 +263,16 @@ export class ApiService
     return `api.rpc.${ext?.section ?? '*'}.${ext?.method ?? '*'}`;
   }
 
+  // Overrides the super function because of the specVer
   async fetchBlocks(
     heights: number[],
     overallSpecVer?: number,
     numAttempts = MAX_RECONNECT_ATTEMPTS,
   ): Promise<LightBlockContent[]> {
-    let reconnectAttempts = 0;
-    while (reconnectAttempts < numAttempts) {
-      try {
-        const apiInstance = this.connectionPoolService.api;
-        return await apiInstance.fetchBlocks(heights, overallSpecVer);
-      } catch (e: any) {
-        logger.error(
-          e,
-          `Failed to fetch blocks ${heights[0]}...${
-            heights[heights.length - 1]
-          }`,
-        );
-
-        reconnectAttempts++;
-      }
-    }
-    throw new Error(`Maximum number of retries (${numAttempts}) reached.`);
+    return this.retryFetch(async () => {
+      // Get the latest fetch function from the provider
+      const apiInstance = this.connectionPoolService.api;
+      return apiInstance.fetchBlocks(heights, overallSpecVer);
+    }, numAttempts);
   }
 }