summaryrefslogtreecommitdiffstats
path: root/src/swarm
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm')
-rw-r--r--src/swarm/blob.rs200
-rw-r--r--src/swarm/stats.rs10
-rw-r--r--src/swarm/sync.rs121
3 files changed, 312 insertions, 19 deletions
diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index 372d964..679e1c6 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -4,7 +4,9 @@
//! 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).
+//! (e.g. the new document of a `replace` or `import` remake), or the changes between two
+//! sets of heads (a delta: what a peer lacks, when one sync message would be too large;
+//! `swarm::sync`).
//!
//! * 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
@@ -26,14 +28,14 @@ use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Duration;
use anyhow::{Result, anyhow, bail};
-use automerge::ChangeHash;
+use automerge::{AutoCommit, ChangeHash};
use tokio::sync::{Semaphore, mpsc};
use tokio::time::Instant;
use tracing::{debug, info, warn};
use crate::locks::MutexExt;
use crate::state::AppState;
-use crate::transport::{BlobBody, BlobId, BlobMessage, ParsedMessage};
+use crate::transport::{BlobBody, BlobId, BlobMessage, DeltaSpec, ParsedMessage};
/// Chunk sends in progress at once, over all downloads served.
const MAX_SERVING: usize = 8;
@@ -76,8 +78,16 @@ pub(crate) struct Blobs {
cache: Mutex<Option<(BlobId, Arc<Prepared>)>>,
/// Answers for downloads in progress, by blob.
fetches: Mutex<HashMap<BlobId, mpsc::UnboundedSender<(String, BlobBody)>>>,
+ /// Peers whose offered delta is being downloaded (`swarm::sync`).
+ deltas: Mutex<HashSet<String>>,
}
+/// A sync message larger than this many chunks is sent as a delta blob instead
+/// (`swarm::sync`): below it, a blob costs more round trips than it saves; above it, a
+/// lost message costs too much to send again whole, and a slow link may not deliver it
+/// within Mycelium's deadline.
+const DELTA_THRESHOLD_CHUNKS: usize = 16;
+
struct Prepared {
bytes: Arc<Vec<u8>>,
hashes: Vec<[u8; 32]>,
@@ -92,9 +102,24 @@ impl Blobs {
dropped: Mutex::new((0, None)),
cache: Mutex::new(None),
fetches: Mutex::new(HashMap::new()),
+ deltas: Mutex::new(HashSet::new()),
}
}
+ /// The largest sync message sent as it is (`DELTA_THRESHOLD_CHUNKS`).
+ pub(crate) fn delta_threshold(&self) -> usize {
+ DELTA_THRESHOLD_CHUNKS * self.chunk_size as usize
+ }
+
+ /// Starts downloading a delta offered by `peer`; false if one from it is running.
+ pub(crate) fn start_delta(&self, peer: &str) -> bool {
+ self.deltas.lock_or_recover().insert(peer.to_owned())
+ }
+
+ pub(crate) fn end_delta(&self, peer: &str) {
+ self.deltas.lock_or_recover().remove(peer);
+ }
+
fn fetches(
&self,
) -> MutexGuard<'_, HashMap<BlobId, mpsc::UnboundedSender<(String, BlobBody)>>> {
@@ -143,7 +168,7 @@ fn chunk_count(size: u64, chunk_size: u32) -> u64 {
/// A blob message from `peer`.
pub(crate) async fn handle(state: &AppState, peer: &str, message: BlobMessage) {
match message.body {
- BlobBody::Request { chunks } => {
+ BlobBody::Request { chunks, delta } => {
let Ok(permit) = Arc::clone(&state.blobs.answering).try_acquire_owned() else {
state.stats.blob_request_dropped();
debug!(peer = %peer, blob = %message.id, "Answering too many blob requests; dropping one");
@@ -160,7 +185,7 @@ pub(crate) async fn handle(state: &AppState, peer: &str, message: BlobMessage) {
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;
+ serve(&state, &peer, message.id, chunks, delta).await;
drop(permit);
});
}
@@ -178,11 +203,21 @@ pub(crate) async fn handle(state: &AppState, peer: &str, message: BlobMessage) {
}
}
-async fn prepare(state: &AppState, id: BlobId) -> Option<Arc<Prepared>> {
+async fn prepare(state: &AppState, id: BlobId, delta: Option<DeltaSpec>) -> Option<Arc<Prepared>> {
if let Some(prepared) = state.blobs.cached(id) {
return Some(prepared);
}
let bytes = match id {
+ BlobId::Delta(delta_id) => {
+ // The spec must be the one the ID names.
+ let spec = delta.filter(|spec| spec.id() == delta_id)?;
+ let doc = Arc::clone(&state.doc);
+ tokio::task::spawn_blocking(move || delta_bytes(&mut doc.blocking_write().doc, &spec))
+ .await
+ .ok()
+ .flatten()
+ .map(Arc::new)?
+ }
BlobId::Change(hash) => {
let doc = Arc::clone(&state.doc);
tokio::task::spawn_blocking(move || {
@@ -225,8 +260,14 @@ async fn reply(state: &AppState, peer: &str, id: BlobId, body: BlobBody) {
}
}
-async fn serve(state: &AppState, peer: &str, id: BlobId, chunks: Vec<u32>) {
- let Some(prepared) = prepare(state, id).await else {
+async fn serve(
+ state: &AppState,
+ peer: &str,
+ id: BlobId,
+ chunks: Vec<u32>,
+ delta: Option<DeltaSpec>,
+) {
+ let Some(prepared) = prepare(state, id, delta).await else {
return reply(state, peer, id, BlobBody::Missing).await;
};
let chunk_size = state.blobs.chunk_size;
@@ -266,13 +307,22 @@ impl Drop for Registration<'_> {
}
}
-async fn request(state: &AppState, peer: &str, id: BlobId, chunks: Vec<u32>) {
+async fn request(
+ state: &AppState,
+ peer: &str,
+ id: BlobId,
+ chunks: Vec<u32>,
+ delta: Option<&DeltaSpec>,
+) {
let (epoch, epoch_seq) = state.epoch_stamp();
let message = ParsedMessage::Blob(BlobMessage {
epoch,
epoch_seq,
id,
- body: BlobBody::Request { chunks },
+ body: BlobBody::Request {
+ chunks,
+ delta: delta.cloned(),
+ },
});
if let Err(error) = state.transport.send_message(peer, &message).await {
debug!(peer = %peer, blob = %id, error = %format!("{error:#}"), "Blob request failed");
@@ -301,6 +351,7 @@ pub(crate) async fn fetch(
id: BlobId,
peers: Vec<String>,
expected_size: Option<u64>,
+ delta: Option<&DeltaSpec>,
) -> Result<Vec<u8>> {
let (sender, mut answers) = mpsc::unbounded_channel();
if state.blobs.fetches().insert(id, sender).is_some() {
@@ -331,7 +382,7 @@ pub(crate) async fn fetch(
let mut manifest = None;
let mut missing: HashSet<String> = HashSet::new();
for peer in &candidates {
- request(state, peer, id, Vec::new()).await;
+ request(state, peer, id, Vec::new(), delta).await;
let deadline = Instant::now() + MANIFEST_TIMEOUT;
loop {
let answer = tokio::time::timeout_at(deadline, answers.recv()).await;
@@ -425,7 +476,7 @@ pub(crate) async fn fetch(
for index in &batch {
source.in_flight.insert(*index, now);
}
- request(state, &source.peer, id, batch).await;
+ request(state, &source.peer, id, batch, delta).await;
}
match tokio::time::timeout(Duration::from_millis(500), answers.recv()).await {
@@ -520,7 +571,7 @@ pub(crate) async fn fetch_content(
peers: Vec<String>,
) -> Result<Vec<u8>> {
let id = BlobId::Content(hash);
- let bytes = fetch(state, id, peers, Some(size)).await?;
+ let bytes = fetch(state, id, peers, Some(size), None).await?;
if *blake3::hash(&bytes).as_bytes() != hash {
bail!("downloaded {id} doesn't match its hash");
}
@@ -532,7 +583,7 @@ pub(crate) async fn fetch_change(
hash: ChangeHash,
peers: Vec<String>,
) -> Result<automerge::Change> {
- let bytes = fetch(state, BlobId::Change(hash), peers, None).await?;
+ let bytes = fetch(state, BlobId::Change(hash), peers, None, None).await?;
let change = automerge::Change::from_bytes(bytes)
.map_err(|error| anyhow!("downloaded change {hash} doesn't load: {error}"))?;
if change.hash() != hash {
@@ -541,9 +592,130 @@ pub(crate) async fn fetch_change(
Ok(change)
}
+/// Downloads the changes of `spec` from `peers` (any that hold its `to`). The caller
+/// loads them and checks that it then has `to`.
+pub(crate) async fn fetch_delta(
+ state: &AppState,
+ spec: &DeltaSpec,
+ peers: Vec<String>,
+) -> Result<Vec<u8>> {
+ fetch(state, BlobId::Delta(spec.id()), peers, None, Some(spec)).await
+}
+
+/// The bytes of delta `spec` from `doc`: the changes in the history of `spec.to` but
+/// not of `spec.from`, each as Automerge encodes it, sorted by hash and concatenated
+/// (`AutoCommit::load_incremental` reads that). The same on every node holding `to`,
+/// whatever else it has. None if `doc` lacks part of `to`.
+pub(crate) fn delta_bytes(doc: &mut AutoCommit, spec: &DeltaSpec) -> Option<Vec<u8>> {
+ if spec
+ .to
+ .iter()
+ .any(|head| doc.get_change_by_hash(head).is_none())
+ {
+ return None;
+ }
+ // Heads of `from` we don't have can't hide anything we have.
+ let from: Vec<ChangeHash> = spec
+ .from
+ .iter()
+ .copied()
+ .filter(|head| doc.get_change_by_hash(head).is_some())
+ .collect();
+ let since: HashMap<ChangeHash, (Vec<ChangeHash>, Vec<u8>)> = doc
+ .get_changes(&from)
+ .iter()
+ .map(|change| {
+ (
+ change.hash(),
+ (change.deps().to_vec(), change.raw_bytes().to_vec()),
+ )
+ })
+ .collect();
+ // Of those, the ones `to` depends on: what we have beyond `to` stays out.
+ let mut included: HashSet<ChangeHash> = HashSet::new();
+ let mut pending: Vec<ChangeHash> = spec.to.clone();
+ while let Some(hash) = pending.pop() {
+ if let Some((deps, _)) = since.get(&hash)
+ && included.insert(hash)
+ {
+ pending.extend(deps.iter().copied());
+ }
+ }
+ let mut hashes: Vec<ChangeHash> = included.into_iter().collect();
+ hashes.sort_unstable();
+ let mut bytes = Vec::new();
+ for hash in hashes {
+ bytes.extend_from_slice(&since[&hash].1);
+ }
+ Some(bytes)
+}
+
#[cfg(test)]
mod tests {
use super::*;
+ use automerge::{ActorId, ROOT, transaction::Transactable};
+
+ fn random_write(rng: &mut crate::testrand::Rng, doc: &mut AutoCommit) {
+ let key = format!("k{}", rng.below(6));
+ doc.put(ROOT, key, rng.below(100) as i64).unwrap();
+ doc.commit();
+ }
+
+ fn sorted(mut heads: Vec<ChangeHash>) -> Vec<ChangeHash> {
+ heads.sort_unstable();
+ heads
+ }
+
+ /// Over random histories: two nodes that reached the same heads through different
+ /// merge orders, one with more on top, produce the same bytes for a delta to those
+ /// heads; a document at `from` that loads them is at `to`.
+ #[test]
+ fn delta_bytes_are_the_same_everywhere_and_complete() {
+ crate::testrand::for_seeds(200, |rng| {
+ let actor = |i: u8| ActorId::from(&[i; 16][..]);
+ let mut base = AutoCommit::new().with_actor(actor(0));
+ for _ in 0..rng.between(0, 5) {
+ random_write(rng, &mut base);
+ }
+ let from = base.get_heads();
+ let mut writers: Vec<AutoCommit> = (1..=rng.between(1, 3))
+ .map(|i| base.fork().with_actor(actor(i as u8)))
+ .collect();
+ for _ in 0..rng.between(1, 15) {
+ let writer = rng.below(writers.len());
+ random_write(rng, &mut writers[writer]);
+ }
+ let mut one = base.fork().with_actor(actor(10));
+ let mut two = base.fork().with_actor(actor(11));
+ for writer in writers.iter_mut() {
+ one.merge(writer).unwrap();
+ }
+ for writer in writers.iter_mut().rev() {
+ two.merge(writer).unwrap();
+ }
+ let to = one.get_heads();
+ assert_eq!(sorted(to.clone()), sorted(two.get_heads()));
+ // Beyond `to`: must stay out of the delta.
+ random_write(rng, &mut two);
+
+ let spec = DeltaSpec::new(from.clone(), to.clone());
+ let bytes = delta_bytes(&mut one, &spec).unwrap();
+ assert_eq!(delta_bytes(&mut two, &spec).unwrap(), bytes);
+
+ let mut behind = base.fork();
+ behind.load_incremental(&bytes).unwrap();
+ assert_eq!(sorted(behind.get_heads()), sorted(to));
+ });
+ }
+
+ #[test]
+ fn no_delta_without_its_heads() {
+ let mut doc = AutoCommit::new();
+ doc.put(ROOT, "k", 1_i64).unwrap();
+ doc.commit();
+ let spec = DeltaSpec::new(Vec::new(), vec![ChangeHash([7; 32])]);
+ assert!(delta_bytes(&mut doc, &spec).is_none());
+ }
#[test]
fn chunking() {
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index 8172233..308c2a5 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -31,6 +31,7 @@ struct Inner {
stale_hello_replies: u64,
relay: Relay,
blob_requests_dropped: u64,
+ largest_sync_received: u64,
saved_size: Option<SavedSize>,
measuring: bool,
/// The document version (`AppState::change_tx`) and epoch the saved size is of.
@@ -122,6 +123,8 @@ pub(crate) struct Snapshot {
pub(crate) stale_hello_replies: u64,
pub(crate) relay: Relay,
pub(crate) blob_requests_dropped: u64,
+ /// The largest sync message received (its Automerge part, in bytes).
+ pub(crate) largest_sync_received: u64,
pub(crate) propagation: Propagation,
}
@@ -181,6 +184,12 @@ impl NodeStats {
self.lock().relay.pulls_queued += 1;
}
+ /// A sync message of `bytes` arrived.
+ pub(crate) fn sync_received(&self, bytes: usize) {
+ let mut inner = self.lock();
+ inner.largest_sync_received = inner.largest_sync_received.max(bytes as u64);
+ }
+
/// A blob request went unanswered: too many were being answered.
pub(crate) fn blob_request_dropped(&self) {
self.lock().blob_requests_dropped += 1;
@@ -237,6 +246,7 @@ impl NodeStats {
stale_hello_replies: inner.stale_hello_replies,
relay: inner.relay,
blob_requests_dropped: inner.blob_requests_dropped,
+ largest_sync_received: inner.largest_sync_received,
propagation: inner.propagation,
}
}
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index ca54fae..70ce28e 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -124,6 +124,11 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
// The retry timer enqueues the peer once it's due.
return Ok(());
}
+ if peer_state.sync.delta_offered {
+ // Its answer, or the reply timeout, brings us back here.
+ trace!("A delta is offered; waiting for the peer");
+ return Ok(());
+ }
}
let (session, encoded) = {
@@ -134,13 +139,14 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
let ready = state.is_ready().await;
// Before this message is counted.
let bytes_before = SyncBytes::of(&state.traffic, peer);
+ let delta_threshold = state.blobs.delta_threshold();
// The session the message comes from, for its send's outcome.
let (session, generated) = tokio::task::spawn_blocking(
move || -> (u64, Result<Option<transport::EncodedMessage>>) {
// Label the message with the epoch of the document it was generated
// from, read under the same lock.
- let (session, epoch, epoch_seq, message) = {
+ let (session, epoch, epoch_seq, body) = {
let mut replica = doc.blocking_write();
let mut peer_state = psm.blocking_lock();
let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
@@ -150,16 +156,33 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
peer_state
.sync
.generate(&mut replica.doc, Instant::now(), bytes_before);
- (peer_state.sync.id(), epoch, epoch_seq, message)
+ let body = message.map(|message| {
+ let payload = transport::SyncPayload::encode(message);
+ if payload.0.len() <= delta_threshold {
+ return transport::SyncBody::Message(payload);
+ }
+ // Too large for one message: all or nothing, and on a slow link
+ // within Mycelium's deadline. The peer fetches it in chunks.
+ let from = peer_state.sync.withdraw_for_delta();
+ let spec = transport::DeltaSpec::new(from, replica.heads().to_vec());
+ info!(
+ message_bytes = payload.0.len(),
+ from = spec.from.len(),
+ to = spec.to.len(),
+ "Sync message too large; offering the changes as a blob"
+ );
+ transport::SyncBody::Delta(spec)
+ });
+ (peer_state.sync.id(), epoch, epoch_seq, body)
};
- let Some(message) = message else {
+ let Some(body) = body else {
return (session, Ok(None));
};
let envelope = transport::SyncEnvelope {
swarm_id,
epoch,
epoch_seq,
- message: transport::SyncPayload::encode(message),
+ body,
ready,
};
let encoded = transport
@@ -316,6 +339,14 @@ pub(crate) async fn handle_sync(
let epoch_label = envelope.epoch;
let from_ready = envelope.ready;
+ let payload = match envelope.body {
+ transport::SyncBody::Message(payload) => payload,
+ transport::SyncBody::Delta(spec) => {
+ fetch_delta(state, peer, envelope.epoch, spec);
+ return Ok(());
+ }
+ };
+ state.stats.sync_received(payload.0.len());
let applied = {
let doc = Arc::clone(&state.doc);
@@ -329,7 +360,7 @@ pub(crate) async fn handle_sync(
// 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 message = payload;
let peer = peer.to_owned();
let clock = Arc::clone(&state.clock);
@@ -427,6 +458,86 @@ pub(crate) async fn handle_sync(
Ok(())
}
+/// The peer offered the changes we lack as a blob (`SyncBody::Delta`): fetch them, from
+/// it and any other peer that has them, apply them, and sync on. One download per peer
+/// at a time: the offer is repeated while it runs.
+fn fetch_delta(state: &AppState, peer: &str, epoch: transport::Epoch, spec: transport::DeltaSpec) {
+ if !state.blobs.start_delta(peer) {
+ trace!(peer = %peer, "Delta offered again while it downloads");
+ return;
+ }
+ let state = state.clone();
+ let peer = peer.to_owned();
+ tokio::spawn(async move {
+ let applied = download_and_apply_delta(&state, &peer, epoch, spec).await;
+ state.blobs.end_delta(&peer);
+ match applied {
+ Ok(changed) => {
+ state.notify_store_changed();
+ if changed {
+ crate::swarm::relay::announce(&state, &peer).await;
+ }
+ }
+ Err(error) => {
+ warn!(peer = %peer, error = %format!("{error:#}"), "Delta download failed; syncing on");
+ }
+ }
+ // Our next message says what we have now; the peer answers with the rest.
+ let _ = state.sync_events.send(SyncEvent::PeerNeedsSync(peer));
+ });
+}
+
+/// Returns whether our heads changed.
+async fn download_and_apply_delta(
+ state: &AppState,
+ peer: &str,
+ epoch: transport::Epoch,
+ spec: transport::DeltaSpec,
+) -> Result<bool> {
+ let mut peers = vec![peer.to_owned()];
+ peers.extend(
+ state
+ .active_peers_on_local_epoch()
+ .await
+ .into_iter()
+ .filter(|other| other != peer),
+ );
+ info!(peer = %peer, to = spec.to.len(), "Fetching changes offered as a blob");
+ let bytes = crate::swarm::blob::fetch_delta(state, &spec, peers).await?;
+
+ let doc = Arc::clone(&state.doc);
+ let peer_state = state.get_peer_state(peer).await;
+ let stats = Arc::clone(&state.stats);
+ tokio::task::spawn_blocking(move || -> Result<bool> {
+ let mut replica = doc.blocking_write();
+ if replica.epoch != epoch {
+ bail!("the epoch changed during the download");
+ }
+ let heads_before = replica.heads().to_vec();
+ replica
+ .doc
+ .load_incremental(&bytes)
+ .context("Loading the downloaded changes")?;
+ replica.refresh_heads();
+ let missing = spec
+ .to
+ .iter()
+ .filter(|head| replica.doc.get_change_by_hash(head).is_none())
+ .count();
+ if missing > 0 {
+ bail!("{missing} of the offered heads are still missing after loading");
+ }
+ stats.heads_seen(replica.heads().len());
+ if let Some(peer_state) = peer_state {
+ // Sync from it made progress: no heal for a while (`keepalive`).
+ peer_state.blocking_lock().last_sync_rx = Some(Instant::now());
+ }
+ Ok(replica.heads() != heads_before.as_slice())
+ })
+ .await
+ .context("Delta apply task panicked")?
+}
+
pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedReceiver<SyncEvent>) {
let mut dirty_peers = HashSet::<String>::new();
let mut document_dirty = false;
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 06:44:13 +0000