feat(client): support region request reuse and route refresh hints (#9540)

* feat(client): support shared region requests without deep cloning Signed-off-by: WenyXu <wenymedia@gmail.com> * feat: propagate route refresh retry hints Signed-off-by: WenyXu <wenymedia@gmail.com> * style: preserve local error variant imports in retry hints Signed-off-by: WenyXu <wenymedia@gmail.com> * fix(mito2): distinguish staging rejection from write backpressure Signed-off-by: WenyXu <wenymedia@gmail.com> * fix(meta): preserve legacy retry hints in instruction replies Signed-off-by: WenyXu <wenymedia@gmail.com> * test(client): inline retry hint compatibility cases Signed-off-by: WenyXu <wenymedia@gmail.com> * test(datanode): align batch-open retry hint assertion Signed-off-by: WenyXu <wenymedia@gmail.com> * style: use imported types in region request tests Signed-off-by: WenyXu <wenymedia@gmail.com> --------- Signed-off-by: WenyXu <wenymedia@gmail.com>
2026-10-13 07:22:32 +00:00 · 2026-10-13 02:29:54 +00:00
parent c1a0ca9595
commit 684c8ff291
19 changed files with 850 additions and 176 deletions
@@ -2122,6 +2122,7 @@ dependencies = [
 "tokio",
 "tokio-stream",
 "tonic 0.14.2",
+ "tonic-prost",
 ]

 [[package]]
@@ -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"
@@ -45,3 +45,4 @@ substrait.workspace = true
 tokio.workspace = true
 tokio-stream = { workspace = true, features = ["net"] }
 tonic.workspace = true
+tonic-prost.workspace = true
@@ -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<Channel>)> {
+    /// 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 = 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))
    }

@@ -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 {
@@ -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> {
@@ -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<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> {
@@ -144,11 +137,49 @@ impl RegionRequester {
        recordbatches_from_flight_message_stream(addr, flight_message_stream).await
    }

-    async fn handle_inner(&self, request: RegionRequest) -> Result<RegionResponse> {
+    /// 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,
+    {
        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<RegionResponse> {
-        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>(
@@ -385,13 +410,22 @@ pub fn check_response_header(header: &Option<ResponseHeader>) -> 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<std::sync::Mutex<Vec<(bool, bool)>>>,
+    #[derive(Clone, Default)]
+    struct RecordingRegionService {
+        requests: Arc<Mutex<Vec<(RegionRequest, bool, bool)>>>,
+        responses: Arc<Mutex<VecDeque<std::result::Result<PbRegionResponse, Status>>>>,
    }

    #[tonic::async_trait]
-    impl Region for CompressionRecordingRegionService {
+    impl Region for RecordingRegionService {
        async fn handle(
            &self,
            request: Request<RegionRequest>,
        ) -> std::result::Result<Response<PbRegionResponse>, 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::<Vec<_>>()
        );
-        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);
@@ -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<R>(PhantomData<R>);
+
+impl<R> Default for RegionCodec<R> {
+    fn default() -> Self {
+        Self(PhantomData)
+    }
+}
+
+impl<R> Codec for RegionCodec<R>
+where
+    R: Borrow<RegionRequest> + Send + 'static,
+{
+    type Encode = R;
+    type Decode = RegionResponse;
+    type Encoder = RegionEncoder<R>;
+    type Decoder = ProstDecoder<RegionResponse>;
+
+    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<R>(PhantomData<R>);
+
+impl<R: Borrow<RegionRequest>> Encoder for RegionEncoder<R> {
+    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()))
+    }
+}
@@ -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
    }
@@ -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");
@@ -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 { .. }
@@ -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<S>(hint: &RetryHint, serializer: S) -> Result<S::Ok, S::Error>
+    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);
@@ -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]
@@ -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 {
@@ -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);
+        }
+    }
+}
@@ -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::<Error>().unwrap(),
        Error::RejectWrite { .. }
@@ -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::<Error>().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::<Error>().unwrap(),
+            Error::RejectWriteInStaging { .. }
+        );
+    }
 }

 #[tokio::test]
@@ -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() {
@@ -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<S> RegionWorkerLoop<S> {
                        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<S> RegionWorkerLoop<S> {
            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<S> RegionWorkerLoop<S> {
                    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<S> RegionWorkerLoop<S> {
                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();