Merge pull request #119 from gavofyork/client

Client state syncing

res/ethereum/frontier.json

@@ -1,4 +1,5 @@
 {
+	"name": "Frontier",
 	"engineName": "Ethash",
 	"params": {
 		"accountStartNonce": "0x00",

res/ethereum/frontier_test.json

@@ -1,4 +1,5 @@
 {
+	"engineName": "Frontier (Test)",
 	"engineName": "Ethash",
 	"params": {
 		"accountStartNonce": "0x00",
@@ -30,4 +31,4 @@
 		"0000000000000000000000000000000000000003": { "builtin": { "name": "ripemd160", "linear": { "base": 600, "word": 120 } } },
 		"0000000000000000000000000000000000000004": { "builtin": { "name": "identity", "linear": { "base": 15, "word": 3 } } }
 	}
-}
+}

res/ethereum/homestead_test.json

@@ -1,4 +1,5 @@
 {
+	"name": "Homestead (Test)",
 	"engineName": "Ethash",
 	"params": {
 		"accountStartNonce": "0x00",

res/ethereum/morden.json

@@ -1,4 +1,5 @@
 {
+	"name": "Morden",
 	"engineName": "Ethash",
 	"params": {
 		"accountStartNonce": "0x0100000",
@@ -31,4 +32,4 @@
 		"0000000000000000000000000000000000000004": { "balance": "1", "nonce": "1048576", "builtin": { "name": "identity", "linear": { "base": 15, "word": 3 } } },
 		"102e61f5d8f9bc71d0ad4a084df4e65e05ce0e1c": { "balance": "1606938044258990275541962092341162602522202993782792835301376", "nonce": "1048576" }
 	}
-}
+}

res/ethereum/olympic.json

@@ -1,4 +1,5 @@
 {
+	"name": "Olympic",
 	"engineName": "Ethash",
 	"params": {
 		"accountStartNonce": "0x00",
@@ -38,4 +39,4 @@
 		"6c386a4b26f73c802f34673f7248bb118f97424a": { "balance": "1606938044258990275541962092341162602522202993782792835301376" },
 		"e4157b34ea9615cfbde6b4fda419828124b70c78": { "balance": "1606938044258990275541962092341162602522202993782792835301376" }
 	}
-}
+}

res/null_morden.json

@@ -1,4 +1,5 @@
 {
+	"name": "Morden",
 	"engineName": "NullEngine",
 	"params": {
 		"accountStartNonce": "0x0100000",
@@ -31,4 +32,4 @@
 		"0000000000000000000000000000000000000004": { "balance": "1", "nonce": "1048576", "builtin": { "name": "identity", "linear": { "base": 15, "word": 3 } } },
 		"102e61f5d8f9bc71d0ad4a084df4e65e05ce0e1c": { "balance": "1606938044258990275541962092341162602522202993782792835301376", "nonce": "1048576" }
 	}
-}
+}

src/bin/client.rs

@@ -1,26 +1,32 @@
 extern crate ethcore_util as util;
 extern crate ethcore;
 extern crate rustc_serialize;
+extern crate log;
 extern crate env_logger;
 
 use std::io::*;
 use std::env;
-use std::sync::Arc;
+use log::{LogLevelFilter};
+use env_logger::LogBuilder;
 use util::hash::*;
-use util::network::{NetworkService};
+use ethcore::service::ClientService;
-use ethcore::client::Client;
-use ethcore::sync::EthSync;
 use ethcore::ethereum;
 
+fn setup_log() {
+	let mut builder = LogBuilder::new();
+	builder.filter(None, LogLevelFilter::Info);
+
+	if env::var("RUST_LOG").is_ok() {
+		builder.parse(&env::var("RUST_LOG").unwrap());
+	}
+
+	builder.init().unwrap();
+}
+
 fn main() {
-	::env_logger::init().ok();
+	setup_log();
-	let mut service = NetworkService::start().unwrap();
-	//TODO: replace with proper genesis and chain params.
 	let spec = ethereum::new_frontier();
-	let mut dir = env::temp_dir();
+	let mut _service = ClientService::start(spec).unwrap();
-	dir.push(H32::random().hex());
-	let client = Arc::new(Client::new(spec, &dir).unwrap());
-	EthSync::register(&mut service, client);
 	loop {
 		let mut cmd = String::new();
 		stdin().read_line(&mut cmd).unwrap();

src/block.rs

@@ -229,6 +229,9 @@ impl<'x, 'y> ClosedBlock<'x, 'y> {
 
 	/// Turn this back into an `OpenBlock`.
 	pub fn reopen(self) -> OpenBlock<'x, 'y> { self.open_block }
+
+	/// Drop this object and return the underlieing database.
+	pub fn drain(self) -> OverlayDB { self.open_block.block.state.drop().1 }
 }
 
 impl SealedBlock {
@@ -251,15 +254,20 @@ impl IsBlock for SealedBlock {
 }
 
 /// Enact the block given by `block_bytes` using `engine` on the database `db` with given `parent` block header
-/// 
+pub fn enact<'x, 'y>(block_bytes: &[u8], engine: &'x Engine, db: OverlayDB, parent: &Header, last_hashes: &'y LastHashes) -> Result<ClosedBlock<'x, 'y>, Error> {
-pub fn enact(block_bytes: &[u8], engine: &Engine, db: OverlayDB, parent: &Header, last_hashes: &LastHashes) -> Result<SealedBlock, Error> {
 	let block = BlockView::new(block_bytes);
 	let header = block.header_view();
 	let mut b = OpenBlock::new(engine, db, parent, last_hashes, header.author(), header.extra_data());
 	b.set_timestamp(header.timestamp());
 	for t in block.transactions().into_iter() { try!(b.push_transaction(t, None)); }
 	for u in block.uncles().into_iter() { try!(b.push_uncle(u)); }
-	Ok(try!(b.close().seal(header.seal())))
+	Ok(b.close())
+}
+
+/// Enact the block given by `block_bytes` using `engine` on the database `db` with given `parent` block header. Seal the block aferwards
+pub fn enact_and_seal(block_bytes: &[u8], engine: &Engine, db: OverlayDB, parent: &Header, last_hashes: &LastHashes) -> Result<SealedBlock, Error> {
+	let header = BlockView::new(block_bytes).header_view();
+	Ok(try!(try!(enact(block_bytes, engine, db, parent, last_hashes)).seal(header.seal())))
 }
 
 #[test]
@@ -289,7 +297,7 @@ fn enact_block() {
 
 	let mut db = OverlayDB::new_temp();
 	engine.spec().ensure_db_good(&mut db);
-	let e = enact(&orig_bytes, engine.deref(), db, &genesis_header, &vec![genesis_header.hash()]).unwrap();
+	let e = enact_and_seal(&orig_bytes, engine.deref(), db, &genesis_header, &vec![genesis_header.hash()]).unwrap();
 
 	assert_eq!(e.rlp_bytes(), orig_bytes);
 	

src/client.rs

@@ -1,4 +1,5 @@
 use util::*;
+use rocksdb::{DB};
 use blockchain::{BlockChain, BlockProvider};
 use views::BlockView;
 use error::*;
@@ -6,6 +7,10 @@ use header::BlockNumber;
 use spec::Spec;
 use engine::Engine;
 use queue::BlockQueue;
+use sync::NetSyncMessage;
+use env_info::LastHashes;
+use verification::*;
+use block::*;
 
 /// General block status
 pub enum BlockStatus {
@@ -41,7 +46,7 @@ pub struct BlockQueueStatus {
 pub type TreeRoute = ::blockchain::TreeRoute;
 
 /// Blockchain database client. Owns and manages a blockchain and a block queue.
-pub trait BlockChainClient : Sync {
+pub trait BlockChainClient : Sync + Send {
 	/// Get raw block header data by block header hash.
 	fn block_header(&self, hash: &H256) -> Option<Bytes>;
 
@@ -94,20 +99,86 @@ pub trait BlockChainClient : Sync {
 /// Blockchain database client backed by a persistent database. Owns and manages a blockchain and a block queue.
 pub struct Client {
 	chain: Arc<RwLock<BlockChain>>,
-	_engine: Arc<Box<Engine>>,
+	engine: Arc<Box<Engine>>,
+	state_db: OverlayDB,
 	queue: BlockQueue,
 }
 
 impl Client {
-	pub fn new(spec: Spec, path: &Path) -> Result<Client, Error> {
+	/// Create a new client with given spec and DB path.
+	pub fn new(spec: Spec, path: &Path, message_channel: IoChannel<NetSyncMessage> ) -> Result<Client, Error> {
 		let chain = Arc::new(RwLock::new(BlockChain::new(&spec.genesis_block(), path)));
 		let engine = Arc::new(try!(spec.to_engine()));
+		let mut state_path = path.to_path_buf();
+		state_path.push("state");
+		let db = DB::open_default(state_path.to_str().unwrap()).unwrap();
+		let mut state_db = OverlayDB::new(db);
+		engine.spec().ensure_db_good(&mut state_db);
+		state_db.commit().expect("Error commiting genesis state to state DB");
+
 		Ok(Client {
 			chain: chain.clone(),
-			_engine: engine.clone(),
+			engine: engine.clone(),
-			queue: BlockQueue::new(chain.clone(), engine.clone()),
+			state_db: state_db,
+			queue: BlockQueue::new(engine.clone(), message_channel),
 		})
 	}
+
+	/// This is triggered by a message coming from a block queue when the block is ready for insertion
+	pub fn import_verified_block(&mut self, bytes: Bytes) {
+		let block = BlockView::new(&bytes);
+		let header = block.header();
+		if let Err(e) = verify_block_family(&header, &bytes, self.engine.deref().deref(), self.chain.read().unwrap().deref()) {
+			warn!(target: "client", "Stage 3 block verification failed for #{} ({})\nError: {:?}", header.number(), header.hash(), e);
+			self.queue.mark_as_bad(&header.hash());
+			return;
+		};
+		let parent = match self.chain.read().unwrap().block_header(&header.parent_hash) {
+			Some(p) => p,
+			None => {
+				warn!(target: "client", "Block import failed for #{} ({}): Parent not found ({}) ", header.number(), header.hash(), header.parent_hash);
+				self.queue.mark_as_bad(&header.hash());
+				return;
+			},
+		};
+		// build last hashes
+		let mut last = self.chain.read().unwrap().best_block_hash();
+		let mut last_hashes = LastHashes::new();
+		last_hashes.resize(256, H256::new());
+		for i in 0..255 {
+			match self.chain.read().unwrap().block_details(&last) {
+				Some(details) => {
+					last_hashes[i + 1] = details.parent.clone();
+					last = details.parent.clone();
+				},
+				None => break,
+			}
+		}
+
+		let result = match enact(&bytes, self.engine.deref().deref(), self.state_db.clone(), &parent, &last_hashes) {
+			Ok(b) => b,
+			Err(e) => {
+				warn!(target: "client", "Block import failed for #{} ({})\nError: {:?}", header.number(), header.hash(), e);
+				self.queue.mark_as_bad(&header.hash());
+				return;
+			}
+		};
+		if let Err(e) = verify_block_final(&header, result.block().header()) {
+			warn!(target: "client", "Stage 4 block verification failed for #{} ({})\nError: {:?}", header.number(), header.hash(), e);
+			self.queue.mark_as_bad(&header.hash());
+			return;
+		}
+
+		self.chain.write().unwrap().insert_block(&bytes); //TODO: err here?
+		match result.drain().commit() {
+			Ok(_) => (),
+			Err(e) => {
+				warn!(target: "client", "State DB commit failed: {:?}", e);
+				return;
+			}
+		}
+		info!(target: "client", "Imported #{} ({})", header.number(), header.hash());
+	}
 }
 
 impl BlockChainClient for Client {
@@ -165,6 +236,10 @@ impl BlockChainClient for Client {
 	}
 
 	fn import_block(&mut self, bytes: &[u8]) -> ImportResult {
+		let header = BlockView::new(bytes).header();
+		if self.chain.read().unwrap().is_known(&header.hash()) {
+			return Err(ImportError::AlreadyInChain);
+		}
 		self.queue.import_block(bytes)
 	}
 

src/engine.rs

@@ -45,7 +45,7 @@ pub trait Engine : Sync + Send {
 
 	/// Phase 3 verification. Check block information against parent and uncles. `block` (the header's full block) 
 	/// may be provided for additional checks. Returns either a null `Ok` or a general error detailing the problem with import.
-	fn verify_block_final(&self, _header: &Header, _parent: &Header, _block: Option<&[u8]>) -> Result<(), Error> { Ok(()) }
+	fn verify_block_family(&self, _header: &Header, _parent: &Header, _block: Option<&[u8]>) -> Result<(), Error> { Ok(()) }
 
 	/// Additional verification for transactions in blocks.
 	// TODO: Add flags for which bits of the transaction to check.

src/error.rs

@@ -2,6 +2,7 @@
 
 use util::*;
 use header::BlockNumber;
+use basic_types::LogBloom;
 
 #[derive(Debug, PartialEq, Eq)]
 pub struct Mismatch<T: fmt::Debug> {
@@ -53,15 +54,15 @@ pub enum BlockError {
 	UncleIsBrother(OutOfBounds<BlockNumber>),
 	UncleInChain(H256),
 	UncleParentNotInChain(H256),
-	InvalidStateRoot,
+	InvalidStateRoot(Mismatch<H256>),
-	InvalidGasUsed,
+	InvalidGasUsed(Mismatch<U256>),
 	InvalidTransactionsRoot(Mismatch<H256>),
 	InvalidDifficulty(Mismatch<U256>),
 	InvalidGasLimit(OutOfBounds<U256>),
-	InvalidReceiptsStateRoot,
+	InvalidReceiptsStateRoot(Mismatch<H256>),
 	InvalidTimestamp(OutOfBounds<u64>),
-	InvalidLogBloom,
+	InvalidLogBloom(Mismatch<LogBloom>),
-	InvalidBlockNonce,
+	InvalidBlockNonce(Mismatch<H256>),
 	InvalidParentHash(Mismatch<H256>),
 	InvalidNumber(OutOfBounds<BlockNumber>),
 	UnknownParent(H256),
@@ -70,14 +71,14 @@ pub enum BlockError {
 
 #[derive(Debug)]
 pub enum ImportError {
-	Bad(Error),
+	Bad(Option<Error>),
 	AlreadyInChain,
 	AlreadyQueued,
 }
 
 impl From<Error> for ImportError {
 	fn from(err: Error) -> ImportError {
-		ImportError::Bad(err)
+		ImportError::Bad(Some(err))
 	}
 }
 
@@ -124,6 +125,18 @@ impl From<DecoderError> for Error {
 	}
 }
 
+impl From<UtilError> for Error {
+	fn from(err: UtilError) -> Error {
+		Error::Util(err)
+	}
+}
+
+impl From<IoError> for Error {
+	fn from(err: IoError) -> Error {
+		Error::Util(From::from(err))
+	}
+}
+
 // TODO: uncomment below once https://github.com/rust-lang/rust/issues/27336 sorted.
 /*#![feature(concat_idents)]
 macro_rules! assimilate {

src/ethereum/ethash.rs

@@ -57,8 +57,9 @@ impl Engine for Ethash {
 		// Bestow uncle rewards
 		let current_number = fields.header.number();
 		for u in fields.uncles.iter() {
-			fields.state.add_balance(u.author(), &(reward * U256::from((8 + u.number() - current_number) / 8)));
+			fields.state.add_balance(u.author(), &(reward * U256::from(8 + u.number() - current_number) / U256::from(8)));
 		}
+		fields.state.commit();
 	}
 
 	fn verify_block_basic(&self, header: &Header,  _block: Option<&[u8]>) -> result::Result<(), Error> {
@@ -75,7 +76,7 @@ impl Engine for Ethash {
 		Ok(())
 	}
 
-	fn verify_block_final(&self, header: &Header, parent: &Header, _block: Option<&[u8]>) -> result::Result<(), Error> {
+	fn verify_block_family(&self, header: &Header, parent: &Header, _block: Option<&[u8]>) -> result::Result<(), Error> {
 		// Check difficulty is correct given the two timestamps.
 		let expected_difficulty = self.calculate_difficuty(header, parent);
 		if header.difficulty != expected_difficulty {
@@ -143,4 +144,23 @@ fn on_close_block() {
 	assert_eq!(b.state().balance(&Address::zero()), U256::from_str("4563918244f40000").unwrap());
 }
 
+#[test]
+fn on_close_block_with_uncle() {
+	use super::*;
+	let engine = new_morden().to_engine().unwrap();
+	let genesis_header = engine.spec().genesis_header();
+	let mut db = OverlayDB::new_temp();
+	engine.spec().ensure_db_good(&mut db);
+	let last_hashes = vec![genesis_header.hash()];
+	let mut b = OpenBlock::new(engine.deref(), db, &genesis_header, &last_hashes, Address::zero(), vec![]);
+	let mut uncle = Header::new();
+	let uncle_author = address_from_hex("ef2d6d194084c2de36e0dabfce45d046b37d1106");
+	uncle.author = uncle_author.clone();
+	b.push_uncle(uncle).unwrap();
+	
+	let b = b.close();
+	assert_eq!(b.state().balance(&Address::zero()), U256::from_str("478eae0e571ba000").unwrap());
+	assert_eq!(b.state().balance(&uncle_author), U256::from_str("3cb71f51fc558000").unwrap());
+}
+
 // TODO: difficulty test

src/lib.rs

@@ -111,6 +111,7 @@ pub mod blockchain;
 pub mod extras;
 pub mod substate;
 pub mod evm;
+pub mod service;
 pub mod executive;
 pub mod externalities;
 

src/queue.rs

@@ -1,23 +1,25 @@
 use util::*;
-use blockchain::*;
-use views::{BlockView};
 use verification::*;
 use error::*;
 use engine::Engine;
+use sync::*;
+use views::*;
 
 /// A queue of blocks. Sits between network or other I/O and the BlockChain.
 /// Sorts them ready for blockchain insertion.
 pub struct BlockQueue {
-	bc: Arc<RwLock<BlockChain>>,
 	engine: Arc<Box<Engine>>,
+	message_channel: IoChannel<NetSyncMessage>,
+	bad: HashSet<H256>,
 }
 
 impl BlockQueue {
 	/// Creates a new queue instance.
-	pub fn new(bc: Arc<RwLock<BlockChain>>, engine: Arc<Box<Engine>>) -> BlockQueue {
+	pub fn new(engine: Arc<Box<Engine>>, message_channel: IoChannel<NetSyncMessage>) -> BlockQueue {
 		BlockQueue {
-			bc: bc,
 			engine: engine,
+			message_channel: message_channel,
+			bad: HashSet::new(),
 		}
 	}
 
@@ -28,14 +30,30 @@ impl BlockQueue {
 	/// Add a block to the queue.
 	pub fn import_block(&mut self, bytes: &[u8]) -> ImportResult {
 		let header = BlockView::new(bytes).header();
-		if self.bc.read().unwrap().is_known(&header.hash()) {
+		if self.bad.contains(&header.hash()) {
-			return Err(ImportError::AlreadyInChain);
+			return Err(ImportError::Bad(None));
 		}
-		try!(verify_block_basic(bytes, self.engine.deref().deref()));
+
-		try!(verify_block_unordered(bytes, self.engine.deref().deref()));
+		if self.bad.contains(&header.parent_hash) {
-		try!(verify_block_final(bytes, self.engine.deref().deref(), self.bc.read().unwrap().deref()));
+			self.bad.insert(header.hash());
-		self.bc.write().unwrap().insert_block(bytes);
+			return Err(ImportError::Bad(None));
+		}
+
+		try!(verify_block_basic(&header, bytes, self.engine.deref().deref()).map_err(|e| {
+			warn!(target: "client", "Stage 1 block verification failed for {}\nError: {:?}", BlockView::new(&bytes).header_view().sha3(), e);
+			e
+		}));
+		try!(verify_block_unordered(&header, bytes, self.engine.deref().deref()).map_err(|e| {
+			warn!(target: "client", "Stage 2 block verification failed for {}\nError: {:?}", BlockView::new(&bytes).header_view().sha3(), e);
+			e
+		}));
+		try!(self.message_channel.send(UserMessage(SyncMessage::BlockVerified(bytes.to_vec()))).map_err(|e| Error::from(e)));
 		Ok(())
 	}
+
+	pub fn mark_as_bad(&mut self, hash: &H256) {
+		self.bad.insert(hash.clone());
+		//TODO: walk the queue
+	}
 }
 

src/receipt.rs

@@ -3,6 +3,7 @@ use basic_types::LogBloom;
 use log_entry::LogEntry;
 
 /// Information describing execution of a transaction.
+#[derive(Debug)]
 pub struct Receipt {
 	pub state_root: H256,
 	pub gas_used: U256,

src/service.rs

@@ -0,0 +1,59 @@
+use util::*;
+use sync::*;
+use spec::Spec;
+use error::*;
+use std::env;
+use client::Client;
+
+/// Client service setup. Creates and registers client and network services with the IO subsystem.
+pub struct ClientService {
+	_net_service: NetworkService<SyncMessage>,
+}
+
+impl ClientService {
+	/// Start the service in a separate thread.
+	pub fn start(spec: Spec) -> Result<ClientService, Error> {
+		let mut net_service = try!(NetworkService::start());
+		info!("Starting {}", net_service.host_info());
+		info!("Configured for {} using {} engine", spec.name, spec.engine_name);
+		let mut dir = env::home_dir().unwrap();
+		dir.push(".parity");
+		dir.push(H64::from(spec.genesis_header().hash()).hex());
+		let client = Arc::new(RwLock::new(try!(Client::new(spec, &dir, net_service.io().channel()))));
+		EthSync::register(&mut net_service, client.clone());
+		let client_io = Box::new(ClientIoHandler {
+			client: client
+		});
+		try!(net_service.io().register_handler(client_io));
+
+		Ok(ClientService {
+			_net_service: net_service,
+		})
+	}
+}
+
+/// IO interface for the Client handler
+struct ClientIoHandler {
+	client: Arc<RwLock<Client>>
+}
+
+impl IoHandler<NetSyncMessage> for ClientIoHandler {
+	fn initialize<'s>(&'s mut self, _io: &mut IoContext<'s, NetSyncMessage>) { }
+
+	fn message<'s>(&'s mut self, _io: &mut IoContext<'s, NetSyncMessage>, net_message: &'s mut NetSyncMessage) {
+		match net_message {
+			&mut UserMessage(ref mut message) =>  {
+				match message {
+					&mut SyncMessage::BlockVerified(ref mut bytes) => {
+						self.client.write().unwrap().import_verified_block(mem::replace(bytes, Bytes::new()));
+					},
+					_ => {}, // ignore other messages
+				}
+
+			}
+			_ => {}, // ignore other messages
+		}
+
+	}
+}
+

src/spec.rs

@@ -65,6 +65,8 @@ impl GenesisAccount {
 /// chain and those to be interpreted by the active chain engine.
 #[derive(Debug)]
 pub struct Spec {
+	// User friendly spec name
+	pub name: String,
 	// What engine are we using for this?
 	pub engine_name: String,
 
@@ -196,6 +198,7 @@ impl FromJson for Spec {
 
 		
 		Spec {
+			name: json.find("name").map(|j| j.as_string().unwrap()).unwrap_or("unknown").to_string(),
 			engine_name: json["engineName"].as_string().unwrap().to_string(),
 			engine_params: json_to_rlp_map(&json["params"]),
 			builtins: builtins,
@@ -218,11 +221,16 @@ impl Spec {
 	/// Ensure that the given state DB has the trie nodes in for the genesis state.
 	pub fn ensure_db_good(&self, db: &mut HashDB) {
 		if !db.contains(&self.state_root()) {
-			let mut root = H256::new();
+			info!("Populating genesis state...");
-			let mut t = SecTrieDBMut::new(db, &mut root);
+			let mut root = H256::new(); 
-			for (address, account) in self.genesis_state.iter() {
+			{
-				t.insert(address.as_slice(), &account.rlp());
+				let mut t = SecTrieDBMut::new(db, &mut root);
+				for (address, account) in self.genesis_state.iter() {
+					t.insert(address.as_slice(), &account.rlp());
+				}
 			}
+			assert!(db.contains(&self.state_root()));
+			info!("Genesis state is ready");
 		}
 	}
 

src/state.rs

@@ -142,7 +142,9 @@ impl State {
 		let e = try!(Executive::new(self, env_info, engine).transact(t));
 		//println!("Executed: {:?}", e);
 		self.commit();
-		Ok(Receipt::new(self.root().clone(), e.cumulative_gas_used, e.logs))
+		let receipt = Receipt::new(self.root().clone(), e.cumulative_gas_used, e.logs);
+		debug!("Transaction receipt: {:?}", receipt);
+		Ok(receipt)
 	}
 
 	pub fn revert(&mut self, backup: State) {

src/sync/chain.rs

@@ -234,7 +234,7 @@ impl ChainSync {
 	}
 
 	/// Called by peer to report status
-	fn on_peer_status(&mut self, io: &mut SyncIo, peer_id: &PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
+	fn on_peer_status(&mut self, io: &mut SyncIo, peer_id: PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
 		let peer = PeerInfo {
 			protocol_version: try!(r.val_at(0)),
 			network_id: try!(r.val_at(1)),
@@ -263,12 +263,13 @@ impl ChainSync {
 		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(())
 	}
 
 	/// Called by peer once it has new block headers during sync
-	fn on_peer_block_headers(&mut self, io: &mut SyncIo, peer_id: &PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
+	fn on_peer_block_headers(&mut self, io: &mut SyncIo, peer_id: PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
 		self.reset_peer_asking(peer_id, PeerAsking::BlockHeaders);
 		let item_count = r.item_count();
 		trace!(target: "sync", "{} -> BlockHeaders ({} entries)", peer_id, item_count);
@@ -351,7 +352,7 @@ impl ChainSync {
 	}
 
 	/// Called by peer once it has new block bodies
-	fn on_peer_block_bodies(&mut self, io: &mut SyncIo, peer_id: &PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
+	fn on_peer_block_bodies(&mut self, io: &mut SyncIo, peer_id: PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
 		use util::triehash::ordered_trie_root;
 		self.reset_peer_asking(peer_id, PeerAsking::BlockBodies);
 		let item_count = r.item_count();
@@ -391,7 +392,7 @@ impl ChainSync {
 	}
 
 	/// Called by peer once it has new block bodies
-	fn on_peer_new_block(&mut self, io: &mut SyncIo, peer_id: &PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
+	fn on_peer_new_block(&mut self, io: &mut SyncIo, peer_id: PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
 		let block_rlp = try!(r.at(0));
 		let header_rlp = try!(block_rlp.at(0));
 		let h = header_rlp.as_raw().sha3();
@@ -430,7 +431,7 @@ 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> {
+	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 {
 			trace!(target: "sync", "Ignoring new hashes since we're already downloading.");
 			return Ok(());
@@ -467,16 +468,18 @@ impl ChainSync {
 	}
 
 	/// Called by peer when it is disconnecting
-	pub fn on_peer_aborting(&mut self, io: &mut SyncIo, peer: &PeerId) {
+	pub fn on_peer_aborting(&mut self, io: &mut SyncIo, peer: PeerId) {
-		trace!(target: "sync", "== Disconnected {}", peer);
+		trace!(target: "sync", "== Disconnecting {}", peer);
 		if self.peers.contains_key(&peer) {
+			info!(target: "sync", "Disconneced {}:{}", peer, io.peer_info(peer));
 			self.clear_peer_download(peer);
+			self.peers.remove(&peer);
 			self.continue_sync(io);
 		}
 	}
 
 	/// Called when a new peer is connected
-	pub fn on_peer_connected(&mut self, io: &mut SyncIo, peer: &PeerId) {
+	pub fn on_peer_connected(&mut self, io: &mut SyncIo, peer: PeerId) {
 		trace!(target: "sync", "== Connected {}", peer);
 		self.send_status(io, peer);
 	}
@@ -486,7 +489,7 @@ impl ChainSync {
 		let mut peers: Vec<(PeerId, U256)> = self.peers.iter().map(|(k, p)| (*k, p.difficulty)).collect();
 		peers.sort_by(|&(_, d1), &(_, d2)| d1.cmp(&d2).reverse()); //TODO: sort by rating
 		for (p, _) in peers {
-			self.sync_peer(io, &p, false);
+			self.sync_peer(io, p, false);
 		}
 	}
 
@@ -504,7 +507,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) {
+	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");
 			if peer.asking != PeerAsking::Nothing {
@@ -534,7 +537,7 @@ impl ChainSync {
 	}
 
 	/// Find some headers or blocks to download for a peer.
-	fn request_blocks(&mut self, io: &mut SyncIo, peer_id: &PeerId) {
+	fn request_blocks(&mut self, io: &mut SyncIo, peer_id: PeerId) {
 		self.clear_peer_download(peer_id);
 
 		if io.chain().queue_status().full {
@@ -564,7 +567,7 @@ impl ChainSync {
 			}
 		}
 		if !needed_bodies.is_empty() {
-			replace(&mut self.peers.get_mut(peer_id).unwrap().asking_blocks, needed_numbers);
+			replace(&mut self.peers.get_mut(&peer_id).unwrap().asking_blocks, needed_numbers);
 			self.request_bodies(io, peer_id, needed_bodies);
 		}
 		else {
@@ -607,7 +610,7 @@ impl ChainSync {
 				if !headers.is_empty() {
 					start = headers[0] as usize;
 					let count = headers.len();
-					replace(&mut self.peers.get_mut(peer_id).unwrap().asking_blocks, headers);
+					replace(&mut self.peers.get_mut(&peer_id).unwrap().asking_blocks, headers);
 					assert!(!self.headers.have_item(&(start as BlockNumber)));
 					self.request_headers_by_number(io, peer_id, start as BlockNumber, count, 0, false);
 				}
@@ -619,7 +622,7 @@ impl ChainSync {
 	}
 
 	/// Clear all blocks/headers marked as being downloaded by a peer.
-	fn clear_peer_download(&mut self, peer_id: &PeerId) {
+	fn clear_peer_download(&mut self, peer_id: PeerId) {
 		let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer");
 		for b in &peer.asking_blocks {
 			self.downloading_headers.remove(&b);
@@ -715,7 +718,7 @@ impl ChainSync {
 	}
 
 	/// Request headers from a peer by block hash
-	fn request_headers_by_hash(&mut self, sync: &mut SyncIo, peer_id: &PeerId, h: &H256, count: usize, skip: usize, reverse: bool) {
+	fn request_headers_by_hash(&mut self, sync: &mut SyncIo, peer_id: PeerId, h: &H256, count: usize, skip: usize, reverse: bool) {
 		trace!(target: "sync", "{} <- GetBlockHeaders: {} entries starting from {}", peer_id, count, h);
 		let mut rlp = RlpStream::new_list(4);
 		rlp.append(h);
@@ -726,7 +729,7 @@ impl ChainSync {
 	}
 
 	/// Request headers from a peer by block number
-	fn request_headers_by_number(&mut self, sync: &mut SyncIo, peer_id: &PeerId, n: BlockNumber, count: usize, skip: usize, reverse: bool) {
+	fn request_headers_by_number(&mut self, sync: &mut SyncIo, peer_id: PeerId, n: BlockNumber, count: usize, skip: usize, reverse: bool) {
 		let mut rlp = RlpStream::new_list(4);
 		trace!(target: "sync", "{} <- GetBlockHeaders: {} entries starting from {}", peer_id, count, n);
 		rlp.append(&n);
@@ -737,7 +740,7 @@ impl ChainSync {
 	}
 
 	/// Request block bodies from a peer
-	fn request_bodies(&mut self, sync: &mut SyncIo, peer_id: &PeerId, hashes: Vec<H256>) {
+	fn request_bodies(&mut self, sync: &mut SyncIo, peer_id: PeerId, hashes: Vec<H256>) {
 		let mut rlp = RlpStream::new_list(hashes.len());
 		trace!(target: "sync", "{} <- GetBlockBodies: {} entries", peer_id, hashes.len());
 		for h in hashes {
@@ -747,7 +750,7 @@ impl ChainSync {
 	}
 
 	/// Reset peer status after request is complete.
-	fn reset_peer_asking(&mut self, peer_id: &PeerId, asking: PeerAsking) {
+	fn reset_peer_asking(&mut self, peer_id: PeerId, asking: PeerAsking) {
 		let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer");
 		if peer.asking != asking {
 			warn!(target:"sync", "Asking {:?} while expected {:?}", peer.asking, asking);
@@ -758,14 +761,14 @@ impl ChainSync {
 	}
 
 	/// Generic request sender
-	fn send_request(&mut self, sync: &mut SyncIo, peer_id: &PeerId, asking: PeerAsking,  packet_id: PacketId, packet: Bytes) {
+	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");
 			if peer.asking != PeerAsking::Nothing {
 				warn!(target:"sync", "Asking {:?} while requesting {:?}", asking, peer.asking);
 			}
 		}
-		match sync.send(*peer_id, packet_id, packet) {
+		match sync.send(peer_id, packet_id, packet) {
 			Err(e) => {
 				warn!(target:"sync", "Error sending request: {:?}", e);
 				sync.disable_peer(peer_id);
@@ -779,12 +782,12 @@ impl ChainSync {
 	}
 
 	/// Called when peer sends us new transactions
-	fn on_peer_transactions(&mut self, _io: &mut SyncIo, _peer_id: &PeerId, _r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
+	fn on_peer_transactions(&mut self, _io: &mut SyncIo, _peer_id: PeerId, _r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
 		Ok(())
 	}
 
 	/// Send Status message
-	fn send_status(&mut self, io: &mut SyncIo, peer_id: &PeerId) {
+	fn send_status(&mut self, io: &mut SyncIo, peer_id: PeerId) {
 		let mut packet = RlpStream::new_list(5);
 		let chain = io.chain().chain_info();
 		packet.append(&(PROTOCOL_VERSION as u32));
@@ -793,7 +796,7 @@ impl ChainSync {
 		packet.append(&chain.best_block_hash);
 		packet.append(&chain.genesis_hash);
 		//TODO: handle timeout for status request
-		match io.send(*peer_id, STATUS_PACKET, packet.out()) {
+		match io.send(peer_id, STATUS_PACKET, packet.out()) {
 			Err(e) => {
 				warn!(target:"sync", "Error sending status request: {:?}", e);
 				io.disable_peer(peer_id);
@@ -940,7 +943,7 @@ impl ChainSync {
 	}
 
 	/// Dispatch incoming requests and responses
-	pub fn on_packet(&mut self, io: &mut SyncIo, peer: &PeerId, packet_id: u8, data: &[u8]) {
+	pub fn on_packet(&mut self, io: &mut SyncIo, peer: PeerId, packet_id: u8, data: &[u8]) {
 		let rlp = UntrustedRlp::new(data);
 		let result = match packet_id {
 			STATUS_PACKET => self.on_peer_status(io, peer, &rlp),

src/sync/io.rs

@@ -1,30 +1,35 @@
 use client::BlockChainClient;
-use util::network::{HandlerIo, PeerId, PacketId,};
+use util::{NetworkContext, PeerId, PacketId,};
 use util::error::UtilError;
+use sync::SyncMessage;
 
 /// IO interface for the syning handler.
 /// Provides peer connection management and an interface to the blockchain client.
 // TODO: ratings
 pub trait SyncIo {
 	/// Disable a peer
-	fn disable_peer(&mut self, peer_id: &PeerId);
+	fn disable_peer(&mut self, peer_id: PeerId);
 	/// Respond to current request with a packet. Can be called from an IO handler for incoming packet.
 	fn respond(&mut self, packet_id: PacketId, data: Vec<u8>) -> Result<(), UtilError>;
 	/// Send a packet to a peer.
 	fn send(&mut self, peer_id: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), UtilError>;
 	/// Get the blockchain
 	fn chain<'s>(&'s mut self) -> &'s mut BlockChainClient;
+	/// Returns peer client identifier string
+	fn peer_info(&self, peer_id: PeerId) -> String {
+		peer_id.to_string()
+	}
 }
 
-/// Wraps `HandlerIo` and the blockchain client
+/// Wraps `NetworkContext` and the blockchain client
-pub struct NetSyncIo<'s, 'h> where 'h:'s {
+pub struct NetSyncIo<'s, 'h, 'io> where 'h: 's, 'io: 'h {
-	network: &'s mut HandlerIo<'h>,
+	network: &'s mut NetworkContext<'h, 'io, SyncMessage>,
 	chain: &'s mut BlockChainClient
 }
 
-impl<'s, 'h> NetSyncIo<'s, 'h> {
+impl<'s, 'h, 'io> NetSyncIo<'s, 'h, 'io> {
-	/// Creates a new instance from the `HandlerIo` and the blockchain client reference.
+	/// Creates a new instance from the `NetworkContext` and the blockchain client reference.
-	pub fn new(network: &'s mut HandlerIo<'h>, chain: &'s mut BlockChainClient) -> NetSyncIo<'s,'h> {
+	pub fn new(network: &'s mut NetworkContext<'h, 'io, SyncMessage>, chain: &'s mut BlockChainClient) -> NetSyncIo<'s,'h,'io> {
 		NetSyncIo {
 			network: network,
 			chain: chain,
@@ -32,9 +37,9 @@ impl<'s, 'h> NetSyncIo<'s, 'h> {
 	}
 }
 
-impl<'s, 'h> SyncIo for NetSyncIo<'s, 'h> {
+impl<'s, 'h, 'op> SyncIo for NetSyncIo<'s, 'h, 'op> {
-	fn disable_peer(&mut self, peer_id: &PeerId) {
+	fn disable_peer(&mut self, peer_id: PeerId) {
-		self.network.disable_peer(*peer_id);
+		self.network.disable_peer(peer_id);
 	}
 
 	fn respond(&mut self, packet_id: PacketId, data: Vec<u8>) -> Result<(), UtilError>{
@@ -48,6 +53,10 @@ impl<'s, 'h> SyncIo for NetSyncIo<'s, 'h> {
 	fn chain<'a>(&'a mut self) -> &'a mut BlockChainClient {
 		self.chain
 	}
+
+	fn peer_info(&self, peer_id: PeerId) -> String {
+		self.network.peer_info(peer_id)
+	}
 }
 
 

src/sync/mod.rs

@@ -22,9 +22,12 @@
 /// }
 /// ```
 
-use std::sync::Arc;
+use std::ops::*;
-use client::BlockChainClient;
+use std::sync::*;
-use util::network::{ProtocolHandler, NetworkService, HandlerIo, TimerToken, PeerId, Message};
+use client::Client;
+use util::network::{NetworkProtocolHandler, NetworkService, NetworkContext, PeerId, NetworkIoMessage};
+use util::TimerToken;
+use util::Bytes;
 use sync::chain::ChainSync;
 use sync::io::NetSyncIo;
 
@@ -35,10 +38,20 @@ mod range_collection;
 #[cfg(test)]
 mod tests;
 
+/// Message type for external events
+pub enum SyncMessage {
+	/// New block has been imported into the blockchain
+	NewChainBlock(Bytes),
+	/// A block is ready 
+	BlockVerified(Bytes),
+}
+
+pub type NetSyncMessage = NetworkIoMessage<SyncMessage>;
+
 /// Ethereum network protocol handler
 pub struct EthSync {
 	/// Shared blockchain client. TODO: this should evetually become an IPC endpoint
-	chain: Arc<BlockChainClient + Send + Sized>,
+	chain: Arc<RwLock<Client>>,
 	/// Sync strategy
 	sync: ChainSync
 }
@@ -47,7 +60,7 @@ pub use self::chain::SyncStatus;
 
 impl EthSync {
 	/// Creates and register protocol with the network service
-	pub fn register(service: &mut NetworkService, chain: Arc<BlockChainClient + Send + Sized>) {
+	pub fn register(service: &mut NetworkService<SyncMessage>, chain: Arc<RwLock<Client>>) {
 		let sync = Box::new(EthSync {
 			chain: chain,
 			sync: ChainSync::new(),
@@ -61,39 +74,39 @@ impl EthSync {
 	}
 
 	/// Stop sync
-	pub fn stop(&mut self, io: &mut HandlerIo) {
+	pub fn stop(&mut self, io: &mut NetworkContext<SyncMessage>) {
-		self.sync.abort(&mut NetSyncIo::new(io, Arc::get_mut(&mut self.chain).unwrap()));
+		self.sync.abort(&mut NetSyncIo::new(io, self.chain.write().unwrap().deref_mut()));
 	}
 
 	/// Restart sync
-	pub fn restart(&mut self, io: &mut HandlerIo) {
+	pub fn restart(&mut self, io: &mut NetworkContext<SyncMessage>) {
-		self.sync.restart(&mut NetSyncIo::new(io, Arc::get_mut(&mut self.chain).unwrap()));
+		self.sync.restart(&mut NetSyncIo::new(io, self.chain.write().unwrap().deref_mut()));
 	}
 }
 
-impl ProtocolHandler for EthSync {
+impl NetworkProtocolHandler<SyncMessage> for EthSync {
-	fn initialize(&mut self, io: &mut HandlerIo) {
+	fn initialize(&mut self, io: &mut NetworkContext<SyncMessage>) {
-		self.sync.restart(&mut NetSyncIo::new(io, Arc::get_mut(&mut self.chain).unwrap()));
+		self.sync.restart(&mut NetSyncIo::new(io, self.chain.write().unwrap().deref_mut()));
 		io.register_timer(1000).unwrap();
 	}
 
-	fn read(&mut self, io: &mut HandlerIo, peer: &PeerId, packet_id: u8, data: &[u8]) {
+	fn read(&mut self, io: &mut NetworkContext<SyncMessage>, peer: &PeerId, packet_id: u8, data: &[u8]) {
-		self.sync.on_packet(&mut NetSyncIo::new(io, Arc::get_mut(&mut self.chain).unwrap()), peer, packet_id, data);
+		self.sync.on_packet(&mut NetSyncIo::new(io, self.chain.write().unwrap().deref_mut()) , *peer, packet_id, data);
 	}
 
-	fn connected(&mut self, io: &mut HandlerIo, peer: &PeerId) {
+	fn connected(&mut self, io: &mut NetworkContext<SyncMessage>, peer: &PeerId) {
-		self.sync.on_peer_connected(&mut NetSyncIo::new(io, Arc::get_mut(&mut self.chain).unwrap()), peer);
+		self.sync.on_peer_connected(&mut NetSyncIo::new(io, self.chain.write().unwrap().deref_mut()), *peer);
 	}
 
-	fn disconnected(&mut self, io: &mut HandlerIo, peer: &PeerId) {
+	fn disconnected(&mut self, io: &mut NetworkContext<SyncMessage>, peer: &PeerId) {
-		self.sync.on_peer_aborting(&mut NetSyncIo::new(io, Arc::get_mut(&mut self.chain).unwrap()), peer);
+		self.sync.on_peer_aborting(&mut NetSyncIo::new(io, self.chain.write().unwrap().deref_mut()), *peer);
 	}
 
-	fn timeout(&mut self, io: &mut HandlerIo, _timer: TimerToken) {
+	fn timeout(&mut self, io: &mut NetworkContext<SyncMessage>, _timer: TimerToken) {
-		self.sync.maintain_sync(&mut NetSyncIo::new(io, Arc::get_mut(&mut self.chain).unwrap()));
+		self.sync.maintain_sync(&mut NetSyncIo::new(io, self.chain.write().unwrap().deref_mut()));
 	}
 
-	fn message(&mut self, _io: &mut HandlerIo, _message: &Message) {
+	fn message(&mut self, _io: &mut NetworkContext<SyncMessage>, _message: &SyncMessage) {
 	}
 }
 

src/sync/tests.rs

@@ -187,7 +187,7 @@ impl<'p> TestIo<'p> {
 }
 
 impl<'p> SyncIo for TestIo<'p> {
-	fn disable_peer(&mut self, _peer_id: &PeerId) {
+	fn disable_peer(&mut self, _peer_id: PeerId) {
 	}
 
 	fn respond(&mut self, packet_id: PacketId, data: Vec<u8>) -> Result<(), UtilError> {
@@ -257,7 +257,7 @@ impl TestNet {
 			for client in 0..self.peers.len() {
 				if peer != client {
 					let mut p = self.peers.get_mut(peer).unwrap();
-					p.sync.on_peer_connected(&mut TestIo::new(&mut p.chain, &mut p.queue, Some(client as PeerId)), &(client as PeerId));
+					p.sync.on_peer_connected(&mut TestIo::new(&mut p.chain, &mut p.queue, Some(client as PeerId)), client as PeerId);
 				}
 			}
 		}
@@ -269,7 +269,7 @@ impl TestNet {
 				Some(packet) => {
 					let mut p = self.peers.get_mut(packet.recipient).unwrap();
 					trace!("--- {} -> {} ---", peer, packet.recipient);
-					p.sync.on_packet(&mut TestIo::new(&mut p.chain, &mut p.queue, Some(peer as PeerId)), &(peer as PeerId), packet.packet_id, &packet.data);
+					p.sync.on_packet(&mut TestIo::new(&mut p.chain, &mut p.queue, Some(peer as PeerId)), peer as PeerId, packet.packet_id, &packet.data);
 					trace!("----------------");
 				},
 				None => {}

src/verification.rs

@@ -10,9 +10,7 @@ use engine::Engine;
 use blockchain::*;
 
 /// Phase 1 quick block verification. Only does checks that are cheap. Operates on a single block
-pub fn verify_block_basic(bytes: &[u8], engine: &Engine) -> Result<(), Error> {
+pub fn verify_block_basic(header: &Header, bytes: &[u8], engine: &Engine) -> Result<(), Error> {
-	let block = BlockView::new(bytes);
-	let header = block.header();
 	try!(verify_header(&header, engine));
 	try!(verify_block_integrity(bytes, &header.transactions_root, &header.uncles_hash));
 	try!(engine.verify_block_basic(&header, Some(bytes)));
@@ -26,9 +24,7 @@ pub fn verify_block_basic(bytes: &[u8], engine: &Engine) -> Result<(), Error> {
 /// Phase 2 verification. Perform costly checks such as transaction signatures and block nonce for ethash.
 /// Still operates on a individual block
 /// TODO: return cached transactions, header hash.
-pub fn verify_block_unordered(bytes: &[u8], engine: &Engine) -> Result<(), Error> {
+pub fn verify_block_unordered(header: &Header, bytes: &[u8], engine: &Engine) -> Result<(), Error> {
-	let block = BlockView::new(bytes);
-	let header = block.header();
 	try!(engine.verify_block_unordered(&header, Some(bytes)));
 	for u in Rlp::new(bytes).at(2).iter().map(|rlp| rlp.as_val::<Header>()) {
 		try!(engine.verify_block_unordered(&u, None));
@@ -37,12 +33,11 @@ pub fn verify_block_unordered(bytes: &[u8], engine: &Engine) -> Result<(), Error
 }
 
 /// Phase 3 verification. Check block information against parent and uncles.
-pub fn verify_block_final<BC>(bytes: &[u8], engine: &Engine, bc: &BC) -> Result<(), Error> where BC: BlockProvider {
+pub fn verify_block_family<BC>(header: &Header, bytes: &[u8], engine: &Engine, bc: &BC) -> Result<(), Error> where BC: BlockProvider {
-	let block = BlockView::new(bytes);
+	// TODO: verify timestamp
-	let header = block.header();
 	let parent = try!(bc.block_header(&header.parent_hash).ok_or::<Error>(From::from(BlockError::UnknownParent(header.parent_hash.clone()))));
 	try!(verify_parent(&header, &parent));
-	try!(engine.verify_block_final(&header, &parent, Some(bytes)));
+	try!(engine.verify_block_family(&header, &parent, Some(bytes)));
 
 	let num_uncles = Rlp::new(bytes).at(2).item_count();
 	if num_uncles != 0 {
@@ -112,12 +107,29 @@ pub fn verify_block_final<BC>(bytes: &[u8], engine: &Engine, bc: &BC) -> Result<
 			}
 
 			try!(verify_parent(&uncle, &uncle_parent));
-			try!(engine.verify_block_final(&uncle, &uncle_parent, Some(bytes)));
+			try!(engine.verify_block_family(&uncle, &uncle_parent, Some(bytes)));
 		}
 	}
 	Ok(())
 }
 
+/// Phase 4 verification. Check block information against transaction enactment results,
+pub fn verify_block_final(expected: &Header, got: &Header) -> Result<(), Error> {
+	if expected.state_root != got.state_root {
+		return Err(From::from(BlockError::InvalidStateRoot(Mismatch { expected: expected.state_root.clone(), found: got.state_root.clone() })))
+	}
+	if expected.receipts_root != got.receipts_root {
+		return Err(From::from(BlockError::InvalidReceiptsStateRoot(Mismatch { expected: expected.receipts_root.clone(), found: got.receipts_root.clone() })))
+	}
+	if expected.log_bloom != got.log_bloom {
+		return Err(From::from(BlockError::InvalidLogBloom(Mismatch { expected: expected.log_bloom.clone(), found: got.log_bloom.clone() })))
+	}
+	if expected.gas_used != got.gas_used {
+		return Err(From::from(BlockError::InvalidGasUsed(Mismatch { expected: expected.gas_used, found: got.gas_used })))
+	}
+	Ok(())
+}
+
 /// Check basic header parameters.
 fn verify_header(header: &Header, engine: &Engine) -> Result<(), Error> {
 	if header.number >= From::from(BlockNumber::max_value()) {
@@ -176,6 +188,7 @@ mod tests {
 	use error::BlockError::*;
 	use views::*;
 	use blockchain::*;
+	use engine::*;
 	use ethereum;
 
 	fn create_test_block(header: &Header) -> Bytes {
@@ -262,6 +275,16 @@ mod tests {
 		}
 	}
 
+	fn basic_test(bytes: &[u8], engine: &Engine) -> Result<(), Error> {
+		let header = BlockView::new(bytes).header();
+		verify_block_basic(&header, bytes, engine)
+	}
+
+	fn family_test<BC>(bytes: &[u8], engine: &Engine, bc: &BC) -> Result<(), Error> where BC: BlockProvider {
+		let header = BlockView::new(bytes).header();
+		verify_block_family(&header, bytes, engine, bc)
+	}
+
 	#[test]
 	fn test_verify_block() {
 		// Test against morden
@@ -330,69 +353,69 @@ mod tests {
 		bc.insert(create_test_block(&parent7));
 		bc.insert(create_test_block(&parent8));
 
-		check_ok(verify_block_basic(&create_test_block(&good), engine.deref()));
+		check_ok(basic_test(&create_test_block(&good), engine.deref()));
 
 		let mut header = good.clone();
 		header.transactions_root = good_transactions_root.clone();
 		header.uncles_hash = good_uncles_hash.clone();
-		check_ok(verify_block_basic(&create_test_block_with_data(&header, &good_transactions, &good_uncles), engine.deref()));
+		check_ok(basic_test(&create_test_block_with_data(&header, &good_transactions, &good_uncles), engine.deref()));
 
 		header.gas_limit = min_gas_limit - From::from(1);
-		check_fail(verify_block_basic(&create_test_block(&header), engine.deref()),
+		check_fail(basic_test(&create_test_block(&header), engine.deref()),
 			InvalidGasLimit(OutOfBounds { min: Some(min_gas_limit), max: None, found: header.gas_limit }));
 
 		header = good.clone();
 		header.number = BlockNumber::max_value();
-		check_fail(verify_block_basic(&create_test_block(&header), engine.deref()),
+		check_fail(basic_test(&create_test_block(&header), engine.deref()),
 			InvalidNumber(OutOfBounds { max: Some(BlockNumber::max_value()), min: None, found: header.number }));
 
 		header = good.clone();
 		header.gas_used = header.gas_limit + From::from(1);
-		check_fail(verify_block_basic(&create_test_block(&header), engine.deref()),
+		check_fail(basic_test(&create_test_block(&header), engine.deref()),
 			TooMuchGasUsed(OutOfBounds { max: Some(header.gas_limit), min: None, found: header.gas_used }));
 
 		header = good.clone();
 		header.extra_data.resize(engine.maximum_extra_data_size() + 1, 0u8);
-		check_fail(verify_block_basic(&create_test_block(&header), engine.deref()),
+		check_fail(basic_test(&create_test_block(&header), engine.deref()),
 			ExtraDataOutOfBounds(OutOfBounds { max: Some(engine.maximum_extra_data_size()), min: None, found: header.extra_data.len() }));
 
 		header = good.clone();
 		header.extra_data.resize(engine.maximum_extra_data_size() + 1, 0u8);
-		check_fail(verify_block_basic(&create_test_block(&header), engine.deref()),
+		check_fail(basic_test(&create_test_block(&header), engine.deref()),
 			ExtraDataOutOfBounds(OutOfBounds { max: Some(engine.maximum_extra_data_size()), min: None, found: header.extra_data.len() }));
 
 		header = good.clone();
 		header.uncles_hash = good_uncles_hash.clone();
-		check_fail(verify_block_basic(&create_test_block_with_data(&header, &good_transactions, &good_uncles), engine.deref()),
+		check_fail(basic_test(&create_test_block_with_data(&header, &good_transactions, &good_uncles), engine.deref()),
 			InvalidTransactionsRoot(Mismatch { expected: good_transactions_root.clone(), found: header.transactions_root }));
 
 		header = good.clone();
 		header.transactions_root = good_transactions_root.clone();
-		check_fail(verify_block_basic(&create_test_block_with_data(&header, &good_transactions, &good_uncles), engine.deref()),
+		check_fail(basic_test(&create_test_block_with_data(&header, &good_transactions, &good_uncles), engine.deref()),
 			InvalidUnclesHash(Mismatch { expected: good_uncles_hash.clone(), found: header.uncles_hash }));
 
-		check_ok(verify_block_final(&create_test_block(&good), engine.deref(), &bc));
+		check_ok(family_test(&create_test_block(&good), engine.deref(), &bc));
-		check_ok(verify_block_final(&create_test_block_with_data(&good, &good_transactions, &good_uncles), engine.deref(), &bc));
+		check_ok(family_test(&create_test_block_with_data(&good, &good_transactions, &good_uncles), engine.deref(), &bc));
 
 		header = good.clone();
 		header.parent_hash = H256::random();
-		check_fail(verify_block_final(&create_test_block_with_data(&header, &good_transactions, &good_uncles), engine.deref(), &bc),
+		check_fail(family_test(&create_test_block_with_data(&header, &good_transactions, &good_uncles), engine.deref(), &bc),
 			UnknownParent(header.parent_hash));
 
 		header = good.clone();
 		header.timestamp = 10;
-		check_fail(verify_block_final(&create_test_block_with_data(&header, &good_transactions, &good_uncles), engine.deref(), &bc),
+		check_fail(family_test(&create_test_block_with_data(&header, &good_transactions, &good_uncles), engine.deref(), &bc),
 			InvalidTimestamp(OutOfBounds { max: None, min: Some(parent.timestamp + 1), found: header.timestamp }));
 
 		header = good.clone();
 		header.number = 9;
-		check_fail(verify_block_final(&create_test_block_with_data(&header, &good_transactions, &good_uncles), engine.deref(), &bc),
+		check_fail(family_test(&create_test_block_with_data(&header, &good_transactions, &good_uncles), engine.deref(), &bc),
 			InvalidNumber(OutOfBounds { max: None, min: Some(parent.number + 1), found: header.number }));
 		
 		header = good.clone();
 		let mut bad_uncles = good_uncles.clone();
 		bad_uncles.push(good_uncle1.clone());
-		check_fail(verify_block_final(&create_test_block_with_data(&header, &good_transactions, &bad_uncles), engine.deref(), &bc),
+		check_fail(family_test(&create_test_block_with_data(&header, &good_transactions, &bad_uncles), engine.deref(), &bc),
 			TooManyUncles(OutOfBounds { max: Some(engine.maximum_uncle_count()), min: None, found: bad_uncles.len() }));
 
 		// TODO: some additional uncle checks