[XCM] Treat recursion limit error as transient in the MQ (#4202)

Changes:
- Add new error variant `ProcessMessageError::StackLimitReached` and
treat XCM error `ExceedsStackLimit` as such.

---------

Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io>
Co-authored-by: Branislav Kontur <bkontur@gmail.com>

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

@@ -765,6 +765,13 @@ enum MessageExecutionStatus {
 	Processed,
 	/// The message was processed and resulted in a, possibly permanent, error.
 	Unprocessable { permanent: bool },
+	/// The stack depth limit was reached.
+	///
+	/// We cannot just return `Unprocessable` in this case, because the processability of the
+	/// message depends on how the function was called. This may be a permanent error if it was
+	/// called by a top-level function, or a transient error if it was already called in a nested
+	/// function.
+	StackLimitReached,
 }
 
 impl<T: Config> Pallet<T> {
@@ -984,7 +991,8 @@ impl<T: Config> Pallet<T> {
 			// additional overweight event being deposited.
 		) {
 			Overweight | InsufficientWeight => Err(Error::<T>::InsufficientWeight),
-			Unprocessable { permanent: false } => Err(Error::<T>::TemporarilyUnprocessable),
+			StackLimitReached | Unprocessable { permanent: false } =>
+				Err(Error::<T>::TemporarilyUnprocessable),
 			Unprocessable { permanent: true } | Processed => {
 				page.note_processed_at_pos(pos);
 				book_state.message_count.saturating_dec();
@@ -1250,7 +1258,7 @@ impl<T: Config> Pallet<T> {
 		let is_processed = match res {
 			InsufficientWeight => return ItemExecutionStatus::Bailed,
 			Unprocessable { permanent: false } => return ItemExecutionStatus::NoProgress,
-			Processed | Unprocessable { permanent: true } => true,
+			Processed | Unprocessable { permanent: true } | StackLimitReached => true,
 			Overweight => false,
 		};
 
@@ -1461,6 +1469,10 @@ impl<T: Config> Pallet<T> {
 				Self::deposit_event(Event::<T>::ProcessingFailed { id: id.into(), origin, error });
 				MessageExecutionStatus::Unprocessable { permanent: true }
 			},
+			Err(error @ StackLimitReached) => {
+				Self::deposit_event(Event::<T>::ProcessingFailed { id: id.into(), origin, error });
+				MessageExecutionStatus::StackLimitReached
+			},
 			Ok(success) => {
 				// Success
 				let weight_used = meter.consumed().saturating_sub(prev_consumed);

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

@@ -198,6 +198,7 @@ impl ProcessMessage for RecordingMessageProcessor {
 
 parameter_types! {
 	pub static Callback: Box<fn (&MessageOrigin, u32)> = Box::new(|_, _| {});
+	pub static IgnoreStackOvError: bool = false;
 }
 
 /// Processed a mocked message. Messages that end with `badformat`, `corrupt`, `unsupported` or
@@ -216,6 +217,8 @@ fn processing_message(msg: &[u8], origin: &MessageOrigin) -> Result<(), ProcessM
 		Err(ProcessMessageError::Unsupported)
 	} else if msg.ends_with("yield") {
 		Err(ProcessMessageError::Yield)
+	} else if msg.ends_with("stacklimitreached") && !IgnoreStackOvError::get() {
+		Err(ProcessMessageError::StackLimitReached)
 	} else {
 		Ok(())
 	}

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

@@ -174,9 +174,10 @@ fn service_queues_failing_messages_works() {
 		MessageQueue::enqueue_message(msg("badformat"), Here);
 		MessageQueue::enqueue_message(msg("corrupt"), Here);
 		MessageQueue::enqueue_message(msg("unsupported"), Here);
+		MessageQueue::enqueue_message(msg("stacklimitreached"), Here);
 		MessageQueue::enqueue_message(msg("yield"), Here);
 		// Starts with four pages.
-		assert_pages(&[0, 1, 2, 3]);
+		assert_pages(&[0, 1, 2, 3, 4]);
 
 		assert_eq!(MessageQueue::service_queues(1.into_weight()), 1.into_weight());
 		assert_last_event::<Test>(
@@ -206,9 +207,9 @@ fn service_queues_failing_messages_works() {
 			.into(),
 		);
 		assert_eq!(MessageQueue::service_queues(1.into_weight()), 1.into_weight());
-		assert_eq!(System::events().len(), 3);
+		assert_eq!(System::events().len(), 4);
 		// Last page with the `yield` stays in.
-		assert_pages(&[3]);
+		assert_pages(&[4]);
 	});
 }
 
@@ -1880,3 +1881,97 @@ fn process_enqueued_on_idle_requires_enough_weight() {
 		assert_eq!(MessagesProcessed::take(), vec![]);
 	})
 }
+
+/// A message that reports `StackLimitReached` will not be put into the overweight queue when
+/// executed from the top level.
+#[test]
+fn process_discards_stack_ov_message() {
+	use MessageOrigin::*;
+	build_and_execute::<Test>(|| {
+		MessageQueue::enqueue_message(msg("stacklimitreached"), Here);
+
+		MessageQueue::service_queues(10.into_weight());
+
+		assert_last_event::<Test>(
+			Event::ProcessingFailed {
+				id: blake2_256(b"stacklimitreached").into(),
+				origin: MessageOrigin::Here,
+				error: ProcessMessageError::StackLimitReached,
+			}
+			.into(),
+		);
+
+		assert!(MessagesProcessed::take().is_empty());
+		// Message is gone and not overweight:
+		assert_pages(&[]);
+	});
+}
+
+/// A message that reports `StackLimitReached` will stay in the overweight queue when it is executed
+/// by `execute_overweight`.
+#[test]
+fn execute_overweight_keeps_stack_ov_message() {
+	use MessageOrigin::*;
+	build_and_execute::<Test>(|| {
+		// We need to create a mocked message that first reports insufficient weight, and then
+		// `StackLimitReached`:
+		IgnoreStackOvError::set(true);
+		MessageQueue::enqueue_message(msg("stacklimitreached"), Here);
+		MessageQueue::service_queues(0.into_weight());
+
+		assert_last_event::<Test>(
+			Event::OverweightEnqueued {
+				id: blake2_256(b"stacklimitreached"),
+				origin: MessageOrigin::Here,
+				message_index: 0,
+				page_index: 0,
+			}
+			.into(),
+		);
+		// Does not count as 'processed':
+		assert!(MessagesProcessed::take().is_empty());
+		assert_pages(&[0]);
+
+		// Now let it return `StackLimitReached`. Note that this case would normally not happen,
+		// since we assume that the top-level execution is the one with the most remaining stack
+		// depth.
+		IgnoreStackOvError::set(false);
+		// Ensure that trying to execute the message does not change any state (besides events).
+		System::reset_events();
+		let storage_noop = StorageNoopGuard::new();
+		assert_eq!(
+			<MessageQueue as ServiceQueues>::execute_overweight(3.into_weight(), (Here, 0, 0)),
+			Err(ExecuteOverweightError::Other)
+		);
+		assert_last_event::<Test>(
+			Event::ProcessingFailed {
+				id: blake2_256(b"stacklimitreached").into(),
+				origin: MessageOrigin::Here,
+				error: ProcessMessageError::StackLimitReached,
+			}
+			.into(),
+		);
+		System::reset_events();
+		drop(storage_noop);
+
+		// Now let's process it normally:
+		IgnoreStackOvError::set(true);
+		assert_eq!(
+			<MessageQueue as ServiceQueues>::execute_overweight(1.into_weight(), (Here, 0, 0))
+				.unwrap(),
+			1.into_weight()
+		);
+
+		assert_last_event::<Test>(
+			Event::Processed {
+				id: blake2_256(b"stacklimitreached").into(),
+				origin: MessageOrigin::Here,
+				weight_used: 1.into_weight(),
+				success: true,
+			}
+			.into(),
+		);
+		assert_pages(&[]);
+		System::reset_events();
+	});
+}

substrate/frame/support/src/traits/messages.rs

@@ -46,6 +46,8 @@ pub enum ProcessMessageError {
 	/// the case that a queue is re-serviced within the same block after *yielding*. A queue is
 	/// not required to *yield* again when it is being re-serviced withing the same block.
 	Yield,
+	/// The message could not be processed for reaching the stack depth limit.
+	StackLimitReached,
 }
 
 /// Can process messages from a specific origin.
@@ -96,6 +98,8 @@ pub trait ServiceQueues {
 	/// - `weight_limit`: The maximum amount of dynamic weight that this call can use.
 	///
 	/// Returns the dynamic weight used by this call; is never greater than `weight_limit`.
+	/// Should only be called in top-level runtime entry points like `on_initialize` or `on_idle`.
+	/// Otherwise, stack depth limit errors may be miss-handled.
 	fn service_queues(weight_limit: Weight) -> Weight;
 
 	/// Executes a message that could not be executed by [`Self::service_queues()`] because it was