summaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/main.rs141
-rw-r--r--src/node.rs168
2 files changed, 199 insertions, 110 deletions
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<dyn Transport>,
+ 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<ActorId>,
+ pub(crate) keepalive: Duration,
+ pub(crate) neighbours_k: usize,
+ pub(crate) alias: Option<String>,
+ pub(crate) electorate_window: Duration,
+ pub(crate) blob_chunk_bytes: u32,
+ pub(crate) swarm_key: Option<Arc<SwarmKey>>,
+ pub(crate) protocol_writers: HashSet<String>,
+ pub(crate) files: Arc<Files>,
+ pub(crate) state_file: Option<Arc<StateFile>>,
+ pub(crate) document_file: Option<Arc<DocumentFile>>,
+ /// 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<SyncEvent>,
+}
+
+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()));
+}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-26 17:01:29 +0000