RUST-679 Use error strings when marking servers as unknown from CMAP

Author: saghm

src/cmap/test/mod.rs

@@ -163,7 +163,7 @@ impl Executor {
         let manager = pool.manager.clone();
         RUNTIME.execute(async move {
             while let Some(update) = update_receiver.recv().await {
-                match update.message() {
+                match update.into_message() {
                     ServerUpdate::Error { .. } => manager.clear(),
                 }
             }

src/cmap/worker.rs

@@ -562,7 +562,7 @@ async fn establish_connection(
                 };
                 handler.handle_connection_closed_event(event);
             }
-            server_updater.handle_error(e.clone(), generation).await;
+            server_updater.handle_error(&e, generation).await;
             manager.handle_connection_failed();
         }
         Ok(ref mut connection) => {

src/runtime/acknowledged_message.rs

@@ -21,8 +21,8 @@ impl<M, R> AcknowledgedMessage<M, R> {
     }
 
     /// Get the message.
-    pub(crate) fn message(&self) -> &M {
-        &self.message
+    pub(crate) fn into_message(self) -> M {
+        self.message
     }
 
     /// Send acknowledgement to the receiver.

src/sdam/monitor.rs

@@ -12,7 +12,7 @@ use super::{
 use crate::{
     bson::doc,
     cmap::{is_master, Command, Connection, Handshaker, PoolGenerationSubscriber},
-    error::{Error, Result},
+    error::Result,
     is_master::IsMasterReply,
     options::{ClientOptions, StreamAddress},
     RUNTIME,
@@ -159,7 +159,7 @@ impl HeartbeatMonitor {
                         .map(|sd| sd.is_available())
                         .unwrap_or(false)
                 {
-                    self.handle_error(e, topology, server).await;
+                    self.handle_error(e.to_string(), topology, server).await;
                     retried = true;
                     self.perform_is_master().await
                 } else {
@@ -174,7 +174,7 @@ impl HeartbeatMonitor {
                     ServerDescription::new(server.address.clone(), Some(Ok(reply)));
                 topology.update(server, server_description).await
             }
-            Err(e) => self.handle_error(e, topology, server).await || retried,
+            Err(e) => self.handle_error(e.to_string(), topology, server).await || retried,
         }
     }
 
@@ -217,8 +217,8 @@ impl HeartbeatMonitor {
         result
     }
 
-    async fn handle_error(&mut self, error: Error, topology: &Topology, server: &Server) -> bool {
-        topology.handle_pre_handshake_error(&error, server).await
+    async fn handle_error(&mut self, error: String, topology: &Topology, server: &Server) -> bool {
+        topology.handle_pre_handshake_error(error, server).await
     }
 }
 
@@ -244,13 +244,13 @@ impl UpdateMonitor {
                 _ => return,
             };
 
-            match update.message() {
+            match update.into_message() {
                 ServerUpdate::Error {
                     error,
                     error_generation,
                 } => {
-                    if *error_generation == self.generation_subscriber.generation() {
-                        topology.handle_pre_handshake_error(&error, &server).await;
+                    if error_generation == self.generation_subscriber.generation() {
+                        topology.handle_pre_handshake_error(error, &server).await;
                     }
                 }
             }

src/sdam/state/mod.rs

@@ -227,10 +227,10 @@ impl Topology {
 
     /// Updates the topology based on an error that occurs before the handshake has completed during
     /// an operation.
-    pub(crate) async fn handle_pre_handshake_error(&self, error: &Error, server: &Server) -> bool {
+    pub(crate) async fn handle_pre_handshake_error(&self, error: String, server: &Server) -> bool {
         let state_lock = self.state.write().await;
         let changed = self
-            .mark_server_as_unknown(&error, &server, state_lock)
+            .mark_server_as_unknown(error, &server, state_lock)
             .await;
         if changed {
             server.pool.clear();
@@ -249,13 +249,13 @@ impl Topology {
         // SDAM spec.
         if error.is_non_timeout_network_error() {
             let state_lock = self.state.write().await;
-            self.mark_server_as_unknown(error, &server, state_lock)
+            self.mark_server_as_unknown(error.to_string(), &server, state_lock)
                 .await;
             server.pool.clear();
         } else if error.is_recovering() || error.is_not_master() {
             let state_lock = self.state.write().await;
 
-            self.mark_server_as_unknown(error, &server, state_lock)
+            self.mark_server_as_unknown(error.to_string(), &server, state_lock)
                 .await;
 
             let wire_version = conn
@@ -279,12 +279,11 @@ impl Topology {
     /// Returns whether the topology changed as a result of the update.
     async fn mark_server_as_unknown(
         &self,
-        error: &Error,
+        error: String,
         server: &Server,
         state_lock: RwLockWriteGuard<'_, TopologyState>,
     ) -> bool {
-        let description =
-            ServerDescription::new(server.address.clone(), Some(Err(error.to_string())));
+        let description = ServerDescription::new(server.address.clone(), Some(Err(error)));
         self.update_and_notify(server, description, state_lock)
             .await
     }

src/sdam/state/server.rs

@@ -74,7 +74,10 @@ impl Server {
 /// TODO: add success cases from application handshakes.
 #[derive(Debug)]
 pub(crate) enum ServerUpdate {
-    Error { error: Error, error_generation: u32 },
+    Error {
+        error: String,
+        error_generation: u32,
+    },
 }
 
 #[derive(Debug)]
@@ -106,9 +109,9 @@ impl ServerUpdateSender {
 
     /// Update the server based on the given error.
     /// This will block until the topology has processed the error.
-    pub(crate) async fn handle_error(&mut self, error: Error, error_generation: u32) {
+    pub(crate) async fn handle_error(&mut self, error: &Error, error_generation: u32) {
         let reason = ServerUpdate::Error {
-            error,
+            error: error.to_string(),
             error_generation,
         };
 

src/test/util/event.rs

@@ -31,12 +31,14 @@ pub type EventQueue<T> = Arc<RwLock<VecDeque<T>>>;
 pub type CmapEvent = crate::cmap::test::event::Event;
 
 #[derive(Clone, Debug, From)]
+#[allow(clippy::large_enum_variant)]
 pub enum Event {
     CmapEvent(CmapEvent),
     CommandEvent(CommandEvent),
 }
 
 #[derive(Clone, Debug)]
+#[allow(clippy::large_enum_variant)]
 pub enum CommandEvent {
     CommandStartedEvent(CommandStartedEvent),
     CommandSucceededEvent(CommandSucceededEvent),