chore(compression): merkleized registration table (#2959)

This pull request introduces a new `Registrations` table to the
compression service, enabling the tracking and merkleization of
registrations for compressed blocks. It includes updates to the database
schema, error handling, and tests to support this new functionality.
Below is a breakdown of the most important changes:

### Addition of `Registrations` Table

* Introduced the `Registrations` table in the compression database,
which maps block heights to registrations made for compressing those
blocks. This table is merkleized for fault-proofing.
(`.changes/added/2959.md`,
`crates/services/compression/src/storage/registrations.rs`)
[[1]](diffhunk://#diff-473ce0bca28d3684eac4b5a9548eef03dd51630a607890106e3bd85aa0da8101R1)
[[2]](diffhunk://#diff-158a54491feb11d974d50f05aaef46446007139e3708e63befe0432312dbd189R1-R54)
* Added a type alias `Registrations` for the merkleized `Registrations`
table in `crates/services/compression/src/storage.rs`.

### Integration with Compression Service

* Updated the `CompressionStorage` implementation to insert data into
the `Registrations` table when the `fault-proving` feature is enabled.
(`crates/services/compression/src/ports/compression_storage.rs`)
* Added a new test to verify that the `Registrations` table is correctly
populated and merkleized during compression.
(`crates/services/compression/src/service.rs`)

### Error Handling Updates

* Added a new `CompressionError` variant, `FailedToWriteRegistrations`,
to handle errors related to writing to the `Registrations` table.
(`crates/services/compression/src/errors.rs`)

### Schema and Column Updates

* Added a new column, `Registrations`, to the `CompressionColumn` enum
for tracking registrations per table.
(`crates/services/compression/src/storage/column.rs`)
* Enabled the `registrations` module in the storage system when the
`fault-proving` feature is active.
(`crates/services/compression/src/storage.rs`)

---------

Co-authored-by: Green Baneling <[email protected]>

Author: rymnc

.changes/added/2959.md

@@ -0,0 +1 @@
+Add `Registrations` table to the compression database. This is merkleized.

crates/services/compression/src/errors.rs

@@ -12,6 +12,9 @@ pub enum CompressionError {
     /// Failed to read compressed block from storage
     #[error("failed to write compressed block to storage: `{0}`")]
     FailedToWriteCompressedBlock(StorageError),
+    /// Failed to write registrations to storage
+    #[error("failed to write registrations to storage: `{0}`")]
+    FailedToWriteRegistrations(StorageError),
     /// Failed to get the size of compressed block
     #[error("failed to get size of compressed block: `{0}`")]
     FailedToGetCompressedBlockSize(StorageError),

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

@@ -1,7 +1,9 @@
 use crate::{
     errors::CompressionError,
-    storage,
-    storage::CompressedBlocks,
+    storage::{
+        self,
+        CompressedBlocks,
+    },
 };
 use fuel_core_storage::{
     self,
@@ -65,6 +67,16 @@ where
             .insert(&height, compressed_block)
             .map_err(CompressionError::FailedToWriteCompressedBlock)?;
 
+        #[cfg(feature = "fault-proving")]
+        self.storage_as_mut::<crate::storage::registrations::Registrations>()
+            .insert(
+                &height,
+                fuel_core_compression::VersionedBlockPayload::registrations(
+                    compressed_block,
+                ),
+            )
+            .map_err(CompressionError::FailedToWriteRegistrations)?;
+
         // this should not hit the db, we get it from the transaction
         let size = StorageSize::<CompressedBlocks>::size_of_value(self, &height)
             .map_err(CompressionError::FailedToGetCompressedBlockSize)?;

crates/services/compression/src/service.rs

@@ -763,4 +763,43 @@ mod tests {
         assert!(maybe_block.is_some());
         service.shutdown().await.unwrap();
     }
+
+    #[cfg(feature = "fault-proving")]
+    #[tokio::test]
+    async fn compression_service__writes_to_registrations_table() {
+        // given: we create the compression service
+        let block_with_metadata = BlockWithMetadata::default();
+        let block_source = MockBlockSource::new(vec![block_with_metadata]);
+        let storage = test_storage();
+        let config_provider = MockConfigProvider::default();
+        let canonical_height_provider = MockCanonicalHeightProvider::default();
+
+        let uninit_service = UninitializedCompressionService::new(
+            block_source,
+            storage,
+            config_provider.config(),
+            canonical_height_provider,
+        );
+        let sync_observer = uninit_service.shared_data();
+
+        // when: the compression service is started
+        let mut service = uninit_service
+            .into_task(&Default::default(), ())
+            .await
+            .unwrap();
+        let _ = service.run(&mut StateWatcher::started()).await;
+
+        // then: we ensure we can get the registrations per block + root of that table
+        let target_block_height = 0;
+        sync_observer
+            .await_synced_until(&target_block_height)
+            .await
+            .unwrap();
+        let maybe_registrations = service
+            .storage
+            .storage_as_ref::<storage::registrations::Registrations>()
+            .get(&target_block_height.into())
+            .unwrap();
+        assert!(maybe_registrations.is_some());
+    }
 }

crates/services/compression/src/storage.rs

@@ -20,6 +20,8 @@ pub mod predicate_code;
 pub mod registry_index;
 pub mod script_code;
 // TODO: https://github.com/FuelLabs/fuel-core/issues/2842
+#[cfg(feature = "fault-proving")]
+pub mod registrations;
 pub mod timestamps;
 
 /// Merkleized Address table type alias
@@ -48,3 +50,7 @@ pub type Timestamps = Merkleized<timestamps::Timestamps>;
 
 /// Re-export to match api
 pub use compressed_blocks::CompressedBlocks;
+
+/// Merkleized Registrations table type alias
+#[cfg(feature = "fault-proving")]
+pub type Registrations = Merkleized<registrations::Registrations>;

crates/services/compression/src/storage/column.rs

@@ -40,6 +40,9 @@ pub enum CompressionColumn {
     EvictorCache = 7,
     /// Keeps track of timestamps, will be removed eventually, see [`Timestamps`](crate::storage::timestamps::Timestamps)
     Timestamps = 8,
+    #[cfg(feature = "fault-proving")]
+    /// Keeps track of registrations per table associated with a compressed block, see [`Registrations`](crate::storage::registrations::Registrations)
+    Registrations = 9,
 }
 
 impl AsU32 for CompressionColumn {

crates/services/compression/src/storage/registrations.rs

@@ -0,0 +1,54 @@
+//! Registrations table.
+//! Maps from block height -> registrations made for compressing that block
+
+use fuel_core_storage::{
+    Mappable,
+    blueprint::plain::Plain,
+    codec::postcard::Postcard,
+    merkle::sparse::MerkleizedTableColumn,
+    structured_storage::TableWithBlueprint,
+};
+
+use super::column::{
+    CompressionColumn,
+    MerkleizedColumnOf,
+};
+
+/// Table that indexes the registrations.
+pub struct Registrations;
+
+impl MerkleizedTableColumn for Registrations {
+    type TableColumn = CompressionColumn;
+
+    fn table_column() -> Self::TableColumn {
+        Self::TableColumn::Registrations
+    }
+}
+
+impl Mappable for Registrations {
+    type Key = Self::OwnedKey;
+    type OwnedKey = fuel_core_types::fuel_types::BlockHeight;
+    type Value = Self::OwnedValue;
+    type OwnedValue = fuel_core_compression::registry::RegistrationsPerTable;
+}
+
+impl TableWithBlueprint for Registrations {
+    type Blueprint = Plain<Postcard, Postcard>;
+    type Column = MerkleizedColumnOf<Self>;
+
+    fn column() -> Self::Column {
+        Self::Column::TableColumn(Self::table_column())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fuel_core_storage::basic_storage_tests!(
+        Registrations,
+        <Registrations as Mappable>::Key::default(),
+        <Registrations as Mappable>::Value::default(),
+        <Registrations as Mappable>::Value::default()
+    );
+}