path: root/src/node.rs

//! Building a node and starting its workers: what `main` does once it has read the
//! configuration and connected its transport, 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 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::document::{ActorId, Document};
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, changes, peer, sync};
use crate::transport::Transport;
use crate::transport::firewall::Firewall;
use crate::transport::identity::{IdentityConfig, IdentityTransport};
use crate::transport::traffic::{CountingTransport, Traffic};

/// What a node is built from.
pub(crate) struct NodeConfig {
    /// The node's transport, without the identity, counting and firewall layers (added
    /// here).
    pub(crate) transport: Arc<dyn Transport>,
    /// The node key and the transport's address scheme: the transport goes under the
    /// identity layer (`transport::identity`), and the node is its node ID. Without it
    /// (simulations only), the node is its transport identity.
    pub(crate) identity: Option<IdentityConfig>,
    pub(crate) initial_epoch: Epoch,
    pub(crate) initial_doc: Document,
    /// Actors this node wrote with in earlier runs (`--document-file`).
    pub(crate) earlier_actors: Vec<ActorId>,
    /// The signatures of `initial_doc`'s changes (`--document-file`).
    pub(crate) initial_signatures: Vec<(crate::document::ChangeHash, [u8; 64])>,
    /// `Timing::new(--keepalive, the transports' sync reply timeout and delivery
    /// limit)`.
    pub(crate) timing: Timing,
    pub(crate) neighbours_k: usize,
    pub(crate) alias: Option<String>,
    /// `--dashboard-url`.
    pub(crate) dashboard_url: Option<String>,
    pub(crate) electorate_window: Duration,
    pub(crate) clock_skew_warn: Duration,
    /// `--stream-heartbeat`.
    pub(crate) stream_heartbeat: Duration,
    /// `--superseded-writes`.
    pub(crate) superseded_writes: crate::remake::Superseded,
    /// `--topology`.
    pub(crate) topology: crate::swarm::topology::Kind,
    /// `--dissemination`.
    pub(crate) dissemination: crate::swarm::dissemination::Kind,
    pub(crate) blob_chunk_bytes: u32,
    pub(crate) swarm_key: Option<Arc<SwarmKey>>,
    pub(crate) protocol_writers: HashSet<String>,
    /// On a genesis node: the swarm's first users, which it records
    /// (`access::registry::first_users`); empty elsewhere, and where none were named
    /// (the simulations: no registry).
    pub(crate) first_users: crate::access::registry::FirstUsers,
    /// To sign the tokens this node issues (`access::token`) and the changes it commits
    /// (`document::signatures`): every node's, its node ID being its public key; none in
    /// the simulations without the identity layer, whose node IDs are names.
    pub(crate) node_key: Option<Arc<crate::identity::NodeKey>>,
    /// `--sign-every`.
    pub(crate) sign_every: crate::document::signatures::SignEvery,
    /// `--token-lifetime`, `--session-lifetime`.
    pub(crate) token_lifetime: Duration,
    pub(crate) session_lifetime: Duration,
    /// `--challenge-lifetime`, `--key-logins-per-user`.
    pub(crate) challenge_lifetime: Duration,
    pub(crate) key_logins_per_user: usize,
    /// `--refused-writes-per-minute`.
    pub(crate) refused_writes_per_minute: usize,
    /// `--genesis-root`.
    pub(crate) genesis_root: Option<crate::document::ChangeHash>,
    /// `--strict-keys`.
    pub(crate) strict_keys: Vec<crate::access::sshsig::PublicKey>,
    /// `--cors-origins`.
    pub(crate) cors_origins: Vec<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_reports: crate::config::Reports,
    /// 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 firewall = Arc::new(Firewall::default());
        let traffic = Arc::new(Traffic::default());
        let initial_actor = crate::document::actor(&config.initial_doc);
        let transport = match config.identity {
            Some(identity) => Arc::new(IdentityTransport::new(
                config.transport,
                identity,
                Arc::clone(&traffic),
                Some(Arc::clone(&firewall)),
                Arc::clone(&config.clock),
                config.timing.keepalive,
            )) as Arc<dyn Transport>,
            None => config.transport,
        };

        let mut replica = Replica::new(config.initial_epoch, config.initial_doc);
        replica.signatures = crate::document::signatures::Signatures::from_kept(
            &replica.doc,
            config.initial_signatures,
        );
        // What this node kept unsigned when it stopped: signed before anything leaves.
        if let Some(key) = &config.node_key {
            let replica = &mut replica;
            let signed =
                replica
                    .signatures
                    .sign_own_actors(&mut replica.doc, key, &config.earlier_actors);
            if signed > 0 {
                tracing::info!(
                    signed,
                    "Signed changes this node kept unsigned when it stopped"
                );
            }
        }
        let stamp = replica.stamps();
        let state = AppState {
            doc: Arc::new(crate::document::lock::DocumentLock::new(replica)),
            stamp,
            lifecycle: Arc::new(RwLock::new(NodeLifecycle::Joining {
                discovered_swarm_id: None,
            })),
            peer_states: Arc::new(RwLock::new(HashMap::new())),
            ways: Arc::new(std::sync::RwLock::new(crate::swarm::topology::Ways::new(
                config.topology,
                config.dissemination,
            ))),
            configured_ways: (config.topology, config.dissemination),
            // Counted above the firewall (in the paths layer, under the identity layer):
            // what it drops outbound counts as sent (it plays a lossy network), what it
            // drops inbound never arrived. Without the identity layer (tests) there is
            // no firewall.
            transport: Arc::new(CountingTransport::new(transport, Arc::clone(&traffic))),
            firewall,
            traffic,
            stats: Default::default(),
            sync_events: sync_tx,
            change_tx,
            initial_sync_done: Arc::new(AtomicBool::new(false)),
            closing: Arc::new(watch::channel(false).0),
            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(),
            handshakes: Default::default(),
            neighbours_k: config.neighbours_k,
            alias: config.alias,
            dashboard_url: config.dashboard_url,
            electorate_window: config.electorate_window,
            clock_skew_warn: config.clock_skew_warn,
            stream_heartbeat: config.stream_heartbeat,
            superseded_writes: config.superseded_writes,
            clock_ahead_ms: Default::default(),
            clock_skew: Default::default(),
            blobs: Arc::new(swarm::blob::Blobs::new(config.blob_chunk_bytes)),
            protocol_writers: Arc::new(config.protocol_writers),
            first_users: Arc::new(config.first_users),
            node_key: config.node_key,
            sign_every: config.sign_every,
            screening: Arc::new(std::sync::Mutex::new(
                crate::access::screen::Screening::pinned(crate::swarm::roots::pin(
                    config.genesis_root,
                    &Default::default(),
                    false,
                )),
            )),
            roots: Default::default(),
            genesis_root: config.genesis_root,
            challenges: Default::default(),
            ended_sessions: Default::default(),
            session_ended: Default::default(),
            token_lifetime: config.token_lifetime,
            session_lifetime: config.session_lifetime,
            challenge_lifetime: config.challenge_lifetime,
            key_logins_per_user: config.key_logins_per_user,
            refused_writes: Default::default(),
            refused_writes_per_minute: config.refused_writes_per_minute,
            strict_keys: Arc::new(config.strict_keys),
            cors_origins: Arc::new(config.cors_origins),
            state_file: config.state_file,
            document_file: config.document_file,
            files: config.files,
            config: Arc::new(config.config_reports),
            neighbours: Default::default(),
            topology_changed: Default::default(),
            join_progress: Default::default(),
            forget_records_changed: Default::default(),
            peers_due: Default::default(),
            pulling: Default::default(),
            announce_pending: Default::default(),
            announce_slow: Default::default(),
            slow_rounds: Default::default(),
            reach: Default::default(),
            working_links: Default::default(),
            trees: 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();
        let lanes = Arc::new(peer::SyncLanes::start(&state));
        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();
                    let lanes = Arc::clone(&lanes);
                    Box::pin(async move {
                        peer::handle_message(&state_clone, &lanes, 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");

        let reachable = Arc::new(tokio::sync::Notify::new());
        state.transport.on_reachable(Arc::clone(&reachable));
        tokio::spawn(peer::probe_when_reachable(state.clone(), reachable));

        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 { changes::run_change_publisher(publisher_state).await });
    tokio::spawn(crate::swarm::document_file::run(state.clone()));
}