use std::{borrow::Cow, sync::Arc};
use thiserror::Error;
use tracing_error::TracedError;
use zakura_chain::serialization::SerializationError;
use crate::{
protocol::external::InventoryHash,
zakura::{ZakuraProtocolError, ZakuraUpgradeError},
};
#[derive(Debug, Clone)]
pub struct SharedPeerError(Arc<TracedError<PeerError>>, NotFoundClass);
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub enum NotFoundClass {
Response,
Registry,
Other,
}
impl std::fmt::Display for SharedPeerError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
std::fmt::Display::fmt(self.0.as_ref(), f)
}
}
impl std::error::Error for SharedPeerError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
std::error::Error::source(self.0.as_ref())
}
}
impl<E> From<E> for SharedPeerError
where
PeerError: From<E>,
{
fn from(source: E) -> Self {
let inner = PeerError::from(source);
let class = match &inner {
PeerError::NotFoundResponse(_) => NotFoundClass::Response,
PeerError::NotFoundRegistry(_) => NotFoundClass::Registry,
_ => NotFoundClass::Other,
};
Self(Arc::new(TracedError::from(inner)), class)
}
}
impl SharedPeerError {
pub fn inner_debug(&self) -> String {
format!("{:?}", self.0.as_ref())
}
pub fn not_found_class(&self) -> Option<NotFoundClass> {
(self.1 != NotFoundClass::Other).then_some(self.1)
}
}
#[derive(Error, Debug)]
#[allow(dead_code)]
pub enum PeerError {
#[error("Peer closed connection")]
ConnectionClosed,
#[error("Internal connection dropped")]
ConnectionDropped,
#[error("Internal client dropped")]
ClientDropped,
#[error("Internal peer connection task exited")]
ConnectionTaskExited,
#[error("Internal client cancelled its heartbeat task")]
ClientCancelledHeartbeatTask,
#[error("Internal heartbeat task exited with message: {0:?}")]
HeartbeatTaskExited(String),
#[error("Sending Client request timed out")]
ConnectionSendTimeout,
#[error("Receiving client response timed out")]
ConnectionReceiveTimeout,
#[error("Serialization error: {0}")]
Serialization(#[from] SerializationError),
#[error("Remote peer sent handshake messages after handshake")]
DuplicateHandshake,
#[error("Internal services over capacity")]
Overloaded,
#[error("No ready peers available")]
NoReadyPeers,
#[error("Internal services timed out")]
InboundTimeout,
#[error("Internal services have failed or shutdown")]
ServiceShutdown,
#[error("Remote peer could not find any of the items: {0:?}")]
NotFoundResponse(Vec<InventoryHash>),
#[error("All ready peers are registered as recently missing these items: {0:?}")]
NotFoundRegistry(Vec<InventoryHash>),
}
impl PeerError {
pub fn kind(&self) -> Cow<'static, str> {
match self {
PeerError::ConnectionClosed => "ConnectionClosed".into(),
PeerError::ConnectionDropped => "ConnectionDropped".into(),
PeerError::ClientDropped => "ClientDropped".into(),
PeerError::ClientCancelledHeartbeatTask => "ClientCancelledHeartbeatTask".into(),
PeerError::HeartbeatTaskExited(_) => "HeartbeatTaskExited".into(),
PeerError::ConnectionTaskExited => "ConnectionTaskExited".into(),
PeerError::ConnectionSendTimeout => "ConnectionSendTimeout".into(),
PeerError::ConnectionReceiveTimeout => "ConnectionReceiveTimeout".into(),
PeerError::Serialization(inner) => format!("Serialization({inner})").into(),
PeerError::DuplicateHandshake => "DuplicateHandshake".into(),
PeerError::Overloaded => "Overloaded".into(),
PeerError::NoReadyPeers => "NoReadyPeers".into(),
PeerError::InboundTimeout => "InboundTimeout".into(),
PeerError::ServiceShutdown => "ServiceShutdown".into(),
PeerError::NotFoundResponse(_) => "NotFoundResponse".into(),
PeerError::NotFoundRegistry(_) => "NotFoundRegistry".into(),
}
}
}
#[derive(Default, Clone)]
pub struct ErrorSlot(Arc<std::sync::Mutex<Option<SharedPeerError>>>);
impl std::fmt::Debug for ErrorSlot {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ErrorSlot")
.field("error", &self.0.try_lock())
.finish()
}
}
impl ErrorSlot {
#[allow(clippy::unwrap_in_result)]
pub fn try_get_error(&self) -> Option<SharedPeerError> {
self.0
.lock()
.expect("error mutex should be unpoisoned")
.as_ref()
.cloned()
}
#[allow(clippy::unwrap_in_result)]
pub fn try_update_error(&self, e: SharedPeerError) -> Result<(), AlreadyErrored> {
let mut guard = self.0.lock().expect("error mutex should be unpoisoned");
if let Some(original_error) = guard.clone() {
Err(AlreadyErrored { original_error })
} else {
*guard = Some(e);
Ok(())
}
}
}
#[derive(Clone, Debug)]
pub struct AlreadyErrored {
pub original_error: SharedPeerError,
}
#[derive(Error, Debug)]
pub enum HandshakeError {
#[error("The remote peer sent an unexpected message: {0:?}")]
UnexpectedMessage(Box<crate::protocol::external::Message>),
#[error("Detected nonce reuse, possible self-connection")]
RemoteNonceReuse,
#[error("Unexpectedly created a duplicate random local nonce")]
LocalDuplicateNonce,
#[error("Peer closed connection")]
ConnectionClosed,
#[error("Underlying IO error: {0}")]
Io(#[from] std::io::Error),
#[error("Serialization error: {0}")]
Serialization(#[from] SerializationError),
#[error("Peer offered obsolete version: {0:?}")]
ObsoleteVersion(crate::protocol::external::types::Version),
#[error("Timeout when sending or receiving a message to peer")]
Timeout,
#[error("Zakura P2P v2 upgrade selected")]
ZakuraUpgradeSelected,
#[error("Zakura P2P v2 upgrade failed: {0}")]
ZakuraUpgrade(#[from] ZakuraUpgradeError),
#[error("Malformed Zakura P2P v2 upgrade prelude: {0}")]
ZakuraUpgradePreludeMalformed(#[source] ZakuraProtocolError),
}
impl HandshakeError {
pub fn is_neutral_disconnect(&self) -> bool {
matches!(
self,
HandshakeError::ZakuraUpgradeSelected | HandshakeError::ZakuraUpgrade(_)
)
}
}
impl From<tokio::time::error::Elapsed> for HandshakeError {
fn from(_source: tokio::time::error::Elapsed) -> Self {
HandshakeError::Timeout
}
}
#[cfg(test)]
mod tests {
use super::{NotFoundClass, PeerError, SharedPeerError};
use crate::protocol::external::InventoryHash;
use zakura_chain::block;
fn block_inv() -> Vec<InventoryHash> {
vec![InventoryHash::Block(block::Hash([0; 32]))]
}
#[test]
fn not_found_class_matches_peer_error_variant() {
assert_eq!(
SharedPeerError::from(PeerError::NotFoundResponse(block_inv())).not_found_class(),
Some(NotFoundClass::Response),
);
assert_eq!(
SharedPeerError::from(PeerError::NotFoundRegistry(block_inv())).not_found_class(),
Some(NotFoundClass::Registry),
);
assert_eq!(
SharedPeerError::from(PeerError::NoReadyPeers).not_found_class(),
None,
);
}
}