feat: introduce poison mechanism for procedure

Cargo.toml

@@ -130,7 +130,7 @@ etcd-client = "0.14"
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "fb8e20ce29afd81835e3ea3c1164c8ce10de2c65" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "36dc57baccb1230091b0e60295911b25201b2c15" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"

src/common/meta/src/ddl/test_util.rs

@@ -80,7 +80,13 @@ pub async fn create_logical_table(
     let tasks = vec![test_create_logical_table_task(table_name)];
     let mut procedure = CreateLogicalTablesProcedure::new(tasks, physical_table_id, ddl_context);
     let status = procedure.on_prepare().await.unwrap();
-    assert_matches!(status, Status::Executing { persist: true });
+    assert_matches!(
+        status,
+        Status::Executing {
+            persist: true,
+            clean_poisons: false
+        }
+    );
     let status = procedure.on_create_metadata().await.unwrap();
     assert_matches!(status, Status::Done { .. });
 

src/common/meta/src/ddl/tests/alter_logical_tables.rs

@@ -180,7 +180,13 @@ async fn test_on_prepare() {
 
     let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, ddl_context);
     let result = procedure.on_prepare().await;
-    assert_matches!(result, Ok(Status::Executing { persist: true }));
+    assert_matches!(
+        result,
+        Ok(Status::Executing {
+            persist: true,
+            clean_poisons: false
+        })
+    );
 }
 
 #[tokio::test]
@@ -205,18 +211,36 @@ async fn test_on_update_metadata() {
 
     let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, ddl_context);
     let mut status = procedure.on_prepare().await.unwrap();
-    assert_matches!(status, Status::Executing { persist: true });
+    assert_matches!(
+        status,
+        Status::Executing {
+            persist: true,
+            clean_poisons: false
+        }
+    );
 
     let ctx = common_procedure::Context {
         procedure_id: ProcedureId::random(),
         provider: Arc::new(MockContextProvider::default()),
     };
     // on_submit_alter_region_requests
     status = procedure.execute(&ctx).await.unwrap();
-    assert_matches!(status, Status::Executing { persist: true });
+    assert_matches!(
+        status,
+        Status::Executing {
+            persist: true,
+            clean_poisons: false
+        }
+    );
     // on_update_metadata
     status = procedure.execute(&ctx).await.unwrap();
-    assert_matches!(status, Status::Executing { persist: true });
+    assert_matches!(
+        status,
+        Status::Executing {
+            persist: true,
+            clean_poisons: false
+        }
+    );
 }
 
 #[tokio::test]
@@ -237,18 +261,36 @@ async fn test_on_part_duplicate_alter_request() {
 
     let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, ddl_context.clone());
     let mut status = procedure.on_prepare().await.unwrap();
-    assert_matches!(status, Status::Executing { persist: true });
+    assert_matches!(
+        status,
+        Status::Executing {
+            persist: true,
+            clean_poisons: false
+        }
+    );
 
     let ctx = common_procedure::Context {
         procedure_id: ProcedureId::random(),
         provider: Arc::new(MockContextProvider::default()),
     };
     // on_submit_alter_region_requests
     status = procedure.execute(&ctx).await.unwrap();
-    assert_matches!(status, Status::Executing { persist: true });
+    assert_matches!(
+        status,
+        Status::Executing {
+            persist: true,
+            clean_poisons: false
+        }
+    );
     // on_update_metadata
     status = procedure.execute(&ctx).await.unwrap();
-    assert_matches!(status, Status::Executing { persist: true });
+    assert_matches!(
+        status,
+        Status::Executing {
+            persist: true,
+            clean_poisons: false
+        }
+    );
 
     // re-alter
     let tasks = vec![
@@ -270,18 +312,36 @@ async fn test_on_part_duplicate_alter_request() {
 
     let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, ddl_context.clone());
     let mut status = procedure.on_prepare().await.unwrap();
-    assert_matches!(status, Status::Executing { persist: true });
+    assert_matches!(
+        status,
+        Status::Executing {
+            persist: true,
+            clean_poisons: false
+        }
+    );
 
     let ctx = common_procedure::Context {
         procedure_id: ProcedureId::random(),
         provider: Arc::new(MockContextProvider::default()),
     };
     // on_submit_alter_region_requests
     status = procedure.execute(&ctx).await.unwrap();
-    assert_matches!(status, Status::Executing { persist: true });
+    assert_matches!(
+        status,
+        Status::Executing {
+            persist: true,
+            clean_poisons: false
+        }
+    );
     // on_update_metadata
     status = procedure.execute(&ctx).await.unwrap();
-    assert_matches!(status, Status::Executing { persist: true });
+    assert_matches!(
+        status,
+        Status::Executing {
+            persist: true,
+            clean_poisons: false
+        }
+    );
 
     let table_name_keys = vec![
         TableNameKey::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "table1"),

src/common/meta/src/ddl/tests/create_logical_tables.rs

@@ -69,7 +69,13 @@ async fn test_on_prepare() {
     let physical_table_id = table_id;
     let mut procedure = CreateLogicalTablesProcedure::new(tasks, physical_table_id, ddl_context);
     let status = procedure.on_prepare().await.unwrap();
-    assert_matches!(status, Status::Executing { persist: true });
+    assert_matches!(
+        status,
+        Status::Executing {
+            persist: true,
+            clean_poisons: false
+        }
+    );
 }
 
 #[tokio::test]
@@ -202,7 +208,13 @@ async fn test_on_prepare_part_logical_tables_exist() {
         ddl_context,
     );
     let status = procedure.on_prepare().await.unwrap();
-    assert_matches!(status, Status::Executing { persist: true });
+    assert_matches!(
+        status,
+        Status::Executing {
+            persist: true,
+            clean_poisons: false
+        }
+    );
 }
 
 #[tokio::test]
@@ -238,7 +250,13 @@ async fn test_on_create_metadata() {
         ddl_context,
     );
     let status = procedure.on_prepare().await.unwrap();
-    assert_matches!(status, Status::Executing { persist: true });
+    assert_matches!(
+        status,
+        Status::Executing {
+            persist: true,
+            clean_poisons: false
+        }
+    );
     let ctx = ProcedureContext {
         procedure_id: ProcedureId::random(),
         provider: Arc::new(MockContextProvider::default()),
@@ -294,7 +312,13 @@ async fn test_on_create_metadata_part_logical_tables_exist() {
         ddl_context,
     );
     let status = procedure.on_prepare().await.unwrap();
-    assert_matches!(status, Status::Executing { persist: true });
+    assert_matches!(
+        status,
+        Status::Executing {
+            persist: true,
+            clean_poisons: false
+        }
+    );
     let ctx = ProcedureContext {
         procedure_id: ProcedureId::random(),
         provider: Arc::new(MockContextProvider::default()),
@@ -339,7 +363,13 @@ async fn test_on_create_metadata_err() {
         ddl_context.clone(),
     );
     let status = procedure.on_prepare().await.unwrap();
-    assert_matches!(status, Status::Executing { persist: true });
+    assert_matches!(
+        status,
+        Status::Executing {
+            persist: true,
+            clean_poisons: false
+        }
+    );
     let ctx = ProcedureContext {
         procedure_id: ProcedureId::random(),
         provider: Arc::new(MockContextProvider::default()),

src/common/meta/src/ddl/tests/create_table.rs

@@ -137,7 +137,13 @@ async fn test_on_prepare_without_create_if_table_exists() {
     task.create_table.create_if_not_exists = true;
     let mut procedure = CreateTableProcedure::new(task, ddl_context);
     let status = procedure.on_prepare().await.unwrap();
-    assert_matches!(status, Status::Executing { persist: true });
+    assert_matches!(
+        status,
+        Status::Executing {
+            persist: true,
+            clean_poisons: false
+        }
+    );
     assert_eq!(procedure.table_id(), 1024);
 }
 

src/common/meta/src/ddl/tests/create_view.rs

@@ -153,7 +153,13 @@ async fn test_on_prepare_without_create_if_table_exists() {
     task.create_view.create_if_not_exists = true;
     let mut procedure = CreateViewProcedure::new(task, ddl_context);
     let status = procedure.on_prepare().await.unwrap();
-    assert_matches!(status, Status::Executing { persist: true });
+    assert_matches!(
+        status,
+        Status::Executing {
+            persist: true,
+            clean_poisons: false
+        }
+    );
     assert_eq!(procedure.view_id(), 1024);
 }
 

src/common/meta/src/ddl_manager.rs

@@ -834,6 +834,7 @@ mod tests {
     use std::sync::Arc;
 
     use common_procedure::local::LocalManager;
+    use common_procedure::test_util::InMemoryPoisonStore;
 
     use super::DdlManager;
     use crate::cache_invalidator::DummyCacheInvalidator;
@@ -883,7 +884,12 @@ mod tests {
         ));
 
         let state_store = Arc::new(KvStateStore::new(kv_backend.clone()));
-        let procedure_manager = Arc::new(LocalManager::new(Default::default(), state_store));
+        let poison_manager = Arc::new(InMemoryPoisonStore::default());
+        let procedure_manager = Arc::new(LocalManager::new(
+            Default::default(),
+            state_store,
+            poison_manager,
+        ));
 
         let _ = DdlManager::try_new(
             DdlContext {

src/common/meta/src/error.rs

@@ -748,6 +748,18 @@ pub enum Error {
         #[snafu(source)]
         error: serde_json::Error,
     },
+
+    #[snafu(display(
+        "Procedure poison key already exists with a different value, key: {}, value: {}",
+        key,
+        value
+    ))]
+    ProcedurePoisonConflict {
+        key: String,
+        value: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -766,7 +778,7 @@ impl ErrorExt for Error {
             | SerializeToJson { .. }
             | DeserializeFromJson { .. } => StatusCode::Internal,
 
-            ValueNotExist { .. } => StatusCode::Unexpected,
+            ValueNotExist { .. } | ProcedurePoisonConflict { .. } => StatusCode::Unexpected,
 
             Unsupported { .. } => StatusCode::Unsupported,
 

src/common/meta/src/key.rs

@@ -156,6 +156,7 @@ use crate::kv_backend::txn::{Txn, TxnOp};
 use crate::kv_backend::KvBackendRef;
 use crate::rpc::router::{region_distribution, LeaderState, RegionRoute};
 use crate::rpc::store::BatchDeleteRequest;
+use crate::state_store::PoisonValue;
 use crate::DatanodeId;
 
 pub const NAME_PATTERN: &str = r"[a-zA-Z_:-][a-zA-Z0-9_:\-\.@#]*";
@@ -1324,7 +1325,8 @@ impl_metadata_value! {
     TableFlowValue,
     NodeAddressValue,
     SchemaNameValue,
-    FlowStateValue
+    FlowStateValue,
+    PoisonValue
 }
 
 impl_optional_metadata_value! {

src/common/meta/src/rpc/procedure.rs

@@ -81,6 +81,7 @@ pub fn procedure_state_to_pb_state(state: &ProcedureState) -> (PbProcedureStatus
         ProcedureState::RollingBack { error } => {
             (PbProcedureStatus::RollingBack, error.to_string())
         }
+        ProcedureState::Poisoned { error, .. } => (PbProcedureStatus::Poisoned, error.to_string()),
     }
 }