diff --git a/Cargo.lock b/Cargo.lock index d6fa5cc..94d0d31 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4652,12 +4652,14 @@ dependencies = [ "jsonrpsee", "lazy_static", "metrics", + "reqwest", "serde_json", "shared_types", "storage", "task_executor", "thiserror", "tokio", + "url", ] [[package]] diff --git a/node/log_entry_sync/Cargo.toml b/node/log_entry_sync/Cargo.toml index c947194..aa07e39 100644 --- a/node/log_entry_sync/Cargo.toml +++ b/node/log_entry_sync/Cargo.toml @@ -24,3 +24,5 @@ futures-util = "0.3.28" thiserror = "1.0.44" lazy_static = "1.4.0" metrics = { workspace = true } +reqwest = {version = "0.11", features = ["json"]} +url = { version = "2.4", default-features = false } diff --git a/node/log_entry_sync/src/sync_manager/config.rs b/node/log_entry_sync/src/sync_manager/config.rs index d9d281f..5b0d190 100644 --- a/node/log_entry_sync/src/sync_manager/config.rs +++ b/node/log_entry_sync/src/sync_manager/config.rs @@ -1,3 +1,5 @@ +use std::time::Duration; + use crate::ContractAddress; pub struct LogSyncConfig { @@ -34,6 +36,9 @@ pub struct LogSyncConfig { pub watch_loop_wait_time_ms: u64, // force to sync log from start block number pub force_log_sync_from_start_block_number: bool, + + // the timeout for blockchain rpc connection + pub blockchain_rpc_timeout: Duration, } #[derive(Clone)] @@ -61,6 +66,7 @@ impl LogSyncConfig { remove_finalized_block_interval_minutes: u64, watch_loop_wait_time_ms: u64, force_log_sync_from_start_block_number: bool, + blockchain_rpc_timeout: Duration, ) -> Self { Self { rpc_endpoint_url, @@ -77,6 +83,7 @@ impl LogSyncConfig { remove_finalized_block_interval_minutes, watch_loop_wait_time_ms, force_log_sync_from_start_block_number, + blockchain_rpc_timeout, } } } diff --git a/node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs b/node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs index 29f8287..1aea564 100644 --- a/node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs +++ b/node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs @@ -1,6 +1,6 @@ use crate::sync_manager::log_query::LogQuery; use crate::sync_manager::RETRY_WAIT_MS; -use crate::ContractAddress; +use crate::{ContractAddress, LogSyncConfig}; use anyhow::{anyhow, bail, Result}; use append_merkle::{Algorithm, Sha3Algorithm}; use contract_interface::{SubmissionNode, SubmitFilter, ZgsFlow}; @@ -12,7 +12,6 @@ use futures::StreamExt; use jsonrpsee::tracing::{debug, error, info, warn}; use shared_types::{DataRoot, Transaction}; use std::collections::{BTreeMap, HashMap}; -use std::str::FromStr; use std::sync::Arc; use std::time::Duration; use storage::log_store::{tx_store::BlockHashAndSubmissionIndex, Store}; @@ -34,28 +33,29 @@ pub struct LogEntryFetcher { } impl LogEntryFetcher { - pub async fn new( - url: &str, - contract_address: ContractAddress, - log_page_size: u64, - confirmation_delay: u64, - rate_limit_retries: u32, - timeout_retries: u32, - initial_backoff: u64, - ) -> Result { + pub async fn new(config: &LogSyncConfig) -> Result { let provider = Arc::new(Provider::new( RetryClientBuilder::default() - .rate_limit_retries(rate_limit_retries) - .timeout_retries(timeout_retries) - .initial_backoff(Duration::from_millis(initial_backoff)) - .build(Http::from_str(url)?, Box::new(HttpRateLimitRetryPolicy)), + .rate_limit_retries(config.rate_limit_retries) + .timeout_retries(config.timeout_retries) + .initial_backoff(Duration::from_millis(config.initial_backoff)) + .build( + Http::new_with_client( + url::Url::parse(&config.rpc_endpoint_url)?, + reqwest::Client::builder() + .timeout(config.blockchain_rpc_timeout) + .connect_timeout(config.blockchain_rpc_timeout) + .build()?, + ), + Box::new(HttpRateLimitRetryPolicy), + ), )); // TODO: `error` types are removed from the ABI json file. Ok(Self { - contract_address, + contract_address: config.contract_address, provider, - log_page_size, - confirmation_delay, + log_page_size: config.log_page_size, + confirmation_delay: config.confirmation_block_count, }) } diff --git a/node/log_entry_sync/src/sync_manager/mod.rs b/node/log_entry_sync/src/sync_manager/mod.rs index 2412990..a37e31a 100644 --- a/node/log_entry_sync/src/sync_manager/mod.rs +++ b/node/log_entry_sync/src/sync_manager/mod.rs @@ -86,16 +86,7 @@ impl LogSyncManager { .expect("shutdown send error") }, async move { - let log_fetcher = LogEntryFetcher::new( - &config.rpc_endpoint_url, - config.contract_address, - config.log_page_size, - config.confirmation_block_count, - config.rate_limit_retries, - config.timeout_retries, - config.initial_backoff, - ) - .await?; + let log_fetcher = LogEntryFetcher::new(&config).await?; let data_cache = DataCache::new(config.cache_config.clone()); let block_hash_cache = Arc::new(RwLock::new( diff --git a/node/src/config/convert.rs b/node/src/config/convert.rs index aae214f..4322fc3 100644 --- a/node/src/config/convert.rs +++ b/node/src/config/convert.rs @@ -146,6 +146,7 @@ impl ZgsConfig { self.remove_finalized_block_interval_minutes, self.watch_loop_wait_time_ms, self.force_log_sync_from_start_block_number, + Duration::from_secs(self.blockchain_rpc_timeout_secs), )) } diff --git a/node/src/config/mod.rs b/node/src/config/mod.rs index 0ed1267..8e9183b 100644 --- a/node/src/config/mod.rs +++ b/node/src/config/mod.rs @@ -49,6 +49,8 @@ build_config! { (remove_finalized_block_interval_minutes, (u64), 30) (watch_loop_wait_time_ms, (u64), 500) + (blockchain_rpc_timeout_secs, (u64), 120) + // chunk pool (chunk_pool_write_window_size, (usize), 4) (chunk_pool_max_cached_chunks_all, (usize), 4*1024*1024) // 1G