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).

4 files changed, 199 insertions, 16 deletions

diff --git a/src/access/screen.rs b/src/access/screen.rs
index 736e04a..a5a8c8a 100644
--- a/src/access/screen.rs
+++ b/src/access/screen.rs
@@ -8,6 +8,11 @@
 //! the policy, bans) is what its writer saw, unless one of those ancestors changed it. So a change that
 //! does (`relevant`) is applied before the next is judged: the rest go in at once.
 //!
+//! A change refused is remembered by its hash (`Refusals`): when it comes again, as its
+//! writer keeps offering what the node never takes, it is refused again without being
+//! decoded or logged, and the sync session with whoever sent only such changes backs off
+//! (`Outcome::refused_again`, `SyncSession::refused_again`).
+//!
 //! A change whose dependencies are neither in the document nor among those received
 //! waits (`Waiting`) until they come, a bounded number; Automerge never sees it before.
 //! With signatures required (`access::policy`), so does one no signature by its
@@ -29,6 +34,9 @@ use crate::document::{Change, ChangeHash, Document, ReadDoc};
 const MAX_WAITING: usize = 10_000;
 /// Refused changes kept for operators, at most: the oldest go.
 pub(crate) const MAX_REFUSED: usize = 1_000;
+/// Hashes of refused changes remembered, at most: the oldest go (and are judged again,
+/// and refused again, should they come).
+const MAX_REFUSALS: usize = 100_000;
 
 /// A refused change, and why.
 #[derive(Debug, Clone)]
@@ -49,6 +57,31 @@ pub(crate) struct Screening {
     epoch_seq: u32,
     pub(crate) waiting: Waiting,
     refused: std::collections::VecDeque<Refused>,
+    refusals: Refusals,
+}
+
+/// The hashes of the changes refused, to refuse them again by hash.
+#[derive(Debug, Default)]
+pub(crate) struct Refusals {
+    known: HashSet<ChangeHash>,
+    order: std::collections::VecDeque<ChangeHash>,
+}
+
+impl Refusals {
+    fn contains(&self, hash: &ChangeHash) -> bool {
+        self.known.contains(hash)
+    }
+
+    fn insert(&mut self, hash: ChangeHash) {
+        if self.known.insert(hash) {
+            self.order.push_back(hash);
+            if self.order.len() > MAX_REFUSALS
+                && let Some(oldest) = self.order.pop_front()
+            {
+                self.known.remove(&oldest);
+            }
+        }
+    }
 }
 
 impl Screening {
@@ -101,29 +134,63 @@ impl Waiting {
 pub(crate) struct Screened {
     pub(crate) applied: usize,
     pub(crate) refused: Vec<Refused>,
+    /// Received changes refused before (`Refusals`), refused again by their hash.
+    pub(crate) refused_again: usize,
 }
 
 /// Judges `received` (and those `waiting` before) and applies to `doc` (of `epoch`)
 /// those rules allow, what they change recorded in `log`. Those already in `doc` are
-/// left out; those whose dependencies are missing wait.
-/// With `signatures` (required), those no signature vouches for wait too.
+/// left out; those whose dependencies are missing wait; those `refusals` holds are
+/// refused again by their hash, and those building on them, and those refused now go
+/// into it. With `signatures` (required), those no signature vouches for wait too.
 pub(crate) fn apply(
     doc: &mut Document,
     epoch: Epoch,
     received: Vec<Change>,
     waiting: &mut Waiting,
+    refusals: &mut Refusals,
     mut signatures: Option<&mut Signatures>,
     log: &mut crate::document::PatchLog,
 ) -> Result<Screened> {
     let mut screened = Screened::default();
-    // Each once, not in the document yet.
+    // Each once, not in the document yet, not refused before.
     let mut seen = HashSet::new();
     let candidates: Vec<Change> = std::mem::take(&mut waiting.0)
         .into_iter()
         .chain(received)
         .filter(|change| seen.insert(change.hash()))
         .filter(|change| doc.get_change_by_hash(&change.hash()).is_none())
+        .filter(|change| {
+            let again = refusals.contains(&change.hash());
+            screened.refused_again += usize::from(again);
+            !again
+        })
         .collect();
+    // Building on one refused before: refused too, without waiting for it (it never
+    // comes) or judging it.
+    let (building, candidates): (Vec<Change>, Vec<Change>) = candidates
+        .into_iter()
+        .partition(|change| change.deps().iter().any(|dep| refusals.contains(dep)));
+    for change in building {
+        let dep = change
+            .deps()
+            .iter()
+            .find(|dep| refusals.contains(dep))
+            .copied()
+            .expect("partitioned so");
+        refusals.insert(change.hash());
+        screened.refused.push(Refused {
+            committer: change.author().and_then(|author| {
+                std::str::from_utf8(author.as_bytes())
+                    .ok()
+                    .map(str::to_owned)
+            }),
+            user: change.message().and_then(user_of).map(str::to_owned),
+            paths: BTreeSet::new(),
+            why: format!("it builds on refused change {dep}"),
+            change,
+        });
+    }
     let (ready, unready) = in_order(doc, candidates);
     waiting.0 = unready;
     if ready.is_empty() {
@@ -180,7 +247,11 @@ pub(crate) fn apply(
             paths: paths[index].clone(),
             why,
         };
-        if let Some(dep) = change.deps().iter().find(|dep| refused.contains(dep)) {
+        if let Some(dep) = change
+            .deps()
+            .iter()
+            .find(|dep| refused.contains(dep) || refusals.contains(dep))
+        {
             refused.insert(hash);
             screened
                 .refused
@@ -250,10 +321,23 @@ pub(crate) fn apply(
     if !accepted.is_empty() {
         doc.apply_changes_log_patches(accepted, log)?;
     }
+    for hash in refused {
+        refusals.insert(hash);
+    }
     waiting.0.truncate(MAX_WAITING);
     Ok(screened)
 }
 
+/// What `apply_bytes` did.
+#[derive(Debug, Default, Clone, Copy)]
+pub(crate) struct Outcome {
+    /// Changes applied (received, or waiting before).
+    pub(crate) applied: usize,
+    /// Whether the bytes carried changes, and all of them refused before: their sender
+    /// offers again what the node never takes, and its sync session backs off.
+    pub(crate) refused_again: bool,
+}
+
 /// [`apply`] of the changes in `bytes`, as sync messages and downloads carry them, with
 /// what `screening` keeps for document `epoch_seq` (of `epoch`). Bytes that don't load
 /// as changes, or with a misplaced author (`changes::misplaced_author`), are refused
@@ -268,10 +352,10 @@ pub(crate) fn apply_bytes(
     screening: &std::sync::Mutex<Screening>,
     signatures: Option<&mut Signatures>,
     log: &mut crate::document::PatchLog,
-) -> Result<usize> {
+) -> Result<Outcome> {
     use crate::locks::MutexExt;
     if bytes.is_empty() && signatures.is_none() {
-        return Ok(0);
+        return Ok(Outcome::default());
     }
     if let Some(misplaced) = crate::document::changes::misplaced_author(bytes) {
         anyhow::bail!("Refused changes: {misplaced}");
@@ -281,12 +365,24 @@ pub(crate) fn apply_bytes(
     let mut screening = screening.lock_or_recover();
     let screening = screening.of(epoch_seq);
     if changes.is_empty() && screening.waiting.len() == 0 {
-        return Ok(0);
+        return Ok(Outcome::default());
     }
-    let screened = apply(doc, epoch, changes, &mut screening.waiting, signatures, log)?;
-    let applied = screened.applied;
+    let received = changes.len();
+    let screened = apply(
+        doc,
+        epoch,
+        changes,
+        &mut screening.waiting,
+        &mut screening.refusals,
+        signatures,
+        log,
+    )?;
+    let outcome = Outcome {
+        applied: screened.applied,
+        refused_again: received > 0 && screened.refused_again == received,
+    };
     screening.record(screened.refused);
-    Ok(applied)
+    Ok(outcome)
 }
 
 /// Whether a change touching `paths` changes what rules read: the members, the user
@@ -382,6 +478,7 @@ mod tests {
             Epoch::GENESIS,
             changes,
             waiting,
+            &mut Refusals::default(),
             None,
             &mut crate::document::PatchLog::inactive(),
         )
@@ -440,6 +537,45 @@ mod tests {
         assert!(waiting.len() == 0);
     }
 
+    /// A refused change offered again: refused by its hash, not judged or kept again,
+    /// and so is a new one building on it.
+    #[test]
+    fn refused_again_by_hash() {
+        let mut theirs = genesis();
+        write(&mut theirs, "a", Some("alice"), "config/x", json!(1));
+        let mut ours = Document::new();
+        let mut waiting = Waiting::default();
+        let mut refusals = Refusals::default();
+        let mut screen_with = |doc: &mut Document, changes: Vec<Change>| {
+            apply(
+                doc,
+                Epoch::GENESIS,
+                changes,
+                &mut waiting,
+                &mut refusals,
+                None,
+                &mut crate::document::PatchLog::inactive(),
+            )
+            .unwrap()
+        };
+        let first = screen_with(&mut ours, received(&theirs, &[]));
+        assert_eq!(
+            (first.applied, first.refused.len(), first.refused_again),
+            (2, 1, 0)
+        );
+        let heads = ours.get_heads();
+        let again = screen_with(&mut ours, received(&theirs, &heads));
+        assert_eq!(
+            (again.applied, again.refused.len(), again.refused_again),
+            (0, 0, 1)
+        );
+        let before = theirs.get_heads();
+        write(&mut theirs, "a", Some("alice"), "notes/x", json!(2));
+        let building = screen_with(&mut ours, received(&theirs, &before));
+        assert_eq!((building.applied, building.refused.len()), (0, 1));
+        assert!(building.refused[0].why.contains("builds on refused change"));
+    }
+
     /// Judged as its writer saw the registry: a grant earlier in the same batch counts.
     #[test]
     fn a_grant_in_the_same_batch_counts() {
@@ -538,6 +674,7 @@ mod tests {
                     Epoch::GENESIS,
                     changes,
                     waiting,
+                    &mut Refusals::default(),
                     Some(signatures),
                     &mut crate::document::PatchLog::inactive(),
                 )
diff --git a/src/document/sync.rs b/src/document/sync.rs
index f124ee8..823d4a3 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -153,6 +153,10 @@ static NEXT_SESSION: AtomicU64 = AtomicU64::new(1);
 /// while it keeps refusing.
 const SEND_RETRY: Duration = Duration::from_millis(500);
 const SEND_RETRY_MAX: Duration = Duration::from_secs(30);
+/// The first wait before answering a peer offering only changes we refused, and the
+/// longest (`SyncSession::refused_again`).
+const REFUSED_WAIT: Duration = Duration::from_secs(1);
+const REFUSED_WAIT_MAX: Duration = Duration::from_secs(60);
 
 /// One Automerge sync with a peer, from a reset to the next. Everything a reset must
 /// forget lives here, so a reset replaces it whole and can't miss a field.
@@ -205,6 +209,10 @@ pub(crate) struct SyncSession {
     /// comes late doesn't count: `older_keeps_newer`).
     said_heads: Option<Vec<crate::document::ChangeHash>>,
 
+    /// The wait before answering a peer that offers again only changes we refused
+    /// (`refused_again`).
+    refused_backoff: Backoff,
+
     /// The answer we expect to our last sync message(s), if any.
     pub(crate) reply: Option<SyncReplyWait>,
     /// How long to wait for it; doubles while answers stay out.
@@ -255,6 +263,7 @@ impl SyncSession {
             known_elsewhere: std::collections::HashMap::new(),
             assumed_lately: std::collections::VecDeque::new(),
             said_heads: None,
+            refused_backoff: Backoff::new(REFUSED_WAIT, REFUSED_WAIT_MAX),
             reply: None,
             reply_backoff,
         }
@@ -975,6 +984,20 @@ impl SyncSession {
         self.id
     }
 
+    /// The peer offered again only changes we refused (`access::screen`): it would as
+    /// often as we answer, its sync state never learning we won't take them. No message
+    /// to it before the returned wait, which doubles while it goes on.
+    pub(crate) fn refused_again(&mut self, now: Instant) -> Duration {
+        let delay = self.refused_backoff.next();
+        self.retry_at = Some(self.retry_at.map_or(now + delay, |at| at.max(now + delay)));
+        delay
+    }
+
+    /// The peer brought changes we took (or none refused before): answered at once again.
+    pub(crate) fn refusals_settled(&mut self) {
+        self.refused_backoff.reset();
+    }
+
     /// Our transport took the message generated from session `session`. Returns whether
     /// to generate again: a pass was requested while it was handed over, or a reset
     /// replaced the session meanwhile (any pass requested then was lost with the old
diff --git a/src/swarm/signatures.rs b/src/swarm/signatures.rs
index 309ccc2..9ea404e 100644
--- a/src/swarm/signatures.rs
+++ b/src/swarm/signatures.rs
@@ -96,7 +96,7 @@ pub(crate) async fn received(state: &AppState, message: SignaturesMessage) {
                 });
                 replica.refresh_heads();
                 return match applied {
-                    Ok(applied) => applied,
+                    Ok(outcome) => outcome.applied,
                     Err(error) => {
                         tracing::warn!(error = %format!("{error:#}"), "Couldn't judge the changes waiting for signatures");
                         0
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);