common_meta/rpc/
procedure.rs1use std::time::Duration;
16
17pub use api::v1::meta::{MigrateRegionResponse, ProcedureStateResponse};
18use api::v1::meta::{
19 ProcedureDetailResponse as PbProcedureDetailResponse, ProcedureId as PbProcedureId,
20 ProcedureMeta as PbProcedureMeta, ProcedureStateResponse as PbProcedureStateResponse,
21 ProcedureStatus as PbProcedureStatus,
22};
23use common_error::ext::ErrorExt;
24use common_procedure::{ProcedureId, ProcedureInfo, ProcedureState};
25use snafu::ResultExt;
26use table::metadata::TableId;
27
28use crate::error::{ParseProcedureIdSnafu, Result};
29
30#[derive(Clone)]
32pub struct MigrateRegionRequest {
33 pub region_id: u64,
34 pub from_peer: u64,
35 pub to_peer: u64,
36 pub timeout: Duration,
37}
38
39#[derive(Debug, Clone)]
41pub struct AddRegionFollowerRequest {
42 pub region_id: u64,
44 pub peer_id: u64,
46}
47
48#[derive(Debug, Clone)]
49pub struct AddTableFollowerRequest {
50 pub catalog_name: String,
51 pub schema_name: String,
52 pub table_name: String,
53 pub table_id: TableId,
54}
55
56#[derive(Debug, Clone)]
57pub struct RemoveTableFollowerRequest {
58 pub catalog_name: String,
59 pub schema_name: String,
60 pub table_name: String,
61 pub table_id: TableId,
62}
63
64#[derive(Debug, Clone)]
65pub enum ManageRegionFollowerRequest {
66 AddRegionFollower(AddRegionFollowerRequest),
67 RemoveRegionFollower(RemoveRegionFollowerRequest),
68 AddTableFollower(AddTableFollowerRequest),
69 RemoveTableFollower(RemoveTableFollowerRequest),
70}
71
72#[derive(Debug, Clone)]
74pub struct RemoveRegionFollowerRequest {
75 pub region_id: u64,
77 pub peer_id: u64,
79}
80
81pub fn pb_pid_to_pid(pid: &PbProcedureId) -> Result<ProcedureId> {
83 ProcedureId::parse_str(&String::from_utf8_lossy(&pid.key)).with_context(|_| {
84 ParseProcedureIdSnafu {
85 key: hex::encode(&pid.key),
86 }
87 })
88}
89
90pub fn pid_to_pb_pid(pid: ProcedureId) -> PbProcedureId {
92 PbProcedureId {
93 key: pid.to_string().into(),
94 }
95}
96
97pub fn procedure_state_to_pb_state(state: &ProcedureState) -> (PbProcedureStatus, String) {
99 match state {
100 ProcedureState::Running => (PbProcedureStatus::Running, String::default()),
101 ProcedureState::Done { .. } => (PbProcedureStatus::Done, String::default()),
102 ProcedureState::Retrying { error } => (PbProcedureStatus::Retrying, error.output_msg()),
103 ProcedureState::Failed { error } => (PbProcedureStatus::Failed, error.output_msg()),
104 ProcedureState::PrepareRollback { error } => {
105 (PbProcedureStatus::PrepareRollback, error.output_msg())
106 }
107 ProcedureState::RollingBack { error } => {
108 (PbProcedureStatus::RollingBack, error.output_msg())
109 }
110 ProcedureState::Poisoned { error, .. } => (PbProcedureStatus::Poisoned, error.output_msg()),
111 }
112}
113
114pub fn procedure_state_to_pb_response(state: &ProcedureState) -> PbProcedureStateResponse {
116 let (status, error) = procedure_state_to_pb_state(state);
117 PbProcedureStateResponse {
118 status: status.into(),
119 error,
120 ..Default::default()
121 }
122}
123
124pub fn procedure_details_to_pb_response(metas: Vec<ProcedureInfo>) -> PbProcedureDetailResponse {
125 let procedures = metas
126 .into_iter()
127 .map(|meta| {
128 let (status, error) = procedure_state_to_pb_state(&meta.state);
129 PbProcedureMeta {
130 id: Some(pid_to_pb_pid(meta.id)),
131 type_name: meta.type_name.to_string(),
132 status: status.into(),
133 start_time_ms: meta.start_time_ms,
134 end_time_ms: meta.end_time_ms,
135 lock_keys: meta.lock_keys,
136 error,
137 }
138 })
139 .collect();
140 PbProcedureDetailResponse {
141 procedures,
142 ..Default::default()
143 }
144}
145
146#[cfg(test)]
147mod tests {
148 use std::sync::Arc;
149
150 use common_procedure::Error;
151 use snafu::Location;
152
153 use super::*;
154
155 #[test]
156 fn test_pid_pb_pid_conversion() {
157 let pid = ProcedureId::random();
158
159 let pb_pid = pid_to_pb_pid(pid);
160
161 assert_eq!(pid, pb_pid_to_pid(&pb_pid).unwrap());
162 }
163
164 #[test]
165 fn test_procedure_state_to_pb_response() {
166 let state = ProcedureState::Running;
167 let resp = procedure_state_to_pb_response(&state);
168 assert_eq!(PbProcedureStatus::Running as i32, resp.status);
169 assert!(resp.error.is_empty());
170
171 let state = ProcedureState::Done { output: None };
172 let resp = procedure_state_to_pb_response(&state);
173 assert_eq!(PbProcedureStatus::Done as i32, resp.status);
174 assert!(resp.error.is_empty());
175
176 let state = ProcedureState::Retrying {
177 error: Arc::new(Error::ManagerNotStart {
178 location: Location::default(),
179 }),
180 };
181 let resp = procedure_state_to_pb_response(&state);
182 assert_eq!(PbProcedureStatus::Retrying as i32, resp.status);
183 assert_eq!("Procedure Manager is stopped", resp.error);
184
185 let state = ProcedureState::Failed {
186 error: Arc::new(Error::ManagerNotStart {
187 location: Location::default(),
188 }),
189 };
190 let resp = procedure_state_to_pb_response(&state);
191 assert_eq!(PbProcedureStatus::Failed as i32, resp.status);
192 assert_eq!("Procedure Manager is stopped", resp.error);
193 }
194}