Changes to serialization + playing around with the concept of event dispatcher

validator-api/src/dkg/dealing_processing.rs

@@ -0,0 +1,41 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::dkg::networking::message::NewDealingMessage;
+use crate::dkg::state::DkgState;
+use futures::channel::mpsc;
+use futures::StreamExt;
+use log::error;
+
+// Once the DKG epoch begins, all parties will begin exchanging dealings with each other.
+// We really don't want to be processing all of them in parallel since we would starve other
+// parts of the process of CPU. Hence we put them in a queue and process each of them one by one.
+pub(crate) type DealingReceiver = mpsc::UnboundedReceiver<NewDealingMessage>;
+pub(crate) type DealingSender = mpsc::UnboundedSender<NewDealingMessage>;
+
+pub(crate) struct Processor {
+    dkg_state: DkgState,
+    receiver: DealingReceiver,
+}
+impl Processor {
+    pub(crate) fn new(dkg_state: DkgState, receiver: DealingReceiver) -> Self {
+        Processor {
+            dkg_state,
+            receiver,
+        }
+    }
+
+    async fn process_dealing(&self, dealing: NewDealingMessage) {
+        todo!()
+    }
+
+    pub(crate) async fn run(&mut self) {
+        while let Some(dealing) = self.receiver.next().await {
+            self.process_dealing(dealing).await
+        }
+
+        // since we have no graceful shutdowns, seeing this error means something bad has happened
+        // as all senders got dropped
+        error!("")
+    }
+}

validator-api/src/dkg/error.rs

@@ -1,7 +1,6 @@
 // Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::dkg::networking::message::InvalidDkgMessageType;
 use std::io;
 use thiserror::Error;
 use validator_client::ValidatorClientError;
@@ -17,12 +16,9 @@ pub enum DkgError {
     #[error("Networking error - {0}")]
     Networking(#[from] io::Error),
 
+    #[error("Failed to serialize message - {0}")]
+    SerializationError(#[from] bincode::Error),
+
     #[error("todo")]
     DeserializationError,
 }
-
-impl From<InvalidDkgMessageType> for DkgError {
-    fn from(err: InvalidDkgMessageType) -> Self {
-        todo!("figure out how it would fit in the DeserializationError")
-    }
-}

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

@@ -0,0 +1,39 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::dkg::dealing_processing::DealingSender;
+use crate::dkg::events::Event;
+use futures::channel::mpsc;
+use futures::StreamExt;
+use log::error;
+
+pub(crate) type DispatcherSender = mpsc::UnboundedSender<Event>;
+pub(crate) type DispatcherReceiver = mpsc::UnboundedReceiver<Event>;
+
+pub(crate) struct Dispatcher {
+    event_receiver: DispatcherReceiver,
+
+    dealing_processor: DealingSender,
+}
+
+impl Dispatcher {
+    fn handle_event(&self, event: Event) {
+        match event {
+            Event::NewDealing(new_dealing_request) => self
+                .dealing_processor
+                .unbounded_send(new_dealing_request)
+                .expect("failed to forward new dealing message"),
+            _ => todo!(),
+        }
+    }
+
+    pub(crate) async fn run(&mut self) {
+        while let Some(new_event) = self.event_receiver.next().await {
+            self.handle_event(new_event)
+        }
+
+        // since we have no graceful shutdowns, seeing this error means something bad has happened
+        // as all senders got dropped
+        error!("")
+    }
+}

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

@@ -0,0 +1,10 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::dkg::networking::message::{NewDealingMessage, RemoteDealingRequestMessage};
+
+#[derive(Debug)]
+pub(crate) enum Event {
+    NewDealing(NewDealingMessage),
+    NewDealingRequest(RemoteDealingRequestMessage),
+}

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

@@ -0,0 +1,8 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+mod dispatcher;
+mod event;
+
+pub(crate) use dispatcher::DispatcherSender;
+pub(crate) use event::Event;

validator-api/src/dkg/mod.rs

@@ -5,3 +5,5 @@ mod contract_watcher;
 pub(crate) mod error;
 pub(crate) mod networking;
 pub(crate) mod state;
+mod dealing_processing;
+pub(crate) mod events;

validator-api/src/dkg/networking/codec.rs

@@ -16,8 +16,7 @@ impl Encoder<OffchainDkgMessage> for DkgCodec {
     type Error = DkgError;
 
     fn encode(&mut self, item: OffchainDkgMessage, dst: &mut BytesMut) -> Result<(), Self::Error> {
-        item.encode(dst);
-        Ok(())
+        item.encode(dst)
     }
 }
 
@@ -54,9 +53,6 @@ impl Decoder for DkgCodec {
         let payload = src[Header::LEN..Header::LEN + header.payload_length as usize].to_vec();
         src.advance(Header::LEN + header.payload_length as usize);
 
-        Ok(Some(OffchainDkgMessage::try_from_bytes(
-            payload,
-            header.message_type,
-        )?))
+        Ok(Some(OffchainDkgMessage::try_from_bytes(payload)?))
     }
 }

validator-api/src/dkg/networking/handler.rs

@@ -1,45 +1,118 @@
 // Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use crate::dkg::events::DispatcherSender;
 use crate::dkg::networking::codec::DkgCodec;
-use crate::dkg::networking::message::{ErrorReason, OffchainDkgMessage};
+use crate::dkg::networking::message::{
+    ErrorReason, NewDealingMessage, OffchainDkgMessage, RemoteDealingRequestMessage,
+};
 use crate::dkg::state::DkgState;
-use futures::StreamExt;
+use futures::{SinkExt, StreamExt};
 use std::net::SocketAddr;
 use std::time::Duration;
 use tokio::net::TcpStream;
 use tokio::time::timeout;
 use tokio_util::codec::Framed;
 
-const DEFAULT_MAX_CONNECTION_DURATION: Duration = Duration::new(2 * 60, 0);
+const DEFAULT_MAX_CONNECTION_DURATION: Duration = Duration::new(2 * 60 * 60, 0);
 
 #[derive(Debug)]
 pub(crate) struct ConnectionHandler {
     // connection cannot exist for more than this time
     max_connection_duration: Duration,
+    dispatcher_sender: DispatcherSender,
     dkg_state: DkgState,
     conn: Framed<TcpStream, DkgCodec>,
     remote: SocketAddr,
 }
 
 impl ConnectionHandler {
-    pub(crate) fn new(dkg_state: DkgState, conn: TcpStream, remote: SocketAddr) -> Self {
+    pub(crate) fn new(
+        dispatcher_sender: DispatcherSender,
+        dkg_state: DkgState,
+        conn: TcpStream,
+        remote: SocketAddr,
+    ) -> Self {
         ConnectionHandler {
             max_connection_duration: DEFAULT_MAX_CONNECTION_DURATION,
+            dispatcher_sender,
             dkg_state,
             remote,
             conn: Framed::new(conn, DkgCodec),
         }
     }
 
+    async fn send_response(&mut self, response_message: OffchainDkgMessage) {
+        self.conn.send(response_message).await;
+    }
+
     async fn send_error_response<S: Into<String>>(
-        &mut self,
+        &self,
         error: ErrorReason,
         additional_info: Option<S>,
     ) {
         //
     }
 
+    async fn handle_new_dealing(&self, id: u64, message: NewDealingMessage) {
+        todo!()
+    }
+
+    async fn handle_remote_dealing_request(
+        &mut self,
+        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;
+        if current_epoch.id != message.epoch_id {
+            return self
+                .send_error_response(
+                    ErrorReason::InvalidEpoch,
+                    Some(format!(
+                        "current epoch is {} and not {}",
+                        current_epoch.id, message.epoch_id
+                    )),
+                )
+                .await;
+        }
+
+        let dealing = self.dkg_state.get_verified_dealing(message.dealer).await;
+
+        self.send_response(todo!()).await;
+        todo!()
+    }
+
+    async fn handle_request(&mut self, request: OffchainDkgMessage) {
+        match request {
+            OffchainDkgMessage::NewDealing { id, message } => {
+                self.handle_new_dealing(id, message).await
+            }
+            OffchainDkgMessage::RemoteDealingRequest { id, message } => {
+                self.handle_remote_dealing_request(id, message).await
+            }
+            OffchainDkgMessage::RemoteDealingResponse { .. } => {
+                self.send_error_response(
+                    ErrorReason::InvalidRequest,
+                    Some("RemoteDealingResponse is not a valid request type"),
+                )
+                .await
+            }
+            OffchainDkgMessage::ErrorResponse { .. } => {
+                self.send_error_response(
+                    ErrorReason::InvalidRequest,
+                    Some("ErrorResponse is not a valid request type"),
+                )
+                .await
+            }
+        }
+    }
+
     async fn _handle_connection(&mut self) {
         debug!("Starting connection handler for {}", self.remote);
 
@@ -92,13 +165,4 @@ impl ConnectionHandler {
             .await;
         }
     }
-
-    async fn handle_request(&self, request: OffchainDkgMessage) {
-        match request {
-            OffchainDkgMessage::NewDealing { .. } => {}
-            OffchainDkgMessage::RemoteDealingRequest { .. } => {}
-            OffchainDkgMessage::RemoteDealingResponse { .. } => {}
-            OffchainDkgMessage::ErrorResponse { .. } => {}
-        }
-    }
 }

validator-api/src/dkg/networking/listener.rs

@@ -24,7 +24,7 @@ impl<A> Listener<A> {
 
     fn on_connect(&self, conn: TcpStream, remote: SocketAddr) {
         tokio::spawn(
-            ConnectionHandler::new(self.dkg_state.clone(), conn, remote).handle_connection(),
+            async move { todo!() }, // ConnectionHandler::new(self.dkg_state.clone(), conn, remote).handle_connection(),
         );
     }
 

validator-api/src/dkg/networking/message.rs

@@ -2,11 +2,15 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::dkg::error::DkgError;
+use crate::dkg::networking::PROTOCOL_VERSION;
 use bytes::{BufMut, BytesMut};
 use crypto::asymmetric::identity;
 use dkg::Dealing;
+use serde::{Deserialize, Serialize};
+use std::fmt::{Display, Formatter};
 use std::io;
 
+#[derive(Debug, Serialize, Deserialize)]
 pub enum OffchainDkgMessage {
     NewDealing {
         id: u64,
@@ -26,82 +30,109 @@ pub enum OffchainDkgMessage {
     },
 }
 
+#[derive(Debug, Serialize, Deserialize)]
 pub struct NewDealingMessage {
-    epoch_id: u32,
+    pub epoch_id: u32,
     // we keep the dealing in its serialized state as that's what is being signed (and hashed)
     // so that it's easier to verify
-    dealing_bytes: Vec<u8>,
-    dealer_signature: identity::Signature,
+    pub dealing_bytes: Vec<u8>,
+    pub dealer_signature: identity::Signature,
 }
 
+#[derive(Debug, Serialize, Deserialize)]
 pub struct RemoteDealingRequestMessage {
-    epoch_id: u32,
-    dealer: identity::PublicKey,
+    pub epoch_id: u32,
+    pub dealer: identity::PublicKey,
 }
 
+#[derive(Debug, Serialize, Deserialize)]
 pub enum RemoteDealingResponseMessage {
     Available {
         epoch_id: u32,
+        #[serde(with = "dealing_bytes")]
         dealing: Box<Dealing>,
         dealer_signature: identity::Signature,
     },
     Unavailable,
 }
 
-pub struct ErrorResponseMessage {
-    reason: ErrorReason,
-    additional_info: Option<String>,
+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)]
+pub struct ErrorResponseMessage {
+    pub reason: ErrorReason,
+    pub additional_info: Option<String>,
+}
+
+impl ErrorResponseMessage {
+    pub fn new(reason: ErrorReason, additional_info: Option<String>) -> Self {
+        ErrorResponseMessage {
+            reason,
+            additional_info,
+        }
+    }
+}
+
+impl Display for ErrorResponseMessage {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        todo!()
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
 pub enum ErrorReason {
+    InvalidEpoch,
     UnknownDealer,
+    InvalidRequest,
     Timeout,
 }
 
+impl Display for ErrorReason {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        todo!()
+    }
+}
+
 impl OffchainDkgMessage {
-    fn frame(self) -> FramedOffchainDkgMessage {
-        todo!()
+    pub(crate) fn try_from_bytes(bytes: Vec<u8>) -> Result<Self, DkgError> {
+        Ok(bincode::deserialize(&bytes)?)
     }
 
-    pub(crate) fn encode(self, dst: &mut BytesMut) {
-        dst.put(self.frame().into_bytes().as_ref());
+    pub(crate) fn try_to_bytes(&self) -> Result<Vec<u8>, DkgError> {
+        Ok(bincode::serialize(&self)?)
     }
 
-    pub(crate) fn try_from_bytes(
-        bytes: Vec<u8>,
-        expected_type: OffchainDkgMessageType,
-    ) -> Result<Self, DkgError> {
-        todo!()
+    fn frame(self) -> Result<FramedOffchainDkgMessage, DkgError> {
+        let payload = self.try_to_bytes()?;
+        Ok(FramedOffchainDkgMessage {
+            header: Header {
+                payload_length: payload.len() as u64,
+                protocol_version: PROTOCOL_VERSION,
+            },
+            payload,
+        })
     }
-}
 
-#[repr(u8)]
-#[derive(Clone, Copy, Debug, PartialEq)]
-pub(crate) enum OffchainDkgMessageType {
-    NewDealing = 0,
-    RemoteDealingRequest = 1,
-
-    RemoteDealingResponse = 128,
-    ErrorResponse = 255,
-}
-
-pub struct InvalidDkgMessageType(u8);
-
-impl TryFrom<u8> for OffchainDkgMessageType {
-    type Error = InvalidDkgMessageType;
-
-    fn try_from(value: u8) -> std::result::Result<Self, Self::Error> {
-        match value {
-            _ if value == (OffchainDkgMessageType::NewDealing as u8) => Ok(Self::NewDealing),
-            _ if value == (OffchainDkgMessageType::RemoteDealingRequest as u8) => {
-                Ok(Self::RemoteDealingRequest)
-            }
-            _ if value == (OffchainDkgMessageType::RemoteDealingResponse as u8) => {
-                Ok(Self::RemoteDealingResponse)
-            }
-            _ if value == (OffchainDkgMessageType::ErrorResponse as u8) => Ok(Self::ErrorResponse),
-            t => Err(InvalidDkgMessageType(t)),
-        }
+    pub(crate) fn encode(self, dst: &mut BytesMut) -> Result<(), DkgError> {
+        dst.put(self.frame()?.into_bytes().as_ref());
+        Ok(())
     }
 }
 
@@ -127,17 +158,15 @@ impl FramedOffchainDkgMessage {
 
 #[derive(Debug, Copy, Clone, PartialEq)]
 pub(crate) struct Header {
-    pub(crate) message_type: OffchainDkgMessageType,
     pub(crate) payload_length: u64,
     pub(crate) protocol_version: u32,
 }
 
 impl Header {
-    pub(crate) const LEN: usize = 13;
+    pub(crate) const LEN: usize = 12;
 
     pub(crate) fn into_bytes(self) -> Vec<u8> {
         let mut out = Vec::with_capacity(Self::LEN);
-        out.push(self.message_type as u8);
         out.extend_from_slice(&self.payload_length.to_be_bytes());
         out.extend_from_slice(&self.protocol_version.to_be_bytes());
 
@@ -157,9 +186,8 @@ impl Header {
             )));
         }
         Ok(Header {
-            message_type: OffchainDkgMessageType::try_from(bytes[0])?,
-            payload_length: u64::from_be_bytes(bytes[1..9].try_into().unwrap()),
-            protocol_version: u32::from_be_bytes(bytes[9..].try_into().unwrap()),
+            payload_length: u64::from_be_bytes(bytes[..8].try_into().unwrap()),
+            protocol_version: u32::from_be_bytes(bytes[8..].try_into().unwrap()),
         })
     }
 }
@@ -172,7 +200,6 @@ mod tests {
     #[test]
     fn header_deserialization() {
         let valid_header = Header {
-            message_type: OffchainDkgMessageType::NewDealing,
             payload_length: 1234,
             protocol_version: PROTOCOL_VERSION,
         };

validator-api/src/dkg/state.rs

@@ -42,7 +42,6 @@ struct DkgStateInner {
 
 impl DkgState {
     // some save/load action here
-
     pub(crate) async fn is_dealers_remote_address(&self, remote: SocketAddr) -> bool {
         let dealers = &self.inner.lock().await.current_epoch_dealers;
         dealers
@@ -50,6 +49,10 @@ impl DkgState {
             .any(|dealer| dealer.remote_address == remote)
     }
 
+    pub(crate) async fn current_epoch(&self) -> Epoch {
+        self.inner.lock().await.current_epoch
+    }
+
     pub(crate) async fn get_verified_dealing(
         &self,
         dealer: identity::PublicKey,
@@ -62,7 +65,3 @@ impl DkgState {
             .cloned()
     }
 }
-
-impl DkgStateInner {
-    //
-}