add timeout for rpc connections (#263)
Some checks failed
abi-consistent-check / build-and-compare (push) Has been cancelled
code-coverage / unittest-cov (push) Has been cancelled
rust / check (push) Has been cancelled
rust / test (push) Has been cancelled
rust / lints (push) Has been cancelled
functional-test / test (push) Has been cancelled

This commit is contained in:
Joel Liu 2024-11-09 13:37:09 +08:00 committed by GitHub
parent fae2d5efb6
commit 0c493880ee
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 33 additions and 28 deletions

2
Cargo.lock generated
View File

@ -4652,12 +4652,14 @@ dependencies = [
"jsonrpsee",
"lazy_static",
"metrics",
"reqwest",
"serde_json",
"shared_types",
"storage",
"task_executor",
"thiserror",
"tokio",
"url",
]
[[package]]

View File

@ -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 }

View File

@ -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,
}
}
}

View File

@ -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<Self> {
pub async fn new(config: &LogSyncConfig) -> Result<Self> {
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,
})
}

View File

@ -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(

View File

@ -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),
))
}

View File

@ -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