modularize committee sync and event stream logic

Author: alxiong

Cargo.lock

@@ -69,6 +69,7 @@ ethereum_ssz = "0.9.0"
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
 generic-array = { version = "0.14.7", features = ["serde", "zeroize"] }
 http = "1.3.1"
+itertools = "0.14.0"
 jiff = { version = "0.2", default-features = false, features = ["serde", "std"] }
 minicbor = { version = "2.1.1", features = ["full"] }
 nohash-hasher = "0.2"

Cargo.toml

@@ -28,6 +28,7 @@ cliquenet = { path = "../cliquenet" }
 committable = { workspace = true }
 futures = { workspace = true }
 http = { workspace = true }
+itertools = { workspace = true }
 metrics = { path = "../metrics", features = ["prometheus"]}
 multisig = { path = "../multisig" }
 prost = { workspace = true }

timeboost/Cargo.toml

@@ -1,20 +1,18 @@
 use std::{iter::repeat_with, path::PathBuf, sync::Arc, time::Duration};
 
 use ::metrics::prometheus::PrometheusMetrics;
-use alloy::providers::{Provider, ProviderBuilder};
 use anyhow::{Context, Result};
 use cliquenet::{Network, NetworkMetrics, Overlay};
 use committable::{Commitment, Committable, RawCommitmentBuilder};
-use multisig::{Committee, CommitteeId, Keypair, x25519};
+use multisig::{CommitteeId, Keypair, x25519};
 use sailfish::{
     Coordinator,
     consensus::{Consensus, ConsensusMetrics},
     rbc::{Rbc, RbcConfig, RbcMetrics},
-    types::{Action, HasTime, Timestamp, UNKNOWN_COMMITTEE_ID},
+    types::{Action, HasTime, Timestamp},
 };
 use serde::{Deserialize, Serialize};
-use timeboost::config::NodeConfig;
-use timeboost_contract::{CommitteeMemberSol, KeyManager};
+use timeboost::{committee::CommitteeInfo, config::NodeConfig};
 use timeboost_utils::types::logging;
 use tokio::{select, signal, time::sleep};
 use tracing::{error, info};
@@ -91,60 +89,33 @@ async fn main() -> Result<()> {
     let dh_keypair = x25519::Keypair::from(config.keys.dh.secret.clone());
 
     // syncing with contract to get peers keys and network addresses
-    let provider = ProviderBuilder::new().connect_http(config.chain.parent.rpc_url);
-    assert_eq!(
-        provider.get_chain_id().await?,
-        config.chain.parent.id,
-        "Parent chain RPC has mismatched chain_id"
-    );
-
-    let contract = KeyManager::new(config.chain.parent.key_manager_contract, &provider);
-    let members: Vec<CommitteeMemberSol> = contract
-        .getCommitteeById(cli.committee_id.into())
-        .call()
-        .await?
-        .members;
+    let comm_info = CommitteeInfo::fetch(
+        config.chain.parent.rpc_url,
+        config.chain.parent.key_manager_contract,
+        cli.committee_id.into(),
+    )
+    .await?;
     info!(label = %config.keys.signing.public, committee_id = %cli.committee_id, "committee info synced");
 
-    let peer_hosts_and_keys = members
-        .iter()
-        .map(|peer| {
-            let sig_key = multisig::PublicKey::try_from(peer.sigKey.as_ref())
-                .expect("Failed to parse sigKey");
-            let dh_key =
-                x25519::PublicKey::try_from(peer.dhKey.as_ref()).expect("Failed to parse dhKey");
-            let sailfish_address = cliquenet::Address::try_from(peer.networkAddress.as_ref())
-                .expect("Failed to parse networkAddress");
-            (sig_key, dh_key, sailfish_address)
-        })
-        .collect::<Vec<_>>();
-
     let prom = Arc::new(PrometheusMetrics::default());
     let sf_metrics = ConsensusMetrics::new(prom.as_ref());
     let net_metrics = NetworkMetrics::new(
         "sailfish",
         prom.as_ref(),
-        peer_hosts_and_keys.iter().map(|(k, ..)| *k),
+        comm_info.signing_keys().iter().cloned(),
     );
     let rbc_metrics = RbcMetrics::new(prom.as_ref());
     let network = Network::create(
         "sailfish",
         config.net.public.address.clone(),
         signing_keypair.public_key(),
         dh_keypair.clone(),
-        peer_hosts_and_keys.clone(),
+        comm_info.group(),
         net_metrics,
     )
     .await?;
 
-    let committee = Committee::new(
-        UNKNOWN_COMMITTEE_ID,
-        peer_hosts_and_keys
-            .iter()
-            .map(|b| b.0)
-            .enumerate()
-            .map(|(i, key)| (i as u8, key)),
-    );
+    let committee = comm_info.committee();
 
     // If the stamp file exists we need to recover from a previous run.
     let recover = if cli.ignore_stamp {

timeboost/src/binaries/sailfish.rs

@@ -1,15 +1,12 @@
 use std::path::PathBuf;
 
-use alloy::providers::{Provider, ProviderBuilder};
 use anyhow::{Context, Result, bail};
-use cliquenet::AddressableCommittee;
 use multisig::CommitteeId;
-use multisig::{Committee, Keypair, x25519};
+use multisig::{Keypair, x25519};
+use timeboost::committee::CommitteeInfo;
 use timeboost::{Timeboost, TimeboostConfig};
 use timeboost_builder::robusta;
-use timeboost_contract::{CommitteeMemberSol, KeyManager};
-use timeboost_crypto::prelude::DkgEncKey;
-use timeboost_types::{KeyStore, ThresholdKeyCell};
+use timeboost_types::ThresholdKeyCell;
 use tokio::select;
 use tokio::signal;
 use tokio::task::spawn;
@@ -73,99 +70,18 @@ async fn main() -> Result<()> {
     let dh_keypair = x25519::Keypair::from(node_config.keys.dh.secret.clone());
 
     // syncing with contract to get peers keys and network addresses
-    let provider = ProviderBuilder::new().connect_http(node_config.chain.parent.rpc_url.clone());
-    let chain_id = provider.get_chain_id().await?;
-
-    assert_eq!(
-        chain_id, node_config.chain.parent.id,
-        "parent chain rpc has mismatched chain_id"
-    );
-
-    let contract = KeyManager::new(node_config.chain.parent.key_manager_contract, &provider);
-
-    let members: Vec<CommitteeMemberSol> = contract
-        .getCommitteeById(cli.committee_id.into())
-        .call()
-        .await?
-        .members;
-
+    let comm_info = CommitteeInfo::fetch(
+        node_config.chain.parent.rpc_url.clone(),
+        node_config.chain.parent.key_manager_contract,
+        cli.committee_id.into(),
+    )
+    .await?;
     info!(label = %sign_keypair.public_key(), committee_id = %cli.committee_id, "committee info synced");
 
-    let peer_hosts_and_keys = members
-        .iter()
-        .map(|peer| -> Result<_> {
-            let sig_key = multisig::PublicKey::try_from(peer.sigKey.as_ref())
-                .with_context(|| "Failed to parse sigKey bytes")?;
-            let dh_key = x25519::PublicKey::try_from(peer.dhKey.as_ref())
-                .with_context(|| "Failed to parse dhKey bytes")?;
-            let dkg_enc_key = DkgEncKey::from_bytes(peer.dkgKey.as_ref())
-                .with_context(|| "Failed to parse dkgKey bytes")?;
-            let sailfish_address = cliquenet::Address::try_from(peer.networkAddress.as_ref())
-                .with_context(|| "Failed to parse networkAddress string")?;
-            Ok((sig_key, dh_key, dkg_enc_key, sailfish_address))
-        })
-        .collect::<Result<Vec<_>>>()?;
-
-    let mut sailfish_peer_hosts_and_keys = Vec::new();
-    let mut decrypt_peer_hosts_and_keys = Vec::new();
-    let mut certifier_peer_hosts_and_keys = Vec::new();
-    let mut dkg_enc_keys = Vec::new();
-
-    for (signing_key, dh_key, dkg_enc_key, sailfish_addr) in peer_hosts_and_keys.iter().cloned() {
-        sailfish_peer_hosts_and_keys.push((signing_key, dh_key, sailfish_addr.clone()));
-        decrypt_peer_hosts_and_keys.push((
-            signing_key,
-            dh_key,
-            sailfish_addr.clone().with_offset(DECRYPTER_PORT_OFFSET),
-        ));
-        certifier_peer_hosts_and_keys.push((
-            signing_key,
-            dh_key,
-            sailfish_addr.clone().with_offset(CERTIFIER_PORT_OFFSET),
-        ));
-        dkg_enc_keys.push(dkg_enc_key.clone());
-    }
-
-    let sailfish_committee = {
-        let c = Committee::new(
-            cli.committee_id,
-            sailfish_peer_hosts_and_keys
-                .iter()
-                .enumerate()
-                .map(|(i, (k, ..))| (i as u8, *k)),
-        );
-        AddressableCommittee::new(c, sailfish_peer_hosts_and_keys.iter().cloned())
-    };
-
-    let decrypt_committee = {
-        let c = Committee::new(
-            cli.committee_id,
-            decrypt_peer_hosts_and_keys
-                .iter()
-                .enumerate()
-                .map(|(i, (k, ..))| (i as u8, *k)),
-        );
-        AddressableCommittee::new(c, decrypt_peer_hosts_and_keys.iter().cloned())
-    };
-
-    let certifier_committee = {
-        let c = Committee::new(
-            cli.committee_id,
-            certifier_peer_hosts_and_keys
-                .iter()
-                .enumerate()
-                .map(|(i, (k, ..))| (i as u8, *k)),
-        );
-        AddressableCommittee::new(c, certifier_peer_hosts_and_keys.iter().cloned())
-    };
-
-    let key_store = KeyStore::new(
-        sailfish_committee.committee().clone(),
-        dkg_enc_keys
-            .into_iter()
-            .enumerate()
-            .map(|(i, k)| (i as u8, k)),
-    );
+    let sailfish_committee = comm_info.sailfish_committee();
+    let decrypt_committee = comm_info.decrypt_committee();
+    let certifier_committee = comm_info.certifier_committee();
+    let key_store = comm_info.dkg_key_store();
 
     let is_recover = !cli.ignore_stamp && node_config.stamp.is_file();