Implement client-side timeouts

docs/source/SUMMARY.md

@@ -22,6 +22,7 @@
     - [Lightweight transaction query (LWT)](queries/lwt.md)
     - [USE keyspace](queries/usekeyspace.md)
     - [Schema agreement](queries/schema_agreement.md)
+    - [Client timeout](queries/client_timeout.md)
 
 - [Data Types](data-types/data-types.md)
     - [Bool, Tinyint, Smallint, Int, Bigint, Float, Double](data-types/primitive.md)

docs/source/queries/client_timeout.md

@@ -0,0 +1,60 @@
+# Client timeouts
+
+In order to make sure that your application can handle network failures and other disasters, database queries issued by
+Scylla Rust Driver can time out on the client side, without waiting (potentially indefinitely) for a server to respond.
+
+## Per-session settings
+
+A default per-session client timeout is set to 30 seconds. This parameter can be modified via the query builder by using the `client_timeout` method:
+```rust
+# extern crate scylla;
+# extern crate tokio;
+use scylla::{Session, SessionBuilder};
+use std::error::Error;
+use std::time::Duration;
+use std::net::{IpAddr, Ipv4Addr, SocketAddr};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let uri = std::env::var("SCYLLA_URI")
+        .unwrap_or_else(|_| "127.0.0.1:9042".to_string());
+
+    let session: Session = SessionBuilder::new()
+        .known_node(uri)
+        .known_node("127.0.0.72:4321")
+        .known_node("localhost:8000")
+        .connection_timeout(Duration::from_secs(3))
+        .client_timeout(Duration::from_millis(1500))
+        .known_node_addr(SocketAddr::new(
+            IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)),
+            9000,
+        ))
+        .build()
+        .await?;
+
+    Ok(())
+}
+```
+
+## Per-query settings
+
+Queries, prepared statements and batches can also have their own, unique client timeout parameter. That allows overriding the default session configuration and specifying a timeout for a single particular type of query. Example:
+
+```rust
+# extern crate scylla;
+# use scylla::Session;
+# use std::error::Error;
+# async fn check_only_compiles(session: &Session) -> Result<(), Box<dyn Error>> {
+use std::time::Duration;
+use scylla::query::Query;
+
+// Create a Query manually to change the client timeout
+let mut my_query: Query = Query::new("INSERT INTO ks.tab (a) VALUES(?) IF NOT EXISTS".to_string());
+my_query.set_client_timeout(Some(Duration::from_secs(3)));
+
+// Insert a value into the table
+let to_insert: i32 = 12345;
+session.query(my_query, (to_insert,)).await?;
+# Ok(())
+# }
+```

docs/source/queries/queries.md

@@ -4,7 +4,7 @@ This driver supports all query types available in Scylla:
 * [Simple queries](simple.md)
     * Easy to use
     * Poor performance
-    * Primitve load balancing
+    * Primitive load balancing
 * [Prepared queries](prepared.md)
     * Need to be prepared before use
     * Fast
@@ -19,7 +19,11 @@ This driver supports all query types available in Scylla:
 Additionaly there is special functionality to enable `USE KEYSPACE` queries:
 [USE keyspace](usekeyspace.md)
 
-Queries are fully asynchronous - you can run as many of them in parallel as you wish.
+Queries can be subject to driver-side timeouts: [client timeout](client_timeout.md)
+
+Queries are fully asynchronous - you can run as many of them in parallel as you wish,
+but be mindful of the per-connection limit of 32768 limit per connection imposed
+by the CQL protocol.
 
 ```eval_rst
 .. toctree::
@@ -35,4 +39,5 @@ Queries are fully asynchronous - you can run as many of them in parallel as you
    usekeyspace
    schema_agreement
    lwt
+   client_timeout
 ```

scylla/src/statement/batch.rs

@@ -115,6 +115,19 @@ impl Batch {
     pub fn get_timestamp(&self) -> Option<i64> {
         self.config.timestamp
     }
+
+    /// Sets the client-side timeout for this statement.
+    /// If not None, the driver will stop waiting for the request
+    /// to finish after `timeout` passed.
+    /// Otherwise, default session client timeout will be applied.
+    pub fn set_client_timeout(&mut self, timeout: Option<std::time::Duration>) {
+        self.config.client_timeout = timeout
+    }
+
+    /// Gets client timeout associated with this query
+    pub fn get_client_timeout(&mut self) -> Option<std::time::Duration> {
+        self.config.client_timeout
+    }
 }
 
 impl Default for Batch {

scylla/src/statement/mod.rs

@@ -20,6 +20,7 @@ pub struct StatementConfig {
 
     pub tracing: bool,
     pub timestamp: Option<i64>,
+    pub client_timeout: Option<std::time::Duration>,
 }
 
 impl Default for StatementConfig {
@@ -32,6 +33,7 @@ impl Default for StatementConfig {
             speculative_execution_policy: None,
             tracing: false,
             timestamp: None,
+            client_timeout: None,
         }
     }
 }
@@ -49,6 +51,7 @@ impl Clone for StatementConfig {
             speculative_execution_policy: self.speculative_execution_policy.clone(),
             tracing: self.tracing,
             timestamp: self.timestamp,
+            client_timeout: self.client_timeout,
         }
     }
 }

scylla/src/statement/prepared_statement.rs

@@ -202,6 +202,19 @@ impl PreparedStatement {
     pub fn get_timestamp(&self) -> Option<i64> {
         self.config.timestamp
     }
+
+    /// Sets the client-side timeout for this statement.
+    /// If not None, the driver will stop waiting for the request
+    /// to finish after `timeout` passed.
+    /// Otherwise, default session client timeout will be applied.
+    pub fn set_client_timeout(&mut self, timeout: Option<std::time::Duration>) {
+        self.config.client_timeout = timeout
+    }
+
+    /// Gets client timeout associated with this query
+    pub fn get_client_timeout(&mut self) -> Option<std::time::Duration> {
+        self.config.client_timeout
+    }
 }
 
 #[derive(Debug, Error, PartialEq, Eq, PartialOrd, Ord)]

scylla/src/statement/query.rs

@@ -117,6 +117,19 @@ impl Query {
     pub fn get_timestamp(&self) -> Option<i64> {
         self.config.timestamp
     }
+
+    /// Sets the client-side timeout for this statement.
+    /// If not None, the driver will stop waiting for the request
+    /// to finish after `timeout` passed.
+    /// Otherwise, default session client timeout will be applied.
+    pub fn set_client_timeout(&mut self, timeout: Option<std::time::Duration>) {
+        self.config.client_timeout = timeout
+    }
+
+    /// Gets client timeout associated with this query
+    pub fn get_client_timeout(&mut self) -> Option<std::time::Duration> {
+        self.config.client_timeout
+    }
 }
 
 impl From<String> for Query {