diff --git a/src/adapters/hotshot.rs b/src/adapters/hotshot.rs index 79b6862..6aea78f 100644 --- a/src/adapters/hotshot.rs +++ b/src/adapters/hotshot.rs @@ -87,6 +87,17 @@ impl Adapter for HotshotAdapter { Ok(rpc_urls) } + async fn ping_endpoint( + &self, + rpc_url: &str, + timeout: Option, + ) -> Result { + let req_id = rand::rng().random::(); + let _ = jrpc::send(rpc_url, req_id, RPC_METHOD, serde_json::json!({}), timeout).await?; + + Ok(true) + } + async fn send_txs( &self, rpc_url: &str, diff --git a/src/adapters/mod.rs b/src/adapters/mod.rs index 203e30b..b5b060e 100644 --- a/src/adapters/mod.rs +++ b/src/adapters/mod.rs @@ -14,6 +14,15 @@ pub trait Adapter { args: HashMap, ) -> impl std::future::Future, common::TestrpcError>> + Send; + /// Ping the given RPC URL to check if it's reachable. + /// This function should be implemented by each adapter to ping the RPC URL. + /// Returns a future that resolves to a boolean indicating if the ping was successful. + fn ping_endpoint( + &self, + rpc_url: &str, + timeout: Option, + ) -> impl std::future::Future> + Send; + /// Send transactions to the given RPC URL. /// This function should be implemented by each adapter to send transactions to the RPC URL. /// Returns a future that resolves to RoundResults. diff --git a/src/bin/testrpc.rs b/src/bin/testrpc.rs index 9d3c648..780de16 100644 --- a/src/bin/testrpc.rs +++ b/src/bin/testrpc.rs @@ -35,7 +35,7 @@ async fn main() -> Result<(), common::TestrpcError> { let ctx = Arc::new(ctx::Context::new()); let start = std::time::Instant::now(); - tracing::info!("XXX1 Starting testrpc with config file: {}", &opts.file); + tracing::info!("Starting testrpc with config file: {}", &opts.file); if opts.dry_run { tracing::info!("Dry run, we will not send any RPCs"); diff --git a/src/runner.rs b/src/runner.rs index b0e10f1..020a669 100644 --- a/src/runner.rs +++ b/src/runner.rs @@ -1,6 +1,6 @@ use futures::future::join_all; use std::collections::HashMap; -use std::sync::{Arc, RwLock}; +use std::sync::{atomic, Arc, RwLock}; use tokio::task; use tokio::time::Duration; @@ -14,10 +14,35 @@ pub async fn load_endpoints(cfg: config::Config) -> Result, TestrpcE return Ok(rpcs); } let adapter = adapters::new_adapter(cfg.adapter)?; - adapter + let endpoints = adapter .load_endpoints(cfg.args.clone()) .await - .map_err(|e| TestrpcError::LoadEndpointsError(e.to_string())) + .map_err(|e| TestrpcError::LoadEndpointsError(e.to_string()))?; + // for each endpoint call adapter.ping_endpoint to check if it's reachable + let reachable_endpoints = Arc::new(atomic::AtomicUsize::new(0)); + let mut handles = Vec::new(); + for endpoint in endpoints.clone() { + let adapter = adapter.clone(); + let endpoint_clone = endpoint.clone(); + let timeout = cfg.timeout.map(|t| Duration::from_secs(t as u64)); + let reachable_endpoints = Arc::clone(&reachable_endpoints); + let handle = tokio::spawn(async move { + let res = adapter.ping_endpoint(&endpoint_clone, timeout).await; + tracing::info!("Pinged endpoint {}: {:?}", endpoint_clone, res); + if res.is_ok() { + reachable_endpoints.fetch_add(1, atomic::Ordering::SeqCst); + } + res + }); + handles.push(handle); + } + join_all(handles).await; + tracing::info!( + "Loaded {} endpoints, {} live", + endpoints.len(), + reachable_endpoints.load(atomic::Ordering::SeqCst) + ); + Ok(endpoints) } /// Run the test flow with the given configuration.