Ability to shuffle committee by stake weight (MystenLabs#2772)

* Ability to shuffle committee by stake weight

* Allow duplicate rand dep because we can't update ed25519_dalek

* Validate committee composition

* Make voting_rights private

* nits

crates/sui-config/src/genesis.rs

@@ -10,6 +10,7 @@ use std::{fs, path::Path};
 use sui_types::{
     base_types::TxContext,
     committee::{Committee, EpochId},
+    error::SuiResult,
     object::Object,
 };
 use tracing::{info, trace};
@@ -43,7 +44,7 @@ impl Genesis {
         &self.validator_set
     }
 
-    pub fn committee(&self) -> Committee {
+    pub fn committee(&self) -> SuiResult<Committee> {
         let voting_rights = self
             .validator_set()
             .iter()

crates/sui-config/src/swarm.rs

@@ -31,7 +31,7 @@ impl NetworkConfig {
     }
 
     pub fn committee(&self) -> Committee {
-        self.genesis.committee()
+        self.genesis.committee().unwrap()
     }
 
     pub fn into_validator_configs(self) -> Vec<NodeConfig> {

crates/sui-core/src/authority/authority_store.rs

@@ -1487,25 +1487,20 @@ pub async fn generate_genesis_system_object<S: Eq + Serialize + for<'de> Deseria
     let mut temporary_store =
         AuthorityTemporaryStore::new(store.clone(), InputObjects::new(vec![]), genesis_digest);
     let mut pubkeys = Vec::new();
-    for name in committee.voting_rights.keys() {
+    for name in committee.names() {
         pubkeys.push(committee.public_key(name)?.to_bytes().to_vec());
     }
     // TODO: May use separate sui address than derived from pubkey.
     let sui_addresses: Vec<AccountAddress> = committee
-        .voting_rights
-        .keys()
+        .names()
         .map(|pk| SuiAddress::from(pk).into())
         .collect();
     // TODO: Allow config to specify human readable validator names.
     let names: Vec<Vec<u8>> = (0..sui_addresses.len())
         .map(|i| Vec::from(format!("Validator{}", i).as_bytes()))
         .collect();
     // TODO: Change voting_rights to use u64 instead of usize.
-    let stakes: Vec<u64> = committee
-        .voting_rights
-        .values()
-        .map(|v| *v as u64)
-        .collect();
+    let stakes: Vec<u64> = committee.stakes().collect();
     adapter::execute(
         move_vm,
         &mut temporary_store,

crates/sui-core/src/authority_active.rs

@@ -125,9 +125,8 @@ impl<A> ActiveAuthority<A> {
         Ok(ActiveAuthority {
             health: Arc::new(Mutex::new(
                 committee
-                    .voting_rights
-                    .iter()
-                    .map(|(name, _)| (*name, AuthorityHealth::default()))
+                    .names()
+                    .map(|name| (*name, AuthorityHealth::default()))
                     .collect(),
             )),
             state: authority,
@@ -245,7 +244,7 @@ where
         // Number of tasks at most "degree" and no more than committee - 1
         // (validators do not follow themselves for gossip)
         let committee = self.state.committee.load().deref().clone();
-        let target_num_tasks = usize::min(committee.voting_rights.len() - 1, degree);
+        let target_num_tasks = usize::min(committee.num_members() - 1, degree);
 
         tokio::task::spawn(async move {
             gossip_process(&self, target_num_tasks).await;
@@ -260,7 +259,7 @@ where
         // nodes follow all validators to ensure they can eventually determine
         // finality of certs. We need to follow 2f+1 _honest_ validators to
         // eventually find finality, therefore we must follow all validators.
-        let target_num_tasks = committee.voting_rights.len();
+        let target_num_tasks = committee.num_members();
 
         tokio::task::spawn(async move {
             node_sync_process(&self, target_num_tasks, node_sync_store).await;

crates/sui-core/src/authority_active/checkpoint_driver/mod.rs

@@ -790,8 +790,7 @@ pub async fn sync_digest<A>(
 where
     A: AuthorityAPI + Send + Sync + 'static + Clone,
 {
-    let mut source_authorities: BTreeSet<AuthorityName> =
-        net.committee.voting_rights.keys().copied().collect();
+    let mut source_authorities: BTreeSet<AuthorityName> = net.committee.names().copied().collect();
     source_authorities.remove(&name);
 
     // Now try to update the destination authority sequentially using

crates/sui-core/src/authority_active/gossip/configurable_batch_action_client.rs

@@ -231,7 +231,7 @@ pub async fn init_configurable_authorities(
         voting_rights.insert(authority_name, 1);
         key_pairs.push((authority_name, key_pair));
     }
-    let committee = Committee::new(0, voting_rights);
+    let committee = Committee::new(0, voting_rights).unwrap();
 
     // Create Authority Clients and States.
     let mut clients = Vec::new();

crates/sui-core/src/authority_active/gossip/mod.rs

@@ -92,7 +92,7 @@ async fn follower_process<A, Handler: DigestHandler<A> + Clone>(
     let mut committee = local_active.state.committee.load().deref().clone();
 
     // Number of tasks at most "degree" and no more than committee - 1
-    let mut target_num_tasks = usize::min(committee.voting_rights.len() - 1, degree);
+    let mut target_num_tasks = usize::min(committee.num_members() - 1, degree);
 
     // If we do not expect to connect to anyone
     if target_num_tasks == 0 {
@@ -115,10 +115,10 @@ async fn follower_process<A, Handler: DigestHandler<A> + Clone>(
             // validators, and let them end naturally.
             local_active = Arc::new(active_authority.clone());
             committee = local_active.state.committee.load().deref().clone();
-            target_num_tasks = usize::min(committee.voting_rights.len() - 1, degree);
+            target_num_tasks = usize::min(committee.num_members() - 1, degree);
             peer_names = peer_names
                 .into_iter()
-                .filter(|name| committee.voting_rights.contains_key(name))
+                .filter(|name| committee.authority_exists(name))
                 .collect();
         }
         let mut k = 0;
@@ -230,7 +230,7 @@ where
 {
     // Make sure we exit loop by limiting the number of tries to choose peer
     // where n is the total number of committee members.
-    let mut tries_remaining = active_authority.state.committee.load().voting_rights.len();
+    let mut tries_remaining = active_authority.state.committee.load().num_members();
     while tries_remaining > 0 {
         let name = *active_authority.state.committee.load().sample();
         if peer_names.contains(&name)

crates/sui-core/src/authority_aggregator.rs

@@ -472,13 +472,12 @@ where
             Result<V, SuiError>,
         ) -> AsyncResult<'a, ReduceOutput<S>, SuiError>,
     {
-        // TODO: shuffle here according to stake
-        let authority_clients = &self.authority_clients;
+        let authorities_shuffled = self.committee.shuffle_by_stake();
 
         // First, execute in parallel for each authority FMap.
-        let mut responses: futures::stream::FuturesUnordered<_> = authority_clients
-            .iter()
-            .map(|(name, client)| {
+        let mut responses: futures::stream::FuturesUnordered<_> = authorities_shuffled
+            .map(|name| {
+                let client = &self.authority_clients[name];
                 let execute = map_each_authority.clone();
                 async move {
                     (
@@ -791,12 +790,7 @@ where
         // We update each object at each authority that does not have it.
         for object_id in objects {
             // Authorities to update.
-            let mut authorities: HashSet<AuthorityName> = self
-                .committee
-                .voting_rights
-                .iter()
-                .map(|(name, _)| *name)
-                .collect();
+            let mut authorities: HashSet<AuthorityName> = self.committee.names().cloned().collect();
 
             let (aggregate_object_info, certificates) = self.get_object_by_id(*object_id).await?;
 

crates/sui-core/src/checkpoints/reconstruction.rs

@@ -125,11 +125,8 @@ struct SpanGraph {
 impl SpanGraph {
     /// Initialize the graph with each authority just pointing to itself.
     pub fn new(committee: &Committee) -> SpanGraph {
-        let nodes: HashMap<AuthorityName, (AuthorityName, StakeUnit)> = committee
-            .voting_rights
-            .iter()
-            .map(|(n, w)| (*n, (*n, *w)))
-            .collect();
+        let nodes: HashMap<AuthorityName, (AuthorityName, StakeUnit)> =
+            committee.members().map(|(n, w)| (*n, (*n, *w))).collect();
 
         SpanGraph { nodes }
     }

crates/sui-core/src/epoch/reconfiguration.rs

@@ -92,7 +92,7 @@ where
                 )
             })
             .collect();
-        let new_committee = Committee::new(next_epoch, votes);
+        let new_committee = Committee::new(next_epoch, votes)?;
         self.state.insert_new_epoch_info(&new_committee)?;
         let new_net = Arc::new(AuthorityAggregator::new(
             new_committee,

crates/sui-core/src/unit_tests/authority_aggregator_tests.rs

@@ -56,7 +56,7 @@ pub async fn init_local_authorities_with_genesis(
         voting_rights.insert(authority_name, 1);
         key_pairs.push((authority_name, key_pair));
     }
-    let committee = Committee::new(0, voting_rights);
+    let committee = Committee::new(0, voting_rights).unwrap();
 
     let mut clients = BTreeMap::new();
     let mut states = Vec::new();

crates/sui-core/src/unit_tests/authority_tests.rs

@@ -1571,7 +1571,7 @@ fn init_state_parameters() -> (Committee, SuiAddress, KeyPair, Arc<AuthorityStor
         /* address */ *authority_key.public_key_bytes(),
         /* voting right */ 1,
     );
-    let committee = Committee::new(0, authorities);
+    let committee = Committee::new(0, authorities).unwrap();
 
     // Create a random directory to store the DB
 

crates/sui-core/src/unit_tests/batch_tests.rs

@@ -39,7 +39,7 @@ where
         /* address */ *authority_key.public_key_bytes(),
         /* voting right */ 1,
     );
-    let committee = Committee::new(0, authorities);
+    let committee = Committee::new(0, authorities).unwrap();
 
     (committee, authority_address, authority_key)
 }
@@ -761,7 +761,7 @@ async fn test_safe_batch_stream() {
     println!("init public key {:?}", public_key_bytes);
 
     authorities.insert(public_key_bytes, 1);
-    let committee = Committee::new(0, authorities);
+    let committee = Committee::new(0, authorities).unwrap();
     // Create an authority
     let store = Arc::new(AuthorityStore::open(&path, None));
     let state = AuthorityState::new(

crates/sui-gateway/src/config.rs

@@ -20,6 +20,7 @@ use sui_core::{
 use sui_types::{
     base_types::AuthorityName,
     committee::{Committee, EpochId},
+    error::SuiResult,
 };
 
 #[derive(Serialize, Deserialize)]
@@ -65,7 +66,7 @@ impl GatewayType {
         Ok(match self {
             GatewayType::Embedded(config) => {
                 let path = config.db_folder_path.clone();
-                let committee = config.make_committee();
+                let committee = config.make_committee()?;
                 let authority_clients = config.make_authority_clients();
                 let metrics = GatewayMetrics::new(&prometheus::Registry::new());
                 Arc::new(GatewayState::new(
@@ -93,7 +94,7 @@ pub struct GatewayConfig {
 impl Config for GatewayConfig {}
 
 impl GatewayConfig {
-    pub fn make_committee(&self) -> Committee {
+    pub fn make_committee(&self) -> SuiResult<Committee> {
         let voting_rights = self
             .validator_set
             .iter()

crates/sui-gateway/src/lib.rs

@@ -22,7 +22,7 @@ pub fn create_client(
             e
         )
     })?;
-    let committee = config.make_committee();
+    let committee = config.make_committee()?;
     let authority_clients = config.make_authority_clients();
     Ok(Arc::new(GatewayState::new(
         config.db_folder_path,

crates/sui-node/src/lib.rs

@@ -64,7 +64,7 @@ impl SuiNode {
         let genesis = config.genesis()?;
 
         let secret = Arc::pin(config.key_pair().copy());
-        let committee = genesis.committee();
+        let committee = genesis.committee()?;
         let store = Arc::new(AuthorityStore::open(config.db_path().join("store"), None));
         let checkpoint_store = if config.consensus_config().is_some() {
             Some(Arc::new(Mutex::new(CheckpointStore::open(

crates/sui-types/Cargo.toml

@@ -37,6 +37,9 @@ tonic = "0.7"
 strum = "^0.24"
 strum_macros = "^0.24"
 
+# This version is incompatible with ed25519-dalek
+rand_latest = { version = "0.8.5", package = "rand" }
+
 name-variant = { git = "https://github.com/MystenLabs/mysten-infra", rev = "94d7da89f6a52d7f60a9802b0a03147a9c89c3e4" }
 typed-store = { git = "https://github.com/MystenLabs/mysten-infra", rev = "94d7da89f6a52d7f60a9802b0a03147a9c89c3e4" }