fix: respect gc mailbox timeout for admin gc (#8363)

Signed-off-by: discord9 <discord9@163.com>

src/common/function/src/admin/gc.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::time::Duration;
-
 use common_error::ext::BoxedError;
 use common_macro::admin_fn;
 use common_meta::rpc::procedure::{GcRegionsRequest, GcTableRequest};
@@ -30,7 +28,6 @@ use snafu::{ResultExt, ensure};
 use crate::handlers::ProcedureServiceHandlerRef;
 use crate::helper::cast_u64;
 
-const DEFAULT_GC_TIMEOUT: Duration = Duration::from_secs(60);
 const DEFAULT_FULL_FILE_LISTING: bool = false;
 
 #[admin_fn(
@@ -50,7 +47,7 @@ pub(crate) async fn gc_regions(
         .gc_regions(GcRegionsRequest {
             region_ids,
             full_file_listing,
-            timeout: DEFAULT_GC_TIMEOUT,
+            timeout: None,
         })
         .await?;
 
@@ -77,7 +74,7 @@ pub(crate) async fn gc_table(
             schema_name,
             table_name,
             full_file_listing,
-            timeout: DEFAULT_GC_TIMEOUT,
+            timeout: None,
         })
         .await?;
 
@@ -180,9 +177,18 @@ fn gc_table_signature() -> Signature {
 
 #[cfg(test)]
 mod tests {
+    use std::sync::{Arc, Mutex};
+
+    use api::v1::meta::ReconcileRequest;
+    use async_trait::async_trait;
+    use catalog::CatalogManagerRef;
+    use common_meta::rpc::procedure::{
+        GcResponse, ManageRegionFollowerRequest, MigrateRegionRequest, ProcedureStateResponse,
+    };
     use session::context::QueryContext;
 
     use super::*;
+    use crate::handlers::ProcedureServiceHandler;
 
     #[test]
     fn test_parse_gc_regions_params_with_full_file_listing() {
@@ -217,4 +223,80 @@ mod tests {
         assert_eq!(table, "t");
         assert!(full_file_listing);
     }
+
+    #[tokio::test]
+    async fn test_gc_regions_uses_meta_gc_timeout_config() {
+        let handler = Arc::new(MockProcedureServiceHandler::default());
+        let handler_ref: ProcedureServiceHandlerRef = handler.clone();
+        let params = vec![ValueRef::UInt64(1), ValueRef::Boolean(true)];
+
+        super::gc_regions(&handler_ref, &QueryContext::arc(), &params)
+            .await
+            .unwrap();
+
+        let request = handler.gc_regions_request.lock().unwrap().clone().unwrap();
+        assert_eq!(request.region_ids, vec![1]);
+        assert!(request.full_file_listing);
+        assert_eq!(request.timeout, None);
+    }
+
+    #[tokio::test]
+    async fn test_gc_table_uses_meta_gc_timeout_config() {
+        let handler = Arc::new(MockProcedureServiceHandler::default());
+        let handler_ref: ProcedureServiceHandlerRef = handler.clone();
+        let params = vec![ValueRef::String("public.t"), ValueRef::Boolean(true)];
+
+        super::gc_table(&handler_ref, &QueryContext::arc(), &params)
+            .await
+            .unwrap();
+
+        let request = handler.gc_table_request.lock().unwrap().clone().unwrap();
+        assert_eq!(request.catalog_name, "greptime");
+        assert_eq!(request.schema_name, "public");
+        assert_eq!(request.table_name, "t");
+        assert!(request.full_file_listing);
+        assert_eq!(request.timeout, None);
+    }
+
+    #[derive(Default)]
+    struct MockProcedureServiceHandler {
+        gc_regions_request: Mutex<Option<GcRegionsRequest>>,
+        gc_table_request: Mutex<Option<GcTableRequest>>,
+    }
+
+    #[async_trait]
+    impl ProcedureServiceHandler for MockProcedureServiceHandler {
+        async fn migrate_region(&self, _request: MigrateRegionRequest) -> Result<Option<String>> {
+            unreachable!()
+        }
+
+        async fn reconcile(&self, _request: ReconcileRequest) -> Result<Option<String>> {
+            unreachable!()
+        }
+
+        async fn query_procedure_state(&self, _pid: &str) -> Result<ProcedureStateResponse> {
+            unreachable!()
+        }
+
+        async fn manage_region_follower(
+            &self,
+            _request: ManageRegionFollowerRequest,
+        ) -> Result<()> {
+            unreachable!()
+        }
+
+        fn catalog_manager(&self) -> &CatalogManagerRef {
+            unreachable!()
+        }
+
+        async fn gc_regions(&self, request: GcRegionsRequest) -> Result<GcResponse> {
+            *self.gc_regions_request.lock().unwrap() = Some(request);
+            Ok(GcResponse::default())
+        }
+
+        async fn gc_table(&self, request: GcTableRequest) -> Result<GcResponse> {
+            *self.gc_table_request.lock().unwrap() = Some(request);
+            Ok(GcResponse::default())
+        }
+    }
 }

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

@@ -82,7 +82,7 @@ pub struct RemoveRegionFollowerRequest {
 pub struct GcRegionsRequest {
     pub region_ids: Vec<u64>,
     pub full_file_listing: bool,
-    pub timeout: Duration,
+    pub timeout: Option<Duration>,
 }
 
 #[derive(Debug, Clone)]
@@ -91,7 +91,7 @@ pub struct GcTableRequest {
     pub schema_name: String,
     pub table_name: String,
     pub full_file_listing: bool,
-    pub timeout: Duration,
+    pub timeout: Option<Duration>,
 }
 
 #[derive(Debug, Clone, Default, PartialEq, Eq)]

src/meta-client/src/client/procedure.rs

@@ -286,7 +286,7 @@ impl Inner {
             }),
             region_ids: request.region_ids,
             full_file_listing: request.full_file_listing,
-            timeout_secs: timeout.as_secs() as u32,
+            timeout_secs: gc_timeout_secs(timeout),
         };
 
         let resp: GcRegionsResponse = self
@@ -294,7 +294,9 @@ impl Inner {
                 "gc_regions",
                 move |mut client| {
                     let mut req = Request::new(req.clone());
-                    req.set_timeout(timeout);
+                    if let Some(timeout) = timeout {
+                        req.set_timeout(timeout);
+                    }
                     async move { client.gc_regions(req).await.map(|res| res.into_inner()) }
                 },
                 |resp: &GcRegionsResponse| &resp.header,
@@ -323,7 +325,7 @@ impl Inner {
             schema_name: request.schema_name,
             table_name: request.table_name,
             full_file_listing: request.full_file_listing,
-            timeout_secs: timeout.as_secs() as u32,
+            timeout_secs: gc_timeout_secs(timeout),
         };
 
         let resp: GcTableResponse = self
@@ -331,7 +333,9 @@ impl Inner {
                 "gc_table",
                 move |mut client| {
                     let mut req = Request::new(req.clone());
-                    req.set_timeout(timeout);
+                    if let Some(timeout) = timeout {
+                        req.set_timeout(timeout);
+                    }
                     async move { client.gc_table(req).await.map(|res| res.into_inner()) }
                 },
                 |resp: &GcTableResponse| &resp.header,
@@ -413,6 +417,12 @@ impl Inner {
     }
 }
 
+fn gc_timeout_secs(timeout: Option<Duration>) -> u32 {
+    timeout
+        .map(|timeout| timeout.as_secs().max(1).try_into().unwrap_or(u32::MAX))
+        .unwrap_or(0)
+}
+
 #[cfg(test)]
 mod tests {
     use std::time::{Duration, Instant};
@@ -438,8 +448,18 @@ mod tests {
     use tonic::codec::CompressionEncoding;
     use tonic::{Request, Response, Status};
 
+    use super::gc_timeout_secs;
     use crate::client::MetaClientBuilder;
 
+    #[test]
+    fn test_gc_timeout_secs() {
+        assert_eq!(gc_timeout_secs(None), 0);
+        assert_eq!(gc_timeout_secs(Some(Duration::from_millis(1))), 1);
+        assert_eq!(gc_timeout_secs(Some(Duration::from_millis(999))), 1);
+        assert_eq!(gc_timeout_secs(Some(Duration::from_secs(1))), 1);
+        assert_eq!(gc_timeout_secs(Some(Duration::from_secs(10))), 10);
+    }
+
     #[derive(Clone)]
     struct MockHeartbeat {
         leader_addr: String,

src/meta-srv/src/service/procedure.rs

@@ -268,7 +268,7 @@ impl procedure_service_server::ProcedureService for Metasrv {
             .handle_gc_regions(MetaGcRegionsRequest {
                 region_ids,
                 full_file_listing,
-                timeout: Duration::from_secs(timeout_secs as u64),
+                timeout: Self::normalize_gc_timeout(Duration::from_secs(timeout_secs as u64)),
             })
             .await?;
 
@@ -295,7 +295,7 @@ impl procedure_service_server::ProcedureService for Metasrv {
                 schema_name,
                 table_name,
                 full_file_listing,
-                timeout: Duration::from_secs(timeout_secs as u64),
+                timeout: Self::normalize_gc_timeout(Duration::from_secs(timeout_secs as u64)),
             })
             .await?;
 
@@ -356,9 +356,8 @@ impl Metasrv {
         &self,
         region_ids: Vec<RegionId>,
         full_file_listing: bool,
-        timeout: Duration,
+        timeout: Option<Duration>,
     ) -> error::Result<GcResponse> {
-        let timeout = Self::normalize_gc_timeout(timeout);
         let gc_ticker = self.gc_ticker().context(error::UnexpectedSnafu {
             violated: "GC ticker not available".to_string(),
         })?;