Update Cumulus for Parachains V1 (#224)

* Start with something

* Whatever

* Update

* MOARE

* Make cumulus-network compile and tests work

* Update more and fixes

* More stuff

* More fixes

* Make collator build

* Make test almost work

* Remove contracts runtime

* More test work

* Make service compile

* Fix test-service

* Fix test client

* More fixes

* Fix collator test

* Fix network tests (again)

* Make everything compile, finally

* Fix tests

* Update to latest masters

* Remove ignore

* Switch to different branch in polkadot for now

* Update reference

* Make it compile with latest changes

* Update collator/src/lib.rs

Co-authored-by: Robert Habermeier <rphmeier@gmail.com>

* Update to latest upstream

* Update to latest master

* Fix test

Co-authored-by: Robert Habermeier <rphmeier@gmail.com>

network/src/lib.rs

@@ -14,9 +14,11 @@
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
-//! Cumulus-specific network implementation.
+//! Parachain specific networking
 //!
-//! Contains message send between collators and logic to process them.
+//! Provides a custom block announcement implementation for parachains
+//! that use the relay chain provided consensus. See [`BlockAnnounceValidator`]
+//! and [`WaitToAnnounce`] for more information about this implementation.
 
 #[cfg(test)]
 mod tests;
@@ -24,7 +26,7 @@ mod tests;
 use sp_api::ProvideRuntimeApi;
 use sp_blockchain::{Error as ClientError, HeaderBackend};
 use sp_consensus::{
-	block_validation::{BlockAnnounceValidator, Validation},
+	block_validation::{BlockAnnounceValidator as BlockAnnounceValidatorT, Validation},
 	SyncOracle,
 };
 use sp_core::traits::SpawnNamed;
@@ -33,35 +35,60 @@ use sp_runtime::{
 	traits::{Block as BlockT, Header as HeaderT},
 };
 
-use polkadot_collator::Network as CollatorNetwork;
-use polkadot_network::legacy::gossip::{GossipMessage, GossipStatement};
-use polkadot_primitives::v0::{Block as PBlock, Hash as PHash, Id as ParaId, ParachainHost};
-use polkadot_statement_table::v0::{SignedStatement, Statement};
-use polkadot_validation::check_statement;
-
-use cumulus_primitives::HeadData;
+use polkadot_node_primitives::{SignedFullStatement, Statement};
+use polkadot_node_subsystem::messages::StatementDistributionMessage;
+use polkadot_overseer::OverseerHandler;
+use polkadot_primitives::v1::{
+	Block as PBlock, Hash as PHash, Id as ParaId, OccupiedCoreAssumption, ParachainHost,
+	SigningContext,
+};
+use polkadot_service::ClientHandle;
 
 use codec::{Decode, Encode};
-use futures::{channel::oneshot, future::FutureExt, pin_mut, select, StreamExt};
-use log::trace;
+use futures::{
+	channel::{mpsc, oneshot},
+	future::{ready, FutureExt},
+	pin_mut, select, Future, StreamExt,
+};
+use log::{trace, warn};
 
-use parking_lot::Mutex;
-use std::{marker::PhantomData, sync::Arc};
+use std::{marker::PhantomData, pin::Pin, sync::Arc};
 
-/// Validate that data is a valid justification from a relay-chain validator that the block is a
-/// valid parachain-block candidate.
-/// Data encoding is just `GossipMessage`, the relay-chain validator candidate statement message is
-/// the justification.
+/// Parachain specific block announce validator.
 ///
-/// Note: if no justification is provided the annouce is considered valid.
-pub struct JustifiedBlockAnnounceValidator<B, P> {
+/// This block announce validator is required if the parachain is running
+/// with the relay chain provided consensus to make sure each node only
+/// imports a reasonable number of blocks per round. The relay chain provided
+/// consensus doesn't have any authorities and so it could happen that without
+/// this special block announce validator a node would need to import *millions*
+/// of blocks per round, which is clearly not doable.
+///
+/// To solve this problem, each block announcement is delayed until a collator
+/// has received a [`Statement::Seconded`] for its `PoV`. This message tells the
+/// collator that its `PoV` was validated successfully by a parachain validator and
+/// that it is very likely that this `PoV` will be included in the relay chain. Every
+/// collator that doesn't receive the message for its `PoV` will not announce its block.
+/// For more information on the block announcement, see [`WaitToAnnounce`].
+///
+/// For each block announcement that is received, the generic block announcement validation
+/// will call this validator and provides the extra data that was attached to the announcement.
+/// We call this extra data `justification`.
+/// It is expected that the attached data is a SCALE encoded [`SignedFullStatement`]. The
+/// statement is checked to be a [`Statement::Seconded`] and that it is signed by an active
+/// parachain validator.
+///
+/// If no justification was provided we check if the block announcement is at the tip of the known
+/// chain. If it is at the tip, it is required to provide a justification or otherwise we reject
+/// it. However, if the announcement is for a block below the tip the announcement is accepted
+/// as it probably comes from a node that is currently syncing the chain.
+pub struct BlockAnnounceValidator<B, P> {
 	phantom: PhantomData<B>,
 	polkadot_client: Arc<P>,
 	para_id: ParaId,
 	polkadot_sync_oracle: Box<dyn SyncOracle + Send>,
 }
 
-impl<B, P> JustifiedBlockAnnounceValidator<B, P> {
+impl<B, P> BlockAnnounceValidator<B, P> {
 	pub fn new(
 		polkadot_client: Arc<P>,
 		para_id: ParaId,
@@ -76,192 +103,239 @@ impl<B, P> JustifiedBlockAnnounceValidator<B, P> {
 	}
 }
 
-impl<B: BlockT, P> BlockAnnounceValidator<B> for JustifiedBlockAnnounceValidator<B, P>
+impl<B: BlockT, P> BlockAnnounceValidatorT<B> for BlockAnnounceValidator<B, P>
 where
-	P: ProvideRuntimeApi<PBlock> + HeaderBackend<PBlock>,
+	P: ProvideRuntimeApi<PBlock> + HeaderBackend<PBlock> + 'static,
 	P::Api: ParachainHost<PBlock>,
 {
 	fn validate(
 		&mut self,
 		header: &B::Header,
 		mut data: &[u8],
-	) -> Result<Validation, Box<dyn std::error::Error + Send>> {
+	) -> Pin<Box<dyn Future<Output = Result<Validation, Box<dyn std::error::Error + Send>>> + Send>>
+	{
 		if self.polkadot_sync_oracle.is_major_syncing() {
-			return Ok(Validation::Success { is_new_best: false });
+			return ready(Ok(Validation::Success { is_new_best: false })).boxed();
 		}
 
 		let runtime_api = self.polkadot_client.runtime_api();
 		let polkadot_info = self.polkadot_client.info();
 
 		if data.is_empty() {
-			// Check if block is equal or higher than best (this requires a justification)
-			let runtime_api_block_id = BlockId::Hash(polkadot_info.best_hash);
-			let block_number = header.number();
+			let polkadot_client = self.polkadot_client.clone();
+			let header = header.clone();
+			let para_id = self.para_id;
 
-			let local_validation_data = runtime_api
-				.local_validation_data(&runtime_api_block_id, self.para_id)
-				.map_err(|e| Box::new(ClientError::Msg(format!("{:?}", e))) as Box<_>)?
-				.ok_or_else(|| {
-					Box::new(ClientError::Msg(
-						"Could not find parachain head in relay chain".into(),
-					)) as Box<_>
-				})?;
-			let parent_head = HeadData::<B>::decode(&mut &local_validation_data.parent_head.0[..])
-				.map_err(|e| {
-					Box::new(ClientError::Msg(format!(
-						"Failed to decode parachain head: {:?}",
-						e
-					))) as Box<_>
-				})?;
-			let known_best_number = parent_head.header.number();
+			return async move {
+				// Check if block is equal or higher than best (this requires a justification)
+				let runtime_api_block_id = BlockId::Hash(polkadot_info.best_hash);
+				let block_number = header.number();
 
-			return Ok(if block_number >= known_best_number {
-				trace!(
-					target: "cumulus-network",
-					"validation failed because a justification is needed if the block at the top of the chain."
-				);
+				let local_validation_data = polkadot_client
+					.runtime_api()
+					.persisted_validation_data(
+						&runtime_api_block_id,
+						para_id,
+						OccupiedCoreAssumption::TimedOut,
+					)
+					.map_err(|e| Box::new(ClientError::Msg(format!("{:?}", e))) as Box<_>)?
+					.ok_or_else(|| {
+						Box::new(ClientError::Msg(
+							"Could not find parachain head in relay chain".into(),
+						)) as Box<_>
+					})?;
+				let parent_head = B::Header::decode(&mut &local_validation_data.parent_head.0[..])
+					.map_err(|e| {
+						Box::new(ClientError::Msg(format!(
+							"Failed to decode parachain head: {:?}",
+							e
+						))) as Box<_>
+					})?;
+				let known_best_number = parent_head.number();
 
-				Validation::Failure
-			} else {
-				Validation::Success { is_new_best: false }
-			});
+				if block_number >= known_best_number {
+					trace!(
+						target: "cumulus-network",
+						"validation failed because a justification is needed if the block at the top of the chain."
+					);
+
+					Ok(Validation::Failure)
+				} else {
+					Ok(Validation::Success { is_new_best: false })
+				}
+			}
+			.boxed();
 		}
 
-		// Check data is a gossip message.
-		let gossip_message = GossipMessage::decode(&mut data).map_err(|_| {
-			Box::new(ClientError::BadJustification(
-				"cannot decode block announced justification, must be a gossip message".to_string(),
-			)) as Box<_>
-		})?;
+		let signed_stmt = match SignedFullStatement::decode(&mut data) {
+			Ok(r) => r,
+			Err(_) => return ready(Err(Box::new(ClientError::BadJustification(
+				"cannot decode block announcement justification, must be a `SignedFullStatement`"
+					.to_string(),
+			)) as Box<_>))
+			.boxed(),
+		};
 
-		// Check message is a gossip statement.
-		let gossip_statement = match gossip_message {
-			GossipMessage::Statement(gossip_statement) => gossip_statement,
+		// Check statement is a candidate statement.
+		let candidate_receipt = match signed_stmt.payload() {
+			Statement::Seconded(ref candidate_receipt) => candidate_receipt,
 			_ => {
-				return Err(Box::new(ClientError::BadJustification(
-					"block announced justification statement must be a gossip statement"
-						.to_string(),
-				)) as Box<_>)
+				return ready(Err(Box::new(ClientError::BadJustification(
+					"block announcement justification must be a `Statement::Seconded`".to_string(),
+				)) as Box<_>))
+				.boxed()
 			}
 		};
 
-		let GossipStatement {
-			relay_chain_leaf,
-			signed_statement: SignedStatement {
-				statement,
-				signature,
-				sender,
-			},
-		} = gossip_statement;
-
 		// Check that the relay chain parent of the block is the relay chain head
 		let best_number = polkadot_info.best_number;
+		let validator_index = signed_stmt.validator_index();
+		let relay_parent = &candidate_receipt.descriptor.relay_parent;
 
-		match self.polkadot_client.number(relay_chain_leaf) {
+		match self.polkadot_client.number(*relay_parent) {
 			Err(err) => {
-				return Err(Box::new(ClientError::Backend(format!(
+				return ready(Err(Box::new(ClientError::Backend(format!(
 					"could not find block number for {}: {}",
-					relay_chain_leaf, err,
-				))));
+					relay_parent, err,
+				))) as Box<_>))
+				.boxed();
 			}
 			Ok(Some(x)) if x == best_number => {}
 			Ok(None) => {
-				return Err(Box::new(ClientError::UnknownBlock(
-					relay_chain_leaf.to_string(),
-				)));
+				return ready(Err(
+					Box::new(ClientError::UnknownBlock(relay_parent.to_string())) as Box<_>,
+				))
+				.boxed();
 			}
 			Ok(Some(_)) => {
 				trace!(
 					target: "cumulus-network",
 					"validation failed because the relay chain parent ({}) is not the relay chain \
 					head ({})",
-					relay_chain_leaf, best_number,
+					relay_parent,
+					best_number,
 				);
 
-				return Ok(Validation::Failure);
+				return ready(Ok(Validation::Failure)).boxed();
 			}
 		}
 
-		let runtime_api_block_id = BlockId::Hash(relay_chain_leaf);
-		let signing_context = runtime_api
-			.signing_context(&runtime_api_block_id)
-			.map_err(|e| Box::new(ClientError::Msg(format!("{:?}", e))) as Box<_>)?;
-
-		// Check that the signer is a legit validator.
-		let authorities = runtime_api
-			.validators(&runtime_api_block_id)
-			.map_err(|e| Box::new(ClientError::Msg(format!("{:?}", e))) as Box<_>)?;
-		let signer = authorities.get(sender as usize).ok_or_else(|| {
-			Box::new(ClientError::BadJustification(
-				"block accounced justification signer is a validator index out of bound"
-					.to_string(),
-			)) as Box<_>
-		})?;
-
-		// Check statement is correctly signed.
-		if !check_statement(&statement, &signature, signer.clone(), &signing_context) {
-			return Err(Box::new(ClientError::BadJustification(
-				"block announced justification signature is invalid".to_string(),
-			)) as Box<_>);
-		}
-
-		// Check statement is a candidate statement.
-		let candidate_receipt = match statement {
-			Statement::Candidate(candidate_receipt) => candidate_receipt,
-			_ => {
-				return Err(Box::new(ClientError::BadJustification(
-					"block announced justification statement must be a candidate statement"
-						.to_string(),
-				)) as Box<_>)
+		let runtime_api_block_id = BlockId::Hash(*relay_parent);
+		let session_index = match runtime_api.session_index_for_child(&runtime_api_block_id) {
+			Ok(r) => r,
+			Err(e) => {
+				return ready(Err(Box::new(ClientError::Msg(format!("{:?}", e))) as Box<_>)).boxed()
 			}
 		};
 
-		// Check the header in the candidate_receipt match header given header.
-		if header.encode() != candidate_receipt.head_data.0 {
-			return Err(Box::new(ClientError::BadJustification(
-				"block announced header does not match the one justified".to_string(),
-			)) as Box<_>);
-		}
+		let signing_context = SigningContext {
+			parent_hash: *relay_parent,
+			session_index,
+		};
 
-		Ok(Validation::Success { is_new_best: true })
-	}
-}
-
-/// A `BlockAnnounceValidator` that will be able to validate data when its internal
-/// `BlockAnnounceValidator` is set.
-pub struct DelayedBlockAnnounceValidator<B: BlockT>(
-	Arc<Mutex<Option<Box<dyn BlockAnnounceValidator<B> + Send>>>>,
-);
-
-impl<B: BlockT> DelayedBlockAnnounceValidator<B> {
-	pub fn new() -> DelayedBlockAnnounceValidator<B> {
-		DelayedBlockAnnounceValidator(Arc::new(Mutex::new(None)))
-	}
-
-	pub fn set(&self, validator: Box<dyn BlockAnnounceValidator<B> + Send>) {
-		*self.0.lock() = Some(validator);
-	}
-}
-
-impl<B: BlockT> Clone for DelayedBlockAnnounceValidator<B> {
-	fn clone(&self) -> DelayedBlockAnnounceValidator<B> {
-		DelayedBlockAnnounceValidator(self.0.clone())
-	}
-}
-
-impl<B: BlockT> BlockAnnounceValidator<B> for DelayedBlockAnnounceValidator<B> {
-	fn validate(
-		&mut self,
-		header: &B::Header,
-		data: &[u8],
-	) -> Result<Validation, Box<dyn std::error::Error + Send>> {
-		match self.0.lock().as_mut() {
-			Some(validator) => validator.validate(header, data),
-			None => {
-				log::warn!("BlockAnnounce validator not yet set, rejecting block announcement");
-				Ok(Validation::Failure)
+		// Check that the signer is a legit validator.
+		let authorities = match runtime_api.validators(&runtime_api_block_id) {
+			Ok(r) => r,
+			Err(e) => {
+				return ready(Err(Box::new(ClientError::Msg(format!("{:?}", e))) as Box<_>)).boxed()
 			}
+		};
+		let signer = match authorities.get(validator_index as usize) {
+			Some(r) => r,
+			None => {
+				return ready(Err(Box::new(ClientError::BadJustification(
+					"block accouncement justification signer is a validator index out of bound"
+						.to_string(),
+				)) as Box<_>))
+				.boxed()
+			}
+		};
+
+		// Check statement is correctly signed.
+		if signed_stmt
+			.check_signature(&signing_context, &signer)
+			.is_err()
+		{
+			return ready(Err(Box::new(ClientError::BadJustification(
+				"block announced justification signature is invalid".to_string(),
+			)) as Box<_>))
+			.boxed();
 		}
+
+		// Check the header in the candidate_receipt match header given header.
+		if header.encode() != candidate_receipt.commitments.head_data.0 {
+			return ready(Err(Box::new(ClientError::BadJustification(
+				"block announced header does not match the one justified".to_string(),
+			)) as Box<_>))
+			.boxed();
+		}
+
+		ready(Ok(Validation::Success { is_new_best: true })).boxed()
+	}
+}
+
+/// Build a block announce validator instance.
+///
+/// Returns a boxed [`BlockAnnounceValidator`].
+pub fn build_block_announce_validator<B: BlockT>(
+	polkadot_client: polkadot_service::Client,
+	para_id: ParaId,
+	polkadot_sync_oracle: Box<dyn SyncOracle + Send>,
+) -> Box<dyn BlockAnnounceValidatorT<B> + Send> {
+	BlockAnnounceValidatorBuilder::new(polkadot_client, para_id, polkadot_sync_oracle).build()
+}
+
+/// Block announce validator builder.
+///
+/// Builds a [`BlockAnnounceValidator`] for a parachain. As this requires
+/// a concrete Polkadot client instance, the builder takes a [`polkadot_service::Client`]
+/// that wraps this concrete instanace. By using [`polkadot_service::ExecuteWithClient`]
+/// the builder gets access to this concrete instance.
+struct BlockAnnounceValidatorBuilder<B> {
+	phantom: PhantomData<B>,
+	polkadot_client: polkadot_service::Client,
+	para_id: ParaId,
+	polkadot_sync_oracle: Box<dyn SyncOracle + Send>,
+}
+
+impl<B: BlockT> BlockAnnounceValidatorBuilder<B> {
+	/// Create a new instance of the builder.
+	fn new(
+		polkadot_client: polkadot_service::Client,
+		para_id: ParaId,
+		polkadot_sync_oracle: Box<dyn SyncOracle + Send>,
+	) -> Self {
+		Self {
+			polkadot_client,
+			para_id,
+			polkadot_sync_oracle,
+			phantom: PhantomData,
+		}
+	}
+
+	/// Build the block announce validator.
+	fn build(self) -> Box<dyn BlockAnnounceValidatorT<B> + Send> {
+		self.polkadot_client.clone().execute_with(self)
+	}
+}
+
+impl<B: BlockT> polkadot_service::ExecuteWithClient for BlockAnnounceValidatorBuilder<B> {
+	type Output = Box<dyn BlockAnnounceValidatorT<B> + Send>;
+
+	fn execute_with_client<PClient, Api, PBackend>(self, client: Arc<PClient>) -> Self::Output
+	where
+		<Api as sp_api::ApiExt<PBlock>>::StateBackend:
+			sp_api::StateBackend<sp_runtime::traits::BlakeTwo256>,
+		PBackend: sc_client_api::Backend<PBlock>,
+		PBackend::State: sp_api::StateBackend<sp_runtime::traits::BlakeTwo256>,
+		Api: polkadot_service::RuntimeApiCollection<StateBackend = PBackend::State>,
+		PClient: polkadot_service::AbstractClient<PBlock, PBackend, Api = Api> + 'static,
+	{
+		Box::new(BlockAnnounceValidator::new(
+			client,
+			self.para_id,
+			self.polkadot_sync_oracle,
+		))
 	}
 }
 
@@ -273,7 +347,7 @@ impl<B: BlockT> BlockAnnounceValidator<B> for DelayedBlockAnnounceValidator<B> {
 pub struct WaitToAnnounce<Block: BlockT> {
 	spawner: Arc<dyn SpawnNamed + Send + Sync>,
 	announce_block: Arc<dyn Fn(Block::Hash, Vec<u8>) + Send + Sync>,
-	collator_network: Arc<dyn CollatorNetwork>,
+	overseer_handler: OverseerHandler,
 	current_trigger: oneshot::Sender<()>,
 }
 
@@ -282,29 +356,24 @@ impl<Block: BlockT> WaitToAnnounce<Block> {
 	pub fn new(
 		spawner: Arc<dyn SpawnNamed + Send + Sync>,
 		announce_block: Arc<dyn Fn(Block::Hash, Vec<u8>) + Send + Sync>,
-		collator_network: Arc<dyn CollatorNetwork>,
+		overseer_handler: OverseerHandler,
 	) -> WaitToAnnounce<Block> {
 		let (tx, _rx) = oneshot::channel();
 
 		WaitToAnnounce {
 			spawner,
 			announce_block,
-			collator_network,
+			overseer_handler,
 			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>,
-	) {
+	pub fn wait_to_announce(&mut self, block_hash: <Block as BlockT>::Hash, pov_hash: PHash) {
 		let (tx, rx) = oneshot::channel();
 		let announce_block = self.announce_block.clone();
-		let collator_network = self.collator_network.clone();
+		let overseer_handler = self.overseer_handler.clone();
 
 		self.current_trigger = tx;
 
@@ -312,11 +381,10 @@ impl<Block: BlockT> WaitToAnnounce<Block> {
 			"cumulus-wait-to-announce",
 			async move {
 				let t1 = wait_to_announce::<Block>(
-					hash,
-					relay_chain_leaf,
+					block_hash,
+					pov_hash,
 					announce_block,
-					collator_network,
-					&head_data,
+					overseer_handler,
 				)
 				.fuse();
 				let t2 = rx.fuse();
@@ -349,24 +417,30 @@ impl<Block: BlockT> WaitToAnnounce<Block> {
 }
 
 async fn wait_to_announce<Block: BlockT>(
-	hash: <Block as BlockT>::Hash,
-	relay_chain_leaf: PHash,
+	block_hash: <Block as BlockT>::Hash,
+	pov_hash: PHash,
 	announce_block: Arc<dyn Fn(Block::Hash, Vec<u8>) + Send + Sync>,
-	collator_network: Arc<dyn CollatorNetwork>,
-	head_data: &Vec<u8>,
+	mut overseer_handler: OverseerHandler,
 ) {
-	let mut checked_statements = collator_network.checked_statements(relay_chain_leaf);
+	let (sender, mut receiver) = mpsc::channel(5);
+	if overseer_handler
+		.send_msg(StatementDistributionMessage::RegisterStatementListener(
+			sender,
+		))
+		.await
+		.is_err()
+	{
+		warn!(
+			target: "cumulus-network",
+			"Failed to register the statement listener!",
+		);
+		return;
+	}
 
-	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());
+	while let Some(statement) = receiver.next().await {
+		match &statement.payload() {
+			Statement::Seconded(c) if &c.descriptor.pov_hash == &pov_hash => {
+				announce_block(block_hash, statement.encode());
 
 				break;
 			}

network/src/tests.rs

@@ -16,21 +16,25 @@
 
 use super::*;
 use cumulus_test_runtime::{Block, Header};
-use polkadot_primitives::v0::{
-	AbridgedCandidateReceipt, Block as PBlock, Chain, CollatorId, DutyRoster, GlobalValidationData,
-	Hash as PHash, Header as PHeader, Id as ParaId, LocalValidationData, ParachainHost, Retriable,
-	SigningContext, ValidationCode, ValidatorId,
+use futures::executor::block_on;
+use polkadot_node_primitives::{SignedFullStatement, Statement};
+use polkadot_primitives::v1::{
+	AuthorityDiscoveryId, Block as PBlock, BlockNumber, CandidateCommitments, CandidateDescriptor,
+	CandidateEvent, CommittedCandidateReceipt, CoreState, GroupRotationInfo, Hash as PHash,
+	HeadData, Header as PHeader, Id as ParaId, OccupiedCoreAssumption, ParachainHost,
+	PersistedValidationData, SessionIndex, SigningContext, ValidationCode, ValidationData,
+	ValidationOutputs, ValidatorId, ValidatorIndex, InboundDownwardMessage,
 };
-use polkadot_test_runtime_client::{
-	DefaultTestClientBuilderExt, TestClient, TestClientBuilder, TestClientBuilderExt,
-};
-use polkadot_validation::{sign_table_statement, Statement};
 use sp_api::{ApiRef, ProvideRuntimeApi};
 use sp_blockchain::{Error as ClientError, HeaderBackend};
-use sp_consensus::block_validation::BlockAnnounceValidator;
+use sp_consensus::block_validation::BlockAnnounceValidator as _;
 use sp_core::H256;
 use sp_keyring::Sr25519Keyring;
-use sp_runtime::traits::{Block as BlockT, NumberFor, Zero};
+use sp_keystore::{testing::KeyStore, SyncCryptoStore, SyncCryptoStorePtr};
+use sp_runtime::{
+	traits::{NumberFor, Zero},
+	RuntimeAppPublic,
+};
 
 #[derive(Clone)]
 struct DummyCollatorNetwork;
@@ -45,26 +49,16 @@ impl SyncOracle for DummyCollatorNetwork {
 	}
 }
 
-fn make_validator() -> JustifiedBlockAnnounceValidator<Block, TestApi> {
-	let (validator, _client) = make_validator_and_client();
-
-	validator
-}
-
-fn make_validator_and_client() -> (
-	JustifiedBlockAnnounceValidator<Block, TestApi>,
-	Arc<TestApi>,
-) {
-	let builder = TestClientBuilder::new();
-	let client = Arc::new(TestApi::new(Arc::new(builder.build())));
+fn make_validator_and_api() -> (BlockAnnounceValidator<Block, TestApi>, Arc<TestApi>) {
+	let api = Arc::new(TestApi::new());
 
 	(
-		JustifiedBlockAnnounceValidator::new(
-			client.clone(),
+		BlockAnnounceValidator::new(
+			api.clone(),
 			ParaId::from(56),
 			Box::new(DummyCollatorNetwork),
 		),
-		client,
+		api,
 	)
 }
 
@@ -79,42 +73,58 @@ fn default_header() -> Header {
 }
 
 fn make_gossip_message_and_header(
-	client: Arc<TestApi>,
-	relay_chain_leaf: H256,
-) -> (GossipMessage, Header) {
-	let key = Sr25519Keyring::Alice.pair().into();
-	let signing_context = client
+	api: Arc<TestApi>,
+	relay_parent: H256,
+	validator_index: u32,
+) -> (SignedFullStatement, Header) {
+	let keystore: SyncCryptoStorePtr = Arc::new(KeyStore::new());
+	let alice_public = SyncCryptoStore::sr25519_generate_new(
+		&*keystore,
+		ValidatorId::ID,
+		Some(&Sr25519Keyring::Alice.to_seed()),
+	)
+	.unwrap();
+	let session_index = api
 		.runtime_api()
-		.signing_context(&BlockId::Hash(relay_chain_leaf))
+		.session_index_for_child(&BlockId::Hash(relay_parent))
 		.unwrap();
-	let header = default_header();
-	let candidate_receipt = AbridgedCandidateReceipt {
-		head_data: header.encode().into(),
-		..AbridgedCandidateReceipt::default()
-	};
-	let statement = Statement::Candidate(candidate_receipt);
-	let signature = sign_table_statement(&statement, &key, &signing_context);
-	let sender = 0;
-	let gossip_statement = GossipStatement {
-		relay_chain_leaf,
-		signed_statement: SignedStatement {
-			statement,
-			signature,
-			sender,
-		},
+	let signing_context = SigningContext {
+		parent_hash: relay_parent,
+		session_index,
 	};
 
-	(GossipMessage::Statement(gossip_statement), header)
+	let header = default_header();
+	let candidate_receipt = CommittedCandidateReceipt {
+		commitments: CandidateCommitments {
+			head_data: header.encode().into(),
+			..Default::default()
+		},
+		descriptor: CandidateDescriptor {
+			relay_parent,
+			..Default::default()
+		},
+	};
+	let statement = Statement::Seconded(candidate_receipt);
+	let signed = block_on(SignedFullStatement::sign(
+		&keystore,
+		statement,
+		&signing_context,
+		validator_index,
+		&alice_public.into(),
+	))
+	.expect("Signing statement");
+
+	(signed, header)
 }
 
 #[test]
 fn valid_if_no_data_and_less_than_best_known_number() {
-	let mut validator = make_validator();
+	let mut validator = make_validator_and_api().0;
 	let header = Header {
 		number: 0,
 		..default_header()
 	};
-	let res = validator.validate(&header, &[]);
+	let res = block_on(validator.validate(&header, &[]));
 
 	assert_eq!(
 		res.unwrap(),
@@ -125,12 +135,12 @@ fn valid_if_no_data_and_less_than_best_known_number() {
 
 #[test]
 fn invalid_if_no_data_exceeds_best_known_number() {
-	let mut validator = make_validator();
+	let mut validator = make_validator_and_api().0;
 	let header = Header {
 		number: 1,
 		..default_header()
 	};
-	let res = validator.validate(&header, &[]);
+	let res = block_on(validator.validate(&header, &[]));
 
 	assert_eq!(
 		res.unwrap(),
@@ -140,28 +150,26 @@ fn invalid_if_no_data_exceeds_best_known_number() {
 }
 
 #[test]
-fn check_gossip_message_is_valid() {
-	let mut validator = make_validator();
+fn check_statement_is_encoded_correctly() {
+	let mut validator = make_validator_and_api().0;
 	let header = default_header();
-	let res = validator.validate(&header, &[0x42]).err();
+	let res = block_on(validator.validate(&header, &[0x42]))
+		.err()
+		.expect("Should fail on invalid encoded statement");
 
-	assert!(
-		res.is_some(),
-		"only data that are gossip message are allowed"
-	);
 	assert!(matches!(
-		*res.unwrap().downcast::<ClientError>().unwrap(),
-		ClientError::BadJustification(x) if x.contains("must be a gossip message")
+		*res.downcast::<ClientError>().unwrap(),
+		ClientError::BadJustification(x) if x.contains("must be a `SignedFullStatement`")
 	));
 }
 
 #[test]
 fn check_relay_parent_is_head() {
-	let (mut validator, client) = make_validator_and_client();
+	let (mut validator, api) = make_validator_and_api();
 	let relay_chain_leaf = H256::zero();
-	let (gossip_message, header) = make_gossip_message_and_header(client, relay_chain_leaf);
+	let (gossip_message, header) = make_gossip_message_and_header(api, relay_chain_leaf, 0);
 	let data = gossip_message.encode();
-	let res = validator.validate(&header, data.as_slice());
+	let res = block_on(validator.validate(&header, data.as_slice()));
 
 	assert_eq!(
 		res.unwrap(),
@@ -172,199 +180,136 @@ fn check_relay_parent_is_head() {
 
 #[test]
 fn check_relay_parent_actually_exists() {
-	let (mut validator, client) = make_validator_and_client();
-	let relay_chain_leaf = H256::from_low_u64_be(42);
-	let (gossip_message, header) = make_gossip_message_and_header(client, relay_chain_leaf);
-	let data = gossip_message.encode();
-	let res = validator.validate(&header, data.as_slice()).err();
+	let (mut validator, api) = make_validator_and_api();
+	let relay_parent = H256::from_low_u64_be(42);
+	let (signed_statement, header) = make_gossip_message_and_header(api, relay_parent, 0);
+	let data = signed_statement.encode();
+	let res = block_on(validator.validate(&header, &data))
+		.err()
+		.expect("Should fail on unknown relay parent");
 
-	assert!(
-		res.is_some(),
-		"validation should fail if the relay chain leaf does not exist"
-	);
 	assert!(matches!(
-		*res.unwrap().downcast::<ClientError>().unwrap(),
+		*res.downcast::<ClientError>().unwrap(),
 		ClientError::UnknownBlock(_)
 	));
 }
 
 #[test]
 fn check_relay_parent_fails_if_cannot_retrieve_number() {
-	let (mut validator, client) = make_validator_and_client();
-	let relay_chain_leaf = H256::from_low_u64_be(0xdead);
-	let (gossip_message, header) = make_gossip_message_and_header(client, relay_chain_leaf);
-	let data = gossip_message.encode();
-	let res = validator.validate(&header, data.as_slice()).err();
+	let (mut validator, api) = make_validator_and_api();
+	let relay_parent = H256::from_low_u64_be(0xdead);
+	let (signed_statement, header) = make_gossip_message_and_header(api, relay_parent, 0);
+	let data = signed_statement.encode();
+	let res = block_on(validator.validate(&header, &data))
+		.err()
+		.expect("Should fail when the relay chain number could not be retrieved");
 
-	assert!(
-		res.is_some(),
-		"validation should fail if the relay chain leaf does not exist"
-	);
 	assert!(matches!(
-		*res.unwrap().downcast::<ClientError>().unwrap(),
+		*res.downcast::<ClientError>().unwrap(),
 		ClientError::Backend(_)
 	));
 }
 
 #[test]
 fn check_signer_is_legit_validator() {
-	let (mut validator, client) = make_validator_and_client();
-	let relay_chain_leaf = H256::from_low_u64_be(1);
+	let (mut validator, api) = make_validator_and_api();
+	let relay_parent = H256::from_low_u64_be(1);
 
-	let key = Sr25519Keyring::Alice.pair().into();
-	let signing_context = client
-		.runtime_api()
-		.signing_context(&BlockId::Hash(relay_chain_leaf))
-		.unwrap();
-	let header = default_header();
-	let candidate_receipt = AbridgedCandidateReceipt {
-		head_data: header.encode().into(),
-		..AbridgedCandidateReceipt::default()
-	};
-	let statement = Statement::Candidate(candidate_receipt);
-	let signature = sign_table_statement(&statement, &key, &signing_context);
-	let sender = 1;
-	let gossip_statement = GossipStatement {
-		relay_chain_leaf,
-		signed_statement: SignedStatement {
-			statement,
-			signature,
-			sender,
-		},
-	};
-	let gossip_message = GossipMessage::Statement(gossip_statement);
+	let (signed_statement, header) = make_gossip_message_and_header(api, relay_parent, 1);
+	let data = signed_statement.encode();
 
-	let data = gossip_message.encode();
-	let res = validator.validate(&header, data.as_slice()).err();
+	let res = block_on(validator.validate(&header, &data))
+		.err()
+		.expect("Should fail on invalid validator");
 
-	assert!(
-		res.is_some(),
-		"validation should fail if the signer is not a legit validator"
-	);
 	assert!(matches!(
-		*res.unwrap().downcast::<ClientError>().unwrap(),
+		*res.downcast::<ClientError>().unwrap(),
 		ClientError::BadJustification(x) if x.contains("signer is a validator")
 	));
 }
 
 #[test]
 fn check_statement_is_correctly_signed() {
-	let (mut validator, client) = make_validator_and_client();
-	let header = default_header();
-	let relay_chain_leaf = H256::from_low_u64_be(1);
+	let (mut validator, api) = make_validator_and_api();
+	let relay_parent = H256::from_low_u64_be(1);
 
-	let key = Sr25519Keyring::Alice.pair().into();
-	let signing_context = client
-		.runtime_api()
-		.signing_context(&BlockId::Hash(relay_chain_leaf))
-		.unwrap();
-	let mut candidate_receipt = AbridgedCandidateReceipt::default();
-	let statement = Statement::Candidate(candidate_receipt.clone());
-	let signature = sign_table_statement(&statement, &key, &signing_context);
+	let (signed_statement, header) = make_gossip_message_and_header(api, relay_parent, 0);
 
-	// alterate statement so the signature doesn't match anymore
-	candidate_receipt = AbridgedCandidateReceipt {
-		parachain_index: candidate_receipt.parachain_index + 1,
-		..candidate_receipt
-	};
-	let statement = Statement::Candidate(candidate_receipt);
+	let mut data = signed_statement.encode();
 
-	let sender = 0;
-	let gossip_statement = GossipStatement {
-		relay_chain_leaf,
-		signed_statement: SignedStatement {
-			statement,
-			signature,
-			sender,
-		},
-	};
-	let gossip_message = GossipMessage::Statement(gossip_statement);
+	// The signature comes at the end of the type, so change a bit to make the signature invalid.
+	let last = data.len() - 1;
+	data[last] = data[last].wrapping_add(1);
 
-	let data = gossip_message.encode();
-	let res = validator.validate(&header, data.as_slice()).err();
+	let res = block_on(validator.validate(&header, &data))
+		.err()
+		.expect("Validation should fail if the statement is not signed correctly");
 
-	assert!(
-		res.is_some(),
-		"validation should fail if the statement is not correctly signed"
-	);
 	assert!(matches!(
-		*res.unwrap().downcast::<ClientError>().unwrap(),
+		*res.downcast::<ClientError>().unwrap(),
 		ClientError::BadJustification(x) if x.contains("signature is invalid")
 	));
 }
 
 #[test]
-fn check_statement_is_a_candidate_message() {
-	let (mut validator, client) = make_validator_and_client();
+fn check_statement_seconded() {
+	let (mut validator, api) = make_validator_and_api();
 	let header = default_header();
-	let relay_chain_leaf = H256::from_low_u64_be(1);
+	let relay_parent = H256::from_low_u64_be(1);
 
-	let key = Sr25519Keyring::Alice.pair().into();
-	let signing_context = client
+	let keystore: SyncCryptoStorePtr = Arc::new(KeyStore::new());
+	let alice_public = SyncCryptoStore::sr25519_generate_new(
+		&*keystore,
+		ValidatorId::ID,
+		Some(&Sr25519Keyring::Alice.to_seed()),
+	)
+	.unwrap();
+	let session_index = api
 		.runtime_api()
-		.signing_context(&BlockId::Hash(relay_chain_leaf))
+		.session_index_for_child(&BlockId::Hash(relay_parent))
 		.unwrap();
-	let statement = Statement::Valid(H256::zero());
-	let signature = sign_table_statement(&statement, &key, &signing_context);
-	let sender = 0;
-	let gossip_statement = GossipStatement {
-		relay_chain_leaf,
-		signed_statement: SignedStatement {
-			statement,
-			signature,
-			sender,
-		},
+	let signing_context = SigningContext {
+		parent_hash: relay_parent,
+		session_index,
 	};
-	let gossip_message = GossipMessage::Statement(gossip_statement);
 
-	let data = gossip_message.encode();
-	let res = validator.validate(&header, data.as_slice()).err();
+	let statement = Statement::Valid(Default::default());
+
+	let signed_statement = block_on(SignedFullStatement::sign(
+		&keystore,
+		statement,
+		&signing_context,
+		0,
+		&alice_public.into(),
+	))
+	.expect("Signs statement");
+	let data = signed_statement.encode();
+
+	let res = block_on(validator.validate(&header, &data))
+		.err()
+		.expect("validation should fail if not seconded statement");
 
-	assert!(
-		res.is_some(),
-		"validation should fail if the statement is not a candidate message"
-	);
 	assert!(matches!(
-		*res.unwrap().downcast::<ClientError>().unwrap(),
-		ClientError::BadJustification(x) if x.contains("must be a candidate statement")
+		*res.downcast::<ClientError>().unwrap(),
+		ClientError::BadJustification(x) if x.contains("must be a `Statement::Seconded`")
 	));
 }
 
 #[test]
 fn check_header_match_candidate_receipt_header() {
-	let (mut validator, client) = make_validator_and_client();
-	let relay_chain_leaf = H256::from_low_u64_be(1);
+	let (mut validator, api) = make_validator_and_api();
+	let relay_parent = H256::from_low_u64_be(1);
 
-	let key = Sr25519Keyring::Alice.pair().into();
-	let signing_context = client
-		.runtime_api()
-		.signing_context(&BlockId::Hash(relay_chain_leaf))
-		.unwrap();
-	let header = default_header();
-	let candidate_receipt = AbridgedCandidateReceipt::default();
-	let statement = Statement::Candidate(candidate_receipt);
-	let signature = sign_table_statement(&statement, &key, &signing_context);
-	let sender = 0;
-	let gossip_statement = GossipStatement {
-		relay_chain_leaf,
-		signed_statement: SignedStatement {
-			statement,
-			signature,
-			sender,
-		},
-	};
-	let gossip_message = GossipMessage::Statement(gossip_statement);
+	let (signed_statement, mut header) = make_gossip_message_and_header(api, relay_parent, 0);
+	let data = signed_statement.encode();
+	header.number = 300;
 
-	let data = gossip_message.encode();
-	let res = validator.validate(&header, data.as_slice()).err();
+	let res = block_on(validator.validate(&header, &data))
+		.err()
+		.expect("validation should fail if the header in doesn't match");
 
-	assert!(
-		res.is_some(),
-		"validation should fail if the header in the candidate_receipt doesn't \
-		match the header provided"
-	);
 	assert!(matches!(
-		*res.unwrap().downcast::<ClientError>().unwrap(),
+		*res.downcast::<ClientError>().unwrap(),
 		ClientError::BadJustification(x) if x.contains("header does not match")
 	));
 }
@@ -372,31 +317,25 @@ fn check_header_match_candidate_receipt_header() {
 #[derive(Default)]
 struct ApiData {
 	validators: Vec<ValidatorId>,
-	duties: Vec<Chain>,
-	active_parachains: Vec<(ParaId, Option<(CollatorId, Retriable)>)>,
 }
 
-#[derive(Clone)]
 struct TestApi {
-	data: Arc<Mutex<ApiData>>,
-	client: Arc<TestClient>,
+	data: Arc<ApiData>,
 }
 
 impl TestApi {
-	fn new(client: Arc<TestClient>) -> Self {
+	fn new() -> Self {
 		Self {
-			client,
-			data: Arc::new(Mutex::new(ApiData {
+			data: Arc::new(ApiData {
 				validators: vec![Sr25519Keyring::Alice.public().into()],
-				..Default::default()
-			})),
+			}),
 		}
 	}
 }
 
 #[derive(Default)]
 struct RuntimeApi {
-	data: Arc<Mutex<ApiData>>,
+	data: Arc<ApiData>,
 }
 
 impl ProvideRuntimeApi<PBlock> for TestApi {
@@ -415,48 +354,59 @@ sp_api::mock_impl_runtime_apis! {
 		type Error = sp_blockchain::Error;
 
 		fn validators(&self) -> Vec<ValidatorId> {
-			self.data.lock().validators.clone()
+			self.data.validators.clone()
 		}
 
-		fn duty_roster(&self) -> DutyRoster {
-			DutyRoster {
-				validator_duty: self.data.lock().duties.clone(),
-			}
+		fn validator_groups(&self) -> (Vec<Vec<ValidatorIndex>>, GroupRotationInfo<BlockNumber>) {
+			(Vec::new(), GroupRotationInfo { session_start_block: 0, group_rotation_frequency: 0, now: 0 })
 		}
 
-		fn active_parachains(&self) -> Vec<(ParaId, Option<(CollatorId, Retriable)>)> {
-			self.data.lock().active_parachains.clone()
+		fn availability_cores(&self) -> Vec<CoreState<BlockNumber>> {
+			Vec::new()
 		}
 
-		fn parachain_code(_: ParaId) -> Option<ValidationCode> {
-			Some(ValidationCode(Vec::new()))
+		fn full_validation_data(&self, _: ParaId, _: OccupiedCoreAssumption) -> Option<ValidationData<BlockNumber>> {
+			None
 		}
 
-		fn global_validation_data() -> GlobalValidationData {
-			Default::default()
-		}
-
-		fn local_validation_data(_: ParaId) -> Option<LocalValidationData> {
-			Some(LocalValidationData {
-				parent_head: HeadData::<Block> { header: default_header() }.encode().into(),
+		fn persisted_validation_data(&self, _: ParaId, _: OccupiedCoreAssumption) -> Option<PersistedValidationData<BlockNumber>> {
+			Some(PersistedValidationData {
+				parent_head: HeadData(default_header().encode()),
 				..Default::default()
 			})
 		}
 
-		fn get_heads(_: Vec<<PBlock as BlockT>::Extrinsic>) -> Option<Vec<AbridgedCandidateReceipt>> {
-			Some(Vec::new())
+		fn session_index_for_child(&self) -> SessionIndex {
+			0
 		}
 
-		fn signing_context() -> SigningContext {
-			SigningContext {
-				session_index: Default::default(),
-				parent_hash: Default::default(),
-			}
+		fn validation_code(&self, _: ParaId, _: OccupiedCoreAssumption) -> Option<ValidationCode> {
+			None
 		}
 
-		fn downward_messages(_: ParaId) -> Vec<polkadot_primitives::v0::DownwardMessage> {
+		fn candidate_pending_availability(&self, _: ParaId) -> Option<CommittedCandidateReceipt<PHash>> {
+			None
+		}
+
+		fn candidate_events(&self) -> Vec<CandidateEvent<PHash>> {
 			Vec::new()
 		}
+
+		fn validator_discovery(_: Vec<ValidatorId>) -> Vec<Option<AuthorityDiscoveryId>> {
+			Vec::new()
+		}
+
+		fn check_validation_outputs(_: ParaId, _: ValidationOutputs) -> bool {
+			false
+		}
+
+		fn dmq_contents(_: ParaId) -> Vec<InboundDownwardMessage<BlockNumber>> {
+			Vec::new()
+		}
+
+		fn historical_validation_code(_: ParaId, _: BlockNumber) -> Option<ValidationCode> {
+			None
+		}
 	}
 }