RUST-945 Check that explicit sessions were created on the correct client

src/client/executor.rs

@@ -72,6 +72,15 @@ impl Client {
         }
         match session.into() {
             Some(session) => {
+                if !Arc::ptr_eq(&self.inner, &session.client().inner) {
+                    return Err(ErrorKind::InvalidArgument {
+                        message: "the session provided to an operation must be created from the \
+                                  same client as the collection/database"
+                            .into(),
+                    }
+                    .into());
+                }
+
                 if let Some(SelectionCriteria::ReadPreference(read_preference)) =
                     op.selection_criteria()
                 {

src/selection_criteria.rs

@@ -1,5 +1,4 @@
-use std::convert::TryInto;
-use std::{collections::HashMap, sync::Arc, time::Duration};
+use std::{collections::HashMap, convert::TryInto, sync::Arc, time::Duration};
 
 use derivative::Derivative;
 use serde::{de::Error as SerdeError, Deserialize, Deserializer};

src/test/spec/sessions.rs

@@ -1,6 +1,10 @@
-use tokio::sync::RwLockWriteGuard;
+use tokio::sync::{RwLockReadGuard, RwLockWriteGuard};
 
-use crate::test::{run_spec_test, LOCK};
+use crate::{
+    bson::doc,
+    error::ErrorKind,
+    test::{run_spec_test, TestClient, LOCK},
+};
 
 use super::{run_unified_format_test, run_v2_test};
 
@@ -17,3 +21,38 @@ async fn run_legacy() {
     let _guard: RwLockWriteGuard<()> = LOCK.run_exclusively().await;
     run_spec_test(&["sessions", "legacy"], run_v2_test).await;
 }
+
+#[cfg_attr(feature = "tokio-runtime", tokio::test(flavor = "multi_thread"))]
+#[cfg_attr(feature = "async-std-runtime", async_std::test)]
+#[function_name::named]
+async fn explicit_session_created_on_same_client() {
+    let _guard: RwLockReadGuard<_> = LOCK.run_concurrently().await;
+
+    let client0 = TestClient::new().await;
+    let client1 = TestClient::new().await;
+
+    let mut session0 = client0.start_session(None).await.unwrap();
+    let mut session1 = client1.start_session(None).await.unwrap();
+
+    let db = client0.database(function_name!());
+    let err = db
+        .list_collections_with_session(None, None, &mut session1)
+        .await
+        .unwrap_err();
+    match *err.kind {
+        ErrorKind::InvalidArgument { message } => assert!(message.contains("session provided")),
+        other => panic!("expected InvalidArgument error, got {:?}", other),
+    }
+
+    let coll = client1
+        .database(function_name!())
+        .collection(function_name!());
+    let err = coll
+        .insert_one_with_session(doc! {}, None, &mut session0)
+        .await
+        .unwrap_err();
+    match *err.kind {
+        ErrorKind::InvalidArgument { message } => assert!(message.contains("session provided")),
+        other => panic!("expected InvalidArgument error, got {:?}", other),
+    }
+}