Simplify storage of BlobStates (#4053)

## Motivation

* Improve code readability
* Fix a particular failure path for clients

## Proposal

* Remove the mysterious `overwrite` argument to `write_blob_states`
* Do not record last_used_by in the blob state if the certificate is not
yet in storage
* Removed unused APIs

## Test Plan

CI

## Release Plan

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

Author: ma2bd

linera-chain/src/block.rs

@@ -139,12 +139,12 @@ impl ConfirmedBlock {
     }
 
     /// Returns a blob state that applies to all blobs used by this block.
-    pub fn to_blob_state(&self) -> BlobState {
+    pub fn to_blob_state(&self, is_stored_block: bool) -> BlobState {
         BlobState {
-            last_used_by: self.0.hash(),
+            last_used_by: is_stored_block.then_some(self.0.hash()),
             chain_id: self.chain_id(),
             block_height: self.height(),
-            epoch: self.epoch(),
+            epoch: is_stored_block.then_some(self.epoch()),
         }
     }
 }

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

@@ -340,12 +340,11 @@ where
         }
 
         // Update the blob state with last used certificate hash.
-        let blob_state = certificate.value().to_blob_state();
-        let overwrite = blobs_result.is_ok(); // Overwrite only if we wrote the certificate.
+        let blob_state = certificate.value().to_blob_state(blobs_result.is_ok());
         let blob_ids = required_blob_ids.into_iter().collect::<Vec<_>>();
         self.state
             .storage
-            .maybe_write_blob_states(&blob_ids, blob_state, overwrite)
+            .maybe_write_blob_states(&blob_ids, blob_state)
             .await?;
         let mut blobs = blobs_result?
             .into_iter()

linera-core/src/unit_tests/test_utils.rs

@@ -596,7 +596,9 @@ where
             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),
+                Some(blob_state) => blob_state
+                    .last_used_by
+                    .ok_or_else(|| NodeError::BlobsNotFound(vec![blob_id])),
             },
         };
 

linera-execution/src/lib.rs

@@ -1283,14 +1283,16 @@ impl From<Vec<u8>> for QueryResponse {
 /// The state of a blob of binary data.
 #[derive(Eq, PartialEq, Debug, Hash, Clone, Serialize, Deserialize)]
 pub struct BlobState {
-    /// Hash of the last `Certificate` that published or used this blob.
-    pub last_used_by: CryptoHash,
+    /// Hash of the last `Certificate` that published or used this blob. If empty, the
+    /// blob is known to be published by a confirmed certificate but we may not have fully
+    /// processed this certificate just yet.
+    pub last_used_by: Option<CryptoHash>,
     /// The `ChainId` of the chain that published the change
     pub chain_id: ChainId,
     /// The `BlockHeight` of the chain that published the change
     pub block_height: BlockHeight,
-    /// Epoch of the `last_used_by` certificate.
-    pub epoch: Epoch,
+    /// Epoch of the `last_used_by` certificate (if any).
+    pub epoch: Option<Epoch>,
 }
 
 /// The runtime to use for running the application.

linera-service/src/proxy/grpc.rs

@@ -632,7 +632,10 @@ where
             .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()))
+        let last_used_by = blob_state
+            .last_used_by
+            .ok_or_else(|| Status::not_found(format!("Blob not found {}", blob_id)))?;
+        Ok(Response::new(last_used_by.into()))
     }
 
     #[instrument(skip_all, err(level = Level::WARN))]

linera-service/src/proxy/main.rs

@@ -370,8 +370,11 @@ where
             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))?;
+                let last_used_by = blob_state
+                    .last_used_by
+                    .ok_or_else(|| anyhow!("Blob not found {}", blob_id))?;
                 Ok(Some(RpcMessage::BlobLastUsedByResponse(Box::new(
-                    blob_state.last_used_by,
+                    last_used_by,
                 ))))
             }
             MissingBlobIds(blob_ids) => Ok(Some(RpcMessage::MissingBlobIdsResponse(

linera-storage/src/db_storage.rs

@@ -9,7 +9,7 @@ use dashmap::DashMap;
 use linera_base::prometheus_util::MeasureLatency as _;
 use linera_base::{
     crypto::CryptoHash,
-    data_types::{Blob, Epoch, NetworkDescription, TimeDelta, Timestamp},
+    data_types::{Blob, NetworkDescription, TimeDelta, Timestamp},
     identifiers::{ApplicationId, BlobId, ChainId, EventId, IndexAndEvent, StreamId},
 };
 use linera_chain::{
@@ -696,35 +696,13 @@ where
         Ok(())
     }
 
-    async fn maybe_write_blob_state(
-        &self,
-        blob_id: BlobId,
-        blob_state: BlobState,
-    ) -> Result<Epoch, ViewError> {
-        let current_blob_state = self.read_blob_state(blob_id).await?;
-        let (should_write, latest_epoch) = match current_blob_state {
-            Some(current_blob_state) => (
-                current_blob_state.epoch < blob_state.epoch,
-                current_blob_state.epoch.max(blob_state.epoch),
-            ),
-            None => (true, blob_state.epoch),
-        };
-
-        if should_write {
-            self.write_blob_state(blob_id, &blob_state).await?;
-        }
-
-        Ok(latest_epoch)
-    }
-
     async fn maybe_write_blob_states(
         &self,
         blob_ids: &[BlobId],
         blob_state: BlobState,
-        overwrite: bool,
-    ) -> Result<Vec<Epoch>, ViewError> {
+    ) -> Result<(), ViewError> {
         if blob_ids.is_empty() {
-            return Ok(Vec::new());
+            return Ok(());
         }
         let blob_state_keys = blob_ids
             .iter()
@@ -734,36 +712,22 @@ where
             .store
             .read_multi_values::<BlobState>(blob_state_keys)
             .await?;
-        let mut latest_epochs = Vec::new();
         let mut batch = Batch::new();
-        let mut need_write = false;
         for (maybe_blob_state, blob_id) in maybe_blob_states.iter().zip(blob_ids) {
-            let (should_write, latest_epoch) = match maybe_blob_state {
-                None => (true, blob_state.epoch),
-                Some(current_blob_state) => (
-                    overwrite && current_blob_state.epoch < blob_state.epoch,
-                    current_blob_state.epoch.max(blob_state.epoch),
-                ),
-            };
-            if should_write {
-                batch.add_blob_state(*blob_id, &blob_state)?;
-                need_write = true;
+            match maybe_blob_state {
+                None => {
+                    batch.add_blob_state(*blob_id, &blob_state)?;
+                }
+                Some(state) => {
+                    if state.epoch < blob_state.epoch {
+                        batch.add_blob_state(*blob_id, &blob_state)?;
+                    }
+                }
             }
-            latest_epochs.push(latest_epoch);
-        }
-        if need_write {
-            self.write_batch(batch).await?;
         }
-        Ok(latest_epochs)
-    }
-
-    async fn write_blob_state(
-        &self,
-        blob_id: BlobId,
-        blob_state: &BlobState,
-    ) -> Result<(), ViewError> {
-        let mut batch = Batch::new();
-        batch.add_blob_state(blob_id, blob_state)?;
+        // We tolerate race conditions because two active chains are likely to
+        // be both from the latest epoch, and otherwise failing to pick the
+        // more recent blob state has limited impact.
         self.write_batch(batch).await?;
         Ok(())
     }
@@ -997,6 +961,9 @@ where
     }
 
     async fn write_batch(&self, batch: Batch) -> Result<(), ViewError> {
+        if batch.key_value_bytes.is_empty() {
+            return Ok(());
+        }
         let mut futures = Vec::new();
         for (key, bytes) in batch.key_value_bytes.into_iter() {
             let store = self.store.clone();

linera-storage/src/lib.rs

@@ -14,8 +14,8 @@ use dashmap::{mapref::entry::Entry, DashMap};
 use linera_base::{
     crypto::CryptoHash,
     data_types::{
-        ApplicationDescription, Blob, ChainDescription, CompressedBytecode, Epoch,
-        NetworkDescription, TimeDelta, Timestamp,
+        ApplicationDescription, Blob, ChainDescription, CompressedBytecode, NetworkDescription,
+        TimeDelta, Timestamp,
     },
     identifiers::{ApplicationId, BlobId, ChainId, EventId, IndexAndEvent, StreamId},
     vm::VmRuntime,
@@ -115,31 +115,16 @@ pub trait Storage: Sized {
         certificate: &ConfirmedBlockCertificate,
     ) -> Result<(), ViewError>;
 
-    /// Writes the given blob state.
-    async fn write_blob_state(
-        &self,
-        blob_id: BlobId,
-        blob_state: &BlobState,
-    ) -> Result<(), ViewError>;
-
     /// Writes the given blobs, but only if they already have a blob state. Returns `true` for the
     /// blobs that were written.
     async fn maybe_write_blobs(&self, blobs: &[Blob]) -> Result<Vec<bool>, ViewError>;
 
-    /// Attempts to write the given blob state. Returns the latest `Epoch` to have used this blob.
-    async fn maybe_write_blob_state(
-        &self,
-        blob_id: BlobId,
-        blob_state: BlobState,
-    ) -> Result<Epoch, ViewError>;
-
     /// Attempts to write the given blob state. Returns the latest `Epoch` to have used this blob.
     async fn maybe_write_blob_states(
         &self,
         blob_ids: &[BlobId],
         blob_state: BlobState,
-        overwrite: bool,
-    ) -> Result<Vec<Epoch>, ViewError>;
+    ) -> Result<(), ViewError>;
 
     /// Writes several blobs.
     async fn write_blobs(&self, blobs: &[Blob]) -> Result<(), ViewError>;