Add support for versioned notification for HRMP pallet

polkadot/runtime/parachains/src/hrmp.rs

@@ -278,6 +278,9 @@ pub mod pallet {
 		/// parachain.
 		type DefaultChannelSizeAndCapacityWithSystem: Get<(u32, u32)>;
 
+		/// Means of converting an `Xcm` into a `VersionedXcm`.
+		type VersionWrapper: xcm::WrapVersion;
+
 		/// Something that provides the weight of this pallet.
 		type WeightInfo: WeightInfo;
 	}
@@ -1499,28 +1502,19 @@ impl<T: Config> Pallet<T> {
 		);
 		HrmpOpenChannelRequestsList::<T>::append(channel_id);
 
-		let notification_bytes = {
-			use parity_scale_codec::Encode as _;
-			use xcm::opaque::{latest::prelude::*, VersionedXcm};
-
-			VersionedXcm::from(Xcm(vec![HrmpNewChannelOpenRequest {
-				sender: u32::from(origin),
-				max_capacity: proposed_max_capacity,
-				max_message_size: proposed_max_message_size,
-			}]))
-			.encode()
-		};
-		if let Err(dmp::QueueDownwardMessageError::ExceedsMaxMessageSize) =
-			dmp::Pallet::<T>::queue_downward_message(&config, recipient, notification_bytes)
-		{
-			// this should never happen unless the max downward message size is configured to a
-			// jokingly small number.
-			log::error!(
-				target: "runtime::hrmp",
-				"sending 'init_open_channel::notification_bytes' failed."
-			);
-			debug_assert!(false);
-		}
+		Self::send_to_para(
+			"init_open_channel",
+			&config,
+			recipient,
+			Self::wrap_notification(|| {
+				use xcm::opaque::latest::{prelude::*, Xcm};
+				Xcm(vec![HrmpNewChannelOpenRequest {
+					sender: u32::from(origin),
+					max_capacity: proposed_max_capacity,
+					max_message_size: proposed_max_message_size,
+				}])
+			}),
+		);
 
 		Ok(())
 	}
@@ -1562,23 +1556,15 @@ impl<T: Config> Pallet<T> {
 		HrmpOpenChannelRequests::<T>::insert(&channel_id, channel_req);
 		HrmpAcceptedChannelRequestCount::<T>::insert(&origin, accepted_cnt + 1);
 
-		let notification_bytes = {
-			use parity_scale_codec::Encode as _;
-			use xcm::opaque::{latest::prelude::*, VersionedXcm};
-			let xcm = Xcm(vec![HrmpChannelAccepted { recipient: u32::from(origin) }]);
-			VersionedXcm::from(xcm).encode()
-		};
-		if let Err(dmp::QueueDownwardMessageError::ExceedsMaxMessageSize) =
-			dmp::Pallet::<T>::queue_downward_message(&config, sender, notification_bytes)
-		{
-			// this should never happen unless the max downward message size is configured to an
-			// jokingly small number.
-			log::error!(
-				target: "runtime::hrmp",
-				"sending 'accept_open_channel::notification_bytes' failed."
-			);
-			debug_assert!(false);
-		}
+		Self::send_to_para(
+			"accept_open_channel",
+			&config,
+			sender,
+			Self::wrap_notification(|| {
+				use xcm::opaque::latest::{prelude::*, Xcm};
+				Xcm(vec![HrmpChannelAccepted { recipient: u32::from(origin) }])
+			}),
+		);
 
 		Ok(())
 	}
@@ -1633,30 +1619,22 @@ impl<T: Config> Pallet<T> {
 		HrmpCloseChannelRequestsList::<T>::append(channel_id.clone());
 
 		let config = configuration::ActiveConfig::<T>::get();
-		let notification_bytes = {
-			use parity_scale_codec::Encode as _;
-			use xcm::opaque::{latest::prelude::*, VersionedXcm};
-
-			VersionedXcm::from(Xcm(vec![HrmpChannelClosing {
-				initiator: u32::from(origin),
-				sender: u32::from(channel_id.sender),
-				recipient: u32::from(channel_id.recipient),
-			}]))
-			.encode()
-		};
 		let opposite_party =
 			if origin == channel_id.sender { channel_id.recipient } else { channel_id.sender };
-		if let Err(dmp::QueueDownwardMessageError::ExceedsMaxMessageSize) =
-			dmp::Pallet::<T>::queue_downward_message(&config, opposite_party, notification_bytes)
-		{
-			// this should never happen unless the max downward message size is configured to an
-			// jokingly small number.
-			log::error!(
-				target: "runtime::hrmp",
-				"sending 'close_channel::notification_bytes' failed."
-			);
-			debug_assert!(false);
-		}
+
+		Self::send_to_para(
+			"close_channel",
+			&config,
+			opposite_party,
+			Self::wrap_notification(|| {
+				use xcm::opaque::latest::{prelude::*, Xcm};
+				Xcm(vec![HrmpChannelClosing {
+					initiator: u32::from(origin),
+					sender: u32::from(channel_id.sender),
+					recipient: u32::from(channel_id.recipient),
+				}])
+			}),
+		);
 
 		Ok(())
 	}
@@ -1875,3 +1853,56 @@ impl<T: Config> Pallet<T> {
 		}
 	}
 }
+
+impl<T: Config> Pallet<T> {
+	/// Wraps HRMP XCM notifications to the most suitable XCM version for the destination para.
+	/// If the XCM version is unknown, the latest XCM version is used as a best effort.
+	fn wrap_notification(
+		mut notification: impl FnMut() -> xcm::opaque::latest::opaque::Xcm,
+	) -> impl FnOnce(ParaId) -> primitives::DownwardMessage {
+		use xcm::{
+			opaque::VersionedXcm,
+			prelude::{Junction, Location},
+			WrapVersion,
+		};
+
+		// Return a closure that can prepare notifications.
+		move |dest| {
+			// Attempt to wrap the notification for the destination parachain.
+			T::VersionWrapper::wrap_version(
+				&Location::new(0, [Junction::Parachain(dest.into())]),
+				notification(),
+			)
+			.unwrap_or_else(|_| {
+				// As a best effort, if we cannot resolve the version, fallback to using the latest
+				// version.
+				VersionedXcm::from(notification())
+			})
+			.encode()
+		}
+	}
+
+	/// Sends/enqueues notification to the destination parachain.
+	fn send_to_para(
+		log_label: &str,
+		config: &HostConfiguration<BlockNumberFor<T>>,
+		dest: ParaId,
+		notification_bytes_for: impl FnOnce(ParaId) -> primitives::DownwardMessage,
+	) {
+		// prepare notification
+		let notification_bytes = notification_bytes_for(dest);
+
+		// try to enqueue
+		if let Err(dmp::QueueDownwardMessageError::ExceedsMaxMessageSize) =
+			dmp::Pallet::<T>::queue_downward_message(&config, dest, notification_bytes)
+		{
+			// this should never happen unless the max downward message size is configured to a
+			// jokingly small number.
+			log::error!(
+				target: "runtime::hrmp",
+				"sending '{log_label}::notification_bytes' failed."
+			);
+			debug_assert!(false);
+		}
+	}
+}

polkadot/runtime/parachains/src/hrmp/tests.rs

@@ -22,13 +22,13 @@ use super::*;
 use crate::{
 	mock::{
 		deregister_parachain, new_test_ext, register_parachain, register_parachain_with_balance,
-		Hrmp, MockGenesisConfig, Paras, ParasShared, RuntimeEvent as MockEvent, RuntimeOrigin,
-		System, Test,
+		Dmp, Hrmp, MockGenesisConfig, Paras, ParasShared, RuntimeEvent as MockEvent, RuntimeOrigin,
+		System, Test, TestUsesOnlyStoredVersionWrapper,
 	},
 	shared,
 };
 use frame_support::{assert_noop, assert_ok, error::BadOrigin};
-use primitives::BlockNumber;
+use primitives::{BlockNumber, InboundDownwardMessage};
 use std::collections::BTreeMap;
 
 pub(crate) fn run_to_block(to: BlockNumber, new_session: Option<Vec<BlockNumber>>) {
@@ -1004,3 +1004,140 @@ fn establish_channel_with_system_with_invalid_args() {
 		Hrmp::assert_storage_consistency_exhaustive();
 	});
 }
+
+#[test]
+fn hrmp_notifications_works() {
+	use xcm::{
+		opaque::{
+			latest::{prelude::*, Xcm},
+			VersionedXcm,
+		},
+		IntoVersion,
+	};
+
+	let para_a = 2001.into();
+	let para_a_origin: crate::Origin = 2001.into();
+	let para_b = 2003.into();
+	let para_b_origin: crate::Origin = 2003.into();
+
+	new_test_ext(GenesisConfigBuilder::default().build()).execute_with(|| {
+		// We need both A & B to be registered and alive parachains.
+		register_parachain(para_a);
+		register_parachain(para_b);
+		run_to_block(5, Some(vec![4, 5]));
+
+		// set XCM versions for wrapper
+
+		// for para_a -> `None`, means we will use latest.
+		TestUsesOnlyStoredVersionWrapper::set_version(
+			Location::new(0, [Junction::Parachain(para_a.into())]),
+			None,
+		);
+		// for para_b -> `Some(latest - 1)`, means we will use latest-1 XCM version.
+		let previous_version = XCM_VERSION - 1;
+		TestUsesOnlyStoredVersionWrapper::set_version(
+			Location::new(0, [Junction::Parachain(para_b.into())]),
+			Some(previous_version),
+		);
+
+		let assert_notification_for = |sent_at, para_id, expected| {
+			assert_eq!(
+				Dmp::dmq_contents(para_id),
+				vec![InboundDownwardMessage { sent_at, msg: expected }]
+			);
+		};
+
+		// init open channel requests
+		assert_ok!(Hrmp::hrmp_init_open_channel(para_a_origin.clone().into(), para_b, 2, 8));
+		assert_ok!(Hrmp::hrmp_init_open_channel(para_b_origin.clone().into(), para_a, 2, 8));
+		Hrmp::assert_storage_consistency_exhaustive();
+
+		// check dmp notications
+		assert_notification_for(
+			5,
+			para_b,
+			VersionedXcm::from(Xcm(vec![HrmpNewChannelOpenRequest {
+				sender: u32::from(para_a),
+				max_capacity: 2,
+				max_message_size: 8,
+			}]))
+			.into_version(previous_version)
+			.expect("compatible")
+			.encode(),
+		);
+		assert_notification_for(
+			5,
+			para_a,
+			VersionedXcm::from(Xcm(vec![HrmpNewChannelOpenRequest {
+				sender: u32::from(para_b),
+				max_capacity: 2,
+				max_message_size: 8,
+			}]))
+			.encode(),
+		);
+		let _ = Dmp::prune_dmq(para_a, 1000);
+		let _ = Dmp::prune_dmq(para_b, 1000);
+
+		// accept open channel requests
+		assert_ok!(Hrmp::hrmp_accept_open_channel(para_a_origin.clone().into(), para_b));
+		assert_ok!(Hrmp::hrmp_accept_open_channel(para_b_origin.clone().into(), para_a));
+		Hrmp::assert_storage_consistency_exhaustive();
+
+		// check dmp notications
+		assert_notification_for(
+			5,
+			para_b,
+			VersionedXcm::from(Xcm(vec![HrmpChannelAccepted { recipient: u32::from(para_a) }]))
+				.into_version(previous_version)
+				.expect("compatible")
+				.encode(),
+		);
+		assert_notification_for(
+			5,
+			para_a,
+			VersionedXcm::from(Xcm(vec![HrmpChannelAccepted { recipient: u32::from(para_b) }]))
+				.encode(),
+		);
+		let _ = Dmp::prune_dmq(para_a, 1000);
+		let _ = Dmp::prune_dmq(para_b, 1000);
+
+		// On Block 6: session change - creates channel.
+		run_to_block(6, Some(vec![6]));
+		assert!(channel_exists(para_a, para_b));
+
+		// close channel requests
+		assert_ok!(Hrmp::hrmp_close_channel(
+			para_a_origin.into(),
+			HrmpChannelId { sender: para_a, recipient: para_b }
+		));
+		assert_ok!(Hrmp::hrmp_close_channel(
+			para_b_origin.into(),
+			HrmpChannelId { sender: para_b, recipient: para_a }
+		));
+		Hrmp::assert_storage_consistency_exhaustive();
+
+		// check dmp notications
+		assert_notification_for(
+			6,
+			para_b,
+			VersionedXcm::from(Xcm(vec![HrmpChannelClosing {
+				initiator: u32::from(para_a),
+				sender: u32::from(para_a),
+				recipient: u32::from(para_b),
+			}]))
+			.into_version(previous_version)
+			.expect("compatible")
+			.encode(),
+		);
+		assert_notification_for(
+			6,
+			para_a,
+			VersionedXcm::from(Xcm(vec![HrmpChannelClosing {
+				initiator: u32::from(para_b),
+				sender: u32::from(para_b),
+				recipient: u32::from(para_a),
+			}]))
+			.encode(),
+		);
+	});
+}