eric/geth
Watch
1
0
Fork
You've already forked geth
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Polish first-run operator workflow

Browse source
This commit is contained in:
Eric Wendland 2026-05-22 15:00:09 +02:00
commit e6879e8949
6 changed files with 208 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

7
crates/geth-cli/src/lib.rs
View file

@ -1923,8 +1923,10 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> {
}
for stream in peer.streams {
println!(
" {}\tcursor={}\tlast_attempt={}\tlast_success={}\timported={}\trejected={}\terror={}",
" {}\tstate={}\tstale={}\tcursor={}\tlast_attempt={}\tlast_success={}\timported={}\trejected={}\terror={}\tnext={}",
stream.stream,
stream.state,
stream.stale,
stream.cursor_ms,
stream
.last_attempt_ms
@ -1936,7 +1938,8 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> {
.unwrap_or_else(|| "never".to_owned()),
stream.last_imported,
stream.last_rejected,
stream.last_error.unwrap_or_else(|| "-".to_owned())
stream.last_error.unwrap_or_else(|| "-".to_owned()),
stream.next_action
);
}
}

8
crates/geth-control/src/lib.rs
View file

@ -1391,11 +1391,15 @@ pub struct SyncPeerStatus {
pub struct SyncStreamStatus {
pub stream: String,
pub cursor_ms: i64,
pub state: String,
pub stale: bool,
pub stale_after_ms: i64,
pub last_attempt_ms: Option<i64>,
pub last_success_ms: Option<i64>,
pub last_error: Option<String>,
pub last_imported: usize,
pub last_rejected: usize,
pub next_action: String,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@ -1614,11 +1618,15 @@ mod tests {
streams: vec![SyncStreamStatus {
stream: "keychain".to_owned(),
cursor_ms: 42,
state: "ok".to_owned(),
stale: false,
stale_after_ms: 120_000,
last_attempt_ms: Some(43),
last_success_ms: Some(43),
last_error: None,
last_imported: 2,
last_rejected: 0,
next_action: "none".to_owned(),
}],
}],
note: "local health".to_owned(),

108
crates/geth-node/src/lib.rs
View file

@ -181,6 +181,7 @@ struct PipeRuntime {
const PUBSUB_RING_LIMIT: usize = 256;
const PIPE_CONNECTION_RING_LIMIT: usize = 256;
const PIPE_MESSAGE_RING_LIMIT: usize = 1024;
const LIVE_SYNC_STALE_AFTER_MS: i64 = 120_000;
#[derive(Debug, serde::Deserialize, serde::Serialize)]
struct LiveSyncCursor {
@ -787,7 +788,7 @@ async fn handle_stream(node: LocalNode, stream: UnixStream) -> Result<(), NodeEr
let response = match handle_request_async(&node, request).await {
Ok(response) => response,
Err(error) => ControlResponse::Error {
message: error.to_string(),
message: operator_error_message(&error),
},
};
let mut stream = reader.into_inner();
@ -797,6 +798,48 @@ async fn handle_stream(node: LocalNode, stream: UnixStream) -> Result<(), NodeEr
Ok(())
}
fn operator_error_message(error: &NodeError) -> String {
let mut message = error.to_string();
if let Some(next) = operator_recovery_hint(error) {
message.push_str("\nnext: ");
message.push_str(&next);
}
message
}
fn operator_recovery_hint(error: &NodeError) -> Option<String> {
match error {
NodeError::PeerNotFound(peer) => Some(format!(
"import a signed peer card for {peer} with `geth peer import <path>`, then verify it with `geth peer list`"
)),
NodeError::IrohEndpointUnavailable => Some(
"run `geth status` to inspect Iroh endpoint state; restart with `geth daemon run` if the daemon was started without an endpoint".to_owned(),
),
NodeError::Unauthorized(_) => Some(
"run `geth auth explain <subject> <resource> <capability>` and then grant the missing capability with `geth auth grant ... --signing-key <admin-key>`".to_owned(),
),
NodeError::ResourceNotFound(resource) => Some(format!(
"run `geth resource list` to inspect resources or create {resource} with the matching module command"
)),
NodeError::DbNotFound(name) => Some(format!(
"register the database with `geth db add {name} <path>` or inspect existing DB resources with `geth resource list`"
)),
NodeError::KvNotFound(name) => Some(format!(
"create the KV store with `geth kv create {name}` or inspect existing resources with `geth resource list`"
)),
NodeError::DocumentNotFound(name) => Some(format!(
"create the document with `geth document create {name}` or inspect existing resources with `geth resource list`"
)),
NodeError::SshCertRequestNotFound(_) => Some(
"run `geth ssh cert requests` on the signing machine and sync metadata with `geth ssh cert sync <node>` if needed".to_owned(),
),
NodeError::InvalidDbPath(_) => Some(
"create the SQLite file first, then run `geth db add <name> <path>` again".to_owned(),
),
_ => None,
}
}
async fn start_peer_card_lan_discovery(node: &LocalNode) -> Option<swarm_discovery::DropGuard> {
if !node.iroh_status.local_discovery || !node.iroh_status.enabled {
return None;
@ -3124,6 +3167,7 @@ fn sync_status_local(store: &Store) -> Result<ControlResponse, NodeError> {
.into_iter()
.map(|peer| {
let prefix = live_sync_status_prefix(&peer.peer_id);
let now_ms = geth_store::now_ms();
let mut streams = store
.list_module_states_with_prefix(&prefix)?
.into_iter()
@ -3131,14 +3175,24 @@ fn sync_status_local(store: &Store) -> Result<ControlResponse, NodeError> {
let stream = state.module.trim_start_matches(&prefix).to_owned();
let health: LiveSyncHealth = serde_json::from_str(&state.state_json)?;
let cursor_ms = load_live_sync_cursor(store, &peer.peer_id, &stream)?;
let stale = health.last_success_ms.is_some_and(|last_success_ms| {
now_ms.saturating_sub(last_success_ms) > LIVE_SYNC_STALE_AFTER_MS
});
let status_state = sync_stream_state(&health, stale).to_owned();
let next_action =
sync_stream_next_action(&peer.peer_id, &stream, &health, stale);
Ok(SyncStreamStatus {
stream,
cursor_ms,
state: status_state,
stale,
stale_after_ms: LIVE_SYNC_STALE_AFTER_MS,
last_attempt_ms: Some(health.last_attempt_ms),
last_success_ms: health.last_success_ms,
last_error: health.last_error,
last_imported: health.last_imported,
last_rejected: health.last_rejected,
next_action,
})
})
.collect::<Result<Vec<_>, NodeError>>()?;
@ -3155,6 +3209,41 @@ fn sync_status_local(store: &Store) -> Result<ControlResponse, NodeError> {
})
}
fn sync_stream_state(health: &LiveSyncHealth, stale: bool) -> &'static str {
if health.last_error.is_some() {
"failed"
} else if health.last_success_ms.is_none() {
"never-succeeded"
} else if stale {
"stale"
} else {
"ok"
}
}
fn sync_stream_next_action(
peer_node: &str,
stream: &str,
health: &LiveSyncHealth,
stale: bool,
) -> String {
if health.last_error.is_some() {
format!(
"run `geth sync now {peer_node}`; if it still fails, run `geth auth explain <subject> <resource> <capability>` for stream {stream}"
)
} else if health.last_success_ms.is_none() {
format!(
"run `geth sync now {peer_node}` to establish the first successful sync for {stream}"
)
} else if stale {
format!(
"run `geth sync now {peer_node}` and check `geth peer ping {peer_node}` if the stream remains stale"
)
} else {
"none".to_owned()
}
}
fn should_live_sync_stream(
store: &Store,
peer_node: &str,
@ -10203,9 +10292,26 @@ mod tests {
assert_eq!(peers[0].streams.len(), 1);
assert_eq!(peers[0].streams[0].stream, "keychain");
assert_eq!(peers[0].streams[0].cursor_ms, 42);
assert_eq!(peers[0].streams[0].state, "ok");
assert!(!peers[0].streams[0].stale);
assert_eq!(peers[0].streams[0].stale_after_ms, LIVE_SYNC_STALE_AFTER_MS);
assert_eq!(peers[0].streams[0].last_imported, 3);
assert_eq!(peers[0].streams[0].last_rejected, 1);
assert!(peers[0].streams[0].last_error.is_none());
assert_eq!(peers[0].streams[0].next_action, "none");
}
#[test]
fn operator_errors_include_recovery_commands() {
let peer_error =
operator_error_message(&NodeError::PeerNotFound("node:missing".to_owned()));
assert!(peer_error.contains("geth peer import"));
assert!(peer_error.contains("geth peer list"));
let denied_error =
operator_error_message(&NodeError::Unauthorized("missing grant".to_owned()));
assert!(denied_error.contains("geth auth explain"));
assert!(denied_error.contains("geth auth grant"));
}
#[test]