Add possibility to submit a list of changes to rocksdb

CHANGELOG.md

@@ -6,6 +6,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [Unreleased]
 
+### Added
+- [2619](https://github.com/FuelLabs/fuel-core/pull/2691): Add possibility to submit list of changes to rocksdb.
+
 ### Changed
 - [2630](https://github.com/FuelLabs/fuel-core/pull/2630): Removed some noisy `tracing::info!` logs
 

crates/fuel-core/src/database.rs

@@ -43,6 +43,7 @@ use fuel_core_storage::{
         ConflictPolicy,
         HistoricalView,
         Modifiable,
+        StorageChanges,
         StorageTransaction,
     },
     Error as StorageError,
@@ -364,72 +365,96 @@ where
 
 impl Modifiable for Database<OnChain> {
     fn commit_changes(&mut self, changes: Changes) -> StorageResult<()> {
-        commit_changes_with_height_update(self, changes, |iter| {
-            iter.iter_all_keys::<FuelBlocks>(Some(IterDirection::Reverse))
-                .try_collect()
-        })
+        commit_changes_with_height_update(
+            self,
+            StorageChanges::Changes(changes),
+            |iter| {
+                iter.iter_all_keys::<FuelBlocks>(Some(IterDirection::Reverse))
+                    .try_collect()
+            },
+        )
     }
 }
 
 impl Modifiable for Database<OffChain> {
     fn commit_changes(&mut self, changes: Changes) -> StorageResult<()> {
-        commit_changes_with_height_update(self, changes, |iter| {
-            iter.iter_all::<FuelBlockIdsToHeights>(Some(IterDirection::Reverse))
-                .map(|result| result.map(|(_, height)| height))
-                .try_collect()
-        })
+        commit_changes_with_height_update(
+            self,
+            StorageChanges::Changes(changes),
+            |iter| {
+                iter.iter_all::<FuelBlockIdsToHeights>(Some(IterDirection::Reverse))
+                    .map(|result| result.map(|(_, height)| height))
+                    .try_collect()
+            },
+        )
     }
 }
 
 impl Modifiable for Database<GasPriceDatabase> {
     fn commit_changes(&mut self, changes: Changes) -> StorageResult<()> {
-        commit_changes_with_height_update(self, changes, |iter| {
-            iter.iter_all_keys::<GasPriceMetadata>(Some(IterDirection::Reverse))
-                .try_collect()
-        })
+        commit_changes_with_height_update(
+            self,
+            StorageChanges::Changes(changes),
+            |iter| {
+                iter.iter_all_keys::<GasPriceMetadata>(Some(IterDirection::Reverse))
+                    .try_collect()
+            },
+        )
     }
 }
 
 #[cfg(feature = "relayer")]
 impl Modifiable for Database<Relayer> {
     fn commit_changes(&mut self, changes: Changes) -> StorageResult<()> {
-        commit_changes_with_height_update(self, changes, |iter| {
-            iter.iter_all_keys::<fuel_core_relayer::storage::EventsHistory>(Some(
-                IterDirection::Reverse,
-            ))
-            .try_collect()
-        })
+        commit_changes_with_height_update(
+            self,
+            StorageChanges::Changes(changes),
+            |iter| {
+                iter.iter_all_keys::<fuel_core_relayer::storage::EventsHistory>(Some(
+                    IterDirection::Reverse,
+                ))
+                .try_collect()
+            },
+        )
     }
 }
 
 #[cfg(not(feature = "relayer"))]
 impl Modifiable for Database<Relayer> {
     fn commit_changes(&mut self, changes: Changes) -> StorageResult<()> {
-        commit_changes_with_height_update(self, changes, |_| Ok(vec![]))
+        commit_changes_with_height_update(self, StorageChanges::Changes(changes), |_| {
+            Ok(vec![])
+        })
     }
 }
 
 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, StorageChanges::Changes(changes))
     }
 }
 
 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, StorageChanges::Changes(changes))
     }
 }
 
 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, StorageChanges::Changes(changes))
     }
 }
 
 pub fn commit_changes_with_height_update<Description>(
     database: &mut Database<Description>,
-    changes: Changes,
+    mut changes: StorageChanges,
     heights_lookup: impl Fn(
         &ChangesIterator<Description::Column>,
     ) -> StorageResult<Vec<Description::Height>>,
@@ -492,14 +517,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>>()
@@ -509,14 +534,21 @@ 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) => {
+                let mut changes = core::mem::take(&mut list);
+                changes.push(transaction.into_changes());
+                StorageChanges::ChangesList(changes)
+            }
+        }
     };
 
     // 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

@@ -3181,6 +3181,7 @@ mod tests {
                 IteratorOverTable,
             },
             tables::FuelBlocks,
+            transactional::StorageChanges,
             StorageAsMut,
         };
         use fuel_core_types::{
@@ -3324,6 +3325,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,
@@ -3933,6 +3935,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);
@@ -3975,6 +3978,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

@@ -178,7 +178,7 @@ mod tests {
         },
         tables::Coins,
         transactional::{
-            Changes,
+            StorageChanges,
             StorageTransaction,
         },
         Result as StorageResult,
@@ -570,7 +570,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;
@@ -55,7 +55,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(
@@ -75,7 +75,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!()
     }