Ignore redundant dispute messages (#4854)

* ignore duplicate dispute votes

* fmt and TODOs

* tests

* fmt

polkadot/node/core/dispute-coordinator/src/real/initialized.rs

@@ -779,14 +779,21 @@ impl Initialized {
 		// There is one exception: A sufficiently sophisticated attacker could prevent
 		// us from seeing the backing votes by witholding arbitrary blocks, and hence we do
 		// not have a `CandidateReceipt` available.
-		let mut votes = match overlay_db
+		let (mut votes, mut votes_changed) = match overlay_db
 			.load_candidate_votes(session, &candidate_hash)?
 			.map(CandidateVotes::from)
 		{
-			Some(votes) => votes,
+			Some(votes) => (votes, false),
 			None =>
 				if let MaybeCandidateReceipt::Provides(candidate_receipt) = candidate_receipt {
-					CandidateVotes { candidate_receipt, valid: Vec::new(), invalid: Vec::new() }
+					(
+						CandidateVotes {
+							candidate_receipt,
+							valid: Vec::new(),
+							invalid: Vec::new(),
+						},
+						true,
+					)
 				} else {
 					tracing::warn!(
 						target: LOG_TARGET,
@@ -841,22 +848,34 @@ impl Initialized {
 
 			match statement.statement().clone() {
 				DisputeStatement::Valid(valid_kind) => {
-					self.metrics.on_valid_vote();
-					insert_into_statement_vec(
+					let fresh = insert_into_statement_vec(
 						&mut votes.valid,
 						valid_kind,
 						*val_index,
 						statement.validator_signature().clone(),
 					);
+
+					if !fresh {
+						continue
+					}
+
+					votes_changed = true;
+					self.metrics.on_valid_vote();
 				},
 				DisputeStatement::Invalid(invalid_kind) => {
-					self.metrics.on_invalid_vote();
-					insert_into_statement_vec(
+					let fresh = insert_into_statement_vec(
 						&mut votes.invalid,
 						invalid_kind,
 						*val_index,
 						statement.validator_signature().clone(),
 					);
+
+					if !fresh {
+						continue
+					}
+
+					votes_changed = true;
+					self.metrics.on_invalid_vote();
 				},
 			}
 		}
@@ -871,7 +890,7 @@ impl Initialized {
 
 		// Potential spam:
 		if !is_confirmed {
-			let mut free_spam_slots = false;
+			let mut free_spam_slots = statements.is_empty();
 			for (statement, index) in statements.iter() {
 				free_spam_slots |= statement.statement().is_backing() ||
 					self.spam_slots.add_unconfirmed(session, candidate_hash, *index);
@@ -988,7 +1007,10 @@ impl Initialized {
 			overlay_db.write_recent_disputes(recent_disputes);
 		}
 
-		overlay_db.write_candidate_votes(session, candidate_hash, votes.into());
+		// Only write when votes have changed.
+		if votes_changed {
+			overlay_db.write_candidate_votes(session, candidate_hash, votes.into());
+		}
 
 		Ok(ImportStatementsResult::ValidImport)
 	}
@@ -1145,18 +1167,21 @@ impl MuxedMessage {
 	}
 }
 
+// Returns 'true' if no other vote by that validator was already
+// present and 'false' otherwise. Same semantics as `HashSet`.
 fn insert_into_statement_vec<T>(
 	vec: &mut Vec<(T, ValidatorIndex, ValidatorSignature)>,
 	tag: T,
 	val_index: ValidatorIndex,
 	val_signature: ValidatorSignature,
-) {
+) -> bool {
 	let pos = match vec.binary_search_by_key(&val_index, |x| x.1) {
-		Ok(_) => return, // no duplicates needed.
+		Ok(_) => return false, // no duplicates needed.
 		Err(p) => p,
 	};
 
 	vec.insert(pos, (tag, val_index, val_signature));
+	true
 }
 
 #[derive(Debug, Clone)]

polkadot/node/core/dispute-coordinator/src/real/tests.rs

@@ -1548,3 +1548,112 @@ fn negative_issue_local_statement_only_triggers_import() {
 		})
 	});
 }
+
+#[test]
+fn empty_import_still_writes_candidate_receipt() {
+	test_harness(|mut test_state, mut virtual_overseer| {
+		Box::pin(async move {
+			let session = 1;
+
+			test_state.handle_resume_sync(&mut virtual_overseer, session).await;
+
+			let candidate_receipt = make_valid_candidate_receipt();
+			let candidate_hash = candidate_receipt.hash();
+
+			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+
+			let (tx, rx) = oneshot::channel();
+			virtual_overseer
+				.send(FromOverseer::Communication {
+					msg: DisputeCoordinatorMessage::ImportStatements {
+						candidate_hash,
+						candidate_receipt: candidate_receipt.clone(),
+						session,
+						statements: Vec::new(),
+						pending_confirmation: tx,
+					},
+				})
+				.await;
+
+			rx.await.unwrap();
+
+			let backend = DbBackend::new(test_state.db.clone(), test_state.config.column_config());
+
+			let votes = backend.load_candidate_votes(session, &candidate_hash).unwrap().unwrap();
+			assert_eq!(votes.invalid.len(), 0);
+			assert_eq!(votes.valid.len(), 0);
+			assert_eq!(votes.candidate_receipt, candidate_receipt);
+
+			virtual_overseer.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
+			assert!(virtual_overseer.try_recv().await.is_none());
+
+			test_state
+		})
+	});
+}
+
+#[test]
+fn redundant_votes_ignored() {
+	test_harness(|mut test_state, mut virtual_overseer| {
+		Box::pin(async move {
+			let session = 1;
+
+			test_state.handle_resume_sync(&mut virtual_overseer, session).await;
+
+			let candidate_receipt = make_valid_candidate_receipt();
+			let candidate_hash = candidate_receipt.hash();
+
+			test_state.activate_leaf_at_session(&mut virtual_overseer, session, 1).await;
+
+			let valid_vote =
+				test_state.issue_statement_with_index(1, candidate_hash, session, true).await;
+
+			let valid_vote_2 =
+				test_state.issue_statement_with_index(1, candidate_hash, session, true).await;
+
+			assert!(valid_vote.validator_signature() != valid_vote_2.validator_signature());
+
+			let (tx, rx) = oneshot::channel();
+			virtual_overseer
+				.send(FromOverseer::Communication {
+					msg: DisputeCoordinatorMessage::ImportStatements {
+						candidate_hash,
+						candidate_receipt: candidate_receipt.clone(),
+						session,
+						statements: vec![(valid_vote.clone(), ValidatorIndex(1))],
+						pending_confirmation: tx,
+					},
+				})
+				.await;
+
+			rx.await.unwrap();
+
+			let (tx, rx) = oneshot::channel();
+			virtual_overseer
+				.send(FromOverseer::Communication {
+					msg: DisputeCoordinatorMessage::ImportStatements {
+						candidate_hash,
+						candidate_receipt: candidate_receipt.clone(),
+						session,
+						statements: vec![(valid_vote_2, ValidatorIndex(1))],
+						pending_confirmation: tx,
+					},
+				})
+				.await;
+
+			rx.await.unwrap();
+
+			let backend = DbBackend::new(test_state.db.clone(), test_state.config.column_config());
+
+			let votes = backend.load_candidate_votes(session, &candidate_hash).unwrap().unwrap();
+			assert_eq!(votes.invalid.len(), 0);
+			assert_eq!(votes.valid.len(), 1);
+			assert_eq!(&votes.valid[0].2, valid_vote.validator_signature());
+
+			virtual_overseer.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
+			assert!(virtual_overseer.try_recv().await.is_none());
+
+			test_state
+		})
+	});
+}