3 files changed, 82 insertions, 0 deletions

diff --git a/docs/peers.md b/docs/peers.md
index 4775e85..fc4b6f1 100644
--- a/docs/peers.md
+++ b/docs/peers.md
@@ -172,6 +172,9 @@ other neighbours offer them too, and they would come once per offer. Offers that
 meanwhile wait; once the delta is in, the node syncs with those that offered, which then
 send only what is still missing. What a delta brought counts as known to its sender, as
 what its sync messages bring does: none of it goes back, by message or as a delta.
+An offer made before its sender heard what the node has now (one that came behind the
+chunks of the delta before it, say) would bring part of that again: such an offer isn't
+fetched; the node's next message tells the sender, which offers afresh if still needed.
 
 Towards a peer on a slow link the limit is lower: a message is offered as a delta if it
 would take more than 5 s at the throughput measured to that peer (`syncThroughput`,
diff --git a/src/document/sync.rs b/src/document/sync.rs
index 2921bb9..e93fa17 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -366,6 +366,40 @@ impl SyncSession {
         }
     }
 
+    /// 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<crate::document::ChangeHash> = 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<crate::document::ChangeHash> =
+            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()
+    }
+
     /// What the sync state knows of the peer (`PeerView`).
     fn peer_view(&self) -> PeerView {
         PeerView {
@@ -1037,6 +1071,32 @@ mod tests {
         }
     }
 
+    /// 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);
+    }
+
     /// Our transport refused a message taken as delivered: the peer's heads are as
     /// before, so the next message carries the change again.
     #[test]
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 8059fe2..aa6a8f9 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -620,6 +620,25 @@ async fn download_and_apply_delta(
             .into_iter()
             .filter(|other| other != peer),
     );
+    if let Some(peer_state) = state.get_peer_state(peer).await {
+        let doc = Arc::clone(&state.doc);
+        let from = spec.from.clone();
+        let known = tokio::task::spawn_blocking(move || {
+            let replica = doc.blocking_read();
+            peer_state
+                .blocking_lock()
+                .sync
+                .known_beyond(&replica.doc, &from)
+        })
+        .await
+        .context("Delta check task panicked")?;
+        if known > 0 {
+            // Offered before it heard what we have: it would bring those again. Our
+            // next message tells it (`fetch_delta`); it offers afresh if still needed.
+            info!(peer = %peer, known, "Offered changes partly here already; not fetched");
+            return Ok(false);
+        }
+    }
     info!(peer = %peer, to = spec.to.len(), "Fetching changes offered as a blob");
     let started = Instant::now();
     let bytes = crate::swarm::blob::fetch_delta(state, &spec, peers).await?;