From 684c8ff2917c10dcf8103664f5eb2b01ef6720e9 Mon Sep 17 00:00:00 2001 From: Weny Xu Date: Tue, 13 Oct 2026 02:29:54 +0000 Subject: [PATCH] feat(client): support region request reuse and route refresh hints (#9540) * feat(client): support shared region requests without deep cloning Signed-off-by: WenyXu * feat: propagate route refresh retry hints Signed-off-by: WenyXu * style: preserve local error variant imports in retry hints Signed-off-by: WenyXu * fix(mito2): distinguish staging rejection from write backpressure Signed-off-by: WenyXu * fix(meta): preserve legacy retry hints in instruction replies Signed-off-by: WenyXu * test(client): inline retry hint compatibility cases Signed-off-by: WenyXu * test(datanode): align batch-open retry hint assertion Signed-off-by: WenyXu * style: use imported types in region request tests Signed-off-by: WenyXu --------- Signed-off-by: WenyXu --- Cargo.lock | 1 + Cargo.toml | 1 + src/client/Cargo.toml | 1 + src/client/src/client.rs | 22 +- src/client/src/client_manager.rs | 25 +- src/client/src/error.rs | 15 +- src/client/src/region.rs | 616 ++++++++++++++++++++++----- src/client/src/region/codec.rs | 63 +++ src/common/error/src/ext.rs | 18 +- src/common/error/tests/ext.rs | 9 + src/common/meta/src/error.rs | 7 +- src/common/meta/src/instruction.rs | 34 +- src/datanode/src/error.rs | 30 +- src/datanode/src/region_server.rs | 2 +- src/metric-engine/src/error.rs | 20 + src/mito2/src/engine/flush_test.rs | 2 + src/mito2/src/engine/staging_test.rs | 73 +++- src/mito2/src/error.rs | 73 +++- src/mito2/src/worker/handle_write.rs | 14 +- 19 files changed, 850 insertions(+), 176 deletions(-) create mode 100644 src/client/src/region/codec.rs diff --git a/Cargo.lock b/Cargo.lock index 61a9c246ed1..f75bde16328 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2122,6 +2122,7 @@ dependencies = [ "tokio", "tokio-stream", "tonic 0.14.2", + "tonic-prost", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index f47907aa797..702498ee5e5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -272,6 +272,7 @@ url = "2.3" # Pin below 0.14.5 to avoid the `connection_timeout_future` "async fn resumed after # completion" panic (greptimedb#8265 / grpc-rust#2522). Remove once tonic ships a fix. tonic = { version = "0.14, <0.14.5", features = ["tls-aws-lc", "gzip", "zstd"] } +tonic-prost = "0.14" tower = "0.5" tracing = "0.1" tracing-appender = "0.2" diff --git a/src/client/Cargo.toml b/src/client/Cargo.toml index 36c3f60a9a4..e5bd479cc7d 100644 --- a/src/client/Cargo.toml +++ b/src/client/Cargo.toml @@ -45,3 +45,4 @@ substrait.workspace = true tokio.workspace = true tokio-stream = { workspace = true, features = ["net"] } tonic.workspace = true +tonic-prost.workspace = true diff --git a/src/client/src/client.rs b/src/client/src/client.rs index dfc0e44fad5..1746854cd26 100644 --- a/src/client/src/client.rs +++ b/src/client/src/client.rs @@ -20,7 +20,6 @@ use api::v1::HealthCheckRequest; use api::v1::flow::flow_client::FlowClient as PbFlowClient; use api::v1::health_check_client::HealthCheckClient; use api::v1::prometheus_gateway_client::PrometheusGatewayClient; -use api::v1::region::region_client::RegionClient as PbRegionClient; use arrow_flight::flight_service_client::FlightServiceClient; use common_grpc::channel_manager::{ ChannelConfig, ChannelManager, ClientTlsOption, load_client_tls_config, @@ -526,23 +525,12 @@ impl Client { Ok(FlightClient { addr, client }) } - pub(crate) fn raw_region_client(&self) -> Result<(String, PbRegionClient)> { + /// 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 = PbRegionClient::new(channel) - .max_decoding_message_size( - self.inner - .control_channel_manager - .config() - .max_recv_message_size - .as_bytes() as usize, - ) - .max_encoding_message_size( - self.inner - .control_channel_manager - .config() - .max_send_message_size - .as_bytes() as usize, - ); + let client = tonic::client::Grpc::new(channel) + .max_decoding_message_size(self.max_grpc_recv_message_size()) + .max_encoding_message_size(self.max_grpc_send_message_size()); Ok((addr, client)) } 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 fb2e4d10ca1..47c477b06b0 100644 --- a/src/client/src/region.rs +++ b/src/client/src/region.rs @@ -12,6 +12,9 @@ // See the License for the specific language governing permissions and // limitations under the License. +mod codec; + +use std::borrow::Borrow; use std::sync::Arc; use std::time::Duration; @@ -49,10 +52,12 @@ use crate::error::{ MissingFieldSnafu, Result, ServerSnafu, }; use crate::flight::{FlightMessageReader, decode_flight_data}; +use crate::region::codec::RegionCodec; 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, @@ -63,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 { @@ -144,11 +137,49 @@ impl RegionRequester { recordbatches_from_flight_message_stream(addr, flight_message_stream).await } - async fn handle_inner(&self, request: RegionRequest) -> Result { + /// 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, + { let request_body = request + .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(_) @@ -168,49 +199,43 @@ impl RegionRequester { } } - let response = client - .handle(request) - .await - .map_err(|e| { - let code = e.code(); - // Uses `Error::RegionServer` instead of `Error::Server` - error::Error::RegionServer { - addr, - code, - source: BoxedError::new(error::Error::from(e)), - location: location!(), - } - })? - .into_inner(); + let response = async { + client + .ready() + .await + .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {e}")))?; + let mut request = tonic::Request::new(request); + request.extensions_mut().insert(tonic::GrpcMethod::new( + "greptime.v1.region.Region", + "Handle", + )); + client + .unary( + request, + tonic::codegen::http::uri::PathAndQuery::from_static( + "/greptime.v1.region.Region/Handle", + ), + RegionCodec::default(), + ) + .await + } + .await + .map_err(|e| { + let code = e.code(); + // Uses `Error::RegionServer` instead of `Error::Server` + error::Error::RegionServer { + addr, + code, + source: BoxedError::new(error::Error::from(e)), + location: location!(), + } + })? + .into_inner(); check_response_header(&response.header)?; Ok(RegionResponse::from_region_response(response)) } - - pub async fn handle(&self, request: RegionRequest) -> Result { - 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( @@ -385,13 +410,22 @@ pub fn check_response_header(header: &Option) -> Result<()> { #[cfg(test)] #[allow(deprecated)] -mod test { - use api::v1::Status as PbStatus; +mod tests { + use std::collections::VecDeque; + use std::sync::Mutex; + use api::v1::region::region_server::{Region, RegionServer}; use api::v1::region::{ - BulkInsertRequest, RegionResponse as PbRegionResponse, RemoteDynFilterUnregister, - RemoteDynFilterUpdate, region_request, remote_dyn_filter_request, + BulkInsertRequest, InsertRequest, InsertRequests, RegionResponse as PbRegionResponse, + RemoteDynFilterUnregister, RemoteDynFilterUpdate, region_request, + remote_dyn_filter_request, }; + use api::v1::value::ValueData; + use api::v1::{ + ColumnDataType, ColumnSchema as PbColumnSchema, Row, Rows, Status as PbStatus, Value, + }; + use common_error::ext::RetryHint; + use common_grpc::channel_manager::{ChannelConfig, ChannelManager}; use common_recordbatch::adapter::RecordBatchMetrics; use datatypes::arrow::array::Int32Array; use datatypes::prelude::{ConcreteDataType, VectorRef}; @@ -399,74 +433,440 @@ mod test { use datatypes::vectors::Int32Vector; use futures_util::stream; use tokio::net::TcpListener; + use tokio::task::JoinHandle; use tokio_stream::wrappers::TcpListenerStream; use tonic::codec::CompressionEncoding; use tonic::{Request, Response, Status}; use super::*; - use crate::Error::{self, IllegalDatabaseResponse, Server}; + use crate::Error; - #[derive(Clone)] - struct CompressionRecordingRegionService { - zstd_headers: Arc>>, + #[derive(Clone, Default)] + struct RecordingRegionService { + requests: Arc>>, + responses: Arc>>>, } #[tonic::async_trait] - impl Region for CompressionRecordingRegionService { + impl Region for RecordingRegionService { async fn handle( &self, request: Request, ) -> std::result::Result, Status> { let metadata = request.metadata(); - let sends_zstd = metadata - .get("grpc-encoding") - .and_then(|v| v.to_str().ok()) - .is_some_and(|value| value == "zstd"); + let sends_zstd = metadata.get("grpc-encoding").is_some_and(|v| v == "zstd"); let accepts_zstd = metadata .get("grpc-accept-encoding") - .and_then(|value| value.to_str().ok()) - .is_some_and(|value| value.split(',').any(|encoding| encoding == "zstd")); - self.zstd_headers + .and_then(|v| v.to_str().ok()) + .is_some_and(|v| v.split(',').any(|encoding| encoding == "zstd")); + self.requests .lock() .unwrap() - .push((sends_zstd, accepts_zstd)); + .push((request.into_inner(), sends_zstd, accepts_zstd)); + self.responses + .lock() + .unwrap() + .pop_front() + .unwrap_or_else(|| Ok(build_response(StatusCode::Success))) + .map(Response::new) + } + } - Ok(Response::new(PbRegionResponse { - header: Some(ResponseHeader { - status: Some(PbStatus { - status_code: StatusCode::Success as u32, - ..Default::default() - }), + struct RegionTestServer { + addr: String, + service: RecordingRegionService, + task: JoinHandle<()>, + } + + impl RegionTestServer { + async fn start() -> Self { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + let service = RecordingRegionService::default(); + let server_service = service.clone(); + let task = tokio::spawn(async move { + tonic::transport::Server::builder() + .add_service( + RegionServer::new(server_service) + .accept_compressed(CompressionEncoding::Zstd) + .send_compressed(CompressionEncoding::Zstd), + ) + .serve_with_incoming(TcpListenerStream::new(listener)) + .await + .unwrap(); + }); + Self { + addr, + service, + task, + } + } + + fn create_requester( + &self, + config: ChannelConfig, + send: bool, + accept: bool, + ) -> RegionRequester { + RegionRequester::new( + Client::with_manager_and_urls( + ChannelManager::with_config(config, None), + [self.addr.clone()], + ), + send, + accept, + ) + } + } + + impl Drop for RegionTestServer { + fn drop(&mut self) { + self.task.abort(); + } + } + + fn build_response(code: StatusCode) -> PbRegionResponse { + PbRegionResponse { + header: Some(ResponseHeader { + status: Some(PbStatus { + status_code: code as u32, + ..Default::default() }), + }), + affected_rows: 1, + ..Default::default() + } + } + + fn build_insert_request() -> RegionRequest { + RegionRequest { + header: Some(RegionRequestHeader { + dbname: "shared-request-test".to_string(), ..Default::default() - })) + }), + body: Some(region_request::Body::Inserts(InsertRequests { + requests: vec![InsertRequest { + region_id: 42, + rows: Some(Rows { + schema: vec![ + PbColumnSchema { + column_name: "message".to_string(), + datatype: ColumnDataType::String as i32, + ..Default::default() + }, + PbColumnSchema { + column_name: "payload".to_string(), + datatype: ColumnDataType::Binary as i32, + ..Default::default() + }, + ], + rows: vec![Row { + values: vec![ + Value { + value_data: Some(ValueData::StringValue( + "hello 世界".to_string(), + )), + }, + Value { + value_data: Some(ValueData::BinaryValue(vec![0, 1, 127, 255])), + }, + ], + }], + }), + ..Default::default() + }], + })), } } #[tokio::test] - async fn test_inserts_and_bulk_insert_use_transport_compression() { - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); - let zstd_headers = Arc::new(std::sync::Mutex::new(Vec::new())); - let service = CompressionRecordingRegionService { - zstd_headers: zstd_headers.clone(), - }; - let server = tokio::spawn(async move { - tonic::transport::Server::builder() - .add_service( - RegionServer::new(service) - .accept_compressed(CompressionEncoding::Zstd) - .send_compressed(CompressionEncoding::Zstd), - ) - .serve_with_incoming(TcpListenerStream::new(listener)) + async fn test_owned_and_shared_requests_preserve_payload_and_compression() { + let server = RegionTestServer::start().await; + let shared = Arc::new(build_insert_request()); + for (send, accept) in [(false, false), (true, false), (false, true), (true, true)] { + let requester = server.create_requester(ChannelConfig::default(), send, accept); + let channel = + tonic::transport::Endpoint::from_shared(format!("http://{}", server.addr)) + .unwrap() + .connect() + .await + .unwrap(); + let mut baseline = api::v1::region::region_client::RegionClient::new(channel); + if send { + baseline = baseline.send_compressed(CompressionEncoding::Zstd); + } + if accept { + baseline = baseline.accept_compressed(CompressionEncoding::Zstd); + } + assert_eq!( + baseline + .handle(build_insert_request()) + .await + .unwrap() + .into_inner() + .affected_rows, + 1 + ); + assert_eq!( + requester + .handle(build_insert_request()) + .await + .unwrap() + .affected_rows, + 1 + ); + assert_eq!( + requester + .handle(Arc::clone(&shared)) + .await + .unwrap() + .affected_rows, + 1 + ); + } + let requests = server.service.requests.lock().unwrap(); + assert_eq!(requests.len(), 12); + for ((request, sends, accepts), (send, accept)) in requests.iter().zip( + [(false, false), (true, false), (false, true), (true, true)] + .into_iter() + .flat_map(|settings| [settings, settings, settings]), + ) { + assert_eq!(request, shared.as_ref()); + assert_eq!((*sends, *accepts), (send, accept)); + } + } + + #[tokio::test] + async fn test_shared_request_can_be_resent_after_transport_error() { + let server = RegionTestServer::start().await; + let mut status = Status::unavailable("migrating"); + status.metadata_mut().insert( + common_error::GREPTIME_DB_HEADER_ERROR_RETRY_HINT, + RetryHint::Retryable.as_str().parse().unwrap(), + ); + server + .service + .responses + .lock() + .unwrap() + .push_back(Err(status)); + let requester = server.create_requester(ChannelConfig::default(), false, false); + let shared = Arc::new(build_insert_request()); + let error = requester.handle(Arc::clone(&shared)).await.unwrap_err(); + assert_eq!(error.retry_hint(), RetryHint::Retryable); + assert!( + matches!(error, Error::RegionServer { addr, code: tonic::Code::Unavailable, .. } if addr == server.addr) + ); + requester.handle(Arc::clone(&shared)).await.unwrap(); + let requests = server.service.requests.lock().unwrap(); + assert_eq!(requests.len(), 2); + assert_eq!(&requests[0].0, shared.as_ref()); + assert_eq!(&requests[1].0, shared.as_ref()); + } + + #[test] + fn test_retry_hint_wire_compatibility() { + for (name, wire_hint, expected_hint) in [ + ("legacy_missing", None, RetryHint::NonRetryable), + ("legacy_retryable", Some("retryable"), RetryHint::Retryable), + ( + "legacy_non_retryable", + Some("non_retryable"), + RetryHint::NonRetryable, + ), + ( + "route_refresh", + Some("retry_after_route_refresh"), + RetryHint::RetryAfterRouteRefresh, + ), + ( + "unknown_future_hint", + Some("future_retry_hint"), + RetryHint::NonRetryable, + ), + ] { + 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) = wire_hint { + status.metadata_mut().insert( + common_error::GREPTIME_DB_HEADER_ERROR_RETRY_HINT, + hint.parse().unwrap(), + ); + } + let error: Error = status.into(); + assert_eq!(error.retry_hint(), expected_hint, "{name}"); + assert_eq!(error.status_code(), StatusCode::RegionNotReady, "{name}"); + assert_eq!( + error.tonic_code(), + Some(tonic::Code::FailedPrecondition), + "{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::RetryAfterRouteRefresh, + RetryHint::NonRetryable, + ] { + let mut status = Status::failed_precondition("region unavailable"); + status.metadata_mut().insert( + common_error::GREPTIME_DB_HEADER_ERROR_RETRY_HINT, + hint.as_str().parse().unwrap(), + ); + status.metadata_mut().insert( + common_error::GREPTIME_DB_HEADER_ERROR_CODE, + (StatusCode::RegionNotFound as u32) + .to_string() + .parse() + .unwrap(), + ); + server + .service + .responses + .lock() + .unwrap() + .extend((0..4).map(|_| Err(status.clone()))); + let mut baseline = api::v1::region::region_client::RegionClient::connect(format!( + "http://{}", + server.addr + )) + .await + .unwrap(); + let baseline: Error = baseline + .handle(build_insert_request()) .await - .unwrap(); - }); - let client = Client::with_urls([addr.to_string()]); - let requester = RegionRequester::new(client.clone(), true, true); - let send_only_requester = RegionRequester::new(client.clone(), true, false); - let accept_only_requester = RegionRequester::new(client.clone(), false, true); - let disabled_requester = RegionRequester::new(client, false, false); + .unwrap_err() + .into(); + assert_eq!(baseline.status_code(), StatusCode::RegionNotFound); + assert_eq!(baseline.retry_hint(), hint); + assert_eq!(baseline.tonic_code(), Some(tonic::Code::FailedPrecondition)); + let owned = requester.handle(build_insert_request()).await.unwrap_err(); + let shared = requester + .handle(Arc::new(build_insert_request())) + .await + .unwrap_err(); + for error in [owned, shared] { + assert_eq!(error.status_code(), baseline.status_code()); + assert_eq!(error.retry_hint(), baseline.retry_hint()); + assert_eq!(error.tonic_code(), baseline.tonic_code()); + } + let error = Datanode::handle(&requester, build_insert_request()) + .await + .unwrap_err(); + assert_eq!(error.status_code(), StatusCode::RegionNotFound); + assert_eq!(error.retry_hint(), hint); + match hint { + RetryHint::Retryable | RetryHint::RetryAfterRouteRefresh => assert!(matches!( + error, + meta_error::Error::RetryLater { + clean_poisons: false, + .. + } + )), + RetryHint::NonRetryable => { + assert!(matches!(error, meta_error::Error::External { .. })) + } + } + } + } + + #[tokio::test] + async fn test_shared_request_preserves_response_header_errors() { + let server = RegionTestServer::start().await; + server.service.responses.lock().unwrap().extend([ + Ok(PbRegionResponse::default()), + Ok(build_response(StatusCode::RegionNotFound)), + ]); + let requester = server.create_requester(ChannelConfig::default(), false, false); + let shared = Arc::new(build_insert_request()); + assert!(matches!( + requester.handle(Arc::clone(&shared)).await.unwrap_err(), + Error::IllegalDatabaseResponse { .. } + )); + assert!(matches!( + requester.handle(shared).await.unwrap_err(), + Error::Server { + code: StatusCode::RegionNotFound, + .. + } + )); + } + + #[tokio::test] + async fn test_shared_request_respects_configured_message_limits() { + let server = RegionTestServer::start().await; + let shared = Arc::new(build_insert_request()); + for send_limit in [true, false] { + let mut config = ChannelConfig::default(); + if send_limit { + config.max_send_message_size = "1B".parse().unwrap(); + } else { + config.max_recv_message_size = "1B".parse().unwrap(); + } + let mut baseline = api::v1::region::region_client::RegionClient::connect(format!( + "http://{}", + server.addr + )) + .await + .unwrap() + .max_encoding_message_size(config.max_send_message_size.as_bytes() as usize) + .max_decoding_message_size(config.max_recv_message_size.as_bytes() as usize); + let baseline: Error = baseline + .handle(build_insert_request()) + .await + .unwrap_err() + .into(); + if !send_limit { + assert_eq!( + baseline.tonic_code(), + Some(tonic::Code::OutOfRange), + "{baseline:?}" + ); + } + let requester = server.create_requester(config, false, false); + let error = requester.handle(Arc::clone(&shared)).await.unwrap_err(); + assert_eq!( + error.tonic_code(), + baseline.tonic_code(), + "shared: {error:?}; generated: {baseline:?}" + ); + assert_eq!( + error.status_code(), + baseline.status_code(), + "shared: {error:?}; generated: {baseline:?}" + ); + assert_eq!( + error.retry_hint(), + baseline.retry_hint(), + "shared: {error:?}; generated: {baseline:?}" + ); + assert_eq!( + server.service.requests.lock().unwrap().len(), + if send_limit { 0 } else { 2 } + ); + } + let requester = server.create_requester(ChannelConfig::default(), false, false); + requester.handle(shared).await.unwrap(); + } + + #[tokio::test] + async fn test_inserts_and_bulk_insert_use_transport_compression() { + let server = RegionTestServer::start().await; + let requester = server.create_requester(ChannelConfig::default(), true, true); + let send_only_requester = server.create_requester(ChannelConfig::default(), true, false); + let accept_only_requester = server.create_requester(ChannelConfig::default(), false, true); + let disabled_requester = server.create_requester(ChannelConfig::default(), false, false); let inserts = || RegionRequest { body: Some(region_request::Body::Inserts(Default::default())), ..Default::default() @@ -502,9 +902,15 @@ mod test { (false, false), (false, false) ], - *zstd_headers.lock().unwrap() + server + .service + .requests + .lock() + .unwrap() + .iter() + .map(|(_, send, accept)| (*send, *accept)) + .collect::>() ); - server.abort(); } #[test] @@ -566,13 +972,13 @@ mod test { 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 { @@ -591,7 +997,7 @@ mod test { })); assert!(matches!( result.unwrap_err(), - IllegalDatabaseResponse { .. } + Error::IllegalDatabaseResponse { .. } )); let result = check_response_header(&Some(ResponseHeader { @@ -600,7 +1006,7 @@ mod test { 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/client/src/region/codec.rs b/src/client/src/region/codec.rs new file mode 100644 index 00000000000..9476385df36 --- /dev/null +++ b/src/client/src/region/codec.rs @@ -0,0 +1,63 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::borrow::Borrow; +use std::marker::PhantomData; + +use api::v1::region::{RegionRequest, RegionResponse}; +use prost::Message; +use tonic::Status; +use tonic::codec::{Codec, EncodeBuf, Encoder}; +use tonic_prost::ProstDecoder; + +/// Encodes owned or shared region requests using the existing protobuf wire format. +pub(crate) struct RegionCodec(PhantomData); + +impl Default for RegionCodec { + fn default() -> Self { + Self(PhantomData) + } +} + +impl Codec for RegionCodec +where + R: Borrow + Send + 'static, +{ + type Encode = R; + type Decode = RegionResponse; + type Encoder = RegionEncoder; + type Decoder = ProstDecoder; + + fn encoder(&mut self) -> Self::Encoder { + RegionEncoder(PhantomData) + } + + fn decoder(&mut self) -> Self::Decoder { + ProstDecoder::default() + } +} + +/// Borrows the protobuf payload while consuming its owned or shared container. +pub(crate) struct RegionEncoder(PhantomData); + +impl> Encoder for RegionEncoder { + type Item = R; + type Error = Status; + + fn encode(&mut self, item: R, buf: &mut EncodeBuf<'_>) -> Result<(), Status> { + item.borrow() + .encode(buf) + .map_err(|error| Status::internal(error.to_string())) + } +} 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..297af792799 100644 --- a/src/common/meta/src/error.rs +++ b/src/common/meta/src/error.rs @@ -1346,8 +1346,11 @@ impl ErrorExt for Error { use Error::*; match self { - RetryLater { .. } - | GetLatestCacheRetryExceeded { .. } + RetryLater { source, .. } => match source.retry_hint() { + RetryHint::RetryAfterRouteRefresh => RetryHint::RetryAfterRouteRefresh, + RetryHint::Retryable | RetryHint::NonRetryable => RetryHint::Retryable, + }, + GetLatestCacheRetryExceeded { .. } | NoLeader { .. } | ElectionNoLeader { .. } | ElectionLeaderLeaseExpired { .. } diff --git a/src/common/meta/src/instruction.rs b/src/common/meta/src/instruction.rs index 6ae074b0ac9..827f06e944d 100644 --- a/src/common/meta/src/instruction.rs +++ b/src/common/meta/src/instruction.rs @@ -49,7 +49,7 @@ pub struct InstructionError { pub message: String, /// Retry hint serialized as a string. #[serde( - serialize_with = "RetryHint::serialize_as_str", + serialize_with = "InstructionError::serialize_retry_hint", deserialize_with = "RetryHint::deserialize_from_str" )] pub retry_hint: RetryHint, @@ -108,6 +108,18 @@ impl InstructionError { retry_hint: error.retry_hint(), } } + + /// Keeps instruction replies readable by peers that only know the legacy hints. + fn serialize_retry_hint(hint: &RetryHint, serializer: S) -> Result + where + S: Serializer, + { + let hint = match hint { + RetryHint::RetryAfterRouteRefresh => RetryHint::Retryable, + hint => *hint, + }; + RetryHint::serialize_as_str(&hint, serializer) + } } impl Display for InstructionError { @@ -1450,6 +1462,26 @@ mod tests { ); } + #[test] + fn test_instruction_error_route_refresh_compatibility() { + let error = InstructionError::new( + StatusCode::RegionNotFound, + "region not found", + RetryHint::RetryAfterRouteRefresh, + ); + let serialized = serde_json::to_string(&error).unwrap(); + assert_eq!( + r#"{"code":4005,"message":"region not found","retry_hint":"retryable"}"#, + serialized + ); + let decoded: InstructionError = serde_json::from_str(&serialized).unwrap(); + assert_eq!(decoded.code, error.code); + assert_eq!(decoded.message, error.message); + assert_eq!(decoded.retry_hint, RetryHint::Retryable); + assert_eq!(error.retry_hint, RetryHint::RetryAfterRouteRefresh); + assert_eq!(error.retry_hint.as_str(), "retry_after_route_refresh"); + } + #[test] fn test_instruction_error_from_error() { let error = MockError::new(StatusCode::RegionNotFound); diff --git a/src/datanode/src/error.rs b/src/datanode/src/error.rs index e9b5d5ec1dc..3f61d4b3d76 100644 --- a/src/datanode/src/error.rs +++ b/src/datanode/src/error.rs @@ -561,8 +561,8 @@ impl ErrorExt for Error { use Error::*; match self { + RegionNotReady { .. } | RegionNotFound { .. } => RetryHint::RetryAfterRouteRefresh, RegionBusy { .. } - | RegionNotReady { .. } | ConcurrentQueryLimiterClosed { .. } | ConcurrentQueryLimiterTimeout { .. } => RetryHint::Retryable, NewPlanDecoder { source, .. } | ExecuteLogicalPlan { source, .. } => { @@ -602,14 +602,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/datanode/src/region_server.rs b/src/datanode/src/region_server.rs index ef333cff04d..a1ccc382ef1 100644 --- a/src/datanode/src/region_server.rs +++ b/src/datanode/src/region_server.rs @@ -2589,7 +2589,7 @@ mod tests { .unwrap_err(); assert_matches!(&err, error::Error::HandleBatchOpenRequest { .. }); assert_eq!(err.status_code(), StatusCode::RegionNotFound); - assert_eq!(err.retry_hint(), RetryHint::NonRetryable); + assert_eq!(err.retry_hint(), RetryHint::RetryAfterRouteRefresh); } struct CurrentEngineTest { diff --git a/src/metric-engine/src/error.rs b/src/metric-engine/src/error.rs index 0803223ba33..f9d61d6af31 100644 --- a/src/metric-engine/src/error.rs +++ b/src/metric-engine/src/error.rs @@ -486,6 +486,9 @@ impl ErrorExt for Error { use Error::*; match self { + PhysicalRegionNotFound { .. } | LogicalRegionNotFound { .. } => { + RetryHint::RetryAfterRouteRefresh + } CreateMitoRegion { source, .. } | OpenMitoRegion { source, .. } | CleanUpMitoRegion { source, .. } @@ -510,3 +513,20 @@ impl ErrorExt for Error { } } } + +#[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/engine/flush_test.rs b/src/mito2/src/engine/flush_test.rs index 313860c0857..c7b4be14587 100644 --- a/src/mito2/src/engine/flush_test.rs +++ b/src/mito2/src/engine/flush_test.rs @@ -24,6 +24,7 @@ use api::v1::Rows; use async_trait::async_trait; use common_base::Plugins; use common_base::readable_size::ReadableSize; +use common_error::ext::{ErrorExt, RetryHint}; use common_recordbatch::RecordBatches; use common_time::util::current_time_millis; use common_wal::options::{KafkaWalOptions, WAL_OPTIONS_KEY, WalOptions}; @@ -1006,6 +1007,7 @@ async fn test_region_write_buffer_rejects_only_full_region_queue() { .expect("the stalled queue should be rejected at the region hard limit") .unwrap() .unwrap_err(); + assert_eq!(err.retry_hint(), RetryHint::Retryable); assert_matches!( err.into_inner().as_any().downcast_ref::().unwrap(), Error::RejectWrite { .. } diff --git a/src/mito2/src/engine/staging_test.rs b/src/mito2/src/engine/staging_test.rs index bd823708509..1b2a7ffee13 100644 --- a/src/mito2/src/engine/staging_test.rs +++ b/src/mito2/src/engine/staging_test.rs @@ -18,11 +18,14 @@ use std::sync::Arc; use std::time::Duration; use std::{assert_matches, fs}; -use api::v1::Rows; -use common_error::ext::ErrorExt; +use api::v1::{ArrowIpc, Rows}; +use common_error::ext::{ErrorExt, RetryHint}; use common_error::status_code::StatusCode; use common_function::utils::partition_expr_version; +use common_grpc::flight; use common_recordbatch::RecordBatches; +use datatypes::arrow::array::{ArrayRef, Float64Array, StringArray, TimestampMillisecondArray}; +use datatypes::arrow::record_batch::RecordBatch; use datatypes::value::Value; use object_store::Buffer; use object_store::layers::mock::{ @@ -34,8 +37,9 @@ use store_api::region_engine::{ RegionEngine, RemapManifestsRequest, SetRegionRoleStateResponse, SettableRegionRoleState, }; use store_api::region_request::{ - ApplyStagingManifestRequest, EnterStagingRequest, RegionAlterRequest, RegionFlushRequest, - RegionPutRequest, RegionRequest, RegionTruncateRequest, StagingPartitionDirective, + ApplyStagingManifestRequest, EnterStagingRequest, RegionAlterRequest, RegionBulkInsertsRequest, + RegionFlushRequest, RegionPutRequest, RegionRequest, RegionTruncateRequest, + StagingPartitionDirective, }; use store_api::storage::{RegionId, ScanRequest}; @@ -254,7 +258,7 @@ fn default_partition_expr() -> String { } #[tokio::test] -async fn test_staging_reject_all_writes_rejects_put() { +async fn test_staging_reject_all_writes_rejects_put_and_bulk() { let mut env = TestEnv::new().await; let engine = env.create_engine(MitoConfig::default()).await; @@ -281,23 +285,50 @@ async fn test_staging_reject_all_writes_rejects_put() { schema: column_schemas, rows: build_rows(0, 2), }; - let err = engine - .handle_request( + let payload = RecordBatch::try_from_iter([ + ( + "tag_0", + Arc::new(StringArray::from(vec!["0", "1"])) as ArrayRef, + ), + ( + "field_0", + Arc::new(Float64Array::from(vec![0.0, 1.0])) as ArrayRef, + ), + ( + "ts", + Arc::new(TimestampMillisecondArray::from(vec![0, 1000])) as ArrayRef, + ), + ]) + .unwrap(); + let (schema, data_header, data) = flight::record_batch_to_ipc(payload.clone()).unwrap(); + for request in [ + RegionRequest::Put(RegionPutRequest { + skip_wal: false, + rows, + hint: None, + partition_expr_version: None, + }), + RegionRequest::BulkInserts(RegionBulkInsertsRequest { + skip_wal: false, region_id, - RegionRequest::Put(RegionPutRequest { - skip_wal: false, - rows, - hint: None, - partition_expr_version: None, - }), - ) - .await - .unwrap_err(); - assert_eq!(err.status_code(), StatusCode::StorageUnavailable); - assert_matches!( - err.into_inner().as_any().downcast_ref::().unwrap(), - Error::RejectWrite { .. } - ); + payload, + raw_data: ArrowIpc { + schema, + data_header, + payload: data, + }, + partition_expr_version: None, + aligned_schema_version: None, + }), + ] { + let err = engine.handle_request(region_id, request).await.unwrap_err(); + assert_eq!(err.status_code(), StatusCode::StorageUnavailable); + assert_eq!(err.retry_hint(), RetryHint::RetryAfterRouteRefresh); + assert_matches!( + err.into_inner().as_any().downcast_ref::().unwrap(), + Error::RejectWriteInStaging { .. } + ); + } } #[tokio::test] diff --git a/src/mito2/src/error.rs b/src/mito2/src/error.rs index f738044c9ef..5af4b7cec09 100644 --- a/src/mito2/src/error.rs +++ b/src/mito2/src/error.rs @@ -641,6 +641,13 @@ pub enum Error { location: Location, }, + #[snafu(display("Region {} rejects writes while in staging", region_id))] + RejectWriteInStaging { + region_id: RegionId, + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("Failed to compact region {}", region_id))] CompactRegion { region_id: RegionId, @@ -1541,7 +1548,7 @@ impl ErrorExt for Error { RegionClosed { .. } => StatusCode::Cancelled, StaleCompactionExecution { .. } => StatusCode::Cancelled, RegionTruncated { .. } => StatusCode::Cancelled, - RejectWrite { .. } => StatusCode::StorageUnavailable, + RejectWrite { .. } | RejectWriteInStaging { .. } => StatusCode::StorageUnavailable, CompactRegion { source, .. } => source.status_code(), EditRegion { source, .. } => source.status_code(), CompatReader { .. } => StatusCode::Unexpected, @@ -1639,6 +1646,13 @@ impl ErrorExt for Error { use Error::*; match self { + PartitionExprVersionMismatch { .. } + | RegionNotFound { .. } + | RejectWriteInStaging { .. } + | RegionState { + state: RegionRoleState::Follower, + .. + } => RetryHint::RetryAfterRouteRefresh, ReadParquet { .. } | WriteParquet { .. } | RejectWrite { .. } @@ -1720,6 +1734,63 @@ mod tests { use snafu::IntoError; use super::*; + use crate::region::RegionLeaderState; + + #[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(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, hint) in [ + ( + RejectWriteSnafu { region_id }.build(), + StatusCode::StorageUnavailable, + RetryHint::Retryable, + ), + ( + RejectWriteInStagingSnafu { region_id }.build(), + StatusCode::StorageUnavailable, + RetryHint::RetryAfterRouteRefresh, + ), + ( + RegionNotFoundSnafu { region_id }.build(), + StatusCode::RegionNotFound, + RetryHint::RetryAfterRouteRefresh, + ), + ] { + assert_eq!(error.status_code(), status); + assert_eq!(error.retry_hint(), hint); + } + } #[test] fn test_manifest_update_persistence() { diff --git a/src/mito2/src/worker/handle_write.rs b/src/mito2/src/worker/handle_write.rs index 41af569c649..1383a3e9a5d 100644 --- a/src/mito2/src/worker/handle_write.rs +++ b/src/mito2/src/worker/handle_write.rs @@ -26,7 +26,7 @@ use store_api::storage::RegionId; use crate::error::{ InvalidRequestSnafu, PartitionExprVersionMismatchSnafu, RegionNotFoundSnafu, RegionStateSnafu, - RejectWriteSnafu, Result, + RejectWriteInStagingSnafu, RejectWriteSnafu, Result, }; use crate::metrics; use crate::metrics::{ @@ -335,7 +335,7 @@ impl RegionWorkerLoop { if region.reject_all_writes_in_staging() { sender_req .sender - .send(RejectWriteSnafu { region_id }.fail()); + .send(RejectWriteInStagingSnafu { region_id }.fail()); continue; } @@ -399,7 +399,7 @@ impl RegionWorkerLoop { if region.reject_all_writes_in_staging() { sender_req .sender - .send(RejectWriteSnafu { region_id }.fail()); + .send(RejectWriteInStagingSnafu { region_id }.fail()); continue; } let expected_version = region.expected_partition_expr_version(); @@ -481,7 +481,9 @@ impl RegionWorkerLoop { RegionRoleState::Leader(RegionLeaderState::Writable) | RegionRoleState::Leader(RegionLeaderState::Staging) => { if region.reject_all_writes_in_staging() { - bulk_req.sender.send(RejectWriteSnafu { region_id }.fail()); + bulk_req + .sender + .send(RejectWriteInStagingSnafu { region_id }.fail()); continue; } let region_ctx = RegionWriteCtx::new( @@ -529,7 +531,9 @@ impl RegionWorkerLoop { continue; }; if region.reject_all_writes_in_staging() { - bulk_req.sender.send(RejectWriteSnafu { region_id }.fail()); + bulk_req + .sender + .send(RejectWriteInStagingSnafu { region_id }.fail()); continue; } let expected_version = region.expected_partition_expr_version();