roots: an epoch's root pinned, and named in every handshake

Review 2, N7, the care it raised: a joining node's empty document took
as the first epoch's root whatever dependency-less change came first,
from whichever peer it joined through; the first epoch's ID is all
zeros, so nothing pinned it. A remake's epoch has its ID, the snapshot
change's hash, which the rules hold its root to.

* Every node names its epoch's root in HELLO and HELLO_REPLY (the
  epoch's ID for a remake's, its genesis change for the first; wire
  version 45). A node whose document starts from another root of the
  same epoch refuses the handshake, logging it: another history under
  the swarm's ID, a forged root or a genesis started again.
* A joining node with an empty document takes as its root only the one
  it was given (--genesis-root, new), else the one its peers name, and
  none while they name different ones (access::screen::RootPin, kept
  in Screening across documents): one peer hands it another root only
  if it is all the joiner hears, or never, with --genesis-root. A root
  refused is listed with the refused, what builds on it with it.
* The genesis node logs the root (`The swarm's root`); GET
  /admin/epochs shows it (`root`).

A genesis started again under a remembered swarm ID used to merge with
the nodes of it still running; they now refuse it (another root), and
the docs say to join them with seeds instead.

The epoch's identity stays as it is (generation 0 all zeros): records
and epoch metadata name it so, and every comparison of epochs with
them.

5 files changed, 197 insertions, 3 deletions

diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index 98e273a..4384edb 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -70,13 +70,14 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start)
         return Ok(());
     }
     if start == Start::Genesis {
-        // A remembered swarm starts again under its ID: nodes of it still running then
-        // merge with this one instead of ending up in another swarm.
+        // A remembered swarm starts again under its ID, from a root of its own: nodes of
+        // it still running hold another, and refuse it (`swarm::roots`); joining them
+        // takes seeds instead.
         let swarm_id = match state.swarm_id().await {
             Some(remembered) => {
                 warn!(
                     swarm_id = %remembered,
-                    "--genesis with a remembered swarm: starting it again, empty; nodes of it still running merge with this one"
+                    "--genesis with a remembered swarm: starting it again, empty, from a new root; nodes of it still running refuse it (another root): give seeds to join them instead"
                 );
                 remembered
             }
@@ -131,6 +132,9 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start)
             .context(
                 "Recording the swarm ID, its neighbours, its topology and dissemination, its first users",
             )?;
+        if let Some(root) = crate::swarm::roots::own(state).await {
+            info!(root = %root, "The swarm's root: a node joining may be given it (--genesis-root), to take no other");
+        }
         state.initial_sync_done.store(true, Ordering::Release);
         state.mark_ready(swarm_id).await?;
         // Remembered members, if any, to find those still running.
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs
index d872474..e08bc62 100644
--- a/src/swarm/mod.rs
+++ b/src/swarm/mod.rs
@@ -17,6 +17,7 @@ pub(crate) mod plumtree;
 pub(crate) mod reach;
 pub(crate) mod refusals;
 pub(crate) mod relay;
+pub(crate) mod roots;
 pub(crate) mod rtt;
 pub(crate) mod signatures;
 pub(crate) mod skew;
diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs
index b964064..8e9b0ef 100644
--- a/src/swarm/peer/handshake.rs
+++ b/src/swarm/peer/handshake.rs
@@ -139,6 +139,9 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> {
                 ready: state.is_ready().await,
                 run_id: state.run_id,
                 handshake,
+                root: crate::swarm::roots::own(state)
+                    .await
+                    .map(|root| root.0.to_vec()),
             }),
         )
         .await;
@@ -236,6 +239,9 @@ pub(super) async fn send_hello_reply(
         run_id,
         away_ms: away.map(|away| away.as_millis() as u64),
         sender_run_id: state.run_id,
+        root: crate::swarm::roots::own(state)
+            .await
+            .map(|root| root.0.to_vec()),
     };
 
     debug!(to = %peer, ready, "Sending HELLO_REPLY");
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index a7633b3..3ddaf08 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -96,6 +96,18 @@ pub(crate) async fn handle_message(
         return Ok(());
     }
 
+    // The root its document starts from: another than ours, of our epoch, is another
+    // history; what a joiner takes as its root follows those named (`swarm::roots`).
+    match &message {
+        ParsedMessage::Hello(hello) => {
+            crate::swarm::roots::heard(state, &peer, epoch, hello.root.as_deref()).await?
+        }
+        ParsedMessage::HelloReply(reply) => {
+            crate::swarm::roots::heard(state, &peer, epoch, reply.root.as_deref()).await?
+        }
+        _ => {}
+    }
+
     remember_peer(state, &peer).await;
 
     let run = match &message {
diff --git a/src/swarm/roots.rs b/src/swarm/roots.rs
new file mode 100644
index 0000000..5089418
--- /dev/null
+++ b/src/swarm/roots.rs
@@ -0,0 +1,171 @@
+//! An epoch's root, pinned: the change without dependencies a document of that epoch
+//! starts from. For a remake's epoch it is the epoch's ID (its snapshot change's hash),
+//! which the rules hold every root to; for the first epoch, whose ID is all zeros, the
+//! genesis change, which a node joining with an empty document would take from whoever
+//! sent it first. So every node names its epoch's root in its handshakes
+//! (`HelloRequest::root`, `HelloReply::root`), and:
+//!
+//! * a node whose document holds another root of the same epoch refuses the handshake:
+//!   another history under this swarm's ID (a forged root, or a genesis started again);
+//! * a joining node takes as its root only the one given (`--genesis-root`), or else the
+//!   one its peers name, and none while they name different ones (`RootPin`), so one
+//!   peer can't hand it another root unless it is all the joiner hears.
+
+use std::collections::{HashMap, HashSet};
+
+use anyhow::{Result, bail};
+use tracing::{error, info, warn};
+
+use crate::access::screen::RootPin;
+use crate::document::replica::Epoch;
+use crate::document::{ChangeHash, Document};
+use crate::locks::MutexExt;
+use crate::state::AppState;
+
+/// What the node heard of its epoch's root, and knows of its own.
+#[derive(Debug, Default)]
+pub(crate) struct Roots {
+    /// Of the first epoch, as each peer named it.
+    named: HashMap<String, ChangeHash>,
+    /// Its own, for the document (`Replica::epoch_seq`), once known.
+    own: Option<(u32, ChangeHash)>,
+    /// Peers whose root differs, logged once each.
+    differing: HashSet<String>,
+}
+
+/// A change hash in hex (`--genesis-root`).
+pub(crate) fn parse(hex: &str) -> Option<ChangeHash> {
+    let hex = hex.trim();
+    if hex.len() != 64 {
+        return None;
+    }
+    let mut bytes = [0u8; 32];
+    for (index, byte) in bytes.iter_mut().enumerate() {
+        *byte = u8::from_str_radix(hex.get(index * 2..index * 2 + 2)?, 16).ok()?;
+    }
+    Some(ChangeHash(bytes))
+}
+
+/// `doc`'s root, of `epoch`: the epoch's ID for a remake's, the change without
+/// dependencies of the first epoch's; None for an empty document.
+pub(crate) fn of(doc: &Document, epoch: Epoch) -> Option<ChangeHash> {
+    if epoch.generation > 0 {
+        return Some(ChangeHash(epoch.id));
+    }
+    if doc.get_heads().is_empty() {
+        return None;
+    }
+    doc.get_changes(&[])
+        .iter()
+        .find(|change| change.deps().is_empty())
+        .map(|change| change.hash())
+}
+
+/// This node's epoch's root, if its document holds one; found once per document.
+pub(crate) async fn own(state: &AppState) -> Option<ChangeHash> {
+    let (epoch, epoch_seq) = state.epoch_stamp();
+    if epoch.generation > 0 {
+        return Some(ChangeHash(epoch.id));
+    }
+    if let Some((seq, root)) = state.roots.lock_or_recover().own
+        && seq == epoch_seq
+    {
+        return Some(root);
+    }
+    let replica = state.doc.read().await;
+    let root = of(&replica.doc, replica.epoch)?;
+    state.roots.lock_or_recover().own = Some((replica.epoch_seq, root));
+    Some(root)
+}
+
+/// `peer`, in a handshake of `epoch`, names `root` as its root. Refused (an error) if
+/// this node's document holds another of the same epoch; of the first epoch, what an
+/// empty document takes as its root follows (`pin`).
+pub(crate) async fn heard(
+    state: &AppState,
+    peer: &str,
+    epoch: Epoch,
+    root: Option<&[u8]>,
+) -> Result<()> {
+    let Some(root) = root.and_then(|bytes| <[u8; 32]>::try_from(bytes).ok()) else {
+        return Ok(());
+    };
+    let root = ChangeHash(root);
+    if epoch != state.epoch_stamp().0 {
+        // Another epoch: catching up is its business.
+        return Ok(());
+    }
+    if let Some(ours) = own(state).await
+        && ours != root
+    {
+        if state
+            .roots
+            .lock_or_recover()
+            .differing
+            .insert(peer.to_owned())
+        {
+            error!(peer = %peer, ours = %ours, theirs = %root, epoch = %epoch, "A peer's document starts from another root of this epoch: another history under this swarm's ID (a forged root, or a genesis started again); not synced with");
+        }
+        bail!("{peer} starts epoch {epoch} from root {root}, ours from {ours}");
+    }
+    if epoch.generation == 0 {
+        let pin = {
+            let mut roots = state.roots.lock_or_recover();
+            roots.named.insert(peer.to_owned(), root);
+            pin(state.genesis_root, &roots.named)
+        };
+        let mut screening = state.screening.lock_or_recover();
+        if screening.root_pin != pin {
+            match &pin {
+                RootPin::Disputed => {
+                    warn!(
+                        "Peers name different roots of the swarm's first epoch: taking none until they agree, or one is given (--genesis-root)"
+                    );
+                }
+                RootPin::Only(root) => {
+                    info!(root = %root, "The swarm's root, as given or as its peers name it")
+                }
+                RootPin::Any => {}
+            }
+            screening.root_pin = pin;
+        }
+    }
+    Ok(())
+}
+
+/// What an empty document of the first epoch takes as its root: the one `given`, else
+/// the one all peers name (`named`), none while they differ, any while none is named.
+pub(crate) fn pin(given: Option<ChangeHash>, named: &HashMap<String, ChangeHash>) -> RootPin {
+    if let Some(given) = given {
+        return RootPin::Only(given);
+    }
+    let mut roots: Vec<&ChangeHash> = named.values().collect();
+    roots.sort_unstable();
+    roots.dedup();
+    match roots.as_slice() {
+        [] => RootPin::Any,
+        [root] => RootPin::Only(**root),
+        _ => RootPin::Disputed,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// The root given wins; else the one every peer names; none while they differ.
+    #[test]
+    fn what_a_joiner_takes_as_its_root() {
+        let (a, b) = (ChangeHash([1; 32]), ChangeHash([2; 32]));
+        let named = |roots: &[(&str, ChangeHash)]| -> HashMap<String, ChangeHash> {
+            roots
+                .iter()
+                .map(|(peer, root)| ((*peer).to_owned(), *root))
+                .collect()
+        };
+        assert_eq!(pin(None, &named(&[])), RootPin::Any);
+        assert_eq!(pin(None, &named(&[("p", a), ("q", a)])), RootPin::Only(a));
+        assert_eq!(pin(None, &named(&[("p", a), ("q", b)])), RootPin::Disputed);
+        assert_eq!(pin(Some(b), &named(&[("p", a)])), RootPin::Only(b));
+    }
+}