session: fix closing semantics

Makefile

@@ -37,6 +37,7 @@ SCYLLA_TEST_FILTER := $(subst ${SPACE},${EMPTY},ClusterTests.*\
 :DisconnectedNullStringApiArgsTest.*\
 :MetricsTests.*\
 :DcAwarePolicyTest.*\
+:AsyncTests.*\
 :-SchemaMetadataTest.Integration_Cassandra_RegularMetadataNotMarkedVirtual\
 :SchemaMetadataTest.Integration_Cassandra_VirtualMetadata\
 :HeartbeatTests.Integration_Cassandra_HeartbeatFailed\
@@ -95,6 +96,7 @@ CASSANDRA_TEST_FILTER := $(subst ${SPACE},${EMPTY},ClusterTests.*\
 :DisconnectedNullStringApiArgsTest.*\
 :MetricsTests.*\
 :DcAwarePolicyTest.*\
+:AsyncTests.*\
 :-PreparedTests.Integration_Cassandra_FailFastWhenPreparedIDChangesDuringReprepare\
 :SchemaMetadataTest.Integration_Cassandra_RegularMetadataNotMarkedVirtual\
 :SchemaMetadataTest.Integration_Cassandra_VirtualMetadata\

scylla-rust-wrapper/Cargo.toml

@@ -28,6 +28,8 @@ tracing = "0.1.37"
 futures = "0.3"
 thiserror = "1.0"
 yoke = { version = "0.8.0", features = ["derive"] }
+# Holds the CassSession.
+arc-swap = "1.3.0"
 
 [build-dependencies]
 bindgen = "0.65"

scylla-rust-wrapper/src/argconv.rs

@@ -240,7 +240,7 @@ impl<'a, T: Sized, P: Properties> CassPtr<'a, T, P> {
 
 /// Pointer constructors.
 impl<T: Sized, P: Properties> CassPtr<'_, T, P> {
-    fn null() -> Self {
+    pub(crate) fn null() -> Self {
         CassPtr {
             ptr: None,
             _phantom: PhantomData,

scylla-rust-wrapper/src/batch.rs

@@ -5,7 +5,7 @@ use crate::argconv::{
 use crate::cass_error::CassError;
 pub use crate::cass_types::CassBatchType;
 use crate::cass_types::{CassConsistency, make_batch_type};
-use crate::config_value::MaybeUnsetConfig;
+use crate::config_value::{MaybeUnsetConfig, RequestTimeout};
 use crate::exec_profile::PerStatementExecProfile;
 use crate::retry_policy::CassRetryPolicy;
 use crate::statement::{BoundStatement, CassStatement};
@@ -18,8 +18,6 @@ use std::sync::Arc;
 
 pub struct CassBatch {
     pub(crate) state: Arc<CassBatchState>,
-    pub(crate) batch_request_timeout_ms: Option<cass_uint64_t>,
-
     pub(crate) exec_profile: Option<PerStatementExecProfile>,
 }
 
@@ -43,7 +41,6 @@ pub unsafe extern "C" fn cass_batch_new(
                 batch: Batch::new(batch_type),
                 bound_values: Vec::new(),
             }),
-            batch_request_timeout_ms: None,
             exec_profile: None,
         }))
     } else {
@@ -215,7 +212,25 @@ pub unsafe extern "C" fn cass_batch_set_request_timeout(
         tracing::error!("Provided null batch pointer to cass_batch_set_request_timeout!");
         return CassError::CASS_ERROR_LIB_BAD_PARAMS;
     };
-    batch.batch_request_timeout_ms = Some(timeout_ms);
+    let maybe_unset_timeout =
+        MaybeUnsetConfig::<RequestTimeout>::from_c_value_infallible(timeout_ms);
+
+    // `Batch::set_request_timeout` expects an Option<Duration> with unusual semantics:
+    // - `None` means "ignore me and use the default timeout from the cluster/execution profile" - this is
+    //   different than other configuration parameters such as retry policy or serial consistency,
+    //   where `None` means "unset this parameter";
+    // - `Some(timeout)` means "use timeout of given value".
+    // Therefore, to acquire "no timeout" semantics, we need to emulate it with an extremely long timeout.
+    let timeout = match maybe_unset_timeout {
+        MaybeUnsetConfig::Unset => None,
+        MaybeUnsetConfig::Set(RequestTimeout(timeout)) => {
+            Some(timeout.unwrap_or(RequestTimeout::INFINITE))
+        }
+    };
+
+    Arc::make_mut(&mut batch.state)
+        .batch
+        .set_request_timeout(timeout);
 
     CassError::CASS_OK
 }

scylla-rust-wrapper/src/config_value.rs

@@ -1,6 +1,8 @@
+use std::{convert::Infallible, time::Duration};
+
 use scylla::statement::{Consistency, SerialConsistency};
 
-use crate::cass_types::CassConsistency;
+use crate::{cass_types::CassConsistency, types::cass_uint64_t};
 
 /// Represents a configuration value that may or may not be set.
 /// If a configuration value is unset, it means that the default value
@@ -44,6 +46,14 @@ impl<T: MaybeUnsetConfigValue> MaybeUnsetConfig<T> {
     }
 }
 
+impl<T: MaybeUnsetConfigValue<Error = Infallible>> MaybeUnsetConfig<T> {
+    /// Converts a maybe unset C value to a Rust value. Available for C values that are guaranteed
+    /// to be valid and thus never return an error.
+    pub(crate) fn from_c_value_infallible(cvalue: T::CValue) -> Self {
+        Self::from_c_value(cvalue).unwrap_or_else(|never| match never {})
+    }
+}
+
 impl MaybeUnsetConfigValue for Consistency {
     type CValue = CassConsistency;
     type Error = ();
@@ -98,3 +108,25 @@ impl MaybeUnsetConfigValue for Option<SerialConsistency> {
         }
     }
 }
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct RequestTimeout(pub(crate) Option<Duration>);
+
+impl RequestTimeout {
+    pub(crate) const INFINITE: Duration = Duration::MAX;
+}
+
+impl MaybeUnsetConfigValue for RequestTimeout {
+    type CValue = cass_uint64_t;
+    type Error = Infallible;
+
+    fn is_unset(cvalue: &Self::CValue) -> bool {
+        *cvalue == cass_uint64_t::MAX
+    }
+
+    fn from_set_c_value(cvalue: Self::CValue) -> Result<Self, Self::Error> {
+        Ok(RequestTimeout(
+            (cvalue != 0).then(|| Duration::from_millis(cvalue)),
+        ))
+    }
+}

scylla-rust-wrapper/src/future.rs

@@ -79,6 +79,10 @@ enum FutureError {
 struct JoinHandleTimeout(JoinHandle<()>);
 
 impl CassFuture {
+    pub(crate) fn make_ready_raw(res: CassFutureResult) -> CassOwnedSharedPtr<CassFuture, CMut> {
+        Self::new_ready(res).into_raw()
+    }
+
     pub(crate) fn make_raw(
         fut: impl Future<Output = CassFutureResult> + Send + 'static,
         #[cfg(cpp_integration_testing)] recording_listener: Option<
@@ -133,8 +137,6 @@ impl CassFuture {
         cass_fut
     }
 
-    // This is left just because it might be useful in tests.
-    #[expect(unused)]
     pub(crate) fn new_ready(r: CassFutureResult) -> Arc<Self> {
         Arc::new(CassFuture {
             state: Mutex::new(CassFutureState::default()),

scylla-rust-wrapper/src/lib.rs

@@ -165,7 +165,13 @@ pub(crate) mod cass_metrics_types {
     include_bindgen_generated!("cppdriver_metrics_types.rs");
 }
 
-pub(crate) static RUNTIME: LazyLock<Runtime> = LazyLock::new(|| Runtime::new().unwrap());
+pub(crate) static RUNTIME: LazyLock<Runtime> = LazyLock::new(|| {
+    tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(2)
+        .enable_all()
+        .build()
+        .unwrap()
+});
 pub(crate) static LOGGER: LazyLock<RwLock<Logger>> = LazyLock::new(|| {
     RwLock::new(Logger {
         cb: Some(stderr_log_callback),