refactor: move ecstore rpc metadata modules (#3933)

crates/ecstore/src/cluster/mod.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 mod control_plane;
+pub(crate) mod rpc;
 
 pub use control_plane::{
     ClusterControlPlane, ClusterControlPlaneSnapshot, ClusterDriveMembership, ClusterEndpointType, ClusterLocalNodeStorage,

crates/ecstore/src/cluster/rpc/client.rs

@@ -0,0 +1,216 @@
+// Copyright 2024 RustFS 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 crate::disk::error::{DiskError, Error as DiskErrorType};
+use crate::rpc::{TONIC_RPC_PREFIX, gen_signature_headers};
+use crate::runtime_sources;
+use http::Method;
+use rustfs_protos::{create_new_channel, proto_gen::node_service::node_service_client::NodeServiceClient};
+use std::{error::Error, io::ErrorKind};
+use tonic::{service::interceptor::InterceptedService, transport::Channel};
+use tracing::debug;
+
+use super::context_propagation::{inject_request_id_into_metadata, inject_trace_context_into_metadata};
+
+/// 3. Subsequent calls will attempt fresh connections
+/// 4. If node is still down, connection will fail fast (3s timeout)
+pub async fn node_service_time_out_client(
+    addr: &String,
+    interceptor: TonicInterceptor,
+) -> Result<NodeServiceClient<InterceptedService<Channel, TonicInterceptor>>, Box<dyn Error>> {
+    // Try to get cached channel
+    let cached_channel = runtime_sources::cached_node_channel(addr).await;
+
+    let channel = match cached_channel {
+        Some(channel) => {
+            debug!("Using cached gRPC channel for: {}", addr);
+            channel
+        }
+        None => {
+            // No cached connection, create new one
+            create_new_channel(addr).await?
+        }
+    };
+
+    Ok(NodeServiceClient::with_interceptor(channel, interceptor))
+}
+
+pub async fn node_service_time_out_client_no_auth(
+    addr: &String,
+) -> Result<NodeServiceClient<InterceptedService<Channel, TonicInterceptor>>, Box<dyn Error>> {
+    node_service_time_out_client(addr, TonicInterceptor::NoOp(NoOpInterceptor)).await
+}
+
+pub(crate) fn is_network_like_disk_error(err: &DiskErrorType) -> bool {
+    match err {
+        DiskError::Timeout => true,
+        DiskError::Io(io_err) => {
+            if matches!(
+                io_err.kind(),
+                ErrorKind::TimedOut
+                    | ErrorKind::ConnectionRefused
+                    | ErrorKind::ConnectionReset
+                    | ErrorKind::BrokenPipe
+                    | ErrorKind::NotConnected
+                    | ErrorKind::ConnectionAborted
+                    | ErrorKind::UnexpectedEof
+            ) {
+                return true;
+            }
+
+            let message = io_err.to_string().to_ascii_lowercase();
+            [
+                "transport error",
+                "unavailable",
+                "error trying to connect",
+                "connection refused",
+                "connection reset",
+                "broken pipe",
+                "not connected",
+                "unexpected eof",
+                "timed out",
+                "deadline has elapsed",
+                "connection closed",
+                "connection aborted",
+                "tcp connect error",
+            ]
+            .iter()
+            .any(|needle| message.contains(needle))
+        }
+        _ => false,
+    }
+}
+
+pub struct TonicSignatureInterceptor;
+
+impl tonic::service::Interceptor for TonicSignatureInterceptor {
+    fn call(&mut self, mut req: tonic::Request<()>) -> Result<tonic::Request<()>, tonic::Status> {
+        let headers = gen_signature_headers(TONIC_RPC_PREFIX, &Method::GET)
+            .map_err(|_| tonic::Status::unauthenticated("No valid auth token"))?;
+        req.metadata_mut().as_mut().extend(headers);
+        inject_trace_context_into_metadata(req.metadata_mut());
+        inject_request_id_into_metadata(req.metadata_mut());
+        Ok(req)
+    }
+}
+
+pub fn gen_tonic_signature_interceptor() -> TonicSignatureInterceptor {
+    TonicSignatureInterceptor
+}
+
+pub struct NoOpInterceptor;
+
+impl tonic::service::Interceptor for NoOpInterceptor {
+    fn call(&mut self, req: tonic::Request<()>) -> Result<tonic::Request<()>, tonic::Status> {
+        Ok(req)
+    }
+}
+
+pub enum TonicInterceptor {
+    Signature(TonicSignatureInterceptor),
+    NoOp(NoOpInterceptor),
+}
+
+impl tonic::service::Interceptor for TonicInterceptor {
+    fn call(&mut self, req: tonic::Request<()>) -> Result<tonic::Request<()>, tonic::Status> {
+        match self {
+            TonicInterceptor::Signature(interceptor) => interceptor.call(req),
+            TonicInterceptor::NoOp(interceptor) => interceptor.call(req),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use opentelemetry::global;
+    use opentelemetry::trace::{SpanContext, TraceContextExt, TraceFlags, TraceId, TraceState, TracerProvider as _};
+    use opentelemetry_sdk::propagation::TraceContextPropagator;
+    use opentelemetry_sdk::trace::SdkTracerProvider;
+    use tonic::service::Interceptor;
+    use tracing_opentelemetry::OpenTelemetrySpanExt;
+    use tracing_subscriber::{Registry, layer::SubscriberExt};
+
+    fn ensure_test_rpc_secret() {
+        runtime_sources::ensure_test_rpc_secret();
+    }
+
+    fn with_trace_parent<F>(trace_id_hex: &str, f: F)
+    where
+        F: FnOnce(),
+    {
+        global::set_text_map_propagator(TraceContextPropagator::new());
+
+        let provider = SdkTracerProvider::builder().build();
+        let tracer = provider.tracer("rpc-client-tests");
+        let subscriber = Registry::default().with(tracing_opentelemetry::layer().with_tracer(tracer));
+
+        tracing::subscriber::with_default(subscriber, || {
+            let span = tracing::info_span!("rpc-client-test-span");
+            let trace_id = TraceId::from_hex(trace_id_hex).expect("trace id should be valid hex");
+            let span_id = opentelemetry::trace::SpanId::from_hex("0102030405060708").expect("span id should be valid hex");
+            let parent = SpanContext::new(trace_id, span_id, TraceFlags::SAMPLED, true, TraceState::default());
+            span.set_parent(opentelemetry::Context::new().with_remote_span_context(parent))
+                .expect("failed to set parent context");
+            let _guard = span.enter();
+
+            f();
+        });
+        let _ = provider.shutdown();
+    }
+
+    #[test]
+    fn test_signature_interceptor_keeps_auth_headers() {
+        ensure_test_rpc_secret();
+        let mut interceptor = TonicSignatureInterceptor;
+        let req = tonic::Request::new(());
+
+        let req = interceptor.call(req).expect("interceptor call should succeed");
+
+        assert!(req.metadata().contains_key("x-rustfs-signature"));
+        assert!(req.metadata().contains_key("x-rustfs-timestamp"));
+    }
+
+    #[test]
+    fn test_signature_interceptor_may_inject_request_id() {
+        ensure_test_rpc_secret();
+        let mut interceptor = TonicSignatureInterceptor;
+        let req = tonic::Request::new(());
+
+        let span = tracing::info_span!("grpc-rpc-test-span");
+        let _guard = span.enter();
+        let req = interceptor.call(req).expect("interceptor call should succeed");
+
+        if let Some(v) = req.metadata().get("x-request-id") {
+            assert!(!v.as_encoded_bytes().is_empty());
+        }
+    }
+
+    #[test]
+    fn test_signature_interceptor_injects_traceparent_metadata() {
+        ensure_test_rpc_secret();
+        let mut interceptor = TonicSignatureInterceptor;
+        let req = tonic::Request::new(());
+
+        with_trace_parent("4bf92f3577b34da6a3ce929d0e0e4736", || {
+            let req = interceptor.call(req).expect("interceptor call should succeed");
+            let traceparent = req
+                .metadata()
+                .get("traceparent")
+                .and_then(|v| v.to_str().ok())
+                .expect("traceparent metadata should be injected");
+            assert!(traceparent.starts_with("00-4bf92f3577b34da6a3ce929d0e0e4736-"));
+        });
+    }
+}

crates/ecstore/src/cluster/rpc/context_propagation.rs

@@ -0,0 +1,222 @@
+// Copyright 2024 RustFS 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 http::{HeaderMap, HeaderValue};
+use opentelemetry::{global, propagation::Injector, trace::TraceContextExt};
+pub(crate) use rustfs_utils::http::headers::REQUEST_ID_HEADER;
+use tracing::Span;
+use tracing_opentelemetry::OpenTelemetrySpanExt;
+
+struct HttpHeaderInjector<'a> {
+    headers: &'a mut HeaderMap,
+}
+
+impl Injector for HttpHeaderInjector<'_> {
+    fn set(&mut self, key: &str, value: String) {
+        let Ok(name) = http::header::HeaderName::from_bytes(key.as_bytes()) else {
+            return;
+        };
+        let Ok(val) = HeaderValue::from_str(&value) else {
+            return;
+        };
+        self.headers.insert(name, val);
+    }
+}
+
+struct MetadataInjector<'a> {
+    metadata: &'a mut tonic::metadata::MetadataMap,
+}
+
+impl Injector for MetadataInjector<'_> {
+    fn set(&mut self, key: &str, value: String) {
+        let Ok(meta_key) = tonic::metadata::MetadataKey::from_bytes(key.as_bytes()) else {
+            return;
+        };
+        let Ok(meta_value) = tonic::metadata::MetadataValue::try_from(value.as_str()) else {
+            return;
+        };
+        self.metadata.insert(meta_key, meta_value);
+    }
+}
+
+fn current_trace_id() -> Option<String> {
+    let current_context = Span::current().context();
+    let current_span = current_context.span();
+    let span_context = current_span.span_context();
+    if !span_context.is_valid() {
+        return None;
+    }
+    Some(span_context.trace_id().to_string())
+}
+
+fn fallback_request_id() -> String {
+    format!("req-{}", &uuid::Uuid::new_v4().to_string()[..8])
+}
+
+fn propagated_request_id() -> String {
+    current_trace_id()
+        .map(|trace_id| format!("trace-{trace_id}"))
+        .unwrap_or_else(fallback_request_id)
+}
+
+pub(crate) fn inject_trace_context_into_http_headers(headers: &mut HeaderMap) {
+    let current_context = Span::current().context();
+    global::get_text_map_propagator(|propagator| {
+        let mut injector = HttpHeaderInjector { headers };
+        propagator.inject_context(&current_context, &mut injector);
+    });
+}
+
+pub(crate) fn inject_request_id_into_http_headers(headers: &mut HeaderMap) {
+    if headers.contains_key(REQUEST_ID_HEADER) {
+        return;
+    }
+    let request_id = propagated_request_id();
+    if let Ok(value) = HeaderValue::from_str(&request_id) {
+        headers.insert(REQUEST_ID_HEADER, value);
+    }
+}
+
+pub(crate) fn inject_trace_context_into_metadata(metadata: &mut tonic::metadata::MetadataMap) {
+    let current_context = Span::current().context();
+    global::get_text_map_propagator(|propagator| {
+        let mut injector = MetadataInjector { metadata };
+        propagator.inject_context(&current_context, &mut injector);
+    });
+}
+
+pub(crate) fn inject_request_id_into_metadata(metadata: &mut tonic::metadata::MetadataMap) {
+    let request_id_key = tonic::metadata::MetadataKey::from_static(REQUEST_ID_HEADER);
+    if metadata.contains_key(&request_id_key) {
+        return;
+    }
+    let request_id = propagated_request_id();
+    let Ok(value) = tonic::metadata::MetadataValue::try_from(request_id.as_str()) else {
+        return;
+    };
+    metadata.insert(request_id_key, value);
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use opentelemetry::trace::{SpanContext, TraceContextExt, TraceFlags, TraceId, TraceState, TracerProvider as _};
+    use opentelemetry_sdk::trace::SdkTracerProvider;
+    use tracing_opentelemetry::OpenTelemetrySpanExt;
+    use tracing_subscriber::{Registry, layer::SubscriberExt};
+
+    fn with_trace_parent<F>(trace_id_hex: &str, f: F)
+    where
+        F: FnOnce(),
+    {
+        let provider = SdkTracerProvider::builder().build();
+        let tracer = provider.tracer("context-propagation-tests");
+        let subscriber = Registry::default().with(tracing_opentelemetry::layer().with_tracer(tracer));
+
+        tracing::subscriber::with_default(subscriber, || {
+            let span = tracing::info_span!("context-propagation-test-span");
+
+            let trace_id = TraceId::from_hex(trace_id_hex).expect("trace id should be valid hex");
+            let span_id = opentelemetry::trace::SpanId::from_hex("0102030405060708").expect("span id should be valid hex");
+            let parent = SpanContext::new(trace_id, span_id, TraceFlags::SAMPLED, true, TraceState::default());
+            span.set_parent(opentelemetry::Context::new().with_remote_span_context(parent))
+                .expect("failed to set parent context");
+            let _guard = span.enter();
+
+            f();
+        });
+        let _ = provider.shutdown();
+    }
+
+    #[test]
+    fn test_inject_request_id_into_http_headers_preserves_existing_value() {
+        let mut headers = HeaderMap::new();
+        headers.insert(REQUEST_ID_HEADER, HeaderValue::from_static("req-upstream-123"));
+
+        with_trace_parent("0123456789abcdef0123456789abcdef", || {
+            inject_request_id_into_http_headers(&mut headers);
+        });
+
+        assert_eq!(headers.get(REQUEST_ID_HEADER).and_then(|v| v.to_str().ok()), Some("req-upstream-123"));
+    }
+
+    #[test]
+    fn test_inject_request_id_into_http_headers_uses_trace_id_when_missing() {
+        let trace_id = "abcdefabcdefabcdefabcdefabcdefab";
+        let mut headers = HeaderMap::new();
+
+        with_trace_parent(trace_id, || {
+            inject_request_id_into_http_headers(&mut headers);
+        });
+
+        assert_eq!(
+            headers.get(REQUEST_ID_HEADER).and_then(|v| v.to_str().ok()),
+            Some(format!("trace-{trace_id}").as_str())
+        );
+    }
+
+    #[test]
+    fn test_inject_request_id_into_metadata_preserves_existing_value() {
+        let mut metadata = tonic::metadata::MetadataMap::new();
+        metadata.insert(
+            tonic::metadata::MetadataKey::from_static(REQUEST_ID_HEADER),
+            tonic::metadata::MetadataValue::from_static("req-upstream-456"),
+        );
+
+        with_trace_parent("fedcba9876543210fedcba9876543210", || {
+            inject_request_id_into_metadata(&mut metadata);
+        });
+
+        assert_eq!(metadata.get(REQUEST_ID_HEADER).and_then(|v| v.to_str().ok()), Some("req-upstream-456"));
+    }
+
+    #[test]
+    fn test_inject_request_id_into_metadata_uses_trace_id_when_missing() {
+        let trace_id = "1234567890abcdef1234567890abcdef";
+        let mut metadata = tonic::metadata::MetadataMap::new();
+
+        with_trace_parent(trace_id, || {
+            inject_request_id_into_metadata(&mut metadata);
+        });
+
+        assert_eq!(
+            metadata.get(REQUEST_ID_HEADER).and_then(|v| v.to_str().ok()),
+            Some(format!("trace-{trace_id}").as_str())
+        );
+    }
+
+    #[test]
+    fn test_inject_request_id_into_http_headers_uses_req_fallback_when_trace_missing() {
+        let mut headers = HeaderMap::new();
+        inject_request_id_into_http_headers(&mut headers);
+
+        let request_id = headers
+            .get(REQUEST_ID_HEADER)
+            .and_then(|v| v.to_str().ok())
+            .expect("request id should be injected");
+        assert!(request_id.starts_with("req-"), "expected req- fallback, got: {request_id}");
+    }
+
+    #[test]
+    fn test_inject_request_id_into_metadata_uses_req_fallback_when_trace_missing() {
+        let mut metadata = tonic::metadata::MetadataMap::new();
+        inject_request_id_into_metadata(&mut metadata);
+
+        let request_id = metadata
+            .get(REQUEST_ID_HEADER)
+            .and_then(|v| v.to_str().ok())
+            .expect("request id should be injected");
+        assert!(request_id.starts_with("req-"), "expected req- fallback, got: {request_id}");
+    }
+}

crates/ecstore/src/cluster/rpc/http_auth.rs

@@ -0,0 +1,609 @@
+// Copyright 2024 RustFS 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 crate::rpc::context_propagation::{inject_request_id_into_http_headers, inject_trace_context_into_http_headers};
+use base64::Engine as _;
+use base64::engine::general_purpose;
+use hmac::{Hmac, KeyInit, Mac};
+use http::{HeaderMap, HeaderValue, Method, Uri};
+#[cfg(test)]
+use rustfs_credentials::{DEFAULT_SECRET_KEY, RPC_SECRET_REQUIRED_MESSAGE};
+use rustfs_credentials::{RPC_SECRET_REQUIRED_OPERATOR_MESSAGE, try_get_rpc_token};
+use sha2::Sha256;
+use std::sync::Once;
+use time::OffsetDateTime;
+use tracing::error;
+
+type HmacSha256 = Hmac<Sha256>;
+
+const SIGNATURE_HEADER: &str = "x-rustfs-signature";
+const TIMESTAMP_HEADER: &str = "x-rustfs-timestamp";
+const SIGNATURE_VALID_DURATION: i64 = 300; // 5 minutes
+pub const TONIC_RPC_PREFIX: &str = "/node_service.NodeService";
+static RPC_SECRET_RESOLUTION_LOG_ONCE: Once = Once::new();
+
+/// Get the shared secret for HMAC signing
+#[cfg(test)]
+fn resolve_shared_secret(env_secret: Option<&str>, global_secret: Option<&str>) -> std::io::Result<String> {
+    if let Some(secret) = env_secret.map(str::trim).filter(|secret| !secret.is_empty()) {
+        return (secret != DEFAULT_SECRET_KEY)
+            .then(|| secret.to_string())
+            .ok_or_else(|| std::io::Error::other(RPC_SECRET_REQUIRED_MESSAGE));
+    }
+
+    global_secret
+        .map(str::trim)
+        .filter(|secret| !secret.is_empty() && *secret != DEFAULT_SECRET_KEY)
+        .map(ToOwned::to_owned)
+        .ok_or_else(|| std::io::Error::other(RPC_SECRET_REQUIRED_MESSAGE))
+}
+
+fn get_shared_secret() -> std::io::Result<String> {
+    try_get_rpc_token().map_err(|err| {
+        RPC_SECRET_RESOLUTION_LOG_ONCE.call_once(|| {
+            error!("RPC auth secret resolution failed: {}; {}", err, RPC_SECRET_REQUIRED_OPERATOR_MESSAGE);
+        });
+        err
+    })
+}
+
+/// Build the canonical payload covered by the RPC HMAC.
+fn signature_payload(url: &str, method: &Method, timestamp: i64) -> String {
+    let uri: Uri = url.parse().expect("Invalid URL");
+
+    let path_and_query = uri.path_and_query().unwrap();
+
+    let url = path_and_query.to_string();
+
+    format!("{url}|{method}|{timestamp}")
+}
+
+/// Generate HMAC-SHA256 signature for the given data
+fn generate_signature(secret: &str, url: &str, method: &Method, timestamp: i64) -> String {
+    let data = signature_payload(url, method, timestamp);
+    let mut mac = <HmacSha256 as KeyInit>::new_from_slice(secret.as_bytes()).expect("HMAC can take key of any size");
+    mac.update(data.as_bytes());
+    let result = mac.finalize();
+    general_purpose::STANDARD.encode(result.into_bytes())
+}
+
+fn verify_signature(secret: &str, url: &str, method: &Method, timestamp: i64, signature: &str) -> bool {
+    let Ok(signature) = general_purpose::STANDARD.decode(signature) else {
+        return false;
+    };
+
+    let data = signature_payload(url, method, timestamp);
+    let mut mac = <HmacSha256 as KeyInit>::new_from_slice(secret.as_bytes()).expect("HMAC can take key of any size");
+    mac.update(data.as_bytes());
+    mac.verify_slice(&signature).is_ok()
+}
+
+/// Build headers with authentication signature
+pub fn build_auth_headers(url: &str, method: &Method, headers: &mut HeaderMap) -> std::io::Result<()> {
+    let auth_headers = gen_signature_headers(url, method)?;
+
+    headers.extend(auth_headers);
+    inject_trace_context_into_http_headers(headers);
+    inject_request_id_into_http_headers(headers);
+    Ok(())
+}
+
+pub fn gen_signature_headers(url: &str, method: &Method) -> std::io::Result<HeaderMap> {
+    let secret = get_shared_secret()?;
+    let timestamp = OffsetDateTime::now_utc().unix_timestamp();
+
+    let signature = generate_signature(&secret, url, method, timestamp);
+
+    let mut headers = HeaderMap::new();
+    headers.insert(SIGNATURE_HEADER, HeaderValue::from_str(&signature).expect("Invalid header value"));
+    headers.insert(
+        TIMESTAMP_HEADER,
+        HeaderValue::from_str(&timestamp.to_string()).expect("Invalid header value"),
+    );
+
+    Ok(headers)
+}
+
+/// Verify the request signature for RPC requests
+pub fn verify_rpc_signature(url: &str, method: &Method, headers: &HeaderMap) -> std::io::Result<()> {
+    // Get signature from header
+    let signature = headers
+        .get(SIGNATURE_HEADER)
+        .and_then(|v| v.to_str().ok())
+        .ok_or_else(|| std::io::Error::other("Missing signature header"))?;
+
+    // Get timestamp from header
+    let timestamp_str = headers
+        .get(TIMESTAMP_HEADER)
+        .and_then(|v| v.to_str().ok())
+        .ok_or_else(|| std::io::Error::other("Missing timestamp header"))?;
+
+    let timestamp: i64 = timestamp_str
+        .parse()
+        .map_err(|_| std::io::Error::other("Invalid timestamp format"))?;
+
+    // Check timestamp validity (prevent replay attacks)
+    let current_time = OffsetDateTime::now_utc().unix_timestamp();
+
+    if current_time.saturating_sub(timestamp) > SIGNATURE_VALID_DURATION
+        || timestamp.saturating_sub(current_time) > SIGNATURE_VALID_DURATION
+    {
+        return Err(std::io::Error::other("Request timestamp expired"));
+    }
+
+    // Verify signature with constant-time HMAC comparison.
+    let secret = get_shared_secret()?;
+
+    if !verify_signature(&secret, url, method, timestamp, signature) {
+        error!(
+            "verify_rpc_signature: Invalid signature: url {}, method {}, timestamp {}, signature_len {}",
+            url,
+            method,
+            timestamp,
+            signature.len()
+        );
+
+        return Err(std::io::Error::other("Invalid signature"));
+    }
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::rpc::context_propagation::REQUEST_ID_HEADER;
+    use crate::runtime_sources;
+    use http::{HeaderMap, Method};
+    use std::io::{self, Write};
+    use std::sync::{Arc, Mutex};
+    use time::OffsetDateTime;
+    use tracing_subscriber::fmt::MakeWriter;
+
+    #[derive(Clone, Default)]
+    struct CapturedLogs {
+        buffer: Arc<Mutex<Vec<u8>>>,
+    }
+
+    struct CapturedLogWriter {
+        buffer: Arc<Mutex<Vec<u8>>>,
+    }
+
+    impl CapturedLogs {
+        fn contents(&self) -> String {
+            let buffer = self
+                .buffer
+                .lock()
+                .expect("captured logs mutex should not be poisoned")
+                .clone();
+            String::from_utf8(buffer).expect("captured logs should be valid UTF-8")
+        }
+    }
+
+    impl Write for CapturedLogWriter {
+        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+            self.buffer
+                .lock()
+                .expect("captured logs mutex should not be poisoned")
+                .extend_from_slice(buf);
+            Ok(buf.len())
+        }
+
+        fn flush(&mut self) -> io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl<'a> MakeWriter<'a> for CapturedLogs {
+        type Writer = CapturedLogWriter;
+
+        fn make_writer(&'a self) -> Self::Writer {
+            CapturedLogWriter {
+                buffer: Arc::clone(&self.buffer),
+            }
+        }
+    }
+
+    fn ensure_test_rpc_secret() {
+        runtime_sources::ensure_test_rpc_secret();
+    }
+
+    #[test]
+    fn test_resolve_shared_secret_rejects_default_fallback() {
+        let err = resolve_shared_secret(None, None).expect_err("default fallback must be rejected");
+        assert_eq!(err.to_string(), RPC_SECRET_REQUIRED_MESSAGE);
+
+        let err = resolve_shared_secret(None, Some(DEFAULT_SECRET_KEY)).expect_err("default global secret must be rejected");
+        assert_eq!(err.to_string(), RPC_SECRET_REQUIRED_MESSAGE);
+    }
+
+    #[test]
+    fn test_get_shared_secret() {
+        ensure_test_rpc_secret();
+        let secret = get_shared_secret().expect("test RPC secret should resolve");
+        assert!(!secret.is_empty(), "Secret should not be empty");
+
+        let url = "http://node1:7000/rustfs/rpc/read_file_stream?disk=http%3A%2F%2Fnode1%3A7000%2Fdata%2Frustfs3&volume=.rustfs.sys&path=pool.bin%2Fdd0fd773-a962-4265-b543-783ce83953e9%2Fpart.1&offset=0&length=44";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        build_auth_headers(url, &method, &mut headers).expect("auth headers should build");
+
+        let url = "/rustfs/rpc/read_file_stream?disk=http%3A%2F%2Fnode1%3A7000%2Fdata%2Frustfs3&volume=.rustfs.sys&path=pool.bin%2Fdd0fd773-a962-4265-b543-783ce83953e9%2Fpart.1&offset=0&length=44";
+
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_ok(), "Valid signature should pass verification");
+    }
+
+    #[test]
+    fn test_generate_signature_deterministic() {
+        let secret = "test-secret";
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let timestamp = 1640995200; // Fixed timestamp
+
+        let signature1 = generate_signature(secret, url, &method, timestamp);
+        let signature2 = generate_signature(secret, url, &method, timestamp);
+
+        assert_eq!(signature1, signature2, "Same inputs should produce same signature");
+        assert!(!signature1.is_empty(), "Signature should not be empty");
+    }
+
+    #[test]
+    fn test_generate_signature_different_inputs() {
+        let secret = "test-secret";
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let timestamp = 1640995200;
+
+        let signature1 = generate_signature(secret, url, &method, timestamp);
+        let signature2 = generate_signature(secret, "http://different.com/api/test2", &method, timestamp);
+        let signature3 = generate_signature(secret, url, &Method::POST, timestamp);
+        let signature4 = generate_signature(secret, url, &method, timestamp + 1);
+
+        assert_ne!(signature1, signature2, "Different URLs should produce different signatures");
+        assert_ne!(signature1, signature3, "Different methods should produce different signatures");
+        assert_ne!(signature1, signature4, "Different timestamps should produce different signatures");
+    }
+
+    #[test]
+    fn test_build_auth_headers() {
+        ensure_test_rpc_secret();
+        let url = "http://example.com/api/test";
+        let method = Method::POST;
+        let mut headers = HeaderMap::new();
+
+        build_auth_headers(url, &method, &mut headers).expect("auth headers should build");
+
+        // Verify headers are present
+        assert!(headers.contains_key(SIGNATURE_HEADER), "Should contain signature header");
+        assert!(headers.contains_key(TIMESTAMP_HEADER), "Should contain timestamp header");
+
+        // Verify header values are not empty
+        let signature = headers.get(SIGNATURE_HEADER).unwrap().to_str().unwrap();
+        let timestamp_str = headers.get(TIMESTAMP_HEADER).unwrap().to_str().unwrap();
+
+        assert!(!signature.is_empty(), "Signature should not be empty");
+        assert!(!timestamp_str.is_empty(), "Timestamp should not be empty");
+
+        // Verify timestamp is a valid integer
+        let timestamp: i64 = timestamp_str.parse().expect("Timestamp should be valid integer");
+        let current_time = OffsetDateTime::now_utc().unix_timestamp();
+
+        // Should be within a reasonable range (within 1 second of current time)
+        assert!((current_time - timestamp).abs() <= 1, "Timestamp should be close to current time");
+    }
+
+    #[test]
+    fn test_build_auth_headers_preserves_existing_request_id() {
+        ensure_test_rpc_secret();
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+        headers.insert(REQUEST_ID_HEADER, HeaderValue::from_static("req-upstream-123"));
+
+        build_auth_headers(url, &method, &mut headers).expect("auth headers should build");
+
+        assert_eq!(headers.get(REQUEST_ID_HEADER).and_then(|v| v.to_str().ok()), Some("req-upstream-123"));
+    }
+
+    #[test]
+    fn test_build_auth_headers_may_set_request_id_from_trace_id() {
+        ensure_test_rpc_secret();
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        let span = tracing::info_span!("rpc-test-span");
+        let _guard = span.enter();
+        build_auth_headers(url, &method, &mut headers).expect("auth headers should build");
+
+        if let Some(value) = headers.get(REQUEST_ID_HEADER).and_then(|v| v.to_str().ok()) {
+            assert!(!value.is_empty(), "request id should not be empty");
+        }
+    }
+
+    #[test]
+    fn test_verify_rpc_signature_success() {
+        ensure_test_rpc_secret();
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        // Build headers with valid signature
+        build_auth_headers(url, &method, &mut headers).expect("auth headers should build");
+
+        // Verify should succeed
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_ok(), "Valid signature should pass verification");
+    }
+
+    #[test]
+    fn test_verify_rpc_signature_invalid_signature() {
+        ensure_test_rpc_secret();
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        // Build headers with valid signature first
+        build_auth_headers(url, &method, &mut headers).expect("auth headers should build");
+
+        // Tamper with the signature
+        headers.insert(SIGNATURE_HEADER, HeaderValue::from_str("invalid-signature").unwrap());
+
+        // Verify should fail
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_err(), "Invalid signature should fail verification");
+
+        let error = result.unwrap_err();
+        assert_eq!(error.to_string(), "Invalid signature");
+    }
+
+    #[test]
+    fn test_verify_signature_uses_hmac_verification() {
+        let secret = "test-secret";
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let timestamp = 1640995200;
+        let signature = generate_signature(secret, url, &method, timestamp);
+        let mut tampered = general_purpose::STANDARD.decode(&signature).unwrap();
+        tampered[0] ^= 1;
+        let tampered_signature = general_purpose::STANDARD.encode(tampered);
+
+        assert!(verify_signature(secret, url, &method, timestamp, &signature));
+        assert!(!verify_signature(secret, url, &method, timestamp, &tampered_signature));
+        assert!(!verify_signature(secret, url, &method, timestamp, "invalid-signature"));
+    }
+
+    #[test]
+    fn test_invalid_signature_log_contract_excludes_secrets() {
+        ensure_test_rpc_secret();
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let timestamp = OffsetDateTime::now_utc().unix_timestamp();
+        let secret = get_shared_secret().expect("test RPC secret should resolve");
+        let expected_signature = generate_signature(&secret, url, &method, timestamp);
+        let invalid_signature = "invalid-signature";
+        let logs = CapturedLogs::default();
+        let subscriber = tracing_subscriber::fmt()
+            .with_max_level(tracing::Level::ERROR)
+            .with_writer(logs.clone())
+            .with_ansi(false)
+            .without_time()
+            .finish();
+
+        let mut headers = HeaderMap::new();
+        headers.insert(SIGNATURE_HEADER, HeaderValue::from_str(invalid_signature).unwrap());
+        headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str(&timestamp.to_string()).unwrap());
+
+        tracing::subscriber::with_default(subscriber, || {
+            let result = verify_rpc_signature(url, &method, &headers);
+            assert!(result.is_err(), "Invalid signature should fail verification");
+        });
+
+        let captured = logs.contents();
+        assert!(captured.contains("Invalid signature"));
+        assert!(!captured.contains(&secret));
+        assert!(!captured.contains(&expected_signature));
+        assert!(!captured.contains(invalid_signature));
+    }
+
+    #[test]
+    fn test_verify_rpc_signature_expired_timestamp() {
+        ensure_test_rpc_secret();
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        // Set expired timestamp (older than SIGNATURE_VALID_DURATION)
+        let expired_timestamp = OffsetDateTime::now_utc().unix_timestamp() - SIGNATURE_VALID_DURATION - 10;
+        let secret = get_shared_secret().expect("test RPC secret should resolve");
+        let signature = generate_signature(&secret, url, &method, expired_timestamp);
+
+        headers.insert(SIGNATURE_HEADER, HeaderValue::from_str(&signature).unwrap());
+        headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str(&expired_timestamp.to_string()).unwrap());
+
+        // Verify should fail due to expired timestamp
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_err(), "Expired timestamp should fail verification");
+
+        let error = result.unwrap_err();
+        assert_eq!(error.to_string(), "Request timestamp expired");
+    }
+
+    #[test]
+    fn test_verify_rpc_signature_future_timestamp_outside_window() {
+        ensure_test_rpc_secret();
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        let future_timestamp = OffsetDateTime::now_utc().unix_timestamp() + SIGNATURE_VALID_DURATION + 10;
+        let secret = get_shared_secret().expect("test RPC secret should resolve");
+        let signature = generate_signature(&secret, url, &method, future_timestamp);
+
+        headers.insert(SIGNATURE_HEADER, HeaderValue::from_str(&signature).unwrap());
+        headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str(&future_timestamp.to_string()).unwrap());
+
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_err(), "Future timestamp outside valid window should fail verification");
+
+        let error = result.unwrap_err();
+        assert_eq!(error.to_string(), "Request timestamp expired");
+    }
+
+    #[test]
+    fn test_verify_rpc_signature_missing_signature_header() {
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        // Add only timestamp header, missing signature
+        let timestamp = OffsetDateTime::now_utc().unix_timestamp();
+        headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str(&timestamp.to_string()).unwrap());
+
+        // Verify should fail
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_err(), "Missing signature header should fail verification");
+
+        let error = result.unwrap_err();
+        assert_eq!(error.to_string(), "Missing signature header");
+    }
+
+    #[test]
+    fn test_verify_rpc_signature_missing_timestamp_header() {
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        // Add only signature header, missing timestamp
+        headers.insert(SIGNATURE_HEADER, HeaderValue::from_str("some-signature").unwrap());
+
+        // Verify should fail
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_err(), "Missing timestamp header should fail verification");
+
+        let error = result.unwrap_err();
+        assert_eq!(error.to_string(), "Missing timestamp header");
+    }
+
+    #[test]
+    fn test_verify_rpc_signature_invalid_timestamp_format() {
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        headers.insert(SIGNATURE_HEADER, HeaderValue::from_str("some-signature").unwrap());
+        headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str("invalid-timestamp").unwrap());
+
+        // Verify should fail
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_err(), "Invalid timestamp format should fail verification");
+
+        let error = result.unwrap_err();
+        assert_eq!(error.to_string(), "Invalid timestamp format");
+    }
+
+    #[test]
+    fn test_verify_rpc_signature_url_mismatch() {
+        ensure_test_rpc_secret();
+        let original_url = "http://example.com/api/test";
+        let different_url = "http://example.com/api/different";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        // Build headers for one URL
+        build_auth_headers(original_url, &method, &mut headers).expect("auth headers should build");
+
+        // Try to verify with a different URL
+        let result = verify_rpc_signature(different_url, &method, &headers);
+        assert!(result.is_err(), "URL mismatch should fail verification");
+
+        let error = result.unwrap_err();
+        assert_eq!(error.to_string(), "Invalid signature");
+    }
+
+    #[test]
+    fn test_verify_rpc_signature_method_mismatch() {
+        ensure_test_rpc_secret();
+        let url = "http://example.com/api/test";
+        let original_method = Method::GET;
+        let different_method = Method::POST;
+        let mut headers = HeaderMap::new();
+
+        // Build headers for one method
+        build_auth_headers(url, &original_method, &mut headers).expect("auth headers should build");
+
+        // Try to verify with a different method
+        let result = verify_rpc_signature(url, &different_method, &headers);
+        assert!(result.is_err(), "Method mismatch should fail verification");
+
+        let error = result.unwrap_err();
+        assert_eq!(error.to_string(), "Invalid signature");
+    }
+
+    #[test]
+    fn test_signature_valid_duration_boundary() {
+        ensure_test_rpc_secret();
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let secret = get_shared_secret().expect("test RPC secret should resolve");
+
+        let mut headers = HeaderMap::new();
+        let current_time = OffsetDateTime::now_utc().unix_timestamp();
+        // Test timestamp just within valid duration
+        let valid_timestamp = current_time - SIGNATURE_VALID_DURATION + 1;
+
+        let signature = generate_signature(&secret, url, &method, valid_timestamp);
+
+        headers.insert(SIGNATURE_HEADER, HeaderValue::from_str(&signature).unwrap());
+        headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str(&valid_timestamp.to_string()).unwrap());
+
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_ok(), "Timestamp within valid duration should pass");
+
+        // Test timestamp just outside valid duration
+        let mut headers = HeaderMap::new();
+        let invalid_timestamp = current_time - SIGNATURE_VALID_DURATION - 15;
+        let signature = generate_signature(&secret, url, &method, invalid_timestamp);
+
+        headers.insert(SIGNATURE_HEADER, HeaderValue::from_str(&signature).unwrap());
+        headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str(&invalid_timestamp.to_string()).unwrap());
+
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_err(), "Timestamp outside valid duration should fail");
+    }
+
+    #[test]
+    fn test_round_trip_authentication() {
+        ensure_test_rpc_secret();
+        let test_cases = vec![
+            ("http://example.com/api/test", Method::GET),
+            ("https://api.rustfs.com/v1/bucket", Method::POST),
+            ("http://localhost:9000/admin/info", Method::PUT),
+            ("https://storage.example.com/path/to/object?query=param", Method::DELETE),
+        ];
+
+        for (url, method) in test_cases {
+            let mut headers = HeaderMap::new();
+
+            // Build authentication headers
+            build_auth_headers(url, &method, &mut headers).expect("auth headers should build");
+
+            // Verify the signature should succeed
+            let result = verify_rpc_signature(url, &method, &headers);
+            assert!(result.is_ok(), "Round-trip test failed for {method} {url}");
+        }
+    }
+}

crates/ecstore/src/cluster/rpc/internode_data_transport.rs

@@ -0,0 +1,363 @@
+// Copyright 2024 RustFS 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 crate::disk::error::{Error, Result};
+use crate::disk::{FileReader, FileWriter};
+use crate::rpc::build_auth_headers;
+use async_trait::async_trait;
+use http::{HeaderMap, HeaderValue, Method, header::CONTENT_TYPE};
+use rustfs_config::{
+    DEFAULT_INTERNODE_DATA_TRANSPORT, ENV_RUSTFS_INTERNODE_DATA_TRANSPORT, INTERNODE_DATA_TRANSPORT_TCP,
+    KNOWN_INTERNODE_DATA_TRANSPORT_BACKENDS,
+};
+use rustfs_rio::{HttpReader, HttpWriter};
+use std::sync::{Arc, OnceLock};
+use std::time::Duration;
+
+static INTERNODE_DATA_TRANSPORT: OnceLock<std::result::Result<Arc<dyn InternodeDataTransport>, String>> = OnceLock::new();
+
+const READ_FILE_STREAM_PATH: &str = "/rustfs/rpc/read_file_stream";
+const PUT_FILE_STREAM_PATH: &str = "/rustfs/rpc/put_file_stream";
+const WALK_DIR_PATH: &str = "/rustfs/rpc/walk_dir";
+const CONTENT_TYPE_JSON: &str = "application/json";
+
+fn unsupported_transport_message(transport: &str) -> String {
+    format!(
+        "invalid {ENV_RUSTFS_INTERNODE_DATA_TRANSPORT}={transport:?}; supported values: {}",
+        KNOWN_INTERNODE_DATA_TRANSPORT_BACKENDS.join(", ")
+    )
+}
+
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
+pub struct InternodeDataTransportCapabilities {
+    /// Backend can open a streaming remote disk reader.
+    pub streaming_read: bool,
+    /// Backend can open a streaming remote disk writer.
+    pub streaming_write: bool,
+    /// Backend can stream walk-dir responses.
+    pub streaming_walk_dir: bool,
+    /// Backend preserves in-order delivery for each opened transfer.
+    pub ordered_delivery: bool,
+    /// Largest payload the backend accepts for one transfer, or no RustFS-level cap.
+    pub max_transfer_size: Option<usize>,
+    /// Backend can participate in the behavior-preserving TCP fallback path.
+    pub fallback_supported: bool,
+}
+
+impl InternodeDataTransportCapabilities {
+    pub const fn tcp_http() -> Self {
+        Self {
+            streaming_read: true,
+            streaming_write: true,
+            streaming_walk_dir: true,
+            ordered_delivery: true,
+            max_transfer_size: None,
+            fallback_supported: true,
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct ReadStreamRequest {
+    pub endpoint: String,
+    pub disk: String,
+    pub volume: String,
+    pub path: String,
+    pub offset: usize,
+    pub length: usize,
+}
+
+#[derive(Debug, Clone)]
+pub struct WriteStreamRequest {
+    pub endpoint: String,
+    pub disk: String,
+    pub volume: String,
+    pub path: String,
+    pub append: bool,
+    pub size: i64,
+}
+
+#[derive(Debug, Clone)]
+pub struct WalkDirStreamRequest {
+    pub endpoint: String,
+    pub disk: String,
+    pub body: Vec<u8>,
+    pub stall_timeout: Option<Duration>,
+}
+
+/// Data-plane stream opener used by `RemoteDisk`.
+///
+/// This boundary is limited to remote disk streams that can move large payloads.
+/// Internode metadata, lock, health, and administrative calls remain on the
+/// existing gRPC control plane.
+///
+/// Buffer ownership, backend selection, and fallback expectations are documented
+/// in `crates/ecstore/docs/internode-transport/`.
+#[async_trait]
+pub trait InternodeDataTransport: Send + Sync + std::fmt::Debug {
+    async fn open_read(&self, request: ReadStreamRequest) -> Result<FileReader>;
+    async fn open_write(&self, request: WriteStreamRequest) -> Result<FileWriter>;
+    async fn open_walk_dir(&self, request: WalkDirStreamRequest) -> Result<FileReader>;
+    fn name(&self) -> &'static str;
+    fn capabilities(&self) -> InternodeDataTransportCapabilities;
+}
+
+#[derive(Debug, Default)]
+pub struct TcpHttpInternodeDataTransport;
+
+#[async_trait]
+impl InternodeDataTransport for TcpHttpInternodeDataTransport {
+    async fn open_read(&self, request: ReadStreamRequest) -> Result<FileReader> {
+        let url = build_read_file_stream_url(&request);
+        let mut headers = json_headers();
+        build_auth_headers(&url, &Method::GET, &mut headers)?;
+        Ok(Box::new(HttpReader::new(url, Method::GET, headers, None).await?))
+    }
+
+    async fn open_write(&self, request: WriteStreamRequest) -> Result<FileWriter> {
+        let url = build_put_file_stream_url(&request);
+        let mut headers = json_headers();
+        build_auth_headers(&url, &Method::PUT, &mut headers)?;
+        Ok(Box::new(HttpWriter::new(url, Method::PUT, headers).await?))
+    }
+
+    async fn open_walk_dir(&self, request: WalkDirStreamRequest) -> Result<FileReader> {
+        let url = build_walk_dir_url(&request);
+        let mut headers = json_headers();
+        build_auth_headers(&url, &Method::GET, &mut headers)?;
+        Ok(Box::new(
+            HttpReader::new_with_stall_timeout(url, Method::GET, headers, Some(request.body), request.stall_timeout).await?,
+        ))
+    }
+
+    fn name(&self) -> &'static str {
+        DEFAULT_INTERNODE_DATA_TRANSPORT
+    }
+
+    fn capabilities(&self) -> InternodeDataTransportCapabilities {
+        InternodeDataTransportCapabilities::tcp_http()
+    }
+}
+
+fn build_read_file_stream_url(request: &ReadStreamRequest) -> String {
+    format!(
+        "{}{}?disk={}&volume={}&path={}&offset={}&length={}",
+        request.endpoint,
+        READ_FILE_STREAM_PATH,
+        urlencoding::encode(&request.disk),
+        urlencoding::encode(&request.volume),
+        urlencoding::encode(&request.path),
+        request.offset,
+        request.length
+    )
+}
+
+fn build_put_file_stream_url(request: &WriteStreamRequest) -> String {
+    format!(
+        "{}{}?disk={}&volume={}&path={}&append={}&size={}",
+        request.endpoint,
+        PUT_FILE_STREAM_PATH,
+        urlencoding::encode(&request.disk),
+        urlencoding::encode(&request.volume),
+        urlencoding::encode(&request.path),
+        request.append,
+        request.size
+    )
+}
+
+fn build_walk_dir_url(request: &WalkDirStreamRequest) -> String {
+    format!("{}{}?disk={}", request.endpoint, WALK_DIR_PATH, urlencoding::encode(&request.disk))
+}
+
+fn json_headers() -> HeaderMap {
+    let mut headers = HeaderMap::new();
+    headers.insert(CONTENT_TYPE, HeaderValue::from_static(CONTENT_TYPE_JSON));
+    headers
+}
+
+fn build_internode_data_transport_result(
+    configured_transport: Option<&str>,
+) -> std::result::Result<Arc<dyn InternodeDataTransport>, String> {
+    match configured_transport.map(str::trim).filter(|transport| !transport.is_empty()) {
+        None => Ok(Arc::new(TcpHttpInternodeDataTransport)),
+        Some(transport)
+            if transport.eq_ignore_ascii_case(DEFAULT_INTERNODE_DATA_TRANSPORT)
+                || transport.eq_ignore_ascii_case(INTERNODE_DATA_TRANSPORT_TCP) =>
+        {
+            Ok(Arc::new(TcpHttpInternodeDataTransport))
+        }
+        Some(transport) => Err(unsupported_transport_message(transport)),
+    }
+}
+
+pub fn build_internode_data_transport(configured_transport: Option<&str>) -> Result<Arc<dyn InternodeDataTransport>> {
+    build_internode_data_transport_result(configured_transport).map_err(Error::other)
+}
+
+pub fn build_internode_data_transport_from_env() -> Result<Arc<dyn InternodeDataTransport>> {
+    let configured_transport = std::env::var(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT).ok();
+    #[cfg(test)]
+    {
+        build_internode_data_transport(configured_transport.as_deref())
+    }
+
+    #[cfg(not(test))]
+    INTERNODE_DATA_TRANSPORT
+        .get_or_init(|| build_internode_data_transport_result(configured_transport.as_deref()))
+        .as_ref()
+        .map(Arc::clone)
+        .map_err(|err| Error::other(err.clone()))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn tcp_http_capabilities_are_behavior_preserving() {
+        let transport = TcpHttpInternodeDataTransport;
+
+        assert_eq!(transport.name(), DEFAULT_INTERNODE_DATA_TRANSPORT);
+        assert_eq!(
+            transport.capabilities(),
+            InternodeDataTransportCapabilities {
+                streaming_read: true,
+                streaming_write: true,
+                streaming_walk_dir: true,
+                ordered_delivery: true,
+                max_transfer_size: None,
+                fallback_supported: true,
+            }
+        );
+    }
+
+    #[test]
+    fn tcp_http_capabilities_are_conservative() {
+        let capabilities = TcpHttpInternodeDataTransport.capabilities();
+
+        assert!(capabilities.ordered_delivery);
+        assert_eq!(capabilities.max_transfer_size, None);
+        assert!(capabilities.fallback_supported);
+    }
+
+    #[test]
+    fn read_file_stream_url_encodes_query_values() {
+        let url = build_read_file_stream_url(&ReadStreamRequest {
+            endpoint: "http://node1:9000".to_string(),
+            disk: "http://node1:9000/data/rustfs0".to_string(),
+            volume: ".rustfs.sys".to_string(),
+            path: "pool.bin/../part.1".to_string(),
+            offset: 7,
+            length: 11,
+        });
+
+        assert_eq!(
+            url,
+            "http://node1:9000/rustfs/rpc/read_file_stream?disk=http%3A%2F%2Fnode1%3A9000%2Fdata%2Frustfs0&volume=.rustfs.sys&path=pool.bin%2F..%2Fpart.1&offset=7&length=11"
+        );
+    }
+
+    #[test]
+    fn put_file_stream_url_encodes_query_values() {
+        let url = build_put_file_stream_url(&WriteStreamRequest {
+            endpoint: "http://node1:9000".to_string(),
+            disk: "http://node1:9000/data/rustfs0".to_string(),
+            volume: "bucket".to_string(),
+            path: "object/part.1".to_string(),
+            append: false,
+            size: 4096,
+        });
+
+        assert_eq!(
+            url,
+            "http://node1:9000/rustfs/rpc/put_file_stream?disk=http%3A%2F%2Fnode1%3A9000%2Fdata%2Frustfs0&volume=bucket&path=object%2Fpart.1&append=false&size=4096"
+        );
+    }
+
+    #[test]
+    fn walk_dir_url_encodes_disk_ref() {
+        let url = build_walk_dir_url(&WalkDirStreamRequest {
+            endpoint: "http://node1:9000".to_string(),
+            disk: "http://node1:9000/data/rustfs0".to_string(),
+            body: Vec::new(),
+            stall_timeout: None,
+        });
+
+        assert_eq!(
+            url,
+            "http://node1:9000/rustfs/rpc/walk_dir?disk=http%3A%2F%2Fnode1%3A9000%2Fdata%2Frustfs0"
+        );
+    }
+
+    #[test]
+    fn transport_config_defaults_to_tcp_http() {
+        let transport = build_internode_data_transport(None).unwrap();
+
+        assert_eq!(transport.name(), DEFAULT_INTERNODE_DATA_TRANSPORT);
+    }
+
+    #[test]
+    fn transport_config_blank_value_falls_back_to_default() {
+        let transport = build_internode_data_transport(Some("   ")).unwrap();
+
+        assert_eq!(transport.name(), DEFAULT_INTERNODE_DATA_TRANSPORT);
+    }
+
+    #[test]
+    fn transport_config_accepts_tcp_aliases() {
+        for configured in [
+            DEFAULT_INTERNODE_DATA_TRANSPORT,
+            INTERNODE_DATA_TRANSPORT_TCP,
+            "TCP-HTTP",
+            "TCP",
+        ] {
+            let transport = build_internode_data_transport(Some(configured)).unwrap();
+
+            assert_eq!(transport.name(), DEFAULT_INTERNODE_DATA_TRANSPORT);
+        }
+    }
+
+    #[test]
+    fn transport_config_known_backends_are_current_oss_values() {
+        assert_eq!(
+            KNOWN_INTERNODE_DATA_TRANSPORT_BACKENDS,
+            &[DEFAULT_INTERNODE_DATA_TRANSPORT, INTERNODE_DATA_TRANSPORT_TCP]
+        );
+
+        for configured in KNOWN_INTERNODE_DATA_TRANSPORT_BACKENDS {
+            let transport = build_internode_data_transport(Some(configured)).unwrap();
+
+            assert_eq!(transport.name(), DEFAULT_INTERNODE_DATA_TRANSPORT);
+        }
+    }
+
+    #[test]
+    fn transport_config_rejects_unknown_backend() {
+        let err = build_internode_data_transport(Some("unsupported-backend")).expect_err("unknown backend should fail closed");
+
+        assert!(err.to_string().contains(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT));
+        assert!(err.to_string().contains("unsupported-backend"));
+        assert!(err.to_string().contains("supported values: tcp-http, tcp"));
+    }
+
+    #[test]
+    fn cached_transport_config_error_uses_raw_message() {
+        let err =
+            build_internode_data_transport_result(Some("unsupported-backend")).expect_err("unknown backend should fail closed");
+
+        assert!(!err.starts_with("io error "));
+        assert!(err.contains(ENV_RUSTFS_INTERNODE_DATA_TRANSPORT));
+        assert!(err.contains("unsupported-backend"));
+    }
+}

crates/ecstore/src/cluster/rpc/mod.rs

@@ -0,0 +1,38 @@
+// Copyright 2024 RustFS 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.
+
+pub(crate) mod client;
+pub(crate) mod context_propagation;
+pub(crate) mod http_auth;
+pub(crate) mod internode_data_transport;
+pub(crate) mod peer_rest_client;
+pub(crate) mod peer_s3_client;
+pub(crate) mod remote_disk;
+pub(crate) mod remote_locker;
+pub(crate) mod runtime_sources;
+
+pub use client::{
+    TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client, node_service_time_out_client_no_auth,
+};
+pub use http_auth::{TONIC_RPC_PREFIX, build_auth_headers, gen_signature_headers, verify_rpc_signature};
+#[cfg(test)]
+pub(crate) use internode_data_transport::TcpHttpInternodeDataTransport;
+pub use internode_data_transport::build_internode_data_transport_from_env;
+pub use peer_rest_client::{
+    PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC,
+};
+pub(crate) use peer_s3_client::heal_bucket_local_on_disks;
+pub use peer_s3_client::{LocalPeerS3Client, PeerS3Client, S3PeerSys};
+pub use remote_disk::RemoteDisk;
+pub use remote_locker::RemoteClient;

crates/ecstore/src/cluster/rpc/remote_locker.rs

@@ -0,0 +1,683 @@
+// Copyright 2024 RustFS 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 crate::rpc::client::{TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client};
+use async_trait::async_trait;
+use bytes::Bytes;
+use rustfs_lock::{
+    LockClient, LockError, LockInfo, LockRequest, LockResponse, LockStats, LockStatus, LockType, Result,
+    types::{LockId, LockMetadata, LockPriority},
+};
+use rustfs_protos::proto_gen::node_service::{BatchGenerallyLockRequest, GenerallyLockRequest, PingRequest};
+use rustfs_protos::{
+    ConnectionEvictionLogLevel, evict_failed_connection_with_log_level, models::PingBodyBuilder,
+    proto_gen::node_service::node_service_client::NodeServiceClient,
+};
+use std::time::Duration;
+use tokio::time::timeout;
+use tonic::Request;
+use tonic::service::interceptor::InterceptedService;
+use tonic::transport::Channel;
+use tracing::{debug, info, warn};
+
+/// Remote lock client implementation
+#[derive(Debug, Clone)]
+pub struct RemoteClient {
+    addr: String,
+}
+
+impl RemoteClient {
+    pub fn new(endpoint: String) -> Self {
+        Self { addr: endpoint }
+    }
+
+    pub fn from_url(url: url::Url) -> Self {
+        Self { addr: url.to_string() }
+    }
+
+    fn build_ping_request() -> PingRequest {
+        let mut fbb = flatbuffers::FlatBufferBuilder::new();
+        let payload = fbb.create_vector(b"health-check");
+        let mut builder = PingBodyBuilder::new(&mut fbb);
+        builder.add_payload(payload);
+        let root = builder.finish();
+        fbb.finish(root, None);
+
+        PingRequest {
+            version: 1,
+            body: Bytes::copy_from_slice(fbb.finished_data()),
+        }
+    }
+
+    /// Create a minimal LockRequest for unlock operations using only lock_id
+    fn create_unlock_request(lock_id: &LockId) -> LockRequest {
+        LockRequest {
+            lock_id: lock_id.clone(),
+            resource: lock_id.resource.clone(),
+            lock_type: LockType::Exclusive, // Type doesn't matter for unlock
+            owner: String::new(),           // Owner not needed, server uses lock_id
+            acquire_timeout: std::time::Duration::from_secs(30),
+            ttl: std::time::Duration::from_secs(300),
+            metadata: LockMetadata::default(),
+            priority: LockPriority::Normal,
+            deadlock_detection: false,
+            suppress_contention_logs: false,
+        }
+    }
+
+    pub async fn get_client(&self) -> Result<NodeServiceClient<InterceptedService<Channel, TonicInterceptor>>> {
+        node_service_time_out_client(&self.addr, TonicInterceptor::Signature(gen_tonic_signature_interceptor()))
+            .await
+            .map_err(|err| LockError::internal(format!("can not get client, err: {err}")))
+    }
+
+    fn is_scanner_leader_lock(resource_summary: &str) -> bool {
+        resource_summary == ".rustfs.sys/leader.lock@latest"
+    }
+
+    async fn evict_connection(&self, op: &'static str, reason: &str, resource_summary: &str) {
+        let log_level = if Self::is_scanner_leader_lock(resource_summary) {
+            debug!(
+                addr = %self.addr,
+                op,
+                reason,
+                resource_summary,
+                "Evicting cached remote lock connection for scanner leader-lock RPC failure"
+            );
+            ConnectionEvictionLogLevel::Debug
+        } else {
+            warn!(
+                addr = %self.addr,
+                op,
+                reason,
+                resource_summary,
+                "Evicting cached remote lock connection after RPC failure"
+            );
+            ConnectionEvictionLogLevel::Warn
+        };
+        evict_failed_connection_with_log_level(&self.addr, log_level).await;
+    }
+
+    fn summarize_resources(requests: &[LockRequest]) -> String {
+        const LIMIT: usize = 3;
+        let mut resources = requests
+            .iter()
+            .take(LIMIT)
+            .map(|request| request.resource.to_string())
+            .collect::<Vec<_>>();
+        if requests.len() > LIMIT {
+            resources.push(format!("... (+{} more)", requests.len() - LIMIT));
+        }
+        resources.join(", ")
+    }
+
+    fn rpc_timeout() -> Duration {
+        Duration::from_millis(
+            rustfs_utils::get_env_u64(
+                rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS,
+                rustfs_config::DEFAULT_OBJECT_LOCK_RPC_TIMEOUT_MS,
+            )
+            .max(1),
+        )
+    }
+
+    async fn execute_rpc<T, F>(&self, op: &'static str, resource_summary: &str, future: F) -> std::result::Result<T, LockError>
+    where
+        F: std::future::Future<Output = std::result::Result<T, tonic::Status>>,
+    {
+        let lock_timeout = Self::rpc_timeout();
+        match timeout(lock_timeout, future).await {
+            Ok(Ok(response)) => Ok(response),
+            Ok(Err(err)) => {
+                let reason = err.to_string();
+                if Self::is_scanner_leader_lock(resource_summary) {
+                    debug!(
+                        addr = %self.addr,
+                        op,
+                        timeout_ms = lock_timeout.as_millis(),
+                        resource_summary,
+                        tonic_code = ?err.code(),
+                        tonic_message = err.message(),
+                        "Remote lock RPC returned tonic error for scanner leader lock"
+                    );
+                } else {
+                    warn!(
+                        addr = %self.addr,
+                        op,
+                        timeout_ms = lock_timeout.as_millis(),
+                        resource_summary,
+                        tonic_code = ?err.code(),
+                        tonic_message = err.message(),
+                        "Remote lock RPC returned tonic error"
+                    );
+                }
+                self.evict_connection(op, &reason, resource_summary).await;
+                Err(LockError::internal(format!("{op} RPC failed: {reason}")))
+            }
+            Err(_) => {
+                let reason = format!("RPC timed out after {:?}", lock_timeout);
+                if Self::is_scanner_leader_lock(resource_summary) {
+                    debug!(
+                        addr = %self.addr,
+                        op,
+                        timeout_ms = lock_timeout.as_millis(),
+                        resource_summary,
+                        "Remote lock RPC timed out for scanner leader lock"
+                    );
+                } else {
+                    warn!(
+                        addr = %self.addr,
+                        op,
+                        timeout_ms = lock_timeout.as_millis(),
+                        resource_summary,
+                        "Remote lock RPC timed out"
+                    );
+                }
+                self.evict_connection(op, &reason, resource_summary).await;
+                Err(LockError::timeout(format!("remote lock RPC {op} on {}", self.addr), lock_timeout))
+            }
+        }
+    }
+
+    fn rpc_timeout_failure_response(request: &LockRequest, err: &LockError) -> LockResponse {
+        LockResponse::failure(format!("Remote lock RPC timed out: {err}"), request.acquire_timeout)
+    }
+
+    fn rpc_failure_response(_request: &LockRequest, err: &LockError) -> LockResponse {
+        LockResponse::failure(format!("Remote lock RPC failed: {err}"), Duration::ZERO)
+    }
+
+    fn rpc_failure_batch(requests: &[LockRequest], err: &LockError) -> Vec<LockResponse> {
+        requests
+            .iter()
+            .map(|request| Self::rpc_failure_response(request, err))
+            .collect()
+    }
+
+    fn rpc_timeout_failure_batch(requests: &[LockRequest], err: &LockError) -> Vec<LockResponse> {
+        requests
+            .iter()
+            .map(|request| Self::rpc_timeout_failure_response(request, err))
+            .collect()
+    }
+
+    fn build_lock_info(request: &LockRequest, lock_info_json: Option<String>) -> LockInfo {
+        if let Some(lock_info_json) = lock_info_json {
+            match serde_json::from_str::<LockInfo>(&lock_info_json) {
+                Ok(info) => info,
+                Err(e) => {
+                    warn!("Failed to deserialize lock_info from response: {}, using request data", e);
+                    LockInfo {
+                        id: request.lock_id.clone(),
+                        resource: request.resource.clone(),
+                        lock_type: request.lock_type,
+                        status: LockStatus::Acquired,
+                        owner: request.owner.clone(),
+                        acquired_at: std::time::SystemTime::now(),
+                        expires_at: std::time::SystemTime::now() + request.ttl,
+                        last_refreshed: std::time::SystemTime::now(),
+                        metadata: request.metadata.clone(),
+                        priority: request.priority,
+                        wait_start_time: None,
+                    }
+                }
+            }
+        } else {
+            LockInfo {
+                id: request.lock_id.clone(),
+                resource: request.resource.clone(),
+                lock_type: request.lock_type,
+                status: LockStatus::Acquired,
+                owner: request.owner.clone(),
+                acquired_at: std::time::SystemTime::now(),
+                expires_at: std::time::SystemTime::now() + request.ttl,
+                last_refreshed: std::time::SystemTime::now(),
+                metadata: request.metadata.clone(),
+                priority: request.priority,
+                wait_start_time: None,
+            }
+        }
+    }
+}
+
+#[async_trait]
+impl LockClient for RemoteClient {
+    async fn acquire_lock(&self, request: &LockRequest) -> Result<LockResponse> {
+        info!("remote acquire_exclusive for {}", request.resource);
+        let mut client = self.get_client().await?;
+        let resource_summary = request.resource.to_string();
+        let req = Request::new(GenerallyLockRequest {
+            args: serde_json::to_string(&request)
+                .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
+        });
+
+        let resp = match self.execute_rpc("lock", &resource_summary, client.lock(req)).await {
+            Ok(resp) => resp.into_inner(),
+            Err(err @ LockError::Timeout { .. }) => return Ok(Self::rpc_timeout_failure_response(request, &err)),
+            Err(err) => return Ok(Self::rpc_failure_response(request, &err)),
+        };
+
+        // Check if the lock acquisition was successful
+        if resp.success {
+            Ok(LockResponse::success(
+                Self::build_lock_info(request, resp.lock_info),
+                std::time::Duration::ZERO,
+            ))
+        } else {
+            // Lock acquisition failed
+            Ok(LockResponse::failure(
+                resp.error_info
+                    .unwrap_or_else(|| "Lock acquisition failed on remote server".to_string()),
+                std::time::Duration::ZERO,
+            ))
+        }
+    }
+
+    async fn acquire_locks_batch(&self, requests: &[LockRequest]) -> Result<Vec<LockResponse>> {
+        if requests.is_empty() {
+            return Ok(Vec::new());
+        }
+
+        let mut client = self.get_client().await?;
+        let resource_summary = Self::summarize_resources(requests);
+        let req = Request::new(BatchGenerallyLockRequest {
+            args: requests
+                .iter()
+                .map(|request| {
+                    serde_json::to_string(request).map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))
+                })
+                .collect::<Result<Vec<_>>>()?,
+        });
+
+        let resp = match self
+            .execute_rpc("lock_batch", &resource_summary, client.lock_batch(req))
+            .await
+        {
+            Ok(resp) => resp.into_inner(),
+            Err(err @ LockError::Timeout { .. }) => return Ok(Self::rpc_timeout_failure_batch(requests, &err)),
+            Err(err) => return Ok(Self::rpc_failure_batch(requests, &err)),
+        };
+
+        Ok(requests
+            .iter()
+            .enumerate()
+            .map(|(idx, request)| match resp.results.get(idx) {
+                Some(result) if result.success => {
+                    LockResponse::success(Self::build_lock_info(request, result.lock_info.clone()), std::time::Duration::ZERO)
+                }
+                Some(result) => LockResponse::failure(
+                    result
+                        .error_info
+                        .clone()
+                        .unwrap_or_else(|| "Lock acquisition failed on remote server".to_string()),
+                    std::time::Duration::ZERO,
+                ),
+                None => LockResponse::failure(
+                    format!("Lock batch response missing entry for request index {idx}"),
+                    std::time::Duration::ZERO,
+                ),
+            })
+            .collect())
+    }
+
+    async fn release(&self, lock_id: &LockId) -> Result<bool> {
+        info!("remote release for {}", lock_id);
+
+        let unlock_request = Self::create_unlock_request(lock_id);
+        let request_string = serde_json::to_string(&unlock_request)
+            .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?;
+        let mut client = self.get_client().await?;
+        let req = Request::new(GenerallyLockRequest { args: request_string });
+        let resp = client
+            .un_lock(req)
+            .await
+            .map_err(|e| LockError::internal(e.to_string()))?
+            .into_inner();
+        if let Some(error_info) = resp.error_info {
+            return Err(LockError::internal(error_info));
+        }
+        Ok(resp.success)
+    }
+
+    async fn release_locks_batch(&self, lock_ids: &[LockId]) -> Result<Vec<bool>> {
+        let mut client = self.get_client().await?;
+        let req = Request::new(BatchGenerallyLockRequest {
+            args: lock_ids
+                .iter()
+                .map(|lock_id| {
+                    serde_json::to_string(&Self::create_unlock_request(lock_id))
+                        .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))
+                })
+                .collect::<Result<Vec<_>>>()?,
+        });
+
+        let resp = client
+            .un_lock_batch(req)
+            .await
+            .map_err(|e| LockError::internal(e.to_string()))?
+            .into_inner();
+
+        Ok(lock_ids
+            .iter()
+            .enumerate()
+            .map(|(idx, _)| resp.results.get(idx).map(|result| result.success).unwrap_or(false))
+            .collect())
+    }
+
+    async fn refresh(&self, lock_id: &LockId) -> Result<bool> {
+        info!("remote refresh for {}", lock_id);
+        let refresh_request = Self::create_unlock_request(lock_id);
+        let mut client = self.get_client().await?;
+        let req = Request::new(GenerallyLockRequest {
+            args: serde_json::to_string(&refresh_request)
+                .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
+        });
+        let resp = client
+            .refresh(req)
+            .await
+            .map_err(|e| LockError::internal(e.to_string()))?
+            .into_inner();
+        if let Some(error_info) = resp.error_info {
+            return Err(LockError::internal(error_info));
+        }
+        Ok(resp.success)
+    }
+
+    async fn force_release(&self, lock_id: &LockId) -> Result<bool> {
+        info!("remote force_release for {}", lock_id);
+        let force_request = Self::create_unlock_request(lock_id);
+        let mut client = self.get_client().await?;
+        let req = Request::new(GenerallyLockRequest {
+            args: serde_json::to_string(&force_request)
+                .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
+        });
+        let resp = client
+            .force_un_lock(req)
+            .await
+            .map_err(|e| LockError::internal(e.to_string()))?
+            .into_inner();
+        if let Some(error_info) = resp.error_info {
+            return Err(LockError::internal(error_info));
+        }
+        Ok(resp.success)
+    }
+
+    async fn check_status(&self, lock_id: &LockId) -> Result<Option<LockInfo>> {
+        info!("remote check_status for {}", lock_id);
+
+        // Since there's no direct status query in the gRPC service,
+        // we attempt a non-blocking lock acquisition to check if the resource is available
+        let status_request = Self::create_unlock_request(lock_id);
+        let mut client = self.get_client().await?;
+
+        // Try to acquire a very short-lived lock to test availability
+        let req = Request::new(GenerallyLockRequest {
+            args: serde_json::to_string(&status_request)
+                .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
+        });
+
+        // Try exclusive lock first with very short timeout
+        let resp = client.lock(req).await;
+
+        match resp {
+            Ok(response) => {
+                let resp = response.into_inner();
+                if resp.success {
+                    // If we successfully acquired the lock, the resource was free
+                    // Immediately release it
+                    let release_req = Request::new(GenerallyLockRequest {
+                        args: serde_json::to_string(&status_request)
+                            .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
+                    });
+                    let _ = client.un_lock(release_req).await; // Best effort release
+
+                    // Return None since no one was holding the lock
+                    Ok(None)
+                } else {
+                    // Lock acquisition failed, meaning someone is holding it
+                    // We can't determine the exact details remotely, so return a generic status
+                    Ok(Some(LockInfo {
+                        id: lock_id.clone(),
+                        resource: lock_id.resource.clone(),
+                        lock_type: LockType::Exclusive, // We can't know the exact type
+                        status: LockStatus::Acquired,
+                        owner: "unknown".to_string(), // Remote client can't determine owner
+                        acquired_at: std::time::SystemTime::now(),
+                        expires_at: std::time::SystemTime::now() + std::time::Duration::from_secs(3600),
+                        last_refreshed: std::time::SystemTime::now(),
+                        metadata: LockMetadata::default(),
+                        priority: LockPriority::Normal,
+                        wait_start_time: None,
+                    }))
+                }
+            }
+            Err(_) => {
+                // Communication error or lock is held
+                Ok(Some(LockInfo {
+                    id: lock_id.clone(),
+                    resource: lock_id.resource.clone(),
+                    lock_type: LockType::Exclusive,
+                    status: LockStatus::Acquired,
+                    owner: "unknown".to_string(),
+                    acquired_at: std::time::SystemTime::now(),
+                    expires_at: std::time::SystemTime::now() + std::time::Duration::from_secs(3600),
+                    last_refreshed: std::time::SystemTime::now(),
+                    metadata: LockMetadata::default(),
+                    priority: LockPriority::Normal,
+                    wait_start_time: None,
+                }))
+            }
+        }
+    }
+
+    async fn get_stats(&self) -> Result<LockStats> {
+        info!("remote get_stats from {}", self.addr);
+
+        // Since there's no direct statistics endpoint in the gRPC service,
+        // we return basic stats indicating this is a remote client
+        let stats = LockStats {
+            last_updated: std::time::SystemTime::now(),
+            ..Default::default()
+        };
+
+        // We could potentially enhance this by:
+        // 1. Keeping local counters of operations performed
+        // 2. Adding a stats gRPC method to the service
+        // 3. Querying server health endpoints
+
+        // For now, return minimal stats indicating remote connectivity
+        Ok(stats)
+    }
+
+    async fn close(&self) -> Result<()> {
+        Ok(())
+    }
+
+    async fn is_online(&self) -> bool {
+        // Use Ping interface to test if remote service is online
+        let mut client = match self.get_client().await {
+            Ok(client) => client,
+            Err(_) => {
+                info!("remote client {} connection failed", self.addr);
+                return false;
+            }
+        };
+
+        let ping_req = Request::new(Self::build_ping_request());
+
+        match client.ping(ping_req).await {
+            Ok(_) => {
+                info!("remote client {} is online", self.addr);
+                true
+            }
+            Err(_) => {
+                info!("remote client {} ping failed", self.addr);
+                false
+            }
+        }
+    }
+
+    async fn is_local(&self) -> bool {
+        false
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::runtime_sources;
+    use rustfs_lock::{ObjectKey, types::LockPriority};
+    use tokio::net::TcpListener;
+    use tokio::task::JoinHandle;
+    use tonic::transport::Endpoint as TonicEndpoint;
+
+    async fn spawn_hanging_listener() -> Option<(String, JoinHandle<()>)> {
+        let listener = match TcpListener::bind("127.0.0.1:0").await {
+            Ok(listener) => listener,
+            Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return None,
+            Err(err) => panic!("test listener should bind: {err}"),
+        };
+        let addr = format!("http://{}", listener.local_addr().expect("listener local address should be available"));
+        let task = tokio::spawn(async move {
+            if let Ok((stream, _)) = listener.accept().await {
+                let _stream = stream;
+                tokio::time::sleep(Duration::from_secs(2)).await;
+            }
+        });
+        Some((addr, task))
+    }
+
+    async fn cache_lazy_channel(addr: &str) {
+        let channel = TonicEndpoint::from_shared(addr.to_string()).unwrap().connect_lazy();
+        runtime_sources::cache_test_node_channel(addr.to_string(), channel).await;
+    }
+
+    fn ensure_test_rpc_secret() {
+        runtime_sources::ensure_test_rpc_secret();
+    }
+
+    fn test_lock_request(timeout_duration: Duration) -> LockRequest {
+        LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner-a")
+            .with_acquire_timeout(timeout_duration)
+            .with_priority(LockPriority::Normal)
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn test_remote_client_acquire_lock_uses_rpc_timeout_and_evicts_connection() {
+        ensure_test_rpc_secret();
+        let Some((addr, accept_task)) = spawn_hanging_listener().await else {
+            return;
+        };
+        cache_lazy_channel(&addr).await;
+        assert!(runtime_sources::test_node_channel_is_cached(&addr).await);
+
+        temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("50"))], async {
+            let client = RemoteClient::new(addr.clone());
+            let request = test_lock_request(Duration::from_millis(5));
+            let started_at = tokio::time::Instant::now();
+
+            let response = client.acquire_lock(&request).await.unwrap();
+            let elapsed = started_at.elapsed();
+
+            assert!(
+                elapsed >= Duration::from_millis(40),
+                "remote lock RPC should use configured transport timeout, got {elapsed:?}"
+            );
+            assert!(
+                elapsed < Duration::from_secs(1),
+                "test RPC timeout should keep the test fast, got {elapsed:?}"
+            );
+            assert!(!response.success, "timed out lock acquisition should fail");
+            assert!(
+                response
+                    .error
+                    .as_deref()
+                    .is_some_and(|error| error.contains("Remote lock RPC timed out")),
+                "expected remote RPC timeout marker, got {:?}",
+                response.error
+            );
+            assert!(
+                !runtime_sources::test_node_channel_is_cached(&addr).await,
+                "transport timeout should evict cached connection"
+            );
+        })
+        .await;
+
+        accept_task.abort();
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn test_remote_client_acquire_locks_batch_uses_rpc_timeout_and_evicts_connection() {
+        ensure_test_rpc_secret();
+        let Some((addr, accept_task)) = spawn_hanging_listener().await else {
+            return;
+        };
+        cache_lazy_channel(&addr).await;
+        assert!(runtime_sources::test_node_channel_is_cached(&addr).await);
+
+        temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("50"))], async {
+            let client = RemoteClient::new(addr.clone());
+            let requests = vec![test_lock_request(Duration::from_millis(5))];
+            let started_at = tokio::time::Instant::now();
+
+            let responses = client.acquire_locks_batch(&requests).await.unwrap();
+            let elapsed = started_at.elapsed();
+
+            assert!(
+                elapsed >= Duration::from_millis(40),
+                "remote batch lock RPC should use configured transport timeout, got {elapsed:?}"
+            );
+            assert!(
+                elapsed < Duration::from_secs(1),
+                "test RPC timeout should keep the test fast, got {elapsed:?}"
+            );
+            assert_eq!(responses.len(), 1);
+            assert!(!responses[0].success, "timed out batch lock acquisition should fail");
+            assert!(
+                responses[0]
+                    .error
+                    .as_deref()
+                    .is_some_and(|error| error.contains("Remote lock RPC timed out")),
+                "expected remote RPC timeout marker, got {:?}",
+                responses[0].error
+            );
+            assert!(
+                !runtime_sources::test_node_channel_is_cached(&addr).await,
+                "batch transport timeout should evict cached connection"
+            );
+        })
+        .await;
+
+        accept_task.abort();
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn test_remote_client_rpc_timeout_honors_configured_deadline() {
+        temp_env::with_var(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, None::<&str>, || {
+            assert_eq!(
+                RemoteClient::rpc_timeout(),
+                Duration::from_millis(rustfs_config::DEFAULT_OBJECT_LOCK_RPC_TIMEOUT_MS)
+            );
+        });
+        temp_env::with_var(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("50"), || {
+            assert_eq!(RemoteClient::rpc_timeout(), Duration::from_millis(50));
+        });
+        temp_env::with_var(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("0"), || {
+            assert_eq!(RemoteClient::rpc_timeout(), Duration::from_millis(1));
+        });
+    }
+}

crates/ecstore/src/cluster/rpc/runtime_sources.rs

@@ -0,0 +1,83 @@
+// Copyright 2024 RustFS 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 rustfs_io_metrics::internode_metrics::{
+    INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_OPERATION_PUT_FILE_STREAM,
+    INTERNODE_TRANSPORT_BACKEND_GRPC, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics,
+};
+
+#[cfg(test)]
+use rustfs_io_metrics::internode_metrics::InternodeMetricsSnapshot;
+
+pub(crate) fn record_remote_disk_open_write_retry(classification: &'static str) {
+    global_internode_metrics().record_retry_for_operation_and_backend(
+        INTERNODE_OPERATION_PUT_FILE_STREAM,
+        INTERNODE_TRANSPORT_BACKEND_TCP_HTTP,
+        classification,
+    );
+}
+
+pub(crate) fn record_remote_disk_open_write_retry_success(classification: &'static str) {
+    global_internode_metrics().record_retry_success_for_operation_and_backend(
+        INTERNODE_OPERATION_PUT_FILE_STREAM,
+        INTERNODE_TRANSPORT_BACKEND_TCP_HTTP,
+        classification,
+    );
+}
+
+pub(crate) fn record_remote_disk_grpc_write_all_error() {
+    global_internode_metrics()
+        .record_error_for_operation_and_backend(INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC);
+}
+
+pub(crate) fn record_remote_disk_grpc_write_all_request() {
+    global_internode_metrics()
+        .record_outgoing_request_for_operation_and_backend(INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC);
+}
+
+pub(crate) fn record_remote_disk_grpc_write_all_sent_bytes(bytes: usize) {
+    global_internode_metrics().record_sent_bytes_for_operation_and_backend(
+        INTERNODE_OPERATION_GRPC_WRITE_ALL,
+        INTERNODE_TRANSPORT_BACKEND_GRPC,
+        bytes,
+    );
+}
+
+pub(crate) fn record_remote_disk_grpc_read_all_error() {
+    global_internode_metrics()
+        .record_error_for_operation_and_backend(INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC);
+}
+
+pub(crate) fn record_remote_disk_grpc_read_all_request() {
+    global_internode_metrics()
+        .record_outgoing_request_for_operation_and_backend(INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC);
+}
+
+pub(crate) fn record_remote_disk_grpc_read_all_recv_bytes(bytes: usize) {
+    global_internode_metrics().record_recv_bytes_for_operation_and_backend(
+        INTERNODE_OPERATION_GRPC_READ_ALL,
+        INTERNODE_TRANSPORT_BACKEND_GRPC,
+        bytes,
+    );
+}
+
+#[cfg(test)]
+pub(crate) fn reset_internode_metrics_for_test() {
+    global_internode_metrics().reset_for_test();
+}
+
+#[cfg(test)]
+pub(crate) fn internode_metrics_snapshot_for_test() -> InternodeMetricsSnapshot {
+    global_internode_metrics().snapshot()
+}