//! One Automerge sync with a peer, from a reset to the next. //! //! Automerge's sync state (`automerge::sync::State`) is private to this module: the //! rest of the node generates, receives and inspects through `SyncSession`'s methods. //! The workarounds for its assumption of reliable delivery (resending after a timeout, //! forgetting what was sent when the peer still needs it) live here with it, so an //! Automerge upgrade that changes the state touches this file only. use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; use automerge::AutomergeError; use automerge::sync::{self, SyncDoc}; use tokio::time::Instant; use tracing::debug; use super::{Document, ReadDoc}; use crate::backoff::Backoff; use crate::swarm::throughput::SyncBytes; use crate::transport::SyncPayload; /// A sync message from a peer, decoded (`decode`), to apply (`SyncSession::receive`). pub(crate) struct Received(sync::Message); impl Received { /// Whether it carries changes (it may carry only heads and what the peer needs). pub(crate) fn carries_changes(&self) -> bool { !self.0.changes.is_empty() } /// The bytes of the changes it carries, and of its bloom filters (`have`). pub(crate) fn sizes(&self) -> (usize, usize) { let changes = self.0.changes.iter().map(<[u8]>::len).sum(); let blooms = self .0 .have .iter() .map(|have| have.bloom.to_bytes().len()) .sum(); (changes, blooms) } } /// `payload` decoded. Before taking the document's lock: decoding a large message takes /// a while. pub(crate) fn decode(payload: &SyncPayload) -> anyhow::Result { sync::Message::decode(&payload.0) .map(Received) .map_err(|error| anyhow::anyhow!("Automerge decode failed: {error:?}")) } /// A sync message being handed to our transport (for Mycelium: the HTTP request is in /// progress). #[derive(Clone)] pub(crate) struct Sending { /// Changes it carries for the first time; taken back if the send fails. pub(crate) sent: Vec, /// A sync pass was requested meanwhile. Once this one is handed over, the peer must /// be enqueued again so the deferred changes go out. pub(crate) regenerate: bool, /// Taken as delivered (`assume_delivered`): what we knew of the peer before, put /// back if the send fails. pub(crate) assumed: Option, } /// What the sync state knows of the peer: its heads, the heads we share, what it has /// and needs. Saved to be put back: by `assume_delivered` if the send fails, and before /// receiving, for a message older than the last (`answered_as_delivered`). #[derive(Clone)] pub(crate) struct PeerView { their_heads: Option>, shared_heads: Vec, their_have: Option>, their_need: Option>, } /// Session IDs, unique within the process. static NEXT_SESSION: AtomicU64 = AtomicU64::new(1); /// First wait before generating again after our transport refused a message; doubles /// while it keeps refusing. const SEND_RETRY: Duration = Duration::from_millis(500); const SEND_RETRY_MAX: Duration = Duration::from_secs(30); /// One Automerge sync with a peer, from a reset to the next. Everything a reset must /// forget lives here, so a reset replaces it whole and can't miss a field. #[derive(Clone)] pub(crate) struct SyncSession { /// Tells this session from the one a reset puts in its place (`NEXT_SESSION`). id: u64, state: sync::State, /// A message being handed to our transport. pub(crate) sending: Option, /// After our transport refused a message: no new one before this. The refused one is /// not kept; the next one is generated afresh and carries everything pending. pub(crate) retry_at: Option, /// 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 delta last offered: its heads (sorted, as `DeltaSpec`), and the size of the /// message it replaced (`offer_again`). last_offer: Option<( Vec, Vec, usize, )>, /// What the sync state knew of the peer before the last message from it /// (`answered_as_delivered`). before_receive: Option, /// What the last `generate` changed in the sync state besides the hashes sent: the /// heads it last sent, and whether it had answered yet. Restored when the message is /// withdrawn for a delta: otherwise Automerge believes the peer heard our heads, and /// once the peer has the delta and says its heads are ours, never answers it, so the /// peer never learns our heads (a joining node then never finishes its initial /// sync). before_generate: (Vec, bool), /// Changes counted as sent because another parent brings them (`known_to_peer`), /// and when: still in flight perhaps, so they go along with a change of ours that /// builds on them (`with_predecessors`). known_elsewhere: std::collections::HashMap, /// The answer we expect to our last sync message(s), if any. pub(crate) reply: Option, /// How long to wait for it; doubles while answers stay out. pub(crate) reply_backoff: Backoff, } /// How long a change counted as sent because another parent brings it may still be in /// flight to the peer (`SyncSession::with_predecessors`): a slow round trip, and more. const KNOWN_ELSEWHERE_FOR: Duration = Duration::from_secs(2); /// How long such a change is remembered as counted so: to tell, should the peer ask for /// it, how long after (`SyncSession::resend_if_needed`). const KNOWN_ELSEWHERE_KEPT: Duration = Duration::from_secs(30); /// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery: /// a lost message leaves both sides waiting (`in_flight`), and changes it carried are /// never sent again (`sent_hashes`). We resend after a timeout instead. #[derive(Clone, Debug)] pub(crate) struct SyncReplyWait { pub(crate) deadline: Instant, /// When the first of the unanswered messages was generated, and the sync bytes /// exchanged with the peer by then (`swarm::throughput`). pub(crate) started: (Instant, SyncBytes), /// Changes first sent in the unanswered messages. pub(crate) sent: Vec, /// The largest of them, in bytes: if they go unanswered, messages that large may /// not make it to the peer whole (`PeerState::whole_failed`). pub(crate) largest: usize, } impl SyncSession { /// `reply_backoff`: waits for sync answers (`Timing::sync_reply_backoff`). pub(crate) fn new(reply_backoff: Backoff) -> Self { Self { id: NEXT_SESSION.fetch_add(1, Ordering::Relaxed), state: sync::State::new(), sending: None, retry_at: None, send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX), delta_offered: false, last_offer: None, before_receive: None, before_generate: (Vec::new(), false), known_elsewhere: std::collections::HashMap::new(), reply: None, reply_backoff, } } /// Whether the peer said anything in this session yet (its heads are known). pub(crate) fn has_opened(&self) -> bool { self.state.their_heads.is_some() } /// The next sync message for the peer from `doc`, encoded, if there is anything to say. A /// message is marked as being handed over (`sending`), and an answer to it is /// expected, unless the peer's heads are ours already: it has everything, and /// Automerge answers only when it has something to say, so waiting for an answer /// would only end in a pointless resend. `bytes`: the sync bytes exchanged with the /// peer so far (`swarm::throughput`). pub(crate) fn generate( &mut self, doc: &Document, now: Instant, bytes: SyncBytes, ) -> Option { let sent_before = self.state.sent_hashes.clone(); self.before_generate = ( self.state.last_sent_heads.clone(), self.state.have_responded, ); let message = doc.generate_sync_message(&mut self.state); let in_step = self.state.their_heads.as_ref() == Some(&doc.get_heads()); let sent: Vec<_> = self .state .sent_hashes .difference(&sent_before) .copied() .collect(); let payload = message.map(|message| SyncPayload(message.encode().into())); if in_step { self.reply_settled(); } else if let Some(payload) = &payload { self.expect_reply(sent.clone(), now, bytes, payload.0.len()); } if payload.is_some() { // Marked together with the generation, so no second message is generated // while this one is handed over. self.sending = Some(Sending { sent, regenerate: false, assumed: None, }); } payload } /// Applies a sync message from the peer to `doc`. Doesn't settle the awaited /// answer: the caller measures the exchange first (`reply`). #[cfg(test)] pub(crate) fn receive( &mut self, doc: &mut Document, message: Received, ) -> Result<(), AutomergeError> { self.receive_logged(doc, message, &mut automerge::PatchLog::inactive()) } /// Applies a sync message from the peer to `doc`, what it changes recorded in `log` /// (`Replica::logged`). Doesn't settle the awaited answer: the caller measures the /// exchange first (`reply`). pub(crate) fn receive_logged( &mut self, doc: &mut Document, message: Received, log: &mut automerge::PatchLog, ) -> Result<(), AutomergeError> { self.delta_offered = false; self.before_receive = Some(self.peer_view()); doc.receive_sync_message_log_patches(&mut self.state, message.0, log)?; self.known_last_sync(doc); Ok(()) } /// The peer's `have` names what it last shared with us by heads (`last_sync`), and /// one that took a message to us as delivered names the heads that message brought. /// Where the next message overtook it, we don't have those heads yet: Automerge /// would take that for a document that lost changes and answer with a reset, and /// the peer would forget all it ever sent us and send it again, what is still on its /// way included (on slow links, every change several times). So heads in it that we /// lack give way to the heads we share with the peer (`shared_heads`). The bloom /// filter stays: it covers the changes since its own heads, which leaves out at /// worst a few that go again. fn known_last_sync(&mut self, doc: &Document) { let Some(haves) = &mut self.state.their_have else { return; }; for have in haves.iter_mut() { if have .last_sync .iter() .all(|hash| doc.get_change_by_hash(hash).is_some()) { continue; } let mut known: Vec = have .last_sync .iter() .filter(|hash| doc.get_change_by_hash(hash).is_some()) .chain(self.state.shared_heads.iter()) .copied() .collect(); known.sort_unstable(); known.dedup(); have.last_sync = known; } } /// 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 where /// the delta starts (`delta_from`). /// /// An answer is awaited even where none was (the peer's heads were ours: `generate`), /// so the offer lapses at the reply timeout at the latest: nothing else ends it but a /// message from the peer, and one that has everything has nothing to say. Without /// that, no message would go to it again. pub(crate) fn withdraw_for_delta( &mut self, doc: &Document, now: Instant, ) -> Vec { if self.reply.is_none() { self.expect_reply(Vec::new(), now, SyncBytes::default(), 0); } 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.state.last_sent_heads, self.state.have_responded) = self.before_generate.clone(); self.delta_offered = true; self.delta_from(doc) } /// Where a delta to the peer starts, sorted: its heads as far as we know them, of /// those we hold. Every node that serves the delta must hold all of them /// (`document::changes::delta_bytes`), so the same bytes come from each; we do. In /// place of heads we lack (changes it has and we don't), the heads we share: the /// delta may then carry some it has, but nothing it lacks. fn delta_from(&self, doc: &Document) -> Vec { let known = |hash: &&crate::document::ChangeHash| doc.get_change_by_hash(hash).is_some(); let theirs = self.state.their_heads.as_deref().unwrap_or_default(); let mut from: Vec<_> = theirs.iter().filter(known).copied().collect(); if from.len() < theirs.len() { from.extend(self.state.shared_heads.iter().filter(known).copied()); } from.sort_unstable(); from.dedup(); from } /// We offered the delta from `from` to `to` in place of a message of `bytes`. pub(crate) fn offered( &mut self, from: &[crate::document::ChangeHash], to: &[crate::document::ChangeHash], bytes: usize, ) { self.last_offer = Some((from.to_vec(), to.to_vec(), bytes)); } /// The delta to offer again, if nothing changed since the last offer (`offered`): /// where it starts (`delta_from`), our `heads`, and the message too large for /// `threshold` still. Then the message would be the same, and so the delta: its /// `from`, offered as by `withdraw_for_delta`, without generating the message. pub(crate) fn offer_again( &mut self, doc: &Document, heads: &[crate::document::ChangeHash], threshold: usize, now: Instant, ) -> Option> { let (from, to, bytes) = self.last_offer.as_ref()?; let sorted = |heads: &[crate::document::ChangeHash]| { let mut heads = heads.to_vec(); heads.sort_unstable(); heads.dedup(); heads }; if *bytes <= threshold || self.delta_from(doc) != *from || sorted(heads) != *to { return None; } let from = from.clone(); if self.reply.is_none() { self.expect_reply(Vec::new(), now, SyncBytes::default(), 0); } // Handed over like a generated message (`send_failed` then retries). self.sending = Some(Sending { sent: Vec::new(), regenerate: false, assumed: None, }); self.delta_offered = true; Some(from) } /// The message just generated, withdrawn for a delta (`withdraw_for_delta`), turns /// out to carry nothing the peer lacks (it has our heads): none goes, and nothing is /// offered or awaited. pub(crate) fn nothing_to_send(&mut self) { self.sending = None; self.delta_offered = false; self.reply_settled(); } /// 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`. pub(crate) fn have_all_theirs(&self, doc: &Document) -> bool { self.state.their_heads.as_ref().is_some_and(|heads| { !heads.is_empty() && heads .iter() .all(|head| doc.get_change_by_hash(head).is_some()) }) } /// The message just generated goes to a peer that had nothing we lack /// (`have_all_theirs`), and no answer was awaited before it: once it has the message, /// it holds exactly our `heads`, and answers nothing (`answered_as_delivered`; the /// envelope says so, `SyncEnvelope::delivered`). So it is taken as delivered: the /// peer's heads are ours, and no answer is awaited (otherwise about half the sync /// messages are those answers: along trees, to pulls, to what is sent again). Should /// it be lost, the peer still lacks it: where it asked (a pull), it asks again after /// its own reply timeout; otherwise the announcements make it pull (the slow round of /// its other neighbours, or ours where it reaches no other member: /// `swarm::tree::Plan::alone`), or our next message, which it can't apply, makes it /// ask. Should the peer have changes we lack after all, it answers with them, as its /// heads differ from ours. Undone if our transport refuses the message /// (`send_failed`). pub(crate) fn assume_delivered(&mut self, heads: &[crate::document::ChangeHash]) { let view = self.peer_view(); let Some(sending) = &mut self.sending else { return; }; sending.assumed = Some(view); self.state.their_heads = Some(heads.to_vec()); self.state.shared_heads = heads.to_vec(); // And it has all up to our heads, needing nothing: the next message carries // what came since, and if what we sent is forgotten (`resend_if_needed`), only // that goes again, not all since the peer's last message, which may be long ago // (no answers come). self.state.their_have = Some(vec![sync::Have { last_sync: heads.to_vec(), bloom: sync::BloomFilter::default(), }]); self.state.their_need = Some(Vec::new()); self.state.in_flight = false; self.reply = None; } /// The changes since `before` in `doc` came from the peer (we just applied its /// message, or a delta it offered): it has them, so they never go back to it. /// Automerge goes by what it last heard the peer has, and a message that overtook /// another makes that older than what the peer sent since: the answer would carry /// the peer's own changes back. A delta comes outside the sync protocol, so /// Automerge would think the peer lacks all of it, and offer it back as a delta of /// our own. Counted as sent (`sent_hashes`), they don't; Automerge forgets them as /// sent once the peer's heads cover them. Returns how many changes that is. (Changes kept until their predecessors came, /// from another peer, count too: should this one lack them, it asks, and gets them, /// `resend_if_needed`.) pub(crate) fn came_from_peer( &mut self, doc: &Document, before: &[crate::document::ChangeHash], ) -> usize { let changes = doc.get_changes_meta(before); for change in &changes { self.state.sent_hashes.insert(change.hash); } changes.len() } /// Of the changes in `doc` beyond `from`, how many the peer has too, as far as we /// know (its heads' history, and what went either way since, `sent_hashes`). A delta /// it offers from `from` brings those again: offered before it heard what we have /// now (while we downloaded its last one, say, and it waited for our answer). pub(crate) fn known_beyond( &self, doc: &Document, from: &[crate::document::ChangeHash], ) -> usize { let from: Vec = from .iter() .copied() .filter(|hash| doc.get_change_by_hash(hash).is_some()) .collect(); let beyond: std::collections::HashMap<_, _> = doc .get_changes_meta(&from) .into_iter() .map(|change| (change.hash, change.deps)) .collect(); // A change's predecessors outside `beyond` are in the history of `from`. let mut known = std::collections::HashSet::new(); let mut pending: Vec = self.state.their_heads.iter().flatten().copied().collect(); pending.extend(self.state.sent_hashes.iter().copied()); while let Some(hash) = pending.pop() { if let Some(deps) = beyond.get(&hash) && known.insert(hash) { pending.extend(deps.iter().copied()); } } known.len() } /// The peer has `changes`, or will without us (another parent in their writer's /// tree brings them, `swarm::tree::forward`): counted as sent, they don't go to it, /// unless it asks for them (`resend_if_needed`). pub(crate) fn known_to_peer(&mut self, changes: &[crate::document::ChangeHash], now: Instant) { self.known_elsewhere .retain(|_, at| now.saturating_duration_since(*at) < KNOWN_ELSEWHERE_KEPT); for hash in changes { if self.state.sent_hashes.insert(*hash) { self.known_elsewhere.insert(*hash, now); } } } /// We pass `changes` on to the peer: of their predecessors, those counted as sent /// lately only because another parent brings them (`known_to_peer`) go along. That /// one may not have arrived yet (with writers in turn, each change builds on the /// last writer's, which comes along another tree): without them, the peer couldn't /// apply ours, and would ask for them, a round trip later. pub(crate) fn with_predecessors( &mut self, doc: &Document, changes: &[crate::document::ChangeHash], now: Instant, ) { for hash in changes { let Some(change) = doc.get_change_by_hash(hash) else { continue; }; for dep in change.deps() { let lately = self .known_elsewhere .get(dep) .is_some_and(|at| now.saturating_duration_since(*at) < KNOWN_ELSEWHERE_FOR); if lately { self.known_elsewhere.remove(dep); self.state.sent_hashes.remove(dep); } } } } /// What the sync state knows of the peer (`PeerView`). fn peer_view(&self) -> PeerView { PeerView { their_heads: self.state.their_heads.clone(), shared_heads: self.state.shared_heads.clone(), their_have: self.state.their_have.clone(), their_need: self.state.their_need.clone(), } } /// We applied a message its sender took as delivered (`assume_delivered`, /// `SyncEnvelope::delivered`) to `doc`: if we hold exactly its sender's heads now, /// Automerge's answer, our heads, would tell it nothing. So none goes: we count as /// having said our `heads`. Returns whether so. /// /// So also where the message is older than the last one from the peer (they /// overtake each other, as no answers are awaited) and we hold exactly the heads /// that one said: the older one tells nothing new of the peer, all its heads are in /// our history, and we are level with the peer as of the newer. Otherwise the /// older one's view of what we have would stand, and the answer to it would carry /// back to the peer the changes the newer one brought us. pub(crate) fn answered_as_delivered( &mut self, doc: &Document, heads: &[crate::document::ChangeHash], ) -> bool { let sorted = |heads: &[crate::document::ChangeHash]| { let mut heads = heads.to_vec(); heads.sort_unstable(); heads }; let ours = sorted(heads); let level = |theirs: &Option>| { theirs.as_deref().map(sorted).as_ref() == Some(&ours) }; let before = self.before_receive.take(); if !level(&self.state.their_heads) { let older = self.state.their_heads.as_ref().is_some_and(|theirs| { theirs .iter() .all(|hash| doc.get_change_by_hash(hash).is_some()) }); if !before.is_some_and(|before| older && level(&before.their_heads)) { return false; } } // Level: the peer has exactly our heads and needs nothing. Said so outright, not // as the message's view (or the one before it): a view whose heads we lacked // names older ones (`known_last_sync`), by which it would seem to lack what came // since. self.state.their_heads = Some(heads.to_vec()); self.state.shared_heads = heads.to_vec(); self.state.their_have = Some(vec![sync::Have { last_sync: heads.to_vec(), bloom: sync::BloomFilter::default(), }]); self.state.their_need = Some(Vec::new()); self.state.last_sent_heads = heads.to_vec(); self.state.have_responded = true; true } /// The peer has heads we lack (its keepalive said so): ask for what they need, /// even though Automerge thinks a message is still in flight. pub(crate) fn pull_from(&mut self, heads: Vec) { self.state.their_heads = Some(heads); self.state.in_flight = false; } /// Automerge never resends a change it believes the peer has received /// (`sent_hashes`). 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 of it is wrong: forget having sent it, and /// what it builds on that we sent too (`forget_sent_for_need`), and the next message /// carries them. Not what we sent after it: messages overtake each other (we don't /// wait for answers to those taken as delivered), so a peer often needs what is only /// late, while the messages after it are still on their way. Call after `receive`. /// /// Returns how many of the changes it asked for we had counted as its because another /// parent brings them (`known_to_peer`): within `KNOWN_ELSEWHERE_FOR` of counting /// them so, and later (`swarm::stats`). pub(crate) fn resend_if_needed( &mut self, doc: &Document, peer: &str, now: Instant, ) -> (usize, usize) { let mut asked = (0, 0); for hash in self.state.their_need.iter().flatten() { if let Some(at) = self.known_elsewhere.remove(hash) { if now.saturating_duration_since(at) < KNOWN_ELSEWHERE_FOR { asked.0 += 1; } else { asked.1 += 1; } } } let forgotten = forget_sent_for_need(&mut self.state, doc); if forgotten > 0 { debug!( peer = %peer, forgotten, "Peer needs changes we had sent; sending them again" ); } if answer_for_everything_since_shared(&mut self.state, doc) { debug!( peer = %peer, need = self.state.their_need.as_ref().map_or(0, Vec::len), "Peer lacks the changes ours build on; sending everything since what we share" ); } asked } /// For the simulator's dump: the peer's heads as Automerge knows them, and whether /// it thinks a message is in flight. #[cfg(test)] pub(crate) fn sync_view(&self) -> (Option<&[crate::document::ChangeHash]>, bool) { (self.state.their_heads.as_deref(), self.state.in_flight) } /// Whether we leave the first message of this session to the peer. /// /// After a reset, both sides usually open at once, and the openings cross. Between /// peers that share history that costs nothing. But a peer with an empty document /// answers our opening with empty heads, which Automerge takes for "lost /// everything", so the whole document would go out a second time. So a ready node /// never opens towards a peer that isn't ready (one that is still catching up); that /// peer opens itself. pub(crate) fn leave_opening(&self, we_ready: bool, peer_ready: Option) -> bool { self.state.their_heads.is_none() && we_ready && peer_ready == Some(false) } /// Which session this is: a message generated from it reports its send's outcome /// with this (`send_succeeded`, `send_failed`). pub(crate) fn id(&self) -> u64 { self.id } /// Our transport took the message generated from session `session`. Returns whether /// to generate again: a pass was requested while it was handed over, or a reset /// replaced the session meanwhile (any pass requested then was lost with the old /// session, and the new one hasn't necessarily opened yet). pub(crate) fn send_succeeded(&mut self, session: u64) -> bool { if session != self.id { return true; } self.retry_at = None; self.send_backoff.reset(); self.sending .take() .is_some_and(|sending| sending.regenerate) } /// Our transport refused the message: as far as the peer is concerned, it was never /// sent. Takes back its changes (they go out with the next message), clears /// `in_flight`, and returns how long to wait before generating again. None if a reset /// already discarded what was being sent (the message came from another `session`). pub(crate) fn send_failed(&mut self, session: u64, now: Instant) -> Option { if session != self.id { return None; } let sending = self.sending.take()?; for hash in &sending.sent { self.state.sent_hashes.remove(hash); } if let Some(assumed) = sending.assumed { self.state.their_heads = assumed.their_heads; self.state.shared_heads = assumed.shared_heads; self.state.their_have = assumed.their_have; self.state.their_need = assumed.their_need; } if let Some(wait) = &mut self.reply { wait.sent.retain(|hash| !sending.sent.contains(hash)); } self.state.in_flight = false; let delay = self.send_backoff.next(); self.retry_at = Some(now + delay); Some(delay) } /// A sync message of `len` bytes went out; `sent` are the changes it carried for the /// first time. `bytes` are the sync bytes exchanged with the peer before it. pub(crate) fn expect_reply( &mut self, sent: Vec, now: Instant, bytes: SyncBytes, len: usize, ) { match &mut self.reply { // Still waiting for an earlier one: keep its deadline, remember these too. Some(wait) => { wait.sent.extend(sent); wait.largest = wait.largest.max(len); } None => { self.reply = Some(SyncReplyWait { deadline: now + self.reply_backoff.current(), started: (now, bytes), sent, largest: len, }); } } } /// No answer is due any more: the peer answered, or keepalives show both documents /// are the same. pub(crate) fn reply_settled(&mut self) { self.reply = None; self.reply_backoff.reset(); } /// If the answer is overdue: forgets that the unanswered changes were sent and that /// a message is in flight, so the next sync pass sends them again. Returns how many /// changes that concerns. pub(crate) fn take_overdue_reply(&mut self, now: Instant) -> Option { if self.reply.as_ref().is_none_or(|wait| now < wait.deadline) { return None; } let wait = self.reply.take()?; for hash in &wait.sent { self.state.sent_hashes.remove(hash); } self.state.in_flight = false; self.delta_offered = false; self.reply_backoff.grow(); Some(wait.sent.len()) } } /// A peer that got a change of ours without the changes it builds on (they were lost, /// a later one arrived) asks for those it lacks, and while it does, Automerge leaves /// out what it has (its bloom filter: `have` waits until all it needs is in our heads). /// Without a `have`, Automerge sends exactly what was asked for, and the peer, finding /// each change's own predecessors missing in turn, asks again: one round trip per /// missing change, hundreds after a short outage. So answer as if it had said it has /// nothing since the heads we share: everything since then goes in one message (a /// delta when large: `swarm::sync`), and what it had already only twice. But what we /// sent and it doesn't need may be on its way still: that stays sent, but for what it /// needs and what that builds on (`forget_sent_for_need`). Returns whether it did. Call /// after receiving. fn answer_for_everything_since_shared(state: &mut sync::State, doc: &Document) -> bool { let asks_without_saying = state.their_have.as_ref().is_some_and(Vec::is_empty) && state .their_need .as_ref() .is_some_and(|need| !need.is_empty()); if !asks_without_saying { return false; } state.their_have = Some(vec![sync::Have { last_sync: state.shared_heads.clone(), bloom: sync::BloomFilter::default(), }]); // What it needs, and what that builds on, may be what was lost. forget_sent_for_need(state, doc); true } /// Forgets having sent what the peer needs (`their_need`) and the changes those build /// on that we sent too, following their predecessors as long as we sent them: so all of /// a run of lost messages goes again at once. Changes sent after them stay sent (they /// may only be late). Returns how many were forgotten. fn forget_sent_for_need(state: &mut sync::State, doc: &Document) -> usize { let Some(need) = &state.their_need else { return 0; }; let mut forgotten = 0; let mut next: Vec = need .iter() .filter(|hash| state.sent_hashes.contains(hash)) .copied() .collect(); while let Some(hash) = next.pop() { if !state.sent_hashes.remove(&hash) { continue; } forgotten += 1; if let Some(change) = doc.get_change_by_hash(&hash) { next.extend( change .deps() .iter() .filter(|dep| state.sent_hashes.contains(dep)) .copied(), ); } } forgotten } #[cfg(test)] mod tests { use super::*; /// After a loss, the changes the peer lacks go in one message, not one per round /// trip (`answer_for_everything_since_shared`). #[test] fn missing_predecessors_go_at_once() { use crate::document::testing::change; use automerge::transaction::Transactable; let mut a = Document::new(); let mut b = Document::new(); let (mut at_a, mut at_b) = (sync::State::new(), sync::State::new()); change(&mut a, |a| a.put(automerge::ROOT, "early", 1).unwrap()); // Until in step. for _ in 0..10 { let to_b = a.generate_sync_message(&mut at_a); let to_a = b.generate_sync_message(&mut at_b); if to_b.is_none() && to_a.is_none() { break; } if let Some(m) = to_b { b.receive_sync_message(&mut at_b, m).unwrap(); } if let Some(m) = to_a { a.receive_sync_message(&mut at_a, m).unwrap(); } } assert_eq!(a.get_heads(), b.get_heads()); // Five changes, whose message is lost; then a sixth, which arrives alone. for i in 0..5 { change(&mut a, |a| { a.put(automerge::ROOT, format!("lost{i}"), i).unwrap() }); } let _lost = a.generate_sync_message(&mut at_a).unwrap(); change(&mut a, |a| a.put(automerge::ROOT, "last", 6).unwrap()); at_a.in_flight = false; // As after our reply timeout. let last = a.generate_sync_message(&mut at_a).unwrap(); assert_eq!(last.changes.len(), 1); b.receive_sync_message(&mut at_b, last).unwrap(); // b asks for the fifth, saying nothing of what it has. let ask = b.generate_sync_message(&mut at_b).unwrap(); assert!(ask.have.is_empty() && ask.need.len() == 1); a.receive_sync_message(&mut at_a, ask).unwrap(); assert!(answer_for_everything_since_shared(&mut at_a, &a)); // All since the shared heads, in one message (here the whole document: more // than a third of it is missing). let answer = a.generate_sync_message(&mut at_a).unwrap(); b.receive_sync_message(&mut at_b, answer).unwrap(); assert_eq!(b.get_heads(), a.get_heads()); // A message with what it has is left alone. assert!(!answer_for_everything_since_shared(&mut at_b, &b)); } fn session() -> SyncSession { SyncSession::new(Backoff::new( Duration::from_secs(1), Duration::from_secs(30), )) } fn sending() -> Option { Some(Sending { sent: Vec::new(), regenerate: false, assumed: None, }) } #[test] fn a_send_from_a_replaced_session_leaves_the_new_one_alone() { let old = session(); let mut new = session(); new.sending = sending(); // A reset discarded the old session while its message was handed over: the // new one's own message stays, and the peer is synced again. assert!(new.send_succeeded(old.id())); assert!(new.sending.is_some()); assert_eq!(new.send_failed(old.id(), Instant::now()), None); assert!(new.sending.is_some()); assert_eq!(new.retry_at, None); // Its own outcome still applies. assert!(!new.send_succeeded(new.id())); assert!(new.sending.is_none()); } /// A delta offered where no answer was awaited (the peer's heads were ours) still /// lapses at the reply timeout: else nothing ends the wait for a peer with nothing /// to say, and no message goes to it again. #[test] fn an_offered_delta_always_lapses() { let mut a = session(); a.sending = sending(); assert!(a.reply.is_none()); let now = Instant::now(); a.withdraw_for_delta(&Document::new(), now); assert!(a.delta_offered); assert!( a.take_overdue_reply(now + Duration::from_secs(3600)) .is_some() ); assert!(!a.delta_offered); } /// An offer that lapsed goes again as it was while nothing changed; otherwise, or /// once the message would fit, it is generated afresh. #[test] fn the_same_offer_goes_again_ungenerated() { use crate::document::testing::change; use automerge::{ROOT, transaction::Transactable}; let mut ours = Document::new(); change(&mut ours, |doc| doc.put(ROOT, "k", 1).unwrap()); let heads = ours.get_heads(); let mut a = session(); a.state.their_heads = Some(Vec::new()); let now = Instant::now(); assert_eq!( a.offer_again(&ours, &heads, 1000, now), None, "nothing offered yet" ); a.offered(&[], &heads, 5000); assert_eq!(a.offer_again(&ours, &heads, 1000, now), Some(Vec::new())); assert!(a.delta_offered && a.reply.is_some() && a.sending.is_some()); assert_eq!( a.offer_again(&ours, &heads, 5000, now), None, "it would fit now" ); change(&mut ours, |doc| doc.put(ROOT, "k", 2).unwrap()); assert_eq!( a.offer_again(&ours, &ours.get_heads(), 1000, now), None, "ours moved" ); a.state.their_heads = Some(heads.clone()); assert_eq!( a.offer_again(&ours, &heads, 1000, now), None, "theirs moved" ); } /// 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 crate::document::testing::change; 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 = Document::new(); change(&mut ours, |ours| ours.put(ROOT, "big", big).unwrap()); let mut theirs = Document::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(&ours, now, bytes).unwrap(); a.send_succeeded(a.id()); b.receive(&mut theirs, decode(&opening).unwrap()).unwrap(); let answer = b.generate(&theirs, now, bytes).unwrap(); b.send_succeeded(b.id()); a.receive(&mut ours, decode(&answer).unwrap()).unwrap(); let carrying = a.generate(&ours, now, bytes).unwrap(); let carried = carrying.0.len(); assert!(carried > 4_000, "the changes took {carried} bytes"); let from = a.withdraw_for_delta(&ours, now); 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(&ours, now, bytes).unwrap(); let carried_again = again.0.len(); assert!(carried_again > 4_000, "sent again in {carried_again} bytes"); } /// Two sessions exchanging messages until neither has anything to say. fn in_step(ours: &mut Document, theirs: &mut Document) -> (SyncSession, SyncSession) { let (mut a, mut b) = (session(), session()); let (now, bytes) = (Instant::now(), SyncBytes::default()); for _ in 0..10 { let to_b = a.generate(ours, now, bytes); a.send_succeeded(a.id()); let to_a = b.generate(theirs, now, bytes); b.send_succeeded(b.id()); if to_b.is_none() && to_a.is_none() { break; } if let Some(m) = to_b { b.receive(theirs, decode(&m).unwrap()).unwrap(); } if let Some(m) = to_a { a.receive(ours, decode(&m).unwrap()).unwrap(); } } assert_eq!(ours.get_heads(), theirs.get_heads()); (a, b) } /// A message to a peer that had everything else: taken as delivered, and not /// answered; neither side has more to say. #[test] fn delivered_messages_need_no_answer() { use crate::document::testing::change; use automerge::{ROOT, transaction::Transactable}; let (mut ours, mut theirs) = (Document::new(), Document::new()); change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap()); let (mut a, mut b) = in_step(&mut ours, &mut theirs); let (now, bytes) = (Instant::now(), SyncBytes::default()); change(&mut ours, |doc| doc.put(ROOT, "news", 2).unwrap()); assert!(a.have_all_theirs(&ours)); let message = a.generate(&ours, now, bytes).unwrap(); a.assume_delivered(&ours.get_heads()); a.send_succeeded(a.id()); assert!(a.reply.is_none(), "no answer awaited"); b.receive(&mut theirs, decode(&message).unwrap()).unwrap(); assert!(b.answered_as_delivered(&theirs, &theirs.get_heads())); assert!( b.generate(&theirs, now, bytes).is_none(), "nothing to answer" ); assert!( a.generate(&ours, now, bytes).is_none(), "nothing more to send" ); assert_eq!(ours.get_heads(), theirs.get_heads()); } /// The peer wrote meanwhile: its heads differ from ours after our message, so it /// answers, with its change. #[test] fn delivered_messages_are_answered_with_changes_we_lack() { use crate::document::testing::change; use automerge::{ROOT, transaction::Transactable}; let (mut ours, mut theirs) = (Document::new(), Document::new()); change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap()); let (mut a, mut b) = in_step(&mut ours, &mut theirs); let (now, bytes) = (Instant::now(), SyncBytes::default()); change(&mut ours, |doc| doc.put(ROOT, "ours", 2).unwrap()); change(&mut theirs, |doc| doc.put(ROOT, "theirs", 3).unwrap()); let message = a.generate(&ours, now, bytes).unwrap(); a.assume_delivered(&ours.get_heads()); a.send_succeeded(a.id()); b.receive(&mut theirs, decode(&message).unwrap()).unwrap(); assert!(!b.answered_as_delivered(&theirs, &theirs.get_heads())); let answer = b.generate(&theirs, now, bytes).expect("b answers"); b.send_succeeded(b.id()); a.receive(&mut ours, decode(&answer).unwrap()).unwrap(); assert_eq!(ours.get_heads(), theirs.get_heads()); } /// A message taken as delivered is lost, and the next one arrives: the peer can't /// apply it, answers what it needs, and gets the lost change then. #[test] fn a_lost_delivered_message_is_asked_for() { use crate::document::testing::change; use automerge::{ROOT, transaction::Transactable}; let (mut ours, mut theirs) = (Document::new(), Document::new()); change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap()); let (mut a, mut b) = in_step(&mut ours, &mut theirs); let (now, bytes) = (Instant::now(), SyncBytes::default()); change(&mut ours, |doc| doc.put(ROOT, "lost", 2).unwrap()); let _lost = a.generate(&ours, now, bytes).unwrap(); a.assume_delivered(&ours.get_heads()); a.send_succeeded(a.id()); change(&mut ours, |doc| doc.put(ROOT, "next", 3).unwrap()); assert!( a.have_all_theirs(&ours), "a takes the lost one as delivered" ); let next = a.generate(&ours, now, bytes).unwrap(); a.assume_delivered(&ours.get_heads()); a.send_succeeded(a.id()); b.receive(&mut theirs, decode(&next).unwrap()).unwrap(); assert!(!b.answered_as_delivered(&theirs, &theirs.get_heads())); let ask = b .generate(&theirs, now, bytes) .expect("b asks for what it lacks"); b.send_succeeded(b.id()); a.receive(&mut ours, decode(&ask).unwrap()).unwrap(); a.resend_if_needed(&ours, "b", Instant::now()); let answer = a.generate(&ours, now, bytes).expect("a sends the lost one"); a.send_succeeded(a.id()); b.receive(&mut theirs, decode(&answer).unwrap()).unwrap(); assert_eq!(ours.get_heads(), theirs.get_heads()); } /// Two messages taken as delivered arrive in reverse order: the peer asks for the /// first, which arrives meanwhile; only the first is sent again (the second was on /// its way), taken as delivered too. Nobody waits for an answer that won't come. #[test] fn reordered_delivered_messages_leave_nothing_awaited() { use crate::document::testing::change; use automerge::{ROOT, transaction::Transactable}; let (mut ours, mut theirs) = (Document::new(), Document::new()); change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap()); let (mut a, mut b) = in_step(&mut ours, &mut theirs); let (now, bytes) = (Instant::now(), SyncBytes::default()); let send = |a: &mut SyncSession, ours: &Document| { assert!(a.reply.is_none() && a.have_all_theirs(ours)); let message = a.generate(ours, now, bytes).unwrap(); a.assume_delivered(&ours.get_heads()); a.send_succeeded(a.id()); message }; change(&mut ours, |doc| doc.put(ROOT, "first", 2).unwrap()); let first = send(&mut a, &ours); change(&mut ours, |doc| doc.put(ROOT, "second", 3).unwrap()); let second = send(&mut a, &ours); b.receive(&mut theirs, decode(&second).unwrap()).unwrap(); assert!(!b.answered_as_delivered(&theirs, &theirs.get_heads())); let ask = b .generate(&theirs, now, bytes) .expect("b asks for the first"); // The second named the heads the first brought as shared, which b lacks: no // reset for that, else a forgets all it sent. let flags = sync::Message::decode(&ask.0).unwrap().flags; assert!( !flags.is_some_and(|flags| flags.contains(sync::MessageFlags::SYNC_RESET)), "b asked with a reset" ); b.send_succeeded(b.id()); b.receive(&mut theirs, decode(&first).unwrap()).unwrap(); assert_eq!(theirs.get_heads(), ours.get_heads(), "b has both now"); // The first is older than the second, whose heads b holds now: level with a as // of that, b says nothing, and doesn't send a the second back. assert!(b.answered_as_delivered(&theirs, &theirs.get_heads())); assert!( b.generate(&theirs, now, bytes).is_none(), "nothing sent back" ); a.receive(&mut ours, decode(&ask).unwrap()).unwrap(); a.resend_if_needed(&ours, "b", Instant::now()); let sent_before = a.state.sent_hashes.clone(); let again = send(&mut a, &ours); // What the message carried: the changes it added to what a counts as sent. let carried = a.state.sent_hashes.difference(&sent_before).count(); assert_eq!( carried, 1, "only the first again, not the second, on its way" ); b.receive(&mut theirs, decode(&again).unwrap()).unwrap(); assert!(b.answered_as_delivered(&theirs, &theirs.get_heads())); assert!( b.generate(&theirs, now, bytes).is_none(), "nothing to answer" ); assert!(a.reply.is_none(), "nothing awaited"); assert_eq!(ours.get_heads(), theirs.get_heads()); } /// After messages taken as delivered (no answers, so nothing new heard of what the /// peer has), forgetting what was sent resends only what came since the last of /// them, not all since the peer's last message. #[test] fn after_delivered_messages_only_news_goes_again() { use crate::document::testing::change; use automerge::{ROOT, transaction::Transactable}; let (mut ours, mut theirs) = (Document::new(), Document::new()); change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap()); let (mut a, mut b) = in_step(&mut ours, &mut theirs); let (now, bytes) = (Instant::now(), SyncBytes::default()); for i in 0..5 { change(&mut ours, |doc| doc.put(ROOT, format!("k{i}"), i).unwrap()); let message = a.generate(&ours, now, bytes).unwrap(); a.assume_delivered(&ours.get_heads()); a.send_succeeded(a.id()); b.receive(&mut theirs, decode(&message).unwrap()).unwrap(); assert!(b.answered_as_delivered(&theirs, &theirs.get_heads())); } // As `resend_if_needed` does when the peer needs what we sent. a.state.sent_hashes.clear(); change(&mut ours, |doc| doc.put(ROOT, "news", 6).unwrap()); a.generate(&ours, now, bytes).unwrap(); // What the message carried: what a counts as sent now (nothing before). assert_eq!(a.state.sent_hashes.len(), 1, "only the news"); } /// Three messages arrive in reverse order: once the first is in, all three apply, /// and whatever goes back to the peer carries none of its own changes. #[test] fn changes_from_the_peer_never_go_back() { use crate::document::testing::change; use automerge::{ROOT, transaction::Transactable}; let (mut ours, mut theirs) = (Document::new(), Document::new()); change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap()); let (mut a, mut b) = in_step(&mut ours, &mut theirs); let (now, bytes) = (Instant::now(), SyncBytes::default()); let mut messages = Vec::new(); for i in 0..3 { change(&mut ours, |doc| doc.put(ROOT, format!("k{i}"), i).unwrap()); messages.push(a.generate(&ours, now, bytes).unwrap()); a.assume_delivered(&ours.get_heads()); a.send_succeeded(a.id()); } for message in messages.iter().rev() { let before = theirs.get_heads(); b.receive(&mut theirs, decode(message).unwrap()).unwrap(); b.came_from_peer(&theirs, &before); b.answered_as_delivered(&theirs, &theirs.get_heads()); } assert_eq!(theirs.get_heads(), ours.get_heads()); if let Some(answer) = b.generate(&theirs, now, bytes) { let (carried, _) = decode(&answer).unwrap().sizes(); assert_eq!(carried, 0, "a's own changes went back to it"); } } /// A delta starts at the peer's heads where we hold them all; where it has one we /// lack, at those we hold and the heads we share, so every server of the delta /// holds where it starts. #[test] fn a_delta_starts_where_we_hold_its_start() { use crate::document::testing::change; use automerge::{ROOT, transaction::Transactable}; let mut doc = Document::new(); change(&mut doc, |doc| doc.put(ROOT, "k", 1).unwrap()); let shared = doc.get_heads(); change(&mut doc, |doc| doc.put(ROOT, "k", 2).unwrap()); let known = doc.get_heads(); let unknown = crate::document::ChangeHash([7; 32]); let mut a = session(); a.state.shared_heads = shared.clone(); a.state.their_heads = Some(known.clone()); assert_eq!(a.delta_from(&doc), known); a.state.their_heads = Some(vec![known[0], unknown]); let mut expected = vec![known[0], shared[0]]; expected.sort_unstable(); assert_eq!(a.delta_from(&doc), expected); } /// A delta offered from heads we since moved past, by changes from the offerer, /// would bring those again; our own, which it lacks, don't count. #[test] fn offers_from_before_count_what_came_since() { use crate::document::changes::{delta_bytes, load_delta}; use crate::document::testing::change; use automerge::{ROOT, transaction::Transactable}; let (mut ours, mut theirs) = (Document::new(), Document::new()); change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap()); let (_a, mut b) = in_step(&mut ours, &mut theirs); let before = theirs.get_heads(); change(&mut ours, |doc| doc.put(ROOT, "news", 2).unwrap()); let delta = delta_bytes(&ours, &before, &ours.get_heads()).unwrap(); load_delta(&mut theirs, &delta).unwrap(); b.came_from_peer(&theirs, &before); change(&mut theirs, |doc| doc.put(ROOT, "own", 3).unwrap()); assert_eq!( b.known_beyond(&theirs, &before), 1, "the delta's change again" ); assert_eq!(b.known_beyond(&theirs, &theirs.get_heads()), 0); } /// Changes the peer gets from elsewhere (another parent in their writer's tree) /// don't go to it, but do if it asks. #[test] fn changes_known_to_the_peer_go_only_when_asked() { use crate::document::testing::change; use automerge::{ROOT, transaction::Transactable}; let (mut ours, mut theirs) = (Document::new(), Document::new()); change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap()); let (mut a, mut b) = in_step(&mut ours, &mut theirs); let (now, bytes) = (Instant::now(), SyncBytes::default()); change(&mut ours, |doc| doc.put(ROOT, "elsewhere", 2).unwrap()); a.known_to_peer(&ours.get_heads(), Instant::now()); let message = a.generate(&ours, now, bytes).expect("our heads moved"); a.send_succeeded(a.id()); let (carried, _) = decode(&message).unwrap().sizes(); assert_eq!(carried, 0, "not sent: the peer gets it from elsewhere"); // It didn't: it asks, and gets it. b.receive(&mut theirs, decode(&message).unwrap()).unwrap(); let asking = b.generate(&theirs, now, bytes).expect("it asks"); b.send_succeeded(b.id()); a.receive(&mut ours, decode(&asking).unwrap()).unwrap(); a.resend_if_needed(&ours, "b", Instant::now()); let answer = a.generate(&ours, now, bytes).expect("answered"); b.receive(&mut theirs, decode(&answer).unwrap()).unwrap(); assert_eq!(theirs.get_heads(), ours.get_heads()); } /// A change counted as known to the peer lately goes along with one of ours that /// builds on it: the peer applies both at once, without asking. #[test] fn predecessors_known_elsewhere_go_along() { use crate::document::testing::change; use automerge::{ROOT, transaction::Transactable}; let (mut ours, mut theirs) = (Document::new(), Document::new()); change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap()); let (mut a, mut b) = in_step(&mut ours, &mut theirs); let (now, bytes) = (Instant::now(), SyncBytes::default()); change(&mut ours, |doc| doc.put(ROOT, "elsewhere", 2).unwrap()); a.known_to_peer(&ours.get_heads(), now); change(&mut ours, |doc| doc.put(ROOT, "along", 3).unwrap()); a.with_predecessors(&ours, &ours.get_heads(), now); let message = a.generate(&ours, now, bytes).expect("our heads moved"); b.receive(&mut theirs, decode(&message).unwrap()).unwrap(); assert_eq!(theirs.get_heads(), ours.get_heads(), "both in one message"); } /// Our transport refused a message taken as delivered: the peer's heads are as /// before, so the next message carries the change again. #[test] fn a_refused_delivered_message_goes_again() { use crate::document::testing::change; use automerge::{ROOT, transaction::Transactable}; let (mut ours, mut theirs) = (Document::new(), Document::new()); change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap()); let (mut a, mut b) = in_step(&mut ours, &mut theirs); let (now, bytes) = (Instant::now(), SyncBytes::default()); change(&mut ours, |doc| doc.put(ROOT, "news", 2).unwrap()); let _refused = a.generate(&ours, now, bytes).unwrap(); a.assume_delivered(&ours.get_heads()); assert!(a.send_failed(a.id(), now).is_some()); let again = a.generate(&ours, now, bytes).expect("sent again"); a.send_succeeded(a.id()); b.receive(&mut theirs, decode(&again).unwrap()).unwrap(); assert_eq!(ours.get_heads(), theirs.get_heads()); } /// A joining node opens, the answer is too large and becomes a delta, the node gets /// the delta and says so: the answer to that carries our heads, so the node knows it /// has everything (`have_all_theirs`). #[test] fn after_a_delta_the_peer_learns_our_heads() { use crate::document::testing::change; use automerge::{ROOT, transaction::Transactable}; let mut rng = crate::testrand::Rng::new(2); let big: String = (0..1024) .map(|_| format!("{:016x}", rng.next_u64())) .collect(); let mut ours = Document::new(); change(&mut ours, |ours| ours.put(ROOT, "big", big).unwrap()); let mut theirs = Document::new(); let (mut a, mut b) = (session(), session()); let now = Instant::now(); let bytes = SyncBytes::default(); // b (joining) opens; a's answer would carry everything: withdrawn for a delta. let opening = b.generate(&theirs, now, bytes).unwrap(); b.send_succeeded(b.id()); a.receive(&mut ours, decode(&opening).unwrap()).unwrap(); assert!(a.generate(&ours, now, bytes).is_some()); a.withdraw_for_delta(&ours, now); a.send_succeeded(a.id()); // b loads the delta and tells a its heads, which are a's now. theirs.merge(&mut ours.clone()).unwrap(); assert!(!b.have_all_theirs(&theirs), "b hasn't heard a's heads yet"); let after_delta = b.generate(&theirs, now, bytes).unwrap(); b.send_succeeded(b.id()); a.receive(&mut ours, decode(&after_delta).unwrap()).unwrap(); let answer = a .generate(&ours, now, bytes) .expect("a answers with its heads"); b.receive(&mut theirs, decode(&answer).unwrap()).unwrap(); assert!(b.have_all_theirs(&theirs)); } }