diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index ef1c8b1d..b2eb9afe 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -724,8 +724,8 @@ where }; let link = session.link; - // If the peer disconnected while we were waiting for a [`Message::FetchOk`], - // return a failure to any potential fetcher. + // If the peer disconnected while we were fetching, return a failure to any + // potential fetcher. for rid in session.fetching() { if let Some(resp) = self.fetch_reqs.remove(&(rid, remote)) { resp.send(FetchResult::Failed { diff --git a/radicle-node/src/wire/protocol.rs b/radicle-node/src/wire/protocol.rs index d9e222a7..c5561cf0 100644 --- a/radicle-node/src/wire/protocol.rs +++ b/radicle-node/src/wire/protocol.rs @@ -229,6 +229,61 @@ impl Peer { } } +struct Peers(HashMap); + +impl Peers { + fn get_mut(&mut self, fd: &RawFd) -> Option<&mut Peer> { + self.0.get_mut(fd) + } + + fn entry(&mut self, fd: RawFd) -> Entry { + self.0.entry(fd) + } + + fn insert(&mut self, fd: RawFd, peer: Peer) { + if self.0.insert(fd, peer).is_some() { + log::warn!(target: "wire", "Replacing existing peer fd={fd}"); + } + } + + fn remove(&mut self, fd: &RawFd) -> Option { + self.0.remove(fd) + } + + fn lookup(&self, node_id: &NodeId) -> Option<(RawFd, &Peer)> { + self.0 + .iter() + .find(|(_, peer)| peer.id() == Some(node_id)) + .map(|(fd, peer)| (*fd, peer)) + } + + fn lookup_mut(&mut self, node_id: &NodeId) -> Option<(RawFd, &mut Peer)> { + self.0 + .iter_mut() + .find(|(_, peer)| peer.id() == Some(node_id)) + .map(|(fd, peer)| (*fd, peer)) + } + + fn active(&self) -> impl Iterator { + self.0.iter().filter_map(|(fd, peer)| match peer { + Peer::Inbound {} => None, + Peer::Outbound { id } => Some((*fd, id)), + Peer::Connected { nid: id, .. } => Some((*fd, id)), + Peer::Disconnecting { .. } => None, + }) + } + + fn connected(&self) -> impl Iterator { + self.0.iter().filter_map(|(fd, peer)| { + if let Peer::Connected { nid: id, .. } = peer { + Some((*fd, id)) + } else { + None + } + }) + } +} + /// Wire protocol implementation for a set of peers. pub struct Wire { /// Backing service instance. @@ -240,7 +295,7 @@ pub struct Wire { /// Internal queue of actions to send to the reactor. actions: VecDeque>, /// Peer sessions. - peers: HashMap, + peers: Peers, /// SOCKS5 proxy address. proxy: net::SocketAddr, } @@ -269,7 +324,7 @@ where signer, proxy, actions: VecDeque::new(), - peers: HashMap::default(), + peers: Peers(HashMap::default()), } } @@ -277,52 +332,6 @@ where self.actions.push_back(Action::RegisterListener(socket)); } - fn fd_by_id(&self, node_id: &NodeId) -> (RawFd, &Peer) { - self.peers - .iter() - .find(|(_, peer)| peer.id() == Some(node_id)) - .map(|(fd, peer)| (*fd, peer)) - .unwrap_or_else(|| panic!("Peer {node_id} was expected to be known to the transport")) - } - - fn fd_by_id_mut(&mut self, node_id: &NodeId) -> (RawFd, &mut Peer) { - self.peers - .iter_mut() - .find(|(_, peer)| peer.id() == Some(node_id)) - .map(|(fd, peer)| (*fd, peer)) - .unwrap_or_else(|| panic!("Peer {node_id} was expected to be known to the transport")) - } - - fn connected_fd_by_id(&self, node_id: &NodeId) -> RawFd { - match self.fd_by_id(node_id) { - (fd, Peer::Connected { .. }) => fd, - (fd, peer) => { - panic!( - "Peer {node_id} (fd={fd}) was expected to be in a connected state ({peer:?})" - ) - } - } - } - - fn active(&self) -> impl Iterator { - self.peers.iter().filter_map(|(fd, peer)| match peer { - Peer::Inbound {} => None, - Peer::Outbound { id } => Some((*fd, id)), - Peer::Connected { nid: id, .. } => Some((*fd, id)), - Peer::Disconnecting { .. } => None, - }) - } - - fn connected(&self) -> impl Iterator { - self.peers.iter().filter_map(|(fd, peer)| { - if let Peer::Connected { nid: id, .. } = peer { - Some((*fd, id)) - } else { - None - } - }) - } - fn disconnect(&mut self, fd: RawFd, reason: DisconnectReason) { match self.peers.get_mut(&fd) { Some(Peer::Disconnecting { .. }) => { @@ -348,25 +357,20 @@ where ); let nid = task.remote; - let Some((fd, peer)) = self - .peers - .iter_mut() - .find(|(_, peer)| peer.id() == Some(&nid)) - .map(|(fd, peer)| (*fd, peer)) else { - log::warn!(target: "wire", "Peer {nid} not found; ignoring fetch result"); - return; - }; + let Some((fd, peer)) = self.peers.lookup_mut(&nid) else { + log::warn!(target: "wire", "Peer {nid} not found; ignoring fetch result"); + return; + }; let Peer::Connected { nid, link, streams, .. } = peer else { log::warn!(target: "wire", "Peer {nid} is not connected; ignoring fetch result"); return; }; - let remote = *nid; // Only call into the service if we initiated this fetch. match task.result { FetchResult::Initiator { rid, result } => { - self.service.fetched(rid, remote, result); + self.service.fetched(rid, *nid, result); } FetchResult::Responder { .. } => { // We don't do anything with upload results for now. @@ -387,14 +391,10 @@ where } fn flush(&mut self, remote: NodeId, stream: StreamId) { - let Some((fd, peer)) = self - .peers - .iter() - .find(|(_, peer)| peer.id() == Some(&remote)) - .map(|(fd, peer)| (*fd, peer)) else { - log::warn!(target: "wire", "Peer {remote} is not known; ignoring flush"); - return; - }; + let Some((fd, peer)) = self.peers.lookup(&remote) else { + log::warn!(target: "wire", "Peer {remote} is not known; ignoring flush"); + return; + }; let Peer::Connected { streams, link, .. } = peer else { log::warn!(target: "wire", "Peer {remote} is not connected; ignoring flush"); return; @@ -483,6 +483,7 @@ where log::debug!(target: "wire", "Session established with {id} (fd={fd})"); let conflicting = self + .peers .active() .filter(|(other, d)| **d == id && *other != fd) .map(|(fd, _)| fd) @@ -734,14 +735,18 @@ where while let Some(ev) = self.service.next() { match ev { Io::Write(node_id, msgs) => { - let (fd, link) = match self.fd_by_id(&node_id) { - (fd, Peer::Connected { link, .. }) => (fd, *link), - (_, peer) => { + let (fd, link) = match self.peers.lookup(&node_id) { + Some((fd, Peer::Connected { link, .. })) => (fd, *link), + Some((_, peer)) => { // If the peer is disconnected by the wire protocol, the service may // not be aware of this yet, and may continue to write messages to it. log::debug!(target: "wire", "Dropping {} message(s) to {node_id} ({peer:?})", msgs.len()); continue; } + None => { + log::error!(target: "wire", "Dropping {} message(s) to {node_id}: unknown peer", msgs.len()); + continue; + } }; log::trace!( target: "wire", "Writing {} message(s) to {}", msgs.len(), node_id @@ -756,7 +761,7 @@ where self.actions.push_back(reactor::Action::Send(fd, data)); } Io::Connect(node_id, addr) => { - if self.connected().any(|(_, id)| id == &node_id) { + if self.peers.connected().any(|(_, id)| id == &node_id) { log::error!( target: "wire", "Attempt to connect to already connected peer {node_id}" @@ -793,9 +798,12 @@ where } } } - Io::Disconnect(node_id, reason) => { - let fd = self.connected_fd_by_id(&node_id); - self.disconnect(fd, reason); + Io::Disconnect(nid, reason) => { + if let Some((fd, Peer::Connected { .. })) = self.peers.lookup(&nid) { + self.disconnect(fd, reason); + } else { + log::warn!(target: "wire", "Peer {nid} is not connected: ignoring disconnect"); + } } Io::Wakeup(d) => { self.actions.push_back(reactor::Action::SetTimer(d.into())); @@ -807,9 +815,14 @@ where } => { log::trace!(target: "wire", "Processing fetch for {rid} from {remote}.."); - let (fd, Peer::Connected { link, streams, .. }) = - self.fd_by_id_mut(&remote) else { - panic!("Wire::next: peer {remote} is not connected"); + let Some((fd, Peer::Connected { link, streams, .. })) = + self.peers.lookup_mut(&remote) else { + // Nb. It's possible that a peer is disconnected while an `Io::Fetch` + // is in the service's i/o buffer. Since the service may not purge the + // buffer on disconnect, we should just ignore i/o actions that don't + // have a connected peer. + log::error!(target: "wire", "Peer {remote} is not connected: dropping fetch"); + continue; }; let (stream, channels) = streams.open();