pull after grace

Before pulling, a node waits one measured round trip to the announcer
(1 s at most) and pulls only what is still missing.  If the round trip
isn't known yet, it pulls at once, as now.  On the clean benchmark links
the round trip is a few milliseconds, so those scenarios shouldn't
change.

2 files changed, 33 insertions, 4 deletions

diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index f4e961f..0f34eb8 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -239,7 +239,7 @@ pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport
         {
             let (state, peer, remote) = (state.clone(), peer.to_owned(), remote.clone());
             tokio::spawn(async move {
-                crate::swarm::relay::pull_if_behind(&state, &peer, &remote).await;
+                crate::swarm::relay::pull_after_grace(&state, &peer, &remote).await;
             });
         }
         if became_ready {
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 99c1381..95abab8 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -6,7 +6,9 @@
 //! neighbour, a node *announces*
 //! its new heads to its other neighbours with a STATUS (heads, about 100 bytes, not the
 //! changes; no keepalive, and answered only by a neighbour that lacks some of them). A
-//! neighbour that sees heads it lacks *pulls*: it enters the
+//! neighbour that sees heads it lacks *pulls* (those it still lacks after a round trip
+//! to the announcer: while writes go on, the newest are usually on their way to it along
+//! a tree, `pull_after_grace`): it enters the
 //! announced heads as the peer's heads in its sync state, Automerge asks for what's
 //! missing (`need`), and the announcer answers with exactly those changes. Regular
 //! keepalives work the same way, so any gap is noticed within a keepalive interval.
@@ -51,6 +53,11 @@ use crate::transport::{self, HeadsSummary};
 /// A pull that brought nothing for this long no longer blocks pulling from another peer.
 const PULL_TIMEOUT: Duration = Duration::from_secs(15);
 
+/// The longest a STATUS's heads we lack wait before we pull them (`pull_after_grace`: a
+/// round trip to its sender, at most this): over a slow transport (Tor), a change that
+/// was missed waits no longer.
+const PULL_GRACE_MAX: Duration = Duration::from_secs(1);
+
 /// How long an announcement waits for further changes to ride along, and a random part
 /// on top (`ANNOUNCE_JITTER`), so that of two neighbours with the same news one usually
 /// announces first and the other, seeing the first's heads are its own, skips it
@@ -224,8 +231,30 @@ pub(crate) async fn lacks(state: &AppState, heads: &HeadsSummary) -> bool {
     !crate::document::changes::missing(&replica.doc, heads.listed()).is_empty()
 }
 
-/// A keepalive from `peer`: if it has heads we lack, pull them (one peer at a time; a
-/// busy slot queues the peer). Returns whether a pull started.
+/// A STATUS from `peer` (a keepalive's, or an announcement): if it has heads we lack,
+/// pull them, but only those we still lack after a round trip to it (`PULL_GRACE_MAX` at
+/// most; at once if we know none). While writes go on, the newest of them are usually
+/// on their way to us along their writers' trees, as fast as the STATUS: they come
+/// meanwhile, and pulling them would only fetch them twice. What is still missing then
+/// was missed.
+pub(crate) async fn pull_after_grace(state: &AppState, peer: &str, remote: &transport::Status) {
+    if HeadsSummary::of(state.stamp.borrow().heads.to_vec()) == remote.heads {
+        return;
+    }
+    let grace = state
+        .lock_rtt()
+        .effective(peer, Instant::now(), state.timing.rtt_fresh_for())
+        .map(|rtt| rtt.min(PULL_GRACE_MAX));
+    if let Some(grace) = grace
+        && lacks(state, &remote.heads).await
+    {
+        tokio::time::sleep(grace).await;
+    }
+    pull_if_behind(state, peer, remote).await;
+}
+
+/// If `peer` has heads we lack (as its STATUS `remote` says), pull them (one peer at a
+/// time; a busy slot queues the peer). Returns whether a pull started.
 pub(crate) async fn pull_if_behind(
     state: &AppState,
     peer: &str,