[stable2409] Backport #6205 (#6239)

Backport #6205 into `stable2409` from bkchr.

See the
[documentation](https://github.com/paritytech/polkadot-sdk/blob/master/docs/BACKPORT.md)
on how to use this bot.

<!--
  # To be used by other automation, do not modify:
  original-pr-number: #${pull_number}
-->

Co-authored-by: Bastian Köcher <git@kchr.de>
Co-authored-by: Egor_P <egor@parity.io>

prdoc/pr_6205.prdoc

@@ -0,0 +1,8 @@
+title: 'pallet-message-queue: Fix max message size calculation'
+doc:
+- audience: Runtime Dev
+  description: |-
+    The max size of a message should not depend on the weight left in a given execution context. Instead the max message size depends on the service weights configured for the pallet. A message that may does not fit into `on_idle` is not automatically overweight, because it may can be executed successfully in `on_initialize` or in another block in `on_idle` when there is more weight left.
+crates:
+- name: pallet-message-queue
+  bump: patch

substrate/frame/message-queue/src/lib.rs

@@ -868,13 +868,26 @@ impl<T: Config> Pallet<T> {
 		}
 	}
 
-	/// The maximal weight that a single message can consume.
+	/// The maximal weight that a single message ever can consume.
 	///
 	/// Any message using more than this will be marked as permanently overweight and not
 	/// automatically re-attempted. Returns `None` if the servicing of a message cannot begin.
 	/// `Some(0)` means that only messages with no weight may be served.
 	fn max_message_weight(limit: Weight) -> Option<Weight> {
-		limit.checked_sub(&Self::single_msg_overhead())
+		let service_weight = T::ServiceWeight::get().unwrap_or_default();
+		let on_idle_weight = T::IdleMaxServiceWeight::get().unwrap_or_default();
+
+		// Whatever weight is set, the one with the biggest one is used as the maximum weight. If a
+		// message is tried in one context and fails, it will be retried in the other context later.
+		let max_message_weight =
+			if service_weight.any_gt(on_idle_weight) { service_weight } else { on_idle_weight };
+
+		if max_message_weight.is_zero() {
+			// If no service weight is set, we need to use the given limit as max message weight.
+			limit.checked_sub(&Self::single_msg_overhead())
+		} else {
+			max_message_weight.checked_sub(&Self::single_msg_overhead())
+		}
 	}
 
 	/// The overhead of servicing a single message.
@@ -896,6 +909,8 @@ impl<T: Config> Pallet<T> {
 	fn do_integrity_test() -> Result<(), String> {
 		ensure!(!MaxMessageLenOf::<T>::get().is_zero(), "HeapSize too low");
 
+		let max_block = T::BlockWeights::get().max_block;
+
 		if let Some(service) = T::ServiceWeight::get() {
 			if Self::max_message_weight(service).is_none() {
 				return Err(format!(
@@ -904,6 +919,31 @@ impl<T: Config> Pallet<T> {
 					Self::single_msg_overhead(),
 				))
 			}
+
+			if service.any_gt(max_block) {
+				return Err(format!(
+					"ServiceWeight {service} is bigger than max block weight {max_block}"
+				))
+			}
+		}
+
+		if let Some(on_idle) = T::IdleMaxServiceWeight::get() {
+			if on_idle.any_gt(max_block) {
+				return Err(format!(
+					"IdleMaxServiceWeight {on_idle} is bigger than max block weight {max_block}"
+				))
+			}
+		}
+
+		if let (Some(service_weight), Some(on_idle)) =
+			(T::ServiceWeight::get(), T::IdleMaxServiceWeight::get())
+		{
+			if !(service_weight.all_gt(on_idle) ||
+				on_idle.all_gt(service_weight) ||
+				service_weight == on_idle)
+			{
+				return Err("One of `ServiceWeight` or `IdleMaxServiceWeight` needs to be `all_gt` or both need to be equal.".into())
+			}
 		}
 
 		Ok(())
@@ -1509,7 +1549,7 @@ impl<T: Config> Pallet<T> {
 		let mut weight = WeightMeter::with_limit(weight_limit);
 
 		// Get the maximum weight that processing a single message may take:
-		let max_weight = Self::max_message_weight(weight_limit).unwrap_or_else(|| {
+		let overweight_limit = Self::max_message_weight(weight_limit).unwrap_or_else(|| {
 			if matches!(context, ServiceQueuesContext::OnInitialize) {
 				defensive!("Not enough weight to service a single message.");
 			}
@@ -1527,7 +1567,8 @@ impl<T: Config> Pallet<T> {
 			let mut last_no_progress = None;
 
 			loop {
-				let (progressed, n) = Self::service_queue(next.clone(), &mut weight, max_weight);
+				let (progressed, n) =
+					Self::service_queue(next.clone(), &mut weight, overweight_limit);
 				next = match n {
 					Some(n) =>
 						if !progressed {

substrate/frame/message-queue/src/mock.rs

@@ -42,7 +42,7 @@ impl frame_system::Config for Test {
 	type Block = Block;
 }
 parameter_types! {
-	pub const HeapSize: u32 = 24;
+	pub const HeapSize: u32 = 40;
 	pub const MaxStale: u32 = 2;
 	pub const ServiceWeight: Option<Weight> = Some(Weight::from_parts(100, 100));
 }