Handle AccountIndices in transaction pool

demo/cli/src/lib.rs

@@ -50,15 +50,15 @@ pub mod error;
 use std::sync::Arc;
 use demo_primitives::Hash;
 use demo_runtime::{GenesisConfig, ConsensusConfig, CouncilConfig, DemocracyConfig,
- SessionConfig, StakingConfig, BuildStorage};
-use demo_runtime::{Block, UncheckedExtrinsic};
+ SessionConfig, StakingConfig};
+use demo_runtime::{Block, BlockId, UncheckedExtrinsic, BuildStorage};
 use futures::{Future, Sink, Stream};
 
 struct DummyPool;
-impl extrinsic_pool::api::ExtrinsicPool<UncheckedExtrinsic, Hash> for DummyPool {
+impl extrinsic_pool::api::ExtrinsicPool<UncheckedExtrinsic, BlockId, Hash> for DummyPool {
  type Error = extrinsic_pool::txpool::Error;
 
- fn submit(&self, _: Vec<UncheckedExtrinsic>)
+ fn submit(&self, _block: BlockId, _: Vec<UncheckedExtrinsic>)
  -> Result<Vec<Hash>, Self::Error>
  {
  Err("unimplemented".into())
@@ -155,7 +155,8 @@ pub fn run<I, T>(args: I) -> error::Result<()> where
  let _rpc_servers = {
  let handler = || {
  let chain = rpc::apis::chain::Chain::new(client.clone(), core.remote());
- rpc::rpc_handler::<Block, _, _, _, _>(client.clone(), chain, Arc::new(DummyPool), DummySystem)
+ let author = rpc::apis::author::Author::new(client.clone(), Arc::new(DummyPool));
+ rpc::rpc_handler::<Block, _, _, _, _>(client.clone(), chain, author, DummySystem)
  };
  let http_address = "127.0.0.1:9933".parse().unwrap();
  let ws_address = "127.0.0.1:9944".parse().unwrap();

demo/runtime/src/lib.rs

@@ -162,6 +162,8 @@ pub type Address = staking::Address<Concrete>;
 pub type Header = generic::Header<BlockNumber, BlakeTwo256, Vec<u8>>;
 /// Block type as expected by this runtime.
 pub type Block = generic::Block<Header, UncheckedExtrinsic>;
+/// BlockId type as expected by this runtime.
+pub type BlockId = generic::BlockId<Block>;
 /// Unchecked extrinsic type as expected by this runtime.
 pub type UncheckedExtrinsic = generic::UncheckedExtrinsic<Address, Index, Call, Signature>;
 /// Extrinsic type as expected by this runtime. This is not the type that is signed.

polkadot/cli/Cargo.toml

@@ -27,12 +27,12 @@ parking_lot = "0.4"
 serde_json = "1.0"
 serde = "1.0"
 substrate-client = { path = "../../substrate/client" }
-substrate-state-machine = { path = "../../substrate/state-machine" }
-substrate-rpc = { path = "../../substrate/rpc" }
-substrate-rpc-servers = { path = "../../substrate/rpc-servers" }
 substrate-network = { path = "../../substrate/network" }
 substrate-primitives = { path = "../../substrate/primitives" }
+substrate-rpc = { path = "../../substrate/rpc" }
+substrate-rpc-servers = { path = "../../substrate/rpc-servers" }
 substrate-runtime-primitives = { path = "../../substrate/runtime/primitives" }
+substrate-state-machine = { path = "../../substrate/state-machine" }
 substrate-telemetry = { path = "../../substrate/telemetry" }
 polkadot-primitives = { path = "../primitives" }
 polkadot-runtime = { path = "../runtime" }

polkadot/cli/src/lib.rs

@@ -34,13 +34,13 @@ extern crate parking_lot;
 extern crate serde;
 extern crate serde_json;
 
-extern crate substrate_primitives;
-extern crate substrate_state_machine as state_machine;
 extern crate substrate_client as client;
 extern crate substrate_network as network;
+extern crate substrate_primitives;
 extern crate substrate_rpc;
 extern crate substrate_rpc_servers as rpc;
 extern crate substrate_runtime_primitives as runtime_primitives;
+extern crate substrate_state_machine as state_machine;
 extern crate polkadot_primitives;
 extern crate polkadot_runtime;
 extern crate polkadot_service as service;
@@ -192,12 +192,12 @@ pub fn run<I, T>(args: I) -> error::Result<()> where
 
  let (mut genesis_storage, boot_nodes) = PresetConfig::from_spec(chain_spec)
  .map(PresetConfig::deconstruct)
- .unwrap_or_else(|f| (Box::new(move || 
+ .unwrap_or_else(|f| (Box::new(move ||
  read_storage_json(&f)
  .map(|s| { info!("{} storage items read from {}", s.len(), f); s })
  .unwrap_or_else(|| panic!("Bad genesis state file: {}", f))
  ), vec![]));
- 
+
  if matches.is_present("build-genesis") {
  info!("Building genesis");
  for (i, (k, v)) in genesis_storage().iter().enumerate() {
@@ -285,10 +285,11 @@ fn run_until_exit<C>(mut core: reactor::Core, service: service::Service<C>, matc
 
  let handler = || {
  let chain = rpc::apis::chain::Chain::new(service.client(), core.remote());
+ let author = rpc::apis::author::Author::new(service.client(), service.transaction_pool());
  rpc::rpc_handler::<Block, _, _, _, _>(
  service.client(),
  chain,
- service.transaction_pool(),
+ author,
  sys_conf.clone(),
  )
  };

polkadot/consensus/src/lib.rs

@@ -74,7 +74,7 @@ use polkadot_primitives::{Hash, Block, BlockId, BlockNumber, Header, Timestamp};
 use polkadot_primitives::parachain::{Id as ParaId, Chain, DutyRoster, BlockData, Extrinsic as ParachainExtrinsic, CandidateReceipt};
 use polkadot_runtime::BareExtrinsic;
 use primitives::AuthorityId;
-use transaction_pool::{Ready, TransactionPool};
+use transaction_pool::{TransactionPool};
 use tokio_core::reactor::{Handle, Timeout, Interval};
 
 use futures::prelude::*;
@@ -226,7 +226,7 @@ pub struct ProposerFactory<C, N, P> {
  /// The client instance.
  pub client: Arc<C>,
  /// The transaction pool.
- pub transaction_pool: Arc<TransactionPool>,
+ pub transaction_pool: Arc<TransactionPool<C>>,
  /// The backing network handle.
  pub network: N,
  /// Parachain collators.
@@ -239,7 +239,8 @@ pub struct ProposerFactory<C, N, P> {
 
 impl<C, N, P> bft::ProposerFactory<Block> for ProposerFactory<C, N, P>
  where
- C: PolkadotApi,
+ C: PolkadotApi + Send + Sync,
+ C::CheckedBlockId: Sync,
  N: Network,
  P: Collators,
 {
@@ -319,12 +320,13 @@ pub struct Proposer<C: PolkadotApi, R, P> {
  random_seed: Hash,
  router: R,
  table: Arc<SharedTable>,
- transaction_pool: Arc<TransactionPool>,
+ transaction_pool: Arc<TransactionPool<C>>,
 }
 
 impl<C, R, P> bft::Proposer<Block> for Proposer<C, R, P>
  where
- C: PolkadotApi,
+ C: PolkadotApi + Send + Sync,
+ C::CheckedBlockId: Sync,
  R: TableRouter,
  P: Collators,
 {
@@ -501,16 +503,19 @@ impl<C, R, P> bft::Proposer<Block> for Proposer<C, R, P>
 
  let local_id = self.local_key.public().0.into();
  let mut next_index = {
- let readiness_evaluator = Ready::create(self.parent_id.clone(), &*self.client);
- let cur_index = self.transaction_pool.cull_and_get_pending(readiness_evaluator, |pending| pending
+ let cur_index = self.transaction_pool.cull_and_get_pending(BlockId::hash(self.parent_hash), |pending| pending
  .filter(|tx| tx.sender().map(|s| s == local_id).unwrap_or(false))
  .last()
  .map(|tx| Ok(tx.index()))
  .unwrap_or_else(|| self.client.index(&self.parent_id, local_id))
  );
 
  match cur_index {
- Ok(cur_index) => cur_index + 1,
+ Ok(Ok(cur_index)) => cur_index + 1,
+ Ok(Err(e)) => {
+ warn!(target: "consensus", "Error computing next transaction index: {}", e);
+ return;
+ }
  Err(e) => {
  warn!(target: "consensus", "Error computing next transaction index: {}", e);
  return;
@@ -549,7 +554,7 @@ impl<C, R, P> bft::Proposer<Block> for Proposer<C, R, P>
  };
  let uxt = UncheckedExtrinsic::new(extrinsic, signature);
 
- self.transaction_pool.import_unchecked_extrinsic(uxt)
+ self.transaction_pool.import_unchecked_extrinsic(BlockId::hash(self.parent_hash), uxt)
  .expect("locally signed extrinsic is valid; qed");
  }
  }
@@ -618,7 +623,7 @@ pub struct CreateProposal<C: PolkadotApi, R, P: Collators> {
  parent_number: BlockNumber,
  parent_id: C::CheckedBlockId,
  client: Arc<C>,
- transaction_pool: Arc<TransactionPool>,
+ transaction_pool: Arc<TransactionPool<C>>,
  collation: CollationFetch<P, C>,
  router: R,
  table: Arc<SharedTable>,
@@ -640,9 +645,8 @@ impl<C, R, P> CreateProposal<C, R, P>
  let mut block_builder = self.client.build_block(&self.parent_id, timestamp, candidates)?;
 
  {
- let readiness_evaluator = Ready::create(self.parent_id.clone(), &*self.client);
  let mut unqueue_invalid = Vec::new();
- self.transaction_pool.cull_and_get_pending(readiness_evaluator, |pending_iterator| {
+ let result = self.transaction_pool.cull_and_get_pending(BlockId::hash(self.parent_hash), |pending_iterator| {
  let mut pending_size = 0;
  for pending in pending_iterator {
  // skip and cull transactions which are too large.
@@ -664,6 +668,9 @@ impl<C, R, P> CreateProposal<C, R, P>
  }
  }
  });
+ if let Err(e) = result {
+ warn!("Unable to get the pending set: {:?}", e);
+ }
 
  self.transaction_pool.remove(&unqueue_invalid, false);
  }

polkadot/consensus/src/service.rs

@@ -235,13 +235,14 @@ impl Service {
  client: Arc<C>,
  api: Arc<A>,
  network: Arc<net::ConsensusService<Block>>,
- transaction_pool: Arc<TransactionPool>,
+ transaction_pool: Arc<TransactionPool<A>>,
  parachain_empty_duration: Duration,
  key: ed25519::Pair,
  ) -> Service
  where
  A: LocalPolkadotApi + Send + Sync + 'static,
  C: BlockchainEvents<Block> + ChainHead<Block> + bft::BlockImport<Block> + bft::Authorities<Block> + Send + Sync + 'static,
+ A::CheckedBlockId: Sync,
  {
  let (signal, exit) = ::exit_future::signal();
  let thread = thread::spawn(move || {

polkadot/service/src/components.rs

@@ -55,11 +55,11 @@ pub trait Components {
  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>)
+ fn build_network_tx_pool(&self, client: Arc<client::Client<Self::Backend, Self::Executor, Block>>, tx_pool: Arc<TransactionPool<Self::Api>>)
  -> 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)
+ fn build_consensus(&self, client: Arc<Client<Self::Backend, Self::Executor, Block>>, network: Arc<network::Service<Block>>, tx_pool: Arc<TransactionPool<Self::Api>>, keystore: &Keystore)
  -> Result<Option<consensus::Service>, error::Error>;
 }
 
@@ -83,17 +83,16 @@ impl Components for FullComponents {
  client
  }
 
- fn build_network_tx_pool(&self, client: Arc<client::Client<Self::Backend, Self::Executor, Block>>, api: Arc<Self::Api>, pool: Arc<TransactionPool>)
+ fn build_network_tx_pool(&self, client: Arc<client::Client<Self::Backend, Self::Executor, Block>>, pool: Arc<TransactionPool<Self::Api>>)
  -> 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)
+ fn build_consensus(&self, client: Arc<client::Client<Self::Backend, Self::Executor, Block>>, network: Arc<network::Service<Block>>, tx_pool: Arc<TransactionPool<Self::Api>>, keystore: &Keystore)
  -> Result<Option<consensus::Service>, error::Error> {
  if !self.is_validator {
  return Ok(None);
@@ -134,17 +133,16 @@ impl Components for LightComponents {
  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>)
+ fn build_network_tx_pool(&self, client: Arc<client::Client<Self::Backend, Self::Executor, Block>>, pool: Arc<TransactionPool<Self::Api>>)
  -> 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)
+ fn build_consensus(&self, _client: Arc<client::Client<Self::Backend, Self::Executor, Block>>, _network: Arc<network::Service<Block>>, _tx_pool: Arc<TransactionPool<Self::Api>>, _keystore: &Keystore)
  -> Result<Option<consensus::Service>, error::Error> {
  Ok(None)
  }
@@ -153,9 +151,25 @@ impl Components for LightComponents {
 /// Transaction pool adapter.
 pub struct TransactionPoolAdapter<B, E, A> where A: Send + Sync, E: Send + Sync {
  imports_external_transactions: bool,
- pool: Arc<TransactionPool>,
+ pool: Arc<TransactionPool<A>>,
  client: Arc<Client<B, E, Block>>,
- api: Arc<A>,
+}
+
+impl<B, E, A> TransactionPoolAdapter<B, E, A>
+ where
+ A: Send + Sync,
+ 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>,
+{
+ fn best_block_id(&self) -> Option<BlockId> {
+ self.client.info()
+ .map(|info| BlockId::hash(info.chain.best_hash))
+ .map_err(|e| {
+ debug!("Error getting best block: {:?}", e);
+ })
+ .ok()
+ }
 }
 
 impl<B, E, A> network::TransactionPool<Block> for TransactionPoolAdapter<B, E, A>
@@ -166,28 +180,20 @@ impl<B, E, A> network::TransactionPool<Block> for TransactionPoolAdapter<B, E, A
  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 best_block_id = match self.best_block_id() {
+ Some(id) => id,
+ None => return vec![],
  };
-
- let ready = transaction_pool::Ready::create(id, &*self.api);
-
- self.pool.cull_and_get_pending(ready, |pending| pending
+ self.pool.cull_and_get_pending(best_block_id, |pending| pending
  .map(|t| {
  let hash = t.hash().clone();
  (hash, t.primitive_extrinsic())
  })
  .collect()
- )
+ ).unwrap_or_else(|e| {
+ warn!("Error retrieving pending set: {}", e);
+ vec![]
+ })
  }
 
  fn import(&self, transaction: &Vec<u8>) -> Option<Hash> {
@@ -197,7 +203,8 @@ impl<B, E, A> network::TransactionPool<Block> for TransactionPoolAdapter<B, E, A
 
  let encoded = transaction.encode();
  if let Some(uxt) = codec::Slicable::decode(&mut &encoded[..]) {
- match self.pool.import_unchecked_extrinsic(uxt) {
+ let best_block_id = self.best_block_id()?;
+ match self.pool.import_unchecked_extrinsic(best_block_id, uxt) {
  Ok(xt) => Some(*xt.hash()),
  Err(e) => match *e.kind() {
  transaction_pool::ErrorKind::AlreadyImported(hash) => Some(hash[..].into()),