summaryrefslogtreecommitdiffstats
path: root/src/swarm
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm')
-rw-r--r--src/swarm/auth.rs31
-rw-r--r--src/swarm/blob.rs508
-rw-r--r--src/swarm/bootstrap.rs13
-rw-r--r--src/swarm/members.rs262
-rw-r--r--src/swarm/mod.rs10
-rw-r--r--src/swarm/pair.rs78
-rw-r--r--src/swarm/peer.rs405
-rw-r--r--src/swarm/relay.rs179
-rw-r--r--src/swarm/rtt.rs179
-rw-r--r--src/swarm/source.rs210
-rw-r--r--src/swarm/stability.rs270
-rw-r--r--src/swarm/stats.rs284
-rw-r--r--src/swarm/sync.rs490
-rw-r--r--src/swarm/throughput.rs168
-rw-r--r--src/swarm/topology.rs521
15 files changed, 3176 insertions, 432 deletions
diff --git a/src/swarm/auth.rs b/src/swarm/auth.rs
index ea95406..31230a5 100644
--- a/src/swarm/auth.rs
+++ b/src/swarm/auth.rs
@@ -8,8 +8,8 @@
//! The tag is `HMAC-SHA256(key, label ‖ swarm_id ‖ from_pk ‖ to_pk)`. It needs no
//! challenge: it is bound to both keys, and only `from_pk` can send as `from_pk`, so a
//! captured tag is useless to anyone else. Different labels for HELLO and HELLO_REPLY
-//! keep one from being passed off as the other. A HELLO has no swarm ID yet (a new node
-//! doesn't know it), so its tag binds an empty one.
+//! keep one from being passed off as the other. A HELLO binds the swarm ID its sender
+//! claims, or an empty one from a new node that doesn't know it yet.
//!
//! Limits of a shared key: every member holds the same secret, so a single member can't
//! be revoked (forgetting it lasts until it says HELLO again), and changing the key
@@ -82,9 +82,9 @@ impl SwarmKey {
Some(std::mem::take(&mut rejections.suppressed))
}
- /// Tag for a HELLO from `from` to `to`.
- pub(crate) fn hello_tag(&self, from: &str, to: &str) -> Vec<u8> {
- self.tag(HELLO_LABEL, "", from, to).to_vec()
+ /// Tag for a HELLO from `from` to `to`, claiming `swarm_id` ("" if unknown).
+ pub(crate) fn hello_tag(&self, swarm_id: &str, from: &str, to: &str) -> Vec<u8> {
+ self.tag(HELLO_LABEL, swarm_id, from, to).to_vec()
}
/// Tag for a HELLO_REPLY from `from` to `to`, naming the swarm.
@@ -92,8 +92,8 @@ impl SwarmKey {
self.tag(HELLO_REPLY_LABEL, swarm_id, from, to).to_vec()
}
- pub(crate) fn verify_hello(&self, tag: &[u8], from: &str, to: &str) -> bool {
- constant_time_eq(tag, &self.tag(HELLO_LABEL, "", from, to))
+ pub(crate) fn verify_hello(&self, tag: &[u8], swarm_id: &str, from: &str, to: &str) -> bool {
+ constant_time_eq(tag, &self.tag(HELLO_LABEL, swarm_id, from, to))
}
pub(crate) fn verify_hello_reply(
@@ -176,18 +176,19 @@ mod tests {
let key = SwarmKey::new(vec![1; 32]);
let other = SwarmKey::new(vec![2; 32]);
- let hello = key.hello_tag("A", "B");
- assert!(key.verify_hello(&hello, "A", "B"));
- assert!(!other.verify_hello(&hello, "A", "B"));
- assert!(!key.verify_hello(&hello, "B", "A"));
- assert!(!key.verify_hello(&hello, "A", "C"));
- assert!(!key.verify_hello(&[], "A", "B"));
+ let hello = key.hello_tag("swarm", "A", "B");
+ assert!(key.verify_hello(&hello, "swarm", "A", "B"));
+ assert!(!key.verify_hello(&hello, "", "A", "B"));
+ assert!(!other.verify_hello(&hello, "swarm", "A", "B"));
+ assert!(!key.verify_hello(&hello, "swarm", "B", "A"));
+ assert!(!key.verify_hello(&hello, "swarm", "A", "C"));
+ assert!(!key.verify_hello(&[], "swarm", "A", "B"));
let reply = key.hello_reply_tag("swarm", "A", "B");
assert!(key.verify_hello_reply(&reply, "swarm", "A", "B"));
assert!(!key.verify_hello_reply(&reply, "other-swarm", "A", "B"));
// A HELLO tag is no HELLO_REPLY tag, and the other way round.
- assert!(!key.verify_hello_reply(&hello, "", "A", "B"));
- assert!(!key.verify_hello(&key.hello_reply_tag("", "A", "B"), "A", "B"));
+ assert!(!key.verify_hello_reply(&hello, "swarm", "A", "B"));
+ assert!(!key.verify_hello(&key.hello_reply_tag("swarm", "A", "B"), "swarm", "A", "B"));
}
}
diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
new file mode 100644
index 0000000..8c891ae
--- /dev/null
+++ b/src/swarm/blob.rs
@@ -0,0 +1,508 @@
+//! Blobs: content-addressed bytes fetched in chunks from several peers at once.
+//!
+//! Large transfers can't go in one Mycelium message (it must arrive within Mycelium's
+//! five-minute deadline), and a blob every node holds identically can come from many
+//! of them. A blob is an Automerge change by its hash (e.g. an epoch's snapshot change,
+//! which every node on that epoch holds byte for byte), or content by its BLAKE3 hash
+//! (e.g. the new document of a `replace` or `import` remake).
+//!
+//! * Serving: a request without chunks gets the manifest (size, chunk size, a BLAKE3
+//! hash per chunk); one with chunks gets each chunk as a message of its own. A node
+//! that doesn't hold the blob answers `Missing`. The bytes of the blob served last
+//! are kept, so serving a download doesn't extract them for every chunk. Serving runs
+//! in tasks of its own, at most [`MAX_SERVING`] chunk sends at a time.
+//! * Fetching: the manifest from the first peer that has it (closest first), then the
+//! chunks, a few in flight per peer from up to [`MAX_SOURCES`] peers. Every chunk is
+//! checked against the manifest; a peer that sends a bad one is dropped, a chunk
+//! that doesn't arrive in time is asked of another peer. The caller checks the whole
+//! (for a change: its hash).
+//!
+//! The chunk size is the serving node's (`--blob-chunk-kib`); the manifest carries it.
+
+use std::collections::{HashMap, HashSet, VecDeque};
+use std::sync::{Arc, Mutex, MutexGuard};
+use std::time::Duration;
+
+use anyhow::{Result, anyhow, bail};
+use automerge::ChangeHash;
+use tokio::sync::{Semaphore, mpsc};
+use tokio::time::Instant;
+use tracing::{debug, info, warn};
+
+use crate::state::AppState;
+use crate::transport::{BlobBody, BlobId, BlobMessage, ParsedMessage};
+
+/// Chunk sends in progress at once, over all downloads served.
+const MAX_SERVING: usize = 8;
+/// Peers a download fetches chunks from at once.
+const MAX_SOURCES: usize = 4;
+/// Chunks in flight per peer.
+const WINDOW: usize = 4;
+/// Largest blob a download accepts.
+const MAX_SIZE: u64 = 4 << 30;
+/// Largest chunk a download accepts.
+const MAX_CHUNK: u32 = 4 << 20;
+/// How long to wait for a manifest from one peer.
+const MANIFEST_TIMEOUT: Duration = Duration::from_secs(15);
+/// How long to wait for a chunk before asking another peer (raised by the peer's
+/// round-trip time).
+const CHUNK_TIMEOUT: Duration = Duration::from_secs(60);
+/// A download gives up after this long without any chunk arriving.
+const STALL_TIMEOUT: Duration = Duration::from_secs(180);
+/// Bad chunks after which a peer is no longer asked.
+const MAX_BAD: u32 = 2;
+
+/// Per node: what is being served and fetched.
+pub(crate) struct Blobs {
+ chunk_size: u32,
+ serving: Arc<Semaphore>,
+ /// The blob served last.
+ cache: Mutex<Option<(BlobId, Arc<Prepared>)>>,
+ /// Answers for downloads in progress, by blob.
+ fetches: Mutex<HashMap<BlobId, mpsc::UnboundedSender<(String, BlobBody)>>>,
+}
+
+struct Prepared {
+ bytes: Arc<Vec<u8>>,
+ hashes: Vec<[u8; 32]>,
+}
+
+impl Blobs {
+ pub(crate) fn new(chunk_size: u32) -> Self {
+ Self {
+ chunk_size: chunk_size.max(1),
+ serving: Arc::new(Semaphore::new(MAX_SERVING)),
+ cache: Mutex::new(None),
+ fetches: Mutex::new(HashMap::new()),
+ }
+ }
+
+ fn fetches(
+ &self,
+ ) -> MutexGuard<'_, HashMap<BlobId, mpsc::UnboundedSender<(String, BlobBody)>>> {
+ self.fetches
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
+ }
+
+ fn cached(&self, id: BlobId) -> Option<Arc<Prepared>> {
+ self.cache
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
+ .as_ref()
+ .filter(|(cached, _)| *cached == id)
+ .map(|(_, prepared)| Arc::clone(prepared))
+ }
+
+ fn remember(&self, id: BlobId, prepared: Arc<Prepared>) {
+ *self
+ .cache
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some((id, prepared));
+ }
+}
+
+fn chunk_hashes(bytes: &[u8], chunk_size: u32) -> Vec<[u8; 32]> {
+ bytes
+ .chunks(chunk_size as usize)
+ .map(|chunk| *blake3::hash(chunk).as_bytes())
+ .collect()
+}
+
+fn chunk_count(size: u64, chunk_size: u32) -> u64 {
+ size.div_ceil(u64::from(chunk_size))
+}
+
+/// A blob message from `peer`.
+pub(crate) async fn handle(state: &AppState, peer: &str, message: BlobMessage) {
+ match message.body {
+ BlobBody::Request { chunks } => {
+ let state = state.clone();
+ let peer = peer.to_owned();
+ // Off the inbound loop: answering may send many large messages.
+ tokio::spawn(async move { serve(&state, &peer, message.id, chunks).await });
+ }
+ body => {
+ let sender = state.blobs.fetches().get(&message.id).cloned();
+ match sender {
+ Some(sender) => {
+ let _ = sender.send((peer.to_owned(), body));
+ }
+ None => {
+ debug!(peer = %peer, blob = %message.id, "Blob answer for no download in progress")
+ }
+ }
+ }
+ }
+}
+
+async fn prepare(state: &AppState, id: BlobId) -> Option<Arc<Prepared>> {
+ if let Some(prepared) = state.blobs.cached(id) {
+ return Some(prepared);
+ }
+ let bytes = match id {
+ 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())
+ })
+ .await
+ .ok()
+ .flatten()
+ .map(Arc::new)?
+ }
+ BlobId::Content(hash) => state.remake.content(&hash)?,
+ };
+ let chunk_size = state.blobs.chunk_size;
+ let prepared = tokio::task::spawn_blocking(move || {
+ let hashes = chunk_hashes(&bytes, chunk_size);
+ Prepared { bytes, hashes }
+ })
+ .await
+ .ok()?;
+ let prepared = Arc::new(prepared);
+ state.blobs.remember(id, Arc::clone(&prepared));
+ Some(prepared)
+}
+
+async fn reply(state: &AppState, peer: &str, id: BlobId, body: BlobBody) {
+ let message = ParsedMessage::Blob(BlobMessage {
+ epoch: state.local_epoch(),
+ id,
+ body,
+ });
+ if let Err(error) = state.transport.send_message(peer, &message).await {
+ debug!(peer = %peer, blob = %id, error = %format!("{error:#}"), "Blob answer failed");
+ }
+}
+
+async fn serve(state: &AppState, peer: &str, id: BlobId, chunks: Vec<u32>) {
+ let Some(prepared) = prepare(state, id).await else {
+ return reply(state, peer, id, BlobBody::Missing).await;
+ };
+ let chunk_size = state.blobs.chunk_size;
+ if chunks.is_empty() {
+ let body = BlobBody::Manifest {
+ size: prepared.bytes.len() as u64,
+ chunk_size,
+ hashes: prepared.hashes.clone(),
+ };
+ return reply(state, peer, id, body).await;
+ }
+ for index in chunks.into_iter().take(WINDOW * 2) {
+ let start = index as usize * chunk_size as usize;
+ if start >= prepared.bytes.len() {
+ continue;
+ }
+ let end = (start + chunk_size as usize).min(prepared.bytes.len());
+ let Ok(_permit) = state.blobs.serving.acquire().await else {
+ return;
+ };
+ let data = bytes::Bytes::copy_from_slice(&prepared.bytes[start..end]);
+ reply(state, peer, id, BlobBody::Chunk { index, data }).await;
+ }
+}
+
+// --- fetching ----------------------------------------------------------------------
+
+/// Removes the download's answer channel when it ends, however it ends.
+struct Registration<'a> {
+ state: &'a AppState,
+ id: BlobId,
+}
+
+impl Drop for Registration<'_> {
+ fn drop(&mut self) {
+ self.state.blobs.fetches().remove(&self.id);
+ }
+}
+
+async fn request(state: &AppState, peer: &str, id: BlobId, chunks: Vec<u32>) {
+ let message = ParsedMessage::Blob(BlobMessage {
+ epoch: state.local_epoch(),
+ id,
+ body: BlobBody::Request { chunks },
+ });
+ if let Err(error) = state.transport.send_message(peer, &message).await {
+ debug!(peer = %peer, blob = %id, error = %format!("{error:#}"), "Blob request failed");
+ }
+}
+
+struct Manifest {
+ size: u64,
+ chunk_size: u32,
+ hashes: Vec<[u8; 32]>,
+}
+
+/// One peer during a download.
+struct Source {
+ peer: String,
+ in_flight: HashMap<u32, Instant>,
+ window: usize,
+ bad: u32,
+ timeout: Duration,
+}
+
+/// Downloads blob `id` from `peers` (all thought to hold it).
+pub(crate) async fn fetch(state: &AppState, id: BlobId, peers: Vec<String>) -> Result<Vec<u8>> {
+ let (sender, mut answers) = mpsc::unbounded_channel();
+ if state.blobs.fetches().insert(id, sender).is_some() {
+ bail!("{id} is being downloaded already");
+ }
+ let _registration = Registration { state, id };
+
+ // Closest first; peers without a round-trip time last.
+ let now = Instant::now();
+ let mut peers: Vec<(String, Option<Duration>)> = {
+ let rtt = state.lock_rtt();
+ peers
+ .into_iter()
+ .map(|peer| {
+ let estimate = rtt.effective(&peer, now, state.timing.rtt_fresh_for());
+ (peer, estimate)
+ })
+ .collect()
+ };
+ peers.sort_by_key(|(_, rtt)| rtt.unwrap_or(Duration::MAX));
+ let rtts: HashMap<String, Option<Duration>> = peers.iter().cloned().collect();
+ let mut candidates: VecDeque<String> = peers.into_iter().map(|(peer, _)| peer).collect();
+ if candidates.is_empty() {
+ bail!("no peer to download {id} from");
+ }
+
+ // The manifest, from the first peer that has the blob.
+ let mut manifest = None;
+ let mut missing: HashSet<String> = HashSet::new();
+ for peer in &candidates {
+ request(state, peer, id, Vec::new()).await;
+ let deadline = Instant::now() + MANIFEST_TIMEOUT;
+ loop {
+ let answer = tokio::time::timeout_at(deadline, answers.recv()).await;
+ match answer {
+ Ok(Some((
+ from,
+ BlobBody::Manifest {
+ size,
+ chunk_size,
+ hashes,
+ },
+ ))) => {
+ if size > MAX_SIZE
+ || chunk_size == 0
+ || chunk_size > MAX_CHUNK
+ || hashes.len() as u64 != chunk_count(size, chunk_size)
+ {
+ warn!(peer = %from, blob = %id, size, chunk_size, "Unusable blob manifest; asking another peer");
+ missing.insert(from);
+ break;
+ }
+ manifest = Some(Manifest {
+ size,
+ chunk_size,
+ hashes,
+ });
+ break;
+ }
+ Ok(Some((from, BlobBody::Missing))) => {
+ missing.insert(from.clone());
+ if &from == peer {
+ break;
+ }
+ }
+ Ok(Some(_)) => {} // chunks of an earlier attempt
+ Ok(None) | Err(_) => break,
+ }
+ }
+ if manifest.is_some() {
+ break;
+ }
+ }
+ let Some(manifest) = manifest else {
+ bail!("no peer sent a manifest for {id}");
+ };
+ candidates.retain(|peer| !missing.contains(peer));
+
+ let total = manifest.hashes.len();
+ info!(blob = %id, bytes = manifest.size, chunks = total, peers = candidates.len(), "Downloading blob");
+ let started = Instant::now();
+ let mut chunks: Vec<Option<bytes::Bytes>> = vec![None; total];
+ let mut done = 0usize;
+ let mut sources: Vec<Source> = Vec::new();
+ let mut last_progress = Instant::now();
+
+ while done < total {
+ // Fill up the sources, then their windows.
+ while sources.len() < MAX_SOURCES {
+ let Some(peer) = candidates.pop_front() else {
+ break;
+ };
+ let timeout =
+ CHUNK_TIMEOUT + rtts.get(&peer).copied().flatten().unwrap_or_default() * 4;
+ sources.push(Source {
+ peer,
+ in_flight: HashMap::new(),
+ window: WINDOW,
+ bad: 0,
+ timeout,
+ });
+ }
+ if sources.is_empty() {
+ bail!("no peer left to download {id} from ({done} of {total} chunks)");
+ }
+ let now = Instant::now();
+ let assigned: HashSet<u32> = sources
+ .iter()
+ .flat_map(|source| source.in_flight.keys().copied())
+ .collect();
+ let mut wanted = (0..total as u32)
+ .filter(|index| chunks[*index as usize].is_none() && !assigned.contains(index));
+ for source in &mut sources {
+ let free = source.window.saturating_sub(source.in_flight.len());
+ let batch: Vec<u32> = wanted.by_ref().take(free).collect();
+ if batch.is_empty() {
+ continue;
+ }
+ for index in &batch {
+ source.in_flight.insert(*index, now);
+ }
+ request(state, &source.peer, id, batch).await;
+ }
+
+ match tokio::time::timeout(Duration::from_millis(500), answers.recv()).await {
+ Ok(Some((from, BlobBody::Chunk { index, data }))) => {
+ let Some(source) = sources.iter_mut().find(|source| source.peer == from) else {
+ continue;
+ };
+ let sent_at = source.in_flight.remove(&index);
+ let slot = index as usize;
+ if slot >= total || chunks[slot].is_some() {
+ continue;
+ }
+ let expected = expected_len(&manifest, slot);
+ if data.len() != expected
+ || *blake3::hash(&data).as_bytes() != manifest.hashes[slot]
+ {
+ source.bad += 1;
+ warn!(peer = %from, blob = %id, index, "Bad blob chunk");
+ continue;
+ }
+ if let Some(sent_at) = sent_at {
+ debug!(peer = %from, blob = %id, index, ms = sent_at.elapsed().as_millis() as u64, "Blob chunk");
+ }
+ chunks[slot] = Some(data);
+ done += 1;
+ last_progress = Instant::now();
+ // A peer that delivers gets its full window back.
+ source.window = WINDOW;
+ }
+ Ok(Some((from, BlobBody::Missing))) => {
+ sources.retain(|source| source.peer != from);
+ }
+ Ok(Some(_)) => {}
+ Ok(None) => bail!("download of {id} cancelled"),
+ Err(_) => {}
+ }
+
+ // Chunks that took too long go to another peer; the slow peer gets a smaller
+ // window. Peers that sent bad chunks are dropped.
+ let now = Instant::now();
+ for source in &mut sources {
+ let timeout = source.timeout;
+ let before = source.in_flight.len();
+ source
+ .in_flight
+ .retain(|_, sent| now.saturating_duration_since(*sent) < timeout);
+ if source.in_flight.len() < before {
+ source.window = 1;
+ debug!(peer = %source.peer, blob = %id, "Blob chunks timed out; asking others");
+ }
+ }
+ sources.retain(|source| {
+ let keep = source.bad < MAX_BAD;
+ if !keep {
+ warn!(peer = %source.peer, blob = %id, "Dropping a peer that sent bad blob chunks");
+ }
+ keep
+ });
+ if last_progress.elapsed() > STALL_TIMEOUT {
+ bail!(
+ "download of {id} stalled ({done} of {total} chunks, nothing for {}s)",
+ STALL_TIMEOUT.as_secs()
+ );
+ }
+ }
+
+ let mut bytes = Vec::with_capacity(manifest.size as usize);
+ for chunk in chunks.into_iter().flatten() {
+ bytes.extend_from_slice(&chunk);
+ }
+ info!(
+ blob = %id,
+ bytes = bytes.len(),
+ secs = started.elapsed().as_secs_f32(),
+ "Downloaded blob"
+ );
+ Ok(bytes)
+}
+
+fn expected_len(manifest: &Manifest, slot: usize) -> usize {
+ let chunk_size = manifest.chunk_size as u64;
+ let start = slot as u64 * chunk_size;
+ (manifest.size - start).min(chunk_size) as usize
+}
+
+/// Downloads change `hash` from `peers` and checks it.
+/// Downloads the content with BLAKE3 hash `hash` from `peers` and checks it.
+pub(crate) async fn fetch_content(
+ state: &AppState,
+ hash: [u8; 32],
+ peers: Vec<String>,
+) -> Result<Vec<u8>> {
+ let id = BlobId::Content(hash);
+ let bytes = fetch(state, id, peers).await?;
+ if *blake3::hash(&bytes).as_bytes() != hash {
+ bail!("downloaded {id} doesn't match its hash");
+ }
+ Ok(bytes)
+}
+
+pub(crate) async fn fetch_change(
+ state: &AppState,
+ hash: ChangeHash,
+ peers: Vec<String>,
+) -> Result<automerge::Change> {
+ let bytes = fetch(state, BlobId::Change(hash), peers).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)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn chunking() {
+ let bytes: Vec<u8> = (0..1000u32).map(|n| n as u8).collect();
+ let hashes = chunk_hashes(&bytes, 300);
+ assert_eq!(hashes.len(), 4);
+ assert_eq!(chunk_count(1000, 300), 4);
+ assert_eq!(chunk_count(900, 300), 3);
+ let manifest = Manifest {
+ size: 1000,
+ chunk_size: 300,
+ hashes,
+ };
+ assert_eq!(expected_len(&manifest, 0), 300);
+ assert_eq!(expected_len(&manifest, 3), 100);
+ assert_eq!(manifest.hashes[3], *blake3::hash(&bytes[900..]).as_bytes());
+ }
+}
diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index 3b6e510..27ffdd8 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -7,7 +7,7 @@ use anyhow::{Result, bail};
use tracing::{debug, error, info, warn};
use uuid::Uuid;
-use crate::state::{AppState, PeerMembership};
+use crate::state::AppState;
use crate::swarm::peer::{remember_peer, send_hello_once};
const BOOTSTRAP_TIMEOUT: Duration = Duration::from_secs(300);
@@ -68,15 +68,10 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()>
for (peer, peer_state_mutex) in &peer_states {
let peer_state = peer_state_mutex.lock().await;
- let validated_swarm = match &peer_state.membership {
- PeerMembership::Unknown => None,
- PeerMembership::Member { swarm_id } => Some(swarm_id.as_str()),
- };
+ let validated_swarm = peer_state.standing.swarm_id();
peer_diagnostics.push(format!(
- "{peer}: validated_swarm={:?}, epoch={:?}, pending_inbound_sync={}",
- validated_swarm,
- peer_state.epoch,
- peer_state.pending_inbound_sync.len()
+ "{peer}: validated_swarm={:?}, epoch={:?}",
+ validated_swarm, peer_state.epoch,
))
}
let local_swarm = state.swarm_id().await;
diff --git a/src/swarm/members.rs b/src/swarm/members.rs
new file mode 100644
index 0000000..221437e
--- /dev/null
+++ b/src/swarm/members.rs
@@ -0,0 +1,262 @@
+//! The member registry: every node that has joined the swarm, under
+//! `_continuum/members/<pk>`, replicated with the document (and carried into new
+//! epochs with the rest of `_continuum`).
+//!
+//! A pure registry: it decides nothing about admission (that's the swarm key). Each
+//! node registers itself once it is ready, and re-registers if its entry disappears
+//! while it is alive. An operator's forget (`DELETE /peers/{pk}`) removes the entry
+//! together with writing the forget record; a forgotten node that turns out to be alive
+//! cancels the record and registers again.
+//!
+//! Besides its tier, an entry holds the alias a node gives itself (`--alias`), a name
+//! for people reading /peers and metrics (not unique and never an identity: the public
+//! key stays that), and an `alive` stamp the node renews now and then: the members
+//! renewed within `--electorate-window-days` are the electorate of a remake.
+
+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::store::{self, StoreError};
+use crate::state::{AppState, unix_millis};
+
+const META_ROOT: &str = "_continuum";
+
+fn members_dir() -> Vec<String> {
+ vec![META_ROOT.to_owned(), "members".to_owned()]
+}
+
+pub(crate) fn member_path(pk: &str) -> Vec<String> {
+ let mut path = members_dir();
+ path.push(pk.to_owned());
+ path
+}
+
+/// Public keys of all registered members, sorted.
+pub(crate) fn members(doc: &AutoCommit) -> Vec<String> {
+ let mut members = store::keys_at(doc, &members_dir()).unwrap_or_default();
+ members.sort();
+ members
+}
+
+/// A member's stability tier, judged by itself (`swarm::stability`).
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub(crate) enum Tier {
+ Stable,
+ #[default]
+ Transient,
+}
+
+impl Tier {
+ pub(crate) fn as_str(self) -> &'static str {
+ match self {
+ Self::Stable => "stable",
+ Self::Transient => "transient",
+ }
+ }
+}
+
+fn tier_path(pk: &str) -> Vec<String> {
+ let mut path = member_path(pk);
+ path.push("tier".to_owned());
+ path
+}
+
+/// `pk`'s tier; transient unless its entry says stable.
+pub(crate) fn tier(doc: &AutoCommit, 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> {
+ store::write_path(doc, &tier_path(pk), &json!(tier.as_str()))
+}
+
+fn alias_path(pk: &str) -> Vec<String> {
+ let mut path = member_path(pk);
+ path.push("alias".to_owned());
+ path
+}
+
+/// `pk`'s alias, if it gave itself one.
+pub(crate) fn alias(doc: &AutoCommit, pk: &str) -> Option<String> {
+ match store::read_path(doc, &alias_path(pk)) {
+ Ok(Some(JsonValue::String(alias))) => Some(alias),
+ _ => None,
+ }
+}
+
+/// The aliases of all members that have one.
+pub(crate) fn aliases(doc: &AutoCommit) -> 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> {
+ match alias {
+ Some(alias) => store::write_path(doc, &alias_path(pk), &json!(alias)),
+ None => match store::delete_path(doc, &alias_path(pk), false) {
+ Err(StoreError::NotFound) => Ok(()),
+ result => result,
+ },
+ }
+}
+
+fn alive_path(pk: &str) -> Vec<String> {
+ let mut path = member_path(pk);
+ path.push("alive".to_owned());
+ path
+}
+
+fn millis_at(doc: &AutoCommit, path: &[String]) -> Option<u64> {
+ match store::read_path(doc, path) {
+ Ok(Some(JsonValue::Number(number))) => number.as_u64(),
+ _ => None,
+ }
+}
+
+/// 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> {
+ 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> {
+ let oldest = now_ms.saturating_sub(window.as_millis() as u64);
+ members(doc)
+ .into_iter()
+ .filter(|pk| alive_at(doc, pk).is_some_and(|at| at >= oldest))
+ .collect()
+}
+
+/// 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.
+fn alive_renewal(window: Duration) -> Duration {
+ (window / 7).clamp(Duration::from_secs(3600), Duration::from_secs(24 * 3600))
+}
+
+/// 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
+/// renews its `alive` stamp now and then (see `electorate`).
+pub(crate) async fn ensure_registered(state: &AppState) {
+ if !state.is_ready().await {
+ return;
+ }
+ let local = state.local_id().to_owned();
+ let own = member_path(&local);
+ let (registered, recorded_alias, alive) = {
+ let replica = state.doc.read().await;
+ let registered = store::read_path(&replica.doc, &own)
+ .ok()
+ .flatten()
+ .is_some();
+ (
+ registered,
+ alias(&replica.doc, &local),
+ millis_at(&replica.doc, &alive_path(&local)),
+ )
+ };
+ let wanted = state.alias.clone();
+ let now = unix_millis();
+
+ if !registered {
+ let mut entry = json!({ "since": now, "alive": now });
+ if let Some(alias) = &wanted {
+ entry["alias"] = json!(alias);
+ }
+ match state
+ .mutate_doc(move |doc| store::write_path(doc, &own, &entry))
+ .await
+ {
+ Ok(()) => info!(
+ alias = wanted.as_deref(),
+ "Registered in the swarm's member list"
+ ),
+ Err(error) => warn!(error = %error, "Failed to register in the member list"),
+ }
+ return;
+ }
+
+ let renewal = alive_renewal(state.electorate_window).as_millis() as u64;
+ if alive.is_none_or(|at| now.saturating_sub(at) >= renewal) {
+ let path = alive_path(&local);
+ match state
+ .mutate_doc(move |doc| store::write_path(doc, &path, &json!(now)))
+ .await
+ {
+ Ok(()) => debug!("Renewed our alive stamp in the member list"),
+ Err(error) => warn!(error = %error, "Failed to renew our alive stamp"),
+ }
+ }
+
+ if recorded_alias == wanted {
+ return;
+ }
+ let alias = wanted.clone();
+ match state
+ .mutate_doc(move |doc| set_alias(doc, &local, alias.as_deref()))
+ .await
+ {
+ Ok(()) => info!(
+ alias = wanted.as_deref(),
+ previous = recorded_alias.as_deref(),
+ "Alias updated in the member list"
+ ),
+ Err(error) => warn!(error = %error, "Failed to record our alias"),
+ }
+}
+
+/// Removes `pk`'s entry (an operator forgot it). Absent is fine.
+pub(crate) fn unregister(doc: &mut AutoCommit, pk: &str) -> Result<(), StoreError> {
+ match store::delete_path(doc, &member_path(pk), false) {
+ Err(StoreError::NotFound) => Ok(()),
+ result => result,
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn electorate_by_alive_stamp() {
+ let day = 24 * 3600 * 1000;
+ let now = 100 * day;
+ let mut doc = AutoCommit::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 })),
+ ];
+ for (pk, entry) in entries {
+ store::write_path(&mut doc, &member_path(pk), &entry).unwrap();
+ }
+ assert_eq!(
+ electorate(&doc, now, Duration::from_secs(7 * 24 * 3600)),
+ ["fresh", "new"]
+ );
+ }
+
+ #[test]
+ fn renewal_interval() {
+ let day = Duration::from_secs(24 * 3600);
+ assert_eq!(alive_renewal(7 * day), day);
+ assert_eq!(alive_renewal(70 * day), day);
+ assert_eq!(alive_renewal(day), day / 7);
+ // The hourly floor only matters for windows shorter than the option allows.
+ let hour = Duration::from_secs(3600);
+ assert_eq!(alive_renewal(2 * hour), hour);
+ }
+}
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs
index a4277c7..adb5d1c 100644
--- a/src/swarm/mod.rs
+++ b/src/swarm/mod.rs
@@ -2,6 +2,16 @@
//! Automerge synchronization with every member (`sync`).
pub(crate) mod auth;
+pub(crate) mod blob;
pub(crate) mod bootstrap;
+pub(crate) mod members;
+pub(crate) mod pair;
pub(crate) mod peer;
+pub(crate) mod relay;
+pub(crate) mod rtt;
+pub(crate) mod source;
+pub(crate) mod stability;
+pub(crate) mod stats;
pub(crate) mod sync;
+pub(crate) mod throughput;
+pub(crate) mod topology;
diff --git a/src/swarm/pair.rs b/src/swarm/pair.rs
new file mode 100644
index 0000000..c1a196b
--- /dev/null
+++ b/src/swarm/pair.rs
@@ -0,0 +1,78 @@
+//! Which side of a pair of nodes leads, for duties only one side needs to do (sending
+//! keepalive PINGs).
+//!
+//! Both sides decide on their own, from the two keys alone, and agree. The leader is
+//! chosen per pair by one bit of a hash of both keys, not by comparing the keys: a
+//! global order would make the node with the smallest key lead every one of its pairs
+//! and the one with the largest lead none, so every such duty would pile up on the same
+//! few nodes. With the hash, each node leads about half of its pairs.
+
+use sha2::{Digest, Sha256};
+
+const LABEL: &[u8] = b"continuum/pair-leader/1";
+
+/// Whether `local` leads its pair with `peer`. Exactly one of `leads(a, b)` and
+/// `leads(b, a)` is true for any two different keys.
+pub(crate) fn leads(local: &str, peer: &str) -> bool {
+ let (low, high) = if local <= peer {
+ (local, peer)
+ } else {
+ (peer, local)
+ };
+ let mut hash = Sha256::new();
+ // Length-prefixed fields, so no two different pairs hash the same input.
+ for field in [LABEL, low.as_bytes(), high.as_bytes()] {
+ hash.update((field.len() as u32).to_be_bytes());
+ hash.update(field);
+ }
+ let leader = if hash.finalize()[0] & 1 == 0 {
+ low
+ } else {
+ high
+ };
+ leader == local
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn key(n: u32) -> String {
+ format!("{:064x}", u64::from(n).wrapping_mul(0x9e37_79b9_7f4a_7c15))
+ }
+
+ #[test]
+ fn exactly_one_side_leads() {
+ for a in 0..50 {
+ for b in 0..50 {
+ if a != b {
+ assert_ne!(leads(&key(a), &key(b)), leads(&key(b), &key(a)));
+ }
+ }
+ }
+ }
+
+ #[test]
+ fn leading_is_spread_across_nodes() {
+ let nodes: Vec<String> = (0..64).map(key).collect();
+ let mut smallest_key_leads = 0;
+ for (i, node) in nodes.iter().enumerate() {
+ let led = nodes
+ .iter()
+ .filter(|peer| *peer != node && leads(node, peer))
+ .count();
+ // 63 pairs per node: about half each; far from all or none.
+ assert!((10..=53).contains(&led), "node {i} leads {led} of 63 pairs");
+ for peer in &nodes {
+ if node < peer && leads(node, peer) {
+ smallest_key_leads += 1;
+ }
+ }
+ }
+ // Not simply the smaller key of each pair (64 * 63 / 2 = 2016 pairs).
+ assert!(
+ (800..=1216).contains(&smallest_key_leads),
+ "{smallest_key_leads}"
+ );
+ }
+}
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index b6c9700..3d7da32 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -2,24 +2,22 @@
//! swarm-wide forget records, and dispatch of inbound messages.
use std::sync::Arc;
+use std::sync::atomic::Ordering;
use std::time::Duration;
use anyhow::{Context, Result, anyhow, bail};
use automerge::AutoCommit;
-use sha2::{Digest, Sha256};
use tokio::sync::Mutex;
use tokio::time::{Instant, MissedTickBehavior};
use tracing::{debug, info, instrument, trace, warn};
use crate::document::replica::{Epoch, EpochRelation};
use crate::document::store::{self, StoreError};
-use crate::state::{AppState, HelloState, Liveness, PeerMembership, PeerState};
+use crate::state::{AppState, HelloState, Liveness, PeerState, Standing};
use crate::swarm::sync::{SyncEvent, handle_sync};
+use crate::swarm::topology;
use crate::transport::{self, ParsedMessage};
-const HELLO_RETRY_INITIAL: Duration = Duration::from_secs(1);
-const HELLO_RETRY_MAX: Duration = Duration::from_secs(30);
-
const PEER_TICK: Duration = Duration::from_millis(500);
const META_ROOT: &str = "_continuum";
@@ -98,24 +96,63 @@ async fn apply_forget_records(state: &AppState) {
// --- keepalive and self-healing ----------------------------------------------
-fn heads_digest(doc: &mut AutoCommit) -> transport::HeadsDigest {
- let mut heads = doc.get_heads();
- heads.sort_unstable();
- let mut hasher = Sha256::new();
- for head in &heads {
- hasher.update(head.0);
- }
- hasher.finalize().into()
+/// Sends `peer` a PING with our keepalive, noting the time for its round trip.
+pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> {
+ let keepalive = local_keepalive(state, peer).await;
+ state.lock_rtt().ping_sent(peer, Instant::now());
+ state
+ .transport
+ .send_message(peer, &ParsedMessage::Ping(keepalive))
+ .await
}
-pub(crate) async fn local_keepalive(state: &AppState) -> transport::Keepalive {
- // AutoCommit::get_heads takes &mut self (it closes the pending transaction).
- // Digest and epoch come from the same locked replica, so they always match.
- let mut replica = state.doc.write().await;
+/// Our keepalive for `peer`.
+pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive {
+ // Heads and epoch come from the same locked replica, so they always match. The heads
+ // are cached (`Replica`), so the read lock is enough.
+ let (heads, epoch) = {
+ let replica = state.doc.read().await;
+ (
+ transport::HeadsSummary::of(replica.heads().to_vec()),
+ replica.epoch,
+ )
+ };
+ let syncs_with_you = match state.get_peer_state(peer).await {
+ Some(peer_state) => {
+ let peer_state = peer_state.lock().await;
+ peer_state.liveness == Liveness::Active
+ && matches!(peer_state.standing, Standing::Member { .. })
+ && peer_state.epoch == Some(epoch)
+ }
+ None => false,
+ };
transport::Keepalive {
- heads_digest: heads_digest(&mut replica.doc),
- epoch: replica.epoch,
+ heads,
+ epoch,
+ ready: state.is_ready().await,
+ syncs_with_you,
+ }
+}
+
+/// What a peer's keepalive says about it.
+async fn note_keepalive(state: &AppState, peer: &str, remote: &transport::Keepalive) {
+ if let Some(peer_state) = state.get_peer_state(peer).await {
+ let mut peer_state = peer_state.lock().await;
+ 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);
+ crate::swarm::relay::pull_if_behind(state, peer, remote).await;
+ 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()));
+ }
}
+ crate::swarm::source::reported(state, peer, remote.ready, Some(remote.syncs_with_you));
}
async fn heal_if_diverged(
@@ -124,8 +161,9 @@ async fn heal_if_diverged(
local: &transport::Keepalive,
remote: &transport::Keepalive,
) {
- // Digests of documents from different epochs are never equal and never converge.
- if local.epoch != remote.epoch {
+ // Heads of documents from different epochs are never equal and never converge.
+ // And a peer that isn't our neighbour isn't ours to keep in step.
+ if local.epoch != remote.epoch || !topology::is_neighbour(state, peer) {
return;
}
@@ -135,18 +173,17 @@ async fn heal_if_diverged(
let heal = {
let mut peer_state = peer_state_mutex.lock().await;
- if local.heads_digest == remote.heads_digest {
+ if local.heads == remote.heads {
// Same document: whatever answer we were waiting for, there is nothing left
// to send. (Automerge answers only when it has something to say, so a sync
// message to a peer that already has everything gets no reply at all.)
- peer_state.sync_in_step();
+ peer_state.sync.reply_settled();
}
- peer_state.needs_heal(
- &local.heads_digest,
- &remote.heads_digest,
- &state.timing,
- Instant::now(),
- )
+ // While we catch up, we hold back sync with everyone but the source on purpose:
+ // differing heads and idle sync are expected then, not a reason to heal.
+ let catching_up = !state.initial_sync_done.load(Ordering::Acquire);
+ !catching_up
+ && peer_state.needs_heal(&local.heads, &remote.heads, &state.timing, Instant::now())
};
if heal {
@@ -171,7 +208,7 @@ pub(crate) async fn remember_peer(state: &AppState, peer: &str) {
let mut peer_states = state.peer_states.write().await;
if !peer_states.contains_key(peer) {
- peer_states.entry(peer.to_owned()).or_default();
+ peer_states.insert(peer.to_owned(), state.new_peer_state(peer));
info!(
peer = %peer,
peer_count = peer_states.len(),
@@ -191,9 +228,10 @@ pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()>
let claimed = {
let mut peer_state = peer_state_mutex.lock().await;
if peer_state.hello_state == HelloState::NotSent {
- peer_state.hello_state = HelloState::Sending {
- next_delay: HELLO_RETRY_INITIAL,
- };
+ // A new handshake: retries start over.
+ peer_state.hello_state = HelloState::Sending;
+ peer_state.hello_backoff.reset();
+ peer_state.hello_attempts = 0;
true
} else {
trace!(peer = %peer, hello_state = ?peer_state.hello_state, "not sending");
@@ -206,7 +244,7 @@ pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()>
}
debug!(to = %peer, "Sending HELLO");
- transmit_hello(state, peer, HELLO_RETRY_INITIAL).await
+ transmit_hello(state, peer).await
}
async fn retry_hello_if_due(state: &AppState, peer: &str) -> Result<()> {
@@ -214,43 +252,41 @@ async fn retry_hello_if_due(state: &AppState, peer: &str) -> Result<()> {
return Ok(());
};
- let delay = {
+ {
let mut peer_state = peer_state_mutex.lock().await;
match peer_state.hello_state {
- HelloState::AwaitingReply {
- next_attempt,
- next_delay,
- } if Instant::now() >= next_attempt => {
- peer_state.hello_state = HelloState::Sending { next_delay };
- next_delay
+ HelloState::AwaitingReply { next_attempt } if Instant::now() >= next_attempt => {
+ peer_state.hello_state = HelloState::Sending;
}
_ => return Ok(()),
}
- };
+ }
- debug!(
- peer = %peer,
- next_retry_ms = delay.as_millis(),
- "No HELLO_REPLY yet; re-sending HELLO"
- );
- transmit_hello(state, peer, delay).await
+ debug!(peer = %peer, "No HELLO_REPLY yet; re-sending HELLO");
+ transmit_hello(state, peer).await
}
/// Sends a HELLO without holding any lock, then schedules the next retry.
/// Success and failure both end in AwaitingReply: Mycelium accepting the message
/// says nothing about whether the peer received it.
-async fn transmit_hello(state: &AppState, peer: &str, delay: Duration) -> Result<()> {
+async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> {
+ let swarm_id = state.swarm_id().await;
+ let auth = state
+ .swarm_key
+ .as_ref()
+ .map(|key| key.hello_tag(swarm_id.as_deref().unwrap_or(""), state.local_id(), peer))
+ .unwrap_or_default();
+ state.lock_rtt().hello_sent(peer, Instant::now());
let result = state
.transport
.send_message(
peer,
&ParsedMessage::Hello(transport::HelloRequest {
+ swarm_id,
epoch: state.local_epoch(),
- auth: state
- .swarm_key
- .as_ref()
- .map(|key| key.hello_tag(state.local_id(), peer))
- .unwrap_or_default(),
+ auth,
+ ready: state.is_ready().await,
+ run_id: state.run_id,
}),
)
.await;
@@ -258,16 +294,14 @@ async fn transmit_hello(state: &AppState, peer: &str, delay: Duration) -> Result
if let Some(peer_state_mutex) = state.get_peer_state(peer).await {
let mut peer_state = peer_state_mutex.lock().await;
- if matches!(peer_state.hello_state, HelloState::Sending { .. }) {
- let (wait, next_delay) = match peer_state.liveness {
- Liveness::Dormant { .. } => {
- (state.timing.dormant_probe, state.timing.dormant_probe)
- }
- Liveness::Active => (delay, delay.saturating_mul(2).min(HELLO_RETRY_MAX)),
+ if peer_state.hello_state == HelloState::Sending {
+ peer_state.hello_sent(Instant::now());
+ let wait = match peer_state.liveness {
+ Liveness::Dormant { .. } => state.timing.dormant_probe,
+ Liveness::Active => peer_state.hello_backoff.next(),
};
peer_state.hello_state = HelloState::AwaitingReply {
next_attempt: Instant::now() + wait,
- next_delay,
};
}
}
@@ -281,36 +315,14 @@ async fn ensure_handshake(state: &AppState, peer: &str) {
}
}
-/// Sends a HELLO_REPLY if one is owed and our swarm ID is known.
-/// Otherwise it stays owed and the peer worker tries again.
-async fn send_owed_hello_reply(state: &AppState, peer: &str) {
- if state.swarm_id().await.is_none() {
- return;
- }
-
- let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
- return;
- };
-
- let owed = std::mem::take(&mut peer_state_mutex.lock().await.reply_owed);
- if !owed {
- return;
- }
-
- if let Err(error) = send_hello_reply(state, peer).await {
- warn!(peer = %peer, error = %format!("{error:#}"), "HELLO_REPLY failed; will retry");
- peer_state_mutex.lock().await.reply_owed = true;
- }
-}
-
-async fn flush_owed_hello_replies(state: &AppState) {
- let peers: Vec<String> = state.peer_states.read().await.keys().cloned().collect();
- for peer in &peers {
- send_owed_hello_reply(state, peer).await;
- }
-}
-
-async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> {
+/// Answers a HELLO that carried `run_id`; `away` is how long before it we had last heard
+/// from the peer.
+async fn send_hello_reply(
+ state: &AppState,
+ peer: &str,
+ run_id: transport::RunId,
+ away: Option<Duration>,
+) -> Result<()> {
let swarm_id = state
.swarm_id()
.await
@@ -330,6 +342,7 @@ async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> {
let peer_state = peer_state_mutex.lock().await;
if peer_state.liveness == Liveness::Active
&& peer_state.hello_state == HelloState::Established
+ && topology::is_neighbour(state, &known)
{
peers.push(known);
}
@@ -340,15 +353,17 @@ async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> {
.as_ref()
.map(|key| key.hello_reply_tag(&swarm_id, state.local_id(), peer))
.unwrap_or_default();
+ let ready = state.is_ready().await;
let reply = transport::HelloReply {
swarm_id,
peers,
epoch: state.local_epoch(),
auth,
+ ready,
+ run_id,
+ away_ms: away.map(|away| away.as_millis() as u64),
};
- let ready = state.is_ready().await;
-
debug!(to = %peer, ready, "Sending HELLO_REPLY");
state
@@ -370,6 +385,9 @@ async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[S
remember_peer(state, discovered_peer).await;
+ if !topology::is_neighbour(state, discovered_peer) {
+ continue;
+ }
if let Err(error) = send_hello_once(state, discovered_peer).await {
warn!(
peer = %discovered_peer,
@@ -380,6 +398,26 @@ async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[S
}
}
+/// New neighbours get our HELLO (we may not have known them before: the member list is
+/// how a partial mesh discovers them). Former neighbours get no more HELLOs, keepalives
+/// or sync openings from us; their session is reset, so we don't keep pushing changes
+/// to them either. They are still answered if they contact us.
+async fn apply_neighbour_change(state: &AppState, change: topology::NeighbourChange) {
+ for peer in &change.added {
+ remember_peer(state, peer).await;
+ if let Err(error) = send_hello_once(state, peer).await {
+ warn!(peer = %peer, error = %format!("{error:#}"), "HELLO to new neighbour failed; will retry");
+ }
+ }
+ for peer in &change.removed {
+ if let Some(peer_state) = state.get_peer_state(peer).await {
+ let mut peer_state = peer_state.lock().await;
+ peer_state.stop_hello_retries();
+ peer_state.reset_sync();
+ }
+ }
+}
+
// --- peer worker -------------------------------------------------------------
pub(crate) async fn run_peer_worker(state: AppState) {
@@ -406,12 +444,17 @@ pub(crate) async fn run_peer_worker(state: AppState) {
for (peer, peer_state_mutex) in &peers {
let now = Instant::now();
+ // Only neighbours are kept alive and judged; others we only answer.
+ let neighbour = topology::is_neighbour(&state, peer);
let (became_dormant, ping_due, unanswered) = {
let mut peer_state = peer_state_mutex.lock().await;
- let became_dormant = ready && peer_state.check_dormancy(&state.timing, now);
- let we_ping = state.local_id() < peer.as_str();
- let ping_due = peer_state.take_keepalive_due(&state.timing, we_ping, now);
+ let became_dormant =
+ ready && neighbour && peer_state.check_dormancy(&state.timing, now);
+ let ping_due = neighbour && peer_state.take_keepalive_due(&state.timing, now);
+ if !neighbour {
+ peer_state.stop_hello_retries();
+ }
let unanswered = peer_state.take_overdue_sync_reply(now);
(became_dormant, ping_due, unanswered)
};
@@ -432,13 +475,10 @@ pub(crate) async fn run_peer_worker(state: AppState) {
);
}
- if ping_due {
- let local = local_keepalive(&state).await;
- if let Err(error) = state
- .transport
- .send_message(peer, &ParsedMessage::Ping(local))
- .await
- {
+ let probe_due =
+ !ping_due && crate::swarm::source::probe_due(&state, peer, Instant::now());
+ if ping_due || probe_due {
+ if let Err(error) = send_ping(&state, peer).await {
debug!(peer = %peer, error = %format!("{error:#}"), "Keepalive PING failed");
}
}
@@ -446,12 +486,25 @@ 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");
}
-
- send_owed_hello_reply(&state, peer).await;
}
if last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL) {
apply_forget_records(&state).await;
+ crate::swarm::members::ensure_registered(&state).await;
+ crate::swarm::stability::tick(&state).await;
+ crate::swarm::stats::measure_saved_size(&state);
+ {
+ let known = state.peer_states.read().await;
+ state.traffic.retain(|peer| known.contains_key(peer));
+ state
+ .throughput
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
+ .retain(|peer| known.contains_key(peer));
+ }
+ if let Some(change) = topology::refresh(&state).await {
+ apply_neighbour_change(&state, change).await;
+ }
last_forget_check = Some(Instant::now());
}
}
@@ -459,11 +512,17 @@ pub(crate) async fn run_peer_worker(state: AppState) {
// --- peer membership ---------------------------------------------------------
+/// Makes `peer` a member of `swarm_id`, after its valid HELLO (`from_hello`) or
+/// HELLO_REPLY. Either completes the handshake: one round trip proves both sides to
+/// each other (the HELLO the sender's key, the reply the swarm ID and the responder's
+/// key), so neither side needs a HELLO of its own. `ready` is what the message said.
async fn establish_peer(
state: &AppState,
peer: &str,
swarm_id: String,
-) -> Result<Vec<transport::SyncEnvelope>> {
+ from_hello: bool,
+ ready: bool,
+) -> Result<()> {
let expected_swarm = state.swarm_id().await;
if let Some(expected) = &expected_swarm {
@@ -479,7 +538,7 @@ async fn establish_peer(
let peer_state_mutex = state.get_or_create_peer_state(peer).await;
let mut peer_state = peer_state_mutex.lock().await;
- if let PeerMembership::Member { swarm_id: existing } = &peer_state.membership {
+ if let Standing::Member { swarm_id: existing } = &peer_state.standing {
if existing != &swarm_id {
bail!(
"Peer {peer} changed swarm identity from {} to {}",
@@ -494,34 +553,30 @@ async fn establish_peer(
state.set_expected_swarm_id(swarm_id.clone()).await?;
}
- if matches!(peer_state.membership, PeerMembership::Unknown) {
- peer_state.membership = PeerMembership::Member {
+ if !matches!(peer_state.standing, Standing::Member { .. }) {
+ peer_state.standing = Standing::Member {
swarm_id: swarm_id.clone(),
};
}
- // First reply completes the handshake: stop retries and sync from a clean slate.
- // Duplicate replies (caused by our own retries) don't reset again.
- if peer_state.hello_state != HelloState::Established {
+ // Stop our retries and sync from a clean slate. A HELLO always resets: the peer
+ // starts over with us. Duplicate replies (caused by our own retries) don't.
+ peer_state.reports_ready = Some(ready);
+ if from_hello || peer_state.hello_state != HelloState::Established {
peer_state.hello_state = HelloState::Established;
peer_state.reset_sync();
- info!(peer = %peer, "Handshake complete; sync state reset");
+ info!(peer = %peer, from_hello, "Handshake complete; sync state reset");
}
- let pending = std::mem::take(&mut peer_state.pending_inbound_sync);
-
- info!(peer = %peer, swarm_id = %swarm_id, pending_messages = pending.len(), "Peer membership established");
- Ok(pending)
+ info!(peer = %peer, swarm_id = %swarm_id, "Peer membership established");
+ Ok(())
}
#[instrument(skip(state), fields(peer = %peer))]
pub(crate) async fn peer_swarm(state: &AppState, peer: &str) -> Option<String> {
let peer_state_mutex = state.get_peer_state(peer).await?;
let peer_state = peer_state_mutex.lock().await;
- match &peer_state.membership {
- PeerMembership::Unknown => None,
- PeerMembership::Member { swarm_id } => Some(swarm_id.clone()),
- }
+ peer_state.standing.swarm_id().map(str::to_owned)
}
// --- epochs ------------------------------------------------------------------
@@ -545,6 +600,8 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool
// (resetting both sides again) while this fresh sync is under way.
peer_state.reset_sync();
peer_state.last_sync_rx = Some(Instant::now());
+ // Whether it would sync with us was said about its old epoch.
+ peer_state.syncs_with_us = None;
}
previous
};
@@ -597,7 +654,12 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool {
let local = state.local_id();
let proof = match message {
- ParsedMessage::Hello(hello) => Some(key.verify_hello(&hello.auth, peer, local)),
+ ParsedMessage::Hello(hello) => Some(key.verify_hello(
+ &hello.auth,
+ hello.swarm_id.as_deref().unwrap_or(""),
+ peer,
+ local,
+ )),
ParsedMessage::HelloReply(reply) => {
Some(key.verify_hello_reply(&reply.auth, &reply.swarm_id, peer, local))
}
@@ -609,8 +671,8 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool {
remember_peer(state, peer).await;
if let Some(peer_state) = state.get_peer_state(peer).await {
let mut peer_state = peer_state.lock().await;
- if !peer_state.authenticated {
- peer_state.authenticated = true;
+ if peer_state.standing == Standing::Stranger {
+ peer_state.standing = Standing::Admitted;
debug!(peer = %peer, "Peer proved the swarm key");
}
}
@@ -622,11 +684,10 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool {
// we stop syncing our document to it, not just listening to it.
if let Some(peer_state) = state.get_peer_state(peer).await {
let mut peer_state = peer_state.lock().await;
- if peer_state.authenticated {
+ if peer_state.standing.is_admitted() {
warn!(peer = %peer, "Peer no longer proves the swarm key; dropping its membership");
}
- peer_state.authenticated = false;
- peer_state.membership = PeerMembership::Unknown;
+ peer_state.standing = Standing::Stranger;
peer_state.hello_state = HelloState::NotSent;
peer_state.reset_sync();
}
@@ -643,7 +704,7 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool {
}
None => {
let admitted = match state.get_peer_state(peer).await {
- Some(peer_state) => peer_state.lock().await.authenticated,
+ Some(peer_state) => peer_state.lock().await.standing.is_admitted(),
None => false,
};
if !admitted {
@@ -666,19 +727,31 @@ pub(crate) async fn handle_message(
return Ok(());
}
+ // A reply to a HELLO from an earlier run of ours (late, or queued in Mycelium while
+ // we were down) says nothing about now; the peer answers our current HELLO.
+ if let transport::ParsedMessage::HelloReply(reply) = &message {
+ if reply.run_id != state.run_id {
+ state.stats.stale_hello_reply();
+ debug!(peer = %peer, "Ignoring a HELLO_REPLY to a HELLO from an earlier run");
+ return Ok(());
+ }
+ }
+
if !admit(state, &peer, &message).await {
return Ok(());
}
remember_peer(state, &peer).await;
- // Any inbound message proves liveness.
- let reactivated = state
- .get_or_create_peer_state(&peer)
- .await
- .lock()
- .await
- .mark_seen(Instant::now());
+ // 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;
+ (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");
@@ -689,50 +762,38 @@ pub(crate) async fn handle_message(
//debug!(from = %peer, message = ?message, "Received message");
match message {
- transport::ParsedMessage::Hello(_) => {
+ transport::ParsedMessage::Hello(hello) => {
// A HELLO means the peer is (re)starting its handshake with us, e.g. after a
- // restart. Whatever our sync state believed about it is stale.
- {
- let peer_state_mutex = state.get_or_create_peer_state(&peer).await;
- let mut peer_state = peer_state_mutex.lock().await;
- peer_state.reset_sync();
- peer_state.reply_owed = true;
+ // restart. It becomes a member right away; our reply makes us one for it.
+ let Some(swarm_id) = state.swarm_id().await else {
+ // We are joining ourselves; the peer retries until we know our swarm.
+ debug!(peer = %peer, "HELLO before we know our swarm; not answering yet");
+ return Ok(());
+ };
+ if let Some(theirs) = &hello.swarm_id {
+ if theirs != &swarm_id {
+ bail!("Incompatible swarm at {peer}: ours is {swarm_id}, it claims {theirs}");
+ }
}
+ establish_peer(state, &peer, swarm_id, true, hello.ready).await?;
- // Sends now if we know our swarm ID; otherwise stays owed.
- send_owed_hello_reply(state, &peer).await;
-
- // Our own HELLO goes out regardless of whether the reply could be sent.
- if let Err(error) = send_hello_once(state, &peer).await {
- warn!(peer = %peer, error = %format!("{error:#}"), "HELLO failed; will retry");
+ let away = previously_seen.map(|seen| seen.elapsed());
+ if let Err(error) = send_hello_reply(state, &peer, hello.run_id, away).await {
+ // The peer retries its HELLO.
+ warn!(peer = %peer, error = %format!("{error:#}"), "HELLO_REPLY failed");
}
- // Restart sync with the fresh state (no-op until the peer is a validated member).
let _ = state
.sync_events
- .send(SyncEvent::PeerNeedsSync(peer.clone()));
+ .send(SyncEvent::PeerValidated(peer.clone()));
}
transport::ParsedMessage::HelloReply(reply) => {
- let pending = establish_peer(state, &peer, reply.swarm_id.clone()).await?;
+ state.lock_rtt().hello_reply_received(&peer, Instant::now());
+ crate::swarm::stability::note_away(state, reply.away_ms.map(Duration::from_millis));
+ establish_peer(state, &peer, reply.swarm_id.clone(), false, reply.ready).await?;
discover_peers(state, &peer, &reply.peers).await;
- // Our swarm ID may have just become known: pay any replies we owe.
- flush_owed_hello_replies(state).await;
-
- if !pending.is_empty() {
- debug!(
- to = %peer,
- pending_messages = pending.len(),
- "Replaying pending sync messages"
- );
- for envelope in pending {
- if let Err(error) = handle_sync(state, &peer, envelope).await {
- warn!(peer = %peer, error = %format!("{error:#}"), "Discarding invalid pending sync envelope");
- }
- }
- }
-
state
.sync_events
.send(SyncEvent::PeerValidated(peer.clone()))
@@ -752,7 +813,8 @@ pub(crate) async fn handle_message(
if reactivated || peer_swarm(state, &peer).await.is_none() {
ensure_handshake(state, &peer).await;
}
- let local = local_keepalive(state).await;
+ note_keepalive(state, &peer, &remote).await;
+ let local = local_keepalive(state, &peer).await;
heal_if_diverged(state, &peer, &local, &remote).await;
if let Err(error) = state
.transport
@@ -762,11 +824,16 @@ pub(crate) async fn handle_message(
debug!(peer = %peer, error = %format!("{error:#}"), "PONG failed");
}
}
+ transport::ParsedMessage::Blob(blob) => {
+ crate::swarm::blob::handle(state, &peer, blob).await;
+ }
transport::ParsedMessage::Pong(remote) => {
+ state.lock_rtt().pong_received(&peer, Instant::now());
if reactivated || peer_swarm(state, &peer).await.is_none() {
ensure_handshake(state, &peer).await;
}
- let local = local_keepalive(state).await;
+ note_keepalive(state, &peer, &remote).await;
+ let local = local_keepalive(state, &peer).await;
heal_if_diverged(state, &peer, &local, &remote).await;
}
}
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
new file mode 100644
index 0000000..703bc3f
--- /dev/null
+++ b/src/swarm/relay.rs
@@ -0,0 +1,179 @@
+//! Passing changes on in a partial mesh (`swarm::topology`).
+//!
+//! A node pushes its own changes to its neighbours, but nothing it receives goes any
+//! further by itself. So after applying changes from one neighbour, a node *announces*
+//! its new heads to its other neighbours with a PING (a keepalive: heads, about 100
+//! bytes, not the changes). A neighbour that sees heads it lacks *pulls*: it enters the
+//! announced heads as the peer's heads in its sync state, Automerge asks for what's
+//! missing (`need`), and the announcer answers with exactly those changes. Each node
+//! receives a change about once; the rest is announcements. Regular keepalives work the
+//! same way, so any gap is noticed within a keepalive interval.
+//!
+//! Announcements are coalesced: the first applied change schedules one announcement
+//! round `ANNOUNCE_DELAY` later, and every change applied until then rides along. A
+//! round costs one PING and one PONG per neighbour (plus the pulls), whatever the
+//! write rate; the delay is the price per hop. Without coalescing, every applied sync
+//! message would announce to every neighbour, and announcements would outnumber the
+//! data they announce (about 15 to 1 at 250 nodes).
+//!
+//! In a full mesh the writer reaches everyone itself, so nothing is announced.
+
+use std::collections::HashSet;
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+
+use tokio::time::Instant;
+use tracing::debug;
+
+use crate::state::{AppState, Standing};
+use crate::swarm::peer::send_ping;
+use crate::swarm::sync::SyncEvent;
+use crate::swarm::topology;
+use crate::transport::{self, HeadsSummary};
+
+/// A pull that brought nothing for this long no longer blocks pulling from another peer.
+const PULL_TIMEOUT: Duration = Duration::from_secs(15);
+
+/// How long an announcement waits for further changes to ride along. Added once per
+/// hop to propagation; well below a round trip over Mycelium.
+const ANNOUNCE_DELAY: Duration = Duration::from_millis(200);
+
+/// After changes from `from` were applied and our heads changed: tell our other
+/// neighbours, in the next announcement round.
+pub(crate) async fn announce(state: &AppState, from: &str) {
+ if !state.initial_sync_done.load(Ordering::Acquire) || topology::is_full_mesh(state) {
+ return;
+ }
+ let schedule = {
+ let mut pending = state
+ .announce_pending
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner());
+ match pending.as_mut() {
+ Some(senders) => {
+ senders.insert(from.to_owned());
+ false
+ }
+ None => {
+ *pending = Some(HashSet::from([from.to_owned()]));
+ true
+ }
+ }
+ };
+ if !schedule {
+ return;
+ }
+ let state = state.clone();
+ tokio::spawn(async move {
+ tokio::time::sleep(ANNOUNCE_DELAY).await;
+ let senders = state
+ .announce_pending
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
+ .take()
+ .unwrap_or_default();
+ announce_now(&state, &senders).await;
+ });
+}
+
+/// One announcement round. The peer whose changes we announce has them already, so a
+/// round with a single sender skips it; with several, each lacks the others' changes,
+/// so every neighbour is told.
+async fn announce_now(state: &AppState, senders: &HashSet<String>) {
+ let skip = match senders.len() {
+ 1 => senders.iter().next().cloned(),
+ _ => None,
+ };
+ let neighbours: Vec<String> = state
+ .peer_states
+ .read()
+ .await
+ .keys()
+ .filter(|peer| {
+ skip.as_deref() != Some(peer.as_str()) && topology::is_neighbour(state, peer)
+ })
+ .cloned()
+ .collect();
+ for peer in neighbours {
+ if let Err(error) = send_ping(state, &peer).await {
+ debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed");
+ }
+ }
+}
+
+/// A keepalive from `peer`: if it has heads we lack, pull them (one peer at a time).
+pub(crate) async fn pull_if_behind(state: &AppState, peer: &str, remote: &transport::Keepalive) {
+ if remote.epoch != state.local_epoch() || !topology::is_neighbour(state, peer) {
+ return;
+ }
+ // The common case, the same heads, needs only the cached heads (read lock).
+ if HeadsSummary::of(state.doc.read().await.heads().to_vec()) == remote.heads {
+ return;
+ }
+ let missing = match &remote.heads {
+ HeadsSummary::Heads(heads) => {
+ // AutoCommit's lookups take &mut self.
+ let mut replica = state.doc.write().await;
+ let missing = heads
+ .iter()
+ .any(|head| replica.doc.get_change_by_hash(head).is_none());
+ if !missing {
+ return;
+ }
+ Some(heads.clone())
+ }
+ // Can't tell which heads are missing; the digest differs from ours, though.
+ HeadsSummary::Digest(_) => None,
+ };
+
+ let now = Instant::now();
+ {
+ let mut pulling = state
+ .pulling
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner());
+ let busy = matches!(
+ &*pulling,
+ Some((other, since)) if other != peer && now < *since + PULL_TIMEOUT
+ );
+ if busy {
+ return;
+ }
+ *pulling = Some((peer.to_owned(), now));
+ }
+
+ let Some(peer_state) = state.get_peer_state(peer).await else {
+ return;
+ };
+ {
+ let mut peer_state = peer_state.lock().await;
+ if !matches!(peer_state.standing, Standing::Member { .. }) {
+ return;
+ }
+ match missing {
+ Some(heads) => {
+ // Automerge asks for whatever these heads need that we don't have.
+ peer_state.sync.state.their_heads = Some(heads);
+ peer_state.sync.state.in_flight = false;
+ }
+ // A fresh session's opening carries our heads and bloom filter; the answer
+ // brings what we lack.
+ None => peer_state.reset_sync(),
+ }
+ }
+ debug!(peer = %peer, "Neighbour has changes we lack; pulling");
+ let _ = state
+ .sync_events
+ .send(SyncEvent::PeerNeedsSync(peer.to_owned()));
+}
+
+/// A sync message from `peer` was applied: a pull from it is answered.
+pub(crate) fn pulled(state: &AppState, peer: &str) {
+ let mut pulling = state
+ .pulling
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner());
+ if pulling.as_ref().is_some_and(|(p, _)| p == peer) {
+ *pulling = None;
+ }
+}
diff --git a/src/swarm/rtt.rs b/src/swarm/rtt.rs
new file mode 100644
index 0000000..ed6cc73
--- /dev/null
+++ b/src/swarm/rtt.rs
@@ -0,0 +1,179 @@
+//! Round-trip times to peers, measured from traffic we send anyway: PING → PONG and
+//! HELLO → HELLO_REPLY. No probes of our own.
+//!
+//! A sample only counts when exactly one request was outstanding: with two, we can't tell
+//! which one the answer belongs to (an answer to the older one would look too fast).
+//! Samples are smoothed (exponential moving average, weight 1/4) to ride out jitter.
+//!
+//! What we use is the *effective* round-trip time: the smoothed value, raised to the age
+//! of the oldest request still unanswered (the round trip takes at least that long), so
+//! a peer that stops answering drifts away by itself. Without a recent sample (nodes
+//! move; an old value says little) it is unknown. Silence alone doesn't count: we don't
+//! ask non-neighbours anything, so their quiet is ours.
+
+use std::collections::HashMap;
+use std::time::Duration;
+
+use tokio::time::Instant;
+
+#[derive(Debug, Default)]
+pub(crate) struct RttTable {
+ pings: HashMap<String, Outstanding>,
+ hellos: HashMap<String, Outstanding>,
+ smoothed: HashMap<String, Sample>,
+}
+
+#[derive(Debug, Clone, Copy)]
+struct Sample {
+ rtt: Duration,
+ at: Instant,
+}
+
+#[derive(Debug, Clone, Copy)]
+struct Outstanding {
+ sent: Instant,
+ count: u32,
+}
+
+impl RttTable {
+ pub(crate) fn ping_sent(&mut self, peer: &str, now: Instant) {
+ Self::sent(&mut self.pings, peer, now);
+ }
+
+ pub(crate) fn pong_received(&mut self, peer: &str, now: Instant) {
+ if let Some(sample) = Self::answered(&mut self.pings, peer, now) {
+ self.record(peer, sample, now);
+ }
+ }
+
+ pub(crate) fn hello_sent(&mut self, peer: &str, now: Instant) {
+ Self::sent(&mut self.hellos, peer, now);
+ }
+
+ pub(crate) fn hello_reply_received(&mut self, peer: &str, now: Instant) {
+ if let Some(sample) = Self::answered(&mut self.hellos, peer, now) {
+ self.record(peer, sample, now);
+ }
+ }
+
+ /// The effective round-trip time: the smoothed value if sampled within `fresh_for`,
+ /// raised to the age of the oldest unanswered request; None without a recent sample.
+ pub(crate) fn effective(
+ &self,
+ peer: &str,
+ now: Instant,
+ fresh_for: Duration,
+ ) -> Option<Duration> {
+ let sample = self
+ .smoothed
+ .get(peer)
+ .filter(|sample| now.saturating_duration_since(sample.at) < fresh_for)?;
+ let waiting = [self.pings.get(peer), self.hellos.get(peer)]
+ .into_iter()
+ .flatten()
+ .map(|outstanding| now.saturating_duration_since(outstanding.sent))
+ .max()
+ .unwrap_or_default();
+ Some(sample.rtt.max(waiting))
+ }
+
+ /// The effective round-trip time of every peer that has one.
+ pub(crate) fn all_effective(
+ &self,
+ now: Instant,
+ fresh_for: Duration,
+ ) -> impl Iterator<Item = (&str, Duration)> {
+ self.smoothed.keys().filter_map(move |peer| {
+ self.effective(peer, now, fresh_for)
+ .map(|rtt| (peer.as_str(), rtt))
+ })
+ }
+
+ fn sent(map: &mut HashMap<String, Outstanding>, peer: &str, now: Instant) {
+ map.entry(peer.to_owned())
+ .and_modify(|outstanding| outstanding.count += 1)
+ .or_insert(Outstanding {
+ sent: now,
+ count: 1,
+ });
+ }
+
+ fn answered(
+ map: &mut HashMap<String, Outstanding>,
+ peer: &str,
+ now: Instant,
+ ) -> Option<Duration> {
+ let outstanding = map.remove(peer)?;
+ (outstanding.count == 1).then(|| now.saturating_duration_since(outstanding.sent))
+ }
+
+ fn record(&mut self, peer: &str, sample: Duration, now: Instant) {
+ let rtt = match self.smoothed.get(peer) {
+ Some(old) => (old.rtt * 3 + sample) / 4,
+ None => sample,
+ };
+ self.smoothed
+ .insert(peer.to_owned(), Sample { rtt, at: now });
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ const FRESH: Duration = Duration::from_secs(300);
+
+ #[test]
+ fn samples_only_unambiguous_answers_and_smooths() {
+ let mut table = RttTable::default();
+ let start = Instant::now();
+ let at = |ms| start + Duration::from_millis(ms);
+ table.ping_sent("a", start);
+ table.pong_received("a", at(100));
+ assert_eq!(
+ table.effective("a", at(100), FRESH),
+ Some(Duration::from_millis(100))
+ );
+
+ table.ping_sent("a", at(100));
+ table.pong_received("a", at(300));
+ assert_eq!(
+ table.effective("a", at(300), FRESH),
+ Some(Duration::from_millis(125))
+ );
+
+ // Two PINGs out: the PONG can't be attributed.
+ table.ping_sent("a", at(300));
+ table.ping_sent("a", at(400));
+ table.pong_received("a", at(450));
+ assert_eq!(
+ table.effective("a", at(450), FRESH),
+ Some(Duration::from_millis(125))
+ );
+
+ // An unsolicited PONG changes nothing.
+ table.pong_received("b", start);
+ assert_eq!(table.effective("b", start, FRESH), None);
+ }
+
+ #[test]
+ fn unanswered_requests_raise_it_and_old_samples_expire() {
+ let mut table = RttTable::default();
+ let start = Instant::now();
+ table.ping_sent("a", start);
+ table.pong_received("a", start + Duration::from_millis(50));
+
+ // A PING unanswered for 40 s: at least 40 s.
+ let sent = start + Duration::from_secs(60);
+ table.ping_sent("a", sent);
+ let later = sent + Duration::from_secs(40);
+ assert_eq!(
+ table.effective("a", later, FRESH),
+ Some(Duration::from_secs(40))
+ );
+
+ // No sample within the freshness window: unknown.
+ let stale = start + FRESH + Duration::from_secs(1);
+ assert_eq!(table.effective("a", stale, FRESH), None);
+ }
+}
diff --git a/src/swarm/source.rs b/src/swarm/source.rs
new file mode 100644
index 0000000..378c40a
--- /dev/null
+++ b/src/swarm/source.rs
@@ -0,0 +1,210 @@
+//! The catch-up source: while a node catches up (joining, or rejoining a newer epoch,
+//! until its initial sync is done), it syncs with one peer only.
+//!
+//! Every peer would otherwise send us what we lack, all at once: a node with an empty
+//! document gets the whole document from each of its peers, and a node that is merely
+//! behind gets the missing changes from each. Automerge can't deduplicate across peers.
+//! So one suitable peer becomes the source, preferably the closest by round-trip time
+//! (a clearly closer one gets a 1 s head start to claim it, else the first to ask);
+//! the others wait until the initial sync is done, and then only exchange heads. Messages from waiting
+//! peers are still applied; we just don't answer them yet.
+//!
+//! A source must be ready and willing to sync with us, as far as we know (its sync
+//! messages and keepalives say so). It counts as alive while any message from it
+//! arrives, so a slow transfer of a large document never looks like a stall. When it
+//! is quiet for `Timing::catch_up_probe`, we PING it; after three unanswered probes, or
+//! when it says it isn't ready, or twice in a row that it wouldn't sync with us (the
+//! first time may be a handshake still under way), the next peer takes over.
+
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+
+use tokio::time::Instant;
+use tracing::{debug, info};
+
+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.
+const CLOSER_PEER_HEAD_START: Duration = Duration::from_secs(1);
+
+/// Probes without an answer before the source is given up.
+const UNANSWERED_PROBES: u32 = 3;
+
+/// Keepalives in a row saying "I wouldn't sync with you" before the source is given up.
+const REFUSALS: u32 = 2;
+
+fn lock_preference(state: &AppState) -> std::sync::MutexGuard<'_, Option<Instant>> {
+ state
+ .catch_up_preference
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
+}
+
+fn catching_up(state: &AppState) -> bool {
+ !state.initial_sync_done.load(Ordering::Acquire)
+}
+
+fn lock(state: &AppState) -> std::sync::MutexGuard<'_, Option<CatchUpSource>> {
+ state
+ .catch_up_source
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
+}
+
+/// Returns how long to hold back sync with `peer`, if at all: while catching up, with
+/// everyone but the source. Picks the source if there is none (or it went quiet).
+pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Option<Duration> {
+ if !catching_up(state) {
+ return None;
+ }
+ let stall = state.timing.catch_up_probe * UNANSWERED_PROBES;
+ if peer_state.reports_ready == Some(false) || peer_state.syncs_with_us == Some(false) {
+ // Not a candidate; its next message re-evaluates this, the timer is a fallback.
+ return Some(stall);
+ }
+
+ let now = Instant::now();
+ let mut source = lock(state);
+ let quiet = match &*source {
+ Some(current) if current.peer == peer => return None,
+ Some(current) if now < current.last_heard + stall => {
+ return Some(current.last_heard + stall - now);
+ }
+ Some(current) => Some(current.peer.clone()),
+ None => None,
+ };
+
+ // Prefer the closest peer: the source sends us everything we lack, so its distance
+ // matters most. If another peer we measured is clearly closer (by a fifth), it gets
+ // a short head start to claim the role; it is asked in the same sync pass. After
+ // that, whoever asks first takes it (the closer one may not be suitable after all).
+ let rtt = {
+ let rtts = state.lock_rtt();
+ let fresh_for = state.timing.rtt_fresh_for();
+ let mine = rtts.effective(peer, now, fresh_for);
+ let closer = rtts.all_effective(now, fresh_for).any(|(other, rtt)| {
+ other != peer
+ && quiet.as_deref() != Some(other)
+ && mine.is_none_or(|mine| rtt * 5 < mine * 4)
+ });
+ if closer {
+ let mut preference = lock_preference(state);
+ let since = *preference.get_or_insert(now);
+ if now < since + CLOSER_PEER_HEAD_START {
+ return Some(since + CLOSER_PEER_HEAD_START - now);
+ }
+ }
+ mine
+ };
+ *lock_preference(state) = None;
+ let rtt_ms = rtt.map(|rtt| rtt.as_millis() as u64);
+
+ match quiet {
+ Some(previous) => info!(
+ peer = %peer,
+ previous = %previous,
+ rtt_ms,
+ quiet_secs = stall.as_secs_f32(),
+ "Catch-up source stopped answering; catching up from another peer"
+ ),
+ None => debug!(peer = %peer, rtt_ms, "Catching up from this peer first"),
+ }
+ state.stats.catch_up_source_chosen();
+ *source = Some(CatchUpSource {
+ peer: peer.to_owned(),
+ last_heard: now,
+ last_probe: None,
+ refusals: 0,
+ });
+ None
+}
+
+/// Any message from `peer` arrived.
+pub(crate) fn heard_from(state: &AppState, peer: &str) {
+ if let Some(current) = lock(state).as_mut().filter(|current| current.peer == peer) {
+ current.last_heard = Instant::now();
+ }
+}
+
+/// What `peer` said about itself, in a sync message (`syncs_with_us: None`) or a
+/// keepalive. Gives up the source if that rules it out.
+pub(crate) fn reported(state: &AppState, peer: &str, ready: bool, syncs_with_us: Option<bool>) {
+ if !catching_up(state) {
+ return;
+ }
+ let reason = {
+ let mut source = lock(state);
+ let Some(current) = source.as_mut().filter(|current| current.peer == peer) else {
+ return;
+ };
+ let reason = if !ready {
+ Some("it isn't ready itself")
+ } else {
+ match syncs_with_us {
+ Some(false) => {
+ current.refusals += 1;
+ (current.refusals >= REFUSALS).then_some("it wouldn't sync with us")
+ }
+ Some(true) => {
+ current.refusals = 0;
+ None
+ }
+ None => None,
+ }
+ };
+ if reason.is_some() {
+ *source = None;
+ }
+ reason
+ };
+ if let Some(reason) = reason {
+ debug!(peer = %peer, reason, "Giving up the catch-up source; trying another peer");
+ resync_everyone(state);
+ }
+}
+
+/// Whether to PING `peer` now: it is the source and has been quiet for a probe interval.
+pub(crate) fn probe_due(state: &AppState, peer: &str, now: Instant) -> bool {
+ if !catching_up(state) {
+ return false;
+ }
+ let probe = state.timing.catch_up_probe;
+ let mut source = lock(state);
+ let Some(current) = source.as_mut().filter(|current| current.peer == peer) else {
+ return false;
+ };
+ let due = now >= current.last_heard + probe
+ && current.last_probe.is_none_or(|last| now >= last + probe);
+ if due {
+ current.last_probe = Some(now);
+ }
+ 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) {
+ *lock(state) = None;
+ *lock_preference(state) = None;
+ 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));
+ }
+ });
+}
+
+/// After giving up a source: let every peer be considered again.
+pub(crate) fn resync_everyone(state: &AppState) {
+ let state = state.clone();
+ 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));
+ }
+ });
+}
diff --git a/src/swarm/stability.rs b/src/swarm/stability.rs
new file mode 100644
index 0000000..9d79b17
--- /dev/null
+++ b/src/swarm/stability.rs
@@ -0,0 +1,270 @@
+//! Whether this node is a stable member of the swarm, judged by itself.
+//!
+//! Long-running, well-connected nodes make a good backbone; laptops that sleep and
+//! phones that lose signal don't. Each node measures what fraction of the last 24 hours
+//! it had at least one live peer, and publishes only a coarse tier in its registry entry
+//! (`_continuum/members/<pk>/tier`): `stable` or `transient`, written only when it
+//! changes, with hysteresis. So the swarm learns about stability from the replicated
+//! document alone, without anyone measuring anyone else, and the tier changes rarely.
+//!
+//! * transient → stable: at least 24 h observed, connected at least 95% of it;
+//! * stable → transient: connected less than 80% (over at least an hour observed).
+//!
+//! After a restart a node has no history. If its entry says `stable` and a peer reports
+//! (in its HELLO_REPLY) that it last heard from us less than an hour ago, the node was
+//! only briefly away and keeps its tier, starting with a full day's credit. Otherwise it
+//! starts over as transient. Time the process didn't run (a laptop asleep) counts as
+//! disconnected.
+
+use std::collections::VecDeque;
+use std::time::Duration;
+
+use tokio::time::Instant;
+use tracing::{info, warn};
+
+use crate::state::{AppState, HelloState};
+use crate::swarm::members::{self, Tier};
+
+const WINDOW: Duration = Duration::from_secs(24 * 3600);
+const BUCKET: Duration = Duration::from_secs(5 * 60);
+const BECOME_STABLE: f64 = 0.95;
+const BECOME_TRANSIENT: f64 = 0.80;
+/// A day observed (the oldest bucket may be partly outside the window).
+const FULL_DAY: Duration = WINDOW.saturating_sub(BUCKET);
+/// Less than this observed is too little to demote a stable node.
+const MIN_TO_DEMOTE: Duration = Duration::from_secs(3600);
+/// An absence shorter than this keeps a stable node stable across a restart.
+const BRIEF_ABSENCE: Duration = Duration::from_secs(3600);
+/// How long after becoming ready we wait for a peer to tell us how long we were away.
+const AWAY_REPORT_GRACE: Duration = Duration::from_secs(30);
+/// A gap between two measurements longer than this means we didn't run (asleep).
+const MAX_TICK_GAP: Duration = Duration::from_secs(30);
+
+#[derive(Debug, Default)]
+pub(crate) struct Stability {
+ buckets: VecDeque<Bucket>,
+ last_tick: Option<Instant>,
+ /// The shortest absence any peer reported to us in this run.
+ away: Option<Duration>,
+ ready_since: Option<Instant>,
+ /// The decision about the tier we found after a restart is made.
+ settled: bool,
+}
+
+#[derive(Debug, Clone, Copy)]
+struct Bucket {
+ start: Instant,
+ observed: Duration,
+ connected: Duration,
+}
+
+impl Stability {
+ fn record(&mut self, now: Instant, connected: bool) {
+ let Some(last) = self.last_tick.replace(now) else {
+ return;
+ };
+ let elapsed = now.saturating_duration_since(last).min(WINDOW);
+ let connected = connected && elapsed <= MAX_TICK_GAP;
+ if self
+ .buckets
+ .back()
+ .is_none_or(|bucket| now.saturating_duration_since(bucket.start) >= BUCKET)
+ {
+ self.buckets.push_back(Bucket {
+ start: now,
+ observed: Duration::ZERO,
+ connected: Duration::ZERO,
+ });
+ }
+ let bucket = self.buckets.back_mut().expect("just ensured");
+ bucket.observed += elapsed;
+ if connected {
+ bucket.connected += elapsed;
+ }
+ while self
+ .buckets
+ .front()
+ .is_some_and(|bucket| now.saturating_duration_since(bucket.start) > WINDOW)
+ {
+ self.buckets.pop_front();
+ }
+ }
+
+ /// A full day of connectivity, for a stable node that was only briefly away.
+ fn credit(&mut self, now: Instant) {
+ self.buckets.push_front(Bucket {
+ start: now,
+ observed: WINDOW,
+ connected: WINDOW,
+ });
+ }
+
+ fn observed(&self) -> Duration {
+ self.buckets.iter().map(|bucket| bucket.observed).sum()
+ }
+
+ fn fraction(&self) -> f64 {
+ let observed = self.observed().as_secs_f64();
+ if observed == 0.0 {
+ return 0.0;
+ }
+ let connected: f64 = self.buckets.iter().map(|b| b.connected.as_secs_f64()).sum();
+ connected / observed
+ }
+
+ /// The tier we should have, given the one we have now.
+ fn assess(&self, current: Tier) -> Tier {
+ match current {
+ Tier::Transient if self.observed() >= FULL_DAY && self.fraction() >= BECOME_STABLE => {
+ Tier::Stable
+ }
+ Tier::Stable
+ if self.observed() >= MIN_TO_DEMOTE && self.fraction() < BECOME_TRANSIENT =>
+ {
+ Tier::Transient
+ }
+ tier => tier,
+ }
+ }
+}
+
+fn lock(state: &AppState) -> std::sync::MutexGuard<'_, Stability> {
+ state
+ .stability
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
+}
+
+/// The fraction of the observed time (up to a day) we had a live peer, and how much
+/// time was observed.
+pub(crate) fn measured(state: &AppState) -> (f64, Duration) {
+ let stability = lock(state);
+ (stability.fraction(), stability.observed())
+}
+
+/// A HELLO_REPLY told us how long ago that peer last heard from us before this run.
+pub(crate) fn note_away(state: &AppState, away: Option<Duration>) {
+ if let Some(away) = away {
+ let mut stability = lock(state);
+ stability.away = Some(stability.away.map_or(away, |known| known.min(away)));
+ }
+}
+
+/// Measures, and updates our tier in the registry when it should change. Called every
+/// few seconds by the peer worker.
+pub(crate) async fn tick(state: &AppState) {
+ let now = Instant::now();
+ let mut connected = false;
+ let peers: Vec<_> = state.peer_states.read().await.values().cloned().collect();
+ for peer_state in peers {
+ let peer_state = peer_state.lock().await;
+ if peer_state.hello_state == HelloState::Established
+ && peer_state.is_alive(&state.timing, now)
+ {
+ connected = true;
+ break;
+ }
+ }
+ lock(state).record(now, connected);
+
+ if !state.is_ready().await {
+ return;
+ }
+ let local = state.local_id().to_owned();
+ let current = members::tier(&state.doc.read().await.doc, &local);
+
+ let desired = {
+ let mut stability = lock(state);
+ let ready_since = *stability.ready_since.get_or_insert(now);
+ if stability.settled {
+ stability.assess(current)
+ } else {
+ let brief = stability.away.is_some_and(|away| away < BRIEF_ABSENCE);
+ if current == Tier::Stable && !brief && now < ready_since + AWAY_REPORT_GRACE {
+ return; // a peer may still tell us we were only briefly away
+ }
+ stability.settled = true;
+ match current {
+ Tier::Stable if brief => {
+ stability.credit(now);
+ info!("Back after a brief absence; staying a stable member");
+ Tier::Stable
+ }
+ // Restarted without a report that we were only briefly away.
+ Tier::Stable => Tier::Transient,
+ Tier::Transient => Tier::Transient,
+ }
+ }
+ };
+ if desired == current {
+ return;
+ }
+
+ let fraction = lock(state).fraction();
+ match state
+ .mutate_doc(move |doc| members::set_tier(doc, &local, desired))
+ .await
+ {
+ Ok(()) => info!(
+ tier = desired.as_str(),
+ connected_percent = (fraction * 100.0).round(),
+ "Stability tier changed"
+ ),
+ Err(error) => warn!(error = %error, "Failed to record our stability tier"),
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn run(stability: &mut Stability, start: Instant, hours: u64, connected: impl Fn(u64) -> bool) {
+ let step = Duration::from_secs(5);
+ let steps = hours * 3600 / 5;
+ for i in 0..=steps {
+ stability.record(start + step * i as u32, connected(i));
+ }
+ }
+
+ #[test]
+ fn becomes_stable_after_a_well_connected_day() {
+ let mut stability = Stability::default();
+ let start = Instant::now();
+ run(&mut stability, start, 12, |_| true);
+ assert_eq!(stability.assess(Tier::Transient), Tier::Transient);
+ run(
+ &mut stability,
+ start + Duration::from_secs(12 * 3600),
+ 13,
+ |_| true,
+ );
+ assert_eq!(stability.assess(Tier::Transient), Tier::Stable);
+ }
+
+ #[test]
+ fn hysteresis_between_the_thresholds() {
+ let mut stability = Stability::default();
+ // Connected 9 of every 10 measurements: 90%.
+ run(&mut stability, Instant::now(), 25, |i| i % 10 != 0);
+ assert_eq!(stability.assess(Tier::Transient), Tier::Transient);
+ assert_eq!(stability.assess(Tier::Stable), Tier::Stable);
+ }
+
+ #[test]
+ fn falls_back_to_transient_when_often_disconnected() {
+ let mut stability = Stability::default();
+ // Connected half the time.
+ run(&mut stability, Instant::now(), 2, |i| i % 2 == 0);
+ assert_eq!(stability.assess(Tier::Stable), Tier::Transient);
+ }
+
+ #[test]
+ fn time_asleep_counts_as_disconnected() {
+ let mut stability = Stability::default();
+ let start = Instant::now();
+ run(&mut stability, start, 1, |_| true);
+ // The process didn't run for two hours, then one more measurement.
+ stability.record(start + Duration::from_secs(3 * 3600), true);
+ assert!(stability.fraction() < 0.5);
+ }
+}
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
new file mode 100644
index 0000000..eb34276
--- /dev/null
+++ b/src/swarm/stats.rs
@@ -0,0 +1,284 @@
+//! Node-wide figures for /admin/stats that no other part keeps. Recorded where things
+//! happen anyway; the one exception is the document's saved size (see
+//! `measure_saved_size`).
+
+use std::sync::{Mutex, MutexGuard};
+use std::time::Duration;
+
+use tokio::time::Instant;
+use tracing::{debug, warn};
+
+use crate::document::replica::Epoch;
+use crate::state::AppState;
+
+/// Saving takes time proportional to the document; we spend at most about 1% of it on
+/// measuring, and measure at least this long apart, and only after a change.
+const SAVED_SIZE_MIN_INTERVAL: Duration = Duration::from_secs(60);
+const SAVED_SIZE_DUTY: u32 = 100;
+
+#[derive(Debug)]
+pub(crate) struct NodeStats {
+ started: Instant,
+ inner: Mutex<Inner>,
+}
+
+#[derive(Debug, Default)]
+struct Inner {
+ caught_up: Option<CaughtUp>,
+ catch_up_sources: u32,
+ max_heads: usize,
+ stale_hello_replies: u64,
+ saved_size: Option<SavedSize>,
+ measuring: bool,
+ /// The document version (`AppState::change_tx`) and epoch the saved size is of.
+ measured_version: Option<(u64, Epoch)>,
+ propagation: Propagation,
+}
+
+/// Upper bounds (inclusive, milliseconds) of the propagation delay buckets; a last
+/// bucket takes the rest.
+const PROPAGATION_BUCKETS_MS: [u64; 11] = [
+ 100, 250, 500, 1_000, 2_500, 5_000, 10_000, 30_000, 60_000, 300_000, 3_600_000,
+];
+
+/// How long changes written elsewhere took to reach us: from the writer's commit (its
+/// clock) to applying them here (ours), so clock skew between the two is part of it.
+#[derive(Debug, Clone, Copy, Default)]
+pub(crate) struct Propagation {
+ /// Per bucket of `PROPAGATION_BUCKETS_MS`, then the rest; not cumulative.
+ pub(crate) buckets: [u64; PROPAGATION_BUCKETS_MS.len() + 1],
+ pub(crate) count: u64,
+ pub(crate) sum_ms: u64,
+ /// Changes stamped later than our clock said on arrival (counted in the first
+ /// bucket): the writer's clock is ahead of ours.
+ pub(crate) ahead: u64,
+}
+
+impl Propagation {
+ fn record(&mut self, delay_ms: i64) {
+ if delay_ms < 0 {
+ self.ahead += 1;
+ }
+ let delay_ms = delay_ms.max(0) as u64;
+ let bucket = PROPAGATION_BUCKETS_MS
+ .iter()
+ .position(|&bound| delay_ms <= bound)
+ .unwrap_or(PROPAGATION_BUCKETS_MS.len());
+ self.buckets[bucket] += 1;
+ self.count += 1;
+ self.sum_ms += delay_ms;
+ }
+
+ /// Each bucket's upper bound (None for the last) and count.
+ pub(crate) fn buckets(&self) -> impl Iterator<Item = (Option<u64>, u64)> + '_ {
+ PROPAGATION_BUCKETS_MS
+ .iter()
+ .copied()
+ .map(Some)
+ .chain([None])
+ .zip(self.buckets.iter().copied())
+ }
+}
+
+/// The document's size as saved (compressed), when it was measured and how long that
+/// took (copy and save).
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct SavedSize {
+ pub(crate) bytes: u64,
+ pub(crate) at: Instant,
+ pub(crate) took: Duration,
+}
+
+/// When the initial sync finished, and what it took.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct CaughtUp {
+ /// Since the process started (handshakes included).
+ pub(crate) after: Duration,
+ /// Sync payload received until then, from all peers.
+ pub(crate) sync_bytes: u64,
+}
+
+/// A copy of the figures.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct Snapshot {
+ pub(crate) uptime: Duration,
+ pub(crate) caught_up: Option<CaughtUp>,
+ pub(crate) catch_up_sources: u32,
+ pub(crate) max_heads: usize,
+ pub(crate) stale_hello_replies: u64,
+ pub(crate) propagation: Propagation,
+}
+
+impl Default for NodeStats {
+ fn default() -> Self {
+ Self {
+ started: Instant::now(),
+ inner: Mutex::default(),
+ }
+ }
+}
+
+impl NodeStats {
+ fn lock(&self) -> MutexGuard<'_, Inner> {
+ self.inner
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
+ }
+
+ pub(crate) fn caught_up(&self, sync_bytes: u64) {
+ let after = self.started.elapsed();
+ self.lock()
+ .caught_up
+ .get_or_insert(CaughtUp { after, sync_bytes });
+ }
+
+ /// A catch-up source was chosen (the first, or a replacement).
+ pub(crate) fn catch_up_source_chosen(&self) {
+ self.lock().catch_up_sources += 1;
+ }
+
+ /// Our document has `count` heads (after applying a sync message).
+ pub(crate) fn heads_seen(&self, count: usize) {
+ let mut inner = self.lock();
+ inner.max_heads = inner.max_heads.max(count);
+ }
+
+ /// Changes from elsewhere were applied, `delays_ms` after they were committed.
+ pub(crate) fn propagated(&self, delays_ms: impl IntoIterator<Item = i64>) {
+ let mut inner = self.lock();
+ for delay_ms in delays_ms {
+ inner.propagation.record(delay_ms);
+ }
+ }
+
+ pub(crate) fn stale_hello_reply(&self) {
+ self.lock().stale_hello_replies += 1;
+ }
+
+ /// The last measurement of the saved size, if any.
+ pub(crate) fn saved_size(&self) -> Option<SavedSize> {
+ self.lock().saved_size
+ }
+
+ /// Whether a measurement of the document at `version` is due; if so, it counts as
+ /// started.
+ fn start_measuring(&self, now: Instant, version: (u64, Epoch)) -> bool {
+ let mut inner = self.lock();
+ let due = !inner.measuring
+ && inner.measured_version != Some(version)
+ && inner.saved_size.is_none_or(|last| {
+ now >= last.at + SAVED_SIZE_MIN_INTERVAL.max(last.took * SAVED_SIZE_DUTY)
+ });
+ if due {
+ inner.measuring = true;
+ }
+ due
+ }
+
+ /// The document at `version` was saved for another reason (an export): a
+ /// measurement for free.
+ pub(crate) fn saved_size_measured(&self, bytes: u64, took: Duration, version: (u64, Epoch)) {
+ let mut inner = self.lock();
+ inner.saved_size = Some(SavedSize {
+ bytes,
+ at: Instant::now(),
+ took,
+ });
+ inner.measured_version = Some(version);
+ }
+
+ fn finish_measuring(&self, measured: Option<(SavedSize, (u64, Epoch))>) {
+ let mut inner = self.lock();
+ inner.measuring = false;
+ if let Some((measured, version)) = measured {
+ inner.saved_size = Some(measured);
+ inner.measured_version = Some(version);
+ }
+ }
+
+ pub(crate) fn snapshot(&self) -> Snapshot {
+ let inner = self.lock();
+ Snapshot {
+ uptime: self.started.elapsed(),
+ caught_up: inner.caught_up,
+ catch_up_sources: inner.catch_up_sources,
+ max_heads: inner.max_heads,
+ stale_hello_replies: inner.stale_hello_replies,
+ propagation: inner.propagation,
+ }
+ }
+}
+
+/// The document's version for measuring: bumped by every write and applied sync
+/// message (`AppState::notify_store_changed`), together with the epoch (a switch
+/// replaces the document).
+pub(crate) fn document_version(state: &AppState) -> (u64, Epoch) {
+ (*state.change_tx.borrow(), state.local_epoch())
+}
+
+/// Measures the document's saved size in the background when due: the first time, then
+/// only after the document changed, at least a minute apart and spending no more than
+/// about 1% of the time on it (a document that takes 2 s to copy and save is measured
+/// at most every 200 s). An idle swarm measures once. Called by the peer worker every
+/// few seconds.
+///
+/// The copy holds the read lock (writes wait for it); the save runs without any lock.
+pub(crate) fn measure_saved_size(state: &AppState) {
+ // Taken before the copy: a change during it makes another measurement due.
+ let version = document_version(state);
+ if !state.stats.start_measuring(Instant::now(), version) {
+ return;
+ }
+ let state = state.clone();
+ tokio::spawn(async move {
+ let start = Instant::now();
+ 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
+ })
+ .await
+ {
+ Ok(bytes) => {
+ let took = start.elapsed();
+ debug!(
+ bytes,
+ took_ms = took.as_millis() as u64,
+ "Measured the saved document size"
+ );
+ Some((
+ SavedSize {
+ bytes,
+ at: Instant::now(),
+ took,
+ },
+ version,
+ ))
+ }
+ Err(error) => {
+ warn!(error = %error, "Measuring the saved document size failed");
+ None
+ }
+ };
+ state.stats.finish_measuring(measured);
+ });
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn propagation_buckets() {
+ let mut propagation = Propagation::default();
+ for delay_ms in [-20, 100, 101, 7_200_000] {
+ propagation.record(delay_ms);
+ }
+ let buckets: Vec<_> = propagation.buckets().filter(|(_, n)| *n > 0).collect();
+ assert_eq!(buckets, [(Some(100), 2), (Some(250), 1), (None, 1)]);
+ assert_eq!(propagation.ahead, 1);
+ assert_eq!(propagation.count, 4);
+ assert_eq!(propagation.sum_ms, 7_200_201);
+ }
+}
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index d82238e..80d6d56 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -4,7 +4,6 @@
use std::collections::HashSet;
use std::sync::Arc;
use std::sync::atomic::Ordering;
-use std::time::Duration;
use anyhow::{Context, Result, anyhow, bail};
use automerge::sync::SyncDoc;
@@ -12,20 +11,18 @@ use tokio::sync::mpsc;
use tokio::time::Instant;
use tracing::{debug, info, trace, warn};
-use crate::state::{AppState, Liveness, PeerMembership, PeerState, PendingOutbound};
+use crate::state::{AppState, Liveness, Sending, Standing};
use crate::swarm::peer::{peer_swarm, send_hello_once};
+use crate::swarm::source;
+use crate::swarm::throughput::SyncBytes;
use crate::transport::{self, ParsedMessage};
-const INITIAL_SYNC_RETRY: Duration = Duration::from_millis(500);
-const MAX_SYNC_RETRY: Duration = Duration::from_secs(30);
-
#[derive(Debug)]
pub(crate) enum SyncEvent {
PeerValidated(String),
DocumentChanged,
MessageReceived(String),
PeerNeedsSync(String),
- RetryDelivery(String),
}
/// Store paths touched by one batch of document changes.
@@ -34,12 +31,20 @@ pub(crate) struct ChangeSet {
pub(crate) paths: Vec<Vec<String>>,
}
-async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool) -> Result<()> {
+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
+ // starts once it is here (`remake::catchup`).
+ trace!("Waiting for the snapshot download; not syncing yet");
+ return Ok(());
+ }
let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
trace!("Peer was forgotten; skipping synchronization");
return Ok(());
};
+ let we_ready = state.is_ready().await;
+
let validated_swarm = {
let peer_state = peer_state_mutex.lock().await;
@@ -58,25 +63,41 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
return Ok(());
}
- let PeerMembership::Member { swarm_id } = &peer_state.membership else {
+ let Standing::Member { swarm_id } = &peer_state.standing else {
return Ok(());
};
- if !retry_delivery {
- let we_open = state.local_id() < peer;
- if let Some(wait) = peer_state.opening_wait(we_open, Instant::now()) {
- trace!(
- wait_ms = wait.as_millis(),
- "Leaving the opening sync message to the peer"
- );
- 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));
- });
- return Ok(());
- }
+ if let Some(wait) = source::defer(state, peer, &peer_state) {
+ trace!(
+ wait_ms = wait.as_millis(),
+ "Catching up from another peer first; not syncing with this one yet"
+ );
+ // 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));
+ });
+ return Ok(());
+ }
+
+ if peer_state
+ .sync
+ .leave_opening(we_ready, peer_state.reports_ready)
+ {
+ // Its own opening (or anything else from it) brings us back here.
+ trace!("Leaving the opening sync message to the peer, which isn't ready");
+ return Ok(());
+ }
+
+ if peer_state.sync.state.their_heads.is_none()
+ && !crate::swarm::topology::is_neighbour(state, peer)
+ {
+ // We only answer peers that aren't our neighbours.
+ trace!("Not opening sync with a peer that isn't our neighbour");
+ return Ok(());
}
swarm_id.clone()
@@ -91,104 +112,113 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
bail!("Peer {peer} belongs to a different swarm");
}
- let existing = {
- let mut guard = peer_state_mutex.lock().await;
- let peer_state = &mut *guard;
- match &peer_state.pending_outbound_sync {
- Some(entry) if retry_delivery => Some((entry.message.clone(), entry.next_attempt)),
- Some(_) => {
- peer_state.needs_generate = true;
- return Ok(());
- }
- None => None,
+ {
+ let mut peer_state = peer_state_mutex.lock().await;
+ if let Some(sending) = &mut peer_state.sync.sending {
+ sending.regenerate = true;
+ return Ok(());
}
- };
-
- let encoded = match existing {
- Some((message, next_attempt)) => {
- if Instant::now() < next_attempt {
- return Ok(());
- }
- message
+ if peer_state
+ .sync
+ .retry_at
+ .is_some_and(|at| Instant::now() < at)
+ {
+ // The retry timer enqueues the peer once it's due.
+ return Ok(());
}
- None if retry_delivery => return Ok(()),
- None => {
- let doc = Arc::clone(&state.doc);
- let psm = Arc::clone(&peer_state_mutex);
- let transport = Arc::clone(&state.transport);
- let peer_owned = peer.to_owned();
+ }
- let generated = tokio::task::spawn_blocking(
- move || -> Result<Option<transport::EncodedMessage>> {
- // Label the message with the epoch of the document it was generated
- // from, read under the same lock.
- let (epoch, message) = {
- let mut replica = doc.blocking_write();
- let mut peer_state = psm.blocking_lock();
- peer_state.needs_generate = false;
- let epoch = replica.epoch;
- let sent_before = peer_state.sync_state.sent_hashes.clone();
- let message = {
- let sync_doc = replica.doc.sync();
- sync_doc.generate_sync_message(&mut peer_state.sync_state)
- };
- // A peer whose heads are ours already has everything; Automerge
- // answers only when it has something to say, so it will stay
- // silent. Waiting for an answer would only end in a pointless
- // resend.
- let in_step = peer_state.sync_state.their_heads.as_ref()
- == Some(&replica.doc.get_heads());
- if in_step {
- peer_state.sync_in_step();
- } else if message.is_some() {
- let sent = peer_state
- .sync_state
- .sent_hashes
- .difference(&sent_before)
- .copied()
- .collect();
- peer_state.expect_sync_reply(sent, Instant::now());
- }
- (epoch, message)
- };
- let Some(message) = message else {
- return Ok(None);
- };
- let envelope = transport::SyncEnvelope {
- swarm_id,
- epoch,
- message,
+ let encoded = {
+ let doc = Arc::clone(&state.doc);
+ let psm = Arc::clone(&peer_state_mutex);
+ let transport = Arc::clone(&state.transport);
+ let peer_owned = peer.to_owned();
+ let ready = state.is_ready().await;
+ // Before this message is counted.
+ let bytes_before = SyncBytes::of(&state.traffic, peer);
+
+ let generated =
+ tokio::task::spawn_blocking(move || -> Result<Option<transport::EncodedMessage>> {
+ // Label the message with the epoch of the document it was generated
+ // from, read under the same lock.
+ let (epoch, message) = {
+ let mut replica = doc.blocking_write();
+ let mut peer_state = psm.blocking_lock();
+ let epoch = replica.epoch;
+ let sent_before = peer_state.sync.state.sent_hashes.clone();
+ let message = {
+ let sync_doc = replica.doc.sync();
+ sync_doc.generate_sync_message(&mut peer_state.sync.state)
};
- transport
- .encode(&peer_owned, &ParsedMessage::Sync(envelope))
- .map(Some)
- },
- )
+ // A peer whose heads are ours already has everything; Automerge
+ // answers only when it has something to say, so it will stay
+ // silent. Waiting for an answer would only end in a pointless
+ // resend.
+ let in_step = peer_state.sync.state.their_heads.as_ref()
+ == Some(&replica.doc.get_heads());
+ let sent: Vec<_> = peer_state
+ .sync
+ .state
+ .sent_hashes
+ .difference(&sent_before)
+ .copied()
+ .collect();
+ if in_step {
+ peer_state.sync.reply_settled();
+ } else if message.is_some() {
+ peer_state
+ .sync
+ .expect_reply(sent.clone(), Instant::now(), bytes_before);
+ }
+ if message.is_some() {
+ // Marked under the same lock as the generation, so no second
+ // message is generated while this one is handed over.
+ peer_state.sync.sending = Some(Sending {
+ sent,
+ regenerate: false,
+ });
+ }
+ (epoch, message)
+ };
+ let Some(message) = message else {
+ return Ok(None);
+ };
+ let envelope = transport::SyncEnvelope {
+ swarm_id,
+ epoch,
+ message,
+ ready,
+ };
+ transport
+ .encode(&peer_owned, &ParsedMessage::Sync(envelope))
+ .map(Some)
+ })
.await
- .context("Sync generation task panicked")??;
-
- let Some(encoded) = generated else {
- return Ok(());
- };
+ .context("Sync generation task panicked")?;
+ let generated = match generated {
+ Ok(generated) => generated,
+ Err(error) => {
+ // Not encodable, so nothing went out: take back what it would have carried.
+ let _ = peer_state_mutex
+ .lock()
+ .await
+ .sync
+ .send_failed(Instant::now());
+ return Err(error);
+ }
+ };
- peer_state_mutex.lock().await.pending_outbound_sync = Some(PendingOutbound {
- message: encoded.clone(),
- retry_delay: INITIAL_SYNC_RETRY,
- next_attempt: Instant::now(),
- });
+ let Some(encoded) = generated else {
+ return Ok(());
+ };
- encoded
- }
+ encoded
};
// No state locks are held across the send.
match state.transport.send_encoded(&encoded).await {
Ok(()) => {
- let regenerate = {
- let mut peer_state = peer_state_mutex.lock().await;
- peer_state.pending_outbound_sync = None;
- std::mem::take(&mut peer_state.needs_generate)
- };
+ let regenerate = peer_state_mutex.lock().await.sync.send_succeeded();
trace!(regenerate, "Outbound sync message accepted by Mycelium");
@@ -202,30 +232,27 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
Ok(())
}
Err(error) => {
- let delay = {
- let mut peer_state = peer_state_mutex.lock().await;
-
- let Some(ref mut entry) = peer_state.pending_outbound_sync else {
- debug!("Pending envelope discarded by a sync reset during delivery");
- return Ok(());
- };
-
- entry.retry_delay = entry.retry_delay.saturating_mul(2).min(MAX_SYNC_RETRY);
- entry.next_attempt = Instant::now() + entry.retry_delay;
- entry.retry_delay
+ let Some(delay) = peer_state_mutex
+ .lock()
+ .await
+ .sync
+ .send_failed(Instant::now())
+ else {
+ debug!("Sync message refused, but a reset already discarded it");
+ return Ok(());
};
warn!(
retry_ms = delay.as_millis(),
error = %format!("{error:#}"),
- "Sync delivery failed; retaining the same outbound envelope"
+ "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::RetryDelivery(peer));
+ let _ = tx.send(SyncEvent::PeerNeedsSync(peer));
});
Ok(())
@@ -239,38 +266,28 @@ pub(crate) async fn handle_sync(
envelope: transport::SyncEnvelope,
) -> Result<()> {
debug!(peer = %peer, "Processing sync envelope");
+ // This message is counted already (`transport::traffic`).
+ let arrived = (Instant::now(), SyncBytes::of(&state.traffic, peer));
let peer_state_mutex = state.get_or_create_peer_state(peer).await;
let needs_hello = {
- let mut peer_state = peer_state_mutex.lock().await;
+ let peer_state = peer_state_mutex.lock().await;
- match &peer_state.membership {
- PeerMembership::Unknown => {
- let duplicate = peer_state.pending_inbound_sync.iter().any(|pending| {
- pending.message == envelope.message
- && pending.swarm_id == envelope.swarm_id
- && pending.epoch == envelope.epoch
- });
- if !duplicate {
- peer_state.pending_inbound_sync.push(envelope.clone());
- info!(
- peer = %peer,
- pending_count = peer_state.pending_inbound_sync.len(),
- "Queued sync envelope pending peer membership"
- );
- } else {
- trace!(peer = %peer, "Ignoring duplicate pending sync envelope");
- }
+ match &peer_state.standing {
+ Standing::Stranger | Standing::Admitted => {
+ // Not processed: our handshake reply resets the session anyway, and the
+ // sender's reply timeout resends what still matters.
+ debug!(peer = %peer, "Dropping sync message from a peer that isn't a member yet");
true
}
- PeerMembership::Member { swarm_id } if swarm_id != &envelope.swarm_id => {
+ Standing::Member { swarm_id } if swarm_id != &envelope.swarm_id => {
bail!(
"Sync swarm mismatch for peer {peer}: validated_swarm={swarm_id}, envelope_swarm={}",
envelope.swarm_id
);
}
- PeerMembership::Member { .. } => false,
+ Standing::Member { .. } => false,
}
};
@@ -296,17 +313,26 @@ pub(crate) async fn handle_sync(
}
let epoch_label = envelope.epoch;
+ let from_ready = envelope.ready;
let applied = {
let doc = Arc::clone(&state.doc);
let peer_state_mutex = Arc::clone(&peer_state_mutex);
+ let stats = Arc::clone(&state.stats);
+ let throughput = Arc::clone(&state.throughput);
+ let rtt = state
+ .lock_rtt()
+ .effective(peer, Instant::now(), state.timing.rtt_fresh_for());
let check_caught_up = !state.initial_sync_done.load(Ordering::Acquire);
+ // While catching up, everything arrives late; that says nothing about propagation.
+ let measure_propagation = !check_caught_up;
let epoch = envelope.epoch;
let message = envelope.message;
let peer = peer.to_owned();
- // Some(caught_up) when applied, None when the message belongs to another epoch.
- tokio::task::spawn_blocking(move || -> Result<Option<bool>> {
+ // Some((caught_up, heads changed)) when applied, None when the message belongs
+ // to another epoch.
+ tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool)>> {
let mut replica = doc.blocking_write();
// Authoritative check, under the same lock as the apply: the document may
@@ -316,52 +342,90 @@ pub(crate) async fn handle_sync(
}
let mut peer_state = peer_state_mutex.blocking_lock();
-
- let heads_before = replica.doc.get_heads();
- let their_heads_before = peer_state.sync_state.their_heads.clone();
+ peer_state.reports_ready = Some(from_ready);
+ let heads_before = replica.heads().to_vec();
replica
.doc
.sync()
- .receive_sync_message(&mut peer_state.sync_state, message)
+ .receive_sync_message(&mut peer_state.sync.state, message)
.context("Automerge synchronization failed")?;
peer_state.last_sync_rx = Some(Instant::now());
- peer_state.sync_reply_received();
+ if let Some(wait) = &peer_state.sync.reply {
+ throughput
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
+ .exchange(&peer, wait.started, arrived, rtt);
+ }
+ peer_state.sync.reply_settled();
- let heads_after = replica.doc.get_heads();
- let unproductive = heads_before == heads_after
- && peer_state.sync_state.their_heads == their_heads_before
- && peer_state.sync_state.their_heads.as_ref() != Some(&heads_after);
- break_sync_stall(&peer, &mut peer_state, unproductive);
+ resend_if_needed(&peer, &mut peer_state.sync.state);
+ replica.refresh_heads();
+ let heads_after = replica.heads().to_vec();
+ stats.heads_seen(heads_after.len());
+ let changed = heads_after != heads_before;
+ if changed && measure_propagation {
+ let now_ms = crate::state::unix_millis() 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();
+ stats.propagated(delays);
+ }
- Ok(Some(
+ Ok(Some((
+ // Only from a ready peer, and only if it has heads at all: empty heads are
+ // "contained" in every document, so they prove nothing. A peer still
+ // 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_state
+ .sync
+ .state
.their_heads
.as_ref()
.is_some_and(|heads| {
- heads
- .iter()
- .all(|h| replica.doc.get_change_by_hash(h).is_some())
+ !heads.is_empty()
+ && heads
+ .iter()
+ .all(|h| replica.doc.get_change_by_hash(h).is_some())
}),
- ))
+ changed,
+ )))
})
.await
.context("Sync task panicked")??
};
- let Some(caught_up) = applied else {
+ let Some((caught_up, changed)) = applied else {
debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch");
return Ok(());
};
+ source::reported(state, peer, from_ready, None);
+
if caught_up && !state.initial_sync_done.swap(true, Ordering::AcqRel) {
info!(peer = %peer, "Initial sync complete");
+ state
+ .stats
+ .caught_up(state.traffic.total().received.sync.bytes);
+ source::finished(state, peer);
}
state.notify_store_changed();
+ crate::swarm::relay::pulled(state, peer);
+ if changed {
+ crate::swarm::relay::announce(state, peer).await;
+ }
+
state
.sync_events
.send(SyncEvent::MessageReceived(peer.to_owned()))
@@ -370,75 +434,38 @@ pub(crate) async fn handle_sync(
Ok(())
}
-/// Exchanges in a row that changed nothing on either side before `sent_hashes` is
-/// cleared, and before a full two-sided reset (re-handshake).
-const STALL_RESEND_NEEDED: u32 = 2;
-const STALL_CLEAR_SENT: u32 = 8;
-const STALL_RESET: u32 = 32;
-
-/// Breaks sync livelocks.
-///
/// Automerge never resends a change it believes the peer has received (`sent_hashes`).
-/// If that change was lost on the way (a dropped or undeliverable message, an epoch
-/// check, a reset of only one side's state), the two sides keep exchanging messages
-/// with differing heads and no changes, as fast as the transport allows. The keepalive
-/// heal never fires then, because sync traffic keeps flowing.
-///
-/// * A change the peer keeps asking for is sent again.
-/// * After a few unproductive exchanges, forget what we sent: the next message carries
-/// everything the peer's bloom filter lacks.
-/// * If that doesn't help either, reset both sides through a fresh handshake.
-fn break_sync_stall(peer: &str, peer_state: &mut PeerState, unproductive: bool) {
- if !unproductive {
- peer_state.stalled_exchanges = 0;
+/// If the message carrying it was lost while later ones arrive, the peer keeps asking for
+/// it (`need`, from the heads we advertise) and we keep not sending it: a livelock the
+/// reply timeout can't see (answers keep coming) and the keepalive heal can't either
+/// (sync never goes idle). So when the peer needs a change we think it has, what we
+/// think it has is wrong: forget all of it, and the next message carries everything its
+/// bloom filter lacks. (A reordered message can make that send a change twice; cheap.)
+fn resend_if_needed(peer: &str, sync_state: &mut automerge::sync::State) {
+ let Some(need) = &sync_state.their_need else {
return;
- }
- peer_state.stalled_exchanges += 1;
-
- let sync_state = &mut peer_state.sync_state;
-
- // Not on the first unproductive exchange: when messages cross, the peer asks for a
- // change that is merely still in flight, and resending it would only duplicate it.
- if peer_state.stalled_exchanges >= STALL_RESEND_NEEDED {
- if let Some(need) = &sync_state.their_need {
- let before = sync_state.sent_hashes.len();
- sync_state.sent_hashes.retain(|hash| !need.contains(hash));
- let resend = before - sync_state.sent_hashes.len();
- if resend > 0 {
- debug!(peer = %peer, resend, "Peer still needs changes we had sent; sending them again");
- }
- }
- }
-
- match peer_state.stalled_exchanges {
- STALL_CLEAR_SENT => {
- warn!(
- peer = %peer,
- sent = sync_state.sent_hashes.len(),
- their_heads = ?sync_state.their_heads,
- shared_heads = ?sync_state.shared_heads,
- "Sync is stalled; forgetting which changes were sent"
- );
- sync_state.sent_hashes.clear();
- }
- STALL_RESET => {
- warn!(peer = %peer, "Sync is still stalled; resetting it on both sides");
- peer_state.stalled_exchanges = 0;
- peer_state.request_resync();
- }
- _ => {}
+ };
+ if need
+ .iter()
+ .any(|hash| sync_state.sent_hashes.contains(hash))
+ {
+ debug!(
+ peer = %peer,
+ sent = sync_state.sent_hashes.len(),
+ "Peer needs changes we had sent; forgetting what was sent"
+ );
+ sync_state.sent_hashes.clear();
}
}
pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedReceiver<SyncEvent>) {
let mut dirty_peers = HashSet::<String>::new();
- let mut retry_peers = HashSet::<String>::new();
let mut document_dirty = false;
info!("Starting synchronization worker");
loop {
- if dirty_peers.is_empty() && retry_peers.is_empty() && !document_dirty {
+ if dirty_peers.is_empty() && !document_dirty {
match rx.recv().await {
Some(event) => match event {
SyncEvent::PeerValidated(peer)
@@ -446,9 +473,6 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
| SyncEvent::PeerNeedsSync(peer) => {
dirty_peers.insert(peer);
}
- SyncEvent::RetryDelivery(peer) => {
- retry_peers.insert(peer);
- }
SyncEvent::DocumentChanged => {
document_dirty = true;
}
@@ -468,9 +492,6 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
| SyncEvent::PeerNeedsSync(peer) => {
dirty_peers.insert(peer);
}
- SyncEvent::RetryDelivery(peer) => {
- retry_peers.insert(peer);
- }
SyncEvent::DocumentChanged => {
document_dirty = true;
}
@@ -489,30 +510,21 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
peers_guard.keys().cloned().collect()
};
+ // Our own changes go to our neighbours; they pass them on. Others only get
+ // answers to what they send us.
for peer in peer_keys {
- if peer_swarm(&state, &peer).await.is_some() {
+ if peer_swarm(&state, &peer).await.is_some()
+ && crate::swarm::topology::is_neighbour(&state, &peer)
+ {
dirty_peers.insert(peer);
}
}
}
- // Retry existing envelopes separately from generating new ones.
- let retries: Vec<String> = retry_peers.drain().collect();
-
- for peer in retries {
- if let Err(error) = deliver_sync_to_peer(&state, &peer, true).await {
- warn!(
- peer = %peer,
- error = %format!("{error:#}"),
- "Sync delivery retry failed"
- );
- }
- }
-
let peers: Vec<String> = dirty_peers.drain().collect();
for peer in peers {
- if let Err(error) = deliver_sync_to_peer(&state, &peer, false).await {
+ if let Err(error) = deliver_sync_to_peer(&state, &peer).await {
warn!(
peer = %peer,
error = %format!("{error:#}"),
diff --git a/src/swarm/throughput.rs b/src/swarm/throughput.rs
new file mode 100644
index 0000000..f1b9b14
--- /dev/null
+++ b/src/swarm/throughput.rs
@@ -0,0 +1,168 @@
+//! Sync throughput with peers, measured from sync exchanges we have anyway: from sending
+//! a sync message until the peer's answer arrives, the sync bytes that went either way
+//! with that peer. No transfers of our own.
+//!
+//! Only exchanges of at least `MIN_BYTES` count; smaller ones measure the round trip,
+//! not the throughput. The round trip (`swarm::rtt`) is taken off the elapsed time, but
+//! never more than half of it, so a stale round trip can't inflate a sample. What
+//! remains also includes the time the peer took to apply and answer: this is the rate a
+//! peer delivers at in practice, not the raw capacity of the path.
+//!
+//! Samples are smoothed (exponential moving average, weight 1/4). Large exchanges are
+//! rare (catching up, big writes), so estimates are kept however old; they come with
+//! their age.
+
+use std::collections::HashMap;
+use std::time::Duration;
+
+use tokio::time::Instant;
+
+use crate::transport::traffic::Traffic;
+
+/// Smallest exchange that counts: some dozens of Mycelium chunks.
+const MIN_BYTES: u64 = 32 * 1024;
+const WEIGHT: f64 = 0.25;
+
+/// Sync bytes exchanged with a peer so far (from the traffic counters).
+#[derive(Debug, Default, Clone, Copy)]
+pub(crate) struct SyncBytes {
+ sent: u64,
+ received: u64,
+}
+
+impl SyncBytes {
+ pub(crate) fn of(traffic: &Traffic, peer: &str) -> Self {
+ traffic
+ .peer(peer)
+ .map(|t| Self {
+ sent: t.sent.sync.bytes,
+ received: t.received.sync.bytes,
+ })
+ .unwrap_or_default()
+ }
+
+ /// Bytes either way since `earlier` (0 if the counters were dropped meanwhile).
+ fn since(&self, earlier: &SyncBytes) -> u64 {
+ self.sent.saturating_sub(earlier.sent) + self.received.saturating_sub(earlier.received)
+ }
+}
+
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct Estimate {
+ pub(crate) bytes_per_sec: f64,
+ pub(crate) samples: u32,
+ /// When the last sample was taken.
+ pub(crate) at: Instant,
+}
+
+#[derive(Debug, Default)]
+pub(crate) struct ThroughputTable {
+ peers: HashMap<String, Estimate>,
+}
+
+impl ThroughputTable {
+ /// A sync exchange with `peer` finished: it started at `started` with the counters
+ /// at `before`, and its answer arrived at `arrived` with the counters at `after`.
+ pub(crate) fn exchange(
+ &mut self,
+ peer: &str,
+ (started, before): (Instant, SyncBytes),
+ (arrived, after): (Instant, SyncBytes),
+ rtt: Option<Duration>,
+ ) {
+ let bytes = after.since(&before);
+ if bytes < MIN_BYTES {
+ return;
+ }
+ let elapsed = arrived.saturating_duration_since(started);
+ let transfer = elapsed
+ .saturating_sub(rtt.unwrap_or_default())
+ .max(elapsed / 2);
+ if transfer.is_zero() {
+ return;
+ }
+ let sample = bytes as f64 / transfer.as_secs_f64();
+ let estimate = self.peers.entry(peer.to_owned()).or_insert(Estimate {
+ bytes_per_sec: sample,
+ samples: 0,
+ at: arrived,
+ });
+ if estimate.samples > 0 {
+ estimate.bytes_per_sec += WEIGHT * (sample - estimate.bytes_per_sec);
+ }
+ estimate.samples += 1;
+ estimate.at = arrived;
+ }
+
+ pub(crate) fn get(&self, peer: &str) -> Option<Estimate> {
+ self.peers.get(peer).copied()
+ }
+
+ pub(crate) fn retain(&mut self, mut keep: impl FnMut(&str) -> bool) {
+ self.peers.retain(|peer, _| keep(peer));
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn bytes(sent: u64, received: u64) -> SyncBytes {
+ SyncBytes { sent, received }
+ }
+
+ #[test]
+ fn small_exchanges_dont_count() {
+ let mut table = ThroughputTable::default();
+ let t0 = Instant::now();
+ table.exchange(
+ "a",
+ (t0, bytes(0, 0)),
+ (t0 + Duration::from_millis(200), bytes(500, 10_000)),
+ None,
+ );
+ assert!(table.get("a").is_none());
+ }
+
+ #[test]
+ fn round_trip_is_taken_off_but_at_most_half() {
+ let mut table = ThroughputTable::default();
+ let t0 = Instant::now();
+ // 1 MB in 2 s, of which 1 s round trip: 1 MB/s.
+ table.exchange(
+ "a",
+ (t0, bytes(100, 0)),
+ (t0 + Duration::from_secs(2), bytes(200, 999_900)),
+ Some(Duration::from_secs(1)),
+ );
+ let a = table.get("a").unwrap();
+ assert!((a.bytes_per_sec - 1_000_000.0).abs() < 1.0);
+
+ // A round trip longer than the exchange takes off only half.
+ table.exchange(
+ "b",
+ (t0, bytes(0, 0)),
+ (t0 + Duration::from_secs(2), bytes(0, 1_000_000)),
+ Some(Duration::from_secs(5)),
+ );
+ assert!((table.get("b").unwrap().bytes_per_sec - 1_000_000.0).abs() < 1.0);
+ }
+
+ #[test]
+ fn smoothed() {
+ let mut table = ThroughputTable::default();
+ let t0 = Instant::now();
+ for received in [500_000u64, 1_000_000] {
+ table.exchange(
+ "a",
+ (t0, bytes(0, 0)),
+ (t0 + Duration::from_secs(1), bytes(0, received)),
+ None,
+ );
+ }
+ // Samples of 500 kB/s, then 1 MB/s: 500 kB/s + (1 MB/s - 500 kB/s) / 4.
+ let a = table.get("a").unwrap();
+ assert_eq!(a.samples, 2);
+ assert!((a.bytes_per_sec - 625_000.0).abs() < 1.0);
+ }
+}
diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
new file mode 100644
index 0000000..57856b6
--- /dev/null
+++ b/src/swarm/topology.rs
@@ -0,0 +1,521 @@
+//! 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.
+
+use std::collections::{BTreeMap, BTreeSet};
+
+use sha2::{Digest, Sha256};
+use tokio::time::Instant;
+use tracing::info;
+
+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>,
+ neighbours: BTreeSet<String>,
+ /// Every reachable member is a neighbour.
+ full_mesh: bool,
+}
+
+/// 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()
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
+ .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()
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
+ .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()
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
+ {
+ 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 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) = {
+ let replica = state.doc.read().await;
+ let current = members::members(&replica.doc);
+ let stable: BTreeSet<String> = current
+ .iter()
+ .filter(|m| members::tier(&replica.doc, m) == members::Tier::Stable)
+ .cloned()
+ .collect();
+ (current, stable)
+ };
+ let k = state.neighbours_k;
+ let now = Instant::now();
+ let mut unreachable = BTreeSet::new();
+ for member in &current {
+ if let Some(peer_state) = state.get_peer_state(member).await {
+ if peer_state.lock().await.is_unreachable(&state.timing, now) {
+ unreachable.insert(member.clone());
+ }
+ }
+ }
+ let previous = {
+ let view = state
+ .neighbours
+ .read()
+ .unwrap_or_else(|poisoned| poisoned.into_inner());
+ if view.as_ref().is_some_and(|view| {
+ view.members == current
+ && view.k == k
+ && view.unreachable == unreachable
+ && view.stable == stable
+ }) {
+ 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 computed = neighbours(
+ &current,
+ state.local_id(),
+ k,
+ &|m| unreachable.contains(m),
+ &|m| stable.contains(m),
+ );
+ 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(),
+ 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()
+ .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(NeighbourView {
+ full_mesh: computed.len() >= reachable_others,
+ members: current,
+ k,
+ unreachable,
+ stable,
+ neighbours: computed,
+ });
+ Some(change)
+}
+
+const RING_LABEL: &[u8] = b"continuum/topology-ring/1";
+const PAIR_LABEL: &[u8] = b"continuum/topology-pair/1";
+
+/// 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 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. Costs about N² hashes (a few milliseconds for
+/// hundreds 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")
+ };
+
+ // How many members have preferred each stable member so far (`live` iterates in key
+ // order, the same on every node).
+ let budget = STABLE_FANIN_FACTOR * k;
+ let mut preferred_by: BTreeMap<&str, usize> = BTreeMap::new();
+
+ for member in &live {
+ let at = position[member];
+ let next = first_live(at, 1);
+ let prev = first_live(at, n - 1);
+ link(member, next);
+ link(member, prev);
+ let mut candidates: Vec<([u8; 32], &str)> = live
+ .iter()
+ .copied()
+ .filter(|other| other != member && *other != next && *other != prev)
+ .map(|other| (pair_score(member, other), other))
+ .collect();
+ candidates.sort_unstable();
+ // 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);
+ let preferred: Vec<&str> = candidates
+ .iter()
+ .map(|(_, other)| *other)
+ .filter(|other| stable(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;
+ }
+ let rest = candidates
+ .iter()
+ .map(|(_, other)| *other)
+ .filter(|other| !preferred.contains(other))
+ .take(k - 2 - preferred.len());
+ for partner in preferred.iter().copied().chain(rest) {
+ link(member, partner);
+ }
+ }
+ links
+}
+
+/// Symmetric: the same for (a, b) and (b, a).
+fn pair_score(a: &str, b: &str) -> [u8; 32] {
+ let (low, high) = if a <= b { (a, b) } else { (b, a) };
+ hash(PAIR_LABEL, &[low, high])
+}
+
+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()
+ }
+
+ #[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"
+ );
+ }
+
+ #[test]
+ fn hub_degree_is_bounded_however_few_stable_members() {
+ // 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.
+ for (n, stable_count) in [(250, 10), (1000, 10), (1000, 20)] {
+ 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"
+ );
+ }
+ }
+}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-24 20:44:38 +0000