Merge pull request #3775 from ethcore/auth-bft-test

AuthorityRound network simulation test

Cargo.lock

@@ -660,12 +660,14 @@ dependencies = [
  "clippy 0.0.103 (registry+https://github.com/rust-lang/crates.io-index)",
  "env_logger 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)",
  "ethcore 1.5.0",
+ "ethcore-devtools 1.5.0",
  "ethcore-io 1.5.0",
  "ethcore-ipc 1.5.0",
  "ethcore-ipc-codegen 1.5.0",
  "ethcore-ipc-nano 1.5.0",
  "ethcore-network 1.5.0",
  "ethcore-util 1.5.0",
+ "ethkey 0.2.0",
  "heapsize 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "parking_lot 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)",

ethcore/res/authority_round.json

@@ -4,7 +4,8 @@
 		"AuthorityRound": {
 			"params": {
 				"gasLimitBoundDivisor": "0x0400",
-				"stepDuration": "1",
+				"stepDuration": 1,
+				"startStep": 2,
 				"authorities" : [
 					"0x7d577a597b2742b498cb5cf0c26cdcd726d39e6e",
 					"0x82a978b3f5962a5b0957d9ee9eef472ee55b42f1"

ethcore/src/client/client.rs

@@ -255,6 +255,11 @@ impl Client {
 		self.notify.write().push(Arc::downgrade(&target));
 	}
 
+	/// Returns engine reference.
+	pub fn engine(&self) -> &Engine {
+		&*self.engine
+	}
+
 	fn notify<F>(&self, f: F) where F: Fn(&ChainNotify) {
 		for np in self.notify.read().iter() {
 			if let Some(n) = np.upgrade() {
@@ -570,6 +575,11 @@ impl Client {
 		results.len()
 	}
 
+	/// Get shared miner reference.
+	pub fn miner(&self) -> Arc<Miner> {
+		self.miner.clone()
+	}
+
 	/// Handle messages from the IO queue
 	pub fn handle_queued_message(&self, message: &Bytes) {
 		if let Err(e) = self.engine.handle_message(UntrustedRlp::new(message)) {

ethcore/src/client/test_client.rs

@@ -255,7 +255,7 @@ impl TestBlockChainClient {
 	}
 
 	/// Make a bad block by setting invalid extra data.
-	pub fn corrupt_block(&mut self, n: BlockNumber) {
+	pub fn corrupt_block(&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.set_extra_data(b"This extra data is way too long to be considered valid".to_vec());
@@ -267,7 +267,7 @@ impl TestBlockChainClient {
 	}
 
 	/// Make a bad block by setting invalid parent hash.
-	pub fn corrupt_block_parent(&mut self, n: BlockNumber) {
+	pub fn corrupt_block_parent(&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.set_parent_hash(H256::from(42));

ethcore/src/engines/authority_round.rs

@@ -49,6 +49,8 @@ pub struct AuthorityRoundParams {
 	pub authorities: Vec<Address>,
 	/// Number of authorities.
 	pub authority_n: usize,
+	/// Starting step,
+	pub start_step: Option<u64>,
 }
 
 impl From<ethjson::spec::AuthorityRoundParams> for AuthorityRoundParams {
@@ -58,6 +60,7 @@ impl From<ethjson::spec::AuthorityRoundParams> for AuthorityRoundParams {
 			step_duration: Duration::from_secs(p.step_duration.into()),
 			authority_n: p.authorities.len(),
 			authorities: p.authorities.into_iter().map(Into::into).collect::<Vec<_>>(),
+			start_step: p.start_step.map(Into::into),
 		}
 	}
 }
@@ -97,7 +100,7 @@ impl AsMillis for Duration {
 impl AuthorityRound {
 	/// Create a new instance of AuthorityRound engine.
 	pub fn new(params: CommonParams, our_params: AuthorityRoundParams, builtins: BTreeMap<Address, Builtin>) -> Result<Arc<Self>, Error> {
-		let initial_step = (unix_now().as_secs() / our_params.step_duration.as_secs()) as usize;
+		let initial_step = our_params.start_step.unwrap_or_else(|| (unix_now().as_secs() / our_params.step_duration.as_secs())) as usize;
 		let engine = Arc::new(
 			AuthorityRound {
 				params: params,
@@ -160,14 +163,7 @@ impl IoHandler<()> for TransitionHandler {
 	fn timeout(&self, io: &IoContext<()>, timer: TimerToken) {
 		if timer == ENGINE_TIMEOUT_TOKEN {
 			if let Some(engine) = self.engine.upgrade() {
-				engine.step.fetch_add(1, AtomicOrdering::SeqCst);
+				engine.step();
-				engine.proposed.store(false, AtomicOrdering::SeqCst);
-				if let Some(ref channel) = *engine.message_channel.lock() {
-					match channel.send(ClientIoMessage::UpdateSealing) {
-						Ok(_) => trace!(target: "poa", "timeout: UpdateSealing message sent for step {}.", engine.step.load(AtomicOrdering::Relaxed)),
-						Err(err) => trace!(target: "poa", "timeout: Could not send a sealing message {} for step {}.", err, engine.step.load(AtomicOrdering::Relaxed)),
-					}
-				}
 				io.register_timer_once(ENGINE_TIMEOUT_TOKEN, engine.remaining_step_duration().as_millis())
 					.unwrap_or_else(|e| warn!(target: "poa", "Failed to restart consensus step timer: {}.", e))
 			}
@@ -184,6 +180,17 @@ impl Engine for AuthorityRound {
 	fn params(&self) -> &CommonParams { &self.params }
 	fn builtins(&self) -> &BTreeMap<Address, Builtin> { &self.builtins }
 
+	fn step(&self) {
+		self.step.fetch_add(1, AtomicOrdering::SeqCst);
+		self.proposed.store(false, AtomicOrdering::SeqCst);
+		if let Some(ref channel) = *self.message_channel.lock() {
+			match channel.send(ClientIoMessage::UpdateSealing) {
+				Ok(_) => trace!(target: "poa", "timeout: UpdateSealing message sent for step {}.", self.step.load(AtomicOrdering::Relaxed)),
+				Err(err) => trace!(target: "poa", "timeout: Could not send a sealing message {} for step {}.", err, self.step.load(AtomicOrdering::Relaxed)),
+			}
+		}
+	}
+
 	/// Additional engine-specific information for the user/developer concerning `header`.
 	fn extra_info(&self, header: &Header) -> BTreeMap<String, String> {
 		map![
@@ -236,6 +243,8 @@ impl Engine for AuthorityRound {
 			} else {
 				warn!(target: "poa", "generate_seal: FAIL: Accounts not provided.");
 			}
+		} else {
+			trace!(target: "poa", "generate_seal: Not a proposer for step {}.", step);
 		}
 		Seal::None
 	}

ethcore/src/engines/mod.rs

@@ -215,4 +215,6 @@ pub trait Engine : Sync + Send {
 
 	/// Add an account provider useful for Engines that sign stuff.
 	fn register_account_provider(&self, _account_provider: Arc<AccountProvider>) {}
+	/// Trigger next step of the consensus engine.
+	fn step(&self) {}
 }

ethcore/src/spec/spec.rs

@@ -274,7 +274,7 @@ impl Spec {
 	pub fn new_instant() -> Spec { load_bundled!("instant_seal") }
 
 	/// Create a new Spec with AuthorityRound consensus which does internal sealing (not requiring work).
-	/// Accounts with secrets "1".sha3() and "2".sha3() are the authorities.
+	/// Accounts with secrets "0".sha3() and "1".sha3() are the authorities.
 	pub fn new_test_round() -> Self { load_bundled!("authority_round") }
 
 	/// Create a new Spec with Tendermint consensus which does internal sealing (not requiring work).

json/src/spec/authority_round.rs

@@ -30,6 +30,10 @@ pub struct AuthorityRoundParams {
 	pub step_duration: Uint,
 	/// Valid authorities
 	pub authorities: Vec<Address>,
+	/// Starting step. Determined automatically if not specified.
+	/// To be used for testing only.
+	#[serde(rename="startStep")]
+	pub start_step: Option<Uint>,
 }
 
 /// Authority engine deserialization.
@@ -50,7 +54,8 @@ mod tests {
 			"params": {
 				"gasLimitBoundDivisor": "0x0400",
 				"stepDuration": "0x02",
-				"authorities" : ["0xc6d9d2cd449a754c494264e1809c50e34d64562b"]
+				"authorities" : ["0xc6d9d2cd449a754c494264e1809c50e34d64562b"],
+				"startStep" : 24
 			}
 		}"#;
 

sync/Cargo.toml

@@ -26,6 +26,8 @@ heapsize = "0.3"
 ethcore-ipc = { path = "../ipc/rpc" }
 semver = "0.2"
 ethcore-ipc-nano = { path = "../ipc/nano" }
+ethcore-devtools = { path = "../devtools" }
+ethkey = { path = "../ethkey" }
 parking_lot = "0.3"
 
 [features]

sync/src/lib.rs

@@ -37,6 +37,9 @@ extern crate semver;
 extern crate parking_lot;
 extern crate rlp;
 
+#[cfg(test)] extern crate ethcore_devtools as devtools;
+#[cfg(test)] extern crate ethkey;
+
 #[macro_use]
 extern crate log;
 #[macro_use]

sync/src/tests/chain.rs

@@ -24,8 +24,8 @@ use SyncConfig;
 fn two_peers() {
 	::env_logger::init().ok();
 	let mut net = TestNet::new(3);
-	net.peer_mut(1).chain.add_blocks(1000, EachBlockWith::Uncle);
+	net.peer(1).chain.add_blocks(1000, EachBlockWith::Uncle);
-	net.peer_mut(2).chain.add_blocks(1000, EachBlockWith::Uncle);
+	net.peer(2).chain.add_blocks(1000, EachBlockWith::Uncle);
 	net.sync();
 	assert!(net.peer(0).chain.block(BlockID::Number(1000)).is_some());
 	assert_eq!(*net.peer(0).chain.blocks.read(), *net.peer(1).chain.blocks.read());
@@ -35,7 +35,7 @@ fn two_peers() {
 fn long_chain() {
 	::env_logger::init().ok();
 	let mut net = TestNet::new(2);
-	net.peer_mut(1).chain.add_blocks(50000, EachBlockWith::Nothing);
+	net.peer(1).chain.add_blocks(50000, EachBlockWith::Nothing);
 	net.sync();
 	assert!(net.peer(0).chain.block(BlockID::Number(50000)).is_some());
 	assert_eq!(*net.peer(0).chain.blocks.read(), *net.peer(1).chain.blocks.read());
@@ -45,8 +45,8 @@ fn long_chain() {
 fn status_after_sync() {
 	::env_logger::init().ok();
 	let mut net = TestNet::new(3);
-	net.peer_mut(1).chain.add_blocks(1000, EachBlockWith::Uncle);
+	net.peer(1).chain.add_blocks(1000, EachBlockWith::Uncle);
-	net.peer_mut(2).chain.add_blocks(1000, EachBlockWith::Uncle);
+	net.peer(2).chain.add_blocks(1000, EachBlockWith::Uncle);
 	net.sync();
 	let status = net.peer(0).sync.read().status();
 	assert_eq!(status.state, SyncState::Idle);
@@ -55,8 +55,8 @@ fn status_after_sync() {
 #[test]
 fn takes_few_steps() {
 	let mut net = TestNet::new(3);
-	net.peer_mut(1).chain.add_blocks(100, EachBlockWith::Uncle);
+	net.peer(1).chain.add_blocks(100, EachBlockWith::Uncle);
-	net.peer_mut(2).chain.add_blocks(100, EachBlockWith::Uncle);
+	net.peer(2).chain.add_blocks(100, EachBlockWith::Uncle);
 	let total_steps = net.sync();
 	assert!(total_steps < 20);
 }
@@ -67,8 +67,8 @@ fn empty_blocks() {
 	let mut net = TestNet::new(3);
 	for n in 0..200 {
 		let with = if n % 2 == 0 { EachBlockWith::Nothing } else { EachBlockWith::Uncle };
-		net.peer_mut(1).chain.add_blocks(5, with.clone());
+		net.peer(1).chain.add_blocks(5, with.clone());
-		net.peer_mut(2).chain.add_blocks(5, with);
+		net.peer(2).chain.add_blocks(5, with);
 	}
 	net.sync();
 	assert!(net.peer(0).chain.block(BlockID::Number(1000)).is_some());
@@ -79,14 +79,14 @@ fn empty_blocks() {
 fn forked() {
 	::env_logger::init().ok();
 	let mut net = TestNet::new(3);
-	net.peer_mut(0).chain.add_blocks(30, EachBlockWith::Uncle);
+	net.peer(0).chain.add_blocks(30, EachBlockWith::Uncle);
-	net.peer_mut(1).chain.add_blocks(30, EachBlockWith::Uncle);
+	net.peer(1).chain.add_blocks(30, EachBlockWith::Uncle);
-	net.peer_mut(2).chain.add_blocks(30, EachBlockWith::Uncle);
+	net.peer(2).chain.add_blocks(30, EachBlockWith::Uncle);
-	net.peer_mut(0).chain.add_blocks(10, EachBlockWith::Nothing); //fork
+	net.peer(0).chain.add_blocks(10, EachBlockWith::Nothing); //fork
-	net.peer_mut(1).chain.add_blocks(20, EachBlockWith::Uncle);
+	net.peer(1).chain.add_blocks(20, EachBlockWith::Uncle);
-	net.peer_mut(2).chain.add_blocks(20, EachBlockWith::Uncle);
+	net.peer(2).chain.add_blocks(20, EachBlockWith::Uncle);
-	net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Uncle); //fork between 1 and 2
+	net.peer(1).chain.add_blocks(10, EachBlockWith::Uncle); //fork between 1 and 2
-	net.peer_mut(2).chain.add_blocks(1, EachBlockWith::Nothing);
+	net.peer(2).chain.add_blocks(1, EachBlockWith::Nothing);
 	// peer 1 has the best chain of 601 blocks
 	let peer1_chain = net.peer(1).chain.numbers.read().clone();
 	net.sync();
@@ -102,12 +102,12 @@ fn forked_with_misbehaving_peer() {
 	let mut net = TestNet::new(3);
 	// peer 0 is on a totally different chain with higher total difficulty
 	net.peer_mut(0).chain = TestBlockChainClient::new_with_extra_data(b"fork".to_vec());
-	net.peer_mut(0).chain.add_blocks(50, EachBlockWith::Nothing);
+	net.peer(0).chain.add_blocks(50, EachBlockWith::Nothing);
-	net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Nothing);
+	net.peer(1).chain.add_blocks(10, EachBlockWith::Nothing);
-	net.peer_mut(2).chain.add_blocks(10, EachBlockWith::Nothing);
+	net.peer(2).chain.add_blocks(10, EachBlockWith::Nothing);
 
-	net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Nothing);
+	net.peer(1).chain.add_blocks(10, EachBlockWith::Nothing);
-	net.peer_mut(2).chain.add_blocks(20, EachBlockWith::Uncle);
+	net.peer(2).chain.add_blocks(20, EachBlockWith::Uncle);
 	// peer 1 should sync to peer 2, others should not change
 	let peer0_chain = net.peer(0).chain.numbers.read().clone();
 	let peer2_chain = net.peer(2).chain.numbers.read().clone();
@@ -124,13 +124,13 @@ fn net_hard_fork() {
 	ref_client.add_blocks(50, EachBlockWith::Uncle);
 	{
 		let mut net = TestNet::new_with_fork(2, Some((50, ref_client.block_hash(BlockID::Number(50)).unwrap())));
-		net.peer_mut(0).chain.add_blocks(100, EachBlockWith::Uncle);
+		net.peer(0).chain.add_blocks(100, EachBlockWith::Uncle);
 		net.sync();
 		assert_eq!(net.peer(1).chain.chain_info().best_block_number, 100);
 	}
 	{
 		let mut net = TestNet::new_with_fork(2, Some((50, ref_client.block_hash(BlockID::Number(50)).unwrap())));
-		net.peer_mut(0).chain.add_blocks(100, EachBlockWith::Nothing);
+		net.peer(0).chain.add_blocks(100, EachBlockWith::Nothing);
 		net.sync();
 		assert_eq!(net.peer(1).chain.chain_info().best_block_number, 0);
 	}
@@ -140,8 +140,8 @@ fn net_hard_fork() {
 fn restart() {
 	::env_logger::init().ok();
 	let mut net = TestNet::new(3);
-	net.peer_mut(1).chain.add_blocks(1000, EachBlockWith::Uncle);
+	net.peer(1).chain.add_blocks(1000, EachBlockWith::Uncle);
-	net.peer_mut(2).chain.add_blocks(1000, EachBlockWith::Uncle);
+	net.peer(2).chain.add_blocks(1000, EachBlockWith::Uncle);
 
 	net.sync();
 
@@ -166,37 +166,37 @@ fn status_empty() {
 #[test]
 fn status_packet() {
 	let mut net = TestNet::new(2);
-	net.peer_mut(0).chain.add_blocks(100, EachBlockWith::Uncle);
+	net.peer(0).chain.add_blocks(100, EachBlockWith::Uncle);
-	net.peer_mut(1).chain.add_blocks(1, EachBlockWith::Uncle);
+	net.peer(1).chain.add_blocks(1, EachBlockWith::Uncle);
 
 	net.start();
 
 	net.sync_step_peer(0);
 
-	assert_eq!(1, net.peer(0).queue.len());
+	assert_eq!(1, net.peer(0).queue.read().len());
-	assert_eq!(0x00, net.peer(0).queue[0].packet_id);
+	assert_eq!(0x00, net.peer(0).queue.read()[0].packet_id);
 }
 
 #[test]
 fn propagate_hashes() {
 	let mut net = TestNet::new(6);
-	net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Uncle);
+	net.peer(1).chain.add_blocks(10, EachBlockWith::Uncle);
 	net.sync();
 
-	net.peer_mut(0).chain.add_blocks(10, EachBlockWith::Uncle);
+	net.peer(0).chain.add_blocks(10, EachBlockWith::Uncle);
 	net.sync();
 	net.trigger_chain_new_blocks(0); //first event just sets the marker
 	net.trigger_chain_new_blocks(0);
 
 	// 5 peers with NewHahses, 4 with blocks
-	assert_eq!(9, net.peer(0).queue.len());
+	assert_eq!(9, net.peer(0).queue.read().len());
 	let mut hashes = 0;
 	let mut blocks = 0;
-	for i in 0..net.peer(0).queue.len() {
+	for i in 0..net.peer(0).queue.read().len() {
-		if net.peer(0).queue[i].packet_id == 0x1 {
+		if net.peer(0).queue.read()[i].packet_id == 0x1 {
 			hashes += 1;
 		}
-		if net.peer(0).queue[i].packet_id == 0x7 {
+		if net.peer(0).queue.read()[i].packet_id == 0x7 {
 			blocks += 1;
 		}
 	}
@@ -207,24 +207,24 @@ fn propagate_hashes() {
 #[test]
 fn propagate_blocks() {
 	let mut net = TestNet::new(20);
-	net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Uncle);
+	net.peer(1).chain.add_blocks(10, EachBlockWith::Uncle);
 	net.sync();
 
-	net.peer_mut(0).chain.add_blocks(10, EachBlockWith::Uncle);
+	net.peer(0).chain.add_blocks(10, EachBlockWith::Uncle);
 	net.trigger_chain_new_blocks(0); //first event just sets the marker
 	net.trigger_chain_new_blocks(0);
 
-	assert!(!net.peer(0).queue.is_empty());
+	assert!(!net.peer(0).queue.read().is_empty());
 	// NEW_BLOCK_PACKET
-	let blocks = net.peer(0).queue.iter().filter(|p| p.packet_id == 0x7).count();
+	let blocks = net.peer(0).queue.read().iter().filter(|p| p.packet_id == 0x7).count();
 	assert!(blocks > 0);
 }
 
 #[test]
 fn restart_on_malformed_block() {
 	let mut net = TestNet::new(2);
-	net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Uncle);
+	net.peer(1).chain.add_blocks(10, EachBlockWith::Uncle);
-	net.peer_mut(1).chain.corrupt_block(6);
+	net.peer(1).chain.corrupt_block(6);
 	net.sync_steps(20);
 
 	assert_eq!(net.peer(0).chain.chain_info().best_block_number, 5);
@@ -233,8 +233,8 @@ fn restart_on_malformed_block() {
 #[test]
 fn restart_on_broken_chain() {
 	let mut net = TestNet::new(2);
-	net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Uncle);
+	net.peer(1).chain.add_blocks(10, EachBlockWith::Uncle);
-	net.peer_mut(1).chain.corrupt_block_parent(6);
+	net.peer(1).chain.corrupt_block_parent(6);
 	net.sync_steps(20);
 
 	assert_eq!(net.peer(0).chain.chain_info().best_block_number, 5);
@@ -243,8 +243,8 @@ fn restart_on_broken_chain() {
 #[test]
 fn high_td_attach() {
 	let mut net = TestNet::new(2);
-	net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Uncle);
+	net.peer(1).chain.add_blocks(10, EachBlockWith::Uncle);
-	net.peer_mut(1).chain.corrupt_block_parent(6);
+	net.peer(1).chain.corrupt_block_parent(6);
 	net.sync_steps(20);
 
 	assert_eq!(net.peer(0).chain.chain_info().best_block_number, 5);
@@ -255,8 +255,8 @@ fn high_td_attach() {
 fn disconnect_on_unrelated_chain() {
 	::env_logger::init().ok();
 	let mut net = TestNet::new(2);
-	net.peer_mut(0).chain.add_blocks(200, EachBlockWith::Uncle);
+	net.peer(0).chain.add_blocks(200, EachBlockWith::Uncle);
-	net.peer_mut(1).chain.add_blocks(100, EachBlockWith::Nothing);
+	net.peer(1).chain.add_blocks(100, EachBlockWith::Nothing);
 	net.sync();
 	assert_eq!(net.disconnect_events, vec![(0, 0)]);
 }

sync/src/tests/consensus.rs

@@ -0,0 +1,78 @@
+// 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/>.
+
+use util::*;
+use ethcore::client::BlockChainClient;
+use ethcore::spec::Spec;
+use ethcore::miner::MinerService;
+use ethcore::transaction::*;
+use ethcore::account_provider::AccountProvider;
+use ethkey::KeyPair;
+use super::helpers::*;
+use SyncConfig;
+
+#[test]
+fn test_authority_round() {
+	::env_logger::init().ok();
+
+	let s1 = KeyPair::from_secret("1".sha3()).unwrap();
+	let s2 = KeyPair::from_secret("0".sha3()).unwrap();
+	let spec_factory = || {
+		let spec = Spec::new_test_round();
+		let account_provider = AccountProvider::transient_provider();
+		account_provider.insert_account(s1.secret().clone(), "").unwrap();
+		account_provider.insert_account(s2.secret().clone(), "").unwrap();
+		spec.engine.register_account_provider(Arc::new(account_provider));
+		spec
+	};
+	let mut net = TestNet::new_with_spec(2, SyncConfig::default(), spec_factory);
+	let mut net = &mut *net;
+	// Push transaction to both clients. Only one of them gets lucky to mine a block.
+	net.peer(0).chain.miner().set_author(s1.address());
+	net.peer(0).chain.engine().set_signer(s1.address(), "".to_owned());
+	net.peer(1).chain.miner().set_author(s2.address());
+	net.peer(1).chain.engine().set_signer(s2.address(), "".to_owned());
+	let tx1 = Transaction {
+		nonce: 0.into(),
+		gas_price: 0.into(),
+		gas: 21000.into(),
+		action: Action::Call(Address::default()),
+		value: 0.into(),
+		data: Vec::new(),
+	}.sign(s1.secret(), None);
+	// exhange statuses
+	net.sync_steps(5);
+	net.peer(0).chain.miner().import_own_transaction(&net.peer(0).chain, tx1).unwrap();
+	net.sync();
+	assert_eq!(net.peer(0).chain.chain_info().best_block_number, 1);
+	assert_eq!(net.peer(1).chain.chain_info().best_block_number, 1);
+
+	let tx2 = Transaction {
+		nonce: 0.into(),
+		gas_price: 0.into(),
+		gas: 21000.into(),
+		action: Action::Call(Address::default()),
+		value: 0.into(),
+		data: Vec::new(),
+	}.sign(s2.secret(), None);
+	net.peer(1).chain.miner().import_own_transaction(&net.peer(1).chain, tx2).unwrap();
+	net.peer(1).chain.engine().step();
+	net.peer(1).chain.miner().update_sealing(&net.peer(1).chain);
+	net.sync();
+	assert_eq!(net.peer(0).chain.chain_info().best_block_number, 2);
+	assert_eq!(net.peer(1).chain.chain_info().best_block_number, 2);
+}
+

sync/src/tests/helpers.rs

@@ -17,16 +17,33 @@
 use util::*;
 use network::*;
 use tests::snapshot::*;
-use ethcore::client::{TestBlockChainClient, BlockChainClient};
+use ethcore::client::{TestBlockChainClient, BlockChainClient, Client as EthcoreClient, ClientConfig, ChainNotify};
 use ethcore::header::BlockNumber;
 use ethcore::snapshot::SnapshotService;
+use ethcore::spec::Spec;
+use ethcore::miner::Miner;
+use ethcore::db::NUM_COLUMNS;
 use sync_io::SyncIo;
+use io::IoChannel;
 use api::WARP_SYNC_PROTOCOL_ID;
 use chain::ChainSync;
 use ::SyncConfig;
+use devtools::{self, GuardedTempResult};
 
-pub struct TestIo<'p> {
+pub trait FlushingBlockChainClient: BlockChainClient {
-	pub chain: &'p mut TestBlockChainClient,
+	fn flush(&self) {}
+}
+
+impl FlushingBlockChainClient for EthcoreClient {
+	fn flush(&self) {
+		self.flush_queue();
+	}
+}
+
+impl FlushingBlockChainClient for TestBlockChainClient {}
+
+pub struct TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
+	pub chain: &'p C,
 	pub snapshot_service: &'p TestSnapshotService,
 	pub queue: &'p mut VecDeque<TestPacket>,
 	pub sender: Option<PeerId>,
@@ -34,8 +51,8 @@ pub struct TestIo<'p> {
 	overlay: RwLock<HashMap<BlockNumber, Bytes>>,
 }
 
-impl<'p> TestIo<'p> {
+impl<'p, C> TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
-	pub fn new(chain: &'p mut TestBlockChainClient, ss: &'p TestSnapshotService, queue: &'p mut VecDeque<TestPacket>, sender: Option<PeerId>) -> TestIo<'p> {
+	pub fn new(chain: &'p C, ss: &'p TestSnapshotService, queue: &'p mut VecDeque<TestPacket>, sender: Option<PeerId>) -> TestIo<'p, C> {
 		TestIo {
 			chain: chain,
 			snapshot_service: ss,
@@ -47,7 +64,7 @@ impl<'p> TestIo<'p> {
 	}
 }
 
-impl<'p> SyncIo for TestIo<'p> {
+impl<'p, C> SyncIo for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
 	fn disable_peer(&mut self, peer_id: PeerId) {
 		self.disconnect_peer(peer_id);
 	}
@@ -99,7 +116,7 @@ impl<'p> SyncIo for TestIo<'p> {
 	}
 
 	fn protocol_version(&self, protocol: &ProtocolId, peer_id: PeerId) -> u8 {
-		if protocol == &WARP_SYNC_PROTOCOL_ID { 1 } else { self.eth_protocol_version(peer_id) }
+		if protocol == &WARP_SYNC_PROTOCOL_ID { 2 } else { self.eth_protocol_version(peer_id) }
 	}
 
 	fn chain_overlay(&self) -> &RwLock<HashMap<BlockNumber, Bytes>> {
@@ -113,31 +130,31 @@ pub struct TestPacket {
 	pub recipient: PeerId,
 }
 
-pub struct TestPeer {
+pub struct TestPeer<C> where C: FlushingBlockChainClient {
-	pub chain: TestBlockChainClient,
+	pub chain: C,
 	pub snapshot_service: Arc<TestSnapshotService>,
 	pub sync: RwLock<ChainSync>,
-	pub queue: VecDeque<TestPacket>,
+	pub queue: RwLock<VecDeque<TestPacket>>,
 }
 
-pub struct TestNet {
+pub struct TestNet<C> where C: FlushingBlockChainClient {
-	pub peers: Vec<TestPeer>,
+	pub peers: Vec<Arc<TestPeer<C>>>,
 	pub started: bool,
 	pub disconnect_events: Vec<(PeerId, PeerId)>, //disconnected (initiated by, to)
 }
 
-impl TestNet {
+impl TestNet<TestBlockChainClient> {
-	pub fn new(n: usize) -> TestNet {
+	pub fn new(n: usize) -> TestNet<TestBlockChainClient> {
 		Self::new_with_config(n, SyncConfig::default())
 	}
 
-	pub fn new_with_fork(n: usize, fork: Option<(BlockNumber, H256)>) -> TestNet {
+	pub fn new_with_fork(n: usize, fork: Option<(BlockNumber, H256)>) -> TestNet<TestBlockChainClient> {
 		let mut config = SyncConfig::default();
 		config.fork_block = fork;
 		Self::new_with_config(n, config)
 	}
 
-	pub fn new_with_config(n: usize, config: SyncConfig) -> TestNet {
+	pub fn new_with_config(n: usize, config: SyncConfig) -> TestNet<TestBlockChainClient> {
 		let mut net = TestNet {
 			peers: Vec::new(),
 			started: false,
@@ -147,31 +164,77 @@ impl TestNet {
 			let chain = TestBlockChainClient::new();
 			let ss = Arc::new(TestSnapshotService::new());
 			let sync = ChainSync::new(config.clone(), &chain);
-			net.peers.push(TestPeer {
+			net.peers.push(Arc::new(TestPeer {
 				sync: RwLock::new(sync),
 				snapshot_service: ss,
 				chain: chain,
-				queue: VecDeque::new(),
+				queue: RwLock::new(VecDeque::new()),
-			});
+			}));
 		}
 		net
 	}
+}
 
-	pub fn peer(&self, i: usize) -> &TestPeer {
+impl TestNet<EthcoreClient> {
+	pub fn new_with_spec<F>(n: usize, config: SyncConfig, spec_factory: F) -> GuardedTempResult<TestNet<EthcoreClient>>
+		where F: Fn() -> Spec
+	{
+		let mut net = TestNet {
+			peers: Vec::new(),
+			started: false,
+			disconnect_events: Vec::new(),
+		};
+		let dir = devtools::RandomTempPath::new();
+		for _ in 0..n {
+			let mut client_dir = dir.as_path().clone();
+			client_dir.push(devtools::random_filename());
+
+			let db_config = DatabaseConfig::with_columns(NUM_COLUMNS);
+
+			let spec = spec_factory();
+			let client = Arc::try_unwrap(EthcoreClient::new(
+				ClientConfig::default(),
+				&spec,
+				client_dir.as_path(),
+				Arc::new(Miner::with_spec(&spec)),
+				IoChannel::disconnected(),
+				&db_config
+			).unwrap()).ok().unwrap();
+
+			let ss = Arc::new(TestSnapshotService::new());
+			let sync = ChainSync::new(config.clone(), &client);
+			let peer = Arc::new(TestPeer {
+				sync: RwLock::new(sync),
+				snapshot_service: ss,
+				chain: client,
+				queue: RwLock::new(VecDeque::new()),
+			});
+			peer.chain.add_notify(peer.clone());
+			net.peers.push(peer);
+		}
+		GuardedTempResult::<TestNet<EthcoreClient>> {
+			_temp: dir,
+			result: Some(net)
+		}
+	}
+}
+
+impl<C> TestNet<C> where C: FlushingBlockChainClient {
+	pub fn peer(&self, i: usize) -> &TestPeer<C> {
 		&self.peers[i]
 	}
 
-	pub fn peer_mut(&mut self, i: usize) -> &mut TestPeer {
+	pub fn peer_mut(&mut self, i: usize) -> &mut TestPeer<C> {
-		&mut self.peers[i]
+		Arc::get_mut(&mut self.peers[i]).unwrap()
 	}
 
 	pub fn start(&mut self) {
 		for peer in 0..self.peers.len() {
 			for client in 0..self.peers.len() {
 				if peer != client {
-					let mut p = &mut self.peers[peer];
+					let p = &self.peers[peer];
 					p.sync.write().update_targets(&p.chain);
-					p.sync.write().on_peer_connected(&mut TestIo::new(&mut p.chain, &p.snapshot_service, &mut p.queue, Some(client as PeerId)), client as PeerId);
+					p.sync.write().on_peer_connected(&mut TestIo::new(&p.chain, &p.snapshot_service, &mut p.queue.write(), Some(client as PeerId)), client as PeerId);
 				}
 			}
 		}
@@ -179,18 +242,20 @@ impl TestNet {
 
 	pub fn sync_step(&mut self) {
 		for peer in 0..self.peers.len() {
-			if let Some(packet) = self.peers[peer].queue.pop_front() {
+			let packet = self.peers[peer].queue.write().pop_front();
+			if let Some(packet) = packet {
 				let disconnecting = {
-					let mut p = &mut self.peers[packet.recipient];
+					let p = &self.peers[packet.recipient];
+					let mut queue = p.queue.write();
 					trace!("--- {} -> {} ---", peer, packet.recipient);
 					let to_disconnect = {
-						let mut io = TestIo::new(&mut p.chain, &p.snapshot_service, &mut p.queue, Some(peer as PeerId));
+						let mut io = TestIo::new(&p.chain, &p.snapshot_service, &mut queue, Some(peer as PeerId));
 						ChainSync::dispatch_packet(&p.sync, &mut io, peer as PeerId, packet.packet_id, &packet.data);
 						io.to_disconnect
 					};
 					for d in &to_disconnect {
 						// notify this that disconnecting peers are disconnecting
-						let mut io = TestIo::new(&mut p.chain, &p.snapshot_service, &mut p.queue, Some(*d));
+						let mut io = TestIo::new(&p.chain, &p.snapshot_service, &mut queue, Some(*d));
 						p.sync.write().on_peer_aborting(&mut io, *d);
 						self.disconnect_events.push((peer, *d));
 					}
@@ -198,8 +263,9 @@ impl TestNet {
 				};
 				for d in &disconnecting {
 					// notify other peers that this peer is disconnecting
-					let mut p = &mut self.peers[*d];
+					let p = &self.peers[*d];
-					let mut io = TestIo::new(&mut p.chain, &p.snapshot_service, &mut p.queue, Some(peer as PeerId));
+					let mut queue = p.queue.write();
+					let mut io = TestIo::new(&p.chain, &p.snapshot_service, &mut queue, Some(peer as PeerId));
 					p.sync.write().on_peer_aborting(&mut io, peer as PeerId);
 				}
 			}
@@ -209,13 +275,17 @@ impl TestNet {
 	}
 
 	pub fn sync_step_peer(&mut self, peer_num: usize) {
-		let mut peer = self.peer_mut(peer_num);
+		let peer = self.peer(peer_num);
-		peer.sync.write().maintain_sync(&mut TestIo::new(&mut peer.chain, &peer.snapshot_service, &mut peer.queue, None));
+		peer.chain.flush();
+		let mut queue = peer.queue.write();
+		peer.sync.write().maintain_peers(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut queue, None));
+		peer.sync.write().maintain_sync(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut queue, None));
+		peer.sync.write().propagate_new_transactions(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut queue, None));
 	}
 
 	pub fn restart_peer(&mut self, i: usize) {
-		let peer = self.peer_mut(i);
+		let peer = self.peer(i);
-		peer.sync.write().restart(&mut TestIo::new(&mut peer.chain, &peer.snapshot_service, &mut peer.queue, None));
+		peer.sync.write().restart(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut peer.queue.write(), None));
 	}
 
 	pub fn sync(&mut self) -> u32 {
@@ -239,11 +309,46 @@ impl TestNet {
 	}
 
 	pub fn done(&self) -> bool {
-		self.peers.iter().all(|p| p.queue.is_empty())
+		self.peers.iter().all(|p| p.queue.read().is_empty())
 	}
 
 	pub fn trigger_chain_new_blocks(&mut self, peer_id: usize) {
-		let mut peer = self.peer_mut(peer_id);
+		let peer = self.peer(peer_id);
-		peer.sync.write().chain_new_blocks(&mut TestIo::new(&mut peer.chain, &peer.snapshot_service, &mut peer.queue, None), &[], &[], &[], &[], &[], &[]);
+		let mut queue = peer.queue.write();
+		peer.sync.write().chain_new_blocks(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut queue, None), &[], &[], &[], &[], &[], &[]);
 	}
 }
+
+impl ChainNotify for TestPeer<EthcoreClient> {
+	fn new_blocks(&self,
+		imported: Vec<H256>,
+		invalid: Vec<H256>,
+		enacted: Vec<H256>,
+		retracted: Vec<H256>,
+		sealed: Vec<H256>,
+		proposed: Vec<Bytes>,
+		_duration: u64)
+	{
+		let mut queue = self.queue.write();
+		let mut io = TestIo::new(&self.chain, &self.snapshot_service, &mut queue, None);
+		self.sync.write().chain_new_blocks(
+			&mut io,
+			&imported,
+			&invalid,
+			&enacted,
+			&retracted,
+			&sealed,
+			&proposed);
+	}
+
+	fn start(&self) {}
+
+	fn stop(&self) {}
+
+	fn broadcast(&self, message: Vec<u8>) {
+		let mut queue = self.queue.write();
+		let mut io = TestIo::new(&self.chain, &self.snapshot_service, &mut queue, None);
+		self.sync.write().propagate_consensus_packet(&mut io, message.clone());
+	}
+}
+

sync/src/tests/mod.rs

@@ -17,4 +17,5 @@
 pub mod helpers;
 pub mod snapshot;
 mod chain;
+mod consensus;
 mod rpc;

sync/src/tests/snapshot.rs

@@ -129,7 +129,7 @@ fn snapshot_sync() {
 	let snapshot_service = Arc::new(TestSnapshotService::new_with_snapshot(16, H256::new(), 500000));
 	for i in 0..4 {
 		net.peer_mut(i).snapshot_service = snapshot_service.clone();
-		net.peer_mut(i).chain.add_blocks(1, EachBlockWith::Nothing);
+		net.peer(i).chain.add_blocks(1, EachBlockWith::Nothing);
 	}
 	net.sync_steps(50);
 	assert_eq!(net.peer(4).snapshot_service.state_restoration_chunks.lock().len(), net.peer(0).snapshot_service.manifest.as_ref().unwrap().state_hashes.len());