node: Get rid of `nakamoto::DisconnectReason`

Signed-off-by: Alexis Sellier <self@cloudhead.io>
This commit is contained in:
Dr. Maxim Orlovsky 2023-01-10 16:18:15 +01:00 committed by Alexis Sellier
parent b7b33acbdf
commit 08334f790b
No known key found for this signature in database
4 changed files with 58 additions and 69 deletions

View File

@ -558,11 +558,7 @@ where
} }
} }
pub fn disconnected( pub fn disconnected(&mut self, remote: NodeId, reason: &DisconnectReason) {
&mut self,
remote: NodeId,
reason: &nakamoto::DisconnectReason<DisconnectReason>,
) {
let since = self.local_time(); let since = self.local_time();
debug!("Disconnected from {} ({})", remote, reason); debug!("Disconnected from {} ({})", remote, reason);
@ -576,10 +572,8 @@ where
if reason.is_dial_err() { if reason.is_dial_err() {
return; return;
} }
if let nakamoto::DisconnectReason::Protocol(r) = reason { if !reason.is_transient() {
if !r.is_transient() { return;
return;
}
} }
// TODO: Eventually we want a delay before attempting a reconnection, // TODO: Eventually we want a delay before attempting a reconnection,
// with exponential back-off. // with exponential back-off.
@ -610,7 +604,7 @@ where
// If there's an error, stop processing messages from this peer. // If there's an error, stop processing messages from this peer.
// However, we still relay messages returned up to this point. // However, we still relay messages returned up to this point.
self.reactor self.reactor
.disconnect(remote, DisconnectReason::Error(err)); .disconnect(remote, DisconnectReason::Session(err));
// FIXME: The peer should be set in a state such that we don'that // FIXME: The peer should be set in a state such that we don'that
// process further messages. // process further messages.
@ -983,8 +977,10 @@ where
.filter(|(_, session)| session.last_active < *now - STALE_CONNECTION_TIMEOUT); .filter(|(_, session)| session.last_active < *now - STALE_CONNECTION_TIMEOUT);
for (_, session) in stale { for (_, session) in stale {
self.reactor self.reactor.disconnect(
.disconnect(session.id, DisconnectReason::Error(session::Error::Timeout)); session.id,
DisconnectReason::Session(session::Error::Timeout),
);
} }
} }
@ -1086,35 +1082,43 @@ where
} }
} }
/// Disconnect reason.
#[derive(Debug)] #[derive(Debug)]
pub enum DisconnectReason { pub enum DisconnectReason {
User, /// Error while dialing the remote. This error occures before a connection is
Peer, /// even established. Errors of this kind are usually not transient.
Error(session::Error), Dial(Arc<dyn std::error::Error + Sync + Send>),
/// Error with an underlying established connection. Sometimes, reconnecting
/// after such an error is possible.
Connection(Arc<dyn std::error::Error + Sync + Send>),
// Session error.
Session(session::Error),
} }
impl DisconnectReason { impl DisconnectReason {
fn is_transient(&self) -> bool { pub fn is_dial_err(&self) -> bool {
match self { matches!(self, Self::Dial(_))
Self::User => false,
Self::Peer => false,
Self::Error(..) => false,
}
} }
}
impl From<DisconnectReason> for nakamoto_net::DisconnectReason<DisconnectReason> { pub fn is_connection_err(&self) -> bool {
fn from(reason: DisconnectReason) -> Self { matches!(self, Self::Connection(_))
nakamoto_net::DisconnectReason::Protocol(reason) }
pub fn is_transient(&self) -> bool {
match self {
Self::Dial(_) => false,
Self::Connection(_) => true,
Self::Session(..) => false,
}
} }
} }
impl fmt::Display for DisconnectReason { impl fmt::Display for DisconnectReason {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self { match self {
Self::User => write!(f, "user"), Self::Dial(err) => write!(f, "{}", err),
Self::Peer => write!(f, "peer"), Self::Connection(err) => write!(f, "{}", err),
Self::Error(err) => write!(f, "error: {}", err), Self::Session(err) => write!(f, "error: {}", err),
} }
} }
} }

View File

@ -6,10 +6,10 @@ use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::marker::PhantomData; use std::marker::PhantomData;
use std::ops::{Deref, DerefMut, Range}; use std::ops::{Deref, DerefMut, Range};
use std::rc::Rc; use std::rc::Rc;
use std::sync::Arc;
use std::{fmt, io}; use std::{fmt, io};
use log::*; use log::*;
use nakamoto_net as nakamoto;
use nakamoto_net::{Link, LocalDuration, LocalTime}; use nakamoto_net::{Link, LocalDuration, LocalTime};
use crate::crypto::{Negotiator, Signer}; use crate::crypto::{Negotiator, Signer};
@ -55,7 +55,7 @@ pub enum Input {
link: Link, link: Link,
}, },
/// Disconnected from peer. /// Disconnected from peer.
Disconnected(NodeId, Rc<nakamoto::DisconnectReason<DisconnectReason>>), Disconnected(NodeId, Rc<DisconnectReason>),
/// Received a message from a remote peer. /// Received a message from a remote peer.
Received(NodeId, Vec<Message>), Received(NodeId, Vec<Message>),
/// Used to advance the state machine after some wall time has passed. /// Used to advance the state machine after some wall time has passed.
@ -493,9 +493,9 @@ impl<S: WriteStorage + 'static, G: Signer + Negotiator> Simulation<S, G> {
remote, remote,
input: Input::Disconnected( input: Input::Disconnected(
remote, remote,
Rc::new(nakamoto::DisconnectReason::ConnectionError( Rc::new(DisconnectReason::Connection(Arc::new(
io::Error::from(io::ErrorKind::UnexpectedEof).into(), io::Error::from(io::ErrorKind::UnexpectedEof),
)), ))),
), ),
}, },
); );
@ -533,7 +533,7 @@ impl<S: WriteStorage + 'static, G: Signer + Negotiator> Simulation<S, G> {
self.priority.push_back(Scheduled { self.priority.push_back(Scheduled {
remote, remote,
node, node,
input: Input::Disconnected(remote, Rc::new(reason.into())), input: Input::Disconnected(remote, Rc::new(reason)),
}); });
// Nb. It's possible for disconnects to happen simultaneously from both ends, hence // Nb. It's possible for disconnects to happen simultaneously from both ends, hence
@ -556,9 +556,9 @@ impl<S: WriteStorage + 'static, G: Signer + Negotiator> Simulation<S, G> {
remote: node, remote: node,
input: Input::Disconnected( input: Input::Disconnected(
node, node,
Rc::new(nakamoto::DisconnectReason::ConnectionError( Rc::new(DisconnectReason::Connection(Arc::new(io::Error::from(
io::Error::from(io::ErrorKind::ConnectionReset).into(), io::ErrorKind::ConnectionReset,
)), )))),
), ),
}, },
); );

View File

@ -3,7 +3,6 @@ use std::io;
use std::sync::Arc; use std::sync::Arc;
use crossbeam_channel as chan; use crossbeam_channel as chan;
use nakamoto_net as nakamoto;
use crate::collections::{HashMap, HashSet}; use crate::collections::{HashMap, HashSet};
use crate::crypto::test::signer::MockSigner; use crate::crypto::test::signer::MockSigner;
@ -379,7 +378,7 @@ fn test_inventory_relay_bad_timestamp() {
); );
assert_matches!( assert_matches!(
alice.outbox().next(), alice.outbox().next(),
Some(Io::Disconnect(addr, DisconnectReason::Error(session::Error::InvalidTimestamp(t)))) Some(Io::Disconnect(addr, DisconnectReason::Session(session::Error::InvalidTimestamp(t))))
if addr == bob.id() && t == timestamp if addr == bob.id() && t == timestamp
); );
} }
@ -735,17 +734,11 @@ fn test_persistent_peer_reconnect() {
// A non-transient disconnect, such as one requested by the user will not trigger // A non-transient disconnect, such as one requested by the user will not trigger
// a reconnection. // a reconnection.
alice.disconnected( alice.disconnected(eve.id(), &DisconnectReason::Dial(error.clone()));
eve.id(),
&nakamoto::DisconnectReason::DialError(error.clone()),
);
assert_matches!(alice.outbox().next(), None); assert_matches!(alice.outbox().next(), None);
for _ in 0..MAX_CONNECTION_ATTEMPTS { for _ in 0..MAX_CONNECTION_ATTEMPTS {
alice.disconnected( alice.disconnected(bob.id(), &DisconnectReason::Connection(error.clone()));
bob.id(),
&nakamoto::DisconnectReason::ConnectionError(error.clone()),
);
assert_matches!(alice.outbox().next(), Some(Io::Connect(a, _)) if a == bob.id()); assert_matches!(alice.outbox().next(), Some(Io::Connect(a, _)) if a == bob.id());
assert_matches!(alice.outbox().next(), None); assert_matches!(alice.outbox().next(), None);
@ -753,10 +746,7 @@ fn test_persistent_peer_reconnect() {
} }
// After the max connection attempts, a disconnect doesn't trigger a reconnect. // After the max connection attempts, a disconnect doesn't trigger a reconnect.
alice.disconnected( alice.disconnected(bob.id(), &DisconnectReason::Connection(error));
bob.id(),
&nakamoto::DisconnectReason::ConnectionError(error),
);
assert_matches!(alice.outbox().next(), None); assert_matches!(alice.outbox().next(), None);
} }
@ -790,10 +780,7 @@ fn test_maintain_connections() {
// A transient error such as this will cause Alice to attempt a reconnection. // A transient error such as this will cause Alice to attempt a reconnection.
let error = Arc::new(io::Error::from(io::ErrorKind::ConnectionReset)); let error = Arc::new(io::Error::from(io::ErrorKind::ConnectionReset));
for peer in connected.iter() { for peer in connected.iter() {
alice.disconnected( alice.disconnected(peer.id(), &DisconnectReason::Connection(error.clone()));
peer.id(),
&nakamoto::DisconnectReason::ConnectionError(error.clone()),
);
let id = alice let id = alice
.outbox() .outbox()

View File

@ -11,7 +11,7 @@ use std::{io, net};
use crossbeam_channel as chan; use crossbeam_channel as chan;
use cyphernet::addr::{Addr as _, HostAddr, PeerAddr}; use cyphernet::addr::{Addr as _, HostAddr, PeerAddr};
use nakamoto_net::{DisconnectReason, Link, LocalTime}; use nakamoto_net::{Link, LocalTime};
use netservices::noise::NoiseXk; use netservices::noise::NoiseXk;
use netservices::wire::{ListenerEvent, NetAccept, NetTransport, SessionEvent}; use netservices::wire::{ListenerEvent, NetAccept, NetTransport, SessionEvent};
use netservices::NetSession; use netservices::NetSession;
@ -23,7 +23,7 @@ use radicle::storage::WriteStorage;
use crate::crypto::Signer; use crate::crypto::Signer;
use crate::service::reactor::{Fetch, Io}; use crate::service::reactor::{Fetch, Io};
use crate::service::{routing, session, Message, Service}; use crate::service::{routing, session, DisconnectReason, Message, Service};
use crate::wire::{Decode, Encode}; use crate::wire::{Decode, Encode};
use crate::worker::{WorkerReq, WorkerResp}; use crate::worker::{WorkerReq, WorkerResp};
use crate::{address, service}; use crate::{address, service};
@ -43,7 +43,7 @@ enum Peer<G: Negotiator> {
/// or once connected. /// or once connected.
Disconnected { Disconnected {
id: NodeId, id: NodeId,
reason: DisconnectReason<service::DisconnectReason>, reason: DisconnectReason,
}, },
/// The state after we've started the process of upgraded the peer for a fetch. /// The state after we've started the process of upgraded the peer for a fetch.
/// The request to handover the socket was made to the reactor. /// The request to handover the socket was made to the reactor.
@ -76,7 +76,7 @@ impl<G: Negotiator> Peer<G> {
} }
/// Switch to disconnected state. /// Switch to disconnected state.
fn disconnected(&mut self, reason: DisconnectReason<service::DisconnectReason>) { fn disconnected(&mut self, reason: DisconnectReason) {
if let Self::Connected { id, .. } = self { if let Self::Connected { id, .. } = self {
*self = Self::Disconnected { id: *id, reason }; *self = Self::Disconnected { id: *id, reason };
} else { } else {
@ -188,7 +188,7 @@ where
}) })
} }
fn disconnect(&mut self, fd: RawFd, reason: DisconnectReason<service::DisconnectReason>) { fn disconnect(&mut self, fd: RawFd, reason: DisconnectReason) {
let Some(peer) = self.peers.get_mut(&fd) else { let Some(peer) = self.peers.get_mut(&fd) else {
log::error!(target: "transport", "Peer with fd {fd} was not found"); log::error!(target: "transport", "Peer with fd {fd} was not found");
return; return;
@ -342,9 +342,9 @@ where
); );
self.disconnect( self.disconnect(
fd, fd,
DisconnectReason::DialError( DisconnectReason::Dial(Arc::new(io::Error::from(
io::Error::from(io::ErrorKind::AlreadyExists).into(), io::ErrorKind::AlreadyExists,
), ))),
); );
} }
@ -385,9 +385,7 @@ where
log::error!(target: "transport", "Invalid message from {}: {err}", id); log::error!(target: "transport", "Invalid message from {}: {err}", id);
self.disconnect( self.disconnect(
fd, fd,
DisconnectReason::Protocol(service::DisconnectReason::Error( DisconnectReason::Session(session::Error::Misbehavior),
session::Error::Misbehavior,
)),
); );
break; break;
} }
@ -398,7 +396,7 @@ where
} }
} }
SessionEvent::Terminated(err) => { SessionEvent::Terminated(err) => {
self.disconnect(fd, DisconnectReason::ConnectionError(Arc::new(err))); self.disconnect(fd, DisconnectReason::Connection(Arc::new(err)));
} }
} }
} }
@ -513,14 +511,14 @@ where
} }
Err(err) => { Err(err) => {
self.service self.service
.disconnected(node_id, &DisconnectReason::DialError(Arc::new(err))); .disconnected(node_id, &DisconnectReason::Dial(Arc::new(err)));
break; break;
} }
} }
} }
Io::Disconnect(node_id, reason) => { Io::Disconnect(node_id, reason) => {
let fd = self.by_id(&node_id); let fd = self.by_id(&node_id);
self.disconnect(fd, DisconnectReason::Protocol(reason)); self.disconnect(fd, reason);
return self.actions.pop_back(); return self.actions.pop_back();
} }