Keep unpinned blocks for a while before announcing

collator/Cargo.toml

@@ -24,8 +24,9 @@ polkadot-validation = { git = "https://github.com/paritytech/polkadot", branch =
 
 # Cumulus dependencies
 cumulus-consensus = { path = "../consensus" }
-cumulus-runtime = { path = "../runtime" }
+cumulus-network = { path = "../network" }
 cumulus-primitives = { path = "../primitives" }
+cumulus-runtime = { path = "../runtime" }
 
 # Other dependencies
 log = "0.4.8"

collator/src/lib.rs

@@ -16,6 +16,7 @@
 
 //! Cumulus Collator implementation for Substrate.
 
+use cumulus_network::WaitToAnnounce;
 use cumulus_primitives::{
 	inherents::VALIDATION_FUNCTION_PARAMS_IDENTIFIER as VFP_IDENT,
 	validation_function_params::ValidationFunctionParams,
@@ -30,6 +31,7 @@ use sp_inherents::{InherentData, InherentDataProviders};
 use sp_runtime::traits::{Block as BlockT, Header as HeaderT, HashFor};
 use sp_api::{ApiExt, ProvideRuntimeApi};
 use sc_client_api::{StateBackend, UsageProvider, Finalizer, BlockchainEvents};
+use sc_service::Configuration;
 
 use polkadot_collator::{
 	BuildParachainContext, InvalidHead, Network as CollatorNetwork, ParachainContext,
@@ -44,7 +46,8 @@ use codec::{Decode, Encode};
 
 use log::{error, trace};
 
-use futures::{task::Spawn, Future, future, FutureExt};
+use futures::task::Spawn;
+use futures::prelude::*;
 
 use std::{fmt::Debug, marker::PhantomData, sync::Arc, time::Duration, pin::Pin};
 
@@ -57,28 +60,39 @@ struct HeadData<Block: BlockT> {
 }
 
 /// The implementation of the Cumulus `Collator`.
-pub struct Collator<Block, PF, BI> {
+pub struct Collator<Block: BlockT, PF, BI> {
 	proposer_factory: Arc<Mutex<PF>>,
 	_phantom: PhantomData<Block>,
 	inherent_data_providers: InherentDataProviders,
 	collator_network: Arc<dyn CollatorNetwork>,
 	block_import: Arc<Mutex<BI>>,
+	wait_to_announce: Arc<Mutex<WaitToAnnounce<Block>>>,
 }
 
-impl<Block, PF, BI> Collator<Block, PF, BI> {
+impl<Block: BlockT, PF, BI> Collator<Block, PF, BI> {
 	/// Create a new instance.
 	fn new(
 		proposer_factory: PF,
 		inherent_data_providers: InherentDataProviders,
 		collator_network: impl CollatorNetwork + Clone + 'static,
 		block_import: BI,
+		spawner: Arc<dyn Spawn + Send + Sync>,
+		announce_block: Arc<dyn Fn(Block::Hash, Vec<u8>) + Send + Sync>,
 	) -> Self {
+		let collator_network = Arc::new(collator_network);
+		let wait_to_announce = Arc::new(Mutex::new(WaitToAnnounce::new(
+			spawner,
+			announce_block,
+			collator_network.clone(),
+		)));
+
 		Self {
 			proposer_factory: Arc::new(Mutex::new(proposer_factory)),
 			inherent_data_providers,
 			_phantom: PhantomData,
-			collator_network: Arc::new(collator_network),
+			collator_network,
 			block_import: Arc::new(Mutex::new(block_import)),
+			wait_to_announce,
 		}
 	}
 
@@ -115,14 +129,15 @@ impl<Block, PF, BI> Collator<Block, PF, BI> {
 	}
 }
 
-impl<Block, PF, BI> Clone for Collator<Block, PF, BI> {
+impl<Block: BlockT, PF, BI> Clone for Collator<Block, PF, BI> {
 	fn clone(&self) -> Self {
 		Self {
 			proposer_factory: self.proposer_factory.clone(),
 			inherent_data_providers: self.inherent_data_providers.clone(),
 			_phantom: PhantomData,
 			collator_network: self.collator_network.clone(),
 			block_import: self.block_import.clone(),
+			wait_to_announce: self.wait_to_announce.clone(),
 		}
 	}
 }
@@ -147,7 +162,7 @@ where
 
 	fn produce_candidate(
 		&mut self,
-		_relay_chain_parent: PHash,
+		relay_chain_parent: PHash,
 		global_validation: GlobalValidationSchedule,
 		local_validation: LocalValidationData,
 	) -> Self::ProduceCandidate {
@@ -169,6 +184,8 @@ where
 			.lock()
 			.init(&last_head.header);
 
+		let wait_to_announce = self.wait_to_announce.clone();
+
 		Box::pin(async move {
 			let parent_state_root = *last_head.header.state_root();
 
@@ -245,15 +262,24 @@ where
 			}
 
 			let block_data = BlockData(b.encode());
+			let header = b.into_header();
+			let encoded_header = header.encode();
+			let hash = header.hash();
 			let head_data = HeadData::<Block> {
-				header: b.into_header(),
+				header,
 			};
 
 			let candidate = (
 				block_data,
 				parachain::HeadData(head_data.encode()),
 			);
 
+			wait_to_announce.lock().wait_to_announce(
+				hash,
+				relay_chain_parent,
+				encoded_header,
+			);
+
 			trace!(target: "cumulus-collator", "Produced candidate: {:?}", candidate);
 
 			Ok(candidate)
@@ -268,6 +294,7 @@ pub struct CollatorBuilder<Block: BlockT, PF, BI, Backend, Client> {
 	block_import: BI,
 	para_id: ParaId,
 	client: Arc<Client>,
+	announce_block: Arc<dyn Fn(Block::Hash, Vec<u8>) + Send + Sync>,
 	_marker: PhantomData<(Block, Backend)>,
 }
 
@@ -281,13 +308,15 @@ impl<Block: BlockT, PF, BI, Backend, Client>
 		block_import: BI,
 		para_id: ParaId,
 		client: Arc<Client>,
+		announce_block: Arc<dyn Fn(Block::Hash, Vec<u8>) + Send + Sync>,
 	) -> Self {
 		Self {
 			proposer_factory,
 			inherent_data_providers,
 			block_import,
 			para_id,
 			client,
+			announce_block,
 			_marker: PhantomData,
 		}
 	}
@@ -313,7 +342,7 @@ where
 		self,
 		polkadot_client: Arc<PClient>,
 		spawner: Spawner,
-		network: impl CollatorNetwork + Clone + 'static,
+		polkadot_network: impl CollatorNetwork + Clone + 'static,
 	) -> Result<Self::ParachainContext, ()>
 	where
 		PClient: ProvideRuntimeApi<PBlock> + Send + Sync + BlockchainEvents<PBlock> + 'static,
@@ -342,12 +371,24 @@ where
 		Ok(Collator::new(
 			self.proposer_factory,
 			self.inherent_data_providers,
-			network,
+			polkadot_network,
 			self.block_import,
+			Arc::new(spawner),
+			self.announce_block,
 		))
 	}
 }
 
+/// Prepare the collator's node condifugration
+///
+/// This function will disable the default announcement of Substrate for the parachain in favor
+/// of the one of Cumulus.
+pub fn prepare_collator_config(mut parachain_config: Configuration) -> Configuration {
+	parachain_config.announce_block = false;
+
+	parachain_config
+}
+
 #[cfg(test)]
 mod tests {
 	use super::*;
@@ -460,13 +501,15 @@ mod tests {
 		let id = ParaId::from(100);
 		let _ = env_logger::try_init();
 		let spawner = futures::executor::ThreadPool::new().unwrap();
+		let announce_block = |_, _| ();
 
 		let builder = CollatorBuilder::new(
 			DummyFactory,
 			InherentDataProviders::default(),
 			TestClientBuilder::new().build(),
 			id,
 			Arc::new(TestClientBuilder::new().build()),
+			Arc::new(announce_block),
 		);
 		let context = builder
 			.build(

network/Cargo.toml

@@ -13,10 +13,13 @@ sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "cumulu
 sp-api = { git = "https://github.com/paritytech/substrate", branch = "cumulus-branch" }
 
 # polkadot deps
+polkadot-collator = { git = "https://github.com/paritytech/polkadot", branch = "cumulus-branch" }
 polkadot-primitives = { git = "https://github.com/paritytech/polkadot", branch = "cumulus-branch" }
 polkadot-statement-table = { git = "https://github.com/paritytech/polkadot", branch = "cumulus-branch" }
 polkadot-validation = { git = "https://github.com/paritytech/polkadot", branch = "cumulus-branch" }
 polkadot-network = { git = "https://github.com/paritytech/polkadot", branch = "cumulus-branch" }
 
 # other deps
 codec = { package = "parity-scale-codec", version = "1.3.0", features = [ "derive" ] }
+futures = { version = "0.3.1", features = ["compat"] }
+log = "0.4.8"

network/src/lib.rs

@@ -23,15 +23,21 @@ use sp_blockchain::Error as ClientError;
 use sp_consensus::block_validation::{BlockAnnounceValidator, Validation};
 use sp_runtime::{generic::BlockId, traits::Block as BlockT};
 
+use polkadot_collator::Network as CollatorNetwork;
 use polkadot_network::legacy::gossip::{GossipMessage, GossipStatement};
 use polkadot_primitives::{
 	parachain::{ParachainHost, ValidatorId},
-	Block as PBlock,
+	Block as PBlock, Hash as PHash,
 };
 use polkadot_statement_table::{SignedStatement, Statement};
 use polkadot_validation::check_statement;
 
 use codec::{Decode, Encode};
+use futures::{pin_mut, select, StreamExt};
+use futures::channel::oneshot;
+use futures::future::FutureExt;
+use futures::task::Spawn;
+use log::{error, trace};
 
 use std::{marker::PhantomData, sync::Arc};
 
@@ -141,3 +147,113 @@ where
 		Ok(Validation::Success)
 	}
 }
+
+/// Wait before announcing a block that a candidate message has been received for this block, then
+/// add this message as justification for the block announcement.
+///
+/// This object will spawn a new task every time the method `wait_to_announce` is called and cancel
+/// the previous task running.
+pub struct WaitToAnnounce<Block: BlockT> {
+	spawner: Arc<dyn Spawn + Send + Sync>,
+	announce_block: Arc<dyn Fn(Block::Hash, Vec<u8>) + Send + Sync>,
+	collator_network: Arc<dyn CollatorNetwork>,
+	current_trigger: oneshot::Sender<()>,
+}
+
+impl<Block: BlockT> WaitToAnnounce<Block> {
+	/// Create the `WaitToAnnounce` object
+	pub fn new(
+		spawner: Arc<dyn Spawn + Send + Sync>,
+		announce_block: Arc<dyn Fn(Block::Hash, Vec<u8>) + Send + Sync>,
+		collator_network: Arc<dyn CollatorNetwork>,
+	) -> WaitToAnnounce<Block> {
+		let (tx, _rx) = oneshot::channel();
+
+		WaitToAnnounce {
+			spawner,
+			announce_block,
+			collator_network,
+			current_trigger: tx,
+		}
+	}
+
+	/// Wait for a candidate message for the block, then announce the block. The candidate
+	/// message will be added as justification to the block announcement.
+	pub fn wait_to_announce(
+		&mut self,
+		hash: <Block as BlockT>::Hash,
+		relay_chain_leaf: PHash,
+		head_data: Vec<u8>,
+	) {
+		let (tx, rx) = oneshot::channel();
+		let announce_block = self.announce_block.clone();
+		let collator_network = self.collator_network.clone();
+
+		self.current_trigger = tx;
+
+		if let Err(err) = self.spawner.spawn_obj(Box::pin(async move {
+			let t1 = wait_to_announce::<Block>(
+				hash,
+				relay_chain_leaf,
+				announce_block,
+				collator_network,
+				&head_data,
+			).fuse();
+			let t2 = rx.fuse();
+
+			pin_mut!(t1, t2);
+
+			trace!(
+				target: "cumulus-network",
+				"waiting for announce block in a background task...",
+			);
+
+			select! {
+				_ = t1 => {
+					trace!(
+						target: "cumulus-network",
+						"block announcement finished",
+					);
+				},
+				_ = t2 => {
+					trace!(
+						target: "cumulus-network",
+						"previous task that waits for announce block has been canceled",
+					);
+				}
+			}
+		}).into()) {
+			error!(
+				target: "cumulus-network",
+				"Could not spawn a new task to wait for the announce block: {:?}",
+				err,
+			);
+		}
+	}
+}
+
+async fn wait_to_announce<Block: BlockT>(
+	hash: <Block as BlockT>::Hash,
+	relay_chain_leaf: PHash,
+	announce_block: Arc<dyn Fn(Block::Hash, Vec<u8>) + Send + Sync>,
+	collator_network: Arc<dyn CollatorNetwork>,
+	head_data: &Vec<u8>,
+) {
+	let mut checked_statements = collator_network.checked_statements(relay_chain_leaf);
+
+	while let Some(statement) = checked_statements.next().await {
+		match &statement.statement {
+			Statement::Candidate(c) if &c.head_data.0 == head_data => {
+				let gossip_message: GossipMessage = GossipStatement {
+					relay_chain_leaf,
+					signed_statement: statement,
+				}.into();
+
+				announce_block(hash, gossip_message.encode());
+
+				break;
+			},
+			_ => {},
+		}
+	}
+}