Allow nodes to gradually drop out during the protocol

Author: fjarri

manul/src/protocol/round.rs

@@ -50,6 +50,11 @@ impl<Id: Ord> IdSet<Id> {
         // TODO: assuming `ids` are a subset of `self.ids`. Can we?
         ids.len() >= self.threshold
     }
+
+    pub(crate) fn is_quorum_possible(&self, banned_ids: &BTreeSet<Id>) -> bool {
+        let ids = self.ids.intersection(banned_ids).collect::<BTreeSet<_>>();
+        self.ids.len() - ids.len() >= self.threshold
+    }
 }
 
 /// Describes what other parties this rounds sends messages to, and what other parties it expects messages from.

manul/src/session/session.rs

@@ -300,7 +300,7 @@ where
     ) -> Result<PreprocessOutcome<SP::Verifier>, LocalError> {
         // Quick preliminary checks, before we proceed with more expensive verification
         let key = self.verifier();
-        if self.transcript.is_banned(from) || accum.is_banned(from) {
+        if accum.is_banned(from) {
             trace!("{key:?} Banned.");
             return Ok(PreprocessOutcome::remote_error("The sender is banned"));
         }
@@ -417,7 +417,10 @@ where
 
     /// Makes an accumulator for a new round.
     pub fn make_accumulator(&self) -> RoundAccumulator<P, SP> {
-        RoundAccumulator::new(&self.communication_info.expecting_messages_from)
+        RoundAccumulator::new(
+            &self.communication_info.expecting_messages_from,
+            self.transcript.banned_ids(),
+        )
     }
 
     fn terminate_inner(
@@ -549,6 +552,7 @@ pub enum CanFinalize {
 /// A mutable accumulator for collecting the results and errors from processing messages for a single round.
 #[derive_where::derive_where(Debug)]
 pub struct RoundAccumulator<P: Protocol<SP::Verifier>, SP: SessionParameters> {
+    banned_ids: BTreeSet<SP::Verifier>,
     still_have_not_sent_messages: BTreeSet<SP::Verifier>,
     expecting_messages_from: IdSet<SP::Verifier>,
     processing: BTreeSet<SP::Verifier>,
@@ -567,8 +571,9 @@ where
     P: Protocol<SP::Verifier>,
     SP: SessionParameters,
 {
-    fn new(expecting_messages_from: &IdSet<SP::Verifier>) -> Self {
+    fn new(expecting_messages_from: &IdSet<SP::Verifier>, banned_ids: BTreeSet<SP::Verifier>) -> Self {
         Self {
+            banned_ids,
             still_have_not_sent_messages: expecting_messages_from.all().clone(),
             expecting_messages_from: expecting_messages_from.clone(),
             processing: BTreeSet::new(),
@@ -589,15 +594,15 @@ where
             .is_quorum(&self.payloads.keys().cloned().collect::<BTreeSet<_>>())
         {
             CanFinalize::Yes
-        } else if !self.still_have_not_sent_messages.is_empty() {
+        } else if self.expecting_messages_from.is_quorum_possible(&self.banned_ids) {
             CanFinalize::NotYet
         } else {
             CanFinalize::Never
         }
     }
 
     fn is_banned(&self, from: &SP::Verifier) -> bool {
-        self.provable_errors.contains_key(from) || self.unprovable_errors.contains_key(from)
+        self.banned_ids.contains(from)
     }
 
     fn message_is_being_processed(&self, from: &SP::Verifier) -> bool {
@@ -619,6 +624,7 @@ where
                 from
             )))
         } else {
+            self.banned_ids.insert(from.clone());
             Ok(())
         }
     }
@@ -630,6 +636,7 @@ where
                 from
             )))
         } else {
+            self.banned_ids.insert(from.clone());
             Ok(())
         }
     }
@@ -728,10 +735,7 @@ where
                 )?;
                 self.register_provable_error(&from, evidence)
             }
-            ReceiveErrorType::Unprovable(error) => {
-                self.unprovable_errors.insert(from.clone(), error);
-                Ok(())
-            }
+            ReceiveErrorType::Unprovable(error) => self.register_unprovable_error(&from, error),
             ReceiveErrorType::Echo(error) => {
                 let (_echo_broadcast, normal_broadcast, _direct_message) = processed.message.into_parts();
                 let evidence = Evidence::new_echo_round_error(&from, normal_broadcast, *error)?;

manul/src/session/transcript.rs

@@ -163,6 +163,12 @@ where
             .ok_or_else(|| LocalError::new(format!("No direct messages registered for {from:?} in {round_id:?}")))
     }
 
+    pub fn banned_ids(&self) -> BTreeSet<SP::Verifier> {
+        let mut banned = self.provable_errors.keys().cloned().collect::<BTreeSet<_>>();
+        banned.extend(self.unprovable_errors.keys().cloned());
+        banned
+    }
+
     pub fn is_banned(&self, from: &SP::Verifier) -> bool {
         self.provable_errors.contains_key(from) || self.unprovable_errors.contains_key(from)
     }