build node in separate module

1 files changed, 31 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);