Use streaming DownloadBlobs RPC for batch blob downloads

CLI.md

@@ -200,7 +200,7 @@ Client implementation and command-line tool for the Linera blockchain
 * `--quorum-grace-period <QUORUM_GRACE_PERIOD>` — An additional delay, after reaching a quorum, to wait for additional validator signatures, as a fraction of time taken to reach quorum
 
   Default value: `0.2`
-* `--blob-download-timeout-ms <BLOB_DOWNLOAD_TIMEOUT>` — The delay when downloading a blob, after which we try a second validator, in milliseconds
+* `--blob-download-timeout-ms <BLOB_DOWNLOAD_TIMEOUT>` — The maximum time without progress (stream opening or a new blob arriving) when downloading blobs from a validator, after which we try the next validator, in milliseconds
 
   Default value: `1000`
 * `--cert-batch-download-timeout-ms <CERTIFICATE_BATCH_DOWNLOAD_TIMEOUT>` — The delay when downloading a batch of certificates, after which we try a second validator, in milliseconds

linera-client/src/client_options.rs

@@ -212,7 +212,9 @@ pub struct Options {
     #[arg(long, default_value_t = DEFAULT_QUORUM_GRACE_PERIOD)]
     pub quorum_grace_period: f64,
 
-    /// The delay when downloading a blob, after which we try a second validator, in milliseconds.
+    /// The maximum time without progress (stream opening or a new blob arriving) when
+    /// downloading blobs from a validator, after which we try the next validator, in
+    /// milliseconds.
     #[arg(
         long = "blob-download-timeout-ms",
         default_value = "1000",

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

@@ -100,7 +100,8 @@ pub struct Options {
     /// An additional delay, after reaching a quorum, to wait for additional validator signatures,
     /// as a fraction of time taken to reach quorum.
     pub quorum_grace_period: f64,
-    /// The delay when downloading a blob, after which we try a second validator.
+    /// The maximum time without progress (stream opening or a new blob arriving) when
+    /// downloading blobs from a validator, after which we try the next validator.
     pub blob_download_timeout: Duration,
     /// The delay when downloading a batch of certificates, after which we try a second validator.
     pub certificate_batch_download_timeout: Duration,

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

@@ -4,7 +4,7 @@
 use std::fmt;
 
 use linera_base::{
-    data_types::{Blob, BlobContent, BlockHeight},
+    data_types::{BlobContent, BlockHeight},
     identifiers::{BlobId, ChainId},
 };
 use linera_chain::types::ConfirmedBlockCertificate;
@@ -21,8 +21,6 @@ pub enum RequestKey {
         chain_id: ChainId,
         heights: Vec<BlockHeight>,
     },
-    /// Download a blob by ID
-    Blob(BlobId),
     /// Download a pending blob
     PendingBlob { chain_id: ChainId, blob_id: BlobId },
     /// Download certificate for a specific blob
@@ -37,7 +35,6 @@ impl fmt::Debug for RequestKey {
                 .field("chain_id", chain_id)
                 .field("heights", &CompressedHeights(heights))
                 .finish(),
-            RequestKey::Blob(blob_id) => f.debug_tuple("Blob").field(blob_id).finish(),
             RequestKey::PendingBlob { chain_id, blob_id } => f
                 .debug_struct("PendingBlob")
                 .field("chain_id", chain_id)
@@ -67,7 +64,7 @@ impl RequestKey {
     ///
     /// # Returns
     /// - `Some((chain_id, heights))` for certificate requests, where heights are sorted
-    /// - `None` for non-certificate requests (Blob, PendingBlob, CertificateForBlob)
+    /// - `None` for non-certificate requests (PendingBlob, CertificateForBlob)
     fn heights(&self) -> Option<Vec<BlockHeight>> {
         match self {
             RequestKey::Certificates { heights, .. } => Some(heights.clone()),
@@ -80,7 +77,6 @@ impl RequestKey {
 #[derive(Debug, Clone)]
 pub enum RequestResult {
     Certificates(Vec<ConfirmedBlockCertificate>),
-    Blob(Option<Blob>),
     BlobContent(BlobContent),
     Certificate(Box<ConfirmedBlockCertificate>),
 }
@@ -90,12 +86,6 @@ pub enum RequestResult {
 pub trait Cacheable: TryFrom<RequestResult> + Into<RequestResult> {}
 impl<T> Cacheable for T where T: TryFrom<RequestResult> + Into<RequestResult> {}
 
-impl From<Option<Blob>> for RequestResult {
-    fn from(blob: Option<Blob>) -> Self {
-        RequestResult::Blob(blob)
-    }
-}
-
 impl From<Vec<ConfirmedBlockCertificate>> for RequestResult {
     fn from(certs: Vec<ConfirmedBlockCertificate>) -> Self {
         RequestResult::Certificates(certs)
@@ -114,17 +104,6 @@ impl From<ConfirmedBlockCertificate> for RequestResult {
     }
 }
 
-impl TryFrom<RequestResult> for Option<Blob> {
-    type Error = ();
-
-    fn try_from(result: RequestResult) -> Result<Self, Self::Error> {
-        match result {
-            RequestResult::Blob(blob) => Ok(blob),
-            _ => Err(()),
-        }
-    }
-}
-
 impl TryFrom<RequestResult> for Vec<ConfirmedBlockCertificate> {
     type Error = ();
 
@@ -322,7 +301,7 @@ mod tests {
         };
 
         // Non-certificate result should return None
-        let blob_result = RequestResult::Blob(None);
+        let blob_result = RequestResult::BlobContent(super::BlobContent::new_data(Vec::new()));
         assert!(req1.try_extract_result(&req2, &blob_result).is_none());
     }
 

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

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use std::{
-    collections::BTreeMap,
+    collections::{BTreeMap, BTreeSet, HashMap},
     future::Future,
     sync::{
         atomic::{AtomicU32, Ordering},
@@ -11,15 +11,18 @@ use std::{
 };
 
 use custom_debug_derive::Debug;
-use futures::stream::{FuturesUnordered, StreamExt};
+use futures::stream::StreamExt;
 use linera_base::{
     crypto::ValidatorPublicKey,
     data_types::{Blob, BlobContent, BlockHeight},
     identifiers::{BlobId, ChainId},
     time::{Duration, Instant},
 };
 use linera_chain::types::ConfirmedBlockCertificate;
-use rand::distributions::{Distribution, WeightedIndex};
+use rand::{
+    distributions::{Distribution, WeightedIndex},
+    seq::SliceRandom,
+};
 use tracing::{instrument, warn};
 
 use super::{
@@ -298,61 +301,168 @@ impl<Env: Environment> RequestsScheduler<Env> {
         .await
     }
 
+    /// Downloads the blobs with the given IDs using the streaming `download_blobs` RPC.
+    ///
+    /// Tries peers in shuffled order: each peer is asked for the IDs that are still
+    /// missing after previous peers' attempts. The remote stream yields blobs in
+    /// requested order; partial progress is preserved across peers when a stream
+    /// errors mid-way (e.g. the validator is missing one of the IDs).
+    ///
+    /// Returns `Ok(None)` if any blob could not be retrieved from any peer.
+    /// Returns `Err` only if every peer attempt failed before returning a single blob.
     #[instrument(level = "trace", skip_all)]
-    async fn download_blob(
+    pub async fn download_blobs(
         &self,
         peers: &[RemoteNode<Env::ValidatorNode>],
-        blob_id: BlobId,
+        blob_ids: &[BlobId],
         timeout: Duration,
-    ) -> Result<Option<Blob>, NodeError> {
-        let key = RequestKey::Blob(blob_id);
-        communicate_concurrently(
-            peers,
-            async move |peer| {
-                self.with_peer(key, peer, move |peer| async move {
-                    peer.download_blob(blob_id).await
-                })
-                .await
-            },
-            timeout,
-        )
-        .await
-        .map_err(|errors| {
-            for (validator, error) in &errors {
-                warn!(
-                    %validator,
-                    %blob_id,
+    ) -> Result<Option<Vec<Blob>>, NodeError> {
+        if blob_ids.is_empty() {
+            return Ok(Some(Vec::new()));
+        }
+
+        // Streaming endpoint aborts if the server echoes a blob ID twice; deduplicate upfront.
+        let blob_ids: Vec<BlobId> = blob_ids
+            .iter()
+            .copied()
+            .collect::<BTreeSet<_>>()
+            .into_iter()
+            .collect();
+
+        let mut shuffled_peers = peers.to_vec();
+        shuffled_peers.shuffle(&mut rand::thread_rng());
+
+        let mut collected: HashMap<BlobId, Blob> = HashMap::with_capacity(blob_ids.len());
+        let mut last_error: Option<NodeError> = None;
+
+        for peer in shuffled_peers {
+            let pending: Vec<BlobId> = blob_ids
+                .iter()
+                .copied()
+                .filter(|id| !collected.contains_key(id))
+                .collect();
+            if pending.is_empty() {
+                break;
+            }
+
+            self.add_peer(peer.clone()).await;
+            let address = peer.address();
+            let (yielded, attempt_error) =
+                self.stream_blobs_from_peer(peer, &pending, timeout).await;
+            for blob in yielded {
+                collected.insert(blob.id(), blob);
+            }
+            if let Some(error) = attempt_error {
+                tracing::debug!(
+                    %address,
+                    pending = pending.len(),
                     %error,
-                    "failed to download blob from validator",
+                    "streaming download_blobs attempt failed",
                 );
+                last_error = Some(error);
             }
-            errors
-                .into_iter()
-                .last()
-                .map_or(NodeError::NoValidators, |(_, error)| error)
-        })
+        }
+
+        if collected.len() == blob_ids.len() {
+            let blobs = blob_ids
+                .iter()
+                .map(|id| collected.remove(id).expect("checked above"))
+                .collect();
+            return Ok(Some(blobs));
+        }
+        if collected.is_empty() {
+            if let Some(error) = last_error {
+                return Err(error);
+            }
+        }
+        Ok(None)
     }
 
-    /// Downloads the blobs with the given IDs. This is done in one concurrent task per blob.
-    /// Uses intelligent peer selection based on scores and load balancing.
-    /// Returns `None` if it couldn't find all blobs.
-    #[instrument(level = "trace", skip_all)]
-    pub async fn download_blobs(
+    /// Streams blobs from a single peer, accepting any blob whose ID is in the
+    /// pending set. The server is allowed to skip blobs it does not have, so
+    /// the caller cannot assume a strict 1:1 correspondence with the request.
+    /// Returns whatever blobs the peer yielded and the error, if any, that
+    /// ended the stream.
+    ///
+    /// `timeout` bounds the wait for the stream to open and for each subsequent
+    /// item — it detects a stalled peer without cancelling a transfer that is
+    /// still making progress. Updates the peer's EMA metrics; an attempt counts
+    /// as failed if it errored, stalled, or the peer sent an unexpected blob.
+    async fn stream_blobs_from_peer(
         &self,
-        peers: &[RemoteNode<Env::ValidatorNode>],
-        blob_ids: &[BlobId],
+        peer: RemoteNode<Env::ValidatorNode>,
+        pending: &[BlobId],
         timeout: Duration,
-    ) -> Result<Option<Vec<Blob>>, NodeError> {
-        let mut stream = blob_ids
-            .iter()
-            .map(|blob_id| self.download_blob(peers, *blob_id, timeout))
-            .collect::<FuturesUnordered<_>>();
+    ) -> (Vec<Blob>, Option<NodeError>) {
+        let start_time = Instant::now();
+        let public_key = peer.public_key;
+        let address = peer.address();
+        let mut yielded = Vec::<Blob>::with_capacity(pending.len());
+        let mut pending_set = pending.iter().copied().collect::<BTreeSet<BlobId>>();
+        let mut peer_misbehaved = false;
+        let attempt = async {
+            let stalled = || NodeError::ClientIoError {
+                error: format!("download_blobs made no progress for {timeout:?}"),
+            };
+            let mut stream =
+                linera_base::time::timer::timeout(timeout, peer.download_blobs(pending.to_vec()))
+                    .await
+                    .map_err(|_| stalled())??;
+            loop {
+                let item = linera_base::time::timer::timeout(timeout, stream.next())
+                    .await
+                    .map_err(|_| stalled())?;
+                let Some(item) = item else { break };
+                let content = item?;
+                let blob = Blob::new(content);
+                let blob_id = blob.id();
+                if !pending_set.remove(&blob_id) {
+                    tracing::debug!(
+                        %address,
+                        %blob_id,
+                        "validator sent an unexpected or duplicate blob; aborting stream",
+                    );
+                    peer_misbehaved = true;
+                    break;
+                }
+                yielded.push(blob);
+            }
+            Ok::<(), NodeError>(())
+        };
 
-        let mut blobs = Vec::new();
-        while let Some(maybe_blob) = stream.next().await {
-            blobs.push(maybe_blob?);
+        let result = attempt.await;
+
+        // Update per-peer EMA metrics; mirror what `track_request` does, but allow
+        // returning partial results separately from the error.
+        #[expect(
+            clippy::cast_possible_truncation,
+            reason = "elapsed millis fits in u64 for any realistic measurement window"
+        )]
+        let response_time_ms = start_time.elapsed().as_millis() as u64;
+        let is_success = result.is_ok() && !peer_misbehaved;
+        {
+            let mut nodes_guard = self.nodes.write().await;
+            if let Some(info) = nodes_guard.get_mut(&public_key) {
+                info.update_metrics(is_success, response_time_ms);
+            }
         }
-        Ok(blobs.into_iter().collect::<Option<Vec<_>>>())
+        #[cfg(with_metrics)]
+        {
+            let validator_name = public_key.to_string();
+            metrics::VALIDATOR_RESPONSE_TIME
+                .with_label_values(&[&validator_name])
+                .observe(response_time_ms as f64);
+            metrics::VALIDATOR_REQUEST_TOTAL
+                .with_label_values(&[&validator_name])
+                .inc();
+            if is_success {
+                metrics::VALIDATOR_REQUEST_SUCCESS
+                    .with_label_values(&[&validator_name])
+                    .inc();
+            }
+        }
+
+        (yielded, result.err())
     }
 
     pub async fn download_certificates(
@@ -1295,7 +1405,7 @@ mod tests {
                 Duration::from_millis(STAGGERED_DELAY_MS),
             ));
 
-        let key = RequestKey::Blob(BlobId::new(
+        let key = RequestKey::CertificateForBlob(BlobId::new(
             CryptoHash::test_hash("test_blob"),
             BlobType::Data,
         ));
@@ -1318,7 +1428,7 @@ mod tests {
                         if let Some(receiver) = rx.lock().await.take() {
                             receiver.await.unwrap();
                         }
-                        Ok(None) // Return Option<Blob>
+                        Ok(Vec::<ConfirmedBlockCertificate>::new())
                     }
                 })
                 .await
@@ -1337,7 +1447,7 @@ mod tests {
                 tokio::spawn(async move {
                     manager_clone
                         .with_peer(key_clone, node, |_peer| async move {
-                            Ok(None) // Return Option<Blob>
+                            Ok(Vec::<ConfirmedBlockCertificate>::new())
                         })
                         .await
                 })