Upload blobs for validated block certificates in separate messages. (#3153)

## Motivation

Ultimately we want to transfer all blobs separately, rather than in a
single message.
(#3048)

This PR is another step towards that goal: The blobs required by a
validated block certificate are now uploaded separately, rather than in
the same message as the certificate itself.

## Proposal

Add a map of missing blobs for the highest-round validated block to the
chain state, and a `HandlePendingBlob` endpoint to populate that map.

## Test Plan

There are already tests covering different scenarios with validated
blocks' blobs. Where necessary, these were updated.

## Release Plan

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

## Links

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

Author: afck

linera-chain/src/chain.rs

@@ -13,12 +13,13 @@ use futures::stream::{self, StreamExt, TryStreamExt};
 use linera_base::{
     crypto::CryptoHash,
     data_types::{
-        Amount, ArithmeticError, BlockHeight, OracleResponse, Timestamp, UserApplicationDescription,
+        Amount, ArithmeticError, Blob, BlockHeight, OracleResponse, Timestamp,
+        UserApplicationDescription,
     },
     ensure,
     identifiers::{
-        ChainId, ChannelName, Destination, GenericApplicationId, MessageId, Owner, StreamId,
-        UserApplicationId,
+        BlobId, ChainId, ChannelName, Destination, GenericApplicationId, MessageId, Owner,
+        StreamId, UserApplicationId,
     },
 };
 use linera_execution::{
@@ -30,6 +31,7 @@ use linera_execution::{
 use linera_views::{
     context::Context,
     log_view::LogView,
+    map_view::MapView,
     queue_view::QueueView,
     reentrant_collection_view::ReentrantCollectionView,
     register_view::RegisterView,
@@ -46,6 +48,7 @@ use crate::{
     inbox::{Cursor, InboxError, InboxStateView},
     manager::ChainManager,
     outbox::OutboxStateView,
+    types::ValidatedBlockCertificate,
     ChainError, ChainExecutionContext, ExecutionResultExt,
 };
 
@@ -196,6 +199,11 @@ where
 
     /// Consensus state.
     pub manager: ChainManager<C>,
+    /// Pending validated block that is still missing blobs.
+    #[graphql(skip)]
+    pub pending_validated_block: RegisterView<C, Option<ValidatedBlockCertificate>>,
+    /// The incomplete set of blobs for the pending validated block.
+    pub pending_validated_blobs: MapView<C, BlobId, Option<Blob>>,
 
     /// Hashes of all certified blocks for this sender.
     /// This ends with `block_hash` and has length `usize::from(next_block_height)`.
@@ -873,6 +881,9 @@ where
         self.execution_state_hash.set(Some(state_hash));
         // Last, reset the consensus state based on the current ownership.
         let maybe_committee = self.execution_state.system.current_committee().into_iter();
+
+        self.pending_validated_blobs.clear();
+        self.pending_validated_block.set(None);
         self.manager.reset(
             self.execution_state.system.ownership.get().clone(),
             block.height.try_add_one()?,

linera-client/src/client_context.rs

@@ -930,9 +930,7 @@ where
         // Replay the certificates locally.
         for certificate in certificates {
             // No required certificates from other chains: This is only used with benchmark.
-            node.handle_certificate(certificate, vec![], &())
-                .await
-                .unwrap();
+            node.handle_certificate(certificate, &()).await.unwrap();
         }
         // Last update the wallet.
         for chain in self.wallet.as_mut().chains_mut() {

linera-core/src/chain_worker/actor.rs

@@ -106,7 +106,6 @@ where
     /// Process a validated block issued for this multi-owner chain.
     ProcessValidatedBlock {
         certificate: ValidatedBlockCertificate,
-        blobs: Vec<Blob>,
         #[debug(skip)]
         callback: oneshot::Sender<Result<(ChainInfoResponse, NetworkActions, bool), WorkerError>>,
     },
@@ -149,6 +148,13 @@ where
         callback: oneshot::Sender<Result<Blob, WorkerError>>,
     },
 
+    /// Handle a blob that belongs to a pending proposal or validated block certificate.
+    HandlePendingBlob {
+        blob: Blob,
+        #[debug(skip)]
+        callback: oneshot::Sender<Result<ChainInfoResponse, WorkerError>>,
+    },
+
     /// Update the received certificate trackers to at least the given values.
     UpdateReceivedCertificateTrackers {
         new_trackers: BTreeMap<ValidatorName, u64>,
@@ -294,14 +300,9 @@ where
                     .is_ok(),
                 ChainWorkerRequest::ProcessValidatedBlock {
                     certificate,
-                    blobs,
                     callback,
                 } => callback
-                    .send(
-                        self.worker
-                            .process_validated_block(certificate, &blobs)
-                            .await,
-                    )
+                    .send(self.worker.process_validated_block(certificate).await)
                     .is_ok(),
                 ChainWorkerRequest::ProcessConfirmedBlock {
                     certificate,
@@ -340,6 +341,9 @@ where
                 ChainWorkerRequest::DownloadPendingBlob { blob_id, callback } => callback
                     .send(self.worker.download_pending_blob(blob_id).await)
                     .is_ok(),
+                ChainWorkerRequest::HandlePendingBlob { blob, callback } => callback
+                    .send(self.worker.handle_pending_blob(blob).await)
+                    .is_ok(),
                 ChainWorkerRequest::UpdateReceivedCertificateTrackers {
                     new_trackers,
                     callback,

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

@@ -164,7 +164,6 @@ where
     pub(super) async fn process_validated_block(
         &mut self,
         certificate: ValidatedBlockCertificate,
-        blobs: &[Blob],
     ) -> Result<(ChainInfoResponse, NetworkActions, bool), WorkerError> {
         let executed_block = certificate.executed_block();
 
@@ -209,11 +208,30 @@ where
             .executed_block_values
             .insert(Cow::Borrowed(certificate.inner().inner()))
             .await;
-        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 maybe_blobs = self
+            .state
+            .maybe_get_required_blobs(executed_block, &[])
+            .await?;
+        let missing_blob_ids = super::missing_blob_ids(&maybe_blobs);
+        if !missing_blob_ids.is_empty() {
+            let chain = &mut self.state.chain;
+            let pending_validated_block = chain.pending_validated_block.get_mut();
+            if !pending_validated_block
+                .as_ref()
+                .is_some_and(|existing_cert| existing_cert.round > certificate.round)
+            {
+                for (blob_id, maybe_blob) in maybe_blobs {
+                    chain.pending_validated_blobs.insert(&blob_id, maybe_blob)?;
+                }
+                *pending_validated_block = Some(certificate);
+                self.save().await?;
+            }
+            return Err(WorkerError::BlobsNotFound(missing_blob_ids));
+        }
+        let blobs = maybe_blobs
+            .into_iter()
+            .filter_map(|(blob_id, maybe_blob)| Some((blob_id, maybe_blob?)))
+            .collect();
         let old_round = self.state.chain.manager.current_round();
         self.state.chain.manager.create_final_vote(
             certificate,
@@ -560,6 +578,24 @@ where
         Ok(())
     }
 
+    pub(super) async fn handle_pending_blob(
+        &mut self,
+        blob: Blob,
+    ) -> Result<ChainInfoResponse, WorkerError> {
+        let chain = &mut self.state.chain;
+        let blob_id = blob.id();
+        if let Some(maybe_blob) = chain.pending_validated_blobs.get_mut(&blob_id).await? {
+            if maybe_blob.is_none() {
+                *maybe_blob = Some(blob);
+            }
+        }
+        self.save().await?;
+        Ok(ChainInfoResponse::new(
+            &self.state.chain,
+            self.state.config.key_pair(),
+        ))
+    }
+
     /// Stores the chain state in persistent storage.
     ///
     /// Waits until the [`ChainStateView`] is no longer shared before persisting the changes.

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

@@ -225,11 +225,10 @@ where
     pub(super) async fn process_validated_block(
         &mut self,
         certificate: ValidatedBlockCertificate,
-        blobs: &[Blob],
     ) -> Result<(ChainInfoResponse, NetworkActions, bool), WorkerError> {
         ChainWorkerStateWithAttemptedChanges::new(self)
             .await
-            .process_validated_block(certificate, blobs)
+            .process_validated_block(certificate)
             .await
     }
 
@@ -300,6 +299,17 @@ where
         maybe_blob.ok_or_else(|| WorkerError::BlobsNotFound(vec![blob_id]))
     }
 
+    /// Adds the blob to pending blocks or validated block certificates that are missing it.
+    pub(super) async fn handle_pending_blob(
+        &mut self,
+        blob: Blob,
+    ) -> Result<ChainInfoResponse, WorkerError> {
+        ChainWorkerStateWithAttemptedChanges::new(&mut *self)
+            .await
+            .handle_pending_blob(blob)
+            .await
+    }
+
     /// Ensures that the current chain is active, returning an error otherwise.
     fn ensure_is_active(&mut self) -> Result<(), WorkerError> {
         if !self.knows_chain_is_active {
@@ -309,61 +319,57 @@ where
         Ok(())
     }
 
-    /// Returns an error if unrelated blobs were provided.
-    fn check_for_unneeded_blobs(
-        &self,
-        required_blob_ids: &HashSet<BlobId>,
-        blobs: &[Blob],
-    ) -> Result<(), WorkerError> {
-        // Find all certificates containing blobs used when executing this block.
-        for blob in blobs {
-            let blob_id = blob.id();
-            ensure!(
-                required_blob_ids.contains(&blob_id),
-                WorkerError::UnneededBlob { blob_id }
-            );
-        }
-
-        Ok(())
-    }
-
     /// Returns the blobs required by the given executed block. The ones that are not passed in
     /// are read from the chain manager or from storage.
     async fn get_required_blobs(
         &self,
         executed_block: &ExecutedBlock,
         blobs: &[Blob],
     ) -> Result<BTreeMap<BlobId, Blob>, WorkerError> {
-        let mut blob_ids = executed_block.required_blob_ids();
-        let mut found_blobs = BTreeMap::new();
+        let maybe_blobs = self.maybe_get_required_blobs(executed_block, blobs).await?;
+        let not_found_blob_ids = missing_blob_ids(&maybe_blobs);
+        ensure!(
+            not_found_blob_ids.is_empty(),
+            WorkerError::BlobsNotFound(not_found_blob_ids)
+        );
+        Ok(maybe_blobs
+            .into_iter()
+            .filter_map(|(blob_id, maybe_blob)| Some((blob_id, maybe_blob?)))
+            .collect())
+    }
 
-        for blob in blobs {
-            if blob_ids.remove(&blob.id()) {
-                found_blobs.insert(blob.id(), blob.clone());
+    /// Returns the blobs required by the given executed block. The ones that are not passed in
+    /// are read from the chain manager or from storage.
+    async fn maybe_get_required_blobs(
+        &self,
+        executed_block: &ExecutedBlock,
+        provided_blobs: &[Blob],
+    ) -> Result<BTreeMap<BlobId, Option<Blob>>, WorkerError> {
+        let required_blob_ids = executed_block.required_blob_ids().into_iter();
+        let mut maybe_blobs = BTreeMap::from_iter(required_blob_ids.map(|blob_id| (blob_id, None)));
+
+        for blob in provided_blobs {
+            if let Some(maybe_blob) = maybe_blobs.get_mut(&blob.id()) {
+                *maybe_blob = Some(blob.clone());
             }
         }
-        let mut missing_blob_ids = Vec::new();
-        for blob_id in blob_ids {
-            if let Some(blob) = self.chain.manager.pending_blob(&blob_id).await? {
-                found_blobs.insert(blob_id, blob);
-            } else {
-                missing_blob_ids.push(blob_id);
+        for (blob_id, maybe_blob) in &mut maybe_blobs {
+            if maybe_blob.is_some() {
+                continue;
+            }
+            if let Some(blob) = self.chain.manager.pending_blob(blob_id).await? {
+                *maybe_blob = Some(blob);
+            } else if let Some(Some(blob)) = self.chain.pending_validated_blobs.get(blob_id).await?
+            {
+                *maybe_blob = Some(blob);
             }
         }
+        let missing_blob_ids = missing_blob_ids(&maybe_blobs);
         let blobs_from_storage = self.storage.read_blobs(&missing_blob_ids).await?;
-        let mut not_found_blob_ids = Vec::new();
         for (blob_id, maybe_blob) in missing_blob_ids.into_iter().zip(blobs_from_storage) {
-            if let Some(blob) = maybe_blob {
-                found_blobs.insert(blob_id, blob);
-            } else {
-                not_found_blob_ids.push(blob_id);
-            }
+            maybe_blobs.insert(blob_id, maybe_blob);
         }
-        ensure!(
-            not_found_blob_ids.is_empty(),
-            WorkerError::BlobsNotFound(not_found_blob_ids)
-        );
-        Ok(found_blobs)
+        Ok(maybe_blobs)
     }
 
     /// Adds any newly created chains to the set of `tracked_chains`, if the parent chain is
@@ -542,6 +548,15 @@ where
     }
 }
 
+/// Returns the keys whose value is `None`.
+fn missing_blob_ids(maybe_blobs: &BTreeMap<BlobId, Option<Blob>>) -> Vec<BlobId> {
+    maybe_blobs
+        .iter()
+        .filter(|(_, maybe_blob)| maybe_blob.is_none())
+        .map(|(chain_id, _)| *chain_id)
+        .collect()
+}
+
 /// Returns an error if the block is not at the expected epoch.
 fn check_block_epoch(chain_epoch: Epoch, block: &Block) -> Result<(), WorkerError> {
     ensure!(

linera-core/src/client/mod.rs

@@ -397,24 +397,27 @@ where
         certificate: GenericCertificate<T>,
         blobs: Vec<Blob>,
     ) -> Result<ChainInfoResponse, LocalNodeError> {
-        if T::KIND == CertificateKind::Confirmed {
-            let result = self
+        let chain_id = certificate.inner().chain_id();
+        let result = self
+            .local_node
+            .handle_certificate(certificate.clone(), &self.notifier)
+            .await;
+        if let Err(LocalNodeError::BlobsNotFound(_)) = &result {
+            match T::KIND {
+                CertificateKind::Confirmed => self.local_node.store_blobs(&blobs).await?,
+                CertificateKind::Validated => {
+                    self.local_node
+                        .handle_pending_blobs(chain_id, blobs)
+                        .await?
+                }
+                CertificateKind::Timeout => return result,
+            }
+            return self
                 .local_node
-                .handle_certificate(certificate.clone(), vec![], &self.notifier)
+                .handle_certificate(certificate, &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
         }
+        result
     }
 }