Resolve some todos in async backing statement-distribution branch

node/core/backing/src/lib.rs

@@ -80,8 +80,8 @@ use futures::{
 
 use error::{Error, FatalResult};
 use polkadot_node_primitives::{
-	AvailableData, InvalidCandidate, PoV, SignedFullStatementWithPVD, StatementWithPVD,
-	ValidationResult, BACKING_EXECUTION_TIMEOUT,
+	minimum_votes, AvailableData, InvalidCandidate, PoV, SignedFullStatementWithPVD,
+	StatementWithPVD, ValidationResult, BACKING_EXECUTION_TIMEOUT,
 };
 use polkadot_node_subsystem::{
 	messages::{
@@ -374,14 +374,6 @@ struct AttestingData {
 	backing: Vec<ValidatorIndex>,
 }
 
-/// How many votes we need to consider a candidate backed.
-///
-/// WARNING: This has to be kept in sync with the runtime check in the inclusion module and
-/// statement distribution.
-fn minimum_votes(n_validators: usize) -> usize {
-	std::cmp::min(2, n_validators)
-}
-
 #[derive(Default)]
 struct TableContext {
 	validator: Option<Validator>,
@@ -1579,9 +1571,9 @@ async fn import_statement<Context>(
 					})
 					.await;
 
-					// TODO [now]: notify statement distribution of backed
-					// candidate. alter control flow so "Share" is always sent
-					// first.
+				// TODO [now]: notify statement distribution of backed
+				// candidate. alter control flow so "Share" is always sent
+				// first.
 				} else {
 					// The provisioner waits on candidate-backing, which means
 					// that we need to send unbounded messages to avoid cycles.

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

@@ -271,7 +271,8 @@ impl<R: rand::Rng> StatementDistributionSubsystem<R> {
 							ctx,
 							unimplemented!(), // TODO [now] state
 							candidate_hash,
-						).await;
+						)
+						.await;
 					},
 				},
 		}

node/network/statement-distribution/src/vstaging/candidates.rs

@@ -263,7 +263,9 @@ impl Candidates {
 
 	/// Note that a candidate is backed. No-op if the candidate is not confirmed.
 	pub fn note_backed(&mut self, candidate_hash: &CandidateHash) {
-		if let Some(&mut CandidateState::Confirmed(ref mut c)) = self.candidates.get_mut(candidate_hash) {
+		if let Some(&mut CandidateState::Confirmed(ref mut c)) =
+			self.candidates.get_mut(candidate_hash)
+		{
 			c.backed = true;
 		}
 	}

node/network/statement-distribution/src/vstaging/cluster.rs

@@ -172,6 +172,7 @@ impl ClusterTracker {
 	}
 
 	/// Note that we accepted an incoming statement. This updates internal structures.
+	///
 	/// Should only be called after a successful `can_receive` call.
 	pub fn note_received(
 		&mut self,
@@ -748,5 +749,55 @@ mod tests {
 		);
 	}
 
-	// TODO [now] ensure statements received with prejudice don't prevent sending later
+	// Ensure statements received with prejudice don't prevent sending later.
+	#[test]
+	fn can_send_statements_received_with_prejudice() {
+		let group =
+			vec![ValidatorIndex(5), ValidatorIndex(200), ValidatorIndex(24), ValidatorIndex(146)];
+
+		let seconding_limit = 1;
+
+		let mut tracker = ClusterTracker::new(group.clone(), seconding_limit).expect("not empty");
+		let hash_a = CandidateHash(Hash::repeat_byte(1));
+		let hash_b = CandidateHash(Hash::repeat_byte(2));
+
+		assert_eq!(
+			tracker.can_receive(
+				ValidatorIndex(200),
+				ValidatorIndex(5),
+				CompactStatement::Seconded(hash_a),
+			),
+			Ok(Accept::Ok),
+		);
+
+		tracker.note_received(
+			ValidatorIndex(200),
+			ValidatorIndex(5),
+			CompactStatement::Seconded(hash_a),
+		);
+
+		assert_eq!(
+			tracker.can_receive(
+				ValidatorIndex(24),
+				ValidatorIndex(5),
+				CompactStatement::Seconded(hash_b),
+			),
+			Ok(Accept::WithPrejudice),
+		);
+
+		tracker.note_received(
+			ValidatorIndex(24),
+			ValidatorIndex(5),
+			CompactStatement::Seconded(hash_b),
+		);
+
+		assert_eq!(
+			tracker.can_send(
+				ValidatorIndex(24),
+				ValidatorIndex(5),
+				CompactStatement::Seconded(hash_a),
+			),
+			Ok(()),
+		);
+	}
 }

node/network/statement-distribution/src/vstaging/grid.rs

@@ -213,9 +213,11 @@ impl GridTracker {
 			//   * They are in the sending set for the group AND we have sent them
 			//     a manifest AND the received manifest is partial.
 			ManifestKind::Full => is_receiver,
-			ManifestKind::Acknowledgement => is_sender && self.confirmed_backed
-				.get(&candidate_hash)
-				.map_or(false, |c| c.has_sent_manifest_to(sender)),
+			ManifestKind::Acknowledgement =>
+				is_sender &&
+					self.confirmed_backed
+						.get(&candidate_hash)
+						.map_or(false, |c| c.has_sent_manifest_to(sender)),
 		};
 
 		if !manifest_allowed {
@@ -267,7 +269,8 @@ impl GridTracker {
 		if let Some(confirmed) = self.confirmed_backed.get_mut(&candidate_hash) {
 			// TODO [now]: send statements they need if this is an ack
 			if is_receiver && !confirmed.has_sent_manifest_to(sender) {
-				self.pending_communication.entry(sender)
+				self.pending_communication
+					.entry(sender)
 					.or_default()
 					.insert(candidate_hash, ManifestKind::Acknowledgement);
 
@@ -317,7 +320,6 @@ impl GridTracker {
 				.and_then(|r| r.candidate_statement_filter(&candidate_hash))
 				.expect("unconfirmed is only populated by validators who have sent manifest; qed");
 
-
 			// No need to send direct statements, because our local knowledge is `None`
 			c.manifest_received_from(v, statement_filter);
 		}
@@ -329,25 +331,25 @@ impl GridTracker {
 
 		// advertise onwards ad accept received advertisements
 
-		let sending_group_manifests = group_topology
-			.sending
-			.iter()
-			.map(|v| (*v, ManifestKind::Full));
+		let sending_group_manifests =
+			group_topology.sending.iter().map(|v| (*v, ManifestKind::Full));
 
-		let receiving_group_manifests = group_topology
-			.receiving
-			.iter()
-			.filter_map(|v| if c.has_received_manifest_from(*v) {
+		let receiving_group_manifests = group_topology.receiving.iter().filter_map(|v| {
+			if c.has_received_manifest_from(*v) {
 				Some((*v, ManifestKind::Acknowledgement))
 			} else {
 				None
-			});
+			}
+		});
 
 		// Note that order is important: if a validator is part of both the sending
 		// and receiving groups, we may overwrite a `Full` manifest with a `Acknowledgement`
 		// one.
 		for (v, manifest_mode) in sending_group_manifests.chain(receiving_group_manifests) {
-			self.pending_communication.entry(v).or_default().insert(candidate_hash, manifest_mode);
+			self.pending_communication
+				.entry(v)
+				.or_default()
+				.insert(candidate_hash, manifest_mode);
 		}
 
 		self.pending_communication
@@ -1064,7 +1066,7 @@ mod tests {
 		};
 
 		let groups = Groups::new(
-			vec![vec![ValidatorIndex(0), ValidatorIndex(1), ValidatorIndex(2)]],
+			vec![vec![ValidatorIndex(0), ValidatorIndex(1), ValidatorIndex(2)]].into(),
 			&[
 				AuthorityDiscoveryPair::generate().0.public(),
 				AuthorityDiscoveryPair::generate().0.public(),
@@ -1090,6 +1092,7 @@ mod tests {
 					seconded_in_group: bitvec::bitvec![u8, Lsb0; 0, 1, 0],
 					validated_in_group: bitvec::bitvec![u8, Lsb0; 1, 0, 1],
 				},
+				ManifestKind::Full,
 				ValidatorIndex(1),
 			),
 			Err(ManifestImportError::Disallowed)
@@ -1109,6 +1112,7 @@ mod tests {
 					seconded_in_group: bitvec::bitvec![u8, Lsb0; 0, 1, 0],
 					validated_in_group: bitvec::bitvec![u8, Lsb0; 1, 0, 1],
 				},
+				ManifestKind::Full,
 				ValidatorIndex(0),
 			),
 			Err(ManifestImportError::Disallowed)
@@ -1131,7 +1135,7 @@ mod tests {
 		};
 
 		let groups = Groups::new(
-			vec![vec![ValidatorIndex(0), ValidatorIndex(1), ValidatorIndex(2)]],
+			vec![vec![ValidatorIndex(0), ValidatorIndex(1), ValidatorIndex(2)]].into(),
 			&[
 				AuthorityDiscoveryPair::generate().0.public(),
 				AuthorityDiscoveryPair::generate().0.public(),
@@ -1155,6 +1159,7 @@ mod tests {
 					seconded_in_group: bitvec::bitvec![u8, Lsb0; 0, 1, 0, 1],
 					validated_in_group: bitvec::bitvec![u8, Lsb0; 1, 0, 1],
 				},
+				ManifestKind::Full,
 				ValidatorIndex(0),
 			),
 			Err(ManifestImportError::Malformed)
@@ -1172,6 +1177,7 @@ mod tests {
 					seconded_in_group: bitvec::bitvec![u8, Lsb0; 0, 1, 0],
 					validated_in_group: bitvec::bitvec![u8, Lsb0; 1, 0, 1, 0],
 				},
+				ManifestKind::Full,
 				ValidatorIndex(0),
 			),
 			Err(ManifestImportError::Malformed)
@@ -1194,7 +1200,7 @@ mod tests {
 		};
 
 		let groups = Groups::new(
-			vec![vec![ValidatorIndex(0), ValidatorIndex(1), ValidatorIndex(2)]],
+			vec![vec![ValidatorIndex(0), ValidatorIndex(1), ValidatorIndex(2)]].into(),
 			&[
 				AuthorityDiscoveryPair::generate().0.public(),
 				AuthorityDiscoveryPair::generate().0.public(),
@@ -1218,6 +1224,7 @@ mod tests {
 					seconded_in_group: bitvec::bitvec![u8, Lsb0; 0, 0, 0],
 					validated_in_group: bitvec::bitvec![u8, Lsb0; 1, 1, 1],
 				},
+				ManifestKind::Full,
 				ValidatorIndex(0),
 			),
 			Err(ManifestImportError::Malformed)
@@ -1240,7 +1247,7 @@ mod tests {
 		};
 
 		let groups = Groups::new(
-			vec![vec![ValidatorIndex(0), ValidatorIndex(1), ValidatorIndex(2)]],
+			vec![vec![ValidatorIndex(0), ValidatorIndex(1), ValidatorIndex(2)]].into(),
 			&[
 				AuthorityDiscoveryPair::generate().0.public(),
 				AuthorityDiscoveryPair::generate().0.public(),
@@ -1266,6 +1273,7 @@ mod tests {
 					seconded_in_group: bitvec::bitvec![u8, Lsb0; 0, 0, 1],
 					validated_in_group: bitvec::bitvec![u8, Lsb0; 0, 0, 0],
 				},
+				ManifestKind::Full,
 				ValidatorIndex(0),
 			),
 			Err(ManifestImportError::Malformed)
@@ -1285,6 +1293,7 @@ mod tests {
 					seconded_in_group: bitvec::bitvec![u8, Lsb0; 0, 0, 1],
 					validated_in_group: bitvec::bitvec![u8, Lsb0; 0, 0, 1],
 				},
+				ManifestKind::Full,
 				ValidatorIndex(0),
 			),
 			Err(ManifestImportError::Malformed)
@@ -1304,9 +1313,10 @@ mod tests {
 					seconded_in_group: bitvec::bitvec![u8, Lsb0; 0, 0, 1],
 					validated_in_group: bitvec::bitvec![u8, Lsb0; 0, 1, 0],
 				},
+				ManifestKind::Full,
 				ValidatorIndex(0),
 			),
-			Ok(())
+			Ok(None)
 		);
 	}
 

node/network/statement-distribution/src/vstaging/groups.rs

@@ -16,6 +16,7 @@
 
 //! A utility for tracking groups and their members within a session.
 
+use polkadot_node_primitives::minimum_votes;
 use polkadot_primitives::vstaging::{AuthorityDiscoveryId, GroupIndex, IndexedVec, ValidatorIndex};
 
 use std::collections::HashMap;
@@ -68,7 +69,7 @@ impl Groups {
 		&self,
 		group_index: GroupIndex,
 	) -> Option<(usize, usize)> {
-		self.get(group_index).map(|g| (g.len(), super::minimum_votes(g.len())))
+		self.get(group_index).map(|g| (g.len(), minimum_votes(g.len())))
 	}
 
 	/// Get the group index for a validator by index.