Moved StorageTransaction to the fuel-core-storage crate

CHANGELOG.md

@@ -6,6 +6,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [Unreleased]
 
+Description of the upcoming release here.
+
 ### Added
 
 - [#1740](https://github.com/FuelLabs/fuel-core/pull/1740): Remove optional fields from genesis configs
@@ -15,6 +17,30 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 ### Changed
 
 #### Breaking
+- [#1694](https://github.com/FuelLabs/fuel-core/pull/1694): The change moves the database transaction logic from the `fuel-core` to the `fuel-core-storage` level. The corresponding [issue](https://github.com/FuelLabs/fuel-core/issues/1589) described the reason behind it.
+
+    ## Technical details of implementation
+
+    - The change splits the `KeyValueStore` into `KeyValueInspect` and `KeyValueMutate`, as well the `Blueprint` into `BlueprintInspect` and `BlueprintMutate`. It allows requiring less restricted constraints for any read-related operations.
+
+    - One of the main ideas of the change is to allow for the actual storage only to implement `KeyValueInspect` and `Modifiable` without the `KeyValueMutate`. It simplifies work with the databases and provides a safe way of interacting with them (Modification into the database can only go through the `Modifiable::commit_changes`). This feature is used to [track the height](https://github.com/FuelLabs/fuel-core/pull/1694/files#diff-c95a3d57a39feac7c8c2f3b193a24eec39e794413adc741df36450f9a4539898) of each database during commits and even limit how commits are done, providing additional safety. This part of the change was done as a [separate commit](https://github.com/FuelLabs/fuel-core/pull/1694/commits/7b1141ac838568e3590f09dd420cb24a6946bd32).
+
+    - The `StorageTransaction` is a `StructuredStorage` that uses `InMemoryTransaction` inside to accumulate modifications. Only `InMemoryTransaction` has a real implementation of the `KeyValueMutate`(Other types only implement it in tests).
+
+    - The implementation of the `Modifiable` for the `Database` contains a business logic that provides additional safety but limits the usage of the database. The `Database` now tracks its height and is responsible for its updates. In the `commit_changes` function, it analyzes the changes that were done and tries to find a new height(For example, in the case of the `OnChain` database, we are looking for a new `Block` in the `FuelBlocks` table).
+
+    - As was planned in the issue, now the executor has full control over how commits to the storage are done.
+
+    - All mutation methods now require `&mut self` - exclusive ownership over the object to be able to write into it. It almost negates the chance of concurrent modification of the storage, but it is still possible since the `Database` implements the `Clone` trait. To be sure that we don't corrupt the state of the database, the `commit_changes` function implements additional safety checks to be sure that we commit updates per each height only once time.
+
+    - Side changes:
+      - The `drop` function was moved from `Database` to `RocksDB` as a preparation for the state rewind since the read view should also keep the drop function until it is destroyed.
+      - The `StatisticTable` table lives in the off-chain worker.
+      - Removed duplication of the `Database` from the `dap::ConcreteStorage` since it is already available from the VM.
+      - The executor return only produced `Changes` instead of the storage transaction, which simplifies the interaction between modules and port definition.
+      - The logic related to the iteration over the storage is moved to the `fuel-core-storage` crate and is now reusable. It provides an `interator` method that duplicates the logic from `MemoryStore` on iterating over the `BTreeMap` and methods like `iter_all`, `iter_all_by_prefix`, etc. It was done in a separate revivable [commit](https://github.com/FuelLabs/fuel-core/pull/1694/commits/5b9bd78320e6f36d0650ec05698f12f7d1b3c7c9).
+      - The `MemoryTransactionView` is fully replaced by the `StorageTransactionInner`.
+      - Removed `flush` method from the `Database` since it is not needed after https://github.com/FuelLabs/fuel-core/pull/1664.
 
 - [#1693](https://github.com/FuelLabs/fuel-core/pull/1693): The change separates the initial chain state from the chain config and stores them in separate files when generating a snapshot. The state snapshot can be generated in a new format where parquet is used for compression and indexing while postcard is used for encoding. This enables importing in a stream like fashion which reduces memory requirements. Json encoding is still supported to enable easy manual setup. However, parquet is prefered for large state files.
 

benches/benches/block_target_gas.rs

@@ -17,6 +17,10 @@ use ed25519_dalek::Signer;
 use ethnum::U256;
 use fuel_core::{
     combined_database::CombinedDatabase,
+    database::{
+        balances::BalancesInitializer,
+        state::StateInitializer,
+    },
     service::{
         config::Trigger,
         Config,
@@ -256,7 +260,7 @@ fn service_with_many_contracts(
         .build()
         .unwrap();
     let _drop = rt.enter();
-    let mut database = Database::rocksdb();
+    let mut database = Database::rocksdb_temp();
     let mut config = Config::local_node();
     config
         .chain_config

benches/benches/state.rs

@@ -5,8 +5,19 @@ use criterion::{
     BenchmarkGroup,
     Criterion,
 };
-use fuel_core::database::Database;
-use fuel_core_storage::vm_storage::VmStorage;
+use fuel_core::database::{
+    database_description::on_chain::OnChain,
+    state::StateInitializer,
+    Database,
+};
+use fuel_core_storage::{
+    transactional::{
+        IntoTransaction,
+        ReadTransaction,
+        WriteTransaction,
+    },
+    vm_storage::VmStorage,
+};
 use fuel_core_types::{
     blockchain::header::GeneratedConsensusFields,
     fuel_tx::Bytes32,
@@ -28,7 +39,10 @@ use std::{
 #[global_allocator]
 static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
 
-fn setup(db: &mut Database, contract: &ContractId, n: usize) {
+fn setup<D>(db: &mut D, contract: &ContractId, n: usize)
+where
+    D: StateInitializer,
+{
     let mut rng_keys = thread_rng();
     let gen_keys = || -> Bytes32 { rng_keys.gen() };
     let state_keys = iter::repeat_with(gen_keys).take(n);
@@ -51,16 +65,16 @@ fn insert_state_single_contract_database(c: &mut Criterion) {
 
     let mut bench_state = |group: &mut BenchmarkGroup<WallTime>, name: &str, n: usize| {
         group.bench_function(name, |b| {
-            let mut db = VmStorage::default();
+            let mut db = Database::<OnChain>::default();
             let contract: ContractId = rng.gen();
-            setup(db.database_mut(), &contract, n);
-            let outer = db.database_mut().transaction();
+            setup(&mut db, &contract, n);
+            let outer = db.write_transaction();
             b.iter_custom(|iters| {
                 let mut elapsed_time = Duration::default();
                 for _ in 0..iters {
-                    let mut inner = outer.transaction();
+                    let inner = outer.read_transaction();
                     let mut inner_db = VmStorage::new::<GeneratedConsensusFields>(
-                        inner.as_mut().clone(),
+                        inner,
                         &Default::default(),
                         Default::default(),
                     );
@@ -112,16 +126,16 @@ fn insert_state_single_contract_transaction(c: &mut Criterion) {
 
     let mut bench_state = |group: &mut BenchmarkGroup<WallTime>, name: &str, n: usize| {
         group.bench_function(name, |b| {
-            let mut db = VmStorage::<Database>::default();
+            let db = Database::<OnChain>::default();
             let contract: ContractId = rng.gen();
-            let mut outer = db.database_mut().transaction();
-            setup(outer.as_mut(), &contract, n);
+            let mut outer = db.into_transaction();
+            setup(&mut outer, &contract, n);
             b.iter_custom(|iters| {
                 let mut elapsed_time = Duration::default();
                 for _ in 0..iters {
-                    let mut inner = outer.transaction();
+                    let inner = outer.read_transaction();
                     let mut inner_db = VmStorage::new::<GeneratedConsensusFields>(
-                        inner.as_mut().clone(),
+                        inner,
                         &Default::default(),
                         Default::default(),
                     );
@@ -173,19 +187,19 @@ fn insert_state_multiple_contracts_database(c: &mut Criterion) {
 
     let mut bench_state = |group: &mut BenchmarkGroup<WallTime>, name: &str, n: usize| {
         group.bench_function(name, |b| {
-            let mut db = VmStorage::<Database>::default();
+            let mut db = Database::<OnChain>::default();
             for _ in 0..n {
                 let contract: ContractId = rng.gen();
-                setup(db.database_mut(), &contract, 1);
+                setup(&mut db, &contract, 1);
             }
-            let outer = db.database_mut().transaction();
+            let outer = db.into_transaction();
             b.iter_custom(|iters| {
                 let mut elapsed_time = Duration::default();
                 let contract: ContractId = rng.gen();
                 for _ in 0..iters {
-                    let mut inner = outer.transaction();
+                    let inner = outer.read_transaction();
                     let mut inner_db = VmStorage::new::<GeneratedConsensusFields>(
-                        inner.as_mut().clone(),
+                        inner,
                         &Default::default(),
                         Default::default(),
                     );
@@ -237,19 +251,19 @@ fn insert_state_multiple_contracts_transaction(c: &mut Criterion) {
 
     let mut bench_state = |group: &mut BenchmarkGroup<WallTime>, name: &str, n: usize| {
         group.bench_function(name, |b| {
-            let mut db = VmStorage::<Database>::default();
-            let mut outer = db.database_mut().transaction();
+            let db = Database::<OnChain>::default();
+            let mut outer = db.into_transaction();
             for _ in 0..n {
                 let contract: ContractId = rng.gen();
-                setup(outer.as_mut(), &contract, 1);
+                setup(&mut outer, &contract, 1);
             }
             b.iter_custom(|iters| {
                 let mut elapsed_time = Duration::default();
                 let contract: ContractId = rng.gen();
                 for _ in 0..iters {
-                    let mut inner = outer.transaction();
+                    let inner = outer.read_transaction();
                     let mut inner_db = VmStorage::new::<GeneratedConsensusFields>(
-                        inner.as_mut().clone(),
+                        inner,
                         &Default::default(),
                         Default::default(),
                     );

benches/benches/vm.rs

@@ -11,10 +11,12 @@ use criterion::{
     BenchmarkGroup,
     Criterion,
 };
+use std::sync::Arc;
 
 use crate::vm_initialization::vm_initialization;
 use contract::*;
 use fuel_core_benches::*;
+use fuel_core_storage::transactional::IntoTransaction;
 use fuel_core_types::fuel_asm::Instruction;
 use vm_set::*;
 
@@ -37,15 +39,19 @@ where
 
             let clock = quanta::Clock::new();
 
+            let original_db = vm.as_mut().database_mut().clone();
+            // During block production/validation for each state, which may affect the state of the database,
+            // we create a new storage transaction. The code here simulates the same behavior to have
+            // the same nesting level and the same performance.
+            let block_database_tx = original_db.clone().into_transaction();
+            let relayer_database_tx = block_database_tx.into_transaction();
+            let thread_database_tx = relayer_database_tx.into_transaction();
+            let tx_database_tx = thread_database_tx.into_transaction();
+            let database = Database::new(Arc::new(tx_database_tx));
+            *vm.as_mut().database_mut() = database.into_transaction();
+
             let mut total = core::time::Duration::ZERO;
             for _ in 0..iters {
-                let original_db = vm.as_mut().database_mut().clone();
-                // Simulates the block production/validation with three levels of database transaction.
-                let block_database_tx = original_db.transaction().as_ref().clone();
-                let tx_database_tx = block_database_tx.transaction().as_ref().clone();
-                let vm_tx_database_tx = tx_database_tx.transaction().as_ref().clone();
-                *vm.as_mut().database_mut() = vm_tx_database_tx;
-
                 let start = black_box(clock.raw());
                 match instruction {
                     Instruction::CALL(call) => {
@@ -60,9 +66,10 @@ where
                 let end = black_box(clock.raw());
                 total += clock.delta(start, end);
                 vm.reset_vm_state(diff);
-                // restore original db
-                *vm.as_mut().database_mut() = original_db;
+                // Reset database changes.
+                vm.as_mut().database_mut().reset_changes();
             }
+            *vm.as_mut().database_mut() = original_db;
             total
         })
     });

benches/benches/vm_set/blockchain.rs

@@ -12,6 +12,11 @@ use criterion::{
     Throughput,
 };
 use fuel_core::{
+    database::{
+        balances::BalancesInitializer,
+        database_description::on_chain::OnChain,
+        state::StateInitializer,
+    },
     service::Config,
     state::rocks_db::{
         RocksDb,
@@ -21,6 +26,10 @@ use fuel_core::{
 use fuel_core_benches::*;
 use fuel_core_storage::{
     tables::FuelBlocks,
+    transactional::{
+        IntoTransaction,
+        StorageTransaction,
+    },
     vm_storage::{
         IncreaseStorageKey,
         VmStorage,
@@ -60,7 +69,8 @@ impl BenchDb {
     fn new(contract_id: &ContractId) -> anyhow::Result<Self> {
         let tmp_dir = ShallowTempDir::new();
 
-        let db = Arc::new(RocksDb::default_open(tmp_dir.path(), None).unwrap());
+        let db =
+            Arc::new(RocksDb::<OnChain>::default_open(tmp_dir.path(), None).unwrap());
         let mut storage_key = primitive_types::U256::zero();
         let mut key_bytes = Bytes32::zeroed();
 
@@ -103,7 +113,6 @@ impl BenchDb {
                 &block.compress(&config.chain_config.consensus_parameters.chain_id),
             )
             .unwrap();
-        database.clone().flush()?;
 
         Ok(Self {
             _tmp_dir: tmp_dir,
@@ -112,14 +121,18 @@ impl BenchDb {
     }
 
     /// Creates a `VmDatabase` instance.
-    fn to_vm_database(&self) -> VmStorage<Database> {
+    fn to_vm_database(&self) -> VmStorage<StorageTransaction<Database>> {
         let header = ConsensusHeader {
             prev_root: Default::default(),
             height: 1.into(),
             time: Tai64::UNIX_EPOCH,
             generated: (),
         };
-        VmStorage::new(self.db.clone(), &header, ContractId::zeroed())
+        VmStorage::new(
+            self.db.clone().into_transaction(),
+            &header,
+            ContractId::zeroed(),
+        )
     }
 }
 

benches/src/lib.rs

@@ -31,12 +31,13 @@ use fuel_core_types::{
     },
 };
 
+use fuel_core_storage::transactional::StorageTransaction;
 pub use rand::Rng;
 use std::iter;
 
 const LARGE_GAS_LIMIT: u64 = u64::MAX - 1001;
 
-fn new_db() -> VmStorage<Database> {
+fn new_db() -> VmStorage<StorageTransaction<Database>> {
     // when rocksdb is enabled, this creates a new db instance with a temporary path
     VmStorage::default()
 }
@@ -89,7 +90,7 @@ pub struct VmBench {
     pub inputs: Vec<Input>,
     pub outputs: Vec<Output>,
     pub witnesses: Vec<Witness>,
-    pub db: Option<VmStorage<Database>>,
+    pub db: Option<VmStorage<StorageTransaction<Database>>>,
     pub instruction: Instruction,
     pub prepare_call: Option<PrepareCall>,
     pub dummy_contract: Option<ContractId>,
@@ -100,7 +101,7 @@ pub struct VmBench {
 
 #[derive(Debug, Clone)]
 pub struct VmBenchPrepared {
-    pub vm: Interpreter<VmStorage<Database>, Script>,
+    pub vm: Interpreter<VmStorage<StorageTransaction<Database>>, Script>,
     pub instruction: Instruction,
     pub diff: diff::Diff<diff::InitialVmState>,
 }
@@ -148,7 +149,7 @@ impl VmBench {
 
     pub fn contract_using_db<R>(
         rng: &mut R,
-        mut db: VmStorage<Database>,
+        mut db: VmStorage<StorageTransaction<Database>>,
         instruction: Instruction,
     ) -> anyhow::Result<Self>
     where
@@ -207,7 +208,7 @@ impl VmBench {
             .with_prepare_call(prepare_call))
     }
 
-    pub fn with_db(mut self, db: VmStorage<Database>) -> Self {
+    pub fn with_db(mut self, db: VmStorage<StorageTransaction<Database>>) -> Self {
         self.db.replace(db);
         self
     }
@@ -468,8 +469,6 @@ impl TryFrom<VmBench> for VmBenchPrepared {
 
         let start_vm = vm.clone();
         let original_db = vm.as_mut().database_mut().clone();
-        let database_tx = original_db.transaction().as_ref().clone();
-        *vm.as_mut().database_mut() = database_tx;
         let mut vm = vm.add_recording();
         match instruction {
             Instruction::CALL(call) => {

bin/fuel-core/src/cli/snapshot.rs

@@ -240,7 +240,7 @@ fn load_chain_config(
 }
 
 fn open_db(path: &Path) -> anyhow::Result<Database> {
-    Database::<OnChain>::open(path, None)
+    Database::<OnChain>::open_rocksdb(path, None)
         .map_err(Into::<anyhow::Error>::into)
         .context(format!("failed to open database at path {path:?}",))
 }