GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Logic for processing dealer set changes

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2022-05-12 11:12:02 +01:00
parent 1318b4425d
commit fb09e97eb6
5 changed files with 233 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
validator-api/src/dkg/main_loop/mod.rs
+77 -6
View File
@@ -1,12 +1,15 @@
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
// 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<watcher::Event>;
// it is driven by events received from the contract watcher
pub(crate) struct ProcessingLoop<C> {
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<C>,
@@ -23,12 +29,77 @@ pub(crate) struct ProcessingLoop<C> {
}
impl<C> ProcessingLoop<C> {
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<DealerChange>, 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) {
validator-api/src/dkg/smart_contract/watcher/event.rs
+12 -6
View File
@@ -1,13 +1,13 @@
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
// 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<DealerChange> },
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"),
}
validator-api/src/dkg/smart_contract/watcher/mod.rs
+42 -14
View File
@@ -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<C> Watcher<C>
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::<HashMap<_, _>>();
// 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!(),
validator-api/src/dkg/state/accessor.rs
+2 -2
View File
@@ -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<Addr> {
pub(crate) async fn get_malformed_dealers(&self) -> HashMap<Addr, MalformedDealer> {
self.dkg_state.get_malformed_dealers().await
}
}
validator-api/src/dkg/state/mod.rs
+100 -17
View File
@@ -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<Self, MalformedDealer> {
pub(crate) fn try_parse_from_raw(contract_value: &DealerDetails) -> Result<Self, Malformation> {
// 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<Mutex<DkgStateInner>>,
@@ -90,9 +110,9 @@ struct DkgStateInner {
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
// 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<Addr>,
bad_dealers: HashMap<Addr, MalformedDealer>,
current_epoch_dealers: HashMap<IdentityBytes, Dealer>,
verified_epoch_dealings: HashMap<IdentityBytes, ReceivedDealing>,
unconfirmed_dealings: HashMap<IdentityBytes, ReceivedDealing>,
@@ -137,7 +157,70 @@ impl DkgState {
self.inner.lock().await.current_epoch_dealers.clone()
}
pub(crate) async fn get_malformed_dealers(&self) -> HashSet<Addr> {
pub(crate) async fn get_malformed_dealers(&self) -> HashMap<Addr, MalformedDealer> {
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
),
}
}
}
}