fault_proving(compression): implement compression in compression service (#2844)

## Linked Issues/PRs
<!-- List of related issues/PRs -->
- part of #2775

## Description
<!-- List of detailed changes -->

ai generated desc:

This pull request introduces significant updates to the
`fuel-core-compression-service`, including the implementation of DA
compression, configuration enhancements, error handling improvements,
and new features for block and storage management. The most important
changes are summarized below:

### Implementation and Configuration Enhancements:
* Implemented DA compression in `fuel-core-compression-service`.
* Added new dependencies and features in `Cargo.toml` for
`fuel-core-chain-config`, `futures`, and `paste`.
[[1]](diffhunk://#diff-3b348e1ded89f349afd45656ad5c57ebbab5e690f2bf64e059dd5d7c32de228dR15-R23)
[[2]](diffhunk://#diff-3b348e1ded89f349afd45656ad5c57ebbab5e690f2bf64e059dd5d7c32de228dR34-R42)
* Introduced `CompressionConfig` with `temporal_registry_retention` and
methods for configuration management.

### Error Handling Improvements:
* Enhanced error handling in `CompressionError` by replacing string
errors with `StorageError` and adding a new error variant for block
compression failures.
[[1]](diffhunk://#diff-4e1e8a438c26f0ee9cde7ff8e033cb1ab58ffd3f5fad9dd5f39a74269a4a971cR1)
[[2]](diffhunk://#diff-4e1e8a438c26f0ee9cde7ff8e033cb1ab58ffd3f5fad9dd5f39a74269a4a971cL13-R21)

### Block and Storage Management:
* Introduced `BlockWithMetadata` struct to encapsulate block data and
associated events.
* Updated `CompressionStorage` trait to include methods for reading and
writing compressed blocks, and getting the latest height.

### Service Enhancements:
* Refactored `CompressionService` to use the new configuration and block
structures.
[[1]](diffhunk://#diff-9fc61045febb6284fa8e01c5c33172787228fe21b8b813c62c8f03d2fec55fe8L1-R124)
[[2]](diffhunk://#diff-9fc61045febb6284fa8e01c5c33172787228fe21b8b813c62c8f03d2fec55fe8L78-L82)
[[3]](diffhunk://#diff-9fc61045febb6284fa8e01c5c33172787228fe21b8b813c62c8f03d2fec55fe8L105-R167)
[[4]](diffhunk://#diff-9fc61045febb6284fa8e01c5c33172787228fe21b8b813c62c8f03d2fec55fe8L124-R191)
[[5]](diffhunk://#diff-9fc61045febb6284fa8e01c5c33172787228fe21b8b813c62c8f03d2fec55fe8L143-R276)
* Added new tests for `CompressionService` to verify its functionality
with different block sources and configurations.

### Miscellaneous:
* Added new module for `temporal_registry` implementations.
* Introduced `timestamps` module and corresponding type alias in
`storage.rs`.
[[1]](diffhunk://#diff-6354c3b240b9d7af0841032b5051a86cf313008738ca64e678a1783912dd0640R22-R23)
[[2]](diffhunk://#diff-6354c3b240b9d7af0841032b5051a86cf313008738ca64e678a1783912dd0640R46-R48)
* Added `Timestamps` to `CompressionColumn` enum.


## Checklist
- [ ] Breaking changes are clearly marked as such in the PR description
and changelog
- [ ] New behavior is reflected in tests
- [ ] [The specification](https://github.com/FuelLabs/fuel-specs/)
matches the implemented behavior (link update PR if changes are needed)

### Before requesting review
- [ ] I have reviewed the code myself
- [ ] I have created follow-up issues caused by this PR and linked them
here

### After merging, notify other teams

[Add or remove entries as needed]

- [ ] [Rust SDK](https://github.com/FuelLabs/fuels-rs/)
- [ ] [Sway compiler](https://github.com/FuelLabs/sway/)
- [ ] [Platform
documentation](https://github.com/FuelLabs/devrel-requests/issues/new?assignees=&labels=new+request&projects=&template=NEW-REQUEST.yml&title=%5BRequest%5D%3A+)
(for out-of-organization contributors, the person merging the PR will do
this)
- [ ] Someone else?

Author: rymnc

.changes/added/2844.md

@@ -0,0 +1 @@
+Implement DA compression in `fuel-core-compression-service`.

Cargo.lock

@@ -12,10 +12,15 @@ repository = { workspace = true }
 anyhow = { workspace = true }
 async-trait = { workspace = true }
 enum-iterator = { workspace = true }
+fuel-core-chain-config = { workspace = true, features = [
+  "test-helpers",
+], optional = true }
 fuel-core-compression = { workspace = true }
 fuel-core-services = { workspace = true }
 fuel-core-storage = { workspace = true, features = ["std"] }
 fuel-core-types = { workspace = true }
+futures = { workspace = true }
+paste = { workspace = true }
 rand = { workspace = true, optional = true }
 serde = { workspace = true, features = ["derive"] }
 strum = { workspace = true, features = ["derive"] }
@@ -26,11 +31,14 @@ tracing = { workspace = true }
 
 [dev-dependencies]
 fuel-core-compression-service = { path = ".", features = ["test-helpers"] }
+tokio-stream = { workspace = true }
 
 [features]
 test-helpers = [
+  "fuel-core-services/test-helpers",
   "fuel-core-types/test-helpers",
   "fuel-core-storage/test-helpers",
   "fuel-core-compression/test-helpers",
   "dep:rand",
+  "dep:fuel-core-chain-config",
 ]

crates/services/compression/Cargo.toml

@@ -1,3 +1,29 @@
+use std::time::Duration;
+
 /// Compression configuration
-#[derive(Debug, Clone)]
-pub struct CompressionConfig {}
+#[derive(Debug, Clone, Copy)]
+pub struct CompressionConfig {
+    temporal_registry_retention: Duration,
+}
+
+impl CompressionConfig {
+    /// Create a new compression configuration
+    pub fn new(temporal_registry_retention: Duration) -> Self {
+        Self {
+            temporal_registry_retention,
+        }
+    }
+
+    /// Get the temporal registry retention
+    pub fn temporal_registry_retention(&self) -> Duration {
+        self.temporal_registry_retention
+    }
+}
+
+impl From<CompressionConfig> for fuel_core_compression::Config {
+    fn from(config: CompressionConfig) -> Self {
+        Self {
+            temporal_registry_retention: config.temporal_registry_retention(),
+        }
+    }
+}

crates/services/compression/src/config.rs

@@ -1,3 +1,4 @@
+use fuel_core_storage::Error as StorageError;
 use thiserror::Error;
 
 /// Errors that can occur during compression
@@ -10,11 +11,24 @@ pub enum CompressionError {
     // Compression storage Errors
     /// Failed to read compressed block from storage
     #[error("failed to write compressed block to storage: `{0}`")]
-    FailedToWriteCompressedBlock(String),
+    FailedToWriteCompressedBlock(StorageError),
+    /// Failed to commit storage transaction
+    #[error("failed to commit transaction: `{0}`")]
+    FailedToCommitTransaction(StorageError),
+    /// Failed to get latest height from storage
+    #[error("failed to get latest height from storage: `{0}`")]
+    FailedToGetLatestHeight(StorageError),
+    // Configuration Errors
+    /// Failed to get config
+    #[error("failed to get configuration: `{0}`")]
+    FailedToGetConfig(String),
     // Service errors
     /// Failed to create service
     #[error("failed to create service: `{0}`")]
     FailedToCreateService(String),
+    /// Failed to compress block
+    #[error("failed to compress block: `{0}`")]
+    FailedToCompressBlock(anyhow::Error),
     /// Failed to handle new block
     #[error("failed to handle new block: `{0}`")]
     FailedToHandleNewBlock(String),

crates/services/compression/src/errors.rs

@@ -5,8 +5,6 @@
 #![deny(unused_crate_dependencies)]
 #![deny(missing_docs)]
 #![deny(warnings)]
-// we allow unused for now while incrementally creating this service.
-#![allow(unused)]
 
 /// Configuration for the compression service
 pub mod config;
@@ -16,9 +14,10 @@ pub mod errors;
 pub mod ports;
 /// Service definition for the compression service
 pub mod service;
-
 /// Storage traits for the compression service
 pub mod storage;
+/// Temporal Registry implementations
+pub mod temporal_registry;
 
 /// Result type for compression operations
 pub type Result<T> = core::result::Result<T, errors::CompressionError>;

crates/services/compression/src/lib.rs

@@ -1,11 +1,41 @@
-/// Type alias for fuel_core_types::blockchain::block::Block
-pub type Block = fuel_core_types::blockchain::block::Block;
+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>,
+}
+
+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) fn height(&self) -> u32 {
+        (*self.block.header().height()).into()
+    }
+
+    pub(crate) fn events(&self) -> &[fuel_core_types::services::executor::Event] {
+        &self.events
+    }
+
+    pub(crate) fn block(&self) -> &fuel_core_types::blockchain::block::Block {
+        &self.block
+    }
+}
+
+/// Type alias for returned value by .subscribe() method on `BlockSource`
+pub type BlockStream = BoxStream<BlockWithMetadata>;
 
 /// Port for L2 blocks source
-/// Each time `.next()` is called, the service expects the same behaviour as an iterator.
 pub trait BlockSource {
-    /// Get the next block from the source
-    fn next(
-        &self,
-    ) -> impl core::future::Future<Output = crate::Result<Option<Block>>> + Send + Sync;
+    /// Should provide a stream of blocks with metadata
+    fn subscribe(&self) -> BoxStream<BlockWithMetadata>;
 }

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

@@ -1,11 +1,50 @@
+use crate::storage;
+use fuel_core_storage::{
+    kv_store::KeyValueInspect,
+    merkle::column::MerkleizedColumn,
+    transactional::{
+        Modifiable,
+        StorageTransaction,
+    },
+    StorageAsMut,
+};
+
 /// Compressed block type alias
 pub type CompressedBlock = fuel_core_compression::VersionedCompressedBlock;
 
 /// Trait for interacting with storage that supports compression
-pub trait CompressionStorage {
-    /// Write a compressed block to storage
+pub trait CompressionStorage:
+    KeyValueInspect<Column = MerkleizedColumn<storage::column::CompressionColumn>>
+    + Modifiable
+{
+}
+
+impl<T> CompressionStorage for T where
+    T: KeyValueInspect<Column = MerkleizedColumn<storage::column::CompressionColumn>>
+        + Modifiable
+{
+}
+
+pub(crate) trait WriteCompressedBlock {
     fn write_compressed_block(
         &mut self,
-        compressed_block: &CompressedBlock,
+        height: &u32,
+        compressed_block: &crate::ports::compression_storage::CompressedBlock,
     ) -> crate::Result<()>;
 }
+
+impl<Storage> WriteCompressedBlock for StorageTransaction<Storage>
+where
+    Storage:
+        KeyValueInspect<Column = MerkleizedColumn<storage::column::CompressionColumn>>,
+{
+    fn write_compressed_block(
+        &mut self,
+        height: &u32,
+        compressed_block: &crate::ports::compression_storage::CompressedBlock,
+    ) -> crate::Result<()> {
+        self.storage_as_mut::<storage::CompressedBlocks>()
+            .insert(&(*height).into(), compressed_block)
+            .map_err(crate::errors::CompressionError::FailedToWriteCompressedBlock)
+    }
+}