Duty logic validation

anchor/client/src/lib.rs

@@ -28,7 +28,7 @@ use eth2::{
 use keygen::{encryption::decrypt, run_keygen, Keygen};
 use message_receiver::NetworkMessageReceiver;
 use message_sender::{impostor::ImpostorMessageSender, MessageSender, NetworkMessageSender};
-use message_validator::Validator;
+use message_validator::{DutiesTracker, Validator};
 use network::Network;
 use openssl::{pkey::Private, rsa::Rsa};
 use parking_lot::RwLock;
@@ -381,9 +381,20 @@ impl Client {
         // Network sender/receiver
         let (network_tx, network_rx) = mpsc::channel::<(SubnetId, Vec<u8>)>(9001);
 
+        let duties_tracker = Arc::new(DutiesTracker::new(
+            beacon_nodes.clone(),
+            spec.clone(),
+            E::slots_per_epoch(),
+            slot_clock.clone(),
+            database.watch(),
+        ));
+        duties_tracker.clone().start(executor.clone());
+
         let message_validator = Arc::new(Validator::new(
             database.watch(),
             E::slots_per_epoch(),
+            spec.epochs_per_sync_committee_period.as_u64(),
+            duties_tracker.clone(),
             slot_clock.clone(),
         ));
 

anchor/common/ssv_types/src/cluster.rs

@@ -70,6 +70,12 @@ pub struct ClusterMember {
 #[ssz(struct_behaviour = "transparent")]
 pub struct ValidatorIndex(pub usize);
 
+impl From<ValidatorIndex> for u64 {
+    fn from(value: ValidatorIndex) -> Self {
+        value.0 as u64
+    }
+}
+
 /// General Metadata about a Validator
 #[derive(Debug, Clone)]
 pub struct ValidatorMetadata {

anchor/database/src/state.rs

@@ -368,4 +368,12 @@ impl NetworkState {
             validator_indices: validator_index.map(|idx| vec![idx]).unwrap_or(vec![]),
         })
     }
+
+    pub fn validator_indices(&self) -> Vec<u64> {
+        self.multi_state
+            .validator_metadata
+            .values()
+            .filter_map(|metadata| metadata.index.map(|idx| idx.into()))
+            .collect()
+    }
 }

anchor/message_receiver/src/manager.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use database::{NetworkState, NonUniqueIndex, UniqueIndex};
 use gossipsub::{Message, MessageAcceptance, MessageId};
 use libp2p::PeerId;
-use message_validator::{ValidatedMessage, ValidatedSSVMessage, Validator};
+use message_validator::{DutiesProvider, ValidatedMessage, ValidatedSSVMessage, Validator};
 use qbft_manager::QbftManager;
 use signature_collector::SignatureCollectorManager;
 use slot_clock::SlotClock;
@@ -22,23 +22,23 @@ pub struct Outcome {
 }
 
 /// A message receiver that passes messages to responsible managers.
-pub struct NetworkMessageReceiver<S: SlotClock> {
+pub struct NetworkMessageReceiver<S: SlotClock, D: DutiesProvider> {
     processor: processor::Senders,
     qbft_manager: Arc<QbftManager>,
     signature_collector: Arc<SignatureCollectorManager>,
     network_state_rx: watch::Receiver<NetworkState>,
     outcome_tx: mpsc::Sender<Outcome>,
-    validator: Arc<Validator<S>>,
+    validator: Arc<Validator<S, D>>,
 }
 
-impl<S: SlotClock + 'static> NetworkMessageReceiver<S> {
+impl<S: SlotClock + 'static, D: DutiesProvider> NetworkMessageReceiver<S, D> {
     pub fn new(
         processor: processor::Senders,
         qbft_manager: Arc<QbftManager>,
         signature_collector: Arc<SignatureCollectorManager>,
         network_state_rx: watch::Receiver<NetworkState>,
         outcome_tx: mpsc::Sender<Outcome>,
-        validator: Arc<Validator<S>>,
+        validator: Arc<Validator<S, D>>,
     ) -> Arc<Self> {
         Arc::new(Self {
             processor,
@@ -51,7 +51,9 @@ impl<S: SlotClock + 'static> NetworkMessageReceiver<S> {
     }
 }
 
-impl<S: SlotClock + 'static> MessageReceiver for Arc<NetworkMessageReceiver<S>> {
+impl<S: SlotClock + 'static, D: DutiesProvider> MessageReceiver
+    for Arc<NetworkMessageReceiver<S, D>>
+{
     fn receive(
         &self,
         propagation_source: PeerId,
@@ -92,12 +94,12 @@ impl<S: SlotClock + 'static> MessageReceiver for Arc<NetworkMessageReceiver<S>>
                 } = match result {
                     Ok(message) => message,
                     Err(failure) => {
-                        debug!(?failure, "Validation failure");
+                        debug!(gosspisub_message_id = ?message_id, ?failure, "Validation failure");
                         return;
                     }
                 };
 
-                let msg_id = signed_ssv_message.ssv_message().msg_id();
+                let msg_id = signed_ssv_message.ssv_message().msg_id().clone();
 
                 match msg_id.duty_executor() {
                     Some(DutyExecutor::Validator(validator)) => {
@@ -109,7 +111,7 @@ impl<S: SlotClock + 'static> MessageReceiver for Arc<NetworkMessageReceiver<S>>
                             .is_none()
                         {
                             // We are not a signer for this validator, return without passing.
-                            trace!(?validator, ?msg_id, "Not interested");
+                            trace!(gosspisub_message_id = ?message_id, ssv_msg_id = ?msg_id, ?validator, "Not interested");
                             return;
                         }
                     }
@@ -123,19 +125,19 @@ impl<S: SlotClock + 'static> MessageReceiver for Arc<NetworkMessageReceiver<S>>
                         let committee = state.clusters().get_all_by(&committee_id);
                         // We only need to check one cluster, as all clusters will have the same set
                         // of operators.
-                        let is_member = committee
+                        let is_member = committee.clone()
                             .and_then(|mut v| v.pop())
                             .map(|c| c.cluster_members.contains(&own_id))
                             .unwrap_or(false);
 
                         if is_member {
                             // We are not a member for this committee, return without passing.
-                            trace!(?committee_id, ?msg_id, "Not interested");
+                            trace!(gosspisub_message_id = ?message_id, ssv_msg_id = ?msg_id, ?committee, "Not interested");
                             return;
                         }
                     }
                     None => {
-                        error!(?msg_id, "Invalid message ID");
+                        error!(gosspisub_message_id = ?message_id, ssv_msg_id = ?msg_id, "Invalid message ID");
                         return;
                     }
                 }
@@ -146,15 +148,15 @@ impl<S: SlotClock + 'static> MessageReceiver for Arc<NetworkMessageReceiver<S>>
                             .qbft_manager
                             .receive_data(signed_ssv_message, qbft_message)
                         {
-                            error!(?err, "Unable to receive QBFT message");
+                            error!(gosspisub_message_id = ?message_id, ssv_msg_id = ?msg_id, ?err, "Unable to receive QBFT message");
                         }
                     }
                     ValidatedSSVMessage::PartialSignatureMessages(messages) => {
                         if let Err(err) = receiver
                             .signature_collector
                             .receive_partial_signatures(messages)
                         {
-                            error!(?err, "Unable to receive partial signature message");
+                            error!(gosspisub_message_id = ?message_id, ssv_msg_id = ?msg_id, ?err, "Unable to receive partial signature message");
                         }
                     }
                 }

anchor/message_sender/src/network.rs

@@ -1,6 +1,6 @@
 use std::sync::Arc;
 
-use message_validator::Validator;
+use message_validator::{DutiesProvider, Validator};
 use openssl::{
     error::ErrorStack,
     hash::MessageDigest,
@@ -22,16 +22,16 @@ use crate::{Error, MessageCallback, MessageSender};
 const SIGNER_NAME: &str = "message_sign_and_send";
 const SENDER_NAME: &str = "message_send";
 
-pub struct NetworkMessageSender<S: SlotClock> {
+pub struct NetworkMessageSender<S: SlotClock, D: DutiesProvider> {
     processor: processor::Senders,
     network_tx: mpsc::Sender<(SubnetId, Vec<u8>)>,
     private_key: PKey<Private>,
     operator_id: OperatorId,
-    validator: Option<Arc<Validator<S>>>,
+    validator: Option<Arc<Validator<S, D>>>,
     subnet_count: usize,
 }
 
-impl<S: SlotClock + 'static> MessageSender for Arc<NetworkMessageSender<S>> {
+impl<S: SlotClock + 'static, D: DutiesProvider> MessageSender for Arc<NetworkMessageSender<S, D>> {
     fn sign_and_send(
         &self,
         message: UnsignedSSVMessage,
@@ -94,13 +94,13 @@ impl<S: SlotClock + 'static> MessageSender for Arc<NetworkMessageSender<S>> {
     }
 }
 
-impl<S: SlotClock> NetworkMessageSender<S> {
+impl<S: SlotClock, D: DutiesProvider> NetworkMessageSender<S, D> {
     pub fn new(
         processor: processor::Senders,
         network_tx: mpsc::Sender<(SubnetId, Vec<u8>)>,
         private_key: Rsa<Private>,
         operator_id: OperatorId,
-        validator: Option<Arc<Validator<S>>>,
+        validator: Option<Arc<Validator<S, D>>>,
         subnet_count: usize,
     ) -> Result<Arc<Self>, String> {
         let private_key = PKey::from_rsa(private_key)

anchor/message_validator/Cargo.toml

@@ -5,20 +5,26 @@ edition = { workspace = true }
 authors = ["Sigma Prime <[email protected]>"]
 
 [dependencies]
+beacon_node_fallback = { workspace = true }
+bls = { workspace = true }
 dashmap = { workspace = true }
 database = { workspace = true }
+eth2 = { workspace = true }
 ethereum_ssz = { workspace = true }
 gossipsub = { workspace = true }
 hex = { workspace = true }
 openssl = { workspace = true }
+parking_lot = { workspace = true }
 processor = { workspace = true }
+safe_arith = { workspace = true }
 sha2 = { workspace = true }
 slot_clock = { workspace = true }
 ssv_types = { workspace = true }
+task_executor = { workspace = true }
 thiserror =  { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
+types = { workspace = true }
 
 [dev-dependencies]
 bls = { workspace = true }
-types = { workspace = true }