From 3963aebc97d0e7877bf573e0446285679f39ec6b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 5 May 2022 11:02:26 +0100 Subject: [PATCH] Simplified error response handling --- validator-api/src/dkg/networking/handler.rs | 66 +++++++++++---------- validator-api/src/dkg/networking/message.rs | 59 ++++++++---------- validator-api/src/dkg/state.rs | 17 ++++-- 3 files changed, 72 insertions(+), 70 deletions(-) diff --git a/validator-api/src/dkg/networking/handler.rs b/validator-api/src/dkg/networking/handler.rs index 0f2369c22c..f2713661fd 100644 --- a/validator-api/src/dkg/networking/handler.rs +++ b/validator-api/src/dkg/networking/handler.rs @@ -4,7 +4,7 @@ use crate::dkg::events::DispatcherSender; use crate::dkg::networking::codec::DkgCodec; use crate::dkg::networking::message::{ - ErrorReason, NewDealingMessage, OffchainDkgMessage, RemoteDealingRequestMessage, + ErrorResponseMessage, NewDealingMessage, OffchainDkgMessage, RemoteDealingRequestMessage, }; use crate::dkg::state::DkgState; use futures::{SinkExt, StreamExt}; @@ -14,7 +14,7 @@ use tokio::net::TcpStream; use tokio::time::timeout; use tokio_util::codec::Framed; -const DEFAULT_MAX_CONNECTION_DURATION: Duration = Duration::new(2 * 60 * 60, 0); +const DEFAULT_MAX_CONNECTION_DURATION: Duration = Duration::from_secs(2 * 60 * 60); #[derive(Debug)] pub(crate) struct ConnectionHandler { @@ -46,12 +46,9 @@ impl ConnectionHandler { self.conn.send(response_message).await; } - async fn send_error_response>( - &self, - error: ErrorReason, - additional_info: Option, - ) { - // + async fn send_error_response(&mut self, id: Option, error: ErrorResponseMessage) { + self.send_response(OffchainDkgMessage::new_error_response(id, error)) + .await } async fn handle_new_dealing(&self, id: u64, message: NewDealingMessage) { @@ -73,11 +70,11 @@ impl ConnectionHandler { 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 - )), + Some(id), + ErrorResponseMessage::InvalidEpoch { + current: current_epoch.id, + requested: message.epoch_id, + }, ) .await; } @@ -96,17 +93,21 @@ impl ConnectionHandler { OffchainDkgMessage::RemoteDealingRequest { id, message } => { self.handle_remote_dealing_request(id, message).await } - OffchainDkgMessage::RemoteDealingResponse { .. } => { + OffchainDkgMessage::RemoteDealingResponse { id, .. } => { self.send_error_response( - ErrorReason::InvalidRequest, - Some("RemoteDealingResponse is not a valid request type"), + Some(id), + ErrorResponseMessage::InvalidRequest { + typ: "RemoteDealingResponse".into(), + }, ) .await } - OffchainDkgMessage::ErrorResponse { .. } => { + OffchainDkgMessage::ErrorResponse { id, .. } => { self.send_error_response( - ErrorReason::InvalidRequest, - Some("ErrorResponse is not a valid request type"), + id, + ErrorResponseMessage::InvalidRequest { + typ: "ErrorResponse".into(), + }, ) .await } @@ -116,14 +117,20 @@ impl ConnectionHandler { async fn _handle_connection(&mut self) { debug!("Starting connection handler for {}", self.remote); - if !self.dkg_state.is_dealers_remote_address(self.remote).await { - let msg = format!( - "{} is not a socket address of any known dealer. Closing the connection.", + let (is_dealer, epoch) = self.dkg_state.is_dealers_remote_address(self.remote).await; + if !is_dealer { + warn!( + "Received a request from an unknown dealer - {}", self.remote ); - warn!("{}", msg); - self.send_error_response(ErrorReason::UnknownDealer, Some(msg)) - .await; + self.send_error_response( + None, + ErrorResponseMessage::UnknownDealer { + sender_address: self.remote, + epoch_id: epoch.id, + }, + ) + .await; return; } @@ -156,11 +163,10 @@ impl ConnectionHandler { remote ); self.send_error_response( - ErrorReason::Timeout, - Some(format!( - "could not resolve connection within {:?}", - self.max_connection_duration - )), + None, + ErrorResponseMessage::Timeout { + timeout: self.max_connection_duration, + }, ) .await; } diff --git a/validator-api/src/dkg/networking/message.rs b/validator-api/src/dkg/networking/message.rs index 9827863daa..d5b56dafbd 100644 --- a/validator-api/src/dkg/networking/message.rs +++ b/validator-api/src/dkg/networking/message.rs @@ -9,6 +9,9 @@ use dkg::Dealing; use serde::{Deserialize, Serialize}; use std::fmt::{Display, Formatter}; use std::io; +use std::net::SocketAddr; +use std::time::Duration; +use thiserror::Error; #[derive(Debug, Serialize, Deserialize)] pub enum OffchainDkgMessage { @@ -25,7 +28,7 @@ pub enum OffchainDkgMessage { message: RemoteDealingResponseMessage, }, ErrorResponse { - id: u64, + id: Option, message: ErrorResponseMessage, }, } @@ -75,42 +78,32 @@ mod dealing_bytes { } } -#[derive(Debug, Serialize, Deserialize)] -pub struct ErrorResponseMessage { - pub reason: ErrorReason, - pub additional_info: Option, -} +#[derive(Debug, Serialize, Deserialize, Error)] +pub enum ErrorResponseMessage { + #[error("Received request for epoch: {requested}, while the current epoch is {current}")] + InvalidEpoch { current: u32, requested: u32 }, -impl ErrorResponseMessage { - pub fn new(reason: ErrorReason, additional_info: Option) -> Self { - ErrorResponseMessage { - reason, - additional_info, - } - } -} + #[error("This sender is not a known dealer for this DKG epoch. Epoch: {epoch_id}, sender: {sender_address}")] + UnknownDealer { + sender_address: SocketAddr, + epoch_id: u32, + }, -impl Display for ErrorResponseMessage { - fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - todo!() - } -} + #[error("{typ} is not a valid request type")] + InvalidRequest { typ: String }, -#[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!() - } + #[error("This request failed to get resolved within {} seconds", .timeout.as_secs())] + Timeout { timeout: Duration }, } impl OffchainDkgMessage { + pub(crate) fn new_error_response( + id: Option, + message: ErrorResponseMessage, + ) -> OffchainDkgMessage { + OffchainDkgMessage::ErrorResponse { id, message } + } + pub(crate) fn try_from_bytes(bytes: Vec) -> Result { Ok(bincode::deserialize(&bytes)?) } @@ -150,10 +143,6 @@ impl FramedOffchainDkgMessage { out.append(&mut self.payload); out } - - fn try_from_bytes(bytes: &[u8]) -> Result { - todo!() - } } #[derive(Debug, Copy, Clone, PartialEq)] diff --git a/validator-api/src/dkg/state.rs b/validator-api/src/dkg/state.rs index 504ccf330c..ff367c2ff3 100644 --- a/validator-api/src/dkg/state.rs +++ b/validator-api/src/dkg/state.rs @@ -42,11 +42,18 @@ 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 - .values() - .any(|dealer| dealer.remote_address == remote) + + pub(crate) async fn is_dealers_remote_address(&self, remote: SocketAddr) -> (bool, Epoch) { + let guard = self.inner.lock().await; + let epoch = guard.current_epoch; + let dealers = &guard.current_epoch_dealers; + + ( + dealers + .values() + .any(|dealer| dealer.remote_address == remote), + epoch, + ) } pub(crate) async fn current_epoch(&self) -> Epoch {