session: move hostname resolution to MetadataReader::new()

This is a refactor that makes reduces number of parameters needed for
MetadataReader::new().

scylla/src/transport/cluster.rs

@@ -17,7 +17,7 @@ use arc_swap::ArcSwap;
 use futures::future::join_all;
 use futures::{future::RemoteHandle, FutureExt};
 use itertools::Itertools;
-use scylla_cql::errors::BadQuery;
+use scylla_cql::errors::{BadQuery, NewSessionError};
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::sync::Arc;
@@ -26,7 +26,7 @@ use tracing::instrument::WithSubscriber;
 use tracing::{debug, warn};
 use uuid::Uuid;
 
-use super::node::{KnownNode, NodeAddr, ResolvedContactPoint};
+use super::node::{KnownNode, NodeAddr};
 
 use super::locator::ReplicaLocator;
 use super::partitioner::calculate_token_for_partition_key;
@@ -140,13 +140,12 @@ struct UseKeyspaceRequest {
 impl Cluster {
     pub(crate) async fn new(
         known_nodes: Vec<KnownNode>,
-        initial_peers: Vec<ResolvedContactPoint>,
         pool_config: PoolConfig,
         keyspaces_to_fetch: Vec<String>,
         fetch_schema_metadata: bool,
         host_filter: Option<Arc<dyn HostFilter>>,
         cluster_metadata_refresh_interval: Duration,
-    ) -> Result<Cluster, QueryError> {
+    ) -> Result<Cluster, NewSessionError> {
         let (refresh_sender, refresh_receiver) = tokio::sync::mpsc::channel(32);
         let (use_keyspace_sender, use_keyspace_receiver) = tokio::sync::mpsc::channel(32);
         let (server_events_sender, server_events_receiver) = tokio::sync::mpsc::channel(32);
@@ -156,14 +155,14 @@ impl Cluster {
         let mut metadata_reader = MetadataReader::new(
             known_nodes,
             control_connection_repair_sender,
-            initial_peers,
             pool_config.connection_config.clone(),
             pool_config.keepalive_interval,
             server_events_sender,
             keyspaces_to_fetch,
             fetch_schema_metadata,
             &host_filter,
-        );
+        )
+        .await?;
 
         let metadata = metadata_reader.read_metadata(true).await?;
         let cluster_data = ClusterData::new(

scylla/src/transport/session.rs

@@ -38,7 +38,6 @@ use super::connection::QueryResponse;
 use super::connection::SslConfig;
 use super::errors::{NewSessionError, QueryError};
 use super::execution_profile::{ExecutionProfile, ExecutionProfileHandle, ExecutionProfileInner};
-use super::node::resolve_contact_points;
 #[cfg(feature = "cloud")]
 use super::node::CloudEndpoint;
 use super::node::KnownNode;
@@ -494,14 +493,6 @@ impl Session {
             return Err(NewSessionError::EmptyKnownNodesList);
         }
 
-        let (initial_peers, resolved_hostnames) = resolve_contact_points(&known_nodes).await;
-        // Ensure there is at least one resolved node
-        if initial_peers.is_empty() {
-            return Err(NewSessionError::FailedToResolveAnyHostname(
-                resolved_hostnames,
-            ));
-        }
-
         let connection_config = ConnectionConfig {
             compression: config.compression,
             tcp_nodelay: config.tcp_nodelay,
@@ -529,7 +520,6 @@ impl Session {
 
         let cluster = Cluster::new(
             known_nodes,
-            initial_peers,
             pool_config,
             config.keyspaces_to_fetch,
             config.fetch_schema_metadata,

scylla/src/transport/topology.rs

@@ -13,6 +13,7 @@ use futures::stream::{self, StreamExt, TryStreamExt};
 use futures::Stream;
 use rand::seq::SliceRandom;
 use rand::{thread_rng, Rng};
+use scylla_cql::errors::NewSessionError;
 use scylla_cql::frame::response::result::Row;
 use scylla_cql::frame::value::ValueList;
 use scylla_macros::FromRow;
@@ -400,21 +401,29 @@ impl Metadata {
 impl MetadataReader {
     /// Creates new MetadataReader, which connects to initially_known_peers in the background
     #[allow(clippy::too_many_arguments)]
-    pub(crate) fn new(
+    pub(crate) async fn new(
         initial_known_nodes: Vec<KnownNode>,
         control_connection_repair_requester: broadcast::Sender<()>,
-        initially_known_peers: Vec<ResolvedContactPoint>,
         mut connection_config: ConnectionConfig,
         keepalive_interval: Option<Duration>,
         server_event_sender: mpsc::Sender<Event>,
         keyspaces_to_fetch: Vec<String>,
         fetch_schema: bool,
         host_filter: &Option<Arc<dyn HostFilter>>,
-    ) -> Self {
+    ) -> Result<Self, NewSessionError> {
+        let (initial_peers, resolved_hostnames) =
+            resolve_contact_points(&initial_known_nodes).await;
+        // Ensure there is at least one resolved node
+        if initial_peers.is_empty() {
+            return Err(NewSessionError::FailedToResolveAnyHostname(
+                resolved_hostnames,
+            ));
+        }
+
         let control_connection_endpoint = UntranslatedEndpoint::ContactPoint(
-            initially_known_peers
+            initial_peers
                 .choose(&mut thread_rng())
-                .expect("Tried to initialize MetadataReader with empty known_peers list!")
+                .expect("Tried to initialize MetadataReader with empty initial_known_nodes list!")
                 .clone(),
         );
 
@@ -430,12 +439,12 @@ impl MetadataReader {
             control_connection_repair_requester.clone(),
         );
 
-        MetadataReader {
+        Ok(MetadataReader {
             control_connection_endpoint,
             control_connection,
             keepalive_interval,
             connection_config,
-            known_peers: initially_known_peers
+            known_peers: initial_peers
                 .into_iter()
                 .map(UntranslatedEndpoint::ContactPoint)
                 .collect(),
@@ -444,7 +453,7 @@ impl MetadataReader {
             host_filter: host_filter.clone(),
             initial_known_nodes,
             control_connection_repair_requester,
-        }
+        })
     }
 
     /// Fetches current metadata from the cluster