1 files changed, 120 insertions, 99 deletions

diff --git a/src/main.rs b/src/main.rs
index 09b365b..bfc45f0 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,37 +1,40 @@
 //! 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;
+mod clock;
 mod config;
 mod document;
+mod files;
+mod force;
 mod logging;
+mod node;
 mod remake;
+#[cfg(test)]
+mod simtest;
 mod state;
 mod swarm;
+#[cfg(test)]
+mod testrand;
 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<()> {
@@ -48,7 +51,7 @@ async fn main() -> Result<()> {
     let swarm_key = match &args.swarm_key_file {
         Some(path) => Some(Arc::new(swarm::auth::SwarmKey::from_file(path)?)),
         None => {
-            warn!("No swarm key: any node that learns the swarm ID can join");
+            warn!("No swarm key: any node that can reach a member may join");
             None
         }
     };
@@ -86,107 +89,124 @@ async fn main() -> Result<()> {
         "Using Mycelium REST API"
     );
 
-    let seeds: Vec<String> = args.seeds.clone();
+    let force = force::Force::parse(args.force.iter().map(String::as_str), config::FORCE)
+        .map_err(anyhow::Error::msg)?;
+    let files = Arc::new(files::Files::default());
+    let mut seeds: Vec<String> = args.seeds.clone();
+    let (state_file, remembered) = match &args.state_file {
+        Some(path) => {
+            let (file, remembered) =
+                swarm::state_file::StateFile::open(path.clone(), Arc::clone(&files));
+            (Some(Arc::new(file)), remembered)
+        }
+        None => (None, None),
+    };
+    if let Some(remembered) = &remembered {
+        info!(
+            swarm_id = %remembered.swarm_id,
+            seeds = remembered.seeds.len(),
+            "Remembering a swarm from the state file"
+        );
+        for seed in &remembered.seeds {
+            if !seeds.contains(seed) {
+                seeds.push(seed.clone());
+            }
+        }
+    } else if args.genesis && state_file.is_none() {
+        warn!(
+            "--genesis without --state-file: every restart with --genesis starts another new swarm"
+        );
+    }
     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 initial_epoch = match args.debug_epoch {
-        Some(generation) => Epoch {
+    let starting = bootstrap::starting_document(
+        &bootstrap::StartOptions {
+            recover_from: args.recover_from.as_deref(),
+            document_file: args.document_file.as_deref(),
+            genesis: args.genesis,
+            remembered: remembered.as_ref().map(|r| r.swarm_id.as_str()),
+        },
+        &files,
+        &force,
+    )?;
+    let (initial_doc, starting_epoch, starting_swarm, earlier_actors) = match starting {
+        Some(starting) => (
+            starting.doc,
+            Some(starting.epoch),
+            Some((starting.swarm_id, starting.origin)),
+            starting.earlier_actors,
+        ),
+        None => (AutoCommit::new(), None, None, Vec::new()),
+    };
+    let mut initial_doc = initial_doc;
+    if let Some(spec) = &args.remake {
+        remake::remake_genesis(spec, &files, &mut initial_doc)
+            .with_context(|| format!("Making the genesis document (--remake {spec})"))?;
+        info!(remake = %spec, "Made the genesis document");
+    }
+    let initial_epoch = match (starting_epoch, args.debug_epoch) {
+        (Some(epoch), _) => epoch,
+        (None, Some(generation)) => Epoch {
             generation,
             id: [0xee; 32],
         },
-        None => Epoch::GENESIS,
+        (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, AutoCommit::new()))),
-        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,
+        timing: state::Timing::from_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()),
-        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");
-
-    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;
+        blob_chunk_bytes: args.blob_chunk_kib * 1024,
+        swarm_key,
+        protocol_writers: args.protocol_writers.iter().cloned().collect(),
+        files,
+        state_file,
+        document_file,
+        config_report: config.report(),
+        clock: Default::default(),
     });
-    trace!("Sync worker task spawned");
+    let state = node.state.clone();
 
-    let peer_worker_state = state.clone();
-    tokio::spawn(async move {
-        peer::run_peer_worker(peer_worker_state).await;
-    });
-    trace!("Peer worker task spawned");
+    // 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).
+    let start = match starting_swarm {
+        Some((swarm_id, origin)) => {
+            state.set_expected_swarm_id(swarm_id).await?;
+            bootstrap::Start::FromDocument(origin)
+        }
+        None => {
+            if let Some(remembered) = remembered {
+                state.set_expected_swarm_id(remembered.swarm_id).await?;
+            }
+            if args.genesis {
+                bootstrap::Start::Genesis
+            } else {
+                bootstrap::Start::Join
+            }
+        }
+    };
 
-    // 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, args.genesis);
+    let joining = bootstrap::bootstrap(&state, &seeds, start);
     tokio::select! {
         joined = joining => joined.context("Joining swarm")?,
         _ = stopped(stop.clone()) => {
@@ -195,16 +215,14 @@ 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,
         ready = state.is_ready().await,
         "Bootstrap completed"
     );
+    swarm::state_file::update(&state, true).await;
 
     let auth_state = AuthState::new(&args.htpasswd);
     let app = api::router(state.clone(), auth_state, args.debug_api);
@@ -223,6 +241,7 @@ async fn main() -> Result<()> {
     });
 
     let server_fut = server_builder.into_future();
+    let stopping = state.clone();
     tokio::pin!(server_fut);
 
     tokio::select! {
@@ -248,6 +267,8 @@ async fn main() -> Result<()> {
     }
 
     info!("HTTP API server stopped");
+    swarm::state_file::update(&stopping, true).await;
+    document::file::flush(&stopping).await;
     Ok(())
 }