#![cfg_attr(not(feature = "std"), no_std)]
pub mod migration;
#[cfg(test)]
mod mock;
#[cfg(test)]
mod tests;
#[cfg(feature = "runtime-benchmarks")]
mod benchmarking;
#[cfg(feature = "bridging")]
pub mod bridging;
pub mod weights;
pub use weights::WeightInfo;
extern crate alloc;
use alloc::vec::Vec;
use bounded_collections::BoundedBTreeSet;
use codec::{Decode, DecodeLimit, Encode, MaxEncodedLen};
use cumulus_primitives_core::{
relay_chain::BlockNumber as RelayBlockNumber, ChannelStatus, GetChannelInfo, MessageSendError,
ParaId, XcmpMessageFormat, XcmpMessageHandler, XcmpMessageSource,
};
use frame_support::{
defensive, defensive_assert,
traits::{Defensive, EnqueueMessage, EnsureOrigin, Get, QueueFootprint, QueuePausedQuery},
weights::{Weight, WeightMeter},
BoundedVec,
};
use pallet_message_queue::OnQueueChanged;
use polkadot_runtime_common::xcm_sender::PriceForMessageDelivery;
use polkadot_runtime_parachains::FeeTracker;
use scale_info::TypeInfo;
use sp_core::MAX_POSSIBLE_ALLOCATION;
use sp_runtime::{FixedU128, RuntimeDebug, Saturating, WeakBoundedVec};
use xcm::{latest::prelude::*, VersionedLocation, VersionedXcm, WrapVersion, MAX_XCM_DECODE_DEPTH};
use xcm_builder::InspectMessageQueues;
use xcm_executor::traits::ConvertOrigin;
pub use pallet::*;
pub type OverweightIndex = u64;
pub type MaxXcmpMessageLenOf<T> =
<<T as Config>::XcmpQueue as EnqueueMessage<ParaId>>::MaxMessageLen;
const LOG_TARGET: &str = "xcmp_queue";
const DEFAULT_POV_SIZE: u64 = 64 * 1024; pub mod delivery_fee_constants {
use super::FixedU128;
pub const THRESHOLD_FACTOR: u32 = 2;
pub const EXPONENTIAL_FEE_BASE: FixedU128 = FixedU128::from_rational(105, 100); pub const MESSAGE_SIZE_FEE_BASE: FixedU128 = FixedU128::from_rational(1, 1000); }
#[frame_support::pallet]
pub mod pallet {
use super::*;
use frame_support::{pallet_prelude::*, Twox64Concat};
use frame_system::pallet_prelude::*;
#[pallet::pallet]
#[pallet::storage_version(migration::STORAGE_VERSION)]
pub struct Pallet<T>(_);
#[pallet::config]
pub trait Config: frame_system::Config {
type RuntimeEvent: From<Event<Self>> + IsType<<Self as frame_system::Config>::RuntimeEvent>;
type ChannelInfo: GetChannelInfo;
type VersionWrapper: WrapVersion;
type XcmpQueue: EnqueueMessage<ParaId>;
#[pallet::constant]
type MaxInboundSuspended: Get<u32>;
#[pallet::constant]
type MaxActiveOutboundChannels: Get<u32>;
#[pallet::constant]
type MaxPageSize: Get<u32>;
type ControllerOrigin: EnsureOrigin<Self::RuntimeOrigin>;
type ControllerOriginConverter: ConvertOrigin<Self::RuntimeOrigin>;
type PriceForSiblingDelivery: PriceForMessageDelivery<Id = ParaId>;
type WeightInfo: WeightInfo;
}
#[pallet::call]
impl<T: Config> Pallet<T> {
#[pallet::call_index(1)]
#[pallet::weight((T::DbWeight::get().writes(1), DispatchClass::Operational,))]
pub fn suspend_xcm_execution(origin: OriginFor<T>) -> DispatchResult {
T::ControllerOrigin::ensure_origin(origin)?;
QueueSuspended::<T>::try_mutate(|suspended| {
if *suspended {
Err(Error::<T>::AlreadySuspended.into())
} else {
*suspended = true;
Ok(())
}
})
}
#[pallet::call_index(2)]
#[pallet::weight((T::DbWeight::get().writes(1), DispatchClass::Operational,))]
pub fn resume_xcm_execution(origin: OriginFor<T>) -> DispatchResult {
T::ControllerOrigin::ensure_origin(origin)?;
QueueSuspended::<T>::try_mutate(|suspended| {
if !*suspended {
Err(Error::<T>::AlreadyResumed.into())
} else {
*suspended = false;
Ok(())
}
})
}
#[pallet::call_index(3)]
#[pallet::weight((T::WeightInfo::set_config_with_u32(), DispatchClass::Operational,))]
pub fn update_suspend_threshold(origin: OriginFor<T>, new: u32) -> DispatchResult {
ensure_root(origin)?;
QueueConfig::<T>::try_mutate(|data| {
data.suspend_threshold = new;
data.validate::<T>()
})
}
#[pallet::call_index(4)]
#[pallet::weight((T::WeightInfo::set_config_with_u32(),DispatchClass::Operational,))]
pub fn update_drop_threshold(origin: OriginFor<T>, new: u32) -> DispatchResult {
ensure_root(origin)?;
QueueConfig::<T>::try_mutate(|data| {
data.drop_threshold = new;
data.validate::<T>()
})
}
#[pallet::call_index(5)]
#[pallet::weight((T::WeightInfo::set_config_with_u32(), DispatchClass::Operational,))]
pub fn update_resume_threshold(origin: OriginFor<T>, new: u32) -> DispatchResult {
ensure_root(origin)?;
QueueConfig::<T>::try_mutate(|data| {
data.resume_threshold = new;
data.validate::<T>()
})
}
}
#[pallet::hooks]
impl<T: Config> Hooks<BlockNumberFor<T>> for Pallet<T> {
fn integrity_test() {
let w = Self::on_idle_weight();
assert!(w != Weight::zero());
assert!(w.all_lte(T::BlockWeights::get().max_block));
}
fn on_idle(_block: BlockNumberFor<T>, limit: Weight) -> Weight {
let mut meter = WeightMeter::with_limit(limit);
if meter.try_consume(Self::on_idle_weight()).is_err() {
log::debug!(
"Not enough weight for on_idle. {} < {}",
Self::on_idle_weight(),
limit
);
return meter.consumed()
}
migration::v3::lazy_migrate_inbound_queue::<T>();
meter.consumed()
}
}
#[pallet::event]
#[pallet::generate_deposit(pub(super) fn deposit_event)]
pub enum Event<T: Config> {
XcmpMessageSent { message_hash: XcmHash },
}
#[pallet::error]
pub enum Error<T> {
BadQueueConfig,
AlreadySuspended,
AlreadyResumed,
TooManyActiveOutboundChannels,
TooBig,
}
#[pallet::storage]
pub type InboundXcmpSuspended<T: Config> =
StorageValue<_, BoundedBTreeSet<ParaId, T::MaxInboundSuspended>, ValueQuery>;
#[pallet::storage]
pub(super) type OutboundXcmpStatus<T: Config> = StorageValue<
_,
BoundedVec<OutboundChannelDetails, T::MaxActiveOutboundChannels>,
ValueQuery,
>;
#[pallet::storage]
pub(super) type OutboundXcmpMessages<T: Config> = StorageDoubleMap<
_,
Blake2_128Concat,
ParaId,
Twox64Concat,
u16,
WeakBoundedVec<u8, T::MaxPageSize>,
ValueQuery,
>;
#[pallet::storage]
pub(super) type SignalMessages<T: Config> =
StorageMap<_, Blake2_128Concat, ParaId, WeakBoundedVec<u8, T::MaxPageSize>, ValueQuery>;
#[pallet::storage]
pub(super) type QueueConfig<T: Config> = StorageValue<_, QueueConfigData, ValueQuery>;
#[pallet::storage]
pub(super) type QueueSuspended<T: Config> = StorageValue<_, bool, ValueQuery>;
#[pallet::type_value]
pub fn InitialFactor() -> FixedU128 {
FixedU128::from_u32(1)
}
#[pallet::storage]
pub(super) type DeliveryFeeFactor<T: Config> =
StorageMap<_, Twox64Concat, ParaId, FixedU128, ValueQuery, InitialFactor>;
}
#[derive(Copy, Clone, Eq, PartialEq, Encode, Decode, RuntimeDebug, TypeInfo, MaxEncodedLen)]
pub enum OutboundState {
Ok,
Suspended,
}
#[derive(Clone, Eq, PartialEq, Encode, Decode, TypeInfo, RuntimeDebug, MaxEncodedLen)]
pub struct OutboundChannelDetails {
recipient: ParaId,
state: OutboundState,
signals_exist: bool,
first_index: u16,
last_index: u16,
}
impl OutboundChannelDetails {
pub fn new(recipient: ParaId) -> OutboundChannelDetails {
OutboundChannelDetails {
recipient,
state: OutboundState::Ok,
signals_exist: false,
first_index: 0,
last_index: 0,
}
}
pub fn with_signals(mut self) -> OutboundChannelDetails {
self.signals_exist = true;
self
}
pub fn with_suspended_state(mut self) -> OutboundChannelDetails {
self.state = OutboundState::Suspended;
self
}
}
#[derive(Copy, Clone, Eq, PartialEq, Encode, Decode, RuntimeDebug, TypeInfo, MaxEncodedLen)]
pub struct QueueConfigData {
suspend_threshold: u32,
drop_threshold: u32,
resume_threshold: u32,
}
impl Default for QueueConfigData {
fn default() -> Self {
Self {
drop_threshold: 48, suspend_threshold: 32, resume_threshold: 8, }
}
}
impl QueueConfigData {
pub fn validate<T: crate::Config>(&self) -> sp_runtime::DispatchResult {
if self.resume_threshold < self.suspend_threshold &&
self.suspend_threshold <= self.drop_threshold &&
self.resume_threshold > 0
{
Ok(())
} else {
Err(Error::<T>::BadQueueConfig.into())
}
}
}
#[derive(PartialEq, Eq, Copy, Clone, Encode, Decode, TypeInfo)]
pub enum ChannelSignal {
Suspend,
Resume,
}
impl<T: Config> Pallet<T> {
fn send_fragment<Fragment: Encode>(
recipient: ParaId,
format: XcmpMessageFormat,
fragment: Fragment,
) -> Result<u32, MessageSendError> {
let encoded_fragment = fragment.encode();
let channel_info =
T::ChannelInfo::get_channel_info(recipient).ok_or(MessageSendError::NoChannel)?;
let max_message_size = channel_info.max_message_size.min(T::MaxPageSize::get()) as usize;
let format_size = format.encoded_size();
let size_to_check = encoded_fragment
.len()
.checked_add(format_size)
.ok_or(MessageSendError::TooBig)?;
if size_to_check > max_message_size {
return Err(MessageSendError::TooBig)
}
let mut all_channels = <OutboundXcmpStatus<T>>::get();
let channel_details = if let Some(details) =
all_channels.iter_mut().find(|channel| channel.recipient == recipient)
{
details
} else {
all_channels.try_push(OutboundChannelDetails::new(recipient)).map_err(|e| {
log::error!("Failed to activate HRMP channel: {:?}", e);
MessageSendError::TooManyChannels
})?;
all_channels
.last_mut()
.expect("can't be empty; a new element was just pushed; qed")
};
let have_active = channel_details.last_index > channel_details.first_index;
let appended_to_last_page = have_active
.then(|| {
<OutboundXcmpMessages<T>>::try_mutate(
recipient,
channel_details.last_index - 1,
|page| {
if XcmpMessageFormat::decode_with_depth_limit(
MAX_XCM_DECODE_DEPTH,
&mut &page[..],
) != Ok(format)
{
defensive!("Bad format in outbound queue; dropping message");
return Err(())
}
if page.len() + encoded_fragment.len() > max_message_size {
return Err(())
}
for frag in encoded_fragment.iter() {
page.try_push(*frag)?;
}
Ok(page.len())
},
)
.ok()
})
.flatten();
let (number_of_pages, last_page_size) = if let Some(size) = appended_to_last_page {
let number_of_pages = (channel_details.last_index - channel_details.first_index) as u32;
(number_of_pages, size)
} else {
let page_index = channel_details.last_index;
channel_details.last_index += 1;
let mut new_page = format.encode();
new_page.extend_from_slice(&encoded_fragment[..]);
let last_page_size = new_page.len();
let number_of_pages = (channel_details.last_index - channel_details.first_index) as u32;
let bounded_page = BoundedVec::<u8, T::MaxPageSize>::try_from(new_page)
.map_err(|_| MessageSendError::TooBig)?;
let bounded_page = WeakBoundedVec::force_from(bounded_page.into_inner(), None);
<OutboundXcmpMessages<T>>::insert(recipient, page_index, bounded_page);
<OutboundXcmpStatus<T>>::put(all_channels);
(number_of_pages, last_page_size)
};
let total_size =
number_of_pages.saturating_sub(1) * max_message_size as u32 + last_page_size as u32;
let threshold = channel_info.max_total_size / delivery_fee_constants::THRESHOLD_FACTOR;
if total_size > threshold {
let message_size_factor = FixedU128::from((encoded_fragment.len() / 1024) as u128)
.saturating_mul(delivery_fee_constants::MESSAGE_SIZE_FEE_BASE);
Self::increase_fee_factor(recipient, message_size_factor);
}
Ok(number_of_pages)
}
fn send_signal(dest: ParaId, signal: ChannelSignal) -> Result<(), Error<T>> {
let mut s = <OutboundXcmpStatus<T>>::get();
if let Some(details) = s.iter_mut().find(|item| item.recipient == dest) {
details.signals_exist = true;
} else {
s.try_push(OutboundChannelDetails::new(dest).with_signals())
.map_err(|_| Error::<T>::TooManyActiveOutboundChannels)?;
}
let page = BoundedVec::<u8, T::MaxPageSize>::try_from(
(XcmpMessageFormat::Signals, signal).encode(),
)
.map_err(|_| Error::<T>::TooBig)?;
let page = WeakBoundedVec::force_from(page.into_inner(), None);
<SignalMessages<T>>::insert(dest, page);
<OutboundXcmpStatus<T>>::put(s);
Ok(())
}
fn suspend_channel(target: ParaId) {
<OutboundXcmpStatus<T>>::mutate(|s| {
if let Some(details) = s.iter_mut().find(|item| item.recipient == target) {
let ok = details.state == OutboundState::Ok;
defensive_assert!(ok, "WARNING: Attempt to suspend channel that was not Ok.");
details.state = OutboundState::Suspended;
} else {
if s.try_push(OutboundChannelDetails::new(target).with_suspended_state()).is_err() {
defensive!("Cannot pause channel; too many outbound channels");
}
}
});
}
fn resume_channel(target: ParaId) {
<OutboundXcmpStatus<T>>::mutate(|s| {
if let Some(index) = s.iter().position(|item| item.recipient == target) {
let suspended = s[index].state == OutboundState::Suspended;
defensive_assert!(
suspended,
"WARNING: Attempt to resume channel that was not suspended."
);
if s[index].first_index == s[index].last_index {
s.remove(index);
} else {
s[index].state = OutboundState::Ok;
}
} else {
defensive!("WARNING: Attempt to resume channel that was not suspended.");
}
});
}
fn enqueue_xcmp_message(
sender: ParaId,
xcm: BoundedVec<u8, MaxXcmpMessageLenOf<T>>,
meter: &mut WeightMeter,
) -> Result<(), ()> {
if meter.try_consume(T::WeightInfo::enqueue_xcmp_message()).is_err() {
defensive!("Out of weight: cannot enqueue XCMP messages; dropping msg");
return Err(())
}
let QueueConfigData { drop_threshold, .. } = <QueueConfig<T>>::get();
let fp = T::XcmpQueue::footprint(sender);
let new_pages = fp.ready_pages.saturating_add(1);
if new_pages > drop_threshold {
log::error!("XCMP queue for sibling {:?} is full; dropping messages.", sender);
return Err(())
}
T::XcmpQueue::enqueue_message(xcm.as_bounded_slice(), sender);
Ok(())
}
pub(crate) fn take_first_concatenated_xcm(
data: &mut &[u8],
meter: &mut WeightMeter,
) -> Result<BoundedVec<u8, MaxXcmpMessageLenOf<T>>, ()> {
if data.is_empty() {
return Err(())
}
if meter.try_consume(T::WeightInfo::take_first_concatenated_xcm()).is_err() {
defensive!("Out of weight; could not decode all; dropping");
return Err(())
}
let xcm = VersionedXcm::<()>::decode_with_depth_limit(MAX_XCM_DECODE_DEPTH, data)
.map_err(|_| ())?;
xcm.encode().try_into().map_err(|_| ())
}
pub fn on_idle_weight() -> Weight {
<T as crate::Config>::WeightInfo::on_idle_good_msg()
.max(<T as crate::Config>::WeightInfo::on_idle_large_msg())
}
#[cfg(feature = "bridging")]
fn is_inbound_channel_suspended(sender: ParaId) -> bool {
<InboundXcmpSuspended<T>>::get().iter().any(|c| c == &sender)
}
#[cfg(feature = "bridging")]
fn outbound_channel_state(target: ParaId) -> Option<(OutboundState, u16)> {
<OutboundXcmpStatus<T>>::get().iter().find(|c| c.recipient == target).map(|c| {
let queued_pages = c.last_index.saturating_sub(c.first_index);
(c.state, queued_pages)
})
}
}
impl<T: Config> OnQueueChanged<ParaId> for Pallet<T> {
fn on_queue_changed(para: ParaId, fp: QueueFootprint) {
let QueueConfigData { resume_threshold, suspend_threshold, .. } = <QueueConfig<T>>::get();
let mut suspended_channels = <InboundXcmpSuspended<T>>::get();
let suspended = suspended_channels.contains(¶);
if suspended && fp.ready_pages <= resume_threshold {
if let Err(err) = Self::send_signal(para, ChannelSignal::Resume) {
log::error!("defensive: Could not send resumption signal to inbound channel of sibling {:?}: {:?}; channel remains suspended.", para, err);
} else {
suspended_channels.remove(¶);
<InboundXcmpSuspended<T>>::put(suspended_channels);
}
} else if !suspended && fp.ready_pages >= suspend_threshold {
log::warn!("XCMP queue for sibling {:?} is full; suspending channel.", para);
if let Err(err) = Self::send_signal(para, ChannelSignal::Suspend) {
log::error!(
"defensive: Could not send suspension signal; future messages may be dropped: {:?}", err
);
} else if let Err(err) = suspended_channels.try_insert(para) {
log::error!("Too many channels suspended; cannot suspend sibling {:?}: {:?}; further messages may be dropped.", para, err);
} else {
<InboundXcmpSuspended<T>>::put(suspended_channels);
}
}
}
}
impl<T: Config> QueuePausedQuery<ParaId> for Pallet<T> {
fn is_paused(para: &ParaId) -> bool {
if !QueueSuspended::<T>::get() {
return false
}
let sender_origin = T::ControllerOriginConverter::convert_origin(
(Parent, Parachain((*para).into())),
OriginKind::Superuser,
);
let is_controller =
sender_origin.map_or(false, |origin| T::ControllerOrigin::try_origin(origin).is_ok());
!is_controller
}
}
impl<T: Config> XcmpMessageHandler for Pallet<T> {
fn handle_xcmp_messages<'a, I: Iterator<Item = (ParaId, RelayBlockNumber, &'a [u8])>>(
iter: I,
max_weight: Weight,
) -> Weight {
let mut meter = WeightMeter::with_limit(max_weight);
for (sender, _sent_at, mut data) in iter {
let format = match XcmpMessageFormat::decode(&mut data) {
Ok(f) => f,
Err(_) => {
defensive!("Unknown XCMP message format - dropping");
continue
},
};
match format {
XcmpMessageFormat::Signals =>
while !data.is_empty() {
if meter
.try_consume(
T::WeightInfo::suspend_channel()
.max(T::WeightInfo::resume_channel()),
)
.is_err()
{
defensive!("Not enough weight to process signals - dropping");
break
}
match ChannelSignal::decode(&mut data) {
Ok(ChannelSignal::Suspend) => Self::suspend_channel(sender),
Ok(ChannelSignal::Resume) => Self::resume_channel(sender),
Err(_) => {
defensive!("Undecodable channel signal - dropping");
break
},
}
},
XcmpMessageFormat::ConcatenatedVersionedXcm =>
while !data.is_empty() {
let Ok(xcm) = Self::take_first_concatenated_xcm(&mut data, &mut meter)
else {
defensive!("HRMP inbound decode stream broke; page will be dropped.",);
break
};
if let Err(()) = Self::enqueue_xcmp_message(sender, xcm, &mut meter) {
defensive!(
"Could not enqueue XCMP messages. Used weight: ",
meter.consumed_ratio()
);
break
}
},
XcmpMessageFormat::ConcatenatedEncodedBlob => {
defensive!("Blob messages are unhandled - dropping");
continue
},
}
}
meter.consumed()
}
}
impl<T: Config> XcmpMessageSource for Pallet<T> {
fn take_outbound_messages(maximum_channels: usize) -> Vec<(ParaId, Vec<u8>)> {
let mut statuses = <OutboundXcmpStatus<T>>::get();
let old_statuses_len = statuses.len();
let max_message_count = statuses.len().min(maximum_channels);
let mut result = Vec::with_capacity(max_message_count);
for status in statuses.iter_mut() {
let OutboundChannelDetails {
recipient: para_id,
state: outbound_state,
mut signals_exist,
mut first_index,
mut last_index,
} = *status;
let (max_size_now, max_size_ever) = match T::ChannelInfo::get_channel_status(para_id) {
ChannelStatus::Closed => {
for i in first_index..last_index {
<OutboundXcmpMessages<T>>::remove(para_id, i);
}
if signals_exist {
<SignalMessages<T>>::remove(para_id);
}
*status = OutboundChannelDetails::new(para_id);
continue
},
ChannelStatus::Full => continue,
ChannelStatus::Ready(n, e) => (n, e),
};
if result.len() == max_message_count {
break
}
let page = if signals_exist {
let page = <SignalMessages<T>>::get(para_id);
defensive_assert!(!page.is_empty(), "Signals must exist");
if page.len() < max_size_now {
<SignalMessages<T>>::remove(para_id);
signals_exist = false;
page
} else {
defensive!("Signals should fit into a single page");
continue
}
} else if outbound_state == OutboundState::Suspended {
continue
} else if last_index > first_index {
let page = <OutboundXcmpMessages<T>>::get(para_id, first_index);
if page.len() < max_size_now {
<OutboundXcmpMessages<T>>::remove(para_id, first_index);
first_index += 1;
page
} else {
continue
}
} else {
continue
};
if first_index == last_index {
first_index = 0;
last_index = 0;
}
if page.len() > max_size_ever {
defensive!("WARNING: oversize message in queue - dropping");
} else {
result.push((para_id, page.into_inner()));
}
let max_total_size = match T::ChannelInfo::get_channel_info(para_id) {
Some(channel_info) => channel_info.max_total_size,
None => {
log::warn!("calling `get_channel_info` with no RelevantMessagingState?!");
MAX_POSSIBLE_ALLOCATION },
};
let threshold = max_total_size.saturating_div(delivery_fee_constants::THRESHOLD_FACTOR);
let remaining_total_size: usize = (first_index..last_index)
.map(|index| OutboundXcmpMessages::<T>::decode_len(para_id, index).unwrap())
.sum();
if remaining_total_size <= threshold as usize {
Self::decrease_fee_factor(para_id);
}
*status = OutboundChannelDetails {
recipient: para_id,
state: outbound_state,
signals_exist,
first_index,
last_index,
};
}
debug_assert!(!statuses.iter().any(|s| s.signals_exist), "Signals should be handled");
result.sort_by_key(|m| m.0);
statuses.retain(|x| {
x.state == OutboundState::Suspended || x.signals_exist || x.first_index < x.last_index
});
let pruned = old_statuses_len - statuses.len();
let _ = statuses.try_rotate_left(result.len().saturating_sub(pruned)).defensive_proof(
"Could not store HRMP channels config. Some HRMP channels may be broken.",
);
<OutboundXcmpStatus<T>>::put(statuses);
result
}
}
impl<T: Config> SendXcm for Pallet<T> {
type Ticket = (ParaId, VersionedXcm<()>);
fn validate(
dest: &mut Option<Location>,
msg: &mut Option<Xcm<()>>,
) -> SendResult<(ParaId, VersionedXcm<()>)> {
let d = dest.take().ok_or(SendError::MissingArgument)?;
match d.unpack() {
(1, [Parachain(id)]) => {
let xcm = msg.take().ok_or(SendError::MissingArgument)?;
let id = ParaId::from(*id);
let price = T::PriceForSiblingDelivery::price_for_delivery(id, &xcm);
let versioned_xcm = T::VersionWrapper::wrap_version(&d, xcm)
.map_err(|()| SendError::DestinationUnsupported)?;
versioned_xcm
.validate_xcm_nesting()
.map_err(|()| SendError::ExceedsMaxMessageSize)?;
Ok(((id, versioned_xcm), price))
},
_ => {
*dest = Some(d);
Err(SendError::NotApplicable)
},
}
}
fn deliver((id, xcm): (ParaId, VersionedXcm<()>)) -> Result<XcmHash, SendError> {
let hash = xcm.using_encoded(sp_io::hashing::blake2_256);
match Self::send_fragment(id, XcmpMessageFormat::ConcatenatedVersionedXcm, xcm) {
Ok(_) => {
Self::deposit_event(Event::XcmpMessageSent { message_hash: hash });
Ok(hash)
},
Err(e) => {
log::error!(target: LOG_TARGET, "Deliver error: {e:?}");
Err(SendError::Transport(e.into()))
},
}
}
}
impl<T: Config> InspectMessageQueues for Pallet<T> {
fn clear_messages() {
let _ = OutboundXcmpMessages::<T>::clear(u32::MAX, None);
}
fn get_messages() -> Vec<(VersionedLocation, Vec<VersionedXcm<()>>)> {
use xcm::prelude::*;
OutboundXcmpMessages::<T>::iter()
.map(|(para_id, _, messages)| {
let mut data = &messages[..];
let decoded_format =
XcmpMessageFormat::decode_with_depth_limit(MAX_XCM_DECODE_DEPTH, &mut data)
.unwrap();
if decoded_format != XcmpMessageFormat::ConcatenatedVersionedXcm {
panic!("Unexpected format.")
}
let mut decoded_messages = Vec::new();
while !data.is_empty() {
let decoded_message = VersionedXcm::<()>::decode_with_depth_limit(
MAX_XCM_DECODE_DEPTH,
&mut data,
)
.unwrap();
decoded_messages.push(decoded_message);
}
(
VersionedLocation::V4((Parent, Parachain(para_id.into())).into()),
decoded_messages,
)
})
.collect()
}
}
impl<T: Config> FeeTracker for Pallet<T> {
type Id = ParaId;
fn get_fee_factor(id: Self::Id) -> FixedU128 {
<DeliveryFeeFactor<T>>::get(id)
}
fn increase_fee_factor(id: Self::Id, message_size_factor: FixedU128) -> FixedU128 {
<DeliveryFeeFactor<T>>::mutate(id, |f| {
*f = f.saturating_mul(
delivery_fee_constants::EXPONENTIAL_FEE_BASE.saturating_add(message_size_factor),
);
*f
})
}
fn decrease_fee_factor(id: Self::Id) -> FixedU128 {
<DeliveryFeeFactor<T>>::mutate(id, |f| {
*f = InitialFactor::get().max(*f / delivery_fee_constants::EXPONENTIAL_FEE_BASE);
*f
})
}
}