pass messages from network to manager

anchor/client/src/lib.rs

@@ -145,11 +145,6 @@ impl Client {
         let subnet_tracker =
             start_subnet_tracker(database.watch(), network::SUBNET_COUNT, &executor);
 
-        // Start the p2p network
-        let network = Network::try_new(&config.network, subnet_tracker, executor.clone()).await?;
-        // Spawn the network listening task
-        executor.spawn(network.run(), "network");
-
         // Initialize slashing protection.
         let slashing_db_path = config.data_dir.join(SLASHING_PROTECTION_FILENAME);
         let slashing_protection =
@@ -362,6 +357,18 @@ impl Client {
         )
         .map_err(|e| format!("Unable to initialize qbft manager: {e:?}"))?;
 
+        // Start the p2p network
+        let network = Network::try_new(
+            &config.network,
+            subnet_tracker,
+            qbft_manager.clone(),
+            signature_collector.clone(),
+            executor.clone(),
+        )
+        .await?;
+        // Spawn the network listening task
+        executor.spawn(network.run(), "network");
+
         let validator_store = AnchorValidatorStore::<_, E>::new(
             database.watch(),
             signature_collector,

anchor/common/ssv_types/Cargo.toml

@@ -5,6 +5,7 @@ edition = { workspace = true }
 authors = ["Sigma Prime <[email protected]>"]
 
 [dependencies]
+alloy = { workspace = true }
 base64 = { workspace = true }
 derive_more = { workspace = true }
 ethereum_ssz = { workspace = true }

anchor/common/ssv_types/src/cluster.rs

@@ -3,6 +3,7 @@ use derive_more::{Deref, From};
 use indexmap::IndexSet;
 use ssz_derive::{Decode, Encode};
 use types::{Address, Graffiti, PublicKeyBytes};
+use crate::committee::CommitteeId;
 
 /// Unique identifier for a cluster
 #[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Hash, From, Deref)]
@@ -36,6 +37,10 @@ impl Cluster {
     pub fn get_f(&self) -> u64 {
         (self.cluster_members.len().saturating_sub(1) / 3) as u64
     }
+
+    pub fn committee_id(&self) -> CommitteeId {
+        self.cluster_members.iter().cloned().collect::<Vec<_>>().into()
+    }
 }
 
 /// A member of a Cluster.

anchor/common/ssv_types/src/committee.rs

@@ -0,0 +1,33 @@
+use alloy::primitives::keccak256;
+use derive_more::{Deref, From};
+use crate::OperatorId;
+
+const COMMITTEE_ID_LEN: usize = 32;
+
+/// Unique identifier for a committee
+#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Hash, From, Deref)]
+pub struct CommitteeId(pub [u8; COMMITTEE_ID_LEN]);
+
+impl From<Vec<OperatorId>> for CommitteeId {
+    fn from(mut operator_ids: Vec<OperatorId>) -> Self {
+        // Sort the operator IDs
+        operator_ids.sort();
+        let mut data: Vec<u8> = Vec::with_capacity(operator_ids.len() * 4);
+
+        // Add the operator IDs as 32 byte values
+        for id in operator_ids {
+            data.extend_from_slice(&id.to_le_bytes());
+        }
+
+        // Hash it all
+        keccak256(data).0.into()
+    }
+}
+
+impl TryFrom<&[u8]> for CommitteeId {
+    type Error = ();
+
+    fn try_from(value: &[u8]) -> Result<Self, ()> {
+        value.try_into().map(CommitteeId).map_err(|_| ())
+    }
+}

anchor/common/ssv_types/src/consensus.rs

@@ -1,6 +1,6 @@
 use crate::message::*;
 use crate::msgid::MessageId;
-use crate::{OperatorId, ValidatorIndex};
+use crate::ValidatorIndex;
 use sha2::{Digest, Sha256};
 use ssz::{Decode, DecodeError, Encode};
 use ssz_derive::{Decode, Encode};
@@ -26,7 +26,7 @@ use types::{
 //   MsgType            FullData
 //  ---------          -----------
 //  ConsensusMsg       QBFTMessage SSZ
-//  PartialSigMsg      PartialSignatureMessage SSZ
+//  PartialSigMsg      PartialSignatureMessages SSZ
 
 pub trait QbftData: Debug + Clone + Encode + Decode {
     type Hash: Debug + Clone + Eq + Hash;
@@ -144,15 +144,6 @@ impl Decode for QbftMessageType {
     }
 }
 
-// A partial signature specific message
-#[derive(Clone, Debug)]
-pub struct PartialSignatureMessage {
-    pub partial_signature: Signature,
-    pub signing_root: Hash256,
-    pub signer: OperatorId,
-    pub validator_index: ValidatorIndex,
-}
-
 #[derive(Clone, Debug, PartialEq, Encode, Decode)]
 pub struct ValidatorConsensusData {
     pub duty: ValidatorDuty,

anchor/common/ssv_types/src/lib.rs

@@ -2,11 +2,13 @@ pub use cluster::{Cluster, ClusterId, ClusterMember, ValidatorIndex, ValidatorMe
 pub use operator::{Operator, OperatorId};
 pub use share::Share;
 mod cluster;
+pub mod committee;
 pub mod consensus;
 pub mod domain_type;
 pub mod message;
 pub mod msgid;
 mod operator;
+pub mod partial_sig;
 mod share;
 mod sql_conversions;
 mod util;

anchor/common/ssv_types/src/message.rs

@@ -92,7 +92,7 @@ impl SSVMessage {
     ///
     /// ```
     /// use ssv_types::message::{MessageId, MsgType, SSVMessage};
-    /// let message_id = MessageId::new([0u8; 56]);
+    /// let message_id = MessageId::from([0u8; 56]);
     /// let msg = SSVMessage::new(MsgType::SSVConsensusMsgType, message_id, vec![1, 2, 3]);
     /// ```
     pub fn new(msg_type: MsgType, msg_id: MessageId, data: Vec<u8>) -> Self {
@@ -154,7 +154,7 @@ impl SignedSSVMessage {
     /// ```
     /// use ssv_types::message::{MessageId, MsgType, SSVMessage, SignedSSVMessage};
     /// use ssv_types::OperatorId;
-    /// let ssv_msg = SSVMessage::new(MsgType::SSVConsensusMsgType, MessageId::new([0u8; 56]), vec![1,2,3]);
+    /// let ssv_msg = SSVMessage::new(MsgType::SSVConsensusMsgType, MessageId::from([0u8; 56]), vec![1,2,3]);
     /// let signed_msg = SignedSSVMessage::new(vec![vec![0; 256]], vec![OperatorId(1)], ssv_msg, vec![4,5,6]).unwrap();
     /// ```
     pub fn new(

anchor/common/ssv_types/src/msgid.rs

@@ -1,5 +1,8 @@
+use derive_more::From;
+use crate::committee::CommitteeId;
 use crate::domain_type::DomainType;
 use ssz::{Decode, DecodeError, Encode};
+use types::PublicKeyBytes;
 
 const MESSAGE_ID_LEN: usize = 56;
 
@@ -37,26 +40,50 @@ impl TryFrom<&[u8]> for Role {
 }
 
 #[derive(Debug, Clone, Hash, Eq, PartialEq)]
-pub enum Executor {
-    Committee([u8; 32]),
-    Validator([u8; 48]),
+pub enum DutyExecutor {
+    Committee(CommitteeId),
+    Validator(PublicKeyBytes),
 }
 
-#[derive(Debug, Clone, Hash, Eq, PartialEq)]
+#[derive(Debug, Clone, Hash, Eq, PartialEq, From)]
 pub struct MessageId([u8; 56]);
 
 impl MessageId {
-    pub fn new(domain: &DomainType, role: Role, duty_executor: &Executor) -> Self {
+    pub fn new(domain: &DomainType, role: Role, duty_executor: &DutyExecutor) -> Self {
         let mut id = [0; 56];
         id[0..4].copy_from_slice(&domain.0);
         id[4..8].copy_from_slice(&<[u8; 4]>::from(role));
         match duty_executor {
-            Executor::Committee(slice) => id[24..].copy_from_slice(slice),
-            Executor::Validator(slice) => id[8..].copy_from_slice(slice),
+            DutyExecutor::Committee(committee_id) => id[24..].copy_from_slice(committee_id.as_slice()),
+            DutyExecutor::Validator(public_key) => id[8..].copy_from_slice(public_key.as_serialized()),
         }
 
         MessageId(id)
     }
+
+    pub fn domain(&self) -> DomainType {
+        DomainType(
+            self.0[0..4]
+                .try_into()
+                .expect("we know the slice has the correct length"),
+        )
+    }
+
+    pub fn role(&self) -> Option<Role> {
+        self.0[4..8].try_into().ok()
+    }
+
+    pub fn duty_executor(&self) -> Option<DutyExecutor> {
+        // which kind of executor we need to get depends on the role
+        match self.role()? {
+            Role::Committee => {
+                self.0[24..].try_into().ok().map(DutyExecutor::Committee)
+            }
+            Role::Aggregator | Role::Proposer | Role::SyncCommittee => {
+                PublicKeyBytes::deserialize(&self.0[8..]).ok().map(DutyExecutor::Validator)
+            }
+        }
+    }
 }
 
 impl AsRef<[u8]> for MessageId {
@@ -65,9 +92,11 @@ impl AsRef<[u8]> for MessageId {
     }
 }
 
-impl From<[u8; MESSAGE_ID_LEN]> for MessageId {
-    fn from(value: [u8; MESSAGE_ID_LEN]) -> Self {
-        MessageId(value)
+impl TryFrom<&[u8]> for MessageId {
+    type Error = ();
+
+    fn try_from(value: &[u8]) -> Result<Self, ()> {
+        value.try_into().map(MessageId).map_err(|_| ())
     }
 }
 

anchor/common/ssv_types/src/partial_sig.rs

@@ -0,0 +1,93 @@
+use ssz::{Decode, DecodeError, Encode};
+use ssz_derive::{Decode, Encode};
+use types::{Hash256, Signature, Slot};
+use crate::{OperatorId, ValidatorIndex};
+
+#[derive(Clone, Copy, Debug)]
+pub enum PartialSignatureKind {
+    // PostConsensusPartialSig is a partial signature over a decided duty (attestation data, block, etc)
+    PostConsensus = 0,
+    // RandaoPartialSig is a partial signature over randao reveal
+    RandaoPartialSig = 1,
+    // SelectionProofPartialSig is a partial signature for aggregator selection proof
+    SelectionProofPartialSig = 2,
+    // ContributionProofs is the partial selection proofs for sync committee contributions (it's an array of sigs)
+    ContributionProofs = 3,
+    // ValidatorRegistrationPartialSig is a partial signature over a ValidatorRegistration object
+    ValidatorRegistration = 4,
+    // VoluntaryExitPartialSig is a partial signature over a VoluntaryExit object
+    VoluntaryExit = 5,
+}
+
+impl TryFrom<u64> for PartialSignatureKind {
+    type Error = ();
+
+    fn try_from(value: u64) -> Result<Self, Self::Error> {
+        match value {
+            0 => Ok(PartialSignatureKind::PostConsensus),
+            1 => Ok(PartialSignatureKind::RandaoPartialSig),
+            2 => Ok(PartialSignatureKind::SelectionProofPartialSig),
+            3 => Ok(PartialSignatureKind::ContributionProofs),
+            4 => Ok(PartialSignatureKind::ValidatorRegistration),
+            5 => Ok(PartialSignatureKind::VoluntaryExit),
+            _ => Err(()),
+        }
+    }
+}
+
+const U64_SIZE: usize = 8; // u64 is 8 bytes
+
+impl Encode for PartialSignatureKind {
+    fn is_ssz_fixed_len() -> bool {
+        true
+    }
+
+    fn ssz_append(&self, buf: &mut Vec<u8>) {
+        buf.extend_from_slice(&(*self as u64).to_le_bytes());
+    }
+
+    fn ssz_fixed_len() -> usize {
+        U64_SIZE
+    }
+
+    fn ssz_bytes_len(&self) -> usize {
+        U64_SIZE
+    }
+}
+
+impl Decode for PartialSignatureKind {
+    fn is_ssz_fixed_len() -> bool {
+        true
+    }
+
+    fn ssz_fixed_len() -> usize {
+        U64_SIZE
+    }
+
+    fn from_ssz_bytes(bytes: &[u8]) -> Result<Self, DecodeError> {
+        if bytes.len() != U64_SIZE {
+            return Err(DecodeError::InvalidByteLength {
+                len: bytes.len(),
+                expected: U64_SIZE,
+            });
+        }
+        let value = u64::from_le_bytes(bytes.try_into().unwrap());
+        value.try_into().map_err(|_| DecodeError::NoMatchingVariant)
+    }
+}
+
+// A partial signature specific message
+#[derive(Clone, Debug, Encode, Decode)]
+pub struct PartialSignatureMessages {
+    pub kind: PartialSignatureKind,
+    pub slot: Slot,
+    pub messages: Vec<PartialSignatureMessage>,
+}
+
+#[derive(Clone, Debug, Encode, Decode)]
+pub struct PartialSignatureMessage {
+    pub partial_signature: Signature,
+    pub signing_root: Hash256,
+    pub signer: OperatorId,
+    pub validator_index: ValidatorIndex,
+}

anchor/network/Cargo.toml

@@ -30,6 +30,9 @@ quick-protobuf = "0.8.1"
 serde = { workspace = true }
 serde_json = "1.0.137"
 ssv_types = { workspace = true }
+qbft_manager = { workspace = true }
+signature_collector = { workspace = true }
+slot_clock = { workspace = true }
 ssz_types = "0.8"
 subnet_tracker = { workspace = true }
 task_executor = { workspace = true }