Initial event emission by the contract watcher

validator-api/src/dkg/error.rs

@@ -3,6 +3,7 @@
 
 use std::io;
 use thiserror::Error;
+use validator_client::nymd::error::NymdError;
 use validator_client::ValidatorClientError;
 
 #[derive(Error, Debug)]
@@ -13,12 +14,18 @@ pub enum DkgError {
     #[error("{0}")]
     ContractClient(#[from] ValidatorClientError),
 
+    #[error("{0}")]
+    NymdClient(#[from] NymdError),
+
     #[error("Networking error - {0}")]
     Networking(#[from] io::Error),
 
     #[error("Failed to serialize message - {0}")]
     SerializationError(#[from] bincode::Error),
 
+    #[error("todo")]
+    MalformedDealerDetails,
+    
     #[error("todo")]
     DeserializationError,
 }

validator-api/src/dkg/events/event.rs

@@ -11,6 +11,12 @@ pub(crate) enum Event {
     DkgContractChange(watcher::Event),
 }
 
+impl Event {
+    pub(crate) fn new_contract_change_event(event: watcher::Event) -> Self {
+        Event::DkgContractChange(event)
+    }
+}
+
 impl Display for Event {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         match self {

validator-api/src/dkg/main_loop/mod.rs

@@ -1,8 +1,10 @@
 // Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use crate::dkg::smart_contract::publisher::Publisher;
 use crate::dkg::smart_contract::watcher;
-use crate::dkg::state::{DkgState, StateAccessor};
+use crate::dkg::state::StateAccessor;
+use coconut_dkg_common::types::DealerDetails;
 use futures::channel::mpsc;
 use futures::StreamExt;
 
@@ -12,12 +14,19 @@ pub(crate) type ContractEventsReceiver = mpsc::UnboundedReceiver<watcher::Event>
 pub(crate) type ContractEventsSender = mpsc::UnboundedSender<watcher::Event>;
 
 // it is driven by events received from the contract watcher
-pub(crate) struct ProcessingLoop {
+pub(crate) struct ProcessingLoop<C> {
     state_accessor: StateAccessor,
     contract_events_receiver: ContractEventsReceiver,
+
+    contract_publisher: Publisher<C>,
+    // network_sender: Sender
 }
 
-impl ProcessingLoop {
+impl<C> ProcessingLoop<C> {
+    fn verify_dealer(&self, contract_dealer_details: &DealerDetails) {
+        //
+    }
+
     async fn process_event(&self, event: watcher::Event) {
         todo!()
     }

validator-api/src/dkg/mod.rs

@@ -9,6 +9,14 @@ pub(crate) mod networking;
 mod smart_contract;
 pub(crate) mod state;
 
+pub(crate) struct Config {
+    // not entirely sure what should go here just yet
+}
+
+pub(crate) fn initiate_dkg(config: Config) {
+    //
+}
+
 // upon startup, the following tasks will need to be spawned:
 // - smart contract watcher
 // - main loop processing

validator-api/src/dkg/smart_contract/publisher/mod.rs

@@ -2,7 +2,18 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::Client;
+use validator_client::nymd::SigningCosmWasmClient;
 
 pub(crate) struct Publisher<C> {
     client: Client<C>,
 }
+
+impl<C> Publisher<C>
+where
+    C: SigningCosmWasmClient + Send + Sync,
+{
+    pub(crate) async fn submit_dealing_commitment(&self) {
+        // self.client.submit_dealing_commitment().await;
+        //
+    }
+}

validator-api/src/dkg/smart_contract/watcher/event.rs

@@ -1,18 +1,45 @@
 // Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use coconut_dkg_common::types::{BlockHeight, DealerDetails};
 use std::fmt::{Display, Formatter};
 
 #[derive(Debug)]
-pub(crate) enum Event {
-    NewDealingCommitment,
+pub(crate) struct Event {
+    height: BlockHeight,
+    event_type: EventType,
+}
+
+impl Event {
+    pub(crate) fn new(height: BlockHeight, event_type: EventType) -> Self {
+        Event { height, event_type }
+    }
 }
 
 impl Display for Event {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "SmartContractWatcherEvent - ")?;
+        write!(
+            f,
+            "SmartContractWatcherEvent at height {}. {}",
+            self.height, self.event_type
+        )
+    }
+}
+
+#[derive(Debug)]
+pub(crate) enum EventType {
+    NewDealerIdentity { details: DealerDetails },
+    NewDealingCommitment,
+}
+
+impl Display for EventType {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "EventType - ")?;
         match self {
-            Event::NewDealingCommitment => write!(f, "NewDealingCommitment"),
+            EventType::NewDealerIdentity { details } => {
+                write!(f, "NewDealerIdentity for {}", details.address)
+            }
+            EventType::NewDealingCommitment => write!(f, "NewDealingCommitment"),
         }
     }
 }

validator-api/src/dkg/smart_contract/watcher/mod.rs

@@ -2,22 +2,93 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::dkg::error::DkgError;
+use crate::dkg::state::StateAccessor;
 use crate::Client;
+use coconut_dkg_common::types::EpochState;
 use log::warn;
 use std::time::Duration;
 use tokio::time::interval;
 
-mod event;
+pub(crate) use event::{Event, EventType};
+use validator_client::nymd::SigningCosmWasmClient;
 
-pub(crate) use event::Event;
+mod event;
 
 pub(crate) struct Watcher<C> {
     client: Client<C>,
     polling_rate: Duration,
+    state_accessor: StateAccessor,
 }
 
-impl<C> Watcher<C> {
+impl<C> Watcher<C>
+where
+    C: SigningCosmWasmClient + Send + Sync,
+{
+    async fn try_get_new_dealers(&self) -> Result<(), DkgError> {
+        let known_dealers = self.state_accessor.get_known_dealers().await;
+        let bad_dealers = self.state_accessor.get_malformed_dealers().await;
+
+        let current_height = self.client.current_block_height().await?.value();
+        let current_dealers = self.client.get_current_dealers().await?;
+        for dealer in current_dealers {
+            let is_bad = bad_dealers.contains(&dealer.address);
+            // ideally we would have been accessing the hashmap by address, but it would make
+            // other parts inconvenient. However, we're extremely unlikely to have more than 100 dealers
+            // anyway, so the overhead of iterating over the entire map is minimal
+            let is_known = known_dealers
+                .values()
+                .any(|known_dealer| known_dealer.chain_address == dealer.address);
+
+            // we had absolutely no idea about this dealer existing
+            if !is_bad || !is_known {
+                let watcher_event = self::Event::new(
+                    current_height,
+                    EventType::NewDealerIdentity { details: dealer },
+                );
+                self.state_accessor
+                    .push_contract_change_event(watcher_event)
+                    .await;
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn perform_epoch_state_based_actions(&self, state: EpochState) -> Result<(), DkgError> {
+        match state {
+            EpochState::PublicKeySubmission { .. } => self.try_get_new_dealers().await,
+            EpochState::DealingExchange { .. } => todo!(),
+            EpochState::ComplaintSubmission { .. } => todo!(),
+            EpochState::ComplaintVoting { .. } => todo!(),
+            EpochState::VerificationKeySubmission { .. } => todo!(),
+            EpochState::VerificationKeyMismatchSubmission { .. } => todo!(),
+            EpochState::VerificationKeyMismatchVoting { .. } => todo!(),
+            EpochState::InProgress { .. } => todo!(),
+        }
+    }
+
     async fn poll_contract(&self) -> Result<(), DkgError> {
+        // based on the current epoch state (assuming it HASN'T CHANGED since last check), the following further actions have to be performed:
+        // (if the epoch state changed, we have to ALSO perform actions as if it was in the previous variants):
+
+        // 1. PublicKeySubmission -> get keys of all submitted dealers and if there are any new ones, update dkg state
+        // 2. DealingExchange -> get commitments to dealings and if there are any new ones, update dkg state
+        // 3. ComplaintSubmission -> look for any complaints and if there is any, grab it and emit an event
+        // 4. ComplaintVoting -> grab information about any votes and if exist, emit an event
+        // 5. VerificationKeySubmission -> get list of who submitted their verification keys and if there are any new ones either update state or emit an event
+        // 6. VerificationKeyMismatchSubmission -> look for any complaints and if there is any, grab it and emit an event,
+        // 7. VerificationKeyMismatchVoting -> grab information about any votes and if exist, emit an event
+        // 8. InProgress -> No need to do anything (... I think? unless maybe there was any information about epoch transition, to be determined)
+
+        // figure out what we need to pay attention to (for example if we're in "waiting for complaints" state,
+        // we don't care about identities of potential new dealers just yet)
+        let prior_epoch = self.state_accessor.current_epoch().await;
+        let current_epoch = self.client.get_dkg_epoch().await?;
+
+        if prior_epoch.state != current_epoch.state {
+            todo!()
+        }
+
         Ok(())
     }
 

validator-api/src/dkg/state/accessor.rs

@@ -2,9 +2,11 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::dkg::events::{DispatcherSender, Event};
-use crate::dkg::state::{DkgState, ReceivedDealing};
-use coconut_dkg_common::types::Epoch;
+use crate::dkg::smart_contract::watcher;
+use crate::dkg::state::{Dealer, DkgState, IdentityBytes, ReceivedDealing};
+use coconut_dkg_common::types::{Addr, Epoch};
 use crypto::asymmetric::identity;
+use std::collections::{HashMap, HashSet};
 use std::net::SocketAddr;
 
 // essentially some intermediary that allows either pushing events to the dispatcher or operating
@@ -25,6 +27,11 @@ impl StateAccessor {
         }
     }
 
+    pub(crate) async fn push_contract_change_event(&self, event: watcher::Event) {
+        self.push_event(Event::new_contract_change_event(event))
+            .await
+    }
+
     pub(crate) async fn current_epoch(&self) -> Epoch {
         self.dkg_state.current_epoch().await
     }
@@ -39,4 +46,12 @@ impl StateAccessor {
     pub(crate) async fn is_dealers_remote_address(&self, remote: SocketAddr) -> (bool, Epoch) {
         self.dkg_state.is_dealers_remote_address(remote).await
     }
+
+    pub(crate) async fn get_known_dealers(&self) -> HashMap<IdentityBytes, Dealer> {
+        self.dkg_state.get_known_dealers().await
+    }
+
+    pub(crate) async fn get_malformed_dealers(&self) -> HashSet<Addr> {
+        self.dkg_state.get_malformed_dealers().await
+    }
 }

validator-api/src/dkg/state/mod.rs

@@ -6,12 +6,13 @@ use crypto::asymmetric::identity;
 use dkg::{bte, Dealing};
 use futures::lock::Mutex;
 use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::net::SocketAddr;
 use std::sync::Arc;
 
 mod accessor;
 
+use crate::dkg::error::DkgError;
 pub(crate) use accessor::StateAccessor;
 
 type IdentityBytes = [u8; identity::PUBLIC_KEY_LENGTH];
@@ -81,12 +82,17 @@ pub struct ReceivedDealing {
 
 #[derive(Debug, Serialize, Deserialize)]
 struct DkgStateInner {
+    last_seen_height: BlockHeight,
     bte_decryption_key: bte::DecryptionKey,
     signing_key: identity::PublicKey,
 
     current_epoch: Epoch,
 
     expected_epoch_dealing_digests: HashMap<IdentityBytes, [u8; 32]>,
+
+    // we need to keep track of all bad dealers as well so that we wouldn't attempt to compalaint about them
+    // repeatedly
+    bad_dealers: HashSet<Addr>,
     current_epoch_dealers: HashMap<IdentityBytes, Dealer>,
     verified_epoch_dealings: HashMap<IdentityBytes, ReceivedDealing>,
     unconfirmed_dealings: HashMap<IdentityBytes, ReceivedDealing>,
@@ -126,4 +132,12 @@ impl DkgState {
             .get(&dealer.to_bytes())
             .cloned()
     }
+
+    pub(crate) async fn get_known_dealers(&self) -> HashMap<IdentityBytes, Dealer> {
+        self.inner.lock().await.current_epoch_dealers.clone()
+    }
+
+    pub(crate) async fn get_malformed_dealers(&self) -> HashSet<Addr> {
+        self.inner.lock().await.bad_dealers.clone()
+    }
 }

validator-api/src/nymd_client.rs

@@ -22,8 +22,8 @@ use validator_client::nymd::error::NymdError;
 use validator_client::nymd::traits::DkgClient;
 use validator_client::nymd::{
     hash::{Hash, SHA256_HASH_SIZE},
-    CosmWasmClient, CosmosCoin, Fee, QueryNymdClient, SigningCosmWasmClient, SigningNymdClient,
-    TendermintTime,
+    CosmWasmClient, CosmosCoin, Fee, Height, QueryNymdClient, SigningCosmWasmClient,
+    SigningNymdClient, TendermintTime,
 };
 use validator_client::ValidatorClientError;
 
@@ -119,6 +119,15 @@ impl<C> Client<C> {
         Ok(time)
     }
 
+    pub(crate) async fn current_block_height(&self) -> Result<Height, ValidatorClientError>
+    where
+        C: CosmWasmClient + Sync,
+    {
+        let height = self.0.read().await.nymd.get_current_block_height().await?;
+
+        Ok(height)
+    }
+
     pub(crate) async fn get_mixnodes(&self) -> Result<Vec<MixNodeBond>, ValidatorClientError>
     where
         C: CosmWasmClient + Sync,