Ping endpoints

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<std::time::Duration>,
+    ) -> Result<bool, crate::common::TestrpcError> {
+        let req_id = rand::rng().random::<u64>();
+        let _ = jrpc::send(rpc_url, req_id, RPC_METHOD, serde_json::json!({}), timeout).await?;
+
+        Ok(true)
+    }
+
     async fn send_txs(
         &self,
         rpc_url: &str,

src/adapters/mod.rs

@@ -14,6 +14,15 @@ pub trait Adapter {
         args: HashMap<String, Value>,
     ) -> impl std::future::Future<Output = Result<Vec<String>, 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<std::time::Duration>,
+    ) -> impl std::future::Future<Output = Result<bool, common::TestrpcError>> + 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.

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");

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<Vec<String>, 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.