Add new DownloadRawCertificatesByHeight endpoint and use it. (#4478)

## Motivation

During our performance tests it was found that proxies' bottleneck is
(de)serialization of certificates when responding to
`DownloadCertificatesByHeights` request.

## Proposal

It is a waste of CPU cycles to deserialize data (after loading from the
DB) only to serialize it for transporting over gRPC protocol.

This PR:
- adds a new method on the storage `read_certificates_raw` which returns
raw bytes of the requested certificates
- adds a new endpoint to `rpc.proto/ValidatorNode` –
`DownloadRawCertificatesByHeight` that responds with (bcs) bytes of the
requested certificates
- updates proxy code to serve the new endpoint
- the gRPC _client_ does NOT get a new method, instead the old
`download_certificates_by_height` is modified to use the new endpoint.
**This is safe:** b/c we will release this as a new SDK version meaning
old clients still use old code paths while next version uses new.

## Test Plan

CI

## Release Plan

- These changes should be backported to the latest `devnet` branch, then
    - be released in a new SDK,
    - be released in a validator hotfix.
- These changes should be backported to the latest `testnet` branch,
then
    - be released in a new SDK,
    - be released in a validator hotfix.

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

Author: deuszx

Cargo.lock

@@ -44,6 +44,7 @@ web-default = ["web"]
 [dependencies]
 anyhow.workspace = true
 async-trait.workspace = true
+bcs.workspace = true
 bincode.workspace = true
 bytes.workspace = true
 cfg-if.workspace = true

linera-rpc/Cargo.toml

@@ -86,9 +86,12 @@ service ValidatorNode {
   // Download a batch of certificates.
   rpc DownloadCertificates(CertificatesBatchRequest) returns (CertificatesBatchResponse);
 
-  /// Download a batch of certificates by height range.
+  /// Download a batch of certificates by heights.
   rpc DownloadCertificatesByHeights(DownloadCertificatesByHeightsRequest) returns (CertificatesBatchResponse);
 
+  // Download a batch of certificates, in serialized form, by their heights.
+  rpc DownloadRawCertificatesByHeights(DownloadCertificatesByHeightsRequest) returns (RawCertificatesBatch);
+
   // Return the hash of the `Certificate` that last used a blob.
   rpc BlobLastUsedBy(BlobId) returns (CryptoHash);
 
@@ -100,6 +103,19 @@ service ValidatorNode {
   rpc MissingBlobIds(BlobIds) returns (BlobIds);
 }
 
+// Batch of raw certificates.
+message RawCertificatesBatch {
+  repeated RawCertificate certificates = 1;
+}
+
+// A confirmed block certificate in a serialized form.
+message RawCertificate {
+  // BCS-serialized bytes of lite certificate part.
+  bytes lite_certificate = 1;
+  // BCS-serialized bytes of confirmed block part.
+  bytes confirmed_block = 2;
+}
+
 // A request for downloading certificates by block heights.
 message DownloadCertificatesByHeightsRequest {
   ChainId chain_id = 1;

linera-rpc/proto/rpc.proto

@@ -14,8 +14,8 @@ use linera_base::{
 use linera_chain::{
     data_types::{self},
     types::{
-        self, Certificate, ConfirmedBlock, ConfirmedBlockCertificate, GenericCertificate, Timeout,
-        ValidatedBlock,
+        self, Certificate, ConfirmedBlock, ConfirmedBlockCertificate, GenericCertificate,
+        LiteCertificate, Timeout, ValidatedBlock,
     },
 };
 use linera_core::{
@@ -32,8 +32,8 @@ use super::{
     transport, GRPC_MAX_MESSAGE_SIZE,
 };
 use crate::{
-    HandleConfirmedCertificateRequest, HandleLiteCertRequest, HandleTimeoutCertificateRequest,
-    HandleValidatedCertificateRequest,
+    grpc::api::RawCertificate, HandleConfirmedCertificateRequest, HandleLiteCertRequest,
+    HandleTimeoutCertificateRequest, HandleValidatedCertificateRequest,
 };
 
 #[derive(Clone)]
@@ -447,15 +447,26 @@ impl ValidatorNode for GrpcClient {
                 chain_id,
                 heights: missing.clone(),
             };
-            let mut received: Vec<ConfirmedBlockCertificate> = Vec::<Certificate>::try_from(
-                client_delegate!(self, download_certificates_by_heights, request)?,
-            )?
-            .into_iter()
-            .map(|cert| {
-                ConfirmedBlockCertificate::try_from(cert)
-                    .map_err(|_| NodeError::UnexpectedCertificateValue)
-            })
-            .collect::<Result<_, _>>()?;
+            let mut received: Vec<ConfirmedBlockCertificate> =
+                client_delegate!(self, download_raw_certificates_by_heights, request)?
+                    .certificates
+                    .into_iter()
+                    .map(
+                        |RawCertificate {
+                             lite_certificate,
+                             confirmed_block,
+                         }| {
+                            let cert = bcs::from_bytes::<LiteCertificate>(&lite_certificate)
+                                .map_err(|_| NodeError::UnexpectedCertificateValue)?;
+
+                            let block = bcs::from_bytes::<ConfirmedBlock>(&confirmed_block)
+                                .map_err(|_| NodeError::UnexpectedCertificateValue)?;
+
+                            cert.with_value(block)
+                                .ok_or(NodeError::UnexpectedCertificateValue)
+                        },
+                    )
+                    .collect::<Result<_, _>>()?;
 
             if received.is_empty() {
                 break;

linera-rpc/src/grpc/client.rs

@@ -353,6 +353,13 @@ impl ValidatorNode for DummyValidator {
         unimplemented!()
     }
 
+    async fn download_raw_certificates_by_heights(
+        &self,
+        _request: Request<linera_rpc::grpc::api::DownloadCertificatesByHeightsRequest>,
+    ) -> Result<Response<linera_rpc::grpc::api::RawCertificatesBatch>, Status> {
+        unimplemented!()
+    }
+
     async fn blob_last_used_by(
         &self,
         _request: Request<linera_rpc::grpc::api::BlobId>,

linera-service/src/exporter/test_utils.rs

@@ -35,7 +35,8 @@ use linera_rpc::{
             BlobContent, BlobId, BlobIds, BlockProposal, Certificate, CertificatesBatchRequest,
             CertificatesBatchResponse, ChainInfoResult, CryptoHash, HandlePendingBlobRequest,
             LiteCertificate, NetworkDescription, Notification, PendingBlobRequest,
-            PendingBlobResult, SubscriptionRequest, VersionInfo,
+            PendingBlobResult, RawCertificate, RawCertificatesBatch, SubscriptionRequest,
+            VersionInfo,
         },
         pool::GrpcConnectionPool,
         GrpcProtoConversionError, GrpcProxyable, GRPC_CHUNKED_MESSAGE_FILL_LIMIT,
@@ -675,6 +676,75 @@ where
             .await
     }
 
+    #[instrument(skip_all, err(Display))]
+    async fn download_raw_certificates_by_heights(
+        &self,
+        request: Request<api::DownloadCertificatesByHeightsRequest>,
+    ) -> Result<Response<api::RawCertificatesBatch>, Status> {
+        let original_request: CertificatesByHeightRequest = request.into_inner().try_into()?;
+        let chain_info_request = ChainInfoQuery::new(original_request.chain_id)
+            .with_sent_certificate_hashes_by_heights(original_request.heights);
+        // Use handle_chain_info_query to get the certificate hashes
+        let chain_info_response = self
+            .handle_chain_info_query(Request::new(chain_info_request.try_into()?))
+            .await?;
+        // Extract the ChainInfoResult from the response
+        let chain_info_result = chain_info_response.into_inner();
+        // Extract the certificate hashes from the ChainInfo
+        let hashes = match chain_info_result.inner {
+            Some(api::chain_info_result::Inner::ChainInfoResponse(response)) => {
+                let chain_info: ChainInfo =
+                    bincode::deserialize(&response.chain_info).map_err(|e| {
+                        Status::internal(format!("Failed to deserialize ChainInfo: {}", e))
+                    })?;
+                chain_info.requested_sent_certificate_hashes
+            }
+            Some(api::chain_info_result::Inner::Error(error)) => {
+                return Err(Status::internal(format!(
+                    "Chain info query failed: {:?}",
+                    error
+                )));
+            }
+            None => {
+                return Err(Status::internal("Empty chain info result"));
+            }
+        };
+
+        // Use 70% of the max message size as a buffer capacity.
+        // Leave 30% as overhead.
+        let mut grpc_message_limiter: GrpcMessageLimiter<linera_chain::types::Certificate> =
+            GrpcMessageLimiter::new(GRPC_CHUNKED_MESSAGE_FILL_LIMIT);
+
+        let mut returned_certificates = vec![];
+
+        'outer: for batch in hashes.chunks(100) {
+            let certificates: Vec<(Vec<u8>, Vec<u8>)> = self
+                .0
+                .storage
+                .read_certificates_raw(batch.to_vec())
+                .await
+                .map_err(Self::view_error_to_status)?
+                .into_iter()
+                .collect();
+            for (lite_cert_bytes, confirmed_block_bytes) in certificates {
+                if grpc_message_limiter
+                    .fits_raw(lite_cert_bytes.len() + confirmed_block_bytes.len())
+                {
+                    returned_certificates.push(RawCertificate {
+                        lite_certificate: lite_cert_bytes,
+                        confirmed_block: confirmed_block_bytes,
+                    });
+                } else {
+                    break 'outer;
+                }
+            }
+        }
+
+        Ok(Response::new(RawCertificatesBatch {
+            certificates: returned_certificates,
+        }))
+    }
+
     #[instrument(skip_all, err(level = Level::WARN))]
     async fn blob_last_used_by(
         &self,
@@ -764,11 +834,18 @@ impl<T> GrpcMessageLimiter<T> {
         U: TryFrom<T, Error = GrpcProtoConversionError> + Message,
     {
         let required = U::try_from(el).map(|proto| proto.encoded_len())?;
-        if required > self.remaining {
-            return Ok(false);
+        Ok(self.fits_raw(required))
+    }
+
+    /// Adds the given number of bytes to the remaining capacity.
+    ///
+    /// Returns whether we managed to fit the element.
+    fn fits_raw(&mut self, bytes_len: usize) -> bool {
+        if self.remaining < bytes_len {
+            return false;
         }
-        self.remaining -= required;
-        Ok(true)
+        self.remaining = self.remaining.saturating_sub(bytes_len);
+        true
     }
 }
 

linera-service/src/proxy/grpc.rs

@@ -824,6 +824,37 @@ where
         Ok(certificates)
     }
 
+    /// Reads certificates by hashes.
+    ///
+    /// Returns a vector of tuples where the first element is a lite certificate
+    /// and the second element is confirmed block.
+    ///
+    /// It does not check if all hashes all returned.
+    async fn read_certificates_raw<I: IntoIterator<Item = CryptoHash> + Send>(
+        &self,
+        hashes: I,
+    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>, ViewError> {
+        let hashes = hashes.into_iter().collect::<Vec<_>>();
+        if hashes.is_empty() {
+            return Ok(Vec::new());
+        }
+        let keys = Self::get_keys_for_certificates(&hashes)?;
+        let store = self.database.open_shared(&[])?;
+        let values = store.read_multi_values_bytes(keys).await?;
+        #[cfg(with_metrics)]
+        metrics::READ_CERTIFICATES_COUNTER
+            .with_label_values(&[])
+            .inc_by(hashes.len() as u64);
+        Ok(values
+            .chunks_exact(2)
+            .filter_map(|chunk| {
+                let lite_cert_bytes = chunk[0].as_ref()?;
+                let confirmed_block_bytes = chunk[1].as_ref()?;
+                Some((lite_cert_bytes.clone(), confirmed_block_bytes.clone()))
+            })
+            .collect())
+    }
+
     async fn read_event(&self, event_id: EventId) -> Result<Option<Vec<u8>>, ViewError> {
         let store = self.database.open_shared(&[])?;
         let event_key = bcs::to_bytes(&BaseKey::Event(event_id.clone()))?;

linera-storage/src/db_storage.rs

@@ -140,6 +140,17 @@ pub trait Storage: Sized {
         hashes: I,
     ) -> Result<Vec<Option<ConfirmedBlockCertificate>>, ViewError>;
 
+    /// Reads certificates by hashes.
+    ///
+    /// Returns a vector of tuples where the first element is a lite certificate
+    /// and the second element is confirmed block.
+    ///
+    /// It does not check if all hashes all returned.
+    async fn read_certificates_raw<I: IntoIterator<Item = CryptoHash> + Send>(
+        &self,
+        hashes: I,
+    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>, ViewError>;
+
     /// Reads the event with the given ID.
     async fn read_event(&self, id: EventId) -> Result<Option<Vec<u8>>, ViewError>;