From 5fb7ebeb63fa9f3bb06637c6031fe6c52133d0ce Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 5 May 2022 12:17:41 +0100 Subject: [PATCH] Reorganised accessing state --- common/crypto/src/asymmetric/identity/mod.rs | 2 +- validator-api/src/dkg/events/event.rs | 9 ++++ validator-api/src/dkg/networking/handler.rs | 44 +++++++++---------- validator-api/src/dkg/networking/message.rs | 41 ++++++----------- validator-api/src/dkg/state/accessor.rs | 42 ++++++++++++++++++ .../src/dkg/{state.rs => state/mod.rs} | 40 +++++++++++++++-- 6 files changed, 123 insertions(+), 55 deletions(-) create mode 100644 validator-api/src/dkg/state/accessor.rs rename validator-api/src/dkg/{state.rs => state/mod.rs} (61%) diff --git a/common/crypto/src/asymmetric/identity/mod.rs b/common/crypto/src/asymmetric/identity/mod.rs index e3c21469e0..2a03d50042 100644 --- a/common/crypto/src/asymmetric/identity/mod.rs +++ b/common/crypto/src/asymmetric/identity/mod.rs @@ -266,7 +266,7 @@ impl PemStorableKey for PrivateKey { } } -#[derive(Debug)] +#[derive(Copy, Clone, Debug)] pub struct Signature(ed25519_dalek::Signature); impl Signature { diff --git a/validator-api/src/dkg/events/event.rs b/validator-api/src/dkg/events/event.rs index dac1d66d5e..afd0f388e6 100644 --- a/validator-api/src/dkg/events/event.rs +++ b/validator-api/src/dkg/events/event.rs @@ -8,3 +8,12 @@ pub(crate) enum Event { NewDealing(NewDealingMessage), NewDealingRequest(RemoteDealingRequestMessage), } + +impl Event { + pub(crate) fn name(&self) -> String { + match self { + Event::NewDealing(..) => "NewDealing".to_string(), + Event::NewDealingRequest(..) => "NewDealingRequest".to_string(), + } + } +} diff --git a/validator-api/src/dkg/networking/handler.rs b/validator-api/src/dkg/networking/handler.rs index f2713661fd..805ccd38d8 100644 --- a/validator-api/src/dkg/networking/handler.rs +++ b/validator-api/src/dkg/networking/handler.rs @@ -1,12 +1,12 @@ // Copyright 2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::dkg::events::DispatcherSender; +use crate::dkg::events::Event; use crate::dkg::networking::codec::DkgCodec; use crate::dkg::networking::message::{ ErrorResponseMessage, NewDealingMessage, OffchainDkgMessage, RemoteDealingRequestMessage, }; -use crate::dkg::state::DkgState; +use crate::dkg::state::StateAccessor; use futures::{SinkExt, StreamExt}; use std::net::SocketAddr; use std::time::Duration; @@ -20,23 +20,16 @@ const DEFAULT_MAX_CONNECTION_DURATION: Duration = Duration::from_secs(2 * 60 * 6 pub(crate) struct ConnectionHandler { // connection cannot exist for more than this time max_connection_duration: Duration, - dispatcher_sender: DispatcherSender, - dkg_state: DkgState, + state_accessor: StateAccessor, conn: Framed, remote: SocketAddr, } impl ConnectionHandler { - pub(crate) fn new( - dispatcher_sender: DispatcherSender, - dkg_state: DkgState, - conn: TcpStream, - remote: SocketAddr, - ) -> Self { + pub(crate) fn new(state_accessor: StateAccessor, conn: TcpStream, remote: SocketAddr) -> Self { ConnectionHandler { max_connection_duration: DEFAULT_MAX_CONNECTION_DURATION, - dispatcher_sender, - dkg_state, + state_accessor, remote, conn: Framed::new(conn, DkgCodec), } @@ -52,7 +45,10 @@ impl ConnectionHandler { } async fn handle_new_dealing(&self, id: u64, message: NewDealingMessage) { - todo!() + // we'll probably need to wait for a response or something, so this is rather temporary + self.state_accessor + .push_event(Event::NewDealing(message)) + .await } async fn handle_remote_dealing_request( @@ -60,13 +56,7 @@ impl ConnectionHandler { id: u64, message: RemoteDealingRequestMessage, ) { - // TODO: when somebody is reviewing this code, what's your opinion on accessing the DkgState here - // vs keeping it slightly more consistent and dispatching an event to request the value from - // something managing it instead? - - // personal note: once more parts are developed, I might change it myself before it even gets to the PR state - - let current_epoch = self.dkg_state.current_epoch().await; + let current_epoch = self.state_accessor.current_epoch().await; if current_epoch.id != message.epoch_id { return self .send_error_response( @@ -79,10 +69,13 @@ impl ConnectionHandler { .await; } - let dealing = self.dkg_state.get_verified_dealing(message.dealer).await; + let dealing = self + .state_accessor + .get_verified_dealing(message.dealer) + .await; - self.send_response(todo!()).await; - todo!() + self.send_response(OffchainDkgMessage::new_remote_dealing_response(id, dealing)) + .await; } async fn handle_request(&mut self, request: OffchainDkgMessage) { @@ -117,7 +110,10 @@ impl ConnectionHandler { async fn _handle_connection(&mut self) { debug!("Starting connection handler for {}", self.remote); - let (is_dealer, epoch) = self.dkg_state.is_dealers_remote_address(self.remote).await; + let (is_dealer, epoch) = self + .state_accessor + .is_dealers_remote_address(self.remote) + .await; if !is_dealer { warn!( "Received a request from an unknown dealer - {}", diff --git a/validator-api/src/dkg/networking/message.rs b/validator-api/src/dkg/networking/message.rs index d5b56dafbd..e5bade5e2c 100644 --- a/validator-api/src/dkg/networking/message.rs +++ b/validator-api/src/dkg/networking/message.rs @@ -3,11 +3,10 @@ use crate::dkg::error::DkgError; use crate::dkg::networking::PROTOCOL_VERSION; +use crate::dkg::state::ReceivedDealing; use bytes::{BufMut, BytesMut}; use crypto::asymmetric::identity; -use dkg::Dealing; use serde::{Deserialize, Serialize}; -use std::fmt::{Display, Formatter}; use std::io; use std::net::SocketAddr; use std::time::Duration; @@ -50,34 +49,10 @@ pub struct RemoteDealingRequestMessage { #[derive(Debug, Serialize, Deserialize)] pub enum RemoteDealingResponseMessage { - Available { - epoch_id: u32, - #[serde(with = "dealing_bytes")] - dealing: Box, - dealer_signature: identity::Signature, - }, + Available { dealing: ReceivedDealing }, Unavailable, } -mod dealing_bytes { - use dkg::Dealing; - use serde::de::Error as SerdeError; - use serde::{Deserialize, Deserializer, Serialize, Serializer}; - use serde_bytes::{ByteBuf as SerdeByteBuf, Bytes as SerdeBytes}; - - pub fn serialize(val: &Dealing, serializer: S) -> Result { - SerdeBytes::new(&val.to_bytes()).serialize(serializer) - } - - pub fn deserialize<'de, D: Deserializer<'de>>( - deserializer: D, - ) -> Result, D::Error> { - let bytes = ::deserialize(deserializer)?; - let dealing = Dealing::try_from_bytes(bytes.as_ref()).map_err(SerdeError::custom)?; - Ok(Box::new(dealing)) - } -} - #[derive(Debug, Serialize, Deserialize, Error)] pub enum ErrorResponseMessage { #[error("Received request for epoch: {requested}, while the current epoch is {current}")] @@ -104,6 +79,18 @@ impl OffchainDkgMessage { OffchainDkgMessage::ErrorResponse { id, message } } + pub(crate) fn new_remote_dealing_response( + id: u64, + dealing: Option, + ) -> OffchainDkgMessage { + let message = match dealing { + Some(dealing) => RemoteDealingResponseMessage::Available { dealing }, + None => RemoteDealingResponseMessage::Unavailable, + }; + + OffchainDkgMessage::RemoteDealingResponse { id, message } + } + pub(crate) fn try_from_bytes(bytes: Vec) -> Result { Ok(bincode::deserialize(&bytes)?) } diff --git a/validator-api/src/dkg/state/accessor.rs b/validator-api/src/dkg/state/accessor.rs new file mode 100644 index 0000000000..9f5bd74d68 --- /dev/null +++ b/validator-api/src/dkg/state/accessor.rs @@ -0,0 +1,42 @@ +// Copyright 2022 - Nym Technologies SA +// 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 crypto::asymmetric::identity; +use std::net::SocketAddr; + +// essentially some intermediary that allows either pushing events to the dispatcher or operating +// directly on the dkg state +// note: it should get renamed, but at the time of writing it, I couldn't come up with anything... +#[derive(Debug, Clone)] +pub(crate) struct StateAccessor { + dkg_state: DkgState, + dispatcher_sender: DispatcherSender, +} + +impl StateAccessor { + pub(crate) async fn push_event(&self, event: Event) { + if let Err(err) = self.dispatcher_sender.unbounded_send(event) { + log::error!("Our event dispatcher failed to receive {} event - it has presumably crashed. Shutting down the API after saving DKG state", err.into_inner().name()); + self.dkg_state.save().await; + std::process::exit(1); + } + } + + pub(crate) async fn current_epoch(&self) -> Epoch { + self.dkg_state.current_epoch().await + } + + pub(crate) async fn get_verified_dealing( + &self, + dealer: identity::PublicKey, + ) -> Option { + self.dkg_state.get_verified_dealing(dealer).await + } + + pub(crate) async fn is_dealers_remote_address(&self, remote: SocketAddr) -> (bool, Epoch) { + self.dkg_state.is_dealers_remote_address(remote).await + } +} diff --git a/validator-api/src/dkg/state.rs b/validator-api/src/dkg/state/mod.rs similarity index 61% rename from validator-api/src/dkg/state.rs rename to validator-api/src/dkg/state/mod.rs index ff367c2ff3..644c9630b0 100644 --- a/validator-api/src/dkg/state.rs +++ b/validator-api/src/dkg/state/mod.rs @@ -10,6 +10,10 @@ use std::collections::HashMap; use std::net::SocketAddr; use std::sync::Arc; +mod accessor; + +pub(crate) use accessor::StateAccessor; + type IdentityBytes = [u8; identity::PUBLIC_KEY_LENGTH]; // TODO: some TryFrom impl to convert from encoded contract data @@ -27,6 +31,33 @@ pub(crate) struct DkgState { inner: Arc>, } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ReceivedDealing { + epoch_id: u32, + #[serde(with = "dealing_bytes")] + dealing: Box, + signature: identity::Signature, +} + +mod dealing_bytes { + use dkg::Dealing; + use serde::de::Error as SerdeError; + use serde::{Deserialize, Deserializer, Serialize, Serializer}; + use serde_bytes::{ByteBuf as SerdeByteBuf, Bytes as SerdeBytes}; + + pub fn serialize(val: &Dealing, serializer: S) -> Result { + SerdeBytes::new(&val.to_bytes()).serialize(serializer) + } + + pub fn deserialize<'de, D: Deserializer<'de>>( + deserializer: D, + ) -> Result, D::Error> { + let bytes = ::deserialize(deserializer)?; + let dealing = Dealing::try_from_bytes(bytes.as_ref()).map_err(SerdeError::custom)?; + Ok(Box::new(dealing)) + } +} + #[derive(Debug)] struct DkgStateInner { bte_decryption_key: bte::DecryptionKey, @@ -36,12 +67,15 @@ struct DkgStateInner { expected_epoch_dealing_digests: HashMap, current_epoch_dealers: HashMap, - verified_epoch_dealings: HashMap, - unconfirmed_dealings: HashMap, + verified_epoch_dealings: HashMap, + unconfirmed_dealings: HashMap, } impl DkgState { // some save/load action here + pub(crate) async fn save(&self) { + todo!() + } pub(crate) async fn is_dealers_remote_address(&self, remote: SocketAddr) -> (bool, Epoch) { let guard = self.inner.lock().await; @@ -63,7 +97,7 @@ impl DkgState { pub(crate) async fn get_verified_dealing( &self, dealer: identity::PublicKey, - ) -> Option { + ) -> Option { self.inner .lock() .await