diff --git a/crates/geth-node/src/daemon.rs b/crates/geth-node/src/daemon.rs index a084a3c..2e666ad 100644 --- a/crates/geth-node/src/daemon.rs +++ b/crates/geth-node/src/daemon.rs @@ -1,8 +1,63 @@ //! Daemon task orchestration helpers. -use crate::{LocalNode, sync::run_live_sync_once}; -use geth_iroh::GethIrohEndpoint; -use std::time::Duration; +use crate::{ + LocalNode, NodeError, handle_stream, init_node, mirror_all_kv_stores_to_iroh_docs, + mirror_local_cas_to_iroh_blobs, start_peer_card_lan_discovery, sync::run_live_sync_once, +}; +use geth_config::{GethConfig, GethPaths, RelayMode}; +use geth_iroh::{EndpointStatus, GethIrohConfig, GethIrohEndpoint, GethRelayMode}; +use geth_store::Store; +use std::{path::Path, time::Duration}; +use tokio::net::UnixListener; + +pub async fn run_daemon(paths: GethPaths) -> Result<(), NodeError> { + let mut node = init_node(&paths)?; + let _iroh_endpoint = start_daemon_iroh_endpoint(&mut node).await?; + if let Some(endpoint) = node + .iroh_endpoint + .lock() + .map_err(|_| NodeError::RuntimeLockPoisoned)? + .clone() + { + spawn_iroh_control_accept_loop(node.clone(), endpoint); + let config = GethConfig::load(&node.paths.config_file())?; + if config.sync.live_sync_enabled { + spawn_background_live_sync( + node.clone(), + Duration::from_millis(config.sync.live_sync_interval_ms), + ); + } + } + let _peer_card_lan_discovery = start_peer_card_lan_discovery(&node).await; + if Path::new(&paths.socket_path()).exists() { + std::fs::remove_file(paths.socket_path())?; + } + let listener = UnixListener::bind(paths.socket_path())?; + tracing::info!(socket = %paths.socket_path().display(), "geth daemon listening"); + + serve_local_control(node, listener).await +} + +async fn serve_local_control(node: LocalNode, listener: UnixListener) -> Result<(), NodeError> { + loop { + tokio::select! { + accepted = listener.accept() => { + let (stream, _) = accepted?; + let node = node.clone(); + tokio::spawn(async move { + if let Err(error) = handle_stream(node, stream).await { + tracing::warn!(%error, "control request failed"); + } + }); + } + signal = tokio::signal::ctrl_c() => { + signal?; + tracing::info!("shutdown signal received"); + return Ok(()); + } + } + } +} pub(crate) fn spawn_iroh_control_accept_loop(node: LocalNode, endpoint: GethIrohEndpoint) { let raw_endpoint = endpoint.endpoint(); @@ -35,3 +90,99 @@ pub(crate) fn spawn_background_live_sync(node: LocalNode, interval_duration: Dur } }); } + +pub(crate) async fn start_daemon_iroh_endpoint( + node: &mut LocalNode, +) -> Result, NodeError> { + let node_config = GethConfig::load(&node.paths.config_file())?; + let relay_mode = node_config.iroh.relay_mode.clone(); + let relay_mode_label = relay_mode.label(); + let local_discovery = node_config.iroh.local_discovery; + let iroh_relay_mode = config_relay_mode_to_iroh(&relay_mode, &node_config.iroh.relay_maps); + let mut config = GethIrohConfig::local_with_relay(node.paths.iroh_key(), iroh_relay_mode); + config.local_discovery = local_discovery; + 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")?; + } + 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 { + enabled: false, + endpoint_id: None, + relay_mode: relay_mode_label, + local_discovery, + note: format!("Iroh endpoint failed to start: {error}"), + }; + Ok(None) + } + } +} + +fn config_relay_mode_to_iroh( + mode: &RelayMode, + relay_maps: &std::collections::BTreeMap, +) -> GethRelayMode { + match mode { + RelayMode::Disabled => GethRelayMode::Disabled, + RelayMode::Default => GethRelayMode::Default, + RelayMode::Staging => GethRelayMode::Staging, + RelayMode::Custom { map } => { + let relay_map = relay_maps + .get(map) + .expect("custom relay map was validated during config load"); + GethRelayMode::Custom { + name: map.clone(), + relay_urls: relay_map.urls.clone(), + } + } + } +} diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index 194898a..23730e0 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -8,14 +8,16 @@ mod sync; mod wire; use base64::Engine; -use daemon::{spawn_background_live_sync, spawn_iroh_control_accept_loop}; +pub use daemon::run_daemon; +#[cfg(test)] +use daemon::{spawn_iroh_control_accept_loop, start_daemon_iroh_endpoint}; use futures::StreamExt; use geth_auth::{AUTH_SIGNATURE_NAMESPACE, AuthExplanation, AuthOp, AuthOpKind, AuthOpSignature}; use geth_cas::{ BlobInfoSummary, CasTreeEntryKind, CasTreeObject, FileConflict, FileConflictKind, FileConflictResolution, FileConflictStatus, FileRoot, FileRootScan, LocalCas, hash_path, }; -use geth_config::{GethConfig, GethPaths, RelayMode}; +use geth_config::{GethConfig, GethPaths}; use geth_control::{ CasBlob, CasProvider, ControlRequest, ControlResponse, KeychainStatusResponse, NativeBackendStatus, NodeIdResponse, OverlayPeer, OverlayWireRequest, OverlayWireResponse, @@ -29,7 +31,7 @@ use geth_discovery::{ discovery_is_untrusted_note, peer_card_from_txt_attributes, peer_card_txt_attributes, }; use geth_document::{DocumentResource, DocumentState}; -use geth_iroh::{EndpointStatus, GethIrohConfig, GethIrohEndpoint, GethRelayMode}; +use geth_iroh::{EndpointStatus, GethIrohEndpoint}; use geth_keychain::{ KeychainOp, KeychainOpKind, KeychainOpSignature, NODE_ENROLLMENT_REQUEST_NAMESPACE, NodeEnrollmentCapability, NodeEnrollmentProvenance, NodeEnrollmentRequest, @@ -413,42 +415,6 @@ fn initialize_owner_keychain( Ok(ops) } -pub async fn run_daemon(paths: GethPaths) -> Result<(), NodeError> { - let mut node = init_node(&paths)?; - let _iroh_endpoint = start_daemon_iroh_endpoint(&mut node).await?; - if let Some(endpoint) = node - .iroh_endpoint - .lock() - .map_err(|_| NodeError::RuntimeLockPoisoned)? - .clone() - { - spawn_iroh_control_accept_loop(node.clone(), endpoint); - let config = GethConfig::load(&node.paths.config_file())?; - if config.sync.live_sync_enabled { - spawn_background_live_sync( - node.clone(), - Duration::from_millis(config.sync.live_sync_interval_ms), - ); - } - } - let _peer_card_lan_discovery = start_peer_card_lan_discovery(&node).await; - if Path::new(&paths.socket_path()).exists() { - std::fs::remove_file(paths.socket_path())?; - } - let listener = UnixListener::bind(paths.socket_path())?; - tracing::info!(socket = %paths.socket_path().display(), "geth daemon listening"); - - loop { - let (stream, _) = listener.accept().await?; - let node = node.clone(); - tokio::spawn(async move { - if let Err(error) = handle_stream(node, stream).await { - tracing::warn!(%error, "control request failed"); - } - }); - } -} - pub async fn send_control( paths: &GethPaths, request: ControlRequest, @@ -11361,102 +11327,6 @@ fn stable_node_id(agent_id: &str) -> String { format!("node:{agent_id}") } -async fn start_daemon_iroh_endpoint( - node: &mut LocalNode, -) -> Result, NodeError> { - let node_config = GethConfig::load(&node.paths.config_file())?; - let relay_mode = node_config.iroh.relay_mode.clone(); - let relay_mode_label = relay_mode.label(); - let local_discovery = node_config.iroh.local_discovery; - let iroh_relay_mode = config_relay_mode_to_iroh(&relay_mode, &node_config.iroh.relay_maps); - let mut config = GethIrohConfig::local_with_relay(node.paths.iroh_key(), iroh_relay_mode); - config.local_discovery = local_discovery; - 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")?; - } - 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 { - enabled: false, - endpoint_id: None, - relay_mode: relay_mode_label, - local_discovery, - note: format!("Iroh endpoint failed to start: {error}"), - }; - Ok(None) - } - } -} - -fn config_relay_mode_to_iroh( - mode: &RelayMode, - relay_maps: &std::collections::BTreeMap, -) -> GethRelayMode { - match mode { - RelayMode::Disabled => GethRelayMode::Disabled, - RelayMode::Default => GethRelayMode::Default, - RelayMode::Staging => GethRelayMode::Staging, - RelayMode::Custom { map } => { - let relay_map = relay_maps - .get(map) - .expect("custom relay map was validated during config load"); - GethRelayMode::Custom { - name: map.clone(), - relay_urls: relay_map.urls.clone(), - } - } - } -} - fn expected_openssh_cert_path(public_key_path: &Path) -> String { let text = public_key_path.display().to_string(); if let Some(prefix) = text.strip_suffix(".pub") { diff --git a/docs/architecture.md b/docs/architecture.md index e5ffbe0..7ec2633 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -6,6 +6,13 @@ the shared Iroh endpoint, resource registry, module routing, and local control socket. Control commands connect to the Unix socket and send typed JSONL requests. +Within `geth-node`, daemon lifecycle code is separated from feature handlers: +`daemon.rs` owns `geth daemon run` startup, local socket binding, shutdown +signal handling, Iroh endpoint startup, the Iroh accept loop, and background +live-sync task spawning. Runtime registries for pubsub, pipes, and overlays live +behind narrow mutex-protected structs in `runtime.rs`. Local control and +protected peer-control feature dispatch remain separate refactor targets. + The local metadata store is SQLite product state. `geth-store` tracks a numeric `schema_version` in the `meta` table and applies ordered migrations up to the crate's current schema version when the store opens. Fresh database creation and diff --git a/docs/production-readiness-roadmap.md b/docs/production-readiness-roadmap.md index 8915bb3..0e767e4 100644 --- a/docs/production-readiness-roadmap.md +++ b/docs/production-readiness-roadmap.md @@ -35,15 +35,15 @@ Goal: make the documented local quality gate pass before deeper refactors. Goal: split `geth-node` into reviewable daemon subsystems without changing behavior. -- `[~]` Extract daemon startup and runtime ownership. +- `[x]` Extract daemon startup and runtime ownership. Acceptance criteria: - - `[ ]` Daemon startup, shutdown, signal handling, socket setup, and Iroh + - `[x]` Daemon startup, shutdown, signal handling, socket setup, and Iroh endpoint ownership live outside the main feature handler module. - `[x]` Iroh accept-loop and background live-sync task spawning live outside the main feature handler module. - `[x]` Runtime state is represented by narrow structs with documented ownership and locking rules. - - `[ ]` Existing daemon startup and status tests pass unchanged. + - `[x]` Existing daemon startup and status tests pass unchanged. - `[ ]` Extract local control routing. Acceptance criteria: