From 8b526c017fa779821436e473bb17c73b4a1aa8cf Mon Sep 17 00:00:00 2001
From: Andreas Fackler <andreas.fackler@linera.io>
Date: Mon, 16 Dec 2024 15:01:10 +0100
Subject: [PATCH] Download blobs from the node that gave us the certificate.

---
 linera-core/src/client/mod.rs | 38 ++++++++++++++++++++++++++---------
 1 file changed, 29 insertions(+), 9 deletions(-)

diff --git a/linera-core/src/client/mod.rs b/linera-core/src/client/mod.rs
index 8ba37248112..21ec2cccb6d 100644
--- a/linera-core/src/client/mod.rs
+++ b/linera-core/src/client/mod.rs
@@ -1185,7 +1185,8 @@ where
         let block_chain_id = certificate.value().chain_id();
         let block_height = certificate.value().height();
 
-        self.receive_certificate_internal(certificate, mode).await?;
+        self.receive_certificate_internal(certificate, mode, None)
+            .await?;
 
         // Make sure a quorum of validators (according to our new local committee) are up-to-date
         // for data availability.
@@ -1207,6 +1208,7 @@ where
         &self,
         certificate: Certificate,
         mode: ReceiveCertificateMode,
+        nodes: Option<Vec<RemoteNode<P::Node>>>,
     ) -> Result<(), ChainClientError> {
         let CertificateValue::ConfirmedBlock { executed_block, .. } = certificate.value() else {
             return Err(ChainClientError::InternalError(
@@ -1227,9 +1229,13 @@ where
         if let ReceiveCertificateMode::NeedsCheck = mode {
             certificate.check(remote_committee)?;
         }
-        // Recover history from the network. We assume that the committee that signed the
-        // certificate is still active.
-        let nodes = self.make_nodes(remote_committee)?;
+        // Recover history from the network.
+        let nodes = if let Some(nodes) = nodes {
+            nodes
+        } else {
+            // We assume that the committee that signed the certificate is still active.
+            self.make_nodes(remote_committee)?
+        };
         self.client
             .download_certificates(&nodes, block.chain_id, block.height)
             .await?;
@@ -1443,7 +1449,10 @@ where
                 for certificate in certificates.into_values() {
                     let hash = certificate.hash();
                     let mode = ReceiveCertificateMode::AlreadyChecked;
-                    if let Err(e) = client.receive_certificate_internal(certificate, mode).await {
+                    if let Err(e) = client
+                        .receive_certificate_internal(certificate, mode, None)
+                        .await
+                    {
                         warn!("Received invalid certificate {hash}: {e}");
                     }
                 }
@@ -1807,15 +1816,26 @@ where
         for blob_id in blob_ids {
             let mut certificate_stream = validators
                 .iter()
-                .map(|remote_node| remote_node.download_certificate_for_blob(blob_id))
+                .map(|remote_node| async move {
+                    let cert = remote_node.download_certificate_for_blob(blob_id).await?;
+                    Ok::<_, NodeError>((remote_node.clone(), cert))
+                })
                 .collect::<FuturesUnordered<_>>();
             loop {
                 let Some(result) = certificate_stream.next().await else {
                     missing_blobs.push(blob_id);
                     break;
                 };
-                if let Ok(cert) = result {
-                    if self.receive_certificate(cert).await.is_ok() {
+                if let Ok((remote_node, cert)) = result {
+                    if self
+                        .receive_certificate_internal(
+                            cert,
+                            ReceiveCertificateMode::NeedsCheck,
+                            Some(vec![remote_node]),
+                        )
+                        .await
+                        .is_ok()
+                    {
                         break;
                     }
                 }
@@ -2578,7 +2598,7 @@ where
         &self,
         certificate: Certificate,
     ) -> Result<(), ChainClientError> {
-        self.receive_certificate_internal(certificate, ReceiveCertificateMode::NeedsCheck)
+        self.receive_certificate_internal(certificate, ReceiveCertificateMode::NeedsCheck, None)
             .await
     }