Wire timeout flag to requests

src/args.rs

@@ -310,7 +310,7 @@ pub struct Args {
     #[clap(long)]
     pub read_consistency: Option<ReadConsistency>,
 
-    /// Timeout for requests in seconds
+    /// Timeout for requests in seconds (applied as both the client channel deadline and the server-side request timeout where supported).
     #[clap(long, value_parser = parse_number)]
     pub timeout: Option<usize>,
 

src/client.rs

@@ -24,8 +24,11 @@ pub fn get_config(args: &Args) -> Vec<QdrantConfig> {
             let api_key = std::env::var("QDRANT_API_KEY").ok();
 
             if let Some(timeout) = args.timeout {
-                config.set_timeout(Duration::from_secs(timeout as u64));
-                config.set_connect_timeout(Duration::from_secs(timeout as u64));
+                // Give the channel a small cushion over the server-side request timeout
+                // so the server's structured timeout error wins the race over a transport deadline.
+                let channel_timeout = Duration::from_secs(timeout as u64 + 5);
+                config.set_timeout(channel_timeout);
+                config.set_connect_timeout(channel_timeout);
             }
 
             if let Some(api_key) = api_key {

src/query.rs

@@ -42,13 +42,13 @@ async fn get_uuids(args: &Args, client: &Qdrant) -> Result<Vec<String>> {
     }
 
     // Retrieve existing UUIDs
-    let res = client
-        .scroll(
-            ScrollPointsBuilder::new(&args.collection_name)
-                .with_payload(true)
-                .limit(args.num_vectors as u32),
-        )
-        .await?;
+    let mut scroll_builder = ScrollPointsBuilder::new(&args.collection_name)
+        .with_payload(true)
+        .limit(args.num_vectors as u32);
+    if let Some(timeout) = args.timeout {
+        scroll_builder = scroll_builder.timeout(timeout as u64);
+    }
+    let res = client.scroll(scroll_builder).await?;
     let uuids: Vec<_> = res
         .result
         .iter()

src/scroll.rs

@@ -91,6 +91,10 @@ impl ScrollProcessor {
             request_builder = request_builder.read_consistency(read_consistency);
         }
 
+        if let Some(timeout) = self.args.timeout {
+            request_builder = request_builder.timeout(timeout as u64);
+        }
+
         let request = request_builder.build();
         let res = retry_with_clients(&self.clients, args, |client| client.scroll(request.clone()))
             .await?;

src/search.rs

@@ -256,8 +256,11 @@ impl SearchProcessor {
             (None, query_points)
         };
 
-        let batch_request_builder =
+        let mut batch_request_builder =
             QueryBatchPointsBuilder::new(self.args.collection_name.clone(), batch_query_points);
+        if let Some(timeout) = self.args.timeout {
+            batch_request_builder = batch_request_builder.timeout(timeout as u64);
+        }
 
         let request = batch_request_builder.build();
         let res_batch = retry_with_clients(&self.clients, args, |client| {
@@ -315,8 +318,11 @@ impl SearchProcessor {
         for point in &mut exact_query_points {
             point.params = Some(exact_search);
         }
-        let exact_request_builder =
+        let mut exact_request_builder =
             QueryBatchPointsBuilder::new(self.args.collection_name.clone(), exact_query_points);
+        if let Some(timeout) = self.args.timeout {
+            exact_request_builder = exact_request_builder.timeout(timeout as u64);
+        }
         let exact_request = exact_request_builder.build();
 
         let exact_res = retry_with_clients(&self.clients, args, |client| {

src/upsert.rs

@@ -182,6 +182,9 @@ impl UpsertProcessor {
         if let Some(shard_key) = &args.shard_key {
             request = request.shard_key_selector(vec![Key::Keyword(shard_key.to_string())]);
         }
+        if let Some(timeout) = self.args.timeout {
+            request = request.timeout(timeout as u64);
+        }
 
         let request = request.build();
         let res = retry_with_clients(&self.clients, args, |client| {
@@ -209,6 +212,9 @@ impl UpsertProcessor {
             if let Some(ordering) = self.args.write_ordering {
                 request_builder = request_builder.ordering(ordering);
             }
+            if let Some(timeout) = self.args.timeout {
+                request_builder = request_builder.timeout(timeout as u64);
+            }
 
             let request = request_builder.build();