process enqueued messages on idle (#3844)

This will make it possible to use remaining weight on idle for
processing enqueued messages.
More context here https://github.com/paritytech/polkadot-sdk/issues/3709

---------

Co-authored-by: Adrian Catangiu <adrian@parity.io>

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

@@ -73,6 +73,7 @@ impl Config for Test {
 	type HeapSize = HeapSize;
 	type MaxStale = MaxStale;
 	type ServiceWeight = ServiceWeight;
+	type IdleMaxServiceWeight = ();
 }
 
 /// Simulates heavy usage by enqueueing and processing large amounts of messages.

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

@@ -525,12 +525,21 @@ pub mod pallet {
 		type MaxStale: Get<u32>;
 
 		/// The amount of weight (if any) which should be provided to the message queue for
-		/// servicing enqueued items.
+		/// servicing enqueued items `on_initialize`.
 		///
 		/// This may be legitimately `None` in the case that you will call
-		/// `ServiceQueues::service_queues` manually.
+		/// `ServiceQueues::service_queues` manually or set [`Self::IdleMaxServiceWeight`] to have
+		/// it run in `on_idle`.
 		#[pallet::constant]
 		type ServiceWeight: Get<Option<Weight>>;
+
+		/// The maximum amount of weight (if any) to be used from remaining weight `on_idle` which
+		/// should be provided to the message queue for servicing enqueued items `on_idle`.
+		/// Useful for parachains to process messages at the same block they are received.
+		///
+		/// If `None`, it will not call `ServiceQueues::service_queues` in `on_idle`.
+		#[pallet::constant]
+		type IdleMaxServiceWeight: Get<Option<Weight>>;
 	}
 
 	#[pallet::event]
@@ -643,6 +652,15 @@ pub mod pallet {
 			}
 		}
 
+		fn on_idle(_n: BlockNumberFor<T>, remaining_weight: Weight) -> Weight {
+			if let Some(weight_limit) = T::IdleMaxServiceWeight::get() {
+				// Make use of the remaining weight to process enqueued messages.
+				Self::service_queues(weight_limit.min(remaining_weight))
+			} else {
+				Weight::zero()
+			}
+		}
+
 		#[cfg(feature = "try-runtime")]
 		fn try_state(_: BlockNumberFor<T>) -> Result<(), sp_runtime::TryRuntimeError> {
 			Self::do_try_state()

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

@@ -56,6 +56,7 @@ impl Config for Test {
 	type HeapSize = HeapSize;
 	type MaxStale = MaxStale;
 	type ServiceWeight = ServiceWeight;
+	type IdleMaxServiceWeight = ServiceWeight;
 }
 
 /// Mocked `WeightInfo` impl with allows to set the weight per call.

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

@@ -1838,3 +1838,45 @@ fn with_service_mutex_works() {
 	with_service_mutex(|| called = 3).unwrap();
 	assert_eq!(called, 3);
 }
+
+#[test]
+fn process_enqueued_on_idle() {
+	use MessageOrigin::*;
+	build_and_execute::<Test>(|| {
+		// Some messages enqueued on previous block.
+		MessageQueue::enqueue_messages(vec![msg("a"), msg("ab"), msg("abc")].into_iter(), Here);
+		assert_eq!(BookStateFor::<Test>::iter().count(), 1);
+
+		// Process enqueued messages from previous block.
+		Pallet::<Test>::on_initialize(1);
+		assert_eq!(
+			MessagesProcessed::take(),
+			vec![(b"a".to_vec(), Here), (b"ab".to_vec(), Here), (b"abc".to_vec(), Here),]
+		);
+
+		MessageQueue::enqueue_messages(vec![msg("x"), msg("xy"), msg("xyz")].into_iter(), There);
+		assert_eq!(BookStateFor::<Test>::iter().count(), 2);
+
+		// Enough weight to process on idle.
+		Pallet::<Test>::on_idle(1, Weight::from_parts(100, 100));
+		assert_eq!(
+			MessagesProcessed::take(),
+			vec![(b"x".to_vec(), There), (b"xy".to_vec(), There), (b"xyz".to_vec(), There)]
+		);
+	})
+}
+
+#[test]
+fn process_enqueued_on_idle_requires_enough_weight() {
+	use MessageOrigin::*;
+	build_and_execute::<Test>(|| {
+		Pallet::<Test>::on_initialize(1);
+
+		MessageQueue::enqueue_messages(vec![msg("x"), msg("xy"), msg("xyz")].into_iter(), There);
+		assert_eq!(BookStateFor::<Test>::iter().count(), 1);
+
+		// Not enough weight to process on idle.
+		Pallet::<Test>::on_idle(1, Weight::from_parts(0, 0));
+		assert_eq!(MessagesProcessed::take(), vec![]);
+	})
+}