Introduce ControlProcessors command

This commit introduces the ControlProcessors command which allows an
external entity to instruct the PartitionProcessorManager to start
and stop PartitionProcessors.

This fixes restatedev#1695.

crates/types/protobuf/restate/common.proto

@@ -42,6 +42,7 @@ enum TargetName {
   ATTACH_RESPONSE = 6;
   GET_PROCESSORS_STATE_REQUEST = 7;
   PROCESSORS_STATE_RESPONSE = 8;
+  CONTROL_PROCESSORS = 9;
 }
 
 enum NodeStatus {

crates/types/src/net/partition_processor_manager.rs

@@ -15,9 +15,10 @@ use serde_with::serde_as;
 
 use crate::cluster::cluster_state::PartitionProcessorStatus;
 use crate::identifiers::PartitionId;
-use crate::net::{RequestId, TargetName};
+use crate::net::{define_message, RequestId, TargetName};
 
 use crate::net::define_rpc;
+use crate::Version;
 
 define_rpc! {
     @request = GetProcessorsState,
@@ -38,3 +39,27 @@ pub struct ProcessorsStateResponse {
     #[serde_as(as = "serde_with::Seq<(_, _)>")]
     pub state: BTreeMap<PartitionId, PartitionProcessorStatus>,
 }
+
+define_message! {
+    @message = ControlProcessors,
+    @target = TargetName::ControlProcessors,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ControlProcessors {
+    pub min_partition_table_version: Version,
+    pub commands: Vec<ControlProcessor>,
+}
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
+pub struct ControlProcessor {
+    pub partition_id: PartitionId,
+    pub command: ProcessorCommand,
+}
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
+pub enum ProcessorCommand {
+    Stop,
+    Follower,
+    Leader,
+}