From fb09e97eb69109e4be84597571edaee8ece33d2a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 12 May 2022 11:12:02 +0100 Subject: [PATCH] Logic for processing dealer set changes --- validator-api/src/dkg/main_loop/mod.rs | 83 ++++++++++++- .../src/dkg/smart_contract/watcher/event.rs | 18 ++- .../src/dkg/smart_contract/watcher/mod.rs | 56 ++++++--- validator-api/src/dkg/state/accessor.rs | 4 +- validator-api/src/dkg/state/mod.rs | 117 +++++++++++++++--- 5 files changed, 233 insertions(+), 45 deletions(-) diff --git a/validator-api/src/dkg/main_loop/mod.rs b/validator-api/src/dkg/main_loop/mod.rs index e6b453456f..dc1418e005 100644 --- a/validator-api/src/dkg/main_loop/mod.rs +++ b/validator-api/src/dkg/main_loop/mod.rs @@ -1,12 +1,15 @@ // Copyright 2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use crate::dkg::events::DispatcherSender; use crate::dkg::smart_contract::publisher::Publisher; use crate::dkg::smart_contract::watcher; -use crate::dkg::state::StateAccessor; -use coconut_dkg_common::types::DealerDetails; +use crate::dkg::smart_contract::watcher::{DealerChange, EventType}; +use crate::dkg::state::{Dealer, DkgState, Malformation, MalformedDealer}; +use coconut_dkg_common::types::{Addr, BlockHeight, DealerDetails}; use futures::channel::mpsc; use futures::StreamExt; +use log::info; // essentially events originating from the contract watcher could only drive our state forward // (TODO: is it actually true? I guess we'll find out soon enough) @@ -15,7 +18,10 @@ pub(crate) type ContractEventsSender = mpsc::UnboundedSender; // it is driven by events received from the contract watcher pub(crate) struct ProcessingLoop { - state_accessor: StateAccessor, + // technically we could have used `StateAccessor` here as well, but I really want to keep + // `StateAccessor` pure from anything that would have to modify `dkg_state` + dkg_state: DkgState, + dispatcher_sender: DispatcherSender, contract_events_receiver: ContractEventsReceiver, contract_publisher: Publisher, @@ -23,12 +29,77 @@ pub(crate) struct ProcessingLoop { } impl ProcessingLoop { - fn verify_dealer(&self, contract_dealer_details: &DealerDetails) { - // + async fn raise_malformed_dealer_complaint( + &self, + dealer_address: Addr, + malformation: Malformation, + ) { + info!( + "here we would be raising complaint about dealer {} being malformed: {:?}", + dealer_address, malformation + ); + // todo!() + } + + async fn deal_with_new_dealer(&self, dealer: Dealer) { + if dealer.bte_public_key.verify() { + self.dkg_state.try_add_new_dealer(dealer).await + } else { + let dealer_address = dealer.chain_address.clone(); + // the dealer failed to provide valid proof of possession + let malformation = Malformation::InvalidBTEPublicKey; + self.dkg_state + .try_add_malformed_dealer(MalformedDealer::Parsed(dealer)) + .await; + self.raise_malformed_dealer_complaint(dealer_address, malformation) + .await; + } + } + + async fn deal_with_malformed_dealer( + &self, + dealer_details: DealerDetails, + malformation: Malformation, + ) { + let dealer_address = dealer_details.address.clone(); + self.dkg_state + .try_add_malformed_dealer(MalformedDealer::Raw(dealer_details)) + .await; + self.raise_malformed_dealer_complaint(dealer_address, malformation) + .await; + } + + async fn process_new_dealer(&self, dealer_details: DealerDetails) { + match Dealer::try_parse_from_raw(&dealer_details) { + Ok(dealer) => self.deal_with_new_dealer(dealer).await, + Err(malformed_dealer) => { + self.deal_with_malformed_dealer(dealer_details, malformed_dealer) + .await + } + } + } + + async fn process_dealer_removal(&self, dealer_address: Addr) { + self.dkg_state.try_remove_dealer(dealer_address).await + } + + async fn process_dealer_changes(&self, changes: Vec, height: BlockHeight) { + for change in changes { + match change { + DealerChange::Addition { details } => self.process_new_dealer(details).await, + DealerChange::Removal { address } => self.process_dealer_removal(address).await, + } + } + self.dkg_state.update_last_seen_height(height).await } async fn process_event(&self, event: watcher::Event) { - todo!() + match event.event_type { + EventType::DealerSetChange { changes } => { + self.process_dealer_changes(changes, event.height).await + } + EventType::NewDealingCommitment => todo!(), + } } pub(crate) async fn run(&mut self) { diff --git a/validator-api/src/dkg/smart_contract/watcher/event.rs b/validator-api/src/dkg/smart_contract/watcher/event.rs index b45e144c00..4d74e9bfcd 100644 --- a/validator-api/src/dkg/smart_contract/watcher/event.rs +++ b/validator-api/src/dkg/smart_contract/watcher/event.rs @@ -1,13 +1,13 @@ // Copyright 2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use coconut_dkg_common::types::{BlockHeight, DealerDetails}; +use coconut_dkg_common::types::{Addr, BlockHeight, DealerDetails}; use std::fmt::{Display, Formatter}; #[derive(Debug)] pub(crate) struct Event { - height: BlockHeight, - event_type: EventType, + pub(crate) height: BlockHeight, + pub(crate) event_type: EventType, } impl Event { @@ -26,9 +26,15 @@ impl Display for Event { } } +#[derive(Debug)] +pub(crate) enum DealerChange { + Addition { details: DealerDetails }, + Removal { address: Addr }, +} + #[derive(Debug)] pub(crate) enum EventType { - NewDealerIdentity { details: DealerDetails }, + DealerSetChange { changes: Vec }, NewDealingCommitment, } @@ -36,8 +42,8 @@ impl Display for EventType { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { write!(f, "EventType - ")?; match self { - EventType::NewDealerIdentity { details } => { - write!(f, "NewDealerIdentity for {}", details.address) + EventType::DealerSetChange { changes } => { + write!(f, "DealerSetChange with {} changes", changes.len()) } EventType::NewDealingCommitment => write!(f, "NewDealingCommitment"), } diff --git a/validator-api/src/dkg/smart_contract/watcher/mod.rs b/validator-api/src/dkg/smart_contract/watcher/mod.rs index 0b853f85e2..5237405c94 100644 --- a/validator-api/src/dkg/smart_contract/watcher/mod.rs +++ b/validator-api/src/dkg/smart_contract/watcher/mod.rs @@ -6,10 +6,11 @@ use crate::dkg::state::StateAccessor; use crate::Client; use coconut_dkg_common::types::EpochState; use log::warn; +use std::collections::HashMap; use std::time::Duration; use tokio::time::interval; -pub(crate) use event::{Event, EventType}; +pub(crate) use event::{DealerChange, Event, EventType}; use validator_client::nymd::SigningCosmWasmClient; mod event; @@ -24,39 +25,66 @@ impl Watcher where C: SigningCosmWasmClient + Send + Sync, { - async fn try_get_new_dealers(&self) -> Result<(), DkgError> { + async fn check_for_dealers(&self) -> Result<(), DkgError> { + // get current state let known_dealers = self.state_accessor.get_known_dealers().await; let bad_dealers = self.state_accessor.get_malformed_dealers().await; + // get data from the contract 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); + let contract_dealers = self.client.get_current_dealers().await?; + let contract_dealers = contract_dealers + .into_iter() + .map(|dealer| (dealer.address.clone(), dealer)) + .collect::>(); + + // TODO: this would probably have to get generalised since we'd need to use the same logic + // for other possible events + + let mut changes = Vec::new(); + + // check for removed dealers (if our lists contain keys that do not exist in the contract, + // it implies they got purged from there) + for dealer in bad_dealers + .keys() + .chain(known_dealers.values().map(|dealer| &dealer.chain_address)) + { + if !contract_dealers.contains_key(dealer) { + changes.push(DealerChange::Removal { + address: dealer.clone(), + }); + } + } + + // check for new dealers + for (dealer, details) in contract_dealers { + let is_bad = bad_dealers.contains_key(&dealer); // 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); + .any(|known_dealer| known_dealer.chain_address == dealer); // 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; + changes.push(DealerChange::Addition { details }); } } + self.state_accessor + .push_contract_change_event(Event::new( + current_height, + EventType::DealerSetChange { changes }, + )) + .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::PublicKeySubmission { .. } => self.check_for_dealers().await, EpochState::DealingExchange { .. } => todo!(), EpochState::ComplaintSubmission { .. } => todo!(), EpochState::ComplaintVoting { .. } => todo!(), diff --git a/validator-api/src/dkg/state/accessor.rs b/validator-api/src/dkg/state/accessor.rs index ce01308f31..bf94966f73 100644 --- a/validator-api/src/dkg/state/accessor.rs +++ b/validator-api/src/dkg/state/accessor.rs @@ -3,7 +3,7 @@ use crate::dkg::events::{DispatcherSender, Event}; use crate::dkg::smart_contract::watcher; -use crate::dkg::state::{Dealer, DkgState, IdentityBytes, ReceivedDealing}; +use crate::dkg::state::{Dealer, DkgState, IdentityBytes, MalformedDealer, ReceivedDealing}; use coconut_dkg_common::types::{Addr, Epoch}; use crypto::asymmetric::identity; use std::collections::{HashMap, HashSet}; @@ -51,7 +51,7 @@ impl StateAccessor { self.dkg_state.get_known_dealers().await } - pub(crate) async fn get_malformed_dealers(&self) -> HashSet { + pub(crate) async fn get_malformed_dealers(&self) -> HashMap { self.dkg_state.get_malformed_dealers().await } } diff --git a/validator-api/src/dkg/state/mod.rs b/validator-api/src/dkg/state/mod.rs index eb433d1112..9dfc7c0ebb 100644 --- a/validator-api/src/dkg/state/mod.rs +++ b/validator-api/src/dkg/state/mod.rs @@ -5,14 +5,14 @@ use coconut_dkg_common::types::{Addr, BlockHeight, DealerDetails, Epoch, NodeInd use crypto::asymmetric::identity; use dkg::{bte, Dealing}; use futures::lock::Mutex; +use log::error; use serde::{Deserialize, Serialize}; -use std::collections::{HashMap, HashSet}; +use std::collections::HashMap; 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]; @@ -27,8 +27,15 @@ pub(crate) struct Dealer { pub(crate) remote_address: SocketAddr, } +impl Dealer { + pub(crate) fn map_key(&self) -> IdentityBytes { + self.identity.to_bytes() + } +} + // TODO: move it elsewhere and propagate it to the contract -pub enum MalformedDealer { +#[derive(Debug)] +pub enum Malformation { MalformedEd25519PublicKey, MalformedBTEPublicKey, InvalidBTEPublicKey, @@ -36,30 +43,28 @@ pub enum MalformedDealer { } impl Dealer { - pub(crate) fn try_parse_from_raw( - contract_value: DealerDetails, - ) -> Result { + pub(crate) fn try_parse_from_raw(contract_value: &DealerDetails) -> Result { // this should be impossible as the contract must have used this key for signature verification - let identity = identity::PublicKey::from_base58_string(contract_value.ed25519_public_key) - .map_err(|_| MalformedDealer::MalformedEd25519PublicKey)?; + let identity = identity::PublicKey::from_base58_string(&contract_value.ed25519_public_key) + .map_err(|_| Malformation::MalformedEd25519PublicKey)?; - let bte_public_key = bs58::decode(contract_value.bte_public_key_with_proof) + let bte_public_key = bs58::decode(&contract_value.bte_public_key_with_proof) .into_vec() .map(|bytes| bte::PublicKeyWithProof::try_from_bytes(&bytes)) - .map_err(|_| MalformedDealer::MalformedBTEPublicKey)? - .map_err(|_| MalformedDealer::MalformedBTEPublicKey)?; + .map_err(|_| Malformation::MalformedBTEPublicKey)? + .map_err(|_| Malformation::MalformedBTEPublicKey)?; if !bte_public_key.verify() { - return Err(MalformedDealer::InvalidBTEPublicKey); + return Err(Malformation::InvalidBTEPublicKey); } let parsed_host = contract_value .host .parse() - .map_err(|_| MalformedDealer::InvalidHostInformation)?; + .map_err(|_| Malformation::InvalidHostInformation)?; Ok(Dealer { - chain_address: contract_value.address, + chain_address: contract_value.address.clone(), node_index: contract_value.assigned_index, bte_public_key, identity, @@ -68,6 +73,21 @@ impl Dealer { } } +#[derive(Clone, Debug, Serialize, Deserialize)] +pub(crate) enum MalformedDealer { + Raw(DealerDetails), + Parsed(Dealer), +} + +impl MalformedDealer { + pub(crate) fn address(&self) -> &Addr { + match self { + MalformedDealer::Raw(dealer) => &dealer.address, + MalformedDealer::Parsed(dealer) => &dealer.chain_address, + } + } +} + #[derive(Debug, Clone)] pub(crate) struct DkgState { inner: Arc>, @@ -90,9 +110,9 @@ struct DkgStateInner { expected_epoch_dealing_digests: HashMap, - // we need to keep track of all bad dealers as well so that we wouldn't attempt to compalaint about them + // we need to keep track of all bad dealers as well so that we wouldn't attempt to complaint about them // repeatedly - bad_dealers: HashSet, + bad_dealers: HashMap, current_epoch_dealers: HashMap, verified_epoch_dealings: HashMap, unconfirmed_dealings: HashMap, @@ -137,7 +157,70 @@ impl DkgState { self.inner.lock().await.current_epoch_dealers.clone() } - pub(crate) async fn get_malformed_dealers(&self) -> HashSet { + pub(crate) async fn get_malformed_dealers(&self) -> HashMap { self.inner.lock().await.bad_dealers.clone() } + + pub(crate) async fn update_last_seen_height(&self, new_last_seen: BlockHeight) { + self.inner.lock().await.last_seen_height = new_last_seen; + } + + pub(crate) async fn try_add_new_dealer(&self, dealer: Dealer) { + // TODO: perhaps we should panic or something instead since this should have never occured in the first place? + if let Some(old_dealer) = self + .inner + .lock() + .await + .current_epoch_dealers + .insert(dealer.map_key(), dealer) + { + error!( + "We have overwritten {} dealer details", + old_dealer.chain_address + ) + } + } + + pub(crate) async fn try_add_malformed_dealer(&self, dealer_details: MalformedDealer) { + // TODO: perhaps we should panic or something instead since this should have never occured in the first place? + if let Some(old_dealer) = self + .inner + .lock() + .await + .bad_dealers + .insert(dealer_details.address().clone(), dealer_details) + { + error!( + "We have overwritten {} dealer details", + old_dealer.address() + ) + } + } + + pub(crate) async fn try_remove_dealer(&self, dealer_address: Addr) { + let mut guard = self.inner.lock().await; + + // dealer is in either bad dealers or known dealers, never both, + // so if we managed to remove it from the former, we don't need to check the latter + if guard.bad_dealers.remove(&dealer_address).is_none() { + // find storage key associated with the entry we want to remove + let storage_key = guard + .current_epoch_dealers + .values() + .find(|&dealer| dealer.chain_address == dealer_address) + .map(|dealer| dealer.map_key()); + + match storage_key { + Some(key) => { + guard.current_epoch_dealers.remove(&key); + } + // this should be impossible as in order to get to this point we must have learned about + // this dealer existing somewhere in our state! + None => error!( + "We failed to remove {} dealer details as it somehow doesn't exist!", + dealer_address + ), + } + } + } }