[price_pusher] Create gas pool for Sui (#923)

* [sui][price_pusher]Implement gas pool

* Add the option to specify gas budget to avoid dry run

* Avoid for loop in constructing transaction block

* Improve error handling

* Implement coin consolidation

* minor tweaks

* k

* cleanup

* fix

---------

Co-authored-by: Chris Li <chris@mystenlabs.com>

price_pusher/README.md

@@ -91,18 +91,19 @@ npm run start -- aptos --endpoint https://fullnode.testnet.aptoslabs.com/v1 \
     [--polling-frequency 5] \
 
 # For Sui
-npm run start -- sui
-  --endpoint https://sui-testnet-rpc.allthatnode.com,
-  --pyth-package-id 0x975e063f398f720af4f33ec06a927f14ea76ca24f7f8dd544aa62ab9d5d15f44,
-  --pyth-state-id 0xd8afde3a48b4ff7212bd6829a150f43f59043221200d63504d981f62bff2e27a,
-  --wormhole-package-id 0xcc029e2810f17f9f43f52262f40026a71fbdca40ed3803ad2884994361910b7e,
-  --wormhole-state-id 0xebba4cc4d614f7a7cdbe883acc76d1cc767922bc96778e7b68be0d15fce27c02,
-  --price-feed-to-price-info-object-table-id 0xf8929174008c662266a1adde78e1e8e33016eb7ad37d379481e860b911e40ed5,
-  --price-service-endpoint https://xc-testnet.pyth.network,
-  --mnemonic-file ./mnemonic,
-  --price-config-file ./price-config.testnet.sample.yaml
+npm run start -- sui \
+  --endpoint https://sui-testnet-rpc.allthatnode.com \
+  --pyth-package-id 0x975e063f398f720af4f33ec06a927f14ea76ca24f7f8dd544aa62ab9d5d15f44 \
+  --pyth-state-id 0xd8afde3a48b4ff7212bd6829a150f43f59043221200d63504d981f62bff2e27a \
+  --wormhole-package-id 0xcc029e2810f17f9f43f52262f40026a71fbdca40ed3803ad2884994361910b7e \
+  --wormhole-state-id 0xebba4cc4d614f7a7cdbe883acc76d1cc767922bc96778e7b68be0d15fce27c02 \
+  --price-feed-to-price-info-object-table-id 0xf8929174008c662266a1adde78e1e8e33016eb7ad37d379481e860b911e40ed5 \
+  --price-service-endpoint https://xc-testnet.pyth.network \
+  --mnemonic-file ./mnemonic \
+  --price-config-file ./price-config.testnet.sample.yaml \
   [--pushing-frequency 10] \
   [--polling-frequency 5] \
+  [--num-gas-objects 30]
 
 
 # Or, run the price pusher docker image instead of building from the source

price_pusher/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@pythnetwork/price-pusher",
-  "version": "5.4.1",
+  "version": "5.4.2",
   "description": "Pyth Price Pusher",
   "homepage": "https://pyth.network",
   "main": "lib/index.js",

price_pusher/src/sui/command.ts

@@ -66,13 +66,25 @@ export default {
       required: true,
       default: 1,
     } as Options,
+    "num-gas-objects": {
+      description: "Number of gas objects in the pool.",
+      type: "number",
+      required: true,
+      default: 30,
+    } as Options,
+    "gas-budget": {
+      description: "Gas budget for each price update",
+      type: "number",
+      required: true,
+      default: 500_000_000,
+    } as Options,
     ...options.priceConfigFile,
     ...options.priceServiceEndpoint,
     ...options.mnemonicFile,
     ...options.pollingFrequency,
     ...options.pushingFrequency,
   },
-  handler: function (argv: any) {
+  handler: async function (argv: any) {
     const {
       endpoint,
       priceConfigFile,
@@ -86,6 +98,8 @@ export default {
       wormholeStateId,
       priceFeedToPriceInfoObjectTableId,
       maxVaasPerPtb,
+      numGasObjects,
+      gasBudget,
     } = argv;
 
     const priceConfigs = readPriceConfigFile(priceConfigFile);
@@ -128,7 +142,7 @@ export default {
       priceItems,
       { pollingFrequency }
     );
-    const suiPusher = new SuiPricePusher(
+    const suiPusher = await SuiPricePusher.createWithAutomaticGasPool(
       priceServiceConnection,
       pythPackageId,
       pythStateId,
@@ -137,7 +151,9 @@ export default {
       priceFeedToPriceInfoObjectTableId,
       maxVaasPerPtb,
       endpoint,
-      mnemonic
+      mnemonic,
+      gasBudget,
+      numGasObjects
     );
 
     const controller = new Controller(

price_pusher/src/sui/sui.ts

@@ -13,8 +13,18 @@ import {
   RawSigner,
   TransactionBlock,
   SUI_CLOCK_OBJECT_ID,
+  getCreatedObjects,
+  SuiObjectRef,
+  getTransactionEffects,
+  getExecutionStatusError,
+  PaginatedCoins,
+  SuiAddress,
 } from "@mysten/sui.js";
 
+const GAS_FEE_FOR_SPLIT = 2_000_000_000;
+// TODO: read this from on chain config
+const MAX_NUM_GAS_OBJECTS_IN_PTB = 256;
+const MAX_NUM_OBJECTS_IN_ARGUMENT = 510;
 export class SuiPriceListener extends ChainPriceListener {
   constructor(
     private pythPackageId: string,
@@ -82,12 +92,8 @@ export class SuiPriceListener extends ChainPriceListener {
 }
 
 export class SuiPricePusher implements IPricePusher {
-  private readonly signer: RawSigner;
-  // Sui transactions can error if they're sent concurrently. This flag tracks whether an update is in-flight,
-  // so we can skip sending another update at the same time.
-  private isAwaitingTx: boolean;
-
   constructor(
+    private readonly signer: RawSigner,
     private priceServiceConnection: PriceServiceConnection,
     private pythPackageId: string,
     private pythStateId: string,
@@ -96,13 +102,58 @@ export class SuiPricePusher implements IPricePusher {
     private priceFeedToPriceInfoObjectTableId: string,
     private maxVaasPerPtb: number,
     endpoint: string,
-    mnemonic: string
-  ) {
-    this.signer = new RawSigner(
+    mnemonic: string,
+    private gasBudget: number,
+    private gasPool: SuiObjectRef[]
+  ) {}
+
+  /**
+   * Create a price pusher with a pool of `numGasObjects` gas coins that will be used to send transactions.
+   * The gas coins of the wallet for the provided mnemonic will be merged and then evenly split into `numGasObjects`.
+   */
+  static async createWithAutomaticGasPool(
+    priceServiceConnection: PriceServiceConnection,
+    pythPackageId: string,
+    pythStateId: string,
+    wormholePackageId: string,
+    wormholeStateId: string,
+    priceFeedToPriceInfoObjectTableId: string,
+    maxVaasPerPtb: number,
+    endpoint: string,
+    mnemonic: string,
+    gasBudget: number,
+    numGasObjects: number
+  ): Promise<SuiPricePusher> {
+    if (numGasObjects > MAX_NUM_OBJECTS_IN_ARGUMENT) {
+      throw new Error(
+        `numGasObjects cannot be greater than ${MAX_NUM_OBJECTS_IN_ARGUMENT} until we implement split chunking`
+      );
+    }
+
+    const signer = new RawSigner(
       Ed25519Keypair.deriveKeypair(mnemonic),
       new JsonRpcProvider(new Connection({ fullnode: endpoint }))
     );
-    this.isAwaitingTx = false;
+
+    const gasPool = await SuiPricePusher.initializeGasPool(
+      signer,
+      numGasObjects
+    );
+
+    return new SuiPricePusher(
+      signer,
+      priceServiceConnection,
+      pythPackageId,
+      pythStateId,
+      wormholePackageId,
+      wormholeStateId,
+      priceFeedToPriceInfoObjectTableId,
+      maxVaasPerPtb,
+      endpoint,
+      mnemonic,
+      gasBudget,
+      gasPool
+    );
   }
 
   async updatePriceFeed(
@@ -116,10 +167,8 @@ export class SuiPricePusher implements IPricePusher {
     if (priceIds.length !== pubTimesToPush.length)
       throw new Error("Invalid arguments");
 
-    if (this.isAwaitingTx) {
-      console.log(
-        "Skipping update: previous price update transaction(s) have not completed."
-      );
+    if (this.gasPool.length === 0) {
+      console.warn("Skipping update: no available gas coin.");
       return;
     }
 
@@ -127,7 +176,7 @@ export class SuiPricePusher implements IPricePusher {
       priceIds
     );
     if (priceFeeds === undefined) {
-      console.log("Failed to fetch price updates. Skipping push.");
+      console.warn("Failed to fetch price updates. Skipping push.");
       return;
     }
 
@@ -161,12 +210,7 @@ export class SuiPricePusher implements IPricePusher {
       }
     }
 
-    try {
-      this.isAwaitingTx = true;
-      await this.sendTransactionBlocks(txs);
-    } finally {
-      this.isAwaitingTx = false;
-    }
+    await this.sendTransactionBlocks(txs);
   }
 
   private async createPriceUpdateTransaction(
@@ -244,34 +288,199 @@ export class SuiPricePusher implements IPricePusher {
     return tx;
   }
 
-  /** Send every transaction in txs sequentially, returning when all transactions have completed. */
-  private async sendTransactionBlocks(txs: TransactionBlock[]): Promise<void> {
-    for (const tx of txs) {
-      try {
-        const result = await this.signer.signAndExecuteTransactionBlock({
-          transactionBlock: tx,
-          options: {
-            showInput: true,
-            showEffects: true,
-            showEvents: true,
-            showObjectChanges: true,
-            showBalanceChanges: true,
-          },
-        });
-
-        console.log(
-          "Successfully updated price with transaction digest ",
-          result.digest
+  /** Send every transaction in txs in parallel, returning when all transactions have completed. */
+  private async sendTransactionBlocks(
+    txs: TransactionBlock[]
+  ): Promise<void[]> {
+    return Promise.all(txs.map((tx) => this.sendTransactionBlock(tx)));
+  }
+
+  /** Send a single transaction block using a gas coin from the pool. */
+  private async sendTransactionBlock(tx: TransactionBlock): Promise<void> {
+    const gasObject = this.gasPool.shift();
+    if (gasObject === undefined) {
+      console.warn("No available gas coin. Skipping push.");
+      return;
+    }
+
+    let nextGasObject: SuiObjectRef | undefined = undefined;
+    try {
+      tx.setGasPayment([gasObject]);
+      tx.setGasBudget(this.gasBudget);
+      const result = await this.signer.signAndExecuteTransactionBlock({
+        transactionBlock: tx,
+        options: {
+          showEffects: true,
+        },
+      });
+
+      nextGasObject = getTransactionEffects(result)
+        ?.mutated?.map((obj) => obj.reference)
+        .find((ref) => ref.objectId === gasObject.objectId);
+
+      console.log(
+        "Successfully updated price with transaction digest ",
+        result.digest
+      );
+    } catch (e) {
+      console.log("Error when signAndExecuteTransactionBlock");
+      if (String(e).includes("GasBalanceTooLow")) {
+        console.warn(
+          `The balance of gas object ${gasObject.objectId} is too low. Removing from pool.`
+        );
+      } else {
+        nextGasObject = gasObject;
+      }
+      console.error(e);
+    }
+
+    if (nextGasObject !== undefined) {
+      this.gasPool.push(nextGasObject);
+    }
+  }
+
+  // This function will smash all coins owned by the signer into one, and then
+  // split them equally into numGasObjects.
+  private static async initializeGasPool(
+    signer: RawSigner,
+    numGasObjects: number
+  ): Promise<SuiObjectRef[]> {
+    const signerAddress = await signer.getAddress();
+    const { totalBalance: balance } = await signer.provider.getBalance({
+      owner: signerAddress,
+    });
+    const splitAmount =
+      (BigInt(balance) - BigInt(GAS_FEE_FOR_SPLIT)) / BigInt(numGasObjects);
+
+    const consolidatedCoin = await SuiPricePusher.mergeGasCoinsIntoOne(
+      signer,
+      signerAddress
+    );
+
+    const gasPool = await SuiPricePusher.splitGasCoinEqually(
+      signer,
+      signerAddress,
+      Number(splitAmount),
+      numGasObjects,
+      consolidatedCoin
+    );
+    console.log("Gas pool is filled with coins: ", gasPool);
+    return gasPool;
+  }
+
+  private static async getAllGasCoins(
+    provider: JsonRpcProvider,
+    owner: SuiAddress
+  ): Promise<SuiObjectRef[]> {
+    let hasNextPage = true;
+    let cursor;
+    const coins = new Set<string>([]);
+    let numCoins = 0;
+    while (hasNextPage) {
+      const paginatedCoins: PaginatedCoins = await provider.getCoins({
+        owner,
+        cursor,
+      });
+      numCoins += paginatedCoins.data.length;
+      paginatedCoins.data.forEach((c) =>
+        coins.add(
+          JSON.stringify({
+            objectId: c.coinObjectId,
+            version: c.version,
+            digest: c.digest,
+          })
+        )
+      );
+      hasNextPage = paginatedCoins.hasNextPage;
+      cursor = paginatedCoins.nextCursor;
+    }
+
+    if (numCoins !== coins.size) {
+      throw new Error("Unexpected getCoins result: duplicate coins found");
+    }
+    return [...coins].map((item) => JSON.parse(item));
+  }
+
+  private static async splitGasCoinEqually(
+    signer: RawSigner,
+    signerAddress: SuiAddress,
+    splitAmount: number,
+    numGasObjects: number,
+    gasCoin: SuiObjectRef
+  ): Promise<SuiObjectRef[]> {
+    // TODO: implement chunking if numGasObjects exceeds MAX_NUM_CREATED_OBJECTS
+    const tx = new TransactionBlock();
+    const coins = tx.splitCoins(
+      tx.gas,
+      Array.from({ length: numGasObjects }, () => tx.pure(splitAmount))
+    );
+
+    tx.transferObjects(
+      Array.from({ length: numGasObjects }, (_, i) => coins[i]),
+      tx.pure(signerAddress)
+    );
+    tx.setGasPayment([gasCoin]);
+    const result = await signer.signAndExecuteTransactionBlock({
+      transactionBlock: tx,
+      options: { showEffects: true },
+    });
+    const error = getExecutionStatusError(result);
+    if (error) {
+      throw new Error(
+        `Failed to initialize gas pool: ${error}. Try re-running the script`
+      );
+    }
+    const newCoins = getCreatedObjects(result)!.map((obj) => obj.reference);
+    if (newCoins.length !== numGasObjects) {
+      throw new Error(
+        `Failed to initialize gas pool. Expected ${numGasObjects}, got: ${newCoins}`
+      );
+    }
+    return newCoins;
+  }
+
+  private static async mergeGasCoinsIntoOne(
+    signer: RawSigner,
+    owner: SuiAddress
+  ): Promise<SuiObjectRef> {
+    const gasCoins = await SuiPricePusher.getAllGasCoins(
+      signer.provider,
+      owner
+    );
+    // skip merging if there is only one coin
+    if (gasCoins.length === 1) {
+      return gasCoins[0];
+    }
+
+    const gasCoinsChunks = chunkArray<SuiObjectRef>(
+      gasCoins,
+      MAX_NUM_GAS_OBJECTS_IN_PTB - 2
+    );
+    let finalCoin;
+
+    for (let i = 0; i < gasCoinsChunks.length; i++) {
+      const mergeTx = new TransactionBlock();
+      let coins = gasCoinsChunks[i];
+      if (finalCoin) {
+        coins = [finalCoin, ...coins];
+      }
+      mergeTx.setGasPayment(coins);
+      const mergeResult = await signer.signAndExecuteTransactionBlock({
+        transactionBlock: mergeTx,
+        options: { showEffects: true },
+      });
+      const error = getExecutionStatusError(mergeResult);
+      if (error) {
+        throw new Error(
+          `Failed to merge coins when initializing gas pool: ${error}. Try re-running the script`
         );
-      } catch (e) {
-        console.log("Error when signAndExecuteTransactionBlock");
-        if (String(e).includes("GasBalanceTooLow")) {
-          console.log("Insufficient Gas Amount. Please top up your account");
-          process.exit();
-        }
-        console.error(e);
       }
+      finalCoin = getTransactionEffects(mergeResult)!.mutated!.map(
+        (obj) => obj.reference
+      )[0];
     }
+
+    return finalCoin as SuiObjectRef;
   }
 }
 
@@ -318,3 +527,13 @@ async function priceIdToPriceInfoObjectId(
 
   return priceInfoObjectId;
 }
+
+function chunkArray<T>(array: Array<T>, size: number): Array<Array<T>> {
+  const chunked = [];
+  let index = 0;
+  while (index < array.length) {
+    chunked.push(array.slice(index, size + index));
+    index += size;
+  }
+  return chunked;
+}