RUST-714 Fix possible deadlock during retry (#318)

Author: patrickfreed

src/client/executor.rs

@@ -125,10 +125,13 @@ impl Client {
                     .topology
                     .handle_application_error(
                         err.clone(),
-                        HandshakePhase::after_completion(conn),
+                        HandshakePhase::after_completion(&conn),
                         &server,
                     )
                     .await;
+                // release the connection to be processed by the connection pool
+                drop(conn);
+                // release the selected server to decrement its operation count
                 drop(server);
 
                 // TODO RUST-90: Do not retry read if session is in a transaction
@@ -173,7 +176,7 @@ impl Client {
                     .topology
                     .handle_application_error(
                         err.clone(),
-                        HandshakePhase::after_completion(conn),
+                        HandshakePhase::after_completion(&conn),
                         &server,
                     )
                     .await;

src/sdam/state/mod.rs

@@ -563,7 +563,7 @@ pub(crate) enum HandshakePhase {
 }
 
 impl HandshakePhase {
-    pub(crate) fn after_completion(handshaked_connection: Connection) -> Self {
+    pub(crate) fn after_completion(handshaked_connection: &Connection) -> Self {
         Self::AfterCompletion {
             generation: handshaked_connection.generation,
             // given that this is a handshaked connection, the stream description should

src/test/spec/retryable_reads.rs

@@ -1,6 +1,20 @@
+use std::time::Duration;
+
+use bson::doc;
 use tokio::sync::RwLockWriteGuard;
 
-use crate::test::{run_spec_test, LOCK};
+use crate::{
+    test::{
+        run_spec_test,
+        FailCommandOptions,
+        FailPoint,
+        FailPointMode,
+        TestClient,
+        CLIENT_OPTIONS,
+        LOCK,
+    },
+    RUNTIME,
+};
 
 use super::run_v2_test;
 
@@ -10,3 +24,37 @@ async fn run() {
     let _guard: RwLockWriteGuard<()> = LOCK.run_exclusively().await;
     run_spec_test(&["retryable-reads"], run_v2_test).await;
 }
+
+/// Test ensures that the connection used in the first attempt of a retry is released back into the
+/// pool before the second attempt.
+#[cfg_attr(feature = "tokio-runtime", tokio::test(flavor = "multi_thread"))]
+#[cfg_attr(feature = "async-std-runtime", async_std::test)]
+async fn retry_releases_connection() {
+    let _guard: RwLockWriteGuard<()> = LOCK.run_exclusively().await;
+
+    let mut client_options = CLIENT_OPTIONS.clone();
+    client_options.hosts.drain(1..);
+    client_options.retry_reads = Some(true);
+    client_options.max_pool_size = Some(1);
+
+    let client = TestClient::with_options(Some(client_options)).await;
+    if !client.supports_fail_command().await {
+        println!("skipping retry_releases_connection due to failCommand not being supported");
+        return;
+    }
+
+    let collection = client
+        .database("retry_releases_connection")
+        .collection("retry_releases_connection");
+    collection.insert_one(doc! { "x": 1 }, None).await.unwrap();
+
+    let options = FailCommandOptions::builder().error_code(91).build();
+    let failpoint = FailPoint::fail_command(&["find"], FailPointMode::Times(1), Some(options));
+    let _fp_guard = client.enable_failpoint(failpoint, None).await.unwrap();
+
+    RUNTIME
+        .timeout(Duration::from_secs(1), collection.find_one(doc! {}, None))
+        .await
+        .expect("operation should not time out")
+        .expect("find should succeed");
+}