From f03f97c609e9edd1578c0fa56a86f93856689a77 Mon Sep 17 00:00:00 2001 From: 0g-peterzhb <158457852+0g-peterzhb@users.noreply.github.com> Date: Mon, 12 Aug 2024 11:35:21 +0800 Subject: [PATCH] more robust provider (#157) * more robust provider --- node/miner/src/config.rs | 32 +++++++++++++++++++++++++++++--- node/pruner/src/lib.rs | 29 ++++++++++++++++++++--------- node/src/config/convert.rs | 6 ++++++ 3 files changed, 55 insertions(+), 12 deletions(-) diff --git a/node/miner/src/config.rs b/node/miner/src/config.rs index 2fefad1..ea8a059 100644 --- a/node/miner/src/config.rs +++ b/node/miner/src/config.rs @@ -1,9 +1,16 @@ +use std::str::FromStr; +use std::sync::Arc; +use std::time::Duration; + use ethereum_types::{Address, H256, U256}; use ethers::core::k256::SecretKey; use ethers::middleware::SignerMiddleware; use ethers::providers::Http; +use ethers::providers::HttpRateLimitRetryPolicy; use ethers::providers::Middleware; use ethers::providers::Provider; +use ethers::providers::RetryClient; +use ethers::providers::RetryClientBuilder; use ethers::signers::LocalWallet; use ethers::signers::Signer; use storage::config::ShardConfig; @@ -18,9 +25,12 @@ pub struct MinerConfig { pub(crate) cpu_percentage: u64, pub(crate) iter_batch: usize, pub(crate) shard_config: ShardConfig, + pub(crate) rate_limit_retries: u32, + pub(crate) timeout_retries: u32, + pub(crate) initial_backoff: u64, } -pub type MineServiceMiddleware = SignerMiddleware, LocalWallet>; +pub type MineServiceMiddleware = SignerMiddleware>>, LocalWallet>; impl MinerConfig { #[allow(clippy::too_many_arguments)] @@ -34,6 +44,9 @@ impl MinerConfig { cpu_percentage: u64, iter_batch: usize, shard_config: ShardConfig, + rate_limit_retries: u32, + timeout_retries: u32, + initial_backoff: u64, ) -> Option { miner_key.map(|miner_key| MinerConfig { miner_id, @@ -45,12 +58,24 @@ impl MinerConfig { cpu_percentage, iter_batch, shard_config, + rate_limit_retries, + timeout_retries, + initial_backoff, }) } pub(crate) async fn make_provider(&self) -> Result { - let provider = Provider::::try_from(&self.rpc_endpoint_url) - .map_err(|e| format!("Can not parse blockchain endpoint: {:?}", e))?; + let provider = Arc::new(Provider::new( + RetryClientBuilder::default() + .rate_limit_retries(self.rate_limit_retries) + .timeout_retries(self.timeout_retries) + .initial_backoff(Duration::from_millis(self.initial_backoff)) + .build( + Http::from_str(&self.rpc_endpoint_url) + .map_err(|e| format!("Cannot parse blockchain endpoint: {:?}", e))?, + Box::new(HttpRateLimitRetryPolicy), + ), + )); let chain_id = provider .get_chainid() .await @@ -59,6 +84,7 @@ impl MinerConfig { .map_err(|e| format!("Cannot parse private key: {:?}", e))?; let signer = LocalWallet::from(secret_key).with_chain_id(chain_id.as_u64()); let middleware = SignerMiddleware::new(provider, signer); + Ok(middleware) } } diff --git a/node/pruner/src/lib.rs b/node/pruner/src/lib.rs index 111b7d7..514ac1e 100644 --- a/node/pruner/src/lib.rs +++ b/node/pruner/src/lib.rs @@ -1,11 +1,13 @@ -use anyhow::{anyhow, bail, Result}; +use anyhow::{bail, Result}; use contract_interface::ChunkLinearReward; use ethereum_types::Address; use ethers::prelude::{Http, Provider}; +use ethers::providers::{HttpRateLimitRetryPolicy, RetryClient, RetryClientBuilder}; use miner::MinerMessage; use rand::Rng; use std::cmp::Ordering; use std::path::PathBuf; +use std::str::FromStr; use std::sync::Arc; use std::time::Duration; use storage::config::{ShardConfig, SHARD_CONFIG_KEY}; @@ -34,17 +36,16 @@ pub struct PrunerConfig { pub rpc_endpoint_url: String, pub reward_address: Address, + + pub rate_limit_retries: u32, + pub timeout_retries: u32, + pub initial_backoff: u64, } impl PrunerConfig { fn start_prune_size(&self) -> u64 { (self.max_num_sectors as f32 * PRUNE_THRESHOLD) as u64 } - - fn make_provider(&self) -> Result> { - Provider::::try_from(&self.rpc_endpoint_url) - .map_err(|e| anyhow!("Can not parse blockchain endpoint: {:?}", e)) - } } pub struct Pruner { @@ -57,7 +58,7 @@ pub struct Pruner { sender: mpsc::UnboundedSender, miner_sender: Option>, - reward_contract: ChunkLinearReward>, + reward_contract: ChunkLinearReward>>>, } impl Pruner { @@ -73,8 +74,18 @@ impl Pruner { let (first_rewardable_chunk, first_tx_seq) = get_first_rewardable_chunk(store.as_ref()) .await? .unwrap_or((0, 0)); - let reward_contract = - ChunkLinearReward::new(config.reward_address, Arc::new(config.make_provider()?)); + + let provider = Arc::new(Provider::new( + RetryClientBuilder::default() + .rate_limit_retries(config.rate_limit_retries) + .timeout_retries(config.timeout_retries) + .initial_backoff(Duration::from_millis(config.initial_backoff)) + .build( + Http::from_str(&config.rpc_endpoint_url)?, + Box::new(HttpRateLimitRetryPolicy), + ), + )); + let reward_contract = ChunkLinearReward::new(config.reward_address, Arc::new(provider)); let (tx, rx) = mpsc::unbounded_channel(); let pruner = Pruner { config, diff --git a/node/src/config/convert.rs b/node/src/config/convert.rs index eff984c..a363984 100644 --- a/node/src/config/convert.rs +++ b/node/src/config/convert.rs @@ -181,6 +181,9 @@ impl ZgsConfig { cpu_percentage, iter_batch, shard_config, + self.rate_limit_retries, + self.timeout_retries, + self.initial_backoff, )) } @@ -216,6 +219,9 @@ impl ZgsConfig { batch_wait_time: Duration::from_millis(self.prune_batch_wait_time_ms), rpc_endpoint_url: self.blockchain_rpc_endpoint.clone(), reward_address, + rate_limit_retries: self.rate_limit_retries, + timeout_retries: self.timeout_retries, + initial_backoff: self.initial_backoff, })) } else { Ok(None)