Repository: nervosnetwork/ckb (language: rust)

Issue:
Guide users to re-sync only when migration fails.
## Feature Request

#### Is your feature request related to a problem? Please describe.
https://github.com/Magickbase/neuron-public-issues/issues/71#issuecomment-1297020609

The message is misleading. User may think re-sync is faster than migration.

#### Describe the solution you'd like

Remove the following line from the prompt.

```
Another choice is to delete all data, then synchronize them again.
```

Instead, tell users that if the migration fails, try to delete all data and sync again.



Discussion / hints:

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index b3a05d75ad..9fca0ac1f2 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -748,12 +748,15 @@ name = "ckb-db-migration"
 version = "0.114.0-pre"
 dependencies = [
  "ckb-app-config",
+ "ckb-channel",
  "ckb-db",
  "ckb-db-schema",
  "ckb-error",
  "ckb-logger",
+ "ckb-stop-handler",
  "console",
  "indicatif",
+ "once_cell",
  "tempfile",
 ]
 
diff --git a/ckb-bin/src/subcommand/export.rs b/ckb-bin/src/subcommand/export.rs
index 04459d3f67..17f2d05acf 100644
--- a/ckb-bin/src/subcommand/export.rs
+++ b/ckb-bin/src/subcommand/export.rs
@@ -10,8 +10,9 @@ pub fn export(args: ExportArgs, async_handle: Handle) -> Result<(), ExitCode> {
         &args.config.db,
         None,
         async_handle,
+        args.consensus,
     )?;
-    let (shared, _) = builder.consensus(args.consensus).build()?;
+    let (shared, _) = builder.build()?;
     Export::new(shared, args.target).execute().map_err(|err| {
         eprintln!("Export error: {err:?}");
         ExitCode::Failure
diff --git a/ckb-bin/src/subcommand/import.rs b/ckb-bin/src/subcommand/import.rs
index 8c03cfd7f9..d6fba348c3 100644
--- a/ckb-bin/src/subcommand/import.rs
+++ b/ckb-bin/src/subcommand/import.rs
@@ -11,8 +11,9 @@ pub fn import(args: ImportArgs, async_handle: Handle) -> Result<(), ExitCode> {
         &args.config.db,
         None,
         async_handle,
+        args.consensus,
     )?;
-    let (shared, mut pack) = builder.consensus(args.consensus).build()?;
+    let (shared, mut pack) = builder.build()?;
 
     let chain_service = ChainService::new(shared, pack.take_proposal_table());
     let chain_controller = chain_service.start::<&str>(Some("ImportChainService"));
diff --git a/ckb-bin/src/subcommand/migrate.rs b/ckb-bin/src/subcommand/migrate.rs
index c62b4dd47a..1bc0db2f43 100644
--- a/ckb-bin/src/subcommand/migrate.rs
+++ b/ckb-bin/src/subcommand/migrate.rs
@@ -6,7 +6,7 @@ use std::cmp::Ordering;
 use crate::helper::prompt;
 
 pub fn migrate(args: MigrateArgs) -> Result<(), ExitCode> {
-    let migrate = Migrate::new(&args.config.db.path);
+    let migrate = Migrate::new(&args.config.db.path, args.consensus.hardfork_switch);
 
     {
         let read_only_db = migrate.open_read_only_db().map_err(|e| {
@@ -42,10 +42,11 @@ pub fn migrate(args: MigrateArgs) -> Result<(), ExitCode> {
                     let input = prompt("\
                     \n\
                     Doing migration will take quite a long time before CKB could work again.\n\
-                    Another choice is to delete all data, then synchronize them again.\n\
                     \n\
                     Once the migration started, the data will be no longer compatible with all older versions CKB,\n\
                     so we strongly recommended you to backup the old data before migrating.\n\
+                    \n\
+                    If the migration failed, try to delete all data and sync from scratch.\n\
                     \nIf you want to migrate the data, please input YES, otherwise, the current process will exit.\n\
                     > ",
                     );
@@ -67,7 +68,7 @@ pub fn migrate(args: MigrateArgs) -> Result<(), ExitCode> {
     })?;
 
     if let Some(db) = bulk_load_db_db {
-        migrate.migrate(db).map_err(|err| {
+        migrate.migrate(db, false).map_err(|err| {
             eprintln!("Run error: {err:?}");
             ExitCode::Failure
         })?;
diff --git a/ckb-bin/src/subcommand/replay.rs b/ckb-bin/src/subcommand/replay.rs
index 4f524f47db..ac7da08fb2 100644
--- a/ckb-bin/src/subcommand/replay.rs
+++ b/ckb-bin/src/subcommand/replay.rs
@@ -17,9 +17,9 @@ pub fn replay(args: ReplayArgs, async_handle: Handle) -> Result<(), ExitCode> {
         &args.config.db,
         None,
         async_handle.clone(),
+        args.consensus.clone(),
     )?;
     let (shared, _) = shared_builder
-        .consensus(args.consensus.clone())
         .tx_pool_config(args.config.tx_pool.clone())
         .build()?;
 
@@ -44,11 +44,9 @@ pub fn replay(args: ReplayArgs, async_handle: Handle) -> Result<(), ExitCode> {
             &tmp_db_config,
             None,
             async_handle,
+            args.consensus,
         )?;
-        let (tmp_shared, mut pack) = shared_builder
-            .consensus(args.consensus)
-            .tx_pool_config(args.config.tx_pool)
-            .build()?;
+        let (tmp_shared, mut pack) = shared_builder.tx_pool_config(args.config.tx_pool).build()?;
         let chain = ChainService::new(tmp_shared, pack.take_proposal_table());
 
         if let Some((from, to)) = args.profile {
diff --git a/ckb-bin/src/subcommand/stats.rs b/ckb-bin/src/subcommand/stats.rs
index e0edc73fc1..ace9e3084b 100644
--- a/ckb-bin/src/subcommand/stats.rs
+++ b/ckb-bin/src/subcommand/stats.rs
@@ -32,8 +32,9 @@ impl Statics {
             &args.config.db,
             None,
             async_handle,
+            args.consensus,
         )?;
-        let (shared, _) = shared_builder.consensus(args.consensus).build()?;
+        let (shared, _) = shared_builder.build()?;
 
         let tip_number = shared.snapshot().tip_number();
 
diff --git a/db-migration/Cargo.toml b/db-migration/Cargo.toml
index 3fec4f0d9e..b93c936d90 100644
--- a/db-migration/Cargo.toml
+++ b/db-migration/Cargo.toml
@@ -15,6 +15,9 @@ ckb-db = { path = "../db", version = "= 0.114.0-pre" }
 ckb-logger = { path = "../util/logger", version = "= 0.114.0-pre" }
 ckb-error = { path = "../error", version = "= 0.114.0-pre" }
 ckb-db-schema = { path = "../db-schema", version = "= 0.114.0-pre" }
+ckb-channel = { path = "../util/channel", version = "= 0.114.0-pre" }
+ckb-stop-handler = { path = "../util/stop-handler", version = "= 0.114.0-pre" }
+once_cell = "1.8.0"
 indicatif = "0.16"
 console = ">=0.9.1, <1.0.0"
 
diff --git a/db-migration/src/lib.rs b/db-migration/src/lib.rs
index e2243f216f..038bbf1d00 100644
--- a/db-migration/src/lib.rs
+++ b/db-migration/src/lib.rs
@@ -1,13 +1,25 @@
 //! TODO(doc): @quake
+use ckb_channel::select;
+use ckb_channel::unbounded;
+use ckb_channel::Receiver;
 use ckb_db::{ReadOnlyDB, RocksDB};
 use ckb_db_schema::{COLUMN_META, META_TIP_HEADER_KEY, MIGRATION_VERSION_KEY};
 use ckb_error::{Error, InternalErrorKind};
 use ckb_logger::{debug, error, info};
+use ckb_stop_handler::register_thread;
 use console::Term;
 pub use indicatif::{HumanDuration, MultiProgress, ProgressBar, ProgressDrawTarget, ProgressStyle};
+use once_cell::sync::OnceCell;
 use std::cmp::Ordering;
 use std::collections::BTreeMap;
+use std::collections::VecDeque;
 use std::sync::Arc;
+use std::sync::Mutex;
+use std::thread;
+use std::thread::JoinHandle;
+
+/// Shutdown flag for background migration.
+pub static SHUTDOWN_BACKGROUND_MIGRATION: OnceCell<bool> = OnceCell::new();
 
 #[cfg(test)]
 mod tests;
@@ -19,7 +31,66 @@ fn internal_error(reason: String) -> Error {
 /// TODO(doc): @quake
 #[derive(Default)]
 pub struct Migrations {
-    migrations: BTreeMap<String, Box<dyn Migration>>,
+    migrations: BTreeMap<String, Arc<dyn Migration>>,
+}
+
+/// Commands
+#[derive(PartialEq, Eq, Debug)]
+enum Command {
+    Start,
+    Stop,
+}
+
+type MigrationTasks = VecDeque<(String, Arc<dyn Migration>)>;
+struct MigrationWorker {
+    tasks: Arc<Mutex<MigrationTasks>>,
+    db: RocksDB,
+    inbox: Receiver<Command>,
+}
+
+impl MigrationWorker {
+    pub fn new(tasks: Arc<Mutex<MigrationTasks>>, db: RocksDB, inbox: Receiver<Command>) -> Self {
+        Self { tasks, db, inbox }
+    }
+
+    pub fn start(self) -> JoinHandle<()> {
+        thread::spawn(move || {
+            if let Ok(Command::Start) = self.inbox.recv() {
+                let mut idx = 0;
+                let migrations_count = self.tasks.lock().unwrap().len() as u64;
+                let mpb = Arc::new(MultiProgress::new());
+
+                while let Some((name, task)) = self.tasks.lock().unwrap().pop_front() {
+                    select! {
+                        recv(self.inbox) -> msg => {
+                            if let Ok(Command::Stop) = msg {
+                                eprintln!("stop to run migrate in background: {}", name);
+                                break;
+                            }
+                        }
+                        default => {
+                            eprintln!("start to run migrate in background: {}", name);
+                            let mpbc = Arc::clone(&mpb);
+                            idx += 1;
+                            let pb = move |count: u64| -> ProgressBar {
+                                let pb = mpbc.add(ProgressBar::new(count));
+                                pb.set_draw_target(ProgressDrawTarget::term(Term::stdout(), None));
+                                pb.set_prefix(format!("[{}/{}]", idx, migrations_count));
+                                pb
+                            };
+                            if let Ok(db) = task.migrate(self.db.clone(), Arc::new(pb)) {
+                                db.put_default(MIGRATION_VERSION_KEY, task.version())
+                                .map_err(|err| {
+                                    internal_error(format!("failed to migrate the database: {err}"))
+                                })
+                                .unwrap();
+                            }
+                        }
+                    }
+                }
+            }
+        })
+    }
 }
 
 impl Migrations {
@@ -31,7 +102,7 @@ impl Migrations {
     }
 
     /// TODO(doc): @quake
-    pub fn add_migration(&mut self, migration: Box<dyn Migration>) {
+    pub fn add_migration(&mut self, migration: Arc<dyn Migration>) {
         self.migrations
             .insert(migration.version().to_string(), migration);
     }
@@ -97,6 +168,28 @@ impl Migrations {
             .any(|m| m.expensive())
     }
 
+    /// Check if all the pending migrations will be executed in background.
+    pub fn can_run_in_background(&self, db: &ReadOnlyDB) -> bool {
+        let db_version = match db
+            .get_pinned_default(MIGRATION_VERSION_KEY)
+            .expect("get the version of database")
+        {
+            Some(version_bytes) => {
+                String::from_utf8(version_bytes.to_vec()).expect("version bytes to utf8")
+            }
+            None => {
+                // if version is none, but db is not empty
+                // patch 220464f
+                return self.is_non_empty_rdb(db);
+            }
+        };
+
+        self.migrations
+            .values()
+            .skip_while(|m| m.version() <= db_version.as_str())
+            .all(|m| m.run_in_background())
+    }
+
     fn is_non_empty_rdb(&self, db: &ReadOnlyDB) -> bool {
         if let Ok(v) = db.get_pinned(COLUMN_META, META_TIP_HEADER_KEY) {
             if v.is_some() {
@@ -139,6 +232,38 @@ impl Migrations {
         Ok(db)
     }
 
+    fn run_migrate_async(&self, db: RocksDB, v: &str) {
+        let migrations: VecDeque<(String, Arc<dyn Migration>)> = self
+            .migrations
+            .iter()
+            .filter(|(mv, _)| mv.as_str() > v)
+            .map(|(mv, m)| (mv.to_string(), Arc::clone(m)))
+            .collect::<VecDeque<_>>();
+
+        let all_can_resume = migrations.iter().all(|(_, m)| m.can_resume());
+        let tasks = Arc::new(Mutex::new(migrations));
+        let (tx, rx) = unbounded();
+        let worker = MigrationWorker::new(tasks, db.clone(), rx);
+
+        let exit_signal = ckb_stop_handler::new_crossbeam_exit_rx();
+        let clone = v.to_string();
+        let tx_clone = tx.clone();
+        let notifier = thread::spawn(move || {
+            let _ = exit_signal.recv();
+            let res = SHUTDOWN_BACKGROUND_MIGRATION.set(true);
+            let _ = tx_clone.send(Command::Stop);
+            info!("set shutdown flag to true: {:?} version: {}", res, clone);
+        });
+        register_thread("migration-notifier", notifier);
+
+        let handler = worker.start();
+        if all_can_resume {
+            info!("register thread: migration ....");
+            register_thread("migration", handler);
+        }
+        tx.send(Command::Start).expect("send start command");
+    }
+
     fn get_migration_version(&self, db: &RocksDB) -> Result<Option<String>, Error> {
         let raw = db
             .get_pinned_default(MIGRATION_VERSION_KEY)
@@ -167,25 +292,18 @@ impl Migrations {
     }
 
     /// TODO(doc): @quake
-    pub fn migrate(&self, db: RocksDB) -> Result<RocksDB, Error> {
+    pub fn migrate(&self, db: RocksDB, run_in_background: bool) -> Result<RocksDB, Error> {
         let db_version = self.get_migration_version(&db)?;
         match db_version {
             Some(ref v) => {
                 info!("Current database version {}", v);
-                if let Some(m) = self.migrations.values().last() {
-                    if m.version() < v.as_str() {
-                        error!(
-                            "Database downgrade detected. \
-                            The database schema version is more recent than the client schema version.\
-                            Please upgrade to the latest client version."
-                        );
-                        return Err(internal_error(
-                            "Database downgrade is not supported".to_string(),
-                        ));
-                    }
-                }
-
-                let db = self.run_migrate(db, v.as_str())?;
+                self.check_migration_downgrade(v)?;
+                let db = if !run_in_background {
+                    self.run_migrate(db, v.as_str())?
+                } else {
+                    self.run_migrate_async(db.clone(), v.as_str());
+                    db
+                };
                 Ok(db)
             }
             None => {
@@ -203,10 +321,26 @@ impl Migrations {
         const V: &str = "20210609195048"; // AddExtraDataHash - 1
         self.run_migrate(db, V)
     }
+
+    fn check_migration_downgrade(&self, cur_version: &str) -> Result<(), Error> {
+        if let Some(m) = self.migrations.values().last() {
+            if m.version() < cur_version {
+                error!(
+                    "Database downgrade detected. \
+                    The database schema version is newer than `ckb` schema version,\
+                    please upgrade `ckb` to the latest version"
+                );
+                return Err(internal_error(
+                    "Database downgrade is not supported".to_string(),
+                ));
+            }
+        }
+        Ok(())
+    }
 }
 
 /// TODO(doc): @quake
-pub trait Migration {
+pub trait Migration: Send + Sync {
     /// TODO(doc): @quake
     fn migrate(
         &self,
@@ -223,6 +357,30 @@ pub trait Migration {
     fn expensive(&self) -> bool {
         true
     }
+
+    /// Will this migration be executed in background.
+    fn run_in_background(&self) -> bool {
+        false
+    }
+
+    /// Check if the background migration should be stopped.
+    /// If a migration need to implement the recovery logic, it should check this flag periodically,
+    /// store the migration progress when exiting and recover from the current progress when restarting.
+    fn stop_background(&self) -> bool {
+        *SHUTDOWN_BACKGROUND_MIGRATION.get().unwrap_or(&false)
+    }
+
+    /// Check if the background migration can be resumed.
+    ///
+    /// If a migration can be resumed, it should implement the recovery logic in `migrate` function.
+    /// and the `MigirateWorker` will add the migration's handler with `register_thread`, so that then
+    /// main thread can wait for the background migration to store the progress and exit.
+    ///
+    /// Otherwise, the migration will be restarted from the beginning.
+    ///
+    fn can_resume(&self) -> bool {
+        false
+    }
 }
 
 /// TODO(doc): @quake
diff --git a/shared/src/shared_builder.rs b/shared/src/shared_builder.rs
index 7f5a0673bb..a0d3476ce0 100644
--- a/shared/src/shared_builder.rs
+++ b/shared/src/shared_builder.rs
@@ -23,6 +23,7 @@ use ckb_logger::{error, info};
 use ckb_migrate::migrate::Migrate;
 use ckb_notify::{NotifyController, NotifyService};
 use ckb_store::{ChainDB, ChainStore, Freezer};
+use ckb_types::core::hardfork::HardForks;
 use ckb_types::core::service::PoolTransactionEntry;
 use ckb_types::core::tx_pool::Reject;
 use ckb_types::core::EpochExt;
@@ -38,7 +39,7 @@ use tempfile::TempDir;
 pub struct SharedBuilder {
     db: RocksDB,
     ancient_path: Option<PathBuf>,
-    consensus: Option<Consensus>,
+    consensus: Consensus,
     tx_pool_config: Option<TxPoolConfig>,
     store_config: Option<StoreConfig>,
     block_assembler_config: Option<BlockAssemblerConfig>,
@@ -51,8 +52,9 @@ pub fn open_or_create_db(
     bin_name: &str,
     root_dir: &Path,
     config: &DBConfig,
+    hardforks: HardForks,
 ) -> Result<RocksDB, ExitCode> {
-    let migrate = Migrate::new(&config.path);
+    let migrate = Migrate::new(&config.path, hardforks);
 
     let read_only_db = migrate.open_read_only_db().map_err(|e| {
         eprintln!("Migration error {e}");
@@ -71,17 +73,27 @@ pub fn open_or_create_db(
             }
             Ordering::Equal => Ok(RocksDB::open(config, COLUMNS)),
             Ordering::Less => {
-                if migrate.require_expensive(&db) {
+                let can_run_in_background = migrate.can_run_in_background(&db);
+                eprintln!("can_run_in_background: {}", can_run_in_background);
+                if migrate.require_expensive(&db) && !can_run_in_background {
                     eprintln!(
                         "For optimal performance, CKB recommends migrating your data into a new format.\n\
                         If you prefer to stick with the older version, \n\
                         it's important to note that they may have unfixed vulnerabilities.\n\
-                        Before migrating, we strongly recommend backuping your data directory.
+                        Before migrating, we strongly recommend backuping your data directory.\n\
                         To migrate, run `\"{}\" migrate -C \"{}\"` and confirm by typing \"YES\".",
                         bin_name,
                         root_dir.display()
                     );
                     Err(ExitCode::Failure)
+                } else if can_run_in_background {
+                    info!("process migrations in background ...");
+                    let db = RocksDB::open(config, COLUMNS);
+                    migrate.migrate(db.clone(), true).map_err(|err| {
+                        eprintln!("Run error: {err:?}");
+                        ExitCode::Failure
+                    })?;
+                    Ok(db)
                 } else {
                     info!("Processing fast migrations ...");
 
@@ -91,7 +103,7 @@ pub fn open_or_create_db(
                     })?;
 
                     if let Some(db) = bulk_load_db_db {
-                        migrate.migrate(db).map_err(|err| {
+                        migrate.migrate(db, false).map_err(|err| {
                             eprintln!("Run error: {err:?}");
                             ExitCode::Failure
                         })?;
@@ -119,13 +131,19 @@ impl SharedBuilder {
         db_config: &DBConfig,
         ancient: Option<PathBuf>,
         async_handle: Handle,
+        consensus: Consensus,
     ) -> Result<SharedBuilder, ExitCode> {
-        let db = open_or_create_db(bin_name, root_dir, db_config)?;
+        let db = open_or_create_db(
+            bin_name,
+            root_dir,
+            db_config,
+            consensus.hardfork_switch.clone(),
+        )?;
 
         Ok(SharedBuilder {
             db,
             ancient_path: ancient,
-            consensus: None,
+            consensus,
             tx_pool_config: None,
             notify_config: None,
             store_config: None,
@@ -170,7 +188,7 @@ impl SharedBuilder {
         RUNTIME_HANDLE.with(|runtime| SharedBuilder {
             db,
             ancient_path: None,
-            consensus: None,
+            consensus: Consensus::default(),
             tx_pool_config: None,
             notify_config: None,
             store_config: None,
@@ -183,7 +201,7 @@ impl SharedBuilder {
 impl SharedBuilder {
     /// TODO(doc): @quake
     pub fn consensus(mut self, value: Consensus) -> Self {
-        self.consensus = Some(value);
+        self.consensus = value;
         self
     }
 
@@ -316,7 +334,7 @@ impl SharedBuilder {
         let tx_pool_config = tx_pool_config.unwrap_or_default();
         let notify_config = notify_config.unwrap_or_default();
         let store_config = store_config.unwrap_or_default();
-        let consensus = Arc::new(consensus.unwrap_or_default());
+        let consensus = Arc::new(consensus);
 
         let notify_controller = start_notify_service(notify_config, async_handle.clone());
 
diff --git a/util/launcher/src/lib.rs b/util/launcher/src/lib.rs
index f00dd801aa..db3af82673 100644
--- a/util/launcher/src/lib.rs
+++ b/util/launcher/src/lib.rs
@@ -195,10 +195,10 @@ impl Launcher {
             &self.args.config.db,
             Some(self.args.config.ancient.clone()),
             self.async_handle.clone(),
+            self.args.consensus.clone(),
         )?;
 
         let (shared, pack) = shared_builder
-            .consensus(self.args.consensus.clone())
             .tx_pool_config(self.args.config.tx_pool.clone())
             .notify_config(self.args.config.notify.clone())
             .store_config(self.args.config.store)
diff --git a/util/migrate/src/migrate.rs b/util/migrate/src/migrate.rs
index 99d6a65b7e..35d3c3a880 100644
--- a/util/migrate/src/migrate.rs
+++ b/util/migrate/src/migrate.rs
@@ -5,8 +5,10 @@ use ckb_db::{ReadOnlyDB, RocksDB};
 use ckb_db_migration::{DefaultMigration, Migrations};
 use ckb_db_schema::{COLUMNS, COLUMN_META};
 use ckb_error::Error;
+use ckb_types::core::hardfork::HardForks;
 use std::cmp::Ordering;
 use std::path::PathBuf;
+use std::sync::Arc;
 
 const INIT_DB_VERSION: &str = "20191127135521";
 
@@ -18,17 +20,18 @@ pub struct Migrate {
 
 impl Migrate {
     /// Construct new migrate
-    pub fn new<P: Into<PathBuf>>(path: P) -> Self {
+    pub fn new<P: Into<PathBuf>>(path: P, hardforks: HardForks) -> Self {
         let mut migrations = Migrations::default();
-        migrations.add_migration(Box::new(DefaultMigration::new(INIT_DB_VERSION)));
-        migrations.add_migration(Box::new(migrations::ChangeMoleculeTableToStruct)); // since v0.35.0
-        migrations.add_migration(Box::new(migrations::CellMigration)); // since v0.37.0
-        migrations.add_migration(Box::new(migrations::AddNumberHashMapping)); // since v0.40.0
-        migrations.add_migration(Box::new(migrations::AddExtraDataHash)); // since v0.43.0
-        migrations.add_migration(Box::new(migrations::AddBlockExtensionColumnFamily)); // since v0.100.0
-        migrations.add_migration(Box::new(migrations::AddChainRootMMR)); // TODO(light-client) update the comment: which version?
-        migrations.add_migration(Box::new(migrations::AddBlockFilterColumnFamily)); // since v0.105.0
-        migrations.add_migration(Box::new(migrations::AddBlockFilterHash)); // since v0.108.0
+        migrations.add_migration(Arc::new(DefaultMigration::new(INIT_DB_VERSION)));
+        migrations.add_migration(Arc::new(migrations::ChangeMoleculeTableToStruct)); // since v0.35.0
+        migrations.add_migration(Arc::new(migrations::CellMigration)); // since v0.37.0
+        migrations.add_migration(Arc::new(migrations::AddNumberHashMapping)); // since v0.40.0
+        migrations.add_migration(Arc::new(migrations::AddExtraDataHash)); // since v0.43.0
+        migrations.add_migration(Arc::new(migrations::AddBlockExtensionColumnFamily)); // since v0.100.0
+        migrations.add_migration(Arc::new(migrations::AddChainRootMMR)); // TODO(light-client) update the comment: which version?
+        migrations.add_migration(Arc::new(migrations::AddBlockFilterColumnFamily)); // since v0.105.0
+        migrations.add_migration(Arc::new(migrations::AddBlockFilterHash)); // since v0.108.0
+        migrations.add_migration(Arc::new(migrations::BlockExt2019ToZero::new(hardforks))); // since v0.111.1
 
         Migrate {
             migrations,
@@ -59,14 +62,19 @@ impl Migrate {
         self.migrations.expensive(db)
     }
 
+    /// Check whether the pending migrations are all background migrations.
+    pub fn can_run_in_background(&self, db: &ReadOnlyDB) -> bool {
+        self.migrations.can_run_in_background(db)
+    }
+
     /// Open bulk load db.
     pub fn open_bulk_load_db(&self) -> Result<Option<RocksDB>, Error> {
         RocksDB::prepare_for_bulk_load_open(&self.path, COLUMNS)
     }
 
     /// Perform migrate.
-    pub fn migrate(self, db: RocksDB) -> Result<RocksDB, Error> {
-        self.migrations.migrate(db)
+    pub fn migrate(self, db: RocksDB, run_in_background: bool) -> Result<RocksDB, Error> {
+        self.migrations.migrate(db, run_in_background)
     }
 
     /// Perform init_db_version.
diff --git a/util/migrate/src/migrations/mod.rs b/util/migrate/src/migrations/mod.rs
index 37e6136371..bc6b1e4338 100644
--- a/util/migrate/src/migrations/mod.rs
+++ b/util/migrate/src/migrations/mod.rs
@@ -5,6 +5,7 @@ mod add_chain_root_mmr;
 mod add_extra_data_hash;
 mod add_number_hash_mapping;
 mod cell;
+mod set_2019_block_cycle_zero;
 mod table_to_struct;
 
 pub use add_block_extension_cf::AddBlockExtensionColumnFamily;
@@ -14,4 +15,5 @@ pub use add_chain_root_mmr::AddChainRootMMR;
 pub use add_extra_data_hash::AddExtraDataHash;
 pub use add_number_hash_mapping::AddNumberHashMapping;
 pub use cell::CellMigration;
+pub use set_2019_block_cycle_zero::BlockExt2019ToZero;
 pub use table_to_struct::ChangeMoleculeTableToStruct;
diff --git a/util/migrate/src/migrations/set_2019_block_cycle_zero.rs b/util/migrate/src/migrations/set_2019_block_cycle_zero.rs
new file mode 100644
index 0000000000..e45f6fd253
--- /dev/null
+++ b/util/migrate/src/migrations/set_2019_block_cycle_zero.rs
@@ -0,0 +1,114 @@
+use ckb_app_config::StoreConfig;
+use ckb_db_migration::{Migration, ProgressBar, ProgressStyle};
+use ckb_db_schema::COLUMN_EPOCH;
+use ckb_error::InternalErrorKind;
+use ckb_store::{ChainDB, ChainStore};
+use ckb_types::{
+    core::hardfork::HardForks,
+    packed,
+    prelude::{Entity, FromSliceShouldBeOk, Pack, Reader},
+};
+
+const VERSION: &str = "20231101000000";
+pub struct BlockExt2019ToZero {
+    hardforks: HardForks,
+}
+
+impl BlockExt2019ToZero {
+    pub fn new(hardforks: HardForks) -> Self {
+        BlockExt2019ToZero { hardforks }
+    }
+}
+
+impl Migration for BlockExt2019ToZero {
+    fn run_in_background(&self) -> bool {
+        true
+    }
+
+    fn migrate(
+        &self,
+        db: ckb_db::RocksDB,
+        pb: std::sync::Arc<dyn Fn(u64) -> ProgressBar + Send + Sync>,
+    ) -> Result<ckb_db::RocksDB, ckb_error::Error> {
+        let chain_db = ChainDB::new(db, StoreConfig::default());
+        let limit_epoch = self.hardforks.ckb2021.rfc_0032();
+
+        eprintln!(
+            "begin to run block_ext 2019 to zero migrate...: {}",
+            limit_epoch
+        );
+
+        if limit_epoch == 0 {
+            return Ok(chain_db.into_inner());
+        }
+
+        let hard_fork_epoch_number: packed::Uint64 = limit_epoch.pack();
+        let tip_header = chain_db.get_tip_header().expect("db must have tip header");
+        let tip_epoch_number = tip_header.epoch().pack();
+
+        let header = if tip_epoch_number < hard_fork_epoch_number {
+            Some(tip_header)
+        } else if let Some(epoch_hash) =
+            chain_db.get(COLUMN_EPOCH, hard_fork_epoch_number.as_slice())
+        {
+            let epoch_ext = chain_db
+                .get_epoch_ext(
+                    &packed::Byte32Reader::from_slice_should_be_ok(epoch_hash.as_ref()).to_entity(),
+                )
+                .expect("db must have epoch ext");
+            let header = chain_db
+                .get_block_header(&epoch_ext.last_block_hash_in_previous_epoch())
+                .expect("db must have header");
+            Some(header)
+        } else {
+            None
+        };
+
+        if let Some(mut header) = header {
+            let pb = ::std::sync::Arc::clone(&pb);
+            let pbi = pb(header.number() + 1);
+            pbi.set_style(
+                        ProgressStyle::default_bar()
+                            .template(
+                                "{prefix:.bold.dim} {spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {pos}/{len} ({eta}) {msg}",
+                            )
+                            .progress_chars("#>-"),
+                    );
+            pbi.set_position(0);
+            pbi.enable_steady_tick(5000);
+
+            loop {
+                let db_txn = chain_db.begin_transaction();
+                if self.stop_background() {
+                    return Err(InternalErrorKind::Database.other("intrupted").into());
+                }
+                for _ in 0..10000 {
+                    let hash = header.hash();
+                    let mut old_block_ext = db_txn.get_block_ext(&hash).unwrap();
+                    old_block_ext.cycles = None;
+                    db_txn.insert_block_ext(&hash, &old_block_ext)?;
+
+                    if header.is_genesis() {
+                        break;
+                    }
+
+                    header = db_txn
+                        .get_block_header(&header.parent_hash())
+                        .expect("db must have header");
+
+                    pbi.inc(1);
+                }
+                db_txn.commit()?;
+
+                if header.is_genesis() {
+                    break;
+                }
+            }
+        }
+
+        Ok(chain_db.into_inner())
+    }
+    fn version(&self) -> &str {
+        VERSION
+    }
+}
diff --git a/util/stop-handler/src/stop_register.rs b/util/stop-handler/src/stop_register.rs
index f508a75dcf..ccc3d9a6f4 100644
--- a/util/stop-handler/src/stop_register.rs
+++ b/util/stop-handler/src/stop_register.rs
@@ -68,7 +68,7 @@ pub fn broadcast_exit_signals() {
         });
 }
 
-/// Register a thread `JoinHandle` to  `CKB_HANDLES`
+/// Register a thread `JoinHandle` to `CKB_HANDLES`
 pub fn register_thread(name: &str, thread_handle: std::thread::JoinHandle<()>) {
     trace!("Registering thread {}", name);
     CKB_HANDLES


Test patch:
diff --git a/db-migration/src/tests.rs b/db-migration/src/tests.rs
index 9d52ea661a..b3a6fa18c3 100644
--- a/db-migration/src/tests.rs
+++ b/db-migration/src/tests.rs
@@ -1,4 +1,5 @@
 use ckb_app_config::DBConfig;
+use ckb_db::ReadOnlyDB;
 use ckb_db::RocksDB;
 use ckb_db_schema::MIGRATION_VERSION_KEY;
 use ckb_error::Error;
@@ -19,10 +20,10 @@ fn test_default_migration() {
     };
     {
         let mut migrations = Migrations::default();
-        migrations.add_migration(Box::new(DefaultMigration::new("20191116225943")));
+        migrations.add_migration(Arc::new(DefaultMigration::new("20191116225943")));
         let db = RocksDB::open(&config, 1);
         migrations.init_db_version(&db).unwrap();
-        let r = migrations.migrate(db).unwrap();
+        let r = migrations.migrate(db, false).unwrap();
         assert_eq!(
             b"20191116225943".to_vec(),
             r.get_pinned_default(MIGRATION_VERSION_KEY)
@@ -33,9 +34,11 @@ fn test_default_migration() {
     }
     {
         let mut migrations = Migrations::default();
-        migrations.add_migration(Box::new(DefaultMigration::new("20191116225943")));
-        migrations.add_migration(Box::new(DefaultMigration::new("20191127101121")));
-        let r = migrations.migrate(RocksDB::open(&config, 1)).unwrap();
+        migrations.add_migration(Arc::new(DefaultMigration::new("20191116225943")));
+        migrations.add_migration(Arc::new(DefaultMigration::new("20191127101121")));
+        let r = migrations
+            .migrate(RocksDB::open(&config, 1), false)
+            .unwrap();
         assert_eq!(
             b"20191127101121".to_vec(),
             r.get_pinned_default(MIGRATION_VERSION_KEY)
@@ -87,10 +90,10 @@ fn test_customized_migration() {
 
     {
         let mut migrations = Migrations::default();
-        migrations.add_migration(Box::new(DefaultMigration::new("20191116225943")));
+        migrations.add_migration(Arc::new(DefaultMigration::new("20191116225943")));
         let db = RocksDB::open(&config, 1);
         migrations.init_db_version(&db).unwrap();
-        let db = migrations.migrate(db).unwrap();
+        let db = migrations.migrate(db, false).unwrap();
 
         let txn = db.transaction();
         txn.put(COLUMN, &[1, 1], &[1, 1, 1]).unwrap();
@@ -99,9 +102,11 @@ fn test_customized_migration() {
     }
     {
         let mut migrations = Migrations::default();
-        migrations.add_migration(Box::new(DefaultMigration::new("20191116225943")));
-        migrations.add_migration(Box::new(CustomizedMigration));
-        let db = migrations.migrate(RocksDB::open(&config, 1)).unwrap();
+        migrations.add_migration(Arc::new(DefaultMigration::new("20191116225943")));
+        migrations.add_migration(Arc::new(CustomizedMigration));
+        let db = migrations
+            .migrate(RocksDB::open(&config, 1), false)
+            .unwrap();
         assert!(
             vec![1u8, 1, 1, 1].as_slice()
                 == db.get_pinned(COLUMN, &[1, 1]).unwrap().unwrap().as_ref()
@@ -120,3 +125,178 @@ fn test_customized_migration() {
         );
     }
 }
+
+#[test]
+fn test_background_migration() {
+    use ckb_stop_handler::broadcast_exit_signals;
+
+    pub struct BackgroundMigration {
+        version: String,
+    }
+
+    impl BackgroundMigration {
+        pub fn new(version: &str) -> Self {
+            BackgroundMigration {
+                version: version.to_string(),
+            }
+        }
+    }
+
+    impl Migration for BackgroundMigration {
+        fn run_in_background(&self) -> bool {
+            true
+        }
+
+        fn migrate(
+            &self,
+            db: RocksDB,
+            _pb: Arc<dyn Fn(u64) -> ProgressBar + Send + Sync>,
+        ) -> Result<RocksDB, Error> {
+            let db_tx = db.transaction();
+            let v = self.version.as_bytes();
+            db_tx.put("1", v, &[1])?;
+            db_tx.commit()?;
+            Ok(db)
+        }
+
+        fn version(&self) -> &str {
+            self.version.as_str()
+        }
+    }
+
+    pub struct RunStopMigration {
+        version: String,
+    }
+    impl Migration for RunStopMigration {
+        fn run_in_background(&self) -> bool {
+            true
+        }
+
+        fn migrate(
+            &self,
+            db: RocksDB,
+            _pb: Arc<dyn Fn(u64) -> ProgressBar + Send + Sync>,
+        ) -> Result<RocksDB, Error> {
+            let db_tx = db.transaction();
+            loop {
+                if self.stop_background() {
+                    let v = self.version.as_bytes();
+                    db_tx.put("1", v, &[2])?;
+                    db_tx.commit()?;
+                    return Ok(db);
+                }
+            }
+        }
+
+        fn version(&self) -> &str {
+            self.version.as_str()
+        }
+    }
+
+    let tmp_dir = tempfile::Builder::new()
+        .prefix("test_default_migration")
+        .tempdir()
+        .unwrap();
+    let config = DBConfig {
+        path: tmp_dir.as_ref().to_path_buf(),
+        ..Default::default()
+    };
+    {
+        let mut migrations = Migrations::default();
+        migrations.add_migration(Arc::new(DefaultMigration::new("20191116225943")));
+        let db = RocksDB::open(&config, 12);
+        migrations.init_db_version(&db).unwrap();
+        let r = migrations.migrate(db, false).unwrap();
+        assert_eq!(
+            b"20191116225943".to_vec(),
+            r.get_pinned_default(MIGRATION_VERSION_KEY)
+                .unwrap()
+                .unwrap()
+                .to_vec()
+        );
+    }
+    {
+        let mut migrations = Migrations::default();
+        migrations.add_migration(Arc::new(DefaultMigration::new("20191116225943")));
+        migrations.add_migration(Arc::new(BackgroundMigration::new("20231127101121")));
+
+        let db = ReadOnlyDB::open_cf(&config.path, vec!["4"])
+            .unwrap()
+            .unwrap();
+
+        assert!(migrations.can_run_in_background(&db));
+        migrations.add_migration(Arc::new(DefaultMigration::new("20191127101121")));
+        assert!(!migrations.can_run_in_background(&db));
+    }
+
+    {
+        let mut migrations = Migrations::default();
+        migrations.add_migration(Arc::new(DefaultMigration::new("20191116225943")));
+        migrations.add_migration(Arc::new(BackgroundMigration::new("20231127101121")));
+        migrations.add_migration(Arc::new(BackgroundMigration::new("20241127101122")));
+
+        let db = ReadOnlyDB::open_cf(&config.path, vec!["4"])
+            .unwrap()
+            .unwrap();
+
+        assert!(migrations.can_run_in_background(&db));
+        let db = migrations
+            .migrate(RocksDB::open(&config, 12), true)
+            .unwrap();
+
+        // wait for background migration to finish
+        std::thread::sleep(std::time::Duration::from_millis(1000));
+        assert_eq!(
+            b"20241127101122".to_vec(),
+            db.get_pinned_default(MIGRATION_VERSION_KEY)
+                .unwrap()
+                .unwrap()
+                .to_vec()
+        );
+
+        // confirm the background migration is executed
+        let db_tx = db.transaction();
+        let v = db_tx
+            .get_pinned("1", "20231127101121".as_bytes())
+            .unwrap()
+            .unwrap()
+            .to_vec();
+        assert_eq!(v, vec![1]);
+
+        let v = db_tx
+            .get_pinned("1", "20241127101122".as_bytes())
+            .unwrap()
+            .unwrap()
+            .to_vec();
+        assert_eq!(v, vec![1]);
+    }
+
+    {
+        let mut migrations = Migrations::default();
+        migrations.add_migration(Arc::new(RunStopMigration {
+            version: "20251116225943".to_string(),
+        }));
+
+        let db = ReadOnlyDB::open_cf(&config.path, vec!["4"])
+            .unwrap()
+            .unwrap();
+
+        assert!(migrations.can_run_in_background(&db));
+        let db = migrations
+            .migrate(RocksDB::open(&config, 12), true)
+            .unwrap();
+
+        std::thread::sleep(std::time::Duration::from_millis(100));
+        //send stop signal
+        broadcast_exit_signals();
+        std::thread::sleep(std::time::Duration::from_millis(200));
+
+        let db_tx = db.transaction();
+        let v = db_tx
+            .get_pinned("1", "20251116225943".as_bytes())
+            .unwrap()
+            .unwrap()
+            .to_vec();
+        assert_eq!(v, vec![2]);
+    }
+}
diff --git a/util/migrate/src/tests.rs b/util/migrate/src/tests.rs
index ba68c542d6..1d48487a01 100644
--- a/util/migrate/src/tests.rs
+++ b/util/migrate/src/tests.rs
@@ -9,7 +9,9 @@ use ckb_db_schema::{
 };
 use ckb_systemtime::unix_time_as_millis;
 use ckb_types::{
-    core::{capacity_bytes, BlockBuilder, BlockExt, Capacity, TransactionBuilder},
+    core::{
+        capacity_bytes, hardfork::HardForks, BlockBuilder, BlockExt, Capacity, TransactionBuilder,
+    },
     packed::{self, Bytes},
     prelude::*,
     utilities::DIFF_TWO,
@@ -149,13 +151,13 @@ fn test_mock_migration() {
     drop(db_txn);
     drop(db);
 
-    let mg = Migrate::new(tmp_dir.as_ref().to_path_buf());
+    let mg = Migrate::new(tmp_dir.as_ref().to_path_buf(), HardForks::new_mirana());
 
     let db = mg.open_bulk_load_db().unwrap().unwrap();
 
-    mg.migrate(db).unwrap();
+    mg.migrate(db, false).unwrap();
 
-    let mg2 = Migrate::new(tmp_dir.as_ref().to_path_buf());
+    let mg2 = Migrate::new(tmp_dir.as_ref().to_path_buf(), HardForks::new_mirana());
 
     let rdb = mg2.open_read_only_db().unwrap().unwrap();