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

Closes: #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: #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 <git@kchr.de>
paritytech · May 7, 2024 · b709dcc · b709dcc
1 parent e434176
commit b709dcc
Show file tree

Hide file tree

Showing 11 changed files with 406 additions and 167 deletions.
diff --git a/polkadot/runtime/parachains/src/hrmp.rs b/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);
+		}
+	}
+}
diff --git a/polkadot/runtime/parachains/src/hrmp/tests.rs b/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(),
+		);
+	});
+}