more robust provider (#157)
Some checks are pending
abi-consistent-check / build-and-compare (push) Waiting to run
code-coverage / unittest-cov (push) Waiting to run
rust / check (push) Waiting to run
rust / test (push) Waiting to run
rust / lints (push) Waiting to run
functional-test / test (push) Waiting to run

* more robust provider
This commit is contained in:
0g-peterzhb 2024-08-12 11:35:21 +08:00 committed by GitHub
parent 53449e1faa
commit f03f97c609
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 55 additions and 12 deletions

View File

@ -1,9 +1,16 @@
use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;
use ethereum_types::{Address, H256, U256}; use ethereum_types::{Address, H256, U256};
use ethers::core::k256::SecretKey; use ethers::core::k256::SecretKey;
use ethers::middleware::SignerMiddleware; use ethers::middleware::SignerMiddleware;
use ethers::providers::Http; use ethers::providers::Http;
use ethers::providers::HttpRateLimitRetryPolicy;
use ethers::providers::Middleware; use ethers::providers::Middleware;
use ethers::providers::Provider; use ethers::providers::Provider;
use ethers::providers::RetryClient;
use ethers::providers::RetryClientBuilder;
use ethers::signers::LocalWallet; use ethers::signers::LocalWallet;
use ethers::signers::Signer; use ethers::signers::Signer;
use storage::config::ShardConfig; use storage::config::ShardConfig;
@ -18,9 +25,12 @@ pub struct MinerConfig {
pub(crate) cpu_percentage: u64, pub(crate) cpu_percentage: u64,
pub(crate) iter_batch: usize, pub(crate) iter_batch: usize,
pub(crate) shard_config: ShardConfig, 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<Provider<Http>, LocalWallet>; pub type MineServiceMiddleware = SignerMiddleware<Arc<Provider<RetryClient<Http>>>, LocalWallet>;
impl MinerConfig { impl MinerConfig {
#[allow(clippy::too_many_arguments)] #[allow(clippy::too_many_arguments)]
@ -34,6 +44,9 @@ impl MinerConfig {
cpu_percentage: u64, cpu_percentage: u64,
iter_batch: usize, iter_batch: usize,
shard_config: ShardConfig, shard_config: ShardConfig,
rate_limit_retries: u32,
timeout_retries: u32,
initial_backoff: u64,
) -> Option<MinerConfig> { ) -> Option<MinerConfig> {
miner_key.map(|miner_key| MinerConfig { miner_key.map(|miner_key| MinerConfig {
miner_id, miner_id,
@ -45,12 +58,24 @@ impl MinerConfig {
cpu_percentage, cpu_percentage,
iter_batch, iter_batch,
shard_config, shard_config,
rate_limit_retries,
timeout_retries,
initial_backoff,
}) })
} }
pub(crate) async fn make_provider(&self) -> Result<MineServiceMiddleware, String> { pub(crate) async fn make_provider(&self) -> Result<MineServiceMiddleware, String> {
let provider = Provider::<Http>::try_from(&self.rpc_endpoint_url) let provider = Arc::new(Provider::new(
.map_err(|e| format!("Can not parse blockchain endpoint: {:?}", e))?; 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 let chain_id = provider
.get_chainid() .get_chainid()
.await .await
@ -59,6 +84,7 @@ impl MinerConfig {
.map_err(|e| format!("Cannot parse private key: {:?}", e))?; .map_err(|e| format!("Cannot parse private key: {:?}", e))?;
let signer = LocalWallet::from(secret_key).with_chain_id(chain_id.as_u64()); let signer = LocalWallet::from(secret_key).with_chain_id(chain_id.as_u64());
let middleware = SignerMiddleware::new(provider, signer); let middleware = SignerMiddleware::new(provider, signer);
Ok(middleware) Ok(middleware)
} }
} }

View File

@ -1,11 +1,13 @@
use anyhow::{anyhow, bail, Result}; use anyhow::{bail, Result};
use contract_interface::ChunkLinearReward; use contract_interface::ChunkLinearReward;
use ethereum_types::Address; use ethereum_types::Address;
use ethers::prelude::{Http, Provider}; use ethers::prelude::{Http, Provider};
use ethers::providers::{HttpRateLimitRetryPolicy, RetryClient, RetryClientBuilder};
use miner::MinerMessage; use miner::MinerMessage;
use rand::Rng; use rand::Rng;
use std::cmp::Ordering; use std::cmp::Ordering;
use std::path::PathBuf; use std::path::PathBuf;
use std::str::FromStr;
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use storage::config::{ShardConfig, SHARD_CONFIG_KEY}; use storage::config::{ShardConfig, SHARD_CONFIG_KEY};
@ -34,17 +36,16 @@ pub struct PrunerConfig {
pub rpc_endpoint_url: String, pub rpc_endpoint_url: String,
pub reward_address: Address, pub reward_address: Address,
pub rate_limit_retries: u32,
pub timeout_retries: u32,
pub initial_backoff: u64,
} }
impl PrunerConfig { impl PrunerConfig {
fn start_prune_size(&self) -> u64 { fn start_prune_size(&self) -> u64 {
(self.max_num_sectors as f32 * PRUNE_THRESHOLD) as u64 (self.max_num_sectors as f32 * PRUNE_THRESHOLD) as u64
} }
fn make_provider(&self) -> Result<Provider<Http>> {
Provider::<Http>::try_from(&self.rpc_endpoint_url)
.map_err(|e| anyhow!("Can not parse blockchain endpoint: {:?}", e))
}
} }
pub struct Pruner { pub struct Pruner {
@ -57,7 +58,7 @@ pub struct Pruner {
sender: mpsc::UnboundedSender<PrunerMessage>, sender: mpsc::UnboundedSender<PrunerMessage>,
miner_sender: Option<broadcast::Sender<MinerMessage>>, miner_sender: Option<broadcast::Sender<MinerMessage>>,
reward_contract: ChunkLinearReward<Provider<Http>>, reward_contract: ChunkLinearReward<Arc<Provider<RetryClient<Http>>>>,
} }
impl Pruner { impl Pruner {
@ -73,8 +74,18 @@ impl Pruner {
let (first_rewardable_chunk, first_tx_seq) = get_first_rewardable_chunk(store.as_ref()) let (first_rewardable_chunk, first_tx_seq) = get_first_rewardable_chunk(store.as_ref())
.await? .await?
.unwrap_or((0, 0)); .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 (tx, rx) = mpsc::unbounded_channel();
let pruner = Pruner { let pruner = Pruner {
config, config,

View File

@ -181,6 +181,9 @@ impl ZgsConfig {
cpu_percentage, cpu_percentage,
iter_batch, iter_batch,
shard_config, 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), batch_wait_time: Duration::from_millis(self.prune_batch_wait_time_ms),
rpc_endpoint_url: self.blockchain_rpc_endpoint.clone(), rpc_endpoint_url: self.blockchain_rpc_endpoint.clone(),
reward_address, reward_address,
rate_limit_retries: self.rate_limit_retries,
timeout_retries: self.timeout_retries,
initial_backoff: self.initial_backoff,
})) }))
} else { } else {
Ok(None) Ok(None)