Merge branch 'master' into create_exec_sequencer_skeleton

Author: AurelienFT

.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.

README.md

@@ -14,7 +14,7 @@ Fuel client implementation.
 | Network  | Version |
 |----------|---------|
 | Fuel Ignition | 0.43.2 |
-| Testnet | 0.43.2 |
+| Testnet | 0.44.0 |
 | Devnet | 0.44.0 |
 
 ## Contributing

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