access: refused once

A refused change is remembered by its hash: offered again it is refused
without decoding, judging or logging, so is one built on it, and a peer
offering nothing else is answered later each time, up to a minute.

Auth plan M6 (re-sending).

1 files changed, 28 insertions, 5 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index ff7fa5a..f5d6bfb 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -554,13 +554,15 @@ pub(crate) async fn handle_sync(
         let ahead_ms = state.clock_ahead_ms.load(Ordering::Relaxed);
 
         // Some((caught_up, heads changed, carried changes, the new changes' authors with
-        // theirs, and whether any has none)) when applied, None when the message belongs
-        // to another epoch.
+        // theirs, and whether any has none, how long to wait before answering if it
+        // brought only changes refused before)) when applied, None when the message
+        // belongs to another epoch.
         type Applied = (
             bool,
             bool,
             bool,
             (Vec<(String, Vec<crate::document::ChangeHash>)>, bool),
+            Option<std::time::Duration>,
         );
         tokio::task::spawn_blocking(move || -> Result<Option<Applied>> {
             // Before the lock: inflating and decoding a large message takes a while. No
@@ -607,6 +609,7 @@ pub(crate) async fn handle_sync(
             let sync = &mut peer_state.sync;
             let epoch_seq = replica.epoch_seq;
             let mut message = message;
+            let mut refused_again = false;
             replica
                 .logged_signed(|doc, known, log| {
                     let result = (|| {
@@ -619,7 +622,7 @@ pub(crate) async fn handle_sync(
                             if required {
                                 known.offer(doc, signatures.clone());
                             }
-                            crate::access::screen::apply_bytes(
+                            refused_again = crate::access::screen::apply_bytes(
                                 doc,
                                 epoch,
                                 epoch_seq,
@@ -627,7 +630,8 @@ pub(crate) async fn handle_sync(
                                 &screening,
                                 required.then_some(&mut *known),
                                 log,
-                            )?;
+                            )?
+                            .refused_again;
                         }
                         sync.receive_logged(doc, message, log)
                     })();
@@ -635,6 +639,16 @@ pub(crate) async fn handle_sync(
                 })
                 .context("Automerge sync failed")?;
             peer_state.last_sync_rx = Some(Instant::now());
+            // It offers again only what we refused, and would as fast as we answer: we
+            // answer later, later each time (`SyncSession::refused_again`).
+            let back_off = if refused_again {
+                Some(peer_state.sync.refused_again(Instant::now()))
+            } else {
+                if carried {
+                    peer_state.sync.refusals_settled();
+                }
+                None
+            };
             if let Some(wait) = &peer_state.sync.reply {
                 throughput
                     .lock_or_recover()
@@ -721,16 +735,25 @@ pub(crate) async fn handle_sync(
                 changed,
                 carried,
                 authors,
+                back_off,
             )))
         })
         .await
         .context("Sync task panicked")??
     };
 
-    let Some((caught_up, changed, carried, (authors, unauthored))) = applied else {
+    let Some((caught_up, changed, carried, (authors, unauthored), back_off)) = applied else {
         debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch");
         return Ok(());
     };
+    if let Some(delay) = back_off {
+        debug!(
+            peer = %peer,
+            wait_ms = delay.as_millis(),
+            "Offered only changes refused before; answering later"
+        );
+        state.request_sync_after(peer, delay);
+    }
 
     source::reported(state, peer, from_ready, None);