1 files changed, 178 insertions, 0 deletions
diff --git a/src/node.rs b/src/node.rs
new file mode 100644
index 0000000..5005d30
--- /dev/null
+++ b/src/node.rs
@@ -0,0 +1,178 @@
+//! 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::clock::Clock;
+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>,
+    /// `Timing::from_keepalive(--keepalive)`.
+    pub(crate) timing: Timing,
+    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,
+    /// The node's wall clock; `Clock::default()` outside tests.
+    pub(crate) clock: Arc<Clock>,
+}
+
+/// 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: config.timing,
+            remake: Arc::new(remake::RemakeState::new(config.clock.now_millis())),
+            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(),
+            topology_changed: 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(),
+            catch_up_avoided: Default::default(),
+            clock: config.clock,
+        };
+        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");
+
+        tokio::spawn(swarm::topology::run_worker(state.clone()));
+        trace!("Topology 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()));
+}