chain sync initial state refact

src/sync/chain.rs

@@ -21,6 +21,7 @@ use client::{BlockChainClient, BlockStatus};
 use sync::range_collection::{RangeCollection, ToUsize, FromUsize};
 use error::*;
 use sync::io::SyncIo;
+use std::option::Option;
 
 impl ToUsize for BlockNumber {
 	fn to_usize(&self) -> usize {
@@ -99,14 +100,14 @@ pub struct SyncStatus {
 	pub protocol_version: u8,
 	/// BlockChain height for the moment the sync started.
 	pub start_block_number: BlockNumber,
-	/// Last fully downloaded and imported block number.
+	/// Last fully downloaded and imported block number (if any).
-	pub last_imported_block_number: BlockNumber,
+	pub last_imported_block_number: Option<BlockNumber>,
-	/// Highest block number in the download queue.
+	/// Highest block number in the download queue (if any).
-	pub highest_block_number: BlockNumber,
+	pub highest_block_number: Option<BlockNumber>,
 	/// Total number of blocks for the sync process.
-	pub blocks_total: usize,
+	pub blocks_total: BlockNumber,
 	/// Number of blocks downloaded so far.
-	pub blocks_received: usize,
+	pub blocks_received: BlockNumber,
 	/// Total number of connected peers
 	pub num_peers: usize,
 	/// Total number of active peers
@@ -147,7 +148,7 @@ pub struct ChainSync {
 	/// Last block number for the start of sync
 	starting_block: BlockNumber,
 	/// Highest block number seen
-	highest_block: BlockNumber,
+	highest_block: Option<BlockNumber>,
 	/// Set of block header numbers being downloaded
 	downloading_headers: HashSet<BlockNumber>,
 	/// Set of block body numbers being downloaded
@@ -161,9 +162,9 @@ pub struct ChainSync {
 	/// Used to map body to header
 	header_ids: HashMap<HeaderId, BlockNumber>,
 	/// Last impoted block number
-	last_imported_block: BlockNumber,
+	last_imported_block: Option<BlockNumber>,
 	/// Last impoted block hash
-	last_imported_hash: H256,
+	last_imported_hash: Option<H256>,
 	/// Syncing total  difficulty
 	syncing_difficulty: U256,
 	/// True if common block for our and remote chain has been found
@@ -177,15 +178,15 @@ impl ChainSync {
 		ChainSync {
 			state: SyncState::NotSynced,
 			starting_block: 0,
-			highest_block: 0,
+			highest_block: None,
 			downloading_headers: HashSet::new(),
 			downloading_bodies: HashSet::new(),
 			headers: Vec::new(),
 			bodies: Vec::new(),
 			peers: HashMap::new(),
 			header_ids: HashMap::new(),
-			last_imported_block: 0,
+			last_imported_block: None,
-			last_imported_hash: H256::new(),
+			last_imported_hash: None,
 			syncing_difficulty: U256::from(0u64),
 			have_common_block: false,
 		}
@@ -199,8 +200,8 @@ impl ChainSync {
 			start_block_number: self.starting_block,
 			last_imported_block_number: self.last_imported_block,
 			highest_block_number: self.highest_block,
-			blocks_received: (self.last_imported_block - self.starting_block) as usize,
+			blocks_received: match self.last_imported_block { None => 0, Some(x) => x - self.starting_block },
-			blocks_total: (self.highest_block - self.starting_block) as usize,
+			blocks_total: match self.highest_block { None => 0, Some(x) => x - self.starting_block },
 			num_peers: self.peers.len(),
 			num_active_peers: self.peers.values().filter(|p| p.asking != PeerAsking::Nothing).count(),
 		}
@@ -229,10 +230,10 @@ impl ChainSync {
 	/// Restart sync
 	pub fn restart(&mut self, io: &mut SyncIo) {
 		self.reset();
-		self.last_imported_block = 0;
+		self.last_imported_block = None;
-		self.last_imported_hash = H256::new();
+		self.last_imported_hash = None;
 		self.starting_block = 0;
-		self.highest_block = 0;
+		self.highest_block = None;
 		self.have_common_block = false;
 		io.chain().clear_queue();
 		self.starting_block = io.chain().chain_info().best_block_number;
@@ -293,25 +294,27 @@ impl ChainSync {
 		for i in 0..item_count {
 			let info: BlockHeader = try!(r.val_at(i));
 			let number = BlockNumber::from(info.number);
-			if number <= self.last_imported_block || self.headers.have_item(&number) {
+			if number <= self.current_base_block() || self.headers.have_item(&number) {
 				trace!(target: "sync", "Skipping existing block header");
 				continue;
 			}
-			if number > self.highest_block {
+
-				self.highest_block = number;
+			if self.highest_block == None || number > self.highest_block.unwrap() {
+				self.highest_block = Some(number);
 			}
 			let hash = info.hash();
 			match io.chain().block_status(&hash) {
 				BlockStatus::InChain => {
 					self.have_common_block = true;
-					self.last_imported_block = number;
+					self.last_imported_block = Some(number);
-					self.last_imported_hash = hash.clone();
+					self.last_imported_hash = Some(hash.clone());
 					trace!(target: "sync", "Found common header {} ({})", number, hash);
 				},
 				_ => {
 					if self.have_common_block {
 						//validate chain
-						if self.have_common_block && number == self.last_imported_block + 1 && info.parent_hash != self.last_imported_hash {
+						let base_hash = self.last_imported_hash.clone().unwrap();
+						if self.have_common_block && number == self.current_base_block() + 1 && info.parent_hash != base_hash {
 							// TODO: lower peer rating
 							debug!(target: "sync", "Mismatched block header {} {}", number, hash);
 							continue;
@@ -407,7 +410,7 @@ impl ChainSync {
 		trace!(target: "sync", "{} -> NewBlock ({})", peer_id, h);
 		let header_view = HeaderView::new(header_rlp.as_raw());
 		// TODO: Decompose block and add to self.headers and self.bodies instead
-		if header_view.number() == From::from(self.last_imported_block + 1) {
+		if header_view.number() == From::from(self.current_base_block() + 1) {
 			match io.chain().import_block(block_rlp.as_raw().to_vec()) {
 				Err(ImportError::AlreadyInChain) => {
 					trace!(target: "sync", "New block already in chain {:?}", h);
@@ -550,6 +553,10 @@ impl ChainSync {
 		}
 	}
 
+	fn current_base_block(&self) -> BlockNumber {
+		match self.last_imported_block { None => 0, Some(x) => x }
+	}
+
 	/// Find some headers or blocks to download for a peer.
 	fn request_blocks(&mut self, io: &mut SyncIo, peer_id: PeerId) {
 		self.clear_peer_download(peer_id);
@@ -563,7 +570,7 @@ impl ChainSync {
 		let mut needed_bodies: Vec<H256> = Vec::new();
 		let mut needed_numbers: Vec<BlockNumber> = Vec::new();
 
-		if self.have_common_block && !self.headers.is_empty() && self.headers.range_iter().next().unwrap().0 == self.last_imported_block + 1 {
+		if self.have_common_block && !self.headers.is_empty() && self.headers.range_iter().next().unwrap().0 == self.current_base_block() + 1 {
 			for (start, ref items) in self.headers.range_iter() {
 				if needed_bodies.len() > MAX_BODIES_TO_REQUEST {
 					break;
@@ -596,12 +603,12 @@ impl ChainSync {
 				}
 				if start == 0 {
 					self.have_common_block = true; //reached genesis
-					self.last_imported_hash = chain_info.genesis_hash;
+					self.last_imported_hash = Some(chain_info.genesis_hash);
 				}
 			}
 			if self.have_common_block {
 				let mut headers: Vec<BlockNumber> = Vec::new();
-				let mut prev = self.last_imported_block + 1;
+				let mut prev = self.current_base_block() + 1;
 				for (next, ref items) in self.headers.range_iter() {
 					if !headers.is_empty() {
 						break;
@@ -656,7 +663,7 @@ impl ChainSync {
 		{
 			let headers = self.headers.range_iter().next().unwrap();
 			let bodies = self.bodies.range_iter().next().unwrap();
-			if headers.0 != bodies.0 || headers.0 != self.last_imported_block + 1 {
+			if headers.0 != bodies.0 || headers.0 != self.current_base_block() + 1 {
 				return;
 			}
 
@@ -672,18 +679,18 @@ impl ChainSync {
 				match io.chain().import_block(block_rlp.out()) {
 					Err(ImportError::AlreadyInChain) => {
 						trace!(target: "sync", "Block already in chain {:?}", h);
-						self.last_imported_block = headers.0 + i as BlockNumber;
+						self.last_imported_block = Some(headers.0 + i as BlockNumber);
-						self.last_imported_hash = h.clone();
+						self.last_imported_hash = Some(h.clone());
 					},
 					Err(ImportError::AlreadyQueued) => {
 						trace!(target: "sync", "Block already queued {:?}", h);
-						self.last_imported_block = headers.0 + i as BlockNumber;
+						self.last_imported_block = Some(headers.0 + i as BlockNumber);
-						self.last_imported_hash = h.clone();
+						self.last_imported_hash = Some(h.clone());
 					},
 					Ok(_) => {
 						trace!(target: "sync", "Block queued {:?}", h);
-						self.last_imported_block = headers.0 + i as BlockNumber;
+						self.last_imported_block = Some(headers.0 + i as BlockNumber);
-						self.last_imported_hash = h.clone();
+						self.last_imported_hash = Some(h.clone());
 						imported += 1;
 					},
 					Err(e) => {
@@ -700,8 +707,8 @@ impl ChainSync {
 			return;
 		}
 
-		self.headers.remove_head(&(self.last_imported_block + 1));
+		self.headers.remove_head(&(self.last_imported_block.unwrap() + 1));
-		self.bodies.remove_head(&(self.last_imported_block + 1));
+		self.bodies.remove_head(&(self.last_imported_block.unwrap() + 1));
 
 		if self.headers.is_empty() {
 			assert!(self.bodies.is_empty());

src/sync/tests.rs

@@ -4,7 +4,7 @@ use block_queue::BlockQueueInfo;
 use header::{Header as BlockHeader, BlockNumber};
 use error::*;
 use sync::io::SyncIo;
-use sync::chain::ChainSync;
+use sync::chain::{ChainSync, SyncState};
 
 struct TestBlockChainClient {
 	blocks: RwLock<HashMap<H256, Bytes>>,
@@ -241,13 +241,15 @@ struct TestPeer {
 }
 
 struct TestNet {
-	peers: Vec<TestPeer>
+	peers: Vec<TestPeer>,
+	started: bool
 }
 
 impl TestNet {
 	pub fn new(n: usize) -> TestNet {
 		let mut net = TestNet {
 			peers: Vec::new(),
+			started: false
 		};
 		for _ in 0..n {
 			net.peers.push(TestPeer {
@@ -291,10 +293,28 @@ impl TestNet {
 		}
 	}
 
-	pub fn sync(&mut self) {
+	pub fn restart_peer(&mut self, i: usize) {
+		let peer = self.peer_mut(i);
+		peer.sync.restart(&mut TestIo::new(&mut peer.chain, &mut peer.queue, None));
+	}
+
+	pub fn sync(&mut self) -> u32 {
 		self.start();
+		let mut total_steps = 0;
 		while !self.done() {
-			self.sync_step()
+			self.sync_step();
+			total_steps = total_steps + 1;
+		}
+		total_steps
+	}
+
+	pub fn sync_steps(&mut self, count: usize) {
+		if !self.started {
+			self.start();
+			self.started = true;
+		}
+		for _ in 0..count {
+			self.sync_step();
 		}
 	}
 
@@ -303,7 +323,6 @@ impl TestNet {
 	}
 }
 
-
 #[test]
 fn full_sync_two_peers() {
 	::env_logger::init().ok();
@@ -315,6 +334,26 @@ fn full_sync_two_peers() {
 	assert_eq!(net.peer(0).chain.blocks.read().unwrap().deref(), net.peer(1).chain.blocks.read().unwrap().deref());
 }
 
+#[test]
+fn status_after_sync() {
+	::env_logger::init().ok();
+	let mut net = TestNet::new(3);
+	net.peer_mut(1).chain.add_blocks(1000, false);
+	net.peer_mut(2).chain.add_blocks(1000, false);
+	net.sync();
+	let status = net.peer(0).sync.status();
+	assert_eq!(status.state, SyncState::Idle);
+}
+
+#[test]
+fn full_sync_takes_few_steps() {
+	let mut net = TestNet::new(3);
+	net.peer_mut(1).chain.add_blocks(100, false);
+	net.peer_mut(2).chain.add_blocks(100, false);
+	let total_steps = net.sync();
+	assert!(total_steps < 7);
+}
+
 #[test]
 fn full_sync_empty_blocks() {
 	::env_logger::init().ok();
@@ -347,3 +386,25 @@ fn forked_sync() {
 	assert_eq!(net.peer(1).chain.numbers.read().unwrap().deref(), &peer1_chain);
 	assert_eq!(net.peer(2).chain.numbers.read().unwrap().deref(), &peer1_chain);
 }
+
+#[test]
+fn chain_sync_restart() {
+	let mut net = TestNet::new(3);
+	net.peer_mut(1).chain.add_blocks(1000, false);
+	net.peer_mut(2).chain.add_blocks(1000, false);
+
+	net.sync_steps(8);
+
+	// make sure that sync has actually happened
+	assert!(net.peer(0).chain.chain_info().best_block_number > 100);
+	net.restart_peer(0);
+
+	let status = net.peer(0).sync.status();
+	assert_eq!(status.state, SyncState::NotSynced);
+}
+
+#[test]
+fn chain_sync_status() {
+	let net = TestNet::new(2);
+	assert_eq!(net.peer(0).sync.status().state, SyncState::NotSynced);
+}