refactor: extract node runtime state

This commit is contained in:
Eric Wendland 2026-07-05 17:27:42 +02:00
commit 87fe47fd7f
3 changed files with 91 additions and 72 deletions

View file

@ -1,3 +1,4 @@
mod runtime;
pub mod service; pub mod service;
use base64::Engine; use base64::Engine;
@ -54,7 +55,11 @@ use geth_types::{
}; };
use iroh::protocol::ProtocolHandler; use iroh::protocol::ProtocolHandler;
use iroh_docs::api::protocol::{AddrInfoOptions, ShareMode}; use iroh_docs::api::protocol::{AddrInfoOptions, ShareMode};
use std::collections::{BTreeMap, BTreeSet, VecDeque}; use runtime::{
KvDocsState, LiveSyncCursor, LiveSyncHealth, NodeRuntime, OverlayTunCounters,
OverlayTunRuntime, PipeRuntime, PubsubGossipTopicRuntime, PubsubRuntime,
};
use std::collections::{BTreeMap, BTreeSet};
use std::path::{Component, Path, PathBuf}; use std::path::{Component, Path, PathBuf};
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
use std::time::Duration; use std::time::Duration;
@ -178,79 +183,11 @@ pub struct LocalNode {
runtime: Arc<NodeRuntime>, runtime: Arc<NodeRuntime>,
} }
#[derive(Debug)]
struct NodeRuntime {
pubsub: Mutex<PubsubRuntime>,
pipes: Mutex<PipeRuntime>,
overlays: Mutex<BTreeMap<String, OverlayTunRuntime>>,
}
#[derive(Debug, Default)]
struct PubsubRuntime {
messages: VecDeque<PubsubMessage>,
gossip_topics: BTreeMap<String, PubsubGossipTopicRuntime>,
}
#[derive(Debug, Clone)]
struct PubsubGossipTopicRuntime {
sender: iroh_gossip::api::GossipSender,
}
#[derive(Debug, Default)]
struct PipeRuntime {
listeners: BTreeMap<String, PipeListener>,
connections: VecDeque<PipeConnection>,
messages: VecDeque<PipeMessage>,
}
#[derive(Debug)]
struct OverlayTunRuntime {
name: String,
interface_name: String,
virtual_ip: String,
cidr: String,
mtu: u16,
started_at_ms: i64,
inject_tx: mpsc::Sender<Vec<u8>>,
stop_tx: Option<oneshot::Sender<()>>,
counters: Arc<Mutex<OverlayTunCounters>>,
}
#[derive(Debug, Default)]
struct OverlayTunCounters {
packets_from_tun: u64,
packets_to_tun: u64,
packets_to_peers: u64,
last_error: Option<String>,
}
const PUBSUB_RING_LIMIT: usize = 256; const PUBSUB_RING_LIMIT: usize = 256;
const PIPE_CONNECTION_RING_LIMIT: usize = 256; const PIPE_CONNECTION_RING_LIMIT: usize = 256;
const PIPE_MESSAGE_RING_LIMIT: usize = 1024; const PIPE_MESSAGE_RING_LIMIT: usize = 1024;
const LIVE_SYNC_STALE_AFTER_MS: i64 = 120_000; const LIVE_SYNC_STALE_AFTER_MS: i64 = 120_000;
#[derive(Debug, serde::Deserialize, serde::Serialize)]
struct LiveSyncCursor {
cursor_ms: i64,
}
#[derive(Debug, serde::Deserialize, serde::Serialize)]
struct LiveSyncHealth {
last_attempt_ms: i64,
last_success_ms: Option<i64>,
last_error: Option<String>,
last_imported: usize,
last_rejected: usize,
}
#[derive(Debug, serde::Deserialize, serde::Serialize)]
struct KvDocsState {
name: String,
namespace_id: String,
read_ticket: String,
updated_at_ms: i64,
}
struct PipeTcpConnectWire { struct PipeTcpConnectWire {
peer_card: PeerCard, peer_card: PeerCard,
target_addr: String, target_addr: String,

View file

@ -0,0 +1,82 @@
//! Daemon-local runtime state.
//!
//! `LocalNode` owns one `Arc<NodeRuntime>` for daemon-lifetime, in-memory
//! registries and task handles. Durable resource state still belongs in
//! `geth-store`; these structs only hold transient pubsub rings, pipe
//! listeners/messages, and active overlay TUN runtimes. Each registry has its
//! own mutex so feature handlers can lock the smallest runtime surface they
//! mutate.
use geth_pipe::{PipeConnection, PipeListener, PipeMessage};
use geth_pubsub::PubsubMessage;
use std::collections::{BTreeMap, VecDeque};
use std::sync::{Arc, Mutex};
use tokio::sync::{mpsc, oneshot};
#[derive(Debug)]
pub(crate) struct NodeRuntime {
pub(crate) pubsub: Mutex<PubsubRuntime>,
pub(crate) pipes: Mutex<PipeRuntime>,
pub(crate) overlays: Mutex<BTreeMap<String, OverlayTunRuntime>>,
}
#[derive(Debug, Default)]
pub(crate) struct PubsubRuntime {
pub(crate) messages: VecDeque<PubsubMessage>,
pub(crate) gossip_topics: BTreeMap<String, PubsubGossipTopicRuntime>,
}
#[derive(Debug, Clone)]
pub(crate) struct PubsubGossipTopicRuntime {
pub(crate) sender: iroh_gossip::api::GossipSender,
}
#[derive(Debug, Default)]
pub(crate) struct PipeRuntime {
pub(crate) listeners: BTreeMap<String, PipeListener>,
pub(crate) connections: VecDeque<PipeConnection>,
pub(crate) messages: VecDeque<PipeMessage>,
}
#[derive(Debug)]
pub(crate) struct OverlayTunRuntime {
pub(crate) name: String,
pub(crate) interface_name: String,
pub(crate) virtual_ip: String,
pub(crate) cidr: String,
pub(crate) mtu: u16,
pub(crate) started_at_ms: i64,
pub(crate) inject_tx: mpsc::Sender<Vec<u8>>,
pub(crate) stop_tx: Option<oneshot::Sender<()>>,
pub(crate) counters: Arc<Mutex<OverlayTunCounters>>,
}
#[derive(Debug, Default)]
pub(crate) struct OverlayTunCounters {
pub(crate) packets_from_tun: u64,
pub(crate) packets_to_tun: u64,
pub(crate) packets_to_peers: u64,
pub(crate) last_error: Option<String>,
}
#[derive(Debug, serde::Deserialize, serde::Serialize)]
pub(crate) struct LiveSyncCursor {
pub(crate) cursor_ms: i64,
}
#[derive(Debug, serde::Deserialize, serde::Serialize)]
pub(crate) struct LiveSyncHealth {
pub(crate) last_attempt_ms: i64,
pub(crate) last_success_ms: Option<i64>,
pub(crate) last_error: Option<String>,
pub(crate) last_imported: usize,
pub(crate) last_rejected: usize,
}
#[derive(Debug, serde::Deserialize, serde::Serialize)]
pub(crate) struct KvDocsState {
pub(crate) name: String,
pub(crate) namespace_id: String,
pub(crate) read_ticket: String,
pub(crate) updated_at_ms: i64,
}

View file

@ -35,12 +35,12 @@ 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. - `[~]` Extract daemon startup and runtime ownership.
Acceptance criteria: Acceptance criteria:
- `[ ]` Daemon startup, shutdown, signal handling, socket setup, Iroh - `[ ]` Daemon startup, shutdown, signal handling, socket setup, Iroh
endpoint ownership, and background task spawning live outside the main endpoint ownership, and background task spawning live outside the main
feature handler module. feature handler module.
- `[ ]` 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. - `[ ]` Existing daemon startup and status tests pass unchanged.
@ -330,7 +330,7 @@ Goal: prove the system works as an actual base layer before broader use.
## Working Order ## Working Order
1. `[~]` Finish Phase 0. 1. `[~]` Finish Phase 0.
2. `[ ]` Refactor `geth-node` into daemon subsystems. 2. `[~]` Refactor `geth-node` into daemon subsystems.
3. `[ ]` Add stable contract and golden JSON tests. 3. `[ ]` Add stable contract and golden JSON tests.
4. `[ ]` Harden store migrations and backup. 4. `[ ]` Harden store migrations and backup.
5. `[ ]` Complete security-boundary test coverage. 5. `[ ]` Complete security-boundary test coverage.