fix(compression_service): post merge reviews

crates/fuel-core/src/database.rs

@@ -825,11 +825,6 @@ mod tests {
         };
         use fuel_core_storage::transactional::WriteTransaction;
 
-        #[test]
-        fn column_keys_not_exceed_count_test() {
-            column_keys_not_exceed_count::<OffChain>();
-        }
-
         #[test]
         fn database_advances_with_a_new_block() {
             // Given

crates/fuel-core/src/database/database_description/compression.rs

@@ -19,7 +19,7 @@ impl DatabaseDescription for CompressionDatabase {
     }
 
     fn metadata_column() -> Self::Column {
-        Self::Column::MerkleMetadataColumn
+        Self::Column::Metadata
     }
 
     fn prefix(_column: &Self::Column) -> Option<usize> {

crates/fuel-core/src/graphql_api/storage.rs

@@ -96,13 +96,13 @@ pub enum Column {
     /// See [`SpentMessages`](messages::SpentMessages)
     SpentMessages = 13,
     /// Coin balances per account and asset.
-    CoinBalances = 14,
+    CoinBalances = 23,
     /// Message balances per account.
-    MessageBalances = 15,
+    MessageBalances = 24,
     /// See [`AssetsInfo`](assets::AssetsInfo)
-    AssetsInfo = 16,
+    AssetsInfo = 25,
     /// Index of the coins that are available to spend.
-    CoinsToSpend = 17,
+    CoinsToSpend = 26,
 }
 
 impl Column {

crates/fuel-core/src/service/adapters/compression_adapters.rs

@@ -12,20 +12,11 @@ use fuel_core_compression_service::{
         configuration,
     },
 };
-use fuel_core_services::stream::IntoBoxStream;
+use fuel_core_types::services::block_importer::SharedImportResult;
 
 impl block_source::BlockSource for BlockImporterAdapter {
-    fn subscribe(
-        &self,
-    ) -> fuel_core_services::stream::BoxStream<block_source::BlockWithMetadata> {
-        use futures::StreamExt;
+    fn subscribe(&self) -> fuel_core_services::stream::BoxStream<SharedImportResult> {
         self.events_shared_result()
-            .map(|result| {
-                let sealed_block = result.sealed_block.clone();
-                let events = result.events.clone();
-                block_source::BlockWithMetadata::new(sealed_block.entity, events)
-            })
-            .into_boxed()
     }
 }
 

crates/fuel-core/src/state/in_memory/memory_store.rs

@@ -63,9 +63,15 @@ where
     Description: DatabaseDescription,
 {
     fn default() -> Self {
-        use strum::EnumCount;
+        use enum_iterator::all;
+
+        let largest_column_idx = all::<Description::Column>()
+            .map(|column| column.as_usize())
+            .max()
+            .expect("there should be atleast 1 column in the storage");
+
         Self {
-            inner: (0..Description::Column::COUNT)
+            inner: (0..=largest_column_idx)
                 .map(|_| Mutex::new(BTreeMap::new()))
                 .collect(),
             _marker: Default::default(),

crates/services/compression/src/ports/block_source.rs

@@ -1,35 +1,34 @@
 use fuel_core_services::stream::BoxStream;
-
-/// Block with metadata needed for compression
-#[cfg_attr(feature = "test-helpers", derive(Default))]
-#[derive(Clone, Debug)]
-pub struct BlockWithMetadata {
-    block: fuel_core_types::blockchain::block::Block,
-    // todo: perhaps slice with lifetime? (https://github.com/FuelLabs/fuel-core/issues/2870)
-    events: Vec<fuel_core_types::services::executor::Event>,
-}
+pub(crate) use fuel_core_types::services::block_importer::SharedImportResult as BlockWithMetadata;
 
 pub(crate) type BlockHeight = u32;
 
-impl BlockWithMetadata {
-    /// Create a new block with metadata
-    pub fn new(
-        block: fuel_core_types::blockchain::block::Block,
-        events: Vec<fuel_core_types::services::executor::Event>,
-    ) -> Self {
-        Self { block, events }
+pub(crate) trait BlockWithMetadataExt {
+    fn height(&self) -> &BlockHeight;
+    fn events(&self) -> &[fuel_core_types::services::executor::Event];
+    fn block(&self) -> &fuel_core_types::blockchain::block::Block;
+    #[cfg(test)]
+    fn default() -> Self;
+}
+
+impl BlockWithMetadataExt for BlockWithMetadata {
+    fn events(&self) -> &[fuel_core_types::services::executor::Event] {
+        self.events.as_ref()
     }
 
-    pub(crate) fn height(&self) -> BlockHeight {
-        (*self.block.header().height()).into()
+    fn height(&self) -> &BlockHeight {
+        self.block().header().height()
     }
 
-    pub(crate) fn events(&self) -> &[fuel_core_types::services::executor::Event] {
-        &self.events
+    fn block(&self) -> &fuel_core_types::blockchain::block::Block {
+        &self.sealed_block.entity
     }
 
-    pub(crate) fn block(&self) -> &fuel_core_types::blockchain::block::Block {
-        &self.block
+    #[cfg(test)]
+    fn default() -> Self {
+        use fuel_core_types::services::block_importer::ImportResult;
+
+        std::sync::Arc::new(ImportResult::default().wrap())
     }
 }
 
@@ -39,5 +38,5 @@ pub type BlockStream = BoxStream<BlockWithMetadata>;
 /// Port for L2 blocks source
 pub trait BlockSource {
     /// Should provide a stream of blocks with metadata
-    fn subscribe(&self) -> BoxStream<BlockWithMetadata>;
+    fn subscribe(&self) -> BlockStream;
 }

crates/services/compression/src/service.rs

@@ -1,7 +1,11 @@
 use crate::{
     config::CompressionConfig,
     ports::{
-        block_source::BlockSource,
+        block_source::{
+            BlockSource,
+            BlockWithMetadata,
+            BlockWithMetadataExt,
+        },
         compression_storage::{
             CompressionStorage,
             WriteCompressedBlock,
@@ -71,7 +75,7 @@ where
 {
     fn compress_block(
         &mut self,
-        block_with_metadata: &crate::ports::block_source::BlockWithMetadata,
+        block_with_metadata: &BlockWithMetadata,
     ) -> crate::Result<()> {
         let mut storage_tx = self.storage.write_transaction();
 
@@ -92,7 +96,7 @@ where
         .map_err(crate::errors::CompressionError::FailedToCompressBlock)?;
 
         storage_tx
-            .write_compressed_block(&block_with_metadata.height(), &compressed_block)?;
+            .write_compressed_block(block_with_metadata.height(), &compressed_block)?;
 
         storage_tx
             .commit()
@@ -103,26 +107,20 @@ where
 
     fn handle_new_block(
         &mut self,
-        block_with_metadata: &crate::ports::block_source::BlockWithMetadata,
+        block_with_metadata: &BlockWithMetadata,
     ) -> crate::Result<()> {
         // set the status to not synced
-        if let Err(err) = self
-            .sync_notifier
+        self.sync_notifier
             .send(crate::sync_state::SyncState::NotSynced)
-        {
-            tracing::error!("Failed to set sync status to not synced: {:?}", err);
-        }
+            .ok();
         // compress the block
         self.compress_block(block_with_metadata)?;
         // set the status to synced
-        if let Err(err) = self
-            .sync_notifier
+        self.sync_notifier
             .send(crate::sync_state::SyncState::Synced(
-                block_with_metadata.height(),
+                *block_with_metadata.height(),
             ))
-        {
-            tracing::error!("Failed to set sync status to synced: {:?}", err);
-        }
+            .ok();
         Ok(())
     }
 }
@@ -203,7 +201,7 @@ where
                         fuel_core_services::TaskNextAction::Stop
                     }
                     Some(block_with_metadata) => {
-                        tracing::debug!("Got new block: {:?}", block_with_metadata.height());
+                        tracing::debug!("Got new block: {:?}", &block_with_metadata.height());
                         if let Err(e) = self.handle_new_block(&block_with_metadata) {
                             tracing::error!("Error handling new block: {:?}", e);
                             return fuel_core_services::TaskNextAction::ErrorContinue(anyhow::anyhow!(e));
@@ -245,7 +243,10 @@ where
 mod tests {
     use super::*;
     use crate::{
-        ports::block_source::BlockWithMetadata,
+        ports::block_source::{
+            BlockWithMetadata,
+            BlockWithMetadataExt,
+        },
         storage,
     };
     use fuel_core_services::{

crates/storage/src/merkle/column.rs

@@ -22,12 +22,14 @@ use alloc::{
     strum_macros::IntoStaticStr,
 )]
 pub enum MerkleizedColumn<TC> {
+    /// The Metadata column (non-merkleized)
+    Metadata,
+    /// The merkleized metadata column.
+    MerkleMetadataColumn,
     /// The column with the specific data.
     TableColumn(TC),
     /// The column with the merkle tree nodes.
     MerkleDataColumn(TC),
-    /// The merkle metadata column.
-    MerkleMetadataColumn,
 }
 
 impl<TC> strum::EnumCount for MerkleizedColumn<TC>
@@ -36,8 +38,8 @@ where
 {
     /// The total count of variants in the enum.
     /// Since we have two columns for each table column and one for the merkle data,
-    /// we have to multiply the count of the table columns by 2 and add one for the merkle metadata.
-    const COUNT: usize = TC::COUNT * 2 + 1;
+    /// we have to multiply the count of the table columns by 2 and add 2 for the metadata tables.
+    const COUNT: usize = TC::COUNT * 2 + 2;
 }
 
 /// The trait to convert the column to the `u32`.
@@ -56,9 +58,12 @@ where
     /// The start of the merkle data columns.
     pub const MERKLE_DATA_COLUMNS_START: u32 = u16::MAX as u32;
 
+    /// The metadata column (non-merkleized)
+    pub const METADATA_COLUMN: u32 = 0;
+
     /// The merkle metadata column
-    // we set it to 0, since any future column updates will be added after this column
-    pub const MERKLE_METADATA_COLUMN: u32 = 0;
+    // we set it to 1, since any future column updates will be added after this column
+    pub const MERKLE_METADATA_COLUMN: u32 = 1;
 
     #[inline]
     fn with_metadata_offset(column: u32) -> u32 {
@@ -68,11 +73,12 @@ where
     /// Returns the `u32` representation of the `Column`.
     pub fn as_u32(&self) -> u32 {
         match self {
+            Self::Metadata => Self::METADATA_COLUMN,
+            Self::MerkleMetadataColumn => Self::MERKLE_METADATA_COLUMN,
             Self::TableColumn(column) => Self::with_metadata_offset(column.as_u32()),
             Self::MerkleDataColumn(column) => Self::with_metadata_offset(
                 Self::MERKLE_DATA_COLUMNS_START.wrapping_add(column.as_u32()),
             ),
-            Self::MerkleMetadataColumn => Self::MERKLE_METADATA_COLUMN,
         }
     }
 }
@@ -83,13 +89,14 @@ where
 {
     fn name(&self) -> String {
         match self {
+            Self::Metadata => "Metadata".to_string(),
+            MerkleizedColumn::MerkleMetadataColumn => "MerkleMetadata".to_string(),
             Self::TableColumn(column) => {
                 format!("{:?}", column)
             }
             Self::MerkleDataColumn(column) => {
                 format!("Merkle{:?}", column)
             }
-            MerkleizedColumn::MerkleMetadataColumn => "MerkleMetadata".to_string(),
         }
     }