XCM: Automatic Version Negotiation (#3736)

* XCM: Automatic Version Negotiation * Introduce the version instructions and subscription trait * Notification and subscription data migration * Version change subscriptions * Fixes * Formatting * Spelling * Fixes * Fixes * Automatic unsubscription * Formatting * Expose remote origin in VM and ensure it is unchanged from actual origin in subscription instructions. * Barrier * Unsubscription extrinsic * Remove top_level param * Formatting * Fixes * Automatic subscription * Formatting * Spelling * Unit tests for XCM executor * Formatting * Spellin * Unit test for XCM pallet subscriber side * Formatting * More tests * Formatting * Fixes * Subscription-side tests * Formatting * Unit tests for XCM pallet * Formatting * Update roadmap/implementers-guide/src/types/overseer-protocol.md Co-authored-by: Shawn Tabrizi <shawntabrizi@gmail.com> * Remove commented code * Grumbles * Multi-stage XCM version migration * Formatting * v1 subscriptions backport * Warning * Spelling * Fix grumbles * Formatting * Avoid running through old notifications * Formatting Co-authored-by: Shawn Tabrizi <shawntabrizi@gmail.com>
2026-05-30 12:51:02 +00:00 · 2021-09-02 18:47:38 +02:00
parent 446aa6777e
commit 82ffe7dd17
27 changed files with 1503 additions and 175 deletions
@@ -23,9 +23,12 @@ mod mock;
 #[cfg(test)]
 mod tests;

-use codec::{Decode, Encode};
+use codec::{Decode, Encode, EncodeLike};
 use frame_support::traits::{Contains, EnsureOrigin, Get, OriginTrait};
-use sp_runtime::{traits::BadOrigin, RuntimeDebug};
+use sp_runtime::{
+	traits::{BadOrigin, Saturating},
+	RuntimeDebug,
+};
 use sp_std::{
 	boxed::Box,
 	convert::{TryFrom, TryInto},
@@ -34,10 +37,7 @@ use sp_std::{
 	result::Result,
 	vec,
 };
-use xcm::{
-	latest::prelude::*, VersionedMultiAssets, VersionedMultiLocation, VersionedResponse,
-	VersionedXcm,
-};
+use xcm::prelude::*;
 use xcm_executor::traits::ConvertOrigin;

 use frame_support::PalletId;
@@ -49,15 +49,25 @@ pub mod pallet {
 	use frame_support::{
 		dispatch::{Dispatchable, GetDispatchInfo, PostDispatchInfo},
 		pallet_prelude::*,
+		parameter_types,
 	};
 	use frame_system::{pallet_prelude::*, Config as SysConfig};
 	use sp_core::H256;
 	use sp_runtime::traits::{AccountIdConversion, BlakeTwo256, BlockNumberProvider, Hash};
 	use xcm_executor::{
-		traits::{ClaimAssets, DropAssets, InvertLocation, OnResponse, WeightBounds},
+		traits::{
+			ClaimAssets, DropAssets, InvertLocation, OnResponse, VersionChangeNotifier,
+			WeightBounds,
+		},
 		Assets,
 	};

+	parameter_types! {
+		/// An implementation of `Get<u32>` which just returns the latest XCM version which we can
+		/// support.
+		pub const CurrentXcmVersion: u32 = XCM_VERSION;
+	}
+
 	#[pallet::pallet]
 	#[pallet::generate_store(pub(super) trait Store)]
 	pub struct Pallet<T>(_);
@@ -106,6 +116,12 @@ pub mod pallet {
 			+ GetDispatchInfo
 			+ IsType<<Self as frame_system::Config>::Call>
 			+ Dispatchable<Origin = <Self as Config>::Origin, PostInfo = PostDispatchInfo>;
+
+		const VERSION_DISCOVERY_QUEUE_SIZE: u32;
+
+		/// The latest supported version that we advertise. Generally just set it to
+		/// `pallet_xcm::CurrentXcmVersion`.
+		type AdvertisedXcmVersion: Get<XcmVersion>;
 	}

 	/// The maximum number of distinct assets allowed to be transferred in a single helper extrinsic.
@@ -160,7 +176,7 @@ pub mod pallet {
 		/// be received and acted upon.
 		///
 		/// \[ origin location, id, expected location \]
-		InvalidResponder(MultiLocation, QueryId, MultiLocation),
+		InvalidResponder(MultiLocation, QueryId, Option<MultiLocation>),
 		/// Expected query response has been received but the expected origin location placed in
 		/// storate by this runtime previously cannot be decoded. The query remains registered.
 		///
@@ -179,6 +195,25 @@ pub mod pallet {
 		///
 		/// \[ hash, origin, assets \]
 		AssetsTrapped(H256, MultiLocation, VersionedMultiAssets),
+		/// An XCM version change notification message has been attempted to be sent.
+		///
+		/// \[ destination, result \]
+		VersionChangeNotified(MultiLocation, XcmVersion),
+		/// The supported version of a location has been changed. This might be through an
+		/// automatic notification or a manual intervention.
+		///
+		/// \[ location, XCM version \]
+		SupportedVersionChanged(MultiLocation, XcmVersion),
+		/// A given location which had a version change subscription was dropped owing to an error
+		/// sending the notification to it.
+		///
+		/// \[ location, query ID, error \]
+		NotifyTargetSendFail(MultiLocation, QueryId, XcmError),
+		/// A given location which had a version change subscription was dropped owing to an error
+		/// migrating the location to our new XCM format.
+		///
+		/// \[ location, query ID \]
+		NotifyTargetMigrationFail(VersionedMultiLocation, QueryId),
 	}

 	#[pallet::origin]
@@ -219,6 +254,13 @@ pub mod pallet {
 		InvalidOrigin,
 		/// The version of the `Versioned` value used is not able to be interpreted.
 		BadVersion,
+		/// The given location could not be used (e.g. because it cannot be expressed in the
+		/// desired version of XCM).
+		BadLocation,
+		/// The referenced subscription could not be found.
+		NoSubscription,
+		/// The location is invalid since it already has a subscription from us.
+		AlreadySubscribed,
 	}

 	/// The status of a query.
@@ -230,16 +272,41 @@ pub mod pallet {
 			maybe_notify: Option<(u8, u8)>,
 			timeout: BlockNumber,
 		},
+		/// The query is for an ongoing version notification subscription.
+		VersionNotifier { origin: VersionedMultiLocation, is_active: bool },
 		/// A response has been received.
 		Ready { response: VersionedResponse, at: BlockNumber },
 	}

-	/// Value of a query, must be unique for each query.
-	pub type QueryId = u64;
+	#[derive(Copy, Clone)]
+	pub(crate) struct LatestVersionedMultiLocation<'a>(pub(crate) &'a MultiLocation);
+	impl<'a> EncodeLike<VersionedMultiLocation> for LatestVersionedMultiLocation<'a> {}
+	impl<'a> Encode for LatestVersionedMultiLocation<'a> {
+		fn encode(&self) -> Vec<u8> {
+			let mut r = VersionedMultiLocation::from(MultiLocation::default()).encode();
+			r.truncate(1);
+			self.0.using_encoded(|d| r.extend_from_slice(d));
+			r
+		}
+	}
+
+	#[derive(Clone, Encode, Decode, Eq, PartialEq, Ord, PartialOrd)]
+	pub enum VersionMigrationStage {
+		MigrateSupportedVersion,
+		MigrateVersionNotifiers,
+		NotifyCurrentTargets(Option<Vec<u8>>),
+		MigrateAndNotifyOldTargets,
+	}
+
+	impl Default for VersionMigrationStage {
+		fn default() -> Self {
+			Self::MigrateSupportedVersion
+		}
+	}

 	/// The latest available query index.
 	#[pallet::storage]
-	pub(super) type QueryCount<T: Config> = StorageValue<_, QueryId, ValueQuery>;
+	pub(super) type QueryCounter<T: Config> = StorageValue<_, QueryId, ValueQuery>;

 	/// The ongoing queries.
 	#[pallet::storage]
@@ -255,8 +322,131 @@ pub mod pallet {
 	#[pallet::getter(fn asset_trap)]
 	pub(super) type AssetTraps<T: Config> = StorageMap<_, Identity, H256, u32, ValueQuery>;

+	/// Default version to encode XCM when latest version of destination is unknown. If `None`,
+	/// then the destinations whose XCM version is unknown are considered unreachable.
+	#[pallet::storage]
+	pub(super) type SafeXcmVersion<T: Config> = StorageValue<_, XcmVersion, OptionQuery>;
+
+	/// Latest versions that we know various locations support.
+	#[pallet::storage]
+	pub(super) type SupportedVersion<T: Config> = StorageDoubleMap<
+		_,
+		Twox64Concat,
+		XcmVersion,
+		Blake2_128Concat,
+		VersionedMultiLocation,
+		XcmVersion,
+		OptionQuery,
+	>;
+
+	/// All locations that we have requested version notifications from.
+	#[pallet::storage]
+	pub(super) type VersionNotifiers<T: Config> = StorageDoubleMap<
+		_,
+		Twox64Concat,
+		XcmVersion,
+		Blake2_128Concat,
+		VersionedMultiLocation,
+		QueryId,
+		OptionQuery,
+	>;
+
+	/// The target locations that are subscribed to our version changes, as well as the most recent
+	/// of our versions we informed them of.
+	#[pallet::storage]
+	pub(super) type VersionNotifyTargets<T: Config> = StorageDoubleMap<
+		_,
+		Twox64Concat,
+		XcmVersion,
+		Blake2_128Concat,
+		VersionedMultiLocation,
+		(QueryId, u64, XcmVersion),
+		OptionQuery,
+	>;
+
+	pub struct VersionDiscoveryQueueSize<T>(PhantomData<T>);
+	impl<T: Config> Get<u32> for VersionDiscoveryQueueSize<T> {
+		fn get() -> u32 {
+			T::VERSION_DISCOVERY_QUEUE_SIZE
+		}
+	}
+
+	/// Destinations whose latest XCM version we would like to know. Duplicates not allowed, and
+	/// the `u32` counter is the number of times that a send to the destination has been attempted,
+	/// which is used as a prioritization.
+	#[pallet::storage]
+	pub(super) type VersionDiscoveryQueue<T: Config> = StorageValue<
+		_,
+		BoundedVec<(VersionedMultiLocation, u32), VersionDiscoveryQueueSize<T>>,
+		ValueQuery,
+	>;
+
+	/// The current migration's stage, if any.
+	#[pallet::storage]
+	pub(super) type CurrentMigration<T: Config> =
+		StorageValue<_, VersionMigrationStage, OptionQuery>;
+
+	#[pallet::genesis_config]
+	pub struct GenesisConfig {
+		/// The default version to encode outgoing XCM messages with.
+		pub safe_xcm_version: Option<XcmVersion>,
+	}
+
+	#[cfg(feature = "std")]
+	impl Default for GenesisConfig {
+		fn default() -> Self {
+			Self { safe_xcm_version: Some(XCM_VERSION) }
+		}
+	}
+
+	#[pallet::genesis_build]
+	impl<T: Config> GenesisBuild<T> for GenesisConfig {
+		fn build(&self) {
+			SafeXcmVersion::<T>::set(self.safe_xcm_version);
+		}
+	}
+
 	#[pallet::hooks]
-	impl<T: Config> Hooks<BlockNumberFor<T>> for Pallet<T> {}
+	impl<T: Config> Hooks<BlockNumberFor<T>> for Pallet<T> {
+		fn on_initialize(_n: BlockNumberFor<T>) -> Weight {
+			let mut weight_used = 0;
+			if let Some(migration) = CurrentMigration::<T>::get() {
+				// Consume 10% of block at most
+				let max_weight = T::BlockWeights::get().max_block / 10;
+				let (w, maybe_migration) = Self::check_xcm_version_change(migration, max_weight);
+				CurrentMigration::<T>::set(maybe_migration);
+				weight_used.saturating_accrue(w);
+			}
+
+			// Here we aim to get one successful version negotiation request sent per block, ordered
+			// by the destinations being most sent to.
+			let mut q = VersionDiscoveryQueue::<T>::take().into_inner();
+			// TODO: correct weights.
+			weight_used += T::DbWeight::get().read + T::DbWeight::get().write;
+			q.sort_by_key(|i| i.1);
+			while let Some((versioned_dest, _)) = q.pop() {
+				if let Ok(dest) = versioned_dest.try_into() {
+					if Self::request_version_notify(dest).is_ok() {
+						// TODO: correct weights.
+						weight_used += T::DbWeight::get().read + T::DbWeight::get().write;
+						break
+					}
+				}
+			}
+			// Should never fail since we only removed items. But better safe than panicking as it's
+			// way better to drop the queue than panic on initialize.
+			if let Ok(q) = BoundedVec::try_from(q) {
+				VersionDiscoveryQueue::<T>::put(q);
+			}
+			weight_used
+		}
+		fn on_runtime_upgrade() -> Weight {
+			// Start a migration (this happens before on_initialize so it'll happen later in this
+			// block, which should be good enough)...
+			CurrentMigration::<T>::put(VersionMigrationStage::default());
+			T::DbWeight::get().write
+		}
+	}

 	#[pallet::call]
 	impl<T: Config> Pallet<T> {
@@ -448,9 +638,263 @@ pub mod pallet {
 			Self::deposit_event(Event::Attempted(outcome));
 			Ok(())
 		}
+
+		/// Extoll that a particular destination can be communicated with through a particular
+		/// version of XCM.
+		///
+		/// - `origin`: Must be Root.
+		/// - `location`: The destination that is being described.
+		/// - `xcm_version`: The latest version of XCM that `location` supports.
+		#[pallet::weight(100_000_000u64)]
+		pub fn force_xcm_version(
+			origin: OriginFor<T>,
+			location: Box<MultiLocation>,
+			xcm_version: XcmVersion,
+		) -> DispatchResult {
+			ensure_root(origin)?;
+			let location = *location;
+			SupportedVersion::<T>::insert(
+				XCM_VERSION,
+				LatestVersionedMultiLocation(&location),
+				xcm_version,
+			);
+			Self::deposit_event(Event::SupportedVersionChanged(location, xcm_version));
+			Ok(())
+		}
+
+		/// Set a safe XCM version (the version that XCM should be encoded with if the most recent
+		/// version a destination can accept is unknown).
+		///
+		/// - `origin`: Must be Root.
+		/// - `maybe_xcm_version`: The default XCM encoding version, or `None` to disable.
+		#[pallet::weight(100_000_000u64)]
+		pub fn force_default_xcm_version(
+			origin: OriginFor<T>,
+			maybe_xcm_version: Option<XcmVersion>,
+		) -> DispatchResult {
+			ensure_root(origin)?;
+			SafeXcmVersion::<T>::set(maybe_xcm_version);
+			Ok(())
+		}
+
+		/// Ask a location to notify us regarding their XCM version and any changes to it.
+		///
+		/// - `origin`: Must be Root.
+		/// - `location`: The location to which we should subscribe for XCM version notifications.
+		#[pallet::weight(100_000_000u64)]
+		pub fn force_subscribe_version_notify(
+			origin: OriginFor<T>,
+			location: Box<VersionedMultiLocation>,
+		) -> DispatchResult {
+			ensure_root(origin)?;
+			let location = (*location).try_into().map_err(|()| Error::<T>::BadLocation)?;
+			Self::request_version_notify(location).map_err(|e| {
+				match e {
+					XcmError::InvalidLocation => Error::<T>::AlreadySubscribed,
+					_ => Error::<T>::InvalidOrigin,
+				}
+				.into()
+			})
+		}
+
+		/// Require that a particular destination should no longer notify us regarding any XCM
+		/// version changes.
+		///
+		/// - `origin`: Must be Root.
+		/// - `location`: The location to which we are currently subscribed for XCM version
+		///   notifications which we no longer desire.
+		#[pallet::weight(100_000_000u64)]
+		pub fn force_unsubscribe_version_notify(
+			origin: OriginFor<T>,
+			location: Box<VersionedMultiLocation>,
+		) -> DispatchResult {
+			ensure_root(origin)?;
+			let location = (*location).try_into().map_err(|()| Error::<T>::BadLocation)?;
+			Self::unrequest_version_notify(location).map_err(|e| {
+				match e {
+					XcmError::InvalidLocation => Error::<T>::NoSubscription,
+					_ => Error::<T>::InvalidOrigin,
+				}
+				.into()
+			})
+		}
 	}

 	impl<T: Config> Pallet<T> {
+		/// Will always make progress, and will do its best not to use much more than `weight_cutoff`
+		/// in doing so.
+		pub(crate) fn check_xcm_version_change(
+			mut stage: VersionMigrationStage,
+			weight_cutoff: Weight,
+		) -> (Weight, Option<VersionMigrationStage>) {
+			let mut weight_used = 0;
+
+			// TODO: Correct weights for the components of this:
+			let todo_sv_migrate_weight: Weight = T::DbWeight::get().read + T::DbWeight::get().write;
+			let todo_vn_migrate_weight: Weight = T::DbWeight::get().read + T::DbWeight::get().write;
+			let todo_vnt_already_notified_weight: Weight = T::DbWeight::get().read;
+			let todo_vnt_notify_weight: Weight =
+				T::DbWeight::get().read + T::DbWeight::get().write * 3;
+			let todo_vnt_migrate_weight: Weight =
+				T::DbWeight::get().read + T::DbWeight::get().write;
+			let todo_vnt_migrate_fail_weight: Weight =
+				T::DbWeight::get().read + T::DbWeight::get().write;
+			let todo_vnt_notify_migrate_weight: Weight =
+				T::DbWeight::get().read + T::DbWeight::get().write * 3;
+
+			use VersionMigrationStage::*;
+
+			if stage == MigrateSupportedVersion {
+				// We assume that supported XCM version only ever increases, so just cycle through lower
+				// XCM versioned from the current.
+				for v in 0..XCM_VERSION {
+					for (old_key, value) in SupportedVersion::<T>::drain_prefix(v) {
+						if let Ok(new_key) = old_key.into_latest() {
+							SupportedVersion::<T>::insert(XCM_VERSION, new_key, value);
+						}
+						weight_used.saturating_accrue(todo_sv_migrate_weight);
+						if weight_used >= weight_cutoff {
+							return (weight_used, Some(stage))
+						}
+					}
+				}
+				stage = MigrateVersionNotifiers;
+			}
+			if stage == MigrateVersionNotifiers {
+				for v in 0..XCM_VERSION {
+					for (old_key, value) in VersionNotifiers::<T>::drain_prefix(v) {
+						if let Ok(new_key) = old_key.into_latest() {
+							VersionNotifiers::<T>::insert(XCM_VERSION, new_key, value);
+						}
+						weight_used.saturating_accrue(todo_vn_migrate_weight);
+						if weight_used >= weight_cutoff {
+							return (weight_used, Some(stage))
+						}
+					}
+				}
+				stage = NotifyCurrentTargets(None);
+			}
+
+			let xcm_version = T::AdvertisedXcmVersion::get();
+
+			if let NotifyCurrentTargets(maybe_last_raw_key) = stage {
+				let mut iter = match maybe_last_raw_key {
+					Some(k) => VersionNotifyTargets::<T>::iter_prefix_from(XCM_VERSION, k),
+					None => VersionNotifyTargets::<T>::iter_prefix(XCM_VERSION),
+				};
+				while let Some((key, value)) = iter.next() {
+					let (query_id, max_weight, target_xcm_version) = value;
+					let new_key: MultiLocation = match key.clone().try_into() {
+						Ok(k) if target_xcm_version != xcm_version => k,
+						_ => {
+							// We don't early return here since we need to be certain that we
+							// make some progress.
+							weight_used.saturating_accrue(todo_vnt_already_notified_weight);
+							continue
+						},
+					};
+					let response = Response::Version(xcm_version);
+					let message = Xcm(vec![QueryResponse { query_id, response, max_weight }]);
+					let event = match T::XcmRouter::send_xcm(new_key.clone(), message) {
+						Ok(()) => {
+							let value = (query_id, max_weight, xcm_version);
+							VersionNotifyTargets::<T>::insert(XCM_VERSION, key, value);
+							Event::VersionChangeNotified(new_key, xcm_version)
+						},
+						Err(e) => {
+							VersionNotifyTargets::<T>::remove(XCM_VERSION, key);
+							Event::NotifyTargetSendFail(new_key, query_id, e.into())
+						},
+					};
+					Self::deposit_event(event);
+					weight_used.saturating_accrue(todo_vnt_notify_weight);
+					if weight_used >= weight_cutoff {
+						let last = Some(iter.last_raw_key().into());
+						return (weight_used, Some(NotifyCurrentTargets(last)))
+					}
+				}
+				stage = MigrateAndNotifyOldTargets;
+			}
+			if stage == MigrateAndNotifyOldTargets {
+				for v in 0..XCM_VERSION {
+					for (old_key, value) in VersionNotifyTargets::<T>::drain_prefix(v) {
+						let (query_id, max_weight, target_xcm_version) = value;
+						let new_key = match MultiLocation::try_from(old_key.clone()) {
+							Ok(k) => k,
+							Err(()) => {
+								Self::deposit_event(Event::NotifyTargetMigrationFail(
+									old_key, value.0,
+								));
+								weight_used.saturating_accrue(todo_vnt_migrate_fail_weight);
+								if weight_used >= weight_cutoff {
+									return (weight_used, Some(stage))
+								}
+								continue
+							},
+						};
+
+						let versioned_key = LatestVersionedMultiLocation(&new_key);
+						if target_xcm_version == xcm_version {
+							VersionNotifyTargets::<T>::insert(XCM_VERSION, versioned_key, value);
+							weight_used.saturating_accrue(todo_vnt_migrate_weight);
+						} else {
+							// Need to notify target.
+							let response = Response::Version(xcm_version);
+							let message =
+								Xcm(vec![QueryResponse { query_id, response, max_weight }]);
+							let event = match T::XcmRouter::send_xcm(new_key.clone(), message) {
+								Ok(()) => {
+									VersionNotifyTargets::<T>::insert(
+										XCM_VERSION,
+										versioned_key,
+										(query_id, max_weight, xcm_version),
+									);
+									Event::VersionChangeNotified(new_key, xcm_version)
+								},
+								Err(e) => Event::NotifyTargetSendFail(new_key, query_id, e.into()),
+							};
+							Self::deposit_event(event);
+							weight_used.saturating_accrue(todo_vnt_notify_migrate_weight);
+						}
+						if weight_used >= weight_cutoff {
+							return (weight_used, Some(stage))
+						}
+					}
+				}
+			}
+			(weight_used, None)
+		}
+
+		/// Request that `dest` informs us of its version.
+		pub fn request_version_notify(dest: MultiLocation) -> XcmResult {
+			let versioned_dest = VersionedMultiLocation::from(dest.clone());
+			let already = VersionNotifiers::<T>::contains_key(XCM_VERSION, &versioned_dest);
+			ensure!(!already, XcmError::InvalidLocation);
+			let query_id = QueryCounter::<T>::mutate(|q| {
+				let r = *q;
+				q.saturating_inc();
+				r
+			});
+			// TODO #3735: Correct weight.
+			let instruction = SubscribeVersion { query_id, max_response_weight: 0 };
+			T::XcmRouter::send_xcm(dest, Xcm(vec![instruction]))?;
+			VersionNotifiers::<T>::insert(XCM_VERSION, &versioned_dest, query_id);
+			let query_status =
+				QueryStatus::VersionNotifier { origin: versioned_dest, is_active: false };
+			Queries::<T>::insert(query_id, query_status);
+			Ok(())
+		}
+
+		/// Request that `dest` ceases informing us of its version.
+		pub fn unrequest_version_notify(dest: MultiLocation) -> XcmResult {
+			let versioned_dest = LatestVersionedMultiLocation(&dest);
+			let query_id = VersionNotifiers::<T>::take(XCM_VERSION, versioned_dest)
+				.ok_or(XcmError::InvalidLocation)?;
+			T::XcmRouter::send_xcm(dest.clone(), Xcm(vec![UnsubscribeVersion]))?;
+			Queries::<T>::remove(query_id);
+			Ok(())
+		}
+
 		/// Relay an XCM `message` from a given `interior` location in this context to a given `dest`
 		/// location. A null `dest` is not handled.
 		pub fn send_xcm(
@@ -475,9 +919,9 @@ pub mod pallet {
 			maybe_notify: Option<(u8, u8)>,
 			timeout: T::BlockNumber,
 		) -> u64 {
-			QueryCount::<T>::mutate(|q| {
+			QueryCounter::<T>::mutate(|q| {
 				let r = *q;
-				*q += 1;
+				q.saturating_inc();
 				Queries::<T>::insert(
 					r,
 					QueryStatus::Pending { responder: responder.into(), maybe_notify, timeout },
@@ -579,8 +1023,68 @@ pub mod pallet {
 				None
 			}
 		}
+
+		/// Note that a particular destination to whom we would like to send a message is unknown
+		/// and queue it for version discovery.
+		fn note_unknown_version(dest: &MultiLocation) {
+			let versioned_dest = VersionedMultiLocation::from(dest.clone());
+			VersionDiscoveryQueue::<T>::mutate(|q| {
+				if let Some(index) = q.iter().position(|i| &i.0 == &versioned_dest) {
+					// exists - just bump the count.
+					q[index].1.saturating_inc();
+				} else {
+					let _ = q.try_push((versioned_dest, 1));
+				}
+			});
+		}
 	}

+	impl<T: Config> WrapVersion for Pallet<T> {
+		fn wrap_version<Call>(
+			dest: &MultiLocation,
+			xcm: impl Into<VersionedXcm<Call>>,
+		) -> Result<VersionedXcm<Call>, ()> {
+			SupportedVersion::<T>::get(XCM_VERSION, LatestVersionedMultiLocation(dest))
+				.or_else(|| {
+					Self::note_unknown_version(dest);
+					SafeXcmVersion::<T>::get()
+				})
+				.ok_or(())
+				.and_then(|v| xcm.into().into_version(v.min(XCM_VERSION)))
+		}
+	}
+
+	impl<T: Config> VersionChangeNotifier for Pallet<T> {
+		/// Start notifying `location` should the XCM version of this chain change.
+		///
+		/// When it does, this type should ensure a `QueryResponse` message is sent with the given
+		/// `query_id` & `max_weight` and with a `response` of `Repsonse::Version`. This should happen
+		/// until/unless `stop` is called with the correct `query_id`.
+		///
+		/// If the `location` has an ongoing notification and when this function is called, then an
+		/// error should be returned.
+		fn start(dest: &MultiLocation, query_id: QueryId, max_weight: u64) -> XcmResult {
+			let versioned_dest = LatestVersionedMultiLocation(dest);
+			let already = VersionNotifyTargets::<T>::contains_key(XCM_VERSION, versioned_dest);
+			ensure!(!already, XcmError::InvalidLocation);
+
+			let xcm_version = T::AdvertisedXcmVersion::get();
+			let response = Response::Version(xcm_version);
+			let instruction = QueryResponse { query_id, response, max_weight };
+			T::XcmRouter::send_xcm(dest.clone(), Xcm(vec![instruction]))?;
+
+			let value = (query_id, max_weight, xcm_version);
+			VersionNotifyTargets::<T>::insert(XCM_VERSION, versioned_dest, value);
+			Ok(())
+		}
+
+		/// Stop notifying `location` should the XCM change. This is a no-op if there was never a
+		/// subscription.
+		fn stop(dest: &MultiLocation) -> XcmResult {
+			VersionNotifyTargets::<T>::remove(XCM_VERSION, LatestVersionedMultiLocation(dest));
+			Ok(())
+		}
+	}
 	impl<T: Config> DropAssets for Pallet<T> {
 		fn drop_assets(origin: &MultiLocation, assets: Assets) -> Weight {
 			if assets.is_empty() {
@@ -590,7 +1094,7 @@ pub mod pallet {
 			let hash = BlakeTwo256::hash_of(&(&origin, &versioned));
 			AssetTraps::<T>::mutate(hash, |n| *n += 1);
 			Self::deposit_event(Event::AssetsTrapped(hash, origin.clone(), versioned));
-			// TODO: Put the real weight in there.
+			// TODO #3735: Put the real weight in there.
 			0
 		}
 	}
@@ -623,10 +1127,13 @@ pub mod pallet {

 	impl<T: Config> OnResponse for Pallet<T> {
 		fn expecting_response(origin: &MultiLocation, query_id: QueryId) -> bool {
-			if let Some(QueryStatus::Pending { responder, .. }) = Queries::<T>::get(query_id) {
-				return MultiLocation::try_from(responder).map_or(false, |r| origin == &r)
+			match Queries::<T>::get(query_id) {
+				Some(QueryStatus::Pending { responder, .. }) =>
+					MultiLocation::try_from(responder).map_or(false, |r| origin == &r),
+				Some(QueryStatus::VersionNotifier { origin: r, .. }) =>
+					MultiLocation::try_from(r).map_or(false, |r| origin == &r),
+				_ => false,
 			}
-			false
 		}

 		fn on_response(
@@ -635,87 +1142,133 @@ pub mod pallet {
 			response: Response,
 			max_weight: Weight,
 		) -> Weight {
-			if let Some(QueryStatus::Pending { responder, maybe_notify, .. }) =
-				Queries::<T>::get(query_id)
-			{
-				if let Ok(responder) = MultiLocation::try_from(responder) {
-					if origin == &responder {
-						return match maybe_notify {
-							Some((pallet_index, call_index)) => {
-								// This is a bit horrible, but we happen to know that the `Call` will
-								// be built by `(pallet_index: u8, call_index: u8, QueryId, Response)`.
-								// So we just encode that and then re-encode to a real Call.
-								let bare = (pallet_index, call_index, query_id, response);
-								if let Ok(call) = bare.using_encoded(|mut bytes| {
-									<T as Config>::Call::decode(&mut bytes)
-								}) {
-									Queries::<T>::remove(query_id);
-									let weight = call.get_dispatch_info().weight;
-									if weight > max_weight {
-										let e = Event::NotifyOverweight(
-											query_id,
-											pallet_index,
-											call_index,
-											weight,
-											max_weight,
-										);
-										Self::deposit_event(e);
-										return 0
-									}
-									let dispatch_origin = Origin::Response(origin.clone()).into();
-									match call.dispatch(dispatch_origin) {
-										Ok(post_info) => {
-											let e =
-												Event::Notified(query_id, pallet_index, call_index);
-											Self::deposit_event(e);
-											post_info.actual_weight
-										},
-										Err(error_and_info) => {
-											let e = Event::NotifyDispatchError(
-												query_id,
-												pallet_index,
-												call_index,
-											);
-											Self::deposit_event(e);
-											// Not much to do with the result as it is. It's up to the parachain to ensure that the
-											// message makes sense.
-											error_and_info.post_info.actual_weight
-										},
-									}
-									.unwrap_or(weight)
-								} else {
-									let e = Event::NotifyDecodeFailed(
-										query_id,
-										pallet_index,
-										call_index,
-									);
-									Self::deposit_event(e);
-									0
-								}
+			match (response, Queries::<T>::get(query_id)) {
+				(
+					Response::Version(v),
+					Some(QueryStatus::VersionNotifier { origin: expected_origin, is_active }),
+				) => {
+					let origin: MultiLocation = match expected_origin.try_into() {
+						Ok(o) if &o == origin => o,
+						Ok(o) => {
+							Self::deposit_event(Event::InvalidResponder(
+								origin.clone(),
+								query_id,
+								Some(o),
+							));
+							return 0
+						},
+						_ => {
+							Self::deposit_event(Event::InvalidResponder(
+								origin.clone(),
+								query_id,
+								None,
+							));
+							// TODO #3735: Correct weight for this.
+							return 0
+						},
+					};
+					// TODO #3735: Check max_weight is correct.
+					if !is_active {
+						Queries::<T>::insert(
+							query_id,
+							QueryStatus::VersionNotifier {
+								origin: origin.clone().into(),
+								is_active: true,
 							},
-							None => {
-								let e = Event::ResponseReady(query_id, response.clone());
-								Self::deposit_event(e);
-								let at = frame_system::Pallet::<T>::current_block_number();
-								let response = response.into();
-								Queries::<T>::insert(query_id, QueryStatus::Ready { response, at });
-								0
-							},
-						}
-					} else {
+						);
+					}
+					// We're being notified of a version change.
+					SupportedVersion::<T>::insert(
+						XCM_VERSION,
+						LatestVersionedMultiLocation(&origin),
+						v,
+					);
+					Self::deposit_event(Event::SupportedVersionChanged(origin, v));
+					0
+				},
+				(response, Some(QueryStatus::Pending { responder, maybe_notify, .. })) => {
+					let responder = match MultiLocation::try_from(responder) {
+						Ok(r) => r,
+						Err(_) => {
+							Self::deposit_event(Event::InvalidResponderVersion(
+								origin.clone(),
+								query_id,
+							));
+							return 0
+						},
+					};
+					if origin != &responder {
 						Self::deposit_event(Event::InvalidResponder(
 							origin.clone(),
 							query_id,
-							responder,
+							Some(responder),
 						));
+						return 0
 					}
-				} else {
-					Self::deposit_event(Event::InvalidResponderVersion(origin.clone(), query_id));
-				}
-			} else {
-				Self::deposit_event(Event::UnexpectedResponse(origin.clone(), query_id));
+					return match maybe_notify {
+						Some((pallet_index, call_index)) => {
+							// This is a bit horrible, but we happen to know that the `Call` will
+							// be built by `(pallet_index: u8, call_index: u8, QueryId, Response)`.
+							// So we just encode that and then re-encode to a real Call.
+							let bare = (pallet_index, call_index, query_id, response);
+							if let Ok(call) = bare
+								.using_encoded(|mut bytes| <T as Config>::Call::decode(&mut bytes))
+							{
+								Queries::<T>::remove(query_id);
+								let weight = call.get_dispatch_info().weight;
+								if weight > max_weight {
+									let e = Event::NotifyOverweight(
+										query_id,
+										pallet_index,
+										call_index,
+										weight,
+										max_weight,
+									);
+									Self::deposit_event(e);
+									return 0
+								}
+								let dispatch_origin = Origin::Response(origin.clone()).into();
+								match call.dispatch(dispatch_origin) {
+									Ok(post_info) => {
+										let e = Event::Notified(query_id, pallet_index, call_index);
+										Self::deposit_event(e);
+										post_info.actual_weight
+									},
+									Err(error_and_info) => {
+										let e = Event::NotifyDispatchError(
+											query_id,
+											pallet_index,
+											call_index,
+										);
+										Self::deposit_event(e);
+										// Not much to do with the result as it is. It's up to the parachain to ensure that the
+										// message makes sense.
+										error_and_info.post_info.actual_weight
+									},
+								}
+								.unwrap_or(weight)
+							} else {
+								let e =
+									Event::NotifyDecodeFailed(query_id, pallet_index, call_index);
+								Self::deposit_event(e);
+								0
+							}
+						},
+						None => {
+							let e = Event::ResponseReady(query_id, response.clone());
+							Self::deposit_event(e);
+							let at = frame_system::Pallet::<T>::current_block_number();
+							let response = response.into();
+							Queries::<T>::insert(query_id, QueryStatus::Ready { response, at });
+							0
+						},
+					}
+				},
+				_ => {
+					Self::deposit_event(Event::UnexpectedResponse(origin.clone(), query_id));
+					return 0
+				},
 			}
-			0
 		}
 	}
 }
@@ -22,11 +22,11 @@ use sp_runtime::{testing::Header, traits::IdentityLookup, AccountId32};
 pub use sp_std::{cell::RefCell, fmt::Debug, marker::PhantomData};
 use xcm::latest::prelude::*;
 use xcm_builder::{
-	AccountId32Aliases, AllowKnownQueryResponses, AllowTopLevelPaidExecutionFrom, Case,
-	ChildParachainAsNative, ChildParachainConvertsVia, ChildSystemParachainAsSuperuser,
-	CurrencyAdapter as XcmCurrencyAdapter, FixedRateOfFungible, FixedWeightBounds, IsConcrete,
-	LocationInverter, SignedAccountId32AsNative, SignedToAccountId32, SovereignSignedViaLocation,
-	TakeWeightCredit,
+	AccountId32Aliases, AllowKnownQueryResponses, AllowSubscriptionsFrom,
+	AllowTopLevelPaidExecutionFrom, Case, ChildParachainAsNative, ChildParachainConvertsVia,
+	ChildSystemParachainAsSuperuser, CurrencyAdapter as XcmCurrencyAdapter, FixedRateOfFungible,
+	FixedWeightBounds, IsConcrete, LocationInverter, SignedAccountId32AsNative,
+	SignedToAccountId32, SovereignSignedViaLocation, TakeWeightCredit,
 };
 use xcm_executor::XcmExecutor;

@@ -133,9 +133,16 @@ construct_runtime!(
 thread_local! {
 	pub static SENT_XCM: RefCell<Vec<(MultiLocation, Xcm<()>)>> = RefCell::new(Vec::new());
 }
-pub fn sent_xcm() -> Vec<(MultiLocation, Xcm<()>)> {
+pub(crate) fn sent_xcm() -> Vec<(MultiLocation, Xcm<()>)> {
 	SENT_XCM.with(|q| (*q.borrow()).clone())
 }
+pub(crate) fn take_sent_xcm() -> Vec<(MultiLocation, Xcm<()>)> {
+	SENT_XCM.with(|q| {
+		let mut r = Vec::new();
+		std::mem::swap(&mut r, &mut *q.borrow_mut());
+		r
+	})
+}
 /// Sender that never returns error, always sends
 pub struct TestSendXcm;
 impl SendXcm for TestSendXcm {
@@ -236,6 +243,7 @@ pub type Barrier = (
 	TakeWeightCredit,
 	AllowTopLevelPaidExecutionFrom<Everything>,
 	AllowKnownQueryResponses<XcmPallet>,
+	AllowSubscriptionsFrom<Everything>,
 );

 pub struct XcmConfig;
@@ -253,10 +261,15 @@ impl xcm_executor::Config for XcmConfig {
 	type ResponseHandler = XcmPallet;
 	type AssetTrap = XcmPallet;
 	type AssetClaims = XcmPallet;
+	type SubscriptionService = XcmPallet;
 }

 pub type LocalOriginToLocation = SignedToAccountId32<Origin, AccountId, AnyNetwork>;

+parameter_types! {
+	pub static AdvertisedXcmVersion: pallet_xcm::XcmVersion = 2;
+}
+
 impl pallet_xcm::Config for Test {
 	type Event = Event;
 	type SendXcmOrigin = xcm_builder::EnsureXcmOrigin<Origin, LocalOriginToLocation>;
@@ -270,6 +283,8 @@ impl pallet_xcm::Config for Test {
 	type LocationInverter = LocationInverter<Ancestry>;
 	type Origin = Origin;
 	type Call = Call;
+	const VERSION_DISCOVERY_QUEUE_SIZE: u32 = 100;
+	type AdvertisedXcmVersion = AdvertisedXcmVersion;
 }

 impl origin::Config for Test {}
@@ -14,13 +14,20 @@
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.

-use crate::{mock::*, AssetTraps, QueryStatus};
-use frame_support::{assert_noop, assert_ok, traits::Currency};
+use crate::{
+	mock::*, AssetTraps, CurrentMigration, Error, LatestVersionedMultiLocation, Queries,
+	QueryStatus, VersionDiscoveryQueue, VersionNotifiers, VersionNotifyTargets,
+};
+use frame_support::{
+	assert_noop, assert_ok,
+	traits::{Currency, Hooks},
+};
 use polkadot_parachain::primitives::{AccountIdConversion, Id as ParaId};
 use sp_runtime::traits::{BlakeTwo256, Hash};
 use std::convert::TryInto;
-use xcm::{latest::prelude::*, VersionedMultiAssets, VersionedXcm};
-use xcm_executor::XcmExecutor;
+use xcm::prelude::*;
+use xcm_builder::AllowKnownQueryResponses;
+use xcm_executor::{traits::ShouldExecute, XcmExecutor};

 const ALICE: AccountId = AccountId::new([0u8; 32]);
 const BOB: AccountId = AccountId::new([1u8; 32]);
@@ -375,3 +382,423 @@ fn trapped_assets_can_be_claimed() {
 		);
 	});
 }
+
+#[test]
+fn fake_latest_versioned_multilocation_works() {
+	use codec::Encode;
+	let remote = Parachain(1000).into();
+	let versioned_remote = LatestVersionedMultiLocation(&remote);
+	assert_eq!(versioned_remote.encode(), VersionedMultiLocation::from(remote.clone()).encode());
+}
+
+#[test]
+fn basic_subscription_works() {
+	new_test_ext_with_balances(vec![]).execute_with(|| {
+		let remote = Parachain(1000).into();
+		assert_ok!(XcmPallet::force_subscribe_version_notify(
+			Origin::root(),
+			Box::new(remote.clone().into()),
+		));
+
+		assert_eq!(
+			Queries::<Test>::iter().collect::<Vec<_>>(),
+			vec![(
+				0,
+				QueryStatus::VersionNotifier { origin: remote.clone().into(), is_active: false }
+			)]
+		);
+		assert_eq!(
+			VersionNotifiers::<Test>::iter().collect::<Vec<_>>(),
+			vec![(2, remote.clone().into(), 0)]
+		);
+
+		assert_eq!(
+			take_sent_xcm(),
+			vec![(
+				remote.clone(),
+				Xcm(vec![SubscribeVersion { query_id: 0, max_response_weight: 0 }]),
+			),]
+		);
+
+		let weight = BaseXcmWeight::get();
+		let mut message = Xcm::<()>(vec![
+			// Remote supports XCM v1
+			QueryResponse { query_id: 0, max_weight: 0, response: Response::Version(1) },
+		]);
+		assert_ok!(AllowKnownQueryResponses::<XcmPallet>::should_execute(
+			&remote,
+			&mut message,
+			weight,
+			&mut 0
+		));
+	});
+}
+
+#[test]
+fn subscriptions_increment_id() {
+	new_test_ext_with_balances(vec![]).execute_with(|| {
+		let remote = Parachain(1000).into();
+		assert_ok!(XcmPallet::force_subscribe_version_notify(
+			Origin::root(),
+			Box::new(remote.clone().into()),
+		));
+
+		let remote2 = Parachain(1001).into();
+		assert_ok!(XcmPallet::force_subscribe_version_notify(
+			Origin::root(),
+			Box::new(remote2.clone().into()),
+		));
+
+		assert_eq!(
+			take_sent_xcm(),
+			vec![
+				(
+					remote.clone(),
+					Xcm(vec![SubscribeVersion { query_id: 0, max_response_weight: 0 }]),
+				),
+				(
+					remote2.clone(),
+					Xcm(vec![SubscribeVersion { query_id: 1, max_response_weight: 0 }]),
+				),
+			]
+		);
+	});
+}
+
+#[test]
+fn double_subscription_fails() {
+	new_test_ext_with_balances(vec![]).execute_with(|| {
+		let remote = Parachain(1000).into();
+		assert_ok!(XcmPallet::force_subscribe_version_notify(
+			Origin::root(),
+			Box::new(remote.clone().into()),
+		));
+		assert_noop!(
+			XcmPallet::force_subscribe_version_notify(
+				Origin::root(),
+				Box::new(remote.clone().into())
+			),
+			Error::<Test>::AlreadySubscribed,
+		);
+	})
+}
+
+#[test]
+fn unsubscribe_works() {
+	new_test_ext_with_balances(vec![]).execute_with(|| {
+		let remote = Parachain(1000).into();
+		assert_ok!(XcmPallet::force_subscribe_version_notify(
+			Origin::root(),
+			Box::new(remote.clone().into()),
+		));
+		assert_ok!(XcmPallet::force_unsubscribe_version_notify(
+			Origin::root(),
+			Box::new(remote.clone().into())
+		));
+		assert_noop!(
+			XcmPallet::force_unsubscribe_version_notify(
+				Origin::root(),
+				Box::new(remote.clone().into())
+			),
+			Error::<Test>::NoSubscription,
+		);
+
+		assert_eq!(
+			take_sent_xcm(),
+			vec![
+				(
+					remote.clone(),
+					Xcm(vec![SubscribeVersion { query_id: 0, max_response_weight: 0 }]),
+				),
+				(remote.clone(), Xcm(vec![UnsubscribeVersion]),),
+			]
+		);
+	});
+}
+
+/// Parachain 1000 is asking us for a version subscription.
+#[test]
+fn subscription_side_works() {
+	new_test_ext_with_balances(vec![]).execute_with(|| {
+		AdvertisedXcmVersion::set(1);
+
+		let remote = Parachain(1000).into();
+		let weight = BaseXcmWeight::get();
+		let message = Xcm(vec![SubscribeVersion { query_id: 0, max_response_weight: 0 }]);
+		let r = XcmExecutor::<XcmConfig>::execute_xcm(remote.clone(), message, weight);
+		assert_eq!(r, Outcome::Complete(weight));
+
+		let instr = QueryResponse { query_id: 0, max_weight: 0, response: Response::Version(1) };
+		assert_eq!(take_sent_xcm(), vec![(remote.clone(), Xcm(vec![instr]))]);
+
+		// A runtime upgrade which doesn't alter the version sends no notifications.
+		XcmPallet::on_runtime_upgrade();
+		XcmPallet::on_initialize(1);
+		assert_eq!(take_sent_xcm(), vec![]);
+
+		// New version.
+		AdvertisedXcmVersion::set(2);
+
+		// A runtime upgrade which alters the version does send notifications.
+		XcmPallet::on_runtime_upgrade();
+		XcmPallet::on_initialize(2);
+		let instr = QueryResponse { query_id: 0, max_weight: 0, response: Response::Version(2) };
+		assert_eq!(take_sent_xcm(), vec![(remote.clone(), Xcm(vec![instr]))]);
+	});
+}
+
+#[test]
+fn subscription_side_upgrades_work_with_notify() {
+	new_test_ext_with_balances(vec![]).execute_with(|| {
+		AdvertisedXcmVersion::set(1);
+
+		// An entry from a previous runtime with v0 XCM.
+		let v0_location = xcm::v0::MultiLocation::X1(xcm::v0::Junction::Parachain(1000));
+		let v0_location = VersionedMultiLocation::from(v0_location);
+		VersionNotifyTargets::<Test>::insert(0, v0_location, (69, 0, 1));
+		let v1_location = Parachain(1001).into().versioned();
+		VersionNotifyTargets::<Test>::insert(1, v1_location, (70, 0, 1));
+		let v2_location = Parachain(1002).into().versioned();
+		VersionNotifyTargets::<Test>::insert(2, v2_location, (71, 0, 1));
+
+		// New version.
+		AdvertisedXcmVersion::set(2);
+
+		// A runtime upgrade which alters the version does send notifications.
+		XcmPallet::on_runtime_upgrade();
+		XcmPallet::on_initialize(1);
+
+		let instr0 = QueryResponse { query_id: 69, max_weight: 0, response: Response::Version(2) };
+		let instr1 = QueryResponse { query_id: 70, max_weight: 0, response: Response::Version(2) };
+		let instr2 = QueryResponse { query_id: 71, max_weight: 0, response: Response::Version(2) };
+		let mut sent = take_sent_xcm();
+		sent.sort_by_key(|k| match (k.1).0[0] {
+			QueryResponse { query_id: q, .. } => q,
+			_ => 0,
+		});
+		assert_eq!(
+			sent,
+			vec![
+				(Parachain(1000).into(), Xcm(vec![instr0])),
+				(Parachain(1001).into(), Xcm(vec![instr1])),
+				(Parachain(1002).into(), Xcm(vec![instr2])),
+			]
+		);
+
+		let mut contents = VersionNotifyTargets::<Test>::iter().collect::<Vec<_>>();
+		contents.sort_by_key(|k| k.2);
+		assert_eq!(
+			contents,
+			vec![
+				(2, Parachain(1000).into().versioned(), (69, 0, 2)),
+				(2, Parachain(1001).into().versioned(), (70, 0, 2)),
+				(2, Parachain(1002).into().versioned(), (71, 0, 2)),
+			]
+		);
+	});
+}
+
+#[test]
+fn subscription_side_upgrades_work_without_notify() {
+	new_test_ext_with_balances(vec![]).execute_with(|| {
+		// An entry from a previous runtime with v0 XCM.
+		let v0_location = xcm::v0::MultiLocation::X1(xcm::v0::Junction::Parachain(1000));
+		let v0_location = VersionedMultiLocation::from(v0_location);
+		VersionNotifyTargets::<Test>::insert(0, v0_location, (69, 0, 2));
+		let v1_location = Parachain(1001).into().versioned();
+		VersionNotifyTargets::<Test>::insert(1, v1_location, (70, 0, 2));
+		let v2_location = Parachain(1002).into().versioned();
+		VersionNotifyTargets::<Test>::insert(2, v2_location, (71, 0, 2));
+
+		// A runtime upgrade which alters the version does send notifications.
+		XcmPallet::on_runtime_upgrade();
+		XcmPallet::on_initialize(1);
+
+		let mut contents = VersionNotifyTargets::<Test>::iter().collect::<Vec<_>>();
+		contents.sort_by_key(|k| k.2);
+		assert_eq!(
+			contents,
+			vec![
+				(2, Parachain(1000).into().versioned(), (69, 0, 2)),
+				(2, Parachain(1001).into().versioned(), (70, 0, 2)),
+				(2, Parachain(1002).into().versioned(), (71, 0, 2)),
+			]
+		);
+	});
+}
+
+#[test]
+fn subscriber_side_subscription_works() {
+	new_test_ext_with_balances(vec![]).execute_with(|| {
+		let remote = Parachain(1000).into();
+		assert_ok!(XcmPallet::force_subscribe_version_notify(
+			Origin::root(),
+			Box::new(remote.clone().into()),
+		));
+		take_sent_xcm();
+
+		// Assume subscription target is working ok.
+
+		let weight = BaseXcmWeight::get();
+		let message = Xcm(vec![
+			// Remote supports XCM v1
+			QueryResponse { query_id: 0, max_weight: 0, response: Response::Version(1) },
+		]);
+		let r = XcmExecutor::<XcmConfig>::execute_xcm(remote.clone(), message, weight);
+		assert_eq!(r, Outcome::Complete(weight));
+		assert_eq!(take_sent_xcm(), vec![]);
+
+		// This message cannot be sent to a v1 remote.
+		let v2_msg = Xcm::<()>(vec![Trap(0)]);
+		assert_eq!(XcmPallet::wrap_version(&remote, v2_msg.clone()), Err(()));
+
+		let message = Xcm(vec![
+			// Remote upgraded to XCM v2
+			QueryResponse { query_id: 0, max_weight: 0, response: Response::Version(2) },
+		]);
+		let r = XcmExecutor::<XcmConfig>::execute_xcm(remote.clone(), message, weight);
+		assert_eq!(r, Outcome::Complete(weight));
+
+		// This message can now be sent to remote as it's v2.
+		assert_eq!(
+			XcmPallet::wrap_version(&remote, v2_msg.clone()),
+			Ok(VersionedXcm::from(v2_msg))
+		);
+	});
+}
+
+/// We should autosubscribe when we don't know the remote's version.
+#[test]
+fn auto_subscription_works() {
+	new_test_ext_with_balances(vec![]).execute_with(|| {
+		let remote0 = Parachain(1000).into();
+		let remote1 = Parachain(1001).into();
+
+		assert_ok!(XcmPallet::force_default_xcm_version(Origin::root(), Some(1)));
+
+		// Wrapping a version for a destination we don't know elicits a subscription.
+		let v1_msg = xcm::v1::Xcm::<()>::QueryResponse {
+			query_id: 1,
+			response: xcm::v1::Response::Assets(vec![].into()),
+		};
+		let v2_msg = Xcm::<()>(vec![Trap(0)]);
+		assert_eq!(
+			XcmPallet::wrap_version(&remote0, v1_msg.clone()),
+			Ok(VersionedXcm::from(v1_msg.clone())),
+		);
+		assert_eq!(XcmPallet::wrap_version(&remote0, v2_msg.clone()), Err(()));
+		let expected = vec![(remote0.clone().into(), 2)];
+		assert_eq!(VersionDiscoveryQueue::<Test>::get().into_inner(), expected);
+
+		assert_eq!(XcmPallet::wrap_version(&remote0, v2_msg.clone()), Err(()));
+		assert_eq!(XcmPallet::wrap_version(&remote1, v2_msg.clone()), Err(()));
+		let expected = vec![(remote0.clone().into(), 3), (remote1.clone().into(), 1)];
+		assert_eq!(VersionDiscoveryQueue::<Test>::get().into_inner(), expected);
+
+		XcmPallet::on_initialize(1);
+		assert_eq!(
+			take_sent_xcm(),
+			vec![(
+				remote0.clone(),
+				Xcm(vec![SubscribeVersion { query_id: 0, max_response_weight: 0 }]),
+			)]
+		);
+
+		// Assume remote0 is working ok and XCM version 2.
+
+		let weight = BaseXcmWeight::get();
+		let message = Xcm(vec![
+			// Remote supports XCM v2
+			QueryResponse { query_id: 0, max_weight: 0, response: Response::Version(2) },
+		]);
+		let r = XcmExecutor::<XcmConfig>::execute_xcm(remote0.clone(), message, weight);
+		assert_eq!(r, Outcome::Complete(weight));
+
+		// This message can now be sent to remote0 as it's v2.
+		assert_eq!(
+			XcmPallet::wrap_version(&remote0, v2_msg.clone()),
+			Ok(VersionedXcm::from(v2_msg.clone()))
+		);
+
+		XcmPallet::on_initialize(2);
+		assert_eq!(
+			take_sent_xcm(),
+			vec![(
+				remote1.clone(),
+				Xcm(vec![SubscribeVersion { query_id: 1, max_response_weight: 0 }]),
+			)]
+		);
+
+		// Assume remote1 is working ok and XCM version 1.
+
+		let weight = BaseXcmWeight::get();
+		let message = Xcm(vec![
+			// Remote supports XCM v1
+			QueryResponse { query_id: 1, max_weight: 0, response: Response::Version(1) },
+		]);
+		let r = XcmExecutor::<XcmConfig>::execute_xcm(remote1.clone(), message, weight);
+		assert_eq!(r, Outcome::Complete(weight));
+
+		// v2 messages cannot be sent to remote1...
+		assert_eq!(XcmPallet::wrap_version(&remote1, v1_msg.clone()), Ok(VersionedXcm::V1(v1_msg)));
+		assert_eq!(XcmPallet::wrap_version(&remote1, v2_msg.clone()), Err(()));
+	})
+}
+
+#[test]
+fn subscription_side_upgrades_work_with_multistage_notify() {
+	new_test_ext_with_balances(vec![]).execute_with(|| {
+		AdvertisedXcmVersion::set(1);
+
+		// An entry from a previous runtime with v0 XCM.
+		let v0_location = xcm::v0::MultiLocation::X1(xcm::v0::Junction::Parachain(1000));
+		let v0_location = VersionedMultiLocation::from(v0_location);
+		VersionNotifyTargets::<Test>::insert(0, v0_location, (69, 0, 1));
+		let v1_location = Parachain(1001).into().versioned();
+		VersionNotifyTargets::<Test>::insert(1, v1_location, (70, 0, 1));
+		let v2_location = Parachain(1002).into().versioned();
+		VersionNotifyTargets::<Test>::insert(2, v2_location, (71, 0, 1));
+
+		// New version.
+		AdvertisedXcmVersion::set(2);
+
+		// A runtime upgrade which alters the version does send notifications.
+		XcmPallet::on_runtime_upgrade();
+		let mut maybe_migration = CurrentMigration::<Test>::take();
+		let mut counter = 0;
+		while let Some(migration) = maybe_migration.take() {
+			counter += 1;
+			let (_, m) = XcmPallet::check_xcm_version_change(migration, 0);
+			maybe_migration = m;
+		}
+		assert_eq!(counter, 4);
+
+		let instr0 = QueryResponse { query_id: 69, max_weight: 0, response: Response::Version(2) };
+		let instr1 = QueryResponse { query_id: 70, max_weight: 0, response: Response::Version(2) };
+		let instr2 = QueryResponse { query_id: 71, max_weight: 0, response: Response::Version(2) };
+		let mut sent = take_sent_xcm();
+		sent.sort_by_key(|k| match (k.1).0[0] {
+			QueryResponse { query_id: q, .. } => q,
+			_ => 0,
+		});
+		assert_eq!(
+			sent,
+			vec![
+				(Parachain(1000).into(), Xcm(vec![instr0])),
+				(Parachain(1001).into(), Xcm(vec![instr1])),
+				(Parachain(1002).into(), Xcm(vec![instr2])),
+			]
+		);
+
+		let mut contents = VersionNotifyTargets::<Test>::iter().collect::<Vec<_>>();
+		contents.sort_by_key(|k| k.2);
+		assert_eq!(
+			contents,
+			vec![
+				(2, Parachain(1000).into().versioned(), (69, 0, 2)),
+				(2, Parachain(1001).into().versioned(), (70, 0, 2)),
+				(2, Parachain(1002).into().versioned(), (71, 0, 2)),
+			]
+		);
+	});
+}