Make CandidateHash a real type (#1916)

* Make `CandidateHash` a real type

This pr adds a new type `CandidateHash` that is used instead of the
opaque `Hash` type. This helps to ensure on the type system level that
we are passing the correct types.

This pr also fixes wrong usage of `relay_parent` as `candidate_hash`
when communicating with the av storage.

* Update core-primitives/src/lib.rs

Co-authored-by: Peter Goodspeed-Niklaus <coriolinus@users.noreply.github.com>

* Wrap the lines

Co-authored-by: Peter Goodspeed-Niklaus <coriolinus@users.noreply.github.com>

polkadot/node/network/availability-distribution/src/lib.rs

@@ -38,7 +38,7 @@ use polkadot_node_network_protocol::{
 use polkadot_node_subsystem_util::metrics::{self, prometheus};
 use polkadot_primitives::v1::{
 	BlakeTwo256, CommittedCandidateReceipt, CoreState, ErasureChunk, Hash, HashT, Id as ParaId,
-	SessionIndex, ValidatorId, ValidatorIndex, PARACHAIN_KEY_TYPE_ID,
+	SessionIndex, ValidatorId, ValidatorIndex, PARACHAIN_KEY_TYPE_ID, CandidateHash,
 };
 use polkadot_subsystem::messages::{
 	AllMessages, AvailabilityDistributionMessage, AvailabilityStoreMessage, ChainApiMessage,
@@ -130,7 +130,7 @@ const BENEFIT_VALID_MESSAGE: Rep = Rep::new(10, "Valid message");
 #[derive(Encode, Decode, Debug, Clone, PartialEq, Eq, Hash)]
 pub struct AvailabilityGossipMessage {
 	/// Anchor hash of the candidate the `ErasureChunk` is associated to.
-	pub candidate_hash: Hash,
+	pub candidate_hash: CandidateHash,
 	/// The erasure chunk, a encoded information part of `AvailabilityData`.
 	pub erasure_chunk: ErasureChunk,
 }
@@ -149,13 +149,13 @@ struct ProtocolState {
 	/// Caches a mapping of relay parents or ancestor to live candidate receipts.
 	/// Allows fast intersection of live candidates with views and consecutive unioning.
 	/// Maps relay parent / ancestor -> live candidate receipts + its hash.
-	receipts: HashMap<Hash, HashSet<(Hash, CommittedCandidateReceipt)>>,
+	receipts: HashMap<Hash, HashSet<(CandidateHash, CommittedCandidateReceipt)>>,
 
 	/// Allow reverse caching of view checks.
 	/// Maps candidate hash -> relay parent for extracting meta information from `PerRelayParent`.
 	/// Note that the presence of this is not sufficient to determine if deletion is OK, i.e.
 	/// two histories could cover this.
-	reverse: HashMap<Hash, Hash>,
+	reverse: HashMap<CandidateHash, Hash>,
 
 	/// Keeps track of which candidate receipts are required due to ancestors of which relay parents
 	/// of our view.
@@ -166,7 +166,7 @@ struct ProtocolState {
 	per_relay_parent: HashMap<Hash, PerRelayParent>,
 
 	/// Track data that is specific to a candidate.
-	per_candidate: HashMap<Hash, PerCandidate>,
+	per_candidate: HashMap<CandidateHash, PerCandidate>,
 }
 
 #[derive(Debug, Clone, Default)]
@@ -176,11 +176,11 @@ struct PerCandidate {
 	/// candidate hash + erasure chunk index -> gossip message
 	message_vault: HashMap<u32, AvailabilityGossipMessage>,
 
-	/// Track received candidate hashes and chunk indices from peers.
-	received_messages: HashMap<PeerId, HashSet<(Hash, ValidatorIndex)>>,
+	/// Track received candidate hashes and validator indices from peers.
+	received_messages: HashMap<PeerId, HashSet<(CandidateHash, ValidatorIndex)>>,
 
 	/// Track already sent candidate hashes and the erasure chunk index to the peers.
-	sent_messages: HashMap<PeerId, HashSet<(Hash, ValidatorIndex)>>,
+	sent_messages: HashMap<PeerId, HashSet<(CandidateHash, ValidatorIndex)>>,
 
 	/// The set of validators.
 	validators: Vec<ValidatorId>,
@@ -221,7 +221,7 @@ impl ProtocolState {
 	fn cached_live_candidates_unioned<'a>(
 		&'a self,
 		relay_parents: impl IntoIterator<Item = &'a Hash> + 'a,
-	) -> HashMap<Hash, CommittedCandidateReceipt> {
+	) -> HashMap<CandidateHash, CommittedCandidateReceipt> {
 		let relay_parents_and_ancestors = self.extend_with_ancestors(relay_parents);
 		relay_parents_and_ancestors
 			.into_iter()
@@ -229,7 +229,7 @@ impl ProtocolState {
 			.map(|receipt_set| receipt_set.into_iter())
 			.flatten()
 			.map(|(receipt_hash, receipt)| (receipt_hash.clone(), receipt.clone()))
-			.collect::<HashMap<Hash, CommittedCandidateReceipt>>()
+			.collect()
 	}
 
 	async fn add_relay_parent<Context>(
@@ -296,8 +296,9 @@ impl ProtocolState {
 				// remove from the ancestry index
 				self.ancestry.remove(relay_parent);
 				// and also remove the actual receipt
-				self.receipts.remove(relay_parent);
-				self.per_candidate.remove(relay_parent);
+				if let Some(candidates) = self.receipts.remove(relay_parent) {
+					candidates.into_iter().for_each(|c| { self.per_candidate.remove(&c.0); });
+				}
 			}
 		}
 		if let Some(per_relay_parent) = self.per_relay_parent.remove(relay_parent) {
@@ -313,8 +314,9 @@ impl ProtocolState {
 						// remove from the ancestry index
 						self.ancestry.remove(&ancestor);
 						// and also remove the actual receipt
-						self.receipts.remove(&ancestor);
-						self.per_candidate.remove(&ancestor);
+						if let Some(candidates) = self.receipts.remove(&ancestor) {
+							candidates.into_iter().for_each(|c| { self.per_candidate.remove(&c.0); });
+						}
 					}
 				}
 			}
@@ -645,8 +647,7 @@ where
 	let live_candidates = state.cached_live_candidates_unioned(state.view.0.iter());
 
 	// check if the candidate is of interest
-	let live_candidate = if let Some(live_candidate) = live_candidates.get(&message.candidate_hash)
-	{
+	let live_candidate = if let Some(live_candidate) = live_candidates.get(&message.candidate_hash) {
 		live_candidate
 	} else {
 		return modify_reputation(ctx, origin, COST_NOT_A_LIVE_CANDIDATE).await;
@@ -862,7 +863,7 @@ async fn query_live_candidates<Context>(
 	ctx: &mut Context,
 	state: &mut ProtocolState,
 	relay_parents: impl IntoIterator<Item = Hash>,
-) -> Result<HashMap<Hash, (Hash, CommittedCandidateReceipt)>>
+) -> Result<HashMap<Hash, (CandidateHash, CommittedCandidateReceipt)>>
 where
 	Context: SubsystemContext<Message = AvailabilityDistributionMessage>,
 {
@@ -871,7 +872,7 @@ where
 
 	let capacity = hint.1.unwrap_or(hint.0) * (1 + AvailabilityDistributionSubsystem::K);
 	let mut live_candidates =
-		HashMap::<Hash, (Hash, CommittedCandidateReceipt)>::with_capacity(capacity);
+		HashMap::<Hash, (CandidateHash, CommittedCandidateReceipt)>::with_capacity(capacity);
 
 	for relay_parent in iter {
 		// register one of relay parents (not the ancestors)
@@ -969,7 +970,7 @@ where
 }
 
 /// Query the proof of validity for a particular candidate hash.
-async fn query_data_availability<Context>(ctx: &mut Context, candidate_hash: Hash) -> Result<bool>
+async fn query_data_availability<Context>(ctx: &mut Context, candidate_hash: CandidateHash) -> Result<bool>
 where
 	Context: SubsystemContext<Message = AvailabilityDistributionMessage>,
 {
@@ -985,7 +986,7 @@ where
 
 async fn query_chunk<Context>(
 	ctx: &mut Context,
-	candidate_hash: Hash,
+	candidate_hash: CandidateHash,
 	validator_index: ValidatorIndex,
 ) -> Result<Option<ErasureChunk>>
 where
@@ -1002,7 +1003,7 @@ where
 
 async fn store_chunk<Context>(
 	ctx: &mut Context,
-	candidate_hash: Hash,
+	candidate_hash: CandidateHash,
 	relay_parent: Hash,
 	validator_index: ValidatorIndex,
 	erasure_chunk: ErasureChunk,
@@ -1012,18 +1013,18 @@ where
 {
 	let (tx, rx) = oneshot::channel();
 	ctx.send_message(
-        AllMessages::AvailabilityStore(
-                AvailabilityStoreMessage::StoreChunk {
-                candidate_hash,
-                relay_parent,
-                validator_index,
-                chunk: erasure_chunk,
-                tx,
-            }
-        )).await
-        .map_err(|e| Error::StoreChunkSendQuery(e))?;
+		AllMessages::AvailabilityStore(
+				AvailabilityStoreMessage::StoreChunk {
+				candidate_hash,
+				relay_parent,
+				validator_index,
+				chunk: erasure_chunk,
+				tx,
+			}
+		)).await
+		.map_err(|e| Error::StoreChunkSendQuery(e))?;
 
-    rx.await.map_err(|e| Error::StoreChunkResponseChannel(e))
+	rx.await.map_err(|e| Error::StoreChunkResponseChannel(e))
 }
 
 /// Request the head data for a particular para.

polkadot/node/network/availability-distribution/src/tests.rs

@@ -254,7 +254,7 @@ fn make_erasure_root(test: &TestState, pov: PoV) -> Hash {
 
 fn make_valid_availability_gossip(
 	test: &TestState,
-	candidate_hash: Hash,
+	candidate_hash: CandidateHash,
 	erasure_chunk_index: u32,
 	pov: PoV,
 ) -> AvailabilityGossipMessage {
@@ -320,7 +320,7 @@ fn helper_integrity() {
 	.build();
 
 	let message =
-		make_valid_availability_gossip(&test_state, dbg!(candidate.hash()), 2, pov_block.clone());
+		make_valid_availability_gossip(&test_state, candidate.hash(), 2, pov_block.clone());
 
 	let root = dbg!(&candidate.commitments.erasure_root);
 

polkadot/node/network/protocol/src/lib.rs

@@ -186,7 +186,7 @@ impl View {
 pub mod v1 {
 	use polkadot_primitives::v1::{
 		Hash, CollatorId, Id as ParaId, ErasureChunk, CandidateReceipt,
-		SignedAvailabilityBitfield, PoV,
+		SignedAvailabilityBitfield, PoV, CandidateHash,
 	};
 	use polkadot_node_primitives::SignedFullStatement;
 	use parity_scale_codec::{Encode, Decode};
@@ -198,7 +198,7 @@ pub mod v1 {
 	pub enum AvailabilityDistributionMessage {
 		/// An erasure chunk for a given candidate hash.
 		#[codec(index = "0")]
-		Chunk(Hash, ErasureChunk),
+		Chunk(CandidateHash, ErasureChunk),
 	}
 
 	/// Network messages used by the bitfield distribution subsystem.

polkadot/node/network/statement-distribution/src/lib.rs

@@ -35,7 +35,7 @@ use polkadot_node_subsystem_util::{
 };
 use node_primitives::SignedFullStatement;
 use polkadot_primitives::v1::{
-	Hash, CompactStatement, ValidatorIndex, ValidatorId, SigningContext, ValidatorSignature,
+	Hash, CompactStatement, ValidatorIndex, ValidatorId, SigningContext, ValidatorSignature, CandidateHash,
 };
 use polkadot_node_network_protocol::{
 	v1 as protocol_v1, View, PeerId, ReputationChange as Rep, NetworkBridgeEvent,
@@ -102,32 +102,32 @@ impl StatementDistribution {
 /// via other means.
 #[derive(Default)]
 struct VcPerPeerTracker {
-	local_observed: arrayvec::ArrayVec<[Hash; VC_THRESHOLD]>,
-	remote_observed: arrayvec::ArrayVec<[Hash; VC_THRESHOLD]>,
+	local_observed: arrayvec::ArrayVec<[CandidateHash; VC_THRESHOLD]>,
+	remote_observed: arrayvec::ArrayVec<[CandidateHash; VC_THRESHOLD]>,
 }
 
 impl VcPerPeerTracker {
-	// Note that the remote should now be aware that a validator has seconded a given candidate (by hash)
-	// based on a message that we have sent it from our local pool.
-	fn note_local(&mut self, h: Hash) {
+	/// Note that the remote should now be aware that a validator has seconded a given candidate (by hash)
+	/// based on a message that we have sent it from our local pool.
+	fn note_local(&mut self, h: CandidateHash) {
 		if !note_hash(&mut self.local_observed, h) {
 			log::warn!("Statement distribution is erroneously attempting to distribute more \
 				than {} candidate(s) per validator index. Ignoring", VC_THRESHOLD);
 		}
 	}
 
-	// Note that the remote should now be aware that a validator has seconded a given candidate (by hash)
-	// based on a message that it has sent us.
-	//
-	// Returns `true` if the peer was allowed to send us such a message, `false` otherwise.
-	fn note_remote(&mut self, h: Hash) -> bool {
+	/// Note that the remote should now be aware that a validator has seconded a given candidate (by hash)
+	/// based on a message that it has sent us.
+	///
+	/// Returns `true` if the peer was allowed to send us such a message, `false` otherwise.
+	fn note_remote(&mut self, h: CandidateHash) -> bool {
 		note_hash(&mut self.remote_observed, h)
 	}
 }
 
 fn note_hash(
-	observed: &mut arrayvec::ArrayVec<[Hash; VC_THRESHOLD]>,
-	h: Hash,
+	observed: &mut arrayvec::ArrayVec<[CandidateHash; VC_THRESHOLD]>,
+	h: CandidateHash,
 ) -> bool {
 	if observed.contains(&h) { return true; }
 
@@ -139,7 +139,7 @@ fn note_hash(
 struct PeerRelayParentKnowledge {
 	/// candidates that the peer is aware of. This indicates that we can
 	/// send other statements pertaining to that candidate.
-	known_candidates: HashSet<Hash>,
+	known_candidates: HashSet<CandidateHash>,
 	/// fingerprints of all statements a peer should be aware of: those that
 	/// were sent to the peer by us.
 	sent_statements: HashSet<(CompactStatement, ValidatorIndex)>,
@@ -149,7 +149,7 @@ struct PeerRelayParentKnowledge {
 	/// How many candidates this peer is aware of for each given validator index.
 	seconded_counts: HashMap<ValidatorIndex, VcPerPeerTracker>,
 	/// How many statements we've received for each candidate that we're aware of.
-	received_message_count: HashMap<Hash, usize>,
+	received_message_count: HashMap<CandidateHash, usize>,
 }
 
 impl PeerRelayParentKnowledge {
@@ -246,7 +246,7 @@ impl PeerRelayParentKnowledge {
 
 		{
 			let received_per_candidate = self.received_message_count
-				.entry(candidate_hash.clone())
+				.entry(*candidate_hash)
 				.or_insert(0);
 
 			if *received_per_candidate >= max_message_count {
@@ -372,7 +372,7 @@ enum NotedStatement<'a> {
 
 struct ActiveHeadData {
 	/// All candidates we are aware of for this head, keyed by hash.
-	candidates: HashSet<Hash>,
+	candidates: HashSet<CandidateHash>,
 	/// Stored statements for circulation to peers.
 	///
 	/// These are iterable in insertion order, and `Seconded` statements are always
@@ -464,7 +464,7 @@ impl ActiveHeadData {
 	}
 
 	/// Get an iterator over all statements for the active head that are for a particular candidate.
-	fn statements_about(&self, candidate_hash: Hash)
+	fn statements_about(&self, candidate_hash: CandidateHash)
 		-> impl Iterator<Item = &'_ StoredStatement> + '_
 	{
 		self.statements().filter(move |s| s.compact().candidate_hash() == &candidate_hash)
@@ -521,10 +521,10 @@ async fn circulate_statement_and_dependents(
 
 		// First circulate the statement directly to all peers needing it.
 		// The borrow of `active_head` needs to encompass only this (Rust) statement.
-		let outputs: Option<(Hash, Vec<PeerId>)> = {
+		let outputs: Option<(CandidateHash, Vec<PeerId>)> = {
 			match active_head.note_statement(statement) {
 				NotedStatement::Fresh(stored) => Some((
-					stored.compact().candidate_hash().clone(),
+					*stored.compact().candidate_hash(),
 					circulate_statement(peers, ctx, relay_parent, stored).await?,
 				)),
 				_ => None,
@@ -602,7 +602,7 @@ async fn send_statements_about(
 	peer_data: &mut PeerData,
 	ctx: &mut impl SubsystemContext<Message = StatementDistributionMessage>,
 	relay_parent: Hash,
-	candidate_hash: Hash,
+	candidate_hash: CandidateHash,
 	active_head: &ActiveHeadData,
 	metrics: &Metrics,
 ) -> SubsystemResult<()> {
@@ -1110,8 +1110,8 @@ mod tests {
 
 	#[test]
 	fn note_local_works() {
-		let hash_a: Hash = [1; 32].into();
-		let hash_b: Hash = [2; 32].into();
+		let hash_a = CandidateHash([1; 32].into());
+		let hash_b = CandidateHash([2; 32].into());
 
 		let mut per_peer_tracker = VcPerPeerTracker::default();
 		per_peer_tracker.note_local(hash_a.clone());
@@ -1126,9 +1126,9 @@ mod tests {
 
 	#[test]
 	fn note_remote_works() {
-		let hash_a: Hash = [1; 32].into();
-		let hash_b: Hash = [2; 32].into();
-		let hash_c: Hash = [3; 32].into();
+		let hash_a = CandidateHash([1; 32].into());
+		let hash_b = CandidateHash([2; 32].into());
+		let hash_c = CandidateHash([3; 32].into());
 
 		let mut per_peer_tracker = VcPerPeerTracker::default();
 		assert!(per_peer_tracker.note_remote(hash_a.clone()));
@@ -1148,7 +1148,7 @@ mod tests {
 	fn per_peer_relay_parent_knowledge_send() {
 		let mut knowledge = PeerRelayParentKnowledge::default();
 
-		let hash_a: Hash = [1; 32].into();
+		let hash_a = CandidateHash([1; 32].into());
 
 		// Sending an un-pinned statement should not work and should have no effect.
 		assert!(knowledge.send(&(CompactStatement::Valid(hash_a), 0)).is_none());
@@ -1180,7 +1180,7 @@ mod tests {
 	fn cant_send_after_receiving() {
 		let mut knowledge = PeerRelayParentKnowledge::default();
 
-		let hash_a: Hash = [1; 32].into();
+		let hash_a = CandidateHash([1; 32].into());
 		assert!(knowledge.receive(&(CompactStatement::Candidate(hash_a), 0), 3).unwrap());
 		assert!(knowledge.send(&(CompactStatement::Candidate(hash_a), 0)).is_none());
 	}
@@ -1189,7 +1189,7 @@ mod tests {
 	fn per_peer_relay_parent_knowledge_receive() {
 		let mut knowledge = PeerRelayParentKnowledge::default();
 
-		let hash_a: Hash = [1; 32].into();
+		let hash_a = CandidateHash([1; 32].into());
 
 		assert_eq!(
 			knowledge.receive(&(CompactStatement::Valid(hash_a), 0), 3),
@@ -1226,8 +1226,8 @@ mod tests {
 		assert_eq!(knowledge.received_statements.len(), 3); // number of prior `Ok`s.
 
 		// Now make sure that the seconding limit is respected.
-		let hash_b: Hash = [2; 32].into();
-		let hash_c: Hash = [3; 32].into();
+		let hash_b = CandidateHash([2; 32].into());
+		let hash_c = CandidateHash([3; 32].into());
 
 		assert_eq!(
 			knowledge.receive(&(CompactStatement::Candidate(hash_b), 0), 3),