asked known elsewhere

5 files changed, 70 insertions, 9 deletions

diff --git a/src/document/sync.rs b/src/document/sync.rs
index 82b37f5..8c04632 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -134,6 +134,9 @@ pub(crate) struct SyncSession {
 /// 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
@@ -490,7 +493,7 @@ impl SyncSession {
     /// 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_FOR);
+            .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);
@@ -507,13 +510,19 @@ impl SyncSession {
         &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() {
-                if self.known_elsewhere.remove(dep).is_some() {
+                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);
                 }
             }
@@ -599,7 +608,26 @@ impl SyncSession {
     /// 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`.
-    pub(crate) fn resend_if_needed(&mut self, doc: &Document, peer: &str) {
+    ///
+    /// 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!(
@@ -615,6 +643,7 @@ impl SyncSession {
                 "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
@@ -1101,7 +1130,7 @@ mod tests {
             .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");
+        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();
@@ -1156,7 +1185,7 @@ mod tests {
         );
 
         a.receive(&mut ours, decode(&ask).unwrap()).unwrap();
-        a.resend_if_needed(&ours, "b");
+        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.
@@ -1305,7 +1334,7 @@ mod tests {
         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");
+        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());
@@ -1325,7 +1354,7 @@ mod tests {
         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());
+        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");
diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs
index 5242633..ba8995c 100644
--- a/src/simtest/bench.rs
+++ b/src/simtest/bench.rs
@@ -293,6 +293,8 @@ impl Totals {
             relay.carrying_change_bytes += stats.relay.carrying_change_bytes;
             relay.carrying_bloom_bytes += stats.relay.carrying_bloom_bytes;
             relay.new_changes += stats.relay.new_changes;
+            relay.asked_known_lately += stats.relay.asked_known_lately;
+            relay.asked_known_later += stats.relay.asked_known_later;
             relay.nothing_new_change_bytes += stats.relay.nothing_new_change_bytes;
             relay.nothing_new_bloom_bytes += stats.relay.nothing_new_bloom_bytes;
             relay.off_tree += stats.relay.off_tree;
@@ -634,6 +636,15 @@ fn report(
             news
         ),
     );
+    let (lately, later) = (
+        r1.asked_known_lately - r0.asked_known_lately,
+        r1.asked_known_later - r0.asked_known_later,
+    );
+    let _ = writeln!(
+        out,
+        "    asked for, though counted as the asker's (another parent brings them): {:.2} per write ({lately} within 2 s, {later} later)",
+        (lately + later) as f64 / writes,
+    );
     let (i0, i1) = (during.0.incidents, during.1.incidents);
     let _ = writeln!(
         out,
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index b51a802..191f4e9 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -146,6 +146,11 @@ pub(crate) struct Relay {
     /// The changes new to us that sync messages brought (beside what they carried that
     /// we had): with `carrying` less `nothing_new`, how many a message with news brings.
     pub(crate) new_changes: u64,
+    /// Changes peers asked for that we had counted as theirs, as another parent in
+    /// their writer's tree brings them (`SyncSession::known_to_peer`): within 2 s of
+    /// counting them so (so they weren't what we passed on built on), and later.
+    pub(crate) asked_known_lately: u64,
+    pub(crate) asked_known_later: u64,
     /// Changes from elsewhere that came not along a writer's tree (pulled, with an
     /// answer, sent again, fetched as a delta), and their delays summed (as `Propagation`).
     pub(crate) off_tree: u64,
@@ -285,6 +290,15 @@ impl NodeStats {
         inner.relay.carrying_bloom_bytes += sizes.1 as u64;
     }
 
+    /// A peer asked for changes we had counted as its (`SyncSession::resend_if_needed`).
+    pub(crate) fn asked_known_elsewhere(&self, (lately, later): (usize, usize)) {
+        if lately + later > 0 {
+            let mut inner = self.lock();
+            inner.relay.asked_known_lately += lately as u64;
+            inner.relay.asked_known_later += later as u64;
+        }
+    }
+
     /// A sync message brought `count` changes new to us.
     pub(crate) fn new_changes(&self, count: usize) {
         self.lock().relay.new_changes += count as u64;
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index c76c16b..6c67e57 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -497,7 +497,10 @@ pub(crate) async fn handle_sync(
             }
             peer_state.sync.reply_settled();
 
-            peer_state.sync.resend_if_needed(&replica.doc, &peer);
+            let asked = peer_state
+                .sync
+                .resend_if_needed(&replica.doc, &peer, Instant::now());
+            stats.asked_known_elsewhere(asked);
             replica.refresh_heads();
             let heads_after = replica.heads().to_vec();
             if delivered {
diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs
index ffdf80e..d6d4a35 100644
--- a/src/swarm/tree.rs
+++ b/src/swarm/tree.rs
@@ -394,7 +394,11 @@ pub(crate) async fn forward(
             let mut peer_state = peer_state.lock().await;
             peer_state.tree_writer = Some(writer);
             if let Some(replica) = &replica {
-                peer_state.sync.with_predecessors(&replica.doc, changes);
+                peer_state.sync.with_predecessors(
+                    &replica.doc,
+                    changes,
+                    tokio::time::Instant::now(),
+                );
             }
         }
         state.stats.tree_forward();