Make a collator send a collation as backup as well (#1353)

Currently a collator will only send a collation to validators it is a
primary for. While testing this could lead to the situation that the
same collator was registered as prime for all Parachain validators but
failed for other reasons to generate a PoVBlock. However no other
collator was sending a collation, which stopped the Parachain until the
faulty collator was stopped.

This pr solves this problem by making sure that every collator sends a
collation to one of his validators he is connected to, but registered as backup.

network/Cargo.toml

@@ -26,6 +26,7 @@ futures-timer = "2.0"
 sp-blockchain = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-api = { git = "https://github.com/paritytech/substrate", branch = "master" }
 wasm-timer = "0.2.4"
+rand = "0.7.3"
 
 [dev-dependencies]
 sp-keyring = { git = "https://github.com/paritytech/substrate", branch = "master" }

network/src/legacy/local_collations.rs

@@ -24,6 +24,7 @@ use crate::legacy::collator_pool::Role;
 use std::collections::{HashMap, HashSet};
 use std::time::Duration;
 use wasm_timer::Instant;
+use rand::seq::SliceRandom;
 
 const LIVE_FOR: Duration = Duration::from_secs(60 * 5);
 
@@ -106,9 +107,7 @@ impl<C: Clone> LocalCollations<C> {
 		relay_parent: Hash,
 		targets: HashSet<ValidatorId>,
 		collation: C
-	)
-		-> impl Iterator<Item=(ValidatorId, C)> + 'a
-	{
+	) -> impl Iterator<Item=(ValidatorId, C)> + 'a {
 		self.local_collations.insert(relay_parent, LocalCollation {
 			targets,
 			collation,
@@ -119,8 +118,17 @@ impl<C: Clone> LocalCollations<C> {
 			.expect("just inserted to this key; qed");
 
 		let borrowed_collation = &local.collation;
+
+		// If we are conected to multiple validators,
+		// make sure we always send the collation to one of the validators
+		// we are registered as backup. This ensures that one collator that
+		// is primary at multiple validators, desn't block the Parachain from progressing.
+		let mut rng = rand::thread_rng();
+		let diff = local.targets.difference(&self.primary_for).collect::<Vec<_>>();
+
 		local.targets
 			.intersection(&self.primary_for)
+			.chain(diff.choose(&mut rng).map(|r| r.clone()))
 			.map(move |k| (k.clone(), borrowed_collation.clone()))
 	}
 
@@ -149,7 +157,7 @@ mod tests {
 		};
 
 		let mut tracker = LocalCollations::new();
-		assert!(tracker.add_collation(relay_parent, targets, 5).next().is_none());
+		assert!(tracker.add_collation(relay_parent, targets, 5).next().is_some());
 		assert_eq!(tracker.note_validator_role(key, Role::Primary), vec![(relay_parent, 5)]);
 	}
 
@@ -165,7 +173,7 @@ mod tests {
 		};
 
 		let mut tracker: LocalCollations<u8> = LocalCollations::new();
-		assert!(tracker.add_collation(relay_parent, targets, 5).next().is_none());
+		assert!(tracker.add_collation(relay_parent, targets, 5).next().is_some());
 		assert!(tracker.note_validator_role(orig_key.clone(), Role::Primary).is_empty());
 		assert_eq!(tracker.fresh_key(&orig_key, &new_key), vec![(relay_parent, 5u8)]);
 	}