Improve syncing on incoming message (#4373)

## Motivation

When inspecting the performance of our web demos (namely
hosted-fungible) we noticed that an incoming transfer triggers 7 network
requests while only 3 (last ones) are about accepting an incoming
transfer (proposing a block that includes the incoming message).

After closer research, it was discovered that we make:
- `HandleChainInfoQuery` - first request to learn about the sender chain
tip
- `HandleChainInfoQuery` - second time we request via
`fetch_sent_certificate_hashes` with a specific `BlockRange` (after
checking the `info.requested_received_log`)
- `DownloadCertificates` - to download the certififcates (by hash)
- `DownloadCertificates` - unnecessary query with empty payload (this
was a bug).
- and finally three requests to propose the block.

## Proposal

This PR proposes an improvement of the situation – decreasing # of
requests from 7 to 5:
- `HandleChainInfoQuery` – to learn about the missing certificates for
block heights that we might be missing
- `DownloadCertificatesByHeights` – new endpoint that we use to download
certificates at specific block heights
- and finally three requests to propose a block.

I don't think we can get it any lower with the way `ChainClient` is
currently structured.

This also improves all calls to `query_certificates_from` which are now
making 1 network query instead of two.

## Test Plan

CI

## Release Plan

- Nothing to do / These changes follow the usual release cycle. **OR**
- These changes should be backported to the latest `devnet` branch, then
    - be released in a new SDK,
    - be released in a validator hotfix.

## Links

- [reviewer
checklist](https://github.com/linera-io/linera-protocol/blob/main/CONTRIBUTING.md#reviewer-checklist)

Author: deuszx

linera-core/src/chain_worker/state.rs

@@ -1294,18 +1294,11 @@ where
 
             info.requested_pending_message_bundles = messages;
         }
-        if let Some(range) = query.request_sent_certificate_hashes_in_range {
-            let start: usize = range.start.try_into()?;
-            let end = match range.limit {
-                None => chain.confirmed_log.count(),
-                Some(limit) => start
-                    .checked_add(usize::try_from(limit).map_err(|_| ArithmeticError::Overflow)?)
-                    .ok_or(ArithmeticError::Overflow)?
-                    .min(chain.confirmed_log.count()),
-            };
-            let keys = chain.confirmed_log.read(start..end).await?;
-            info.requested_sent_certificate_hashes = keys;
+        let mut hashes = Vec::new();
+        for height in query.request_sent_certificate_hashes_by_heights {
+            hashes.extend(chain.block_hashes(height..=height).await?);
         }
+        info.requested_sent_certificate_hashes = hashes;
         if let Some(start) = query.request_received_log_excluding_first_n {
             let start = usize::try_from(start).map_err(|_| ArithmeticError::Overflow)?;
             info.requested_received_log = chain.received_log.read(start..).await?;

linera-core/src/client/mod.rs

@@ -69,9 +69,7 @@ use tokio_stream::wrappers::UnboundedReceiverStream;
 use tracing::{debug, error, info, instrument, warn, Instrument as _};
 
 use crate::{
-    data_types::{
-        BlockHeightRange, ChainInfo, ChainInfoQuery, ChainInfoResponse, ClientOutcome, RoundTimeout,
-    },
+    data_types::{ChainInfo, ChainInfoQuery, ChainInfoResponse, ClientOutcome, RoundTimeout},
     environment::Environment,
     local_node::{LocalChainInfoExt as _, LocalNodeClient, LocalNodeError},
     node::{
@@ -786,7 +784,7 @@ impl<Env: Environment> Client<Env> {
         // put all their sent messages into the inbox.
         let mut other_sender_chains = Vec::new();
 
-        let certificate_hashes = future::try_join_all(remote_heights.into_iter().filter_map(
+        let certificates = future::try_join_all(remote_heights.into_iter().filter_map(
             |(sender_chain_id, remote_heights)| {
                 let local_next = *local_next_heights.get(&sender_chain_id)?;
                 if let Ok(height) = local_next.try_sub_one() {
@@ -803,20 +801,11 @@ impl<Env: Environment> Client<Env> {
                     other_sender_chains.push(sender_chain_id);
                     return None;
                 };
-                let height0 = *remote_heights.first()?;
-                let height1 = *remote_heights.last()?;
-
-                // Find the hashes of the blocks we need.
-                let range = BlockHeightRange::multi(height0, height1.0 + 1 - height0.0);
                 Some(async move {
-                    let hashes = remote_node
-                        .fetch_sent_certificate_hashes(sender_chain_id, range)
+                    let certificates = remote_node
+                        .download_certificates_by_heights(sender_chain_id, remote_heights)
                         .await?;
-                    Ok::<_, ChainClientError>(
-                        remote_heights
-                            .into_iter()
-                            .filter_map(move |h| hashes.get((h.0 - height0.0) as usize).copied()),
-                    )
+                    Ok::<Vec<_>, ChainClientError>(certificates)
                 })
             },
         ))
@@ -825,41 +814,10 @@ impl<Env: Environment> Client<Env> {
         .flatten()
         .collect::<Vec<_>>();
 
-        let local_certificates = self
-            .storage_client()
-            .read_certificates(certificate_hashes.clone())
-            .await?
-            .into_iter()
-            .flatten()
-            .collect::<Vec<_>>();
-        let local_certificate_hashes = local_certificates
-            .iter()
-            .map(|cert| cert.hash())
-            .collect::<HashSet<_>>();
-
-        // Download the block certificates.
-        let remote_certificates = remote_node
-            .download_certificates(
-                certificate_hashes
-                    .into_iter()
-                    .filter(|hash| !local_certificate_hashes.contains(hash))
-                    .collect(),
-            )
-            .await?;
         let mut certificates_by_height_by_chain = BTreeMap::new();
 
-        for confirmed_block_certificate in local_certificates {
-            let block_header = &confirmed_block_certificate.inner().block().header;
-            let sender_chain_id = block_header.chain_id;
-            let height = block_header.height;
-            certificates_by_height_by_chain
-                .entry(sender_chain_id)
-                .or_insert_with(BTreeMap::new)
-                .insert(height, confirmed_block_certificate);
-        }
-
         // Check the signatures and keep only the ones that are valid.
-        for confirmed_block_certificate in remote_certificates {
+        for confirmed_block_certificate in certificates {
             let block_header = &confirmed_block_certificate.inner().block().header;
             let sender_chain_id = block_header.chain_id;
             let height = block_header.height;

linera-core/src/data_types.rs

@@ -76,6 +76,8 @@ pub struct ChainInfoQuery {
     #[debug(skip_if = Not::not)]
     pub request_pending_message_bundles: bool,
     /// Query a range of certificate hashes sent from the chain.
+    //  dev: this field is left and unused to maintain backwards compatibility
+    //  after hotfixing Testnet Conway.
     #[debug(skip_if = Option::is_none)]
     pub request_sent_certificate_hashes_in_range: Option<BlockHeightRange>,
     /// Query new certificate sender chain IDs and block heights received from the chain.
@@ -132,8 +134,8 @@ impl ChainInfoQuery {
         self
     }
 
-    pub fn with_sent_certificate_hashes_in_range(mut self, range: BlockHeightRange) -> Self {
-        self.request_sent_certificate_hashes_in_range = Some(range);
+    pub fn with_sent_certificate_hashes_by_heights(mut self, heights: Vec<BlockHeight>) -> Self {
+        self.request_sent_certificate_hashes_by_heights = heights;
         self
     }
 
@@ -314,6 +316,13 @@ impl ChainInfoResponse {
 
 impl BcsSignable<'_> for ChainInfo {}
 
+/// Request for downloading certificates by heights.
+#[derive(Debug, Clone)]
+pub struct CertificatesByHeightRequest {
+    pub chain_id: ChainId,
+    pub heights: Vec<BlockHeight>,
+}
+
 /// The outcome of trying to commit a list of operations to the chain.
 #[derive(Debug)]
 pub enum ClientOutcome<T> {

linera-core/src/node.rs

@@ -145,6 +145,13 @@ pub trait ValidatorNode {
         hashes: Vec<CryptoHash>,
     ) -> Result<Vec<ConfirmedBlockCertificate>, NodeError>;
 
+    /// Requests a batch of certificates from a specific chain by heights.
+    async fn download_certificates_by_heights(
+        &self,
+        chain_id: ChainId,
+        heights: Vec<BlockHeight>,
+    ) -> Result<Vec<ConfirmedBlockCertificate>, NodeError>;
+
     /// Returns the hash of the `Certificate` that last used a blob.
     async fn blob_last_used_by(&self, blob_id: BlobId) -> Result<CryptoHash, NodeError>;
 

linera-core/src/remote_node.rs

@@ -1,12 +1,15 @@
 // Copyright (c) Zefchain Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-use std::{collections::HashSet, time::Duration};
+use std::{
+    collections::{HashSet, VecDeque},
+    time::Duration,
+};
 
 use custom_debug_derive::Debug;
 use futures::{future::try_join_all, stream::FuturesUnordered, StreamExt};
 use linera_base::{
-    crypto::{CryptoHash, ValidatorPublicKey},
+    crypto::ValidatorPublicKey,
     data_types::{Blob, BlockHeight},
     ensure,
     identifiers::{BlobId, ChainId},
@@ -22,7 +25,7 @@ use rand::seq::SliceRandom as _;
 use tracing::{instrument, warn};
 
 use crate::{
-    data_types::{BlockHeightRange, ChainInfo, ChainInfoQuery, ChainInfoResponse},
+    data_types::{ChainInfo, ChainInfoQuery, ChainInfoResponse},
     node::{CrossChainMessageDelivery, NodeError, ValidatorNode},
 };
 
@@ -169,25 +172,11 @@ impl<N: ValidatorNode> RemoteNode<N> {
         limit: u64,
     ) -> Result<Vec<ConfirmedBlockCertificate>, NodeError> {
         tracing::debug!(name = ?self.public_key, ?chain_id, ?start, ?limit, "Querying certificates");
-        let range = BlockHeightRange {
-            start,
-            limit: Some(limit),
-        };
-        let query = ChainInfoQuery::new(chain_id).with_sent_certificate_hashes_in_range(range);
-        let info = self.handle_chain_info_query(query).await?;
-        self.node
-            .download_certificates(info.requested_sent_certificate_hashes)
-            .await?
-            .into_iter()
-            .map(|c| {
-                ensure!(
-                    c.inner().chain_id() == chain_id,
-                    NodeError::UnexpectedCertificateValue
-                );
-                ConfirmedBlockCertificate::try_from(c)
-                    .map_err(|_| NodeError::InvalidChainInfoResponse)
-            })
-            .collect()
+        let heights = (start.0..start.0 + limit)
+            .map(BlockHeight)
+            .collect::<Vec<_>>();
+        self.download_certificates_by_heights(chain_id, heights)
+            .await
     }
 
     #[instrument(level = "trace")]
@@ -246,51 +235,46 @@ impl<N: ValidatorNode> RemoteNode<N> {
         }
     }
 
-    /// Returns the list of certificate hashes on the given chain in the given range of heights.
-    /// Returns an error if the number of hashes does not match the size of the range.
+    /// Downloads a list of certificates from the given chain.
     #[instrument(level = "trace")]
-    pub(crate) async fn fetch_sent_certificate_hashes(
+    pub async fn download_certificates_by_heights(
         &self,
         chain_id: ChainId,
-        range: BlockHeightRange,
-    ) -> Result<Vec<CryptoHash>, NodeError> {
-        let query =
-            ChainInfoQuery::new(chain_id).with_sent_certificate_hashes_in_range(range.clone());
-        let response = self.handle_chain_info_query(query).await?;
-        let hashes = response.requested_sent_certificate_hashes;
+        heights: Vec<BlockHeight>,
+    ) -> Result<Vec<ConfirmedBlockCertificate>, NodeError> {
+        let mut expected_heights = VecDeque::from(heights.clone());
+        let certificates = self
+            .node
+            .download_certificates_by_heights(chain_id, heights)
+            .await?;
 
-        if range
-            .limit
-            .is_some_and(|limit| hashes.len() as u64 != limit)
-        {
-            warn!(
-                ?range,
-                received_num = hashes.len(),
-                "Validator sent invalid number of certificate hashes."
-            );
-            return Err(NodeError::InvalidChainInfoResponse);
+        if certificates.len() > expected_heights.len() {
+            return Err(NodeError::TooManyCertificatesReturned {
+                chain_id,
+                remote_node: Box::new(self.public_key),
+            });
         }
-        Ok(hashes)
-    }
 
-    #[instrument(level = "trace")]
-    pub async fn download_certificates(
-        &self,
-        hashes: Vec<CryptoHash>,
-    ) -> Result<Vec<ConfirmedBlockCertificate>, NodeError> {
-        if hashes.is_empty() {
-            return Ok(Vec::new());
+        for certificate in &certificates {
+            ensure!(
+                certificate.inner().chain_id() == chain_id,
+                NodeError::UnexpectedCertificateValue
+            );
+            if let Some(expected_height) = expected_heights.pop_front() {
+                ensure!(
+                    expected_height == certificate.inner().height(),
+                    NodeError::UnexpectedCertificateValue
+                );
+            } else {
+                return Err(NodeError::UnexpectedCertificateValue);
+            }
         }
-        let certificates = self.node.download_certificates(hashes.clone()).await?;
-        let returned = certificates
-            .iter()
-            .map(ConfirmedBlockCertificate::hash)
-            .collect();
+
         ensure!(
-            returned == hashes,
-            NodeError::UnexpectedCertificates {
-                returned,
-                requested: hashes
+            expected_heights.is_empty(),
+            NodeError::MissingCertificatesByHeights {
+                chain_id,
+                heights: expected_heights.into_iter().collect(),
             }
         );
         Ok(certificates)

linera-core/src/unit_tests/test_utils.rs

@@ -238,6 +238,17 @@ where
         .await
     }
 
+    async fn download_certificates_by_heights(
+        &self,
+        chain_id: ChainId,
+        heights: Vec<BlockHeight>,
+    ) -> Result<Vec<ConfirmedBlockCertificate>, NodeError> {
+        self.spawn_and_receive(move |validator, sender| {
+            validator.do_download_certificates_by_heights(chain_id, heights, sender)
+        })
+        .await
+    }
+
     async fn blob_last_used_by(&self, blob_id: BlobId) -> Result<CryptoHash, NodeError> {
         self.spawn_and_receive(move |validator, sender| {
             validator.do_blob_last_used_by(blob_id, sender)
@@ -596,6 +607,51 @@ where
         sender.send(certificates)
     }
 
+    async fn do_download_certificates_by_heights(
+        self,
+        chain_id: ChainId,
+        heights: Vec<BlockHeight>,
+        sender: oneshot::Sender<Result<Vec<ConfirmedBlockCertificate>, NodeError>>,
+    ) -> Result<(), Result<Vec<ConfirmedBlockCertificate>, NodeError>> {
+        // First, use do_handle_chain_info_query to get the certificate hashes
+        let (query_sender, query_receiver) = oneshot::channel();
+        let query = ChainInfoQuery::new(chain_id).with_sent_certificate_hashes_by_heights(heights);
+
+        let self_clone = self.clone();
+        self.do_handle_chain_info_query(query, query_sender)
+            .await
+            .expect("Failed to handle chain info query");
+
+        // Get the response from the chain info query
+        let chain_info_response = query_receiver.await.map_err(|_| {
+            Err(NodeError::ClientIoError {
+                error: "Failed to receive chain info response".to_string(),
+            })
+        })?;
+
+        let hashes = match chain_info_response {
+            Ok(response) => response.info.requested_sent_certificate_hashes,
+            Err(e) => {
+                return sender.send(Err(e));
+            }
+        };
+
+        // Now use do_download_certificates to get the actual certificates
+        let (cert_sender, cert_receiver) = oneshot::channel();
+        self_clone
+            .do_download_certificates(hashes, cert_sender)
+            .await?;
+
+        // Forward the result to the original sender
+        let result = cert_receiver.await.map_err(|_| {
+            Err(NodeError::ClientIoError {
+                error: "Failed to receive certificates".to_string(),
+            })
+        })?;
+
+        sender.send(result)
+    }
+
     async fn do_blob_last_used_by(
         self,
         blob_id: BlobId,
@@ -1020,11 +1076,8 @@ where
         block_height: BlockHeight,
         target_count: usize,
     ) -> Option<ConfirmedBlockCertificate> {
-        let query =
-            ChainInfoQuery::new(chain_id).with_sent_certificate_hashes_in_range(BlockHeightRange {
-                start: block_height,
-                limit: Some(1),
-            });
+        let query = ChainInfoQuery::new(chain_id)
+            .with_sent_certificate_hashes_by_heights(vec![block_height]);
         let mut count = 0;
         let mut certificate = None;
         for validator in self.validator_clients.clone() {