Add support for versioned notification for HRMP pallet (paritytech#4281)

Closes: paritytech#4003 (please
see for the problem description)

## TODO
- [x] add more tests covering `WrapVersion` corner cases (e.g. para has
lower version, ...)
- [x] regenerate benchmarks `runtime_parachains::hrmp` (fix for Rococo
is here: paritytech#4332)

## Questions / possible improvements
- [ ] A `WrapVersion` implementation for `pallet_xcm` initiates version
discovery with
[note_unknown_version](https://github.com/paritytech/polkadot-sdk/blob/master/polkadot/xcm/pallet-xcm/src/lib.rs#L2527C5-L2527C25),
there is possibility to avoid this overhead in this HRMP case to create
new `WrapVersion` adapter for `pallet_xcm` which would not use
`note_unknown_version`. Is it worth to do it or not?
- [ ] There's a possibility to decouple XCM functionality from the HRMP
pallet, allowing any relay chain to generate its own notifications. This
approach wouldn't restrict notifications solely to the XCM. However,
it's uncertain whether it's worthwhile or desirable to do so? It means
making HRMP pallet more generic. E.g. hiding HRMP notifications behind
some trait:
	```
	trait HrmpNotifications {

		fn on_channel_open_request(
			sender: ParaId,
			proposed_max_capacity: u32,
			proposed_max_message_size: u32) -> primitives::DownwardMessage;

fn on_channel_accepted(recipient: ParaId) ->
primitives::DownwardMessage;

fn on_channel_closing(initiator: ParaId, sender: ParaId, recipient:
ParaId) -> primitives::DownwardMessage;
	}
	```
and then we could have whatever adapter, `impl HrmpNotifications for
VersionedXcmHrmpNotifications {...}`,
	```
	impl parachains_hrmp::Config for Runtime {
	..
		type HrmpNotifications = VersionedXcmHrmpNotifications;
	..
	}
	```

---------

Co-authored-by: command-bot <>
Co-authored-by: Bastian Köcher <[email protected]>

Author: 2 people

polkadot/runtime/parachains/src/hrmp.rs

@@ -278,6 +278,14 @@ pub mod pallet {
 		/// parachain.
 		type DefaultChannelSizeAndCapacityWithSystem: Get<(u32, u32)>;
 
+		/// Means of converting an `Xcm` into a `VersionedXcm`. This pallet sends HRMP XCM
+		/// notifications to the channel-related parachains, while the `WrapVersion` implementation
+		/// attempts to wrap them into the most suitable XCM version for the destination parachain.
+		///
+		/// NOTE: For example, `pallet_xcm` provides an accurate implementation (recommended), or
+		/// the default `()` implementation uses the latest XCM version for all parachains.
+		type VersionWrapper: xcm::WrapVersion;
+
 		/// Something that provides the weight of this pallet.
 		type WeightInfo: WeightInfo;
 	}
@@ -1499,28 +1507,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: origin.into(),
+					max_capacity: proposed_max_capacity,
+					max_message_size: proposed_max_message_size,
+				}])
+			}),
+		);
 
 		Ok(())
 	}
@@ -1562,23 +1561,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: origin.into() }])
+			}),
+		);
 
 		Ok(())
 	}
@@ -1633,30 +1624,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: origin.into(),
+					sender: channel_id.sender.into(),
+					recipient: channel_id.recipient.into(),
+				}])
+			}),
+		);
 
 		Ok(())
 	}
@@ -1875,3 +1858,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(),
+		);
+	});
+}