asked known elsewhere

1 files changed, 36 insertions, 7 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");