fault_proving(compression): glue code for integ into fuel-core

.changes/changed/2859.md

@@ -0,0 +1 @@
+Swap out off-chain worker compression for dedicated compression service in `fuel-core-bin`.

.changes/removed/2859.md

@@ -0,0 +1 @@
+Removed DA compression from off-chain worker in favor of dedicated compression service in `fuel-core-bin`.

Cargo.toml

@@ -72,6 +72,7 @@ fuel-core-keygen-bin = { version = "0.42.0", path = "./bin/keygen" }
 fuel-core-chain-config = { version = "0.42.0", path = "./crates/chain-config", default-features = false }
 fuel-core-client = { version = "0.42.0", path = "./crates/client" }
 fuel-core-compression = { version = "0.42.0", path = "./crates/compression" }
+fuel-core-compression-service = { version = "0.42.0", path = "./crates/services/compression" }
 fuel-core-database = { version = "0.42.0", path = "./crates/database" }
 fuel-core-metrics = { version = "0.42.0", path = "./crates/metrics" }
 fuel-core-services = { version = "0.42.0", path = "./crates/services" }

benches/benches/block_target_gas.rs

@@ -359,6 +359,7 @@ fn service_with_many_contracts(
                 Default::default(),
                 Default::default(),
                 Default::default(),
+                Default::default(),
             ),
             config.clone(),
         )

bin/fuel-core/Cargo.toml

@@ -27,7 +27,6 @@ dirs = "4.0"
 dotenvy = { version = "0.15", optional = true }
 fuel-core = { workspace = true, features = ["wasm-executor"] }
 fuel-core-chain-config = { workspace = true }
-fuel-core-compression = { workspace = true }
 fuel-core-metrics = { workspace = true }
 fuel-core-shared-sequencer = { workspace = true, optional = true }
 fuel-core-types = { workspace = true, features = ["std"] }
@@ -84,7 +83,6 @@ production = [
 ]
 parallel-executor = ["fuel-core/parallel-executor"]
 fault-proving = [
-  "fuel-core-compression/fault-proving",
   "fuel-core-storage/fault-proving",
   "fuel-core-types/fault-proving",
   "fuel-core-chain-config/fault-proving",

bin/fuel-core/src/cli/run.rs

@@ -22,13 +22,15 @@ use fuel_core::{
         CombinedDatabaseConfig,
     },
     fuel_core_graphql_api::{
-        worker_service::DaCompressionConfig,
         Costs,
         ServiceConfig as GraphQLConfig,
     },
     producer::Config as ProducerConfig,
     service::{
-        config::Trigger,
+        config::{
+            DaCompressionMode,
+            Trigger,
+        },
         genesis::NotifyCancel,
         Config,
         DbType,
@@ -526,12 +528,12 @@ impl Command {
         let gas_price_metrics = metrics.is_enabled(Module::GasPrice);
 
         let da_compression = match da_compression {
-            Some(retention) => {
-                DaCompressionConfig::Enabled(fuel_core_compression::Config {
-                    temporal_registry_retention: retention.into(),
-                })
-            }
-            None => DaCompressionConfig::Disabled,
+            Some(retention_duration) => DaCompressionMode::Enabled(
+                fuel_core::service::config::DaCompressionConfig {
+                    retention_duration: retention_duration.into(),
+                },
+            ),
+            None => DaCompressionMode::Disabled,
         };
 
         let TxPoolArgs {

crates/fuel-core/Cargo.toml

@@ -21,7 +21,7 @@ cosmrs = { version = "0.21", optional = true }
 derive_more = { version = "0.99" }
 enum-iterator = { workspace = true }
 fuel-core-chain-config = { workspace = true, features = ["std"] }
-fuel-core-compression = { workspace = true }
+fuel-core-compression-service = { workspace = true }
 fuel-core-consensus-module = { workspace = true }
 fuel-core-database = { workspace = true }
 fuel-core-executor = { workspace = true, features = ["std"] }
@@ -49,7 +49,6 @@ itertools = { workspace = true }
 mockall = { workspace = true, optional = true }
 num_cpus = { version = "1.16.0", optional = true }
 parking_lot = { workspace = true }
-paste = { workspace = true }
 postcard = { workspace = true }
 rand = { workspace = true }
 rocksdb = { version = "0.21", default-features = false, features = [
@@ -110,7 +109,6 @@ test-helpers = [
   "fuel-core-p2p?/test-helpers",
   "fuel-core-storage/test-helpers",
   "fuel-core-chain-config/test-helpers",
-  "fuel-core-compression/test-helpers",
   "fuel-core-txpool/test-helpers",
   "fuel-core-tx-status-manager/test-helpers",
   "fuel-core-services/test-helpers",
@@ -123,7 +121,6 @@ rocksdb-production = ["rocksdb", "rocksdb/jemalloc"]
 wasm-executor = ["fuel-core-upgradable-executor/wasm-executor"]
 parallel-executor = ["fuel-core-parallel-executor"]
 fault-proving = [
-  "fuel-core-compression/fault-proving",
   "fuel-core-types/fault-proving",
   "fuel-core-executor/fault-proving",
   "fuel-core-storage/fault-proving",

crates/fuel-core/src/combined_database.rs

@@ -7,6 +7,7 @@ use crate::state::{
 use crate::{
     database::{
         database_description::{
+            compression::CompressionDatabase,
             gas_price::GasPriceDatabase,
             off_chain::OffChain,
             on_chain::OnChain,
@@ -55,6 +56,7 @@ pub struct CombinedDatabase {
     off_chain: Database<OffChain>,
     relayer: Database<Relayer>,
     gas_price: Database<GasPriceDatabase>,
+    compression: Database<CompressionDatabase>,
 }
 
 impl CombinedDatabase {
@@ -63,12 +65,14 @@ impl CombinedDatabase {
         off_chain: Database<OffChain>,
         relayer: Database<Relayer>,
         gas_price: Database<GasPriceDatabase>,
+        compression: Database<CompressionDatabase>,
     ) -> Self {
         Self {
             on_chain,
             off_chain,
             relayer,
             gas_price,
+            compression,
         }
     }
 
@@ -78,6 +82,7 @@ impl CombinedDatabase {
         crate::state::rocks_db::RocksDb::<OffChain>::prune(path)?;
         crate::state::rocks_db::RocksDb::<Relayer>::prune(path)?;
         crate::state::rocks_db::RocksDb::<GasPriceDatabase>::prune(path)?;
+        crate::state::rocks_db::RocksDb::<CompressionDatabase>::prune(path)?;
         Ok(())
     }
 
@@ -118,6 +123,9 @@ impl CombinedDatabase {
         crate::state::rocks_db::RocksDb::<GasPriceDatabase>::backup(db_dir, temp_dir)
             .trace_err("Failed to backup gas-price database")?;
 
+        crate::state::rocks_db::RocksDb::<CompressionDatabase>::backup(db_dir, temp_dir)
+            .trace_err("Failed to backup compression database")?;
+
         Ok(())
     }
 
@@ -149,22 +157,28 @@ impl CombinedDatabase {
         temp_restore_dir: &std::path::Path,
     ) -> crate::database::Result<()> {
         crate::state::rocks_db::RocksDb::<OnChain>::restore(temp_restore_dir, backup_dir)
-            .trace_err("Failed to backup on-chain database")?;
+            .trace_err("Failed to restore on-chain database")?;
 
         crate::state::rocks_db::RocksDb::<OffChain>::restore(
             temp_restore_dir,
             backup_dir,
         )
-        .trace_err("Failed to backup off-chain database")?;
+        .trace_err("Failed to restore off-chain database")?;
 
         crate::state::rocks_db::RocksDb::<Relayer>::restore(temp_restore_dir, backup_dir)
-            .trace_err("Failed to backup relayer database")?;
+            .trace_err("Failed to restore relayer database")?;
 
         crate::state::rocks_db::RocksDb::<GasPriceDatabase>::restore(
             temp_restore_dir,
             backup_dir,
         )
-        .trace_err("Failed to backup gas-price database")?;
+        .trace_err("Failed to restore gas-price database")?;
+
+        crate::state::rocks_db::RocksDb::<CompressionDatabase>::restore(
+            temp_restore_dir,
+            backup_dir,
+        )
+        .trace_err("Failed to restore compression database")?;
 
         Ok(())
     }
@@ -215,11 +229,20 @@ impl CombinedDatabase {
                 ..database_config
             },
         )?;
+        let compression = Database::open_rocksdb(
+            path,
+            state_rewind_policy,
+            DatabaseConfig {
+                max_fds,
+                ..database_config
+            },
+        )?;
         Ok(Self {
             on_chain,
             off_chain,
             relayer,
             gas_price,
+            compression,
         })
     }
 
@@ -234,6 +257,7 @@ impl CombinedDatabase {
             off_chain: Database::rocksdb_temp(state_rewind_policy, database_config)?,
             relayer: Default::default(),
             gas_price: Default::default(),
+            compression: Default::default(),
         })
     }
 
@@ -278,6 +302,7 @@ impl CombinedDatabase {
             Database::in_memory(),
             Database::in_memory(),
             Database::in_memory(),
+            Database::in_memory(),
         )
     }
 
@@ -286,13 +311,18 @@ impl CombinedDatabase {
         self.off_chain.check_version()?;
         self.relayer.check_version()?;
         self.gas_price.check_version()?;
+        self.compression.check_version()?;
         Ok(())
     }
 
     pub fn on_chain(&self) -> &Database<OnChain> {
         &self.on_chain
     }
 
+    pub fn compression(&self) -> &Database<CompressionDatabase> {
+        &self.compression
+    }
+
     #[cfg(any(feature = "test-helpers", test))]
     pub fn on_chain_mut(&mut self) -> &mut Database<OnChain> {
         &mut self.on_chain

crates/fuel-core/src/database.rs

@@ -24,6 +24,7 @@ use crate::{
         KeyValueView,
     },
 };
+use database_description::compression::CompressionDatabase;
 use fuel_core_chain_config::TableEntry;
 pub use fuel_core_database::Error;
 use fuel_core_gas_price_service::common::fuel_core_storage_adapter::storage::GasPriceMetadata;
@@ -446,6 +447,15 @@ impl Modifiable for Database<Relayer> {
     }
 }
 
+impl Modifiable for Database<CompressionDatabase> {
+    fn commit_changes(&mut self, changes: Changes) -> StorageResult<()> {
+        commit_changes_with_height_update(self, changes, |iter| {
+            iter.iter_all_keys::<fuel_core_compression_service::storage::CompressedBlocks>(Some(IterDirection::Reverse))
+                .try_collect()
+        })
+    }
+}
+
 #[cfg(not(feature = "relayer"))]
 impl Modifiable for Database<Relayer> {
     fn commit_changes(&mut self, changes: Changes) -> StorageResult<()> {

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

@@ -7,6 +7,7 @@ use fuel_core_types::{
 use std::collections::HashSet;
 use strum::IntoEnumIterator;
 
+pub mod compression;
 pub mod gas_price;
 pub mod off_chain;
 pub mod on_chain;

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

@@ -0,0 +1,28 @@
+use crate::database::database_description::DatabaseDescription;
+use fuel_core_compression_service::storage::column::CompressionColumn;
+use fuel_core_storage::merkle::column::MerkleizedColumn;
+use fuel_core_types::fuel_types::BlockHeight;
+
+#[derive(Clone, Copy, Debug)]
+pub struct CompressionDatabase;
+
+impl DatabaseDescription for CompressionDatabase {
+    type Column = MerkleizedColumn<CompressionColumn>;
+    type Height = BlockHeight;
+
+    fn version() -> u32 {
+        0
+    }
+
+    fn name() -> String {
+        "compression".to_string()
+    }
+
+    fn metadata_column() -> Self::Column {
+        Self::Column::MerkleMetadataColumn
+    }
+
+    fn prefix(_column: &Self::Column) -> Option<usize> {
+        None
+    }
+}

crates/fuel-core/src/graphql_api.rs

@@ -11,7 +11,6 @@ use std::{
 
 pub mod api_service;
 pub(crate) mod block_height_subscription;
-pub mod da_compression;
 pub mod database;
 pub(crate) mod extensions;
 pub(crate) mod indexation;

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

@@ -100,6 +100,7 @@ use super::{
     block_height_subscription,
     ports::{
         worker,
+        DatabaseDaCompressedBlocks,
         OnChainDatabaseAt,
     },
 };
@@ -116,6 +117,8 @@ pub type GasPriceProvider = Box<dyn GasPriceEstimate>;
 
 pub type ChainInfoProvider = Box<dyn ChainStateProviderTrait>;
 
+pub type DaCompressionProvider = Box<dyn DatabaseDaCompressedBlocks>;
+
 #[derive(Clone)]
 pub struct SharedState {
     pub bound_address: SocketAddr,
@@ -245,6 +248,7 @@ pub fn new_service<OnChain, OffChain>(
     chain_state_info_provider: ChainInfoProvider,
     memory_pool: SharedMemoryPool,
     block_height_subscriber: block_height_subscription::Subscriber,
+    da_compression_provider: DaCompressionProvider,
 ) -> anyhow::Result<Service>
 where
     OnChain: HistoricalView<Height = BlockHeight> + 'static,
@@ -303,6 +307,7 @@ where
         .data(gas_price_provider)
         .data(chain_state_info_provider)
         .data(memory_pool)
+        .data(da_compression_provider)
         .data(block_height_subscriber.clone())
         .extension(ValidationExtension::new(
             max_queries_resolver_recursive_depth,