replay: support massive replay (#3177)

* replay: support massive replay

* fix clippy

Cargo.toml

@@ -99,6 +99,7 @@ members = [
     "cmd/merkle-generator",
     "cmd/peer-watcher",
     "cmd/airdrop",
+    "cmd/replay",
     "stratum",
     "cmd/miner_client/api",
     "cmd/db-exporter",
@@ -202,6 +203,7 @@ default-members = [
     "cmd/merkle-generator",
     "cmd/peer-watcher",
     "cmd/airdrop",
+    "cmd/replay",
     "cmd/genesis-nft-miner",
     "stratum",
     "cmd/miner_client/api",

cmd/replay/Cargo.toml

@@ -7,7 +7,7 @@ publish = false
 edition = "2021"
 
 [[bin]]
-name = "starcoin_replay"
+name = "starcoin-replay"
 path = "src/main.rs"
 
 [dependencies]

cmd/replay/README.md

@@ -7,14 +7,14 @@ A tools for replay data from a network to a new chain.
 
 
 ```bash
-$ .target/release/starcoin_replay  -n proxima -f $source -t $target -c 10000
+$ .target/release/starcoin-replay  -n proxima -f $source -t $target -c 10000
 ```
 
 
 ```
 
 USAGE:
-    starcoin_replay [OPTIONS] --from <from> --to <to>
+    starcoin-replay [OPTIONS] --from <from> --to <to>
 
 FLAGS:
     -h, --help       Prints help information

cmd/replay/src/main.rs

@@ -15,7 +15,7 @@ use starcoin_storage::{BlockStore, Storage};
 use starcoin_types::startup_info::StartupInfo;
 use std::path::PathBuf;
 use std::sync::Arc;
-use std::time::SystemTime;
+use std::time::Instant;
 use structopt::StructOpt;
 
 #[derive(Debug, StructOpt)]
@@ -30,6 +30,9 @@ pub struct ReplayOpt {
     #[structopt(short = "t", long, parse(from_os_str))]
     /// Target dir.
     pub to: PathBuf,
+    #[structopt(long)]
+    /// replay from start to this block, include this block. default to all.
+    pub end_block: Option<u64>,
     #[structopt(long, short = "c", default_value = "20000")]
     /// Number of block.
     pub block_num: u64,
@@ -44,7 +47,7 @@ pub struct ReplayOpt {
 // deprecated use starcoin_db_exporter replace
 fn main() -> anyhow::Result<()> {
     let _logger = starcoin_logger::init();
-    let opts = ReplayOpt::from_args();
+    let opts: ReplayOpt = ReplayOpt::from_args();
 
     let network = match opts.net {
         Some(network) => network,
@@ -53,7 +56,8 @@ fn main() -> anyhow::Result<()> {
     let net = ChainNetwork::new_builtin(network);
 
     let from_dir = opts.from;
-    let block_num = opts.block_num;
+    let end_block = opts.end_block;
+
     let to_dir = opts.to;
     // start watching
     if opts.watch {
@@ -78,19 +82,6 @@ fn main() -> anyhow::Result<()> {
         .expect("init storage by genesis fail.");
     let chain = BlockChain::new(net.time_service(), chain_info.head().id(), storage, None)
         .expect("create block chain should success.");
-    //read from first chain
-    let begin = SystemTime::now();
-    let mut block_vec = vec![];
-    for i in 1..block_num {
-        if let Ok(Some(block)) = chain.get_block_by_number(i) {
-            block_vec.push(block);
-        } else {
-            println!("read block err, number : {:?}", i);
-            break;
-        }
-    }
-    let use_time = SystemTime::now().duration_since(begin).unwrap();
-    println!("read use time: {:?}", use_time.as_nanos());
 
     let storage2 = Arc::new(
         Storage::new(StorageInstance::new_cache_and_db_instance(
@@ -109,34 +100,79 @@ fn main() -> anyhow::Result<()> {
         None,
     )
     .expect("create block chain should success.");
-    let mut last_block_hash = None;
-    if let Some(last_block) = block_vec.last() {
-        last_block_hash = Some(last_block.header.id());
-    }
-    let begin = SystemTime::now();
-    for block in block_vec {
-        match opts.verifier {
-            Verifier::Basic => {
-                chain2.apply_with_verifier::<BasicVerifier>(block).unwrap();
-            }
-            Verifier::Consensus => {
-                chain2
-                    .apply_with_verifier::<ConsensusVerifier>(block)
-                    .unwrap();
-            }
-            Verifier::None => {
-                chain2.apply_with_verifier::<NoneVerifier>(block).unwrap();
+
+    //read from first chain
+
+    let time_begin = Instant::now();
+
+    let end_block = end_block.unwrap_or_else(|| chain.current_header().number());
+
+    {
+        let b = chain2.get_block_by_number(end_block)?;
+        if let Some(h) = b {
+            if h.id() == chain.current_header().id() {
+                println!("target chain already synced with source chain");
+            } else {
+                println!("target chain have different block with source chain at latest block: {}, target: {}, source: {}",
+                             end_block,
+                             h.id(),
+                             chain.current_header().id(),
+                    );
             }
-            Verifier::Full => {
-                chain2.apply_with_verifier::<FullVerifier>(block).unwrap();
+            return Ok(());
+        }
+    }
+    let start_block = chain2.current_header().number() + 1;
+    let mut last_block_hash = None;
+    for i in start_block..=end_block {
+        if let Ok(Some(block)) = chain.get_block_by_number(i) {
+            let start = Instant::now();
+            let expected_state_root = block.header().state_root();
+            let block_id = block.id();
+            let block_height = block.header().number();
+            match opts.verifier {
+                Verifier::Basic => {
+                    chain2.apply_with_verifier::<BasicVerifier>(block).unwrap();
+                }
+                Verifier::Consensus => {
+                    chain2
+                        .apply_with_verifier::<ConsensusVerifier>(block)
+                        .unwrap();
+                }
+                Verifier::None => {
+                    chain2.apply_with_verifier::<NoneVerifier>(block).unwrap();
+                }
+                Verifier::Full => {
+                    chain2.apply_with_verifier::<FullVerifier>(block).unwrap();
+                }
+            };
+            println!(
+                "apply block {} at height: {}, time_used: {:?}, source state root: {}, target state root: {}",
+                block_id,
+                block_height,
+                start.elapsed(),
+                expected_state_root,
+                chain2.chain_state_reader().state_root()
+            );
+            last_block_hash = Some(block_id);
+
+            // save start up info every 100 blocks
+            if i % 100 == 0 {
+                if let Some(last_block_hash) = last_block_hash {
+                    let startup_info = StartupInfo::new(last_block_hash);
+                    storage2.save_startup_info(startup_info)?;
+                }
             }
-        };
+        } else {
+            println!("read source block err, number : {:?}", i);
+            break;
+        }
     }
+
     if let Some(last_block_hash) = last_block_hash {
         let startup_info = StartupInfo::new(last_block_hash);
         storage2.save_startup_info(startup_info)?;
     }
-    let use_time = SystemTime::now().duration_since(begin).unwrap();
-    println!("apply use time: {:?}", use_time.as_nanos());
+    println!("apply use time: {:?}", time_begin.elapsed());
     Ok(())
 }