Introduce RequestsScheduler to track (and deduplicate) all requests and validators' scores (#4856)

Backport of #4752 

## Motivation

The Linera client needs to interact with multiple validator nodes
efficiently. Previously, the
  client would make individual requests to validators without:

1. Performance tracking: No mechanism to prefer faster, more reliable
validators
2. Request deduplication: Concurrent requests for the same data would
all hit the network, wasting
  bandwidth and validator resources
3. Response caching: Repeated requests for the same data would always go
to validators
  4. Load balancing: No rate limiting per validator, risking overload
5. Resilience: No fallback mechanism when a validator is slow or
unresponsive

  This led to:
  - Unnecessary network traffic and validator load
  - Poor user experience with redundant waiting
  - No optimization based on validator performance
  - Risk of overwhelming validators with too many concurrent requests
  - No recovery mechanism when validators are slow

##  Proposal

This PR introduces `RequestsScheduler`, a sophisticated request
orchestration layer that provides
intelligent peer selection, request deduplication, caching, and
performance-based routing.

  Key Features

  1. Performance Tracking with Exponential Moving Averages (EMA)
  - Tracks latency, success rate, and current load for each validator
  - Uses configurable weights to compute a composite performance score
  - Intelligently selects the best available validator for each request
  - Weighted random selection from top performers to avoid hotspots

  2. Request Deduplication
- Exact matching: Multiple concurrent requests for identical data are
deduplicated
- Subsumption-based matching: Smaller requests are satisfied by larger
in-flight requests that
contain the needed data (e.g., a request for blocks 10-12 can be
satisfied by an in-flight request
  for blocks 10-20)
- Broadcast mechanism ensures all waiting requesters receive the result
when the request completes
- Timeout handling: Stale in-flight requests (>200ms) are not
deduplicated against, allowing fresh
  attempts

  3. Response Caching
- Successfully completed requests are cached with configurable TTL
(default: 2 seconds)
  - LRU eviction when cache reaches maximum size (default: 1000 entries)
  - Works with both exact and subsumption matching
  - Only successful results are cached

  4. Slot-Based Rate Limiting
  - Each validator has a maximum concurrent request limit (default: 100)
- Async await mechanism: requests wait for available slots without
polling
  - Prevents overloading individual validators
  - Automatic slot release on request completion

  5. Alternative Peer Handling
- When multiple callers request the same data, they register as
"alternative peers"
- If the original request times out (>200ms), any alternative peer can
complete the request
  - The result is broadcast to all waiting requesters
  - Provides resilience against slow validators

  6. Modular Architecture

Created a new ` requests_scheduler` module with clear separation of
concerns:
```
 requests_scheduler/
  ├── mod.rs              - Module exports and constants
  ├── scheduler.rs          - RequestsScheduler orchestration logic
  ├── in_flight_tracker.rs - In-flight request tracking and deduplication
  ├── node_info.rs        - Per-validator performance tracking
  ├── request.rs          - Request types and result extraction
  └── scoring.rs          - Configurable scoring weights
```
  API

  High-level APIs:
```rust
  // Execute with best available validator
  scheduler.with_best(request_key, |peer| async {
      peer.download_certificates(chain_id, start, limit).await
  }).await

  // Execute with specific validator
  scheduler.with_peer(request_key, peer, |peer| async {
      peer.download_blob(blob_id).await
  }).await

  Configuration:
  let manager = RequestsScheduler::with_config(
      validator_nodes,
      max_requests_per_node: 100,
      weights: ScoringWeights { latency: 0.4, success: 0.4, load: 0.2 },
      alpha: 0.1,              // EMA smoothing factor
      max_expected_latency_ms: 5000.0,
      cache_ttl: Duration::from_secs(2),
      max_cache_size: 1000,
  );
```

  Benefits

- Reduced network load: Deduplication and caching eliminate redundant
requests
- Better performance: Intelligent peer selection routes to fastest
validators
  - Improved reliability: Alternative peer mechanism provides resilience
  - Protection for validators: Rate limiting prevents overload
- Efficient resource usage: EMA-based scoring optimizes validator
selection
- Clean architecture: Modular design makes code maintainable and
testable

  Metrics

  In production usage, this should significantly reduce:
  - Network traffic between clients and validators
  - Validator CPU/memory usage from redundant requests
  - Client request latency through caching and smart routing
  - Failed requests through performance tracking and rate limiting

The following metrics have been added to Prometheus (with compiled with
`--features metrics`):
- `requests_scheduler_response_time_ms` - Response time for requests to
validators in milliseconds
- `requests_scheduler_request_total` - Total number of requests made to
each validator
- `requests_scheduler_request_success` - Number of successful requests
to each validator (`(requests_scheduler_request_total -
requests_scheduler_request_success) / requests_scheduler_request_total`
is an error rate)
- `requests_scheduler_request_deduplication_total` - Number of requests
that were deduplicated by joining an in-flight request
- `requests_scheduler_request_cache_hit_total` - Number of requests that
were served from cache

## Test Plan

Existing CI makes sure we maintain backwards compatibility. Some tests
have been added to the new modules.

## Release Plan

- Nothing to do / These changes follow the usual release cycle.

## Links

<!--
Optional section for related PRs, related issues, and other references.

If needed, please create issues to track future improvements and link
them here.
-->
- [reviewer
checklist](https://github.com/linera-io/linera-protocol/blob/main/CONTRIBUTING.md#reviewer-checklist)

Author: deuszx

CLI.md

@@ -190,6 +190,21 @@ Client implementation and command-line tool for the Linera blockchain
 * `--max-joined-tasks <MAX_JOINED_TASKS>` — Maximum number of tasks that can are joined concurrently in the client
 
   Default value: `100`
+* `--max-accepted-latency-ms <MAX_ACCEPTED_LATENCY_MS>` — Maximum expected latency in milliseconds for score normalization
+
+  Default value: `5000`
+* `--cache-ttl-ms <CACHE_TTL_MS>` — Time-to-live for cached responses in milliseconds
+
+  Default value: `2000`
+* `--cache-max-size <CACHE_MAX_SIZE>` — Maximum number of entries in the cache
+
+  Default value: `1000`
+* `--max-request-ttl-ms <MAX_REQUEST_TTL_MS>` — Maximum latency for an in-flight request before we stop deduplicating it (in milliseconds)
+
+  Default value: `200`
+* `--alpha <ALPHA>` — Smoothing factor for Exponential Moving Averages (0 < alpha < 1). Higher values give more weight to recent observations. Typical values are between 0.01 and 0.5. A value of 0.1 means that 10% of the new observation is considered and 90% of the previous average is retained
+
+  Default value: `0.1`
 * `--storage <STORAGE_CONFIG>` — Storage configuration for the blockchain history
 * `--storage-max-concurrent-queries <STORAGE_MAX_CONCURRENT_QUERIES>` — The maximal number of simultaneous queries to the database
 * `--storage-max-stream-queries <STORAGE_MAX_STREAM_QUERIES>` — The maximal number of simultaneous stream queries to the database

linera-client/src/client_context.rs

@@ -301,6 +301,7 @@ where
             options.to_chain_client_options(),
             block_cache_size,
             execution_state_cache_size,
+            options.to_requests_scheduler_config(),
         );
 
         #[cfg(not(web))]
@@ -369,6 +370,7 @@ where
             },
             block_cache_size,
             execution_state_cache_size,
+            linera_core::client::RequestsSchedulerConfig::default(),
         );
 
         ClientContext {

linera-client/src/client_options.rs

@@ -186,6 +186,50 @@ pub struct ClientContextOptions {
     /// Maximum number of tasks that can are joined concurrently in the client.
     #[arg(long, default_value = "100")]
     pub max_joined_tasks: usize,
+
+    /// Maximum expected latency in milliseconds for score normalization.
+    #[arg(
+        long,
+        default_value_t = linera_core::client::requests_scheduler::MAX_ACCEPTED_LATENCY_MS,
+        env = "LINERA_REQUESTS_SCHEDULER_MAX_ACCEPTED_LATENCY_MS"
+    )]
+    pub max_accepted_latency_ms: f64,
+
+    /// Time-to-live for cached responses in milliseconds.
+    #[arg(
+        long,
+        default_value_t = linera_core::client::requests_scheduler::CACHE_TTL_MS,
+        env = "LINERA_REQUESTS_SCHEDULER_CACHE_TTL_MS"
+    )]
+    pub cache_ttl_ms: u64,
+
+    /// Maximum number of entries in the cache.
+    #[arg(
+        long,
+        default_value_t = linera_core::client::requests_scheduler::CACHE_MAX_SIZE,
+        env = "LINERA_REQUESTS_SCHEDULER_CACHE_MAX_SIZE"
+    )]
+    pub cache_max_size: usize,
+
+    /// Maximum latency for an in-flight request before we stop deduplicating it (in milliseconds).
+    #[arg(
+        long,
+        default_value_t = linera_core::client::requests_scheduler::MAX_REQUEST_TTL_MS,
+        env = "LINERA_REQUESTS_SCHEDULER_MAX_REQUEST_TTL_MS"
+    )]
+    pub max_request_ttl_ms: u64,
+
+    /// Smoothing factor for Exponential Moving Averages (0 < alpha < 1).
+    /// Higher values give more weight to recent observations.
+    /// Typical values are between 0.01 and 0.5.
+    /// A value of 0.1 means that 10% of the new observation is considered
+    /// and 90% of the previous average is retained.
+    #[arg(
+        long,
+        default_value_t = linera_core::client::requests_scheduler::ALPHA_SMOOTHING_FACTOR,
+        env = "LINERA_REQUESTS_SCHEDULER_ALPHA"
+    )]
+    pub alpha: f64,
 }
 
 impl ClientContextOptions {
@@ -218,6 +262,19 @@ impl ClientContextOptions {
             report_interval_secs: self.timing_interval,
         }
     }
+
+    /// Creates [`RequestsSchedulerConfig`] with the corresponding values.
+    pub(crate) fn to_requests_scheduler_config(
+        &self,
+    ) -> linera_core::client::RequestsSchedulerConfig {
+        linera_core::client::RequestsSchedulerConfig {
+            max_accepted_latency_ms: self.max_accepted_latency_ms,
+            cache_ttl_ms: self.cache_ttl_ms,
+            cache_max_size: self.cache_max_size,
+            max_request_ttl_ms: self.max_request_ttl_ms,
+            alpha: self.alpha,
+        }
+    }
 }
 
 #[derive(Debug, Clone, clap::Args)]

linera-client/src/unit_tests/chain_listener.rs

@@ -125,6 +125,7 @@ async fn test_chain_listener() -> anyhow::Result<()> {
             chain_client::Options::test_default(),
             5_000,
             10_000,
+            linera_core::client::RequestsSchedulerConfig::default(),
         )),
     };
     context
@@ -213,6 +214,7 @@ async fn test_chain_listener_admin_chain() -> anyhow::Result<()> {
             chain_client::Options::test_default(),
             5_000,
             10_000,
+            linera_core::client::RequestsSchedulerConfig::default(),
         )),
     };
     let context = Arc::new(Mutex::new(context));

linera-core/src/client/mod.rs

@@ -51,7 +51,7 @@ use crate::{
     data_types::{ChainInfo, ChainInfoQuery, ChainInfoResponse, RoundTimeout},
     environment::Environment,
     local_node::{LocalChainInfoExt as _, LocalNodeClient, LocalNodeError},
-    node::{CrossChainMessageDelivery, NodeError, ValidatorNode, ValidatorNodeProvider as _},
+    node::{CrossChainMessageDelivery, NodeError, ValidatorNodeProvider as _},
     notifier::ChannelNotifier,
     remote_node::RemoteNode,
     updater::{communicate_with_quorum, CommunicateAction, ValidatorUpdater},
@@ -67,6 +67,9 @@ pub use crate::data_types::ClientOutcome;
 #[cfg(test)]
 #[path = "../unit_tests/client_tests.rs"]
 mod client_tests;
+pub mod requests_scheduler;
+
+pub use requests_scheduler::{RequestsScheduler, RequestsSchedulerConfig, ScoringWeights};
 mod received_log;
 mod validator_trackers;
 
@@ -257,6 +260,8 @@ pub struct Client<Env: Environment> {
     /// Local node to manage the execution state and the local storage of the chains that we are
     /// tracking.
     local_node: LocalNodeClient<Env::Storage>,
+    /// Manages the requests sent to validator nodes.
+    requests_scheduler: RequestsScheduler<Env>,
     /// The admin chain ID.
     admin_id: ChainId,
     /// Chains that should be tracked by the client.
@@ -285,6 +290,7 @@ impl<Env: Environment> Client<Env> {
         options: chain_client::Options,
         block_cache_size: usize,
         execution_state_cache_size: usize,
+        requests_scheduler_config: requests_scheduler::RequestsSchedulerConfig,
     ) -> Self {
         let tracked_chains = Arc::new(RwLock::new(tracked_chains.into_iter().collect()));
         let state = WorkerState::new_for_client(
@@ -300,10 +306,12 @@ impl<Env: Environment> Client<Env> {
         .with_chain_worker_ttl(chain_worker_ttl)
         .with_sender_chain_worker_ttl(sender_chain_worker_ttl);
         let local_node = LocalNodeClient::new(state);
+        let requests_scheduler = RequestsScheduler::new(vec![], requests_scheduler_config);
 
         Self {
             environment,
             local_node,
+            requests_scheduler,
             chains: papaya::HashMap::new(),
             admin_id,
             tracked_chains,
@@ -484,8 +492,10 @@ impl<Env: Environment> Client<Env> {
                 .checked_sub(u64::from(next_height))
                 .ok_or(ArithmeticError::Overflow)?
                 .min(self.options.certificate_download_batch_size);
-            let certificates = remote_node
-                .query_certificates_from(chain_id, next_height, limit)
+
+            let certificates = self
+                .requests_scheduler
+                .download_certificates(remote_node, chain_id, next_height, limit)
                 .await?;
             let Some(info) = self.process_certificates(remote_node, certificates).await? else {
                 break;
@@ -499,28 +509,25 @@ impl<Env: Environment> Client<Env> {
 
     async fn download_blobs(
         &self,
-        remote_node: &RemoteNode<impl ValidatorNode>,
-        blob_ids: impl IntoIterator<Item = BlobId>,
+        remote_nodes: &[RemoteNode<Env::ValidatorNode>],
+        blob_ids: &[BlobId],
     ) -> Result<(), chain_client::Error> {
-        self.local_node
-            .store_blobs(
-                &futures::stream::iter(blob_ids.into_iter().map(|blob_id| async move {
-                    remote_node.try_download_blob(blob_id).await.unwrap()
-                }))
-                .buffer_unordered(self.options.max_joined_tasks)
-                .collect::<Vec<_>>()
-                .await,
-            )
-            .await
-            .map_err(Into::into)
+        let blobs = &self
+            .requests_scheduler
+            .download_blobs(remote_nodes, blob_ids, self.options.blob_download_timeout)
+            .await?
+            .ok_or_else(|| {
+                chain_client::Error::RemoteNodeError(NodeError::BlobsNotFound(blob_ids.to_vec()))
+            })?;
+        self.local_node.store_blobs(blobs).await.map_err(Into::into)
     }
 
     /// Tries to process all the certificates, requesting any missing blobs from the given node.
     /// Returns the chain info of the last successfully processed certificate.
     #[instrument(level = "trace", skip_all)]
     async fn process_certificates(
         &self,
-        remote_node: &RemoteNode<impl ValidatorNode>,
+        remote_node: &RemoteNode<Env::ValidatorNode>,
         certificates: Vec<ConfirmedBlockCertificate>,
     ) -> Result<Option<Box<ChainInfo>>, chain_client::Error> {
         let mut info = None;
@@ -535,7 +542,8 @@ impl<Env: Environment> Client<Env> {
             .await
         {
             Err(LocalNodeError::BlobsNotFound(blob_ids)) => {
-                self.download_blobs(remote_node, blob_ids).await?;
+                self.download_blobs(&[remote_node.clone()], &blob_ids)
+                    .await?;
             }
             x => {
                 x?;
@@ -546,7 +554,8 @@ impl<Env: Environment> Client<Env> {
             info = Some(
                 match self.handle_certificate(certificate.clone()).await {
                     Err(LocalNodeError::BlobsNotFound(blob_ids)) => {
-                        self.download_blobs(remote_node, blob_ids).await?;
+                        self.download_blobs(&[remote_node.clone()], &blob_ids)
+                            .await?;
                         self.handle_certificate(certificate).await?
                     }
                     x => x?,
@@ -802,7 +811,6 @@ impl<Env: Environment> Client<Env> {
     ) -> Result<(), chain_client::Error> {
         let certificate = Box::new(certificate);
         let block = certificate.block();
-
         // Recover history from the network.
         self.download_certificates(block.header.chain_id, block.header.height)
             .await?;
@@ -811,14 +819,9 @@ impl<Env: Environment> Client<Env> {
         if let Err(err) = self.process_certificate(certificate.clone()).await {
             match &err {
                 LocalNodeError::BlobsNotFound(blob_ids) => {
-                    let blobs = RemoteNode::download_blobs(
-                        blob_ids,
-                        &self.validator_nodes().await?,
-                        self.options.blob_download_timeout,
-                    )
-                    .await
-                    .ok_or(err)?;
-                    self.local_node.store_blobs(&blobs).await?;
+                    self.download_blobs(&self.validator_nodes().await?, blob_ids)
+                        .await
+                        .map_err(|_| err)?;
                     self.process_certificate(certificate).await?;
                 }
                 _ => {
@@ -855,14 +858,7 @@ impl<Env: Environment> Client<Env> {
         if let Err(err) = self.handle_certificate(certificate.clone()).await {
             match &err {
                 LocalNodeError::BlobsNotFound(blob_ids) => {
-                    let blobs = RemoteNode::download_blobs(
-                        blob_ids,
-                        &nodes,
-                        self.options.blob_download_timeout,
-                    )
-                    .await
-                    .ok_or(err)?;
-                    self.local_node.store_blobs(&blobs).await?;
+                    self.download_blobs(&nodes, blob_ids).await?;
                     self.handle_certificate(certificate.clone()).await?;
                 }
                 _ => {
@@ -916,8 +912,13 @@ impl<Env: Environment> Client<Env> {
                         // anything from the validator - let the function try the other validators
                         return Err(());
                     }
-                    let certificates = remote_node
-                        .download_certificates_by_heights(sender_chain_id, remote_heights)
+                    let certificates = self
+                        .requests_scheduler
+                        .download_certificates_by_heights(
+                            &remote_node,
+                            sender_chain_id,
+                            remote_heights,
+                        )
                         .await
                         .map_err(|_| ())?;
                     let mut certificates_with_check_results = vec![];
@@ -1073,8 +1074,13 @@ impl<Env: Environment> Client<Env> {
         // Stop if we've reached the height we've already processed.
         while current_height >= next_outbox_height {
             // Download the certificate for this height.
-            let downloaded = remote_node
-                .download_certificates_by_heights(sender_chain_id, vec![current_height])
+            let downloaded = self
+                .requests_scheduler
+                .download_certificates_by_heights(
+                    remote_node,
+                    sender_chain_id,
+                    vec![current_height],
+                )
                 .await?;
             let Some(certificate) = downloaded.into_iter().next() else {
                 return Err(chain_client::Error::CannotDownloadMissingSenderBlock {
@@ -1258,9 +1264,9 @@ impl<Env: Environment> Client<Env> {
                     if !required_blob_ids.is_empty() {
                         let mut blobs = Vec::new();
                         for blob_id in required_blob_ids {
-                            let blob_content = match remote_node
-                                .node
-                                .download_pending_blob(chain_id, blob_id)
+                            let blob_content = match self
+                                .requests_scheduler
+                                .download_pending_blob(remote_node, chain_id, blob_id)
                                 .await
                             {
                                 Ok(content) => content,
@@ -1356,9 +1362,9 @@ impl<Env: Environment> Client<Env> {
             Err(LocalNodeError::BlobsNotFound(blob_ids)) => {
                 let mut blobs = Vec::new();
                 for blob_id in blob_ids {
-                    let blob_content = remote_node
-                        .node
-                        .download_pending_blob(chain_id, blob_id)
+                    let blob_content = self
+                        .requests_scheduler
+                        .download_pending_blob(remote_node, chain_id, blob_id)
                         .await?;
                     blobs.push(Blob::new(blob_content));
                 }
@@ -1387,7 +1393,10 @@ impl<Env: Environment> Client<Env> {
             communicate_concurrently(
                 remote_nodes,
                 async move |remote_node| {
-                    let certificate = remote_node.download_certificate_for_blob(blob_id).await?;
+                    let certificate = self
+                        .requests_scheduler
+                        .download_certificate_for_blob(&remote_node, blob_id)
+                        .await?;
                     self.receive_sender_certificate(
                         certificate,
                         ReceiveCertificateMode::NeedsCheck,
@@ -1494,7 +1503,7 @@ impl<Env: Environment> Client<Env> {
 }
 
 /// Performs `f` in parallel on multiple nodes, starting with a quadratically increasing delay on
-/// each subsequent node. Returns error `err` is all of the nodes fail.
+/// each subsequent node. Returns error `err` if all of the nodes fail.
 async fn communicate_concurrently<'a, A, E1, E2, F, G, R, V>(
     nodes: &[RemoteNode<A>],
     f: F,