Handel: parameterize the complete condition

handel/src/aggregation.rs

@@ -187,7 +187,7 @@ where
     }
 
     fn is_complete_aggregate(&self, aggregate: &P::Contribution) -> bool {
-        self.num_contributors(aggregate) == self.protocol.partitioner().size()
+        self.protocol.evaluator().is_complete(aggregate)
     }
 
     /// Check if the given level was completed. If so mark it as such.

handel/src/evaluator.rs

@@ -16,6 +16,12 @@ use crate::{
     Identifier,
 };
 
+type IsCompleteFn<TId, TProtocol> = fn(
+    &<TProtocol as Protocol<TId>>::Contribution,
+    &<TProtocol as Protocol<TId>>::Registry,
+    &<TProtocol as Protocol<TId>>::Partitioner,
+) -> bool;
+
 /// Trait for scoring or evaluating a contribution or signature.
 pub trait Evaluator<TId, TProtocol>
 where
@@ -31,6 +37,9 @@ where
 
     /// Returns whether a level contains a specific peer ID.
     fn verify(&self, msg: &LevelUpdate<TProtocol::Contribution>) -> Result<(), VerificationError>;
+
+    /// Returns whether the given aggregate is considered complete.
+    fn is_complete(&self, aggregate: &TProtocol::Contribution) -> bool;
 }
 
 /// A signature counts as it was signed N times, where N is the signers weight
@@ -48,6 +57,9 @@ where
 
     /// Partitioner that registers the handel levels and its IDs.
     partitioner: Arc<TProtocol::Partitioner>,
+
+    /// Function to test if the given aggregate is complete.
+    is_complete_fn: IsCompleteFn<TId, TProtocol>,
 }
 
 impl<TId, TProtocol> WeightedVote<TId, TProtocol>
@@ -76,11 +88,13 @@ where
         store: Arc<RwLock<TProtocol::Store>>,
         weights: Arc<TProtocol::Registry>,
         partitioner: Arc<TProtocol::Partitioner>,
+        is_complete_fn: IsCompleteFn<TId, TProtocol>,
     ) -> Self {
         Self {
             store,
             weights,
             partitioner,
+            is_complete_fn,
         }
     }
 }
@@ -91,10 +105,7 @@ pub enum VerificationError {
         level: usize,
         num_levels: usize,
     },
-    InvalidFullAggregate {
-        weight: usize,
-        expected_weight: usize,
-    },
+    InvalidFullAggregate,
     InvalidOrigin {
         origin: usize,
         allowed_contributors: Identity,
@@ -264,18 +275,12 @@ where
         }
 
         // Special case for full aggregations, which are sent at level `num_levels`.
-        // They are only valid if they contain all signers.
+        // They are only valid if they are complete.
         let contributors = self.weights.signers_identity(&msg.aggregate.contributors());
         if level == num_levels {
-            let weight = contributors.len();
-            let expected_weight = self.partitioner.size();
-            if weight != expected_weight {
-                return Err(InvalidFullAggregate {
-                    weight,
-                    expected_weight,
-                });
+            if !self.is_complete(&msg.aggregate) {
+                return Err(InvalidFullAggregate);
             }
-
             return Ok(());
         }
 
@@ -318,4 +323,8 @@ where
 
         Ok(())
     }
+
+    fn is_complete(&self, aggregate: &TProtocol::Contribution) -> bool {
+        (self.is_complete_fn)(aggregate, &self.weights, &self.partitioner)
+    }
 }

handel/tests/mod.rs

@@ -11,7 +11,7 @@ use nimiq_handel::{
     evaluator::WeightedVote,
     identity::{Identity, IdentityRegistry, WeightRegistry},
     network::Network,
-    partitioner::BinomialPartitioner,
+    partitioner::{BinomialPartitioner, Partitioner},
     protocol,
     store::ReplaceStore,
     update::LevelUpdate,
@@ -113,6 +113,9 @@ impl Protocol {
             store.clone(),
             Arc::clone(&registry),
             partitioner.clone(),
+            |aggregate: &Contribution, registry: &Registry, partitioner: &BinomialPartitioner| {
+                registry.signers_identity(&aggregate.contributors()).len() == partitioner.size()
+            },
         ));
 
         Protocol {

validator/src/aggregation/skip_block.rs

@@ -20,7 +20,7 @@ use nimiq_network_interface::{
     network::CloseReason,
     request::{MessageMarker, RequestCommon},
 };
-use nimiq_primitives::{policy, slots_allocation::Validators, Message};
+use nimiq_primitives::{policy::Policy, slots_allocation::Validators, Message};
 use nimiq_validator_network::ValidatorNetwork;
 use parking_lot::RwLock;
 use serde::{Deserialize, Serialize};
@@ -144,6 +144,9 @@ impl SkipBlockAggregationProtocol {
             Arc::clone(&store),
             Arc::clone(&registry),
             Arc::clone(&partitioner),
+            |aggregate: &SignedSkipBlockMessage, registry: &ValidatorRegistry, _| {
+                registry.signature_weight(aggregate).unwrap_or(0) >= Policy::TWO_F_PLUS_ONE as usize
+            },
         ));
 
         SkipBlockAggregationProtocol {
@@ -286,12 +289,12 @@ impl SkipBlockAggregation {
                 block_number = skip_block_info.block_number,
                 "New skip block aggregate weight {}/{} with signers {}",
                 aggregate_weight,
-                policy::Policy::TWO_F_PLUS_ONE,
+                Policy::TWO_F_PLUS_ONE,
                 &msg.contributors(),
             );
 
             // Check if the combined weight of the aggregation is at least 2f+1.
-            if aggregate_weight >= policy::Policy::TWO_F_PLUS_ONE as usize {
+            if aggregate_weight >= Policy::TWO_F_PLUS_ONE as usize {
                 // Create SkipBlockProof from the aggregate.
                 let skip_block_proof = SkipBlockProof { sig: msg.proof };
 

validator/src/aggregation/tendermint/protocol.rs

@@ -4,7 +4,8 @@ use nimiq_handel::{
     evaluator::WeightedVote, partitioner::BinomialPartitioner, protocol::Protocol,
     store::ReplaceStore,
 };
-use nimiq_primitives::TendermintIdentifier;
+use nimiq_primitives::{policy::Policy, TendermintIdentifier};
+use nimiq_tendermint::Aggregation;
 use parking_lot::RwLock;
 
 use super::{
@@ -26,30 +27,35 @@ pub(crate) struct TendermintAggregationProtocol {
 
 impl TendermintAggregationProtocol {
     pub(crate) fn new(
-        validators: Arc<ValidatorRegistry>,
+        registry: Arc<ValidatorRegistry>,
         node_id: usize,
         id: TendermintIdentifier,
     ) -> Self {
-        let partitioner = Arc::new(BinomialPartitioner::new(node_id, validators.len()));
+        let partitioner = Arc::new(BinomialPartitioner::new(node_id, registry.len()));
 
         let store = Arc::new(RwLock::new(
             ReplaceStore::<TendermintIdentifier, Self>::new(Arc::clone(&partitioner)),
         ));
 
         let evaluator = Arc::new(WeightedVote::new(
             Arc::clone(&store),
-            validators.clone(),
+            Arc::clone(&registry),
             Arc::clone(&partitioner),
+            |aggregate: &TendermintContribution, _, _| {
+                aggregate.proposals().iter().any(|(_, contributor_count)| {
+                    *contributor_count >= Policy::TWO_F_PLUS_ONE as usize
+                })
+            },
         ));
 
-        let verifier = Arc::new(TendermintVerifier::new(validators.clone(), id.clone()));
+        let verifier = Arc::new(TendermintVerifier::new(registry.clone(), id.clone()));
 
         Self {
             verifier,
             partitioner,
             evaluator,
             store,
-            registry: validators,
+            registry,
             node_id,
             id,
         }