diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/api/stats.rs | 4 | ||||
| -rw-r--r-- | src/api/stats/prometheus.rs | 6 | ||||
| -rw-r--r-- | src/simtest/deltas.rs | 98 | ||||
| -rw-r--r-- | src/simtest/mod.rs | 1 | ||||
| -rw-r--r-- | src/state/session.rs | 75 | ||||
| -rw-r--r-- | src/swarm/blob.rs | 200 | ||||
| -rw-r--r-- | src/swarm/stats.rs | 10 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 121 | ||||
| -rw-r--r-- | src/transport/mod.rs | 72 | ||||
| -rw-r--r-- | src/transport/wire.rs | 30 |
10 files changed, 587 insertions, 30 deletions
diff --git a/src/api/stats.rs b/src/api/stats.rs index a34d4ed..fabbffd 100644 --- a/src/api/stats.rs +++ b/src/api/stats.rs @@ -114,6 +114,9 @@ struct SyncStats { last_from_neighbour_secs: Option<u64>, /// Peers whose heads differ from ours while sync is idle (`peers.md`, healing). diverged_peers: usize, + /// The largest sync message received since start (its Automerge part). Larger + /// transfers come as deltas (blobs). + largest_received_bytes: u64, } /// Passing changes on (`swarm::relay`) and serving blobs, since start. @@ -256,6 +259,7 @@ async fn collect(state: &AppState) -> StatsResponse { sync: SyncStats { last_from_neighbour_secs: last_from_neighbour.map(|at| at.elapsed().as_secs()), diverged_peers, + largest_received_bytes: node.largest_sync_received, }, relay: RelayStats { announce_rounds: node.relay.announce_rounds, diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs index 30bb966..f9eb3bf 100644 --- a/src/api/stats/prometheus.rs +++ b/src/api/stats/prometheus.rs @@ -323,6 +323,12 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { ); } out.single( + "largest_sync_received_bytes", + "gauge", + "The largest sync message received since start; larger transfers come as deltas.", + stats.sync.largest_received_bytes as f64, + ); + out.single( "diverged_peers", "gauge", "Peers whose heads differ from ours while sync is idle.", diff --git a/src/simtest/deltas.rs b/src/simtest/deltas.rs new file mode 100644 index 0000000..ce74607 --- /dev/null +++ b/src/simtest/deltas.rs @@ -0,0 +1,98 @@ +//! Large transfers as deltas (`docs/peers.md`): a node joining a swarm with much +//! history, and one returning after much was written, get the bulk as a blob, and no +//! sync message they receive is larger than the threshold. +//! +//! The nodes use 1 KiB chunks, so the threshold is 16 KiB, and a few hundred KiB of +//! writes are well above it. + +use std::time::Duration; + +use serde_json::json; + +use super::{Sim, read, write}; +use crate::node::NodeConfig; +use crate::state::AppState; +use crate::swarm::bootstrap::Start; + +const SOON: Duration = Duration::from_secs(10); +/// Catching up a few hundred KiB in 1 KiB chunks. +const CATCH_UP: Duration = Duration::from_secs(60); +const CHUNK: u32 = 1024; +/// 16 chunks (`swarm::blob`). +const THRESHOLD: u64 = 16 * CHUNK as u64; + +fn small_chunks(config: &mut NodeConfig) { + config.blob_chunk_bytes = CHUNK; +} + +/// About 1 KB that doesn't compress (Automerge compresses large changes). +fn noise(seed: u64) -> String { + let mut rng = crate::testrand::Rng::new(seed); + (0..63) + .map(|_| format!("{:016x}", rng.next_u64())) + .collect() +} + +/// `count` writes of about 1 KB each, under `prefix`. +async fn write_much(node: &AppState, prefix: &str, count: usize) { + for i in 0..count { + write(node, &format!("{prefix}/{i}"), json!(noise(i as u64))).await; + } +} + +fn largest_sync_received(node: &AppState) -> u64 { + node.stats.snapshot().largest_sync_received +} + +fn blob_bytes_received(node: &AppState) -> u64 { + node.traffic.total().received.blob.bytes +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn joining_node_gets_the_history_as_a_delta() { + let sim = Sim::new(41); + let a = sim.start_with("a", &[], Start::Genesis, small_chunks); + sim.wait_ready(&[&a], SOON).await; + write_much(&a, "history", 200).await; + + let b = sim.start_with("b", &["a"], Start::Join, small_chunks); + sim.wait_ready(&[&b], CATCH_UP).await; + sim.wait_converged(&[&a, &b], CATCH_UP).await; + assert_eq!(read(&b, "history/199").await, Some(json!(noise(199)))); + + assert!( + blob_bytes_received(&b) > 100 * 1_000, + "{} blob bytes received", + blob_bytes_received(&b) + ); + assert!( + largest_sync_received(&b) <= THRESHOLD, + "a sync message of {} bytes", + largest_sync_received(&b) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn returning_node_gets_what_it_missed_as_a_delta() { + let sim = Sim::new(42); + let a = sim.start_with("a", &[], Start::Genesis, small_chunks); + let b = sim.start_with("b", &["a"], Start::Join, small_chunks); + let (a, b) = (&a, &b); + sim.wait_ready(&[a, b], SOON).await; + write(a, "early", json!(1)).await; + sim.wait_converged(&[a, b], SOON).await; + + sim.net.take_down("b"); + write_much(a, "while/away", 200).await; + let blobs_before = blob_bytes_received(b); + sim.net.bring_up("b"); + + sim.wait_converged(&[a, b], CATCH_UP).await; + assert_eq!(read(b, "while/away/199").await, Some(json!(noise(199)))); + assert!(blob_bytes_received(b) - blobs_before > 100 * 1_000); + assert!( + largest_sync_received(b) <= THRESHOLD, + "a sync message of {} bytes", + largest_sync_received(b) + ); +} diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs index f7af61b..5e9cc2f 100644 --- a/src/simtest/mod.rs +++ b/src/simtest/mod.rs @@ -9,6 +9,7 @@ //! `RUST_LOG=continuum=debug cargo test simtest -- --nocapture` shows the nodes' logs //! (all nodes interleaved; messages name peers by ID). +mod deltas; mod remakes; mod soak; mod timing; diff --git a/src/state/session.rs b/src/state/session.rs index 618d65b..6efe538 100644 --- a/src/state/session.rs +++ b/src/state/session.rs @@ -52,6 +52,11 @@ pub(crate) struct SyncSession { /// The wait after the next refusal. pub(crate) send_backoff: Backoff, + /// We offered the peer a delta instead of a message too large to send + /// (`withdraw_for_delta`): no more messages until it answers, or the answer is + /// overdue (then the offer goes out again). + pub(crate) delta_offered: bool, + /// The answer we expect to our last sync message(s), if any. pub(crate) reply: Option<SyncReplyWait>, /// How long to wait for it; doubles while answers stay out. @@ -80,6 +85,7 @@ impl SyncSession { sending: None, retry_at: None, send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX), + delta_offered: false, reply: None, reply_backoff, } @@ -134,9 +140,30 @@ impl SyncSession { doc: &mut AutoCommit, message: sync::Message, ) -> Result<(), AutomergeError> { + self.delta_offered = false; doc.sync().receive_sync_message(&mut self.state, message) } + /// The message just generated is too large to send in one piece: take back what it + /// carried, as if it had never been generated, and offer a delta instead. The + /// message being handed over becomes the offer (it carries nothing); the answer is + /// still awaited, and if it doesn't come, the offer is made again. Returns the + /// peer's heads as far as we know them: where the delta starts. + pub(crate) fn withdraw_for_delta(&mut self) -> Vec<automerge::ChangeHash> { + if let Some(sending) = &mut self.sending { + let sent = std::mem::take(&mut sending.sent); + for hash in &sent { + self.state.sent_hashes.remove(hash); + } + if let Some(wait) = &mut self.reply { + wait.sent.retain(|hash| !sent.contains(hash)); + } + } + self.state.in_flight = false; + self.delta_offered = true; + self.state.their_heads.clone().unwrap_or_default() + } + /// Whether `doc` holds everything the peer had when it last told us its heads: they /// are known, not empty (empty heads are "contained" in every document, so they /// prove nothing), and all in `doc`. @@ -282,6 +309,7 @@ impl SyncSession { self.state.sent_hashes.remove(hash); } self.state.in_flight = false; + self.delta_offered = false; self.reply_backoff.grow(); Some(wait.sent.len()) } @@ -323,4 +351,51 @@ mod tests { assert!(!new.send_succeeded(new.id())); assert!(new.sending.is_none()); } + + /// A message too large to send is taken back: the peer, after the offer and an + /// overdue answer, gets the same changes in the next message. + #[test] + fn withdrawing_for_a_delta_keeps_the_changes_for_later() { + use automerge::{ROOT, transaction::Transactable}; + + // 16 KiB of hex, about 8 KiB of information: Automerge compresses what it sends + // (a peer with nothing gets the document saved), so this stays above 4 KB. + let mut rng = crate::testrand::Rng::new(1); + let big: String = (0..1024) + .map(|_| format!("{:016x}", rng.next_u64())) + .collect(); + let mut ours = AutoCommit::new(); + ours.put(ROOT, "big", big).unwrap(); + ours.commit(); + let mut theirs = AutoCommit::new(); + let (mut a, mut b) = (session(), session()); + let now = Instant::now(); + let bytes = SyncBytes::default(); + + // a opens, b answers with its (empty) heads, then a sends the changes. + let opening = a.generate(&mut ours, now, bytes).unwrap(); + a.send_succeeded(a.id()); + b.receive(&mut theirs, opening).unwrap(); + let answer = b.generate(&mut theirs, now, bytes).unwrap(); + b.send_succeeded(b.id()); + a.receive(&mut ours, answer).unwrap(); + let carrying = a.generate(&mut ours, now, bytes).unwrap(); + let carried = carrying.clone().encode().len(); + assert!(carried > 4_000, "the changes took {carried} bytes"); + + let from = a.withdraw_for_delta(); + assert!(from.is_empty()); + assert!(a.delta_offered); + assert!(!a.send_succeeded(a.id())); + + // No answer to the offer: it lapses, and the changes go out again. + assert!( + a.take_overdue_reply(now + Duration::from_secs(3600)) + .is_some() + ); + assert!(!a.delta_offered); + let again = a.generate(&mut ours, now, bytes).unwrap(); + let carried_again = again.encode().len(); + assert!(carried_again > 4_000, "sent again in {carried_again} bytes"); + } } 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; diff --git a/src/transport/mod.rs b/src/transport/mod.rs index c109b0c..3bc88e4 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -147,6 +147,9 @@ pub(crate) enum BlobId { Change(#[serde(with = "crate::transport::wire::codec::change_hash")] ChangeHash), /// Bytes by their BLAKE3 hash: e.g. the content of a remake. Content(#[serde(with = "crate::transport::wire::codec::bytes32")] [u8; 32]), + /// The changes of a `DeltaSpec`, by its ID (`DeltaSpec::id`). The requests carry + /// the spec, so any node holding its `to` can serve it. + Delta(#[serde(with = "crate::transport::wire::codec::bytes32")] [u8; 32]), } impl std::fmt::Display for BlobId { @@ -157,6 +160,10 @@ impl std::fmt::Display for BlobId { write!(f, "content:")?; hash.iter().try_for_each(|byte| write!(f, "{byte:02x}")) } + BlobId::Delta(id) => { + write!(f, "delta:")?; + id.iter().try_for_each(|byte| write!(f, "{byte:02x}")) + } } } } @@ -176,8 +183,12 @@ pub(crate) struct BlobMessage { #[derive(Clone, Serialize, Deserialize)] pub(crate) enum BlobBody { - /// Asks for these chunks; for the manifest if empty. - Request { chunks: Vec<u32> }, + /// Asks for these chunks; for the manifest if empty. `delta`: the spec of a + /// `BlobId::Delta`, which the ID alone doesn't say. + Request { + chunks: Vec<u32>, + delta: Option<DeltaSpec>, + }, /// Size and per-chunk BLAKE3 hashes; the last chunk may be shorter. Manifest { size: u64, @@ -198,9 +209,11 @@ pub(crate) enum BlobBody { impl std::fmt::Debug for BlobBody { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - BlobBody::Request { chunks } => { - f.debug_struct("Request").field("chunks", chunks).finish() - } + BlobBody::Request { chunks, delta } => f + .debug_struct("Request") + .field("chunks", chunks) + .field("delta", &delta.is_some()) + .finish(), BlobBody::Manifest { size, chunk_size, @@ -325,14 +338,59 @@ pub(crate) struct SyncEnvelope { /// (`Replica::epoch_seq`). Messages may arrive late and out of order: the receiver /// ignores an epoch older than the newest it has seen in the same session. pub(crate) epoch_seq: u32, - #[serde(with = "crate::transport::wire::codec::sync_message")] - pub(crate) message: SyncPayload, + pub(crate) body: SyncBody, /// The sender was ready when it generated the message: its document is the swarm's, /// not one it is still filling (joining, rejoining). Only such a peer can tell a /// catching-up node that it has everything; an empty document says nothing else. pub(crate) ready: bool, } +/// What a sync envelope carries. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) enum SyncBody { + /// An Automerge sync message. + Message(#[serde(with = "crate::transport::wire::codec::sync_message")] SyncPayload), + /// Instead of a message too large to send in one piece: fetch these changes as a + /// blob (`BlobId::Delta`), then sync on. + Delta(DeltaSpec), +} + +/// The changes in the history of `to` but not of `from`: what a peer at `from` lacks of +/// a document at `to`. Every node holding `to` produces the same bytes for it +/// (`swarm::blob::delta_bytes`). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct DeltaSpec { + /// Sorted. + #[serde(with = "crate::transport::wire::codec::change_hashes")] + pub(crate) from: Vec<ChangeHash>, + /// Sorted. + #[serde(with = "crate::transport::wire::codec::change_hashes")] + pub(crate) to: Vec<ChangeHash>, +} + +impl DeltaSpec { + pub(crate) fn new(mut from: Vec<ChangeHash>, mut to: Vec<ChangeHash>) -> Self { + from.sort_unstable(); + from.dedup(); + to.sort_unstable(); + to.dedup(); + Self { from, to } + } + + /// Names the spec as a blob: SHA-256 over both lists. + pub(crate) fn id(&self) -> [u8; 32] { + let mut hasher = Sha256::new(); + hasher.update(b"continuum/delta/1"); + for heads in [&self.from, &self.to] { + hasher.update((heads.len() as u32).to_be_bytes()); + for head in heads { + hasher.update(head.0); + } + } + hasher.finalize().into() + } +} + /// An Automerge sync message in Automerge's own encoding. Decoded only where it is /// applied (`decode`), after the sender proved to be a member and off the consumer: a /// large message from anyone else costs nothing but its bytes. diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 67d2fb3..009380f 100644 --- a/src/transport/wire.rs +++ b/src/transport/wire.rs @@ -14,7 +14,7 @@ use super::{MessageKind, ParsedMessage}; /// The wire format's version, the first byte of every payload. Bump it with every change /// to a message type or a codec (`golden_bytes` fails then). -pub(crate) const VERSION: u8 = 2; +pub(crate) const VERSION: u8 = 3; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -233,6 +233,28 @@ pub(crate) mod codec { } } + /// Change hashes, each a 32-byte binary. + pub(crate) mod change_hashes { + use super::*; + use automerge::ChangeHash; + + pub(crate) fn serialize<S: Serializer>( + value: &[ChangeHash], + s: S, + ) -> Result<S::Ok, S::Error> { + s.collect_seq(value.iter().map(|hash| Bytes::new(&hash.0))) + } + + pub(crate) fn deserialize<'de, D: Deserializer<'de>>( + d: D, + ) -> Result<Vec<ChangeHash>, D::Error> { + Vec::<ByteBuf>::deserialize(d)? + .iter() + .map(|hash| Ok(ChangeHash(fixed(hash, "a change hash")?))) + .collect() + } + } + /// An Automerge sync message, in Automerge's own encoding (left encoded: /// `SyncPayload`). pub(crate) mod sync_message { @@ -416,7 +438,7 @@ mod tests { }); assert_eq!( hex(&encode(&hello).unwrap()), - format!("020097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) + format!("030097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) ); let reply = ParsedMessage::HelloReply(HelloReply { @@ -433,7 +455,7 @@ mod tests { assert_eq!( hex(&encode(&reply).unwrap()), format!( - "020199a5737761726d91a170{epoch_hex}01c400c2{}05{}", + "030199a5737761726d91a170{epoch_hex}01c400c2{}05{}", run("04"), run("06") ) @@ -448,7 +470,7 @@ mod tests { }); assert_eq!( hex(&encode(&ping).unwrap()), - format!("0203959290c400{epoch_hex}01c3c2") + format!("0303959290c400{epoch_hex}01c3c2") ); } } |