feat(warp): added warp update to madara and other changes

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Next release
 
+- feat(warp): added warp update to madara
 - fix(log): define RUST_LOG=info by default
 - fix(tracing): RUST_LOG filtering support
 - fix(fgw): fetch class

README.md

@@ -32,7 +32,8 @@ Madara is a powerful Starknet client written in Rust.
 - 🌐 Interactions
   - [Supported JSON-RPC Methods](#supported-json-rpc-methods)
   - [Example of Calling a JSON-RPC Method](#example-of-calling-a-json-rpc-method)
-- Supported Features
+- 🔄 Migration
+- ℹ️ Supported Features
 - 👍 Contribute
 
 ## ⬇️ Installation
@@ -486,7 +487,50 @@ Madara comes packed with OTEL integration, supporting export of traces, metrics
 - Signoz Docker Standalone can be setup following this [guide](https://signoz.io/docs/install/docker/).
 - Ensure to configure the correct service_name after importing the json to the dashboard.
 
-## ✔ Supported Features
+## 🔄 Migration
+
+When migration to a newer version of Madara which introduces breaking changes,
+you might need to update your database. Instead of re-synchronizing the entirety
+of your chain's state from genesis, you can use Madara's **warp update**
+feature.
+
+> [!NOTE]
+> Warp update requires you to already have an existing node with a working
+> database as this will serve as the source for your database migration.
+
+Starting the migration:
+
+```bash
+cargo run --releasae -- \
+  --name Sender         \
+  --full                \ # This also works with other types of nodes
+  --network mainnet     \
+  --warp-sync-sender
+```
+
+Then run:
+
+```bash
+cargo run --releasae --       \
+  --name Receiver             \
+  --base-path /tmp/madara_new \ # Where you want the new database to be stored
+  --full                      \
+  --network mainnet           \
+  --l1-endpoint https://***   \
+  --warp-sync-receiver
+```
+
+This will start generating a new up-to-date database under `/tmp/madara_new`.
+Once this process is over, the warp update sender node will automatically
+shutdown while the warp update receiver will take its place.
+
+> [!WARNING]
+> As of now, the warp update receiver has its rpc disabled, even after the
+> migration process has completed. This will be fixed in the future, so that
+> services that would otherwise conflict with the sender node will automatically
+> start after the migration has finished.
+
+## ℹ️ Supported Features
 
 Madara offers numerous features and is constantly improving to stay at the cutting edge of Starknet technology.
 

crates/client/block_import/src/lib.rs

@@ -119,16 +119,11 @@ pub struct BlockImporter {
     backend: Arc<MadaraBackend>,
     verify_apply: VerifyApply,
     metrics: BlockMetrics,
-    always_force_flush: bool,
 }
 
 impl BlockImporter {
     /// The starting block is used for metrics. Setting it to None means it will look at the database latest block number.
-    pub fn new(
-        backend: Arc<MadaraBackend>,
-        starting_block: Option<u64>,
-        always_force_flush: bool,
-    ) -> anyhow::Result<Self> {
+    pub fn new(backend: Arc<MadaraBackend>, starting_block: Option<u64>) -> anyhow::Result<Self> {
         let pool = Arc::new(RayonPool::new());
         let starting_block = if let Some(n) = starting_block {
             n
@@ -145,7 +140,6 @@ impl BlockImporter {
             pool,
             metrics: BlockMetrics::register(starting_block).context("Registering metrics for block import")?,
             backend,
-            always_force_flush,
         })
     }
 
@@ -176,11 +170,6 @@ impl BlockImporter {
         validation: BlockValidationContext,
     ) -> Result<BlockImportResult, BlockImportError> {
         let result = self.verify_apply.verify_apply(block, validation).await?;
-        // Flush step.
-        let force = self.always_force_flush;
-        self.backend
-            .maybe_flush(force)
-            .map_err(|err| BlockImportError::Internal(format!("DB flushing error: {err:#}").into()))?;
         self.metrics.update(&result.header, &self.backend);
         Ok(result)
     }

crates/client/db/src/lib.rs

@@ -1,19 +1,18 @@
 //! Madara database
 
-use anyhow::{Context, Result};
+use anyhow::Context;
 use bonsai_db::{BonsaiDb, DatabaseKeyMapping};
 use bonsai_trie::id::BasicId;
 use bonsai_trie::{BonsaiStorage, BonsaiStorageConfig};
 use db_metrics::DbMetrics;
 use mp_chain_config::ChainConfig;
-use mp_utils::service::Service;
+use mp_utils::service::{MadaraCapability, Service};
 use rocksdb::backup::{BackupEngine, BackupEngineOptions};
 use rocksdb::{BoundColumnFamily, ColumnFamilyDescriptor, DBWithThreadMode, Env, FlushOptions, MultiThreaded};
 use rocksdb_options::rocksdb_global_options;
 use starknet_types_core::hash::{Pedersen, Poseidon, StarkHash};
 use std::path::{Path, PathBuf};
-use std::sync::{Arc, Mutex};
-use std::time::{Duration, Instant};
+use std::sync::Arc;
 use std::{fmt, fs};
 use tokio::sync::{mpsc, oneshot};
 
@@ -37,7 +36,7 @@ pub type WriteBatchWithTransaction = rocksdb::WriteBatchWithTransaction<false>;
 
 const DB_UPDATES_BATCH_SIZE: usize = 1024;
 
-pub fn open_rocksdb(path: &Path) -> Result<Arc<DB>> {
+pub fn open_rocksdb(path: &Path) -> anyhow::Result<Arc<DB>> {
     let opts = rocksdb_global_options()?;
     tracing::debug!("opening db at {:?}", path.display());
     let db = DB::open_cf_descriptors(
@@ -56,7 +55,7 @@ fn spawn_backup_db_task(
     db_path: &Path,
     db_restored_cb: oneshot::Sender<()>,
     mut recv: mpsc::Receiver<BackupRequest>,
-) -> Result<()> {
+) -> anyhow::Result<()> {
     let mut backup_opts = BackupEngineOptions::new(backup_dir).context("Creating backup options")?;
     let cores = std::thread::available_parallelism().map(|e| e.get() as i32).unwrap_or(1);
     backup_opts.set_max_background_operations(cores);
@@ -254,7 +253,6 @@ impl DatabaseExt for DB {
 pub struct MadaraBackend {
     backup_handle: Option<mpsc::Sender<BackupRequest>>,
     db: Arc<DB>,
-    last_flush_time: Mutex<Option<Instant>>,
     chain_config: Arc<ChainConfig>,
     db_metrics: DbMetrics,
     sender_block_info: tokio::sync::broadcast::Sender<mp_block::MadaraBlockInfo>,
@@ -305,7 +303,11 @@ impl DatabaseService {
     }
 }
 
-impl Service for DatabaseService {}
+impl Service for DatabaseService {
+    fn id(&self) -> MadaraCapability {
+        MadaraCapability::Database
+    }
+}
 
 struct BackupRequest {
     callback: oneshot::Sender<()>,
@@ -315,7 +317,7 @@ struct BackupRequest {
 impl Drop for MadaraBackend {
     fn drop(&mut self) {
         tracing::info!("⏳ Gracefully closing the database...");
-        self.maybe_flush(true).expect("Error when flushing the database"); // flush :)
+        self.flush().expect("Error when flushing the database"); // flush :)
     }
 }
 
@@ -330,7 +332,6 @@ impl MadaraBackend {
         Arc::new(Self {
             backup_handle: None,
             db: open_rocksdb(temp_dir.as_ref()).unwrap(),
-            last_flush_time: Default::default(),
             chain_config,
             db_metrics: DbMetrics::register().unwrap(),
             sender_block_info: tokio::sync::broadcast::channel(100).0,
@@ -344,7 +345,7 @@ impl MadaraBackend {
         backup_dir: Option<PathBuf>,
         restore_from_latest_backup: bool,
         chain_config: Arc<ChainConfig>,
-    ) -> Result<Arc<MadaraBackend>> {
+    ) -> anyhow::Result<Arc<MadaraBackend>> {
         let db_path = db_config_dir.join("db");
 
         // when backups are enabled, a thread is spawned that owns the rocksdb BackupEngine (it is not thread safe) and it receives backup requests using a mpsc channel
@@ -374,7 +375,6 @@ impl MadaraBackend {
             db_metrics: DbMetrics::register().context("Registering db metrics")?,
             backup_handle,
             db,
-            last_flush_time: Default::default(),
             chain_config: Arc::clone(&chain_config),
             sender_block_info: tokio::sync::broadcast::channel(100).0,
             #[cfg(feature = "testing")]
@@ -384,30 +384,21 @@ impl MadaraBackend {
         Ok(backend)
     }
 
-    pub fn maybe_flush(&self, force: bool) -> Result<bool> {
-        let mut inst = self.last_flush_time.lock().expect("poisoned mutex");
-        let will_flush = force
-            || match *inst {
-                Some(inst) => inst.elapsed() >= Duration::from_secs(5),
-                None => true,
-            };
-        if will_flush {
-            tracing::debug!("doing a db flush");
-            let mut opts = FlushOptions::default();
-            opts.set_wait(true);
-            // we have to collect twice here :/
-            let columns = Column::ALL.iter().map(|e| self.db.get_column(*e)).collect::<Vec<_>>();
-            let columns = columns.iter().collect::<Vec<_>>();
-            self.db.flush_cfs_opt(&columns, &opts).context("Flushing database")?;
-
-            *inst = Some(Instant::now());
-        }
+    pub fn flush(&self) -> anyhow::Result<()> {
+        tracing::debug!("doing a db flush");
+        let mut opts = FlushOptions::default();
+        opts.set_wait(true);
+        // we have to collect twice here :/
+        let columns = Column::ALL.iter().map(|e| self.db.get_column(*e)).collect::<Vec<_>>();
+        let columns = columns.iter().collect::<Vec<_>>();
 
-        Ok(will_flush)
+        self.db.flush_cfs_opt(&columns, &opts).context("Flushing database")?;
+
+        Ok(())
     }
 
     #[tracing::instrument(skip(self))]
-    pub async fn backup(&self) -> Result<()> {
+    pub async fn backup(&self) -> anyhow::Result<()> {
         let (callback_sender, callback_recv) = oneshot::channel();
         let _res = self
             .backup_handle

crates/client/devnet/src/lib.rs

@@ -302,7 +302,7 @@ mod tests {
         let chain_config = Arc::new(ChainConfig::madara_devnet());
         let block = g.build(&chain_config).unwrap();
         let backend = MadaraBackend::open_for_testing(Arc::clone(&chain_config));
-        let importer = Arc::new(BlockImporter::new(Arc::clone(&backend), None, true).unwrap());
+        let importer = Arc::new(BlockImporter::new(Arc::clone(&backend), None).unwrap());
 
         println!("{:?}", block.state_diff);
         tokio::runtime::Runtime::new()

crates/client/eth/Cargo.toml

@@ -84,3 +84,4 @@ httpmock = { workspace = true }
 tracing-test = "0.2.5"
 serial_test = { workspace = true }
 lazy_static = { workspace = true }
+mp-utils = { workspace = true, features = ["testing"] }

crates/client/eth/src/l1_gas_price.rs

@@ -5,7 +5,7 @@ use anyhow::Context;
 use mc_mempool::{GasPriceProvider, L1DataProvider};
 use std::time::{Duration, UNIX_EPOCH};
 
-use mp_utils::wait_or_graceful_shutdown;
+use mp_utils::{service::ServiceContext, wait_or_graceful_shutdown};
 use std::time::SystemTime;
 
 pub async fn gas_price_worker_once(
@@ -36,12 +36,12 @@ pub async fn gas_price_worker(
     eth_client: &EthereumClient,
     l1_gas_provider: GasPriceProvider,
     gas_price_poll_ms: Duration,
-    cancellation_token: tokio_util::sync::CancellationToken,
+    ctx: ServiceContext,
 ) -> anyhow::Result<()> {
     l1_gas_provider.update_last_update_timestamp();
     let mut interval = tokio::time::interval(gas_price_poll_ms);
     interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
-    while wait_or_graceful_shutdown(interval.tick(), &cancellation_token).await.is_some() {
+    while wait_or_graceful_shutdown(interval.tick(), &ctx).await.is_some() {
         gas_price_worker_once(eth_client, l1_gas_provider.clone(), gas_price_poll_ms).await?;
     }
     Ok(())
@@ -135,7 +135,7 @@ mod eth_client_gas_price_worker_test {
                     &eth_client,
                     l1_gas_provider,
                     Duration::from_millis(200),
-                    tokio_util::sync::CancellationToken::new(),
+                    ServiceContext::new_for_testing(),
                 )
                 .await
             }
@@ -280,7 +280,7 @@ mod eth_client_gas_price_worker_test {
                 &eth_client,
                 l1_gas_provider.clone(),
                 Duration::from_millis(200),
-                tokio_util::sync::CancellationToken::new(),
+                ServiceContext::new_for_testing(),
             ),
         )
         .await;