Fix state_subscribeRuntimeVersion for parachains (paritytech#9617)

The old implementation was listening for storage changes and every time
a block changed the `CODE` storage field, it checked if the runtime
version changed. It used the best block to compare against the latest
known runtime version. It could happen that you processed the storage
notification of block Y and checked the runtime version of block X (the
current best block). This is also what happened on parachains.
Parachains import blocks and set the new best block in a later step.
This means we imported the block that changed the code, got notified and
checked the runtime version of the current best block (which would still
be the parent of the block that changed the runtime). As the parent did
not changed the runtime, the runtime version also did not changed and we
never notified the subscribers.

The new implementation now switches to listen for best imported blocks.
Every time we import a new best block, we check its runtime version
against the latest known runtime version. As we also send a notification
when the parachains sets a block as new best block, we will trigger this
code path correctly. It moves some computation from checking if the key
was modified to getting the runtime version. As fetching the runtime
version is a rather common pattern, it should not make any big
difference performancewise.

client/rpc/src/state/state_full.rs

@@ -35,8 +35,7 @@ use sp_blockchain::{
 };
 use sp_core::{
 	storage::{
-		well_known_keys, ChildInfo, ChildType, PrefixedStorageKey, StorageChangeSet, StorageData,
-		StorageKey,
+		ChildInfo, ChildType, PrefixedStorageKey, StorageChangeSet, StorageData, StorageKey,
 	},
 	Bytes,
 };
@@ -470,17 +469,6 @@ where
 		_meta: crate::Metadata,
 		subscriber: Subscriber<RuntimeVersion>,
 	) {
-		let stream = match self.client.storage_changes_notification_stream(
-			Some(&[StorageKey(well_known_keys::CODE.to_vec())]),
-			None,
-		) {
-			Ok(stream) => stream,
-			Err(err) => {
-				let _ = subscriber.reject(Error::from(client_err(err)).into());
-				return
-			},
-		};
-
 		self.subscriptions.add(subscriber, |sink| {
 			let version = self
 				.block_or_best(None)
@@ -493,12 +481,16 @@ where
 			let client = self.client.clone();
 			let mut previous_version = version.clone();
 
-			let stream = stream.filter_map(move |_| {
-				let info = client.info();
+			// A stream of all best blocks.
+			let stream =
+				client.import_notification_stream().filter(|n| future::ready(n.is_new_best));
+
+			let stream = stream.filter_map(move |n| {
 				let version = client
-					.runtime_version_at(&BlockId::hash(info.best_hash))
+					.runtime_version_at(&BlockId::hash(n.hash))
 					.map_err(|e| Error::Client(Box::new(e)))
 					.map_err(Into::into);
+
 				if previous_version != version {
 					previous_version = version.clone();
 					future::ready(Some(Ok::<_, ()>(version)))