RUST-1328 Only emit SDAM changed events when servers change (#679)

This also adds a test for the fix for RUST-1274.

Author: patrickfreed

src/sdam/description/server.rs

@@ -122,6 +122,7 @@ impl PartialEq for ServerDescription {
 
                 self_response == other_response
             }
+            (Err(self_err), Err(other_err)) => self_err == other_err,
             _ => false,
         }
     }

src/sdam/description/topology/test/sdam.rs

@@ -646,7 +646,7 @@ async fn heartbeat_events() {
         .await
         .expect("should see server heartbeat succeeded event");
 
-    if !client.supports_fail_command_appname() {
+    if !client.supports_fail_command_appname_initial_handshake() {
         return;
     }
 

src/sdam/test.rs

@@ -43,16 +43,10 @@ async fn min_heartbeat_frequency() {
 
     let setup_client = TestClient::with_options(Some(setup_client_options.clone())).await;
 
-    if !setup_client.supports_fail_command() {
+    if !setup_client.supports_fail_command_appname_initial_handshake() {
         log_uncaptured(
-            "skipping min_heartbeat_frequency test due to server not supporting fail points",
-        );
-        return;
-    }
-
-    if setup_client.server_version_lt(4, 9) {
-        log_uncaptured(
-            "skipping min_heartbeat_frequency test due to server version being less than 4.9",
+            "skipping min_heartbeat_frequency test due to server not supporting failcommand \
+             appname",
         );
         return;
     }

src/sdam/topology.rs

@@ -477,23 +477,25 @@ impl TopologyWorker {
         let old_description = latest_state.description.clone();
 
         if let Some(expected_name) = &self.options.repl_set_name {
-            let got_name = sd.set_name();
-            if latest_state.description.topology_type() == TopologyType::Single
-                && got_name.as_ref().map(|opt| opt.as_ref()) != Ok(Some(expected_name))
-            {
-                let got_display = match got_name {
-                    Ok(Some(s)) => format!("{:?}", s),
-                    Ok(None) => "<none>".to_string(),
-                    Err(s) => format!("<error: {}>", s),
-                };
-                // Mark server as unknown.
-                sd = ServerDescription::new(
-                    sd.address,
-                    Some(Err(format!(
-                        "Connection string replicaSet name {:?} does not match actual name {}",
-                        expected_name, got_display,
-                    ))),
-                );
+            if sd.is_available() {
+                let got_name = sd.set_name();
+                if latest_state.description.topology_type() == TopologyType::Single
+                    && got_name.as_ref().map(|opt| opt.as_ref()) != Ok(Some(expected_name))
+                {
+                    let got_display = match got_name {
+                        Ok(Some(s)) => format!("{:?}", s),
+                        Ok(None) => "<none>".to_string(),
+                        Err(s) => format!("<error: {}>", s),
+                    };
+                    // Mark server as unknown.
+                    sd = ServerDescription::new(
+                        sd.address,
+                        Some(Err(format!(
+                            "Connection string replicaSet name {:?} does not match actual name {}",
+                            expected_name, got_display,
+                        ))),
+                    );
+                }
             }
         }
 
@@ -513,18 +515,18 @@ impl TopologyWorker {
         let topology_changed =
             self.process_topology_diff(&old_description, &latest_state.description);
 
-        if topology_changed {
-            if server_type.is_data_bearing()
+        if topology_changed
+            && (server_type.is_data_bearing()
                 || (server_type != ServerType::Unknown
-                    && latest_state.description.topology_type() == TopologyType::Single)
-            {
-                if let Some(s) = latest_state.servers.get(&server_address) {
-                    s.pool.mark_as_ready().await;
-                }
+                    && latest_state.description.topology_type() == TopologyType::Single))
+        {
+            if let Some(s) = latest_state.servers.get(&server_address) {
+                s.pool.mark_as_ready().await;
             }
-            self.publisher.publish_new_state(latest_state)
         }
 
+        self.publisher.publish_new_state(latest_state);
+
         topology_changed
     }
 

src/test/client.rs

@@ -1,4 +1,4 @@
-use std::{borrow::Cow, collections::HashMap, time::Duration};
+use std::{borrow::Cow, collections::HashMap, sync::Arc, time::Duration};
 
 use bson::Document;
 use serde::Deserialize;
@@ -7,11 +7,25 @@ use tokio::sync::{RwLockReadGuard, RwLockWriteGuard};
 use crate::{
     bson::{doc, Bson},
     error::{CommandError, Error, ErrorKind},
+    hello::LEGACY_HELLO_COMMAND_NAME,
     options::{AuthMechanism, ClientOptions, Credential, ListDatabasesOptions, ServerAddress},
     runtime,
     selection_criteria::{ReadPreference, ReadPreferenceOptions, SelectionCriteria},
-    test::{log_uncaptured, util::TestClient, CLIENT_OPTIONS, LOCK},
+    test::{
+        log_uncaptured,
+        util::TestClient,
+        CmapEvent,
+        Event,
+        EventHandler,
+        FailCommandOptions,
+        FailPoint,
+        FailPointMode,
+        SdamEvent,
+        CLIENT_OPTIONS,
+        LOCK,
+    },
     Client,
+    ServerType,
 };
 
 #[derive(Debug, Deserialize)]
@@ -691,3 +705,134 @@ async fn plain_auth() {
         }
     );
 }
+
+/// Test verifies that retrying a commitTransaction operation after a checkOut
+/// failure works.
+#[cfg_attr(feature = "tokio-runtime", tokio::test(flavor = "multi_thread"))]
+#[cfg_attr(feature = "async-std-runtime", async_std::test)]
+async fn retry_commit_txn_check_out() {
+    let _guard: RwLockWriteGuard<_> = LOCK.run_exclusively().await;
+
+    let setup_client = TestClient::new().await;
+    if !setup_client.is_replica_set() {
+        log_uncaptured("skipping retry_commit_txn_check_out due to non-replicaset topology");
+        return;
+    }
+
+    if !setup_client.supports_transactions() {
+        log_uncaptured("skipping retry_commit_txn_check_out due to lack of transaction support");
+        return;
+    }
+
+    if !setup_client.supports_fail_command_appname_initial_handshake() {
+        log_uncaptured(
+            "skipping retry_commit_txn_check_out due to insufficient failCommand support",
+        );
+        return;
+    }
+
+    // ensure namespace exists
+    setup_client
+        .database("retry_commit_txn_check_out")
+        .collection("retry_commit_txn_check_out")
+        .insert_one(doc! {}, None)
+        .await
+        .unwrap();
+
+    let mut options = CLIENT_OPTIONS.get().await.clone();
+    let handler = Arc::new(EventHandler::new());
+    options.cmap_event_handler = Some(handler.clone());
+    options.sdam_event_handler = Some(handler.clone());
+    options.heartbeat_freq = Some(Duration::from_secs(120));
+    options.app_name = Some("retry_commit_txn_check_out".to_string());
+    let client = Client::with_options(options).unwrap();
+
+    let mut session = client.start_session(None).await.unwrap();
+    session.start_transaction(None).await.unwrap();
+    // transition transaction to "in progress" so that the commit
+    // actually executes an operation.
+    client
+        .database("retry_commit_txn_check_out")
+        .collection("retry_commit_txn_check_out")
+        .insert_one_with_session(doc! {}, None, &mut session)
+        .await
+        .unwrap();
+
+    // enable a fail point that clears the connection pools so that
+    // commitTransaction will create a new connection during check out.
+    let fp = FailPoint::fail_command(
+        &["ping"],
+        FailPointMode::Times(1),
+        FailCommandOptions::builder().error_code(11600).build(),
+    );
+    let _guard = setup_client.enable_failpoint(fp, None).await.unwrap();
+
+    let mut subscriber = handler.subscribe();
+    client
+        .database("foo")
+        .run_command(doc! { "ping": 1 }, None)
+        .await
+        .unwrap_err();
+
+    // failing with a state change error will request an immediate check
+    // wait for the mark unknown and subsequent succeeded heartbeat
+    let mut primary = None;
+    subscriber
+        .wait_for_event(Duration::from_secs(1), |e| {
+            if let Event::Sdam(SdamEvent::ServerDescriptionChanged(event)) = e {
+                if event.is_marked_unknown_event() {
+                    primary = Some(event.address.clone());
+                    return true;
+                }
+            }
+            false
+        })
+        .await
+        .expect("should see marked unknown event");
+
+    subscriber
+        .wait_for_event(Duration::from_secs(1), |e| {
+            if let Event::Sdam(SdamEvent::ServerDescriptionChanged(event)) = e {
+                if &event.address == primary.as_ref().unwrap()
+                    && event.previous_description.server_type() == ServerType::Unknown
+                {
+                    return true;
+                }
+            }
+            false
+        })
+        .await
+        .expect("should see mark available event");
+
+    // enable a failpoint on the handshake to cause check_out
+    // to fail with a retryable error
+    let fp = FailPoint::fail_command(
+        &[LEGACY_HELLO_COMMAND_NAME, "hello"],
+        FailPointMode::Times(1),
+        FailCommandOptions::builder()
+            .error_code(11600)
+            .app_name("retry_commit_txn_check_out".to_string())
+            .build(),
+    );
+    let _guard2 = setup_client.enable_failpoint(fp, None).await.unwrap();
+
+    // finally, attempt the commit.
+    // this should succeed due to retry
+    session.commit_transaction().await.unwrap();
+
+    // ensure the first check out attempt fails
+    subscriber
+        .wait_for_event(Duration::from_secs(1), |e| {
+            matches!(e, Event::Cmap(CmapEvent::ConnectionCheckOutFailed(_)))
+        })
+        .await
+        .expect("should see check out failed event");
+
+    // ensure the second one succeeds
+    subscriber
+        .wait_for_event(Duration::from_secs(1), |e| {
+            matches!(e, Event::Cmap(CmapEvent::ConnectionCheckedOut(_)))
+        })
+        .await
+        .expect("should see checked out event");
+}

src/test/util/event.rs

@@ -394,7 +394,7 @@ pub struct EventSubscriber<'a> {
     receiver: tokio::sync::broadcast::Receiver<Event>,
 }
 
-impl<'a> EventSubscriber<'a> {
+impl EventSubscriber<'_> {
     pub async fn wait_for_event<F>(&mut self, timeout: Duration, mut filter: F) -> Option<Event>
     where
         F: FnMut(&Event) -> bool,

src/test/util/mod.rs

@@ -244,9 +244,19 @@ impl TestClient {
         version.matches(&self.server_version)
     }
 
-    pub fn supports_fail_command_appname(&self) -> bool {
-        let version = VersionReq::parse(">= 4.2.9").unwrap();
-        version.matches(&self.server_version)
+    /// Whether the deployment supports failing the initial handshake
+    /// only when it uses a specified appName.
+    ///
+    /// See SERVER-49336 for more info.
+    pub fn supports_fail_command_appname_initial_handshake(&self) -> bool {
+        let requirements = [
+            VersionReq::parse(">= 4.2.15, < 4.3.0").unwrap(),
+            VersionReq::parse(">= 4.4.7, < 4.5.0").unwrap(),
+            VersionReq::parse(">= 4.9.0").unwrap(),
+        ];
+        requirements
+            .iter()
+            .any(|req| req.matches(&self.server_version))
     }
 
     pub fn supports_transactions(&self) -> bool {