feat(frontend): retry transient distributed write failures (#9543)

* refactor(operator): route shared requests through NodeManager

Remove the concrete NodeClients requester API and use the existing node manager for owned and shared requests. Preserve the original write hint while the legacy error adapter remains in place.

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix(meta): preserve retry policy through error hints

Propagate client errors through External and let DDL read retry hints while preserving cleanup and terminal-child policies. Region transport failures retain the legacy Retryable classification, which now also participates in bounded write retries.

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix(operator): restore sparse partition tags when rebuilding writes

Decode retained primary keys when refreshed partition expressions need different tags. Keep the partitioner and multi-region sparse write regression tests with the fix.

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix(operator): stop detached write retries when callers cancel

Signed-off-by: WenyXu <wenymedia@gmail.com>

* feat(frontend): configure retries across distributed write paths

Wire the shared dispatcher through frontend and logical batching, preserving partition columns and request headers. Keep standalone retries disabled and include configuration, batching, and storage regression coverage.

Signed-off-by: WenyXu <wenymedia@gmail.com>

---------

Signed-off-by: WenyXu <wenymedia@gmail.com>

src/common/meta/src/ddl/drop_table.rs

@@ -19,6 +19,8 @@ use std::collections::HashMap;
 
 use async_trait::async_trait;
 use common_error::ext::BoxedError;
+#[cfg(feature = "enterprise")]
+use common_error::ext::ErrorExt;
 use common_event_recorder::Event;
 use common_procedure::error::{ExternalSnafu, FromJsonSnafu, ToJsonSnafu};
 use common_procedure::{
@@ -53,7 +55,7 @@ use crate::rpc::router::{RegionRoute, operating_leader_region_roles};
 
 #[cfg(feature = "enterprise")]
 fn ensure_retry_later(err: error::Error) -> error::Error {
-    if err.is_retry_later() {
+    if err.retry_hint().is_retryable() {
         err
     } else {
         error::Error::retry_later(err)

src/common/meta/src/ddl/undrop_table.rs

@@ -17,8 +17,8 @@ use api::v1::region::{
 use async_trait::async_trait;
 use common_procedure::error::{FromJsonSnafu, ToJsonSnafu};
 use common_procedure::{
-    Context as ProcedureContext, EventContext, EventTrigger, LockKey, Procedure,
-    Result as ProcedureResult, Status,
+    Context as ProcedureContext, Error as ProcedureError, EventContext, EventTrigger, LockKey,
+    Procedure, Result as ProcedureResult, Status,
 };
 use common_telemetry::tracing_context::TracingContext;
 use common_telemetry::warn;
@@ -156,6 +156,7 @@ impl UndropTableProcedure {
         }
         .await;
         if let Err(err) = restore_result {
+            // Keep this classification aligned with map_restore_error().
             let should_cleanup_opened_regions = !err.is_retry_later();
             let err = self.map_restore_metadata_error(err);
             if should_cleanup_opened_regions
@@ -274,6 +275,19 @@ impl UndropTableProcedure {
         Ok(())
     }
 
+    fn map_restore_error(err: error::Error) -> ProcedureError {
+        // Restore retries keep opened regions, while other failures close them. Preserve
+        // this policy even when the underlying metadata error carries a retry hint:
+        // retrying after cleanup would restore metadata without reopening the regions.
+        if err.is_retry_later() {
+            map_to_procedure_error(err)
+        } else if err.need_clean_poisons() {
+            ProcedureError::external_and_clean_poisons(err)
+        } else {
+            ProcedureError::external(err)
+        }
+    }
+
     fn map_restore_metadata_error(&self, err: error::Error) -> error::Error {
         match err {
             error::Error::TombstoneTargetAlreadyExists { .. } => {
@@ -413,7 +427,12 @@ impl Procedure for UndropTableProcedure {
     async fn execute(&mut self, _: &ProcedureContext) -> ProcedureResult<Status> {
         match self.data.state {
             UndropTableState::Prepare => self.on_prepare().await,
-            UndropTableState::RestoreMetadata => self.on_restore_metadata().await,
+            UndropTableState::RestoreMetadata => {
+                return self
+                    .on_restore_metadata()
+                    .await
+                    .map_err(Self::map_restore_error);
+            }
             UndropTableState::OpenRegions => self.on_open_regions().await,
             UndropTableState::InvalidateTableCache => self.on_broadcast().await,
         }
@@ -606,7 +625,8 @@ mod tests {
     use api::v1::region::region_request;
     use async_trait::async_trait;
     use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
-    use common_error::ext::ErrorExt;
+    use common_error::ext::{BoxedError, ErrorExt};
+    use common_error::mock::MockError;
     use common_error::status_code::StatusCode;
     use store_api::storage::RegionId;
     use table::table_name::TableName;
@@ -635,6 +655,38 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_restore_metadata_retry_policy() {
+        let source = error::Error::retry_later(MockError::new(StatusCode::Internal));
+        let err = Err::<(), _>(BoxedError::new(source))
+            .context(error::ExternalSnafu)
+            .unwrap_err();
+        assert!(err.retry_hint().is_retryable());
+        assert!(!err.is_retry_later());
+        let err = UndropTableProcedure::map_restore_error(err);
+        assert!(!err.is_retry_later());
+        assert!(!err.need_clean_poisons());
+
+        for retryable in [false, true] {
+            for clean_poisons in [false, true] {
+                let source = BoxedError::new(MockError::new(StatusCode::Internal));
+                let err = if retryable {
+                    error::Error::RetryLater {
+                        source,
+                        clean_poisons,
+                    }
+                } else {
+                    Err::<(), _>(source)
+                        .context(error::AbortProcedureSnafu { clean_poisons })
+                        .unwrap_err()
+                };
+                let err = UndropTableProcedure::map_restore_error(err);
+                assert_eq!(err.is_retry_later(), retryable);
+                assert_eq!(err.need_clean_poisons(), clean_poisons);
+            }
+        }
+    }
+
     #[test]
     fn test_map_restore_metadata_error_without_table_name() {
         let context = new_ddl_context(Arc::new(MockDatanodeManager::new(())));

src/common/meta/src/ddl/utils.rs

@@ -28,7 +28,7 @@ use api::v1::region::{
     region_request,
 };
 use common_catalog::consts::{METRIC_ENGINE, MITO_ENGINE};
-use common_error::ext::BoxedError;
+use common_error::ext::{BoxedError, ErrorExt};
 use common_procedure::error::Error as ProcedureError;
 use common_telemetry::tracing_context::TracingContext;
 use common_telemetry::{error, info, warn};
@@ -61,7 +61,7 @@ use crate::wal_provider::RegionWalOptions;
 pub fn add_peer_context_if_needed(datanode: Peer) -> impl FnOnce(Error) -> Error {
     move |err| {
         error!(err; "Failed to operate datanode, peer: {}", datanode);
-        if !err.is_retry_later() {
+        if !err.retry_hint().is_retryable() {
             return Err::<(), BoxedError>(BoxedError::new(err))
                 .context(OperateDatanodeSnafu { peer: datanode })
                 .unwrap_err();
@@ -84,7 +84,7 @@ pub(crate) fn is_metric_engine_logical_table(
 /// This function determines whether the error should be retried and if poison cleanup is needed,
 /// then maps it to the appropriate procedure error variant.
 pub fn map_to_procedure_error(e: Error) -> ProcedureError {
-    match (e.is_retry_later(), e.need_clean_poisons()) {
+    match (e.retry_hint().is_retryable(), e.need_clean_poisons()) {
         (true, true) => ProcedureError::retry_later_and_clean_poisons(e),
         (true, false) => ProcedureError::retry_later(e),
         (false, true) => ProcedureError::external_and_clean_poisons(e),
@@ -256,7 +256,7 @@ pub fn handle_multiple_results<T: Debug>(results: Vec<Result<T>>) -> MultipleRes
         match result {
             Ok(value) => ok_results.push(value),
             Err(err) => {
-                if err.is_retry_later() {
+                if err.retry_hint().is_retryable() {
                     retryable_results.push(err);
                 } else {
                     non_retryable_results.push(err);
@@ -475,8 +475,74 @@ pub fn extract_column_metadatas(
 
 #[cfg(test)]
 mod tests {
+    use common_error::mock::MockError;
+    use common_error::status_code::StatusCode;
+
     use super::*;
 
+    fn external_error(retryable: bool) -> Error {
+        let source = MockError::new(StatusCode::Internal);
+        let source = if retryable {
+            BoxedError::new(Error::retry_later(source))
+        } else {
+            BoxedError::new(source)
+        };
+        Err::<(), _>(source)
+            .context(error::ExternalSnafu)
+            .unwrap_err()
+    }
+
+    #[test]
+    fn test_map_to_procedure_error() {
+        for retryable in [false, true] {
+            for clean_poisons in [false, true] {
+                let err = if retryable {
+                    Error::RetryLater {
+                        source: BoxedError::new(external_error(false)),
+                        clean_poisons,
+                    }
+                } else {
+                    Err::<(), _>(BoxedError::new(external_error(true)))
+                        .context(error::AbortProcedureSnafu { clean_poisons })
+                        .unwrap_err()
+                };
+                let err = map_to_procedure_error(err);
+                assert_eq!(err.is_retry_later(), retryable);
+                assert_eq!(err.need_clean_poisons(), clean_poisons);
+            }
+            let err = map_to_procedure_error(external_error(retryable));
+            assert_eq!(err.is_retry_later(), retryable);
+            assert!(!err.need_clean_poisons());
+        }
+    }
+
+    #[test]
+    fn test_handle_multiple_results_retry_hint() {
+        assert!(matches!(
+            handle_multiple_results(vec![Ok(()), Err(external_error(true))]),
+            MultipleResults::PartialRetryable(_)
+        ));
+        assert!(matches!(
+            handle_multiple_results::<()>(vec![
+                Err(external_error(true)),
+                Err(external_error(true))
+            ]),
+            MultipleResults::AllRetryable(_)
+        ));
+        assert!(matches!(
+            handle_multiple_results(vec![
+                Ok(()),
+                Err(external_error(true)),
+                Err(external_error(false))
+            ]),
+            MultipleResults::PartialNonRetryable(_)
+        ));
+        assert!(matches!(
+            handle_multiple_results::<()>(vec![Err(external_error(false))]),
+            MultipleResults::AllNonRetryable(_)
+        ));
+    }
+
     #[test]
     fn test_get_catalog_and_schema() {
         let test_catalog = "my_catalog";

src/common/meta/src/error.rs

@@ -1373,15 +1373,15 @@ impl ErrorExt for Error {
             | StartProcedureManager { source, .. }
             | StopProcedureManager { source, .. }
             | RegisterProcedureLoader { source, .. }
-            | PutPoison { source, .. }
-            | ProcedureStateReceiver { source, .. } => source.retry_hint(),
+            | PutPoison { source, .. } => source.retry_hint(),
             External { source, .. }
             | ResponseExceededSizeLimit { source, .. }
             | OperateDatanode { source, .. }
-            | AbortProcedure { source, .. }
             | RegisterRepartitionProcedureLoader { source, .. }
             | CreateRepartitionProcedure { source, .. }
             | PersistRepartitionGcRequirement { source, .. } => source.retry_hint(),
+            // These errors end an operation; their sources cannot make it retryable.
+            AbortProcedure { .. } | ProcedureStateReceiver { .. } => RetryHint::NonRetryable,
             Table { source, .. } => source.retry_hint(),
             ConvertAlterTableRequest { source, .. } => source.retry_hint(),
             ConvertColumnDef { source, .. } => source.retry_hint(),
@@ -1466,6 +1466,7 @@ mod retry_hint_tests {
     use std::sync::Arc;
 
     use common_error::mock::MockError;
+    use snafu::ResultExt;
 
     use super::*;
 
@@ -1483,6 +1484,42 @@ mod retry_hint_tests {
         assert_eq!(err.retry_hint(), RetryHint::Retryable);
     }
 
+    #[test]
+    fn test_leader_retry_hints() {
+        for (error, expected) in [
+            (
+                NoLeaderSnafu { table_id: 1_u32 }.build(),
+                RetryHint::Retryable,
+            ),
+            (ElectionNoLeaderSnafu.build(), RetryHint::Retryable),
+            (
+                ElectionLeaderLeaseExpiredSnafu.build(),
+                RetryHint::Retryable,
+            ),
+            (
+                ElectionLeaderLeaseChangedSnafu.build(),
+                RetryHint::Retryable,
+            ),
+        ] {
+            assert_eq!(error.retry_hint(), expected, "{error:?}");
+        }
+    }
+
+    #[test]
+    fn test_procedure_error_retry_hints() {
+        let source = common_procedure::Error::retry_later(MockError::new(StatusCode::Internal));
+        let err = Err::<(), _>(source).context(PutPoisonSnafu).unwrap_err();
+        assert_eq!(err.retry_hint(), RetryHint::Retryable);
+
+        let source = common_procedure::Error::retry_later(MockError::new(StatusCode::Internal));
+        let err = Err::<(), _>(source)
+            .context(ProcedureStateReceiverSnafu {
+                procedure_id: ProcedureId::random(),
+            })
+            .unwrap_err();
+        assert_eq!(err.retry_hint(), RetryHint::NonRetryable);
+    }
+
     #[test]
     fn test_get_cache_forwards_retry_hint() {
         let source = Arc::new(Error::retry_later(MockError::new(StatusCode::Internal)));

src/common/meta/src/node_manager.rs

@@ -28,12 +28,22 @@ use crate::peer::Peer;
 #[async_trait::async_trait]
 pub trait Datanode: Send + Sync {
     /// Handles DML, and DDL requests.
-    async fn handle(&self, request: RegionRequest) -> Result<RegionResponse>;
+    async fn handle_request(&self, request: Arc<RegionRequest>) -> Result<RegionResponse>;
 
     /// Handles query requests
     async fn handle_query(&self, request: QueryRequest) -> Result<SendableRecordBatchStream>;
 }
 
+impl dyn Datanode + '_ {
+    /// Handles an owned or shared request without cloning its payload.
+    pub async fn handle<R>(&self, request: R) -> Result<RegionResponse>
+    where
+        R: Into<Arc<RegionRequest>> + Send,
+    {
+        self.handle_request(request.into()).await
+    }
+}
+
 pub type DatanodeRef = Arc<dyn Datanode>;
 
 /// The trait for handling requests to flownode

src/common/meta/src/reconciliation/reconcile_catalog.rs

@@ -16,6 +16,7 @@ use std::any::Any;
 use std::fmt::Debug;
 use std::time::Instant;
 
+use common_error::ext::ErrorExt;
 use common_procedure::error::FromJsonSnafu;
 use common_procedure::{
     Context as ProcedureContext, Error as ProcedureError, EventContext, EventTrigger, LockKey,
@@ -191,7 +192,7 @@ impl Procedure for ReconcileCatalogProcedure {
                 Ok(status)
             }
             Err(e) => {
-                if e.is_retry_later() {
+                if e.retry_hint().is_retryable() {
                     metrics::METRIC_META_RECONCILIATION_PROCEDURE_ERROR
                         .with_label_values(&[procedure_name, step, metrics::ERROR_TYPE_RETRYABLE])
                         .inc();

src/common/meta/src/reconciliation/reconcile_database.rs

@@ -23,6 +23,7 @@ use std::fmt::Debug;
 use std::time::Instant;
 
 use async_trait::async_trait;
+use common_error::ext::ErrorExt;
 use common_procedure::error::{FromJsonSnafu, ToJsonSnafu};
 use common_procedure::{
     Context as ProcedureContext, Error as ProcedureError, EventContext, EventTrigger, LockKey,
@@ -231,7 +232,7 @@ impl Procedure for ReconcileDatabaseProcedure {
                 Ok(status)
             }
             Err(e) => {
-                if e.is_retry_later() {
+                if e.retry_hint().is_retryable() {
                     metrics::METRIC_META_RECONCILIATION_PROCEDURE_ERROR
                         .with_label_values(&[procedure_name, step, metrics::ERROR_TYPE_RETRYABLE])
                         .inc();

src/common/meta/src/reconciliation/reconcile_logical_tables.rs

@@ -22,6 +22,7 @@ use std::any::Any;
 use std::fmt::Debug;
 
 use async_trait::async_trait;
+use common_error::ext::ErrorExt;
 use common_procedure::error::{FromJsonSnafu, ToJsonSnafu};
 use common_procedure::{
     Context as ProcedureContext, Error as ProcedureError, EventContext, EventTrigger, LockKey,
@@ -207,7 +208,7 @@ impl Procedure for ReconcileLogicalTablesProcedure {
                 Ok(status)
             }
             Err(e) => {
-                if e.is_retry_later() {
+                if e.retry_hint().is_retryable() {
                     metrics::METRIC_META_RECONCILIATION_PROCEDURE_ERROR
                         .with_label_values(&[procedure_name, step, metrics::ERROR_TYPE_RETRYABLE])
                         .inc();

src/common/meta/src/reconciliation/reconcile_table.rs

@@ -20,6 +20,7 @@ pub(crate) mod update_table_info;
 
 use std::fmt::Debug;
 
+use common_error::ext::ErrorExt;
 use common_procedure::error::{FromJsonSnafu, ToJsonSnafu};
 use common_procedure::{
     Context as ProcedureContext, Error as ProcedureError, EventContext, EventTrigger, LockKey,
@@ -318,7 +319,7 @@ impl Procedure for ReconcileTableProcedure {
                 Ok(status)
             }
             Err(e) => {
-                if e.is_retry_later() {
+                if e.retry_hint().is_retryable() {
                     metrics::METRIC_META_RECONCILIATION_PROCEDURE_ERROR
                         .with_label_values(&[procedure_name, step, metrics::ERROR_TYPE_RETRYABLE])
                         .inc();

src/common/meta/src/test_util.rs

@@ -111,7 +111,8 @@ struct MockNode<T> {
 
 #[async_trait::async_trait]
 impl<T: MockDatanodeHandler> Datanode for MockNode<T> {
-    async fn handle(&self, request: RegionRequest) -> Result<RegionResponse> {
+    async fn handle_request(&self, request: Arc<RegionRequest>) -> Result<RegionResponse> {
+        let request = Arc::try_unwrap(request).expect("mock request must be exclusively owned");
         self.handler.handle(&self.peer, request).await
     }
 

src/common/procedure/src/error.rs

@@ -306,7 +306,8 @@ impl ErrorExt for Error {
             | Error::DeletePoison { source, .. }
             | Error::GetPoison { source, .. }
             | Error::CheckStatus { source, .. } => source.retry_hint(),
-            Error::ProcedureExec { source, .. } => source.retry_hint(),
+            // The watcher reports this only after the child has failed or been poisoned.
+            Error::ProcedureExec { .. } => RetryHint::NonRetryable,
             Error::StartRemoveOutdatedMetaTask { source, .. }
             | Error::StopRemoveOutdatedMetaTask { source, .. } => source.retry_hint(),
             Error::DeleteState { error, .. } => retry_hint_from_opendal_error(error),
@@ -379,8 +380,11 @@ mod tests {
     use std::sync::Arc;
 
     use common_error::mock::MockError;
+    use tokio::sync::watch;
 
     use super::*;
+    use crate::procedure::ProcedureState;
+    use crate::watcher;
 
     #[test]
     fn test_retry_later_hint_is_retryable() {
@@ -397,6 +401,21 @@ mod tests {
         assert_eq!(err.retry_hint(), RetryHint::Retryable);
     }
 
+    #[tokio::test]
+    async fn test_finished_procedure_hint_is_non_retryable() {
+        let source = Arc::new(Error::retry_later(MockError::new(StatusCode::Internal)));
+        for state in [
+            ProcedureState::failed(source.clone()),
+            ProcedureState::poisoned(Default::default(), source),
+        ] {
+            let (sender, mut receiver) = watch::channel(ProcedureState::Running);
+            sender.send(state).unwrap();
+            let err = watcher::wait(&mut receiver).await.unwrap_err();
+
+            assert_eq!(err.retry_hint(), RetryHint::NonRetryable);
+        }
+    }
+
     #[test]
     fn test_retry_exceeded_hint_is_non_retryable() {
         let source = Arc::new(Error::retry_later(MockError::new(StatusCode::Internal)));