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

Degrade gracefully when Iroh modules fail

Browse source
This commit is contained in:
Eric Wendland 2026-07-18 16:00:13 +02:00
commit 7d67da9dfe
4 changed files with 193 additions and 55 deletions
Download patch file Download diff file Expand all files Collapse all files

173
crates/geth-node/src/daemon.rs
View file

@ -88,6 +88,14 @@ async fn shutdown_iroh_runtime(
node: &LocalNode,
endpoint: Option<GethIrohEndpoint>,
) -> Result<(), NodeError> {
clear_iroh_runtime_handles(node)?;
if let Some(endpoint) = endpoint {
endpoint.shutdown().await;
}
Ok(())
}
fn clear_iroh_runtime_handles(node: &LocalNode) -> Result<(), NodeError> {
node.iroh_docs
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)?
@ -104,9 +112,6 @@ async fn shutdown_iroh_runtime(
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)?
.take();
if let Some(endpoint) = endpoint {
endpoint.shutdown().await;
}
Ok(())
}
@ -198,55 +203,24 @@ pub(crate) async fn start_daemon_iroh_endpoint(
match geth_iroh::start_endpoint(&config).await {
Ok(endpoint) => {
let status = endpoint.status();
if let Some(endpoint_id) = &status.endpoint_id {
let store = Store::open(&node.paths.metadata_db())?;
store.upsert_node_endpoint(endpoint_id, &node.node_id, &node.agent_id, "iroh")?;
match initialize_daemon_iroh_runtime(node, &endpoint, status).await {
Ok(()) => Ok(Some(endpoint)),
Err(error) => {
clear_iroh_runtime_handles(node)?;
endpoint.shutdown().await;
node.iroh_status = EndpointStatus {
enabled: false,
endpoint_id: None,
relay_mode: relay_mode_label,
local_discovery,
note: format!(
"Iroh endpoint started but native runtime initialization failed: {error}"
),
};
tracing::warn!(%error, "Iroh runtime unavailable; continuing with local control");
Ok(None)
}
}
let iroh_blobs_path = node.paths.cas_dir().join("iroh-blobs");
std::fs::create_dir_all(&iroh_blobs_path)?;
let iroh_docs_path = node.paths.home().join("iroh-docs");
std::fs::create_dir_all(&iroh_docs_path)?;
let blob_store = iroh_blobs::store::fs::FsStore::load(iroh_blobs_path)
.await
.map_err(|error| {
NodeError::IrohPeer(format!("failed to open iroh-blobs store: {error}"))
})?;
let gossip = iroh_gossip::net::Gossip::builder().spawn(endpoint.endpoint());
let blob_api: iroh_blobs::api::Store = blob_store.clone().into();
let docs = iroh_docs::protocol::Docs::persistent(iroh_docs_path)
.spawn(endpoint.endpoint(), blob_api, gossip.clone())
.await
.map_err(|error| {
NodeError::IrohPeer(format!("failed to start iroh-docs: {error}"))
})?;
let mirrored = mirror_local_cas_to_iroh_blobs(&node.paths, &blob_store).await?;
tracing::info!(
mirrored_blobs = mirrored,
"native iroh-blobs store is ready"
);
node.iroh_status = status;
*node
.iroh_endpoint
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)? = Some(endpoint.clone());
*node
.iroh_blob_store
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)? = Some(blob_store);
*node
.iroh_gossip
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)? = Some(gossip);
*node
.iroh_docs
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)? = Some(docs);
let mirrored_kv = mirror_all_kv_stores_to_iroh_docs(node).await?;
tracing::info!(
mirrored_kv_stores = mirrored_kv,
"native iroh-docs KV store is ready"
);
Ok(Some(endpoint))
}
Err(error) => {
node.iroh_status = EndpointStatus {
@ -261,6 +235,60 @@ pub(crate) async fn start_daemon_iroh_endpoint(
}
}
async fn initialize_daemon_iroh_runtime(
node: &mut LocalNode,
endpoint: &GethIrohEndpoint,
status: EndpointStatus,
) -> Result<(), NodeError> {
if let Some(endpoint_id) = &status.endpoint_id {
let store = Store::open(&node.paths.metadata_db())?;
store.upsert_node_endpoint(endpoint_id, &node.node_id, &node.agent_id, "iroh")?;
}
let iroh_blobs_path = node.paths.cas_dir().join("iroh-blobs");
std::fs::create_dir_all(&iroh_blobs_path)?;
let iroh_docs_path = node.paths.home().join("iroh-docs");
std::fs::create_dir_all(&iroh_docs_path)?;
let blob_store = iroh_blobs::store::fs::FsStore::load(iroh_blobs_path)
.await
.map_err(|error| {
NodeError::IrohPeer(format!("failed to open iroh-blobs store: {error}"))
})?;
let gossip = iroh_gossip::net::Gossip::builder().spawn(endpoint.endpoint());
let blob_api: iroh_blobs::api::Store = blob_store.clone().into();
let docs = iroh_docs::protocol::Docs::persistent(iroh_docs_path)
.spawn(endpoint.endpoint(), blob_api, gossip.clone())
.await
.map_err(|error| NodeError::IrohPeer(format!("failed to start iroh-docs: {error}")))?;
let mirrored = mirror_local_cas_to_iroh_blobs(&node.paths, &blob_store).await?;
tracing::info!(
mirrored_blobs = mirrored,
"native iroh-blobs store is ready"
);
node.iroh_status = status;
*node
.iroh_endpoint
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)? = Some(endpoint.clone());
*node
.iroh_blob_store
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)? = Some(blob_store);
*node
.iroh_gossip
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)? = Some(gossip);
*node
.iroh_docs
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)? = Some(docs);
let mirrored_kv = mirror_all_kv_stores_to_iroh_docs(node).await?;
tracing::info!(
mirrored_kv_stores = mirrored_kv,
"native iroh-docs KV store is ready"
);
Ok(())
}
fn config_relay_mode_to_iroh(
mode: &RelayMode,
relay_maps: &std::collections::BTreeMap<String, geth_config::RelayMapConfig>,
@ -338,6 +366,49 @@ mod tests {
);
}
#[tokio::test]
async fn native_runtime_failure_degrades_to_local_only_status() {
if std::env::var_os("GETH_TEST_SKIP_IROH").is_some() {
eprintln!("skipping Iroh degradation test because GETH_TEST_SKIP_IROH is set");
return;
}
let dir = tempfile::tempdir().expect("tempdir");
let paths = GethPaths::from_home(dir.path());
let mut node = init_node(&paths).expect("init node");
std::fs::write(
paths.config_file(),
"[iroh]\nrelay_mode = \"disabled\"\nlocal_discovery = false\n",
)
.expect("offline config");
std::fs::write(paths.cas_dir().join("iroh-blobs"), b"not a directory")
.expect("invalid native store fixture");
let endpoint = start_daemon_iroh_endpoint(&mut node)
.await
.expect("runtime failure is degraded");
if node
.iroh_status
.note
.starts_with("Iroh endpoint failed to start")
{
eprintln!("skipping native runtime assertion because UDP bind is unavailable");
return;
}
assert!(endpoint.is_none());
assert!(!node.iroh_status.enabled);
assert!(
node.iroh_status
.note
.contains("native runtime initialization failed")
);
assert!(node.iroh_endpoint.lock().expect("endpoint lock").is_none());
assert!(node.iroh_blob_store.lock().expect("blob lock").is_none());
assert!(node.iroh_docs.lock().expect("docs lock").is_none());
assert!(node.iroh_gossip.lock().expect("gossip lock").is_none());
}
#[tokio::test]
async fn control_socket_binding_rejects_a_second_daemon_and_recovers_stale_paths() {
let dir = tempfile::tempdir().expect("tempdir");

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

@ -5566,7 +5566,7 @@ pub fn handle_request(
iroh_relay_mode: node.iroh_status.relay_mode.clone(),
iroh_local_discovery: node.iroh_status.local_discovery,
iroh: node.iroh_status.note.clone(),
native_backends: native_backend_statuses(),
native_backends: native_backend_statuses(node)?,
}))
}
ControlRequest::NodeId => Ok(ControlResponse::NodeId(NodeIdResponse {
@ -7579,10 +7579,46 @@ pub fn handle_request(
}
}
fn native_backend_statuses() -> Vec<NativeBackendStatus> {
geth_iroh::native_iroh_libraries()
fn native_backend_statuses(node: &LocalNode) -> Result<Vec<NativeBackendStatus>, NodeError> {
let endpoint_ready = node
.iroh_endpoint
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)?
.is_some();
let blobs_ready = node
.iroh_blob_store
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)?
.is_some();
let docs_ready = node
.iroh_docs
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)?
.is_some();
let gossip_ready = node
.iroh_gossip
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)?
.is_some();
Ok(geth_iroh::native_iroh_libraries()
.into_iter()
.map(|library| {
let runtime_ready = match library.module.as_str() {
"cas" => endpoint_ready && blobs_ready,
"kv" => endpoint_ready && docs_ready,
"pubsub" => endpoint_ready && gossip_ready,
_ => endpoint_ready,
};
if !runtime_ready {
return NativeBackendStatus {
module: library.module,
current_backend: "unavailable".to_owned(),
target_crate: library.crate_name,
target_version: library.crate_version,
status: "unavailable".to_owned(),
blocker: node.iroh_status.note.clone(),
};
}
let (status, blocker) = match library.module.as_str() {
"cas" => (
"wired",
@ -7623,7 +7659,7 @@ fn native_backend_statuses() -> Vec<NativeBackendStatus> {
blocker,
}
})
.collect()
.collect())
}
fn stored_resource_to_descriptor(stored: StoredResource) -> Result<ResourceDescriptor, NodeError> {
@ -11151,6 +11187,21 @@ mod tests {
);
}
#[test]
fn native_backend_status_reflects_runtime_availability() {
let dir = tempfile::tempdir().expect("tempdir");
let node = init_node(&GethPaths::from_home(dir.path())).expect("init node");
let statuses = native_backend_statuses(&node).expect("backend status");
assert!(!statuses.is_empty());
assert!(statuses.iter().all(|status| status.status == "unavailable"));
assert!(
statuses
.iter()
.all(|status| status.current_backend == "unavailable")
);
}
#[derive(Debug, PartialEq, Eq)]
enum RemoteGuardKind {
Capability,