sync: a member refused as a whole, synced with again once it changed document

A node refusing a member as a whole sent it nothing, and lifted that
only when a change of the member's was taken again. But a member that
dropped what was refused (REFUSALS) starts its sync sessions over, and
a fresh session sends no changes until the other side answers: the two
waited for each other, and the member stayed refused.

The refusal now notes the member's run and epoch_seq when it began; a
member with another document since (it dropped what was refused, or
rejoined) is synced with again.

3 files changed, 34 insertions, 3 deletions

diff --git a/src/access/screen.rs b/src/access/screen.rs
index 44d96ec..be4bb9c 100644
--- a/src/access/screen.rs
+++ b/src/access/screen.rs
@@ -145,6 +145,14 @@ impl Screening {
         self.refusing.get(node)
     }
 
+    /// `node` refused as a whole no more: it has another document since, without what
+    /// was refused, or it would build on it (`swarm::sync`).
+    pub(crate) fn forgive(&mut self, node: &str) {
+        if self.refusing.remove(node).is_some() {
+            tracing::info!(node = %node, "A member refused as a whole has another document since: refused no longer");
+        }
+    }
+
     /// The members refused as a whole and taken again of `screened`, logged.
     fn judge_members(&mut self, screened: &Screened) {
         for node in &screened.taken_from {
diff --git a/src/state/peer.rs b/src/state/peer.rs
index e0b161d..f6e4a62 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -141,6 +141,10 @@ pub(crate) struct PeerState {
     pub(crate) epoch_seq: Option<u32>,
     /// The peer's run, from its HELLO or HELLO_REPLY: a new one starts `epoch_seq` over.
     pub(crate) run_id: Option<crate::transport::RunId>,
+    /// While we refuse the peer as a whole (`access::screen`): its run and `epoch_seq`
+    /// when we began to. Another means it has another document since (it dropped what
+    /// we refused, `swarm::refusals`, or rejoined): we sync with it again.
+    pub(crate) refused_with: Option<(Option<crate::transport::RunId>, Option<u32>)>,
     /// The newest handshake the peer started with us: its run and number
     /// (`HelloRequest::handshake`). A HELLO repeating it is a retry or a late copy.
     pub(crate) last_hello: Option<(crate::transport::RunId, u32)>,
@@ -319,6 +323,7 @@ impl PeerState {
             epoch: None,
             epoch_seq: None,
             run_id: None,
+            refused_with: None,
             last_hello: None,
             handshake: 0,
             reports_ready: None,
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 7601d69..ef2c1b3 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -38,10 +38,28 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
     };
     {
         use crate::locks::MutexExt;
-        if state.screening.lock_or_recover().refusing(peer).is_some() {
+        let refusing = state.screening.lock_or_recover().refusing(peer).is_some();
+        let mut ps = peer_state_mutex.lock().await;
+        let now_with = (ps.run_id, ps.epoch_seq);
+        match (refusing, ps.refused_with) {
+            (false, _) => ps.refused_with = None,
             // All it commits builds on what we refused (`access::screen`).
-            trace!("Refusing the peer as a whole; not syncing with it");
-            return Ok(());
+            (true, None) => {
+                ps.refused_with = Some(now_with);
+                trace!("Refusing the peer as a whole; not syncing with it");
+                return Ok(());
+            }
+            (true, Some(with)) if with == now_with => {
+                trace!("Refusing the peer as a whole; not syncing with it");
+                return Ok(());
+            }
+            // Another document since: without what we refused, or what it builds on it
+            // is refused again.
+            (true, Some(_)) => {
+                ps.refused_with = None;
+                drop(ps);
+                state.screening.lock_or_recover().forgive(peer);
+            }
         }
     }