add limitations for sei

src/block/block.service.ts

@@ -5,6 +5,7 @@ import { Repository } from 'typeorm';
 import * as _ from 'lodash';
 import Web3 from 'web3';
 import { Deployment } from '../deployment/deployment.service';
+import { sleep } from '../utilities';
 
 export interface BlocksDictionary {
   [id: number]: Date;
@@ -36,29 +37,32 @@ export class BlockService {
 
   private async fetchAndStore(blocks: Array<number>, deployment: Deployment): Promise<void> {
     const batches = _.chunk(blocks, 100);
+    const limit = (await import('p-limit')).default;
+    const concurrencyLimit = limit(deployment.harvestConcurrency);
 
     for (let i = 0; i < batches.length; i++) {
-      const promises = [];
       const newBlocks = [];
 
-      for (let x = 0; x < batches[i].length; x++) {
-        const promise = new Promise(async (resolve) => {
-          try {
-            const blockchainData = await this.getBlockchainData(batches[i][x], deployment);
-            const newBlock = this.block.create({
-              id: Number(blockchainData.number),
-              timestamp: new Date(parseInt(blockchainData.timestamp) * 1000),
-              blockchainType: deployment.blockchainType,
-            });
-            newBlocks.push(newBlock);
-          } catch (error) {
-            console.log('error detected:', error);
-          }
-          resolve(true);
-        });
-        promises.push(promise);
-      }
-      await Promise.all(promises);
+      await Promise.all(
+        batches[i].map((blockNumber) =>
+          concurrencyLimit(async () => {
+            try {
+              const blockchainData = await this.getBlockchainData(blockNumber, deployment);
+              const newBlock = this.block.create({
+                id: Number(blockchainData.number),
+                timestamp: new Date(parseInt(blockchainData.timestamp) * 1000),
+                blockchainType: deployment.blockchainType,
+              });
+              newBlocks.push(newBlock);
+
+              await sleep(deployment.harvestSleep || 0);
+            } catch (error) {
+              console.log('error detected:', error);
+            }
+          }),
+        ),
+      );
+
       await this.block.save(newBlocks);
     }
   }

src/deployment/deployment.service.ts

@@ -41,6 +41,7 @@ export interface Deployment {
   rpcEndpoint: string;
   harvestEventsBatchSize: number;
   harvestConcurrency: number;
+  harvestSleep?: number;
   multicallAddress: string;
   gasToken: GasToken;
   startBlock: number;
@@ -125,6 +126,7 @@ export class DeploymentService {
         rpcEndpoint: this.configService.get('SEI_RPC_ENDPOINT'),
         harvestEventsBatchSize: 1000,
         harvestConcurrency: 1,
+        harvestSleep: 1000,
         multicallAddress: '0x51aA24A9230e62CfaF259c47DE3133578cE36317',
         startBlock: 79146720,
         gasToken: {

src/harvester/harvester.service.ts

@@ -20,6 +20,7 @@ import { TokensByAddress } from '../token/token.service';
 import { BigNumber } from '@ethersproject/bignumber';
 import { BlockchainType, Deployment } from '../deployment/deployment.service';
 import { ConfigService } from '@nestjs/config';
+import { sleep } from '../utilities';
 
 export const VERSIONS = {
   // PoolMigrator: [{ terminatesAt: 14830503, version: 1 }, { version: 2 }],
@@ -196,7 +197,7 @@ export class HarvesterService {
   }
 
   async processEvents(args: ProcessEventsArgs): Promise<any[]> {
-    const { deployment } = args; // Extract deployment from args
+    const { deployment } = args;
     const key = `${deployment.blockchainType}-${deployment.exchangeId}-${args.entity}`;
     const lastProcessedBlock = await this.lastProcessedBlockService.getOrInit(key, deployment.startBlock);
     const result = [];
@@ -205,6 +206,9 @@ export class HarvesterService {
       await this.preClear(args.repository, lastProcessedBlock, deployment);
     }
 
+    const limit = (await import('p-limit')).default;
+    const concurrencyLimit = limit(deployment.harvestConcurrency);
+
     for (
       let rangeStart = lastProcessedBlock + 1;
       rangeStart <= args.endBlock;
@@ -222,7 +226,7 @@ export class HarvesterService {
         rangeEnd,
         args.contractAddress,
         deployment,
-      ); // Pass deployment
+      );
 
       if (events.length > 0) {
         let blocksDictionary: BlocksDictionary;
@@ -234,16 +238,13 @@ export class HarvesterService {
 
         const newEvents = await Promise.all(
           events.map(async (e) => {
-            if (e.transactionHash === '0x4274ad534b26c72588faae1331dac752e7f4facf0fff9333a4b9964b751331a1') {
-              console.log(rangeStart, rangeEnd);
-            }
             let newEvent = args.repository.create({
               block: { id: Number(e.blockNumber) },
               transactionIndex: Number(e.transactionIndex),
               transactionHash: e.transactionHash,
               logIndex: Number(e.logIndex),
-              blockchainType: deployment.blockchainType, // Include blockchainType
-              exchangeId: deployment.exchangeId, // Include exchangeId
+              blockchainType: deployment.blockchainType,
+              exchangeId: deployment.exchangeId,
             });
 
             if (args.constants) {
@@ -301,9 +302,13 @@ export class HarvesterService {
             }
 
             if (args.fetchCallerId) {
-              const web3 = new Web3(deployment.rpcEndpoint);
-              const transaction = await web3.eth.getTransaction(e.transactionHash);
-              newEvent['callerId'] = transaction.from;
+              await concurrencyLimit(async () => {
+                const web3 = new Web3(deployment.rpcEndpoint);
+                const transaction = await web3.eth.getTransaction(e.transactionHash);
+                newEvent['callerId'] = transaction.from;
+
+                await sleep(deployment.harvestSleep || 0);
+              });
             }
 
             if (args.customFns) {

src/token/token.service.ts

@@ -43,7 +43,7 @@ export class TokenService {
     const lastProcessedBlockNumber = await this.lastProcessedBlockService.getOrInit(lastProcessedEntity, 1);
 
     // Define batch size
-    const batchSize = 100000;
+    const batchSize = 10000;
     let currentBlock = lastProcessedBlockNumber;
 
     while (currentBlock < endBlock) {

src/utilities.ts

@@ -22,3 +22,5 @@ export function convertKeysToCamelCase(obj) {
   }
   return obj; // If it's neither an array nor an object, return the value directly
 }
+
+export const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));