feat: chunk broadcasting support (#2533)

* feat: chunk broadcasting support

* chore: prettier lint

* fix: non-chunking support

* chore: bump version

---------

Co-authored-by: Ali Behjati <bahjatia@gmail.com>

Author: bangjelkoski

apps/price_pusher/README.md

@@ -114,6 +114,7 @@ pnpm run start injective --grpc-endpoint https://grpc-endpoint.com \
     --network testnet \
     [--gas-price 160000000] \
     [--gas-multiplier 1.1] \
+    [--priceIds-process-chunk-size 100] \
     [--pushing-frequency 10] \
     [--polling-frequency 5]
 

apps/price_pusher/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@pythnetwork/price-pusher",
-  "version": "9.1.5",
+  "version": "9.2.0",
   "description": "Pyth Price Pusher",
   "homepage": "https://pyth.network",
   "main": "lib/index.js",
@@ -29,7 +29,7 @@
     "dev": "ts-node src/index.ts",
     "prepublishOnly": "pnpm run build && pnpm run test:lint",
     "preversion": "pnpm run test:lint",
-    "version": "pnpm run format && git add -A src"
+    "version": "pnpm run test:format && pnpm run test:lint && git add -A src"
   },
   "keywords": [
     "pyth",
@@ -60,8 +60,9 @@
   },
   "dependencies": {
     "@coral-xyz/anchor": "^0.30.0",
-    "@injectivelabs/networks": "^1.14.6",
-    "@injectivelabs/sdk-ts": "1.10.72",
+    "@injectivelabs/networks": "1.14.47",
+    "@injectivelabs/utils": "^1.14.47",
+    "@injectivelabs/sdk-ts": "1.14.49",
     "@mysten/sui": "^1.3.0",
     "@pythnetwork/hermes-client": "^1.3.1",
     "@pythnetwork/price-service-sdk": "workspace:^",

apps/price_pusher/src/injective/command.ts

@@ -27,13 +27,19 @@ export default {
       required: true,
     } as Options,
     "gas-price": {
-      description: "Gas price to be used for each transasction",
+      description: "Gas price to be used for each transaction",
       type: "number",
     } as Options,
     "gas-multiplier": {
-      description: "Gas multiplier to be used for each transasction",
+      description: "Gas multiplier to be used for each transaction",
       type: "number",
     } as Options,
+    "price-ids-process-chunk-size": {
+      description:
+        "Set in case we wanna split price feeds updates into chunks to have smaller transactions. Set to -1 to disable chunking.",
+      type: "number",
+      required: false,
+    } as Options,
     ...options.priceConfigFile,
     ...options.priceServiceEndpoint,
     ...options.mnemonicFile,
@@ -46,18 +52,19 @@ export default {
   handler: async function (argv: any) {
     // FIXME: type checks for this
     const {
+      network,
+      logLevel,
       gasPrice,
-      gasMultiplier,
       grpcEndpoint,
-      priceConfigFile,
-      priceServiceEndpoint,
       mnemonicFile,
-      pythContractAddress,
-      pushingFrequency,
+      gasMultiplier,
+      priceConfigFile,
       pollingFrequency,
-      network,
-      logLevel,
+      pushingFrequency,
       controllerLogLevel,
+      pythContractAddress,
+      priceServiceEndpoint,
+      priceIdsProcessChunkSize,
     } = argv;
 
     const logger = pino({ level: logLevel });
@@ -111,6 +118,7 @@ export default {
         chainId: getNetworkInfo(network).chainId,
         gasPrice,
         gasMultiplier,
+        priceIdsProcessChunkSize,
       },
     );
 

apps/price_pusher/src/injective/injective.ts

@@ -7,20 +7,22 @@ import {
 } from "../interface";
 import { DurationInSeconds } from "../utils";
 import {
+  Msgs,
+  Account,
+  TxResponse,
+  PrivateKey,
+  TxGrpcApi,
   ChainGrpcAuthApi,
   ChainGrpcWasmApi,
   MsgExecuteContract,
-  Msgs,
-  PrivateKey,
-  TxGrpcClient,
-  TxResponse,
   createTransactionFromMsg,
 } from "@injectivelabs/sdk-ts";
+import { splitArrayToChunks } from "@injectivelabs/utils";
 import { Logger } from "pino";
-import { Account } from "@injectivelabs/sdk-ts/dist/cjs/client/chain/types/auth";
 
 const DEFAULT_GAS_PRICE = 160000000;
 const DEFAULT_GAS_MULTIPLIER = 1.05;
+const DEFAULT_PRICE_IDS_PROCESS_CHUNK_SIZE = -1;
 const INJECTIVE_TESTNET_CHAIN_ID = "injective-888";
 
 type PriceQueryResponse = {
@@ -90,6 +92,7 @@ type InjectiveConfig = {
   chainId: string;
   gasMultiplier: number;
   gasPrice: number;
+  priceIdsProcessChunkSize: number;
 };
 export class InjectivePricePusher implements IPricePusher {
   private wallet: PrivateKey;
@@ -110,6 +113,9 @@ export class InjectivePricePusher implements IPricePusher {
       chainId: chainConfig?.chainId ?? INJECTIVE_TESTNET_CHAIN_ID,
       gasMultiplier: chainConfig?.gasMultiplier ?? DEFAULT_GAS_MULTIPLIER,
       gasPrice: chainConfig?.gasPrice ?? DEFAULT_GAS_PRICE,
+      priceIdsProcessChunkSize:
+        chainConfig?.priceIdsProcessChunkSize ??
+        DEFAULT_PRICE_IDS_PROCESS_CHUNK_SIZE,
     };
   }
 
@@ -119,6 +125,7 @@ export class InjectivePricePusher implements IPricePusher {
 
   private async signAndBroadcastMsg(msg: Msgs): Promise<TxResponse> {
     const chainGrpcAuthApi = new ChainGrpcAuthApi(this.grpcEndpoint);
+
     // Fetch the latest account details only if it's not stored.
     this.account ??= await chainGrpcAuthApi.fetchAccount(
       this.injectiveAddress(),
@@ -132,7 +139,7 @@ export class InjectivePricePusher implements IPricePusher {
       pubKey: this.wallet.toPublicKey().toBase64(),
     });
 
-    const txService = new TxGrpcClient(this.grpcEndpoint);
+    const txService = new TxGrpcApi(this.grpcEndpoint);
     // simulation
     try {
       const {
@@ -207,12 +214,33 @@ export class InjectivePricePusher implements IPricePusher {
     if (priceIds.length !== pubTimesToPush.length)
       throw new Error("Invalid arguments");
 
+    const priceIdChunks =
+      this.chainConfig.priceIdsProcessChunkSize === -1
+        ? [priceIds]
+        : splitArrayToChunks({
+            array: priceIds,
+            chunkSize: this.chainConfig.priceIdsProcessChunkSize,
+          });
+
+    for (const [chunkIndex, priceIdChunk] of priceIdChunks.entries()) {
+      await this.updatePriceFeedChunk(priceIdChunk, chunkIndex);
+    }
+  }
+
+  private async updatePriceFeedChunk(
+    priceIds: string[],
+    chunkIndex: number,
+  ): Promise<void> {
     let priceFeedUpdateObject;
+
     try {
       // get the latest VAAs for updatePriceFeed and then push them
       priceFeedUpdateObject = await this.getPriceFeedUpdateObject(priceIds);
     } catch (err) {
-      this.logger.error(err, "Error fetching the latest vaas to push");
+      this.logger.error(
+        err,
+        `Error fetching the latest vaas to push for chunk ${chunkIndex}`,
+      );
       return;
     }
 
@@ -233,7 +261,10 @@ export class InjectivePricePusher implements IPricePusher {
       const json = Buffer.from(data).toString();
       updateFeeQueryResponse = JSON.parse(json);
     } catch (err) {
-      this.logger.error(err, "Error fetching update fee");
+      this.logger.error(
+        err,
+        `Error fetching update fee for chunk ${chunkIndex}`,
+      );
       // Throwing an error because it is likely an RPC issue
       throw err;
     }
@@ -247,21 +278,27 @@ export class InjectivePricePusher implements IPricePusher {
       });
 
       const rs = await this.signAndBroadcastMsg(executeMsg);
-      this.logger.info({ hash: rs.txHash }, "Succesfully broadcasted txHash");
+      this.logger.info(
+        { hash: rs.txHash },
+        `Successfully broadcasted txHash for chunk ${chunkIndex}`,
+      );
     } catch (err: any) {
       if (err.message.match(/account inj[a-zA-Z0-9]+ not found/) !== null) {
-        this.logger.error(err, "Account not found");
+        this.logger.error(err, `Account not found for chunk ${chunkIndex}`);
         throw new Error("Please check the mnemonic");
       }
 
       if (
         err.message.match(/insufficient/) !== null &&
         err.message.match(/funds/) !== null
       ) {
-        this.logger.error(err, "Insufficient funds");
+        this.logger.error(err, `Insufficient funds for chunk ${chunkIndex}`);
         throw new Error("Insufficient funds");
       }
-      this.logger.error(err, "Error executing messages");
+      this.logger.error(
+        err,
+        `Error executing messages for chunk ${chunkIndex}`,
+      );
     }
   }
 }