Relaying tx/blocks by light clients (#190)

* do not import external transactions into light tx pool

* do not announce blocks on light clients

* blocks_are_not_announced_by_light_nodes

polkadot/cli/src/informant.rs

@@ -18,7 +18,7 @@
 
 use std::time::{Duration, Instant};
 use futures::stream::Stream;
-use service::Service;
+use service::{Service, Components};
 use tokio_core::reactor;
 use network::{SyncState, SyncProvider};
 use polkadot_primitives::Block;
@@ -28,11 +28,10 @@ use client::{self, BlockchainEvents};
 const TIMER_INTERVAL_MS: u64 = 5000;
 
 /// Spawn informant on the event loop
-pub fn start<B, E>(service: &Service<B, E>, handle: reactor::Handle)
+pub fn start<C>(service: &Service<C>, handle: reactor::Handle)
 	where
-		B: client::backend::Backend<Block> + Send + Sync + 'static,
-		E: client::CallExecutor<Block> + Send + Sync + 'static,
-		client::error::Error: From<<<B as client::backend::Backend<Block>>::State as state_machine::backend::Backend>::Error>
+		C: Components,
+		client::error::Error: From<<<<C as Components>::Backend as client::backend::Backend<Block>>::State as state_machine::Backend>::Error>,
 {
 	let interval = reactor::Interval::new_at(Instant::now(), Duration::from_millis(TIMER_INTERVAL_MS), &handle)
 		.expect("Error creating informant timer");

polkadot/cli/src/lib.rs

@@ -213,11 +213,10 @@ pub fn run<I, T>(args: I) -> error::Result<()> where
 	}
 }
 
-fn run_until_exit<B, E>(mut core: reactor::Core, service: service::Service<B, E>, matches: &clap::ArgMatches, config: service::Configuration) -> error::Result<()>
+fn run_until_exit<C>(mut core: reactor::Core, service: service::Service<C>, matches: &clap::ArgMatches, config: service::Configuration) -> error::Result<()>
 	where
-		B: client::backend::Backend<Block> + Send + Sync + 'static,
-		E: client::CallExecutor<Block> + Send + Sync + 'static,
-		client::error::Error: From<<<B as client::backend::Backend<Block>>::State as state_machine::backend::Backend>::Error>
+		C: service::Components,
+		client::error::Error: From<<<<C as service::Components>::Backend as client::backend::Backend<Block>>::State as state_machine::Backend>::Error>,
 {
 	let exit = {
 		// can't use signal directly here because CtrlC takes only `Fn`.

polkadot/service/src/components.rs

@@ -0,0 +1,232 @@
+// Copyright 2017 Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot 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.
+
+// Polkadot 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 Polkadot.  If not, see <http://www.gnu.org/licenses/>.?
+
+//! Polkadot service components.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use client::{self, genesis, Client};
+use client_db;
+use codec::{self, Slicable};
+use consensus;
+use keystore::Store as Keystore;
+use network;
+use polkadot_api;
+use polkadot_executor::Executor as LocalDispatch;
+use polkadot_runtime::{GenesisConfig, BuildExternalities};
+use polkadot_primitives::{Block, BlockId, Hash, Header};
+use state_machine;
+use substrate_executor::NativeExecutor;
+use transaction_pool::{self, TransactionPool};
+use error;
+
+/// Code executor.
+pub type CodeExecutor = NativeExecutor<LocalDispatch>;
+
+/// Polkadot service components.
+pub trait Components {
+	/// Client backend type.
+	type Backend: 'static + client::backend::Backend<Block>;
+
+	/// Polkadot API type.
+	type Api: 'static + polkadot_api::PolkadotApi + Send + Sync;
+
+	/// Code executor type.
+	type Executor: 'static + client::CallExecutor<Block> + Send + Sync;
+
+	/// Create client.
+	fn build_client(&self, settings: client_db::DatabaseSettings, executor: CodeExecutor, genesis: GenesisBuilder)
+		-> Result<(Arc<Client<Self::Backend, Self::Executor, Block>>, Option<Arc<network::OnDemand<Block, network::Service<Block>>>>), error::Error>;
+
+	/// Create api.
+	fn build_api(&self, client: Arc<Client<Self::Backend, Self::Executor, Block>>) -> Arc<Self::Api>;
+
+	/// Create network transaction pool adapter.
+	fn build_network_tx_pool(&self, client: Arc<client::Client<Self::Backend, Self::Executor, Block>>, api: Arc<Self::Api>, tx_pool: Arc<TransactionPool>)
+		-> Arc<network::TransactionPool<Block>>;
+
+	/// Create consensus service.
+	fn build_consensus(&self, client: Arc<Client<Self::Backend, Self::Executor, Block>>, network: Arc<network::Service<Block>>, tx_pool: Arc<TransactionPool>, keystore: &Keystore)
+		-> Result<Option<consensus::Service>, error::Error>;
+}
+
+/// Genesis block builder.
+pub struct GenesisBuilder {
+	pub config: GenesisConfig,
+}
+
+impl client::GenesisBuilder<Block> for GenesisBuilder {
+	fn build(self) -> (Header, Vec<(Vec<u8>, Vec<u8>)>) {
+		let storage = self.config.build_externalities();
+		let block = genesis::construct_genesis_block::<Block>(&storage);
+		(block.header, storage.into_iter().collect())
+	}
+}
+
+/// Components for full Polkadot service.
+pub struct FullComponents {
+	/// Is this a validator node?
+	pub is_validator: bool,
+}
+
+impl Components for FullComponents {
+	type Backend = client_db::Backend<Block>;
+	type Api = Client<Self::Backend, Self::Executor, Block>;
+	type Executor = client::LocalCallExecutor<client_db::Backend<Block>, NativeExecutor<LocalDispatch>>;
+
+	fn build_client(&self, db_settings: client_db::DatabaseSettings, executor: CodeExecutor, genesis: GenesisBuilder)
+		-> Result<(Arc<client::Client<Self::Backend, Self::Executor, Block>>, Option<Arc<network::OnDemand<Block, network::Service<Block>>>>), error::Error> {
+		Ok((Arc::new(client_db::new_client(db_settings, executor, genesis)?), None))
+	}
+
+	fn build_api(&self, client: Arc<client::Client<Self::Backend, Self::Executor, Block>>) -> Arc<Self::Api> {
+		client
+	}
+
+	fn build_network_tx_pool(&self, client: Arc<client::Client<Self::Backend, Self::Executor, Block>>, api: Arc<Self::Api>, pool: Arc<TransactionPool>)
+		-> Arc<network::TransactionPool<Block>> {
+		Arc::new(TransactionPoolAdapter {
+			imports_external_transactions: true,
+			pool,
+			client,
+			api,
+		})
+	}
+
+	fn build_consensus(&self, client: Arc<client::Client<Self::Backend, Self::Executor, Block>>, network: Arc<network::Service<Block>>, tx_pool: Arc<TransactionPool>, keystore: &Keystore)
+		-> Result<Option<consensus::Service>, error::Error> {
+		if !self.is_validator {
+			return Ok(None);
+		}
+
+		// Load the first available key
+		let key = keystore.load(&keystore.contents()?[0], "")?;
+		info!("Using authority key {:?}", key.public());
+		Ok(Some(consensus::Service::new(
+			client.clone(),
+			client.clone(),
+			network.clone(),
+			tx_pool.clone(),
+			::std::time::Duration::from_millis(4000), // TODO: dynamic
+			key,
+		)))
+	}
+}
+
+/// Components for light Polkadot service.
+pub struct LightComponents;
+
+impl Components for LightComponents {
+	type Backend = client::light::Backend<Block>;
+	type Api = polkadot_api::light::RemotePolkadotApiWrapper<Self::Backend, Self::Executor>;
+	type Executor = client::RemoteCallExecutor<client::light::Backend<Block>, network::OnDemand<Block, network::Service<Block>>>;
+
+	fn build_client(&self, _settings: client_db::DatabaseSettings, executor: CodeExecutor, genesis: GenesisBuilder)
+		-> Result<(Arc<client::Client<Self::Backend, Self::Executor, Block>>, Option<Arc<network::OnDemand<Block, network::Service<Block>>>>), error::Error> {
+		let client_backend = client::light::new_light_backend();
+		let fetch_checker = Arc::new(client::light::new_fetch_checker(client_backend.clone(), executor));
+		let fetcher = Arc::new(network::OnDemand::new(fetch_checker));
+		let client = client::light::new_light(client_backend, fetcher.clone(), genesis)?;
+		Ok((Arc::new(client), Some(fetcher)))
+	}
+
+	fn build_api(&self, client: Arc<client::Client<Self::Backend, Self::Executor, Block>>) -> Arc<Self::Api> {
+		Arc::new(polkadot_api::light::RemotePolkadotApiWrapper(client.clone()))
+	}
+
+	fn build_network_tx_pool(&self, client: Arc<client::Client<Self::Backend, Self::Executor, Block>>, api: Arc<Self::Api>, pool: Arc<TransactionPool>)
+		-> Arc<network::TransactionPool<Block>> {
+		Arc::new(TransactionPoolAdapter {
+			imports_external_transactions: false,
+			pool,
+			client,
+			api,
+		})
+	}
+
+	fn build_consensus(&self, _client: Arc<client::Client<Self::Backend, Self::Executor, Block>>, _network: Arc<network::Service<Block>>, _tx_pool: Arc<TransactionPool>, _keystore: &Keystore)
+		-> Result<Option<consensus::Service>, error::Error> {
+		Ok(None)
+	}
+}
+
+/// Transaction pool adapter.
+pub struct TransactionPoolAdapter<B, E, A> where A: Send + Sync, E: Send + Sync {
+	imports_external_transactions: bool,
+	pool: Arc<TransactionPool>,
+	client: Arc<Client<B, E, Block>>,
+	api: Arc<A>,
+}
+
+impl<B, E, A> network::TransactionPool<Block> for TransactionPoolAdapter<B, E, A>
+	where
+		B: client::backend::Backend<Block> + Send + Sync,
+		E: client::CallExecutor<Block> + Send + Sync,
+		client::error::Error: From<<<B as client::backend::Backend<Block>>::State as state_machine::backend::Backend>::Error>,
+		A: polkadot_api::PolkadotApi + Send + Sync,
+{
+	fn transactions(&self) -> Vec<(Hash, Vec<u8>)> {
+		let best_block = match self.client.info() {
+			Ok(info) => info.chain.best_hash,
+			Err(e) => {
+				debug!("Error getting best block: {:?}", e);
+				return Vec::new();
+			}
+		};
+
+		let id = match self.api.check_id(BlockId::hash(best_block)) {
+			Ok(id) => id,
+			Err(_) => return Vec::new(),
+		};
+
+		let ready = transaction_pool::Ready::create(id, &*self.api);
+
+		self.pool.cull_and_get_pending(ready, |pending| pending
+			.map(|t| {
+				let hash = t.hash().clone();
+				(hash, t.primitive_extrinsic())
+			})
+			.collect()
+		)
+	}
+
+	fn import(&self, transaction: &Vec<u8>) -> Option<Hash> {
+		if !self.imports_external_transactions {
+			return None;
+		}
+
+		let encoded = transaction.encode();
+		if let Some(uxt) = codec::Slicable::decode(&mut &encoded[..]) {
+			match self.pool.import_unchecked_extrinsic(uxt) {
+				Ok(xt) => Some(*xt.hash()),
+				Err(e) => match *e.kind() {
+					transaction_pool::ErrorKind::AlreadyImported(hash) => Some(hash[..].into()),
+					_ => {
+						debug!("Error adding transaction to the pool: {:?}", e);
+						None
+					},
+				}
+			}
+		} else {
+			debug!("Error decoding transaction");
+			None
+		}
+	}
+
+	fn on_broadcasted(&self, propagations: HashMap<Hash, Vec<String>>) {
+		self.pool.on_broadcasted(propagations)
+	}
+}

polkadot/service/src/lib.rs

@@ -51,106 +51,39 @@ extern crate log;
 #[macro_use]
 extern crate hex_literal;
 
+mod components;
 mod error;
 mod config;
 
-use std::collections::HashMap;
 use std::sync::Arc;
 use std::thread;
 use futures::prelude::*;
 use tokio_core::reactor::Core;
-use codec::Slicable;
 use primitives::AuthorityId;
 use transaction_pool::TransactionPool;
-use substrate_executor::NativeExecutor;
-use polkadot_executor::Executor as LocalDispatch;
 use keystore::Store as Keystore;
 use polkadot_api::PolkadotApi;
-use polkadot_primitives::{Block, BlockId, Hash, Header};
+use polkadot_primitives::{Block, BlockId, Hash};
 use polkadot_runtime::{GenesisConfig, ConsensusConfig, CouncilConfig, DemocracyConfig,
-	SessionConfig, StakingConfig, BuildExternalities};
-use client::backend::Backend;
-use client::{genesis, Client, BlockchainEvents, CallExecutor};
+	SessionConfig, StakingConfig};
+use client::{Client, BlockchainEvents};
 use network::ManageNetwork;
 use exit_future::Signal;
 
 pub use self::error::{ErrorKind, Error};
+pub use self::components::{Components, FullComponents, LightComponents};
 pub use config::{Configuration, Role, OptionChainSpec, ChainSpec};
 
-type CodeExecutor = NativeExecutor<LocalDispatch>;
-
 /// Polkadot service.
-pub struct Service<B, E> {
+pub struct Service<Components: components::Components> {
 	thread: Option<thread::JoinHandle<()>>,
-	client: Arc<Client<B, E, Block>>,
+	client: Arc<Client<Components::Backend, Components::Executor, Block>>,
 	network: Arc<network::Service<Block>>,
 	transaction_pool: Arc<TransactionPool>,
 	signal: Option<Signal>,
 	_consensus: Option<consensus::Service>,
 }
 
-struct TransactionPoolAdapter<B, E, A> where A: Send + Sync, E: Send + Sync {
-	pool: Arc<TransactionPool>,
-	client: Arc<Client<B, E, Block>>,
-	api: Arc<A>,
-}
-
-impl<B, E, A> network::TransactionPool<Block> for TransactionPoolAdapter<B, E, A>
-	where
-		B: Backend<Block> + Send + Sync,
-		E: client::CallExecutor<Block> + Send + Sync,
-		client::error::Error: From<<<B as Backend<Block>>::State as state_machine::backend::Backend>::Error>,
-		A: PolkadotApi + Send + Sync,
-{
-	fn transactions(&self) -> Vec<(Hash, Vec<u8>)> {
-		let best_block = match self.client.info() {
-			Ok(info) => info.chain.best_hash,
-			Err(e) => {
-				debug!("Error getting best block: {:?}", e);
-				return Vec::new();
-			}
-		};
-
-		let id = match self.api.check_id(BlockId::hash(best_block)) {
-			Ok(id) => id,
-			Err(_) => return Vec::new(),
-		};
-
-		let ready = transaction_pool::Ready::create(id, &*self.api);
-
-		self.pool.cull_and_get_pending(ready, |pending| pending
-			.map(|t| {
-				let hash = t.hash().clone();
-				(hash, t.primitive_extrinsic())
-			})
-			.collect()
-		)
-	}
-
-	fn import(&self, transaction: &Vec<u8>) -> Option<Hash> {
-		let encoded = transaction.encode();
-		if let Some(uxt) = codec::Slicable::decode(&mut &encoded[..]) {
-			match self.pool.import_unchecked_extrinsic(uxt) {
-				Ok(xt) => Some(*xt.hash()),
-				Err(e) => match *e.kind() {
-					transaction_pool::ErrorKind::AlreadyImported(hash) => Some(hash[..].into()),
-					_ => {
-						debug!("Error adding transaction to the pool: {:?}", e);
-						None
-					},
-				}
-			}
-		} else {
-			debug!("Error decoding transaction");
-			None
-		}
-	}
-
-	fn on_broadcasted(&self, propagations: HashMap<Hash, Vec<String>>) {
-		self.pool.on_broadcasted(propagations)
-	}
-}
-
 pub struct ChainConfig {
 	genesis_config: GenesisConfig,
 	boot_nodes: Vec<String>,
@@ -294,90 +227,24 @@ fn local_testnet_config() -> ChainConfig {
 	])
 }
 
-struct GenesisBuilder {
-	config: GenesisConfig,
-}
-
-impl client::GenesisBuilder<Block> for GenesisBuilder {
-	fn build(self) -> (Header, Vec<(Vec<u8>, Vec<u8>)>) {
-		let storage = self.config.build_externalities();
-		let block = genesis::construct_genesis_block::<Block>(&storage);
-		(block.header, storage.into_iter().collect())
-	}
-}
-
 /// Creates light client and register protocol with the network service
-pub fn new_light(config: Configuration)
-	-> Result<
-		Service<
-			client::light::Backend<Block>,
-			client::RemoteCallExecutor<client::light::Backend<Block>, network::OnDemand<Block, network::Service<Block>>>
-		>,
-		error::Error,
-	> {
-	Service::new(move |_, executor, genesis_builder: GenesisBuilder| {
-			let client_backend = client::light::new_light_backend();
-			let fetch_checker = Arc::new(client::light::new_fetch_checker(client_backend.clone(), executor));
-			let fetcher = Arc::new(network::OnDemand::new(fetch_checker));
-			let client = client::light::new_light(client_backend, fetcher.clone(), genesis_builder)?;
-			Ok((Arc::new(client), Some(fetcher)))
-		},
-		|client| Arc::new(polkadot_api::light::RemotePolkadotApiWrapper(client.clone())),
-		|_client, _network, _tx_pool, _keystore| Ok(None),
-		config)
+pub fn new_light(config: Configuration) -> Result<Service<components::LightComponents>, error::Error> {
+	Service::new(components::LightComponents, config)
 }
 
 /// Creates full client and register protocol with the network service
-pub fn new_full(config: Configuration) -> Result<Service<client_db::Backend<Block>, client::LocalCallExecutor<client_db::Backend<Block>, CodeExecutor>>, error::Error> {
+pub fn new_full(config: Configuration) -> Result<Service<components::FullComponents>, error::Error> {
 	let is_validator = (config.roles & Role::VALIDATOR) == Role::VALIDATOR;
-	Service::new(|db_settings, executor, genesis_builder: GenesisBuilder|
-		Ok((Arc::new(client_db::new_client(db_settings, executor, genesis_builder)?), None)),
-		|client| client,
-		|client, network, tx_pool, keystore| {
-			if !is_validator {
-				return Ok(None);
-			}
-
-			// Load the first available key. Code above makes sure it exisis.
-			let key = keystore.load(&keystore.contents()?[0], "")?;
-			info!("Using authority key {:?}", key.public());
-			Ok(Some(consensus::Service::new(
-				client.clone(),
-				client.clone(),
-				network.clone(),
-				tx_pool.clone(),
-				::std::time::Duration::from_millis(4000), // TODO: dynamic
-				key,
-			)))
-		},
-		config)
+	Service::new(components::FullComponents { is_validator }, config)
 }
 
-impl<B, E> Service<B, E>
+impl<Components> Service<Components>
 	where
-		B: Backend<Block> + Send + Sync + 'static,
-		E: CallExecutor<Block> + Send + Sync + 'static,
-		client::error::Error: From<<<B as Backend<Block>>::State as state_machine::backend::Backend>::Error>
+		Components: components::Components,
+		client::error::Error: From<<<<Components as components::Components>::Backend as client::backend::Backend<Block>>::State as state_machine::Backend>::Error>,
 {
 	/// Creates and register protocol with the network service
-	fn new<F, G, C, A>(client_creator: F, api_creator: G, consensus_creator: C, mut config: Configuration) -> Result<Self, error::Error>
-		where
-			F: FnOnce(
-					client_db::DatabaseSettings,
-					CodeExecutor,
-					GenesisBuilder,
-				) -> Result<(Arc<Client<B, E, Block>>, Option<Arc<network::OnDemand<Block, network::Service<Block>>>>), error::Error>,
-			G: Fn(
-					Arc<Client<B, E, Block>>,
-				) -> Arc<A>,
-			C: Fn(
-					Arc<Client<B, E, Block>>,
-					Arc<network::Service<Block>>,
-					Arc<TransactionPool>,
-					&Keystore
-				) -> Result<Option<consensus::Service>, error::Error>,
-			A: PolkadotApi + Send + Sync + 'static,
-	{
+	fn new(components: Components, mut config: Configuration) -> Result<Self, error::Error> {
 		use std::sync::Barrier;
 
 		let (signal, exit) = ::exit_future::signal();
@@ -402,7 +269,7 @@ impl<B, E> Service<B, E>
 		};
 		config.network.boot_nodes.extend(boot_nodes);
 
-		let genesis_builder = GenesisBuilder {
+		let genesis_builder = components::GenesisBuilder {
 			config: genesis_config,
 		};
 
@@ -411,17 +278,15 @@ impl<B, E> Service<B, E>
 			path: config.database_path.into(),
 		};
 
-		let (client, on_demand) = client_creator(db_settings, executor, genesis_builder)?;
-		let api = api_creator(client.clone());
+		let (client, on_demand) = components.build_client(db_settings, executor, genesis_builder)?;
+		let api = components.build_api(client.clone());
 		let best_header = client.best_block_header()?;
+
 		info!("Best block is #{}", best_header.number);
 		telemetry!("node.start"; "height" => best_header.number, "best" => ?best_header.hash());
+
 		let transaction_pool = Arc::new(TransactionPool::new(config.transaction_pool));
-		let transaction_pool_adapter = Arc::new(TransactionPoolAdapter {
-			pool: transaction_pool.clone(),
-			client: client.clone(),
-			api: api.clone(),
-		});
+		let transaction_pool_adapter = components.build_network_tx_pool(client.clone(), api.clone(), transaction_pool.clone());
 		let network_params = network::Params {
 			config: network::ProtocolConfig {
 				roles: config.roles,
@@ -481,7 +346,7 @@ impl<B, E> Service<B, E>
 		barrier.wait();
 
 		// Spin consensus service if configured
-		let consensus_service = consensus_creator(client.clone(), network.clone(), transaction_pool.clone(), &keystore)?;
+		let consensus_service = components.build_consensus(client.clone(), network.clone(), transaction_pool.clone(), &keystore)?;
 
 		Ok(Service {
 			thread: Some(thread),
@@ -494,7 +359,7 @@ impl<B, E> Service<B, E>
 	}
 
 	/// Get shared client instance.
-	pub fn client(&self) -> Arc<Client<B, E, Block>> {
+	pub fn client(&self) -> Arc<Client<Components::Backend, Components::Executor, Block>> {
 		self.client.clone()
 	}
 
@@ -523,7 +388,7 @@ pub fn prune_imported<A>(api: &A, pool: &TransactionPool, hash: Hash)
 	}
 }
 
-impl<B, E> Drop for Service<B, E> {
+impl<Components> Drop for Service<Components> where Components: components::Components {
 	fn drop(&mut self) {
 		self.network.stop_network();