implement custom proposer

Cargo.toml

@@ -43,6 +43,7 @@ members = [
 	"service",
 	"validation",
 
+	"node/core/proposer",
 	"node/network/bridge",
 	"node/overseer",
 	"node/primitives",

node/core/proposer/Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "polkadot-node-core-proposer"
+version = "0.1.0"
+authors = ["Parity Technologies <[email protected]>"]
+edition = "2018"
+
+[dependencies]
+futures = "0.3.4"
+futures-timer = "3.0.1"
+log = "0.4.8"
+parity-scale-codec = "1.3.0"
+polkadot-node-subsystem = { path = "../../subsystem" }
+polkadot-overseer = { path = "../../overseer" }
+polkadot-primitives = { path = "../../../primitives" }
+sc-basic-authorship = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sc-block-builder = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sc-client-api = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sc-telemetry = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-api = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-blockchain = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-consensus = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-inherents = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-transaction-pool = { git = "https://github.com/paritytech/substrate", branch = "master" }
+tokio-executor = { version = "0.2.0-alpha.6", features = ["blocking"] }
+wasm-timer = "0.2.4"

node/core/proposer/src/lib.rs

@@ -0,0 +1,215 @@
+use futures::prelude::*;
+use futures::select;
+use polkadot_node_subsystem::{messages::{AllMessages, ProvisionerInherentData, ProvisionerMessage}, SubsystemError};
+use polkadot_overseer::OverseerHandler;
+use polkadot_primitives::{
+	inclusion_inherent,
+	parachain::ParachainHost,
+	Block, Header,
+};
+use sc_block_builder::{BlockBuilderApi, BlockBuilderProvider};
+use sp_api::{ApiExt, ProvideRuntimeApi};
+use sp_blockchain::HeaderBackend;
+use sp_consensus::{Proposal, RecordProof};
+use sp_inherents::InherentData;
+use sp_runtime::traits::{DigestFor, HashFor};
+use sp_transaction_pool::TransactionPool;
+use std::{fmt, pin::Pin, sync::Arc, time};
+
+/// How long proposal can take before we give up and err out
+const PROPOSE_TIMEOUT: core::time::Duration = core::time::Duration::from_secs(2);
+
+/// Custom Proposer factory for Polkadot
+pub struct ProposerFactory<TxPool, Backend, Client> {
+	inner: sc_basic_authorship::ProposerFactory<TxPool, Backend, Client>,
+	overseer: OverseerHandler,
+}
+
+impl<TxPool, Backend, Client> ProposerFactory<TxPool, Backend, Client> {
+	pub fn new(
+		client: Arc<Client>,
+		transaction_pool: Arc<TxPool>,
+		overseer: OverseerHandler,
+	) -> Self {
+		ProposerFactory {
+			inner: sc_basic_authorship::ProposerFactory::new(
+				client,
+				transaction_pool,
+				None,
+			),
+			overseer,
+		}
+	}
+}
+
+impl<TxPool, Backend, Client> sp_consensus::Environment<Block>
+	for ProposerFactory<TxPool, Backend, Client>
+where
+	TxPool: 'static + TransactionPool<Block = Block>,
+	Client: 'static
+		+ BlockBuilderProvider<Backend, Block, Client>
+		+ ProvideRuntimeApi<Block>
+		+ HeaderBackend<Block>
+		+ Send
+		+ Sync,
+	Client::Api:
+		ParachainHost<Block> + BlockBuilderApi<Block> + ApiExt<Block, Error = sp_blockchain::Error>,
+	Backend:
+		'static + sc_client_api::Backend<Block, State = sp_api::StateBackendFor<Client, Block>>,
+	// Rust bug: https://github.com/rust-lang/rust/issues/24159
+	sp_api::StateBackendFor<Client, Block>: sp_api::StateBackend<HashFor<Block>> + Send,
+{
+	type CreateProposer = Pin<Box<
+		dyn Future<Output = Result<Self::Proposer, Self::Error>> + Send + 'static,
+	>>;
+	type Proposer = Proposer<TxPool, Backend, Client>;
+	type Error = Error;
+
+	fn init(&mut self, parent_header: &Header) -> Self::CreateProposer {
+		// create the inner proposer
+		let proposer = self.inner.init(parent_header).into_inner();
+
+		// clone the overseer handle (lightweight) so it can be moved into the future
+		let mut overseer = self.overseer.clone();
+
+		// we know that this function will be called at least once per proposed block,
+		// because this is where the parent header is supplied. Therefore, we can send
+		// the request for authorship data here.
+		let (sender, receiver) = futures::channel::oneshot::channel();
+		let parent_header_hash = parent_header.hash();
+
+		async move {
+			overseer.send_msg(AllMessages::Provisioner(
+				ProvisionerMessage::RequestInherentData(parent_header_hash, sender),
+			)).await?;
+			Ok(Proposer {
+				inner: proposer?,
+				provisioner_inherent_data: receiver,
+			})
+		}.boxed()
+	}
+}
+
+pub struct Proposer<TxPool: TransactionPool<Block = Block>, Backend, Client> {
+	inner: sc_basic_authorship::Proposer<Backend, Block, Client, TxPool>,
+	provisioner_inherent_data: futures::channel::oneshot::Receiver<ProvisionerInherentData>,
+}
+
+impl<TxPool, Backend, Client> sp_consensus::Proposer<Block> for Proposer<TxPool, Backend, Client>
+where
+	TxPool: 'static + TransactionPool<Block = Block>,
+	Client: 'static
+		+ BlockBuilderProvider<Backend, Block, Client>
+		+ ProvideRuntimeApi<Block>
+		+ HeaderBackend<Block>
+		+ Send
+		+ Sync,
+	Client::Api:
+		ParachainHost<Block> + BlockBuilderApi<Block> + ApiExt<Block, Error = sp_blockchain::Error>,
+	Backend:
+		'static + sc_client_api::Backend<Block, State = sp_api::StateBackendFor<Client, Block>>,
+	// Rust bug: https://github.com/rust-lang/rust/issues/24159
+	sp_api::StateBackendFor<Client, Block>: sp_api::StateBackend<HashFor<Block>> + Send,
+{
+	type Transaction = sc_client_api::TransactionFor<Backend, Block>;
+	type Proposal = Pin<Box<
+		dyn Future<Output = Result<Proposal<Block, sp_api::TransactionFor<Client, Block>>, Error>> + Send,
+	>>;
+	type Error = Error;
+
+	fn propose(
+		self,
+		mut inherent_data: InherentData,
+		inherent_digests: DigestFor<Block>,
+		max_duration: time::Duration,
+		record_proof: RecordProof,
+	) -> Self::Proposal {
+		let mut proposal = async move {
+			let provisioner_inherent_data = self.provisioner_inherent_data.await.map_err(Error::ClosedChannelFromProvisioner)?;
+
+			inherent_data.put_data(
+				inclusion_inherent::INHERENT_IDENTIFIER,
+				&provisioner_inherent_data,
+			)?;
+
+			self.inner
+				.propose(inherent_data, inherent_digests, max_duration, record_proof)
+				.await
+				.map_err(Into::into)
+		}
+		.boxed()
+		.fuse();
+
+		let mut timeout = wasm_timer::Delay::new(PROPOSE_TIMEOUT).fuse();
+
+		async move {
+			select! {
+				proposal_result = proposal => proposal_result,
+				_ = timeout => Err(Error::Timeout),
+			}
+		}
+		.boxed()
+	}
+}
+
+// It would have been more ergonomic to use thiserror to derive the
+// From implementations, Display, and std::error::Error, but unfortunately
+// two of the wrapped errors (sp_inherents::Error, SubsystemError) also
+// don't impl std::error::Error, which breaks the thiserror derive.
+#[derive(Debug)]
+pub enum Error {
+	Consensus(sp_consensus::Error),
+	Blockchain(sp_blockchain::Error),
+	Inherent(sp_inherents::Error),
+	Timeout,
+	ClosedChannelFromProvisioner(futures::channel::oneshot::Canceled),
+	Subsystem(SubsystemError)
+}
+
+impl From<sp_consensus::Error> for Error {
+	fn from(e: sp_consensus::Error) -> Error {
+		Error::Consensus(e)
+	}
+}
+
+impl From<sp_blockchain::Error> for Error {
+	fn from(e: sp_blockchain::Error) -> Error {
+		Error::Blockchain(e)
+	}
+}
+
+impl From<sp_inherents::Error> for Error {
+	fn from(e: sp_inherents::Error) -> Error {
+		Error::Inherent(e)
+	}
+}
+
+impl From<SubsystemError> for Error {
+	fn from(e: SubsystemError) -> Error {
+		Error::Subsystem(e)
+	}
+}
+
+impl fmt::Display for Error {
+	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+		match self {
+			Self::Consensus(err) => write!(f, "consensus error: {}", err),
+			Self::Blockchain(err) => write!(f, "blockchain error: {}", err),
+			Self::Inherent(err) => write!(f, "inherent error: {:?}", err),
+			Self::Timeout => write!(f, "timeout: provisioner did not return inherent data after {:?}", PROPOSE_TIMEOUT),
+			Self::ClosedChannelFromProvisioner(err) => write!(f, "provisioner closed inherent data channel before sending: {}", err),
+			Self::Subsystem(err) => write!(f, "subsystem error: {:?}", err),
+		}
+	}
+}
+
+impl std::error::Error for Error {
+	fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+		match self {
+			Self::Consensus(err) => Some(err),
+			Self::Blockchain(err) => Some(err),
+			Self::ClosedChannelFromProvisioner(err) => Some(err),
+			_ => None
+		}
+	}
+}

node/service/Cargo.toml

@@ -20,6 +20,7 @@ kusama-runtime = { path = "../../runtime/kusama" }
 westend-runtime = { path = "../../runtime/westend" }
 polkadot-network = { path = "../../network", optional = true }
 polkadot-rpc = { path = "../../rpc" }
+polkadot-node-core-proposer = { path = "../core/proposer" }
 sp-io = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-api = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "master" }
@@ -56,7 +57,6 @@ sp-session = { git = "https://github.com/paritytech/substrate", branch = "master
 sp-offchain = { package = "sp-offchain", git = "https://github.com/paritytech/substrate", branch = "master" }
 prometheus-endpoint = { package = "substrate-prometheus-endpoint", git = "https://github.com/paritytech/substrate", branch = "master" }
 frame-benchmarking = { git = "https://github.com/paritytech/substrate", branch = "master" }
-sc-basic-authorship = { git = "https://github.com/paritytech/substrate", branch = "master" }
 
 [dev-dependencies]
 polkadot-test-runtime-client = { path = "../../runtime/test-runtime/client" }

node/service/src/lib.rs

@@ -34,6 +34,7 @@ use polkadot_subsystem::{
 	Subsystem, SubsystemContext, SpawnedSubsystem,
 	messages::{CandidateValidationMessage, CandidateBackingMessage},
 };
+use polkadot_node_core_proposer::ProposerFactory;
 pub use service::{
 	Role, PruningMode, TransactionPoolOptions, Error, RuntimeGenesis,
 	TFullClient, TLightClient, TFullBackend, TLightBackend, TFullCallExecutor, TLightCallExecutor,
@@ -360,6 +361,7 @@ macro_rules! new_full {
 			.collect();
 
 		let (overseer, handler) = real_overseer(leaves, spawner)?;
+		let handler_clone = handler.clone();
 
 		task_manager.spawn_essential_handle().spawn_blocking("overseer", Box::pin(async move {
 			use futures::{pin_mut, select, FutureExt};
@@ -386,11 +388,10 @@ macro_rules! new_full {
 			let can_author_with =
 				consensus_common::CanAuthorWithNativeVersion::new(client.executor().clone());
 
-			// TODO: custom proposer (https://github.com/paritytech/polkadot/issues/1248)
-			let proposer = sc_basic_authorship::ProposerFactory::new(
+			let proposer = ProposerFactory::new(
 				client.clone(),
 				transaction_pool,
-				None,
+				handler_clone,
 			);
 
 			let babe_config = babe::BabeParams {

node/subsystem/src/messages.rs

@@ -28,7 +28,7 @@ use sc_network::{ObservedRole, ReputationChange, PeerId};
 use polkadot_primitives::{BlockNumber, Hash, Signature};
 use polkadot_primitives::parachain::{
 	AbridgedCandidateReceipt, PoVBlock, ErasureChunk, BackedCandidate, Id as ParaId,
-	SignedAvailabilityBitfield, SigningContext, ValidatorId, ValidationCode, ValidatorIndex,
+	SignedAvailabilityBitfield, SignedAvailabilityBitfields, SigningContext, ValidatorId, ValidationCode, ValidatorIndex,
 };
 use polkadot_node_primitives::{
 	MisbehaviorReport, SignedFullStatement, View, ProtocolId,
@@ -190,6 +190,11 @@ pub enum ProvisionableData {
 	Dispute(Hash, Signature),
 }
 
+/// This data needs to make its way from the provisioner into the InherentData.
+///
+/// There, it is used to construct the InclusionInherent.
+pub type ProvisionerInherentData = (SignedAvailabilityBitfields, Vec<BackedCandidate>);
+
 /// Message to the Provisioner.
 ///
 /// In all cases, the Hash is that of the relay parent.
@@ -198,6 +203,12 @@ pub enum ProvisionerMessage {
 	/// This message allows potential block authors to be kept updated with all new authorship data
 	/// as it becomes available.
 	RequestBlockAuthorshipData(Hash, mpsc::Sender<ProvisionableData>),
+	/// This message allows external subsystems to request the set of bitfields and backed candidates
+	/// associated with a particular potential block hash.
+	///
+	/// This is expected to be used by a proposer, to inject that information into the InherentData
+	/// where it can be assembled into the InclusionInherent.
+	RequestInherentData(Hash, oneshot::Sender<ProvisionerInherentData>),
 	/// This data should become part of a relay chain block
 	ProvisionableData(ProvisionableData),
 }

roadmap/implementors-guide/src/node/utility/provisioner.md

@@ -38,6 +38,12 @@ At initialization, this subsystem has no outputs. Block authors can send a `Prov
 
 Note that block authors must re-send a `ProvisionerMessage::RequestBlockAuthorshipData` for each relay parent they are interested in receiving provisionable data for.
 
+## Block Production
+
+When a validator is selected by BABE to author a block, it becomes a block producer. The provisioner is the subsystem best suited to choosing which specific backed candidates and availability bitfields should be assembled into the block. To engage this functionality, a `ProvisionerMessage::RequestInherentData` is sent; the response is a set of non-conflicting candidates and the appropriate bitfields. Non-conflicting generally means that there are never two distinct parachain candidates included for the same parachain.
+
+One might ask: given `ProvisionerMessage::RequestInherentData`, what's the point of `ProvisionerMessage::RequestBlockAuthorshipData`? The answer is that the block authorship data includes more information than is present in the inherent data; disputes, for example.
+
 ## Functionality
 
 The subsystem should maintain a set of handles to Block Authorship Provisioning Jobs that are currently live.