extend overseer to send priority messages (#8834)

Extend overseer to send priority messages, the new functionality is used
for sending messages on the grandpa call path when we call
dispute-coordinator and approval-voting in
finality_target_with_longest_chain to make sure we don't block
unnecessarily.

Depends on: paritytech/orchestra#87.

---------

Signed-off-by: Alexandru Gheorghe <[email protected]>
Co-authored-by: cmd[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Javier Viola <[email protected]>
Co-authored-by: Javier Viola <[email protected]>

Author: 3 people

Cargo.lock

@@ -347,7 +347,7 @@ async fn run_main_loop<Context>(
 						// The message the approval voting subsystem would've handled.
 						ApprovalVotingParallelMessage::ApprovedAncestor(_, _,_) |
 						ApprovalVotingParallelMessage::GetApprovalSignaturesForCandidate(_, _)  => {
-							to_approval_voting_worker.send_message(
+							to_approval_voting_worker.send_message_with_priority::<overseer::HighPriority>(
 								msg.try_into().expect(
 									"Message is one of ApprovedAncestor, GetApprovalSignaturesForCandidate
 									 and that can be safely converted to ApprovalVotingMessage; qed"

polkadot/node/core/approval-voting-parallel/src/lib.rs

@@ -190,9 +190,20 @@ impl Handle {
 		self.send_and_log_error(Event::BlockImported(block)).await
 	}
 
-	/// Send some message to one of the `Subsystem`s.
+	/// Send some message with normal priority to one of the `Subsystem`s.
 	pub async fn send_msg(&mut self, msg: impl Into<AllMessages>, origin: &'static str) {
-		self.send_and_log_error(Event::MsgToSubsystem { msg: msg.into(), origin }).await
+		self.send_msg_with_priority(msg, origin, PriorityLevel::Normal).await
+	}
+
+	/// Send some message with the specified priority to one of the `Subsystem`s.
+	pub async fn send_msg_with_priority(
+		&mut self,
+		msg: impl Into<AllMessages>,
+		origin: &'static str,
+		priority: PriorityLevel,
+	) {
+		self.send_and_log_error(Event::MsgToSubsystem { msg: msg.into(), origin, priority })
+			.await
 	}
 
 	/// Send a message not providing an origin.
@@ -296,6 +307,8 @@ pub enum Event {
 		msg: AllMessages,
 		/// The originating subsystem name.
 		origin: &'static str,
+		/// The priority of the message.
+		priority: PriorityLevel,
 	},
 	/// A request from the outer world.
 	ExternalRequest(ExternalRequest),
@@ -764,8 +777,15 @@ where
 			select! {
 				msg = self.events_rx.select_next_some() => {
 					match msg {
-						Event::MsgToSubsystem { msg, origin } => {
-							self.route_message(msg.into(), origin).await?;
+						Event::MsgToSubsystem { msg, origin, priority } => {
+							match priority {
+								PriorityLevel::Normal => {
+									self.route_message(msg.into(), origin).await?;
+								},
+								PriorityLevel::High => {
+									self.route_message_with_priority::<HighPriority>(msg.into(), origin).await?;
+								},
+							}
 							self.metrics.on_message_relayed();
 						}
 						Event::Stop => {

polkadot/node/overseer/src/lib.rs

@@ -43,7 +43,7 @@ use polkadot_node_subsystem::messages::{
 	ChainSelectionMessage, DisputeCoordinatorMessage, HighestApprovedAncestorBlock,
 };
 use polkadot_node_subsystem_util::metrics::{self, prometheus};
-use polkadot_overseer::{AllMessages, Handle};
+use polkadot_overseer::{AllMessages, Handle, PriorityLevel};
 use polkadot_primitives::{Block as PolkadotBlock, BlockNumber, Hash, Header as PolkadotHeader};
 use sp_consensus::{Error as ConsensusError, SelectChain};
 use std::sync::Arc;
@@ -238,7 +238,7 @@ pub struct SelectRelayChainInner<B, OH> {
 impl<B, OH> SelectRelayChainInner<B, OH>
 where
 	B: HeaderProviderProvider<PolkadotBlock>,
-	OH: OverseerHandleT,
+	OH: OverseerHandleT + OverseerHandleWithPriorityT,
 {
 	/// Create a new [`SelectRelayChainInner`] wrapping the given chain backend
 	/// and a handle to the overseer.
@@ -286,7 +286,7 @@ where
 impl<B, OH> Clone for SelectRelayChainInner<B, OH>
 where
 	B: HeaderProviderProvider<PolkadotBlock> + Send + Sync,
-	OH: OverseerHandleT,
+	OH: OverseerHandleT + OverseerHandleWithPriorityT,
 {
 	fn clone(&self) -> Self {
 		SelectRelayChainInner {
@@ -325,17 +325,40 @@ pub trait OverseerHandleT: Clone + Send + Sync {
 	async fn send_msg<M: Send + Into<AllMessages>>(&mut self, msg: M, origin: &'static str);
 }
 
+/// Trait for the overseer handle that allows sending messages with the specified priority level.
+#[async_trait::async_trait]
+pub trait OverseerHandleWithPriorityT: Clone + Send + Sync {
+	async fn send_msg_with_priority<M: Send + Into<AllMessages>>(
+		&mut self,
+		msg: M,
+		origin: &'static str,
+		priority: PriorityLevel,
+	);
+}
+
 #[async_trait::async_trait]
 impl OverseerHandleT for Handle {
 	async fn send_msg<M: Send + Into<AllMessages>>(&mut self, msg: M, origin: &'static str) {
 		Handle::send_msg(self, msg, origin).await
 	}
 }
 
+#[async_trait::async_trait]
+impl OverseerHandleWithPriorityT for Handle {
+	async fn send_msg_with_priority<M: Send + Into<AllMessages>>(
+		&mut self,
+		msg: M,
+		origin: &'static str,
+		priority: PriorityLevel,
+	) {
+		Handle::send_msg_with_priority(self, msg, origin, priority).await
+	}
+}
+
 impl<B, OH> SelectRelayChainInner<B, OH>
 where
 	B: HeaderProviderProvider<PolkadotBlock>,
-	OH: OverseerHandleT + 'static,
+	OH: OverseerHandleT + OverseerHandleWithPriorityT + 'static,
 {
 	/// Get all leaves of the chain, i.e. block hashes that are suitable to
 	/// build upon and have no suitable children.
@@ -472,9 +495,10 @@ where
 					.await;
 			} else {
 				overseer
-					.send_msg(
+					.send_msg_with_priority(
 						ApprovalVotingMessage::ApprovedAncestor(subchain_head, target_number, tx),
 						std::any::type_name::<Self>(),
+						PriorityLevel::High,
 					)
 					.await;
 			}
@@ -503,16 +527,18 @@ where
 			let lag_update_task = async move {
 				if approval_voting_parallel_enabled {
 					overseer_handle
-						.send_msg(
+						.send_msg_with_priority(
 							ApprovalVotingParallelMessage::ApprovalCheckingLagUpdate(lag),
 							std::any::type_name::<Self>(),
+							PriorityLevel::High,
 						)
 						.await;
 				} else {
 					overseer_handle
-						.send_msg(
+						.send_msg_with_priority(
 							ApprovalDistributionMessage::ApprovalCheckingLagUpdate(lag),
 							std::any::type_name::<Self>(),
+							PriorityLevel::High,
 						)
 						.await;
 				}
@@ -542,13 +568,14 @@ where
 			// 3. Constrain according to disputes:
 			let (tx, rx) = oneshot::channel();
 			overseer
-				.send_msg(
+				.send_msg_with_priority(
 					DisputeCoordinatorMessage::DetermineUndisputedChain {
 						base: (target_number, target_hash),
 						block_descriptions: subchain_block_descriptions,
 						tx,
 					},
 					std::any::type_name::<Self>(),
+					PriorityLevel::High,
 				)
 				.await;
 

polkadot/node/service/src/relay_chain_selection.rs

@@ -20,6 +20,7 @@ use futures::channel::oneshot::Receiver;
 use polkadot_node_primitives::approval::v2::VrfSignature;
 use polkadot_node_subsystem::messages::{AllMessages, BlockDescription};
 use polkadot_node_subsystem_util::TimeoutExt;
+use polkadot_overseer::{HighPriority, PriorityLevel};
 use polkadot_test_client::Sr25519Keyring;
 use sp_consensus_babe::{
 	digests::{CompatibleDigestItem, PreDigest, SecondaryVRFPreDigest},
@@ -55,6 +56,26 @@ impl OverseerHandleT for TestSubsystemSender {
 	}
 }
 
+#[async_trait::async_trait]
+impl OverseerHandleWithPriorityT for TestSubsystemSender {
+	async fn send_msg_with_priority<M: Send + Into<AllMessages>>(
+		&mut self,
+		msg: M,
+		_origin: &'static str,
+		priority: PriorityLevel,
+	) {
+		match priority {
+			PriorityLevel::High => {
+				TestSubsystemSender::send_message_with_priority::<HighPriority>(self, msg.into())
+					.await;
+			},
+			PriorityLevel::Normal => {
+				TestSubsystemSender::send_message(self, msg.into()).await;
+			},
+		}
+	}
+}
+
 struct TestHarness {
 	virtual_overseer: VirtualOverseer,
 	case_vars: CaseVars,

polkadot/node/service/src/tests.rs

@@ -0,0 +1,15 @@
+title: extend overseer to send priority messages
+doc:
+- audience: Node Dev
+  description: |-
+    Extend overseer to send priority messages, the new functionality is used for sending messages
+    on the grandpa call path when we call dispute-coordinator and approval-voting in
+    finality_target_with_longest_chain to make sure we don't block unnecessarily.
+
+crates:
+- name: polkadot-node-core-approval-voting-parallel
+  bump: patch
+- name: polkadot-overseer
+  bump: patch
+- name: polkadot-service
+  bump: patch