feat(procedure): Max retry time (#1095)

* feat: procedure config

* fix: modify config

* feat: add retry logic

* feat: add error

* feat: add it

* feat: add it

* feat: add it

* feat: rm retry from runner

* feat: use backon

* feat: add retry_interval

* feat: add retry_interval

* fix: conflict

* fix: cr

* feat: add retry error and id

* feat: rename

* refactor: execute

* feat: use config dir

* fix: cr

* fix: cr

* fix: fmt

* fix: fmt

* fix: pr

* fix: it

* fix: rm unless cmd params

* feat: add toml

* fix: ut

* feat: add rolling back

* fix: cr

* fix: cr

* fix: cr

* fix: ci

* fix: ci

* fix: ci

* chore: Apply suggestions from code review

---------

Co-authored-by: Yingwen <realevenyag@gmail.com>

src/datanode/src/datanode.rs

@@ -149,11 +149,22 @@ impl From<&DatanodeOptions> for StorageEngineConfig {
 pub struct ProcedureConfig {
     /// Storage config for procedure manager.
     pub store: ObjectStoreConfig,
+    /// Max retry times of procedure.
+    pub max_retry_times: usize,
+    /// Initial retry delay of procedures, increases exponentially.
+    #[serde(with = "humantime_serde")]
+    pub retry_delay: Duration,
 }
 
 impl Default for ProcedureConfig {
     fn default() -> ProcedureConfig {
-        ProcedureConfig::from_file_path("/tmp/greptimedb/procedure/".to_string())
+        ProcedureConfig {
+            store: ObjectStoreConfig::File(FileConfig {
+                data_dir: "/tmp/greptimedb/procedure/".to_string(),
+            }),
+            max_retry_times: 3,
+            retry_delay: Duration::from_millis(500),
+        }
     }
 }
 
@@ -161,6 +172,7 @@ impl ProcedureConfig {
     pub fn from_file_path(path: String) -> ProcedureConfig {
         ProcedureConfig {
             store: ObjectStoreConfig::File(FileConfig { data_dir: path }),
+            ..Default::default()
         }
     }
 }

src/datanode/src/instance.rs

@@ -460,7 +460,11 @@ pub(crate) async fn create_procedure_manager(
     );
 
     let object_store = new_object_store(&procedure_config.store).await?;
-    let manager_config = ManagerConfig { object_store };
+    let manager_config = ManagerConfig {
+        object_store,
+        max_retry_times: procedure_config.max_retry_times,
+        retry_delay: procedure_config.retry_delay,
+    };
 
     Ok(Some(Arc::new(LocalManager::new(manager_config))))
 }

src/datanode/src/tests/test_util.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::sync::Arc;
+use std::time::Duration;
 
 use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, MIN_USER_TABLE_ID};
 use common_query::Output;
@@ -64,6 +65,8 @@ impl MockInstance {
             store: ObjectStoreConfig::File(FileConfig {
                 data_dir: procedure_dir.path().to_str().unwrap().to_string(),
             }),
+            max_retry_times: 3,
+            retry_delay: Duration::from_millis(500),
         });
 
         let instance = Instance::with_mock_meta_client(&opts).await.unwrap();