Merge branch 'master' of github.com:ethcore/parity

.travis.yml

@@ -6,6 +6,8 @@ branches:
   - master
   - /^beta-.*$/
   - /^stable-.*$/
+  - /^beta$/
+  - /^stable$/
 matrix:
   fast_finish: true
   include:

Cargo.lock

@@ -19,10 +19,10 @@ dependencies = [
 
 [[package]]
 name = "aho-corasick"
-version = "0.4.1"
+version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
- "memchr 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)",
+ "memchr 0.1.10 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -64,7 +64,7 @@ name = "clippy"
 version = "0.0.41"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
- "regex-syntax 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "regex-syntax 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)",
  "semver 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "unicode-normalization 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
@@ -113,7 +113,7 @@ name = "docopt"
 version = "0.6.78"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
- "regex 0.1.48 (registry+https://github.com/rust-lang/crates.io-index)",
+ "regex 0.1.51 (registry+https://github.com/rust-lang/crates.io-index)",
  "rustc-serialize 0.3.18 (registry+https://github.com/rust-lang/crates.io-index)",
  "strsim 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
@@ -137,7 +137,7 @@ version = "0.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "log 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)",
- "regex 0.1.48 (registry+https://github.com/rust-lang/crates.io-index)",
+ "regex 0.1.51 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -146,11 +146,11 @@ version = "0.5.4"
 source = "git+https://github.com/arkpar/rust-secp256k1.git#321e6c22a83606d1875f89cb61c9cb37c7d249ae"
 dependencies = [
  "arrayvec 0.3.15 (registry+https://github.com/rust-lang/crates.io-index)",
- "gcc 0.3.23 (registry+https://github.com/rust-lang/crates.io-index)",
+ "gcc 0.3.24 (registry+https://github.com/rust-lang/crates.io-index)",
  "libc 0.1.12 (registry+https://github.com/rust-lang/crates.io-index)",
  "rand 0.3.14 (registry+https://github.com/rust-lang/crates.io-index)",
  "rustc-serialize 0.3.18 (registry+https://github.com/rust-lang/crates.io-index)",
- "serde 0.6.12 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde 0.6.13 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde_json 0.6.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
@@ -193,7 +193,7 @@ dependencies = [
  "jsonrpc-core 1.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "jsonrpc-http-server 1.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "rustc-serialize 0.3.18 (registry+https://github.com/rust-lang/crates.io-index)",
- "serde 0.6.12 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde 0.6.13 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde_json 0.6.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde_macros 0.6.13 (registry+https://github.com/rust-lang/crates.io-index)",
  "target_info 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -220,7 +220,7 @@ dependencies = [
  "rocksdb 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "rust-crypto 0.2.34 (registry+https://github.com/rust-lang/crates.io-index)",
  "rustc-serialize 0.3.18 (registry+https://github.com/rust-lang/crates.io-index)",
- "serde 0.6.12 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde 0.6.13 (registry+https://github.com/rust-lang/crates.io-index)",
  "sha3 0.1.0",
  "slab 0.1.4 (git+https://github.com/arkpar/slab.git)",
  "target_info 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -250,7 +250,7 @@ dependencies = [
 
 [[package]]
 name = "gcc"
-version = "0.3.23"
+version = "0.3.24"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 
 [[package]]
@@ -326,7 +326,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "hyper 0.6.16 (registry+https://github.com/rust-lang/crates.io-index)",
  "rand 0.3.14 (registry+https://github.com/rust-lang/crates.io-index)",
- "regex 0.1.48 (registry+https://github.com/rust-lang/crates.io-index)",
+ "regex 0.1.51 (registry+https://github.com/rust-lang/crates.io-index)",
  "xml-rs 0.1.26 (registry+https://github.com/rust-lang/crates.io-index)",
  "xmltree 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
@@ -349,7 +349,7 @@ name = "jsonrpc-core"
 version = "1.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
- "serde 0.6.12 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde 0.6.13 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde_json 0.6.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde_macros 0.6.13 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
@@ -425,7 +425,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 
 [[package]]
 name = "memchr"
-version = "0.1.7"
+version = "0.1.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "libc 0.2.7 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -437,7 +437,7 @@ version = "0.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "log 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)",
- "serde 0.6.12 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde 0.6.13 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -556,17 +556,18 @@ dependencies = [
 
 [[package]]
 name = "regex"
-version = "0.1.48"
+version = "0.1.51"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
- "aho-corasick 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "aho-corasick 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)",
- "memchr 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)",
+ "memchr 0.1.10 (registry+https://github.com/rust-lang/crates.io-index)",
- "regex-syntax 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "regex-syntax 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)",
+ "utf8-ranges 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
 name = "regex-syntax"
-version = "0.2.2"
+version = "0.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 
 [[package]]
@@ -582,7 +583,7 @@ name = "rust-crypto"
 version = "0.2.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
- "gcc 0.3.23 (registry+https://github.com/rust-lang/crates.io-index)",
+ "gcc 0.3.24 (registry+https://github.com/rust-lang/crates.io-index)",
  "libc 0.1.12 (registry+https://github.com/rust-lang/crates.io-index)",
  "rand 0.3.14 (registry+https://github.com/rust-lang/crates.io-index)",
  "rustc-serialize 0.3.18 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -617,7 +618,7 @@ dependencies = [
 
 [[package]]
 name = "serde"
-version = "0.6.12"
+version = "0.6.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "num 0.1.30 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -639,7 +640,7 @@ version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "num 0.1.30 (registry+https://github.com/rust-lang/crates.io-index)",
- "serde 0.6.12 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde 0.6.13 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -654,7 +655,7 @@ dependencies = [
 name = "sha3"
 version = "0.1.0"
 dependencies = [
- "gcc 0.3.23 (registry+https://github.com/rust-lang/crates.io-index)",
+ "gcc 0.3.24 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -754,6 +755,11 @@ dependencies = [
  "uuid 0.1.18 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "utf8-ranges"
+version = "0.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
 [[package]]
 name = "uuid"
 version = "0.1.18"

ethcore/src/error.rs

@@ -20,26 +20,6 @@ use util::*;
 use header::BlockNumber;
 use basic_types::LogBloom;
 
-#[derive(Debug, PartialEq, Eq)]
-/// Error indicating an expected value was not found.
-pub struct Mismatch<T: fmt::Debug> {
-	/// Value expected.
-	pub expected: T,
-	/// Value found.
-	pub found: T,
-}
-
-#[derive(Debug, PartialEq, Eq)]
-/// Error indicating value found is outside of a valid range.
-pub struct OutOfBounds<T: fmt::Debug> {
-	/// Minimum allowed value.
-	pub min: Option<T>,
-	/// Maximum allowed value.
-	pub max: Option<T>,
-	/// Value found.
-	pub found: T,
-}
-
 /// Result of executing the transaction.
 #[derive(PartialEq, Debug)]
 pub enum ExecutionError {

sync/src/chain.rs

@@ -155,13 +155,17 @@ struct PeerInfo {
 	/// Peer network id
 	network_id: U256,
 	/// Peer best block hash
-	latest: H256,
+	latest_hash: H256,
+	/// Peer best block number if known
+	latest_number: Option<BlockNumber>,
 	/// Peer total difficulty
 	difficulty: U256,
 	/// Type of data currenty being requested from peer.
 	asking: PeerAsking,
 	/// A set of block numbers being requested
 	asking_blocks: Vec<BlockNumber>,
+	/// Holds requested header hash if currently requesting block header by hash
+	asking_hash: Option<H256>,
 	/// Request timestamp
 	ask_time: f64,
 }
@@ -179,6 +183,8 @@ pub struct ChainSync {
 	downloading_headers: HashSet<BlockNumber>,
 	/// Set of block body numbers being downloaded
 	downloading_bodies: HashSet<BlockNumber>,
+	/// Set of block headers being downloaded by hash
+	downloading_hashes: HashSet<H256>,
 	/// Downloaded headers.
 	headers: Vec<(BlockNumber, Vec<Header>)>, //TODO: use BTreeMap once range API is sable. For now it is a vector sorted in descending order
 	/// Downloaded bodies
@@ -195,6 +201,8 @@ pub struct ChainSync {
 	syncing_difficulty: U256,
 	/// True if common block for our and remote chain has been found
 	have_common_block: bool,
+	/// Last propagated block number
+	last_send_block_number: BlockNumber,
 }
 
 type RlpResponseResult = Result<Option<(PacketId, RlpStream)>, PacketDecodeError>;
@@ -208,6 +216,7 @@ impl ChainSync {
 			highest_block: None,
 			downloading_headers: HashSet::new(),
 			downloading_bodies: HashSet::new(),
+			downloading_hashes: HashSet::new(),
 			headers: Vec::new(),
 			bodies: Vec::new(),
 			peers: HashMap::new(),
@@ -216,6 +225,7 @@ impl ChainSync {
 			last_imported_hash: None,
 			syncing_difficulty: U256::from(0u64),
 			have_common_block: false,
+			last_send_block_number: 0,
 		}
 	}
 
@@ -250,6 +260,7 @@ impl ChainSync {
 		self.bodies.clear();
 		for (_, ref mut p) in &mut self.peers {
 			p.asking_blocks.clear();
+			p.asking_hash = None;
 		}
 		self.header_ids.clear();
 		self.syncing_difficulty = From::from(0u64);
@@ -275,15 +286,21 @@ impl ChainSync {
 			protocol_version: try!(r.val_at(0)),
 			network_id: try!(r.val_at(1)),
 			difficulty: try!(r.val_at(2)),
-			latest: try!(r.val_at(3)),
+			latest_hash: try!(r.val_at(3)),
+			latest_number: None,
 			genesis: try!(r.val_at(4)),
 			asking: PeerAsking::Nothing,
 			asking_blocks: Vec::new(),
+			asking_hash: None,
 			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);
+		trace!(target: "sync", "New peer {} (protocol: {}, network: {:?}, difficulty: {:?}, latest:{}, genesis:{})", peer_id, peer.protocol_version, peer.network_id, peer.difficulty, peer.latest_hash, peer.genesis);
 
+		if self.peers.contains_key(&peer_id) {
+			warn!("Unexpected status packet from {}:{}", peer_id, io.peer_info(peer_id));
+			return Ok(());
+		}
 		let chain_info = io.chain().chain_info();
 		if peer.genesis != chain_info.genesis_hash {
 			io.disable_peer(peer_id);
@@ -296,10 +313,7 @@ impl ChainSync {
 			return Ok(());
 		}
 
-		let old = self.peers.insert(peer_id.clone(), peer);
+		self.peers.insert(peer_id.clone(), peer);
-		if old.is_some() {
-			panic!("ChainSync: new peer already exists");
-		}
 		info!(target: "sync", "Connected {}:{}", peer_id, io.peer_info(peer_id));
 		self.sync_peer(io, peer_id, false);
 		Ok(())
@@ -439,6 +453,11 @@ impl ChainSync {
 		trace!(target: "sync", "{} -> NewBlock ({})", peer_id, h);
 		let header: BlockHeader = try!(header_rlp.as_val());
  		let mut unknown = false;
+		{
+			let peer = self.peers.get_mut(&peer_id).unwrap();
+			peer.latest_hash = header.hash();
+			peer.latest_number = Some(header.number());
+		}
 		// TODO: Decompose block and add to self.headers and self.bodies instead
 		if header.number == From::from(self.current_base_block() + 1) {
 			match io.chain().import_block(block_rlp.as_raw().to_vec()) {
@@ -449,6 +468,7 @@ impl ChainSync {
 					trace!(target: "sync", "New block already queued {:?}", h);
 				},
 				Ok(_) => {
+					self.last_imported_block = Some(header.number);
 					trace!(target: "sync", "New block queued {:?}", h);
 				},
 				Err(ImportError::UnknownParent) => {
@@ -468,13 +488,9 @@ impl ChainSync {
 			trace!(target: "sync", "New block unknown {:?}", h);
 			//TODO: handle too many unknown blocks
 			let difficulty: U256 = try!(r.val_at(1));
-			let peer_difficulty = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer").difficulty;
+			let peer_difficulty = self.peers.get_mut(&peer_id).unwrap().difficulty;
 			if difficulty > peer_difficulty {
 				trace!(target: "sync", "Received block {:?}  with no known parent. Peer needs syncing...", h);
-				{
-					let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer");
-					peer.latest = header.hash();
-				}
 				self.sync_peer(io, peer_id, true);
 			}
 		}
@@ -483,16 +499,19 @@ impl ChainSync {
 
 	/// Handles NewHashes packet. Initiates headers download for any unknown hashes.
 	fn on_peer_new_hashes(&mut self, io: &mut SyncIo, peer_id: PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
-		if self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer").asking != PeerAsking::Nothing {
+		if self.peers.get_mut(&peer_id).unwrap().asking != PeerAsking::Nothing {
 			trace!(target: "sync", "Ignoring new hashes since we're already downloading.");
 			return Ok(());
 		}
 		trace!(target: "sync", "{} -> NewHashes ({} entries)", peer_id, r.item_count());
-		let hashes = r.iter().map(|item| (item.val_at::<H256>(0), item.val_at::<U256>(1)));
+		let hashes = r.iter().map(|item| (item.val_at::<H256>(0), item.val_at::<BlockNumber>(1)));
-		let mut max_height: U256 = From::from(0);
+		let mut max_height: BlockNumber = 0;
 		for (rh, rd) in hashes {
 			let h = try!(rh);
 			let d = try!(rd);
+			if self.downloading_hashes.contains(&h) {
+				continue;
+			}
 			match io.chain().block_status(BlockId::Hash(h.clone())) {
 				BlockStatus::InChain  => {
 					trace!(target: "sync", "New block hash already in chain {:?}", h);
@@ -501,10 +520,11 @@ impl ChainSync {
 					trace!(target: "sync", "New hash block already queued {:?}", h);
 				},
 				BlockStatus::Unknown => {
-					trace!(target: "sync", "New unknown block hash {:?}", h);
 					if d > max_height {
-						let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer");
+						trace!(target: "sync", "New unknown block hash {:?}", h);
-						peer.latest = h.clone();
+						let peer = self.peers.get_mut(&peer_id).unwrap();
+						peer.latest_hash = h.clone();
+						peer.latest_number = Some(d);
 						max_height = d;
 					}
 				},
@@ -515,7 +535,7 @@ impl ChainSync {
 				}
 			}
 		};
-		if max_height != x!(0) {
+		if max_height != 0 {
 			self.sync_peer(io, peer_id, true);
 		}
 		Ok(())
@@ -525,7 +545,7 @@ impl ChainSync {
 	pub fn on_peer_aborting(&mut self, io: &mut SyncIo, peer: PeerId) {
 		trace!(target: "sync", "== Disconnecting {}", peer);
 		if self.peers.contains_key(&peer) {
-			info!(target: "sync", "Disconnected {}:{}", peer, io.peer_info(peer));
+			info!(target: "sync", "Disconnected {}", peer);
 			self.clear_peer_download(peer);
 			self.peers.remove(&peer);
 			self.continue_sync(io);
@@ -563,7 +583,7 @@ impl ChainSync {
 	/// Find something to do for a peer. Called for a new peer or when a peer is done with it's task.
 	fn sync_peer(&mut self, io: &mut SyncIo,  peer_id: PeerId, force: bool) {
 		let (peer_latest, peer_difficulty) = {
-			let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer");
+			let peer = self.peers.get_mut(&peer_id).unwrap();
 			if peer.asking != PeerAsking::Nothing {
 				return;
 			}
@@ -571,7 +591,7 @@ impl ChainSync {
 				trace!(target: "sync", "Waiting for block queue");
 				return;
 			}
-			(peer.latest.clone(), peer.difficulty.clone())
+			(peer.latest_hash.clone(), peer.difficulty.clone())
 		};
 
 		let td = io.chain().chain_info().pending_total_difficulty;
@@ -583,6 +603,8 @@ impl ChainSync {
 				self.state = SyncState::Blocks;
 			}
 			trace!(target: "sync", "Starting sync with better chain");
+			self.peers.get_mut(&peer_id).unwrap().asking_hash = Some(peer_latest.clone());
+			self.downloading_hashes.insert(peer_latest.clone());
 			self.request_headers_by_hash(io, peer_id, &peer_latest, 1, 0, false);
 		}
 		else if self.state == SyncState::Blocks && io.chain().block_status(BlockId::Hash(peer_latest)) == BlockStatus::Unknown {
@@ -675,6 +697,8 @@ impl ChainSync {
 				}
 			}
 			else {
+				// continue search for common block
+				self.downloading_headers.insert(start as BlockNumber);
 				self.request_headers_by_number(io, peer_id, start as BlockNumber, 1, 0, false);
 			}
 		}
@@ -682,7 +706,10 @@ impl ChainSync {
 
 	/// Clear all blocks/headers marked as being downloaded by a peer.
 	fn clear_peer_download(&mut self, peer_id: PeerId) {
-		let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer");
+		let peer = self.peers.get_mut(&peer_id).unwrap();
+		if let Some(hash) = peer.asking_hash.take() {
+			self.downloading_hashes.remove(&hash);
+		}
 		for b in &peer.asking_blocks {
 			self.downloading_headers.remove(&b);
 			self.downloading_bodies.remove(&b);
@@ -815,7 +842,7 @@ impl ChainSync {
 
 	/// Reset peer status after request is complete.
 	fn reset_peer_asking(&mut self, peer_id: PeerId, asking: PeerAsking) {
-		let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer");
+		let peer = self.peers.get_mut(&peer_id).unwrap();
 		if peer.asking != asking {
 			warn!(target:"sync", "Asking {:?} while expected {:?}", peer.asking, asking);
 		}
@@ -827,9 +854,9 @@ impl ChainSync {
 	/// Generic request sender
 	fn send_request(&mut self, sync: &mut SyncIo, peer_id: PeerId, asking: PeerAsking,  packet_id: PacketId, packet: Bytes) {
 		{
-			let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer");
+			let peer = self.peers.get_mut(&peer_id).unwrap();
 			if peer.asking != PeerAsking::Nothing {
-				warn!(target:"sync", "Asking {:?} while requesting {:?}", asking, peer.asking);
+				warn!(target:"sync", "Asking {:?} while requesting {:?}", peer.asking, asking);
 			}
 		}
 		match sync.send(peer_id, packet_id, packet) {
@@ -846,6 +873,14 @@ impl ChainSync {
 		}
 	}
 
+	/// Generic packet sender
+	fn send_packet(&mut self, sync: &mut SyncIo, peer_id: PeerId, packet_id: PacketId, packet: Bytes) {
+		if let Err(e) = sync.send(peer_id, packet_id, packet) {
+			warn!(target:"sync", "Error sending packet: {:?}", e);
+			sync.disable_peer(peer_id);
+			self.on_peer_aborting(sync, peer_id);
+		}
+	}
 	/// Called when peer sends us new transactions
 	fn on_peer_transactions(&mut self, _io: &mut SyncIo, _peer_id: PeerId, _r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
 		Ok(())
@@ -1002,6 +1037,11 @@ impl ChainSync {
 	/// Dispatch incoming requests and responses
 	pub fn on_packet(&mut self, io: &mut SyncIo, peer: PeerId, packet_id: u8, data: &[u8]) {
 		let rlp = UntrustedRlp::new(data);
+
+		if packet_id != STATUS_PACKET && !self.peers.contains_key(&peer) {
+			warn!(target:"sync", "Unexpected packet from unregistered peer: {}:{}", peer, io.peer_info(peer));
+			return;
+		}
 		let result = match packet_id {
 			STATUS_PACKET => self.on_peer_status(io, peer, &rlp),
 			TRANSACTIONS_PACKET => self.on_peer_transactions(io, peer, &rlp),
@@ -1063,7 +1103,6 @@ impl ChainSync {
 						for block_hash in route.blocks {
 							let mut hash_rlp = RlpStream::new_list(2);
 							let difficulty = chain.block_total_difficulty(BlockId::Hash(block_hash.clone())).expect("Mallformed block without a difficulty on the chain!");
-
 							hash_rlp.append(&block_hash);
 							hash_rlp.append(&difficulty);
 							rlp_stream.append_raw(&hash_rlp.out(), 1);
@@ -1079,32 +1118,34 @@ impl ChainSync {
 	/// creates latest block rlp for the given client
 	fn create_latest_block_rlp(chain: &BlockChainClient) -> Bytes {
 		let mut rlp_stream = RlpStream::new_list(2);
-		rlp_stream.append_raw(&chain.block(BlockId::Hash(chain.chain_info().best_block_hash)).expect("Creating latest block when there is none"), 1);
+		rlp_stream.append_raw(&chain.block(BlockId::Hash(chain.chain_info().best_block_hash)).unwrap(), 1);
 		rlp_stream.append(&chain.chain_info().total_difficulty);
 		rlp_stream.out()
 	}
 
 	/// returns peer ids that have less blocks than our chain
-	fn get_lagging_peers(&self, io: &SyncIo) -> Vec<PeerId> {
+	fn get_lagging_peers(&mut self, io: &SyncIo) -> Vec<(PeerId, BlockNumber)> {
 		let chain = io.chain();
 		let chain_info = chain.chain_info();
 		let latest_hash = chain_info.best_block_hash;
 		let latest_number = chain_info.best_block_number;
-		self.peers.iter().filter(|&(_, peer_info)|
+		self.peers.iter_mut().filter_map(|(&id, ref mut peer_info)|
-			match io.chain().block_status(BlockId::Hash(peer_info.latest.clone()))
+			match io.chain().block_status(BlockId::Hash(peer_info.latest_hash.clone())) {
-			{
 				BlockStatus::InChain => {
-					let peer_number = HeaderView::new(&io.chain().block_header(BlockId::Hash(peer_info.latest.clone())).unwrap()).number();
+					if peer_info.latest_number.is_none() {
-					peer_info.latest != latest_hash && latest_number > peer_number && latest_number - peer_number < MAX_PEER_LAG_PROPAGATION
+						peer_info.latest_number = Some(HeaderView::new(&io.chain().block_header(BlockId::Hash(peer_info.latest_hash.clone())).unwrap()).number());
+					}
+					if peer_info.latest_hash != latest_hash && latest_number > peer_info.latest_number.unwrap() {
+						Some((id, peer_info.latest_number.unwrap()))
+					} else { None }
 				},
-				_ => false
+				_ => None
 			})
-			.map(|(peer_id, _)| peer_id)
+			.collect::<Vec<_>>()
-			.cloned().collect::<Vec<PeerId>>()
 	}
 
 	/// propagades latest block to lagging peers
-	fn propagade_blocks(&mut self, io: &mut SyncIo) -> usize {
+	fn propagade_blocks(&mut self, local_best: &H256, best_number: BlockNumber, io: &mut SyncIo) -> usize {
 		let updated_peers = {
 			let lagging_peers = self.get_lagging_peers(io);
 
@@ -1112,37 +1153,43 @@ impl ChainSync {
 			let fraction = (self.peers.len() as f64).powf(-0.5).mul(u32::max_value() as f64).round() as u32;
 			let lucky_peers = match lagging_peers.len() {
 				0 ... MIN_PEERS_PROPAGATION => lagging_peers,
-				_ => lagging_peers.iter().filter(|_| ::rand::random::<u32>() < fraction).cloned().collect::<Vec<PeerId>>()
+				_ => lagging_peers.into_iter().filter(|_| ::rand::random::<u32>() < fraction).collect::<Vec<_>>()
 			};
 
 			// taking at max of MAX_PEERS_PROPAGATION
-			lucky_peers.iter().take(min(lucky_peers.len(), MAX_PEERS_PROPAGATION)).cloned().collect::<Vec<PeerId>>()
+			lucky_peers.iter().map(|&(id, _)| id.clone()).take(min(lucky_peers.len(), MAX_PEERS_PROPAGATION)).collect::<Vec<PeerId>>()
 		};
 
 		let mut sent = 0;
-		let local_best = io.chain().chain_info().best_block_hash;
 		for peer_id in updated_peers {
 			let rlp = ChainSync::create_latest_block_rlp(io.chain());
-			self.send_request(io, peer_id, PeerAsking::Nothing, NEW_BLOCK_PACKET, rlp);
+			self.send_packet(io, peer_id, NEW_BLOCK_PACKET, rlp);
-			self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer").latest = local_best.clone();
+			self.peers.get_mut(&peer_id).unwrap().latest_hash = local_best.clone();
+			self.peers.get_mut(&peer_id).unwrap().latest_number = Some(best_number);
 			sent = sent + 1;
 		}
 		sent
 	}
 
 	/// propagades new known hashes to all peers
-	fn propagade_new_hashes(&mut self, io: &mut SyncIo) -> usize {
+	fn propagade_new_hashes(&mut self, local_best: &H256, best_number: BlockNumber, io: &mut SyncIo) -> usize {
 		let updated_peers = self.get_lagging_peers(io);
 		let mut sent = 0;
-		let local_best = io.chain().chain_info().best_block_hash;
+		let last_parent = HeaderView::new(&io.chain().block_header(BlockId::Hash(local_best.clone())).unwrap()).parent_hash();
-		for peer_id in updated_peers {
+		for (peer_id, peer_number) in updated_peers {
-			sent = sent + match ChainSync::create_new_hashes_rlp(io.chain(), &self.peers.get(&peer_id).expect("ChainSync: unknown peer").latest, &local_best) {
+			let mut peer_best = self.peers.get(&peer_id).unwrap().latest_hash.clone();
+			if best_number - peer_number > MAX_PEERS_PROPAGATION as BlockNumber {
+				// If we think peer is too far behind just end one latest hash
+				peer_best = last_parent.clone();
+			}
+			sent = sent + match ChainSync::create_new_hashes_rlp(io.chain(), &peer_best, &local_best) {
 				Some(rlp) => {
 					{
-						let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer");
+						let peer = self.peers.get_mut(&peer_id).unwrap();
-						peer.latest = local_best.clone();
+						peer.latest_hash = local_best.clone();
+						peer.latest_number = Some(best_number);
 					}
-					self.send_request(io, peer_id, PeerAsking::Nothing, NEW_BLOCK_HASHES_PACKET, rlp);
+					self.send_packet(io, peer_id, NEW_BLOCK_HASHES_PACKET, rlp);
 					1
 				},
 				None => 0
@@ -1154,15 +1201,19 @@ impl ChainSync {
 	/// Maintain other peers. Send out any new blocks and transactions
 	pub fn maintain_sync(&mut self, io: &mut SyncIo) {
 		self.check_resume(io);
-
-		let peers = self.propagade_new_hashes(io);
-		trace!(target: "sync", "Sent new hashes to peers: {:?}", peers);
 	}
 
 	/// should be called once chain has new block, triggers the latest block propagation
 	pub fn chain_blocks_verified(&mut self, io: &mut SyncIo) {
-		let peers = self.propagade_blocks(io);
+		let chain = io.chain().chain_info();
-		trace!(target: "sync", "Sent latest block to peers: {:?}", peers);
+		if (((chain.best_block_number as i64) - (self.last_send_block_number as i64)).abs() as BlockNumber) < MAX_PEER_LAG_PROPAGATION {
+			let blocks = self.propagade_blocks(&chain.best_block_hash, chain.best_block_number, io);
+			let hashes = self.propagade_new_hashes(&chain.best_block_hash, chain.best_block_number, io);
+			if blocks != 0 || hashes != 0 {
+				trace!(target: "sync", "Sent latest {} blocks and {} hashes to peers.", blocks, hashes);
+			}
+		}
+		self.last_send_block_number = chain.best_block_number;
 	}
 }
 
@@ -1248,9 +1299,9 @@ mod tests {
 		// the length of two rlp-encoded receipts
 		assert_eq!(597, rlp_result.unwrap().1.out().len());
 
-		let mut sync = ChainSync::new();
+		let mut sync = dummy_sync_with_peer(H256::new());
 		io.sender = Some(2usize);
-		sync.on_packet(&mut io, 1usize, super::GET_RECEIPTS_PACKET, &receipts_request);
+		sync.on_packet(&mut io, 0usize, super::GET_RECEIPTS_PACKET, &receipts_request);
 		assert_eq!(1, io.queue.len());
 	}
 
@@ -1276,9 +1327,9 @@ mod tests {
 		// the length of one rlp-encoded hashe
 		assert_eq!(34, rlp_result.unwrap().1.out().len());
 
-		let mut sync = ChainSync::new();
+		let mut sync = dummy_sync_with_peer(H256::new());
 		io.sender = Some(2usize);
-		sync.on_packet(&mut io, 1usize, super::GET_NODE_DATA_PACKET, &node_request);
+		sync.on_packet(&mut io, 0usize, super::GET_NODE_DATA_PACKET, &node_request);
 		assert_eq!(1, io.queue.len());
 	}
 
@@ -1289,10 +1340,12 @@ mod tests {
 				protocol_version: 0,
 				genesis: H256::zero(),
 				network_id: U256::zero(),
-				latest: peer_latest_hash,
+				latest_hash: peer_latest_hash,
+				latest_number: None,
 				difficulty: U256::zero(),
 				asking: PeerAsking::Nothing,
 				asking_blocks: Vec::<BlockNumber>::new(),
+				asking_hash: None,
 				ask_time: 0f64,
 	  		});
 		sync
@@ -1303,7 +1356,7 @@ mod tests {
 		let mut client = TestBlockChainClient::new();
 		client.add_blocks(100, false);
 		let mut queue = VecDeque::new();
-		let sync = dummy_sync_with_peer(client.block_hash_delta_minus(10));
+		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(10));
 		let io = TestIo::new(&mut client, &mut queue, None);
 
 		let lagging_peers = sync.get_lagging_peers(&io);
@@ -1334,9 +1387,11 @@ mod tests {
 		client.add_blocks(100, false);
 		let mut queue = VecDeque::new();
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5));
+		let best_hash = client.chain_info().best_block_hash.clone();
+		let best_number = client.chain_info().best_block_number;
 		let mut io = TestIo::new(&mut client, &mut queue, None);
 
-		let peer_count = sync.propagade_new_hashes(&mut io);
+		let peer_count = sync.propagade_new_hashes(&best_hash, best_number, &mut io);
 
 		// 1 message should be send
 		assert_eq!(1, io.queue.len());
@@ -1352,9 +1407,11 @@ mod tests {
 		client.add_blocks(100, false);
 		let mut queue = VecDeque::new();
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5));
+		let best_hash = client.chain_info().best_block_hash.clone();
+		let best_number = client.chain_info().best_block_number;
 		let mut io = TestIo::new(&mut client, &mut queue, None);
 
-		let peer_count = sync.propagade_blocks(&mut io);
+		let peer_count = sync.propagade_blocks(&best_hash, best_number, &mut io);
 
 		// 1 message should be send
 		assert_eq!(1, io.queue.len());
@@ -1456,9 +1513,11 @@ mod tests {
 		client.add_blocks(100, false);
 		let mut queue = VecDeque::new();
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5));
+		let best_hash = client.chain_info().best_block_hash.clone();
+		let best_number = client.chain_info().best_block_number;
 		let mut io = TestIo::new(&mut client, &mut queue, None);
 
-		sync.propagade_new_hashes(&mut io);
+		sync.propagade_new_hashes(&best_hash, best_number, &mut io);
 
 		let data = &io.queue[0].data.clone();
 		let result = sync.on_peer_new_hashes(&mut io, 0, &UntrustedRlp::new(&data));
@@ -1473,9 +1532,11 @@ mod tests {
 		client.add_blocks(100, false);
 		let mut queue = VecDeque::new();
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5));
+		let best_hash = client.chain_info().best_block_hash.clone();
+		let best_number = client.chain_info().best_block_number;
 		let mut io = TestIo::new(&mut client, &mut queue, None);
 
-		sync.propagade_blocks(&mut io);
+		sync.propagade_blocks(&best_hash, best_number, &mut io);
 
 		let data = &io.queue[0].data.clone();
 		let result = sync.on_peer_new_block(&mut io, 0, &UntrustedRlp::new(&data));

sync/src/tests/chain.rs

@@ -109,7 +109,7 @@ fn status_empty() {
 #[test]
 fn status_packet() {
 	let mut net = TestNet::new(2);
-	net.peer_mut(0).chain.add_blocks(1000, false);
+	net.peer_mut(0).chain.add_blocks(100, false);
 	net.peer_mut(1).chain.add_blocks(1, false);
 
 	net.start();
@@ -122,18 +122,28 @@ fn status_packet() {
 
 #[test]
 fn propagade_hashes() {
-	let mut net = TestNet::new(3);
+	let mut net = TestNet::new(6);
-	net.peer_mut(1).chain.add_blocks(1000, false);
+	net.peer_mut(1).chain.add_blocks(10, false);
-	net.peer_mut(2).chain.add_blocks(1000, false);
 	net.sync();
 
 	net.peer_mut(0).chain.add_blocks(10, false);
-	net.sync_step_peer(0);
+	net.sync();
+	net.trigger_block_verified(0); //first event just sets the marker
+	net.trigger_block_verified(0);
 
-	// 2 peers to sync
+	// 5 peers to sync
-	assert_eq!(2, net.peer(0).queue.len());
+	assert_eq!(5, net.peer(0).queue.len());
-	// NEW_BLOCK_HASHES_PACKET
+	let mut hashes = 0;
-	assert_eq!(0x01, net.peer(0).queue[0].packet_id);
+	let mut blocks = 0;
+	for i in 0..5 {
+		if net.peer(0).queue[i].packet_id == 0x1 {
+			hashes += 1;
+		}
+		if net.peer(0).queue[i].packet_id == 0x7 {
+			blocks += 1;
+		}
+	}
+	assert!(blocks + hashes == 5);
 }
 
 #[test]
@@ -143,9 +153,21 @@ fn propagade_blocks() {
 	net.sync();
 
 	net.peer_mut(0).chain.add_blocks(10, false);
+	net.trigger_block_verified(0); //first event just sets the marker
 	net.trigger_block_verified(0);
 
 	assert!(!net.peer(0).queue.is_empty());
 	// NEW_BLOCK_PACKET
 	assert_eq!(0x07, net.peer(0).queue[0].packet_id);
 }
+
+#[test]
+fn restart_on_malformed_block() {
+	let mut net = TestNet::new(2);
+	net.peer_mut(1).chain.add_blocks(10, false);
+	net.peer_mut(1).chain.corrupt_block(6);
+	net.sync_steps(10);
+
+	assert_eq!(net.peer(0).chain.chain_info().best_block_number, 4);
+}
+

sync/src/tests/helpers.rs

@@ -71,6 +71,17 @@ impl TestBlockChainClient {
 		}
 	}
 
+	pub fn corrupt_block(&mut self, n: BlockNumber) {
+		let hash = self.block_hash(BlockId::Number(n)).unwrap();
+		let mut header: BlockHeader = decode(&self.block_header(BlockId::Number(n)).unwrap());
+		header.parent_hash = H256::new();
+		let mut rlp = RlpStream::new_list(3);
+		rlp.append(&header);
+		rlp.append_raw(&rlp::NULL_RLP, 1);
+		rlp.append_raw(&rlp::NULL_RLP, 1);
+		self.blocks.write().unwrap().insert(hash, rlp.out());
+	}
+
 	pub fn block_hash_delta_minus(&mut self, delta: usize) -> H256 {
 		let blocks_read = self.numbers.read().unwrap();
 		let index = blocks_read.len() - delta;

util/src/error.rs

@@ -20,6 +20,7 @@ use rustc_serialize::hex::FromHexError;
 use network::NetworkError;
 use rlp::DecoderError;
 use io;
+use std::fmt;
 
 #[derive(Debug)]
 /// Error in database subsystem.
@@ -55,6 +56,27 @@ pub enum UtilError {
 	BadSize,
 }
 
+
+#[derive(Debug, PartialEq, Eq)]
+/// Error indicating an expected value was not found.
+pub struct Mismatch<T: fmt::Debug> {
+	/// Value expected.
+	pub expected: T,
+	/// Value found.
+	pub found: T,
+}
+
+#[derive(Debug, PartialEq, Eq)]
+/// Error indicating value found is outside of a valid range.
+pub struct OutOfBounds<T: fmt::Debug> {
+	/// Minimum allowed value.
+	pub min: Option<T>,
+	/// Maximum allowed value.
+	pub max: Option<T>,
+	/// Value found.
+	pub found: T,
+}
+
 impl From<FromHexError> for UtilError {
 	fn from(err: FromHexError) -> UtilError {
 		UtilError::FromHex(err)

util/src/keys/mod.rs

@@ -0,0 +1,19 @@
+// Copyright 2015, 2016 Ethcore (UK) Ltd.
+// This file is part of Parity.
+
+// Parity is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Key management module
+
+pub mod directory;

util/src/lib.rs

@@ -141,6 +141,7 @@ pub mod io;
 pub mod network;
 pub mod log;
 pub mod panics;
+pub mod keys;
 
 pub use common::*;
 pub use misc::*;
@@ -161,3 +162,6 @@ pub use semantic_version::*;
 pub use network::*;
 pub use io::*;
 pub use log::*;
+
+#[cfg(test)]
+mod tests;

util/src/network/host.rs

@@ -217,6 +217,7 @@ impl<'s, Message> NetworkContext<'s, Message> where Message: Send + Sync + Clone
 					s.send_packet(self.protocol, packet_id as u8, &data).unwrap_or_else(|e| {
 						warn!(target: "net", "Send error: {:?}", e);
 					}); //TODO: don't copy vector data
+					try!(self.io.update_registration(peer));
 				},
 				_ => warn!(target: "net", "Send: Peer is not connected yet")
 			}

util/src/tests/helpers.rs

@@ -0,0 +1,31 @@
+use common::*;
+use std::path::PathBuf;
+use std::fs::{remove_dir_all};
+use std::env;
+
+pub struct RandomTempPath {
+	path: PathBuf
+}
+
+impl RandomTempPath {
+	pub fn create_dir() -> RandomTempPath {
+		let mut dir = env::temp_dir();
+		dir.push(H32::random().hex());
+		fs::create_dir_all(dir.as_path()).unwrap();
+		RandomTempPath {
+			path: dir.clone()
+		}
+	}
+
+	pub fn as_path(&self) -> &PathBuf {
+		&self.path
+	}
+}
+
+impl Drop for RandomTempPath {
+	fn drop(&mut self) {
+		if let Err(e) = remove_dir_all(self.as_path()) {
+			panic!("failed to remove temp directory, probably something failed to destroyed ({})", e);
+		}
+	}
+}

util/src/tests/mod.rs

@@ -0,0 +1 @@
+pub mod helpers;