From 51973b974d76507d3bc748fced6750b651953e80 Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 25 Oct 2026 21:29:39 +0100 Subject: build node in separate module --- src/main.rs | 141 +++++++++++--------------------------------------- src/node.rs | 168 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 199 insertions(+), 110 deletions(-) create mode 100644 src/node.rs (limited to 'src') diff --git a/src/main.rs b/src/main.rs index 321fe91..0a978fe 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,18 +1,16 @@ //! continuum: a replicated Automerge key-value store synchronized over Mycelium. //! -//! main.rs only wires things together: arguments, state construction, background -//! tasks, bootstrap and the HTTP server. +//! main.rs only wires things together: arguments, the starting document, the node +//! (`node`), bootstrap and the HTTP server. -use std::collections::HashMap; use std::future::IntoFuture; use std::sync::Arc; -use std::sync::atomic::AtomicBool; use std::time::Duration; use anyhow::{Context, Result}; use automerge::AutoCommit; -use tokio::sync::{RwLock, broadcast, mpsc, watch}; -use tracing::{debug, error, info, trace, warn}; +use tokio::sync::watch; +use tracing::{debug, error, info, warn}; mod api; mod backoff; @@ -21,19 +19,17 @@ mod document; mod files; mod force; mod logging; +mod node; mod remake; mod state; mod swarm; mod transport; use api::auth::AuthState; -use document::replica::{Epoch, Replica}; -use state::{AppState, NodeLifecycle, Timing}; -use swarm::{bootstrap, peer, sync}; +use document::replica::Epoch; +use swarm::bootstrap; use transport::Transport; -use transport::firewall::{Firewall, FirewallTransport}; use transport::mycelium::{self, MyceliumSetup, MyceliumSpawnConfig}; -use transport::traffic::{CountingTransport, Traffic}; #[tokio::main] async fn main() -> Result<()> { @@ -118,10 +114,6 @@ async fn main() -> Result<()> { } debug!(seed_count = seeds.len(), "Loaded bootstrap configuration"); - let (sync_tx, sync_rx) = mpsc::unbounded_channel(); - let (change_tx, _) = watch::channel(0_u64); - let (change_events, _) = broadcast::channel(256); - let starting = bootstrap::starting_document( &bootstrap::StartOptions { recover_from: args.recover_from.as_deref(), @@ -147,7 +139,6 @@ async fn main() -> Result<()> { .with_context(|| format!("Making the genesis document (--remake {spec})"))?; info!(remake = %spec, "Made the genesis document"); } - let initial_actor = initial_doc.get_actor().clone(); let initial_epoch = match (starting_epoch, args.debug_epoch) { (Some(epoch), _) => epoch, (None, Some(generation)) => Epoch { @@ -156,74 +147,37 @@ async fn main() -> Result<()> { }, (None, None) => Epoch::GENESIS, }; - let (epoch_tx, _) = watch::channel(initial_epoch); debug!(epoch = %initial_epoch, "Local epoch"); - let firewall = Arc::new(Firewall::default()); - let traffic = Arc::new(Traffic::default()); - let mycelium_transport = Arc::new(mycelium::MyceliumTransport::connect(mycelium_setup).await?); debug!(local_id = mycelium_transport.local_id(), "Transport ready"); - let state = AppState { - doc: Arc::new(RwLock::new(Replica::new(initial_epoch, initial_doc))), - epoch_tx, - lifecycle: Arc::new(RwLock::new(NodeLifecycle::Joining { - discovered_swarm_id: None, - })), - peer_states: Arc::new(RwLock::new(HashMap::new())), - // Counted outside the firewall: what it drops outbound counts as sent (it plays a - // lossy network), what it drops inbound never arrived. - transport: Arc::new(CountingTransport::new( - Arc::new(FirewallTransport::new( - mycelium_transport, - Arc::clone(&firewall), - )), - Arc::clone(&traffic), - )), - firewall, - traffic, - stats: Default::default(), - sync_events: sync_tx, - change_tx, - initial_sync_done: Arc::new(AtomicBool::new(false)), - change_events, - timing: Timing::from_keepalive(Duration::from_secs(args.keepalive_secs)), - remake: Arc::new(remake::RemakeState::default()), - swarm_key, - run_id: uuid::Uuid::new_v4().into_bytes(), + let document_file = args.document_file.clone().map(|path| { + Arc::new(document::file::DocumentFile::new( + path, + Arc::clone(&files), + Duration::from_secs(args.document_append_secs), + Duration::from_secs(args.document_rewrite_secs), + )) + }); + let node = node::Node::new(node::NodeConfig { + transport: mycelium_transport, + initial_epoch, + initial_doc, + earlier_actors, + keepalive: Duration::from_secs(args.keepalive_secs), neighbours_k: args.neighbours, alias: args.alias.clone(), electorate_window: Duration::from_secs(args.electorate_window_days * 24 * 3600), - blobs: Arc::new(swarm::blob::Blobs::new(args.blob_chunk_kib * 1024)), - protocol_writers: Arc::new(args.protocol_writers.iter().cloned().collect()), - state_file, - document_file: args.document_file.clone().map(|path| { - Arc::new(document::file::DocumentFile::new( - path, - Arc::clone(&files), - Duration::from_secs(args.document_append_secs), - Duration::from_secs(args.document_rewrite_secs), - )) - }), + blob_chunk_bytes: args.blob_chunk_kib * 1024, + swarm_key, + protocol_writers: args.protocol_writers.iter().cloned().collect(), files, - config: Arc::new(config.report()), - neighbours: Default::default(), - pulling: Default::default(), - announce_pending: Default::default(), - rtt: Default::default(), - throughput: Default::default(), - stability: Default::default(), - stale: Default::default(), - catch_up_source: Default::default(), - catch_up_preference: Default::default(), - }; - - trace!("Application state initialized"); - for actor in earlier_actors { - state.remake.add_own_actor(actor); - } - state.remake.add_own_actor(initial_actor); + state_file, + document_file, + config_report: config.report(), + }); + let state = node.state.clone(); // Join the remembered swarm and no other; HELLOs claim it. A document to start on // names its swarm itself (checked against the remembered one above). @@ -244,36 +198,7 @@ async fn main() -> Result<()> { } }; - let consumer_state = state.clone(); - tokio::spawn(async move { - let inner_state = consumer_state.clone(); - let _ = consumer_state - .transport - .start_consumer(Box::new(move |sender, msg, _payload_len| { - let state_clone = inner_state.clone(); - Box::pin(async move { peer::handle_message(&state_clone, sender, msg).await }) - })) - .await; - }); - trace!("Mycelium consumer task spawned"); - - let sync_state = state.clone(); - tokio::spawn(async move { - sync::run_sync_worker(sync_state, sync_rx).await; - }); - trace!("Sync worker task spawned"); - - let peer_worker_state = state.clone(); - tokio::spawn(async move { - peer::run_peer_worker(peer_worker_state).await; - }); - trace!("Peer worker task spawned"); - - // Switches to committed epochs and catches up with newer ones. Starts before - // bootstrapping: a node joining a compacted swarm can only finish its initial sync - // after fetching the current epoch from a peer. - tokio::spawn(remake::run_worker(state.clone())); - trace!("Remake worker task spawned"); + node.spawn_workers(); let joining = bootstrap::bootstrap(&state, &seeds, start); tokio::select! { @@ -284,10 +209,7 @@ async fn main() -> Result<()> { } } - // Start the change publisher after bootstrapping: streams only exist from then on, - // and it avoids diffing the initial sync. - let publisher_state = state.clone(); - tokio::spawn(async move { sync::run_change_publisher(publisher_state).await }); + node::spawn_ready_workers(&state); info!( swarm_id = ?state.swarm_id().await, @@ -295,7 +217,6 @@ async fn main() -> Result<()> { "Bootstrap completed" ); swarm::state_file::update(&state, true).await; - tokio::spawn(document::file::run(state.clone())); let auth_state = AuthState::new(&args.htpasswd); let app = api::router(state.clone(), auth_state, args.debug_api); diff --git a/src/node.rs b/src/node.rs new file mode 100644 index 0000000..b3e7463 --- /dev/null +++ b/src/node.rs @@ -0,0 +1,168 @@ +//! Building a node and starting its workers: what `main` does once it has read the +//! configuration and connected to Mycelium, and what the simulator (`transport::sim`) +//! does for every simulated node. Everything here is independent of where the +//! transport and the settings come from. + +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::time::Duration; + +use automerge::{ActorId, AutoCommit}; +use tokio::sync::{RwLock, broadcast, mpsc, watch}; +use tracing::trace; + +use crate::document::file::DocumentFile; +use crate::document::replica::{Epoch, Replica}; +use crate::files::Files; +use crate::remake; +use crate::state::{AppState, NodeLifecycle, Timing}; +use crate::swarm::auth::SwarmKey; +use crate::swarm::state_file::StateFile; +use crate::swarm::sync::SyncEvent; +use crate::swarm::{self, peer, sync}; +use crate::transport::Transport; +use crate::transport::firewall::{Firewall, FirewallTransport}; +use crate::transport::traffic::{CountingTransport, Traffic}; + +/// What a node is built from. +pub(crate) struct NodeConfig { + /// The node's transport, without the counting and firewall layers (added here). + pub(crate) transport: Arc, + pub(crate) initial_epoch: Epoch, + pub(crate) initial_doc: AutoCommit, + /// Actors this node wrote with in earlier runs (`--document-file`). + pub(crate) earlier_actors: Vec, + pub(crate) keepalive: Duration, + pub(crate) neighbours_k: usize, + pub(crate) alias: Option, + pub(crate) electorate_window: Duration, + pub(crate) blob_chunk_bytes: u32, + pub(crate) swarm_key: Option>, + pub(crate) protocol_writers: HashSet, + pub(crate) files: Arc, + pub(crate) state_file: Option>, + pub(crate) document_file: Option>, + /// For `GET /admin/config`. + pub(crate) config_report: serde_json::Value, +} + +/// A built node whose workers haven't started yet. +pub(crate) struct Node { + pub(crate) state: AppState, + sync_rx: mpsc::UnboundedReceiver, +} + +impl Node { + pub(crate) fn new(config: NodeConfig) -> Self { + let (sync_tx, sync_rx) = mpsc::unbounded_channel(); + let (change_tx, _) = watch::channel(0_u64); + let (change_events, _) = broadcast::channel(256); + let (epoch_tx, _) = watch::channel(config.initial_epoch); + let firewall = Arc::new(Firewall::default()); + let traffic = Arc::new(Traffic::default()); + let initial_actor = config.initial_doc.get_actor().clone(); + + let state = AppState { + doc: Arc::new(RwLock::new(Replica::new( + config.initial_epoch, + config.initial_doc, + ))), + epoch_tx, + lifecycle: Arc::new(RwLock::new(NodeLifecycle::Joining { + discovered_swarm_id: None, + })), + peer_states: Arc::new(RwLock::new(HashMap::new())), + // Counted outside the firewall: what it drops outbound counts as sent (it + // plays a lossy network), what it drops inbound never arrived. + transport: Arc::new(CountingTransport::new( + Arc::new(FirewallTransport::new( + config.transport, + Arc::clone(&firewall), + )), + Arc::clone(&traffic), + )), + firewall, + traffic, + stats: Default::default(), + sync_events: sync_tx, + change_tx, + initial_sync_done: Arc::new(AtomicBool::new(false)), + change_events, + timing: Timing::from_keepalive(config.keepalive), + remake: Arc::new(remake::RemakeState::default()), + swarm_key: config.swarm_key, + run_id: uuid::Uuid::new_v4().into_bytes(), + neighbours_k: config.neighbours_k, + alias: config.alias, + electorate_window: config.electorate_window, + blobs: Arc::new(swarm::blob::Blobs::new(config.blob_chunk_bytes)), + protocol_writers: Arc::new(config.protocol_writers), + state_file: config.state_file, + document_file: config.document_file, + files: config.files, + config: Arc::new(config.config_report), + neighbours: Default::default(), + pulling: Default::default(), + announce_pending: Default::default(), + rtt: Default::default(), + throughput: Default::default(), + stability: Default::default(), + stale: Default::default(), + catch_up_source: Default::default(), + catch_up_preference: Default::default(), + }; + for actor in config.earlier_actors { + state.remake.add_own_actor(actor); + } + state.remake.add_own_actor(initial_actor); + trace!("Application state initialized"); + Self { state, sync_rx } + } + + /// Starts what runs from the start: the message consumer and the sync, peer and + /// remake workers. The remake worker starts before bootstrapping: a node joining a + /// compacted swarm can only finish its initial sync after fetching the current epoch + /// from a peer. + pub(crate) fn spawn_workers(self) -> AppState { + let Self { state, sync_rx } = self; + + let consumer_state = state.clone(); + tokio::spawn(async move { + let inner_state = consumer_state.clone(); + let _ = consumer_state + .transport + .start_consumer(Box::new(move |sender, msg, _payload_len| { + let state_clone = inner_state.clone(); + Box::pin(async move { peer::handle_message(&state_clone, sender, msg).await }) + })) + .await; + }); + trace!("Message consumer task spawned"); + + let sync_state = state.clone(); + tokio::spawn(async move { + sync::run_sync_worker(sync_state, sync_rx).await; + }); + trace!("Sync worker task spawned"); + + let peer_worker_state = state.clone(); + tokio::spawn(async move { + peer::run_peer_worker(peer_worker_state).await; + }); + trace!("Peer worker task spawned"); + + tokio::spawn(remake::run_worker(state.clone())); + trace!("Remake worker task spawned"); + + state + } +} + +/// Starts what runs once bootstrap is done: the change publisher (streams only exist +/// from then on, and it avoids diffing the initial sync) and the document file writer. +pub(crate) fn spawn_ready_workers(state: &AppState) { + let publisher_state = state.clone(); + tokio::spawn(async move { sync::run_change_publisher(publisher_state).await }); + tokio::spawn(crate::document::file::run(state.clone())); +} -- cgit v1.3.1