roots: a joiner takes a root once its seeds named it

Review 3, R2: the root pin was set by the first HELLO_REPLY, and the
first sync, a round trip later, took that root; a peer naming another
afterwards was refused as another history. So the seed that answered
first decided, not "none while they disagree", and a malicious seed
makes itself fastest.

A node joining with an empty document of the first epoch now takes no
root (RootPin::Awaiting, roots left for later) until each of its seeds
named one, or ROOT_WAIT (10 s) passed since the first did; then the
one they name, none while they differ, as before. With --genesis-root
it takes that one alone, at once. access.md, the option's help and
example.toml say so; switching roots later, while the joiner holds
nothing of its own, is postponed.

1 files changed, 19 insertions, 1 deletions

diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index 4384edb..6d3a4ad 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -150,6 +150,14 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start)
         info!(seeds = seeds.len(), "Joining existing swarm");
     }
 
+    // An empty document of the first epoch takes a root once its seeds named it, or a
+    // while after the first did (`swarm::roots`).
+    {
+        let replica = state.doc.read().await;
+        if replica.epoch.generation == 0 && replica.doc.get_heads().is_empty() {
+            crate::swarm::roots::await_seeds(state, seeds.len());
+        }
+    }
     let mut unresolved: Vec<String> = seeds.to_vec();
     let mut failing = HashSet::new();
     let mut resolve_retry = RESOLVE_RETRY;
@@ -191,7 +199,10 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start)
             next_report = now + BOOTSTRAP_REPORT;
             report_joining(state, seeds, &unresolved).await;
         }
-        // Until something changes (`join_progress`), or it's time to resolve or report.
+        // The wait for the seeds to name the swarm's root, over by now?
+        crate::swarm::roots::repin(state);
+        // Until something changes (`join_progress`), or it's time to resolve, report or
+        // stop waiting for the seeds.
         let resolving = async {
             if unresolved.is_empty() {
                 std::future::pending().await
@@ -199,9 +210,16 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start)
                 tokio::time::sleep_until(next_resolve).await
             }
         };
+        let root_wait = async {
+            match crate::swarm::roots::waiting_until(state) {
+                Some(until) => tokio::time::sleep_until(until).await,
+                None => std::future::pending().await,
+            }
+        };
         tokio::select! {
             () = state.join_progress.notified() => {}
             () = resolving => {}
+            () = root_wait => {}
             () = tokio::time::sleep_until(next_report) => {}
         }
     }