Repository: scylladb/scylla-rust-driver
Pull request #797 (2023-08-24 15:11:29)

proxy: add an example configurable from cmdline
The example is useful for manual testing of drivers with regard to address translation of DNS related functionalities, such as hostname mapping alterations combined with address changes.

## Pre-review checklist
- [x] I have split my patch into logically separate commits.
- [x] All commit messages clearly explain what they change and why.
- ~~[ ] I added relevant tests for new features and bug fixes.~~
- [x] All commits compile, pass static checks and pass test.
- [x] PR description sums up the changes and reasons why they should be introduced.
- ~~[ ] I have provided docstrings for the public items that I want to introduce.~~
- ~~[ ] I have adjusted the documentation in `./docs/source/`.~~
- ~~[ ] I added appropriate `Fixes:` annotations to PR description.~~



Patch:
diff --git a/scylla-cql/Cargo.toml b/scylla-cql/Cargo.toml
index 9a4a64f0..a4c21a4a 100644
--- a/scylla-cql/Cargo.toml
+++ b/scylla-cql/Cargo.toml
@@ -27,7 +27,7 @@ async-trait = "0.1.57"
 serde = { version = "1.0", optional = true }
 
 [dev-dependencies]
-criterion = "0.4" # Note: v0.5 needs at least rust 1.70.0
+criterion = "0.5"
 
 [[bench]]
 name = "benchmark"
diff --git a/scylla-proxy/examples/cmdline.rs b/scylla-proxy/examples/cmdline.rs
new file mode 100644
index 00000000..9ac8fd4c
--- /dev/null
+++ b/scylla-proxy/examples/cmdline.rs
@@ -0,0 +1,60 @@
+//! Command line-configurable transparent proxy.
+//!
+//! Useful for testing e.g. address translation or DNS related functionalities,
+//! such as alterations of hostname mapping combined with IP changes.
+//!
+//! How to use:
+//! cargo run --example cmdline -- [node_addr] [proxy_addr]
+//! This starts a proxy that forwards all traffic between a driver and [proxy_addr] to [node_addr].
+
+use std::{
+    net::{Ipv4Addr, SocketAddr},
+    str::FromStr,
+};
+
+use scylla_proxy::{Node, Proxy, ShardAwareness};
+use tracing::instrument::WithSubscriber;
+
+fn init_logger() {
+    tracing_subscriber::fmt::fmt()
+        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+        .without_time()
+        .init();
+}
+
+async fn pause() {
+    println!("Press Ctrl-C to stop the proxy...");
+    tokio::signal::ctrl_c().await.unwrap();
+}
+
+#[tokio::main]
+async fn main() {
+    init_logger();
+    let mut args = std::env::args();
+    args.next(); // skip name
+    let real_addr = args.next();
+    let proxy_addr = args.next();
+    let node_real_addr = SocketAddr::new(
+        std::net::IpAddr::V4(
+            Ipv4Addr::from_str(real_addr.as_deref().unwrap_or("127.0.0.2")).unwrap(),
+        ),
+        9042,
+    );
+    let node_proxy_addr = SocketAddr::new(
+        std::net::IpAddr::V4(
+            Ipv4Addr::from_str(proxy_addr.as_deref().unwrap_or("127.0.0.66")).unwrap(),
+        ),
+        9042,
+    );
+    let proxy = Proxy::new([Node::new(
+        node_real_addr,
+        node_proxy_addr,
+        ShardAwareness::QueryNode,
+        None,
+        None,
+    )]);
+    let running_proxy = proxy.run().with_current_subscriber().await.unwrap();
+
+    pause().await;
+    running_proxy.finish().await.unwrap();
+}
diff --git a/scylla-proxy/examples/identity_proxy.rs b/scylla-proxy/examples/identity_proxy.rs
index 3ebed037..33b6ea74 100644
--- a/scylla-proxy/examples/identity_proxy.rs
+++ b/scylla-proxy/examples/identity_proxy.rs
@@ -1,4 +1,4 @@
-use std::{io::Read as _, net::SocketAddr, str::FromStr};
+use std::{net::SocketAddr, str::FromStr};
 
 use scylla_proxy::{Node, Proxy, ShardAwareness};
 use tracing::instrument::WithSubscriber;
@@ -10,10 +10,9 @@ fn init_logger() {
         .init()
 }
 
-fn pause() {
-    println!("Press Enter to stop proxy...");
-    std::io::stdin().read_exact(&mut [0]).unwrap();
-    println!();
+async fn pause() {
+    println!("Press Ctrl-C to stop the proxy...");
+    tokio::signal::ctrl_c().await.unwrap();
 }
 
 #[tokio::main]
@@ -33,6 +32,6 @@ async fn main() {
         .build();
     let running_proxy = proxy.run().with_current_subscriber().await.unwrap();
 
-    pause();
+    pause().await;
     running_proxy.finish().await.unwrap();
 }
diff --git a/scylla-proxy/examples/identity_shard_aware_proxy.rs b/scylla-proxy/examples/identity_shard_aware_proxy.rs
index bcc5f0e7..6f94ad21 100644
--- a/scylla-proxy/examples/identity_shard_aware_proxy.rs
+++ b/scylla-proxy/examples/identity_shard_aware_proxy.rs
@@ -1,4 +1,4 @@
-use std::{io::Read as _, net::SocketAddr, str::FromStr};
+use std::{net::SocketAddr, str::FromStr};
 
 use scylla_proxy::{Node, Proxy, ShardAwareness};
 use tracing::instrument::WithSubscriber;
@@ -10,10 +10,9 @@ fn init_logger() {
         .init();
 }
 
-fn pause() {
-    println!("Press Enter to stop proxy...");
-    std::io::stdin().read_exact(&mut [0]).unwrap();
-    println!();
+async fn pause() {
+    println!("Press Ctrl-C to stop the proxy...");
+    tokio::signal::ctrl_c().await.unwrap();
 }
 
 #[tokio::main]
@@ -30,6 +29,6 @@ async fn main() {
     )]);
     let running_proxy = proxy.run().with_current_subscriber().await.unwrap();
 
-    pause();
+    pause().await;
     running_proxy.finish().await.unwrap();
 }
diff --git a/scylla-proxy/src/actions.rs b/scylla-proxy/src/actions.rs
index d571afd4..ef0945ba 100644
--- a/scylla-proxy/src/actions.rs
+++ b/scylla-proxy/src/actions.rs
@@ -4,9 +4,8 @@ use bytes::Bytes;
 use rand::{Rng, RngCore};
 use tokio::sync::mpsc;
 
-use crate::{
-    frame::{FrameOpcode, FrameParams, RequestFrame, RequestOpcode, ResponseFrame, ResponseOpcode},
-    TargetShard,
+use crate::frame::{
+    FrameOpcode, FrameParams, RequestFrame, RequestOpcode, ResponseFrame, ResponseOpcode,
 };
 use scylla_cql::{
     errors::{DbError, WriteType},
@@ -172,10 +171,7 @@ pub trait Reaction: Sized {
 
     /// Adds sending the matching frame as feedback using the provided channel.
     /// Modifies the existing `Reaction`.
-    fn with_feedback_when_performed(
-        self,
-        tx: mpsc::UnboundedSender<(Self::Incoming, Option<TargetShard>)>,
-    ) -> Self;
+    fn with_feedback_when_performed(self, tx: mpsc::UnboundedSender<Self::Incoming>) -> Self;
 }
 
 fn fmt_reaction(
@@ -206,7 +202,7 @@ pub struct RequestReaction {
     pub to_addressee: Option<Action<RequestFrame, RequestFrame>>,
     pub to_sender: Option<Action<RequestFrame, ResponseFrame>>,
     pub drop_connection: Option<Option<Duration>>,
-    pub feedback_channel: Option<mpsc::UnboundedSender<(RequestFrame, Option<TargetShard>)>>,
+    pub feedback_channel: Option<mpsc::UnboundedSender<RequestFrame>>,
 }
 
 impl fmt::Debug for RequestReaction {
@@ -227,7 +223,7 @@ pub struct ResponseReaction {
     pub to_addressee: Option<Action<ResponseFrame, ResponseFrame>>,
     pub to_sender: Option<Action<ResponseFrame, RequestFrame>>,
     pub drop_connection: Option<Option<Duration>>,
-    pub feedback_channel: Option<mpsc::UnboundedSender<(ResponseFrame, Option<TargetShard>)>>,
+    pub feedback_channel: Option<mpsc::UnboundedSender<ResponseFrame>>,
 }
 
 impl fmt::Debug for ResponseReaction {
@@ -325,10 +321,7 @@ impl Reaction for RequestReaction {
         }
     }
 
-    fn with_feedback_when_performed(
-        self,
-        tx: mpsc::UnboundedSender<(Self::Incoming, Option<TargetShard>)>,
-    ) -> Self {
+    fn with_feedback_when_performed(self, tx: mpsc::UnboundedSender<Self::Incoming>) -> Self {
         Self {
             feedback_channel: Some(tx),
             ..self
@@ -669,10 +662,7 @@ impl Reaction for ResponseReaction {
         }
     }
 
-    fn with_feedback_when_performed(
-        self,
-        tx: mpsc::UnboundedSender<(Self::Incoming, Option<TargetShard>)>,
-    ) -> Self {
+    fn with_feedback_when_performed(self, tx: mpsc::UnboundedSender<Self::Incoming>) -> Self {
         Self {
             feedback_channel: Some(tx),
             ..self
diff --git a/scylla-proxy/src/lib.rs b/scylla-proxy/src/lib.rs
index 5a4a12f5..8e91ae97 100644
--- a/scylla-proxy/src/lib.rs
+++ b/scylla-proxy/src/lib.rs
@@ -2,9 +2,6 @@ mod actions;
 mod errors;
 mod frame;
 mod proxy;
-
-pub type TargetShard = u16;
-
 pub use actions::{
     Action, Condition, Reaction, RequestReaction, RequestRule, ResponseReaction, ResponseRule,
 };
diff --git a/scylla-proxy/src/proxy.rs b/scylla-proxy/src/proxy.rs
index ae3eea3c..31a9c0cd 100644
--- a/scylla-proxy/src/proxy.rs
+++ b/scylla-proxy/src/proxy.rs
@@ -3,7 +3,7 @@ use crate::errors::{DoorkeeperError, ProxyError, WorkerError};
 use crate::frame::{
     self, read_response_frame, write_frame, FrameOpcode, FrameParams, RequestFrame, ResponseFrame,
 };
-use crate::{RequestOpcode, TargetShard};
+use crate::RequestOpcode;
 use bytes::Bytes;
 use scylla_cql::frame::types::read_string_multimap;
 use std::collections::HashMap;
@@ -206,18 +206,6 @@ impl Display for DisplayableRealAddrOption {
     }
 }
 
-#[derive(Clone, Copy)]
-struct DisplayableShard(Option<TargetShard>);
-impl Display for DisplayableShard {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        if let Some(shard) = self.0 {
-            write!(f, "shard {}", shard)
-        } else {
-            write!(f, "unknown shard")
-        }
-    }
-}
-
 enum InternalNode {
     Real {
         real_addr: SocketAddr,
@@ -612,7 +600,6 @@ impl Doorkeeper {
         connection_no: usize,
         driver_stream: TcpStream,
         cluster_stream: Option<TcpStream>,
-        shard: Option<TargetShard>,
     ) {
         let (driver_read, driver_write) = driver_stream.into_split();
 
@@ -624,7 +611,6 @@ impl Doorkeeper {
             driver_addr,
             real_addr: self.node.real_addr(),
             proxy_addr: self.node.proxy_addr(),
-            shard,
         };
 
         let (tx_request, rx_request) = mpsc::unbounded_channel::<RequestFrame>();
@@ -681,13 +667,11 @@ impl Doorkeeper {
         connection_no: usize,
     ) -> Result<(), DoorkeeperError> {
         let (driver_stream, driver_addr) = self.make_driver_stream(connection_no).await?;
-        let (cluster_stream, shard) = match self.node {
+        let cluster_stream = match self.node {
             InternalNode::Real { real_addr, .. } => {
-                let (cluster_stream, shard) =
-                    self.make_cluster_stream(driver_addr, real_addr).await?;
-                (Some(cluster_stream), shard)
+                Some(self.make_cluster_stream(driver_addr, real_addr).await?)
             }
-            InternalNode::Simulated { .. } => (None, None),
+            InternalNode::Simulated { .. } => None,
         };
 
         self.spawn_workers(
@@ -696,7 +680,6 @@ impl Doorkeeper {
             connection_no,
             driver_stream,
             cluster_stream,
-            shard,
         )
         .await;
 
@@ -724,8 +707,8 @@ impl Doorkeeper {
         &mut self,
         driver_addr: SocketAddr,
         real_addr: SocketAddr,
-    ) -> Result<(TcpStream, Option<TargetShard>), DoorkeeperError> {
-        let mut cluster_stream = if let Some(shards) = self.shards_count {
+    ) -> Result<TcpStream, DoorkeeperError> {
+        let cluster_stream = if let Some(shards) = self.shards_count {
             let socket = match self.node.proxy_addr().ip() {
                 std::net::IpAddr::V4(_) => TcpSocket::new_v4(),
                 std::net::IpAddr::V6(_) => TcpSocket::new_v6(),
@@ -748,7 +731,7 @@ impl Doorkeeper {
 
             socket.connect(real_addr).await.map(|ok| {
                 info!(
-                    "Connected to the cluster from {} at {}, intended shard {}.",
+                    "Connected to the cluster from {} at {}, shard {}.",
                     ok.local_addr().unwrap(),
                     real_addr,
                     shard_preserving_addr.port() % shards
@@ -763,53 +746,32 @@ impl Doorkeeper {
         }
         .map_err(|err| DoorkeeperError::NodeConnectionAttempt(real_addr, err))?;
 
-        // If ShardAwareness is aware (QueryNode or FixedNum variants) and the
-        // proxy succeeded to know the shards count (in FixedNum we get it for
-        // free, in QueryNode the initial Options query succceeded and Supported
-        // contained SCYLLA_SHARDS_NUM), then upon opening each connection to the
-        // node, the proxy issues another Options requests and acknowledges the
-        // shard it got connected to.
-        let shard = if self.shards_count.is_some() {
-            self.obtain_shard_number(real_addr, &mut cluster_stream)
-                .await?
-        } else {
-            None
-        };
-
-        Ok((cluster_stream, shard))
+        Ok(cluster_stream)
     }
 
     fn next_port_to_same_shard(&self, port: u16) -> u16 {
         port.wrapping_add(self.shards_count.unwrap())
     }
 
-    async fn get_supported_options(
-        connection: &mut TcpStream,
-    ) -> Result<HashMap<String, Vec<String>>, DoorkeeperError> {
+    async fn obtain_shards_count(&self, real_addr: SocketAddr) -> Result<u16, DoorkeeperError> {
+        let mut connection = TcpStream::connect(real_addr)
+            .await
+            .map_err(|err| DoorkeeperError::NodeConnectionAttempt(real_addr, err))?;
         write_frame(
             HARDCODED_OPTIONS_PARAMS,
             FrameOpcode::Request(RequestOpcode::Options),
             &Bytes::new(),
-            connection,
+            &mut connection,
         )
         .await
         .map_err(DoorkeeperError::ObtainingShardNumber)?;
 
-        let supported_frame = read_response_frame(connection)
+        let supported_frame = read_response_frame(&mut connection)
             .await
             .map_err(DoorkeeperError::ObtainingShardNumberFrame)?;
 
         let options = read_string_multimap(&mut supported_frame.body.as_ref())
             .map_err(DoorkeeperError::ObtainingShardNumberParseOptions)?;
-
-        Ok(options)
-    }
-
-    async fn obtain_shards_count(&self, real_addr: SocketAddr) -> Result<u16, DoorkeeperError> {
-        let mut connection = TcpStream::connect(real_addr)
-            .await
-            .map_err(|err| DoorkeeperError::NodeConnectionAttempt(real_addr, err))?;
-        let options = Self::get_supported_options(&mut connection).await?;
         let nr_shards_entry = options.get("SCYLLA_NR_SHARDS");
         let shards = match nr_shards_entry
             .and_then(|vec| vec.first())
@@ -823,24 +785,6 @@ impl Doorkeeper {
         info!("Obtained shards number on node {}: {}", real_addr, shards);
         Ok(shards)
     }
-
-    async fn obtain_shard_number(
-        &self,
-        real_addr: SocketAddr,
-        connection: &mut TcpStream,
-    ) -> Result<Option<TargetShard>, DoorkeeperError> {
-        let options = Self::get_supported_options(connection).await?;
-        let shard_entry = options.get("SCYLLA_SHARD");
-        let shard = shard_entry
-            .and_then(|vec| vec.first())
-            .map(|s| {
-                s.parse::<u16>()
-                    .map_err(DoorkeeperError::ObtainingShardNumberParseShardNumber)
-            })
-            .transpose()?;
-        info!("Connected to node {}, shard {:?}", real_addr, shard);
-        Ok(shard)
-    }
 }
 
 struct ProxyWorker {
@@ -851,17 +795,15 @@ struct ProxyWorker {
     driver_addr: SocketAddr,
     real_addr: Option<SocketAddr>,
     proxy_addr: SocketAddr,
-    shard: Option<TargetShard>,
 }
 
 impl ProxyWorker {
     fn exit(self, duty: &'static str) {
         debug!(
-            "Worker exits: [driver: {}, proxy: {}, node: {}, {}]::{}.",
+            "Worker exits: [driver: {}, proxy: {}, node: {}]::{}.",
             self.driver_addr,
             self.proxy_addr,
             DisplayableRealAddrOption(self.real_addr),
-            DisplayableShard(self.shard),
             duty
         );
         std::mem::drop(self.finish_guard);
@@ -892,7 +834,6 @@ impl ProxyWorker {
         mut read_half: (impl AsyncRead + Unpin),
         request_processor_tx: mpsc::UnboundedSender<RequestFrame>,
     ) {
-        let shard = self.shard;
         self.run_until_interrupted(
             "receiver_from_driver",
             |driver_addr, proxy_addr, _real_addr| async move {
@@ -905,11 +846,8 @@ impl ProxyWorker {
                         })?;
 
                     debug!(
-                        "Intercepted Driver ({}) -> Cluster ({}) ({}) frame. opcode: {:?}.",
-                        driver_addr,
-                        proxy_addr,
-                        DisplayableShard(shard),
-                        &frame.opcode
+                        "Intercepted Driver ({}) -> Cluster ({}) frame. opcode: {:?}.",
+                        driver_addr, proxy_addr, &frame.opcode
                     );
                     if request_processor_tx.send(frame).is_err() {
                         warn!("request_processor had exited.");
@@ -926,7 +864,6 @@ impl ProxyWorker {
         mut read_half: (impl AsyncRead + Unpin),
         response_processor_tx: mpsc::UnboundedSender<ResponseFrame>,
     ) {
-        let shard = self.shard;
         self.run_until_interrupted(
             "receiver_from_cluster",
             |driver_addr, _proxy_addr, real_addr| async move {
@@ -941,11 +878,8 @@ impl ProxyWorker {
                             })?;
 
                     debug!(
-                        "Intercepted Cluster ({}) -> Driver ({}) ({}) frame. opcode: {:?}.",
-                        real_addr,
-                        driver_addr,
-                        DisplayableShard(shard),
-                        &frame.opcode
+                        "Intercepted Cluster ({}) -> Driver ({}) frame. opcode: {:?}.",
+                        real_addr, driver_addr, &frame.opcode
                     );
 
                     if response_processor_tx.send(frame).is_err() {
@@ -965,7 +899,6 @@ impl ProxyWorker {
         mut connection_close_notifier: ConnectionCloseNotifier,
         mut terminate_notifier: TerminateNotifier,
     ) {
-        let shard = self.shard;
         self.run_until_interrupted(
             "sender_to_driver",
             |driver_addr, proxy_addr, _real_addr| async move {
@@ -983,11 +916,8 @@ impl ProxyWorker {
                     };
 
                     debug!(
-                        "Sending Proxy ({}) -> Driver ({}) ({}) frame. opcode: {:?}.",
-                        proxy_addr,
-                        driver_addr,
-                        DisplayableShard(shard),
-                        &response.opcode
+                        "Sending Proxy ({}) -> Driver ({}) frame. opcode: {:?}.",
+                        proxy_addr, driver_addr, &response.opcode
                     );
                     if response.write(&mut write_half).await.is_err() {
                         if terminate_notifier.try_recv().is_err()
@@ -1011,7 +941,6 @@ impl ProxyWorker {
         mut connection_close_notifier: ConnectionCloseNotifier,
         mut terminate_notifier: TerminateNotifier,
     ) {
-        let shard = self.shard;
         self.run_until_interrupted(
             "sender_to_driver",
             |_driver_addr, proxy_addr, real_addr| async move {
@@ -1030,11 +959,8 @@ impl ProxyWorker {
                     };
 
                     debug!(
-                        "Sending Proxy ({}) -> Cluster ({}) ({}) frame. opcode: {:?}.",
-                        proxy_addr,
-                        real_addr,
-                        DisplayableShard(shard),
-                        &request.opcode
+                        "Sending Proxy ({}) -> Cluster ({}) frame. opcode: {:?}.",
+                        proxy_addr, real_addr, &request.opcode
                     );
 
                     if request.write(&mut write_half).await.is_err() {
@@ -1061,7 +987,6 @@ impl ProxyWorker {
         request_rules: Arc<Mutex<Vec<RequestRule>>>,
         connection_close_signaler: ConnectionCloseSignaler,
     ) {
-        let shard = self.shard;
         self.run_until_interrupted("request_processor", |driver_addr, _, real_addr| async move {
             'mainloop: loop {
                 match requests_rx.recv().await {
@@ -1074,13 +999,13 @@ impl ProxyWorker {
                         let mut guard = request_rules.lock().unwrap();
                         '_ruleloop: for (i, request_rule) in guard.iter_mut().enumerate() {
                             if request_rule.0.eval(&ctx) {
-                                info!("Applying rule no={} to request ({} -> {} ({})).", i, driver_addr, DisplayableRealAddrOption(real_addr), DisplayableShard(shard));
+                                info!("Applying rule no={} to request ({} -> {}).", i, driver_addr, DisplayableRealAddrOption(real_addr));
                                 debug!("-> Applied rule: {:?}", request_rule);
                                 debug!("-> To request: {:?}", ctx.opcode);
                                 trace!("{:?}", request);
 
                                 if let Some(ref tx) = request_rule.1.feedback_channel {
-                                    tx.send((request.clone(), shard)).unwrap_or_else(|err|
+                                    tx.send(request.clone()).unwrap_or_else(|err|
                                         warn!("Could not send received request as feedback: {}", err)
                                     );
                                 }
@@ -1130,11 +1055,8 @@ impl ProxyWorker {
                                         }
                                         // close connection.
                                         info!(
-                                            "Dropping connection between {} and {} ({}) (as requested by a proxy rule)!",
-                                            driver_addr,
-                                            DisplayableRealAddrOption(real_addr),
-                                            DisplayableShard(shard),
-                                        );
+                                            "Dropping connection between {} and {} (as requested by a proxy rule)!"
+                                        , driver_addr, DisplayableRealAddrOption(real_addr));
                                         let _ = connection_close_signaler_clone.send(());
                                     }
                                 };
@@ -1164,7 +1086,6 @@ impl ProxyWorker {
         response_rules: Arc<Mutex<Vec<ResponseRule>>>,
         connection_close_signaler: ConnectionCloseSignaler,
     ) {
-        let shard = self.shard;
         self.run_until_interrupted("request_processor", |driver_addr, _, real_addr| async move {
             'mainloop: loop {
                 match responses_rx.recv().await {
@@ -1177,13 +1098,13 @@ impl ProxyWorker {
                         let mut guard = response_rules.lock().unwrap();
                         '_ruleloop: for (i, response_rule) in guard.iter_mut().enumerate() {
                             if response_rule.0.eval(&ctx) {
-                                info!("Applying rule no={} to request ({} -> {} ({})).", i, DisplayableRealAddrOption(real_addr), driver_addr, DisplayableShard(shard));
+                                info!("Applying rule no={} to request ({} -> {}).", i, DisplayableRealAddrOption(real_addr), driver_addr);
                                 debug!("-> Applied rule: {:?}", response_rule);
                                 debug!("-> To response: {:?}", ctx.opcode);
                                 trace!("{:?}", response);
 
                                 if let Some(ref tx) = response_rule.1.feedback_channel {
-                                    tx.send((response.clone(), shard)).unwrap_or_else(|err| warn!(
+                                    tx.send(response.clone()).unwrap_or_else(|err| warn!(
                                         "Could not send received response as feedback: {}", err
                                     ));
                                 }
@@ -1233,11 +1154,8 @@ impl ProxyWorker {
                                         }
                                         // close connection.
                                         info!(
-                                            "Dropping connection between {} and {} ({}) (as requested by a proxy rule)!",
-                                            driver_addr,
-                                            real_addr.expect("BUG: response rules are unavailable for dry-mode proxy!"),
-                                            DisplayableShard(shard)
-                                        );
+                                            "Dropping connection between {} and {} (as requested by a proxy rule)!"
+                                        , driver_addr, real_addr.expect("BUG: response rules are unavailable for dry-mode proxy!"));
                                         let _ = connection_close_signaler_clone.send(());
                                     }
                                 };
@@ -1285,7 +1203,7 @@ mod tests {
     use crate::{Condition, Reaction as _, RequestReaction, ResponseOpcode, ResponseReaction};
     use assert_matches::assert_matches;
     use bytes::{BufMut, BytesMut};
-    use futures::future::{join, join3};
+    use futures::future::join;
     use rand::RngCore;
     use scylla_cql::frame::frame_errors::FrameError;
     use scylla_cql::frame::types::write_string_multimap;
@@ -1312,21 +1230,23 @@ mod tests {
         body.freeze()
     }
 
-    async fn respond_with_supported(
-        conn: &mut TcpStream,
-        supported_options: &HashMap<String, Vec<String>>,
-    ) {
+    async fn respond_with_shards_number(mut conn: TcpStream, shards_num: u16) {
         let RequestFrame {
             params: recvd_params,
             opcode: recvd_opcode,
             body: recvd_body,
-        } = read_request_frame(conn).await.unwrap();
+        } = read_request_frame(&mut conn).await.unwrap();
         assert_eq!(recvd_params, HARDCODED_OPTIONS_PARAMS);
         assert_eq!(recvd_opcode, RequestOpcode::Options);
         assert_eq!(recvd_body, Bytes::new()); // body should be empty
 
         let mut body = BytesMut::new();
-        write_string_multimap(supported_options, &mut body).unwrap();
+        let mut sharded_info = HashMap::new();
+        sharded_info.insert(
+            String::from("SCYLLA_NR_SHARDS"),
+            vec![shards_num.to_string()],
+        );
+        write_string_multimap(&sharded_info, &mut body).unwrap();
 
         let body = body.freeze();
 
@@ -1334,35 +1254,12 @@ mod tests {
             HARDCODED_OPTIONS_PARAMS.for_response(),
             FrameOpcode::Response(ResponseOpcode::Supported),
             &body,
-            conn,
+            &mut conn,
         )
         .await
         .unwrap();
     }
 
-    fn supported_shards_count(shards_count: u16) -> HashMap<String, Vec<String>> {
-        let mut sharded_info = HashMap::new();
-        sharded_info.insert(
-            String::from("SCYLLA_NR_SHARDS"),
-            vec![shards_count.to_string()],
-        );
-        sharded_info
-    }
-
-    fn supported_shard_number(shard_num: TargetShard) -> HashMap<String, Vec<String>> {
-        let mut sharded_info = HashMap::new();
-        sharded_info.insert(String::from("SCYLLA_SHARD"), vec![shard_num.to_string()]);
-        sharded_info
-    }
-
-    async fn respond_with_shards_count(conn: &mut TcpStream, shards_count: u16) {
-        respond_with_supported(conn, &supported_shards_count(shards_count)).await;
-    }
-
-    async fn respond_with_shard_num(conn: &mut TcpStream, shard_num: TargetShard) {
-        respond_with_supported(conn, &supported_shard_number(shard_num)).await;
-    }
-
     fn next_local_address_with_port(port: u16) -> SocketAddr {
         SocketAddr::new(get_exclusive_local_address(), port)
     }
@@ -1390,7 +1287,7 @@ mod tests {
 
         let body = random_body();
 
-        let send_frame_to_shard = async {
+        let mock_driver_action = async {
             let mut conn = TcpStream::connect(node1_proxy_addr).await.unwrap();
 
             write_frame(params, opcode, &body, &mut conn).await.unwrap();
@@ -1399,13 +1296,9 @@ mod tests {
 
         let mock_node_action = async {
             if let ShardAwareness::QueryNode = shard_awareness {
-                respond_with_shards_count(&mut mock_node_listener.accept().await.unwrap().0, 1)
-                    .await;
+                respond_with_shards_number(mock_node_listener.accept().await.unwrap().0, 1).await;
             }
             let (mut conn, _) = mock_node_listener.accept().await.unwrap();
-            if shard_awareness.is_aware() {
-                respond_with_shard_num(&mut conn, 1).await;
-            }
             let RequestFrame {
                 params: recvd_params,
                 opcode: recvd_opcode,
@@ -1418,7 +1311,7 @@ mod tests {
         };
 
         // we keep the connections open until proxy finishes to let it perform clean exit with no disconnects
-        let (_node_conn, _driver_conn) = join(mock_node_action, send_frame_to_shard).await;
+        let (_node_conn, _driver_conn) = join(mock_node_action, mock_driver_action).await;
         running_proxy.finish().await.unwrap();
     }
 
@@ -1454,7 +1347,7 @@ mod tests {
 
             let (driver_addr_tx, driver_addr_rx) = oneshot::channel::<SocketAddr>();
 
-            let send_frame_to_shard = async {
+            let mock_driver_action = async {
                 let socket = TcpSocket::new_v4().unwrap();
                 socket
                     .bind(SocketAddr::from_str("0.0.0.0:0").unwrap())
@@ -1475,7 +1368,7 @@ mod tests {
             };
 
             // we keep the connections open until proxy finishes to let it perform clean exit with no disconnects
-            let (_node_conn, _driver_conn) = join(mock_node_action, send_frame_to_shard).await;
+            let (_node_conn, _driver_conn) = join(mock_node_action, mock_driver_action).await;
             running_proxy.finish().await.unwrap();
         }
 
@@ -1505,7 +1398,7 @@ mod tests {
                 let (driver_addr_tx, driver_addr_rx) = oneshot::channel::<SocketAddr>();
 
                 let mock_driver_addr = next_local_address_with_port(shards_num * 1234 + shard_num);
-                let send_frame_to_shard = async {
+                let mock_driver_action = async {
                     let socket = TcpSocket::new_v4().unwrap();
                     socket
                         .bind(mock_driver_addr)
@@ -1515,8 +1408,8 @@ mod tests {
                 };
 
                 let mock_node_action = async {
-                    respond_with_shards_count(
-                        &mut mock_node_listener.accept().await.unwrap().0,
+                    respond_with_shards_number(
+                        mock_node_listener.accept().await.unwrap().0,
                         shards_num,
                     )
                     .await;
@@ -1529,7 +1422,7 @@ mod tests {
                     conn
                 };
 
-                let (_node_conn, _driver_conn) = join(mock_node_action, send_frame_to_shard).await;
+                let (_node_conn, _driver_conn) = join(mock_node_action, mock_driver_action).await;
                 running_proxy.finish().await.unwrap();
             }
         }
@@ -1614,7 +1507,7 @@ mod tests {
             body.freeze()
         };
 
-        let send_frame_to_shard = async {
+        let mock_driver_action = async {
             let mut conn = TcpStream::connect(node1_proxy_addr).await.unwrap();
 
             write_frame(params1, opcode1, &body1, &mut conn)
@@ -1662,7 +1555,7 @@ mod tests {
             conn
         };
 
-        let (mut node_conn, mut driver_conn) = join(mock_node_action, send_frame_to_shard).await;
+        let (mut node_conn, mut driver_conn) = join(mock_node_action, mock_driver_action).await;
 
         running_proxy.finish().await.unwrap();
 
@@ -1900,7 +1793,7 @@ mod tests {
 
         let body = random_body();
 
-        let send_frame_to_shard = async {
+        let mock_driver_action = async {
             let mut conn = TcpStream::connect(node1_proxy_addr).await.unwrap();
             write_frame(params, request_opcode, &body, &mut conn)
                 .await
@@ -1917,16 +1810,16 @@ mod tests {
         };
 
         // we keep the connections open until proxy finishes to let it perform clean exit with no disconnects
-        let (_node_conn, _driver_conn) = join(mock_node_action, send_frame_to_shard).await;
+        let (_node_conn, _driver_conn) = join(mock_node_action, mock_driver_action).await;
 
-        let (feedback_request, _shard) = request_feedback_rx.recv().await.unwrap();
+        let feedback_request = request_feedback_rx.recv().await.unwrap();
         assert_eq!(feedback_request.params, params);
         assert_eq!(
             FrameOpcode::Request(feedback_request.opcode),
             request_opcode
         );
         assert_eq!(feedback_request.body, body);
-        let (feedback_response, _shard) = response_feedback_rx.recv().await.unwrap();
+        let feedback_response = response_feedback_rx.recv().await.unwrap();
         assert_eq!(feedback_response.params, params.for_response());
         assert_eq!(
             FrameOpcode::Response(feedback_response.opcode),
@@ -1953,7 +1846,7 @@ mod tests {
 
         let mock_node_listener = TcpListener::bind(node1_real_addr).await.unwrap();
 
-        let send_frame_to_shard = async {
+        let mock_driver_action = async {
             let mut conn = TcpStream::connect(node1_proxy_addr).await.unwrap();
 
             conn.write_all(b"uselessJunk").await.unwrap();
@@ -1965,7 +1858,7 @@ mod tests {
             conn
         };
 
-        let (node_conn, driver_conn) = join(mock_node_action, send_frame_to_shard).await;
+        let (node_conn, driver_conn) = join(mock_node_action, mock_driver_action).await;
 
         running_proxy.sanity_check().unwrap();
 
@@ -2028,7 +1921,7 @@ mod tests {
 
         let body2 = random_body();
 
-        let send_frame_to_shard = async {
+        let mock_driver_action = async {
             let mut conn = TcpStream::connect(node1_proxy_addr).await.unwrap();
 
             write_frame(params1, opcode1, &body1, &mut conn)
@@ -2055,7 +1948,7 @@ mod tests {
 
         // we keep the connections open until proxy finishes to let it perform clean exit with no disconnects
         let (_node_conn, _driver_conn) =
-            tokio::time::timeout(delay, join(mock_node_action, send_frame_to_shard))
+            tokio::time::timeout(delay, join(mock_node_action, mock_driver_action))
                 .await
                 .expect("Request processing was not concurrent");
         running_proxy.finish().await.unwrap();
@@ -2188,155 +2081,4 @@ mod tests {
 
         assert_matches!(conn.read(&mut [0u8; 1]).await, Ok(0));
     }
-
-    // The test asserts that once a (mock) driver connects to the proxy from some port,
-    // the proxy will connect to a shard corresponding to that port and that the target
-    // shard number will be sent through the feedback channel.
-    #[tokio::test]
-    #[ntest::timeout(1000)]
-    async fn proxy_reports_target_shard_as_feedback() {
-        init_logger();
-
-        let node_port = 10101;
-        let node_real_addr = next_local_address_with_port(node_port);
-        let mock_node_listener = TcpListener::bind(node_real_addr).await.unwrap();
-
-        let params = FrameParams {
-            flags: 0,
-            version: 0x04,
-            stream: 0,
-        };
-        let request_opcode = FrameOpcode::Request(RequestOpcode::Options);
-        let response_opcode = FrameOpcode::Response(ResponseOpcode::Ready);
-
-        let body = random_body();
-
-        for shards_count in 2..9 {
-            // Two driver connections are simulated, each to a different shard.
-            let driver1_shard = shards_count - 1;
-            let driver2_shard = shards_count - 2;
-            let node_proxy_addr = next_local_address_with_port(node_port);
-
-            let (request_feedback_tx, mut request_feedback_rx) = mpsc::unbounded_channel();
-            let (response_feedback_tx, mut response_feedback_rx) = mpsc::unbounded_channel();
-
-            let proxy = Proxy::new([Node::new(
-                node_real_addr,
-                node_proxy_addr,
-                ShardAwareness::FixedNum(shards_count),
-                Some(vec![RequestRule(
-                    Condition::True,
-                    RequestReaction::drop_frame().with_feedback_when_performed(request_feedback_tx),
-                )]),
-                Some(vec![ResponseRule(
-                    Condition::True,
-                    ResponseReaction::drop_frame()
-                        .with_feedback_when_performed(response_feedback_tx),
-                )]),
-            )]);
-            let running_proxy = proxy.run().await.unwrap();
-
-            /// Choose a source port `p` such that `shard == shard_of_source_port(p)`.
-            fn draw_source_port_for_shard(shards_count: u16, shard: u16) -> u16 {
-                assert!(shard < shards_count);
-                (49152 + shards_count - 1) / shards_count * shards_count + shard
-            }
-
-            async fn bind_socket_for_shard(shards_count: u16, shard: u16) -> TcpSocket {
-                let socket = TcpSocket::new_v4().unwrap();
-                let initial_port = draw_source_port_for_shard(shards_count, shard);
-
-                let mut desired_addr =
-                    SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), initial_port);
-                while socket.bind(desired_addr).is_err() {
-                    // in search for a port that translates to the desired shard
-                    let next_port = desired_addr.port().wrapping_add(shards_count);
-                    if next_port == initial_port {
-                        panic!("No more ports left");
-                    }
-                    desired_addr.set_port(next_port);
-                }
-
-                socket
-            }
-
-            let body_ref = &body;
-            let send_frame_to_shard = |driver_shard: u16| async move {
-                let socket = bind_socket_for_shard(shards_count, driver_shard).await;
-                let mut conn = socket.connect(node_proxy_addr).await.unwrap();
-
-                write_frame(params, request_opcode, body_ref, &mut conn)
-                    .await
-                    .unwrap();
-                conn
-            };
-
-            let mock_driver1_action = send_frame_to_shard(driver1_shard);
-            let mock_driver2_action = send_frame_to_shard(driver2_shard);
-
-            // Accepts two connections and sends a response to each of them.
-            let mock_node_action = async {
-                let mut conns_futs = (0..2)
-                    .map(|_| async {
-                        let (mut conn, driver_addr) = mock_node_listener.accept().await.unwrap();
-                        respond_with_shard_num(&mut conn, driver_addr.port() % shards_count).await;
-                        write_frame(params.for_response(), response_opcode, body_ref, &mut conn)
-                            .await
-                            .unwrap();
-                        conn
-                    })
-                    .collect::<Vec<_>>();
-                let conn2 = conns_futs.pop().unwrap().await;
-                let conn1 = conns_futs.pop().unwrap().await;
-                (conn1, conn2)
-            };
-
-            // we keep the connections open until proxy finishes to let it perform clean exit with no disconnects
-            let (_node_conns, _driver1_conn, _driver2_conn) =
-                join3(mock_node_action, mock_driver1_action, mock_driver2_action).await;
-
-            let assert_feedback_request = |feedback_request: RequestFrame| {
-                assert_eq!(feedback_request.params, params);
-                assert_eq!(
-                    FrameOpcode::Request(feedback_request.opcode),
-                    request_opcode
-                );
-                assert_eq!(feedback_request.body, body);
-            };
-
-            let assert_feedback_response = |feedback_response: ResponseFrame| {
-                assert_eq!(feedback_response.params, params.for_response());
-                assert_eq!(
-                    FrameOpcode::Response(feedback_response.opcode),
-                    response_opcode
-                );
-                assert_eq!(feedback_response.body, body);
-            };
-
-            let (feedback_request, shard1) = request_feedback_rx.recv().await.unwrap();
-            assert_feedback_request(feedback_request);
-            let (feedback_request, shard2) = request_feedback_rx.recv().await.unwrap();
-            assert_feedback_request(feedback_request);
-            let (feedback_response, shard3) = response_feedback_rx.recv().await.unwrap();
-            assert_feedback_response(feedback_response);
-            let (feedback_response, shard4) = response_feedback_rx.recv().await.unwrap();
-            assert_feedback_response(feedback_response);
-
-            // expected: {driver1_shard request, driver1_shard response, driver2_shard request, driver2_shard response}
-            let mut expected_shards = [driver1_shard, driver1_shard, driver2_shard, driver2_shard];
-            expected_shards.sort_unstable();
-
-            let mut got_shards = [
-                shard1.unwrap(),
-                shard2.unwrap(),
-                shard3.unwrap(),
-                shard4.unwrap(),
-            ];
-            got_shards.sort_unstable();
-
-            assert_eq!(expected_shards, got_shards);
-
-            running_proxy.finish().await.unwrap();
-        }
-    }
 }
diff --git a/scylla/Cargo.toml b/scylla/Cargo.toml
index 2460e020..f83bdbd2 100644
--- a/scylla/Cargo.toml
+++ b/scylla/Cargo.toml
@@ -56,7 +56,7 @@ socket2 = { version = "0.5.3", features = ["all"] }
 [dev-dependencies]
 scylla-proxy = { version = "0.0.3", path = "../scylla-proxy" }
 ntest = "0.9.0"
-criterion = "0.4" # Note: v0.5 needs at least rust 1.70.0
+criterion = "0.5"
 tracing-subscriber = { version = "0.3.14", features = ["env-filter"] }
 assert_matches = "1.5.0"
 rand_chacha = "0.3.1"
diff --git a/scylla/src/transport/cluster.rs b/scylla/src/transport/cluster.rs
index 503d1451..4e436d29 100644
--- a/scylla/src/transport/cluster.rs
+++ b/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, NewSessionError};
+use scylla_cql::errors::BadQuery;
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::sync::Arc;
@@ -26,7 +26,14 @@ use tracing::instrument::WithSubscriber;
 use tracing::{debug, warn};
 use uuid::Uuid;
 
-use super::node::{KnownNode, NodeAddr};
+use super::node::NodeAddr;
+
+#[derive(Debug, Clone)]
+#[non_exhaustive]
+pub struct ContactPoint {
+    pub address: SocketAddr,
+    pub datacenter: Option<String>,
+}
 
 use super::locator::ReplicaLocator;
 use super::partitioner::calculate_token_for_partition_key;
@@ -111,9 +118,6 @@ struct ClusterWorker {
     // Channel used to receive server events
     server_events_channel: tokio::sync::mpsc::Receiver<Event>,
 
-    // Channel used to receive signals that control connection is broken
-    control_connection_repair_channel: tokio::sync::broadcast::Receiver<()>,
-
     // Keyspace send in "USE <keyspace name>" when opening each connection
     used_keyspace: Option<VerifiedKeyspaceName>,
 
@@ -139,30 +143,26 @@ struct UseKeyspaceRequest {
 
 impl Cluster {
     pub(crate) async fn new(
-        known_nodes: Vec<KnownNode>,
+        initial_peers: Vec<ContactPoint>,
         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, NewSessionError> {
+    ) -> Result<Cluster, QueryError> {
         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);
-        let (control_connection_repair_sender, control_connection_repair_receiver) =
-            tokio::sync::broadcast::channel(32);
 
         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(
@@ -185,7 +185,6 @@ impl Cluster {
 
             refresh_channel: refresh_receiver,
             server_events_channel: server_events_receiver,
-            control_connection_repair_channel: control_connection_repair_receiver,
 
             use_keyspace_channel: use_keyspace_receiver,
             used_keyspace: None,
@@ -480,20 +479,14 @@ impl ClusterWorker {
     pub(crate) async fn work(mut self) {
         use tokio::time::Instant;
 
-        let control_connection_repair_duration = Duration::from_secs(1); // Attempt control connection repair every second
         let mut last_refresh_time = Instant::now();
-        let mut control_connection_works = true;
 
         loop {
             let mut cur_request: Option<RefreshRequest> = None;
 
             // Wait until it's time for the next refresh
             let sleep_until: Instant = last_refresh_time
-                .checked_add(if control_connection_works {
-                    self.cluster_metadata_refresh_interval
-                } else {
-                    control_connection_repair_duration
-                })
+                .checked_add(self.cluster_metadata_refresh_interval)
                 .unwrap_or_else(Instant::now);
 
             let sleep_future = tokio::time::sleep_until(sleep_until);
@@ -544,26 +537,6 @@ impl ClusterWorker {
 
                     continue; // Don't go to refreshing, wait for the next event
                 }
-                recv_res = self.control_connection_repair_channel.recv() => {
-                    match recv_res {
-                        Ok(()) => {
-                            // The control connection was broken. Acknowledge that and start attempting to reconnect.
-                            // The first reconnect attempt will be immediate (by attempting metadata refresh below),
-                            // and if it does not succeed, then `control_connection_works` will be set to `false`,
-                            // so subsequent attempts will be issued every second.
-                        }
-                        Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
-                            // This is very unlikely; we would have to have a lot of concurrent
-                            // control connections opened and broken at the same time.
-                            // The best we can do is ignoring this.
-                        }
-                        Err(tokio::sync::broadcast::error::RecvError::Closed) => {
-                            // If control_connection_repair_channel was closed then MetadataReader was dropped,
-                            // we can stop working.
-                            return;
-                        }
-                    }
-                }
             }
 
             // Perform the refresh
@@ -571,8 +544,6 @@ impl ClusterWorker {
             last_refresh_time = Instant::now();
             let refresh_res = self.perform_refresh().await;
 
-            control_connection_works = refresh_res.is_ok();
-
             // Send refresh result if there was a request
             if let Some(request) = cur_request {
                 // We can ignore sending error - if no one waits for the response we can drop it
diff --git a/scylla/src/transport/connection.rs b/scylla/src/transport/connection.rs
index bd12ed06..2eb66f7e 100644
--- a/scylla/src/transport/connection.rs
+++ b/scylla/src/transport/connection.rs
@@ -1764,8 +1764,8 @@ mod tests {
 
     use super::ConnectionConfig;
     use crate::query::Query;
+    use crate::transport::cluster::ContactPoint;
     use crate::transport::connection::open_connection;
-    use crate::transport::node::ResolvedContactPoint;
     use crate::transport::topology::UntranslatedEndpoint;
     use crate::utils::test_utils::unique_keyspace_name;
     use crate::{IntoTypedRows, SessionBuilder};
@@ -1801,7 +1801,7 @@ mod tests {
         let addr: SocketAddr = resolve_hostname(&uri).await;
 
         let (connection, _) = super::open_connection(
-            UntranslatedEndpoint::ContactPoint(ResolvedContactPoint {
+            UntranslatedEndpoint::ContactPoint(ContactPoint {
                 address: addr,
                 datacenter: None,
             }),
@@ -1919,7 +1919,7 @@ mod tests {
 
         let subtest = |enable_coalescing: bool, ks: String| async move {
             let (connection, _) = super::open_connection(
-                UntranslatedEndpoint::ContactPoint(ResolvedContactPoint {
+                UntranslatedEndpoint::ContactPoint(ContactPoint {
                     address: addr,
                     datacenter: None,
                 }),
@@ -2048,16 +2048,16 @@ mod tests {
             .unwrap();
 
         // We must interrupt the driver's full connection opening, because our proxy does not interact further after Startup.
-        let (startup_without_lwt_optimisation, _shard) = select! {
-            _ = open_connection(UntranslatedEndpoint::ContactPoint(ResolvedContactPoint{address: proxy_addr, datacenter: None}), None, config.clone()) => unreachable!(),
+        let startup_without_lwt_optimisation = select! {
+            _ = open_connection(UntranslatedEndpoint::ContactPoint(ContactPoint{address: proxy_addr, datacenter: None}), None, config.clone()) => 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.clone()) => unreachable!(),
+        let startup_with_lwt_optimisation = select! {
+            _ = open_connection(UntranslatedEndpoint::ContactPoint(ContactPoint{address: proxy_addr, datacenter: None}), None, config.clone()) => unreachable!(),
             startup = startup_rx.recv() => startup.unwrap(),
         };
 
@@ -2112,7 +2112,7 @@ mod tests {
 
         // Setup connection normally, without obstruction
         let (conn, mut error_receiver) = open_connection(
-            UntranslatedEndpoint::ContactPoint(ResolvedContactPoint {
+            UntranslatedEndpoint::ContactPoint(ContactPoint {
                 address: proxy_addr,
                 datacenter: None,
             }),
diff --git a/scylla/src/transport/connection_pool.rs b/scylla/src/transport/connection_pool.rs
index f26ea36a..0ee16c28 100644
--- a/scylla/src/transport/connection_pool.rs
+++ b/scylla/src/transport/connection_pool.rs
@@ -9,10 +9,10 @@ use crate::transport::{
 };
 
 #[cfg(feature = "cloud")]
-use super::node::resolve_hostname;
+use super::session::resolve_hostname;
 
 #[cfg(feature = "cloud")]
-use super::node::ResolvedContactPoint;
+use super::cluster::ContactPoint;
 use super::topology::{PeerEndpoint, UntranslatedEndpoint};
 use super::NodeAddr;
 
@@ -25,8 +25,7 @@ use std::num::NonZeroUsize;
 use std::pin::Pin;
 use std::sync::{Arc, RwLock, Weak};
 use std::time::Duration;
-
-use tokio::sync::{broadcast, mpsc, Notify};
+use tokio::sync::{mpsc, Notify};
 use tracing::instrument::WithSubscriber;
 use tracing::{debug, trace, warn};
 
@@ -170,7 +169,6 @@ impl NodeConnectionPool {
         endpoint: UntranslatedEndpoint,
         #[allow(unused_mut)] mut pool_config: PoolConfig, // `mut` needed only with "cloud" feature
         current_keyspace: Option<VerifiedKeyspaceName>,
-        pool_empty_notifier: broadcast::Sender<()>,
     ) -> Self {
         let (use_keyspace_request_sender, use_keyspace_request_receiver) = mpsc::channel(1);
         let pool_updated_notify = Arc::new(Notify::new());
@@ -178,7 +176,7 @@ impl NodeConnectionPool {
         #[cfg(feature = "cloud")]
         if pool_config.connection_config.cloud_config.is_some() {
             let (host_id, address, dc) = match endpoint {
-                UntranslatedEndpoint::ContactPoint(ResolvedContactPoint {
+                UntranslatedEndpoint::ContactPoint(ContactPoint {
                     address,
                     ref datacenter,
                 }) => (None, address, datacenter.as_deref()), // FIXME: Pass DC in ContactPoint
@@ -207,7 +205,6 @@ impl NodeConnectionPool {
             pool_config,
             current_keyspace,
             pool_updated_notify.clone(),
-            pool_empty_notifier,
         );
 
         let conns = refiller.get_shared_connections();
@@ -475,9 +472,6 @@ struct PoolRefiller {
 
     // Signaled when the connection pool is updated
     pool_updated_notify: Arc<Notify>,
-
-    // Signaled when the connection pool becomes empty
-    pool_empty_notifier: broadcast::Sender<()>,
 }
 
 #[derive(Debug)]
@@ -492,7 +486,6 @@ impl PoolRefiller {
         pool_config: PoolConfig,
         current_keyspace: Option<VerifiedKeyspaceName>,
         pool_updated_notify: Arc<Notify>,
-        pool_empty_notifier: broadcast::Sender<()>,
     ) -> Self {
         // At the beginning, we assume the node does not have any shards
         // and assume that the node is a Cassandra node
@@ -520,7 +513,6 @@ impl PoolRefiller {
             current_keyspace,
 
             pool_updated_notify,
-            pool_empty_notifier,
         }
     }
 
@@ -1045,9 +1037,6 @@ impl PoolRefiller {
                 self.conns[shard_id].len(),
                 self.active_connection_count(),
             );
-            if !self.has_connections() {
-                let _ = self.pool_empty_notifier.send(());
-            }
             self.update_shared_conns(Some(last_error));
             return;
         }
@@ -1259,8 +1248,8 @@ async fn open_connection_to_shard_aware_port(
 mod tests {
     use super::open_connection_to_shard_aware_port;
     use crate::routing::{ShardCount, Sharder};
+    use crate::transport::cluster::ContactPoint;
     use crate::transport::connection::ConnectionConfig;
-    use crate::transport::node::ResolvedContactPoint;
     use crate::transport::topology::UntranslatedEndpoint;
     use std::net::{SocketAddr, ToSocketAddrs};
 
@@ -1297,7 +1286,7 @@ mod tests {
 
         for _ in 0..connections_number {
             conns.push(open_connection_to_shard_aware_port(
-                UntranslatedEndpoint::ContactPoint(ResolvedContactPoint {
+                UntranslatedEndpoint::ContactPoint(ContactPoint {
                     address: connect_address,
                     datacenter: None,
                 }),
diff --git a/scylla/src/transport/mod.rs b/scylla/src/transport/mod.rs
index 6025bf63..3b6bea38 100644
--- a/scylla/src/transport/mod.rs
+++ b/scylla/src/transport/mod.rs
@@ -35,4 +35,4 @@ mod cql_types_test;
 mod cql_value_test;
 
 pub use cluster::ClusterData;
-pub use node::{KnownNode, Node, NodeAddr, NodeRef};
+pub use node::{Node, NodeAddr, NodeRef};
diff --git a/scylla/src/transport/node.rs b/scylla/src/transport/node.rs
index 97b26794..303473d7 100644
--- a/scylla/src/transport/node.rs
+++ b/scylla/src/transport/node.rs
@@ -1,5 +1,3 @@
-use tokio::net::lookup_host;
-use tracing::warn;
 use uuid::Uuid;
 
 /// Node represents a cluster node along with it's data and connections
@@ -10,7 +8,6 @@ use crate::transport::connection_pool::{NodeConnectionPool, PoolConfig};
 use crate::transport::errors::QueryError;
 
 use std::fmt::Display;
-use std::io;
 use std::net::IpAddr;
 use std::{
     hash::{Hash, Hasher},
@@ -103,16 +100,8 @@ impl Node {
         let address = peer.address;
         let datacenter = peer.datacenter.clone();
         let rack = peer.rack.clone();
-
-        // We aren't interested in the fact that the pool becomes empty, so we immediately drop the receiving part.
-        let (pool_empty_notifier, _) = tokio::sync::broadcast::channel(1);
         let pool = enabled.then(|| {
-            NodeConnectionPool::new(
-                UntranslatedEndpoint::Peer(peer),
-                pool_config,
-                keyspace_name,
-                pool_empty_notifier,
-            )
+            NodeConnectionPool::new(UntranslatedEndpoint::Peer(peer), pool_config, keyspace_name)
         });
 
         Node {
@@ -226,110 +215,6 @@ impl Hash for Node {
     }
 }
 
-/// Describes a database server known on `Session` startup.
-///
-/// The name derives from SessionBuilder's `known_node()` family of methods.
-#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)]
-#[non_exhaustive]
-pub enum KnownNode {
-    Hostname(String),
-    Address(SocketAddr),
-    #[cfg(feature = "cloud")]
-    CloudEndpoint(CloudEndpoint),
-}
-
-/// Describes a database server in the serverless Scylla Cloud.
-#[cfg(feature = "cloud")]
-#[non_exhaustive]
-#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)]
-pub struct CloudEndpoint {
-    pub hostname: String,
-    pub datacenter: String,
-}
-
-/// Describes a database server known on Session startup, with already resolved address.
-#[derive(Debug, Clone)]
-#[non_exhaustive]
-pub struct ResolvedContactPoint {
-    pub address: SocketAddr,
-    pub datacenter: Option<String>,
-}
-
-// Resolve the given hostname using a DNS lookup if necessary.
-// The resolution may return multiple IPs and the function returns one of them.
-// It prefers to return IPv4s first, and only if there are none, IPv6s.
-pub(crate) async fn resolve_hostname(hostname: &str) -> Result<SocketAddr, io::Error> {
-    let mut ret = None;
-    let addrs: Vec<SocketAddr> = match lookup_host(hostname).await {
-        Ok(addrs) => addrs.collect(),
-        // Use a default port in case of error, but propagate the original error on failure
-        Err(e) => lookup_host((hostname, 9042)).await.or(Err(e))?.collect(),
-    };
-    for a in addrs {
-        match a {
-            SocketAddr::V4(_) => return Ok(a),
-            _ => {
-                ret = Some(a);
-            }
-        }
-    }
-
-    ret.ok_or_else(|| {
-        io::Error::new(
-            io::ErrorKind::Other,
-            format!("Empty address list returned by DNS for {}", hostname),
-        )
-    })
-}
-
-/// Transforms the given [`KnownNode`]s into [`ContactPoint`]s.
-///
-/// In case of a hostname, resolves it using a DNS lookup.
-/// In case of a plain IP address, parses it and uses straight.
-pub(crate) async fn resolve_contact_points(
-    known_nodes: &[KnownNode],
-) -> (Vec<ResolvedContactPoint>, Vec<String>) {
-    // Find IP addresses of all known nodes passed in the config
-    let mut initial_peers: Vec<ResolvedContactPoint> = Vec::with_capacity(known_nodes.len());
-
-    let mut to_resolve: Vec<(&String, Option<String>)> = Vec::new();
-    let mut hostnames: Vec<String> = Vec::new();
-
-    for node in known_nodes.iter() {
-        match node {
-            KnownNode::Hostname(hostname) => {
-                to_resolve.push((hostname, None));
-                hostnames.push(hostname.clone());
-            }
-            KnownNode::Address(address) => initial_peers.push(ResolvedContactPoint {
-                address: *address,
-                datacenter: None,
-            }),
-            #[cfg(feature = "cloud")]
-            KnownNode::CloudEndpoint(CloudEndpoint {
-                hostname,
-                datacenter,
-            }) => to_resolve.push((hostname, Some(datacenter.clone()))),
-        };
-    }
-    let resolve_futures = to_resolve.iter().map(|(hostname, datacenter)| async move {
-        match resolve_hostname(hostname).await {
-            Ok(address) => Some(ResolvedContactPoint {
-                address,
-                datacenter: datacenter.clone(),
-            }),
-            Err(e) => {
-                warn!("Hostname resolution failed for {}: {}", hostname, &e);
-                None
-            }
-        }
-    });
-    let resolved: Vec<_> = futures::future::join_all(resolve_futures).await;
-    initial_peers.extend(resolved.into_iter().flatten());
-
-    (initial_peers, hostnames)
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/scylla/src/transport/session.rs b/scylla/src/transport/session.rs
index 790f8404..28fa0593 100644
--- a/scylla/src/transport/session.rs
+++ b/scylla/src/transport/session.rs
@@ -21,6 +21,7 @@ use std::borrow::Borrow;
 use std::collections::HashMap;
 use std::fmt::Display;
 use std::future::Future;
+use std::io;
 use std::net::SocketAddr;
 use std::num::NonZeroU32;
 use std::str::FromStr;
@@ -28,19 +29,19 @@ use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::Duration;
+use tokio::net::lookup_host;
 use tokio::time::timeout;
+use tracing::warn;
 use tracing::{debug, trace, trace_span, Instrument};
 use uuid::Uuid;
 
+use super::cluster::ContactPoint;
 use super::connection::NonErrorQueryResponse;
 use super::connection::QueryResponse;
 #[cfg(feature = "ssl")]
 use super::connection::SslConfig;
 use super::errors::{NewSessionError, QueryError};
 use super::execution_profile::{ExecutionProfile, ExecutionProfileHandle, ExecutionProfileInner};
-#[cfg(feature = "cloud")]
-use super::node::CloudEndpoint;
-use super::node::KnownNode;
 use super::partitioner::PartitionerName;
 use super::topology::UntranslatedPeer;
 use super::NodeRef;
@@ -288,6 +289,24 @@ pub struct SessionConfig {
     pub cluster_metadata_refresh_interval: Duration,
 }
 
+/// Describes database server known on Session startup.
+#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)]
+#[non_exhaustive]
+pub enum KnownNode {
+    Hostname(String),
+    Address(SocketAddr),
+    #[cfg(feature = "cloud")]
+    CloudEndpoint(CloudEndpoint),
+}
+
+#[cfg(feature = "cloud")]
+#[non_exhaustive]
+#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)]
+pub struct CloudEndpoint {
+    pub hostname: String,
+    pub datacenter: String,
+}
+
 impl SessionConfig {
     /// Creates a [`SessionConfig`] with default configuration
     /// # Default configuration
@@ -457,7 +476,7 @@ impl Session {
     /// # use std::error::Error;
     /// # async fn check_only_compiles() -> Result<(), Box<dyn Error>> {
     /// use scylla::{Session, SessionConfig};
-    /// use scylla::transport::KnownNode;
+    /// use scylla::transport::session::KnownNode;
     ///
     /// let mut config = SessionConfig::new();
     /// config.known_nodes.push(KnownNode::Hostname("127.0.0.1:9042".to_string()));
@@ -493,6 +512,50 @@ impl Session {
             return Err(NewSessionError::EmptyKnownNodesList);
         }
 
+        // Find IP addresses of all known nodes passed in the config
+        let mut initial_peers: Vec<ContactPoint> = Vec::with_capacity(known_nodes.len());
+
+        let mut to_resolve: Vec<(String, Option<String>)> = Vec::new();
+
+        for node in known_nodes {
+            match node {
+                KnownNode::Hostname(hostname) => to_resolve.push((hostname, None)),
+                KnownNode::Address(address) => initial_peers.push(ContactPoint {
+                    address,
+                    datacenter: None,
+                }),
+                #[cfg(feature = "cloud")]
+                KnownNode::CloudEndpoint(CloudEndpoint {
+                    hostname,
+                    datacenter,
+                }) => to_resolve.push((hostname, Some(datacenter))),
+            };
+        }
+        let resolve_futures = to_resolve.iter().map(|(hostname, datacenter)| async move {
+            match resolve_hostname(hostname).await {
+                Ok(address) => Some(ContactPoint {
+                    address,
+                    datacenter: datacenter.clone(),
+                }),
+                Err(e) => {
+                    warn!("Hostname resolution failed for {}: {}", hostname, &e);
+                    None
+                }
+            }
+        });
+        let resolved: Vec<_> = futures::future::join_all(resolve_futures).await;
+        initial_peers.extend(resolved.into_iter().flatten());
+
+        // Ensure there is at least one resolved node
+        if initial_peers.is_empty() {
+            return Err(NewSessionError::FailedToResolveAnyHostname(
+                to_resolve
+                    .into_iter()
+                    .map(|(hostname, _datacenter)| hostname)
+                    .collect(),
+            ));
+        }
+
         let connection_config = ConnectionConfig {
             compression: config.compression,
             tcp_nodelay: config.tcp_nodelay,
@@ -519,7 +582,7 @@ impl Session {
         };
 
         let cluster = Cluster::new(
-            known_nodes,
+            initial_peers,
             pool_config,
             config.keyspaces_to_fetch,
             config.fetch_schema_metadata,
@@ -1803,6 +1866,33 @@ impl Session {
     }
 }
 
+// Resolve the given hostname using a DNS lookup if necessary.
+// The resolution may return multiple IPs and the function returns one of them.
+// It prefers to return IPv4s first, and only if there are none, IPv6s.
+pub(crate) async fn resolve_hostname(hostname: &str) -> Result<SocketAddr, io::Error> {
+    let mut ret = None;
+    let addrs: Vec<SocketAddr> = match lookup_host(hostname).await {
+        Ok(addrs) => addrs.collect(),
+        // Use a default port in case of error, but propagate the original error on failure
+        Err(e) => lookup_host((hostname, 9042)).await.or(Err(e))?.collect(),
+    };
+    for a in addrs {
+        match a {
+            SocketAddr::V4(_) => return Ok(a),
+            _ => {
+                ret = Some(a);
+            }
+        }
+    }
+
+    ret.ok_or_else(|| {
+        io::Error::new(
+            io::ErrorKind::Other,
+            format!("Empty address list returned by DNS for {}", hostname),
+        )
+    })
+}
+
 // run_query, execute_query, etc have a template type called ResT.
 // There was a bug where ResT was set to QueryResponse, which could
 // be an error response. This was not caught by retry policy which
diff --git a/scylla/src/transport/session_builder.rs b/scylla/src/transport/session_builder.rs
index 394a348c..7460fdb1 100644
--- a/scylla/src/transport/session_builder.rs
+++ b/scylla/src/transport/session_builder.rs
@@ -950,7 +950,7 @@ mod tests {
 
     use super::SessionBuilder;
     use crate::transport::execution_profile::{defaults, ExecutionProfile};
-    use crate::transport::node::KnownNode;
+    use crate::transport::session::KnownNode;
     use crate::transport::Compression;
     use std::net::{IpAddr, Ipv4Addr, SocketAddr};
     use std::time::Duration;
diff --git a/scylla/src/transport/topology.rs b/scylla/src/transport/topology.rs
index 63ee14f5..ffad6283 100644
--- a/scylla/src/transport/topology.rs
+++ b/scylla/src/transport/topology.rs
@@ -5,7 +5,6 @@ use crate::transport::connection::{Connection, ConnectionConfig};
 use crate::transport::connection_pool::{NodeConnectionPool, PoolConfig, PoolSize};
 use crate::transport::errors::{DbError, QueryError};
 use crate::transport::host_filter::HostFilter;
-use crate::transport::node::resolve_contact_points;
 use crate::utils::parse::{ParseErrorCause, ParseResult, ParserState};
 
 use futures::future::{self, FutureExt};
@@ -13,7 +12,6 @@ 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;
@@ -28,11 +26,12 @@ use std::str::FromStr;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 use strum_macros::EnumString;
-use tokio::sync::{broadcast, mpsc};
+use tokio::sync::mpsc;
 use tracing::{debug, error, trace, warn};
 use uuid::Uuid;
 
-use super::node::{KnownNode, NodeAddr, ResolvedContactPoint};
+use super::cluster::ContactPoint;
+use super::NodeAddr;
 
 /// Allows to read current metadata from the cluster
 pub(crate) struct MetadataReader {
@@ -47,14 +46,6 @@ pub(crate) struct MetadataReader {
     keyspaces_to_fetch: Vec<String>,
     fetch_schema: bool,
     host_filter: Option<Arc<dyn HostFilter>>,
-
-    // When no known peer is reachable, initial known nodes are resolved once again as a fallback
-    // and establishing control connection to them is attempted.
-    initial_known_nodes: Vec<KnownNode>,
-
-    // When a control connection breaks, the PoolRefiller of its pool uses the requester
-    // to signal ClusterWorker that an immediate metadata refresh is advisable.
-    control_connection_repair_requester: broadcast::Sender<()>,
 }
 
 /// Describes all metadata retrieved from the cluster
@@ -78,7 +69,7 @@ pub struct Peer {
 #[derive(Clone, Debug)]
 pub enum UntranslatedEndpoint {
     /// Provided by user in SessionConfig (initial contact points).
-    ContactPoint(ResolvedContactPoint),
+    ContactPoint(ContactPoint),
     /// Fetched in Metadata with `query_peers()`
     Peer(PeerEndpoint),
 }
@@ -86,7 +77,7 @@ pub enum UntranslatedEndpoint {
 impl UntranslatedEndpoint {
     pub(crate) fn address(&self) -> NodeAddr {
         match *self {
-            UntranslatedEndpoint::ContactPoint(ResolvedContactPoint { address, .. }) => {
+            UntranslatedEndpoint::ContactPoint(ContactPoint { address, .. }) => {
                 NodeAddr::Untranslatable(address)
             }
             UntranslatedEndpoint::Peer(PeerEndpoint { address, .. }) => address,
@@ -94,7 +85,7 @@ impl UntranslatedEndpoint {
     }
     pub(crate) fn set_port(&mut self, port: u16) {
         let inner_addr = match self {
-            UntranslatedEndpoint::ContactPoint(ResolvedContactPoint { address, .. }) => address,
+            UntranslatedEndpoint::ContactPoint(ContactPoint { address, .. }) => address,
             UntranslatedEndpoint::Peer(PeerEndpoint { address, .. }) => address.inner_mut(),
         };
         inner_addr.set_port(port);
@@ -401,29 +392,19 @@ impl Metadata {
 impl MetadataReader {
     /// Creates new MetadataReader, which connects to initially_known_peers in the background
     #[allow(clippy::too_many_arguments)]
-    pub(crate) async fn new(
-        initial_known_nodes: Vec<KnownNode>,
-        control_connection_repair_requester: broadcast::Sender<()>,
+    pub(crate) fn new(
+        initially_known_peers: Vec<ContactPoint>,
         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>>,
-    ) -> 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,
-            ));
-        }
-
+    ) -> Self {
         let control_connection_endpoint = UntranslatedEndpoint::ContactPoint(
-            initial_peers
+            initially_known_peers
                 .choose(&mut thread_rng())
-                .expect("Tried to initialize MetadataReader with empty initial_known_nodes list!")
+                .expect("Tried to initialize MetadataReader with empty known_peers list!")
                 .clone(),
         );
 
@@ -436,43 +417,33 @@ impl MetadataReader {
             control_connection_endpoint.clone(),
             connection_config.clone(),
             keepalive_interval,
-            control_connection_repair_requester.clone(),
         );
 
-        Ok(MetadataReader {
+        MetadataReader {
             control_connection_endpoint,
             control_connection,
             keepalive_interval,
             connection_config,
-            known_peers: initial_peers
+            known_peers: initially_known_peers
                 .into_iter()
                 .map(UntranslatedEndpoint::ContactPoint)
                 .collect(),
             keyspaces_to_fetch,
             fetch_schema,
             host_filter: host_filter.clone(),
-            initial_known_nodes,
-            control_connection_repair_requester,
-        })
+        }
     }
 
     /// Fetches current metadata from the cluster
     pub(crate) async fn read_metadata(&mut self, initial: bool) -> Result<Metadata, QueryError> {
         let mut result = self.fetch_metadata(initial).await;
-        let prev_err = match result {
-            Ok(metadata) => {
-                debug!("Fetched new metadata");
-                self.update_known_peers(&metadata);
-                if initial {
-                    self.handle_unaccepted_host_in_control_connection(&metadata);
-                }
-                return Ok(metadata);
+        if let Ok(metadata) = result {
+            self.update_known_peers(&metadata);
+            if initial {
+                self.handle_unaccepted_host_in_control_connection(&metadata);
             }
-            Err(err) => err,
-        };
-
-        // At this point, we known that fetching metadata on currect control connection failed.
-        // Therefore, we try to fetch metadata from other known peers, in order.
+            return Ok(metadata);
+        }
 
         // shuffle known_peers to iterate through them in random order later
         self.known_peers.shuffle(&mut thread_rng());
@@ -488,61 +459,12 @@ impl MetadataReader {
         let address_of_failed_control_connection = self.control_connection_endpoint.address();
         let filtered_known_peers = self
             .known_peers
-            .clone()
-            .into_iter()
-            .filter(|peer| peer.address() != address_of_failed_control_connection);
+            .iter()
+            .filter(|&peer| peer.address() != address_of_failed_control_connection);
 
         // if fetching metadata on current control connection failed,
         // try to fetch metadata from other known peer
-        result = self
-            .retry_fetch_metadata_on_nodes(initial, filtered_known_peers, prev_err)
-            .await;
-
-        if let Err(prev_err) = result {
-            if !initial {
-                // If no known peer is reachable, try falling back to initial contact points, in hope that
-                // there are some hostnames there which will resolve to reachable new addresses.
-                warn!("Failed to establish control connection and fetch metadata on all known peers. Falling back to initial contact points.");
-                let (initial_peers, _hostnames) =
-                    resolve_contact_points(&self.initial_known_nodes).await;
-                result = self
-                    .retry_fetch_metadata_on_nodes(
-                        initial,
-                        initial_peers
-                            .into_iter()
-                            .map(UntranslatedEndpoint::ContactPoint),
-                        prev_err,
-                    )
-                    .await;
-            } else {
-                // No point in falling back as this is an initial connection attempt.
-                result = Err(prev_err);
-            }
-        }
-
-        match &result {
-            Ok(metadata) => {
-                self.update_known_peers(metadata);
-                self.handle_unaccepted_host_in_control_connection(metadata);
-                debug!("Fetched new metadata");
-            }
-            Err(error) => error!(
-                error = %error,
-                "Could not fetch metadata"
-            ),
-        }
-
-        result
-    }
-
-    async fn retry_fetch_metadata_on_nodes(
-        &mut self,
-        initial: bool,
-        nodes: impl Iterator<Item = UntranslatedEndpoint>,
-        prev_err: QueryError,
-    ) -> Result<Metadata, QueryError> {
-        let mut result = Err(prev_err);
-        for peer in nodes {
+        for peer in filtered_known_peers {
             let err = match result {
                 Ok(_) => break,
                 Err(err) => err,
@@ -563,7 +485,6 @@ impl MetadataReader {
                 self.control_connection_endpoint.clone(),
                 self.connection_config.clone(),
                 self.keepalive_interval,
-                self.control_connection_repair_requester.clone(),
             );
 
             debug!(
@@ -572,6 +493,19 @@ impl MetadataReader {
             );
             result = self.fetch_metadata(initial).await;
         }
+
+        match &result {
+            Ok(metadata) => {
+                self.update_known_peers(metadata);
+                self.handle_unaccepted_host_in_control_connection(metadata);
+                debug!("Fetched new metadata");
+            }
+            Err(error) => error!(
+                error = %error,
+                "Could not fetch metadata"
+            ),
+        }
+
         result
     }
 
@@ -663,7 +597,6 @@ impl MetadataReader {
                         self.control_connection_endpoint.clone(),
                         self.connection_config.clone(),
                         self.keepalive_interval,
-                        self.control_connection_repair_requester.clone(),
                     );
                 }
             }
@@ -674,7 +607,6 @@ impl MetadataReader {
         endpoint: UntranslatedEndpoint,
         connection_config: ConnectionConfig,
         keepalive_interval: Option<Duration>,
-        refresh_requester: broadcast::Sender<()>,
     ) -> NodeConnectionPool {
         let pool_config = PoolConfig {
             connection_config,
@@ -688,7 +620,7 @@ impl MetadataReader {
             can_use_shard_aware_port: false,
         };
 
-        NodeConnectionPool::new(endpoint, pool_config, None, refresh_requester)
+        NodeConnectionPool::new(endpoint, pool_config, None)
     }
 }
 


Test patch:
diff --git a/scylla/tests/integration/consistency.rs b/scylla/tests/integration/consistency.rs
index 93a0ad8f..9eaabf03 100644
--- a/scylla/tests/integration/consistency.rs
+++ b/scylla/tests/integration/consistency.rs
@@ -20,7 +20,7 @@ use scylla_cql::Consistency;
 use scylla_proxy::ShardAwareness;
 use scylla_proxy::{
     Condition, ProxyError, Reaction, RequestFrame, RequestOpcode, RequestReaction, RequestRule,
-    TargetShard, WorkerError,
+    WorkerError,
 };
 use std::sync::Arc;
 use std::time::Duration;
@@ -296,9 +296,9 @@ async fn consistency_is_correctly_set_in_cql_requests() {
             async fn check_consistencies(
                 consistency: Consistency,
                 serial_consistency: Option<SerialConsistency>,
-                mut request_rx: UnboundedReceiver<(RequestFrame, Option<TargetShard>)>,
-            ) -> UnboundedReceiver<(RequestFrame, Option<TargetShard>)> {
-                let (request_frame, _shard) = request_rx.recv().await.unwrap();
+                mut request_rx: UnboundedReceiver<RequestFrame>,
+            ) -> UnboundedReceiver<RequestFrame> {
+                let request_frame = request_rx.recv().await.unwrap();
                 let deserialized_request = request_frame.deserialize().unwrap();
                 assert_eq!(deserialized_request.get_consistency().unwrap(), consistency);
                 assert_eq!(
diff --git a/scylla/tests/integration/lwt_optimisation.rs b/scylla/tests/integration/lwt_optimisation.rs
index 19ed45be..c96b47ba 100644
--- a/scylla/tests/integration/lwt_optimisation.rs
+++ b/scylla/tests/integration/lwt_optimisation.rs
@@ -9,7 +9,7 @@ use tokio::sync::mpsc;
 
 use scylla_proxy::{
     Condition, ProxyError, Reaction, RequestFrame, RequestOpcode, RequestReaction, RequestRule,
-    ResponseOpcode, ResponseReaction, ResponseRule, ShardAwareness, TargetShard, WorkerError,
+    ResponseOpcode, ResponseReaction, ResponseRule, ShardAwareness, WorkerError,
 };
 
 #[tokio::test]
@@ -56,7 +56,7 @@ async fn if_lwt_optimisation_mark_offered_then_negotiatied_and_lwt_routed_optima
             .await
             .unwrap();
 
-        let (supported_frame, _shard) = supported_rx.recv().await.unwrap();
+        let supported_frame = supported_rx.recv().await.unwrap();
         let supported_options = types::read_string_multimap(&mut &*supported_frame.body).unwrap();
         let supported_features = ProtocolFeatures::parse_from_supported(&supported_options);
 
@@ -74,20 +74,20 @@ async fn if_lwt_optimisation_mark_offered_then_negotiatied_and_lwt_routed_optima
             .unwrap();
 
 
-        fn clear_rxs(rxs: &mut [mpsc::UnboundedReceiver<(RequestFrame, Option<TargetShard>)>; 3]) {
+        fn clear_rxs(rxs: &mut [mpsc::UnboundedReceiver<RequestFrame>; 3]) {
             for rx in rxs.iter_mut() {
                 while rx.try_recv().is_ok() {}
             }
         }
 
-        async fn assert_all_replicas_queried(rxs: &mut [mpsc::UnboundedReceiver<(RequestFrame, Option<TargetShard>)>; 3]) {
+        async fn assert_all_replicas_queried(rxs: &mut [mpsc::UnboundedReceiver<RequestFrame>; 3]) {
             for rx in rxs.iter_mut() {
                 rx.recv().await.unwrap();
             }
             clear_rxs(rxs);
         }
 
-        fn assert_one_replica_queried(rxs: &mut [mpsc::UnboundedReceiver<(RequestFrame, Option<TargetShard>)>; 3]) {
+        fn assert_one_replica_queried(rxs: &mut [mpsc::UnboundedReceiver<RequestFrame>; 3]) {
             let mut found_queried = false;
             for rx in rxs.iter_mut() {
                 if rx.try_recv().is_ok() {
@@ -100,7 +100,7 @@ async fn if_lwt_optimisation_mark_offered_then_negotiatied_and_lwt_routed_optima
         }
 
         #[allow(unused)]
-        fn who_was_queried(rxs: &mut [mpsc::UnboundedReceiver<(RequestFrame, Option<TargetShard>)>; 3]) {
+        fn who_was_queried(rxs: &mut [mpsc::UnboundedReceiver<RequestFrame>; 3]) {
             for (i, rx) in rxs.iter_mut().enumerate() {
                 if rx.try_recv().is_ok() {
                     println!("{} was queried.", i);
diff --git a/scylla/tests/integration/main.rs b/scylla/tests/integration/main.rs
index ab64db0c..c2cb8fc6 100644
--- a/scylla/tests/integration/main.rs
+++ b/scylla/tests/integration/main.rs
@@ -4,5 +4,4 @@ mod hygiene;
 mod lwt_optimisation;
 mod new_session;
 mod retries;
-mod shards;
 pub(crate) mod utils;
diff --git a/scylla/tests/integration/shards.rs b/scylla/tests/integration/shards.rs
deleted file mode 100644
index 7bee6a3a..00000000
--- a/scylla/tests/integration/shards.rs
+++ /dev/null
@@ -1,82 +0,0 @@
-use std::sync::Arc;
-
-use crate::utils::test_with_3_node_cluster;
-use scylla::{test_utils::unique_keyspace_name, SessionBuilder};
-use tokio::sync::mpsc;
-
-use scylla_proxy::TargetShard;
-use scylla_proxy::{
-    Condition, Reaction, RequestOpcode, RequestReaction, RequestRule, ShardAwareness,
-};
-use scylla_proxy::{ProxyError, RequestFrame, WorkerError};
-
-#[tokio::test]
-#[ntest::timeout(30000)]
-#[cfg(not(scylla_cloud_tests))]
-async fn test_consistent_shard_awareness() {
-    use std::collections::HashSet;
-
-    let res = test_with_3_node_cluster(ShardAwareness::QueryNode, |proxy_uris, translation_map, mut running_proxy| async move {
-
-        let (feedback_txs, mut feedback_rxs): (Vec<_>, Vec<_>) = (0..3).map(|_| {
-            mpsc::unbounded_channel::<(RequestFrame, Option<TargetShard>)>()
-        }).unzip();
-        for (i, tx) in feedback_txs.iter().cloned().enumerate() {
-            running_proxy.running_nodes[i].change_request_rules(Some(vec![
-                RequestRule(Condition::RequestOpcode(RequestOpcode::Execute), RequestReaction::noop().with_feedback_when_performed(tx))
-            ]));
-        }
-
-        let session = SessionBuilder::new()
-            .known_node(proxy_uris[0].as_str())
-            .address_translator(Arc::new(translation_map))
-            .build()
-            .await
-            .unwrap();
-        let ks = unique_keyspace_name();
-
-        /* Prepare schema */
-        session.query(format!("CREATE KEYSPACE IF NOT EXISTS {} WITH REPLICATION = {{'class' : 'NetworkTopologyStrategy', 'replication_factor' : 3}}", ks), &[]).await.unwrap();
-        session
-            .query(
-                format!(
-                    "CREATE TABLE IF NOT EXISTS {}.t (a int, b int, c text, primary key (a, b))",
-                    ks
-                ),
-                &[],
-            )
-            .await
-            .unwrap();
-
-        let prepared = session.prepare(format!("INSERT INTO {}.t (a, b, c) VALUES (?, ?, 'abc')", ks)).await.unwrap();
-
-        let value_lists = [
-            (4, 2),
-            (2, 1),
-            (3, 7),
-        ];
-
-        fn assert_one_shard_queried(rx: &mut mpsc::UnboundedReceiver<(RequestFrame, Option<TargetShard>)>) {
-            let shards = std::iter::from_fn(|| rx.try_recv().ok().map(|(_frame, shard)| shard)).collect::<HashSet<_>>();
-            if !shards.is_empty() {
-                assert_eq!(shards.len(), 1);
-            }
-        }
-
-        for values in value_lists {
-            for _ in 0..10 {
-                session.execute(&prepared, values).await.unwrap();
-            }
-            for rx in feedback_rxs.iter_mut() {
-                assert_one_shard_queried(rx);
-            }
-        }
-
-        running_proxy
-    }).await;
-    match res {
-        Ok(()) => (),
-        Err(ProxyError::Worker(WorkerError::DriverDisconnected(_))) => (),
-        Err(err) => panic!("{}", err),
-    }
-}