Merge pull request #608 from eco-stake/add-autostake-throttle

Add configurable throttle to autostake

README.md

@@ -181,9 +181,11 @@ Create a `src/networks.local.json` file and specify the networks you want to ove
     "gasPrice": "0.0025uosmo",
     "autostake": {
       "retries": 3,
+      "batchPageSize": 100,
+      "batchQueries": 25,
       "batchTxs": 50,
-      "batchQueries": 50,
-      "delegatorTimeout": 5000,
+      "queryTimeout": 5000,
+      "queryThrottle": 100,
       "gasModifier": 1.1
     },
     "healthCheck": {

src/autostake/NetworkRunner.mjs

@@ -11,11 +11,20 @@ export default class NetworkRunner {
     this.operator = operator
     this.signingClient = signingClient
     this.queryClient = network.queryClient
-    this.errors = {}
-    this.opts = opts || {}
+    this.opts = {
+      batchPageSize: 100,
+      batchQueries: 25,
+      batchTxs: 50,
+      queryTimeout: network.data.autostake?.delegatorTimeout || 5000, // deprecate
+      queryThrottle: 100,
+      gasModifier: 1.1,
+      ...network.data.autostake,
+      ...opts
+    }
     this.batch = {}
     this.messages = []
     this.processed = {}
+    this.errors = {}
     this.results = []
   }
 
@@ -50,6 +59,7 @@ export default class NetworkRunner {
 
   async run(addresses) {
     try {
+      timeStamp("Running with options:", this.opts)
       this.balance = await this.getBalance() || 0
 
       if (addresses) {
@@ -80,32 +90,40 @@ export default class NetworkRunner {
   }
 
   getBalance() {
-    return this.queryClient.getBalance(this.operator.botAddress, this.network.denom)
+    let timeout = this.opts.queryTimeout
+    return this.queryClient.getBalance(this.operator.botAddress, this.network.denom, { timeout })
       .then(
         (balance) => {
           timeStamp("Bot balance is", balance.amount, balance.denom)
           return balance.amount
         },
         (error) => {
-          throw new Error("Failed to get balance:", error.message || error)
+          throw new Error(`Failed to get balance: ${error.message || error}`)
         }
       )
   }
 
   getDelegations() {
-    let batchSize = this.network.data.autostake?.batchQueries || 100
-    return this.queryClient.getAllValidatorDelegations(this.operator.address, batchSize, (pages) => {
+    let timeout = this.opts.queryTimeout
+    let pageSize = this.opts.batchPageSize
+    return this.queryClient.getAllValidatorDelegations(this.operator.address, pageSize, { timeout }, (pages) => {
       timeStamp("...batch", pages.length)
+      return this.throttleQuery()
     }).catch(error => {
       throw new Error(`Failed to get delegations: ${error.message || error}`)
     })
   }
 
   async getGrantedAddresses(addresses) {
     const { botAddress, address } = this.operator
+    let timeout = this.opts.queryTimeout
+    let pageSize = this.opts.batchPageSize
     let allGrants
     try {
-      allGrants = await this.queryClient.getGranteeGrants(botAddress)
+      allGrants = await this.queryClient.getGranteeGrants(botAddress, { timeout, pageSize }, (pages) => {
+      timeStamp("...batch", pages.length)
+      return this.throttleQuery()
+    })
     } catch (e) { }
     let grantCalls = addresses.map(item => {
       return async () => {
@@ -117,16 +135,16 @@ export default class NetworkRunner {
         }
       }
     })
-    let batchSize = this.network.data.autostake?.batchQueries || 50
-    let grantedAddresses = await mapSync(grantCalls, batchSize, (batch, index) => {
+    let grantedAddresses = await mapSync(grantCalls, this.opts.batchQueries, (batch, index) => {
       timeStamp('...batch', index + 1)
+      return this.throttleQuery()
     })
     return _.compact(grantedAddresses.flat())
   }
 
   getGrants(delegatorAddress) {
     const { botAddress, address } = this.operator
-    let timeout = this.network.data.autostake?.delegatorTimeout || 5000
+    let timeout = this.opts.queryTimeout
     return this.queryClient.getGrants(botAddress, delegatorAddress, { timeout })
       .then(
         (result) => {
@@ -165,7 +183,7 @@ export default class NetworkRunner {
   async getAutostakeMessage(grantAddress) {
     const { address, grant } = grantAddress
 
-    let timeout = this.network.data.autostake?.delegatorTimeout || 5000
+    let timeout = this.opts.queryTimeout
     let withdrawAddress, totalRewards
     try {
       withdrawAddress = await this.queryClient.getWithdrawAddress(address, { timeout })
@@ -210,7 +228,7 @@ export default class NetworkRunner {
   }
 
   async autostake(grantedAddresses) {
-    let batchSize = this.network.data.autostake?.batchTxs || 50
+    let batchSize = this.opts.batchTxs
     timeStamp('Calculating and autostaking in batches of', batchSize)
 
     const calls = grantedAddresses.map((item, index) => {
@@ -226,8 +244,7 @@ export default class NetworkRunner {
         await this.sendInBatches(item.address, messages, batchSize, grantedAddresses.length)
       }
     })
-    let querySize = this.network.data.autostake?.batchQueries || _.clamp(batchSize / 2, 50)
-    await executeSync(calls, querySize)
+    await executeSync(calls, this.opts.batchQueries)
 
     this.results = await Promise.all(this.messages)
   }
@@ -258,7 +275,7 @@ export default class NetworkRunner {
     try {
       const execMsg = this.buildExecMessage(this.operator.botAddress, messages)
       const memo = 'REStaked by ' + this.operator.moniker
-      const gasModifier = this.network.data.autostake?.gasModifier || 1.1
+      const gasModifier = this.opts.gasModifier
       const gas = await this.signingClient.simulate(this.operator.botAddress, [execMsg], memo, gasModifier);
       const fee = this.signingClient.getFee(gas).amount[0]
       if (smaller(bignumber(this.balance), bignumber(fee.amount))) {
@@ -290,6 +307,12 @@ export default class NetworkRunner {
     }
   }
 
+  async throttleQuery(){
+    if(!this.opts.queryThrottle) return 
+
+    await new Promise(r => setTimeout(r, this.opts.queryThrottle));
+  }
+
   buildExecMessage(botAddress, messages) {
     return {
       typeUrl: "/cosmos.authz.v1beta1.MsgExec",
@@ -312,7 +335,7 @@ export default class NetworkRunner {
   }
 
   totalRewards(address) {
-    let timeout = this.network.data.autostake?.delegatorTimeout || 5000
+    let timeout = this.opts.queryTimeout
     return this.queryClient.getRewards(address, { timeout })
       .then(
         (rewards) => {

src/autostake/index.mjs

@@ -142,7 +142,8 @@ export default function Autostake(mnemonic, opts) {
 
     if (usingDirectory) {
       timeStamp('You are using public nodes, they may not be reliable. Check the README to use your own')
-      timeStamp('Delaying briefly to reduce load...')
+      timeStamp('Delaying briefly and adjusting config to reduce load...')
+      opts = {...opts, batchPageSize: 50, batchQueries: 10, queryThrottle: 2500}
       await new Promise(r => setTimeout(r, (Math.random() * 31) * 1000));
     }
 

src/utils/Helpers.mjs

@@ -112,7 +112,7 @@ export async function mapSync(calls, count, batchCallback) {
   for (const batchCall of batchCalls) {
     const batchResults = await mapAsync(batchCall, call => call())
     results.push(batchResults)
-    if (batchCallback) batchCallback(batchResults, index)
+    if (batchCallback) await batchCallback(batchResults, index)
     index++
   }
   return results.flat()

src/utils/QueryClient.mjs

@@ -37,16 +37,17 @@ const QueryClient = async (chainId, restUrls) => {
   const getAllValidatorDelegations = (
     validatorAddress,
     pageSize,
+    opts,
     pageCallback
   ) => {
     return getAllPages((nextKey) => {
-      return getValidatorDelegations(validatorAddress, pageSize, nextKey);
+      return getValidatorDelegations(validatorAddress, pageSize, opts, nextKey);
     }, pageCallback).then((pages) => {
       return pages.map((el) => el.delegation_responses).flat();
     });
   };
 
-  const getValidatorDelegations = (validatorAddress, pageSize, nextKey) => {
+  const getValidatorDelegations = (validatorAddress, pageSize, opts, nextKey) => {
     const searchParams = new URLSearchParams();
     if (pageSize) searchParams.append("pagination.limit", pageSize);
     if (nextKey) searchParams.append("pagination.key", nextKey);
@@ -57,14 +58,15 @@ const QueryClient = async (chainId, restUrls) => {
           "/cosmos/staking/v1beta1/validators/" +
           validatorAddress +
           "/delegations?" +
-          searchParams.toString()
+          searchParams.toString(),
+          opts
       )
       .then((res) => res.data);
   };
 
-  const getBalance = (address, denom) => {
+  const getBalance = (address, denom, opts) => {
     return axios
-      .get(restUrl + "/cosmos/bank/v1beta1/balances/" + address)
+      .get(restUrl + "/cosmos/bank/v1beta1/balances/" + address, opts)
       .then((res) => res.data)
       .then((result) => {
         const balance = result.balances?.find(
@@ -137,7 +139,7 @@ const QueryClient = async (chainId, restUrls) => {
       .then((res) => res.data)
   };
 
-  const getGranteeGrants = (grantee, opts) => {
+  const getGranteeGrants = (grantee, opts, pageCallback) => {
     const { pageSize } = opts || {}
     return getAllPages((nextKey) => {
       const searchParams = new URLSearchParams();
@@ -148,12 +150,12 @@ const QueryClient = async (chainId, restUrls) => {
         .get(restUrl + "/cosmos/authz/v1beta1/grants/grantee/" + grantee + "?" +
           searchParams.toString(), opts)
         .then((res) => res.data)
-    }).then((pages) => {
+    }, pageCallback).then((pages) => {
       return pages.map(el => el.grants).flat();
     });
   };
 
-  const getGranterGrants = (granter, opts) => {
+  const getGranterGrants = (granter, opts, pageCallback) => {
     const { pageSize } = opts || {}
     return getAllPages((nextKey) => {
       const searchParams = new URLSearchParams();
@@ -164,7 +166,7 @@ const QueryClient = async (chainId, restUrls) => {
         .get(restUrl + "/cosmos/authz/v1beta1/grants/granter/" + granter + "?" +
           searchParams.toString(), opts)
         .then((res) => res.data)
-    }).then((pages) => {
+    }, pageCallback).then((pages) => {
       return pages.map(el => el.grants).flat();
     });
   };
@@ -202,7 +204,7 @@ const QueryClient = async (chainId, restUrls) => {
       const result = await getPage(nextKey);
       pages.push(result);
       nextKey = result.pagination.next_key;
-      if (pageCallback) pageCallback(pages);
+      if (pageCallback) await pageCallback(pages);
     } while (nextKey);
     return pages;
   };