From 2f5832dfa031db96199f5181120eb416781656c1 Mon Sep 17 00:00:00 2001 From: WenyXu Date: Mon, 12 Oct 2026 10:58:23 +0000 Subject: [PATCH] feat: propagate route refresh retry hints Signed-off-by: WenyXu --- Cargo.lock | 1 + src/client/Cargo.toml | 3 + src/client/src/client.rs | 1 + src/client/src/client_manager.rs | 25 +-- src/client/src/error.rs | 15 +- src/client/src/region.rs | 158 ++++++++++------ src/common/error/src/ext.rs | 18 +- src/common/error/tests/ext.rs | 9 + src/common/meta/src/error.rs | 91 ++++----- src/datanode/src/error.rs | 80 ++++---- src/metric-engine/src/error.rs | 58 ++++-- src/mito2/src/error.rs | 175 ++++++++++++------ tests/compatibility/README.md | 9 +- .../compatibility/wire/write_retry/README.md | 25 +++ .../wire/write_retry/retry_hints.json | 7 + 15 files changed, 446 insertions(+), 229 deletions(-) create mode 100644 tests/compatibility/wire/write_retry/README.md create mode 100644 tests/compatibility/wire/write_retry/retry_hints.json diff --git a/Cargo.lock b/Cargo.lock index f75bde16328..6653e94357d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2115,6 +2115,7 @@ dependencies = [ "prost 0.14.1", "query", "rand 0.9.4", + "serde", "serde_json", "snafu 0.8.6", "store-api", diff --git a/src/client/Cargo.toml b/src/client/Cargo.toml index e5bd479cc7d..236776cc59c 100644 --- a/src/client/Cargo.toml +++ b/src/client/Cargo.toml @@ -46,3 +46,6 @@ tokio.workspace = true tokio-stream = { workspace = true, features = ["net"] } tonic.workspace = true tonic-prost.workspace = true + +[dev-dependencies] +serde.workspace = true diff --git a/src/client/src/client.rs b/src/client/src/client.rs index 787dc3bde6b..1746854cd26 100644 --- a/src/client/src/client.rs +++ b/src/client/src/client.rs @@ -525,6 +525,7 @@ impl Client { Ok(FlightClient { addr, client }) } + /// Creates the region gRPC transport with the configured message limits. pub(crate) fn raw_region_client(&self) -> Result<(String, tonic::client::Grpc)> { let (addr, channel) = self.find_channel()?; let client = tonic::client::Grpc::new(channel) diff --git a/src/client/src/client_manager.rs b/src/client/src/client_manager.rs index 4d723361e80..ef26c107cd5 100644 --- a/src/client/src/client_manager.rs +++ b/src/client/src/client_manager.rs @@ -52,18 +52,7 @@ impl Debug for NodeClients { #[async_trait::async_trait] impl DatanodeManager for NodeClients { async fn datanode(&self, datanode: &Peer) -> DatanodeRef { - let client = self.get_client(datanode).await; - - let ChannelConfig { - send_compression, - accept_compression, - .. - } = self.control_channel_manager.config(); - Arc::new(RegionRequester::new( - client, - *send_compression, - *accept_compression, - )) + Arc::new(self.make_region_requester(datanode).await) } } @@ -89,6 +78,18 @@ impl NodeClients { } } + /// Creates a [`RegionRequester`] with the cached client and control channel compression settings. + pub async fn make_region_requester(&self, datanode: &Peer) -> RegionRequester { + let client = self.get_client(datanode).await; + + let ChannelConfig { + send_compression, + accept_compression, + .. + } = self.control_channel_manager.config(); + RegionRequester::new(client, *send_compression, *accept_compression) + } + pub async fn get_client(&self, datanode: &Peer) -> Client { self.clients .get_with_by_ref(datanode, async move { diff --git a/src/client/src/error.rs b/src/client/src/error.rs index 168ba422715..5c70e5fbe91 100644 --- a/src/client/src/error.rs +++ b/src/client/src/error.rs @@ -18,7 +18,7 @@ use common_error::define_from_tonic_status; use common_error::ext::{BoxedError, ErrorExt, RetryHint}; use common_error::status_code::StatusCode; use common_macro::stack_trace_debug; -use snafu::{Location, Snafu}; +use snafu::{Location, Snafu, location}; use tonic::Code; use tonic::metadata::errors::InvalidMetadataValue; @@ -205,6 +205,19 @@ impl ErrorExt for Error { define_from_tonic_status!(Error, Tonic); +impl From for common_meta::error::Error { + fn from(error: Error) -> Self { + if error.should_retry() { + Self::retry_later(error) + } else { + Self::External { + source: BoxedError::new(error), + location: location!(), + } + } + } +} + impl Error { /// Returns the gRPC status code if this error is caused by a gRPC request failure. pub fn tonic_code(&self) -> Option { diff --git a/src/client/src/region.rs b/src/client/src/region.rs index 42124286b17..19efb60599d 100644 --- a/src/client/src/region.rs +++ b/src/client/src/region.rs @@ -57,6 +57,7 @@ use crate::{Client, metrics}; const FLIGHT_DO_GET_TIMEOUT: Duration = Duration::from_secs(10); +/// Sends region operations through a configured gRPC client. #[derive(Debug)] pub struct RegionRequester { client: Client, @@ -67,19 +68,7 @@ pub struct RegionRequester { #[async_trait] impl Datanode for RegionRequester { async fn handle(&self, request: RegionRequest) -> MetaResult { - self.handle_inner(request).await.map_err(|err| { - if err.should_retry() { - meta_error::Error::RetryLater { - source: BoxedError::new(err), - clean_poisons: false, - } - } else { - meta_error::Error::External { - source: BoxedError::new(err), - location: location!(), - } - } - }) + self.handle_inner(request).await.map_err(Into::into) } async fn handle_query(&self, request: QueryRequest) -> MetaResult { @@ -148,6 +137,40 @@ impl RegionRequester { recordbatches_from_flight_message_stream(addr, flight_message_stream).await } + /// Sends an owned or shared request without deep-cloning its protobuf payload. + /// + /// Accepts both [`RegionRequest`] and [`Arc`]. Callers can retain a + /// shared request for another attempt; this method does not retry or determine + /// whether replaying the request is safe. + pub async fn handle(&self, request: R) -> Result + where + R: Borrow + Send + Sync + 'static, + { + self.handle_inner(request).await + } + + /// Sends a remote dynamic filter update to the region service. + pub async fn handle_remote_dyn_filter_update( + &self, + query_id: impl Into, + update: RemoteDynFilterUpdate, + ) -> Result { + self.handle_inner(build_remote_dyn_filter_update_request(query_id, update)) + .await + } + + /// Unregisters a remote dynamic filter in the region service. + pub async fn handle_remote_dyn_filter_unregister( + &self, + query_id: impl Into, + unregister: RemoteDynFilterUnregister, + ) -> Result { + self.handle_inner(build_remote_dyn_filter_unregister_request( + query_id, unregister, + )) + .await + } + async fn handle_inner(&self, request: R) -> Result where R: Borrow + Send + Sync + 'static, @@ -156,7 +179,7 @@ impl RegionRequester { .borrow() .body .as_ref() - .with_context(|| MissingFieldSnafu { field: "body" })?; + .context(MissingFieldSnafu { field: "body" })?; let is_insert = matches!( request_body, region_request::Body::Inserts(_) | region_request::Body::BulkInsert(_) @@ -213,38 +236,6 @@ impl RegionRequester { Ok(RegionResponse::from_region_response(response)) } - - /// Sends an owned or shared request without deep-cloning its protobuf payload. - /// - /// Accepts both [`RegionRequest`] and [`Arc`]. Callers can retain a - /// shared request for another attempt; this method does not retry or determine - /// whether replaying the request is safe. - pub async fn handle(&self, request: R) -> Result - where - R: Borrow + Send + Sync + 'static, - { - self.handle_inner(request).await - } - - pub async fn handle_remote_dyn_filter_update( - &self, - query_id: impl Into, - update: RemoteDynFilterUpdate, - ) -> Result { - self.handle_inner(build_remote_dyn_filter_update_request(query_id, update)) - .await - } - - pub async fn handle_remote_dyn_filter_unregister( - &self, - query_id: impl Into, - unregister: RemoteDynFilterUnregister, - ) -> Result { - self.handle_inner(build_remote_dyn_filter_unregister_request( - query_id, unregister, - )) - .await - } } async fn recordbatches_from_flight_message_stream( @@ -433,7 +424,6 @@ mod tests { use api::v1::{ ColumnDataType, ColumnSchema as PbColumnSchema, Row, Rows, Status as PbStatus, Value, }; - use common_error::GREPTIME_DB_HEADER_ERROR_RETRY_HINT; use common_error::ext::RetryHint; use common_grpc::channel_manager::{ChannelConfig, ChannelManager}; use common_recordbatch::adapter::RecordBatchMetrics; @@ -449,7 +439,7 @@ mod tests { use tonic::{Request, Response, Status}; use super::*; - use crate::Error::{self, IllegalDatabaseResponse, Server}; + use crate::Error; #[derive(Clone, Default)] struct RecordingRegionService { @@ -651,7 +641,7 @@ mod tests { let server = RegionTestServer::start().await; let mut status = Status::unavailable("migrating"); status.metadata_mut().insert( - GREPTIME_DB_HEADER_ERROR_RETRY_HINT, + common_error::GREPTIME_DB_HEADER_ERROR_RETRY_HINT, RetryHint::Retryable.as_str().parse().unwrap(), ); server @@ -674,14 +664,69 @@ mod tests { assert_eq!(&requests[1].0, shared.as_ref()); } + #[test] + fn test_retry_hint_wire_compatibility() { + #[derive(serde::Deserialize)] + struct Fixture { + name: String, + wire_hint: Option, + expected_hint: String, + } + + let fixtures: Vec = serde_json::from_str(include_str!( + "../../../tests/compatibility/wire/write_retry/retry_hints.json" + )) + .unwrap(); + assert_eq!(fixtures.len(), 5); + for fixture in fixtures { + let mut status = Status::failed_precondition("region unavailable"); + status.metadata_mut().insert( + common_error::GREPTIME_DB_HEADER_ERROR_CODE, + (StatusCode::RegionNotReady as u32) + .to_string() + .parse() + .unwrap(), + ); + if let Some(hint) = fixture.wire_hint { + status.metadata_mut().insert( + common_error::GREPTIME_DB_HEADER_ERROR_RETRY_HINT, + hint.parse().unwrap(), + ); + } + let error: error::Error = status.into(); + assert_eq!( + error.retry_hint().as_str(), + fixture.expected_hint, + "{}", + fixture.name + ); + assert_eq!( + error.status_code(), + StatusCode::RegionNotReady, + "{}", + fixture.name + ); + assert_eq!( + error.tonic_code(), + Some(tonic::Code::FailedPrecondition), + "{}", + fixture.name + ); + } + } + #[tokio::test] async fn test_transport_error_parity_and_datanode_mapping() { let server = RegionTestServer::start().await; let requester = server.create_requester(ChannelConfig::default(), false, false); - for hint in [RetryHint::Retryable, RetryHint::NonRetryable] { + for hint in [ + RetryHint::Retryable, + RetryHint::RetryAfterRouteRefresh, + RetryHint::NonRetryable, + ] { let mut status = Status::failed_precondition("region unavailable"); status.metadata_mut().insert( - GREPTIME_DB_HEADER_ERROR_RETRY_HINT, + common_error::GREPTIME_DB_HEADER_ERROR_RETRY_HINT, hint.as_str().parse().unwrap(), ); status.metadata_mut().insert( @@ -725,8 +770,9 @@ mod tests { .await .unwrap_err(); assert_eq!(error.status_code(), StatusCode::RegionNotFound); + assert_eq!(error.retry_hint(), hint); match hint { - RetryHint::Retryable => assert!(matches!( + RetryHint::Retryable | RetryHint::RetryAfterRouteRefresh => assert!(matches!( error, meta_error::Error::RetryLater { clean_poisons: false, @@ -931,13 +977,13 @@ mod tests { let result = check_response_header(&None); assert!(matches!( result.unwrap_err(), - IllegalDatabaseResponse { .. } + Error::IllegalDatabaseResponse { .. } )); let result = check_response_header(&Some(ResponseHeader { status: None })); assert!(matches!( result.unwrap_err(), - IllegalDatabaseResponse { .. } + Error::IllegalDatabaseResponse { .. } )); let result = check_response_header(&Some(ResponseHeader { @@ -956,7 +1002,7 @@ mod tests { })); assert!(matches!( result.unwrap_err(), - IllegalDatabaseResponse { .. } + Error::IllegalDatabaseResponse { .. } )); let result = check_response_header(&Some(ResponseHeader { @@ -965,7 +1011,7 @@ mod tests { err_msg: "blabla".to_string(), }), })); - let Server { code, msg, .. } = result.unwrap_err() else { + let Error::Server { code, msg, .. } = result.unwrap_err() else { unreachable!() }; assert_eq!(code, StatusCode::Internal); diff --git a/src/common/error/src/ext.rs b/src/common/error/src/ext.rs index 0c267dc15b2..f2d01914818 100644 --- a/src/common/error/src/ext.rs +++ b/src/common/error/src/ext.rs @@ -46,6 +46,10 @@ use crate::status_code::StatusCode; pub enum RetryHint { /// The operation may succeed if retried later. Retryable, + /// The operation may succeed after refreshing routing and partition information. + /// The request may need rebuilding when its partition assignment changes. + /// This hint does not imply that a failed operation had no side effects. + RetryAfterRouteRefresh, /// Retrying the same operation is not expected to help. /// /// This is the default for errors that do not explicitly opt in to retry. @@ -53,16 +57,23 @@ pub enum RetryHint { } const RETRY_HINT_RETRYABLE: &str = "retryable"; +const RETRY_HINT_AFTER_ROUTE_REFRESH: &str = "retry_after_route_refresh"; const RETRY_HINT_NON_RETRYABLE: &str = "non_retryable"; impl RetryHint { + /// Returns whether another attempt may succeed, possibly after refreshing routes. pub fn is_retryable(self) -> bool { - matches!(self, RetryHint::Retryable) + matches!( + self, + RetryHint::Retryable | RetryHint::RetryAfterRouteRefresh + ) } + /// Returns the retry hint encoded in gRPC error metadata. pub fn as_str(self) -> &'static str { match self { RetryHint::Retryable => RETRY_HINT_RETRYABLE, + RetryHint::RetryAfterRouteRefresh => RETRY_HINT_AFTER_ROUTE_REFRESH, RetryHint::NonRetryable => RETRY_HINT_NON_RETRYABLE, } } @@ -90,6 +101,7 @@ impl FromStr for RetryHint { fn from_str(s: &str) -> Result { match s { RETRY_HINT_RETRYABLE => Ok(RetryHint::Retryable), + RETRY_HINT_AFTER_ROUTE_REFRESH => Ok(RetryHint::RetryAfterRouteRefresh), RETRY_HINT_NON_RETRYABLE => Ok(RetryHint::NonRetryable), _ => Err(()), } @@ -133,7 +145,9 @@ pub trait ErrorExt: StackError { /// /// Implementations should return [`RetryHint::Retryable`] only when retrying the /// same operation may succeed without changing the request. The default is - /// [`RetryHint::NonRetryable`] to avoid accidental retry loops. + /// [`RetryHint::NonRetryable`] to avoid accidental retry loops. Use + /// [`RetryHint::RetryAfterRouteRefresh`] when routing or partition information + /// must be refreshed before another attempt. fn retry_hint(&self) -> RetryHint { RetryHint::NonRetryable } diff --git a/src/common/error/tests/ext.rs b/src/common/error/tests/ext.rs index 8be8afa4748..c34a47c400f 100644 --- a/src/common/error/tests/ext.rs +++ b/src/common/error/tests/ext.rs @@ -164,6 +164,15 @@ fn test_transparent_flag() { #[test] fn test_retry_hint_helpers() { assert!(RetryHint::Retryable.is_retryable()); + assert!(RetryHint::RetryAfterRouteRefresh.is_retryable()); + assert_eq!( + RetryHint::RetryAfterRouteRefresh.as_str(), + "retry_after_route_refresh" + ); + assert_eq!( + RetryHint::from_str("retry_after_route_refresh").unwrap(), + RetryHint::RetryAfterRouteRefresh + ); assert!(!RetryHint::NonRetryable.is_retryable()); assert_eq!(RetryHint::Retryable.as_str(), "retryable"); diff --git a/src/common/meta/src/error.rs b/src/common/meta/src/error.rs index 5c1febb9b4d..24159a825c2 100644 --- a/src/common/meta/src/error.rs +++ b/src/common/meta/src/error.rs @@ -1343,59 +1343,62 @@ impl ErrorExt for Error { } fn retry_hint(&self) -> RetryHint { - use Error::*; - match self { - RetryLater { .. } - | GetLatestCacheRetryExceeded { .. } - | NoLeader { .. } - | ElectionNoLeader { .. } - | ElectionLeaderLeaseExpired { .. } - | ElectionLeaderLeaseChanged { .. } => RetryHint::Retryable, - ConnectEtcd { error, .. } | EtcdFailed { error, .. } | EtcdTxnFailed { error, .. } => { - retry_hint_from_etcd_error(error) - } - WriteObject { error, .. } | ReadObject { error, .. } => { + Error::RetryLater { source, .. } => match source.retry_hint() { + RetryHint::RetryAfterRouteRefresh => RetryHint::RetryAfterRouteRefresh, + RetryHint::Retryable | RetryHint::NonRetryable => RetryHint::Retryable, + }, + Error::GetLatestCacheRetryExceeded { .. } + | Error::NoLeader { .. } + | Error::ElectionNoLeader { .. } + | Error::ElectionLeaderLeaseExpired { .. } + | Error::ElectionLeaderLeaseChanged { .. } => RetryHint::Retryable, + Error::ConnectEtcd { error, .. } + | Error::EtcdFailed { error, .. } + | Error::EtcdTxnFailed { error, .. } => retry_hint_from_etcd_error(error), + Error::WriteObject { error, .. } | Error::ReadObject { error, .. } => { retry_hint_from_opendal_error(error) } - BuildKafkaClient { error, .. } - | BuildKafkaCtrlClient { error, .. } - | KafkaPartitionClient { error, .. } - | KafkaGetOffset { error, .. } - | ProduceRecord { error, .. } - | CreateKafkaWalTopic { error, .. } => rskafka_client_error_to_retry_hint(error), - SubmitProcedure { source, .. } - | QueryProcedure { source, .. } - | WaitProcedure { source, .. } - | StartProcedureManager { source, .. } - | StopProcedureManager { source, .. } - | RegisterProcedureLoader { source, .. } - | PutPoison { source, .. } - | ProcedureStateReceiver { source, .. } => source.retry_hint(), - External { source, .. } - | ResponseExceededSizeLimit { source, .. } - | OperateDatanode { source, .. } - | AbortProcedure { source, .. } - | RegisterRepartitionProcedureLoader { source, .. } - | CreateRepartitionProcedure { source, .. } - | PersistRepartitionGcRequirement { source, .. } => source.retry_hint(), - Table { source, .. } => source.retry_hint(), - ConvertAlterTableRequest { source, .. } => source.retry_hint(), - ConvertColumnDef { source, .. } => source.retry_hint(), - GetCache { source, .. } => source.retry_hint(), + Error::BuildKafkaClient { error, .. } + | Error::BuildKafkaCtrlClient { error, .. } + | Error::KafkaPartitionClient { error, .. } + | Error::KafkaGetOffset { error, .. } + | Error::ProduceRecord { error, .. } + | Error::CreateKafkaWalTopic { error, .. } => rskafka_client_error_to_retry_hint(error), + Error::SubmitProcedure { source, .. } + | Error::QueryProcedure { source, .. } + | Error::WaitProcedure { source, .. } + | Error::StartProcedureManager { source, .. } + | Error::StopProcedureManager { source, .. } + | Error::RegisterProcedureLoader { source, .. } + | Error::PutPoison { source, .. } + | Error::ProcedureStateReceiver { source, .. } => source.retry_hint(), + Error::External { source, .. } + | Error::ResponseExceededSizeLimit { source, .. } + | Error::OperateDatanode { source, .. } + | Error::AbortProcedure { source, .. } + | Error::RegisterRepartitionProcedureLoader { source, .. } + | Error::CreateRepartitionProcedure { source, .. } + | Error::PersistRepartitionGcRequirement { source, .. } => source.retry_hint(), + Error::Table { source, .. } => source.retry_hint(), + Error::ConvertAlterTableRequest { source, .. } => source.retry_hint(), + Error::ConvertColumnDef { source, .. } => source.retry_hint(), + Error::GetCache { source, .. } => source.retry_hint(), #[cfg(feature = "pg_kvbackend")] - PostgresExecution { error, .. } | PostgresTransaction { error, .. } => { + Error::PostgresExecution { error, .. } | Error::PostgresTransaction { error, .. } => { retry_hint_from_postgres_error(error) } #[cfg(feature = "pg_kvbackend")] - GetPostgresClient { error, .. } => retry_hint_from_postgres_pool_error(error), + Error::GetPostgresClient { error, .. } => retry_hint_from_postgres_pool_error(error), #[cfg(feature = "mysql_kvbackend")] - MySqlExecution { error, .. } - | CreateMySqlPool { error, .. } - | AcquireMySqlClient { error, .. } - | MySqlTransaction { error, .. } => retry_hint_from_sqlx_error(error), + Error::MySqlExecution { error, .. } + | Error::CreateMySqlPool { error, .. } + | Error::AcquireMySqlClient { error, .. } + | Error::MySqlTransaction { error, .. } => retry_hint_from_sqlx_error(error), #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))] - RdsTransactionRetryFailed { .. } | SqlExecutionTimeout { .. } => RetryHint::Retryable, + Error::RdsTransactionRetryFailed { .. } | Error::SqlExecutionTimeout { .. } => { + RetryHint::Retryable + } _ => RetryHint::NonRetryable, } } diff --git a/src/datanode/src/error.rs b/src/datanode/src/error.rs index e9b5d5ec1dc..e51c438939e 100644 --- a/src/datanode/src/error.rs +++ b/src/datanode/src/error.rs @@ -558,36 +558,38 @@ impl ErrorExt for Error { } fn retry_hint(&self) -> RetryHint { - use Error::*; - match self { - RegionBusy { .. } - | RegionNotReady { .. } - | ConcurrentQueryLimiterClosed { .. } - | ConcurrentQueryLimiterTimeout { .. } => RetryHint::Retryable, - NewPlanDecoder { source, .. } | ExecuteLogicalPlan { source, .. } => { + Error::RegionNotReady { .. } | Error::RegionNotFound { .. } => { + RetryHint::RetryAfterRouteRefresh + } + Error::RegionBusy { .. } + | Error::ConcurrentQueryLimiterClosed { .. } + | Error::ConcurrentQueryLimiterTimeout { .. } => RetryHint::Retryable, + Error::NewPlanDecoder { source, .. } | Error::ExecuteLogicalPlan { source, .. } => { source.retry_hint() } - HandleHeartbeatResponse { source, .. } | GetMetadata { source, .. } => { + Error::HandleHeartbeatResponse { source, .. } | Error::GetMetadata { source, .. } => { source.retry_hint() } - DecodeLogicalPlan { source, .. } => source.retry_hint(), - Delete { source, .. } => source.retry_hint(), - AsyncTaskExecute { source, .. } => source.retry_hint(), - StartServer { source, .. } | ShutdownServer { source, .. } => source.retry_hint(), - OpenLogStore { source, .. } => source.retry_hint(), - MetaClientInit { source, .. } => source.retry_hint(), - HandleRegionRequest { source, .. } - | GetRegionMetadata { source, .. } - | HandleBatchOpenRequest { source, .. } - | HandleBatchDdlRequest { source, .. } - | StopRegionEngine { source, .. } => source.retry_hint(), - FindLogicalRegions { source, .. } => source.retry_hint(), - BuildMitoEngine { source, .. } => source.retry_hint(), - GcMitoEngine { source, .. } => source.retry_hint(), - BuildMetricEngine { source, .. } => source.retry_hint(), - ListStorageSsts { source, .. } => source.retry_hint(), - ObjectStore { source, .. } => source.retry_hint(), + Error::DecodeLogicalPlan { source, .. } => source.retry_hint(), + Error::Delete { source, .. } => source.retry_hint(), + Error::AsyncTaskExecute { source, .. } => source.retry_hint(), + Error::StartServer { source, .. } | Error::ShutdownServer { source, .. } => { + source.retry_hint() + } + Error::OpenLogStore { source, .. } => source.retry_hint(), + Error::MetaClientInit { source, .. } => source.retry_hint(), + Error::HandleRegionRequest { source, .. } + | Error::GetRegionMetadata { source, .. } + | Error::HandleBatchOpenRequest { source, .. } + | Error::HandleBatchDdlRequest { source, .. } + | Error::StopRegionEngine { source, .. } => source.retry_hint(), + Error::FindLogicalRegions { source, .. } => source.retry_hint(), + Error::BuildMitoEngine { source, .. } => source.retry_hint(), + Error::GcMitoEngine { source, .. } => source.retry_hint(), + Error::BuildMetricEngine { source, .. } => source.retry_hint(), + Error::ListStorageSsts { source, .. } => source.retry_hint(), + Error::ObjectStore { source, .. } => source.retry_hint(), _ => RetryHint::NonRetryable, } } @@ -602,14 +604,28 @@ mod tests { use super::*; #[test] - fn test_region_state_hints_are_retryable() { + fn test_region_state_retry_hints() { let region_id = RegionId::new(1024, 1); - - let err = RegionBusySnafu { region_id }.build(); - assert_eq!(err.retry_hint(), RetryHint::Retryable); - - let err = RegionNotReadySnafu { region_id }.build(); - assert_eq!(err.retry_hint(), RetryHint::Retryable); + for (error, status, hint) in [ + ( + RegionBusySnafu { region_id }.build(), + StatusCode::RegionBusy, + RetryHint::Retryable, + ), + ( + RegionNotReadySnafu { region_id }.build(), + StatusCode::RegionNotReady, + RetryHint::RetryAfterRouteRefresh, + ), + ( + RegionNotFoundSnafu { region_id }.build(), + StatusCode::RegionNotFound, + RetryHint::RetryAfterRouteRefresh, + ), + ] { + assert_eq!(error.status_code(), status); + assert_eq!(error.retry_hint(), hint); + } } #[test] diff --git a/src/metric-engine/src/error.rs b/src/metric-engine/src/error.rs index 0803223ba33..25bc76b7e55 100644 --- a/src/metric-engine/src/error.rs +++ b/src/metric-engine/src/error.rs @@ -483,30 +483,48 @@ impl ErrorExt for Error { } fn retry_hint(&self) -> RetryHint { - use Error::*; - match self { - CreateMitoRegion { source, .. } - | OpenMitoRegion { source, .. } - | CleanUpMitoRegion { source, .. } - | BatchOpenMitoRegion { source, .. } - | BatchCatchupMitoRegion { source, .. } - | CloseMitoRegion { source, .. } - | MitoReadOperation { source, .. } - | MitoWriteOperation { source, .. } - | MitoFlushOperation { source, .. } - | MitoTruncateOperation { source, .. } - | MitoSyncOperation { source, .. } - | MitoEnterStagingOperation { source, .. } - | MitoCopyRegionFromOperation { source, .. } - | MitoEditRegion { source, .. } => source.retry_hint(), + Error::PhysicalRegionNotFound { .. } | Error::LogicalRegionNotFound { .. } => { + RetryHint::RetryAfterRouteRefresh + } + Error::CreateMitoRegion { source, .. } + | Error::OpenMitoRegion { source, .. } + | Error::CleanUpMitoRegion { source, .. } + | Error::BatchOpenMitoRegion { source, .. } + | Error::BatchCatchupMitoRegion { source, .. } + | Error::CloseMitoRegion { source, .. } + | Error::MitoReadOperation { source, .. } + | Error::MitoWriteOperation { source, .. } + | Error::MitoFlushOperation { source, .. } + | Error::MitoTruncateOperation { source, .. } + | Error::MitoSyncOperation { source, .. } + | Error::MitoEnterStagingOperation { source, .. } + | Error::MitoCopyRegionFromOperation { source, .. } + | Error::MitoEditRegion { source, .. } => source.retry_hint(), - EncodePrimaryKey { source, .. } => source.retry_hint(), - CollectRecordBatchStream { source, .. } => source.retry_hint(), - StartRepeatedTask { source, .. } => source.retry_hint(), - CacheGet { source, .. } => source.retry_hint(), + Error::EncodePrimaryKey { source, .. } => source.retry_hint(), + Error::CollectRecordBatchStream { source, .. } => source.retry_hint(), + Error::StartRepeatedTask { source, .. } => source.retry_hint(), + Error::CacheGet { source, .. } => source.retry_hint(), _ => RetryHint::NonRetryable, } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_missing_region_retry_hints() { + let region_id = RegionId::new(1024, 1); + for error in [ + PhysicalRegionNotFoundSnafu { region_id }.build(), + LogicalRegionNotFoundSnafu { region_id }.build(), + ] { + assert_eq!(error.status_code(), StatusCode::RegionNotFound); + assert_eq!(error.retry_hint(), RetryHint::RetryAfterRouteRefresh); + } + } +} diff --git a/src/mito2/src/error.rs b/src/mito2/src/error.rs index f738044c9ef..0066b8db47c 100644 --- a/src/mito2/src/error.rs +++ b/src/mito2/src/error.rs @@ -1636,79 +1636,83 @@ impl ErrorExt for Error { } fn retry_hint(&self) -> RetryHint { - use Error::*; - match self { - ReadParquet { .. } - | WriteParquet { .. } - | RejectWrite { .. } - | Download { .. } - | Upload { .. } - | RegionState { .. } - | UpdateManifest { .. } - | RegionStopped { .. } - | RegionBusy { .. } - | ManualCompactionAlreadyRunning { .. } - | FlushableRegionState { .. } - | ManifestDeltaNotFound { .. } => RetryHint::Retryable, + Error::PartitionExprVersionMismatch { .. } + | Error::RegionNotFound { .. } + | Error::RejectWrite { .. } + | Error::RegionState { + state: RegionRoleState::Follower, + .. + } => RetryHint::RetryAfterRouteRefresh, + Error::ReadParquet { .. } + | Error::WriteParquet { .. } + | Error::Download { .. } + | Error::Upload { .. } + | Error::RegionState { .. } + | Error::UpdateManifest { .. } + | Error::RegionStopped { .. } + | Error::RegionBusy { .. } + | Error::ManualCompactionAlreadyRunning { .. } + | Error::FlushableRegionState { .. } + | Error::ManifestDeltaNotFound { .. } => RetryHint::Retryable, - OpenDal { error, .. } - | DeleteSsts { error, .. } - | DeleteIndex { error, .. } - | DeleteIndexes { error, .. } => retry_hint_from_opendal_error(error), + Error::OpenDal { error, .. } + | Error::DeleteSsts { error, .. } + | Error::DeleteIndex { error, .. } + | Error::DeleteIndexes { error, .. } => retry_hint_from_opendal_error(error), - WriteWal { source, .. } - | ReadWal { source, .. } - | DeleteWal { source, .. } - | WaitWalDurable { source, .. } - | FetchManifests { source, .. } - | External { source, .. } => source.retry_hint(), + Error::WriteWal { source, .. } + | Error::ReadWal { source, .. } + | Error::DeleteWal { source, .. } + | Error::WaitWalDurable { source, .. } + | Error::FetchManifests { source, .. } + | Error::External { source, .. } => source.retry_hint(), - OpenRegion { source, .. } - | WriteGroup { source, .. } - | FlushRegion { source, .. } - | BuildIndexAsync { source, .. } - | CompactRegion { source, .. } - | EditRegion { source, .. } - | ScanSeries { source, .. } - | PruneFile { source, .. } => source.retry_hint(), + Error::OpenRegion { source, .. } + | Error::WriteGroup { source, .. } + | Error::FlushRegion { source, .. } + | Error::BuildIndexAsync { source, .. } + | Error::CompactRegion { source, .. } + | Error::EditRegion { source, .. } + | Error::ScanSeries { source, .. } + | Error::PruneFile { source, .. } => source.retry_hint(), - DataTypeMismatch { source, .. } - | ConvertVector { source, .. } - | ConvertValue { source, .. } - | IndexOptions { source, .. } - | CastVector { source, .. } => source.retry_hint(), + Error::DataTypeMismatch { source, .. } + | Error::ConvertVector { source, .. } + | Error::ConvertValue { source, .. } + | Error::IndexOptions { source, .. } + | Error::CastVector { source, .. } => source.retry_hint(), - BuildIndexApplier { source, .. } - | PushIndexValue { source, .. } - | ApplyInvertedIndex { source, .. } - | IndexFinish { source, .. } => source.retry_hint(), + Error::BuildIndexApplier { source, .. } + | Error::PushIndexValue { source, .. } + | Error::ApplyInvertedIndex { source, .. } + | Error::IndexFinish { source, .. } => source.retry_hint(), - ApplyBloomFilterIndex { source, .. } - | PushBloomFilterValue { source, .. } - | BloomFilterFinish { source, .. } => source.retry_hint(), + Error::ApplyBloomFilterIndex { source, .. } + | Error::PushBloomFilterValue { source, .. } + | Error::BloomFilterFinish { source, .. } => source.retry_hint(), - PuffinReadBlob { source, .. } - | PuffinAddBlob { source, .. } - | PuffinInitStager { source, .. } - | PuffinBuildReader { source, .. } - | PuffinPurgeStager { source, .. } => source.retry_hint(), + Error::PuffinReadBlob { source, .. } + | Error::PuffinAddBlob { source, .. } + | Error::PuffinInitStager { source, .. } + | Error::PuffinBuildReader { source, .. } + | Error::PuffinPurgeStager { source, .. } => source.retry_hint(), - CreateFulltextCreator { source, .. } - | FulltextPushText { source, .. } - | FulltextFinish { source, .. } - | ApplyFulltextIndex { source, .. } => source.retry_hint(), + Error::CreateFulltextCreator { source, .. } + | Error::FulltextPushText { source, .. } + | Error::FulltextFinish { source, .. } + | Error::ApplyFulltextIndex { source, .. } => source.retry_hint(), - InvalidRegionRequest { source, .. } => source.retry_hint(), - InvalidPartitionExpr { source, .. } => source.retry_hint(), - RecordBatch { source, .. } => source.retry_hint(), - GetSchemaMetadata { source, .. } => source.retry_hint(), - CompactionMemoryExhausted { source, .. } => source.retry_hint(), - ConvertBulkWalEntry { source, .. } => source.retry_hint(), - Encode { source, .. } | Decode { source, .. } => source.retry_hint(), + Error::InvalidRegionRequest { source, .. } => source.retry_hint(), + Error::InvalidPartitionExpr { source, .. } => source.retry_hint(), + Error::RecordBatch { source, .. } => source.retry_hint(), + Error::GetSchemaMetadata { source, .. } => source.retry_hint(), + Error::CompactionMemoryExhausted { source, .. } => source.retry_hint(), + Error::ConvertBulkWalEntry { source, .. } => source.retry_hint(), + Error::Encode { source, .. } | Error::Decode { source, .. } => source.retry_hint(), #[cfg(feature = "enterprise")] - ScanExternalRange { source, .. } => source.retry_hint(), + Error::ScanExternalRange { source, .. } => source.retry_hint(), _ => RetryHint::NonRetryable, } @@ -1721,6 +1725,55 @@ mod tests { use super::*; + #[test] + fn test_partition_version_mismatch_retry_hint() { + let error = PartitionExprVersionMismatchSnafu { + region_id: RegionId::new(1, 1), + request_version: 1u64, + expected_version: 2u64, + } + .build(); + assert_eq!(error.status_code(), StatusCode::InvalidArguments); + assert_eq!(error.retry_hint(), RetryHint::RetryAfterRouteRefresh); + } + + #[test] + fn test_region_state_retry_hint() { + let region_id = RegionId::new(1, 1); + let writable = RegionRoleState::Leader(crate::region::RegionLeaderState::Writable); + let follower_error = RegionStateSnafu { + region_id, + state: RegionRoleState::Follower, + expect: writable, + } + .build(); + assert_eq!( + follower_error.retry_hint(), + RetryHint::RetryAfterRouteRefresh + ); + assert_eq!(follower_error.status_code(), StatusCode::RegionNotReady); + let leader_error = RegionStateSnafu { + region_id, + state: writable, + expect: RegionRoleState::Follower, + } + .build(); + assert_eq!(leader_error.retry_hint(), RetryHint::Retryable); + for (error, status) in [ + ( + RejectWriteSnafu { region_id }.build(), + StatusCode::StorageUnavailable, + ), + ( + RegionNotFoundSnafu { region_id }.build(), + StatusCode::RegionNotFound, + ), + ] { + assert_eq!(error.status_code(), status); + assert_eq!(error.retry_hint(), RetryHint::RetryAfterRouteRefresh); + } + } + #[test] fn test_manifest_update_persistence() { let rejected_kinds = [ diff --git a/tests/compatibility/README.md b/tests/compatibility/README.md index 404b6332b94..51c87698ff2 100644 --- a/tests/compatibility/README.md +++ b/tests/compatibility/README.md @@ -6,7 +6,7 @@ SQL restart tests are run via `cargo sqlness compat` and reuse the sqlness-runne ## Wire fixtures -Frozen protobuf fixtures are exercised by the Rust tests that own their context +Frozen wire fixtures are exercised by the Rust tests that own their context conversion, not by SQLness case discovery. The [legacy query-context fixture](wire/read_preference/README.md) checks omitted read-preference defaulting and unrelated metadata preservation: @@ -15,6 +15,13 @@ read-preference defaulting and unrelated metadata preservation: cargo nextest run -p session -E 'test(test_legacy_query_context_wire_defaults_to_leader)' ``` +The [write-retry metadata fixture](wire/write_retry/README.md) checks legacy, +route-refresh, missing, and unknown retry-hint decoding in the client: + +```shell +cargo nextest run -p client -E 'test(test_retry_hint_wire_compatibility)' +``` + ## Quick Start ```shell diff --git a/tests/compatibility/wire/write_retry/README.md b/tests/compatibility/wire/write_retry/README.md new file mode 100644 index 00000000000..8b6499ed5a8 --- /dev/null +++ b/tests/compatibility/wire/write_retry/README.md @@ -0,0 +1,25 @@ +# Write-retry metadata compatibility + +`retry_hints.json` freezes gRPC `x-greptime-err-retry-hint` metadata values +and the expected current client decoding. These are constructed protocol fixtures, +not captures from released binaries. + +The client test converts real `tonic::Status` metadata through the production +error decoder. Missing and unknown values default to `NonRetryable`; existing +`retryable` and `non_retryable` values retain their meaning; the new +`retry_after_route_refresh` value preserves its distinct hint. The application +status code and gRPC status code remain unchanged. + +```shell +cargo nextest run -p client -E 'test(test_retry_hint_wire_compatibility)' +``` + +`test_transport_error_parity_and_datanode_mapping` separately exercises these +recognized values over an actual gRPC client/server connection. + +An older decoder that recognizes only the two legacy strings treats the new +string as unknown and uses its existing `NonRetryable` fallback. This is a source +compatibility observation, not a claim that an old binary was run. These tests do +not promise that an old client will refresh routing or perform automatic retries. + +The fixture is hand-authored and must not be regenerated from test output. diff --git a/tests/compatibility/wire/write_retry/retry_hints.json b/tests/compatibility/wire/write_retry/retry_hints.json new file mode 100644 index 00000000000..d24da878b3e --- /dev/null +++ b/tests/compatibility/wire/write_retry/retry_hints.json @@ -0,0 +1,7 @@ +[ + {"name": "legacy_missing", "wire_hint": null, "expected_hint": "non_retryable"}, + {"name": "legacy_retryable", "wire_hint": "retryable", "expected_hint": "retryable"}, + {"name": "legacy_non_retryable", "wire_hint": "non_retryable", "expected_hint": "non_retryable"}, + {"name": "route_refresh", "wire_hint": "retry_after_route_refresh", "expected_hint": "retry_after_route_refresh"}, + {"name": "unknown_future_hint", "wire_hint": "future_retry_hint", "expected_hint": "non_retryable"} +]