Updated block importer to allow more blocks to be queue

CHANGELOG.md

@@ -7,11 +7,14 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 ## [Unreleased]
 
 ### Added
+- [#1996](https://github.com/FuelLabs/fuel-core/pull/1996): Added support for rollback command when state rewind feature is enabled. The command allows the rollback of the state of the blockchain several blocks behind until the end of the historical window. The default historical window it 7 days.
+- [#1996](https://github.com/FuelLabs/fuel-core/pull/1996): Added support for the state rewind feature. The feature allows the execution of the blocks in the past and the same execution results to be received. Together with forkless upgrades, execution of any block from the past is possible if historical data exist for the target block height.
 - [#1994](https://github.com/FuelLabs/fuel-core/pull/1994): Added the actual implementation for the `AtomicView::latest_view`.
 - [#1972](https://github.com/FuelLabs/fuel-core/pull/1972): Implement `AlgorithmUpdater` for `GasPriceService`
 - [#1948](https://github.com/FuelLabs/fuel-core/pull/1948): Add new `AlgorithmV1` and `AlgorithmUpdaterV1` for the gas price. Include tools for analysis
 
 ### Changed
+- [#2010](https://github.com/FuelLabs/fuel-core/pull/2010): Updated the block importer to allow more blocks to be in the queue. It improves synchronization speed and mitigate the impact of other services on synchronization speed.
 - [#2002](https://github.com/FuelLabs/fuel-core/pull/2002): Adapted the block producer to react to checked transactions that were using another version of consensus parameters during validation in the TxPool. After an upgrade of the consensus parameters of the network, TxPool could store invalid `Checked` transactions. This change fixes that by tracking the version that was used to validate the transactions.
 - [#1999](https://github.com/FuelLabs/fuel-core/pull/1999): Minimize the number of panics in the codebase.
 - [#1990](https://github.com/FuelLabs/fuel-core/pull/1990): Use latest view for mutate GraphQL queries after modification of the node.

benches/benches/vm_set/blockchain.rs

@@ -20,9 +20,12 @@ use fuel_core::{
         GenesisDatabase,
     },
     service::Config,
-    state::rocks_db::{
-        RocksDb,
-        ShallowTempDir,
+    state::{
+        historical_rocksdb::HistoricalRocksDB,
+        rocks_db::{
+            RocksDb,
+            ShallowTempDir,
+        },
     },
 };
 use fuel_core_benches::*;
@@ -74,8 +77,8 @@ impl BenchDb {
     fn new(contract_id: &ContractId) -> anyhow::Result<Self> {
         let tmp_dir = ShallowTempDir::new();
 
-        let db =
-            Arc::new(RocksDb::<OnChain>::default_open(tmp_dir.path(), None).unwrap());
+        let db = RocksDb::<OnChain>::default_open(tmp_dir.path(), None).unwrap();
+        let db = Arc::new(HistoricalRocksDB::new(db, Default::default()).unwrap());
         let mut storage_key = primitive_types::U256::zero();
         let mut key_bytes = Bytes32::zeroed();
 

bin/fuel-core/Cargo.toml

@@ -59,6 +59,6 @@ p2p = ["fuel-core/p2p", "const_format"]
 relayer = ["fuel-core/relayer", "dep:url"]
 parquet = ["fuel-core-chain-config/parquet", "fuel-core-types/serde"]
 rocksdb = ["fuel-core/rocksdb"]
-rocksdb-production = ["fuel-core/rocksdb-production"]
+rocksdb-production = ["fuel-core/rocksdb-production", "rocksdb"]
 # features to enable in production, but increase build times
 production = ["env", "relayer", "rocksdb-production", "p2p", "parquet"]

bin/fuel-core/src/cli.rs

@@ -26,9 +26,12 @@ pub fn default_db_path() -> PathBuf {
 }
 
 pub mod fee_contract;
+#[cfg(feature = "rocksdb")]
+pub mod rollback;
 pub mod run;
-#[cfg(any(feature = "rocksdb", feature = "rocksdb-production"))]
+#[cfg(feature = "rocksdb")]
 pub mod snapshot;
+
 // Default database cache is 1 GB
 pub const DEFAULT_DATABASE_CACHE_SIZE: usize = 1024 * 1024 * 1024;
 
@@ -48,8 +51,10 @@ pub struct Opt {
 #[derive(Debug, Parser)]
 pub enum Fuel {
     Run(run::Command),
-    #[cfg(any(feature = "rocksdb", feature = "rocksdb-production"))]
+    #[cfg(feature = "rocksdb")]
     Snapshot(snapshot::Command),
+    #[cfg(feature = "rocksdb")]
+    Rollback(rollback::Command),
     GenerateFeeContract(fee_contract::Command),
 }
 
@@ -128,9 +133,10 @@ pub async fn run_cli() -> anyhow::Result<()> {
     match opt {
         Ok(opt) => match opt.command {
             Fuel::Run(command) => run::exec(command).await,
-            #[cfg(any(feature = "rocksdb", feature = "rocksdb-production"))]
+            #[cfg(feature = "rocksdb")]
             Fuel::Snapshot(command) => snapshot::exec(command).await,
             Fuel::GenerateFeeContract(command) => fee_contract::exec(command).await,
+            Fuel::Rollback(command) => rollback::exec(command).await,
         },
         Err(e) => {
             // Prints the error and exits.
@@ -213,7 +219,7 @@ impl NotifyCancel for ShutdownListener {
     }
 }
 
-#[cfg(any(feature = "rocksdb", feature = "rocksdb-production"))]
+#[cfg(feature = "rocksdb")]
 #[cfg(test)]
 mod tests {
     use anyhow::anyhow;

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

@@ -0,0 +1,93 @@
+use crate::cli::default_db_path;
+use anyhow::Context;
+use clap::Parser;
+use fuel_core::{
+    combined_database::CombinedDatabase,
+    service::genesis::NotifyCancel,
+    state::historical_rocksdb::StateRewindPolicy,
+};
+use std::path::PathBuf;
+
+/// Rollbacks the state of the blockchain to a specific block height.
+#[derive(Debug, Clone, Parser)]
+pub struct Command {
+    /// The path to the database.
+    #[clap(
+        name = "DB_PATH",
+        long = "db-path",
+        value_parser,
+        default_value = default_db_path().into_os_string()
+    )]
+    pub database_path: PathBuf,
+
+    /// The path to the database.
+    #[clap(long = "target-block-height")]
+    pub target_block_height: u32,
+}
+
+pub async fn exec(command: Command) -> anyhow::Result<()> {
+    use crate::cli::ShutdownListener;
+
+    let path = command.database_path.as_path();
+    let db = CombinedDatabase::open(
+        path,
+        64 * 1024 * 1024,
+        StateRewindPolicy::RewindFullRange,
+    )
+    .map_err(Into::<anyhow::Error>::into)
+    .context(format!("failed to open combined database at path {path:?}"))?;
+
+    let shutdown_listener = ShutdownListener::spawn();
+    let target_block_height = command.target_block_height.into();
+
+    while !shutdown_listener.is_cancelled() {
+        let on_chain_height = db
+            .on_chain()
+            .latest_height()?
+            .ok_or(anyhow::anyhow!("on-chain database doesn't have height"))?;
+
+        let off_chain_height = db
+            .off_chain()
+            .latest_height()?
+            .ok_or(anyhow::anyhow!("on-chain database doesn't have height"))?;
+
+        if on_chain_height == target_block_height
+            && off_chain_height == target_block_height
+        {
+            break;
+        }
+
+        if off_chain_height == target_block_height
+            && on_chain_height < target_block_height
+        {
+            return Err(anyhow::anyhow!(
+                "on-chain database height is less than target height"
+            ));
+        }
+
+        if on_chain_height == target_block_height
+            && off_chain_height < target_block_height
+        {
+            return Err(anyhow::anyhow!(
+                "off-chain database height is less than target height"
+            ));
+        }
+
+        if on_chain_height > target_block_height {
+            db.on_chain().rollback_last_block()?;
+            tracing::info!(
+                "Rolled back on-chain database to height {:?}",
+                on_chain_height.pred()
+            );
+        }
+
+        if off_chain_height > target_block_height {
+            db.off_chain().rollback_last_block()?;
+            tracing::info!(
+                "Rolled back off-chain database to height {:?}",
+                on_chain_height.pred()
+            );
+        }
+    }
+    Ok(())
+}

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

@@ -58,6 +58,7 @@ use pyroscope_pprofrs::{
 use std::{
     env,
     net,
+    num::NonZeroU64,
     path::PathBuf,
     str::FromStr,
 };
@@ -67,6 +68,9 @@ use tracing::{
     warn,
 };
 
+#[cfg(feature = "rocksdb")]
+use fuel_core::state::historical_rocksdb::StateRewindPolicy;
+
 use super::DEFAULT_DATABASE_CACHE_SIZE;
 
 pub const CONSENSUS_KEY_ENV: &str = "CONSENSUS_KEY_SECRET";
@@ -114,6 +118,20 @@ pub struct Command {
     )]
     pub database_type: DbType,
 
+    #[cfg(feature = "rocksdb")]
+    /// Defines the state rewind policy for the database when RocksDB is enabled.
+    ///
+    /// The duration defines how many blocks back the rewind feature works.
+    /// Assuming each block requires one second to produce.
+    ///
+    /// The default value is 7 days = 604800 blocks.
+    ///
+    /// The `BlockHeight` is `u32`, meaning the maximum possible number of blocks
+    /// is less than 137 years. `2^32 / 24 / 60 / 60/ 365` = `136.1925195332` years.
+    /// If the value is 136 years or more, the rewind feature is enabled for all blocks.
+    #[clap(long = "state-rewind-duration", default_value = "7d", env)]
+    pub state_rewind_duration: humantime::Duration,
+
     /// Snapshot from which to do (re)genesis. Defaults to local testnet configuration.
     #[arg(name = "SNAPSHOT", long = "snapshot", env)]
     pub snapshot: Option<PathBuf>,
@@ -214,6 +232,8 @@ impl Command {
             max_database_cache_size,
             database_path,
             database_type,
+            #[cfg(feature = "rocksdb")]
+            state_rewind_duration,
             db_prune,
             snapshot,
             vm_backtrace,
@@ -297,10 +317,36 @@ impl Command {
             max_wait_time: max_wait_time.into(),
         };
 
+        #[cfg(feature = "rocksdb")]
+        let state_rewind_policy = {
+            if database_type != DbType::RocksDb {
+                tracing::warn!("State rewind policy is only supported with RocksDB");
+            }
+
+            let blocks = state_rewind_duration.as_secs();
+
+            if blocks == 0 {
+                StateRewindPolicy::NoRewind
+            } else {
+                let maximum_blocks: humantime::Duration = "136y".parse()?;
+
+                if blocks >= maximum_blocks.as_secs() {
+                    StateRewindPolicy::RewindFullRange
+                } else {
+                    StateRewindPolicy::RewindRange {
+                        size: NonZeroU64::new(blocks)
+                            .expect("The value is not zero above"),
+                    }
+                }
+            }
+        };
+
         let combined_db_config = CombinedDatabaseConfig {
             database_path,
             database_type,
             max_database_cache_size,
+            #[cfg(feature = "rocksdb")]
+            state_rewind_policy,
         };
 
         let block_importer =
@@ -376,7 +422,7 @@ impl Command {
 }
 
 pub fn get_service(command: Command) -> anyhow::Result<FuelService> {
-    #[cfg(any(feature = "rocksdb", feature = "rocksdb-production"))]
+    #[cfg(feature = "rocksdb")]
     if command.db_prune && command.database_path.exists() {
         fuel_core::combined_database::CombinedDatabase::prune(&command.database_path)?;
     }

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

@@ -6,10 +6,10 @@ use clap::{
 };
 use fuel_core::{
     combined_database::CombinedDatabase,
+    state::historical_rocksdb::StateRewindPolicy,
     types::fuel_types::ContractId,
 };
 use fuel_core_chain_config::ChainConfig;
-
 use std::path::{
     Path,
     PathBuf,
@@ -112,7 +112,7 @@ pub enum SubCommands {
     },
 }
 
-#[cfg(any(feature = "rocksdb", feature = "rocksdb-production"))]
+#[cfg(feature = "rocksdb")]
 pub async fn exec(command: Command) -> anyhow::Result<()> {
     use fuel_core::service::genesis::Exporter;
     use fuel_core_chain_config::{
@@ -181,9 +181,13 @@ fn load_chain_config_or_use_testnet(path: Option<&Path>) -> anyhow::Result<Chain
 }
 
 fn open_db(path: &Path, capacity: Option<usize>) -> anyhow::Result<CombinedDatabase> {
-    CombinedDatabase::open(path, capacity.unwrap_or(1024 * 1024 * 1024))
-        .map_err(Into::<anyhow::Error>::into)
-        .context(format!("failed to open combined database at path {path:?}",))
+    CombinedDatabase::open(
+        path,
+        capacity.unwrap_or(1024 * 1024 * 1024),
+        StateRewindPolicy::NoRewind,
+    )
+    .map_err(Into::<anyhow::Error>::into)
+    .context(format!("failed to open combined database at path {path:?}",))
 }
 
 #[cfg(test)]

ci_checks.sh

@@ -13,8 +13,8 @@
 # - `cargo install cargo-insta`
 # - `npm install prettier prettier-plugin-toml`
 
-npx prettier --check "**/Cargo.toml" &&
-cargo +nightly fmt --all -- --check &&
+npx prettier --write "**/Cargo.toml" &&
+cargo +nightly fmt --all &&
 cargo sort -w --check &&
 source .github/workflows/scripts/verify_openssl.sh &&
 cargo clippy -p fuel-core-wasm-executor --target wasm32-unknown-unknown --no-default-features &&

crates/database/src/lib.rs

@@ -17,6 +17,7 @@ use fuel_core_types::services::executor::Error as ExecutorError;
 /// The error occurred during work with any of databases.
 #[derive(Debug, derive_more::Display, derive_more::From)]
 #[non_exhaustive]
+#[allow(missing_docs)]
 pub enum Error {
     /// Error occurred during serialization or deserialization of the entity.
     #[display(fmt = "error performing serialization or deserialization")]
@@ -58,6 +59,18 @@ pub enum Error {
         /// The old height known by the database.
         prev_height: u64,
     },
+    #[display(fmt = "The historical database doesn't have any history yet")]
+    NoHistoryIsAvailable,
+    #[display(
+        fmt = "The historical database doesn't have history for the requested height {requested_height:#x}, \
+            the oldest available height is {oldest_available_height:#x}"
+    )]
+    NoHistoryForRequestedHeight {
+        requested_height: u64,
+        oldest_available_height: u64,
+    },
+    #[display(fmt = "Reached the end of the history")]
+    ReachedEndOfHistory,
 
     /// Not related to database error.
     #[from]