Avoid blocking in the PartitionProcessorManager when being in the control loop

Summary:
- Improve tracing of ControlProcessors commands
- Run the spawn of partition processors concurrently
- Avoid blocking partition processor manager main loop

While testing I noticed that InvokerService creations takes
around 25ms (on my machine) to be created with `InvokerService::from_options`

I traced the delay to a single line of code here
https://github.com/restatedev/restate/blob/1cb01203f18bfc6ddb5fe4c45f6697799a79c27e/crates/service-client/src/http.rs#L73

My first thought was to cache the `HttpsConnectorBuilder` but this won't work because `HttpsConnectorBuilder` is not
clonable.

Fixes #2146

Author: muhamadazmy

crates/admin/src/cluster_controller/cluster_state_refresher.rs

@@ -170,7 +170,7 @@ impl<T: TransportConnect> ClusterStateRefresher<T> {
                             NodeState::Alive(AliveNode {
                                 last_heartbeat_at: MillisSinceEpoch::now(),
                                 generational_node_id: peer,
-                                partitions: msg.paritions_processor_state.unwrap_or_default(),
+                                partitions: msg.partition_processor_state.unwrap_or_default(),
                             }),
                         );
                     }

crates/admin/src/cluster_controller/service.rs

@@ -746,7 +746,7 @@ mod tests {
 
             let state = [(PartitionId::MIN, partition_processor_status)].into();
             let response = msg.to_rpc_response(NodeStateResponse {
-                paritions_processor_state: Some(state),
+                partition_processor_state: Some(state),
             });
 
             // We are not really sending something back to target, we just need to provide a known

crates/core/src/metadata/mod.rs

@@ -18,6 +18,7 @@ use std::sync::{Arc, OnceLock};
 use arc_swap::{ArcSwap, AsRaw};
 use enum_map::EnumMap;
 use tokio::sync::{mpsc, oneshot, watch};
+use tracing::instrument;
 
 use restate_types::live::{Live, Pinned};
 use restate_types::logs::metadata::Logs;
@@ -175,6 +176,7 @@ impl Metadata {
     }
 
     /// Returns when the metadata kind is at the provided version (or newer)
+    #[instrument(level = "debug", skip(self))]
     pub async fn wait_for_version(
         &self,
         metadata_kind: MetadataKind,

crates/core/src/network/partition_processor_rpc_client.rs

@@ -46,6 +46,8 @@ pub enum PartitionProcessorRpcClientError {
     Busy,
     #[error("internal error: {0}")]
     Internal(String),
+    #[error("partition processor starting")]
+    Starting,
 }
 
 impl PartitionProcessorRpcClientError {
@@ -60,7 +62,8 @@ impl PartitionProcessorRpcClientError {
             )
             | PartitionProcessorRpcClientError::UnknownPartition(_)
             | PartitionProcessorRpcClientError::UnknownNode(_)
-            | PartitionProcessorRpcClientError::NotLeader(_) => {
+            | PartitionProcessorRpcClientError::NotLeader(_)
+            | PartitionProcessorRpcClientError::Starting => {
                 // These are pre-flight error that we can distinguish,
                 // and for which we know for certain that no message was proposed yet to the log.
                 true
@@ -83,6 +86,7 @@ impl From<PartitionProcessorRpcError> for PartitionProcessorRpcClientError {
             PartitionProcessorRpcError::Internal(msg) => {
                 PartitionProcessorRpcClientError::Internal(msg)
             }
+            PartitionProcessorRpcError::Starting => PartitionProcessorRpcClientError::Starting,
         }
     }
 }

crates/node/src/roles/base.rs

@@ -73,7 +73,7 @@ impl BaseRole {
         &self,
         msg: Incoming<GetNodeState>,
     ) -> Result<(), ShutdownError> {
-        let parition_state = if let Some(ref handle) = self.processor_manager_handle {
+        let partition_state = if let Some(ref handle) = self.processor_manager_handle {
             Some(handle.get_state().await?)
         } else {
             None
@@ -82,7 +82,7 @@ impl BaseRole {
         // only return error if Shutdown
         if let Err(NetworkError::Shutdown(err)) = msg
             .to_rpc_response(NodeStateResponse {
-                paritions_processor_state: parition_state,
+                partition_processor_state: partition_state,
             })
             .try_send()
             .map_err(|err| err.source)

crates/types/src/cluster/cluster_state.rs

@@ -97,7 +97,9 @@ pub struct DeadNode {
     pub last_seen_alive: Option<MillisSinceEpoch>,
 }
 
-#[derive(Debug, Clone, Copy, Serialize, Deserialize, Eq, PartialEq, IntoProto)]
+#[derive(
+    Debug, Clone, Copy, Serialize, Deserialize, Eq, PartialEq, IntoProto, derive_more::Display,
+)]
 #[proto(target = "crate::protobuf::cluster::RunMode")]
 pub enum RunMode {
     Leader,

crates/types/src/net/node.rs

@@ -31,5 +31,5 @@ pub struct GetNodeState {}
 pub struct NodeStateResponse {
     /// State of paritions processor per parition. Is set to None if this node is not a `Worker` node
     #[serde_as(as = "Option<serde_with::Seq<(_, _)>>")]
-    pub paritions_processor_state: Option<BTreeMap<PartitionId, PartitionProcessorStatus>>,
+    pub partition_processor_state: Option<BTreeMap<PartitionId, PartitionProcessorStatus>>,
 }

crates/types/src/net/partition_processor.rs

@@ -79,6 +79,8 @@ pub enum PartitionProcessorRpcError {
     Busy,
     #[error("internal error: {0}")]
     Internal(String),
+    #[error("partition processor starting")]
+    Starting,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

crates/types/src/net/partition_processor_manager.rs

@@ -34,7 +34,7 @@ pub struct ControlProcessor {
     pub command: ProcessorCommand,
 }
 
-#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
+#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize, derive_more::Display)]
 pub enum ProcessorCommand {
     Stop,
     Follower,