Change the API of read_blob_state to return options. (#4018)

## Motivation

Following PR #4014 we
look at the `read_blob_state` and change their API as options. The same
design considerations apply here.

## Proposal

The `read_blob_state(s)` were returning a ViewError that contained a
`NotFound`. This is inadequate and in any
case the `NotFound` is never processed.

In all use cases of those functions, we had some `BlobsNotFound` variant
in the relevant types, the error type was thus put there. It could
change the behavior of some function. However, this is better than
having some error buried in `ViewError`.

The `NotFound` cannot yet be changed because it is used by the
`read_certificate(s)`.

Other changes:
* After PR 4104, some of the metrics were placed before the storage
calls. This has been corrected.
* In some of the accounting of the calls, the plural calls were
sometimes counting several calls as one, and in other cases as the
number of entries in the call. This has been corrected.

## Test Plan

The CI.

## Release Plan

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

## Links

None.

Author: MathieuDutSik

linera-core/src/local_node.rs

@@ -18,7 +18,7 @@ use linera_chain::{
     types::{Block, GenericCertificate, LiteCertificate},
     ChainStateView,
 };
-use linera_execution::{committee::Committee, Query, QueryOutcome};
+use linera_execution::{committee::Committee, BlobState, Query, QueryOutcome};
 use linera_storage::Storage;
 use linera_views::ViewError;
 use thiserror::Error;
@@ -170,6 +170,31 @@ where
         Ok(storage.read_blobs(blob_ids).await?.into_iter().collect())
     }
 
+    /// Reads blob states from storage
+    pub async fn read_blob_states_from_storage(
+        &self,
+        blob_ids: &[BlobId],
+    ) -> Result<Vec<BlobState>, LocalNodeError> {
+        let storage = self.storage_client();
+        let mut blobs_not_found = Vec::new();
+        let mut blob_states = Vec::new();
+        for (blob_state, blob_id) in storage
+            .read_blob_states(blob_ids)
+            .await?
+            .into_iter()
+            .zip(blob_ids)
+        {
+            match blob_state {
+                None => blobs_not_found.push(*blob_id),
+                Some(blob_state) => blob_states.push(blob_state),
+            }
+        }
+        if !blobs_not_found.is_empty() {
+            return Err(LocalNodeError::BlobsNotFound(blobs_not_found));
+        }
+        Ok(blob_states)
+    }
+
     /// Looks for the specified blobs in the local chain manager's locking blobs.
     /// Returns `Ok(None)` if any of the blobs is not found.
     pub async fn get_locking_blobs(

linera-core/src/unit_tests/test_utils.rs

@@ -588,13 +588,19 @@ where
         sender: oneshot::Sender<Result<CryptoHash, NodeError>>,
     ) -> Result<(), Result<CryptoHash, NodeError>> {
         let validator = self.client.lock().await;
-        let certificate_hash = validator
+        let blob_state = validator
             .state
             .storage_client()
             .read_blob_state(blob_id)
             .await
-            .map(|blob_state| blob_state.last_used_by)
             .map_err(Into::into);
+        let certificate_hash = match blob_state {
+            Err(err) => Err(err),
+            Ok(blob_state) => match blob_state {
+                None => Err(NodeError::BlobsNotFound(vec![blob_id])),
+                Some(blob_state) => Ok(blob_state.last_used_by),
+            },
+        };
 
         sender.send(certificate_hash)
     }

linera-core/src/updater.rs

@@ -317,8 +317,10 @@ where
                         .node
                         .missing_blob_ids(mem::take(&mut blob_ids))
                         .await?;
-                    let local_storage = self.local_node.storage_client();
-                    let blob_states = local_storage.read_blob_states(&missing_blob_ids).await?;
+                    let blob_states = self
+                        .local_node
+                        .read_blob_states_from_storage(&missing_blob_ids)
+                        .await?;
                     let mut chain_heights = BTreeMap::new();
                     for blob_state in blob_states {
                         let block_chain_id = blob_state.chain_id;

linera-execution/src/test_utils/solidity.rs

@@ -104,10 +104,10 @@ pub fn load_solidity_example(path: &str) -> anyhow::Result<Vec<u8>> {
         .lines()
         .filter_map(|line| line.trim_start().strip_prefix("contract "))
         .next()
-        .ok_or(anyhow::anyhow!("Not matching"))?;
+        .ok_or_else(|| anyhow::anyhow!("Not matching"))?;
     let contract_name: &str = contract_name
         .strip_suffix(" {")
-        .ok_or(anyhow::anyhow!("Not matching"))?;
+        .ok_or_else(|| anyhow::anyhow!("Not matching"))?;
     get_bytecode(&source_code, contract_name)
 }
 

linera-service/src/cli/main.rs

@@ -1325,7 +1325,7 @@ impl ClientOptions {
     }
 
     fn config_path(&self) -> Result<PathBuf, Error> {
-        let mut config_dir = dirs::config_dir().ok_or(anyhow!(
+        let mut config_dir = dirs::config_dir().ok_or_else(|| anyhow!(
             "Default wallet directory is not supported in this platform: please specify storage and wallet paths"
         ))?;
         config_dir.push("linera");

linera-service/src/proxy/grpc.rs

@@ -473,9 +473,7 @@ where
             .read_network_description()
             .await
             .map_err(Self::error_to_status)?
-            .ok_or(Status::not_found(
-                "Cannot find network description in the database",
-            ))?;
+            .ok_or_else(|| Status::not_found("Cannot find network description in the database"))?;
         Ok(Response::new(description.into()))
     }
 
@@ -504,7 +502,7 @@ where
             .read_blob(blob_id)
             .await
             .map_err(Self::error_to_status)?;
-        let blob = blob.ok_or(Status::not_found(format!("Blob not found {}", blob_id)))?;
+        let blob = blob.ok_or_else(|| Status::not_found(format!("Blob not found {}", blob_id)))?;
         Ok(Response::new(blob.into_content().try_into()?))
     }
 
@@ -626,6 +624,8 @@ where
             .read_blob_state(blob_id)
             .await
             .map_err(Self::error_to_status)?;
+        let blob_state =
+            blob_state.ok_or_else(|| Status::not_found(format!("Blob not found {}", blob_id)))?;
         Ok(Response::new(blob_state.last_used_by.into()))
     }
 

linera-service/src/proxy/main.rs

@@ -319,7 +319,7 @@ where
                     .storage
                     .read_network_description()
                     .await?
-                    .ok_or(anyhow!("Cannot find network description in the database"))?;
+                    .ok_or_else(|| anyhow!("Cannot find network description in the database"))?;
                 Ok(Some(RpcMessage::NetworkDescriptionResponse(Box::new(
                     description,
                 ))))
@@ -335,7 +335,7 @@ where
             }
             DownloadBlob(blob_id) => {
                 let blob = self.storage.read_blob(*blob_id).await?;
-                let blob = blob.ok_or(anyhow!("Blob not found {}", blob_id))?;
+                let blob = blob.ok_or_else(|| anyhow!("Blob not found {}", blob_id))?;
                 let content = blob.into_content();
                 Ok(Some(RpcMessage::DownloadBlobResponse(Box::new(content))))
             }
@@ -346,9 +346,13 @@ where
                 let certificates = self.storage.read_certificates(hashes).await?;
                 Ok(Some(RpcMessage::DownloadCertificatesResponse(certificates)))
             }
-            BlobLastUsedBy(blob_id) => Ok(Some(RpcMessage::BlobLastUsedByResponse(Box::new(
-                self.storage.read_blob_state(*blob_id).await?.last_used_by,
-            )))),
+            BlobLastUsedBy(blob_id) => {
+                let blob_state = self.storage.read_blob_state(*blob_id).await?;
+                let blob_state = blob_state.ok_or_else(|| anyhow!("Blob not found {}", blob_id))?;
+                Ok(Some(RpcMessage::BlobLastUsedByResponse(Box::new(
+                    blob_state.last_used_by,
+                ))))
+            }
             MissingBlobIds(blob_ids) => Ok(Some(RpcMessage::MissingBlobIdsResponse(
                 self.storage.missing_blobs(&blob_ids).await?,
             ))),

linera-storage/src/db_storage.rs

@@ -637,41 +637,35 @@ where
             .collect())
     }
 
-    async fn read_blob_state(&self, blob_id: BlobId) -> Result<BlobState, ViewError> {
+    async fn read_blob_state(&self, blob_id: BlobId) -> Result<Option<BlobState>, ViewError> {
         let blob_state_key = bcs::to_bytes(&BaseKey::BlobState(blob_id))?;
-        let maybe_blob_state = self.store.read_value::<BlobState>(&blob_state_key).await?;
+        let blob_state = self.store.read_value::<BlobState>(&blob_state_key).await?;
         #[cfg(with_metrics)]
         metrics::READ_BLOB_STATE_COUNTER
             .with_label_values(&[])
             .inc();
-        let blob_state = maybe_blob_state
-            .ok_or_else(|| ViewError::not_found("blob state for blob ID", blob_id))?;
         Ok(blob_state)
     }
 
-    async fn read_blob_states(&self, blob_ids: &[BlobId]) -> Result<Vec<BlobState>, ViewError> {
+    async fn read_blob_states(
+        &self,
+        blob_ids: &[BlobId],
+    ) -> Result<Vec<Option<BlobState>>, ViewError> {
         if blob_ids.is_empty() {
             return Ok(Vec::new());
         }
         let blob_state_keys = blob_ids
             .iter()
             .map(|blob_id| bcs::to_bytes(&BaseKey::BlobState(*blob_id)))
             .collect::<Result<_, _>>()?;
-        let maybe_blob_states = self
+        let blob_states = self
             .store
             .read_multi_values::<BlobState>(blob_state_keys)
             .await?;
         #[cfg(with_metrics)]
         metrics::READ_BLOB_STATES_COUNTER
             .with_label_values(&[])
-            .inc();
-        let blob_states = maybe_blob_states
-            .into_iter()
-            .zip(blob_ids)
-            .map(|(blob_state, blob_id)| {
-                blob_state.ok_or_else(|| ViewError::not_found("blob state for blob ID", blob_id))
-            })
-            .collect::<Result<_, _>>()?;
+            .inc_by(blob_ids.len() as u64);
         Ok(blob_states)
     }
 
@@ -705,14 +699,13 @@ where
         blob_id: BlobId,
         blob_state: BlobState,
     ) -> Result<Epoch, ViewError> {
-        let current_blob_state = self.read_blob_state(blob_id).await;
+        let current_blob_state = self.read_blob_state(blob_id).await?;
         let (should_write, latest_epoch) = match current_blob_state {
-            Ok(current_blob_state) => (
+            Some(current_blob_state) => (
                 current_blob_state.epoch < blob_state.epoch,
                 current_blob_state.epoch.max(blob_state.epoch),
             ),
-            Err(ViewError::NotFound(_)) => (true, blob_state.epoch),
-            Err(err) => return Err(err),
+            None => (true, blob_state.epoch),
         };
 
         if should_write {
@@ -856,7 +849,7 @@ where
             #[cfg(with_metrics)]
             metrics::READ_CERTIFICATES_COUNTER
                 .with_label_values(&[])
-                .inc();
+                .inc_by(hashes.len() as u64);
         }
         let values = values?;
         let mut certificates = Vec::new();
@@ -869,9 +862,10 @@ where
 
     async fn read_event(&self, event_id: EventId) -> Result<Option<Vec<u8>>, ViewError> {
         let event_key = bcs::to_bytes(&BaseKey::Event(event_id.clone()))?;
+        let event = self.store.read_value_bytes(&event_key).await?;
         #[cfg(with_metrics)]
         metrics::READ_EVENT_COUNTER.with_label_values(&[]).inc();
-        Ok(self.store.read_value_bytes(&event_key).await?)
+        Ok(event)
     }
 
     async fn contains_event(&self, event_id: EventId) -> Result<bool, ViewError> {

linera-storage/src/lib.rs

@@ -90,10 +90,13 @@ pub trait Storage: Sized {
     async fn read_blobs(&self, blob_ids: &[BlobId]) -> Result<Vec<Option<Blob>>, ViewError>;
 
     /// Reads the blob state with the given blob ID.
-    async fn read_blob_state(&self, blob_id: BlobId) -> Result<BlobState, ViewError>;
+    async fn read_blob_state(&self, blob_id: BlobId) -> Result<Option<BlobState>, ViewError>;
 
     /// Reads the blob states with the given blob IDs.
-    async fn read_blob_states(&self, blob_ids: &[BlobId]) -> Result<Vec<BlobState>, ViewError>;
+    async fn read_blob_states(
+        &self,
+        blob_ids: &[BlobId],
+    ) -> Result<Vec<Option<BlobState>>, ViewError>;
 
     /// Reads the hashed certificate values in descending order from the given hash.
     async fn read_confirmed_blocks_downward(