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

Reorganised accessing state

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2022-05-05 12:17:41 +01:00
parent 3963aebc97
commit 5fb7ebeb63
6 changed files with 123 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
validator-api/src/dkg/events/event.rs
+9
View File
@@ -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(),
}
}
}
validator-api/src/dkg/networking/handler.rs
+20 -24
View File
@@ -1,12 +1,12 @@
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
// 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<TcpStream, DkgCodec>,
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 - {}",
validator-api/src/dkg/networking/message.rs
+14 -27
View File
@@ -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<Dealing>,
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<S: Serializer>(val: &Dealing, serializer: S) -> Result<S::Ok, S::Error> {
SerdeBytes::new(&val.to_bytes()).serialize(serializer)
}
pub fn deserialize<'de, D: Deserializer<'de>>(
deserializer: D,
) -> Result<Box<Dealing>, D::Error> {
let bytes = <SerdeByteBuf>::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<ReceivedDealing>,
) -> 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<u8>) -> Result<Self, DkgError> {
Ok(bincode::deserialize(&bytes)?)
}
validator-api/src/dkg/state/accessor.rs
+42
View File
@@ -0,0 +1,42 @@
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
// 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<ReceivedDealing> {
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
}
}
validator-api/src/dkg/state.rs → validator-api/src/dkg/state/mod.rs
+37 -3
View File
@@ -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<Mutex<DkgStateInner>>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReceivedDealing {
epoch_id: u32,
#[serde(with = "dealing_bytes")]
dealing: Box<Dealing>,
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<S: Serializer>(val: &Dealing, serializer: S) -> Result<S::Ok, S::Error> {
SerdeBytes::new(&val.to_bytes()).serialize(serializer)
}
pub fn deserialize<'de, D: Deserializer<'de>>(
deserializer: D,
) -> Result<Box<Dealing>, D::Error> {
let bytes = <SerdeByteBuf>::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<IdentityBytes, [u8; 32]>,
current_epoch_dealers: HashMap<IdentityBytes, Dealer>,
verified_epoch_dealings: HashMap<IdentityBytes, Dealing>,
unconfirmed_dealings: HashMap<IdentityBytes, Dealing>,
verified_epoch_dealings: HashMap<IdentityBytes, ReceivedDealing>,
unconfirmed_dealings: HashMap<IdentityBytes, ReceivedDealing>,
}
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<Dealing> {
) -> Option<ReceivedDealing> {
self.inner
.lock()
.await