Merge pull request 'Fixes to garage_net peering manager' (#786) from net-fixes into next-0.10

Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/786

2 files changed, 95 insertions, 105 deletions

diff --git a/src/net/netapp.rs b/src/net/netapp.rs
index faa51a99..6480a126 100644
--- a/src/net/netapp.rs
+++ b/src/net/netapp.rs
@@ -292,13 +292,7 @@ impl NetApp {
 	/// the other node with `Netapp::request`
 	pub async fn try_connect(self: Arc<Self>, ip: SocketAddr, id: NodeID) -> Result<(), Error> {
 		// Don't connect to ourself, we don't care
-		// but pretend we did
 		if id == self.id {
-			tokio::spawn(async move {
-				if let Some(h) = self.on_connected_handler.load().as_ref() {
-					h(id, ip, false);
-				}
-			});
 			return Ok(());
 		}
 
@@ -327,31 +321,32 @@ impl NetApp {
 	/// Close the outgoing connection we have to a node specified by its public key,
 	/// if such a connection is currently open.
 	pub fn disconnect(self: &Arc<Self>, id: &NodeID) {
+		let conn = self.client_conns.write().unwrap().remove(id);
+
 		// If id is ourself, we're not supposed to have a connection open
-		if *id != self.id {
-			let conn = self.client_conns.write().unwrap().remove(id);
-			if let Some(c) = conn {
-				debug!(
-					"Closing connection to {} ({})",
-					hex::encode(&c.peer_id[..8]),
-					c.remote_addr
-				);
-				c.close();
-			} else {
-				return;
-			}
+		if *id == self.id {
+			// sanity check
+			assert!(conn.is_none(), "had a connection to local node");
+			return;
 		}
 
-		// call on_disconnected_handler immediately, since the connection
-		// was removed
-		// (if id == self.id, we pretend we disconnected)
-		let id = *id;
-		let self2 = self.clone();
-		tokio::spawn(async move {
-			if let Some(h) = self2.on_disconnected_handler.load().as_ref() {
-				h(id, false);
-			}
-		});
+		if let Some(c) = conn {
+			debug!(
+				"Closing connection to {} ({})",
+				hex::encode(&c.peer_id[..8]),
+				c.remote_addr
+			);
+			c.close();
+
+			// call on_disconnected_handler immediately, since the connection was removed
+			let id = *id;
+			let self2 = self.clone();
+			tokio::spawn(async move {
+				if let Some(h) = self2.on_disconnected_handler.load().as_ref() {
+					h(id, false);
+				}
+			});
+		}
 	}
 
 	// Called from conn.rs when an incoming connection is successfully established
diff --git a/src/net/peering.rs b/src/net/peering.rs
index 61882a18..b4271231 100644
--- a/src/net/peering.rs
+++ b/src/net/peering.rs
@@ -54,12 +54,8 @@ impl Message for PeerListMessage {
 
 #[derive(Debug)]
 struct PeerInfoInternal {
-	// addr is the currently connected address,
-	// or the last address we were connected to,
-	// or an arbitrary address some other peer gave us
-	addr: SocketAddr,
-	// all_addrs contains all of the addresses everyone gave us
-	all_addrs: Vec<SocketAddr>,
+	// known_addrs contains all of the addresses everyone gave us
+	known_addrs: Vec<SocketAddr>,
 
 	state: PeerConnState,
 	last_send_ping: Option<Instant>,
@@ -69,10 +65,9 @@ struct PeerInfoInternal {
 }
 
 impl PeerInfoInternal {
-	fn new(addr: SocketAddr, state: PeerConnState) -> Self {
+	fn new(state: PeerConnState, known_addr: Option<SocketAddr>) -> Self {
 		Self {
-			addr,
-			all_addrs: vec![addr],
+			known_addrs: known_addr.map(|x| vec![x]).unwrap_or_default(),
 			state,
 			last_send_ping: None,
 			last_seen: None,
@@ -81,8 +76,8 @@ impl PeerInfoInternal {
 		}
 	}
 	fn add_addr(&mut self, addr: SocketAddr) -> bool {
-		if !self.all_addrs.contains(&addr) {
-			self.all_addrs.push(addr);
+		if !self.known_addrs.contains(&addr) {
+			self.known_addrs.push(addr);
 			// If we are learning a new address for this node,
 			// we want to retry connecting
 			self.state = match self.state {
@@ -90,7 +85,7 @@ impl PeerInfoInternal {
 				PeerConnState::Waiting(_, _) | PeerConnState::Abandonned => {
 					PeerConnState::Waiting(0, Instant::now())
 				}
-				x @ (PeerConnState::Ourself | PeerConnState::Connected) => x,
+				x @ (PeerConnState::Ourself | PeerConnState::Connected { .. }) => x,
 			};
 			true
 		} else {
@@ -104,8 +99,6 @@ impl PeerInfoInternal {
 pub struct PeerInfo {
 	/// The node's identifier (its public key)
 	pub id: NodeID,
-	/// The node's network address
-	pub addr: SocketAddr,
 	/// The current status of our connection to this node
 	pub state: PeerConnState,
 	/// The last time at which the node was seen
@@ -136,7 +129,7 @@ pub enum PeerConnState {
 	Ourself,
 
 	/// We currently have a connection to this peer
-	Connected,
+	Connected { addr: SocketAddr },
 
 	/// Our next connection tentative (the nth, where n is the first value of the tuple)
 	/// will be at given Instant
@@ -152,7 +145,7 @@ pub enum PeerConnState {
 impl PeerConnState {
 	/// Returns true if we can currently send requests to this peer
 	pub fn is_up(&self) -> bool {
-		matches!(self, Self::Ourself | Self::Connected)
+		matches!(self, Self::Ourself | Self::Connected { .. })
 	}
 }
 
@@ -164,29 +157,42 @@ struct KnownHosts {
 impl KnownHosts {
 	fn new() -> Self {
 		let list = HashMap::new();
-		let hash = Self::calculate_hash(vec![]);
-		Self { list, hash }
+		let mut ret = Self {
+			list,
+			hash: hash::Digest::from_slice(&[0u8; 64][..]).unwrap(),
+		};
+		ret.update_hash();
+		ret
 	}
 	fn update_hash(&mut self) {
-		self.hash = Self::calculate_hash(self.connected_peers_vec());
-	}
-	fn connected_peers_vec(&self) -> Vec<(NodeID, SocketAddr)> {
-		let mut list = Vec::with_capacity(self.list.len());
-		for (id, peer) in self.list.iter() {
-			if peer.state.is_up() {
-				list.push((*id, peer.addr));
-			}
-		}
-		list
-	}
-	fn calculate_hash(mut list: Vec<(NodeID, SocketAddr)>) -> hash::Digest {
+		// The hash is a value that is exchanged between nodes when they ping one
+		// another.  Nodes compare their known hosts hash to know if they are connected
+		// to the same set of nodes. If the hashes differ, they are connected to
+		// different nodes and they trigger an exchange of the full list of active
+		// connections.  The hash value only represents the set of node IDs and not
+		// their actual socket addresses, because nodes can be connected via different
+		// addresses and that shouldn't necessarily trigger a full peer exchange.
+		let mut list = self
+			.list
+			.iter()
+			.filter(|(_, peer)| peer.state.is_up())
+			.map(|(id, _)| *id)
+			.collect::<Vec<_>>();
 		list.sort();
 		let mut hash_state = hash::State::new();
-		for (id, addr) in list {
+		for id in list {
 			hash_state.update(&id[..]);
-			hash_state.update(&format!("{}\n", addr).into_bytes()[..]);
 		}
-		hash_state.finalize()
+		self.hash = hash_state.finalize();
+	}
+	fn connected_peers_vec(&self) -> Vec<(NodeID, SocketAddr)> {
+		self.list
+			.iter()
+			.filter_map(|(id, peer)| match peer.state {
+				PeerConnState::Connected { addr } => Some((*id, addr)),
+				_ => None,
+			})
+			.collect::<Vec<_>>()
 	}
 }
 
@@ -220,18 +226,16 @@ impl PeeringManager {
 			if id != netapp.id {
 				known_hosts.list.insert(
 					id,
-					PeerInfoInternal::new(addr, PeerConnState::Waiting(0, Instant::now())),
+					PeerInfoInternal::new(PeerConnState::Waiting(0, Instant::now()), Some(addr)),
 				);
 			}
 		}
 
-		if let Some(addr) = our_addr {
-			known_hosts.list.insert(
-				netapp.id,
-				PeerInfoInternal::new(addr, PeerConnState::Ourself),
-			);
-			known_hosts.update_hash();
-		}
+		known_hosts.list.insert(
+			netapp.id,
+			PeerInfoInternal::new(PeerConnState::Ourself, our_addr),
+		);
+		known_hosts.update_hash();
 
 		// TODO for v0.10 / v1.0 : rename the endpoint (it will break compatibility)
 		let strat = Arc::new(Self {
@@ -276,7 +280,7 @@ impl PeeringManager {
 				for (id, info) in known_hosts.list.iter() {
 					trace!("{}, {:?}", hex::encode(&id[..8]), info);
 					match info.state {
-						PeerConnState::Connected => {
+						PeerConnState::Connected { .. } => {
 							let must_ping = match info.last_send_ping {
 								None => true,
 								Some(t) => Instant::now() - t > PING_INTERVAL,
@@ -319,7 +323,7 @@ impl PeeringManager {
 							info!(
 								"Retrying connection to {} at {} ({})",
 								hex::encode(&id[..8]),
-								h.all_addrs
+								h.known_addrs
 									.iter()
 									.map(|x| format!("{}", x))
 									.collect::<Vec<_>>()
@@ -328,13 +332,8 @@ impl PeeringManager {
 							);
 							h.state = PeerConnState::Trying(i);
 
-							let alternate_addrs = h
-								.all_addrs
-								.iter()
-								.filter(|x| **x != h.addr)
-								.cloned()
-								.collect::<Vec<_>>();
-							tokio::spawn(self.clone().try_connect(id, h.addr, alternate_addrs));
+							let addresses = h.known_addrs.clone();
+							tokio::spawn(self.clone().try_connect(id, addresses));
 						}
 					}
 				}
@@ -362,27 +361,24 @@ impl PeeringManager {
 	fn update_public_peer_list(&self, known_hosts: &KnownHosts) {
 		let mut pub_peer_list = Vec::with_capacity(known_hosts.list.len());
 		for (id, info) in known_hosts.list.iter() {
+			if *id == self.netapp.id {
+				// sanity check
+				assert!(matches!(info.state, PeerConnState::Ourself));
+			}
 			let mut pings = info.ping.iter().cloned().collect::<Vec<_>>();
 			pings.sort();
 			if !pings.is_empty() {
 				pub_peer_list.push(PeerInfo {
 					id: *id,
-					addr: info.addr,
 					state: info.state,
 					last_seen: info.last_seen,
-					avg_ping: Some(
-						pings
-							.iter()
-							.fold(Duration::from_secs(0), |x, y| x + *y)
-							.div_f64(pings.len() as f64),
-					),
+					avg_ping: Some(pings.iter().sum::<Duration>().div_f64(pings.len() as f64)),
 					max_ping: pings.last().cloned(),
 					med_ping: Some(pings[pings.len() / 2]),
 				});
 			} else {
 				pub_peer_list.push(PeerInfo {
 					id: *id,
-					addr: info.addr,
 					state: info.state,
 					last_seen: info.last_seen,
 					avg_ping: None,
@@ -495,15 +491,10 @@ impl PeeringManager {
 		}
 	}
 
-	async fn try_connect(
-		self: Arc<Self>,
-		id: NodeID,
-		default_addr: SocketAddr,
-		alternate_addrs: Vec<SocketAddr>,
-	) {
+	async fn try_connect(self: Arc<Self>, id: NodeID, addresses: Vec<SocketAddr>) {
 		let conn_addr = {
 			let mut ret = None;
-			for addr in [default_addr].iter().chain(alternate_addrs.iter()) {
+			for addr in addresses.iter() {
 				debug!("Trying address {} for peer {}", addr, hex::encode(&id[..8]));
 				match self.netapp.clone().try_connect(*addr, id).await {
 					Ok(()) => {
@@ -529,7 +520,7 @@ impl PeeringManager {
 			warn!(
 				"Could not connect to peer {} ({} addresses tried)",
 				hex::encode(&id[..8]),
-				1 + alternate_addrs.len()
+				addresses.len()
 			);
 			let mut known_hosts = self.known_hosts.write().unwrap();
 			if let Some(host) = known_hosts.list.get_mut(&id) {
@@ -549,6 +540,14 @@ impl PeeringManager {
 	}
 
 	fn on_connected(self: &Arc<Self>, id: NodeID, addr: SocketAddr, is_incoming: bool) {
+		if id == self.netapp.id {
+			// sanity check
+			panic!(
+				"on_connected from local node, id={:?}, addr={}, incoming={}",
+				id, addr, is_incoming
+			);
+		}
+
 		let mut known_hosts = self.known_hosts.write().unwrap();
 		if is_incoming {
 			if let Some(host) = known_hosts.list.get_mut(&id) {
@@ -563,13 +562,13 @@ impl PeeringManager {
 				addr
 			);
 			if let Some(host) = known_hosts.list.get_mut(&id) {
-				host.state = PeerConnState::Connected;
-				host.addr = addr;
+				host.state = PeerConnState::Connected { addr };
 				host.add_addr(addr);
 			} else {
-				known_hosts
-					.list
-					.insert(id, PeerInfoInternal::new(addr, PeerConnState::Connected));
+				known_hosts.list.insert(
+					id,
+					PeerInfoInternal::new(PeerConnState::Connected { addr }, Some(addr)),
+				);
 			}
 		}
 		known_hosts.update_hash();
@@ -589,12 +588,8 @@ impl PeeringManager {
 	}
 
 	fn new_peer(&self, id: &NodeID, addr: SocketAddr) -> PeerInfoInternal {
-		let state = if *id == self.netapp.id {
-			PeerConnState::Ourself
-		} else {
-			PeerConnState::Waiting(0, Instant::now())
-		};
-		PeerInfoInternal::new(addr, state)
+		assert!(*id != self.netapp.id);
+		PeerInfoInternal::new(PeerConnState::Waiting(0, Instant::now()), Some(addr))
 	}
 }