tree messages need no answer

5 files changed, 235 insertions, 7 deletions

diff --git a/src/document/sync.rs b/src/document/sync.rs
index 4a3fe5e..c432173 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -46,6 +46,12 @@ pub(crate) struct Sending {
     /// 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`): the peer's heads and the shared heads as
+    /// they were before, put back if the send fails.
+    pub(crate) assumed: Option<(
+        Option<Vec<crate::document::ChangeHash>>,
+        Vec<crate::document::ChangeHash>,
+    )>,
 }
 
 /// Session IDs, unique within the process.
@@ -165,6 +171,7 @@ impl SyncSession {
             self.sending = Some(Sending {
                 sent,
                 regenerate: false,
+                assumed: None,
             });
         }
         payload
@@ -227,6 +234,50 @@ impl SyncSession {
         })
     }
 
+    /// The message just generated passes a change on along a writer's tree, 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_along_tree`). So it is taken as delivered: the peer's heads are ours,
+    /// and no answer is awaited (about half the sync messages, otherwise, are those
+    /// answers). Should it be lost, the announcements make the peer pull: the slow round
+    /// of its other neighbours, or ours where it reaches no other member
+    /// (`swarm::tree::Plan::alone`); 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 Some(sending) = &mut self.sending else {
+            return;
+        };
+        sending.assumed = Some((
+            self.state.their_heads.clone(),
+            self.state.shared_heads.clone(),
+        ));
+        self.state.their_heads = Some(heads.to_vec());
+        self.state.shared_heads = heads.to_vec();
+        self.state.in_flight = false;
+        self.reply = None;
+    }
+
+    /// We applied a message passing a change on along a writer's tree: if we hold
+    /// exactly its sender's heads now, its sender took it as delivered
+    /// (`assume_delivered`), and Automerge's answer, our heads, would tell it nothing.
+    /// So none goes: we count as having said our `heads`. Returns whether so.
+    pub(crate) fn answered_along_tree(&mut self, heads: &[crate::document::ChangeHash]) -> bool {
+        let mut ours = heads.to_vec();
+        ours.sort_unstable();
+        let theirs = self.state.their_heads.as_ref().map(|theirs| {
+            let mut theirs = theirs.clone();
+            theirs.sort_unstable();
+            theirs
+        });
+        if theirs.as_ref() != Some(&ours) {
+            return false;
+        }
+        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<crate::document::ChangeHash>) {
@@ -318,6 +369,10 @@ impl SyncSession {
         for hash in &sending.sent {
             self.state.sent_hashes.remove(hash);
         }
+        if let Some((their_heads, shared_heads)) = sending.assumed {
+            self.state.their_heads = their_heads;
+            self.state.shared_heads = shared_heads;
+        }
         if let Some(wait) = &mut self.reply {
             wait.sent.retain(|hash| !sending.sent.contains(hash));
         }
@@ -471,6 +526,7 @@ mod tests {
         Some(Sending {
             sent: Vec::new(),
             regenerate: false,
+            assumed: None,
         })
     }
 
@@ -540,6 +596,144 @@ mod tests {
         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 along a tree to a peer that had everything else: taken as delivered,
+    /// and not answered; neither side has more to say.
+    #[test]
+    fn tree_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_along_tree(&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 tree_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_along_tree(&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 along a tree 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_tree_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_along_tree(&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("b");
+        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());
+    }
+
+    /// 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_tree_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`).
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index f141dbb..98d4ca0 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -126,8 +126,15 @@ pub(crate) async fn announce(state: &AppState, from: &str, writers: &[u64], unau
         let slow = slow.get_or_insert_with(Pending::default);
         slow.senders.insert(from.to_owned());
         slow.all = true;
+        // Our children were sent the changes, but those that reach no member but us
+        // hear of a lost message from nobody else (`Plan::alone`).
         for plan in &plans {
-            slow.sent.extend(plan.children.iter().cloned());
+            slow.sent.extend(
+                plan.children
+                    .iter()
+                    .filter(|child| !plan.alone.contains(child))
+                    .cloned(),
+            );
         }
         schedule
     };
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index b20f7ef..02c6379 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -148,6 +148,13 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                     let replica = doc.blocking_read();
                     let mut peer_state = psm.blocking_lock();
                     let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
+                    // Whose change it passes on, if any: this message carries it.
+                    let writer = peer_state.tree_writer.take();
+                    // Along a tree, to a peer that had nothing we lack, with nothing
+                    // awaited: taken as delivered (`SyncSession::assume_delivered`).
+                    let level = writer.is_some()
+                        && peer_state.sync.reply.is_none()
+                        && peer_state.sync.have_all_theirs(&replica.doc);
                     // Under the same locks as the epoch label: the message belongs to
                     // this document.
                     let message =
@@ -171,8 +178,9 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                         );
                         transport::SyncBody::Delta(spec)
                     });
-                    // Whose change it passes on, if any: this message carries it.
-                    let writer = peer_state.tree_writer.take();
+                    if level && matches!(body, Some(transport::SyncBody::Message(_))) {
+                        peer_state.sync.assume_delivered(replica.heads());
+                    }
                     (peer_state.sync.id(), epoch, epoch_seq, body, writer)
                 };
                 let Some(body) = body else {
@@ -411,6 +419,11 @@ pub(crate) async fn handle_sync(
             peer_state.sync.resend_if_needed(&peer);
             replica.refresh_heads();
             let heads_after = replica.heads().to_vec();
+            if writer.is_some() {
+                // Level with the sender now: it takes the message as delivered, so
+                // there is nothing to answer (`SyncSession::answered_along_tree`).
+                peer_state.sync.answered_along_tree(&heads_after);
+            }
             stats.heads_seen(heads_after.len());
             let changed = heads_after != heads_before;
             // Whose they are, to pass them on along their writers' trees, however they
diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs
index b21908e..c474754 100644
--- a/src/swarm/tree.rs
+++ b/src/swarm/tree.rs
@@ -16,9 +16,10 @@
 //! pair key here): whoever applies changes passes them on to its own children in each of
 //! their writers' trees (`forward`), however they came; so does the writer, to all its
 //! neighbours, with its own writes. A sync message along a tree also says whose
-//! (`SyncEnvelope::writer`), for one that had them already (`swarm::sync`). A member no
-//! member it reaches comes before (or whose parent didn't get the change) gets it from
-//! the announcements (`swarm::relay`), as before.
+//! (`SyncEnvelope::writer`), for one that had them already (`swarm::sync`), and needs no
+//! answer where the child had nothing the parent lacks (`SyncSession::assume_delivered`).
+//! A member no member it reaches comes before (or whose parent didn't get the change)
+//! gets it from the announcements (`swarm::relay`), as before.
 //!
 //! The graph is computed once per member list (tens of milliseconds for thousands of
 //! members), the distances once per writer and graph (`MAX_WRITERS` kept).
@@ -176,6 +177,10 @@ pub(crate) fn whole(
 /// out (`uncovered`).
 pub(crate) struct Plan {
     pub(crate) children: Vec<String>,
+    /// Those of our children that reach no member but us, as they said: nobody else's
+    /// slow round tells them of a change, should our message to them be lost (it is
+    /// taken as delivered, `SyncSession::assume_delivered`), so ours does.
+    pub(crate) alone: Vec<String>,
     /// Our neighbours that no member they reach comes before (the tree reaches them only
     /// by announcement), and those not in the member list's graph (whom no tree has).
     pub(crate) uncovered: Vec<String>,
@@ -205,7 +210,9 @@ pub(crate) async fn plan(state: &AppState, writer: u64) -> Option<Plan> {
             .is_none_or(|keys| keys.contains(&topology::pair_key(member)))
     };
     let my_rank = rank(&graph, &distance, &reached, me)?;
+    let my_key = topology::pair_key(me);
     let mut children = Vec::new();
+    let mut alone = Vec::new();
     let mut uncovered = Vec::new();
     for member in graph.get(me).into_iter().flatten() {
         let Some(its_rank) = rank(&graph, &distance, &reached, member) else {
@@ -219,6 +226,12 @@ pub(crate) async fn plan(state: &AppState, writer: u64) -> Option<Plan> {
             Some(parent)
                 if parent == me && its_rank > my_rank && topology::reaches(state, member).await =>
             {
+                if reaches
+                    .get(member.as_str())
+                    .is_some_and(|keys| keys.iter().all(|key| *key == my_key))
+                {
+                    alone.push(member.clone());
+                }
                 children.push(member.clone());
             }
             Some(_) => {}
@@ -235,6 +248,7 @@ pub(crate) async fn plan(state: &AppState, writer: u64) -> Option<Plan> {
     }
     Some(Plan {
         children,
+        alone,
         uncovered,
     })
 }
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 67f6de0..899454c 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{EncodedMessage, 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 = 23;
+pub(crate) const VERSION: u8 = 24;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {