Set timeout for the first request

Author: deuszx

linera-core/src/client/requests_scheduler/mod.rs

@@ -21,7 +21,7 @@ pub const CACHE_TTL_MS: u64 = 2000;
 pub const CACHE_MAX_SIZE: usize = 1000;
 pub const MAX_REQUEST_TTL_MS: u64 = 200;
 pub const ALPHA_SMOOTHING_FACTOR: f64 = 0.1;
-pub const STAGGERED_DELAY_MS: u64 = 75;
+pub const STAGGERED_DELAY_MS: u64 = 150;
 
 /// Configuration for the `RequestsScheduler`.
 #[derive(Debug, Clone)]

linera-core/src/client/requests_scheduler/scheduler.rs

@@ -4,7 +4,10 @@
 use std::{cmp::Ordering, collections::BTreeMap, future::Future, sync::Arc};
 
 use custom_debug_derive::Debug;
-use futures::stream::{FuturesUnordered, StreamExt};
+use futures::{
+    stream::{FuturesUnordered, StreamExt},
+    FutureExt,
+};
 use linera_base::{
     crypto::ValidatorPublicKey,
     data_types::{Blob, BlobContent, BlockHeight},
@@ -663,7 +666,15 @@ impl<Env: Environment> RequestsScheduler<Env> {
 
         // Execute the actual request
         tracing::trace!(key = ?key, peer = ?peer, "executing new request");
-        let result = operation(peer.clone()).await;
+        let result = futures::select! {
+            _timeout = linera_base::time::timer::sleep(self.retry_delay).fuse() => {
+                tracing::trace!(key = ?key, "retry delay elapsed, proceeding with request");
+                Err(NodeError::WorkerError { error: "timeout".to_string() }) // Placeholder error to trigger retries
+            },
+            result = operation(peer.clone()).fuse() => {
+                result
+            }
+        };
 
         // If the first request failed, try alternative peers in staggered parallel
         let result = if result.is_err() {
@@ -754,7 +765,8 @@ impl<Env: Environment> RequestsScheduler<Env> {
         }
 
         // Wait for first success or collect all failures
-        let mut last_error = None;
+        // Placeholder for the last error. If we error, this will be updated.
+        let mut last_error = NodeError::UnexpectedMessage;
 
         while let Some((peer_idx, result)) = futures.next().await {
             match result {
@@ -773,7 +785,7 @@ impl<Env: Environment> RequestsScheduler<Env> {
                         error = %e,
                         "staggered parallel retry attempt failed"
                     );
-                    last_error = Some(e);
+                    last_error = e;
                 }
             }
         }
@@ -783,7 +795,7 @@ impl<Env: Environment> RequestsScheduler<Env> {
             key = ?key,
             "all staggered parallel retry attempts failed"
         );
-        Err(last_error.unwrap_or(NodeError::UnexpectedMessage))
+        Err(last_error)
     }
 
     /// Returns all peers ordered by their score (highest first).