Dk/schema query increased timeouts

.github/workflows/authenticate_test.yml

@@ -21,7 +21,8 @@ jobs:
     runs-on: ubuntu-latest
     services:
       scylladb:
-        image: scylladb/scylla-passauth
+        image: scylladb/scylla
+        command: --authenticator PasswordAuthenticator
         ports:
           - 9042:9042
         options: --health-cmd "cqlsh --username cassandra --password cassandra --debug" --health-interval 5s --health-retries 30

.github/workflows/tls.yml

@@ -20,7 +20,8 @@ jobs:
     timeout-minutes: 60
     services:
       scylladb:
-        image: scylladb/scylla-tls
+        image: scylla-tls
+        build: ./test/tls
         ports:
           - 9042:9042
           - 9142:9142

scylla-cql/src/types/serialize/batch.rs

@@ -384,7 +384,7 @@ where
     fn is_empty_next(&mut self) -> Option<bool> {
         self.0
             .next_serialized()
-            .map(|sv| sv.map_or(false, |sv| sv.len() == 0))
+            .map(|sv| sv.is_ok_and(|sv| sv.len() == 0))
     }
 
     #[inline]

scylla/src/transport/cluster.rs

@@ -150,11 +150,13 @@ struct UseKeyspaceRequest {
 }
 
 impl Cluster {
+    #[allow(clippy::too_many_arguments)]
     pub(crate) async fn new(
         known_nodes: Vec<InternalKnownNode>,
         pool_config: PoolConfig,
         keyspaces_to_fetch: Vec<String>,
         fetch_schema_metadata: bool,
+        metadata_request_serverside_timeout: Option<Duration>,
         host_filter: Option<Arc<dyn HostFilter>>,
         cluster_metadata_refresh_interval: Duration,
         tablet_receiver: tokio::sync::mpsc::Receiver<(TableSpec<'static>, RawTablet)>,
@@ -170,6 +172,7 @@ impl Cluster {
             control_connection_repair_sender,
             pool_config.connection_config.clone(),
             pool_config.keepalive_interval,
+            metadata_request_serverside_timeout,
             server_events_sender,
             keyspaces_to_fetch,
             fetch_schema_metadata,

scylla/src/transport/connection.rs

@@ -2377,7 +2377,7 @@ impl VerifiedKeyspaceName {
 }
 
 #[cfg(test)]
-mod tests {
+pub(crate) mod tests {
     use assert_matches::assert_matches;
     use scylla_cql::frame::protocol_features::{
         LWT_OPTIMIZATION_META_BIT_MASK_KEY, SCYLLA_LWT_ADD_METADATA_MARK_EXTENSION,
@@ -2406,7 +2406,7 @@ mod tests {
     use std::time::Duration;
 
     // Just like resolve_hostname in session.rs
-    async fn resolve_hostname(hostname: &str) -> SocketAddr {
+    pub(crate) async fn resolve_hostname(hostname: &str) -> SocketAddr {
         match tokio::net::lookup_host(hostname).await {
             Ok(mut addrs) => addrs.next().unwrap(),
             Err(_) => {

scylla/src/transport/session.rs

@@ -277,6 +277,9 @@ pub struct SessionConfig {
     /// If true, full schema is fetched with every metadata refresh.
     pub fetch_schema_metadata: bool,
 
+    /// Custom timeout for requests that query metadata.
+    pub metadata_request_serverside_timeout: Option<Duration>,
+
     /// Interval of sending keepalive requests.
     /// If `None`, keepalives are never sent, so `Self::keepalive_timeout` has no effect.
     pub keepalive_interval: Option<Duration>,
@@ -385,6 +388,7 @@ impl SessionConfig {
             disallow_shard_aware_port: false,
             keyspaces_to_fetch: Vec::new(),
             fetch_schema_metadata: true,
+            metadata_request_serverside_timeout: Some(Duration::from_secs(2)),
             keepalive_interval: Some(Duration::from_secs(30)),
             keepalive_timeout: Some(Duration::from_secs(30)),
             schema_agreement_timeout: Duration::from_secs(60),
@@ -1103,6 +1107,7 @@ where
             pool_config,
             config.keyspaces_to_fetch,
             config.fetch_schema_metadata,
+            config.metadata_request_serverside_timeout,
             config.host_filter,
             config.cluster_metadata_refresh_interval,
             tablet_receiver,

scylla/src/transport/session_builder.rs

@@ -594,7 +594,7 @@ impl<K: SessionBuilderKind> GenericSessionBuilder<K> {
     /// # Ok(())
     /// # }
     /// ```
-    pub fn connection_timeout(mut self, duration: std::time::Duration) -> Self {
+    pub fn connection_timeout(mut self, duration: Duration) -> Self {
         self.config.connect_timeout = duration;
         self
     }
@@ -706,6 +706,30 @@ impl<K: SessionBuilderKind> GenericSessionBuilder<K> {
         self
     }
 
+    /// Set the server-side timeout for metadata queries.
+    /// The default is `Some(Duration::from_secs(2))`. It means that
+    /// the all metadata queries will be set the 2 seconds timeout
+    /// no matter what timeout is set as a cluster default.
+    /// This prevents timeouts of schema queries when the schema is large
+    /// and the default timeout is configured as tight.
+    ///
+    /// # Example
+    /// ```
+    /// # use scylla::{Session, SessionBuilder};
+    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
+    /// let session: Session = SessionBuilder::new()
+    ///     .known_node("127.0.0.1:9042")
+    ///     .metadata_request_serverside_timeout(std::time::Duration::from_secs(5))
+    ///     .build()
+    ///     .await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn metadata_request_serverside_timeout(mut self, timeout: Duration) -> Self {
+        self.config.metadata_request_serverside_timeout = Some(timeout);
+        self
+    }
+
     /// Set the keepalive interval.
     /// The default is `Some(Duration::from_secs(30))`, which corresponds
     /// to keepalive CQL messages being sent every 30 seconds.