connection[_pool]: propagate UntranslatedEndpoint borrowing

As address translation no longer needs owned Strings, it's enough to
pass references to UntranslatedEndpoint to `open_connection` & friends.

Note that, unfortunately, we can't fully elide the clone in
`start_opening_connection()`, because the endpoint may be mutated (in
the shard-aware case) for the purpose of opening a connection to the
shard-aware port. Nonetheless, now the limitation that requires us to
clone is in internal code, not in the user-facing API, which is better
for us.

Author: wprzytula

scylla/src/network/connection.rs

@@ -2045,16 +2045,16 @@ impl Connection {
 }
 
 async fn maybe_translated_addr(
-    endpoint: UntranslatedEndpoint,
+    endpoint: &UntranslatedEndpoint,
     address_translator: Option<&dyn AddressTranslator>,
 ) -> Result<SocketAddr, TranslationError> {
-    match endpoint {
-        UntranslatedEndpoint::ContactPoint(addr) => Ok(addr.address),
+    match *endpoint {
+        UntranslatedEndpoint::ContactPoint(ref addr) => Ok(addr.address),
         UntranslatedEndpoint::Peer(PeerEndpoint {
             host_id,
             address,
-            datacenter,
-            rack,
+            ref datacenter,
+            ref rack,
         }) => match address {
             NodeAddr::Translatable(addr) => {
                 // In this case, addr is subject to AddressTranslator.
@@ -2089,7 +2089,7 @@ async fn maybe_translated_addr(
 ///
 /// At the beginning, translates node's address, if it is subject to address translation.
 pub(super) async fn open_connection(
-    endpoint: UntranslatedEndpoint,
+    endpoint: &UntranslatedEndpoint,
     source_port: Option<u16>,
     config: &HostConnectionConfig,
 ) -> Result<(Connection, ErrorReceiver), ConnectionError> {
@@ -2207,7 +2207,7 @@ pub(super) async fn open_connection(
 }
 
 pub(super) async fn open_connection_to_shard_aware_port(
-    endpoint: UntranslatedEndpoint,
+    endpoint: &UntranslatedEndpoint,
     shard: Shard,
     sharder: Sharder,
     config: &HostConnectionConfig,
@@ -2216,7 +2216,7 @@ pub(super) async fn open_connection_to_shard_aware_port(
     let source_port_iter = sharder.iter_source_ports_for_shard(shard);
 
     for port in source_port_iter {
-        let connect_result = open_connection(endpoint.clone(), Some(port), config).await;
+        let connect_result = open_connection(endpoint, Some(port), config).await;
 
         match connect_result {
             Err(err) if err.is_address_unavailable_for_use() => continue, // If we can't use this port, try the next one
@@ -2525,7 +2525,7 @@ mod tests {
         let addr: SocketAddr = resolve_hostname(&uri).await;
 
         let (connection, _) = super::open_connection(
-            UntranslatedEndpoint::ContactPoint(ResolvedContactPoint {
+            &UntranslatedEndpoint::ContactPoint(ResolvedContactPoint {
                 address: addr,
                 datacenter: None,
             }),
@@ -2650,7 +2650,7 @@ mod tests {
 
         let subtest = |enable_coalescing: bool, ks: String| async move {
             let (connection, _) = super::open_connection(
-                UntranslatedEndpoint::ContactPoint(ResolvedContactPoint {
+                &UntranslatedEndpoint::ContactPoint(ResolvedContactPoint {
                     address: addr,
                     datacenter: None,
                 }),
@@ -2782,16 +2782,20 @@ mod tests {
             .unwrap();
 
         // We must interrupt the driver's full connection opening, because our proxy does not interact further after Startup.
+        let endpoint = UntranslatedEndpoint::ContactPoint(ResolvedContactPoint {
+            address: proxy_addr,
+            datacenter: None,
+        });
         let (startup_without_lwt_optimisation, _shard) = select! {
-            _ = open_connection(UntranslatedEndpoint::ContactPoint(ResolvedContactPoint{address: proxy_addr, datacenter: None}), None, &config) => unreachable!(),
+            _ = open_connection(&endpoint, None, &config) => unreachable!(),
             startup = startup_rx.recv() => startup.unwrap(),
         };
 
         proxy.running_nodes[0]
             .change_request_rules(Some(make_rules(options_with_lwt_optimisation_support)));
 
         let (startup_with_lwt_optimisation, _shard) = select! {
-            _ = open_connection(UntranslatedEndpoint::ContactPoint(ResolvedContactPoint{address: proxy_addr, datacenter: None}), None, &config) => unreachable!(),
+            _ = open_connection(&endpoint, None, &config) => unreachable!(),
             startup = startup_rx.recv() => startup.unwrap(),
         };
 
@@ -2850,7 +2854,7 @@ mod tests {
 
         // Setup connection normally, without obstruction
         let (conn, mut error_receiver) = open_connection(
-            UntranslatedEndpoint::ContactPoint(ResolvedContactPoint {
+            &UntranslatedEndpoint::ContactPoint(ResolvedContactPoint {
                 address: proxy_addr,
                 datacenter: None,
             }),

scylla/src/network/connection_pool.rs

@@ -871,7 +871,7 @@ impl PoolRefiller {
                     endpoint
                 };
                 let result = open_connection_to_shard_aware_port(
-                    shard_aware_endpoint,
+                    &shard_aware_endpoint,
                     shard,
                     sharder.clone(),
                     &cfg,
@@ -886,7 +886,7 @@ impl PoolRefiller {
             .boxed(),
             _ => async move {
                 let non_shard_aware_endpoint = endpoint;
-                let result = open_connection(non_shard_aware_endpoint, None, &cfg).await;
+                let result = open_connection(&non_shard_aware_endpoint, None, &cfg).await;
                 OpenedConnectionEvent {
                     result,
                     requested_shard: None,
@@ -1170,20 +1170,15 @@ mod tests {
         // to the right shard
         let sharder = Sharder::new(ShardCount::new(3).unwrap(), 12);
 
+        let endpoint = UntranslatedEndpoint::ContactPoint(ResolvedContactPoint {
+            address: connect_address,
+            datacenter: None,
+        });
+
         // Open the connections
-        let mut conns = Vec::new();
-
-        for _ in 0..connections_number {
-            conns.push(open_connection_to_shard_aware_port(
-                UntranslatedEndpoint::ContactPoint(ResolvedContactPoint {
-                    address: connect_address,
-                    datacenter: None,
-                }),
-                0,
-                sharder.clone(),
-                &connection_config,
-            ));
-        }
+        let conns = (0..connections_number).map(|_| {
+            open_connection_to_shard_aware_port(&endpoint, 0, sharder.clone(), &connection_config)
+        });
 
         let joined = futures::future::join_all(conns).await;