From 98ee9c46a9bebe3ad43da688811201b6e41beb2f Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Wed, 15 Mar 2023 15:11:37 +0100 Subject: [PATCH] node: Fix state mismatch between service and wire Due to never removing peers from the peer map in the wire protocol module, it was possible for there to be multiple connected peers with the same Node ID. Some might be in a "connected" state, while others might be "disconnected". This caused a bug where a peer, although connected would be considered disconnected by the wire module, because when searching for a peer with that Node ID, the disconnected one would return first. To fix this, we make sure to remove peers from the map once the connection is dropped. --- radicle-node/src/wire/protocol.rs | 99 +++++++++++++++++-------------- 1 file changed, 55 insertions(+), 44 deletions(-) diff --git a/radicle-node/src/wire/protocol.rs b/radicle-node/src/wire/protocol.rs index 41c4fc59..1b9a6ca2 100644 --- a/radicle-node/src/wire/protocol.rs +++ b/radicle-node/src/wire/protocol.rs @@ -2,6 +2,7 @@ //! //! We use the Noise NN handshake pattern to establish an encrypted stream with a remote peer. //! The handshake itself is implemented in the external [`cyphernet`] and [`netservices`] crates. +use std::collections::hash_map::Entry; use std::collections::VecDeque; use std::os::unix::io::AsRawFd; use std::os::unix::prelude::RawFd; @@ -80,9 +81,9 @@ enum Peer { id: NodeId, inbox: VecDeque, }, - /// The state after a peer was disconnected, either during handshake, - /// or once connected. - Disconnected { + /// The peer was scheduled for disconnection. Once the transport is handed over + /// by the reactor, we can consider it disconnected. + Disconnecting { id: Option, reason: DisconnectReason, }, @@ -104,7 +105,7 @@ impl std::fmt::Debug for Peer { Self::Inbound {} => write!(f, "Inbound"), Self::Outbound { id } => write!(f, "Outbound({id})"), Self::Connected { link, id, .. } => write!(f, "Connected({link:?}, {id})"), - Self::Disconnected { .. } => write!(f, "Disconnected"), + Self::Disconnecting { .. } => write!(f, "Disconnecting"), Self::Upgrading { fetch, link, id, .. } => write!( @@ -123,12 +124,12 @@ impl Peer { match self { Peer::Outbound { id } | Peer::Connected { id, .. } - | Peer::Disconnected { id: Some(id), .. } + | Peer::Disconnecting { id: Some(id), .. } | Peer::Upgrading { id, .. } | Peer::Upgraded { id, .. } => Some(id), Peer::Inbound {} => None, - Peer::Disconnected { id: None, .. } => None, + Peer::Disconnecting { id: None, .. } => None, } } @@ -168,17 +169,17 @@ impl Peer { } } - /// Switch to disconnected state. - fn disconnected(&mut self, reason: DisconnectReason) { + /// Switch to disconnecting state. + fn disconnecting(&mut self, reason: DisconnectReason) { if let Self::Connected { id, .. } = self { - *self = Self::Disconnected { + *self = Self::Disconnecting { id: Some(*id), reason, }; } else if let Self::Inbound {} = self { - *self = Self::Disconnected { id: None, reason }; + *self = Self::Disconnecting { id: None, reason }; } else if let Self::Outbound { id } = self { - *self = Self::Disconnected { + *self = Self::Disconnecting { id: Some(*id), reason, }; @@ -320,7 +321,7 @@ where Peer::Connected { id, .. } => Some((*fd, id)), Peer::Upgrading { id, .. } => Some((*fd, id)), Peer::Upgraded { id, .. } => Some((*fd, id)), - Peer::Disconnected { .. } => None, + Peer::Disconnecting { .. } => None, }) } @@ -335,21 +336,26 @@ where } fn disconnect(&mut self, fd: RawFd, reason: DisconnectReason) { - let peer = self.peer_mut_by_fd(fd); - if let Peer::Disconnected { .. } = peer { - log::error!(target: "wire", "Peer (fd={fd}) is already disconnected"); - return; - }; - log::debug!(target: "wire", "Disconnecting peer (fd={fd}): {reason}"); - peer.disconnected(reason); + match self.peers.get_mut(&fd) { + Some(Peer::Disconnecting { .. }) => { + log::error!(target: "wire", "Peer (fd={fd}) is already disconnecting"); + } + Some(peer) => { + log::debug!(target: "wire", "Disconnecting peer (fd={fd}): {reason}"); - self.actions.push_back(Action::UnregisterTransport(fd)); + peer.disconnecting(reason); + self.actions.push_back(Action::UnregisterTransport(fd)); + } + None => { + log::error!(target: "wire", "Unknown peer (fd={fd}) cannot be disconnected"); + } + } } fn upgrade(&mut self, fd: RawFd, fetch: Fetch) { let peer = self.peer_mut_by_fd(fd); - if let Peer::Disconnected { .. } = peer { - log::error!(target: "wire", "Peer (fd={fd}) is already disconnected"); + if let Peer::Disconnecting { .. } = peer { + log::error!(target: "wire", "Peer (fd={fd}) is disconnecting"); return; }; log::debug!(target: "wire", "Requesting transport handover from reactor for peer (fd={fd})"); @@ -387,8 +393,8 @@ where let fd = session.as_connection().as_raw_fd(); let peer = self.peer_mut_by_fd(fd); - let session = if let Peer::Disconnected { .. } = peer { - log::error!(target: "wire", "Peer with fd {fd} is already disconnected"); + let session = if let Peer::Disconnecting { .. } = peer { + log::error!(target: "wire", "Peer with fd {fd} is disconnecting"); return; } else if let Peer::Upgraded { link, .. } = peer { match NetTransport::with_session(session, *link) { @@ -587,7 +593,10 @@ where reactor::Error::TransportDisconnect(fd, _, _) => { log::error!(target: "wire", "Received error: peer (fd={fd}) disconnected"); - match self.peers.get_mut(fd) { + // The peer transport is already disconnected and removed from the reactor; + // therefore there is no need to initiate a disconnection. We simply remove + // the peer from the map. + match self.peers.remove(fd) { Some(peer) => { let reason = DisconnectReason::Connection(Arc::new(io::Error::from( io::ErrorKind::ConnectionReset, @@ -598,7 +607,6 @@ where } else { log::debug!(target: "wire", "Inbound disconnection before handshake; ignoring..") } - peer.disconnected(reason); } None => { log::warn!(target: "wire", "Peer with fd {fd} is unknown"); @@ -617,33 +625,36 @@ where } fn handover_listener(&mut self, _listener: Self::Listener) { - panic!("Transport::handover_listener: listener handover is not supported"); + panic!("Wire::handover_listener: listener handover is not supported"); } fn handover_transport(&mut self, transport: Self::Transport) { let fd = transport.as_raw_fd(); log::debug!(target: "wire", "Received transport handover (fd={fd})"); - match self.peers.get(&fd) { - Some(Peer::Disconnected { id, reason, .. }) => { - // Disconnect TCP stream. - drop(transport); + match self.peers.entry(fd) { + Entry::Occupied(e) => { + match e.get() { + Peer::Disconnecting { id, reason, .. } => { + // Disconnect TCP stream. + drop(transport); - if let Some(id) = id { - self.service.disconnected(*id, reason); - } else { - // TODO: Handle this case by calling `disconnected` with the address instead of - // the node id. + // If there is no ID, the service is not aware of the peer. + if let Some(id) = id { + self.service.disconnected(*id, reason); + } + e.remove(); + } + Peer::Upgrading { .. } => { + self.upgraded(transport); + } + _ => { + panic!("Wire::handover_transport: Unexpected peer with fd {fd} handed over from the reactor"); + } } } - Some(Peer::Upgrading { .. }) => { - self.upgraded(transport); - } - Some(_) => { - panic!("Transport::handover_transport: Unexpected peer with fd {fd} handed over from the reactor"); - } - None => { - panic!("Transport::handover_transport: Unknown peer with fd {fd} handed over"); + Entry::Vacant(_) => { + panic!("Wire::handover_transport: Unknown peer with fd {fd} handed over"); } } }