rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-20 19:02:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: propagate route refresh retry hints

Browse Source 
Signed-off-by: WenyXu <wenymedia@gmail.com>
This commit is contained in:
WenyXu
2026-10-12 10:58:23 +00:00
parent 690872b62a
commit 2f5832dfa0
15 changed files with 446 additions and 229 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -2115,6 +2115,7 @@ dependencies = [
"prost 0.14.1",
"query",
"rand 0.9.4",
"serde",
"serde_json",
"snafu 0.8.6",
"store-api",
src/client/Cargo.toml
+3
View File
@@ -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
src/client/src/client.rs
+1
View File
@@ -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<Channel>)> {
let (addr, channel) = self.find_channel()?;
let client = tonic::client::Grpc::new(channel)
src/client/src/client_manager.rs
+13 -12
View File
@@ -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 {
src/client/src/error.rs
+14 -1
View File
@@ -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<Error> 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<Code> {
src/client/src/region.rs
+102 -56
View File
@@ -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<RegionResponse> {
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<SendableRecordBatchStream> {
@@ -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<RegionRequest>`]. 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<R>(&self, request: R) -> Result<RegionResponse>
where
R: Borrow<RegionRequest> + 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<String>,
update: RemoteDynFilterUpdate,
) -> Result<RegionResponse> {
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<String>,
unregister: RemoteDynFilterUnregister,
) -> Result<RegionResponse> {
self.handle_inner(build_remote_dyn_filter_unregister_request(
query_id, unregister,
))
.await
}
async fn handle_inner<R>(&self, request: R) -> Result<RegionResponse>
where
R: Borrow<RegionRequest> + 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<RegionRequest>`]. 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<R>(&self, request: R) -> Result<RegionResponse>
where
R: Borrow<RegionRequest> + Send + Sync + 'static,
{
self.handle_inner(request).await
}
pub async fn handle_remote_dyn_filter_update(
&self,
query_id: impl Into<String>,
update: RemoteDynFilterUpdate,
) -> Result<RegionResponse> {
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<String>,
unregister: RemoteDynFilterUnregister,
) -> Result<RegionResponse> {
self.handle_inner(build_remote_dyn_filter_unregister_request(
query_id, unregister,
))
.await
}
}
async fn recordbatches_from_flight_message_stream<S>(
@@ -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<String>,
expected_hint: String,
}
let fixtures: Vec<Fixture> = 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);
src/common/error/src/ext.rs
+16 -2
View File
@@ -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<Self, Self::Err> {
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
}
src/common/error/tests/ext.rs
+9
View File
@@ -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");
src/common/meta/src/error.rs
+47 -44
View File
@@ -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,
}
}
src/datanode/src/error.rs
+48 -32
View File
@@ -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]
src/metric-engine/src/error.rs
+38 -20
View File
@@ -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);
}
}
}
src/mito2/src/error.rs
+114 -61
View File
@@ -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 = [
tests/compatibility/README.md
+8 -1
View File
@@ -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
tests/compatibility/wire/write_retry/README.md
+25
View File
@@ -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.
tests/compatibility/wire/write_retry/retry_hints.json
+7
View File
@@ -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"}
]