Upload blobs for confirmed certificates separately. (#3108)

## Motivation

Including blobs with the gRPC message that contains a block proposal or
certificate severely limits the total size of the blobs. (See #3048.)

## Proposal

As a first step, remove the blobs from the
`handle_confirmed_certificate` functions and messages.

Instead, when a validator sees a fully signed confirmed block it creates
the blob states in its local storage even if it doesn't have the blobs
yet. The client can then upload the blobs one by one, and the validator
will accept them. Finally, the client can retry sending the certificate.

We don't do this for block proposals or validated blocks yet: These will
need to be handled differently, because in these cases the blob has not
necessarily been successfully published yet, so we should _not_ create a
blob state. Instead, we will put these blobs into a temporary cache.

## Test Plan

The existing tests are now using the new flow for confirmed block
certificates.

## Release Plan

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

## Links

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

Author: afck

linera-core/src/chain_worker/actor.rs

@@ -114,7 +114,6 @@ where
     /// Process a confirmed block (a commit).
     ProcessConfirmedBlock {
         certificate: ConfirmedBlockCertificate,
-        blobs: Vec<Blob>,
         #[debug(with = "elide_option")]
         notify_when_messages_are_delivered: Option<oneshot::Sender<()>>,
         #[debug(skip)]
@@ -299,15 +298,13 @@ where
                     .is_ok(),
                 ChainWorkerRequest::ProcessConfirmedBlock {
                     certificate,
-                    blobs,
                     notify_when_messages_are_delivered,
                     callback,
                 } => callback
                     .send(
                         self.worker
                             .process_confirmed_block(
                                 certificate,
-                                &blobs,
                                 notify_when_messages_are_delivered,
                             )
                             .await,

linera-core/src/chain_worker/state/attempted_changes.rs

@@ -239,7 +239,6 @@ where
     pub(super) async fn process_confirmed_block(
         &mut self,
         certificate: ConfirmedBlockCertificate,
-        blobs: &[Blob],
         notify_when_messages_are_delivered: Option<oneshot::Sender<()>>,
     ) -> Result<(ChainInfoResponse, NetworkActions), WorkerError> {
         let executed_block = certificate.executed_block();
@@ -298,33 +297,33 @@ where
         );
 
         let required_blob_ids = executed_block.required_blob_ids();
-        // Verify that no unrelated blobs were provided.
-        self.state
-            .check_for_unneeded_blobs(&required_blob_ids, blobs)?;
-        let blobs = self.state.get_required_blobs(executed_block, blobs).await?;
-        let blobs = blobs.into_values().collect::<Vec<_>>();
-
-        let certificate_hash = certificate.hash();
+        let blobs_result = self
+            .state
+            .get_required_blobs(executed_block, &[])
+            .await
+            .map(|blobs| blobs.into_values().collect::<Vec<_>>());
 
-        self.state
-            .storage
-            .write_blobs_and_certificate(&blobs, &certificate)
-            .await?;
+        if let Ok(blobs) = &blobs_result {
+            self.state
+                .storage
+                .write_blobs_and_certificate(blobs, &certificate)
+                .await?;
+        }
 
         // Update the blob state with last used certificate hash.
         let blob_state = BlobState {
-            last_used_by: certificate_hash,
+            last_used_by: certificate.hash(),
             chain_id: block.chain_id,
             block_height,
             epoch: block.epoch,
         };
+        let overwrite = blobs_result.is_ok(); // Overwrite only if we wrote the certificate.
+        let blob_ids = required_blob_ids.into_iter().collect::<Vec<_>>();
         self.state
             .storage
-            .maybe_write_blob_states(
-                &required_blob_ids.into_iter().collect::<Vec<_>>(),
-                blob_state,
-            )
+            .maybe_write_blob_states(&blob_ids, blob_state, overwrite)
             .await?;
+        blobs_result?;
 
         // Execute the block and update inboxes.
         self.state.chain.remove_bundles_from_inboxes(block).await?;

linera-core/src/chain_worker/state/mod.rs

@@ -237,12 +237,11 @@ where
     pub(super) async fn process_confirmed_block(
         &mut self,
         certificate: ConfirmedBlockCertificate,
-        blobs: &[Blob],
         notify_when_messages_are_delivered: Option<oneshot::Sender<()>>,
     ) -> Result<(ChainInfoResponse, NetworkActions), WorkerError> {
         ChainWorkerStateWithAttemptedChanges::new(self)
             .await
-            .process_confirmed_block(certificate, blobs, notify_when_messages_are_delivered)
+            .process_confirmed_block(certificate, notify_when_messages_are_delivered)
             .await
     }
 

linera-core/src/client/mod.rs

@@ -45,8 +45,9 @@ use linera_chain::{
         MessageAction,
     },
     types::{
-        CertificateValue, ConfirmedBlock, ConfirmedBlockCertificate, GenericCertificate,
-        LiteCertificate, Timeout, TimeoutCertificate, ValidatedBlock, ValidatedBlockCertificate,
+        CertificateKind, CertificateValue, ConfirmedBlock, ConfirmedBlockCertificate,
+        GenericCertificate, LiteCertificate, Timeout, TimeoutCertificate, ValidatedBlock,
+        ValidatedBlockCertificate,
     },
     ChainError, ChainExecutionContext, ChainStateView,
 };
@@ -395,9 +396,24 @@ where
         certificate: GenericCertificate<T>,
         blobs: Vec<Blob>,
     ) -> Result<ChainInfoResponse, LocalNodeError> {
-        self.local_node
-            .handle_certificate(certificate, blobs, &self.notifier)
-            .await
+        if T::KIND == CertificateKind::Confirmed {
+            let result = self
+                .local_node
+                .handle_certificate(certificate.clone(), vec![], &self.notifier)
+                .await;
+            if let Err(LocalNodeError::BlobsNotFound(_)) = &result {
+                self.local_node.store_blobs(&blobs).await?;
+                return self
+                    .local_node
+                    .handle_certificate(certificate, vec![], &self.notifier)
+                    .await;
+            }
+            result
+        } else {
+            self.local_node
+                .handle_certificate(certificate, blobs, &self.notifier)
+                .await
+        }
     }
 }
 
@@ -1751,7 +1767,7 @@ where
         if let Some(cert) = info.manager.requested_locked {
             let hash = cert.hash();
             let blobs = info.manager.locked_blobs.clone();
-            if let Err(err) = self.client.handle_certificate(*cert.clone(), blobs).await {
+            if let Err(err) = self.process_certificate(*cert.clone(), blobs).await {
                 warn!(
                     "Skipping certificate {hash} from validator {}: {err}",
                     remote_node.name

linera-core/src/local_node.rs

@@ -210,6 +210,13 @@ where
             .collect())
     }
 
+    /// Writes the given blobs to storage if there is an appropriate blob state.
+    pub async fn store_blobs(&self, blobs: &[Blob]) -> Result<(), LocalNodeError> {
+        let storage = self.storage_client();
+        storage.maybe_write_blobs(blobs).await?;
+        Ok(())
+    }
+
     /// Returns a read-only view of the [`ChainStateView`] of a chain referenced by its
     /// [`ChainId`].
     ///

linera-core/src/node.rs

@@ -71,7 +71,6 @@ pub trait ValidatorNode {
     async fn handle_confirmed_certificate(
         &self,
         certificate: GenericCertificate<ConfirmedBlock>,
-        blobs: Vec<Blob>,
         delivery: CrossChainMessageDelivery,
     ) -> Result<ChainInfoResponse, NodeError>;
 
@@ -103,6 +102,10 @@ pub trait ValidatorNode {
     /// Subscribes to receiving notifications for a collection of chains.
     async fn subscribe(&self, chains: Vec<ChainId>) -> Result<Self::NotificationStream, NodeError>;
 
+    // Uploads a blob content. Returns an error if the validator has not seen a
+    // certificate using this blob.
+    async fn upload_blob(&self, content: BlobContent) -> Result<BlobId, NodeError>;
+
     async fn download_blob_content(&self, blob_id: BlobId) -> Result<BlobContent, NodeError>;
 
     async fn download_certificate(

linera-core/src/remote_node.rs

@@ -4,7 +4,7 @@
 use std::collections::HashSet;
 
 use custom_debug_derive::Debug;
-use futures::{stream::FuturesUnordered, StreamExt};
+use futures::{future::try_join_all, stream::FuturesUnordered, StreamExt};
 use linera_base::{
     crypto::CryptoHash,
     data_types::{Blob, BlockHeight},
@@ -68,13 +68,12 @@ impl<N: ValidatorNode> RemoteNode<N> {
     pub(crate) async fn handle_confirmed_certificate(
         &self,
         certificate: ConfirmedBlockCertificate,
-        blobs: Vec<Blob>,
         delivery: CrossChainMessageDelivery,
     ) -> Result<Box<ChainInfo>, NodeError> {
         let chain_id = certificate.inner().chain_id();
         let response = self
             .node
-            .handle_confirmed_certificate(certificate, blobs, delivery)
+            .handle_confirmed_certificate(certificate, delivery)
             .await?;
         self.check_and_return_info(response, chain_id)
     }
@@ -148,7 +147,7 @@ impl<N: ValidatorNode> RemoteNode<N> {
                 _ => return result,
             }
         }
-        self.handle_confirmed_certificate(certificate.clone(), vec![], delivery)
+        self.handle_confirmed_certificate(certificate.clone(), delivery)
             .await
     }
 
@@ -217,6 +216,16 @@ impl<N: ValidatorNode> RemoteNode<N> {
         Ok(certificate)
     }
 
+    /// Uploads the blobs to the validator.
+    #[instrument(level = "trace")]
+    pub(crate) async fn upload_blobs(&self, blobs: Vec<Blob>) -> Result<(), NodeError> {
+        let tasks = blobs
+            .into_iter()
+            .map(|blob| self.node.upload_blob(blob.into()));
+        try_join_all(tasks).await?;
+        Ok(())
+    }
+
     /// Tries to download the given blobs from this node. Returns `None` if not all could be found.
     #[instrument(level = "trace")]
     pub(crate) async fn try_download_blobs(&self, blob_ids: &[BlobId]) -> Option<Vec<Blob>> {

linera-core/src/unit_tests/test_utils.rs

@@ -148,11 +148,10 @@ where
     async fn handle_confirmed_certificate(
         &self,
         certificate: GenericCertificate<ConfirmedBlock>,
-        blobs: Vec<Blob>,
         _delivery: CrossChainMessageDelivery,
     ) -> Result<ChainInfoResponse, NodeError> {
         self.spawn_and_receive(move |validator, sender| {
-            validator.do_handle_certificate(certificate, blobs, sender)
+            validator.do_handle_certificate(certificate, vec![], sender)
         })
         .await
     }
@@ -180,6 +179,11 @@ where
         Ok(CryptoHash::test_hash("genesis config"))
     }
 
+    async fn upload_blob(&self, content: BlobContent) -> Result<BlobId, NodeError> {
+        self.spawn_and_receive(move |validator, sender| validator.do_upload_blob(content, sender))
+            .await
+    }
+
     async fn download_blob_content(&self, blob_id: BlobId) -> Result<BlobContent, NodeError> {
         self.spawn_and_receive(move |validator, sender| {
             validator.do_download_blob_content(blob_id, sender)
@@ -456,6 +460,23 @@ where
         sender.send(Ok(stream))
     }
 
+    async fn do_upload_blob(
+        self,
+        content: BlobContent,
+        sender: oneshot::Sender<Result<BlobId, NodeError>>,
+    ) -> Result<(), Result<BlobId, NodeError>> {
+        let validator = self.client.lock().await;
+        let blob = Blob::new(content);
+        let id = blob.id();
+        let storage = validator.state.storage_client();
+        let result = match storage.maybe_write_blobs(&[blob]).await {
+            Ok(has_state) if has_state.first() == Some(&true) => Ok(id),
+            Ok(_) => Err(NodeError::BlobsNotFound(vec![id])),
+            Err(error) => Err(error.into()),
+        };
+        sender.send(result)
+    }
+
     async fn do_download_blob_content(
         self,
         blob_id: BlobId,

linera-core/src/unit_tests/wasm_worker_tests.rs

@@ -12,6 +12,7 @@
 
 use std::collections::BTreeSet;
 
+use assert_matches::assert_matches;
 use linera_base::{
     crypto::KeyPair,
     data_types::{
@@ -46,6 +47,7 @@ use linera_views::{memory::MemoryStore, views::CryptoHashView};
 use test_case::test_case;
 
 use super::{init_worker_with_chains, make_certificate};
+use crate::worker::WorkerError;
 
 #[cfg_attr(feature = "wasmer", test_case(WasmRuntime::Wasmer ; "wasmer"))]
 #[cfg_attr(feature = "wasmtime", test_case(WasmRuntime::Wasmtime ; "wasmtime"))]
@@ -103,7 +105,7 @@ where
     let creator_key_pair = KeyPair::generate();
     let creator_chain = ChainDescription::Root(2);
     let (committee, worker) = init_worker_with_chains(
-        storage,
+        storage.clone(),
         vec![
             (publisher_chain, publisher_key_pair.public(), Amount::ZERO),
             (creator_chain, creator_key_pair.public(), Amount::ZERO),
@@ -153,12 +155,17 @@ where
     ));
     let publish_certificate = make_certificate(&committee, &worker, publish_block_proposal);
 
+    assert_matches!(
+        worker
+            .fully_handle_certificate_with_notifications(publish_certificate.clone(), vec![], &())
+            .await,
+        Err(WorkerError::BlobsNotFound(_))
+    );
+    storage
+        .write_blobs(&[contract_blob.clone(), service_blob.clone()])
+        .await?;
     let info = worker
-        .fully_handle_certificate_with_notifications(
-            publish_certificate.clone(),
-            vec![contract_blob.clone(), service_blob.clone()],
-            &(),
-        )
+        .fully_handle_certificate_with_notifications(publish_certificate.clone(), vec![], &())
         .await
         .unwrap()
         .info;

linera-core/src/unit_tests/worker_tests.rs

@@ -728,7 +728,7 @@ where
     drop(chain);
 
     worker
-        .handle_confirmed_certificate(certificate0, vec![], None)
+        .handle_confirmed_certificate(certificate0, None)
         .await?;
     let chain = worker.chain_state_view(ChainId::root(1)).await?;
     drop(chain);
@@ -858,7 +858,7 @@ where
     // Missing earlier blocks
     assert_matches!(
         worker
-            .handle_confirmed_certificate(certificate1.clone(), vec![], None)
+            .handle_confirmed_certificate(certificate1.clone(), None)
             .await,
         Err(WorkerError::MissingEarlierBlocks { .. })
     );
@@ -1088,7 +1088,7 @@ where
             )),
         );
         worker
-            .handle_confirmed_certificate(certificate.clone(), vec![], None)
+            .handle_confirmed_certificate(certificate.clone(), None)
             .await?;
 
         // Then receive the next two messages.
@@ -3922,7 +3922,7 @@ where
     ));
     let certificate = make_certificate(&committee, &worker, value);
     worker
-        .handle_confirmed_certificate(certificate, vec![], None)
+        .handle_confirmed_certificate(certificate, None)
         .await?;
 
     for query_context in query_contexts_after_new_block.clone() {