feat: add automation wait commands
This commit is contained in:
parent
448a831a98
commit
e49b945259
5 changed files with 261 additions and 7 deletions
|
|
@ -3,10 +3,11 @@ use base64::Engine;
|
|||
use clap::{Args, CommandFactory, Parser, Subcommand, ValueEnum};
|
||||
use clap_complete::{Shell, generate};
|
||||
use geth_config::GethPaths;
|
||||
use geth_control::{ControlRequest, ControlResponse};
|
||||
use geth_control::{ControlRequest, ControlResponse, SyncStreamStatus};
|
||||
use geth_node::service::{ServiceInstallOptions, ServiceManager, ServiceReport};
|
||||
use std::io::{Read, stdout};
|
||||
use std::path::PathBuf;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
const TOP_LEVEL_AFTER_HELP: &str = r#"Common starts:
|
||||
geth guide init
|
||||
|
|
@ -363,6 +364,10 @@ pub enum Command {
|
|||
#[command(subcommand)]
|
||||
command: SyncCommand,
|
||||
},
|
||||
Wait {
|
||||
#[command(subcommand)]
|
||||
command: WaitCommand,
|
||||
},
|
||||
Node {
|
||||
#[command(subcommand)]
|
||||
command: NodeCommand,
|
||||
|
|
@ -482,6 +487,30 @@ pub enum SyncCommand {
|
|||
Now { node: Option<String> },
|
||||
}
|
||||
|
||||
#[derive(Debug, Subcommand)]
|
||||
pub enum WaitCommand {
|
||||
Daemon {
|
||||
#[arg(long, default_value_t = 30_000)]
|
||||
timeout_ms: u64,
|
||||
#[arg(long, default_value_t = 250)]
|
||||
interval_ms: u64,
|
||||
},
|
||||
Peer {
|
||||
node: String,
|
||||
#[arg(long, default_value_t = 30_000)]
|
||||
timeout_ms: u64,
|
||||
#[arg(long, default_value_t = 500)]
|
||||
interval_ms: u64,
|
||||
},
|
||||
Sync {
|
||||
node: String,
|
||||
#[arg(long, default_value_t = 30_000)]
|
||||
timeout_ms: u64,
|
||||
#[arg(long, default_value_t = 500)]
|
||||
interval_ms: u64,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Debug, Subcommand)]
|
||||
pub enum NodeCommand {
|
||||
Id,
|
||||
|
|
@ -1284,6 +1313,13 @@ async fn run_inner(cli: Cli) -> Result<()> {
|
|||
run_service_command(&paths, command).context("manage geth user service")?;
|
||||
print_service_report(report, cli.json || cli.jsonl)?;
|
||||
}
|
||||
Command::Wait { command } => {
|
||||
let report = run_wait_command(&paths, command).await?;
|
||||
print_wait_report(&report, cli.json || cli.jsonl)?;
|
||||
if !report.ready {
|
||||
std::process::exit(1);
|
||||
}
|
||||
}
|
||||
Command::Ssh {
|
||||
command:
|
||||
SshCommand::Proxy {
|
||||
|
|
@ -2180,12 +2216,183 @@ fn request_for_command(command: Command) -> Result<ControlRequest> {
|
|||
},
|
||||
},
|
||||
},
|
||||
Command::Guide { .. } | Command::Init { .. } | Command::Daemon { .. } => {
|
||||
Command::Guide { .. }
|
||||
| Command::Init { .. }
|
||||
| Command::Daemon { .. }
|
||||
| Command::Wait { .. } => {
|
||||
bail!("command is handled directly")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct WaitReport {
|
||||
target: String,
|
||||
ready: bool,
|
||||
elapsed_ms: u128,
|
||||
attempts: u64,
|
||||
reason: String,
|
||||
}
|
||||
|
||||
async fn run_wait_command(paths: &GethPaths, command: WaitCommand) -> Result<WaitReport> {
|
||||
match command {
|
||||
WaitCommand::Daemon {
|
||||
timeout_ms,
|
||||
interval_ms,
|
||||
} => wait_for_daemon(paths, timeout_ms, interval_ms).await,
|
||||
WaitCommand::Peer {
|
||||
node,
|
||||
timeout_ms,
|
||||
interval_ms,
|
||||
} => wait_for_peer(paths, node, timeout_ms, interval_ms).await,
|
||||
WaitCommand::Sync {
|
||||
node,
|
||||
timeout_ms,
|
||||
interval_ms,
|
||||
} => wait_for_sync(paths, node, timeout_ms, interval_ms).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn wait_for_daemon(
|
||||
paths: &GethPaths,
|
||||
timeout_ms: u64,
|
||||
interval_ms: u64,
|
||||
) -> Result<WaitReport> {
|
||||
wait_loop("daemon".to_owned(), timeout_ms, interval_ms, || async {
|
||||
match geth_node::send_control(paths, ControlRequest::Status).await {
|
||||
Ok(ControlResponse::Status(_)) => Ok(Some("daemon control is ready".to_owned())),
|
||||
Ok(other) => Ok(Some(format!("unexpected response: {:?}", other))),
|
||||
Err(error) => Err(anyhow::anyhow!(error.to_string())),
|
||||
}
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
async fn wait_for_peer(
|
||||
paths: &GethPaths,
|
||||
node: String,
|
||||
timeout_ms: u64,
|
||||
interval_ms: u64,
|
||||
) -> Result<WaitReport> {
|
||||
wait_loop(format!("peer:{node}"), timeout_ms, interval_ms, || {
|
||||
let node = node.clone();
|
||||
async move {
|
||||
match geth_node::send_control(paths, ControlRequest::PeerPing { node }).await {
|
||||
Ok(ControlResponse::PeerPinged { note, .. }) => Ok(Some(note)),
|
||||
Ok(ControlResponse::Error { message }) => Err(anyhow::anyhow!(message)),
|
||||
Ok(_) => Ok(None),
|
||||
Err(error) => Err(anyhow::anyhow!(error.to_string())),
|
||||
}
|
||||
}
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
async fn wait_for_sync(
|
||||
paths: &GethPaths,
|
||||
node: String,
|
||||
timeout_ms: u64,
|
||||
interval_ms: u64,
|
||||
) -> Result<WaitReport> {
|
||||
wait_loop(format!("sync:{node}"), timeout_ms, interval_ms, || {
|
||||
let node = node.clone();
|
||||
async move {
|
||||
match geth_node::send_control(paths, ControlRequest::SyncStatus).await {
|
||||
Ok(ControlResponse::SyncStatus { peers, .. }) => {
|
||||
let Some(peer) = peers.into_iter().find(|peer| peer.peer_node_id == node)
|
||||
else {
|
||||
return Ok(None);
|
||||
};
|
||||
if peer.streams.is_empty() {
|
||||
return Ok(None);
|
||||
}
|
||||
if peer.streams.iter().all(sync_stream_ready) {
|
||||
Ok(Some("sync streams are healthy or stale".to_owned()))
|
||||
} else {
|
||||
Ok(None)
|
||||
}
|
||||
}
|
||||
Ok(ControlResponse::Error { message }) => Err(anyhow::anyhow!(message)),
|
||||
Ok(_) => Ok(None),
|
||||
Err(error) => Err(anyhow::anyhow!(error.to_string())),
|
||||
}
|
||||
}
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
fn sync_stream_ready(stream: &SyncStreamStatus) -> bool {
|
||||
stream.state == "ok" || stream.stale
|
||||
}
|
||||
|
||||
async fn wait_loop<F, Fut>(
|
||||
target: String,
|
||||
timeout_ms: u64,
|
||||
interval_ms: u64,
|
||||
mut check: F,
|
||||
) -> Result<WaitReport>
|
||||
where
|
||||
F: FnMut() -> Fut,
|
||||
Fut: std::future::Future<Output = Result<Option<String>>>,
|
||||
{
|
||||
let started = Instant::now();
|
||||
let timeout = Duration::from_millis(timeout_ms);
|
||||
let interval = Duration::from_millis(interval_ms.max(1));
|
||||
let mut attempts = 0;
|
||||
let mut last_error: Option<String> = None;
|
||||
loop {
|
||||
attempts += 1;
|
||||
match check().await {
|
||||
Ok(Some(reason)) => {
|
||||
return Ok(WaitReport {
|
||||
target,
|
||||
ready: true,
|
||||
elapsed_ms: started.elapsed().as_millis(),
|
||||
attempts,
|
||||
reason,
|
||||
});
|
||||
}
|
||||
Ok(None) => {}
|
||||
Err(error) => {
|
||||
last_error = Some(error.to_string());
|
||||
}
|
||||
}
|
||||
if started.elapsed() >= timeout {
|
||||
return Ok(WaitReport {
|
||||
target,
|
||||
ready: false,
|
||||
elapsed_ms: started.elapsed().as_millis(),
|
||||
attempts,
|
||||
reason: last_error.unwrap_or_else(|| "timeout waiting for readiness".to_owned()),
|
||||
});
|
||||
}
|
||||
tokio::time::sleep(interval).await;
|
||||
}
|
||||
}
|
||||
|
||||
fn print_wait_report(report: &WaitReport, json: bool) -> Result<()> {
|
||||
if json {
|
||||
println!(
|
||||
"{}",
|
||||
serde_json::to_string_pretty(&serde_json::json!({
|
||||
"type": "wait",
|
||||
"target": report.target,
|
||||
"ready": report.ready,
|
||||
"elapsed_ms": report.elapsed_ms,
|
||||
"attempts": report.attempts,
|
||||
"reason": report.reason,
|
||||
}))?
|
||||
);
|
||||
return Ok(());
|
||||
}
|
||||
println!("target: {}", report.target);
|
||||
println!("ready: {}", report.ready);
|
||||
println!("elapsed_ms: {}", report.elapsed_ms);
|
||||
println!("attempts: {}", report.attempts);
|
||||
println!("reason: {}", report.reason);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn run_service_command(paths: &GethPaths, command: ServiceCommand) -> Result<ServiceReport> {
|
||||
Ok(match command {
|
||||
ServiceCommand::Install {
|
||||
|
|
|
|||
Loading…
Reference in a new issue