fix(compression_service): prevent syncing from genesis if compression db is empty (#2978)

improving error handling, and refining synchronization logic. Key
changes include adding an override for the starting block height,
improving synchronization methods, and updating error reporting.

### Enhancements to Configurability:
* Added an `override_starting_height` field to `DaCompressionConfig` and
`CompressionConfig`, allowing the service to start syncing from a
specific block height. (`crates/fuel-core/src/service/config.rs`
[[1]](diffhunk://#diff-184f5e785485e92ad26d50d0bbca115a93cd4e76d7d0bafb7052225eb31f0245R366)
`crates/services/compression/src/config.rs`
[[2]](diffhunk://#diff-5d0b8937548df3be41160e9752c4d383b66f498014e123054aa2ec486b93967aL1-R23)
[[3]](diffhunk://#diff-5d0b8937548df3be41160e9752c4d383b66f498014e123054aa2ec486b93967aR37-R41)

### Improvements to Synchronization Logic:
* Updated the `sync_previously_produced_blocks` method to support early
exit and handle the `override_starting_height` configuration.
(`crates/services/compression/src/service.rs`
[crates/services/compression/src/service.rsL224-R259](diffhunk://#diff-9fc61045febb6284fa8e01c5c33172787228fe21b8b813c62c8f03d2fec55fe8L224-R259))
* Modified the `await_synced` method to `await_synced_until`, enabling
synchronization up to a specified block height.
(`crates/fuel-core/src/service.rs`
[[1]](diffhunk://#diff-57d6c5d4dadbd1198f75f2070852e4c27a251e756ed40fb6c214b4d78e4a1ecbL232-R242)
`crates/services/compression/src/service.rs`
[[2]](diffhunk://#diff-9fc61045febb6284fa8e01c5c33172787228fe21b8b813c62c8f03d2fec55fe8L274-R306)

### Error Handling Enhancements:
* Updated `get_block` methods in various components to return
`anyhow::Result` instead of `Option`, providing more detailed error
information.
(`crates/fuel-core/src/service/adapters/compression_adapters.rs`
[[1]](diffhunk://#diff-9e66cb09255fa99a40c07b24f3002b040d023c09b9469521b2c8ad8736bf33bfL63-R77)
`crates/services/compression/src/ports/block_source.rs`
[[2]](diffhunk://#diff-b35ece6e136e2fe3793da56f4456d50692646e9d371946e26024ada643151182L76-R76)
`crates/services/compression/src/service.rs`
[[3]](diffhunk://#diff-9fc61045febb6284fa8e01c5c33172787228fe21b8b813c62c8f03d2fec55fe8L247-R278)

### Test Suite Updates:
* Added new test cases to validate the behavior of the
`override_starting_height` configuration and ensure proper
synchronization logic. (`crates/services/compression/src/service.rs`
[[1]](diffhunk://#diff-9fc61045febb6284fa8e01c5c33172787228fe21b8b813c62c8f03d2fec55fe8L607-R665)
[[2]](diffhunk://#diff-9fc61045febb6284fa8e01c5c33172787228fe21b8b813c62c8f03d2fec55fe8L638-R739)

### Miscellaneous Changes:
* Improved error messages across the codebase to include more context,
such as the specific block height or initialization errors.
(`crates/services/src/service.rs`
[[1]](diffhunk://#diff-daa67623825920672e831440a58a430d60cd713481c0e3b956292030162f569cL405-R405)
`crates/services/compression/src/service.rs`
[[2]](diffhunk://#diff-9fc61045febb6284fa8e01c5c33172787228fe21b8b813c62c8f03d2fec55fe8L247-R278)

Author: rymnc

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

@@ -82,7 +82,10 @@ use rlimit::{
 use std::{
     env,
     net,
-    num::NonZeroU64,
+    num::{
+        NonZeroU32,
+        NonZeroU64,
+    },
     path::PathBuf,
     str::FromStr,
     time::Duration,
@@ -238,6 +241,10 @@ pub struct Command {
     #[arg(long = "da-compression", env)]
     pub da_compression: Option<humantime::Duration>,
 
+    /// The starting height of the compression service.
+    #[arg(long = "da-compression-starting-height", env)]
+    pub da_compression_starting_height: Option<NonZeroU32>,
+
     /// A new block is produced instantly when transactions are available.
     #[clap(flatten)]
     pub poa_trigger: PoATriggerArgs,
@@ -341,6 +348,7 @@ impl Command {
             #[cfg(feature = "aws-kms")]
             consensus_aws_kms,
             da_compression,
+            da_compression_starting_height,
             poa_trigger,
             predefined_blocks_path,
             coinbase_recipient,
@@ -551,6 +559,7 @@ impl Command {
             Some(retention_duration) => DaCompressionMode::Enabled(
                 fuel_core::service::config::DaCompressionConfig {
                     retention_duration: retention_duration.into(),
+                    starting_height: da_compression_starting_height,
                     metrics: metrics.is_enabled(Module::Compression),
                 },
             ),

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

@@ -237,6 +237,7 @@ impl KeypairArg {
             return Ok(KeypairArg::InlineSecret(secret))
         }
         let path = PathBuf::from_str(s);
+        #[allow(irrefutable_let_patterns)]
         if let Ok(pathbuf) = path {
             if pathbuf.exists() {
                 return Ok(KeypairArg::Path(pathbuf))

crates/fuel-core/src/service.rs

@@ -40,7 +40,10 @@ use fuel_core_storage::{
     IsNotFound,
     StorageAsMut,
 };
-use fuel_core_types::blockchain::consensus::Consensus;
+use fuel_core_types::{
+    blockchain::consensus::Consensus,
+    fuel_types::BlockHeight,
+};
 
 use crate::{
     combined_database::{
@@ -229,10 +232,14 @@ impl FuelService {
         Ok(())
     }
 
-    /// Wait for the compression service to be in sync with L2 height
-    pub async fn await_compression_synced(&self) -> anyhow::Result<()> {
+    /// Waits until the compression service has synced
+    /// with the given block height
+    pub async fn await_compression_synced_until(
+        &self,
+        block_height: &BlockHeight,
+    ) -> anyhow::Result<()> {
         if let Some(sync_observer) = &self.runner.shared.compression {
-            sync_observer.await_synced().await?;
+            sync_observer.await_synced_until(block_height).await?;
         }
         Ok(())
     }

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

@@ -60,18 +60,21 @@ impl block_source::BlockSource for CompressionBlockImporterAdapter {
         self.block_importer.events_shared_result()
     }
 
-    fn get_block(&self, height: BlockAt) -> Option<SharedImportResult> {
+    fn get_block(&self, height: BlockAt) -> anyhow::Result<SharedImportResult> {
         self.import_result_provider_adapter
             .result_at_height(height.into())
-            .ok()
     }
 }
 
 impl configuration::CompressionConfigProvider
     for crate::service::config::DaCompressionConfig
 {
     fn config(&self) -> config::CompressionConfig {
-        config::CompressionConfig::new(self.retention_duration, self.metrics)
+        config::CompressionConfig::new(
+            self.retention_duration,
+            self.starting_height,
+            self.metrics,
+        )
     }
 }
 

crates/fuel-core/src/service/config.rs

@@ -1,6 +1,9 @@
 use clap::ValueEnum;
 use std::{
-    num::NonZeroU64,
+    num::{
+        NonZeroU32,
+        NonZeroU64,
+    },
     path::PathBuf,
     time::Duration,
 };
@@ -360,6 +363,7 @@ impl GasPriceConfig {
 pub struct DaCompressionConfig {
     pub retention_duration: Duration,
     pub metrics: bool,
+    pub starting_height: Option<NonZeroU32>,
 }
 
 #[derive(Debug, Clone)]

crates/services/compression/src/config.rs

@@ -1,17 +1,26 @@
-use std::time::Duration;
+use std::{
+    num::NonZeroU32,
+    time::Duration,
+};
 
 /// Compression configuration
 #[derive(Debug, Clone, Copy)]
 pub struct CompressionConfig {
     temporal_registry_retention: Duration,
+    starting_height: Option<NonZeroU32>,
     metrics: bool,
 }
 
 impl CompressionConfig {
     /// Create a new compression configuration
-    pub fn new(temporal_registry_retention: Duration, metrics: bool) -> Self {
+    pub fn new(
+        temporal_registry_retention: Duration,
+        starting_height: Option<NonZeroU32>,
+        metrics: bool,
+    ) -> Self {
         Self {
             temporal_registry_retention,
+            starting_height,
             metrics,
         }
     }
@@ -25,6 +34,11 @@ impl CompressionConfig {
     pub fn metrics(&self) -> bool {
         self.metrics
     }
+
+    /// Get the override starting height
+    pub fn starting_height(&self) -> Option<u32> {
+        self.starting_height.map(|height| height.get())
+    }
 }
 
 impl From<&CompressionConfig> for fuel_core_compression::Config {

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

@@ -73,5 +73,5 @@ pub trait BlockSource: Send + Sync {
     /// Should provide a stream of blocks with metadata
     fn subscribe(&self) -> BlockStream;
     /// Should provide the block at a given height
-    fn get_block(&self, height: BlockAt) -> Option<BlockWithMetadata>;
+    fn get_block(&self, height: BlockAt) -> anyhow::Result<BlockWithMetadata>;
 }