Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions src/adapters/hotshot.rs
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,11 @@ impl Adapter for HotshotAdapter {
.map(|ip| format!("http://{ip}:{rpc_port}"))
.collect::<Vec<_>>();

if rpc_urls.is_empty() {
return Err(TestrpcError::LoadEndpointsError(
"No RPC endpoints found".to_string(),
));
}
Ok(rpc_urls)
}

Expand Down Expand Up @@ -134,6 +139,7 @@ fn parse_endpoints(endpoints: &str) -> Result<Vec<String>, TestrpcError> {
// "/ip4/*.*.*.*" -> "*.*.*.*"
ip_addr[5..].to_string()
})
.filter(|s| !s.is_empty())
.collect::<Vec<_>>();

Ok(addrs)
Expand Down
26 changes: 23 additions & 3 deletions src/bin/testrpc.rs
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@ struct Opts {
log_file: Option<String>,
#[clap(long, default_value = "debug")]
log_level: String,
#[clap(long, default_value = "10")]
init_retries: u32,
}

#[tokio::main]
Expand All @@ -33,23 +35,41 @@ async fn main() -> Result<(), common::TestrpcError> {
let ctx = Arc::new(ctx::Context::new());
let start = std::time::Instant::now();

tracing::info!("Starting testrpc with config file: {}", &opts.file);
tracing::info!("XXX1 Starting testrpc with config file: {}", &opts.file);

if opts.dry_run {
tracing::info!("Dry run, we will not send any RPCs");
env::set_var("DRY_RUN", "true");
}

let cfg = config::load_config(opts.file.as_str()).unwrap();
let rpc_urls = if opts.dry_run && opts.gen_mock_rpcs {
let retries = opts.init_retries;
let cfg_rpcs = cfg.clone().rpcs.unwrap_or_default();
let rpc_urls = if cfg_rpcs.is_empty() {
cfg_rpcs
} else if opts.dry_run && opts.gen_mock_rpcs {
let num_of_nodes = cfg.num_of_nodes.unwrap_or(4);
let mut urls = Vec::new();
for i in 0..num_of_nodes {
urls.push(format!("http://dummy:{}", 5000 + i));
}
urls
} else {
runner::load_endpoints(cfg.clone()).await.unwrap()
let cfg_clone = cfg.clone();
let urls = common::retry(
retries as usize,
std::time::Duration::from_secs(1),
|| Box::pin(runner::load_endpoints(cfg_clone.clone())),
true,
)
.await
.unwrap_or(Vec::new());
if urls.is_empty() {
return Err(common::TestrpcError::LoadEndpointsError(format!(
"Failed to load RPC endpoints after {retries} retries"
)));
}
urls
};

if let Some(num_of_nodes) = cfg.num_of_nodes {
Expand Down
29 changes: 28 additions & 1 deletion src/common.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
use serde::{Deserialize, Serialize};
use std::time::Duration;
use std::{future::Future, pin::Pin, time::Duration};
use thiserror::Error;

#[derive(Error, Debug)]
Expand Down Expand Up @@ -59,3 +59,30 @@ impl FlowResults {
}
}
}

/// Retry a given operation a specified number of times with a delay between attempts
/// and optional exponential backoff.
/// Returns Ok(()) if successful, or an error if all retries fail.
pub async fn retry<T: Send + 'static>(
retries: usize,
delay: Duration,
mut operation: impl FnMut() -> Pin<Box<dyn Future<Output = Result<T, TestrpcError>> + Send>>,
exponential_backoff: bool,
) -> Result<T, TestrpcError> {
let mut delay = delay;
for _ in 0..retries {
match operation().await {
Ok(result) => return Ok(result),
Err(e) => {
tracing::warn!("Operation failed: {}. Retrying in {:?}...", e, delay);
tokio::time::sleep(delay).await;
if exponential_backoff {
delay *= 2;
}
}
}
}
Err(TestrpcError::ExecutionError(
"Max retries reached".to_string(),
))
}
2 changes: 2 additions & 0 deletions src/jrpc.rs
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,8 @@ pub async fn send(
start_time.elapsed().as_millis()
);

tracing::debug!("Raw RPC response: {:?}", response);

let response: RpcResponse = response
.json()
.await
Expand Down