network-gossip: Do not report peer on duplicate message if its the first time (#13508)

Two peers can send us the same gossip message. Before this pr we would have reported the second peer
for sending a duplicate message. However, this isn't correct if we haven't seen the message from
this peer yet. So, we should not report them as they can not be aware of our internal state.

substrate/client/network-gossip/src/state_machine.rs

@@ -354,7 +354,15 @@ impl<B: BlockT> ConsensusGossip<B> {
 					protocol = %self.protocol,
 					"Ignored already known message",
 				);
-				network.report_peer(who, rep::DUPLICATE_GOSSIP);
+
+				// If the peer already send us the message once, let's report them.
+				if self
+					.peers
+					.get_mut(&who)
+					.map_or(false, |p| !p.known_messages.insert(message_hash))
+				{
+					network.report_peer(who, rep::DUPLICATE_GOSSIP);
+				}
 				continue
 			}
 
@@ -814,4 +822,30 @@ mod tests {
 			to_forward,
 		);
 	}
+
+	// Two peers can send us the same gossip message. We should not report the second peer
+	// sending the gossip message as long as its the first time the peer send us this message.
+	#[test]
+	fn do_not_report_peer_for_first_time_duplicate_gossip_message() {
+		let mut consensus = ConsensusGossip::<Block>::new(Arc::new(AllowAll), "/foo".into(), None);
+
+		let mut network = NoOpNetwork::default();
+
+		let peer_id = PeerId::random();
+		consensus.new_peer(&mut network, peer_id, ObservedRole::Full);
+		assert!(consensus.peers.contains_key(&peer_id));
+
+		let peer_id2 = PeerId::random();
+		consensus.new_peer(&mut network, peer_id2, ObservedRole::Full);
+		assert!(consensus.peers.contains_key(&peer_id2));
+
+		let message = vec![vec![1, 2, 3]];
+		consensus.on_incoming(&mut network, peer_id, message.clone());
+		consensus.on_incoming(&mut network, peer_id2, message.clone());
+
+		assert_eq!(
+			vec![(peer_id, rep::GOSSIP_SUCCESS)],
+			network.inner.lock().unwrap().peer_reports
+		);
+	}
 }