Integrate transaction pool to the proposal logic (#80)

* reshuffle consensus libraries

* polkadot-useful type definitions for statement table

* begin BftService

* primary selection logic

* bft service implementation without I/O

* extract out `BlockImport` trait

* allow bft primitives to compile on wasm

* Block builder (substrate)

* take polkadot-consensus down to the core.

* test for preemption

* fix test build

* Fix wasm build

* Bulid on any block

* Test for block builder.

* Block import tests for client.

* Tidy ups

* clean up block builder instantiation

* justification verification logic

* JustifiedHeader and import

* Propert block generation for tests

* network and tablerouter trait

* use statement import to drive creation of further statements

* Fixed rpc tests

* custom error type for consensus

* create proposer

* asynchronous proposal evaluation

* inherent transactions in polkadot runtime

* fix tests to match real polkadot block constraints

* implicitly generate inherent functions

* add inherent transaction functionality to block body

* block builder logic for polkadot

* some tests for the polkadot API

* avoid redundancy in native code compatibility check

* helper for extracting nonce

* transaction pool implementation

* transaction pool

* integrate transaction pool with proposer

* indentation

* kill storage keys module

* accept new transactions to replace old

polkadot/consensus/Cargo.toml

@@ -13,6 +13,7 @@ polkadot-api = { path = "../api" }
 polkadot-collator = { path = "../collator" }
 polkadot-primitives = { path = "../primitives" }
 polkadot-statement-table = { path = "../statement-table" }
+polkadot-transaction-pool = { path = "../transaction-pool" }
 substrate-bft = { path = "../../substrate/bft" }
 substrate-codec = { path = "../../substrate/codec" }
 substrate-primitives = { path = "../../substrate/primitives" }

polkadot/consensus/src/error.rs

@@ -41,6 +41,13 @@ error_chain! {
 			description("Proposal had wrong parent hash."),
 			display("Proposal had wrong parent hash. Expected {:?}, got {:?}", expected, got),
 		}
+		ProposalTooLarge(size: usize) {
+			description("Proposal exceeded the maximum size."),
+			display(
+				"Proposal exceeded the maximum size of {} by {} bytes.",
+				::MAX_TRANSACTIONS_SIZE, ::MAX_TRANSACTIONS_SIZE.saturating_sub(*size)
+			),
+		}
 	}
 }
 

polkadot/consensus/src/lib.rs

@@ -37,6 +37,7 @@ extern crate polkadot_api;
 extern crate polkadot_collator as collator;
 extern crate polkadot_statement_table as table;
 extern crate polkadot_primitives;
+extern crate polkadot_transaction_pool as transaction_pool;
 extern crate substrate_bft as bft;
 extern crate substrate_codec as codec;
 extern crate substrate_primitives as primitives;
@@ -56,6 +57,7 @@ use polkadot_primitives::block::Block as PolkadotBlock;
 use polkadot_primitives::parachain::{Id as ParaId, DutyRoster, BlockData, Extrinsic, CandidateReceipt};
 use primitives::block::{Block as SubstrateBlock, Header as SubstrateHeader, HeaderHash, Id as BlockId};
 use primitives::AuthorityId;
+use transaction_pool::TransactionPool;
 
 use futures::prelude::*;
 use futures::future;
@@ -65,6 +67,9 @@ pub use self::error::{ErrorKind, Error};
 
 mod error;
 
+// block size limit.
+const MAX_TRANSACTIONS_SIZE: usize = 4 * 1024 * 1024;
+
 /// A handle to a statement table router.
 pub trait TableRouter {
 	/// Errors when fetching data from the network.
@@ -455,6 +460,8 @@ fn make_group_info(roster: DutyRoster, authorities: &[AuthorityId]) -> Result<Ha
 pub struct ProposerFactory<C, N> {
 	/// The client instance.
 	pub client: Arc<C>,
+	/// The transaction pool.
+	pub transaction_pool: Arc<Mutex<TransactionPool>>,
 	/// The backing network handle.
 	pub network: N,
 }
@@ -465,7 +472,9 @@ impl<C: PolkadotApi, N: Network> bft::ProposerFactory for ProposerFactory<C, N>
 
 	fn init(&self, parent_header: &SubstrateHeader, authorities: &[AuthorityId], sign_with: Arc<ed25519::Pair>) -> Result<Self::Proposer, Error> {
 		let parent_hash = parent_header.hash();
-		let duty_roster = self.client.duty_roster(&BlockId::Hash(parent_hash))?;
+
+		let checked_id = self.client.check_id(BlockId::Hash(parent_hash))?;
+		let duty_roster = self.client.duty_roster(&checked_id)?;
 
 		let group_info = make_group_info(duty_roster, authorities)?;
 		let table = Arc::new(SharedTable::new(group_info, sign_with, parent_hash));
@@ -474,9 +483,11 @@ impl<C: PolkadotApi, N: Network> bft::ProposerFactory for ProposerFactory<C, N>
 		// TODO [PoC-2]: kick off collation process.
 		Ok(Proposer {
 			parent_hash,
+			parent_id: checked_id,
 			_table: table,
 			_router: router,
 			client: self.client.clone(),
+			transaction_pool: self.transaction_pool.clone(),
 		})
 	}
 }
@@ -490,9 +501,11 @@ fn current_timestamp() -> Timestamp {
 }
 
 /// The Polkadot proposer logic.
-pub struct Proposer<C, R> {
+pub struct Proposer<C: PolkadotApi, R> {
 	parent_hash: HeaderHash,
+	parent_id: C::CheckedBlockId,
 	client: Arc<C>,
+	transaction_pool: Arc<Mutex<TransactionPool>>,
 	_table: Arc<SharedTable>,
 	_router: R,
 }
@@ -503,14 +516,45 @@ impl<C: PolkadotApi, R: TableRouter> bft::Proposer for Proposer<C, R> {
 	type Evaluate = Result<bool, Error>;
 
 	fn propose(&self) -> Result<SubstrateBlock, Error> {
+		use transaction_pool::Ready;
+
 		// TODO: handle case when current timestamp behind that in state.
-		let polkadot_block = self.client.build_block(
-			&BlockId::Hash(self.parent_hash),
+		let mut block_builder = self.client.build_block(
+			&self.parent_id,
 			current_timestamp()
-		)?.bake();
+		)?;
 
-		// TODO: integrate transaction queue and `push_transaction`s.
+		let readiness_evaluator = Ready::create(self.parent_id.clone(), &*self.client);
 
+		{
+			let mut pool = self.transaction_pool.lock();
+			let mut unqueue_invalid = Vec::new();
+			let mut pending_size = 0;
+			for pending in pool.pending(readiness_evaluator.clone()) {
+				// skip and cull transactions which are too large.
+				if pending.encoded_size() > MAX_TRANSACTIONS_SIZE {
+					unqueue_invalid.push(pending.hash().clone());
+					continue
+				}
+
+				if pending_size + pending.encoded_size() >= MAX_TRANSACTIONS_SIZE { break }
+
+				match block_builder.push_transaction(pending.as_transaction().clone()) {
+					Ok(()) => {
+						pending_size += pending.encoded_size();
+					}
+					Err(_) => {
+						unqueue_invalid.push(pending.hash().clone());
+					}
+				}
+			}
+
+			for tx_hash in unqueue_invalid {
+				pool.remove(&tx_hash, false);
+			}
+		}
+
+		let polkadot_block = block_builder.bake();
 		let substrate_block = Slicable::decode(&mut polkadot_block.encode().as_slice())
 			.expect("polkadot blocks defined to serialize to substrate blocks correctly; qed");
 
@@ -519,7 +563,7 @@ impl<C: PolkadotApi, R: TableRouter> bft::Proposer for Proposer<C, R> {
 
 	// TODO: certain kinds of errors here should lead to a misbehavior report.
 	fn evaluate(&self, proposal: &SubstrateBlock) -> Result<bool, Error> {
-		evaluate_proposal(proposal, &*self.client, current_timestamp(), &self.parent_hash)
+		evaluate_proposal(proposal, &*self.client, current_timestamp(), &self.parent_hash, &self.parent_id)
 	}
 }
 
@@ -528,6 +572,7 @@ fn evaluate_proposal<C: PolkadotApi>(
 	client: &C,
 	now: Timestamp,
 	parent_hash: &HeaderHash,
+	parent_id: &C::CheckedBlockId,
 ) -> Result<bool, Error> {
 	const MAX_TIMESTAMP_DRIFT: Timestamp = 4;
 
@@ -535,6 +580,14 @@ fn evaluate_proposal<C: PolkadotApi>(
 	let proposal = PolkadotBlock::decode(&mut &encoded[..])
 		.ok_or_else(|| ErrorKind::ProposalNotForPolkadot)?;
 
+	let transactions_size = proposal.body.transactions.iter().fold(0, |a, tx| {
+		a + Slicable::encode(tx).len()
+	});
+
+	if transactions_size > MAX_TRANSACTIONS_SIZE {
+		bail!(ErrorKind::ProposalTooLarge(transactions_size))
+	}
+
 	if proposal.header.parent_hash != *parent_hash {
 		bail!(ErrorKind::WrongParentHash(*parent_hash, proposal.header.parent_hash));
 	}
@@ -551,6 +604,6 @@ fn evaluate_proposal<C: PolkadotApi>(
 	}
 
 	// execute the block.
-	client.evaluate_block(&BlockId::Hash(*parent_hash), proposal)?;
+	client.evaluate_block(parent_id, proposal)?;
 	Ok(true)
 }