diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/document/sync.rs | 90 | ||||
| -rw-r--r-- | src/swarm/peer/inbound.rs | 4 | ||||
| -rw-r--r-- | src/swarm/peer/keepalive.rs | 28 |
3 files changed, 121 insertions, 1 deletions
diff --git a/src/document/sync.rs b/src/document/sync.rs index 5d139e2..5669ce3 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -825,6 +825,56 @@ impl SyncSession { true } + /// The peer said its `heads` outside the sync (a keepalive STATUS): it has every + /// change in their history. Automerge forgets changes as sent only when the peer's + /// sync message says it has them, and a neighbour another parent brings the changes + /// to (`known_to_peer`) sends us none while it has nothing to write: counted as sent, + /// every change would stay so. So those in the history of `heads` count as sent no + /// more, and the peer as having all up to `heads`: the next message to it carries + /// what lies beyond them, not those. Only forward: where `heads` aren't all ours, or + /// don't cover what the peer said before or what we take it to have (a message of + /// ours still on its way, `assume_delivered`), nothing changes. Returns how many + /// changes count as sent no more. + pub(crate) fn heard_heads( + &mut self, + doc: &Document, + heads: &[crate::document::ChangeHash], + ) -> usize { + use crate::document::changes::{covers, holds_all}; + if heads.is_empty() || !holds_all(doc, heads) { + return 0; + } + let behind = |older: &[crate::document::ChangeHash]| !covers(doc, heads, older); + if self.said_heads.as_deref().is_some_and(behind) + || self.state.their_heads.as_deref().is_some_and(behind) + || self + .state + .their_have + .iter() + .flatten() + .any(|have| behind(&have.last_sync)) + { + return 0; + } + let beyond: HashSet<crate::document::ChangeHash> = doc + .get_changes_meta(heads) + .iter() + .map(|change| change.hash) + .collect(); + let before = self.state.sent_hashes.len(); + self.state + .sent_hashes + .retain(|hash| beyond.contains(hash) || doc.get_change_by_hash(hash).is_none()); + 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.said_heads = Some(heads.to_vec()); + before - self.state.sent_hashes.len() + } + /// 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<crate::document::ChangeHash>) { @@ -1340,6 +1390,46 @@ mod tests { assert_eq!(a.sent_hashes_len(), before); } + /// A neighbour another parent brings our changes to sends us no sync message: what + /// counts as sent to it is forgotten once its keepalive says its heads, and nothing + /// goes to it again. + #[test] + fn heads_heard_forget_what_counts_as_sent() { + 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, _b) = in_step(&mut ours, &mut theirs); + let (now, bytes) = (Instant::now(), SyncBytes::default()); + for i in 0..100 { + change(&mut ours, |doc| doc.put(ROOT, "key", i).unwrap()); + a.known_to_peer(&ours.get_heads(), now); + } + theirs.merge(&mut ours.clone()).unwrap(); + assert_eq!(a.sent_hashes_len(), 100); + + // Heads it said before we had them all: nothing changes. + assert_eq!( + a.heard_heads(&ours, &[crate::document::ChangeHash([7; 32])]), + 0 + ); + assert_eq!(a.heard_heads(&ours, &theirs.get_heads()), 100); + assert_eq!(a.sent_hashes_len(), 0); + let carries = |message: Option<SyncPayload>| { + message.is_some_and(|message| decode(&message).unwrap().carries_changes()) + }; + assert!(!carries(a.generate(&ours, now, bytes))); + + // Older heads than those, said later (a STATUS overtaken): nothing changes. + change(&mut ours, |doc| doc.put(ROOT, "key", 100).unwrap()); + a.known_to_peer(&ours.get_heads(), now); + let older = theirs.get_heads(); + theirs.merge(&mut ours.clone()).unwrap(); + assert_eq!(a.heard_heads(&ours, &theirs.get_heads()), 1); + assert_eq!(a.heard_heads(&ours, &older), 0); + assert!(!carries(a.generate(&ours, now, bytes))); + } + /// Automerge, as we rely on it: a message with no heads says the peer lost all it /// had, and what we counted as its goes again (why a push never goes without heads: /// `Received::pushed`, and an opening never goes empty: `leave_opening`). diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs index 1542d4f..8a424b8 100644 --- a/src/swarm/peer/inbound.rs +++ b/src/swarm/peer/inbound.rs @@ -12,7 +12,8 @@ use super::handshake::{ discover_peers, ensure_handshake, establish_peer, peer_swarm, remember_peer, send_hello_reply, }; use super::keepalive::{ - heal_if_diverged, heard_from, local_status, note_alive, note_status, status_received, + heal_if_diverged, heard_from, heard_heads, local_status, note_alive, note_status, + status_received, }; use super::lanes::SyncLanes; use crate::locks::MutexExt; @@ -199,6 +200,7 @@ pub(crate) async fn handle_message( // no reason, as changes on their way to us make them differ while writes go on. if !remote.announcing { status_received(state, &peer).await; + heard_heads(state, &peer, &remote); } let local = on_status(state, &peer, &remote, reactivated).await; if remote.announcing { diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index d698585..693a370 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -218,6 +218,34 @@ pub(super) async fn status_received(state: &AppState, peer: &str) { } } +/// A keepalive STATUS said the peer's heads: what it has counts as sent to it no more +/// (`SyncSession::heard_heads`). Off the consumer: it takes the document's lock. +pub(super) fn heard_heads(state: &AppState, peer: &str, remote: &transport::Status) { + let Some(heads) = remote.heads.heads() else { + return; + }; + if heads.is_empty() || remote.epoch != state.local_epoch() { + return; + } + let (state, peer, heads, epoch) = + (state.clone(), peer.to_owned(), heads.to_vec(), remote.epoch); + tokio::spawn(async move { + let Some(peer_state) = state.get_peer_state(&peer).await else { + return; + }; + let replica = state.doc.read().await; + // The same epoch still: a switch meanwhile makes the heads another document's. + if replica.epoch != epoch { + return; + } + peer_state + .lock() + .await + .sync + .heard_heads(&replica.doc, &heads); + }); +} + /// What a peer's status says about it. pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport::Status) { if let Some(peer_state) = state.get_peer_state(peer).await { |