aboutsummaryrefslogtreecommitdiffstats
path: root/src/swarm
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm')
-rw-r--r--src/swarm/auth.rs53
-rw-r--r--src/swarm/blob.rs166
-rw-r--r--src/swarm/bootstrap.rs35
-rw-r--r--src/swarm/changes.rs79
-rw-r--r--src/swarm/document_file.rs110
-rw-r--r--src/swarm/links.rs453
-rw-r--r--src/swarm/members.rs175
-rw-r--r--src/swarm/mod.rs4
-rw-r--r--src/swarm/peer/admit.rs3
-rw-r--r--src/swarm/peer/epoch.rs8
-rw-r--r--src/swarm/peer/forget.rs11
-rw-r--r--src/swarm/peer/handshake.rs94
-rw-r--r--src/swarm/peer/inbound.rs48
-rw-r--r--src/swarm/peer/keepalive.rs86
-rw-r--r--src/swarm/peer/mod.rs7
-rw-r--r--src/swarm/peer/worker.rs107
-rw-r--r--src/swarm/relay.rs18
-rw-r--r--src/swarm/skew.rs277
-rw-r--r--src/swarm/source.rs33
-rw-r--r--src/swarm/state_file.rs2
-rw-r--r--src/swarm/stats.rs4
-rw-r--r--src/swarm/sync.rs203
-rw-r--r--src/swarm/topology.rs843
-rw-r--r--src/swarm/topology/graph.rs236
-rw-r--r--src/swarm/topology/mod.rs377
-rw-r--r--src/swarm/topology/tests.rs313
26 files changed, 2429 insertions, 1316 deletions
diff --git a/src/swarm/auth.rs b/src/swarm/auth.rs
index 088f615..98a004c 100644
--- a/src/swarm/auth.rs
+++ b/src/swarm/auth.rs
@@ -21,7 +21,7 @@ use std::sync::Mutex;
use std::time::{Duration, Instant};
use crate::locks::MutexExt;
-use anyhow::{Context, Result, bail};
+use anyhow::{Result, bail};
use sha2::{Digest, Sha256};
const HELLO_LABEL: &[u8] = b"continuum/hello/1";
@@ -40,37 +40,25 @@ struct Rejections {
suppressed: u64,
}
-/// A key is a secret: a file others may read is worth a warning (not an error: a
-/// group may be deliberate, e.g. shared with a deployment tool). `what`: which key.
-pub(crate) fn warn_if_readable_by_others(path: &Path, what: &str) {
- #[cfg(unix)]
- {
- use std::os::unix::fs::PermissionsExt;
- if let Ok(metadata) = std::fs::metadata(path) {
- let mode = metadata.permissions().mode();
- if mode & 0o077 != 0 {
- tracing::warn!(
- path = %path.display(),
- mode = format!("{:o}", mode & 0o777),
- "The {what} file is readable by others than its owner; chmod 600 it"
- );
- }
- }
- }
- #[cfg(not(unix))]
- let _ = (path, what);
-}
-
/// At most one rejection log line per this interval.
const REJECTION_LOG_INTERVAL: Duration = Duration::from_secs(10);
impl SwarmKey {
/// Reads the key from a file; its bytes are the key, as they are. Generate one with
/// e.g. `head -c 32 /dev/urandom > swarm.key`.
- pub(crate) fn from_file(path: &Path) -> Result<Self> {
- let key = std::fs::read(path)
- .with_context(|| format!("Reading the swarm key from {}", path.display()))?;
- warn_if_readable_by_others(path, "swarm key");
+ /// With `create`, a new one (32 random bytes) if there is no file at `path`.
+ pub(crate) fn from_file(path: &Path, create: bool) -> Result<Self> {
+ let (key, created) = crate::files::key_file(
+ path,
+ "swarm key",
+ create,
+ "swarm-key-create",
+ "on the swarm's first node; the others need a copy of it",
+ 32,
+ )?;
+ if created {
+ tracing::info!(path = %path.display(), "Created a new swarm key: copy it to the swarm's other nodes");
+ }
if key.len() < 16 {
bail!(
"The swarm key in {} has {} bytes; use at least 16 (32 random bytes recommended)",
@@ -172,7 +160,18 @@ mod tests {
use super::*;
fn hex(bytes: &[u8]) -> String {
- bytes.iter().map(|byte| format!("{byte:02x}")).collect()
+ crate::transport::Hex(bytes).to_string()
+ }
+
+ /// A swarm key file is made only when asked, then read as any other.
+ #[test]
+ fn swarm_key_created_only_when_asked() {
+ let path = crate::files::scratch_dir("swarm-key-create").join("swarm.key");
+ assert!(SwarmKey::from_file(&path, false).is_err());
+ assert!(!path.exists());
+ let created = SwarmKey::from_file(&path, true).unwrap();
+ assert_eq!(created.key.len(), 32);
+ assert_eq!(SwarmKey::from_file(&path, false).unwrap().key, created.key);
}
#[test]
diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index ec99c93..ce9a2bd 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -27,12 +27,12 @@ use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Duration;
-use anyhow::{Result, anyhow, bail};
-use automerge::{AutoCommit, ChangeHash};
+use anyhow::{Context, Result, bail};
use tokio::sync::{Semaphore, mpsc};
use tokio::time::Instant;
use tracing::{debug, info, warn};
+use crate::document::{ChangeHash, changes};
use crate::locks::MutexExt;
use crate::state::AppState;
use crate::transport::{BlobBody, BlobId, BlobMessage, DeltaSpec, ParsedMessage};
@@ -87,6 +87,10 @@ pub(crate) struct Blobs {
/// lost message costs too much to send again whole, and a slow link may not deliver it
/// within the transport's delivery limit.
const DELTA_THRESHOLD_CHUNKS: usize = 16;
+/// The longest a sync message should take whole at the throughput measured to its peer
+/// (`delta_threshold_for`): well within what a transport allows a message before its send
+/// fails (10 s, plus its size at 1 MiB/s, on libp2p).
+const WHOLE_WITHIN: Duration = Duration::from_secs(5);
/// The largest sync message accepted: the threshold of a sender with the largest chunks
/// it may have configured. Anything larger comes from a node that doesn't keep to it.
@@ -115,6 +119,25 @@ impl Blobs {
DELTA_THRESHOLD_CHUNKS * self.chunk_size as usize
}
+ /// [`delta_threshold`](Self::delta_threshold) for one peer: lower where a message
+ /// that large wouldn't make it whole. `rate`: the measured throughput to it, bytes
+ /// per second (`swarm::throughput`), at which a message whole should take at most
+ /// `WHOLE_WITHIN`; `failed`: the largest message to it that didn't make it lately
+ /// (`PeerState::failed_whole`), of which messages half as large go as deltas, if it
+ /// was larger than two chunks (smaller ones are lost for other reasons than their
+ /// size). Never below a chunk: a delta then is no smaller per message.
+ pub(crate) fn delta_threshold_for(&self, rate: Option<f64>, failed: Option<usize>) -> usize {
+ let chunk = self.chunk_size as usize;
+ let mut threshold = self.delta_threshold();
+ if let Some(rate) = rate.filter(|rate| rate.is_finite() && *rate > 0.0) {
+ threshold = threshold.min((rate * WHOLE_WITHIN.as_secs_f64()) as usize);
+ }
+ if let Some(failed) = failed.filter(|failed| *failed > 2 * chunk) {
+ threshold = threshold.min(failed / 2);
+ }
+ threshold.max(chunk)
+ }
+
/// Starts downloading a delta offered by `peer`; false if one from it is running.
pub(crate) fn start_delta(&self, peer: &str) -> bool {
self.deltas.lock_or_recover().insert(peer.to_owned())
@@ -216,21 +239,18 @@ async fn prepare(state: &AppState, id: BlobId, delta: Option<DeltaSpec>) -> Opti
// The spec must be the one the ID names.
let spec = delta.filter(|spec| spec.id() == delta_id)?;
let doc = Arc::clone(&state.doc);
- tokio::task::spawn_blocking(move || delta_bytes(&mut doc.blocking_write().doc, &spec))
- .await
- .ok()
- .flatten()
- .map(Arc::new)?
+ tokio::task::spawn_blocking(move || {
+ changes::delta_bytes(&doc.blocking_read().doc, &spec.from, &spec.to)
+ })
+ .await
+ .ok()
+ .flatten()
+ .map(Arc::new)?
}
BlobId::Change(hash) => {
let doc = Arc::clone(&state.doc);
tokio::task::spawn_blocking(move || {
- // get_change_by_hash needs the document mutably.
- let mut replica = doc.blocking_write();
- replica
- .doc
- .get_change_by_hash(&hash)
- .map(|change| change.raw_bytes().to_vec())
+ changes::change_bytes(&doc.blocking_read().doc, &hash)
})
.await
.ok()
@@ -586,14 +606,9 @@ pub(crate) async fn fetch_change(
state: &AppState,
hash: ChangeHash,
peers: Vec<String>,
-) -> Result<automerge::Change> {
+) -> Result<crate::document::Change> {
let bytes = fetch(state, BlobId::Change(hash), peers, None, None).await?;
- let change = automerge::Change::from_bytes(bytes)
- .map_err(|error| anyhow!("downloaded change {hash} doesn't load: {error}"))?;
- if change.hash() != hash {
- bail!("downloaded change has hash {}, not {hash}", change.hash());
- }
- Ok(change)
+ changes::change_from_bytes(bytes, hash).context("downloaded")
}
/// Downloads the changes of `spec` from `peers` (any that hold its `to`). The caller
@@ -606,120 +621,9 @@ pub(crate) async fn fetch_delta(
fetch(state, BlobId::Delta(spec.id()), peers, None, Some(spec)).await
}
-/// The bytes of delta `spec` from `doc`: the changes in the history of `spec.to` but
-/// not of `spec.from`, each as Automerge encodes it, sorted by hash and concatenated
-/// (`AutoCommit::load_incremental` reads that). The same on every node holding `to`,
-/// whatever else it has. None if `doc` lacks part of `to`.
-pub(crate) fn delta_bytes(doc: &mut AutoCommit, spec: &DeltaSpec) -> Option<Vec<u8>> {
- if spec
- .to
- .iter()
- .any(|head| doc.get_change_by_hash(head).is_none())
- {
- return None;
- }
- // Heads of `from` we don't have can't hide anything we have.
- let from: Vec<ChangeHash> = spec
- .from
- .iter()
- .copied()
- .filter(|head| doc.get_change_by_hash(head).is_some())
- .collect();
- let since: HashMap<ChangeHash, (Vec<ChangeHash>, Vec<u8>)> = doc
- .get_changes(&from)
- .iter()
- .map(|change| {
- (
- change.hash(),
- (change.deps().to_vec(), change.raw_bytes().to_vec()),
- )
- })
- .collect();
- // Of those, the ones `to` depends on: what we have beyond `to` stays out.
- let mut included: HashSet<ChangeHash> = HashSet::new();
- let mut pending: Vec<ChangeHash> = spec.to.clone();
- while let Some(hash) = pending.pop() {
- if let Some((deps, _)) = since.get(&hash)
- && included.insert(hash)
- {
- pending.extend(deps.iter().copied());
- }
- }
- let mut hashes: Vec<ChangeHash> = included.into_iter().collect();
- hashes.sort_unstable();
- let mut bytes = Vec::new();
- for hash in hashes {
- bytes.extend_from_slice(&since[&hash].1);
- }
- Some(bytes)
-}
-
#[cfg(test)]
mod tests {
use super::*;
- use automerge::{ActorId, ROOT, transaction::Transactable};
-
- fn random_write(rng: &mut crate::testrand::Rng, doc: &mut AutoCommit) {
- let key = format!("k{}", rng.below(6));
- doc.put(ROOT, key, rng.below(100) as i64).unwrap();
- doc.commit();
- }
-
- fn sorted(mut heads: Vec<ChangeHash>) -> Vec<ChangeHash> {
- heads.sort_unstable();
- heads
- }
-
- /// Over random histories: two nodes that reached the same heads through different
- /// merge orders, one with more on top, produce the same bytes for a delta to those
- /// heads; a document at `from` that loads them is at `to`.
- #[test]
- fn delta_bytes_are_the_same_everywhere_and_complete() {
- crate::testrand::for_seeds(200, |rng| {
- let actor = |i: u8| ActorId::from(&[i; 16][..]);
- let mut base = AutoCommit::new().with_actor(actor(0));
- for _ in 0..rng.between(0, 5) {
- random_write(rng, &mut base);
- }
- let from = base.get_heads();
- let mut writers: Vec<AutoCommit> = (1..=rng.between(1, 3))
- .map(|i| base.fork().with_actor(actor(i as u8)))
- .collect();
- for _ in 0..rng.between(1, 15) {
- let writer = rng.below(writers.len());
- random_write(rng, &mut writers[writer]);
- }
- let mut one = base.fork().with_actor(actor(10));
- let mut two = base.fork().with_actor(actor(11));
- for writer in writers.iter_mut() {
- one.merge(writer).unwrap();
- }
- for writer in writers.iter_mut().rev() {
- two.merge(writer).unwrap();
- }
- let to = one.get_heads();
- assert_eq!(sorted(to.clone()), sorted(two.get_heads()));
- // Beyond `to`: must stay out of the delta.
- random_write(rng, &mut two);
-
- let spec = DeltaSpec::new(from.clone(), to.clone());
- let bytes = delta_bytes(&mut one, &spec).unwrap();
- assert_eq!(delta_bytes(&mut two, &spec).unwrap(), bytes);
-
- let mut behind = base.fork();
- behind.load_incremental(&bytes).unwrap();
- assert_eq!(sorted(behind.get_heads()), sorted(to));
- });
- }
-
- #[test]
- fn no_delta_without_its_heads() {
- let mut doc = AutoCommit::new();
- doc.put(ROOT, "k", 1_i64).unwrap();
- doc.commit();
- let spec = DeltaSpec::new(Vec::new(), vec![ChangeHash([7; 32])]);
- assert!(delta_bytes(&mut doc, &spec).is_none());
- }
#[test]
fn chunking() {
diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index a9ab1b2..4f6c072 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -18,7 +18,6 @@ use std::collections::HashSet;
use std::path::Path;
use anyhow::{Context, Result, bail};
-use automerge::{ActorId, AutoCommit};
use serde_json::Value as JsonValue;
use tokio::time::Instant;
use tracing::{debug, info, warn};
@@ -28,13 +27,13 @@ use crate::document;
use crate::document::meta::{self, SWARM};
use crate::document::replica::Epoch;
use crate::document::store;
+use crate::document::{ActorId, Document};
use crate::files::{FileKind, Files};
use crate::force::Force;
use crate::remake::snapshot;
use crate::state::AppState;
use crate::swarm::peer::{remember_peer, send_hello_once};
-const BOOTSTRAP_POLL: Duration = Duration::from_millis(500);
/// How often a node that is still joining says why.
const BOOTSTRAP_REPORT: Duration = Duration::from_secs(300);
/// Retrying seeds that didn't resolve: first after this, doubling up to the maximum.
@@ -112,6 +111,7 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start)
let mut resolve_retry = RESOLVE_RETRY;
let mut next_resolve = Instant::now();
let mut next_report = Instant::now() + BOOTSTRAP_REPORT;
+ state.transport.on_seeds_found(state.join_progress.clone());
loop {
if let Some(swarm_id) = state.swarm_id().await {
// A node joining a compacted swarm rejoins the newer epoch first; it is
@@ -147,7 +147,19 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start)
next_report = now + BOOTSTRAP_REPORT;
report_joining(state, seeds, &unresolved).await;
}
- tokio::time::sleep(BOOTSTRAP_POLL).await;
+ // Until something changes (`join_progress`), or it's time to resolve or report.
+ let resolving = async {
+ if unresolved.is_empty() {
+ std::future::pending().await
+ } else {
+ tokio::time::sleep_until(next_resolve).await
+ }
+ };
+ tokio::select! {
+ () = state.join_progress.notified() => {}
+ () = resolving => {}
+ () = tokio::time::sleep_until(next_report) => {}
+ }
}
}
@@ -156,7 +168,7 @@ fn swarm_id_path() -> Vec<String> {
}
/// The swarm ID the document names, if any.
-fn read_swarm_id(doc: &AutoCommit) -> Option<String> {
+fn read_swarm_id(doc: &Document) -> Option<String> {
match store::read_path(doc, &swarm_id_path()) {
Ok(Some(JsonValue::String(id))) if !id.is_empty() => Some(id),
_ => None,
@@ -184,7 +196,7 @@ impl Origin {
/// A document the node starts from, ready at once: on the swarm and epoch it names.
pub(crate) struct StartingDocument {
/// Under a fresh actor.
- pub(crate) doc: AutoCommit,
+ pub(crate) doc: Document,
pub(crate) epoch: Epoch,
pub(crate) swarm_id: String,
pub(crate) origin: Origin,
@@ -281,7 +293,7 @@ pub(crate) fn starting_document(
/// The swarm and epoch `doc` is on.
fn identify(
- mut doc: AutoCommit,
+ doc: Document,
origin: Origin,
earlier_actors: Vec<ActorId>,
) -> Result<StartingDocument> {
@@ -289,8 +301,7 @@ fn identify(
let Some(swarm_id) = read_swarm_id(&doc) else {
bail!("{what} names no swarm (no _continuum/swarm); it predates swarm IDs in documents");
};
- let epoch =
- snapshot::epoch_of(&mut doc).map_err(|reason| anyhow::anyhow!("{what}: {reason}"))?;
+ let epoch = snapshot::epoch_of(&doc).map_err(|reason| anyhow::anyhow!("{what}: {reason}"))?;
Ok(StartingDocument {
doc,
epoch,
@@ -300,15 +311,13 @@ fn identify(
})
}
-/// A backup to recover from, as `GET /admin/document` saves it, under a fresh actor.
-fn load_backup(path: &Path, files: &Files) -> Result<AutoCommit> {
+/// A backup to recover from, as `GET /admin/document` saves it (`document::load`).
+fn load_backup(path: &Path, files: &Files) -> Result<Document> {
let bytes = files
.read(FileKind::Backup, path)
.with_context(|| format!("Reading the backup {}", path.display()))?
.with_context(|| format!("The backup {} doesn't exist", path.display()))?;
- Ok(AutoCommit::load(&bytes)
- .with_context(|| format!("Loading the backup {}", path.display()))?
- .with_actor(ActorId::random()))
+ crate::document::load(&bytes).with_context(|| format!("Loading the backup {}", path.display()))
}
/// Resolves the seeds in `unresolved`, says HELLO to each that resolves, and keeps the
diff --git a/src/swarm/changes.rs b/src/swarm/changes.rs
new file mode 100644
index 0000000..0f22f52
--- /dev/null
+++ b/src/swarm/changes.rs
@@ -0,0 +1,79 @@
+//! The change publisher: after document changes, which store paths they touched
+//! (`AppState::change_events`, for the API's streams, the dashboard and remakes), and
+//! a nudge to topology when the member list changed, and to the peer worker when the
+//! forget records did.
+
+use std::sync::Arc;
+
+use tracing::{info, warn};
+
+use crate::document::changes;
+use crate::document::store::{ChangeSet, InLists};
+use crate::state::AppState;
+
+pub(crate) async fn run_change_publisher(state: AppState) {
+ let mut changed = state.change_tx.subscribe();
+ // The replica's heads (sorted, as of its last refresh: every change refreshes them).
+ let mut last_heads = state.doc.read().await.heads().to_vec();
+
+ info!("Starting change publisher");
+
+ while changed.changed().await.is_ok() {
+ let before = last_heads.clone();
+
+ let doc = Arc::clone(&state.doc);
+ let (heads, paths, conflicts) = tokio::task::spawn_blocking(move || {
+ let replica = doc.blocking_read();
+ let heads = replica.heads().to_vec();
+ if heads == before {
+ return (heads, Vec::new(), Vec::new());
+ }
+
+ // After an epoch switch the previous heads belong to the old document:
+ // report the root as changed, which every stream is related to.
+ if !changes::missing(&replica.doc, &before).is_empty() {
+ return (heads, vec![Vec::new()], Vec::new());
+ }
+
+ let mut paths = Vec::new();
+ let mut conflicts = Vec::new();
+ for changed in replica.changed_since(&before, InLists::Index) {
+ if changed.conflict {
+ conflicts.push(changed.path.clone());
+ }
+ paths.push(changed.path);
+ }
+ (heads, paths, conflicts)
+ })
+ .await
+ .expect("change publisher task panicked");
+
+ if heads == last_heads {
+ continue;
+ }
+ last_heads = heads;
+
+ for path in &conflicts {
+ warn!(path = ?path, "Concurrent writes produced a conflict");
+ }
+
+ if paths
+ .iter()
+ .any(|path| crate::swarm::topology::affects_neighbours(path))
+ {
+ state.topology_changed.notify_one();
+ }
+ if paths
+ .iter()
+ .any(|path| crate::swarm::peer::affects_forget_records(path))
+ {
+ state
+ .forget_records_changed
+ .store(true, std::sync::atomic::Ordering::Release);
+ state.peers_due.now();
+ }
+
+ // Fails only when nobody is subscribed.
+ let _ = state.change_events.send(Arc::new(ChangeSet { paths }));
+ }
+}
diff --git a/src/swarm/document_file.rs b/src/swarm/document_file.rs
new file mode 100644
index 0000000..e9deca0
--- /dev/null
+++ b/src/swarm/document_file.rs
@@ -0,0 +1,110 @@
+//! The worker that keeps the document file (`document::file`) up to date: it looks at
+//! the node (ready, the document's epoch and heads, this node's actors), and the file
+//! says what is due. While a ready node rejoins, the file follows the document before
+//! the rejoin, which takes the node's writes meanwhile (`state::StaleView`): a restart
+//! then resumes on it, and rejoins again.
+
+use std::sync::Arc;
+
+use crate::document::Document;
+use crate::document::file::Due;
+use crate::document::replica::Epoch;
+use crate::locks::RwLockExt;
+use crate::state::AppState;
+
+/// Keeps the document file up to date. Runs for as long as the node does.
+pub(crate) async fn run(state: AppState) {
+ let Some(file) = state.document_file.clone() else {
+ return;
+ };
+ let mut tick = tokio::time::interval(file.append_every());
+ tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+ loop {
+ tick.tick().await;
+ flush(&state).await;
+ }
+}
+
+/// Writes what is due now (`DocumentFile::due`). Also called at shutdown.
+pub(crate) async fn flush(state: &AppState) {
+ let Some(file) = state.document_file.clone() else {
+ return;
+ };
+ if state.remake.is_rejoining() {
+ if let Some((epoch, stale)) = state.stale_document() {
+ flush_before_rejoin(&file, epoch, stale).await;
+ }
+ return;
+ }
+ if !state.is_ready().await {
+ return;
+ }
+ let own = state.remake.own_actors();
+ let (epoch, heads) = {
+ let replica = state.doc.read().await;
+ (replica.epoch, replica.heads().to_vec())
+ };
+ match file.due(epoch, own.len(), &heads) {
+ Due::Rewrite(reason) => {
+ let (epoch, copy) = {
+ let replica = state.doc.read().await;
+ (replica.epoch, replica.doc.clone())
+ };
+ file.rewrite(epoch, copy, own, reason).await;
+ }
+ Due::Append { epoch, since } => {
+ let (heads, increment) = {
+ let replica = state.doc.read().await;
+ // Switched since: the next flush rewrites the file instead.
+ if replica.epoch != epoch {
+ return;
+ }
+ let heads = replica.heads().to_vec();
+ (heads, crate::document::save_since(&replica.doc, &since))
+ };
+ file.append(since, heads, increment).await;
+ }
+ Due::Nothing => {}
+ }
+}
+
+/// [`flush`] of the document before a rejoin (`epoch`'s), while it takes this node's
+/// writes. Only appends: a rewrite (another epoch, or the increments outgrew the full
+/// save) waits for the end of the rejoin, so the file keeps the actors it had.
+async fn flush_before_rejoin(
+ file: &Arc<crate::document::file::DocumentFile>,
+ epoch: Epoch,
+ stale: Arc<std::sync::RwLock<Document>>,
+) {
+ let heads = {
+ let stale = Arc::clone(&stale);
+ // Sorted, as the live document's (`Replica::heads`) the file compares with.
+ tokio::task::spawn_blocking(move || {
+ let mut heads = stale.read_or_recover().get_heads();
+ heads.sort_unstable();
+ heads
+ })
+ .await
+ .expect("document read task panicked")
+ };
+ let Some(actors) = file.actors() else {
+ return;
+ };
+ if let Due::Append {
+ epoch: written,
+ since,
+ } = file.due(epoch, actors, &heads)
+ && written == epoch
+ {
+ let since_then = since.clone();
+ let (heads, increment) = tokio::task::spawn_blocking(move || {
+ let doc = stale.read_or_recover();
+ let mut heads = doc.get_heads();
+ heads.sort_unstable();
+ (heads, crate::document::save_since(&doc, &since_then))
+ })
+ .await
+ .expect("document read task panicked");
+ file.append(since, heads, increment).await;
+ }
+}
diff --git a/src/swarm/links.rs b/src/swarm/links.rs
new file mode 100644
index 0000000..b84f57a
--- /dev/null
+++ b/src/swarm/links.rs
@@ -0,0 +1,453 @@
+//! How members are linked, for a map of the swarm (the dashboard, `api::dashboard`): a
+//! node asks a member with LINKS, and the member answers with a LINKS_REPLY listing its
+//! neighbours and the peers it holds dormant (`own_links`), each with the transport its
+//! messages go over, whether only through a relay, the round trip and whether it is
+//! dormant: as the member measures them, which no other node can.
+//!
+//! Every node answers, whatever its build: one with a dashboard asks any member, which
+//! may have none. Only members (in the member list) are answered, as the links name
+//! members and where they're reached; and each at most every `ANSWER_EVERY`. Nothing
+//! is written to the document: asked for while someone looks, forgotten after.
+//!
+//! A LINKS may subscribe (`subscribe_secs`): the member then sends a LINKS_REPLY again
+//! whenever its links change (`changed`: a peer, a transport, relayed, dormant, or a
+//! round trip that moved by a fifth), looked at every `WATCH_EVERY` while anyone
+//! subscribes (round trips change without telling), until the subscription runs out
+//! (the asker renews it with its next LINKS, so one that is gone stops it by not
+//! renewing) or a LINKS with 0 ends it.
+
+use std::collections::HashMap;
+use std::sync::Mutex;
+use std::time::Duration;
+
+use tokio::time::Instant;
+use tracing::debug;
+
+use crate::locks::MutexExt;
+use crate::state::AppState;
+use crate::swarm::{members, topology};
+use crate::transport::paths::{self, ClockOffset};
+use crate::transport::{
+ Connection, Destination, Link, LinksReply, LinksRequest, OtherPath, ParsedMessage,
+};
+
+/// Links one reply lists at most (a member has about `--neighbours`, more if others
+/// prefer it).
+pub(crate) const MAX_LINKS: usize = 256;
+/// Other transports in a link at most.
+pub(crate) const MAX_OTHERS: usize = 8;
+/// A member answers each other one at most this often.
+const ANSWER_EVERY: Duration = Duration::from_secs(5);
+/// Replies kept from members at most (one each), and requests out.
+const MAX_KEPT: usize = 4096;
+/// Subscriptions a member keeps at most, and how long one lasts at most.
+const MAX_SUBSCRIBERS: usize = 64;
+const MAX_SUBSCRIPTION: Duration = Duration::from_secs(300);
+/// How often a member with subscribers looks whether its links changed.
+const WATCH_EVERY: Duration = Duration::from_secs(2);
+/// A reply to a LINKS that doesn't subscribe is taken for this long.
+const REPLY_WITHIN: Duration = Duration::from_secs(30);
+
+/// The requests out and the replies in, and when each asker was last answered.
+#[derive(Default)]
+pub(crate) struct Links {
+ inner: Mutex<Inner>,
+ /// Someone subscribed (`answer`): `watch` looks again.
+ subscribed: tokio::sync::Notify,
+}
+
+#[derive(Default)]
+struct Inner {
+ /// Our LINKS out, by member: its nonce, and until when replies to it are taken (its
+ /// subscription's end, or `REPLY_WITHIN`).
+ asked: HashMap<String, (u64, Instant)>,
+ /// The latest LINKS_REPLY of each member: when, and its links.
+ replies: HashMap<String, (Instant, Vec<Link>)>,
+ /// When we last answered each member.
+ answered: HashMap<String, Instant>,
+ /// Who subscribed to our links: the nonce to answer with, and until when.
+ subscribers: HashMap<String, (u64, Instant)>,
+ /// The links subscribers were last sent (`watch`).
+ published: Option<Vec<Link>>,
+}
+
+impl Links {
+ /// The latest links each member replied with, and how long ago.
+ #[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
+ pub(crate) fn replies(&self) -> Vec<(String, Duration, Vec<Link>)> {
+ let now = Instant::now();
+ self.inner
+ .lock_or_recover()
+ .replies
+ .iter()
+ .map(|(member, (at, links))| {
+ (
+ member.clone(),
+ now.saturating_duration_since(*at),
+ links.clone(),
+ )
+ })
+ .collect()
+ }
+
+ /// The members our subscriptions are with (not run out).
+ #[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
+ pub(crate) fn subscribed(&self) -> Vec<String> {
+ let now = Instant::now();
+ self.inner
+ .lock_or_recover()
+ .asked
+ .iter()
+ .filter(|(_, (_, until))| now < *until)
+ .map(|(member, _)| member.clone())
+ .collect()
+ }
+}
+
+/// This node's links: its neighbours, and the peers it holds dormant, as it measures
+/// them. A neighbour that goes dormant is one no more (it is unreachable now), but its
+/// link stays listed, dormant, until it is heard from again: left out, the link would
+/// show only as that peer last reported it, from before.
+pub(crate) async fn own_links(state: &AppState) -> Vec<Link> {
+ let mut links = Vec::new();
+ for (peer, peer_state) in state.peers_snapshot().await {
+ let dormant = peer_state.lock().await.liveness.is_dormant();
+ if !dormant && !topology::is_neighbour(state, &peer) {
+ continue;
+ }
+ let destination = state.transport.destination(&peer);
+ let transport = destination
+ .as_ref()
+ .and_then(|destination| scheme_of(&destination.address));
+ links.push(Link {
+ others: others(destination.as_ref(), transport.as_deref()),
+ transport,
+ relayed: destination
+ .as_ref()
+ .is_some_and(|destination| destination.connection == Some(Connection::Relayed)),
+ // Ranked up before down: the first is down only if all are.
+ down: destination
+ .as_ref()
+ .and_then(|destination| destination.paths.first())
+ .is_some_and(|path| path.down),
+ // The path in use's (`Transport::rtt`, the first of `destination.paths`).
+ rtt_us: state
+ .transport
+ .rtt(&peer)
+ .map(|rtt| u32::try_from(rtt.as_micros()).unwrap_or(u32::MAX)),
+ dormant,
+ clock: state.transport.clock_offset(&peer),
+ peer,
+ });
+ }
+ links.sort_by(|a, b| a.peer.cmp(&b.peer));
+ links.truncate(MAX_LINKS);
+ links
+}
+
+/// The scheme of `address` (`mycelium:…`): its transport (`paths::scheme_of`), if it
+/// has one.
+fn scheme_of(address: &str) -> Option<String> {
+ Some(paths::scheme_of(address))
+ .filter(|scheme| !scheme.is_empty())
+ .map(str::to_owned)
+}
+
+/// The transports besides `used` the peer proved itself at, with a path up: each with
+/// the round trip of its fastest path, if any is measured, and whether all of them go
+/// through a relay.
+fn others(destination: Option<&Destination>, used: Option<&str>) -> Vec<OtherPath> {
+ let mut others: Vec<OtherPath> = Vec::new();
+ for path in destination.map_or(&[][..], |destination| &destination.paths[..]) {
+ let Some(transport) = scheme_of(&path.address) else {
+ continue;
+ };
+ if path.down || used == Some(transport.as_str()) {
+ continue;
+ }
+ let rtt_us = path
+ .rtt_ms
+ .filter(|ms| ms.is_finite() && *ms >= 0.0)
+ .map(|ms| (ms * 1000.0).min(f64::from(u32::MAX)) as u32);
+ let relayed = path.connection == Some(Connection::Relayed);
+ match others.iter_mut().find(|other| other.transport == transport) {
+ Some(other) => {
+ other.rtt_us = match (other.rtt_us, rtt_us) {
+ (Some(a), Some(b)) => Some(a.min(b)),
+ (a, b) => a.or(b),
+ };
+ other.relayed &= relayed;
+ }
+ None => others.push(OtherPath {
+ transport,
+ rtt_us,
+ relayed,
+ }),
+ }
+ }
+ others.sort_by(|a, b| a.transport.cmp(&b.transport));
+ others.truncate(MAX_OTHERS);
+ others
+}
+
+/// Asks `member` for its links (a LINKS), and with `subscribe` for its changes for
+/// that long; its replies land in `AppState::links`.
+#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
+pub(crate) async fn ask(state: &AppState, member: &str, subscribe: Duration) {
+ let nonce = crate::transport::paths::random_nonce();
+ {
+ let mut inner = state.links.inner.lock_or_recover();
+ if inner.asked.len() >= MAX_KEPT && !inner.asked.contains_key(member) {
+ inner.asked.clear();
+ }
+ let until = Instant::now() + subscribe.max(REPLY_WITHIN);
+ inner.asked.insert(member.to_owned(), (nonce, until));
+ }
+ let (epoch, epoch_seq) = state.epoch_stamp();
+ let request = ParsedMessage::Links(LinksRequest {
+ epoch,
+ epoch_seq,
+ nonce,
+ subscribe_secs: u32::try_from(subscribe.as_secs()).unwrap_or(u32::MAX),
+ });
+ if let Err(error) = state.transport.send_message(member, &request).await {
+ debug!(member = %member, error = %format!("{error:#}"), "LINKS failed");
+ }
+}
+
+/// Ends our subscription with `member` (a LINKS with 0); its replies aren't taken any
+/// more.
+#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
+pub(crate) async fn unsubscribe(state: &AppState, member: &str) {
+ let nonce = {
+ let mut inner = state.links.inner.lock_or_recover();
+ match inner.asked.remove(member) {
+ Some((nonce, _)) => nonce,
+ None => return,
+ }
+ };
+ let (epoch, epoch_seq) = state.epoch_stamp();
+ let request = ParsedMessage::Links(LinksRequest {
+ epoch,
+ epoch_seq,
+ nonce,
+ subscribe_secs: 0,
+ });
+ if let Err(error) = state.transport.send_message(member, &request).await {
+ debug!(member = %member, error = %format!("{error:#}"), "LINKS failed");
+ }
+}
+
+/// A LINKS from `peer`: answered if it's a member, and not answered lately.
+pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest) {
+ let member = {
+ let replica = state.doc.read().await;
+ members::members(&replica.doc)
+ .iter()
+ .any(|member| member == peer)
+ };
+ if !member {
+ debug!(peer = %peer, "Not answering LINKS from a peer that isn't a member");
+ return;
+ }
+ {
+ let now = Instant::now();
+ let mut inner = state.links.inner.lock_or_recover();
+ // A subscription is taken, renewed or ended even when the reply is due later.
+ if request.subscribe_secs == 0 {
+ inner.subscribers.remove(peer);
+ } else if inner.subscribers.len() < MAX_SUBSCRIBERS || inner.subscribers.contains_key(peer)
+ {
+ let lasts =
+ Duration::from_secs(u64::from(request.subscribe_secs)).min(MAX_SUBSCRIPTION);
+ inner
+ .subscribers
+ .insert(peer.to_owned(), (request.nonce, now + lasts));
+ state.links.subscribed.notify_one();
+ }
+ if inner
+ .answered
+ .get(peer)
+ .is_some_and(|at| now.saturating_duration_since(*at) < ANSWER_EVERY)
+ {
+ return;
+ }
+ if inner.answered.len() >= MAX_KEPT {
+ inner
+ .answered
+ .retain(|_, at| now.saturating_duration_since(*at) < ANSWER_EVERY);
+ }
+ inner.answered.insert(peer.to_owned(), now);
+ }
+ let (epoch, epoch_seq) = state.epoch_stamp();
+ let reply = ParsedMessage::LinksReply(LinksReply {
+ epoch,
+ epoch_seq,
+ nonce: request.nonce,
+ links: own_links(state).await,
+ });
+ if let Err(error) = state.transport.send_message(peer, &reply).await {
+ debug!(peer = %peer, error = %format!("{error:#}"), "LINKS_REPLY failed");
+ }
+}
+
+/// A LINKS_REPLY from `peer`: kept if it answers our latest LINKS to it, while its
+/// replies are taken (a subscription's go on).
+pub(crate) fn received(state: &AppState, peer: &str, mut reply: LinksReply) {
+ let mut inner = state.links.inner.lock_or_recover();
+ let taken = inner
+ .asked
+ .get(peer)
+ .is_some_and(|(nonce, until)| *nonce == reply.nonce && Instant::now() < *until);
+ if !taken {
+ return;
+ }
+ reply.links.truncate(MAX_LINKS);
+ for link in &mut reply.links {
+ link.others.truncate(MAX_OTHERS);
+ }
+ if inner.replies.len() >= MAX_KEPT && !inner.replies.contains_key(peer) {
+ inner.replies.clear();
+ }
+ inner
+ .replies
+ .insert(peer.to_owned(), (Instant::now(), reply.links));
+}
+
+/// For subscribers (`answer`): every `WATCH_EVERY`, while there are any, this node's
+/// links; sent to each of them when they changed since they were last sent. Without
+/// any, it waits for one. Runs for the node's life.
+pub(crate) async fn watch(state: AppState) {
+ let mut every = tokio::time::interval(WATCH_EVERY);
+ every.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+ loop {
+ every.tick().await;
+ let subscribers: Vec<(String, u64)> = {
+ let now = Instant::now();
+ let mut inner = state.links.inner.lock_or_recover();
+ inner.subscribers.retain(|_, (_, until)| now < *until);
+ if inner.subscribers.is_empty() {
+ inner.published = None;
+ }
+ inner
+ .subscribers
+ .iter()
+ .map(|(peer, (nonce, _))| (peer.clone(), *nonce))
+ .collect()
+ };
+ if subscribers.is_empty() {
+ // A new subscriber was just answered (`answer`, or within `ANSWER_EVERY`):
+ // the next look is due an interval later.
+ state.links.subscribed.notified().await;
+ every.reset();
+ continue;
+ }
+ let links = own_links(&state).await;
+ {
+ let mut inner = state.links.inner.lock_or_recover();
+ if inner
+ .published
+ .as_ref()
+ .is_some_and(|published| !changed(published, &links))
+ {
+ continue;
+ }
+ inner.published = Some(links.clone());
+ }
+ let (epoch, epoch_seq) = state.epoch_stamp();
+ for (peer, nonce) in subscribers {
+ let reply = ParsedMessage::LinksReply(LinksReply {
+ epoch,
+ epoch_seq,
+ nonce,
+ links: links.clone(),
+ });
+ if let Err(error) = state.transport.send_message(&peer, &reply).await {
+ debug!(peer = %peer, error = %format!("{error:#}"), "LINKS_REPLY failed");
+ }
+ }
+ }
+}
+
+/// Whether `new` says something `old` didn't: another peer, transport, relay or
+/// dormancy, a round trip measured or moved by more than a fifth (and 1 ms), or a clock
+/// measured or moved beyond both bounds. Round trips and clock samples wobble; small
+/// moves aren't news.
+fn changed(old: &[Link], new: &[Link]) -> bool {
+ let clock_moved = |a: Option<ClockOffset>, b: Option<ClockOffset>| match (a, b) {
+ (Some(a), Some(b)) => {
+ a.offset_ms.abs_diff(b.offset_ms) > a.within_ms.saturating_add(b.within_ms)
+ }
+ (a, b) => a.is_some() != b.is_some(),
+ };
+ let moved = |a: Option<u32>, b: Option<u32>| match (a, b) {
+ (Some(a), Some(b)) => {
+ let by = a.abs_diff(b);
+ by > 1000 && by > a.max(b) / 5
+ }
+ (a, b) => a.is_some() != b.is_some(),
+ };
+ old.len() != new.len()
+ || old.iter().zip(new).any(|(a, b)| {
+ a.peer != b.peer
+ || a.transport != b.transport
+ || a.relayed != b.relayed
+ || a.down != b.down
+ || a.dormant != b.dormant
+ || moved(a.rtt_us, b.rtt_us)
+ || clock_moved(a.clock, b.clock)
+ || a.others.len() != b.others.len()
+ || a.others.iter().zip(&b.others).any(|(x, y)| {
+ x.transport != y.transport
+ || x.relayed != y.relayed
+ || moved(x.rtt_us, y.rtt_us)
+ })
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn link(peer: &str, rtt_us: Option<u32>) -> Link {
+ Link {
+ peer: peer.into(),
+ transport: Some("libp2p".into()),
+ relayed: false,
+ down: false,
+ rtt_us,
+ dormant: false,
+ others: Vec::new(),
+ clock: None,
+ }
+ }
+
+ #[test]
+ fn only_news_is_a_change() {
+ let old = [link("a", Some(10_000))];
+ assert!(!changed(&old, &[link("a", Some(11_000))]));
+ assert!(changed(&old, &[link("a", Some(13_000))]));
+ assert!(changed(&old, &[link("a", None)]));
+ assert!(changed(&old, &[link("b", Some(10_000))]));
+ assert!(changed(&old, &[]));
+ let mut dormant = link("a", Some(10_000));
+ dormant.dormant = true;
+ assert!(changed(&old, &[dormant]));
+ let mut down = link("a", Some(10_000));
+ down.down = true;
+ assert!(changed(&old, &[down]));
+ // Small round trips: a millisecond's wobble isn't news.
+ assert!(!changed(&[link("a", Some(400))], &[link("a", Some(1_300))]));
+ // A clock: news when measured, and when it moved beyond both bounds.
+ let clocked = |offset_ms| Link {
+ clock: Some(ClockOffset {
+ offset_ms,
+ within_ms: 50,
+ }),
+ ..link("a", Some(10_000))
+ };
+ assert!(changed(&old, &[clocked(0)]));
+ assert!(!changed(&[clocked(0)], &[clocked(80)]));
+ assert!(changed(&[clocked(0)], &[clocked(5_000)]));
+ }
+}
diff --git a/src/swarm/members.rs b/src/swarm/members.rs
index 3bfb9cf..05c6501 100644
--- a/src/swarm/members.rs
+++ b/src/swarm/members.rs
@@ -18,12 +18,12 @@
use std::collections::HashMap;
use std::time::Duration;
-use automerge::AutoCommit;
use serde_json::{Value as JsonValue, json};
use tracing::{debug, info, warn};
use crate::document::meta;
use crate::document::store::{self, StoreError};
+use crate::document::{Document, ReadDoc, Transactable};
use crate::state::AppState;
pub(crate) fn members_dir() -> Vec<String> {
@@ -37,7 +37,7 @@ pub(crate) fn member_path(pk: &str) -> Vec<String> {
}
/// Public keys of all registered members, sorted.
-pub(crate) fn members(doc: &AutoCommit) -> Vec<String> {
+pub(crate) fn members(doc: &impl ReadDoc) -> Vec<String> {
let mut members = store::keys_at(doc, &members_dir()).unwrap_or_default();
members.sort();
members
@@ -67,14 +67,18 @@ fn tier_path(pk: &str) -> Vec<String> {
}
/// `pk`'s tier; transient unless its entry says stable.
-pub(crate) fn tier(doc: &AutoCommit, pk: &str) -> Tier {
+pub(crate) fn tier(doc: &Document, pk: &str) -> Tier {
match store::read_path(doc, &tier_path(pk)) {
Ok(Some(JsonValue::String(tier))) if tier == "stable" => Tier::Stable,
_ => Tier::Transient,
}
}
-pub(crate) fn set_tier(doc: &mut AutoCommit, pk: &str, tier: Tier) -> Result<(), StoreError> {
+pub(crate) fn set_tier(
+ doc: &mut impl Transactable,
+ pk: &str,
+ tier: Tier,
+) -> Result<(), StoreError> {
store::write_path(doc, &tier_path(pk), &json!(tier.as_str()))
}
@@ -85,7 +89,7 @@ fn alias_path(pk: &str) -> Vec<String> {
}
/// `pk`'s alias, if it gave itself one.
-pub(crate) fn alias(doc: &AutoCommit, pk: &str) -> Option<String> {
+pub(crate) fn alias(doc: &Document, pk: &str) -> Option<String> {
match store::read_path(doc, &alias_path(pk)) {
Ok(Some(JsonValue::String(alias))) => Some(alias),
_ => None,
@@ -93,14 +97,14 @@ pub(crate) fn alias(doc: &AutoCommit, pk: &str) -> Option<String> {
}
/// The aliases of all members that have one.
-pub(crate) fn aliases(doc: &AutoCommit) -> HashMap<String, String> {
+pub(crate) fn aliases(doc: &Document) -> HashMap<String, String> {
members(doc)
.into_iter()
.filter_map(|pk| alias(doc, &pk).map(|alias| (pk, alias)))
.collect()
}
-fn set_alias(doc: &mut AutoCommit, pk: &str, alias: Option<&str>) -> Result<(), StoreError> {
+fn set_alias(doc: &mut impl Transactable, pk: &str, alias: Option<&str>) -> Result<(), StoreError> {
match alias {
Some(alias) => store::write_path(doc, &alias_path(pk), &json!(alias)),
None => match store::delete_path(doc, &alias_path(pk), false) {
@@ -116,12 +120,14 @@ fn addresses_path(pk: &str) -> Vec<String> {
path
}
-/// Addresses a member most likely has, at most: one per transport.
-const MAX_ADDRESSES: usize = 8;
+/// Addresses a member lists, at most: a few per transport (libp2p's up to
+/// `PUBLISHED_DIRECT` and `PUBLISHED_RELAYED`, and where it relays, again as many to
+/// reserve a slot at), taken in turns by transport (`Paths::local_addresses`).
+const MAX_ADDRESSES: usize = 16;
/// Where `pk` says it can be reached (at most `MAX_ADDRESSES`); none if its entry
/// doesn't say.
-pub(crate) fn addresses(doc: &AutoCommit, pk: &str) -> Vec<String> {
+pub(crate) fn addresses(doc: &Document, pk: &str) -> Vec<String> {
match store::read_path(doc, &addresses_path(pk)) {
Ok(Some(JsonValue::Array(items))) => items
.into_iter()
@@ -135,9 +141,31 @@ pub(crate) fn addresses(doc: &AutoCommit, pk: &str) -> Vec<String> {
}
}
+/// What the member list tells this node besides who its neighbours are: where members
+/// are (`learn_addresses`), what they call themselves (`set_aliases`), and the relays
+/// listed (`relays`). Done whenever the neighbours are recomputed (`topology::refresh`).
+pub(crate) fn apply(state: &AppState, doc: &Document) {
+ learn_addresses(state, doc);
+ set_aliases(state, aliases(doc));
+ state.transport.learn_relays(&relays(doc));
+}
+
+/// Hands the members' `aliases` to the logger, under every name that stands for each:
+/// its node ID, and its addresses (`logging`, which renames them bare too).
+fn set_aliases(state: &AppState, aliases: HashMap<String, String>) {
+ let mut names = HashMap::new();
+ for (member, alias) in aliases {
+ for address in state.transport.identities_of(&member) {
+ names.insert(address, alias.clone());
+ }
+ names.insert(member, alias);
+ }
+ crate::logging::set_aliases(names);
+}
+
/// Hands every other member's addresses to the transport (`Transport::learn_addresses`):
/// the member list is how a node finds where members it never met are.
-pub(crate) fn learn_addresses(state: &AppState, doc: &AutoCommit) {
+pub(crate) fn learn_addresses(state: &AppState, doc: &Document) {
let me = state.local_id();
for pk in members(doc) {
if pk != me {
@@ -153,7 +181,7 @@ const MAX_RELAYS: usize = 32;
/// addresses (`libp2p:<multiaddr>`, with or without the relay's peer ID) of relays that
/// aren't members, e.g. nodes that only relay. Every member's transport may
/// pick them (`Transport::learn_relays`), after the relays it is configured with.
-pub(crate) fn relays(doc: &AutoCommit) -> Vec<String> {
+pub(crate) fn relays(doc: &Document) -> Vec<String> {
match store::read_path(doc, &meta::path(&[meta::RELAYS])) {
Ok(Some(JsonValue::Array(items))) => items
.into_iter()
@@ -173,7 +201,7 @@ fn alive_path(pk: &str) -> Vec<String> {
path
}
-fn millis_at(doc: &AutoCommit, path: &[String]) -> Option<u64> {
+fn millis_at(doc: &impl ReadDoc, path: &[String]) -> Option<u64> {
match store::read_path(doc, path) {
Ok(Some(JsonValue::Number(number))) => number.as_u64(),
_ => None,
@@ -181,29 +209,57 @@ fn millis_at(doc: &AutoCommit, path: &[String]) -> Option<u64> {
}
/// When `pk` last said it is alive (Unix ms): its last renewal, or its registration.
-pub(crate) fn alive_at(doc: &AutoCommit, pk: &str) -> Option<u64> {
+pub(crate) fn alive_at(doc: &impl ReadDoc, pk: &str) -> Option<u64> {
let mut since = member_path(pk);
since.push("since".to_owned());
millis_at(doc, &alive_path(pk)).max(millis_at(doc, &since))
}
/// The members that said they are alive within `window` before `now_ms` (sorted): the
-/// electorate of a remake (`remake::coordinator`).
-pub(crate) fn electorate(doc: &AutoCommit, now_ms: u64, window: Duration) -> Vec<String> {
+/// electorate of a remake (`remake::coordinator`). A stamp further in the future than a
+/// renewal interval (`alive_renewal`) doesn't count: its node's clock is that far ahead
+/// (it warns about it, `swarm::skew`), and the stamp would never age out.
+pub(crate) fn electorate(doc: &impl ReadDoc, now_ms: u64, window: Duration) -> Vec<String> {
let oldest = now_ms.saturating_sub(window.as_millis() as u64);
+ let newest = now_ms.saturating_add(alive_renewal(window).as_millis() as u64);
members(doc)
.into_iter()
- .filter(|pk| alive_at(doc, pk).is_some_and(|at| at >= oldest))
+ .filter(|pk| alive_at(doc, pk).is_some_and(|at| (oldest..=newest).contains(&at)))
.collect()
}
+fn window_path(pk: &str) -> Vec<String> {
+ let mut path = member_path(pk);
+ path.push("window".to_owned());
+ path
+}
+
+/// The shortest `--electorate-window` among this node's (`own`) and those the members
+/// publish in their entries (ms, `window`): every node renews its alive stamp by it
+/// (`alive_renewal`), so a coordinator with a shorter window than others' still counts
+/// them all. Below `MIN_ELECTORATE_WINDOW` (which no node accepts) counts as that.
+pub(crate) fn shortest_window(doc: &impl ReadDoc, own: Duration) -> Duration {
+ members(doc)
+ .iter()
+ .filter_map(|pk| millis_at(doc, &window_path(pk)))
+ .map(|ms| Duration::from_millis(ms).max(MIN_ELECTORATE_WINDOW))
+ .fold(own, Duration::min)
+}
+
/// How often a node renews its `alive` stamp: often enough that a live node never
-/// drops out of the electorate (a seventh of the window), but at most hourly and at
-/// least daily, so idle swarms stay quiet.
+/// drops out of the electorate (a seventh of the shortest window, `shortest_window`),
+/// but at most hourly and at least daily, so idle swarms stay quiet.
fn alive_renewal(window: Duration) -> Duration {
- (window / 7).clamp(Duration::from_secs(3600), Duration::from_secs(24 * 3600))
+ (window / 7).clamp(RENEWAL_MOST_OFTEN, Duration::from_secs(24 * 3600))
}
+const RENEWAL_MOST_OFTEN: Duration = Duration::from_secs(3600);
+
+/// The shortest `--electorate-window`: seven renewals at most often, so a live node
+/// renews within a seventh of it, as with any longer one.
+pub(crate) const MIN_ELECTORATE_WINDOW: Duration =
+ Duration::from_secs(7 * RENEWAL_MOST_OFTEN.as_secs());
+
/// Registers this node if it is ready and its entry is missing, keeps its alias as
/// configured (set, changed or removed after a restart with another `--alias`) and its
/// addresses as the transport gives them, and renews its `alive` stamp now and then
@@ -214,7 +270,8 @@ pub(crate) async fn ensure_registered(state: &AppState) {
}
let local = state.local_id().to_owned();
let own = member_path(&local);
- let (registered, recorded_alias, recorded_addresses, alive) = {
+ let window = state.electorate_window.as_millis() as u64;
+ let (registered, recorded_alias, recorded_addresses, alive, recorded_window, shortest) = {
let replica = state.doc.read().await;
let registered = store::read_path(&replica.doc, &own)
.ok()
@@ -225,6 +282,8 @@ pub(crate) async fn ensure_registered(state: &AppState) {
alias(&replica.doc, &local),
addresses(&replica.doc, &local),
millis_at(&replica.doc, &alive_path(&local)),
+ millis_at(&replica.doc, &window_path(&local)),
+ shortest_window(&replica.doc, state.electorate_window),
)
};
let wanted = state.alias.clone();
@@ -233,7 +292,12 @@ pub(crate) async fn ensure_registered(state: &AppState) {
let now = state.clock.now_millis().as_u64();
if !registered {
- let mut entry = json!({ "since": now, "alive": now, "addresses": own_addresses });
+ let mut entry = json!({
+ "since": now,
+ "alive": now,
+ "addresses": own_addresses,
+ "window": window,
+ });
if let Some(alias) = &wanted {
entry["alias"] = json!(alias);
}
@@ -250,7 +314,21 @@ pub(crate) async fn ensure_registered(state: &AppState) {
return;
}
- let renewal = alive_renewal(state.electorate_window).as_millis() as u64;
+ if recorded_window != Some(window) {
+ let path = window_path(&local);
+ match state
+ .mutate_doc(move |doc| store::write_path(doc, &path, &json!(window)))
+ .await
+ {
+ Ok(()) => debug!(
+ window_ms = window,
+ "Electorate window published in the member list"
+ ),
+ Err(error) => warn!(error = %error, "Failed to record our electorate window"),
+ }
+ }
+
+ let renewal = alive_renewal(shortest).as_millis() as u64;
if alive.is_none_or(|at| now.saturating_sub(at) >= renewal) {
let path = alive_path(&local);
match state
@@ -292,7 +370,7 @@ pub(crate) async fn ensure_registered(state: &AppState) {
}
/// Removes `pk`'s entry (an operator forgot it). Absent is fine.
-pub(crate) fn unregister(doc: &mut AutoCommit, pk: &str) -> Result<(), StoreError> {
+pub(crate) fn unregister(doc: &mut impl Transactable, pk: &str) -> Result<(), StoreError> {
match store::delete_path(doc, &member_path(pk), false) {
Err(StoreError::NotFound) => Ok(()),
result => result,
@@ -302,39 +380,72 @@ pub(crate) fn unregister(doc: &mut AutoCommit, pk: &str) -> Result<(), StoreErro
#[cfg(test)]
mod tests {
use super::*;
+ use crate::document::testing::change;
#[test]
fn electorate_by_alive_stamp() {
let day = 24 * 3600 * 1000;
let now = 100 * day;
- let mut doc = AutoCommit::new();
+ let mut doc = Document::new();
let entries = [
("fresh", json!({ "since": 1, "alive": now - day })),
("stale", json!({ "since": 1, "alive": now - 8 * day })),
("new", json!({ "since": now - day })),
("old", json!({ "since": 1 })),
+ // A clock a day ahead is tolerated (one renewal interval); a week isn't.
+ ("ahead", json!({ "since": 1, "alive": now + day })),
+ ("future", json!({ "since": 1, "alive": now + 7 * day })),
];
- for (pk, entry) in entries {
- store::write_path(&mut doc, &member_path(pk), &entry).unwrap();
- }
+ change(&mut doc, |doc| {
+ for (pk, entry) in entries {
+ store::write_path(doc, &member_path(pk), &entry).unwrap();
+ }
+ });
assert_eq!(
electorate(&doc, now, Duration::from_secs(7 * 24 * 3600)),
- ["fresh", "new"]
+ ["ahead", "fresh", "new"]
);
}
#[test]
fn listed_relays() {
- let mut doc = AutoCommit::new();
+ let mut doc = Document::new();
assert!(relays(&doc).is_empty());
let relay = "libp2p:/ip6/fd00::1/tcp/9661";
- store::write_path(&mut doc, &meta::path(&[meta::RELAYS]), &json!([relay, 7])).unwrap();
+ change(&mut doc, |doc| {
+ store::write_path(doc, &meta::path(&[meta::RELAYS]), &json!([relay, 7])).unwrap()
+ });
assert_eq!(relays(&doc), [relay]); // Anything but text is ignored.
- store::write_path(&mut doc, &meta::path(&[meta::RELAYS]), &json!(relay)).unwrap();
+ change(&mut doc, |doc| {
+ store::write_path(doc, &meta::path(&[meta::RELAYS]), &json!(relay)).unwrap()
+ });
assert!(relays(&doc).is_empty()); // Not a list.
}
#[test]
+ fn shortest_published_window() {
+ let day = Duration::from_secs(24 * 3600);
+ let mut doc = Document::new();
+ assert_eq!(shortest_window(&doc, 7 * day), 7 * day);
+ change(&mut doc, |doc| {
+ store::write_path(
+ doc,
+ &member_path("a"),
+ &json!({ "window": 30 * 86_400_000_u64 }),
+ )
+ .unwrap();
+ store::write_path(doc, &member_path("b"), &json!({ "window": 86_400_000 })).unwrap();
+ store::write_path(doc, &member_path("c"), &json!({ "since": 1 })).unwrap();
+ });
+ assert_eq!(shortest_window(&doc, 7 * day), day);
+ // Below what any node accepts: as the shortest it accepts.
+ change(&mut doc, |doc| {
+ store::write_path(doc, &window_path("c"), &json!(1)).unwrap();
+ });
+ assert_eq!(shortest_window(&doc, 7 * day), MIN_ELECTORATE_WINDOW);
+ }
+
+ #[test]
fn renewal_interval() {
let day = Duration::from_secs(24 * 3600);
assert_eq!(alive_renewal(7 * day), day);
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs
index d424529..8259ed5 100644
--- a/src/swarm/mod.rs
+++ b/src/swarm/mod.rs
@@ -4,11 +4,15 @@
pub(crate) mod auth;
pub(crate) mod blob;
pub(crate) mod bootstrap;
+pub(crate) mod changes;
+pub(crate) mod document_file;
+pub(crate) mod links;
pub(crate) mod members;
pub(crate) mod pair;
pub(crate) mod peer;
pub(crate) mod relay;
pub(crate) mod rtt;
+pub(crate) mod skew;
pub(crate) mod source;
pub(crate) mod stability;
pub(crate) mod state_file;
diff --git a/src/swarm/peer/admit.rs b/src/swarm/peer/admit.rs
index 1800775..6e02150 100644
--- a/src/swarm/peer/admit.rs
+++ b/src/swarm/peer/admit.rs
@@ -4,6 +4,7 @@ use tracing::{debug, info, trace, warn};
use super::handshake::remember_peer;
use crate::state::{AppState, HelloState, Standing};
+use crate::swarm::topology;
use crate::transport::ParsedMessage;
/// With a swarm key, only peers that proved it get past this point. A HELLO or
@@ -50,7 +51,7 @@ pub(super) async fn admit(state: &AppState, peer: &str, message: &ParsedMessage)
if peer_state.standing.is_admitted() {
warn!(peer = %peer, "Peer no longer proves the swarm key; dropping its membership");
}
- peer_state.standing = Standing::Stranger;
+ topology::change_peer(state, &mut peer_state, |p| p.standing = Standing::Stranger);
peer_state.hello_state = HelloState::NotSent;
peer_state.reset_sync();
}
diff --git a/src/swarm/peer/epoch.rs b/src/swarm/peer/epoch.rs
index 56711b1..d00a903 100644
--- a/src/swarm/peer/epoch.rs
+++ b/src/swarm/peer/epoch.rs
@@ -6,7 +6,6 @@ use tracing::{debug, info, warn};
use crate::document::replica::{Epoch, EpochRelation};
use crate::state::AppState;
-use crate::swarm::sync::SyncEvent;
use crate::transport;
/// Records the epoch a peer advertised and compares it with ours. Returns true when the
@@ -14,7 +13,8 @@ use crate::transport;
///
/// `seq` is the peer's `epoch_seq` with it. A message with a lower one than we have was
/// sent before the peer's last epoch change and delivered late: its epoch isn't
-/// recorded, so it can't flip our view of the peer back. The count starts over when the
+/// recorded, so it can't flip our view of the peer back. The message itself still
+/// counts: if its epoch is ours, its sync is applied (it is data of our epoch). The count starts over when the
/// peer restarts, which its HELLO or HELLO_REPLY tells by a new run (`run`).
pub(super) async fn observe_peer_epoch(
state: &AppState,
@@ -66,9 +66,7 @@ pub(super) async fn observe_peer_epoch(
EpochRelation::Same => {
if previous.is_some() {
info!(peer = %peer, epoch = %remote, "Peer is on our epoch again; resuming sync");
- let _ = state
- .sync_events
- .send(SyncEvent::PeerNeedsSync(peer.to_owned()));
+ state.request_sync(peer);
}
}
// Expected after a remake: either side switches or catches up.
diff --git a/src/swarm/peer/forget.rs b/src/swarm/peer/forget.rs
index 6386e3e..68266d5 100644
--- a/src/swarm/peer/forget.rs
+++ b/src/swarm/peer/forget.rs
@@ -5,11 +5,11 @@
use std::collections::HashMap;
use std::time::Duration;
-use automerge::AutoCommit;
use serde_json::Value as JsonValue;
use tokio::time::Instant;
use tracing::{info, warn};
+use crate::document::Document;
use crate::document::meta;
use crate::document::store::{self, StoreError};
use crate::state::{AppState, Standing};
@@ -24,7 +24,14 @@ pub(crate) fn forgotten_path(pk: &str) -> Vec<String> {
path
}
-fn forgotten_peers(doc: &AutoCommit) -> Vec<String> {
+/// Whether a change at `path` may change the forget records: one among them, or above
+/// them (the root, after an epoch switch).
+pub(crate) fn affects_forget_records(path: &[String]) -> bool {
+ let dir = forgotten_dir();
+ path.starts_with(&dir) || dir.starts_with(path)
+}
+
+fn forgotten_peers(doc: &Document) -> Vec<String> {
store::keys_at(doc, &forgotten_dir()).unwrap_or_default()
}
diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs
index 3e8fbf4..072545b 100644
--- a/src/swarm/peer/handshake.rs
+++ b/src/swarm/peer/handshake.rs
@@ -10,7 +10,6 @@ use tokio::time::Instant;
use tracing::{debug, info, instrument, trace, warn};
use crate::state::{AppState, HelloState, PeerState, Standing};
-use crate::swarm::sync::SyncEvent;
use crate::swarm::topology;
use crate::transport::{self, ParsedMessage};
@@ -148,15 +147,11 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> {
let mut peer_state = peer_state_mutex.lock().await;
if peer_state.hello_state == HelloState::Sending {
- let was_unreachable = peer_state.unreachable_since.is_some();
- peer_state.hello_sent(Instant::now());
- if !was_unreachable && peer_state.unreachable_since.is_some() {
- state.topology_changed.notify_one();
- }
+ topology::change_peer(state, &mut peer_state, |p| p.hello_sent(Instant::now()));
let wait = peer_state.next_hello_wait();
- peer_state.hello_state = HelloState::AwaitingReply {
- next_attempt: Instant::now() + wait,
- };
+ let next_attempt = Instant::now() + wait;
+ peer_state.hello_state = HelloState::AwaitingReply { next_attempt };
+ state.peers_due.by(next_attempt);
}
}
@@ -260,6 +255,11 @@ pub(super) async fn send_hello_reply(
/// (it says whom we greet; a listed peer that isn't one would get nothing from us, so
/// it isn't remembered either). Where the sender reaches them is a hint to the
/// transport (at most `MAX_LISTED_ADDRESSES` each).
+///
+/// A listed peer we can't reach (dormant, or unanswered HELLOs) is probed at once
+/// rather than when its probe is due (`probe_now`): the sender reaches it, so it's likely
+/// back for us too, as after a partition, whose sides would otherwise find each other
+/// pair by pair, each at its own probe (up to `dormant-probe-max` apart).
pub(super) async fn discover_peers(
state: &AppState,
from_peer: &str,
@@ -272,9 +272,6 @@ pub(super) async fn discover_peers(
if discovered_peer == from_peer || !state.transport.is_peer_id(discovered_peer) {
continue;
}
- if view_known && !topology::is_neighbour(state, discovered_peer) {
- continue;
- }
if taken == MAX_LISTED_PEERS {
debug!(
from = %from_peer,
@@ -283,9 +280,18 @@ pub(super) async fn discover_peers(
);
break;
}
+ let addresses = &listed.addresses[..listed.addresses.len().min(MAX_LISTED_ADDRESSES)];
+ if probe_now(state, discovered_peer).await {
+ taken += 1;
+ state.transport.learn_addresses(discovered_peer, addresses);
+ debug!(peer = %discovered_peer, by = %from_peer, "A peer we can't reach is reached by another; probing it now");
+ continue;
+ }
+ if view_known && !topology::is_neighbour(state, discovered_peer) {
+ continue;
+ }
taken += 1;
- let addresses = &listed.addresses[..listed.addresses.len().min(MAX_LISTED_ADDRESSES)];
state.transport.learn_addresses(discovered_peer, addresses);
remember_peer(state, discovered_peer).await;
if let Err(error) = send_hello_once(state, discovered_peer).await {
@@ -298,6 +304,53 @@ pub(super) async fn discover_peers(
}
}
+/// Whenever others may reach this node where they couldn't before
+/// (`Transport::on_reachable`: a relay granted it a slot), the members it can't reach
+/// are probed at once (`probe_now`): they may have given up on it meanwhile, waiting
+/// for their next probe, up to `dormant-probe-max` away. Two nodes behind NATs that
+/// restart together each fail to reach the other until it holds a slot; the second to
+/// get one reaches the first this way. Runs for the node's life.
+pub(crate) async fn probe_when_reachable(state: AppState, reachable: Arc<tokio::sync::Notify>) {
+ loop {
+ reachable.notified().await;
+ let mut probed = 0;
+ for (peer, _) in state.peers_snapshot().await {
+ if probe_now(&state, &peer).await {
+ probed += 1;
+ }
+ }
+ if probed > 0 {
+ debug!(
+ probed,
+ "Reachable anew; probing the members we can't reach now"
+ );
+ }
+ }
+}
+
+/// Brings `peer`'s next probe forward to now, if it is a member we can't reach (dormant,
+/// or unanswered HELLOs) whose probe is pending: the peer worker sends it (woken for
+/// it). Whether it was.
+async fn probe_now(state: &AppState, peer: &str) -> bool {
+ let Some(peer_state) = state.get_peer_state(peer).await else {
+ return false;
+ };
+ let mut peer_state = peer_state.lock().await;
+ if !matches!(peer_state.standing, Standing::Member { .. })
+ || !(peer_state.liveness.is_dormant() || peer_state.is_unreachable())
+ {
+ return false;
+ }
+ match &mut peer_state.hello_state {
+ HelloState::AwaitingReply { next_attempt } => {
+ *next_attempt = Instant::now();
+ state.peers_due.now();
+ true
+ }
+ _ => false,
+ }
+}
+
/// New neighbours get our HELLO (we may not have known them before: the member list is
/// how a partial mesh discovers them) and a sync opening. The opening matters for a
/// neighbour whose handshake is already complete, e.g. one that returned from dormancy
@@ -314,9 +367,7 @@ pub(crate) async fn apply_neighbour_change(state: &AppState, change: topology::N
}
// Skipped there unless the handshake is complete; otherwise its reply restarts
// sync anyway.
- let _ = state
- .sync_events
- .send(SyncEvent::PeerNeedsSync(peer.clone()));
+ state.request_sync(peer);
}
for peer in &change.removed {
if let Some(peer_state) = state.get_peer_state(peer).await {
@@ -372,11 +423,12 @@ pub(super) async fn establish_peer(
}
if !matches!(peer_state.standing, Standing::Member { .. }) {
- peer_state.standing = Standing::Member {
- swarm_id: swarm_id.clone(),
- };
// A member our list may lack yet: a neighbour then (`topology::refresh`).
- state.topology_changed.notify_one();
+ topology::change_peer(state, &mut peer_state, |p| {
+ p.standing = Standing::Member {
+ swarm_id: swarm_id.clone(),
+ }
+ });
}
// Stop our retries and sync from a clean slate. A new handshake from the peer always
@@ -386,6 +438,8 @@ pub(super) async fn establish_peer(
if from_hello || peer_state.hello_state != HelloState::Established {
peer_state.hello_state = HelloState::Established;
peer_state.reset_sync();
+ // Keepalives start.
+ state.peers_due.now();
info!(peer = %peer, from_hello, "Handshake complete; sync state reset");
}
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index 2e89739..18d29d5 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -4,14 +4,16 @@ use std::time::Duration;
use anyhow::{Result, anyhow, bail};
use tokio::time::Instant;
-use tracing::{debug, info, warn};
+use tracing::{debug, warn};
use super::admit::admit;
use super::epoch::observe_peer_epoch;
use super::handshake::{
discover_peers, ensure_handshake, establish_peer, peer_swarm, remember_peer, send_hello_reply,
};
-use super::keepalive::{heal_if_diverged, local_status, note_alive, note_status, status_received};
+use super::keepalive::{
+ heal_if_diverged, heard_from, local_status, note_alive, note_status, status_received,
+};
use super::lanes::SyncLanes;
use crate::state::AppState;
use crate::swarm::sync::SyncEvent;
@@ -31,7 +33,7 @@ pub(crate) async fn handle_message(
return Ok(());
}
// PING and PONG only say the peer is alive (`keepalive::note_alive`).
- if matches!(message, ParsedMessage::Ping(_) | ParsedMessage::Pong(_)) {
+ if matches!(message, ParsedMessage::Ping(_) | ParsedMessage::Pong { .. }) {
note_alive(state, &peer, &message).await;
return Ok(());
}
@@ -56,32 +58,24 @@ pub(crate) async fn handle_message(
remember_peer(state, &peer).await;
- // Any inbound message proves liveness. When we heard from it before tells a
- // restarted peer how long it was away (in our HELLO_REPLY).
- let (reactivated, previously_seen) = {
- let peer_state_mutex = state.get_or_create_peer_state(&peer).await;
- let mut peer_state = peer_state_mutex.lock().await;
- let previously_seen = peer_state.last_seen;
- if peer_state.unreachable_since.is_some() {
- // Reachable again (`mark_seen` clears it).
- state.topology_changed.notify_one();
- }
- (peer_state.mark_seen(Instant::now()), previously_seen)
- };
- crate::swarm::source::heard_from(state, &peer);
-
- if reactivated {
- info!(peer = %peer, "Dormant peer is active again; re-handshaking");
- }
-
let run = match &message {
transport::ParsedMessage::Hello(hello) => Some(hello.run_id),
transport::ParsedMessage::HelloReply(reply) => Some(reply.sender_run_id),
_ => None,
};
- let same_epoch = observe_peer_epoch(state, &peer, epoch, epoch_seq, run).await;
- //debug!(from = %peer, message = ?message, "Received message");
+ // Any inbound message proves liveness (but from a run that said farewell: its next
+ // one does). When we heard from it before tells a restarted peer how long it was
+ // away (in our HELLO_REPLY).
+ let peer_state = state.get_or_create_peer_state(&peer).await;
+ let (reactivated, previously_seen) =
+ heard_from(state, &peer, &peer_state, Instant::now(), |peer_state| {
+ if let Some(run) = run {
+ peer_state.returned(run);
+ }
+ })
+ .await;
+ let same_epoch = observe_peer_epoch(state, &peer, epoch, epoch_seq, run).await;
match message {
transport::ParsedMessage::Hello(hello) => {
@@ -162,10 +156,16 @@ pub(crate) async fn handle_message(
transport::ParsedMessage::StatusReply(remote) => {
on_status(state, &peer, &remote, reactivated).await;
}
+ transport::ParsedMessage::Links(request) => {
+ crate::swarm::links::answer(state, &peer, request).await;
+ }
+ transport::ParsedMessage::LinksReply(reply) => {
+ crate::swarm::links::received(state, &peer, reply);
+ }
transport::ParsedMessage::Who(_)
| transport::ParsedMessage::Me(_)
| transport::ParsedMessage::Ping(_)
- | transport::ParsedMessage::Pong(_) => {}
+ | transport::ParsedMessage::Pong { .. } => {}
}
Ok(())
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index ea7ade2..8597dcd 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -16,8 +16,7 @@ use tokio::time::Instant;
use tracing::{debug, info, warn};
use super::handshake::ensure_handshake;
-use crate::state::{AppState, Standing};
-use crate::swarm::sync::SyncEvent;
+use crate::state::{AppState, PeerState};
use crate::swarm::topology;
use crate::transport::{self, ParsedMessage};
@@ -49,12 +48,15 @@ pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMes
let now = Instant::now();
match message {
ParsedMessage::Ping(nonce) if !state.transport.answers_pings() => {
- let pong = ParsedMessage::Pong(*nonce);
+ let pong = ParsedMessage::Pong {
+ nonce: *nonce,
+ at: state.clock.now_millis(),
+ };
if let Err(error) = state.transport.send_message(peer, &pong).await {
debug!(peer = %peer, error = %format!("{error:#}"), "PONG failed");
}
}
- ParsedMessage::Pong(_) => {
+ ParsedMessage::Pong { .. } => {
if let Some(rtt) = state.transport.rtt(peer) {
state.lock_rtt().measured(peer, rtt, now);
}
@@ -64,23 +66,46 @@ pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMes
let Some(peer_state) = state.get_peer_state(peer).await else {
return;
};
- let reactivated = {
- let mut peer_state = peer_state.lock().await;
- if matches!(message, ParsedMessage::Ping(_)) {
+ let ping = matches!(message, ParsedMessage::Ping(_));
+ let (reactivated, _) = heard_from(state, peer, &peer_state, now, |peer_state| {
+ if ping {
// Its keepalive (or a path probe): the next is ours.
peer_state.keepalive_received_at = Some(now);
+ if let Some(at) = peer_state.keepalive_due_at(&state.timing) {
+ state.peers_due.by(at);
+ }
}
- if peer_state.unreachable_since.is_some() {
- // Reachable again (`mark_seen` clears it).
- state.topology_changed.notify_one();
- }
- peer_state.mark_seen(now)
+ })
+ .await;
+ if reactivated {
+ ensure_handshake(state, peer).await;
+ }
+}
+
+/// A message from `peer` came in: it is seen (`PeerState::mark_seen`: active, and
+/// reachable, again), and the catch-up source hears of it. `also` runs on its state
+/// under the same lock. Returns whether it was dormant (the caller re-handshakes), and
+/// when it was last seen before.
+pub(super) async fn heard_from(
+ state: &AppState,
+ peer: &str,
+ peer_state: &tokio::sync::Mutex<PeerState>,
+ now: Instant,
+ also: impl FnOnce(&mut PeerState),
+) -> (bool, Option<Instant>) {
+ let (reactivated, previously_seen) = {
+ let mut peer_state = peer_state.lock().await;
+ also(&mut peer_state);
+ let previously_seen = peer_state.last_seen;
+ // Reachable (again).
+ let reactivated = topology::change_peer(state, &mut peer_state, |p| p.mark_seen(now));
+ (reactivated, previously_seen)
};
crate::swarm::source::heard_from(state, peer);
if reactivated {
info!(peer = %peer, "Dormant peer is active again; re-handshaking");
- ensure_handshake(state, peer).await;
}
+ (reactivated, previously_seen)
}
/// How long shutting down waits for the last STATUSes (`farewell`).
@@ -120,7 +145,8 @@ pub(crate) fn announce(state: &AppState) {
/// catching up from us picks another source now, instead of once we go dormant. Bounded
/// in time: our transport may be gone already.
pub(crate) async fn farewell(state: &AppState) {
- state.stopping.store(true, Ordering::Release);
+ state.closing.send_replace(true);
+ info!("Saying farewell to our neighbours");
status_to_neighbours(state, Some(FAREWELL_TIMEOUT)).await;
}
@@ -137,20 +163,16 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta
)
};
let syncs_with_you = match state.get_peer_state(peer).await {
- Some(peer_state) => {
- let peer_state = peer_state.lock().await;
- !peer_state.liveness.is_dormant()
- && matches!(peer_state.standing, Standing::Member { .. })
- && peer_state.epoch == Some(epoch)
- }
+ Some(peer_state) => peer_state.lock().await.syncs_on(epoch),
None => false,
};
transport::Status {
heads,
epoch,
epoch_seq,
- ready: state.is_ready().await && !state.stopping.load(Ordering::Acquire),
+ ready: state.is_ready().await && !state.is_closing(),
syncs_with_you,
+ leaving: state.is_closing(),
}
}
@@ -162,6 +184,9 @@ pub(super) async fn status_received(state: &AppState, peer: &str) {
let now = Instant::now();
peer_state.keepalive_received_at = Some(now);
peer_state.last_status_at = Some(now);
+ if let Some(at) = peer_state.keepalive_due_at(&state.timing) {
+ state.peers_due.by(at);
+ }
}
}
@@ -169,11 +194,24 @@ pub(super) async fn status_received(state: &AppState, peer: &str) {
pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport::Status) {
if let Some(peer_state) = state.get_peer_state(peer).await {
let mut peer_state = peer_state.lock().await;
+ let left = topology::change_peer(state, &mut peer_state, |p| {
+ // A neighbour that only connects out stays one as long as it says so.
+ p.syncs_with_us = Some(remote.syncs_with_you);
+ // Its farewell: dormant now, as if silent for long (a message from it, as
+ // its HELLO when it's back, makes it active again).
+ remote.leaving && p.say_farewell(Instant::now())
+ });
+ if left {
+ // Probed from now on.
+ state.peers_due.now();
+ }
let became_ready = remote.ready && peer_state.reports_ready != Some(true);
peer_state.reports_ready = Some(remote.ready);
- peer_state.syncs_with_us = Some(remote.syncs_with_you);
peer_state.reported_heads = remote.heads.heads().map(<[_]>::len);
drop(peer_state);
+ if left {
+ info!(peer = %peer, "Peer said farewell; dormant until it is heard from again");
+ }
// Off the consumer: telling which heads we lack takes the document's lock.
{
let (state, peer, remote) = (state.clone(), peer.to_owned(), remote.clone());
@@ -184,9 +222,7 @@ pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport
if became_ready {
// We may have left the opening to it while it wasn't ready, and nothing else
// would make us look again.
- let _ = state
- .sync_events
- .send(SyncEvent::PeerNeedsSync(peer.to_owned()));
+ state.request_sync(peer);
}
}
crate::swarm::source::reported(state, peer, remote.ready, Some(remote.syncs_with_you));
diff --git a/src/swarm/peer/mod.rs b/src/swarm/peer/mod.rs
index 326f3a9..cf427a2 100644
--- a/src/swarm/peer/mod.rs
+++ b/src/swarm/peer/mod.rs
@@ -20,11 +20,12 @@ mod keepalive;
mod lanes;
mod worker;
-pub(crate) use forget::{forget_unless_alive, forgotten_path};
+pub(crate) use forget::{affects_forget_records, forget_unless_alive, forgotten_path};
pub(crate) use handshake::{
- apply_neighbour_change, peer_swarm, rehandshake, remember_peer, send_hello_once,
+ apply_neighbour_change, peer_swarm, probe_when_reachable, rehandshake, remember_peer,
+ send_hello_once,
};
pub(crate) use inbound::handle_message;
pub(crate) use keepalive::{announce, farewell, send_status};
pub(crate) use lanes::SyncLanes;
-pub(crate) use worker::run_peer_worker;
+pub(crate) use worker::{Due, run_peer_worker};
diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
index f3d0e84..6196571 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -1,11 +1,12 @@
-//! The peer worker: every `PEER_TICK`, keepalives due, dormancy, HELLO retries and
-//! overdue sync replies; every `FORGET_CHECK_INTERVAL` (and at once when the node
-//! becomes ready), forget records, the member registry, stability, measurements and
-//! the state file.
+//! The peer worker: keepalives due, dormancy, HELLO retries and overdue sync replies,
+//! when the earliest of them is due (or something may have made one due sooner,
+//! `Due`); every `FORGET_CHECK_INTERVAL` (and at once when the node becomes ready),
+//! forget records, the member registry, stability, measurements and the state file;
+//! forget records also at once when they changed.
use std::time::Duration;
-use tokio::time::{Instant, MissedTickBehavior};
+use tokio::time::Instant;
use tracing::{debug, info, warn};
use super::forget::{AppliedForgets, apply_forget_records, forget_strangers};
@@ -13,17 +14,51 @@ use super::handshake::retry_hello_if_due;
use super::keepalive::{send_ping, send_status};
use crate::locks::MutexExt;
use crate::state::{AppState, Keepalive};
-use crate::swarm::sync::SyncEvent;
use crate::swarm::topology;
-const PEER_TICK: Duration = Duration::from_millis(500);
+/// The least time between two passes, however often something is due.
+const MIN_PASS_GAP: Duration = Duration::from_millis(20);
const FORGET_CHECK_INTERVAL: Duration = Duration::from_secs(5);
-pub(crate) async fn run_peer_worker(state: AppState) {
- let mut tick = tokio::time::interval(PEER_TICK);
- tick.set_missed_tick_behavior(MissedTickBehavior::Delay);
+/// When the peer worker is next due. What makes something of its due sooner than it
+/// knows (a HELLO sent, a sync message waiting for its answer, a keepalive received, a
+/// new peer, new neighbours) says so (`by`); what only puts things off needn't.
+#[derive(Default)]
+pub(crate) struct Due {
+ /// Its next pass while it waits; None while it runs.
+ next: std::sync::Mutex<Option<Instant>>,
+ wake: tokio::sync::Notify,
+}
+
+impl Due {
+ /// Something is due at `at`: a pass then at the latest.
+ pub(crate) fn by(&self, at: Instant) {
+ if self.next.lock_or_recover().is_none_or(|next| at < next) {
+ self.wake.notify_one();
+ }
+ }
+ /// A pass now.
+ pub(crate) fn now(&self) {
+ self.wake.notify_one();
+ }
+
+ fn passing(&self) {
+ *self.next.lock_or_recover() = None;
+ }
+
+ /// Until `at`, or until something is due sooner.
+ async fn until(&self, at: Instant) {
+ *self.next.lock_or_recover() = Some(at);
+ tokio::select! {
+ () = self.wake.notified() => {}
+ () = tokio::time::sleep_until(at) => {}
+ }
+ }
+}
+
+pub(crate) async fn run_peer_worker(state: AppState) {
info!("Starting peer maintenance worker");
let mut last_forget_check: Option<Instant> = None;
@@ -31,7 +66,8 @@ pub(crate) async fn run_peer_worker(state: AppState) {
let mut was_ready = false;
loop {
- tick.tick().await;
+ state.peers_due.passing();
+ let mut next = Instant::now() + FORGET_CHECK_INTERVAL;
// While joining, keep handshakes fast; dormancy would slow probes to minutes.
let ready = state.is_ready().await;
@@ -52,8 +88,11 @@ pub(crate) async fn run_peer_worker(state: AppState) {
let (became_dormant, keepalive, unanswered) = {
let mut peer_state = peer_state_mutex.lock().await;
- let became_dormant =
- ready && neighbour && peer_state.check_dormancy(&state.timing, now);
+ let became_dormant = ready
+ && neighbour
+ && topology::change_peer(&state, &mut peer_state, |p| {
+ p.check_dormancy(&state.timing, now)
+ });
let keepalive = if neighbour {
peer_state.take_keepalive_due(&state.timing, now)
} else {
@@ -69,13 +108,10 @@ pub(crate) async fn run_peer_worker(state: AppState) {
if let Some(changes) = unanswered {
debug!(peer = %peer, changes, "No answer to our sync message; sending again");
crate::swarm::source::reply_timed_out(&state, peer);
- let _ = state
- .sync_events
- .send(SyncEvent::PeerNeedsSync(peer.clone()));
+ state.request_sync(peer);
}
if became_dormant {
- state.topology_changed.notify_one();
warn!(
peer = %peer,
silent_secs = state.timing.dormant_after.as_secs(),
@@ -104,19 +140,45 @@ pub(crate) async fn run_peer_worker(state: AppState) {
if let Err(error) = retry_hello_if_due(&state, peer).await {
warn!(peer = %peer, error = %format!("{error:#}"), "HELLO retry failed");
}
+
+ let due = peer_state_mutex
+ .lock()
+ .await
+ .next_due(&state.timing, neighbour, ready);
+ let probe = crate::swarm::source::probe_at(&state, peer);
+ next = [Some(next), due, probe]
+ .into_iter()
+ .flatten()
+ .min()
+ .unwrap_or(next);
}
- if last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL) {
+ let records_changed = state
+ .forget_records_changed
+ .swap(false, std::sync::atomic::Ordering::AcqRel);
+ let periodic = last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL);
+ if records_changed || periodic {
apply_forget_records(&state, &mut forgets_applied).await;
+ }
+ if periodic {
forget_strangers(&state).await;
crate::swarm::members::ensure_registered(&state).await;
crate::swarm::stability::tick(&state).await;
+ crate::swarm::skew::check(&state).await;
crate::swarm::stats::measure_saved_size(&state);
crate::swarm::source::warn_if_stalled(&state);
crate::swarm::state_file::update(&state, false).await;
{
let known = state.peer_states.read().await;
- state.traffic.retain(|peer| known.contains_key(peer));
+ // And their addresses, where the layers below the node count
+ // (`Traffic::node`).
+ let addresses: std::collections::HashSet<String> = known
+ .keys()
+ .flat_map(|peer| state.transport.identities_of(peer))
+ .collect();
+ state
+ .traffic
+ .retain(|peer| known.contains_key(peer) || addresses.contains(peer));
state
.throughput
.lock_or_recover()
@@ -126,5 +188,12 @@ pub(crate) async fn run_peer_worker(state: AppState) {
state.topology_changed.notify_one();
last_forget_check = Some(Instant::now());
}
+ if let Some(checked) = last_forget_check {
+ next = next.min(checked + FORGET_CHECK_INTERVAL);
+ }
+
+ let passed = Instant::now();
+ state.peers_due.until(next).await;
+ tokio::time::sleep_until(passed + MIN_PASS_GAP).await;
}
}
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 6172f67..1736cec 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -28,7 +28,6 @@ use tracing::debug;
use crate::locks::MutexExt;
use crate::state::{AppState, Standing};
use crate::swarm::peer::send_status;
-use crate::swarm::sync::SyncEvent;
use crate::swarm::topology;
use crate::transport::{self, HeadsSummary};
@@ -131,16 +130,9 @@ pub(crate) async fn pull_if_behind(
}
// Those of the listed heads we lack. With a digest only a sample is listed; if we
// have all of it, a later keepalive lists others.
- let missing: Vec<_> = {
- // AutoCommit's lookups take &mut self.
- let mut replica = state.doc.write().await;
- remote
- .heads
- .listed()
- .iter()
- .filter(|head| replica.doc.get_change_by_hash(head).is_none())
- .copied()
- .collect()
+ let missing = {
+ let replica = state.doc.read().await;
+ crate::document::changes::missing(&replica.doc, remote.heads.listed())
};
if missing.is_empty() {
return false;
@@ -183,9 +175,7 @@ pub(crate) async fn pull_if_behind(
peer_state.lock().await.sync.pull_from(theirs);
state.stats.pull_started();
debug!(peer = %peer, "Neighbour has changes we lack; pulling");
- let _ = state
- .sync_events
- .send(SyncEvent::PeerNeedsSync(peer.to_owned()));
+ state.request_sync(peer);
true
}
diff --git a/src/swarm/skew.rs b/src/swarm/skew.rs
new file mode 100644
index 0000000..69c85f5
--- /dev/null
+++ b/src/swarm/skew.rs
@@ -0,0 +1,277 @@
+//! Whether this node's clock agrees with its peers' (`Transport::clock_offset`, from
+//! their PONGs). Each peer's clock is known within half a round trip; this node's own
+//! is off when it differs from most of its peers' by more than `--clock-skew-warn` for
+//! sure, then the log says so (every `WARN_EVERY` while it lasts). A clock set anew
+//! between two checks (by NTP, a resumed machine, by hand) by more than the limit shows
+//! as a jump, this node's when most peers' clocks moved alike, else those peers'; the
+//! log says so too. Clocks are the operating system's to set (NTP); the propagation
+//! figures only take this node's own offset off (`AppState::clock_ahead_ms`).
+//!
+//! Two nodes alone can't tell whose clock is off, or jumped: each says it is its own.
+
+use std::collections::HashMap;
+
+use std::time::Duration;
+
+use serde::Serialize;
+use tokio::time::Instant;
+use tracing::{info, warn};
+
+use crate::locks::MutexExt;
+use crate::state::AppState;
+use crate::transport::paths::ClockOffset;
+
+/// While this node's clock is off, the log says so this often.
+const WARN_EVERY: Duration = Duration::from_secs(600);
+
+/// How this node's clock compares with its peers'.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Skew {
+ /// Peers whose clocks were compared (active ones with a PONG lately).
+ pub(crate) peers: usize,
+ /// How far this node's clock is ahead of theirs (behind if negative): the median.
+ pub(crate) offset_ms: Option<i64>,
+ /// Peers whose clocks are behind ours by more than the limit for sure, and ahead.
+ pub(crate) behind: usize,
+ pub(crate) ahead: usize,
+ /// Whether ours is off: more than half of them differ from it beyond the limit,
+ /// the same way. Never with a limit of 0.
+ pub(crate) off: bool,
+ /// The limit (`--clock-skew-warn`).
+ pub(crate) limit_ms: u64,
+}
+
+/// Ours against the peers' clocks `offsets` (each ahead of ours), with `limit`.
+pub(crate) fn of(offsets: &[ClockOffset], limit: Duration) -> Skew {
+ let limit_ms = u64::try_from(limit.as_millis()).unwrap_or(u64::MAX);
+ let beyond = |offset_ms: i64, within_ms: u64| {
+ offset_ms > 0 && offset_ms.unsigned_abs() > within_ms.saturating_add(limit_ms)
+ };
+ let ahead = offsets
+ .iter()
+ .filter(|o| beyond(o.offset_ms, o.within_ms))
+ .count();
+ let behind = offsets
+ .iter()
+ .filter(|o| beyond(-o.offset_ms, o.within_ms))
+ .count();
+ let mut ours: Vec<i64> = offsets.iter().map(|o| -o.offset_ms).collect();
+ ours.sort_unstable();
+ let peers = offsets.len();
+ Skew {
+ peers,
+ offset_ms: ours.get(peers / 2).copied(),
+ behind,
+ ahead,
+ off: limit_ms > 0 && (ahead * 2 > peers || behind * 2 > peers),
+ limit_ms,
+ }
+}
+
+/// The clocks of the peers that are active now, by peer.
+async fn offsets(state: &AppState) -> Vec<(String, ClockOffset)> {
+ let mut offsets = Vec::new();
+ for (peer, peer_state) in state.peers_snapshot().await {
+ if peer_state.lock().await.liveness.is_dormant() {
+ continue;
+ }
+ if let Some(offset) = state.transport.clock_offset(&peer) {
+ offsets.push((peer, offset));
+ }
+ }
+ offsets
+}
+
+/// Ours against the clocks of the peers that are active now.
+pub(crate) async fn measure(state: &AppState) -> Skew {
+ let offsets: Vec<ClockOffset> = offsets(state).await.into_iter().map(|(_, o)| o).collect();
+ of(&offsets, state.clock_skew_warn)
+}
+
+/// What the checks remember between them (`AppState::clock_skew`).
+#[derive(Debug, Default)]
+pub(crate) struct Watch {
+ /// When the log last said this node's clock is off, while it is.
+ warned: Option<Instant>,
+ /// Each peer's clock at the last check.
+ last: HashMap<String, ClockOffset>,
+}
+
+/// Clocks set anew between two checks, by more than the limit for sure.
+#[derive(Debug, Default, PartialEq, Eq)]
+struct Jumps {
+ /// Ours, by how much (forward if positive): most peers' clocks moved alike.
+ ours: Option<i64>,
+ /// Otherwise the peers' whose clocks moved, each by how much.
+ peers: Vec<(String, i64)>,
+}
+
+/// Which clocks jumped from `last` to `now` (each peer's ahead of ours), by more than
+/// `limit_ms` beyond both bounds. None with a limit of 0.
+fn jumps(
+ last: &HashMap<String, ClockOffset>,
+ now: &[(String, ClockOffset)],
+ limit_ms: u64,
+) -> Jumps {
+ if limit_ms == 0 {
+ return Jumps::default();
+ }
+ let moves: Vec<(&String, i64, bool)> = now
+ .iter()
+ .filter_map(|(peer, offset)| {
+ let before = last.get(peer)?;
+ let moved = offset.offset_ms - before.offset_ms;
+ let bound = offset
+ .within_ms
+ .saturating_add(before.within_ms)
+ .saturating_add(limit_ms);
+ Some((peer, moved, moved.unsigned_abs() > bound))
+ })
+ .collect();
+ let forward = moves
+ .iter()
+ .filter(|(_, moved, sure)| *sure && *moved > 0)
+ .count();
+ let back = moves
+ .iter()
+ .filter(|(_, moved, sure)| *sure && *moved < 0)
+ .count();
+ if forward * 2 > moves.len() || back * 2 > moves.len() {
+ // Ours moved the other way, by the median.
+ let mut ours: Vec<i64> = moves.iter().map(|(_, moved, _)| -moved).collect();
+ ours.sort_unstable();
+ return Jumps {
+ ours: ours.get(ours.len() / 2).copied(),
+ peers: Vec::new(),
+ };
+ }
+ Jumps {
+ ours: None,
+ peers: moves
+ .into_iter()
+ .filter(|(_, _, sure)| *sure)
+ .map(|(peer, moved, _)| (peer.clone(), moved))
+ .collect(),
+ }
+}
+
+/// Notes how far this node's clock is ahead of its peers' (`AppState::clock_ahead_ms`),
+/// and says in the log when it is off, and when it agrees again. From the peer worker,
+/// every few seconds.
+pub(crate) async fn check(state: &AppState) {
+ let offsets = offsets(state).await;
+ let skew = of(
+ &offsets.iter().map(|(_, o)| *o).collect::<Vec<_>>(),
+ state.clock_skew_warn,
+ );
+ if let Some(offset_ms) = skew.offset_ms {
+ state
+ .clock_ahead_ms
+ .store(offset_ms, std::sync::atomic::Ordering::Relaxed);
+ }
+ let now = Instant::now();
+ let mut watch = state.clock_skew.lock_or_recover();
+ let jumped = jumps(&watch.last, &offsets, skew.limit_ms);
+ if let Some(ms) = jumped.ours {
+ warn!(
+ jump_ms = ms,
+ peers = offsets.len(),
+ "This node's clock jumped (set anew)"
+ );
+ }
+ for (peer, ms) in &jumped.peers {
+ info!(peer = %peer, jump_ms = ms, "A peer's clock jumped (set anew)");
+ }
+ watch.last = offsets.into_iter().collect();
+ let warned = &mut watch.warned;
+ if skew.off {
+ if warned.is_none_or(|at| now.saturating_duration_since(at) >= WARN_EVERY) {
+ warn!(
+ offset_ms = skew.offset_ms,
+ peers = skew.peers,
+ behind = skew.behind,
+ ahead = skew.ahead,
+ limit_ms = state.clock_skew_warn.as_millis() as u64,
+ "This node's clock differs from most of its peers'; check its time synchronisation (NTP)"
+ );
+ *warned = Some(now);
+ }
+ } else if warned.take().is_some() {
+ info!(
+ offset_ms = skew.offset_ms,
+ peers = skew.peers,
+ "This node's clock agrees with its peers' again"
+ );
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn offset(offset_ms: i64, within_ms: u64) -> ClockOffset {
+ ClockOffset {
+ offset_ms,
+ within_ms,
+ }
+ }
+
+ /// Off only beyond the limit and the bound, and only against most peers.
+ #[test]
+ fn off_against_most_peers_for_sure() {
+ let limit = Duration::from_secs(2);
+ // Every peer 5 s ahead: ours is 5 s behind.
+ let behind = of(
+ &[offset(5_000, 100), offset(5_100, 50), offset(4_900, 300)],
+ limit,
+ );
+ assert_eq!(behind.offset_ms, Some(-5_000));
+ assert_eq!((behind.ahead, behind.behind, behind.off), (3, 0, true));
+ // One peer off: that peer's clock, not ours.
+ let one = of(
+ &[offset(5_000, 100), offset(10, 50), offset(-20, 30)],
+ limit,
+ );
+ assert_eq!((one.ahead, one.off), (1, false));
+ // Within the bound: not for sure.
+ let unsure = of(&[offset(2_500, 600), offset(2_500, 600)], limit);
+ assert!(!unsure.off);
+ // A limit of 0: never.
+ assert!(!of(&[offset(60_000, 10)], Duration::ZERO).off);
+ // No peers: nothing known.
+ let none = of(&[], limit);
+ assert_eq!((none.peers, none.offset_ms, none.off), (0, None, false));
+ }
+
+ /// Most peers' clocks moved alike: ours jumped; a few moved: theirs.
+ #[test]
+ fn jumps_are_ours_when_most_moved_alike() {
+ let peers = |offsets: &[i64]| -> Vec<(String, ClockOffset)> {
+ offsets
+ .iter()
+ .enumerate()
+ .map(|(i, &ms)| (format!("p{i}"), offset(ms, 50)))
+ .collect()
+ };
+ let last: HashMap<_, _> = peers(&[0, 10, -20]).into_iter().collect();
+ // Ours went 30 s forward: theirs are 30 s behind it now.
+ let ours = jumps(&last, &peers(&[-30_000, -29_990, -30_020]), 2_000);
+ assert_eq!(ours.ours, Some(30_000));
+ assert!(ours.peers.is_empty());
+ // One peer's went 10 s back.
+ let one = jumps(&last, &peers(&[-10_000, 10, -20]), 2_000);
+ assert_eq!(one.ours, None);
+ assert_eq!(one.peers, vec![("p0".to_owned(), -10_000)]);
+ // Within the bounds and the limit: nothing.
+ assert_eq!(
+ jumps(&last, &peers(&[1_000, 10, -20]), 2_000),
+ Jumps::default()
+ );
+ // A limit of 0: never.
+ assert_eq!(
+ jumps(&last, &peers(&[-30_000, -30_000, -30_000]), 0),
+ Jumps::default()
+ );
+ }
+}
diff --git a/src/swarm/source.rs b/src/swarm/source.rs
index d80e4d7..1288d86 100644
--- a/src/swarm/source.rs
+++ b/src/swarm/source.rs
@@ -37,7 +37,6 @@ use tracing::{debug, info, warn};
use crate::locks::MutexExt;
use crate::state::{AppState, CatchUpSource, PeerState};
-use crate::swarm::sync::SyncEvent;
/// How long a clearly closer peer (by round-trip time) gets to claim the source role
/// before whoever asks next takes it.
@@ -147,6 +146,7 @@ pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Opt
last_probe: None,
refusals: 0,
});
+ state.peers_due.by(now + state.timing.catch_up_probe);
None
}
@@ -286,18 +286,39 @@ pub(crate) fn probe_due(state: &AppState, peer: &str, now: Instant) -> bool {
due
}
-/// The initial sync is done: no source any more, and the peers we held back only need
-/// to learn our heads now.
-pub(crate) fn finished(state: &AppState, from: &str) {
+/// When a STATUS to `peer` is next due (`probe_due`), if it is the source.
+pub(crate) fn probe_at(state: &AppState, peer: &str) -> Option<Instant> {
+ if !catching_up(state) {
+ return None;
+ }
+ let probe = state.timing.catch_up_probe;
+ let source = lock(state);
+ let current = source.as_ref().filter(|current| current.peer == peer)?;
+ let heard = current.last_heard + probe;
+ Some(match current.last_probe {
+ Some(last) => heard.max(last + probe),
+ None => heard,
+ })
+}
+
+/// No catch-up going on: no source, no head start, nobody avoided. When one ends, and
+/// before one starts (a rejoin: `Host::reset_initial_sync`).
+pub(crate) fn reset(state: &AppState) {
*lock(state) = None;
*lock_preference(state) = None;
*lock_avoided(state) = None;
+}
+
+/// The initial sync is done: no source any more, and the peers we held back only need
+/// to learn our heads now.
+pub(crate) fn finished(state: &AppState, from: &str) {
+ reset(state);
let state = state.clone();
let from = from.to_owned();
tokio::spawn(async move {
let peers: Vec<String> = state.peer_states.read().await.keys().cloned().collect();
for other in peers.into_iter().filter(|other| *other != from) {
- let _ = state.sync_events.send(SyncEvent::PeerNeedsSync(other));
+ state.request_sync(&other);
}
});
}
@@ -308,7 +329,7 @@ pub(crate) fn resync_everyone(state: &AppState) {
tokio::spawn(async move {
let peers: Vec<String> = state.peer_states.read().await.keys().cloned().collect();
for other in peers {
- let _ = state.sync_events.send(SyncEvent::PeerNeedsSync(other));
+ state.request_sync(&other);
}
});
}
diff --git a/src/swarm/state_file.rs b/src/swarm/state_file.rs
index a65fa5b..415fb1a 100644
--- a/src/swarm/state_file.rs
+++ b/src/swarm/state_file.rs
@@ -185,7 +185,7 @@ pub(crate) async fn update(state: &AppState, now: bool) {
/// Where to rejoin through: the addresses of members alive within the electorate
/// window, stable ones first, of at most `MAX_SEEDS` members, sorted. Changes only when
/// members join, leave, move or change tier, so the file rarely does.
-fn seeds(doc: &automerge::AutoCommit, me: &str, window: Duration, now: u64) -> Vec<String> {
+fn seeds(doc: &crate::document::Document, me: &str, window: Duration, now: u64) -> Vec<String> {
let (mut stable, transient): (Vec<String>, Vec<String>) = members::electorate(doc, now, window)
.into_iter()
.filter(|pk| pk != me)
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index b5548d5..04f41b2 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -278,8 +278,8 @@ pub(crate) fn measure_saved_size(state: &AppState) {
let doc = std::sync::Arc::clone(&state.doc);
// Both steps take time proportional to the document: off the async threads.
let measured = match tokio::task::spawn_blocking(move || {
- let mut copy = doc.blocking_read().doc.clone();
- copy.save().len() as u64
+ let copy = doc.blocking_read().doc.clone();
+ crate::document::save(&copy).len() as u64
})
.await
{
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index df4bb2d..3bfbf7d 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -11,7 +11,7 @@ use tokio::time::Instant;
use tracing::{debug, info, trace, warn};
use crate::locks::MutexExt;
-use crate::state::{AppState, Liveness, PeerState, Standing};
+use crate::state::{AppState, PeerState, Standing};
use crate::swarm::peer::{peer_swarm, send_hello_once};
use crate::swarm::source;
use crate::swarm::throughput::SyncBytes;
@@ -25,12 +25,6 @@ pub(crate) enum SyncEvent {
PeerNeedsSync(String),
}
-/// Store paths touched by one batch of document changes.
-#[derive(Debug)]
-pub(crate) struct ChangeSet {
- pub(crate) paths: Vec<Vec<String>>,
-}
-
async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
if state.remake.fetching_snapshot() {
// Rejoining: the snapshot comes as a download, not in a sync message. Sync
@@ -48,7 +42,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
let validated_swarm = {
let peer_state = peer_state_mutex.lock().await;
- if matches!(peer_state.liveness, Liveness::Dormant { .. }) {
+ if peer_state.liveness.is_dormant() {
// Reactivation re-handshakes, and PeerValidated then restarts sync.
trace!("Skipping sync with dormant peer");
return Ok(());
@@ -74,12 +68,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
);
// Done earlier when the initial sync completes (every peer is enqueued
// then); this only matters if the source goes quiet.
- let tx = state.sync_events.clone();
- let peer = peer.to_owned();
- tokio::spawn(async move {
- tokio::time::sleep(wait).await;
- let _ = tx.send(SyncEvent::PeerNeedsSync(peer));
- });
+ state.request_sync_after(peer, wait);
return Ok(());
}
@@ -139,7 +128,16 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
let ready = state.is_ready().await;
// Before this message is counted.
let bytes_before = SyncBytes::of(&state.traffic, peer);
- let delta_threshold = state.blobs.delta_threshold();
+ // Lower towards a peer a message that large wouldn't reach whole: a slow link
+ // takes a delta chunk by chunk, where a message whole would fail and be sent
+ // again, whole, every time.
+ let rate = state
+ .throughput
+ .lock_or_recover()
+ .get(peer)
+ .map(|estimate| estimate.bytes_per_sec);
+ let failed = peer_state_mutex.lock().await.failed_whole(Instant::now());
+ let delta_threshold = state.blobs.delta_threshold_for(rate, failed);
// The session the message comes from, for its send's outcome.
let (session, generated) = tokio::task::spawn_blocking(
@@ -147,7 +145,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
// Label the message with the epoch of the document it was generated
// from, read under the same lock.
let (session, epoch, epoch_seq, body) = {
- let mut replica = doc.blocking_write();
+ let replica = doc.blocking_read();
let mut peer_state = psm.blocking_lock();
let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
// Under the same locks as the epoch label: the message belongs to
@@ -155,9 +153,8 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
let message =
peer_state
.sync
- .generate(&mut replica.doc, Instant::now(), bytes_before);
- let body = message.map(|message| {
- let payload = transport::SyncPayload::encode(message);
+ .generate(&replica.doc, Instant::now(), bytes_before);
+ let body = message.map(|payload| {
if payload.0.len() <= delta_threshold {
return transport::SyncBody::Message(payload);
}
@@ -210,6 +207,17 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
let Some(encoded) = generated else {
return Ok(());
};
+ // Its answer is due.
+ let reply_due = peer_state_mutex
+ .lock()
+ .await
+ .sync
+ .reply
+ .as_ref()
+ .map(|wait| wait.deadline);
+ if let Some(at) = reply_due {
+ state.peers_due.by(at);
+ }
(session, encoded)
};
@@ -247,27 +255,22 @@ async fn send_generated(
);
if regenerate {
- state
- .sync_events
- .send(SyncEvent::PeerNeedsSync(peer.to_owned()))
- .map_err(|error| anyhow!("Failed to notify sync worker: {error}"))?;
+ state.request_sync(peer);
}
Ok(())
}
Err(error) => {
- let Some(delay) = peer_state_mutex
- .lock()
- .await
- .sync
- .send_failed(session, Instant::now())
- else {
+ let delay = {
+ let mut peer_state = peer_state_mutex.lock().await;
+ let now = Instant::now();
+ peer_state.whole_failed(encoded.payload_len, now);
+ peer_state.sync.send_failed(session, now)
+ };
+ let Some(delay) = delay else {
// The new session may not have opened yet.
debug!("Sync message refused, but a reset already discarded it");
- state
- .sync_events
- .send(SyncEvent::PeerNeedsSync(peer.to_owned()))
- .map_err(|error| anyhow!("Failed to notify sync worker: {error}"))?;
+ state.request_sync(peer);
return Ok(());
};
@@ -277,12 +280,7 @@ async fn send_generated(
"Sync delivery failed; generating a fresh message later"
);
- let tx = state.sync_events.clone();
- let peer = peer.to_owned();
- tokio::spawn(async move {
- tokio::time::sleep(delay).await;
- let _ = tx.send(SyncEvent::PeerNeedsSync(peer));
- });
+ state.request_sync_after(peer, delay);
Ok(())
}
@@ -372,12 +370,13 @@ pub(crate) async fn handle_sync(
let message = payload;
let peer = peer.to_owned();
let clock = Arc::clone(&state.clock);
+ let ahead_ms = state.clock_ahead_ms.load(Ordering::Relaxed);
// Some((caught_up, heads changed)) when applied, None when the message belongs
// to another epoch.
tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool)>> {
// Before the lock: decoding a large message takes a while.
- let message = message.decode()?;
+ let message = crate::document::sync::decode(&message)?;
let mut replica = doc.blocking_write();
// Authoritative check, under the same lock as the apply: the document may
@@ -390,9 +389,9 @@ pub(crate) async fn handle_sync(
peer_state.reports_ready = Some(from_ready);
let heads_before = replica.heads().to_vec();
- peer_state
- .sync
- .receive(&mut replica.doc, message)
+ let sync = &mut peer_state.sync;
+ replica
+ .logged(|doc, log| (sync.receive_logged(doc, message, log), true))
.context("Automerge sync failed")?;
peer_state.last_sync_rx = Some(Instant::now());
if let Some(wait) = &peer_state.sync.reply {
@@ -408,16 +407,14 @@ pub(crate) async fn handle_sync(
stats.heads_seen(heads_after.len());
let changed = heads_after != heads_before;
if changed && measure_propagation {
- let now_ms = clock.now_millis().as_u64() as i64;
- // Time 0: not stamped (snapshots, older nodes).
- let delays: Vec<i64> = replica
- .doc
- .get_changes(&heads_before)
- .iter()
- .map(|change| change.timestamp())
- .filter(|&time| time > 0)
- .map(|time| now_ms - time)
- .collect();
+ // Now by the peers' clocks (their median): this node's own clock skew
+ // doesn't count; the writer's still does.
+ let now_ms = clock.now_millis().as_u64() as i64 - ahead_ms;
+ let delays: Vec<i64> =
+ crate::document::changes::commit_times(&replica.doc, &heads_before)
+ .into_iter()
+ .map(|time| now_ms - time)
+ .collect();
stats.propagated(delays);
}
@@ -427,7 +424,7 @@ pub(crate) async fn handle_sync(
// catching up has them, and so does a ready peer whose document is empty
// (a misconfigured second swarm, or a genesis node in its first seconds,
// before it registered itself).
- check_caught_up && from_ready && peer_state.sync.have_all_theirs(&mut replica.doc),
+ check_caught_up && from_ready && peer_state.sync.have_all_theirs(&replica.doc),
changed,
)))
})
@@ -449,6 +446,7 @@ pub(crate) async fn handle_sync(
.stats
.caught_up(state.traffic.total().received.sync.bytes);
source::finished(state, peer);
+ state.join_progress.notify_one();
}
state.notify_store_changed();
@@ -492,7 +490,7 @@ fn fetch_delta(state: &AppState, peer: &str, epoch: transport::Epoch, spec: tran
}
}
// Our next message says what we have now; the peer answers with the rest.
- let _ = state.sync_events.send(SyncEvent::PeerNeedsSync(peer));
+ state.request_sync(&peer);
});
}
@@ -523,16 +521,9 @@ async fn download_and_apply_delta(
bail!("the epoch changed during the download");
}
let heads_before = replica.heads().to_vec();
- replica
- .doc
- .load_incremental(&bytes)
- .context("Loading the downloaded changes")?;
+ crate::document::changes::load_delta(&mut replica.doc, &bytes)?;
replica.refresh_heads();
- let missing = spec
- .to
- .iter()
- .filter(|head| replica.doc.get_change_by_hash(head).is_none())
- .count();
+ let missing = crate::document::changes::missing(&replica.doc, &spec.to).len();
if missing > 0 {
bail!("{missing} of the offered heads are still missing after loading");
}
@@ -623,87 +614,3 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
}
}
}
-
-// --- change publisher --------------------------------------------------------
-
-fn patch_store_path(patch: &automerge::Patch) -> Vec<String> {
- let mut path: Vec<String> = patch
- .path
- .iter()
- .map(|(_, prop)| match prop {
- automerge::Prop::Map(key) => key.clone(),
- automerge::Prop::Seq(index) => index.to_string(),
- })
- .collect();
-
- match &patch.action {
- automerge::PatchAction::PutMap { key, .. }
- | automerge::PatchAction::DeleteMap { key, .. } => path.push(key.clone()),
- // List, text and counter changes: the containing object changed.
- _ => {}
- }
- path
-}
-
-pub(crate) fn paths_related(a: &[String], b: &[String]) -> bool {
- a.starts_with(b) || b.starts_with(a)
-}
-
-pub(crate) async fn run_change_publisher(state: AppState) {
- let mut changed = state.change_tx.subscribe();
- let mut last_heads = state.write_doc(|doc| doc.get_heads()).await;
-
- info!("Starting change publisher");
-
- while changed.changed().await.is_ok() {
- let before = last_heads.clone();
-
- let (heads, paths, conflicts) = state
- .write_doc(move |doc| {
- let heads = doc.get_heads();
- if heads == before {
- return (heads, Vec::new(), Vec::new());
- }
-
- // After an epoch switch the previous heads belong to the old document:
- // report the root as changed, which every stream is related to.
- if !before.iter().all(|h| doc.get_change_by_hash(h).is_some()) {
- return (heads, vec![Vec::new()], Vec::new());
- }
-
- let mut paths = Vec::new();
- let mut conflicts = Vec::new();
- for patch in doc.diff(&before, &heads) {
- let path = patch_store_path(&patch);
- if matches!(
- patch.action,
- automerge::PatchAction::PutMap { conflict: true, .. }
- ) {
- conflicts.push(path.clone());
- }
- paths.push(path);
- }
- (heads, paths, conflicts)
- })
- .await;
-
- if heads == last_heads {
- continue;
- }
- last_heads = heads;
-
- for path in &conflicts {
- warn!(path = ?path, "Concurrent writes produced a conflict");
- }
-
- if paths
- .iter()
- .any(|path| crate::swarm::topology::affects_neighbours(path))
- {
- state.topology_changed.notify_one();
- }
-
- // Fails only when nobody is subscribed.
- let _ = state.change_events.send(Arc::new(ChangeSet { paths }));
- }
-}
diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
deleted file mode 100644
index 1d98637..0000000
--- a/src/swarm/topology.rs
+++ /dev/null
@@ -1,843 +0,0 @@
-//! Which members a node syncs with: its neighbours.
-//!
-//! In a full mesh every change reaches every node from every other node (Automerge
-//! decides per pair, so nobody knows that a third peer already passed it on), so the
-//! copies of one write grow with N². Above `k` members, each node keeps only a few
-//! neighbours, and changes travel over several hops.
-//!
-//! Every node computes the graph from the member list and which members it can reach,
-//! with no coordination:
-//! * a ring ordered by key hash: each member's nearest reachable member in both
-//! directions, which keeps the reachable members connected;
-//! * plus, for each member, the `k - 2` reachable others with the lowest pair score (a
-//! hash of both keys, so both ends agree on it), which keeps paths short.
-//!
-//! A link exists if either end chose it, so links are symmetric and a node has at least
-//! `k` neighbours. With at most `k + 1` reachable members the graph is the full mesh:
-//! small swarms behave exactly as before. A member joining, leaving or going offline
-//! changes only links near it.
-//!
-//! Members on several transports (Mycelium, libp2p) get such a graph per transport,
-//! among the members on it (as their addresses in the member list say), and a node's
-//! neighbours are the union of its neighbours on each transport it is on
-//! (`neighbours_across`). Each of those graphs is connected (its ring), and a member on
-//! two transports is in both: as long as members on several transports link the
-//! transports, the swarm stays connected, and such a member has about `k` neighbours per
-//! transport. With one transport, it is the one graph as before.
-
-use std::collections::{BTreeMap, BTreeSet};
-
-use sha2::{Digest, Sha256};
-use tracing::info;
-
-use crate::locks::RwLockExt;
-use crate::state::AppState;
-use crate::swarm::members;
-
-/// This node's current neighbours, and the member list they were computed from.
-#[derive(Debug, Default)]
-pub(crate) struct NeighbourView {
- members: Vec<String>,
- k: usize,
- unreachable: BTreeSet<String>,
- stable: BTreeSet<String>,
- /// The transports (address schemes) of each member, and ours.
- transports: BTreeMap<String, BTreeSet<String>>,
- local_transports: BTreeSet<String>,
- /// Peers that proved they are of our swarm but aren't in our member list (yet): their
- /// registration hasn't reached us. They are neighbours too (`refresh`).
- unlisted: BTreeSet<String>,
- neighbours: BTreeSet<String>,
- /// Every reachable member is a neighbour.
- full_mesh: bool,
-}
-
-/// Whether the neighbours are computed from the member list yet (before, everyone
-/// counts as a neighbour).
-pub(crate) fn view_known(state: &AppState) -> bool {
- state.neighbours.read_or_recover().is_some()
-}
-
-/// Whether every reachable member is our neighbour (also while the view is unknown).
-pub(crate) fn is_full_mesh(state: &AppState) -> bool {
- state
- .neighbours
- .read_or_recover()
- .as_ref()
- .is_none_or(|view| view.full_mesh)
-}
-
-/// The current view in numbers, once known.
-#[derive(Debug, Clone, Copy)]
-pub(crate) struct Summary {
- /// Registered members, us included.
- pub(crate) members: usize,
- /// Members other than us not currently judged unreachable.
- pub(crate) reachable: usize,
- pub(crate) stable: usize,
- pub(crate) neighbours: usize,
- pub(crate) stable_neighbours: usize,
- pub(crate) full_mesh: bool,
-}
-
-pub(crate) fn summary(state: &AppState) -> Option<Summary> {
- let local = state.local_id();
- state
- .neighbours
- .read_or_recover()
- .as_ref()
- .map(|view| Summary {
- members: view.members.len(),
- reachable: view
- .members
- .iter()
- .filter(|m| m.as_str() != local && !view.unreachable.contains(*m))
- .count(),
- stable: view.stable.len(),
- neighbours: view.neighbours.len(),
- stable_neighbours: view
- .neighbours
- .iter()
- .filter(|m| view.stable.contains(*m))
- .count(),
- full_mesh: view.full_mesh,
- })
-}
-
-/// Whether we initiate contact with `peer`: HELLOs, keepalives, dormancy checks, sync
-/// openings. Everyone is a neighbour until the view is known (while joining, or before
-/// any member registered), so joining works as in a full mesh. We always *answer*
-/// anyone who contacts us.
-pub(crate) fn is_neighbour(state: &AppState, peer: &str) -> bool {
- match &*state.neighbours.read_or_recover() {
- Some(view) => view.neighbours.contains(peer),
- None => true,
- }
-}
-
-/// How the neighbours changed.
-pub(crate) struct NeighbourChange {
- pub(crate) added: Vec<String>,
- pub(crate) removed: Vec<String>,
-}
-
-/// Recomputes the neighbours whenever something they depend on may have changed
-/// (`AppState::topology_changed`): the member list (`swarm::sync::run_change_publisher`),
-/// a peer becoming unreachable or reachable again (`swarm::peer`), and every 5 s as a
-/// fallback. A burst of notifications coalesces into one recomputation.
-pub(crate) async fn run_worker(state: AppState) {
- loop {
- state.topology_changed.notified().await;
- if let Some(change) = refresh(&state).await {
- crate::swarm::peer::apply_neighbour_change(&state, change).await;
- }
- }
-}
-
-/// Whether a change at `path` may change the neighbours: one in the member list, or the
-/// root (the whole document replaced, as after an epoch switch).
-pub(crate) fn affects_neighbours(path: &[String]) -> bool {
- path.is_empty() || path.starts_with(&members::members_dir())
-}
-
-/// Hands the members' `aliases` to the logger, under every name that stands for each:
-/// its node ID, and its addresses (`logging`, which renames them bare too).
-fn set_aliases(state: &AppState, aliases: std::collections::HashMap<String, String>) {
- let mut names = std::collections::HashMap::new();
- for (member, alias) in aliases {
- for address in state.transport.identities_of(&member) {
- names.insert(address, alias.clone());
- }
- names.insert(member, alias);
- }
- crate::logging::set_aliases(names);
-}
-
-/// Recomputes the neighbours once ready, if the member list, the reachable members or
-/// `k` changed.
-pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
- if !state.is_ready().await {
- return None;
- }
- let (current, stable, transports, alive) = {
- let replica = state.doc.read().await;
- members::learn_addresses(state, &replica.doc);
- set_aliases(state, members::aliases(&replica.doc));
- state.transport.learn_relays(&members::relays(&replica.doc));
- let current = members::members(&replica.doc);
- // Those that said they are alive within the electorate window.
- let alive: BTreeSet<String> = members::electorate(
- &replica.doc,
- state.clock.now_millis().0,
- state.electorate_window,
- )
- .into_iter()
- .collect();
- let stable: BTreeSet<String> = current
- .iter()
- .filter(|m| members::tier(&replica.doc, m) == members::Tier::Stable)
- .cloned()
- .collect();
- let transports: BTreeMap<String, BTreeSet<String>> = current
- .iter()
- .map(|m| (m.clone(), schemes(&members::addresses(&replica.doc, m))))
- .collect();
- (current, stable, transports, alive)
- };
- let local_transports: BTreeSet<String> = state.transport.transports().into_iter().collect();
- let k = state.neighbours_k;
- // Those that don't answer, those on none of our transports, and those that haven't
- // said they are alive for longer than the electorate window, unless we heard from
- // them since we started (one that comes back says so to us first: its HELLO).
- let mut unreachable = BTreeSet::new();
- for member in &current {
- if member == state.local_id() {
- continue;
- }
- if !state.transport.can_reach(member) {
- unreachable.insert(member.clone());
- continue;
- }
- let (gone, heard) = match state.get_peer_state(member).await {
- Some(peer_state) => {
- let peer_state = peer_state.lock().await;
- (peer_state.is_unreachable(), peer_state.last_seen.is_some())
- }
- None => (false, false),
- };
- if gone || (!heard && !alive.contains(member)) {
- unreachable.insert(member.clone());
- }
- }
- // A node that joined while the genesis node hadn't registered yet has the genesis
- // document without the genesis node in it, and registers itself there: then neither
- // lists the other, and with each the other's neighbour only by the member list,
- // neither would ever sync with the other, and their registrations never pass. So a
- // peer that proved itself a member of our swarm counts until the list says otherwise:
- // syncing brings each the other's registration.
- let mut unlisted = BTreeSet::new();
- for (peer, peer_state) in state.peers_snapshot().await {
- if current.contains(&peer) || peer == state.local_id() {
- continue;
- }
- let peer_state = peer_state.lock().await;
- if matches!(peer_state.standing, crate::state::Standing::Member { .. })
- && !peer_state.is_unreachable()
- {
- unlisted.insert(peer);
- }
- }
- let previous = {
- let view = state.neighbours.read_or_recover();
- if view.as_ref().is_some_and(|view| {
- view.members == current
- && view.unlisted == unlisted
- && view.k == k
- && view.unreachable == unreachable
- && view.stable == stable
- && view.transports == transports
- && view.local_transports == local_transports
- }) {
- return None;
- }
- view.as_ref().map(|view| view.neighbours.clone())
- };
-
- // Before the first view everyone known counted as a neighbour.
- let previous = match previous {
- Some(previous) => previous,
- None => state.peer_states.read().await.keys().cloned().collect(),
- };
- let mut computed = neighbours_across(
- &transports,
- state.local_id(),
- &local_transports,
- k,
- &|m| unreachable.contains(m),
- &|m| stable.contains(m),
- );
- let listed_neighbours = computed.len();
- computed.extend(unlisted.iter().cloned());
- let change = NeighbourChange {
- added: computed
- .iter()
- .filter(|m| !previous.contains(*m))
- .cloned()
- .collect(),
- removed: previous
- .iter()
- .filter(|m| !computed.contains(*m))
- .cloned()
- .collect(),
- };
- info!(
- members = current.len(),
- unreachable = unreachable.len(),
- stable = stable.len(),
- k,
- neighbours = computed.len(),
- unlisted = unlisted.len(),
- added = change.added.len(),
- removed = change.removed.len(),
- "Neighbours computed from the member list"
- );
- let reachable_others = current
- .iter()
- .filter(|m| m.as_str() != state.local_id() && !unreachable.contains(*m))
- .count();
- *state.neighbours.write_or_recover() = Some(NeighbourView {
- full_mesh: listed_neighbours >= reachable_others,
- members: current,
- k,
- unreachable,
- stable,
- transports,
- local_transports,
- unlisted,
- neighbours: computed,
- });
- Some(change)
-}
-
-const RING_LABEL: &[u8] = b"continuum/topology-ring/1";
-const PAIR_LABEL: &[u8] = b"continuum/topology-pair/2";
-
-/// How many members may prefer the same stable member, as a multiple of `k`. Without a
-/// bound, 20 stable members among 2000 get about 500 neighbours each: every
-/// announcement in the swarm lands on them, and the backbone saturates first. Members
-/// that find a stable member full prefer the next one, then go by pair score, so the
-/// graph stays connected and symmetric.
-const STABLE_FANIN_FACTOR: usize = 4;
-
-/// The transports `addresses` are on: their schemes (`mycelium` of `mycelium:<key>`);
-/// the empty one for an address without (a transport whose identities are addresses).
-pub(crate) fn schemes(addresses: &[String]) -> BTreeSet<String> {
- addresses
- .iter()
- .map(|address| {
- address
- .split_once(':')
- .map_or("", |(scheme, _)| scheme)
- .to_owned()
- })
- .collect()
-}
-
-/// The neighbours of `local` among `members` (by the transports each is on), on each of
-/// `local_transports`: on each, its neighbours among the members on it
-/// (`neighbours`), all together.
-pub(crate) fn neighbours_across(
- members: &BTreeMap<String, BTreeSet<String>>,
- local: &str,
- local_transports: &BTreeSet<String>,
- k: usize,
- unreachable: &dyn Fn(&str) -> bool,
- stable: &dyn Fn(&str) -> bool,
-) -> BTreeSet<String> {
- let mut all = BTreeSet::new();
- for transport in local_transports {
- let on_it: Vec<String> = members
- .iter()
- .filter(|(member, transports)| *member != local && transports.contains(transport))
- .map(|(member, _)| member.clone())
- .collect();
- all.extend(neighbours(&on_it, local, k, unreachable, stable));
- }
- all
-}
-
-/// The neighbours of `local` among `members` (which may or may not contain `local`),
-/// skipping members for which `unreachable` holds.
-pub(crate) fn neighbours(
- members: &[String],
- local: &str,
- k: usize,
- unreachable: &dyn Fn(&str) -> bool,
- stable: &dyn Fn(&str) -> bool,
-) -> BTreeSet<String> {
- let mut all: Vec<String> = members.to_vec();
- all.push(local.to_owned());
- graph(&all, k, &|m: &str| m != local && unreachable(m), stable)
- .remove(local)
- .unwrap_or_default()
-}
-
-/// Every reachable member's neighbours. Hashes each member once and scores pairs by
-/// mixing the two hashes (`pair_score`), so the N² scores are a few nanoseconds each:
-/// tens of milliseconds for thousands of members. `stable` members (`swarm::stability`)
-/// are preferred for half of each member's scored slots, each by at most
-/// `STABLE_FANIN_FACTOR * k` members.
-///
-/// Each reachable member walks the ring (by key hash) both ways to the first reachable
-/// member, which gives a ring over the reachable members, and adds the `k - 2`
-/// reachable members with the lowest pair score. A link exists if either end chose it.
-/// With at most `k + 1` reachable members, the reachable members form a full mesh.
-///
-/// Each node judges reachability by itself, so nodes may briefly disagree; that only
-/// adds links (whoever chose a link initiates, the other answers), it never cuts one.
-/// The preference budget is spent in key order, which every node shares, so nodes that
-/// agree on reachability compute the same links.
-pub(crate) fn graph(
- members: &[String],
- k: usize,
- unreachable: &dyn Fn(&str) -> bool,
- stable: &dyn Fn(&str) -> bool,
-) -> BTreeMap<String, BTreeSet<String>> {
- let k = k.max(2);
- let all: BTreeSet<&str> = members.iter().map(String::as_str).collect();
- let live: BTreeSet<&str> = all.iter().copied().filter(|m| !unreachable(m)).collect();
- let mut links: BTreeMap<String, BTreeSet<String>> = live
- .iter()
- .map(|m| ((*m).to_owned(), BTreeSet::new()))
- .collect();
-
- let mut link = |a: &str, b: &str| {
- if a != b {
- links.get_mut(a).expect("live member").insert(b.to_owned());
- links.get_mut(b).expect("live member").insert(a.to_owned());
- }
- };
-
- if live.len() <= k + 1 {
- for a in &live {
- for b in &live {
- link(a, b);
- }
- }
- return links;
- }
-
- // The ring over all members, so its order doesn't change when one goes offline.
- let mut ring: Vec<([u8; 32], &str)> =
- all.iter().map(|m| (hash(RING_LABEL, &[*m]), *m)).collect();
- ring.sort_unstable();
- let n = ring.len();
- let position: BTreeMap<&str, usize> =
- ring.iter().enumerate().map(|(i, (_, m))| (*m, i)).collect();
- let first_live = |from: usize, step: usize| {
- (1..n)
- .map(|d| ring[(from + d * step) % n].1)
- .find(|m| live.contains(m))
- .expect("more than k + 1 live members")
- };
-
- // Live members in key order, with their pair keys; and those of them that are stable.
- let keyed: Vec<(&str, u64)> = live.iter().map(|m| (*m, pair_key(m))).collect();
- let stable_keyed: Vec<(&str, u64)> = keyed.iter().copied().filter(|(m, _)| stable(m)).collect();
-
- // How many members have preferred each stable member so far (`keyed` is in key
- // order, the same on every node).
- let budget = STABLE_FANIN_FACTOR * k;
- let mut preferred_by: BTreeMap<&str, usize> = BTreeMap::new();
- // Half the slots (rounded up) prefer stable members, so long-running nodes form a
- // backbone; the rest go by pair score alone, keeping paths short. A stable member
- // enough others prefer already is skipped, so no hub grows with the swarm.
- let stable_slots = (k - 2).div_ceil(2);
-
- for &(member, key) in &keyed {
- let at = position[member];
- let next = first_live(at, 1);
- let prev = first_live(at, n - 1);
- link(member, next);
- link(member, prev);
- let skip = [member, next, prev];
- // The stable candidates in score order (there are few).
- let mut stable_candidates = scored(&stable_keyed, key, skip);
- stable_candidates.sort_unstable();
- let preferred: Vec<&str> = stable_candidates
- .iter()
- .map(|(_, other)| *other)
- .filter(|other| preferred_by.get(other).copied().unwrap_or(0) < budget)
- .take(stable_slots)
- .collect();
- for other in preferred.iter().copied() {
- *preferred_by.entry(other).or_default() += 1;
- }
- // The rest: the lowest scores among the others that aren't preferred, which are
- // among the `need + preferred` lowest overall. Selecting those is linear; only
- // they are sorted.
- let need = k - 2 - preferred.len();
- let mut candidates = scored(&keyed, key, skip);
- let lowest = (need + preferred.len()).min(candidates.len());
- if lowest < candidates.len() {
- candidates.select_nth_unstable(lowest);
- candidates.truncate(lowest);
- }
- candidates.sort_unstable();
- let rest = candidates
- .iter()
- .map(|(_, other)| *other)
- .filter(|other| !preferred.contains(other))
- .take(need);
- for partner in preferred.iter().copied().chain(rest) {
- link(member, partner);
- }
- }
- links
-}
-
-/// `others` but those in `skip`, scored as pairs with the member whose key is `key`.
-fn scored<'a>(others: &[(&'a str, u64)], key: u64, skip: [&str; 3]) -> Vec<(u64, &'a str)> {
- others
- .iter()
- .filter(|(other, _)| !skip.contains(other))
- .map(|&(other, other_key)| (mix_pair(key, other_key), other))
- .collect()
-}
-
-/// Symmetric: the same for (a, b) and (b, a).
-pub(crate) fn pair_score(a: &str, b: &str) -> u64 {
- mix_pair(pair_key(a), pair_key(b))
-}
-
-/// What a member contributes to its pair scores: computed once per member.
-fn pair_key(member: &str) -> u64 {
- let digest = hash(PAIR_LABEL, &[member]);
- u64::from_be_bytes(digest[..8].try_into().expect("8 bytes"))
-}
-
-/// The score of a pair from its members' keys, symmetric: SplitMix64's finalizer over
-/// the lower key and the mixed higher one. Keys are uniform (SHA-256), so for each member
-/// the others' scores are in an order unrelated to anyone else's.
-fn mix_pair(a: u64, b: u64) -> u64 {
- let (low, high) = if a <= b { (a, b) } else { (b, a) };
- mix(low ^ mix(high))
-}
-
-fn mix(mut z: u64) -> u64 {
- z = z.wrapping_add(0x9e37_79b9_7f4a_7c15);
- z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
- z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
- z ^ (z >> 31)
-}
-
-fn hash(label: &[u8], fields: &[&str]) -> [u8; 32] {
- let mut hasher = Sha256::new();
- for field in [label]
- .into_iter()
- .chain(fields.iter().map(|f| f.as_bytes()))
- {
- hasher.update((field.len() as u32).to_be_bytes());
- hasher.update(field);
- }
- hasher.finalize().into()
-}
-
-#[cfg(test)]
-mod tests {
- use super::*;
- use std::collections::VecDeque;
-
- fn members(n: usize) -> Vec<String> {
- (0..n).map(|i| format!("{:064x}", i * 7919 + 13)).collect()
- }
-
- fn all_reachable(_: &str) -> bool {
- false
- }
-
- fn none_stable(_: &str) -> bool {
- false
- }
-
- fn graph(members: &[String], k: usize) -> BTreeMap<String, BTreeSet<String>> {
- super::graph(members, k, &all_reachable, &none_stable)
- }
-
- fn connected(g: &BTreeMap<String, BTreeSet<String>>) -> bool {
- let Some(start) = g.keys().next() else {
- return true;
- };
- let mut seen = BTreeSet::from([start.clone()]);
- let mut queue = VecDeque::from([start.clone()]);
- while let Some(m) = queue.pop_front() {
- for next in &g[&m] {
- if seen.insert(next.clone()) {
- queue.push_back(next.clone());
- }
- }
- }
- seen.len() == g.len()
- }
-
- fn edges(graph: &BTreeMap<String, BTreeSet<String>>) -> BTreeSet<(String, String)> {
- graph
- .iter()
- .flat_map(|(a, ns)| {
- ns.iter().map(move |b| {
- if a < b {
- (a.clone(), b.clone())
- } else {
- (b.clone(), a.clone())
- }
- })
- })
- .collect()
- }
-
- fn hex(bytes: [u8; 32]) -> String {
- bytes.iter().map(|byte| format!("{byte:02x}")).collect()
- }
-
- /// Over random swarms (size, `k`, who is reachable, who is stable): links only
- /// between reachable members, never to oneself, symmetric; a full mesh while small,
- /// otherwise at least `k` links each and connected; independent of the order members
- /// are listed in; and each member's own view (`neighbours`) agrees with the graph.
- #[test]
- fn random_swarms() {
- crate::testrand::for_seeds(300, |rng| {
- let n = rng.between(1, 60);
- let k = rng.between(2, 14);
- let mut all: Vec<String> = (0..n).map(|_| hex(rng.bytes32())).collect();
- let unreachable: BTreeSet<String> =
- all.iter().filter(|_| rng.chance(0.3)).cloned().collect();
- let stable: BTreeSet<String> =
- all.iter().filter(|_| rng.chance(0.3)).cloned().collect();
- let is_unreachable = |m: &str| unreachable.contains(m);
- let is_stable = |m: &str| stable.contains(m);
-
- let g = super::graph(&all, k, &is_unreachable, &is_stable);
- // Owned: `all` is shuffled below.
- let live: BTreeSet<String> = all
- .iter()
- .filter(|m| !unreachable.contains(*m))
- .cloned()
- .collect();
- assert_eq!(g.keys().cloned().collect::<BTreeSet<_>>(), live);
- for (a, ns) in &g {
- assert!(!ns.contains(a), "{a} links to itself");
- for b in ns {
- assert!(live.contains(b), "{a} links to unreachable {b}");
- assert!(g[b].contains(a), "{a} -> {b} but not back");
- }
- }
- if live.len() <= k + 1 {
- assert!(
- g.values().all(|ns| ns.len() + 1 == live.len()),
- "not a full mesh"
- );
- } else {
- assert!(
- g.values().all(|ns| ns.len() >= k),
- "a member with fewer than k links"
- );
- assert!(connected(&g), "not connected");
- }
-
- rng.shuffle(&mut all);
- assert_eq!(super::graph(&all, k, &is_unreachable, &is_stable), g);
-
- if let Some(local) = live.iter().nth(rng.below(live.len())) {
- let others: Vec<String> = all.iter().filter(|m| *m != local).cloned().collect();
- assert_eq!(
- neighbours(&others, local, k, &is_unreachable, &is_stable),
- g[local]
- );
- }
- });
- }
-
- #[test]
- fn small_swarms_are_a_full_mesh() {
- let ms = members(13);
- for m in &ms {
- assert_eq!(
- neighbours(&ms, m, 12, &all_reachable, &none_stable).len(),
- 12
- );
- }
- }
-
- #[test]
- fn links_are_symmetric_and_degree_at_least_k() {
- let ms = members(250);
- let g = graph(&ms, 8);
- for (a, ns) in &g {
- assert!(ns.len() >= 8, "{a} has {} neighbours", ns.len());
- assert!(!ns.contains(a));
- for b in ns {
- assert!(g[b].contains(a), "{a} -> {b} not symmetric");
- }
- }
- }
-
- #[test]
- fn graph_is_connected() {
- for (n, k) in [(250, 8), (250, 2), (60, 3)] {
- assert!(connected(&graph(&members(n), k)), "n={n} k={k}");
- }
- }
-
- #[test]
- fn one_member_joining_changes_few_links() {
- let ms = members(200);
- let before = edges(&graph(&ms, 8));
- let mut more = ms.clone();
- more.push(format!("{:064x}", 999_999));
- let after = edges(&graph(&more, 8));
- let changed = before.symmetric_difference(&after).count();
- assert!(changed <= 4 * 8, "{changed} links changed");
- }
-
- #[test]
- fn reachable_members_stay_connected_when_most_are_offline() {
- let ms = members(100);
- // Every tenth member is online.
- let online: BTreeSet<String> = ms.iter().step_by(10).cloned().collect();
- let offline = |m: &str| !online.contains(m);
- let g = super::graph(&ms, 12, &offline, &none_stable);
- assert_eq!(g.len(), 10);
- // Ten reachable members and k = 12: a full mesh among them.
- assert!(g.values().all(|ns| ns.len() == 9));
-
- let online: BTreeSet<String> = ms.iter().step_by(4).cloned().collect();
- let offline = |m: &str| !online.contains(m);
- let g = super::graph(&ms, 8, &offline, &none_stable);
- assert_eq!(g.len(), 25);
- assert!(connected(&g));
- assert!(
- g.values()
- .all(|ns| ns.len() >= 8 && ns.iter().all(|n| online.contains(n)))
- );
- }
-
- #[test]
- fn stable_members_become_hubs_with_bounded_degree() {
- let ms = members(250);
- let stable: BTreeSet<String> = ms.iter().step_by(12).take(20).cloned().collect();
- let is_stable = |m: &str| stable.contains(m);
- let g = super::graph(&ms, 12, &all_reachable, &is_stable);
- assert!(connected(&g));
-
- let degree = |pred: &dyn Fn(&String) -> bool| -> (usize, f64) {
- let degrees: Vec<usize> = g
- .iter()
- .filter(|(m, _)| pred(m))
- .map(|(_, ns)| ns.len())
- .collect();
- let max = *degrees.iter().max().unwrap();
- (
- max,
- degrees.iter().sum::<usize>() as f64 / degrees.len() as f64,
- )
- };
- let (stable_max, stable_mean) = degree(&|m| stable.contains(m));
- let (_, transient_mean) = degree(&|m| !stable.contains(m));
- // 230 transient members put about 5 of their 10 scored slots on 20 stable ones.
- assert!(
- stable_mean > 3.0 * transient_mean,
- "{stable_mean} vs {transient_mean}"
- );
- assert!(
- stable_max <= 100,
- "a stable member has {stable_max} neighbours"
- );
- }
-
- /// A hub's degree in a swarm of `n` with `stable_count` stable members stays within
- /// the bound.
- fn hub_degree_is_bounded(n: usize, stable_count: usize) {
- // Few stable members among many: without the fan-in budget each would be chosen
- // by most of the swarm (about 130 neighbours with 10 of 250). With it, a hub has
- // at most the budget plus the links others choose by pair score and the ring.
- let k = 12;
- let ms = members(n);
- let stable: BTreeSet<String> = ms.iter().step_by(n / stable_count).cloned().collect();
- let is_stable = |m: &str| stable.contains(m);
- let g = super::graph(&ms, k, &all_reachable, &is_stable);
- assert!(connected(&g), "n={n} stable={stable_count}");
- let hub_max = g
- .iter()
- .filter(|(m, _)| stable.contains(*m))
- .map(|(_, ns)| ns.len())
- .max()
- .unwrap();
- assert!(
- hub_max <= 8 * k,
- "n={n} stable={stable_count}: a stable member has {hub_max} neighbours"
- );
- }
-
- /// Two transports with members on one or the other, and two members on both: every
- /// link is over a shared transport, and the two members on both connect the rest.
- #[test]
- fn members_on_two_transports_bridge_the_others() {
- let k = 8;
- let ms = members(201);
- let on = |names: &[&str]| -> BTreeSet<String> {
- names.iter().map(|name| (*name).to_owned()).collect()
- };
- let transports: BTreeMap<String, BTreeSet<String>> = ms
- .iter()
- .enumerate()
- .map(|(i, m)| {
- let transports = match i {
- 0 | 1 => on(&["a", "b"]),
- i if i % 2 == 0 => on(&["a"]),
- _ => on(&["b"]),
- };
- (m.clone(), transports)
- })
- .collect();
- let no_stable = |_: &str| false;
- let mut graph: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
- for (m, own) in &transports {
- for n in neighbours_across(&transports, m, own, k, &all_reachable, &no_stable) {
- assert!(
- !transports[&n].is_disjoint(own),
- "{m} and {n} share no transport"
- );
- graph.entry(m.clone()).or_default().insert(n.clone());
- graph.entry(n).or_default().insert(m.clone());
- }
- }
- assert_eq!(graph.len(), ms.len());
- assert!(connected(&graph));
- let bridge = &ms[0];
- assert!(graph[bridge].iter().any(|n| !transports[n].contains("b")));
- assert!(graph[bridge].iter().any(|n| !transports[n].contains("a")));
-
- // One transport: the one graph, as without transports.
- let one: BTreeMap<String, BTreeSet<String>> =
- ms.iter().map(|m| (m.clone(), on(&[""]))).collect();
- for m in &ms[..5] {
- assert_eq!(
- neighbours_across(&one, m, &on(&[""]), k, &all_reachable, &no_stable),
- neighbours(&ms, m, k, &all_reachable, &no_stable)
- );
- }
- }
-
- #[test]
- fn schemes_of_addresses() {
- let addresses =
- |all: &[&str]| -> Vec<String> { all.iter().map(|a| (*a).to_owned()).collect() };
- assert_eq!(
- schemes(&addresses(&[
- "mycelium:ab",
- "libp2p:/ip4/1.2.3.4/tcp/1",
- "libp2p:/ip6/::1/tcp/1"
- ])),
- BTreeSet::from(["libp2p".to_owned(), "mycelium".to_owned()])
- );
- assert_eq!(schemes(&addresses(&["a"])), BTreeSet::from([String::new()]));
- assert!(schemes(&[]).is_empty());
- }
-
- #[test]
- fn hub_degree_is_bounded_however_few_stable_members() {
- for (n, stable_count) in [(250, 10), (400, 5)] {
- hub_degree_is_bounded(n, stable_count);
- }
- }
-
- /// The same at the sizes the bound was designed for; slow in debug builds, so run
- /// on request (`cargo test --release -- --ignored`).
- #[test]
- #[ignore]
- fn hub_degree_is_bounded_in_large_swarms() {
- for (n, stable_count) in [(1000, 10), (1000, 20), (2000, 20)] {
- hub_degree_is_bounded(n, stable_count);
- }
- }
-}
diff --git a/src/swarm/topology/graph.rs b/src/swarm/topology/graph.rs
new file mode 100644
index 0000000..fb33170
--- /dev/null
+++ b/src/swarm/topology/graph.rs
@@ -0,0 +1,236 @@
+//! The graph itself, from the members alone: a ring by hash, plus each member's
+//! lowest-scoring pairs, stable members preferred; per transport, and across them. Pure:
+//! the same members give every node the same graph.
+
+use sha2::{Digest, Sha256};
+
+use super::*;
+
+pub(super) const RING_LABEL: &[u8] = b"continuum/topology-ring/1";
+
+pub(super) const PAIR_LABEL: &[u8] = b"continuum/topology-pair/2";
+
+/// How many members may prefer the same stable member, as a multiple of `k`. Without a
+/// bound, 20 stable members among 2000 get about 500 neighbours each: every
+/// announcement in the swarm lands on them, and the backbone saturates first. Members
+/// that find a stable member full prefer the next one, then go by pair score, so the
+/// graph stays connected and symmetric.
+pub(super) const STABLE_FANIN_FACTOR: usize = 4;
+
+/// The transports `addresses` are on: their schemes (`mycelium` of `mycelium:<key>`);
+/// the empty one for an address without (a transport whose identities are addresses).
+pub(crate) fn schemes(addresses: &[String]) -> BTreeSet<String> {
+ addresses
+ .iter()
+ .map(|address| {
+ address
+ .split_once(':')
+ .map_or("", |(scheme, _)| scheme)
+ .to_owned()
+ })
+ .collect()
+}
+
+/// The neighbours of `local` among `members` (by the transports each is on), on each of
+/// `local_transports`: on each, its neighbours among the members on it
+/// (`neighbours`), all together.
+pub(crate) fn neighbours_across(
+ members: &BTreeMap<String, BTreeSet<String>>,
+ local: &str,
+ local_transports: &BTreeSet<String>,
+ k: usize,
+ unreachable: &dyn Fn(&str) -> bool,
+ stable: &dyn Fn(&str) -> bool,
+) -> BTreeSet<String> {
+ let mut all = BTreeSet::new();
+ for transport in local_transports {
+ let on_it: Vec<String> = members
+ .iter()
+ .filter(|(member, transports)| *member != local && transports.contains(transport))
+ .map(|(member, _)| member.clone())
+ .collect();
+ all.extend(neighbours(&on_it, local, k, unreachable, stable));
+ }
+ all
+}
+
+/// The neighbours of `local` among `members` (which may or may not contain `local`),
+/// skipping members for which `unreachable` holds.
+pub(crate) fn neighbours(
+ members: &[String],
+ local: &str,
+ k: usize,
+ unreachable: &dyn Fn(&str) -> bool,
+ stable: &dyn Fn(&str) -> bool,
+) -> BTreeSet<String> {
+ let mut all: Vec<String> = members.to_vec();
+ all.push(local.to_owned());
+ graph(&all, k, &|m: &str| m != local && unreachable(m), stable)
+ .remove(local)
+ .unwrap_or_default()
+}
+
+/// Every reachable member's neighbours. Hashes each member once and scores pairs by
+/// mixing the two hashes (`pair_score`), so the N² scores are a few nanoseconds each:
+/// tens of milliseconds for thousands of members. `stable` members (`swarm::stability`)
+/// are preferred for half of each member's scored slots, each by at most
+/// `STABLE_FANIN_FACTOR * k` members.
+///
+/// Each reachable member walks the ring (by key hash) both ways to the first reachable
+/// member, which gives a ring over the reachable members, and adds the `k - 2`
+/// reachable members with the lowest pair score. A link exists if either end chose it.
+/// With at most `k + 1` reachable members, the reachable members form a full mesh.
+///
+/// Each node judges reachability by itself, so nodes may briefly disagree; that only
+/// adds links (whoever chose a link initiates, the other answers), it never cuts one.
+/// The preference budget is spent in key order, which every node shares, so nodes that
+/// agree on reachability compute the same links.
+pub(crate) fn graph(
+ members: &[String],
+ k: usize,
+ unreachable: &dyn Fn(&str) -> bool,
+ stable: &dyn Fn(&str) -> bool,
+) -> BTreeMap<String, BTreeSet<String>> {
+ let k = k.max(2);
+ let all: BTreeSet<&str> = members.iter().map(String::as_str).collect();
+ let live: BTreeSet<&str> = all.iter().copied().filter(|m| !unreachable(m)).collect();
+ let mut links: BTreeMap<String, BTreeSet<String>> = live
+ .iter()
+ .map(|m| ((*m).to_owned(), BTreeSet::new()))
+ .collect();
+
+ let mut link = |a: &str, b: &str| {
+ if a != b {
+ links.get_mut(a).expect("live member").insert(b.to_owned());
+ links.get_mut(b).expect("live member").insert(a.to_owned());
+ }
+ };
+
+ if live.len() <= k + 1 {
+ for a in &live {
+ for b in &live {
+ link(a, b);
+ }
+ }
+ return links;
+ }
+
+ // The ring over all members, so its order doesn't change when one goes offline.
+ let mut ring: Vec<([u8; 32], &str)> =
+ all.iter().map(|m| (hash(RING_LABEL, &[*m]), *m)).collect();
+ ring.sort_unstable();
+ let n = ring.len();
+ let position: BTreeMap<&str, usize> =
+ ring.iter().enumerate().map(|(i, (_, m))| (*m, i)).collect();
+ let first_live = |from: usize, step: usize| {
+ (1..n)
+ .map(|d| ring[(from + d * step) % n].1)
+ .find(|m| live.contains(m))
+ .expect("more than k + 1 live members")
+ };
+
+ // Live members in key order, with their pair keys; and those of them that are stable.
+ let keyed: Vec<(&str, u64)> = live.iter().map(|m| (*m, pair_key(m))).collect();
+ let stable_keyed: Vec<(&str, u64)> = keyed.iter().copied().filter(|(m, _)| stable(m)).collect();
+
+ // How many members have preferred each stable member so far (`keyed` is in key
+ // order, the same on every node).
+ let budget = STABLE_FANIN_FACTOR * k;
+ let mut preferred_by: BTreeMap<&str, usize> = BTreeMap::new();
+ // Half the slots (rounded up) prefer stable members, so long-running nodes form a
+ // backbone; the rest go by pair score alone, keeping paths short. A stable member
+ // enough others prefer already is skipped, so no hub grows with the swarm.
+ let stable_slots = (k - 2).div_ceil(2);
+
+ for &(member, key) in &keyed {
+ let at = position[member];
+ let next = first_live(at, 1);
+ let prev = first_live(at, n - 1);
+ link(member, next);
+ link(member, prev);
+ let skip = [member, next, prev];
+ // The stable candidates in score order (there are few).
+ let mut stable_candidates = scored(&stable_keyed, key, skip);
+ stable_candidates.sort_unstable();
+ let preferred: Vec<&str> = stable_candidates
+ .iter()
+ .map(|(_, other)| *other)
+ .filter(|other| preferred_by.get(other).copied().unwrap_or(0) < budget)
+ .take(stable_slots)
+ .collect();
+ for other in preferred.iter().copied() {
+ *preferred_by.entry(other).or_default() += 1;
+ }
+ // The rest: the lowest scores among the others that aren't preferred, which are
+ // among the `need + preferred` lowest overall. Selecting those is linear; only
+ // they are sorted.
+ let need = k - 2 - preferred.len();
+ let mut candidates = scored(&keyed, key, skip);
+ let lowest = (need + preferred.len()).min(candidates.len());
+ if lowest < candidates.len() {
+ candidates.select_nth_unstable(lowest);
+ candidates.truncate(lowest);
+ }
+ candidates.sort_unstable();
+ let rest = candidates
+ .iter()
+ .map(|(_, other)| *other)
+ .filter(|other| !preferred.contains(other))
+ .take(need);
+ for partner in preferred.iter().copied().chain(rest) {
+ link(member, partner);
+ }
+ }
+ links
+}
+
+/// `others` but those in `skip`, scored as pairs with the member whose key is `key`.
+pub(super) fn scored<'a>(
+ others: &[(&'a str, u64)],
+ key: u64,
+ skip: [&str; 3],
+) -> Vec<(u64, &'a str)> {
+ others
+ .iter()
+ .filter(|(other, _)| !skip.contains(other))
+ .map(|&(other, other_key)| (mix_pair(key, other_key), other))
+ .collect()
+}
+
+/// Symmetric: the same for (a, b) and (b, a).
+pub(crate) fn pair_score(a: &str, b: &str) -> u64 {
+ mix_pair(pair_key(a), pair_key(b))
+}
+
+/// What a member contributes to its pair scores: computed once per member.
+pub(super) fn pair_key(member: &str) -> u64 {
+ let digest = hash(PAIR_LABEL, &[member]);
+ u64::from_be_bytes(digest[..8].try_into().expect("8 bytes"))
+}
+
+/// The score of a pair from its members' keys, symmetric: SplitMix64's finalizer over
+/// the lower key and the mixed higher one. Keys are uniform (SHA-256), so for each member
+/// the others' scores are in an order unrelated to anyone else's.
+pub(super) fn mix_pair(a: u64, b: u64) -> u64 {
+ let (low, high) = if a <= b { (a, b) } else { (b, a) };
+ mix(low ^ mix(high))
+}
+
+pub(super) fn mix(mut z: u64) -> u64 {
+ z = z.wrapping_add(0x9e37_79b9_7f4a_7c15);
+ z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
+ z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
+ z ^ (z >> 31)
+}
+
+pub(super) fn hash(label: &[u8], fields: &[&str]) -> [u8; 32] {
+ let mut hasher = Sha256::new();
+ for field in [label]
+ .into_iter()
+ .chain(fields.iter().map(|f| f.as_bytes()))
+ {
+ hasher.update((field.len() as u32).to_be_bytes());
+ hasher.update(field);
+ }
+ hasher.finalize().into()
+}
diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs
new file mode 100644
index 0000000..30fa56e
--- /dev/null
+++ b/src/swarm/topology/mod.rs
@@ -0,0 +1,377 @@
+//! Which members a node syncs with: its neighbours.
+//!
+//! In a full mesh every change reaches every node from every other node (Automerge
+//! decides per pair, so nobody knows that a third peer already passed it on), so the
+//! copies of one write grow with N². Above `k` members, each node keeps only a few
+//! neighbours, and changes travel over several hops.
+//!
+//! Every node computes the graph from the member list and which members it can reach,
+//! with no coordination:
+//! * a ring ordered by key hash: each member's nearest reachable member in both
+//! directions, which keeps the reachable members connected;
+//! * plus, for each member, the `k - 2` reachable others with the lowest pair score (a
+//! hash of both keys, so both ends agree on it), which keeps paths short.
+//!
+//! A link exists if either end chose it, so links are symmetric and a node has at least
+//! `k` neighbours. With at most `k + 1` reachable members the graph is the full mesh:
+//! small swarms behave exactly as before. A member joining, leaving or going offline
+//! changes only links near it.
+//!
+//! Members on several transports (Mycelium, libp2p) get such a graph per transport,
+//! among the members on it (as their addresses in the member list say), and a node's
+//! neighbours are the union of its neighbours on each transport it is on
+//! (`neighbours_across`). Each of those graphs is connected (its ring), and a member on
+//! two transports is in both: as long as members on several transports link the
+//! transports, the swarm stays connected, and such a member has about `k` neighbours per
+//! transport. With one transport, it is the one graph as before.
+
+use std::collections::{BTreeMap, BTreeSet};
+
+use tracing::info;
+
+use crate::locks::RwLockExt;
+use crate::state::{AppState, PeerState};
+use crate::swarm::members;
+
+mod graph;
+
+pub(crate) use graph::pair_score;
+use graph::*;
+
+/// This node's current neighbours, and the member list they were computed from.
+#[derive(Debug, Default)]
+pub(crate) struct NeighbourView {
+ members: Vec<String>,
+ k: usize,
+ unreachable: BTreeSet<String>,
+ stable: BTreeSet<String>,
+ /// The transports (address schemes) of each member, and ours.
+ transports: BTreeMap<String, BTreeSet<String>>,
+ local_transports: BTreeSet<String>,
+ /// Peers that proved they are of our swarm but aren't in our member list (yet): their
+ /// registration hasn't reached us. They are neighbours too (`refresh`).
+ unlisted: BTreeSet<String>,
+ /// Members the member list gives no address for on our transports, that reach us
+ /// anyway: they only connect out (no listen address). They are neighbours too while
+ /// they say they sync with us (`refresh`).
+ outbound: BTreeSet<String>,
+ neighbours: BTreeSet<String>,
+ /// Every reachable member is a neighbour.
+ full_mesh: bool,
+}
+
+/// Whether the neighbours are computed from the member list yet (before, everyone
+/// counts as a neighbour).
+pub(crate) fn view_known(state: &AppState) -> bool {
+ state.neighbours.read_or_recover().is_some()
+}
+
+/// Whether every reachable member is our neighbour (also while the view is unknown).
+pub(crate) fn is_full_mesh(state: &AppState) -> bool {
+ state
+ .neighbours
+ .read_or_recover()
+ .as_ref()
+ .is_none_or(|view| view.full_mesh)
+}
+
+/// The current view in numbers, once known.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct Summary {
+ /// Registered members, us included.
+ pub(crate) members: usize,
+ /// Members other than us not currently judged unreachable.
+ pub(crate) reachable: usize,
+ pub(crate) stable: usize,
+ pub(crate) neighbours: usize,
+ pub(crate) stable_neighbours: usize,
+ pub(crate) full_mesh: bool,
+}
+
+pub(crate) fn summary(state: &AppState) -> Option<Summary> {
+ let local = state.local_id();
+ state
+ .neighbours
+ .read_or_recover()
+ .as_ref()
+ .map(|view| Summary {
+ members: view.members.len(),
+ reachable: view
+ .members
+ .iter()
+ .filter(|m| m.as_str() != local && !view.unreachable.contains(*m))
+ .count(),
+ stable: view.stable.len(),
+ neighbours: view.neighbours.len(),
+ stable_neighbours: view
+ .neighbours
+ .iter()
+ .filter(|m| view.stable.contains(*m))
+ .count(),
+ full_mesh: view.full_mesh,
+ })
+}
+
+/// Whether we initiate contact with `peer`: HELLOs, keepalives, dormancy checks, sync
+/// openings. Everyone is a neighbour until the view is known (while joining, or before
+/// any member registered), so joining works as in a full mesh. We always *answer*
+/// anyone who contacts us.
+pub(crate) fn is_neighbour(state: &AppState, peer: &str) -> bool {
+ match &*state.neighbours.read_or_recover() {
+ Some(view) => view.neighbours.contains(peer),
+ None => true,
+ }
+}
+
+/// Every member's neighbours as the member list makes them, as this node sees it (whom
+/// it can reach, who is stable): the graph each node computes its own part of. Ours as
+/// they are (with members that only connect out, or aren't listed yet). For a map of
+/// the swarm (the dashboard); None until the view is known. A link exists if either end
+/// chose it: draw one if either lists the other. About `members²` work: compute it when
+/// the member list may have changed, not on every look.
+#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
+pub(crate) fn intended_graph(state: &AppState) -> Option<BTreeMap<String, BTreeSet<String>>> {
+ let view = state.neighbours.read_or_recover();
+ let view = view.as_ref()?;
+ let unreachable = |member: &str| view.unreachable.contains(member);
+ let stable = |member: &str| view.stable.contains(member);
+ let graph = view
+ .members
+ .iter()
+ .map(|member| {
+ let neighbours = if member == state.local_id() {
+ view.neighbours.clone()
+ } else {
+ let own = view.transports.get(member).cloned().unwrap_or_default();
+ neighbours_across(
+ &view.transports,
+ member,
+ &own,
+ view.k,
+ &unreachable,
+ &stable,
+ )
+ };
+ (member.clone(), neighbours)
+ })
+ .collect();
+ Some(graph)
+}
+
+/// How the neighbours changed.
+pub(crate) struct NeighbourChange {
+ pub(crate) added: Vec<String>,
+ pub(crate) removed: Vec<String>,
+}
+
+/// Runs `change` on a peer's state, and has the neighbours recomputed (`run_worker`) if
+/// it changed what they are chosen from (`PeerState::neighbour_inputs`). Changes to a
+/// peer that may matter to the neighbours go through here.
+pub(crate) fn change_peer<T>(
+ state: &AppState,
+ peer_state: &mut PeerState,
+ change: impl FnOnce(&mut PeerState) -> T,
+) -> T {
+ let before = peer_state.neighbour_inputs();
+ let result = change(peer_state);
+ if peer_state.neighbour_inputs() != before {
+ state.topology_changed.notify_one();
+ }
+ result
+}
+
+/// Recomputes the neighbours whenever something they depend on may have changed
+/// (`AppState::topology_changed`): the member list (`swarm::changes::run_change_publisher`),
+/// a peer's state (`change_peer`), and every 5 s as a fallback (`swarm::peer`). A burst
+/// of notifications coalesces into one recomputation.
+pub(crate) async fn run_worker(state: AppState) {
+ loop {
+ state.topology_changed.notified().await;
+ if let Some(change) = refresh(&state).await {
+ crate::swarm::peer::apply_neighbour_change(&state, change).await;
+ }
+ }
+}
+
+/// Whether a change at `path` may change the neighbours: one in the member list, or the
+/// root (the whole document replaced, as after an epoch switch).
+pub(crate) fn affects_neighbours(path: &[String]) -> bool {
+ path.is_empty() || path.starts_with(&members::members_dir())
+}
+
+/// Recomputes the neighbours once ready, if the member list, the reachable members or
+/// `k` changed.
+pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
+ if !state.is_ready().await {
+ return None;
+ }
+ let (current, stable, transports, alive) = {
+ let replica = state.doc.read().await;
+ members::apply(state, &replica.doc);
+ let current = members::members(&replica.doc);
+ // Those that said they are alive within the electorate window.
+ let alive: BTreeSet<String> = members::electorate(
+ &replica.doc,
+ state.clock.now_millis().0,
+ state.electorate_window,
+ )
+ .into_iter()
+ .collect();
+ let stable: BTreeSet<String> = current
+ .iter()
+ .filter(|m| members::tier(&replica.doc, m) == members::Tier::Stable)
+ .cloned()
+ .collect();
+ let transports: BTreeMap<String, BTreeSet<String>> = current
+ .iter()
+ .map(|m| (m.clone(), schemes(&members::addresses(&replica.doc, m))))
+ .collect();
+ (current, stable, transports, alive)
+ };
+ let local_transports: BTreeSet<String> = state.transport.transports().into_iter().collect();
+ let k = state.neighbours_k;
+ // Those that don't answer, those on none of our transports, and those that haven't
+ // said they are alive for longer than the electorate window, unless we heard from
+ // them since we started (one that comes back says so to us first: its HELLO).
+ let mut unreachable = BTreeSet::new();
+ for member in &current {
+ if member == state.local_id() {
+ continue;
+ }
+ if !state.transport.can_reach(member) {
+ unreachable.insert(member.clone());
+ continue;
+ }
+ let (gone, heard) = match state.get_peer_state(member).await {
+ Some(peer_state) => {
+ let peer_state = peer_state.lock().await;
+ (peer_state.is_unreachable(), peer_state.last_seen.is_some())
+ }
+ None => (false, false),
+ };
+ if gone || (!heard && !alive.contains(member)) {
+ unreachable.insert(member.clone());
+ }
+ }
+ // A node that joined while the genesis node hadn't registered yet has the genesis
+ // document without the genesis node in it, and registers itself there: then neither
+ // lists the other, and with each the other's neighbour only by the member list,
+ // neither would ever sync with the other, and their registrations never pass. So a
+ // peer that proved itself a member of our swarm counts until the list says otherwise:
+ // syncing brings each the other's registration.
+ let mut unlisted = BTreeSet::new();
+ for (peer, peer_state) in state.peers_snapshot().await {
+ if current.contains(&peer) || peer == state.local_id() {
+ continue;
+ }
+ let peer_state = peer_state.lock().await;
+ if matches!(peer_state.standing, crate::state::Standing::Member { .. })
+ && !peer_state.is_unreachable()
+ {
+ unlisted.insert(peer);
+ }
+ }
+ // A member that lists no address on our transports, but reached us anyway, only
+ // connects out: it isn't in the graph on our side (it's on none of our transports, as
+ // far as the list goes), while on its side we are, so it would sync with us while we
+ // never sent it our changes (only answers to its). It picks its neighbours; those it
+ // picked keep it, as it says (STATUS).
+ let mut outbound = BTreeSet::new();
+ for member in &current {
+ if member == state.local_id()
+ || unreachable.contains(member)
+ || transports
+ .get(member)
+ .is_some_and(|on| !on.is_disjoint(&local_transports))
+ {
+ continue;
+ }
+ if let Some(peer_state) = state.get_peer_state(member).await {
+ let peer_state = peer_state.lock().await;
+ if matches!(peer_state.standing, crate::state::Standing::Member { .. })
+ && peer_state.syncs_with_us != Some(false)
+ {
+ outbound.insert(member.clone());
+ }
+ }
+ }
+ let previous = {
+ let view = state.neighbours.read_or_recover();
+ if view.as_ref().is_some_and(|view| {
+ view.members == current
+ && view.unlisted == unlisted
+ && view.outbound == outbound
+ && view.k == k
+ && view.unreachable == unreachable
+ && view.stable == stable
+ && view.transports == transports
+ && view.local_transports == local_transports
+ }) {
+ return None;
+ }
+ view.as_ref().map(|view| view.neighbours.clone())
+ };
+
+ // Before the first view everyone known counted as a neighbour.
+ let previous = match previous {
+ Some(previous) => previous,
+ None => state.peer_states.read().await.keys().cloned().collect(),
+ };
+ let mut computed = neighbours_across(
+ &transports,
+ state.local_id(),
+ &local_transports,
+ k,
+ &|m| unreachable.contains(m),
+ &|m| stable.contains(m),
+ );
+ computed.extend(outbound.iter().cloned());
+ let listed_neighbours = computed.len();
+ computed.extend(unlisted.iter().cloned());
+ let change = NeighbourChange {
+ added: computed
+ .iter()
+ .filter(|m| !previous.contains(*m))
+ .cloned()
+ .collect(),
+ removed: previous
+ .iter()
+ .filter(|m| !computed.contains(*m))
+ .cloned()
+ .collect(),
+ };
+ info!(
+ members = current.len(),
+ unreachable = unreachable.len(),
+ stable = stable.len(),
+ k,
+ neighbours = computed.len(),
+ unlisted = unlisted.len(),
+ outbound = outbound.len(),
+ added = change.added.len(),
+ removed = change.removed.len(),
+ "Neighbours computed from the member list"
+ );
+ let reachable_others = current
+ .iter()
+ .filter(|m| m.as_str() != state.local_id() && !unreachable.contains(*m))
+ .count();
+ *state.neighbours.write_or_recover() = Some(NeighbourView {
+ full_mesh: listed_neighbours >= reachable_others,
+ members: current,
+ k,
+ unreachable,
+ stable,
+ transports,
+ local_transports,
+ unlisted,
+ outbound,
+ neighbours: computed,
+ });
+ // New neighbours are kept alive and judged.
+ state.peers_due.now();
+ Some(change)
+}
+
+#[cfg(test)]
+mod tests;
diff --git a/src/swarm/topology/tests.rs b/src/swarm/topology/tests.rs
new file mode 100644
index 0000000..d3a3e38
--- /dev/null
+++ b/src/swarm/topology/tests.rs
@@ -0,0 +1,313 @@
+use super::*;
+use std::collections::VecDeque;
+
+fn members(n: usize) -> Vec<String> {
+ (0..n).map(|i| format!("{:064x}", i * 7919 + 13)).collect()
+}
+
+fn all_reachable(_: &str) -> bool {
+ false
+}
+
+fn none_stable(_: &str) -> bool {
+ false
+}
+
+fn graph(members: &[String], k: usize) -> BTreeMap<String, BTreeSet<String>> {
+ super::graph(members, k, &all_reachable, &none_stable)
+}
+
+fn connected(g: &BTreeMap<String, BTreeSet<String>>) -> bool {
+ let Some(start) = g.keys().next() else {
+ return true;
+ };
+ let mut seen = BTreeSet::from([start.clone()]);
+ let mut queue = VecDeque::from([start.clone()]);
+ while let Some(m) = queue.pop_front() {
+ for next in &g[&m] {
+ if seen.insert(next.clone()) {
+ queue.push_back(next.clone());
+ }
+ }
+ }
+ seen.len() == g.len()
+}
+
+fn edges(graph: &BTreeMap<String, BTreeSet<String>>) -> BTreeSet<(String, String)> {
+ graph
+ .iter()
+ .flat_map(|(a, ns)| {
+ ns.iter().map(move |b| {
+ if a < b {
+ (a.clone(), b.clone())
+ } else {
+ (b.clone(), a.clone())
+ }
+ })
+ })
+ .collect()
+}
+
+fn hex(bytes: [u8; 32]) -> String {
+ crate::transport::Hex(&bytes).to_string()
+}
+
+/// Over random swarms (size, `k`, who is reachable, who is stable): links only
+/// between reachable members, never to oneself, symmetric; a full mesh while small,
+/// otherwise at least `k` links each and connected; independent of the order members
+/// are listed in; and each member's own view (`neighbours`) agrees with the graph.
+#[test]
+fn random_swarms() {
+ crate::testrand::for_seeds(300, |rng| {
+ let n = rng.between(1, 60);
+ let k = rng.between(2, 14);
+ let mut all: Vec<String> = (0..n).map(|_| hex(rng.bytes32())).collect();
+ let unreachable: BTreeSet<String> =
+ all.iter().filter(|_| rng.chance(0.3)).cloned().collect();
+ let stable: BTreeSet<String> = all.iter().filter(|_| rng.chance(0.3)).cloned().collect();
+ let is_unreachable = |m: &str| unreachable.contains(m);
+ let is_stable = |m: &str| stable.contains(m);
+
+ let g = super::graph(&all, k, &is_unreachable, &is_stable);
+ // Owned: `all` is shuffled below.
+ let live: BTreeSet<String> = all
+ .iter()
+ .filter(|m| !unreachable.contains(*m))
+ .cloned()
+ .collect();
+ assert_eq!(g.keys().cloned().collect::<BTreeSet<_>>(), live);
+ for (a, ns) in &g {
+ assert!(!ns.contains(a), "{a} links to itself");
+ for b in ns {
+ assert!(live.contains(b), "{a} links to unreachable {b}");
+ assert!(g[b].contains(a), "{a} -> {b} but not back");
+ }
+ }
+ if live.len() <= k + 1 {
+ assert!(
+ g.values().all(|ns| ns.len() + 1 == live.len()),
+ "not a full mesh"
+ );
+ } else {
+ assert!(
+ g.values().all(|ns| ns.len() >= k),
+ "a member with fewer than k links"
+ );
+ assert!(connected(&g), "not connected");
+ }
+
+ rng.shuffle(&mut all);
+ assert_eq!(super::graph(&all, k, &is_unreachable, &is_stable), g);
+
+ if let Some(local) = live.iter().nth(rng.below(live.len())) {
+ let others: Vec<String> = all.iter().filter(|m| *m != local).cloned().collect();
+ assert_eq!(
+ neighbours(&others, local, k, &is_unreachable, &is_stable),
+ g[local]
+ );
+ }
+ });
+}
+
+#[test]
+fn small_swarms_are_a_full_mesh() {
+ let ms = members(13);
+ for m in &ms {
+ assert_eq!(
+ neighbours(&ms, m, 12, &all_reachable, &none_stable).len(),
+ 12
+ );
+ }
+}
+
+#[test]
+fn links_are_symmetric_and_degree_at_least_k() {
+ let ms = members(250);
+ let g = graph(&ms, 8);
+ for (a, ns) in &g {
+ assert!(ns.len() >= 8, "{a} has {} neighbours", ns.len());
+ assert!(!ns.contains(a));
+ for b in ns {
+ assert!(g[b].contains(a), "{a} -> {b} not symmetric");
+ }
+ }
+}
+
+#[test]
+fn graph_is_connected() {
+ for (n, k) in [(250, 8), (250, 2), (60, 3)] {
+ assert!(connected(&graph(&members(n), k)), "n={n} k={k}");
+ }
+}
+
+#[test]
+fn one_member_joining_changes_few_links() {
+ let ms = members(200);
+ let before = edges(&graph(&ms, 8));
+ let mut more = ms.clone();
+ more.push(format!("{:064x}", 999_999));
+ let after = edges(&graph(&more, 8));
+ let changed = before.symmetric_difference(&after).count();
+ assert!(changed <= 4 * 8, "{changed} links changed");
+}
+
+#[test]
+fn reachable_members_stay_connected_when_most_are_offline() {
+ let ms = members(100);
+ // Every tenth member is online.
+ let online: BTreeSet<String> = ms.iter().step_by(10).cloned().collect();
+ let offline = |m: &str| !online.contains(m);
+ let g = super::graph(&ms, 12, &offline, &none_stable);
+ assert_eq!(g.len(), 10);
+ // Ten reachable members and k = 12: a full mesh among them.
+ assert!(g.values().all(|ns| ns.len() == 9));
+
+ let online: BTreeSet<String> = ms.iter().step_by(4).cloned().collect();
+ let offline = |m: &str| !online.contains(m);
+ let g = super::graph(&ms, 8, &offline, &none_stable);
+ assert_eq!(g.len(), 25);
+ assert!(connected(&g));
+ assert!(
+ g.values()
+ .all(|ns| ns.len() >= 8 && ns.iter().all(|n| online.contains(n)))
+ );
+}
+
+#[test]
+fn stable_members_become_hubs_with_bounded_degree() {
+ let ms = members(250);
+ let stable: BTreeSet<String> = ms.iter().step_by(12).take(20).cloned().collect();
+ let is_stable = |m: &str| stable.contains(m);
+ let g = super::graph(&ms, 12, &all_reachable, &is_stable);
+ assert!(connected(&g));
+
+ let degree = |pred: &dyn Fn(&String) -> bool| -> (usize, f64) {
+ let degrees: Vec<usize> = g
+ .iter()
+ .filter(|(m, _)| pred(m))
+ .map(|(_, ns)| ns.len())
+ .collect();
+ let max = *degrees.iter().max().unwrap();
+ (
+ max,
+ degrees.iter().sum::<usize>() as f64 / degrees.len() as f64,
+ )
+ };
+ let (stable_max, stable_mean) = degree(&|m| stable.contains(m));
+ let (_, transient_mean) = degree(&|m| !stable.contains(m));
+ // 230 transient members put about 5 of their 10 scored slots on 20 stable ones.
+ assert!(
+ stable_mean > 3.0 * transient_mean,
+ "{stable_mean} vs {transient_mean}"
+ );
+ assert!(
+ stable_max <= 100,
+ "a stable member has {stable_max} neighbours"
+ );
+}
+
+/// A hub's degree in a swarm of `n` with `stable_count` stable members stays within
+/// the bound.
+fn hub_degree_is_bounded(n: usize, stable_count: usize) {
+ // Few stable members among many: without the fan-in budget each would be chosen
+ // by most of the swarm (about 130 neighbours with 10 of 250). With it, a hub has
+ // at most the budget plus the links others choose by pair score and the ring.
+ let k = 12;
+ let ms = members(n);
+ let stable: BTreeSet<String> = ms.iter().step_by(n / stable_count).cloned().collect();
+ let is_stable = |m: &str| stable.contains(m);
+ let g = super::graph(&ms, k, &all_reachable, &is_stable);
+ assert!(connected(&g), "n={n} stable={stable_count}");
+ let hub_max = g
+ .iter()
+ .filter(|(m, _)| stable.contains(*m))
+ .map(|(_, ns)| ns.len())
+ .max()
+ .unwrap();
+ assert!(
+ hub_max <= 8 * k,
+ "n={n} stable={stable_count}: a stable member has {hub_max} neighbours"
+ );
+}
+
+/// Two transports with members on one or the other, and two members on both: every
+/// link is over a shared transport, and the two members on both connect the rest.
+#[test]
+fn members_on_two_transports_bridge_the_others() {
+ let k = 8;
+ let ms = members(201);
+ let on = |names: &[&str]| -> BTreeSet<String> {
+ names.iter().map(|name| (*name).to_owned()).collect()
+ };
+ let transports: BTreeMap<String, BTreeSet<String>> = ms
+ .iter()
+ .enumerate()
+ .map(|(i, m)| {
+ let transports = match i {
+ 0 | 1 => on(&["a", "b"]),
+ i if i % 2 == 0 => on(&["a"]),
+ _ => on(&["b"]),
+ };
+ (m.clone(), transports)
+ })
+ .collect();
+ let no_stable = |_: &str| false;
+ let mut graph: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
+ for (m, own) in &transports {
+ for n in neighbours_across(&transports, m, own, k, &all_reachable, &no_stable) {
+ assert!(
+ !transports[&n].is_disjoint(own),
+ "{m} and {n} share no transport"
+ );
+ graph.entry(m.clone()).or_default().insert(n.clone());
+ graph.entry(n).or_default().insert(m.clone());
+ }
+ }
+ assert_eq!(graph.len(), ms.len());
+ assert!(connected(&graph));
+ let bridge = &ms[0];
+ assert!(graph[bridge].iter().any(|n| !transports[n].contains("b")));
+ assert!(graph[bridge].iter().any(|n| !transports[n].contains("a")));
+
+ // One transport: the one graph, as without transports.
+ let one: BTreeMap<String, BTreeSet<String>> =
+ ms.iter().map(|m| (m.clone(), on(&[""]))).collect();
+ for m in &ms[..5] {
+ assert_eq!(
+ neighbours_across(&one, m, &on(&[""]), k, &all_reachable, &no_stable),
+ neighbours(&ms, m, k, &all_reachable, &no_stable)
+ );
+ }
+}
+
+#[test]
+fn schemes_of_addresses() {
+ let addresses = |all: &[&str]| -> Vec<String> { all.iter().map(|a| (*a).to_owned()).collect() };
+ assert_eq!(
+ schemes(&addresses(&[
+ "mycelium:ab",
+ "libp2p:/ip4/1.2.3.4/tcp/1",
+ "libp2p:/ip6/::1/tcp/1"
+ ])),
+ BTreeSet::from(["libp2p".to_owned(), "mycelium".to_owned()])
+ );
+ assert_eq!(schemes(&addresses(&["a"])), BTreeSet::from([String::new()]));
+ assert!(schemes(&[]).is_empty());
+}
+
+#[test]
+fn hub_degree_is_bounded_however_few_stable_members() {
+ for (n, stable_count) in [(250, 10), (400, 5)] {
+ hub_degree_is_bounded(n, stable_count);
+ }
+}
+
+/// The same at the sizes the bound was designed for; slow in debug builds, so run
+/// on request (`cargo test --release -- --ignored`).
+#[test]
+#[ignore]
+fn hub_degree_is_bounded_in_large_swarms() {
+ for (n, stable_count) in [(1000, 10), (1000, 20), (2000, 20)] {
+ hub_degree_is_bounded(n, stable_count);
+ }
+}