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.

1 files changed, 21 insertions, 3 deletions

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);
+            }
         }
     }