feat: Support rolling back relayer DB (#3022)

Closes #3019 

This PR extends the existing rollback functionality to also roll back
the relayer database. The DA height to roll back to is inferred based on
the DA height of the target block during the rollback.

---------

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

Author: 4 people

.changes/added/3027.md

@@ -0,0 +1 @@
+Support state rewind for compression DB.

.changes/breaking/3022.md

@@ -0,0 +1 @@
+Support rolling back Relayer DB.

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

@@ -42,7 +42,11 @@ pub struct Command {
 
     /// The path to the database.
     #[clap(long = "target-block-height")]
-    pub target_block_height: u32,
+    pub target_block_height: Option<u32>,
+
+    /// The path to the database.
+    #[clap(long = "target-da-block-height")]
+    pub target_da_block_height: Option<u64>,
 }
 
 fn get_default_max_fds() -> i32 {
@@ -68,9 +72,13 @@ pub async fn exec(command: Command) -> anyhow::Result<()> {
     .context(format!("failed to open combined database at path {path:?}"))?;
 
     let mut shutdown_listener = ShutdownListener::spawn();
-    let target_block_height = command.target_block_height.into();
+    if let Some(target_block_height) = command.target_block_height {
+        db.rollback_to(target_block_height.into(), &mut shutdown_listener)?;
+    }
 
-    db.rollback_to(target_block_height, &mut shutdown_listener)?;
+    if let Some(target_da_height) = command.target_da_block_height {
+        db.rollback_relayer_to(target_da_height.into(), &mut shutdown_listener)?;
+    }
 
     Ok(())
 }

ci_checks.sh

@@ -14,7 +14,6 @@
 # - `cargo install cargo-nextest`
 # - `npm install prettier prettier-plugin-toml`
 
-npx prettier --write "**/Cargo.toml" &&
 cargo +nightly fmt --all &&
 cargo sort -w --check &&
 source .github/workflows/scripts/verify_openssl.sh &&

crates/fuel-core/src/combined_database.rs

@@ -36,7 +36,10 @@ use fuel_core_storage::tables::{
     ContractsState,
     Messages,
 };
-use fuel_core_types::fuel_types::BlockHeight;
+use fuel_core_types::{
+    blockchain::primitives::DaBlockHeight,
+    fuel_types::BlockHeight,
+};
 use std::path::PathBuf;
 
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -215,7 +218,7 @@ impl CombinedDatabase {
         )?;
         let relayer = Database::open_rocksdb(
             path,
-            StateRewindPolicy::NoRewind,
+            state_rewind_policy,
             DatabaseConfig {
                 max_fds,
                 ..database_config
@@ -421,14 +424,18 @@ impl CombinedDatabase {
 
             let gas_price_chain_height =
                 self.gas_price().latest_height_from_metadata()?;
+            let gas_price_rolled_back =
+                is_equal_or_none(gas_price_chain_height, target_block_height);
 
-            let gas_price_rolled_back = gas_price_chain_height.is_none()
-                || gas_price_chain_height.expect("We checked height before")
-                    == target_block_height;
+            let compression_db_height =
+                self.compression().latest_height_from_metadata()?;
+            let compression_db_rolled_back =
+                is_equal_or_none(compression_db_height, target_block_height);
 
             if on_chain_height == target_block_height
                 && off_chain_height == target_block_height
                 && gas_price_rolled_back
+                && compression_db_rolled_back
             {
                 break;
             }
@@ -450,7 +457,16 @@ impl CombinedDatabase {
             if let Some(gas_price_chain_height) = gas_price_chain_height {
                 if gas_price_chain_height < target_block_height {
                     return Err(anyhow::anyhow!(
-                        "gas-price-chain database height({gas_price_chain_height}) \
+                        "gas-price database height({gas_price_chain_height}) \
+                        is less than target height({target_block_height})"
+                    ));
+                }
+            }
+
+            if let Some(compression_db_height) = compression_db_height {
+                if compression_db_height < target_block_height {
+                    return Err(anyhow::anyhow!(
+                        "compression database height({compression_db_height}) \
                         is less than target height({target_block_height})"
                     ));
                 }
@@ -469,6 +485,55 @@ impl CombinedDatabase {
                     self.gas_price().rollback_last_block()?;
                 }
             }
+
+            if let Some(compression_db_height) = compression_db_height {
+                if compression_db_height > target_block_height {
+                    self.compression().rollback_last_block()?;
+                }
+            }
+        }
+
+        if shutdown_listener.is_cancelled() {
+            return Err(anyhow::anyhow!(
+                "Stop the rollback due to shutdown signal received"
+            ));
+        }
+
+        Ok(())
+    }
+
+    /// Rollbacks the state of the relayer to a specific block height.
+    pub fn rollback_relayer_to<S>(
+        &self,
+        target_da_height: DaBlockHeight,
+        shutdown_listener: &mut S,
+    ) -> anyhow::Result<()>
+    where
+        S: ShutdownListener,
+    {
+        while !shutdown_listener.is_cancelled() {
+            let relayer_db_height = self.relayer().latest_height_from_metadata()?;
+            let relayer_db_rolled_back =
+                is_equal_or_none(relayer_db_height, target_da_height);
+
+            if relayer_db_rolled_back {
+                break;
+            }
+
+            if let Some(relayer_db_height) = relayer_db_height {
+                if relayer_db_height < target_da_height {
+                    return Err(anyhow::anyhow!(
+                        "relayer database height({relayer_db_height}) \
+                        is less than target height({target_da_height})"
+                    ));
+                }
+            }
+
+            if let Some(relayer_db_height) = relayer_db_height {
+                if relayer_db_height > target_da_height {
+                    self.relayer().rollback_last_block()?;
+                }
+            }
         }
 
         if shutdown_listener.is_cancelled() {
@@ -557,12 +622,19 @@ impl CombinedGenesisDatabase {
     }
 }
 
+fn is_equal_or_none<T: PartialEq>(maybe_left: Option<T>, right: T) -> bool {
+    maybe_left.map(|left| left == right).unwrap_or(true)
+}
+
 #[allow(non_snake_case)]
 #[cfg(feature = "backup")]
 #[cfg(test)]
 mod tests {
     use super::*;
-    use fuel_core_storage::StorageAsMut;
+    use fuel_core_storage::{
+        StorageAsMut,
+        StorageAsRef,
+    };
     use fuel_core_types::{
         entities::coins::coin::CompressedCoin,
         fuel_tx::UtxoId,
@@ -603,7 +675,7 @@ mod tests {
         )
         .unwrap();
 
-        let mut restored_on_chain_db = restored_db.on_chain();
+        let restored_on_chain_db = restored_db.on_chain();
         let restored_value = restored_on_chain_db
             .storage::<Coins>()
             .get(&key)

tests/tests/da_compression.rs

@@ -1,6 +1,9 @@
+use clap::Parser;
 use core::time::Duration;
 use fuel_core::{
     chain_config::TESTNET_WALLET_SECRETS,
+    combined_database::CombinedDatabase,
+    database::database_description::DatabaseHeight,
     p2p_test_helpers::*,
     service::{
         Config,
@@ -10,6 +13,13 @@ use fuel_core::{
             DaCompressionMode,
         },
     },
+    state::{
+        historical_rocksdb::StateRewindPolicy,
+        rocks_db::{
+            ColumnsPolicy,
+            DatabaseConfig,
+        },
+    },
 };
 use fuel_core_client::client::{
     FuelClient,
@@ -253,6 +263,57 @@ async fn da_compression__starts_and_compresses_blocks_correctly_from_empty_datab
     driver.kill().await;
 }
 
+#[tokio::test]
+async fn da_compression__db_can_be_rewinded() {
+    // given
+    let rollback_target_height = 0;
+    let blocks_to_produce = 10;
+
+    let args = vec!["--da-compression", "7d", "--debug"];
+    let driver = FuelCoreDriver::spawn(&args).await.unwrap();
+    let current_height = driver
+        .client
+        .produce_blocks(blocks_to_produce, None)
+        .await
+        .unwrap();
+    assert_eq!(current_height, blocks_to_produce.into());
+
+    let db_dir = driver.kill().await;
+
+    // when
+    let target_block_height = rollback_target_height.to_string();
+    let args = [
+        "_IGNORED_",
+        "--db-path",
+        db_dir.path().to_str().unwrap(),
+        "--target-block-height",
+        target_block_height.as_str(),
+    ];
+
+    let command = fuel_core_bin::cli::rollback::Command::parse_from(args);
+    fuel_core_bin::cli::rollback::exec(command).await.unwrap();
+
+    let db = CombinedDatabase::open(
+        db_dir.path(),
+        StateRewindPolicy::RewindFullRange,
+        DatabaseConfig {
+            cache_capacity: Some(16 * 1024 * 1024 * 1024),
+            max_fds: -1,
+            columns_policy: ColumnsPolicy::Lazy,
+        },
+    )
+    .expect("Failed to create database");
+
+    let compression_db_block_height_after_rollback =
+        db.compression().latest_height_from_metadata().unwrap();
+
+    // then
+    assert_eq!(
+        compression_db_block_height_after_rollback.unwrap().as_u64(),
+        rollback_target_height
+    );
+}
+
 #[tokio::test]
 async fn da_compression__starts_and_compresses_blocks_correctly_with_overridden_height() {
     // given: the node starts without compression enabled, and produces blocks