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, 147 insertions, 10 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(),
                 )