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

refactor: extract daemon task spawning

Browse source
This commit is contained in:
Eric Wendland 2026-07-05 17:37:12 +02:00
commit e7502a7237
3 changed files with 43 additions and 35 deletions
Download patch file Download diff file Expand all files Collapse all files

37
crates/geth-node/src/daemon.rs Normal file
View file

@ -0,0 +1,37 @@
//! Daemon task orchestration helpers.
use crate::LocalNode;
use geth_iroh::GethIrohEndpoint;
use std::time::Duration;
pub(crate) fn spawn_iroh_control_accept_loop(node: LocalNode, endpoint: GethIrohEndpoint) {
let raw_endpoint = endpoint.endpoint();
tokio::spawn(async move {
tracing::debug!("iroh accept loop started");
while let Some(incoming) = raw_endpoint.accept().await {
tracing::debug!("iroh incoming connection accepted by endpoint loop");
let node = node.clone();
tokio::spawn(async move {
if let Err(error) = super::handle_iroh_control_connection(node, incoming).await {
tracing::warn!(%error, "iroh control request failed");
}
});
}
tracing::debug!("iroh accept loop ended");
});
}
pub(crate) fn spawn_background_live_sync(node: LocalNode, interval_duration: Duration) {
tokio::spawn(async move {
if let Err(error) = super::run_live_sync_once(&node).await {
tracing::debug!(%error, "initial live sync tick failed");
}
let mut interval = tokio::time::interval(interval_duration);
loop {
interval.tick().await;
if let Err(error) = super::run_live_sync_once(&node).await {
tracing::debug!(%error, "live sync tick failed");
}
}
});
}

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

@ -1,9 +1,11 @@
mod daemon;
mod runtime;
pub mod service;
mod sync;
mod wire;
use base64::Engine;
use daemon::{spawn_background_live_sync, spawn_iroh_control_accept_loop};
use futures::StreamExt;
use geth_auth::{AUTH_SIGNATURE_NAMESPACE, AuthExplanation, AuthOp, AuthOpKind, AuthOpSignature};
use geth_cas::{
@ -4334,38 +4336,6 @@ async fn request_overlay_wire(
}
}
fn spawn_iroh_control_accept_loop(node: LocalNode, endpoint: GethIrohEndpoint) {
let raw_endpoint = endpoint.endpoint();
tokio::spawn(async move {
tracing::debug!("iroh accept loop started");
while let Some(incoming) = raw_endpoint.accept().await {
tracing::debug!("iroh incoming connection accepted by endpoint loop");
let node = node.clone();
tokio::spawn(async move {
if let Err(error) = handle_iroh_control_connection(node, incoming).await {
tracing::warn!(%error, "iroh control request failed");
}
});
}
tracing::debug!("iroh accept loop ended");
});
}
fn spawn_background_live_sync(node: LocalNode, interval_duration: Duration) {
tokio::spawn(async move {
if let Err(error) = run_live_sync_once(&node).await {
tracing::debug!(%error, "initial live sync tick failed");
}
let mut interval = tokio::time::interval(interval_duration);
loop {
interval.tick().await;
if let Err(error) = run_live_sync_once(&node).await {
tracing::debug!(%error, "live sync tick failed");
}
}
});
}
async fn run_sync_for_peer(node: &LocalNode, peer_node: &str) -> Result<SyncPeerRun, NodeError> {
let kv_stores = Store::open(&node.paths.metadata_db())?.list_kv_stores()?;
let documents = Store::open(&node.paths.metadata_db())?.list_document_resources()?;

7
docs/production-readiness-roadmap.md
View file

@ -37,9 +37,10 @@ behavior.
- `[~]` Extract daemon startup and runtime ownership.
Acceptance criteria:
- `[ ]` Daemon startup, shutdown, signal handling, socket setup, Iroh
endpoint ownership, and background task spawning live outside the main
feature handler module.
- `[ ]` 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.