diff --git a/linera-base/src/data_types.rs b/linera-base/src/data_types.rs index ecafc780b1f..db423a02d70 100644 --- a/linera-base/src/data_types.rs +++ b/linera-base/src/data_types.rs @@ -824,13 +824,6 @@ pub struct HashedBlob { pub blob: Blob, } -/// 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, -} - impl HashedBlob { /// Loads a hashed blob from a file. pub async fn load_from_file(path: impl AsRef) -> std::io::Result { diff --git a/linera-base/src/identifiers.rs b/linera-base/src/identifiers.rs index 556bf238e51..f11cb62c019 100644 --- a/linera-base/src/identifiers.rs +++ b/linera-base/src/identifiers.rs @@ -166,12 +166,6 @@ impl BlobId { pub fn new(blob: &Blob) -> Self { BlobId(CryptoHash::new(blob)) } - - /// Returns the blob ID of `TestString(s)`, for testing purposes. - #[cfg(with_testing)] - pub fn test_blob_id(s: impl Into) -> Self { - BlobId(CryptoHash::test_hash(s)) - } } impl Display for BlobId { diff --git a/linera-chain/src/data_types.rs b/linera-chain/src/data_types.rs index 896ab8ef57d..ba7b7b7dc3f 100644 --- a/linera-chain/src/data_types.rs +++ b/linera-chain/src/data_types.rs @@ -7,7 +7,7 @@ use std::{borrow::Cow, collections::HashSet}; use async_graphql::SimpleObject; use linera_base::{ crypto::{BcsHashable, BcsSignable, CryptoError, CryptoHash, KeyPair, PublicKey, Signature}, - data_types::{Amount, BlockHeight, HashedBlob, OracleRecord, Round, Timestamp}, + data_types::{Amount, BlockHeight, HashedBlob, OracleRecord, OracleResponse, Round, Timestamp}, doc_scalar, ensure, identifiers::{ Account, BlobId, ChainId, ChannelName, Destination, GenericApplicationId, MessageId, Owner, @@ -744,6 +744,19 @@ impl ExecutedBlock { index, } } + + pub fn required_blob_ids(&self) -> HashSet { + let mut required_blob_ids = HashSet::new(); + for record in &self.outcome.oracle_records { + for response in &record.responses { + if let OracleResponse::Blob(blob_id) = response { + required_blob_ids.insert(*blob_id); + } + } + } + + required_blob_ids + } } impl BlockExecutionOutcome { diff --git a/linera-core/src/chain_worker/state.rs b/linera-core/src/chain_worker/state.rs index 1195d904a15..0e703650434 100644 --- a/linera-core/src/chain_worker/state.rs +++ b/linera-core/src/chain_worker/state.rs @@ -15,9 +15,7 @@ use futures::{ }; use linera_base::{ crypto::CryptoHash, - data_types::{ - ArithmeticError, BlockHeight, HashedBlob, OracleRecord, OracleResponse, Timestamp, - }, + data_types::{ArithmeticError, BlockHeight, HashedBlob, Timestamp}, ensure, identifiers::{BlobId, ChainId}, }; @@ -31,8 +29,8 @@ use linera_chain::{ }; use linera_execution::{ committee::{Committee, Epoch}, - BytecodeLocation, ExecutionRequest, Query, QueryContext, Response, ServiceRuntimeRequest, - UserApplicationDescription, UserApplicationId, + BlobState, BytecodeLocation, ExecutionRequest, Query, QueryContext, Response, + ServiceRuntimeRequest, UserApplicationDescription, UserApplicationId, }; use linera_storage::Storage; use linera_views::{ @@ -309,13 +307,14 @@ where async fn check_no_missing_blobs( &self, block: &Block, + blobs_in_block: HashSet, hashed_certificate_values: &[HashedCertificateValue], hashed_blobs: &[HashedBlob], ) -> Result<(), WorkerError> { let missing_bytecodes = self .get_missing_bytecodes(block, hashed_certificate_values) .await?; - let missing_blobs = self.get_missing_blobs(block, hashed_blobs).await?; + let missing_blobs = self.get_missing_blobs(blobs_in_block, hashed_blobs).await?; if missing_bytecodes.is_empty() && missing_blobs.is_empty() { return Ok(()); @@ -330,10 +329,9 @@ where /// Returns the blobs required by the block that we don't have, or an error if unrelated blobs were provided. async fn get_missing_blobs( &self, - block: &Block, + mut required_blob_ids: HashSet, hashed_blobs: &[HashedBlob], ) -> Result, WorkerError> { - let mut required_blob_ids = block.published_blob_ids(); // Find all certificates containing blobs used when executing this block. for hashed_blob in hashed_blobs { let blob_id = hashed_blob.id(); @@ -344,13 +342,26 @@ where } let pending_blobs = &self.chain.manager.get().pending_blobs; - Ok(self + let tasks = self .recent_hashed_blobs .subtract_cached_items_from::<_, Vec<_>>(required_blob_ids, |id| id) .await .into_iter() .filter(|blob_id| !pending_blobs.contains_key(blob_id)) - .collect()) + .map(|blob_id| { + self.storage + .contains_blob(blob_id) + .map(move |result| (blob_id, result)) + }); + + let mut missing_blobs = vec![]; + for (blob_id, result) in future::join_all(tasks).await { + if !result? { + missing_blobs.push(blob_id); + } + } + + Ok(missing_blobs) } /// Returns the blobs requested by their `blob_ids` that are either in pending in the @@ -406,10 +417,8 @@ where .collect::>(); let mut missing_locations = vec![]; for (location, result) in future::join_all(tasks).await { - match result { - Ok(true) => {} - Ok(false) => missing_locations.push(location), - Err(err) => Err(err)?, + if !result? { + missing_locations.push(location); } } @@ -685,7 +694,12 @@ where // Verify that all required bytecode hashed certificate values and blobs are available, and no // unrelated ones provided. self.0 - .check_no_missing_blobs(block, hashed_certificate_values, hashed_blobs) + .check_no_missing_blobs( + block, + block.published_blob_ids(), + hashed_certificate_values, + hashed_blobs, + ) .await?; // Write the values so that the bytecode is available during execution. // TODO(#2199): We should not persist anything in storage before the block is confirmed. @@ -1070,7 +1084,12 @@ where // Verify that all required bytecode hashed certificate values and blobs are available, and no // unrelated ones provided. self.state - .check_no_missing_blobs(block, hashed_certificate_values, hashed_blobs) + .check_no_missing_blobs( + block, + required_blob_ids.clone(), + hashed_certificate_values, + hashed_blobs, + ) .await?; // Persist certificate and hashed certificate values. self.state @@ -1083,31 +1102,7 @@ where .await; } - let blob_ids_in_oracle_records = oracle_records - .iter() - .flat_map(|record| { - record.responses.iter().filter_map(|response| { - if let OracleResponse::Blob(blob_id) = response { - Some(blob_id) - } else { - None - } - }) - }) - .collect::>(); - - let block_blob_ids = block.published_blob_ids(); - let missing_blobs_from_oracles = block_blob_ids - .iter() - .filter(|block_blob_id| !blob_ids_in_oracle_records.contains(block_blob_id)) - .cloned() - .collect::>(); - ensure!( - missing_blobs_from_oracles.is_empty(), - WorkerError::MissingBlockBlobsInOracles(missing_blobs_from_oracles) - ); - - let blobs_in_block = self.state.get_blobs(block_blob_ids).await?; + let blobs_in_block = self.state.get_blobs(required_blob_ids.clone()).await?; let certificate_hash = certificate.hash(); let (result_hashed_certificate_value, result_blobs, result_certificate) = tokio::join!( @@ -1122,10 +1117,14 @@ where result_certificate?; // Update the blob state with last used certificate hash. - try_join_all(blob_ids_in_oracle_records.into_iter().map(|blob_id| { - self.state - .storage - .write_blob_state(*blob_id, certificate_hash) + try_join_all(required_blob_ids.into_iter().map(|blob_id| { + self.state.storage.maybe_write_blob_state( + blob_id, + BlobState { + last_used_by: certificate_hash, + epoch: certificate.value().epoch(), + }, + ) })) .await?; diff --git a/linera-core/src/unit_tests/client_tests.rs b/linera-core/src/unit_tests/client_tests.rs index 6179aa8dd02..5f51283d2f5 100644 --- a/linera-core/src/unit_tests/client_tests.rs +++ b/linera-core/src/unit_tests/client_tests.rs @@ -1445,6 +1445,16 @@ where client_a.synchronize_from_validators().await.unwrap(); assert_eq!(expected_blob_id, blob_id); assert_eq!(certificate.round, Round::MultiLeader(0)); + assert!(certificate + .value() + .executed_block() + .unwrap() + .outcome + .oracle_records + .iter() + .any(|record| record + .responses + .contains(&OracleResponse::Blob(expected_blob_id)))); let previous_block_hash = client_a.block_hash().unwrap(); // Validators goes back up diff --git a/linera-core/src/unit_tests/wasm_client_tests.rs b/linera-core/src/unit_tests/wasm_client_tests.rs index 7a84d2593fd..534b7e116dd 100644 --- a/linera-core/src/unit_tests/wasm_client_tests.rs +++ b/linera-core/src/unit_tests/wasm_client_tests.rs @@ -114,25 +114,49 @@ where let contract_blob = HashedBlob::load_from_file(contract_path.clone()).await?; let expected_contract_blob_id = contract_blob.id(); - let (blob_id, _) = publisher + let (blob_id, certificate) = publisher .publish_blob(contract_blob.clone()) .await .unwrap() .unwrap(); assert_eq!(expected_contract_blob_id, blob_id); + assert!(certificate + .value() + .executed_block() + .unwrap() + .outcome + .oracle_records + .iter() + .any(|record| record.responses.contains(&OracleResponse::Blob(blob_id)))); let service_blob = HashedBlob::load_from_file(service_path.clone()).await?; let expected_service_blob_id = service_blob.id(); - let (blob_id, _) = publisher.publish_blob(service_blob).await.unwrap().unwrap(); + let (blob_id, certificate) = publisher.publish_blob(service_blob).await.unwrap().unwrap(); assert_eq!(expected_service_blob_id, blob_id); + assert!(certificate + .value() + .executed_block() + .unwrap() + .outcome + .oracle_records + .iter() + .any(|record| record.responses.contains(&OracleResponse::Blob(blob_id)))); // If I try to upload the contract blob again, I should get the same blob ID - let (blob_id, _) = publisher + let (blob_id, certificate) = publisher .publish_blob(contract_blob) .await .unwrap() .unwrap(); assert_eq!(expected_contract_blob_id, blob_id); + assert!(certificate + .value() + .executed_block() + .unwrap() + .outcome + .oracle_records + .iter() + .any(|record| record.responses.contains(&OracleResponse::Blob(blob_id)))); let (bytecode_id, cert) = publisher .publish_bytecode( diff --git a/linera-core/src/worker.rs b/linera-core/src/worker.rs index 2ca24d85c30..7a27e6e4392 100644 --- a/linera-core/src/worker.rs +++ b/linera-core/src/worker.rs @@ -263,9 +263,6 @@ pub enum WorkerError { ApplicationBytecodesOrBlobsNotFound(Vec, Vec), #[error("The block proposal is invalid: {0}")] InvalidBlockProposal(String), - - #[error("Block blobs missing from oracles: {0:?}")] - MissingBlockBlobsInOracles(Vec), } impl From for WorkerError { diff --git a/linera-execution/src/execution_state_actor.rs b/linera-execution/src/execution_state_actor.rs index 732fa5af2de..9a617b7fda4 100644 --- a/linera-execution/src/execution_state_actor.rs +++ b/linera-execution/src/execution_state_actor.rs @@ -548,8 +548,8 @@ impl Debug for ExecutionRequest { .field("content_type", content_type) .finish_non_exhaustive(), - Request::ReadBlob { blob_id, .. } => formatter - .debug_struct("Request::ReadBlob") + ExecutionRequest::ReadBlob { blob_id, .. } => formatter + .debug_struct("ExecutionRequest::ReadBlob") .field("blob_id", blob_id) .finish_non_exhaustive(), } diff --git a/linera-execution/src/lib.rs b/linera-execution/src/lib.rs index 6789f4b7382..330d3ccfbf9 100644 --- a/linera-execution/src/lib.rs +++ b/linera-execution/src/lib.rs @@ -23,6 +23,7 @@ use std::{fmt, str::FromStr, sync::Arc}; use async_graphql::SimpleObject; use async_trait::async_trait; +use committee::Epoch; use custom_debug_derive::Debug; use dashmap::DashMap; use derive_more::Display; @@ -1017,6 +1018,15 @@ impl std::fmt::Debug for Bytecode { } } +/// 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, + /// Epoch of the `last_used_by` certificate. + pub epoch: Epoch, +} + /// The runtime to use for running the application. #[derive(Clone, Copy, Display)] #[cfg_attr(with_wasm_runtime, derive(Debug, Default))] diff --git a/linera-execution/src/runtime.rs b/linera-execution/src/runtime.rs index 690c895776b..97921849258 100644 --- a/linera-execution/src/runtime.rs +++ b/linera-execution/src/runtime.rs @@ -987,7 +987,7 @@ impl BaseRuntime for SyncRuntimeInternal { } let blob = self .execution_state_sender - .send_request(|callback| Request::ReadBlob { + .send_request(|callback| ExecutionRequest::ReadBlob { blob_id: *blob_id, callback, })? diff --git a/linera-execution/tests/test_execution.rs b/linera-execution/tests/test_execution.rs index d177383e78b..59c583cc42e 100644 --- a/linera-execution/tests/test_execution.rs +++ b/linera-execution/tests/test_execution.rs @@ -10,8 +10,8 @@ use futures::{stream, StreamExt, TryStreamExt}; use linera_base::{ crypto::PublicKey, data_types::{ - Amount, ApplicationPermissions, BlockHeight, HashedBlob, OracleRecord, Resources, - SendMessageRequest, Timestamp, + Amount, ApplicationPermissions, BlockHeight, OracleRecord, Resources, SendMessageRequest, + Timestamp, }, identifiers::{Account, ChainDescription, ChainId, Destination, MessageId, Owner}, ownership::ChainOwnership, @@ -1625,52 +1625,3 @@ async fn test_close_chain() { .unwrap(); assert!(view.system.closed.get()); } - -/// Tests the system API call `read_blob``. -#[tokio::test] -async fn test_failing_read_blob() { - let committee = Committee::make_simple(vec![PublicKey::test_key(0).into()]); - let committees = BTreeMap::from([(Epoch::ZERO, committee)]); - let ownership = ChainOwnership::single(PublicKey::test_key(1)); - let state = SystemExecutionState { - committees: committees.clone(), - ownership: ownership.clone(), - balance: Amount::from_tokens(5), - ..SystemExecutionState::new(Epoch::ZERO, ChainDescription::Root(0), ChainId::root(0)) - }; - let mut view = state.into_view().await; - let mut applications = register_mock_applications(&mut view, 1).await.unwrap(); - let (application_id, application) = applications.next().unwrap(); - - // The application is not authorized to close the chain. - let context = make_operation_context(); - let blob = HashedBlob::test_blob("test"); - let blob_id = blob.id(); - application.expect_call(ExpectedCall::execute_operation( - move |runtime, _context, _operation| { - assert_matches!( - runtime.read_blob(&blob_id), - Err(ExecutionError::MissingRuntimeResponse) - ); - Ok(vec![]) - }, - )); - application.expect_call(ExpectedCall::default_finalize()); - - let mut controller = ResourceController::default(); - let operation = Operation::User { - application_id, - bytes: vec![], - }; - assert_matches!( - view.execute_operation( - context, - Timestamp::from(0), - operation, - None, - &mut controller, - ) - .await, - Err(ExecutionError::BlobNotFoundOnRead(_)) - ); -} diff --git a/linera-sdk/src/service/conversions_from_wit.rs b/linera-sdk/src/service/conversions_from_wit.rs index f4ebc1506d7..bd916501fe4 100644 --- a/linera-sdk/src/service/conversions_from_wit.rs +++ b/linera-sdk/src/service/conversions_from_wit.rs @@ -5,8 +5,8 @@ use linera_base::{ crypto::CryptoHash, - data_types::{Amount, BlockHeight, Timestamp}, - identifiers::{ApplicationId, BytecodeId, ChainId, MessageId, Owner}, + data_types::{Amount, Blob, BlockHeight, HashedBlob, Timestamp}, + identifiers::{ApplicationId, BlobId, BytecodeId, ChainId, MessageId, Owner}, }; use super::wit::service_system_api as wit_system_api; @@ -23,6 +23,27 @@ impl From for Owner { } } +impl From for HashedBlob { + fn from(hashed_blob: wit_system_api::HashedBlob) -> Self { + HashedBlob { + id: hashed_blob.id.into(), + blob: hashed_blob.blob.into(), + } + } +} + +impl From for BlobId { + fn from(blob_id: wit_system_api::BlobId) -> Self { + BlobId(blob_id.inner0.into()) + } +} + +impl From for Blob { + fn from(blob: wit_system_api::Blob) -> Self { + Blob { bytes: blob.bytes } + } +} + impl From for Amount { fn from(balance: wit_system_api::Amount) -> Self { let (lower_half, upper_half) = balance.inner0; diff --git a/linera-sdk/src/service/conversions_to_wit.rs b/linera-sdk/src/service/conversions_to_wit.rs index 7221077c06b..70d503a845c 100644 --- a/linera-sdk/src/service/conversions_to_wit.rs +++ b/linera-sdk/src/service/conversions_to_wit.rs @@ -6,7 +6,7 @@ use linera_base::{ crypto::CryptoHash, data_types::BlockHeight, - identifiers::{ApplicationId, BytecodeId, ChainId, MessageId, Owner}, + identifiers::{ApplicationId, BlobId, BytecodeId, ChainId, MessageId, Owner}, }; use super::wit::service_system_api as wit_system_api; @@ -44,6 +44,14 @@ impl From for wit_system_api::Owner { } } +impl From for wit_system_api::BlobId { + fn from(blob_id: BlobId) -> Self { + wit_system_api::BlobId { + inner0: blob_id.0.into(), + } + } +} + impl From for wit_system_api::BlockHeight { fn from(block_height: BlockHeight) -> Self { wit_system_api::BlockHeight { diff --git a/linera-sdk/src/service/runtime.rs b/linera-sdk/src/service/runtime.rs index b5611036790..57731e6e07e 100644 --- a/linera-sdk/src/service/runtime.rs +++ b/linera-sdk/src/service/runtime.rs @@ -7,8 +7,8 @@ use std::cell::Cell; use linera_base::{ abi::ServiceAbi, - data_types::{Amount, BlockHeight, Timestamp}, - identifiers::{ApplicationId, ChainId, Owner}, + data_types::{Amount, BlockHeight, HashedBlob, Timestamp}, + identifiers::{ApplicationId, BlobId, ChainId, Owner}, }; use super::wit::service_system_api as wit; @@ -144,4 +144,9 @@ where cell.set(Some(value.clone())); value } + + /// Reads a blob with the given `BlobId` from storage. + pub fn read_blob(&mut self, blob_id: BlobId) -> HashedBlob { + wit::read_blob(blob_id.into()).into() + } } diff --git a/linera-storage/src/db_storage.rs b/linera-storage/src/db_storage.rs index 57c970f8e7a..2849316965b 100644 --- a/linera-storage/src/db_storage.rs +++ b/linera-storage/src/db_storage.rs @@ -7,7 +7,7 @@ use async_trait::async_trait; use dashmap::DashMap; use linera_base::{ crypto::CryptoHash, - data_types::{Blob, BlobState, HashedBlob, TimeDelta, Timestamp}, + data_types::{Blob, HashedBlob, TimeDelta, Timestamp}, identifiers::{BlobId, ChainId}, }; use linera_chain::{ @@ -15,7 +15,8 @@ use linera_chain::{ ChainStateView, }; use linera_execution::{ - ExecutionRuntimeConfig, UserApplicationId, UserContractCode, UserServiceCode, WasmRuntime, + committee::Epoch, BlobState, ExecutionRuntimeConfig, UserApplicationId, UserContractCode, + UserServiceCode, WasmRuntime, }; use linera_views::{ batch::Batch, @@ -62,6 +63,17 @@ static CONTAINS_BLOB_COUNTER: Lazy = Lazy::new(|| { .expect("Counter creation should not fail") }); +/// The metric counting how often a blob state is tested for existence from storage +#[cfg(with_metrics)] +static CONTAINS_BLOB_STATE_COUNTER: Lazy = Lazy::new(|| { + prometheus_util::register_int_counter_vec( + "contains_blob_state", + "The metric counting how often a blob state is tested for existence from storage", + &[], + ) + .expect("Counter creation should not fail") +}); + /// The metric counting how often a certificate is tested for existence from storage. #[cfg(with_metrics)] static CONTAINS_CERTIFICATE_COUNTER: Lazy = Lazy::new(|| { @@ -428,6 +440,14 @@ where Ok(test) } + async fn contains_blob_state(&self, blob_id: BlobId) -> Result { + let blob_key = bcs::to_bytes(&BaseKey::BlobStateId(blob_id))?; + let test = self.client.client.contains_key(&blob_key).await?; + #[cfg(with_metrics)] + CONTAINS_BLOB_STATE_COUNTER.with_label_values(&[]).inc(); + Ok(test) + } + async fn read_hashed_certificate_value( &self, hash: CryptoHash, @@ -506,13 +526,35 @@ where Ok(()) } + async fn maybe_write_blob_state( + &self, + blob_id: BlobId, + blob_state: BlobState, + ) -> Result { + let current_blob_state = self.read_blob_state(blob_id).await; + let (should_write, latest_epoch) = match current_blob_state { + Ok(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), + }; + + if should_write { + self.write_blob_state(blob_id, &blob_state).await?; + } + + Ok(latest_epoch) + } + async fn write_blob_state( &self, blob_id: BlobId, - last_used_by: CryptoHash, + blob_state: &BlobState, ) -> Result<(), ViewError> { let mut batch = Batch::new(); - self.add_blob_state_to_batch(blob_id, last_used_by, &mut batch)?; + self.add_blob_state_to_batch(blob_id, blob_state, &mut batch)?; self.write_batch(batch).await?; Ok(()) } @@ -626,11 +668,11 @@ where fn add_blob_state_to_batch( &self, blob_id: BlobId, - last_used_by: CryptoHash, + blob_state: &BlobState, batch: &mut Batch, ) -> Result<(), ViewError> { let blob_state_key = bcs::to_bytes(&BaseKey::BlobStateId(blob_id))?; - batch.put_key_value(blob_state_key.to_vec(), &BlobState { last_used_by })?; + batch.put_key_value(blob_state_key.to_vec(), blob_state)?; Ok(()) } diff --git a/linera-storage/src/lib.rs b/linera-storage/src/lib.rs index ec5c87e16b7..d5122821912 100644 --- a/linera-storage/src/lib.rs +++ b/linera-storage/src/lib.rs @@ -25,7 +25,7 @@ use dashmap::{mapref::entry::Entry, DashMap}; use futures::future; use linera_base::{ crypto::{CryptoHash, PublicKey}, - data_types::{Amount, BlobState, BlockHeight, HashedBlob, Timestamp}, + data_types::{Amount, BlockHeight, HashedBlob, Timestamp}, identifiers::{BlobId, ChainDescription, ChainId, GenericApplicationId}, ownership::ChainOwnership, }; @@ -36,8 +36,9 @@ use linera_chain::{ use linera_execution::{ committee::{Committee, Epoch}, system::SystemChannel, - ChannelSubscription, ExecutionError, ExecutionRuntimeConfig, ExecutionRuntimeContext, - UserApplicationDescription, UserApplicationId, UserContractCode, UserServiceCode, WasmRuntime, + BlobState, ChannelSubscription, ExecutionError, ExecutionRuntimeConfig, + ExecutionRuntimeContext, UserApplicationDescription, UserApplicationId, UserContractCode, + UserServiceCode, WasmRuntime, }; use linera_views::{ common::Context, @@ -96,6 +97,9 @@ pub trait Storage: Sized { /// Tests existence of a blob with the given hash. async fn contains_blob(&self, blob_id: BlobId) -> Result; + /// Tests existence of a blob state with the given hash. + async fn contains_blob_state(&self, blob_id: BlobId) -> Result; + /// Reads the hashed certificate value with the given hash. async fn read_hashed_certificate_value( &self, @@ -128,9 +132,16 @@ pub trait Storage: Sized { async fn write_blob_state( &self, blob_id: BlobId, - last_used_by: CryptoHash, + blob_state: &BlobState, ) -> Result<(), 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; + /// Writes several hashed certificate values. async fn write_hashed_certificate_values( &self,