future: expose recorded attempted hosts

This commit exposes the `attempted_hosts` method on `CassFuture`,
allowing users to retrieve the hosts that were attempted during
the execution of a future. This can be used in proxy tests in the
wrapper itself, but the main use case is in the C++ integration
tests. The necessary bridging code is implemented in the next commit.

Author: wprzytula

scylla-rust-wrapper/src/future.rs

@@ -60,6 +60,8 @@ pub struct CassFuture {
     state: Mutex<CassFutureState>,
     result: OnceLock<CassFutureResult>,
     wait_for_value: Condvar,
+    #[cfg(cpp_integration_testing)]
+    recording_listener: Option<Arc<crate::integration_testing::RecordingHistoryListener>>,
 }
 
 impl FFI for CassFuture {
@@ -79,17 +81,30 @@ struct JoinHandleTimeout(JoinHandle<()>);
 impl CassFuture {
     pub(crate) fn make_raw(
         fut: impl Future<Output = CassFutureResult> + Send + 'static,
+        #[cfg(cpp_integration_testing)] recording_listener: Option<
+            Arc<crate::integration_testing::RecordingHistoryListener>,
+        >,
     ) -> CassOwnedSharedPtr<CassFuture, CMut> {
-        Self::new_from_future(fut).into_raw()
+        Self::new_from_future(
+            fut,
+            #[cfg(cpp_integration_testing)]
+            recording_listener,
+        )
+        .into_raw()
     }
 
     pub(crate) fn new_from_future(
         fut: impl Future<Output = CassFutureResult> + Send + 'static,
+        #[cfg(cpp_integration_testing)] recording_listener: Option<
+            Arc<crate::integration_testing::RecordingHistoryListener>,
+        >,
     ) -> Arc<CassFuture> {
         let cass_fut = Arc::new(CassFuture {
             state: Mutex::new(Default::default()),
             result: OnceLock::new(),
             wait_for_value: Condvar::new(),
+            #[cfg(cpp_integration_testing)]
+            recording_listener,
         });
         let cass_fut_clone = Arc::clone(&cass_fut);
         let join_handle = RUNTIME.spawn(async move {
@@ -125,6 +140,8 @@ impl CassFuture {
             state: Mutex::new(CassFutureState::default()),
             result: OnceLock::from(r),
             wait_for_value: Condvar::new(),
+            #[cfg(cpp_integration_testing)]
+            recording_listener: None,
         })
     }
 
@@ -300,6 +317,15 @@ impl CassFuture {
     fn into_raw(self: Arc<Self>) -> CassOwnedSharedPtr<Self, CMut> {
         ArcFFI::into_ptr(self)
     }
+
+    #[cfg(cpp_integration_testing)]
+    pub(crate) fn attempted_hosts(&self) -> Vec<std::net::SocketAddr> {
+        if let Some(listener) = &self.recording_listener {
+            listener.get_attempted_hosts()
+        } else {
+            vec![]
+        }
+    }
 }
 
 // Do not remove; this asserts that `CassFuture` implements Send + Sync,
@@ -527,7 +553,11 @@ mod tests {
             tokio::time::sleep(Duration::from_millis(10)).await;
             Err((CassError::CASS_OK, ERROR_MSG.into()))
         };
-        let cass_fut = CassFuture::make_raw(fut);
+        let cass_fut = CassFuture::make_raw(
+            fut,
+            #[cfg(cpp_integration_testing)]
+            None,
+        );
 
         struct PtrWrapper(CassBorrowedSharedPtr<'static, CassFuture, CMut>);
         unsafe impl Send for PtrWrapper {}
@@ -562,7 +592,11 @@ mod tests {
             tokio::time::sleep(Duration::from_micros(HUNDRED_MILLIS_IN_MICROS)).await;
             Err((CassError::CASS_OK, ERROR_MSG.into()))
         };
-        let cass_fut = CassFuture::make_raw(fut);
+        let cass_fut = CassFuture::make_raw(
+            fut,
+            #[cfg(cpp_integration_testing)]
+            None,
+        );
 
         unsafe {
             // This should timeout on tokio::time::timeout.
@@ -609,7 +643,11 @@ mod tests {
                 tokio::time::sleep(Duration::from_micros(HUNDRED_MILLIS_IN_MICROS)).await;
                 Err((CassError::CASS_OK, ERROR_MSG.into()))
             };
-            let cass_fut = CassFuture::make_raw(fut);
+            let cass_fut = CassFuture::make_raw(
+                fut,
+                #[cfg(cpp_integration_testing)]
+                None,
+            );
             let flag = Box::new(false);
             let flag_ptr = Box::into_raw(flag);
 

scylla-rust-wrapper/src/integration_testing.rs

@@ -193,7 +193,6 @@ impl RecordingHistoryListener {
         }
     }
 
-    #[expect(unused)] // <- next commit removes this
     pub(crate) fn get_attempted_hosts(&self) -> Vec<SocketAddr> {
         self.attempted_hosts.lock().unwrap().clone()
     }

scylla-rust-wrapper/src/session.rs

@@ -81,17 +81,21 @@ impl CassSessionInner {
         let exec_profile_map = cluster.execution_profile_map().clone();
         let host_filter = cluster.build_host_filter();
 
-        CassFuture::make_raw(Self::connect_fut(
-            session_opt,
-            session_builder,
-            exec_profile_map,
-            host_filter,
-            cluster
-                .get_client_id()
-                // If user did not set a client id, generate a random uuid v4.
-                .unwrap_or_else(uuid::Uuid::new_v4),
-            keyspace,
-        ))
+        CassFuture::make_raw(
+            Self::connect_fut(
+                session_opt,
+                session_builder,
+                exec_profile_map,
+                host_filter,
+                cluster
+                    .get_client_id()
+                    // If user did not set a client id, generate a random uuid v4.
+                    .unwrap_or_else(uuid::Uuid::new_v4),
+                keyspace,
+            ),
+            #[cfg(cpp_integration_testing)]
+            None,
+        )
     }
 
     async fn connect_fut(
@@ -255,10 +259,16 @@ pub unsafe extern "C" fn cass_session_execute_batch(
     };
 
     match request_timeout_ms {
-        Some(timeout_ms) => {
-            CassFuture::make_raw(async move { request_with_timeout(timeout_ms, future).await })
-        }
-        None => CassFuture::make_raw(future),
+        Some(timeout_ms) => CassFuture::make_raw(
+            async move { request_with_timeout(timeout_ms, future).await },
+            #[cfg(cpp_integration_testing)]
+            None,
+        ),
+        None => CassFuture::make_raw(
+            future,
+            #[cfg(cpp_integration_testing)]
+            None,
+        ),
     }
 }
 
@@ -297,19 +307,24 @@ pub unsafe extern "C" fn cass_session_execute(
     let mut statement = statement_opt.statement.clone();
 
     #[cfg(cpp_integration_testing)]
-    statement_opt.record_hosts.then(|| {
+    let recording_listener = statement_opt.record_hosts.then(|| {
         let recording_listener =
             Arc::new(crate::integration_testing::RecordingHistoryListener::new());
         match statement {
             BoundStatement::Simple(ref mut unprepared) => {
-                unprepared.query.set_history_listener(recording_listener);
+                unprepared
+                    .query
+                    .set_history_listener(Arc::clone(&recording_listener) as _);
             }
             BoundStatement::Prepared(ref mut prepared) => {
+                // It's extremely interesting to me that this `as _` cast is required
+                // for the type checker to accept this code.
                 Arc::make_mut(&mut prepared.statement)
                     .statement
-                    .set_history_listener(recording_listener);
+                    .set_history_listener(Arc::clone(&recording_listener) as _);
             }
         };
+        recording_listener
     });
 
     let statement_exec_profile = statement_opt.exec_profile.clone();
@@ -426,10 +441,16 @@ pub unsafe extern "C" fn cass_session_execute(
     };
 
     match request_timeout_ms {
-        Some(timeout_ms) => {
-            CassFuture::make_raw(async move { request_with_timeout(timeout_ms, future).await })
-        }
-        None => CassFuture::make_raw(future),
+        Some(timeout_ms) => CassFuture::make_raw(
+            async move { request_with_timeout(timeout_ms, future).await },
+            #[cfg(cpp_integration_testing)]
+            recording_listener,
+        ),
+        None => CassFuture::make_raw(
+            future,
+            #[cfg(cpp_integration_testing)]
+            recording_listener,
+        ),
     }
 }
 
@@ -449,31 +470,35 @@ pub unsafe extern "C" fn cass_session_prepare_from_existing(
 
     let statement = cass_statement.statement.clone();
 
-    CassFuture::make_raw(async move {
-        let query = match &statement {
-            BoundStatement::Simple(q) => q,
-            BoundStatement::Prepared(ps) => {
-                return Ok(CassResultValue::Prepared(ps.statement.clone()));
-            }
-        };
-
-        let session_guard = session.read().await;
-        if session_guard.is_none() {
-            return Err((
-                CassError::CASS_ERROR_LIB_NO_HOSTS_AVAILABLE,
-                "Session is not connected".msg(),
-            ));
-        }
-        let session = &session_guard.as_ref().unwrap().session;
-        let prepared = session
-            .prepare(query.query.clone())
-            .await
-            .map_err(|err| (err.to_cass_error(), err.msg()))?;
+    CassFuture::make_raw(
+        async move {
+            let query = match &statement {
+                BoundStatement::Simple(q) => q,
+                BoundStatement::Prepared(ps) => {
+                    return Ok(CassResultValue::Prepared(Arc::clone(&ps.statement)));
+                }
+            };
 
-        Ok(CassResultValue::Prepared(Arc::new(
-            CassPrepared::new_from_prepared_statement(prepared),
-        )))
-    })
+            let session_guard = session.read().await;
+            if session_guard.is_none() {
+                return Err((
+                    CassError::CASS_ERROR_LIB_NO_HOSTS_AVAILABLE,
+                    "Session is not connected".msg(),
+                ));
+            }
+            let session = &session_guard.as_ref().unwrap().session;
+            let prepared = session
+                .prepare(query.query.clone())
+                .await
+                .map_err(|err| (err.to_cass_error(), err.msg()))?;
+
+            Ok(CassResultValue::Prepared(Arc::new(
+                CassPrepared::new_from_prepared_statement(prepared),
+            )))
+        },
+        #[cfg(cpp_integration_testing)]
+        None,
+    )
 }
 
 #[unsafe(no_mangle)]
@@ -503,7 +528,7 @@ pub unsafe extern "C" fn cass_session_prepare_n(
         .unwrap_or_default();
     let query = Statement::new(query_str.to_string());
 
-    CassFuture::make_raw(async move {
+    let fut = async move {
         let session_guard = cass_session.read().await;
         if session_guard.is_none() {
             return Err((
@@ -524,7 +549,13 @@ pub unsafe extern "C" fn cass_session_prepare_n(
         Ok(CassResultValue::Prepared(Arc::new(
             CassPrepared::new_from_prepared_statement(prepared),
         )))
-    })
+    };
+
+    CassFuture::make_raw(
+        fut,
+        #[cfg(cpp_integration_testing)]
+        None,
+    )
 }
 
 #[unsafe(no_mangle)]
@@ -541,19 +572,23 @@ pub unsafe extern "C" fn cass_session_close(
         return ArcFFI::null();
     };
 
-    CassFuture::make_raw(async move {
-        let mut session_guard = session_opt.write().await;
-        if session_guard.is_none() {
-            return Err((
-                CassError::CASS_ERROR_LIB_UNABLE_TO_CLOSE,
-                "Already closing or closed".msg(),
-            ));
-        }
+    CassFuture::make_raw(
+        async move {
+            let mut session_guard = session_opt.write().await;
+            if session_guard.is_none() {
+                return Err((
+                    CassError::CASS_ERROR_LIB_UNABLE_TO_CLOSE,
+                    "Already closing or closed".msg(),
+                ));
+            }
 
-        *session_guard = None;
+            *session_guard = None;
 
-        Ok(CassResultValue::Empty)
-    })
+            Ok(CassResultValue::Empty)
+        },
+        #[cfg(cpp_integration_testing)]
+        None,
+    )
 }
 
 #[unsafe(no_mangle)]