Transfer asset via bridge pallet xcm with dynamic fees and back-pressure

Cargo.toml

@@ -7,6 +7,7 @@ members = [
 	"bridges/modules/messages",
 	"bridges/modules/parachains",
 	"bridges/modules/relayers",
+	"bridges/modules/xcm-bridge-hub-router",
 	"client/cli",
 	"client/collator",
 	"client/consensus/aura",

pallets/xcmp-queue/Cargo.toml

@@ -28,6 +28,9 @@ cumulus-primitives-core = { path = "../../primitives/core", default-features = f
 # Optional import for benchmarking
 frame-benchmarking = { default-features = false, optional = true, git = "https://github.com/paritytech/substrate", branch = "master" }
 
+# Bridges
+bp-xcm-bridge-hub-router = { path = "../../bridges/primitives/xcm-bridge-hub-router", default-features = false, optional = true }
+
 [dev-dependencies]
 
 # Substrate
@@ -55,6 +58,7 @@ std = [
 	"sp-std/std",
 	"xcm-executor/std",
 	"xcm/std",
+	"bp-xcm-bridge-hub-router/std",
 ]
 
 runtime-benchmarks = [
@@ -64,3 +68,7 @@ runtime-benchmarks = [
 	"xcm-builder/runtime-benchmarks",
 ]
 try-runtime = ["frame-support/try-runtime"]
+
+bridging = [
+	"bp-xcm-bridge-hub-router",
+]

pallets/xcmp-queue/src/bridging.rs

@@ -0,0 +1,116 @@
+// Copyright (C) Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: Apache-2.0
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// 	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::pallet;
+use cumulus_primitives_core::ParaId;
+use frame_support::pallet_prelude::Get;
+
+/// Adapter implementation for `bp_xcm_bridge_hub_router::XcmChannelStatusProvider` which checks
+/// both `OutboundXcmpStatus` and `InboundXcmpStatus` for defined `ParaId` if any of those is
+/// suspended.
+pub struct InboundAndOutboundXcmpChannelCongestionStatusProvider<SiblingBridgeHubParaId, Runtime>(
+	sp_std::marker::PhantomData<(SiblingBridgeHubParaId, Runtime)>,
+);
+impl<SiblingBridgeHubParaId: Get<ParaId>, Runtime: crate::Config>
+	bp_xcm_bridge_hub_router::XcmChannelStatusProvider
+	for InboundAndOutboundXcmpChannelCongestionStatusProvider<SiblingBridgeHubParaId, Runtime>
+{
+	fn is_congested() -> bool {
+		// if the outbound channel with recipient is suspended, it means that one of further
+		// bridge queues (e.g. bridge queue between two bridge hubs) is overloaded, so we shall
+		// take larger fee for our outbound messages
+		let sibling_bridge_hub_id: ParaId = SiblingBridgeHubParaId::get();
+		let outbound_channels = pallet::OutboundXcmpStatus::<Runtime>::get();
+		let outbound_channel =
+			outbound_channels.iter().find(|c| c.recipient == sibling_bridge_hub_id);
+		let is_outbound_channel_suspended =
+			outbound_channel.map(|c| c.is_suspended()).unwrap_or(false);
+		if is_outbound_channel_suspended {
+			return true
+		}
+
+		// if the inbound channel with recipient is suspended, it means that we are unable to
+		// receive congestion reports from the bridge hub. So we assume the bridge pipeline is
+		// congested too
+		let inbound_channels = pallet::InboundXcmpStatus::<Runtime>::get();
+		let inbound_channel = inbound_channels.iter().find(|c| c.sender == sibling_bridge_hub_id);
+		let is_inbound_channel_suspended =
+			inbound_channel.map(|c| c.is_suspended()).unwrap_or(false);
+		if is_inbound_channel_suspended {
+			return true
+		}
+
+		// TODO: https://github.com/paritytech/cumulus/pull/2342 - once this PR is merged, we may
+		// remove the following code
+		//
+		// if the outbound channel has at least `N` pages enqueued, let's assume it is congested.
+		// Normally, the chain with a few opened HRMP channels, will "send" pages at every block.
+		// Having `N` pages means that for last `N` blocks we either have not sent any messages,
+		// or have sent signals.
+		const MAX_OUTBOUND_PAGES_BEFORE_CONGESTION: u16 = 4;
+		let is_outbound_channel_congested = outbound_channel.map(|c| c.queued_pages()).unwrap_or(0);
+		is_outbound_channel_congested > MAX_OUTBOUND_PAGES_BEFORE_CONGESTION
+	}
+}
+
+/// Adapter implementation for `bp_xcm_bridge_hub_router::XcmChannelStatusProvider` which checks
+/// only `OutboundXcmpStatus` for defined `SiblingParaId` if is suspended.
+pub struct OutboundXcmpChannelCongestionStatusProvider<SiblingBridgeHubParaId, Runtime>(
+	sp_std::marker::PhantomData<(SiblingBridgeHubParaId, Runtime)>,
+);
+impl<SiblingParaId: Get<ParaId>, Runtime: crate::Config>
+	bp_xcm_bridge_hub_router::XcmChannelStatusProvider
+	for OutboundXcmpChannelCongestionStatusProvider<SiblingParaId, Runtime>
+{
+	fn is_congested() -> bool {
+		// let's find the channel with the sibling parachain
+		let sibling_para_id: cumulus_primitives_core::ParaId = SiblingParaId::get();
+		let outbound_channels = pallet::OutboundXcmpStatus::<Runtime>::get();
+		let channel_with_sibling_parachain =
+			outbound_channels.iter().find(|c| c.recipient == sibling_para_id);
+
+		// no channel => it is empty, so not congested
+		let channel_with_sibling_parachain = match channel_with_sibling_parachain {
+			Some(channel_with_sibling_parachain) => channel_with_sibling_parachain,
+			None => return false,
+		};
+
+		// suspended channel => it is congested
+		if channel_with_sibling_parachain.is_suspended() {
+			return true
+		}
+
+		// TODO: the following restriction is arguable, we may live without that, assuming that
+		// there can't be more than some `N` messages queued at the bridge queue (at the source BH)
+		// AND before accepting next (or next-after-next) delivery transaction, we'll receive the
+		// suspension signal from the target parachain and stop accepting delivery transactions
+
+		// it takes some time for target parachain to suspend inbound channel with the target BH and
+		// during that we will keep accepting new message delivery transactions. Let's also reject
+		// new deliveries if there are too many "pages" (concatenated XCM messages) in the target BH
+		// -> target parachain queue.
+		const MAX_QUEUED_PAGES_BEFORE_DEACTIVATION: u16 = 4;
+		if channel_with_sibling_parachain.queued_pages() > MAX_QUEUED_PAGES_BEFORE_DEACTIVATION {
+			return true
+		}
+
+		true
+	}
+}
+
+#[cfg(feature = "runtime-benchmarks")]
+pub fn suspend_channel_for_benchmarks<T: crate::Config>(target: ParaId) {
+	pallet::Pallet::<T>::suspend_channel(target)
+}

pallets/xcmp-queue/src/lib.rs

@@ -35,6 +35,8 @@ mod tests;
 
 #[cfg(feature = "runtime-benchmarks")]
 mod benchmarking;
+#[cfg(feature = "bridging")]
+pub mod bridging;
 pub mod weights;
 pub use weights::WeightInfo;
 
@@ -400,6 +402,13 @@ pub struct InboundChannelDetails {
 	message_metadata: Vec<(RelayBlockNumber, XcmpMessageFormat)>,
 }
 
+impl InboundChannelDetails {
+	#[cfg(feature = "bridging")]
+	pub(crate) fn is_suspended(&self) -> bool {
+		self.state == InboundState::Suspended
+	}
+}
+
 /// Struct containing detailed information about the outbound channel.
 #[derive(Clone, Eq, PartialEq, Encode, Decode, TypeInfo)]
 pub struct OutboundChannelDetails {
@@ -435,6 +444,16 @@ impl OutboundChannelDetails {
 		self.state = OutboundState::Suspended;
 		self
 	}
+
+	#[cfg(feature = "bridging")]
+	pub(crate) fn is_suspended(&self) -> bool {
+		self.state == OutboundState::Suspended
+	}
+
+	#[cfg(feature = "bridging")]
+	pub(crate) fn queued_pages(&self) -> u16 {
+		self.last_index.saturating_sub(self.first_index)
+	}
 }
 
 #[derive(Copy, Clone, Eq, PartialEq, Encode, Decode, RuntimeDebug, TypeInfo)]

parachains/common/src/xcm_config.rs

@@ -1,14 +1,19 @@
 use crate::impls::AccountIdOf;
+use codec::Decode;
 use core::marker::PhantomData;
 use frame_support::{
-	log,
-	traits::{fungibles::Inspect, tokens::ConversionToAssetBalance, ContainsPair},
+	ensure, log,
+	traits::{
+		fungibles::Inspect, tokens::ConversionToAssetBalance, Contains, ContainsPair,
+		ProcessMessageError,
+	},
 	weights::Weight,
 	DefaultNoBound,
 };
 use sp_runtime::traits::Get;
-use xcm::latest::prelude::*;
-use xcm_builder::ExporterFor;
+use xcm::{latest::prelude::*, DoubleEncoded};
+use xcm_builder::{CreateMatcher, ExporterFor, MatchXcm};
+use xcm_executor::traits::ShouldExecute;
 
 /// A `ChargeFeeInFungibles` implementation that converts the output of
 /// a given WeightToFee implementation an amount charged in
@@ -156,6 +161,59 @@ impl<
 	}
 }
 
+/// Allows execution from `origin` if it is contained in `AllowedOrigin`
+/// and if it is just a straight `Transact` which contains `AllowedCall`.
+pub struct AllowUnpaidTransactsFrom<RuntimeCall, AllowedCall, AllowedOrigin>(
+	sp_std::marker::PhantomData<(RuntimeCall, AllowedCall, AllowedOrigin)>,
+);
+impl<
+		RuntimeCall: Decode,
+		AllowedCall: Contains<RuntimeCall>,
+		AllowedOrigin: Contains<MultiLocation>,
+	> ShouldExecute for AllowUnpaidTransactsFrom<RuntimeCall, AllowedCall, AllowedOrigin>
+{
+	fn should_execute<Call>(
+		origin: &MultiLocation,
+		instructions: &mut [Instruction<Call>],
+		max_weight: Weight,
+		_properties: &mut xcm_executor::traits::Properties,
+	) -> Result<(), ProcessMessageError> {
+		log::trace!(
+			target: "xcm::barriers",
+			"AllowUnpaidTransactFrom origin: {:?}, instructions: {:?}, max_weight: {:?}, properties: {:?}",
+			origin, instructions, max_weight, _properties,
+		);
+
+		// we only allow from configured origins
+		ensure!(AllowedOrigin::contains(origin), ProcessMessageError::Unsupported);
+
+		// we expect an XCM program with single `Transact` call
+		instructions
+			.matcher()
+			.assert_remaining_insts(1)?
+			.match_next_inst(|inst| match inst {
+				Transact { origin_kind: OriginKind::Xcm, call: encoded_call, .. } => {
+					// this is a hack - don't know if there's a way to do that properly
+					// or else we can simply allow all calls
+					let mut decoded_call = DoubleEncoded::<RuntimeCall>::from(encoded_call.clone());
+					ensure!(
+						AllowedCall::contains(
+							decoded_call
+								.ensure_decoded()
+								.map_err(|_| ProcessMessageError::BadFormat)?
+						),
+						ProcessMessageError::BadFormat,
+					);
+
+					Ok(())
+				},
+				_ => Err(ProcessMessageError::BadFormat),
+			})?;
+
+		Ok(())
+	}
+}
+
 #[cfg(test)]
 mod tests {
 	use super::*;