diff --git a/README.md b/README.md index c75c3cc..0b5157c 100644 --- a/README.md +++ b/README.md @@ -99,6 +99,7 @@ The bootstrap implementation provides: - `geth daemon run` - `geth daemon service install|uninstall|start|stop|status|print` - `geth status` +- `geth wait daemon|peer|sync --timeout-ms ` - `geth node id` - `geth node list` - `geth node enroll request --node-name --capability [--out ]` @@ -152,6 +153,9 @@ The bootstrap implementation provides: - `geth auth sync ` - `geth sync status` - `geth sync now [node-id-or-name]` +- `geth wait daemon --timeout-ms ` +- `geth wait peer --timeout-ms ` +- `geth wait sync --timeout-ms ` - `geth secret status` - `geth secret create ` - `geth secret rotate ` diff --git a/crates/geth-cli/src/lib.rs b/crates/geth-cli/src/lib.rs index 55d0f2e..ed78b3a 100644 --- a/crates/geth-cli/src/lib.rs +++ b/crates/geth-cli/src/lib.rs @@ -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 }, } +#[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 { }, }, }, - 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 { + 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 { + 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 { + 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 { + 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( + target: String, + timeout_ms: u64, + interval_ms: u64, + mut check: F, +) -> Result +where + F: FnMut() -> Fut, + Fut: std::future::Future>>, +{ + 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 = 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 { Ok(match command { ServiceCommand::Install { diff --git a/crates/geth/tests/bootstrap.rs b/crates/geth/tests/bootstrap.rs index a5b98a8..bdcb953 100644 --- a/crates/geth/tests/bootstrap.rs +++ b/crates/geth/tests/bootstrap.rs @@ -258,6 +258,18 @@ fn geth_status_against_running_daemon() { let output = run_geth(home.path(), &["status"]); let json_output = run_geth(home.path(), &["--json", "status"]); + let wait_output = run_geth( + home.path(), + &[ + "--json", + "wait", + "daemon", + "--timeout-ms", + "1000", + "--interval-ms", + "10", + ], + ); let _ = daemon.kill(); let _ = daemon.wait(); @@ -305,6 +317,11 @@ fn geth_status_against_running_daemon() { assert_eq!(status_json["store_synchronous"], "normal"); assert_eq!(status_json["store_status"], "ok"); assert!(status_json["native_backends"].is_array()); + + let wait_json = command_json(&wait_output); + assert_eq!(wait_json["type"], "wait"); + assert_eq!(wait_json["target"], "daemon"); + assert_eq!(wait_json["ready"], true); } #[test] @@ -362,6 +379,31 @@ fn json_errors_include_stable_code_for_common_failures() { ); } +#[test] +fn wait_daemon_reports_json_timeout_without_daemon() { + let home = tempfile::tempdir().expect("tempdir"); + let output = run_geth( + home.path(), + &[ + "--json", + "wait", + "daemon", + "--timeout-ms", + "1", + "--interval-ms", + "1", + ], + ); + + assert!(!output.status.success()); + let report = + serde_json::from_slice::(&output.stdout).expect("decode wait report"); + assert_eq!(report["type"], "wait"); + assert_eq!(report["target"], "daemon"); + assert_eq!(report["ready"], false); + assert!(report["attempts"].as_u64().expect("attempts") >= 1); +} + #[test] fn peer_ping_uses_daemon_owned_iroh_endpoint() { if skip_iroh_integration_tests() { diff --git a/docs/command-stability.md b/docs/command-stability.md index fad257d..03c9b65 100644 --- a/docs/command-stability.md +++ b/docs/command-stability.md @@ -32,6 +32,7 @@ The following command families are intended to be stable automation surfaces: - `geth keychain sigchain|verify-sigchain|import-sigchain|verify-checkpoint|fetch|explain|explain-signer` - `geth auth explain|grant|revoke|sync` - `geth sync status|now` +- `geth wait daemon|peer|sync` - `geth secret status|create|rotate|bearer` - Local CAS object commands: `geth cas add|get|hash|has|pin|unpin|cleanup|providers|list|fetch` - File-root metadata commands: `geth cas root add|list|scan|sync|apply` diff --git a/docs/production-readiness-roadmap.md b/docs/production-readiness-roadmap.md index 5c2b40d..5cfbce1 100644 --- a/docs/production-readiness-roadmap.md +++ b/docs/production-readiness-roadmap.md @@ -243,14 +243,14 @@ Goal: make `geth` ergonomic and stable as a base layer for custom automation. - `[x]` Tests assert both code and operator-facing hint for representative failures. -- `[ ]` Add wait commands for automation. +- `[x]` Add wait commands for automation. Acceptance criteria: - - `[ ]` `geth wait daemon` can block until local control is ready. - - `[ ]` `geth wait peer ` can block until peer control succeeds or + - `[x]` `geth wait daemon` can block until local control is ready. + - `[x]` `geth wait peer ` can block until peer control succeeds or times out. - - `[ ]` `geth wait sync ` can block until required streams are healthy + - `[x]` `geth wait sync ` can block until required streams are healthy or stale. - - `[ ]` Wait commands support JSON output and timeout flags. + - `[x]` Wait commands support JSON output and timeout flags. - `[ ]` Make common commands idempotent. Acceptance criteria: