announce only to those behind

6 files changed, 83 insertions, 15 deletions

diff --git a/docs/peers.md b/docs/peers.md
index 9ef570b..0f7635e 100644
--- a/docs/peers.md
+++ b/docs/peers.md
@@ -299,7 +299,8 @@ has the changes already sends nothing back, even if its heads differ, as they do
 writes go on and newer changes are still on their way to it; nor is a neighbour told whose latest heads, as its
 own announcement or STATUS brought them, are this node's: announcements wait 200 to
 400 ms, at random, so of two neighbours with the same news one is usually first, and
-the other skips it), and a neighbour that lacks those heads pulls them (it sets them as the peer's heads in its sync state, so Automerge asks
+the other skips it; nor one whose latest heads, as it said them in a STATUS or a sync
+message, have all of this node's in their history: it is as far or further), and a neighbour that lacks those heads pulls them (it sets them as the peer's heads in its sync state, so Automerge asks
 for what's missing), if it still lacks them after a round trip to the announcer (at
 most 1 s): while writes go on, the newest are usually on their way to it along a
 writer's tree, and would otherwise come twice. It pulls from one neighbour at a time
diff --git a/src/document/changes.rs b/src/document/changes.rs
index 19d4db6..1091261 100644
--- a/src/document/changes.rs
+++ b/src/document/changes.rs
@@ -8,6 +8,25 @@ use anyhow::{Context, Result, anyhow, bail};
 
 use super::{Change, ChangeHash, Document, ReadDoc};
 
+/// Whether the history of `theirs` (of them, those `doc` has) holds all of `ours`
+/// (heads of `doc`): a peer at `theirs` has every change we have.
+pub(crate) fn covers(doc: &Document, theirs: &[ChangeHash], ours: &[ChangeHash]) -> bool {
+    let known: Vec<ChangeHash> = theirs
+        .iter()
+        .copied()
+        .filter(|head| doc.get_change_by_hash(head).is_some())
+        .collect();
+    if known.is_empty() {
+        return ours.is_empty();
+    }
+    let beyond: std::collections::HashSet<ChangeHash> = doc
+        .get_changes_meta(&known)
+        .into_iter()
+        .map(|change| change.hash)
+        .collect();
+    ours.iter().all(|head| !beyond.contains(head))
+}
+
 /// Those of `heads` that `doc` doesn't have.
 pub(crate) fn missing(doc: &impl ReadDoc, heads: &[ChangeHash]) -> Vec<ChangeHash> {
     heads
@@ -176,6 +195,30 @@ mod tests {
         });
     }
 
+    /// Heads cover ours when their history holds all of ours: as far as ours, or
+    /// further; not when behind or beside; heads we don't know count for nothing.
+    #[test]
+    fn heads_that_cover_ours() {
+        let mut doc = Document::new();
+        change(&mut doc, |doc| doc.put(ROOT, "a", 1_i64).unwrap());
+        let early = doc.get_heads();
+        change(&mut doc, |doc| doc.put(ROOT, "b", 2_i64).unwrap());
+        let ours = doc.get_heads();
+        let mut further = doc.fork();
+        change(&mut further, |doc| doc.put(ROOT, "c", 3_i64).unwrap());
+        doc.merge(&mut further.clone()).unwrap();
+        let unknown = ChangeHash([7; 32]);
+
+        assert!(covers(&doc, &ours, &ours));
+        assert!(covers(&doc, &further.get_heads(), &ours), "further");
+        assert!(!covers(&doc, &early, &ours), "behind");
+        assert!(!covers(&doc, &[unknown], &ours), "unknown");
+        assert!(
+            covers(&doc, &[unknown, ours[0]], &ours),
+            "the known one does"
+        );
+    }
+
     #[test]
     fn no_delta_without_its_heads() {
         let mut doc = Document::new();
diff --git a/src/document/sync.rs b/src/document/sync.rs
index 8c04632..e3ff157 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -28,6 +28,11 @@ impl Received {
         !self.0.changes.is_empty()
     }
 
+    /// The sender's heads, as it says.
+    pub(crate) fn heads(&self) -> &[crate::document::ChangeHash] {
+        &self.0.heads
+    }
+
     /// 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();
diff --git a/src/state/peer.rs b/src/state/peer.rs
index a84e069..231f320 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -150,8 +150,8 @@ pub(crate) struct PeerState {
     /// How many heads its most recent keepalive listed (None if it sent a digest).
     pub(crate) reported_heads: Option<usize>,
     /// The heads its most recent STATUS or STATUS_REPLY carried (an announcement's too),
-    /// with its epoch: once they are ours, it has every change we would announce to it
-    /// (`swarm::relay`).
+    /// or its latest sync message, with its epoch: what it said it has. Once those cover
+    /// ours, it has every change we would announce to it (`swarm::relay`).
     pub(crate) last_heads: Option<(transport::Epoch, transport::HeadsSummary)>,
     /// How long its sync messages took to come (`swarm::hop`).
     pub(crate) hop_delays: crate::swarm::hop::HopDelays,
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index e049846..31df0d3 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -251,8 +251,11 @@ async fn announce_now(state: &AppState, pending: &Pending, slow: bool) {
 /// The neighbours to tell of the changes `senders` sent us. The peer whose changes we
 /// announce has them already, so with a single sender it is skipped; with several, each
 /// lacks the others' changes. So is a neighbour whose latest heads (its STATUS, an
-/// announcement of its own, or its STATUS_REPLY) are ours now: it has all we have; and
-/// one `sent` the changes along a writer's tree, which has them already.
+/// announcement of its own, its STATUS_REPLY, or its latest sync message) are ours, or
+/// have all of ours in their history: it has all we have; and one `sent` the changes
+/// along a writer's tree, which has them already. Only what the neighbour said counts,
+/// not what we sent it: a message along a tree may be lost, which is what the slow
+/// round is for.
 async fn unreached(
     state: &AppState,
     senders: &HashSet<String>,
@@ -262,11 +265,13 @@ async fn unreached(
         1 => senders.iter().next().cloned(),
         _ => None,
     };
-    let ours = {
+    let (epoch, heads) = {
         let stamp = state.stamp.borrow();
-        (stamp.epoch, HeadsSummary::of(stamp.heads.to_vec()))
+        (stamp.epoch, stamp.heads.to_vec())
     };
-    let mut unreached = Vec::new();
+    let ours = HeadsSummary::of(heads.clone());
+    // What each neighbour said it has.
+    let mut said = Vec::new();
     for (peer, peer_state) in state.peers_snapshot().await {
         if skip.as_deref() == Some(peer.as_str())
             || sent.contains(&peer)
@@ -274,17 +279,28 @@ async fn unreached(
         {
             continue;
         }
-        let has_ours = peer_state
+        let last = peer_state
             .lock()
             .await
             .last_heads
-            .as_ref()
-            .is_some_and(|(epoch, heads)| *epoch == ours.0 && *heads == ours.1);
-        if !has_ours {
-            unreached.push(peer);
-        }
+            .clone()
+            .filter(|(theirs, _)| *theirs == epoch)
+            .map(|(_, heads)| heads);
+        said.push((peer, last));
+    }
+    let replica = state.doc.read().await;
+    if replica.epoch != epoch {
+        return said.into_iter().map(|(peer, _)| peer).collect();
     }
-    unreached
+    said.into_iter()
+        .filter(|(_, last)| {
+            !last.as_ref().is_some_and(|theirs| {
+                *theirs == ours
+                    || crate::document::changes::covers(&replica.doc, theirs.listed(), &heads)
+            })
+        })
+        .map(|(peer, _)| peer)
+        .collect()
 }
 
 /// Pulls in progress: one neighbour at a time, so each change arrives about once. The
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index ed4c4da..8660742 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -484,6 +484,9 @@ pub(crate) async fn handle_sync(
 
             let mut peer_state = peer_state_mutex.blocking_lock();
             peer_state.reports_ready = Some(from_ready);
+            // What it says it has (`swarm::relay::unreached`).
+            peer_state.last_heads =
+                Some((epoch, transport::HeadsSummary::of(message.heads().to_vec())));
             let heads_before = replica.heads().to_vec();
 
             let sync = &mut peer_state.sync;