Import snapshot block with body

remagpie · foriequal0 · commit abe60a324665 · 2019-12-10T12:11:51.000+09:00
diff --git a/core/src/blockchain/blockchain.rs b/core/src/blockchain/blockchain.rs
@@ -98,18 +98,6 @@ impl BlockChain {
         }
     }
 
-    pub fn insert_bootstrap_header(&self, batch: &mut DBTransaction, header: &HeaderView) {
-        self.headerchain.insert_bootstrap_header(batch, header);
-
-        let hash = header.hash();
-
-        *self.pending_best_block_hash.write() = Some(hash);
-        batch.put(db::COL_EXTRA, BEST_BLOCK_KEY, &hash);
-
-        *self.pending_best_proposal_block_hash.write() = Some(hash);
-        batch.put(db::COL_EXTRA, BEST_PROPOSAL_BLOCK_KEY, &hash);
-    }
-
     pub fn insert_header(
         &self,
         batch: &mut DBTransaction,
@@ -122,6 +110,34 @@ impl BlockChain {
         }
     }
 
+    pub fn insert_bootstrap_block(&self, batch: &mut DBTransaction, bytes: &[u8]) {
+        let block = BlockView::new(bytes);
+        let header = block.header_view();
+        let hash = header.hash();
+
+        ctrace!(BLOCKCHAIN, "Inserting bootstrap block #{}({}) to the blockchain.", header.number(), hash);
+
+        if self.is_known(&hash) {
+            cdebug!(BLOCKCHAIN, "Block #{}({}) is already known.", header.number(), hash);
+            return
+        }
+
+        assert!(self.pending_best_block_hash.read().is_none());
+        assert!(self.pending_best_proposal_block_hash.read().is_none());
+
+        self.headerchain.insert_bootstrap_header(batch, &header);
+        self.body_db.insert_body(batch, &block);
+        self.body_db.update_best_block(batch, &BestBlockChanged::CanonChainAppended {
+            best_block: bytes.to_vec(),
+        });
+
+        *self.pending_best_block_hash.write() = Some(hash);
+        batch.put(db::COL_EXTRA, BEST_BLOCK_KEY, &hash);
+
+        *self.pending_best_proposal_block_hash.write() = Some(hash);
+        batch.put(db::COL_EXTRA, BEST_PROPOSAL_BLOCK_KEY, &hash);
+    }
+
     /// Inserts the block into backing cache database.
     /// Expects the block to be valid and already verified.
     /// If the block is already known, does nothing.
diff --git a/core/src/client/client.rs b/core/src/client/client.rs
@@ -27,7 +27,7 @@ use cstate::{
 };
 use ctimer::{TimeoutHandler, TimerApi, TimerScheduleError, TimerToken};
 use ctypes::transaction::{AssetTransferInput, PartialHashing, ShardTransaction};
-use ctypes::{BlockHash, BlockNumber, CommonParams, Header, ShardId, Tracker, TxHash};
+use ctypes::{BlockHash, BlockNumber, CommonParams, ShardId, Tracker, TxHash};
 use cvm::{decode, execute, ChainTimeInfo, ScriptResult, VMConfig};
 use hashdb::AsHashDB;
 use journaldb;
@@ -42,7 +42,7 @@ use super::{
     ClientConfig, DatabaseClient, EngineClient, EngineInfo, Error as ClientError, ExecuteClient, ImportBlock,
     ImportResult, MiningBlockChainClient, Shard, StateInfo, StateOrBlock, TextClient,
 };
-use crate::block::{ClosedBlock, IsBlock, OpenBlock, SealedBlock};
+use crate::block::{Block, ClosedBlock, IsBlock, OpenBlock, SealedBlock};
 use crate::blockchain::{BlockChain, BlockProvider, BodyProvider, HeaderProvider, InvoiceProvider, TransactionAddress};
 use crate::client::{ConsensusClient, SnapshotClient, TermInfo};
 use crate::consensus::{CodeChainEngine, EngineError};
@@ -655,13 +655,13 @@ impl ImportBlock for Client {
         Ok(self.importer.header_queue.import(unverified)?)
     }
 
-    fn import_bootstrap_header(&self, header: &Header) -> Result<BlockHash, BlockImportError> {
-        if self.block_chain().is_known_header(&header.hash()) {
+    fn import_bootstrap_block(&self, block: &Block) -> Result<BlockHash, BlockImportError> {
+        if self.block_chain().is_known(&block.header.hash()) {
             return Err(BlockImportError::Import(ImportError::AlreadyInChain))
         }
         let import_lock = self.importer.import_lock.lock();
-        self.importer.import_bootstrap_header(header, self, &import_lock);
-        Ok(header.hash())
+        self.importer.import_bootstrap_block(block, self, &import_lock);
+        Ok(block.header.hash())
     }
 
     fn import_sealed_block(&self, block: &SealedBlock) -> ImportResult {
diff --git a/core/src/client/importer.rs b/core/src/client/importer.rs
@@ -19,14 +19,14 @@ use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 
 use cio::IoChannel;
-use ctypes::header::Header;
+use ctypes::header::{Header, Seal};
 use ctypes::BlockHash;
 use kvdb::DBTransaction;
 use parking_lot::{Mutex, MutexGuard};
 use rlp::Encodable;
 
 use super::{BlockChainTrait, Client, ClientConfig};
-use crate::block::{enact, IsBlock, LockedBlock};
+use crate::block::{enact, Block, IsBlock, LockedBlock};
 use crate::blockchain::{BodyProvider, HeaderProvider, ImportRoute};
 use crate::client::EngineInfo;
 use crate::consensus::CodeChainEngine;
@@ -362,19 +362,21 @@ impl Importer {
         imported.len()
     }
 
-    pub fn import_bootstrap_header<'a>(&'a self, header: &'a Header, client: &Client, _importer_lock: &MutexGuard<()>) {
+    pub fn import_bootstrap_block<'a>(&'a self, block: &'a Block, client: &Client, _importer_lock: &MutexGuard<()>) {
+        let header = &block.header;
         let hash = header.hash();
-        ctrace!(CLIENT, "Importing bootstrap header {}-{:?}", header.number(), hash);
+        ctrace!(CLIENT, "Importing bootstrap block #{}-{:?}", header.number(), hash);
 
         {
             let chain = client.block_chain();
             let mut batch = DBTransaction::new();
-            chain.insert_bootstrap_header(&mut batch, &HeaderView::new(&header.rlp_bytes()));
+            chain.insert_bootstrap_block(&mut batch, &block.rlp_bytes(&Seal::With));
             client.db().write_buffered(batch);
             chain.commit();
         }
-
         client.new_headers(&[hash], &[], &[hash], &[], &[], Some(hash));
+        self.miner.chain_new_blocks(client, &[hash], &[], &[hash], &[]);
+        client.new_blocks(&[hash], &[], &[hash], &[], &[]);
 
         client.db().flush().expect("DB flush failed.");
     }
diff --git a/core/src/client/mod.rs b/core/src/client/mod.rs
@@ -38,12 +38,12 @@ use cmerkle::Result as TrieResult;
 use cnetwork::NodeId;
 use cstate::{AssetScheme, FindActionHandler, OwnedAsset, StateResult, Text, TopLevelState, TopStateView};
 use ctypes::transaction::{AssetTransferInput, PartialHashing, ShardTransaction};
-use ctypes::{BlockHash, BlockNumber, CommonParams, Header, ShardId, Tracker, TxHash};
+use ctypes::{BlockHash, BlockNumber, CommonParams, ShardId, Tracker, TxHash};
 use cvm::ChainTimeInfo;
 use kvdb::KeyValueDB;
 use primitives::{Bytes, H160, H256, U256};
 
-use crate::block::{ClosedBlock, OpenBlock, SealedBlock};
+use crate::block::{Block, ClosedBlock, OpenBlock, SealedBlock};
 use crate::blockchain_info::BlockChainInfo;
 use crate::consensus::EngineError;
 use crate::encoded;
@@ -204,7 +204,7 @@ pub trait ImportBlock {
 
     /// Import a trusted bootstrap header into the blockchain
     /// Bootstrap headers don't execute any verifications
-    fn import_bootstrap_header(&self, bytes: &Header) -> Result<BlockHash, BlockImportError>;
+    fn import_bootstrap_block(&self, bytes: &Block) -> Result<BlockHash, BlockImportError>;
 
     /// Import sealed block. Skips all verifications.
     fn import_sealed_block(&self, block: &SealedBlock) -> ImportResult;
diff --git a/core/src/client/test_client.rs b/core/src/client/test_client.rs
@@ -52,7 +52,7 @@ use parking_lot::RwLock;
 use primitives::{Bytes, H256, U256};
 use rlp::*;
 
-use crate::block::{ClosedBlock, OpenBlock, SealedBlock};
+use crate::block::{Block, ClosedBlock, OpenBlock, SealedBlock};
 use crate::blockchain_info::BlockChainInfo;
 use crate::client::{
     AccountData, BlockChainClient, BlockChainTrait, BlockProducer, BlockStatus, ConsensusClient, EngineInfo,
@@ -509,7 +509,7 @@ impl ImportBlock for TestBlockChainClient {
         unimplemented!()
     }
 
-    fn import_bootstrap_header(&self, _header: &BlockHeader) -> Result<BlockHash, BlockImportError> {
+    fn import_bootstrap_block(&self, _header: &Block) -> Result<BlockHash, BlockImportError> {
         unimplemented!()
     }
 
diff --git a/sync/src/block/extension.rs b/sync/src/block/extension.rs
@@ -27,7 +27,7 @@ use ccore::{
 };
 use cmerkle::snapshot::ChunkDecompressor;
 use cmerkle::snapshot::Restore as SnapshotRestore;
-use cmerkle::TrieFactory;
+use cmerkle::{skewed_merkle_root, TrieFactory};
 use cnetwork::{Api, EventSender, NetworkExtension, NodeId};
 use cstate::FindActionHandler;
 use ctimer::TimerToken;
@@ -64,7 +64,7 @@ pub struct TokenInfo {
 enum State {
     SnapshotHeader(BlockHash, u64),
     SnapshotBody {
-        block: BlockHash,
+        header: EncodedHeader,
         prev_root: H256,
     },
     SnapshotChunk {
@@ -151,7 +151,7 @@ impl Extension {
             let parent =
                 client.block_header(&parent_hash.into()).expect("Parent header of the snapshot header must exist");
             return State::SnapshotBody {
-                block: hash,
+                header,
                 prev_root: parent.transactions_root(),
             }
         }
@@ -437,8 +437,29 @@ impl NetworkExtension<Event> for Extension {
                         }
                     }
                     State::SnapshotBody {
+                        ref header,
                         ..
-                    } => unimplemented!(),
+                    } => {
+                        for id in &peer_ids {
+                            if let Some(requests) = self.requests.get_mut(id) {
+                                ctrace!(SYNC, "Send snapshot body request to {}", id);
+                                let request = RequestMessage::Bodies(vec![header.hash()]);
+                                let request_id = self.last_request;
+                                self.last_request += 1;
+                                requests.push((request_id, request.clone()));
+                                self.api.send(id, Arc::new(Message::Request(request_id, request).rlp_bytes()));
+
+                                let token = &self.tokens[id];
+                                let token_info = self.tokens_info.get_mut(token).unwrap();
+
+                                let _ = self.api.clear_timer(*token);
+                                self.api
+                                    .set_timer_once(*token, Duration::from_millis(SYNC_EXPIRE_REQUEST_INTERVAL))
+                                    .expect("Timer set succeeds");
+                                token_info.request_id = Some(request_id);
+                            }
+                        }
+                    }
                     State::SnapshotChunk {
                         block,
                         ref mut restore,
@@ -815,20 +836,11 @@ impl Extension {
         match self.state {
             State::SnapshotHeader(hash, _) => match headers {
                 [parent, header] if header.hash() == hash => {
-                    match self.client.import_bootstrap_header(&header) {
-                        Ok(_) | Err(BlockImportError::Import(ImportError::AlreadyInChain)) => {
-                            self.state = State::SnapshotBody {
-                                block: hash,
-                                prev_root: *parent.transactions_root(),
-                            };
-                            cdebug!(SYNC, "Transitioning state to {:?}", self.state);
-                        }
-                        Err(BlockImportError::Import(ImportError::AlreadyQueued)) => {}
-                        // FIXME: handle import errors
-                        Err(err) => {
-                            cwarn!(SYNC, "Cannot import header({}): {:?}", header.hash(), err);
-                        }
-                    }
+                    self.state = State::SnapshotBody {
+                        header: EncodedHeader::new(header.rlp_bytes().to_vec()),
+                        prev_root: *parent.transactions_root(),
+                    };
+                    cdebug!(SYNC, "Transitioning state to {:?}", self.state);
                 }
                 _ => cdebug!(
                     SYNC,
@@ -887,42 +899,75 @@ impl Extension {
 
     fn on_body_response(&mut self, hashes: Vec<BlockHash>, bodies: Vec<Vec<UnverifiedTransaction>>) {
         ctrace!(SYNC, "Received body response with lenth({}) {:?}", hashes.len(), hashes);
-        {
-            self.body_downloader.import_bodies(hashes, bodies);
-            let completed = self.body_downloader.drain();
-            for (hash, transactions) in completed {
-                let header = self
-                    .client
-                    .block_header(&BlockId::Hash(hash))
-                    .expect("Downloaded body's header must exist")
-                    .decode();
-                let block = Block {
-                    header,
-                    transactions,
-                };
-                cdebug!(SYNC, "Body download completed for #{}({})", block.header.number(), hash);
-                match self.client.import_block(block.rlp_bytes(&Seal::With)) {
-                    Err(BlockImportError::Import(ImportError::AlreadyInChain)) => {
-                        cwarn!(SYNC, "Downloaded already existing block({})", hash)
-                    }
-                    Err(BlockImportError::Import(ImportError::AlreadyQueued)) => {
-                        cwarn!(SYNC, "Downloaded already queued in the verification queue({})", hash)
-                    }
-                    Err(err) => {
+
+        match self.state {
+            State::SnapshotBody {
+                ref header,
+                prev_root,
+            } => {
+                let body = bodies.first().expect("Body response in SnapshotBody state has only one body");
+                let new_root = skewed_merkle_root(prev_root, body.iter().map(Encodable::rlp_bytes));
+                if header.transactions_root() == new_root {
+                    let block = Block {
+                        header: header.decode(),
+                        transactions: body.clone(),
+                    };
+                    match self.client.import_bootstrap_block(&block) {
+                        Ok(_) | Err(BlockImportError::Import(ImportError::AlreadyInChain)) => {
+                            self.state = State::SnapshotChunk {
+                                block: header.hash(),
+                                restore: SnapshotRestore::new(header.state_root()),
+                            };
+                            cdebug!(SYNC, "Transitioning state to {:?}", self.state);
+                        }
+                        Err(BlockImportError::Import(ImportError::AlreadyQueued)) => {}
                         // FIXME: handle import errors
-                        cwarn!(SYNC, "Cannot import block({}): {:?}", hash, err);
-                        break
+                        Err(err) => {
+                            cwarn!(SYNC, "Cannot import header({}): {:?}", header.hash(), err);
+                        }
                     }
-                    _ => {}
                 }
             }
-        }
+            State::Full => {
+                {
+                    self.body_downloader.import_bodies(hashes, bodies);
+                    let completed = self.body_downloader.drain();
+                    for (hash, transactions) in completed {
+                        let header = self
+                            .client
+                            .block_header(&BlockId::Hash(hash))
+                            .expect("Downloaded body's header must exist")
+                            .decode();
+                        let block = Block {
+                            header,
+                            transactions,
+                        };
+                        cdebug!(SYNC, "Body download completed for #{}({})", block.header.number(), hash);
+                        match self.client.import_block(block.rlp_bytes(&Seal::With)) {
+                            Err(BlockImportError::Import(ImportError::AlreadyInChain)) => {
+                                cwarn!(SYNC, "Downloaded already existing block({})", hash)
+                            }
+                            Err(BlockImportError::Import(ImportError::AlreadyQueued)) => {
+                                cwarn!(SYNC, "Downloaded already queued in the verification queue({})", hash)
+                            }
+                            Err(err) => {
+                                // FIXME: handle import errors
+                                cwarn!(SYNC, "Cannot import block({}): {:?}", hash, err);
+                                break
+                            }
+                            _ => {}
+                        }
+                    }
+                }
 
-        let mut peer_ids: Vec<_> = self.header_downloaders.keys().cloned().collect();
-        peer_ids.shuffle(&mut thread_rng());
+                let mut peer_ids: Vec<_> = self.header_downloaders.keys().cloned().collect();
+                peer_ids.shuffle(&mut thread_rng());
 
-        for id in peer_ids {
-            self.send_body_request(&id);
+                for id in peer_ids {
+                    self.send_body_request(&id);
+                }
+            }
+            _ => {}
         }
     }
 

Original file line number	Diff line number	Diff line change
`@@ -52,7 +52,7 @@ use parking_lot::RwLock;`
`52`	`52`	`use primitives::{Bytes, H256, U256};`
`53`	`53`	`use rlp::*;`
`54`	`54`
`55`		`-use crate::block::{ClosedBlock, OpenBlock, SealedBlock};`
	`55`	`+use crate::block::{Block, ClosedBlock, OpenBlock, SealedBlock};`
`56`	`56`	`use crate::blockchain_info::BlockChainInfo;`
`57`	`57`	`use crate::client::{`
`58`	`58`	`AccountData, BlockChainClient, BlockChainTrait, BlockProducer, BlockStatus, ConsensusClient, EngineInfo,`
`@@ -509,7 +509,7 @@ impl ImportBlock for TestBlockChainClient {`
`509`	`509`	`unimplemented!()`
`510`	`510`	`}`
`511`	`511`
`512`		`- fn import_bootstrap_header(&self, _header: &BlockHeader) -> Result<BlockHash, BlockImportError> {`
	`512`	`+ fn import_bootstrap_block(&self, _header: &Block) -> Result<BlockHash, BlockImportError> {`
`513`	`513`	`unimplemented!()`
`514`	`514`	`}`
`515`	`515`