feat: improve migration progress display

ckb-bin/src/subcommand/replay.rs

@@ -44,7 +44,6 @@ pub fn replay(args: ReplayArgs) -> Result<(), ExitCode> {
         if let Some((from, to)) = args.profile {
             profile(shared, chain, from, to);
         } else if args.sanity_check {
-            ckb_logger_service::silent();
             sanity_check(shared, chain, args.full_verfication);
         }
     }

db-migration/Cargo.toml

@@ -11,6 +11,8 @@ edition = "2018"
 ckb-db = { path = "../db" }
 ckb-logger = { path = "../util/logger" }
 ckb-error = { path = "../error" }
+indicatif = "0.15"
+console = "0.12"
 
 [dev-dependencies]
 tempfile = "3.0"

db-migration/src/lib.rs

@@ -1,6 +1,10 @@
 use ckb_db::RocksDB;
 use ckb_error::{Error, InternalErrorKind};
-use ckb_logger::info;
+use ckb_logger::{error, info};
+use console::Term;
+pub use indicatif::{HumanDuration, MultiProgress, ProgressBar, ProgressDrawTarget, ProgressStyle};
+use std::collections::BTreeMap;
+use std::rc::Rc;
 
 pub const VERSION_KEY: &[u8] = b"db-version";
 
@@ -10,16 +14,19 @@ fn internal_error(reason: String) -> Error {
 
 #[derive(Default)]
 pub struct Migrations {
-    migrations: Vec<Box<dyn Migration>>,
+    migrations: BTreeMap<String, Box<dyn Migration>>,
 }
 
 impl Migrations {
     pub fn new() -> Self {
-        Migrations { migrations: vec![] }
+        Migrations {
+            migrations: BTreeMap::new(),
+        }
     }
 
     pub fn add_migration(&mut self, migration: Box<dyn Migration>) {
-        self.migrations.push(migration);
+        self.migrations
+            .insert(migration.version().to_string(), migration);
     }
 
     pub fn migrate(&self, mut db: RocksDB) -> Result<RocksDB, Error> {
@@ -33,19 +40,48 @@ impl Migrations {
             });
 
         match db_version {
-            Some(v) => {
-                for m in self.migrations.iter().filter(|m| m.version() > v.as_str()) {
-                    db = m.migrate(db)?;
-                    db.put(VERSION_KEY, m.version()).map_err(|err| {
+            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 newer than client schema version,\
+                            please upgrade to the newer version"
+                        );
+                        return Err(internal_error(
+                            "Database downgrade is not supported".to_string(),
+                        ));
+                    }
+                }
+
+                let mpb = Rc::new(MultiProgress::new());
+                let migrations: BTreeMap<_, _> = self
+                    .migrations
+                    .iter()
+                    .filter(|(mv, _)| mv.as_str() > v.as_str())
+                    .collect();
+                let migrations_count = migrations.len();
+                for (idx, (_, m)) in migrations.iter().enumerate() {
+                    let mpbc = Rc::clone(&mpb);
+                    let pb = move |count: u64| -> ProgressBar {
+                        let pb = mpbc.add(ProgressBar::new(count));
+                        pb.set_draw_target(ProgressDrawTarget::to_term(Term::stdout(), None));
+                        pb.set_prefix(&format!("[{}/{}]", idx + 1, migrations_count));
+                        pb
+                    };
+                    db = m.migrate(db, Box::new(pb))?;
+                    db.put_default(VERSION_KEY, m.version()).map_err(|err| {
                         internal_error(format!("failed to migrate the database: {}", err))
                     })?;
                 }
+                mpb.join_and_clear().expect("MultiProgress join");
                 Ok(db)
             }
             None => {
-                if let Some(m) = self.migrations.last() {
+                if let Some(m) = self.migrations.values().last() {
                     info!("Init database version {}", m.version());
-                    db.put(VERSION_KEY, m.version()).map_err(|err| {
+                    db.put_default(VERSION_KEY, m.version()).map_err(|err| {
                         internal_error(format!("failed to migrate the database: {}", err))
                     })?;
                 }
@@ -56,9 +92,13 @@ impl Migrations {
 }
 
 pub trait Migration {
-    fn migrate(&self, _db: RocksDB) -> Result<RocksDB, Error>;
+    fn migrate(
+        &self,
+        _db: RocksDB,
+        pb: Box<dyn FnMut(u64) -> ProgressBar>,
+    ) -> Result<RocksDB, Error>;
 
-    /// returns migration version, use `yyyymmddhhmmss` timestamp format
+    /// returns migration version, use `date +'%Y%m%d%H%M%S'` timestamp format
     fn version(&self) -> &str;
 }
 
@@ -75,7 +115,11 @@ impl DefaultMigration {
 }
 
 impl Migration for DefaultMigration {
-    fn migrate(&self, db: RocksDB) -> Result<RocksDB, Error> {
+    fn migrate(
+        &self,
+        db: RocksDB,
+        _pb: Box<dyn FnMut(u64) -> ProgressBar>,
+    ) -> Result<RocksDB, Error> {
         Ok(db)
     }
 

db/src/db.rs

@@ -131,7 +131,7 @@ impl RocksDB {
         self.inner.get_pinned(&key).map_err(internal_error)
     }
 
-    pub fn put<K, V>(&self, key: K, value: V) -> Result<()>
+    pub fn put_default<K, V>(&self, key: K, value: V) -> Result<()>
     where
         K: AsRef<[u8]>,
         V: AsRef<[u8]>,

indexer/src/migrations.rs

@@ -1,6 +1,6 @@
 use crate::types::LockHashIndex;
 use ckb_db::{Col, Result, RocksDB};
-use ckb_db_migration::Migration;
+use ckb_db_migration::{Migration, ProgressBar};
 use ckb_shared::shared::Shared;
 use ckb_store::ChainStore;
 
@@ -20,7 +20,7 @@ impl AddFieldsToLiveCell {
 }
 
 impl Migration for AddFieldsToLiveCell {
-    fn migrate(&self, db: RocksDB) -> Result<RocksDB> {
+    fn migrate(&self, db: RocksDB, _pb: Box<dyn FnMut(u64) -> ProgressBar>) -> Result<RocksDB> {
         const COLUMN_LOCK_HASH_LIVE_CELL: Col = "1";
 
         let snapshot = self.shared.snapshot();

miner/Cargo.toml

@@ -22,6 +22,6 @@ futures = "0.1"
 lru-cache = { git = "https://github.com/nervosnetwork/lru-cache", rev = "a35fdb8" }
 ckb-stop-handler = { path = "../util/stop-handler" }
 failure = "0.1.5"
-indicatif = "0.11"
+indicatif = "0.15"
 console = "0.8.0"
 eaglesong = "0.1"

shared/src/migrations/table_to_struct.rs

@@ -1,39 +1,59 @@
 use ckb_db::{Result, RocksDB};
-use ckb_db_migration::Migration;
+use ckb_db_migration::{Migration, ProgressBar, ProgressStyle};
 use ckb_store::{
     COLUMN_BLOCK_HEADER, COLUMN_EPOCH, COLUMN_META, COLUMN_TRANSACTION_INFO, COLUMN_UNCLES,
     META_CURRENT_EPOCH_KEY,
 };
 
 pub struct ChangeMoleculeTableToStruct;
 
-const VERSION: &str = "20200703124523";
+const VERSION: &str = "20200703124524";
 
 impl Migration for ChangeMoleculeTableToStruct {
-    fn migrate(&self, db: RocksDB) -> Result<RocksDB> {
-        let txn = db.transaction();
+    fn migrate(&self, db: RocksDB, mut pb: Box<dyn FnMut(u64) -> ProgressBar>) -> Result<RocksDB> {
+        let pb = pb(9);
+        let spinner_style = ProgressStyle::default_spinner()
+            .tick_chars("⠁⠂⠄⡀⢀⠠⠐⠈ ")
+            .template("{prefix:.bold.dim} {spinner} {wide_msg}");
+        pb.set_style(spinner_style);
 
+        pb.set_message("migrating: block header");
+        pb.inc(1);
+        let txn = db.transaction();
         let header_view_migration = |key: &[u8], value: &[u8]| -> Result<()> {
             // (1 total size field + 2 fields) * 4 byte per field
             txn.put(COLUMN_BLOCK_HEADER, key, &value[12..])?;
+
             Ok(())
         };
         db.traverse(COLUMN_BLOCK_HEADER, header_view_migration)?;
+        pb.set_message("finish: block header");
+        pb.inc(1);
 
+        pb.set_message("migrating: uncles");
+        pb.inc(1);
         let uncles_migration = |key: &[u8], value: &[u8]| -> Result<()> {
             // (1 total size field + 2 fields) * 4 byte per field
             txn.put(COLUMN_UNCLES, key, &value[12..])?;
             Ok(())
         };
         db.traverse(COLUMN_UNCLES, uncles_migration)?;
+        pb.set_message("finish: uncles");
+        pb.inc(1);
 
+        pb.set_message("migrating: transaction info");
+        pb.inc(1);
         let transaction_info_migration = |key: &[u8], value: &[u8]| -> Result<()> {
             // (1 total size field + 3 fields) * 4 byte per field
             txn.put(COLUMN_TRANSACTION_INFO, key, &value[16..])?;
             Ok(())
         };
         db.traverse(COLUMN_TRANSACTION_INFO, transaction_info_migration)?;
+        pb.set_message("finish: transaction info");
+        pb.inc(1);
 
+        pb.set_message("migrating: epoch");
+        pb.inc(1);
         let epoch_ext_migration = |key: &[u8], value: &[u8]| -> Result<()> {
             // COLUMN_EPOCH stores epoch_number => last_block_hash_in_previous_epoch and last_block_hash_in_previous_epoch => epoch_ext
             // only migrates epoch_ext
@@ -44,11 +64,18 @@ impl Migration for ChangeMoleculeTableToStruct {
             Ok(())
         };
         db.traverse(COLUMN_EPOCH, epoch_ext_migration)?;
+        pb.set_message("finish: epoch");
+        pb.inc(1);
+
         if let Some(current_epoch) = txn.get(COLUMN_META, META_CURRENT_EPOCH_KEY)? {
             txn.put(COLUMN_META, META_CURRENT_EPOCH_KEY, &current_epoch[36..])?;
         }
 
         txn.commit()?;
+
+        pb.set_message("commit changes");
+        pb.inc(1);
+        pb.finish_with_message("waiting...");
         Ok(db)
     }
 

util/instrument/Cargo.toml

@@ -12,7 +12,7 @@ ckb-chain-iter = { path = "../chain-iter" }
 ckb-shared = { path = "../../shared" }
 ckb-jsonrpc-types = { path = "../jsonrpc-types" }
 serde_json = "1.0"
-indicatif = { version = "0.11", optional = true }
+indicatif = { version = "0.15", optional = true }
 
 [features]
 progress_bar = ["indicatif"]

util/logger-service/src/lib.rs

@@ -467,10 +467,6 @@ pub fn init(config: Config) -> Result<LoggerInitGuard, SetLoggerError> {
     })
 }
 
-pub fn silent() {
-    log::set_max_level(LevelFilter::Off);
-}
-
 pub fn flush() {
     log::logger().flush()
 }