summaryrefslogtreecommitdiffstats
path: root/src/swarm
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm')
-rw-r--r--src/swarm/blob.rs508
-rw-r--r--src/swarm/members.rs109
-rw-r--r--src/swarm/mod.rs1
-rw-r--r--src/swarm/peer.rs11
-rw-r--r--src/swarm/relay.rs62
-rw-r--r--src/swarm/source.rs2
-rw-r--r--src/swarm/stats.rs56
-rw-r--r--src/swarm/sync.rs25
-rw-r--r--src/swarm/topology.rs50
9 files changed, 781 insertions, 43 deletions
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/members.rs b/src/swarm/members.rs
index e02dba4..221437e 100644
--- a/src/swarm/members.rs
+++ b/src/swarm/members.rs
@@ -9,14 +9,16 @@
//! 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, metrics and logs. Not unique and never an identity: the
-//! public key stays that.
+//! 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::{info, warn};
+use tracing::{debug, info, warn};
use crate::document::store::{self, StoreError};
use crate::state::{AppState, unix_millis};
@@ -107,26 +109,69 @@ fn set_alias(doc: &mut AutoCommit, pk: &str, alias: Option<&str>) -> Result<(),
}
}
-/// Registers this node if it is ready and its entry is missing, and keeps its alias
-/// as configured (set, changed or removed after a restart with another `--alias`).
+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) = {
+ 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))
+ (
+ 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": unix_millis() });
+ let mut entry = json!({ "since": now, "alive": now });
if let Some(alias) = &wanted {
entry["alias"] = json!(alias);
}
@@ -143,6 +188,18 @@ pub(crate) async fn ensure_registered(state: &AppState) {
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;
}
@@ -167,3 +224,39 @@ pub(crate) fn unregister(doc: &mut AutoCommit, pk: &str) -> Result<(), StoreErro
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 6f194dc..adb5d1c 100644
--- a/src/swarm/mod.rs
+++ b/src/swarm/mod.rs
@@ -2,6 +2,7 @@
//! 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;
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index 3b85539..3d7da32 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -108,12 +108,12 @@ pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> {
/// Our keepalive for `peer`.
pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive {
- // AutoCommit::get_heads takes &mut self (it closes the pending transaction).
- // Heads and epoch come from the same locked replica, so they always match.
+ // 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 mut replica = state.doc.write().await;
+ let replica = state.doc.read().await;
(
- transport::HeadsSummary::of(replica.doc.get_heads()),
+ transport::HeadsSummary::of(replica.heads().to_vec()),
replica.epoch,
)
};
@@ -824,6 +824,9 @@ 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() {
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 454a46a..703bc3f 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -9,8 +9,16 @@
//! 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;
@@ -26,18 +34,64 @@ 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.
+/// 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| peer.as_str() != from && topology::is_neighbour(state, peer))
+ .filter(|peer| {
+ skip.as_deref() != Some(peer.as_str()) && topology::is_neighbour(state, peer)
+ })
.cloned()
.collect();
for peer in neighbours {
@@ -52,6 +106,10 @@ pub(crate) async fn pull_if_behind(state: &AppState, peer: &str, remote: &transp
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.
diff --git a/src/swarm/source.rs b/src/swarm/source.rs
index 4025f0e..378c40a 100644
--- a/src/swarm/source.rs
+++ b/src/swarm/source.rs
@@ -199,7 +199,7 @@ pub(crate) fn finished(state: &AppState, from: &str) {
}
/// After giving up a source: let every peer be considered again.
-fn resync_everyone(state: &AppState) {
+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();
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index 786734a..eb34276 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -8,10 +8,11 @@ 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.
+/// 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;
@@ -29,6 +30,8 @@ struct Inner {
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,
}
@@ -157,10 +160,12 @@ impl NodeStats {
self.lock().saved_size
}
- /// Whether a measurement is due; if so, it counts as started.
- fn start_measuring(&self, now: Instant) -> bool {
+ /// 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)
});
@@ -170,20 +175,24 @@ impl NodeStats {
due
}
- /// The document was saved for another reason (an export): a measurement for free.
- pub(crate) fn saved_size_measured(&self, bytes: u64, took: Duration) {
- self.lock().saved_size = Some(SavedSize {
+ /// 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>) {
+ fn finish_measuring(&self, measured: Option<(SavedSize, (u64, Epoch))>) {
let mut inner = self.lock();
inner.measuring = false;
- if let Some(measured) = measured {
+ if let Some((measured, version)) = measured {
inner.saved_size = Some(measured);
+ inner.measured_version = Some(version);
}
}
@@ -200,14 +209,24 @@ impl NodeStats {
}
}
+/// 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
-/// 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 every 200 s). Called by the
-/// peer worker every few seconds.
+/// 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) {
- if !state.stats.start_measuring(Instant::now()) {
+ // 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();
@@ -228,11 +247,14 @@ pub(crate) fn measure_saved_size(state: &AppState) {
took_ms = took.as_millis() as u64,
"Measured the saved document size"
);
- Some(SavedSize {
- bytes,
- at: Instant::now(),
- took,
- })
+ Some((
+ SavedSize {
+ bytes,
+ at: Instant::now(),
+ took,
+ },
+ version,
+ ))
}
Err(error) => {
warn!(error = %error, "Measuring the saved document size failed");
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 3284015..80d6d56 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -32,6 +32,12 @@ pub(crate) struct ChangeSet {
}
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(());
@@ -337,7 +343,7 @@ pub(crate) async fn handle_sync(
let mut peer_state = peer_state_mutex.blocking_lock();
peer_state.reports_ready = Some(from_ready);
- let heads_before = replica.doc.get_heads();
+ let heads_before = replica.heads().to_vec();
replica
.doc
@@ -354,7 +360,8 @@ pub(crate) async fn handle_sync(
peer_state.sync.reply_settled();
resend_if_needed(&peer, &mut peer_state.sync.state);
- let heads_after = replica.doc.get_heads();
+ 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 {
@@ -372,8 +379,11 @@ pub(crate) async fn handle_sync(
}
Ok(Some((
- // Only from a ready peer: a peer still catching up has empty heads, which
- // every document "contains".
+ // 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
@@ -382,9 +392,10 @@ pub(crate) async fn handle_sync(
.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,
)))
diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
index c5a02f1..57856b6 100644
--- a/src/swarm/topology.rs
+++ b/src/swarm/topology.rs
@@ -204,6 +204,13 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
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(
@@ -222,7 +229,7 @@ pub(crate) fn neighbours(
/// 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 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`
@@ -231,6 +238,8 @@ pub(crate) fn neighbours(
///
/// 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,
@@ -275,6 +284,11 @@ pub(crate) fn graph(
.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);
@@ -289,15 +303,18 @@ pub(crate) fn graph(
.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 and the load
- // on stable members bounded.
+ // 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))
+ .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)
@@ -476,4 +493,29 @@ mod tests {
"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:59:09 +0000