refactor: move daemon startup ownership

This commit is contained in:
Eric Wendland 2026-07-05 23:22:53 +02:00
commit d5a91c2396
4 changed files with 169 additions and 141 deletions

View file

@ -1,8 +1,63 @@
//! Daemon task orchestration helpers. //! Daemon task orchestration helpers.
use crate::{LocalNode, sync::run_live_sync_once}; use crate::{
use geth_iroh::GethIrohEndpoint; LocalNode, NodeError, handle_stream, init_node, mirror_all_kv_stores_to_iroh_docs,
use std::time::Duration; 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) { pub(crate) fn spawn_iroh_control_accept_loop(node: LocalNode, endpoint: GethIrohEndpoint) {
let raw_endpoint = endpoint.endpoint(); 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<Option<GethIrohEndpoint>, 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<String, geth_config::RelayMapConfig>,
) -> 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(),
}
}
}
}

View file

@ -8,14 +8,16 @@ mod sync;
mod wire; mod wire;
use base64::Engine; 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 futures::StreamExt;
use geth_auth::{AUTH_SIGNATURE_NAMESPACE, AuthExplanation, AuthOp, AuthOpKind, AuthOpSignature}; use geth_auth::{AUTH_SIGNATURE_NAMESPACE, AuthExplanation, AuthOp, AuthOpKind, AuthOpSignature};
use geth_cas::{ use geth_cas::{
BlobInfoSummary, CasTreeEntryKind, CasTreeObject, FileConflict, FileConflictKind, BlobInfoSummary, CasTreeEntryKind, CasTreeObject, FileConflict, FileConflictKind,
FileConflictResolution, FileConflictStatus, FileRoot, FileRootScan, LocalCas, hash_path, FileConflictResolution, FileConflictStatus, FileRoot, FileRootScan, LocalCas, hash_path,
}; };
use geth_config::{GethConfig, GethPaths, RelayMode}; use geth_config::{GethConfig, GethPaths};
use geth_control::{ use geth_control::{
CasBlob, CasProvider, ControlRequest, ControlResponse, KeychainStatusResponse, CasBlob, CasProvider, ControlRequest, ControlResponse, KeychainStatusResponse,
NativeBackendStatus, NodeIdResponse, OverlayPeer, OverlayWireRequest, OverlayWireResponse, 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, discovery_is_untrusted_note, peer_card_from_txt_attributes, peer_card_txt_attributes,
}; };
use geth_document::{DocumentResource, DocumentState}; use geth_document::{DocumentResource, DocumentState};
use geth_iroh::{EndpointStatus, GethIrohConfig, GethIrohEndpoint, GethRelayMode}; use geth_iroh::{EndpointStatus, GethIrohEndpoint};
use geth_keychain::{ use geth_keychain::{
KeychainOp, KeychainOpKind, KeychainOpSignature, NODE_ENROLLMENT_REQUEST_NAMESPACE, KeychainOp, KeychainOpKind, KeychainOpSignature, NODE_ENROLLMENT_REQUEST_NAMESPACE,
NodeEnrollmentCapability, NodeEnrollmentProvenance, NodeEnrollmentRequest, NodeEnrollmentCapability, NodeEnrollmentProvenance, NodeEnrollmentRequest,
@ -413,42 +415,6 @@ fn initialize_owner_keychain(
Ok(ops) 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( pub async fn send_control(
paths: &GethPaths, paths: &GethPaths,
request: ControlRequest, request: ControlRequest,
@ -11361,102 +11327,6 @@ fn stable_node_id(agent_id: &str) -> String {
format!("node:{agent_id}") format!("node:{agent_id}")
} }
async fn start_daemon_iroh_endpoint(
node: &mut LocalNode,
) -> Result<Option<GethIrohEndpoint>, 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<String, geth_config::RelayMapConfig>,
) -> 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 { fn expected_openssh_cert_path(public_key_path: &Path) -> String {
let text = public_key_path.display().to_string(); let text = public_key_path.display().to_string();
if let Some(prefix) = text.strip_suffix(".pub") { if let Some(prefix) = text.strip_suffix(".pub") {

View file

@ -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 socket. Control commands connect to the Unix socket and send typed JSONL
requests. 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 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 `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 crate's current schema version when the store opens. Fresh database creation and

View file

@ -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 Goal: split `geth-node` into reviewable daemon subsystems without changing
behavior. behavior.
- `[~]` Extract daemon startup and runtime ownership. - `[x]` Extract daemon startup and runtime ownership.
Acceptance criteria: 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. endpoint ownership live outside the main feature handler module.
- `[x]` Iroh accept-loop and background live-sync task spawning live outside - `[x]` Iroh accept-loop and background live-sync task spawning live outside
the main feature handler module. the main feature handler module.
- `[x]` Runtime state is represented by narrow structs with documented - `[x]` Runtime state is represented by narrow structs with documented
ownership and locking rules. 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. - `[ ]` Extract local control routing.
Acceptance criteria: Acceptance criteria: