fix(indexer): resolve sync issues with fastnear (#837)

Author: cherrybarry

apps/indexer-base/package.json

@@ -16,14 +16,15 @@
     "lint:check": "tsc --noEmit && eslint ./"
   },
   "dependencies": {
+    "@aws-sdk/client-s3": "3.750.0",
     "@near-lake/framework": "0.1.5",
     "@scure/base": "1.1.3",
     "@sentry/node": "7.74.1",
+    "@smithy/node-http-handler": "4.0.3",
     "axios": "1.5.1",
     "envalid": "8.0.0",
     "ethers": "6.13.2",
     "lodash-es": "4.17.21",
-    "near-lake-framework": "2.0.0",
     "stream-json": "1.8.0"
   },
   "devDependencies": {

apps/indexer-base/src/config.ts

@@ -1,5 +1,5 @@
+import { types } from '@near-lake/framework';
 import { bool, cleanEnv, num, str, url } from 'envalid';
-import { types } from 'near-lake-framework';
 
 import { Network } from 'nb-types';
 

apps/indexer-base/src/libs/utils.ts

@@ -3,9 +3,9 @@ import { createRequire } from 'module';
 import { base58 } from '@scure/base';
 import { decodeBase64, hexlify, Transaction } from 'ethers';
 import { snakeCase, toUpper } from 'lodash-es';
-import { types } from 'near-lake-framework';
 
 import { logger } from 'nb-logger';
+import { Action, ExecutionStatus, ReceiptEnum } from 'nb-neardata';
 import {
   AccessKeyPermissionKind,
   ActionKind,
@@ -38,7 +38,7 @@ export const decodeArgs = <T>(args: string): T =>
   json.parse(Buffer.from(args, 'base64').toString());
 
 export const mapExecutionStatus = (
-  status: types.ExecutionStatus,
+  status: ExecutionStatus,
 ): ExecutionOutcomeStatus => {
   const key = toUpper(
     snakeCase(Object.keys(status)[0]),
@@ -47,13 +47,13 @@ export const mapExecutionStatus = (
   return ExecutionOutcomeStatus[key];
 };
 
-export const mapReceiptKind = (receipt: types.ReceiptEnum): ReceiptKind => {
+export const mapReceiptKind = (receipt: ReceiptEnum): ReceiptKind => {
   const key = toUpper(Object.keys(receipt)[0]) as keyof typeof ReceiptKind;
 
   return ReceiptKind[key];
 };
 
-export const mapActionKind = (action: types.Action): ReceiptAction => {
+export const mapActionKind = (action: Action): ReceiptAction => {
   let kind = ActionKind.UNKNOWN;
   let args = {};
   let rlpHash = null;
@@ -172,12 +172,12 @@ export const mapActionKind = (action: types.Action): ReceiptAction => {
 };
 
 export const camelCaseToSnakeCase = (
-  value: Exclude<types.Action, 'CreateAccount'>,
+  value: Exclude<Action, 'CreateAccount'>,
 ) => {
   const newValue: Record<string, unknown> = {};
 
   for (const key in value) {
-    newValue[snakeCase(key)] = value[key as keyof types.Action];
+    newValue[snakeCase(key)] = value[key as keyof Action];
   }
 
   return newValue;
@@ -193,7 +193,7 @@ export const publicKeyFromImplicitAccount = (account: string) => {
   }
 };
 
-export const isExecutionSuccess = (status: types.ExecutionStatus) => {
+export const isExecutionSuccess = (status: ExecutionStatus) => {
   if ('SuccessValue' in status || 'SuccessReceiptId' in status) {
     return true;
   }

apps/indexer-base/src/services/s3.ts

@@ -1,4 +1,5 @@
 import { PutObjectCommand, S3Client } from '@aws-sdk/client-s3';
+import { NodeHttpHandler } from '@smithy/node-http-handler';
 
 import { Message } from 'nb-neardata';
 
@@ -11,7 +12,12 @@ const s3Client = new S3Client({
   },
   endpoint: config.s3Endpoint,
   forcePathStyle: true,
+  maxAttempts: 3,
   region: config.s3Region,
+  requestHandler: new NodeHttpHandler({
+    connectionTimeout: 5000,
+    requestTimeout: 10000,
+  }),
 });
 
 export const uploadJson = async (message: Message) => {

apps/indexer-base/src/types/types.ts

@@ -1,5 +1,6 @@
-import { types } from 'near-lake-framework';
+import { types } from '@near-lake/framework';
 
+import { Action } from 'nb-neardata';
 import { AccessKeyPermissionKind, ActionKind, Network } from 'nb-types';
 
 export interface Config {
@@ -38,7 +39,7 @@ export interface Config {
 
 export type ActionReceipt = {
   Action: {
-    actions: types.Action[];
+    actions: Action[];
     gasPrice: string;
     inputDataIds: string[];
     outputDataReceivers: DataReceiver[];

packages/nb-neardata/package.json

@@ -1,6 +1,6 @@
 {
   "name": "nb-neardata",
-  "version": "0.2.2",
+  "version": "0.2.3",
   "author": "NearBlocks",
   "license": "Business Source License 1.1",
   "type": "module",

packages/nb-neardata/src/index.ts

@@ -64,7 +64,7 @@ const fetchBlock = async (url: string, block: number): Promise<Message> => {
 
       if (!response.ok) {
         if (response.status === 404) {
-          await sleep(700);
+          await sleep(100);
         }
 
         throw new Error(`status: ${response.status}`);
@@ -78,7 +78,6 @@ const fetchBlock = async (url: string, block: number): Promise<Message> => {
   );
 };
 
-/*
 const fetchFinal = async (url: string): Promise<Message> => {
   return await retry(
     async () => {
@@ -98,14 +97,13 @@ const fetchFinal = async (url: string): Promise<Message> => {
     { exponential: true, logger: retryLogger, retries },
   );
 };
-*/
 
 export const streamBlock = (config: BlockStreamConfig) => {
   const url = config.url ?? endpoint(config.network);
   const limit = config.limit ?? 10;
   let isFetching = false;
   let block = config.start;
-  const highWaterMark = limit * 2;
+  const highWaterMark = limit * 5;
 
   const readable = new Readable({
     highWaterMark,
@@ -119,31 +117,47 @@ export const streamBlock = (config: BlockStreamConfig) => {
     isFetching = true;
 
     try {
-      /*
       const remaining = highWaterMark - readable.readableLength;
 
+      logger.warn({ fetchingBlock: block, queueSize: readable.readableLength });
+
       if (block % 10 === 0 && remaining >= 5) {
-        const final = (await fetchFinal(url)).block.header.height;
+        const finalBlocks = [];
+
+        for (let i = 0; i < 2; i++) {
+          finalBlocks.push((await fetchFinal(url)).block.header.height);
+
+          if (i === 0) {
+            await sleep(100);
+          }
+        }
 
+        const final = Math.max(...finalBlocks);
         const promises: Promise<Message>[] = [];
         const concurrency = Math.min(limit, final - block, remaining);
 
-        for (let i = 0; i < concurrency; i++) {
-          promises.push(fetchBlock(url, block + i));
-        }
+        logger.warn({ concurrency, finalBlock: final });
 
-        const results = await Promise.all(promises);
+        if (concurrency > 0) {
+          for (let i = 0; i < concurrency; i++) {
+            promises.push(fetchBlock(url, block + i));
+          }
+
+          const results = await Promise.all(promises);
+
+          for (const result of results) {
+            if (result) {
+              const message: Message = camelCaseKeys(result);
 
-        for (const result of results) {
-          if (result && !readable.push(camelCaseKeys(result))) {
-            return;
+              if (!readable.push(message)) {
+                return;
+              }
+
+              block = message.block.header.height + 1;
+            }
           }
         }
-
-        block += concurrency;
-        return;
       }
-      */
 
       const result = await fetchBlock(url, block);
 
@@ -164,7 +178,7 @@ export const streamBlock = (config: BlockStreamConfig) => {
     }
   };
 
-  const interval = setInterval(fetchBlocks, 500);
+  const interval = setInterval(fetchBlocks, 100);
 
   readable.on('close', () => clearInterval(interval));