Add possibility to submit a list of changes to rocksdb

.changes/added/2619.md

@@ -0,0 +1 @@
+Add possibility to submit list of changes to rocksdb.

crates/database/src/lib.rs

@@ -11,7 +11,10 @@
 #![deny(warnings)]
 #![deny(unused_variables)]
 
-use fuel_core_storage::Error as StorageError;
+use fuel_core_storage::{
+    transactional::ReferenceBytesKey,
+    Error as StorageError,
+};
 use fuel_core_types::services::executor::Error as ExecutorError;
 
 /// The error occurred during work with any of databases.
@@ -84,6 +87,10 @@ pub enum Error {
     #[display(fmt = "Restore error: {}", _0)]
     RestoreError(anyhow::Error),
 
+    #[display(fmt = "During committing of the changes found conflicting \
+        modifications at column {column} for the key {key:?}")]
+    ConflictingChanges { column: u32, key: ReferenceBytesKey },
+
     /// Not related to database error.
     #[from]
     Other(anyhow::Error),

crates/fuel-core/src/database.rs

@@ -44,6 +44,7 @@ use fuel_core_storage::{
         ConflictPolicy,
         HistoricalView,
         Modifiable,
+        StorageChanges,
         StorageTransaction,
     },
     Error as StorageError,
@@ -452,23 +453,23 @@ impl Modifiable for Database<Relayer> {
 
 impl Modifiable for GenesisDatabase<OnChain> {
     fn commit_changes(&mut self, changes: Changes) -> StorageResult<()> {
-        self.data.as_ref().commit_changes(None, changes)
+        self.data.as_ref().commit_changes(None, changes.into())
     }
 }
 
 impl Modifiable for GenesisDatabase<OffChain> {
     fn commit_changes(&mut self, changes: Changes) -> StorageResult<()> {
-        self.data.as_ref().commit_changes(None, changes)
+        self.data.as_ref().commit_changes(None, changes.into())
     }
 }
 
 impl Modifiable for GenesisDatabase<Relayer> {
     fn commit_changes(&mut self, changes: Changes) -> StorageResult<()> {
-        self.data.as_ref().commit_changes(None, changes)
+        self.data.as_ref().commit_changes(None, changes.into())
     }
 }
 
-pub fn commit_changes_with_height_update<Description>(
+pub fn commit_changes_with_height_update<Description, Changes>(
     database: &mut Database<Description>,
     changes: Changes,
     heights_lookup: impl Fn(
@@ -480,8 +481,10 @@ where
     Description::Height: Debug + PartialOrd + DatabaseHeight,
     for<'a> StorageTransaction<&'a &'a mut Database<Description>>:
         StorageMutate<MetadataTable<Description>, Error = StorageError>,
+    Changes: Into<StorageChanges>,
 {
     // Gets the all new heights from the `changes`
+    let mut changes = changes.into();
     let iterator = ChangesIterator::<Description::Column>::new(&changes);
     let new_heights = heights_lookup(&iterator)?;
 
@@ -533,14 +536,14 @@ where
         }
     };
 
-    let updated_changes = if let Some(new_height) = new_height {
+    if let Some(new_height) = new_height {
         // We want to update the metadata table to include a new height.
-        // For that, we are building a new storage transaction around `changes`.
-        // Modifying this transaction will include all required updates into the `changes`.
+        // For that, we are building a new storage transaction.
+        // We get the changes from the database and add to our list of changes.
         let mut transaction = StorageTransaction::transaction(
             &database,
             ConflictPolicy::Overwrite,
-            changes,
+            Default::default(),
         );
         let maybe_current_metadata = transaction
             .storage_as_mut::<MetadataTable<Description>>()
@@ -550,14 +553,20 @@ where
             .storage_as_mut::<MetadataTable<Description>>()
             .insert(&(), &metadata)?;
 
-        transaction.into_changes()
-    } else {
-        changes
+        changes = match changes {
+            StorageChanges::Changes(c) => {
+                StorageChanges::ChangesList(vec![c, transaction.into_changes()])
+            }
+            StorageChanges::ChangesList(mut list) => {
+                list.push(transaction.into_changes());
+                StorageChanges::ChangesList(list)
+            }
+        }
     };
 
     // Atomically commit the changes to the database, and to the mutex-protected field.
     let mut guard = database.stage.height.lock();
-    database.data.commit_changes(new_height, updated_changes)?;
+    database.data.commit_changes(new_height, changes)?;
 
     // Update the block height
     if let Some(new_height) = new_height {

crates/fuel-core/src/executor.rs

@@ -3182,6 +3182,7 @@ mod tests {
                 IteratorOverTable,
             },
             tables::FuelBlocks,
+            transactional::StorageChanges,
             StorageAsMut,
         };
         use fuel_core_types::{
@@ -3325,6 +3326,7 @@ mod tests {
             let (result, changes) = producer.produce_without_commit(block.into())?.into();
 
             // Then
+            let changes = StorageChanges::Changes(changes);
             let view = ChangesIterator::<Column>::new(&changes);
             assert_eq!(
                 view.iter_all::<Messages>(None).count() as u64,
@@ -3934,6 +3936,7 @@ mod tests {
                 .into();
 
             // Then
+            let changes = StorageChanges::Changes(changes);
             let view = ChangesIterator::<Column>::new(&changes);
             assert!(result.skipped_transactions.is_empty());
             assert_eq!(view.iter_all::<Messages>(None).count() as u64, 0);
@@ -3976,6 +3979,7 @@ mod tests {
                 .into();
 
             // Then
+            let changes = StorageChanges::Changes(changes);
             let view = ChangesIterator::<Column>::new(&changes);
             assert!(result.skipped_transactions.is_empty());
             assert_eq!(view.iter_all::<Messages>(None).count() as u64, 0);

crates/fuel-core/src/service.rs

@@ -32,6 +32,7 @@ use fuel_core_storage::{
     transactional::{
         AtomicView,
         ReadTransaction,
+        StorageChanges,
     },
     IsNotFound,
     StorageAsMut,
@@ -311,11 +312,10 @@ impl FuelService {
                     &Consensus::Genesis(initialized_genesis),
                 )?;
 
-            self.shared
-                .database
-                .on_chain()
-                .data
-                .commit_changes(Some(genesis_block_height), database_tx.into_changes())?;
+            self.shared.database.on_chain().data.commit_changes(
+                Some(genesis_block_height),
+                StorageChanges::Changes(database_tx.into_changes()),
+            )?;
         }
 
         Ok(())

crates/fuel-core/src/service/genesis/importer/import_task.rs

@@ -182,7 +182,7 @@ mod tests {
         },
         tables::Coins,
         transactional::{
-            Changes,
+            StorageChanges,
             StorageTransaction,
         },
         Result as StorageResult,
@@ -574,7 +574,7 @@ mod tests {
         fn commit_changes(
             &self,
             _: Option<BlockHeight>,
-            _: Changes,
+            _: StorageChanges,
         ) -> StorageResult<()> {
             Err(anyhow::anyhow!("I refuse to work!").into())
         }

crates/fuel-core/src/state.rs

@@ -12,7 +12,7 @@ use fuel_core_storage::{
         IterableStore,
     },
     kv_store::StorageColumn,
-    transactional::Changes,
+    transactional::StorageChanges,
     Result as StorageResult,
 };
 use std::fmt::Debug;
@@ -57,7 +57,7 @@ pub trait TransactableStorage<Height>: IterableStore + Debug + Send + Sync {
     fn commit_changes(
         &self,
         height: Option<Height>,
-        changes: Changes,
+        changes: StorageChanges,
     ) -> StorageResult<()>;
 
     fn view_at_height(
@@ -77,7 +77,7 @@ impl<Height, S> TransactableStorage<Height>
 where
     S: IterableStore + Debug + Send + Sync,
 {
-    fn commit_changes(&self, _: Option<Height>, _: Changes) -> StorageResult<()> {
+    fn commit_changes(&self, _: Option<Height>, _: StorageChanges) -> StorageResult<()> {
         unimplemented!()
     }