diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/bootstrap.rs | 20 | ||||
| -rw-r--r-- | src/swarm/roots.rs | 130 |
2 files changed, 121 insertions, 29 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) => {} } } diff --git a/src/swarm/roots.rs b/src/swarm/roots.rs index 5089418..9b17218 100644 --- a/src/swarm/roots.rs +++ b/src/swarm/roots.rs @@ -8,10 +8,13 @@ //! * 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. +//! one its peers name, and none while they name different ones (`RootPin`); and none +//! before each of its seeds named one, or `ROOT_WAIT` passed since the first did, so +//! the seed that answers first doesn't decide alone. One peer hands a joiner another +//! root only if it is all the joiner hears within the wait. use std::collections::{HashMap, HashSet}; +use std::time::Duration; use anyhow::{Result, bail}; use tracing::{error, info, warn}; @@ -22,11 +25,18 @@ use crate::document::{ChangeHash, Document}; use crate::locks::MutexExt; use crate::state::AppState; +/// How long a joining node waits, after the first peer named the swarm's root, for its +/// other seeds to name theirs, before it takes the one named. +pub(crate) const ROOT_WAIT: Duration = Duration::from_secs(10); + /// 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>, + /// Joining with an empty document: how many seeds it waits for, and until when once + /// the first named a root. + awaiting: Option<(usize, Option<tokio::time::Instant>)>, /// Its own, for the document (`Replica::epoch_seq`), once known. own: Option<(u32, ChangeHash)>, /// Peers whose root differs, logged once each. @@ -109,36 +119,92 @@ pub(crate) async fn heard( 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") + state + .roots + .lock_or_recover() + .named + .insert(peer.to_owned(), root); + repin(state); + } + Ok(()) +} + +/// A node joining with an empty document through `seeds` seeds: it takes no root until +/// each named one, or `ROOT_WAIT` passed since the first did (`repin`). Not with a root +/// given, which it takes alone. +pub(crate) fn await_seeds(state: &AppState, seeds: usize) { + if state.genesis_root.is_some() || seeds == 0 { + return; + } + state.roots.lock_or_recover().awaiting = Some((seeds, None)); + repin(state); +} + +/// Until when a joining node still waits for its seeds to name the swarm's root, if it +/// waits on a clock (`await_seeds`): then `repin`. +pub(crate) fn waiting_until(state: &AppState) -> Option<tokio::time::Instant> { + state + .roots + .lock_or_recover() + .awaiting + .and_then(|(_, until)| until) +} + +/// What an empty document of the first epoch takes as its root, from what the node was +/// given and heard, into `Screening`: logged when it changes. +pub(crate) fn repin(state: &AppState) { + let now = tokio::time::Instant::now(); + let pin = { + let mut roots = state.roots.lock_or_recover(); + let named = roots.named.len(); + let awaiting = match &mut roots.awaiting { + Some((seeds, until)) => { + if until.is_none() && named > 0 { + *until = Some(now + ROOT_WAIT); } - RootPin::Any => {} + named < *seeds && until.is_none_or(|until| now < until) } - screening.root_pin = pin; + None => false, + }; + if !awaiting { + roots.awaiting = None; } + pin(state.genesis_root, &roots.named, awaiting) + }; + 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::Awaiting => { + info!(wait = ?ROOT_WAIT, "Taking the swarm's root once every seed named it, or a while after the first did") + } + 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 { +/// What an empty document of the first epoch takes as its root: the one `given`, else, +/// while `awaiting` its seeds, none; 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>, + awaiting: bool, +) -> RootPin { if let Some(given) = given { return RootPin::Only(given); } + if awaiting { + return RootPin::Awaiting; + } let mut roots: Vec<&ChangeHash> = named.values().collect(); roots.sort_unstable(); roots.dedup(); @@ -153,7 +219,8 @@ pub(crate) fn pin(given: Option<ChangeHash>, named: &HashMap<String, ChangeHash> mod tests { use super::*; - /// The root given wins; else the one every peer names; none while they differ. + /// The root given wins; else none while waiting for the seeds; 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])); @@ -163,9 +230,16 @@ mod tests { .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)); + assert_eq!(pin(None, &named(&[]), false), RootPin::Any); + assert_eq!(pin(None, &named(&[("p", a)]), true), RootPin::Awaiting); + assert_eq!( + pin(None, &named(&[("p", a), ("q", a)]), false), + RootPin::Only(a) + ); + assert_eq!( + pin(None, &named(&[("p", a), ("q", b)]), false), + RootPin::Disputed + ); + assert_eq!(pin(Some(b), &named(&[("p", a)]), true), RootPin::Only(b)); } } |