remerge of 264

sync/src/chain.rs

@@ -15,12 +15,14 @@
 
 use util::*;
 use std::mem::{replace};
-use views::{HeaderView};
+use ethcore::views::{HeaderView};
-use header::{BlockNumber, Header as BlockHeader};
+use ethcore::header::{BlockNumber, Header as BlockHeader};
-use client::{BlockChainClient, BlockStatus};
+use ethcore::client::{BlockChainClient, BlockStatus};
-use sync::range_collection::{RangeCollection, ToUsize, FromUsize};
+use range_collection::{RangeCollection, ToUsize, FromUsize};
-use error::*;
+use ethcore::error::*;
-use sync::io::SyncIo;
+use ethcore::block::Block;
+use io::SyncIo;
+use time;
 use std::option::Option;
 
 impl ToUsize for BlockNumber {
@@ -61,6 +63,8 @@ const RECEIPTS_PACKET: u8 = 0x10;
 
 const NETWORK_ID: U256 = ONE_U256; //TODO: get this from parent
 
+const CONNECTION_TIMEOUT_SEC: f64 = 30f64;
+
 struct Header {
 	/// Header data
 	data: Bytes,
@@ -138,6 +142,8 @@ struct PeerInfo {
 	asking: PeerAsking,
 	/// A set of block numbers being requested
 	asking_blocks: Vec<BlockNumber>,
+	/// Request timestamp
+	ask_time: f64,
 }
 
 /// Blockchain sync handler.
@@ -250,6 +256,7 @@ impl ChainSync {
 			genesis: try!(r.val_at(4)),
 			asking: PeerAsking::Nothing,
 			asking_blocks: Vec::new(),
+			ask_time: 0f64,
 		};
 
 		trace!(target: "sync", "New peer {} (protocol: {}, network: {:?}, difficulty: {:?}, latest:{}, genesis:{})", peer_id, peer.protocol_version, peer.network_id, peer.difficulty, peer.latest, peer.genesis);
@@ -409,6 +416,7 @@ impl ChainSync {
 
 		trace!(target: "sync", "{} -> NewBlock ({})", peer_id, h);
 		let header_view = HeaderView::new(header_rlp.as_raw());
+ 		let mut unknown = false;
 		// TODO: Decompose block and add to self.headers and self.bodies instead
 		if header_view.number() == From::from(self.current_base_block() + 1) {
 			match io.chain().import_block(block_rlp.as_raw().to_vec()) {
@@ -421,6 +429,10 @@ impl ChainSync {
 				Ok(_) => {
 					trace!(target: "sync", "New block queued {:?}", h);
 				},
+				Err(ImportError::UnknownParent) => {
+					unknown = true;
+					trace!(target: "sync", "New block with unknown parent {:?}", h);
+				},
 				Err(e) => {
 					debug!(target: "sync", "Bad new block {:?} : {:?}", h, e);
 					io.disable_peer(peer_id);
@@ -428,6 +440,9 @@ impl ChainSync {
 			};
 		}
   		else {
+			unknown = true;
+		}
+		if unknown {
 			trace!(target: "sync", "New block unknown {:?}", h);
 			//TODO: handle too many unknown blocks
 			let difficulty: U256 = try!(r.val_at(1));
@@ -676,6 +691,14 @@ impl ChainSync {
 				block_rlp.append_raw(body.at(0).as_raw(), 1);
 				block_rlp.append_raw(body.at(1).as_raw(), 1);
 				let h = &headers.1[i].hash;
+
+				// Perform basic block verification
+				if !Block::is_good(block_rlp.as_raw()) {
+					debug!(target: "sync", "Bad block rlp {:?} : {:?}", h, block_rlp.as_raw());
+					restart = true;
+					break;
+				}
+
 				match io.chain().import_block(block_rlp.out()) {
 					Err(ImportError::AlreadyInChain) => {
 						trace!(target: "sync", "Block already in chain {:?}", h);
@@ -795,6 +818,7 @@ impl ChainSync {
 			Ok(_) => {
 				let mut peer = self.peers.get_mut(&peer_id).unwrap();
 				peer.asking = asking;
+				peer.ask_time = time::precise_time_s();
 			}
 		}
 	}
@@ -972,4 +996,13 @@ impl ChainSync {
 	/// Maintain other peers. Send out any new blocks and transactions
 	pub fn _maintain_sync(&mut self, _io: &mut SyncIo) {
 	}
+
+	pub fn maintain_peers(&self, io: &mut SyncIo) {
+		let tick = time::precise_time_s();
+		for (peer_id, peer) in &self.peers {
+			if peer.asking != PeerAsking::Nothing && (tick - peer.ask_time) > CONNECTION_TIMEOUT_SEC {
+				io.disconnect_peer(*peer_id);
+			}
+		}
+	}
 }

sync/src/tests.rs

@@ -4,7 +4,7 @@ use ethcore::block_queue::BlockQueueInfo;
 use ethcore::header::{Header as BlockHeader, BlockNumber};
 use ethcore::error::*;
 use io::SyncIo;
-use chain::ChainSync;
+use chain::{ChainSync, SyncState};
 
 struct TestBlockChainClient {
 	blocks: RwLock<HashMap<H256, Bytes>>,
@@ -248,13 +248,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 {
@@ -298,10 +300,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();
 		}
 	}
 
@@ -310,9 +330,8 @@ impl TestNet {
 	}
 }
 
-
 #[test]
-fn full_sync_two_peers() {
+fn chain_two_peers() {
 	::env_logger::init().ok();
 	let mut net = TestNet::new(3);
 	net.peer_mut(1).chain.add_blocks(1000, false);
@@ -323,7 +342,27 @@ fn full_sync_two_peers() {
 }
 
 #[test]
-fn full_sync_empty_blocks() {
+fn chain_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 chain_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 chain_empty_blocks() {
 	::env_logger::init().ok();
 	let mut net = TestNet::new(3);
 	for n in 0..200 {
@@ -336,7 +375,7 @@ fn full_sync_empty_blocks() {
 }
 
 #[test]
-fn forked_sync() {
+fn chain_forged() {
 	::env_logger::init().ok();
 	let mut net = TestNet::new(3);
 	net.peer_mut(0).chain.add_blocks(300, false);
@@ -354,3 +393,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_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_status_empty() {
+	let net = TestNet::new(2);
+	assert_eq!(net.peer(0).sync.status().state, SyncState::NotSynced);
+}