perf(storage): optimize internode RPC transfer path (#2262)

Co-authored-by: momoda693 <momoda693@gmail.com>

Cargo.lock

@@ -7914,6 +7914,7 @@ name = "rustfs-rio"
 version = "0.0.5"
 dependencies = [
  "aes-gcm",
+ "axum",
  "base64 0.22.1",
  "bytes",
  "crc-fast",
@@ -7921,10 +7922,12 @@ dependencies = [
  "futures",
  "hex-simd",
  "http 1.4.0",
+ "http-body-util",
  "md-5 0.11.0-rc.5",
  "pin-project-lite",
  "rand 0.10.0",
  "reqwest 0.13.2",
+ "rustfs-common",
  "rustfs-config",
  "rustfs-utils",
  "s3s",

crates/audit/src/registry.rs

@@ -313,6 +313,12 @@ impl AuditRegistry {
                 }
             }
 
+            if &new_config == config {
+                info!("Audit target configuration unchanged, skip persisting server config");
+                info!(count = successful_targets.len(), "All target processing completed");
+                return Ok(successful_targets);
+            }
+
             let Some(store) = rustfs_ecstore::global::new_object_layer_fn() else {
                 return Err(AuditError::StorageNotAvailable(
                     "Failed to save target configuration: server storage not initialized".to_string(),

crates/audit/tests/performance_test.rs

@@ -159,7 +159,8 @@ async fn test_audit_log_dispatch_performance() {
 
     let start = Instant::now();
 
-    // Dispatch audit log (should be fast since no targets are configured)
+    // Dispatch audit log against an unstarted system state. Empty config keeps
+    // the audit system stopped, so dispatch should fail fast without targets.
     let result = system.dispatch(Arc::new(audit_entry)).await;
     let elapsed = start.elapsed();
 
@@ -168,8 +169,10 @@ async fn test_audit_log_dispatch_performance() {
     // Should be very fast (sub-millisecond for no targets)
     assert!(elapsed < Duration::from_millis(100), "Dispatch took too long: {elapsed:?}");
 
-    // Should succeed even with no targets
-    assert!(result.is_ok(), "Dispatch should succeed with no targets");
+    assert!(
+        matches!(result, Err(AuditError::NotInitialized(_))),
+        "Dispatch on a stopped system should return NotInitialized, got: {result:?}"
+    );
 
     // Clean up
     let _ = system.close().await;
@@ -186,11 +189,11 @@ async fn test_system_state_transitions() {
     let config = rustfs_ecstore::config::Config(std::collections::HashMap::new());
     let start_result = system.start(config).await;
 
-    // Should be running (or failed due to server storage)
+    // Empty config keeps the audit system stopped even when start() succeeds.
     let state = system.get_state().await;
     match start_result {
         Ok(_) => {
-            assert_eq!(state, rustfs_audit::system::AuditSystemState::Running);
+            assert_eq!(state, rustfs_audit::system::AuditSystemState::Stopped);
         }
         Err(_) => {
             // Expected in test environment due to server storage not being initialized

crates/audit/tests/system_integration_test.rs

@@ -29,27 +29,21 @@ async fn test_complete_audit_system_lifecycle() {
     assert_eq!(system.get_state().await, system::AuditSystemState::Stopped);
     assert!(!system.is_running().await);
 
-    // 2. Start with empty config (will fail due to no server storage in test)
+    // 2. Start with empty config. The current implementation returns Ok(())
+    // but keeps the system stopped when no audit targets are enabled.
     let config = Config(HashMap::new());
     let start_result = system.start(config).await;
 
-    // Should fail in test environment but state handling should work
+    // State handling should remain consistent for both empty-config success and
+    // storage-unavailable failure paths.
     match start_result {
         Err(AuditError::StorageNotAvailable(_)) => {
             // Expected in test environment
             assert_eq!(system.get_state().await, system::AuditSystemState::Stopped);
         }
         Ok(_) => {
-            // If it somehow succeeds, verify running state
-            assert_eq!(system.get_state().await, system::AuditSystemState::Running);
-            assert!(system.is_running().await);
-
-            // Test pause/resume
-            system.pause().await.expect("Should pause successfully");
-            assert_eq!(system.get_state().await, system::AuditSystemState::Paused);
-
-            system.resume().await.expect("Should resume successfully");
-            assert_eq!(system.get_state().await, system::AuditSystemState::Running);
+            assert_eq!(system.get_state().await, system::AuditSystemState::Stopped);
+            assert!(!system.is_running().await);
         }
         Err(e) => {
             panic!("Unexpected error: {e}");

crates/common/src/internode_metrics.rs

@@ -0,0 +1,170 @@
+// 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 metrics::{counter, gauge};
+use std::sync::{
+    Arc, LazyLock,
+    atomic::{AtomicU64, Ordering},
+};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct InternodeMetricsSnapshot {
+    pub sent_bytes_total: u64,
+    pub recv_bytes_total: u64,
+    pub outgoing_requests_total: u64,
+    pub incoming_requests_total: u64,
+    pub errors_total: u64,
+    pub dial_errors_total: u64,
+    pub dial_avg_time_nanos: u64,
+    pub last_dial_unix_millis: u64,
+}
+
+#[derive(Debug, Default)]
+pub struct InternodeMetrics {
+    sent_bytes_total: AtomicU64,
+    recv_bytes_total: AtomicU64,
+    outgoing_requests_total: AtomicU64,
+    incoming_requests_total: AtomicU64,
+    errors_total: AtomicU64,
+    dial_errors_total: AtomicU64,
+    dial_total_time_nanos: AtomicU64,
+    dial_samples_total: AtomicU64,
+    last_dial_unix_millis: AtomicU64,
+}
+
+impl InternodeMetrics {
+    pub fn record_sent_bytes(&self, bytes: usize) {
+        let bytes = bytes as u64;
+        if bytes == 0 {
+            return;
+        }
+        self.sent_bytes_total.fetch_add(bytes, Ordering::Relaxed);
+        counter!("rustfs.internode.sent.bytes.total").increment(bytes);
+    }
+
+    pub fn record_recv_bytes(&self, bytes: usize) {
+        let bytes = bytes as u64;
+        if bytes == 0 {
+            return;
+        }
+        self.recv_bytes_total.fetch_add(bytes, Ordering::Relaxed);
+        counter!("rustfs.internode.recv.bytes.total").increment(bytes);
+    }
+
+    pub fn record_outgoing_request(&self) {
+        self.outgoing_requests_total.fetch_add(1, Ordering::Relaxed);
+        counter!("rustfs.internode.requests.outgoing.total").increment(1);
+    }
+
+    pub fn record_incoming_request(&self) {
+        self.incoming_requests_total.fetch_add(1, Ordering::Relaxed);
+        counter!("rustfs.internode.requests.incoming.total").increment(1);
+    }
+
+    pub fn record_error(&self) {
+        self.errors_total.fetch_add(1, Ordering::Relaxed);
+        counter!("rustfs.internode.errors.total").increment(1);
+    }
+
+    pub fn record_dial_result(&self, duration: Duration, success: bool) {
+        let elapsed_nanos = duration.as_nanos().min(u128::from(u64::MAX)) as u64;
+        self.dial_total_time_nanos.fetch_add(elapsed_nanos, Ordering::Relaxed);
+        let samples = self.dial_samples_total.fetch_add(1, Ordering::Relaxed) + 1;
+        let total = self.dial_total_time_nanos.load(Ordering::Relaxed);
+        gauge!("rustfs.internode.dial.avg_time.nanos").set(total as f64 / samples as f64);
+
+        if !success {
+            self.dial_errors_total.fetch_add(1, Ordering::Relaxed);
+            counter!("rustfs.internode.dial.errors.total").increment(1);
+        }
+
+        let now_ms = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_millis()
+            .min(u128::from(u64::MAX)) as u64;
+        self.last_dial_unix_millis.store(now_ms, Ordering::Relaxed);
+    }
+
+    pub fn snapshot(&self) -> InternodeMetricsSnapshot {
+        let dial_samples_total = self.dial_samples_total.load(Ordering::Relaxed);
+        let dial_total_time_nanos = self.dial_total_time_nanos.load(Ordering::Relaxed);
+        let dial_avg_time_nanos = if dial_samples_total == 0 {
+            0
+        } else {
+            dial_total_time_nanos / dial_samples_total
+        };
+
+        InternodeMetricsSnapshot {
+            sent_bytes_total: self.sent_bytes_total.load(Ordering::Relaxed),
+            recv_bytes_total: self.recv_bytes_total.load(Ordering::Relaxed),
+            outgoing_requests_total: self.outgoing_requests_total.load(Ordering::Relaxed),
+            incoming_requests_total: self.incoming_requests_total.load(Ordering::Relaxed),
+            errors_total: self.errors_total.load(Ordering::Relaxed),
+            dial_errors_total: self.dial_errors_total.load(Ordering::Relaxed),
+            dial_avg_time_nanos,
+            last_dial_unix_millis: self.last_dial_unix_millis.load(Ordering::Relaxed),
+        }
+    }
+
+    #[doc(hidden)]
+    pub fn reset_for_test(&self) {
+        self.sent_bytes_total.store(0, Ordering::Relaxed);
+        self.recv_bytes_total.store(0, Ordering::Relaxed);
+        self.outgoing_requests_total.store(0, Ordering::Relaxed);
+        self.incoming_requests_total.store(0, Ordering::Relaxed);
+        self.errors_total.store(0, Ordering::Relaxed);
+        self.dial_errors_total.store(0, Ordering::Relaxed);
+        self.dial_total_time_nanos.store(0, Ordering::Relaxed);
+        self.dial_samples_total.store(0, Ordering::Relaxed);
+        self.last_dial_unix_millis.store(0, Ordering::Relaxed);
+    }
+}
+
+pub fn global_internode_metrics() -> &'static Arc<InternodeMetrics> {
+    static GLOBAL_INTERNODE_METRICS: LazyLock<Arc<InternodeMetrics>> = LazyLock::new(|| Arc::new(InternodeMetrics::default()));
+    &GLOBAL_INTERNODE_METRICS
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn snapshot_reports_recorded_values() {
+        let metrics = global_internode_metrics();
+        metrics.reset_for_test();
+
+        metrics.record_sent_bytes(64);
+        metrics.record_recv_bytes(32);
+        metrics.record_outgoing_request();
+        metrics.record_incoming_request();
+        metrics.record_error();
+        metrics.record_dial_result(Duration::from_millis(9), true);
+        metrics.record_dial_result(Duration::from_millis(3), false);
+
+        let snapshot = metrics.snapshot();
+        assert_eq!(snapshot.sent_bytes_total, 64);
+        assert_eq!(snapshot.recv_bytes_total, 32);
+        assert_eq!(snapshot.outgoing_requests_total, 1);
+        assert_eq!(snapshot.incoming_requests_total, 1);
+        assert_eq!(snapshot.errors_total, 1);
+        assert_eq!(snapshot.dial_errors_total, 1);
+        assert_eq!(snapshot.dial_avg_time_nanos, 6_000_000);
+        assert!(snapshot.last_dial_unix_millis > 0);
+
+        metrics.reset_for_test();
+    }
+}

crates/common/src/lib.rs

@@ -17,6 +17,7 @@ pub mod bucket_stats;
 pub mod data_usage;
 pub mod globals;
 pub mod heal_channel;
+pub mod internode_metrics;
 pub mod last_minute;
 pub mod metrics;
 mod readiness;

crates/e2e_test/src/reliant/grpc_lock_client.rs

@@ -61,6 +61,7 @@ impl GrpcLockClient {
             metadata: LockMetadata::default(),
             priority: LockPriority::Normal,
             deadlock_detection: false,
+            suppress_contention_logs: false,
         }
     }
 }

crates/ecstore/src/config/com.rs

@@ -15,6 +15,7 @@
 use crate::config::{Config, GLOBAL_STORAGE_CLASS, storageclass};
 use crate::disk::{MIGRATING_META_BUCKET, RUSTFS_META_BUCKET};
 use crate::error::{Error, Result};
+use crate::global::is_first_cluster_node_local;
 use crate::store_api::{ObjectInfo, ObjectOptions, PutObjReader, StorageAPI};
 use http::HeaderMap;
 use rustfs_config::{DEFAULT_DELIMITER, RUSTFS_REGION};
@@ -44,6 +45,19 @@ pub async fn read_config<S: StorageAPI>(api: Arc<S>, file: &str) -> Result<Vec<u
     Ok(data)
 }
 
+pub async fn read_config_no_lock<S: StorageAPI>(api: Arc<S>, file: &str) -> Result<Vec<u8>> {
+    let (data, _obj) = read_config_with_metadata(
+        api,
+        file,
+        &ObjectOptions {
+            no_lock: true,
+            ..Default::default()
+        },
+    )
+    .await?;
+    Ok(data)
+}
+
 pub async fn read_config_with_metadata<S: StorageAPI>(
     api: Arc<S>,
     file: &str,
@@ -287,7 +301,14 @@ fn is_object_not_found(err: &Error) -> bool {
 pub async fn try_migrate_server_config<S: StorageAPI>(api: Arc<S>) {
     let config_file = get_config_file();
     match api
-        .get_object_info(RUSTFS_META_BUCKET, &config_file, &ObjectOptions::default())
+        .get_object_info(
+            RUSTFS_META_BUCKET,
+            &config_file,
+            &ObjectOptions {
+                no_lock: true,
+                ..Default::default()
+            },
+        )
         .await
     {
         Ok(_) => {
@@ -360,7 +381,13 @@ pub async fn try_migrate_server_config<S: StorageAPI>(api: Arc<S>) {
 /// Handle the situation where the configuration file does not exist, create and save a new configuration
 async fn handle_missing_config<S: StorageAPI>(api: Arc<S>, context: &str) -> Result<Config> {
     warn!("Configuration not found ({}): Start initializing new configuration", context);
-    let cfg = new_and_save_server_config(api).await?;
+    let cfg = if is_first_cluster_node_local().await {
+        new_and_save_server_config(api.clone()).await?
+    } else {
+        let mut cfg = new_server_config();
+        lookup_configs(&mut cfg, api).await;
+        cfg
+    };
     warn!("Configuration initialization complete ({})", context);
     Ok(cfg)
 }
@@ -375,7 +402,7 @@ pub async fn read_config_without_migrate<S: StorageAPI>(api: Arc<S>) -> Result<C
     let config_file = get_config_file();
 
     // Try to read the configuration file
-    match read_config(api.clone(), &config_file).await {
+    match read_config_no_lock(api.clone(), &config_file).await {
         Ok(data) => read_server_config(api, &data).await,
         Err(Error::ConfigNotFound) => handle_missing_config(api, "Read the main configuration").await,
         Err(err) => handle_config_read_error(err, &config_file),
@@ -389,7 +416,7 @@ async fn read_server_config<S: StorageAPI>(api: Arc<S>, data: &[u8]) -> Result<C
         warn!("Received empty configuration data, try to reread from '{}'", config_file);
 
         // Try to read the configuration again
-        match read_config(api.clone(), &config_file).await {
+        match read_config_no_lock(api.clone(), &config_file).await {
             Ok(cfg_data) => {
                 // TODO: decrypt
                 let cfg = decode_server_config_blob(&cfg_data)?;

crates/ecstore/src/config/mod.rs

@@ -79,7 +79,7 @@ pub async fn try_migrate_server_config(api: Arc<ECStore>) {
     com::try_migrate_server_config(api).await
 }
 
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, PartialEq, Eq)]
 pub struct KV {
     pub key: String,
     pub value: String,
@@ -87,7 +87,7 @@ pub struct KV {
     pub hidden_if_empty: bool,
 }
 
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, PartialEq, Eq)]
 pub struct KVS(pub Vec<KV>);
 
 impl Default for KVS {
@@ -163,7 +163,7 @@ impl KVS {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Config(pub HashMap<String, HashMap<String, KVS>>);
 
 impl Default for Config {

crates/ecstore/src/disk/disk_store.rs

@@ -18,6 +18,7 @@ use crate::disk::{
     WalkDirOptions,
     local::{LocalDisk, ScanGuard},
 };
+use crate::global::GLOBAL_LOCAL_DISK_ID_MAP;
 use bytes::Bytes;
 use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo};
 use std::{
@@ -410,7 +411,19 @@ impl LocalDiskWrapper {
     /// Set the disk ID
     pub async fn set_disk_id_internal(&self, id: Option<Uuid>) -> Result<()> {
         let mut disk_id = self.disk_id.write().await;
+        let previous = *disk_id;
         *disk_id = id;
+        drop(disk_id);
+
+        if self.disk.is_local() {
+            let mut disk_id_map = GLOBAL_LOCAL_DISK_ID_MAP.write().await;
+            if let Some(previous_id) = previous {
+                disk_id_map.remove(&previous_id);
+            }
+            if let Some(current_id) = id {
+                disk_id_map.insert(current_id, self.disk.endpoint().to_string());
+            }
+        }
         Ok(())
     }
 

crates/ecstore/src/erasure_coding/bitrot.rs

@@ -15,6 +15,7 @@
 use bytes::Bytes;
 use pin_project_lite::pin_project;
 use rustfs_utils::HashAlgorithm;
+use std::io::IoSlice;
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use tracing::error;
 use uuid::Uuid;
@@ -155,23 +156,49 @@ where
                 error!("bitrot writer write hash error: hash is empty");
                 return Err(std::io::Error::new(std::io::ErrorKind::InvalidInput, "hash is empty"));
             }
-            self.inner.write_all(hash.as_ref()).await?;
+            write_all_vectored(&mut self.inner, hash.as_ref(), buf).await?;
+        } else {
+            self.inner.write_all(buf).await?;
         }
 
-        self.inner.write_all(buf).await?;
-
-        self.inner.flush().await?;
-
         let n = buf.len();
 
         Ok(n)
     }
 
     pub async fn shutdown(&mut self) -> std::io::Result<()> {
+        self.inner.flush().await?;
         self.inner.shutdown().await
     }
 }
 
+async fn write_all_vectored<W>(writer: &mut W, hash: &[u8], data: &[u8]) -> std::io::Result<()>
+where
+    W: AsyncWrite + Unpin,
+{
+    let mut hash_offset = 0;
+    let mut data_offset = 0;
+
+    while hash_offset < hash.len() || data_offset < data.len() {
+        let slices = [IoSlice::new(&hash[hash_offset..]), IoSlice::new(&data[data_offset..])];
+        let written = writer.write_vectored(&slices).await?;
+        if written == 0 {
+            return Err(std::io::Error::new(std::io::ErrorKind::WriteZero, "failed to write hash and data"));
+        }
+
+        let hash_remaining = hash.len() - hash_offset;
+        if written < hash_remaining {
+            hash_offset += written;
+            continue;
+        }
+
+        hash_offset = hash.len();
+        data_offset += written - hash_remaining;
+    }
+
+    Ok(())
+}
+
 pub fn bitrot_shard_file_size(size: usize, shard_size: usize, algo: HashAlgorithm) -> usize {
     if algo != HashAlgorithm::HighwayHash256S && algo != HashAlgorithm::HighwayHash256SLegacy {
         return size;
@@ -292,6 +319,33 @@ impl AsyncWrite for CustomWriter {
             }
         }
     }
+
+    fn poll_write_vectored(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        bufs: &[IoSlice<'_>],
+    ) -> std::task::Poll<std::io::Result<usize>> {
+        match self.get_mut() {
+            Self::InlineBuffer(data) => {
+                let total = bufs.iter().map(|buf| buf.len()).sum::<usize>();
+                for buf in bufs {
+                    data.extend_from_slice(buf);
+                }
+                std::task::Poll::Ready(Ok(total))
+            }
+            Self::Other(writer) => {
+                let pinned_writer = std::pin::Pin::new(writer.as_mut());
+                pinned_writer.poll_write_vectored(cx, bufs)
+            }
+        }
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        match self {
+            Self::InlineBuffer(_) => true,
+            Self::Other(writer) => writer.is_write_vectored(),
+        }
+    }
 }
 
 /// Wrapper around BitrotWriter that uses our custom writer
@@ -361,7 +415,74 @@ mod tests {
     use super::BitrotReader;
     use super::BitrotWriter;
     use rustfs_utils::HashAlgorithm;
-    use std::io::Cursor;
+    use std::io::{Cursor, IoSlice};
+    use std::sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    };
+    use std::task::{Context, Poll};
+    use tokio::io::AsyncWrite;
+
+    #[derive(Default)]
+    struct VectoredCountingWriter {
+        vectored_writes: Arc<AtomicUsize>,
+        writes: Vec<u8>,
+    }
+
+    impl AsyncWrite for VectoredCountingWriter {
+        fn poll_write(self: std::pin::Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &[u8]) -> Poll<std::io::Result<usize>> {
+            Poll::Ready(Err(std::io::Error::other("poll_write should not be used")))
+        }
+
+        fn poll_flush(self: std::pin::Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_shutdown(self: std::pin::Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_write_vectored(
+            mut self: std::pin::Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+            bufs: &[IoSlice<'_>],
+        ) -> Poll<std::io::Result<usize>> {
+            self.vectored_writes.fetch_add(1, Ordering::SeqCst);
+            let total = bufs.iter().map(|buf| buf.len()).sum::<usize>();
+            for buf in bufs {
+                self.writes.extend_from_slice(buf);
+            }
+            Poll::Ready(Ok(total))
+        }
+
+        fn is_write_vectored(&self) -> bool {
+            true
+        }
+    }
+
+    #[derive(Default)]
+    struct CountingWriter {
+        flushes: Arc<AtomicUsize>,
+        shutdowns: Arc<AtomicUsize>,
+        writes: Vec<u8>,
+    }
+
+    impl AsyncWrite for CountingWriter {
+        fn poll_write(mut self: std::pin::Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::io::Result<usize>> {
+            self.writes.extend_from_slice(buf);
+            Poll::Ready(Ok(buf.len()))
+        }
+
+        fn poll_flush(self: std::pin::Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            self.flushes.fetch_add(1, Ordering::SeqCst);
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_shutdown(self: std::pin::Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            self.shutdowns.fetch_add(1, Ordering::SeqCst);
+            Poll::Ready(Ok(()))
+        }
+    }
 
     #[tokio::test]
     async fn test_bitrot_read_write_ok() {
@@ -471,4 +592,41 @@ mod tests {
         assert_eq!(n, data_size);
         assert_eq!(data, &out[..]);
     }
+
+    #[tokio::test]
+    async fn test_bitrot_writer_flushes_once_on_shutdown() {
+        let flushes = Arc::new(AtomicUsize::new(0));
+        let shutdowns = Arc::new(AtomicUsize::new(0));
+        let writer = CountingWriter {
+            flushes: flushes.clone(),
+            shutdowns: shutdowns.clone(),
+            writes: Vec::new(),
+        };
+        let mut bitrot_writer = BitrotWriter::new(writer, 8, HashAlgorithm::None);
+
+        bitrot_writer.write(b"12345678").await.unwrap();
+        bitrot_writer.write(b"abc").await.unwrap();
+
+        assert_eq!(flushes.load(Ordering::SeqCst), 0);
+        assert_eq!(shutdowns.load(Ordering::SeqCst), 0);
+
+        bitrot_writer.shutdown().await.unwrap();
+
+        assert_eq!(flushes.load(Ordering::SeqCst), 1);
+        assert_eq!(shutdowns.load(Ordering::SeqCst), 1);
+    }
+
+    #[tokio::test]
+    async fn test_bitrot_writer_uses_vectored_write_for_hash_and_data() {
+        let vectored_writes = Arc::new(AtomicUsize::new(0));
+        let writer = VectoredCountingWriter {
+            vectored_writes: vectored_writes.clone(),
+            writes: Vec::new(),
+        };
+        let mut bitrot_writer = BitrotWriter::new(writer, 8, HashAlgorithm::HighwayHash256);
+
+        bitrot_writer.write(b"payload").await.unwrap();
+
+        assert!(vectored_writes.load(Ordering::SeqCst) > 0);
+    }
 }

crates/ecstore/src/erasure_coding/encode.rs

@@ -112,11 +112,66 @@ impl<'a> MultiWriter<'a> {
         )))
     }
 
-    pub async fn _shutdown(&mut self) -> std::io::Result<()> {
-        for writer in self.writers.iter_mut().flatten() {
-            writer.shutdown().await?;
+    async fn shutdown_writer(writer_opt: &mut Option<BitrotWriterWrapper>, err: &mut Option<Error>) {
+        match writer_opt {
+            Some(writer) => match writer.shutdown().await {
+                Ok(()) => {
+                    *err = None;
+                }
+                Err(e) => {
+                    *err = Some(Error::from(e));
+                    *writer_opt = None;
+                }
+            },
+            None => {
+                *err = Some(Error::DiskNotFound);
+            }
         }
-        Ok(())
+    }
+
+    pub async fn shutdown(&mut self) -> std::io::Result<()> {
+        {
+            let mut futures = FuturesUnordered::new();
+            for (writer_opt, err) in self.writers.iter_mut().zip(self.errs.iter_mut()) {
+                if err.is_some() {
+                    continue;
+                }
+                futures.push(Self::shutdown_writer(writer_opt, err));
+            }
+            while let Some(()) = futures.next().await {}
+        }
+
+        let nil_count = self.errs.iter().filter(|&e| e.is_none()).count();
+        if nil_count >= self.write_quorum {
+            return Ok(());
+        }
+
+        if let Some(write_err) = reduce_write_quorum_errs(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum) {
+            error!(
+                "reduce_write_quorum_errs during shutdown: {:?}, offline-disks={}/{}, errs={:?}",
+                write_err,
+                count_errs(&self.errs, &Error::DiskNotFound),
+                self.writers.len(),
+                self.errs
+            );
+            return Err(std::io::Error::other(format!(
+                "Failed to shutdown writers: {} (offline-disks={}/{})",
+                write_err,
+                count_errs(&self.errs, &Error::DiskNotFound),
+                self.writers.len()
+            )));
+        }
+
+        Err(std::io::Error::other(format!(
+            "Failed to shutdown writers: (offline-disks={}/{}): {}",
+            count_errs(&self.errs, &Error::DiskNotFound),
+            self.writers.len(),
+            self.errs
+                .iter()
+                .map(|e| e.as_ref().map_or("<nil>".to_string(), |e| e.to_string()))
+                .collect::<Vec<_>>()
+                .join(", ")
+        )))
     }
 }
 
@@ -176,7 +231,69 @@ impl Erasure {
         }
 
         let (reader, total) = task.await??;
-        // writers.shutdown().await?;
+        writers.shutdown().await?;
         Ok((reader, total))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::erasure_coding::{BitrotWriterWrapper, CustomWriter};
+    use rustfs_utils::HashAlgorithm;
+    use std::pin::Pin;
+    use std::sync::{Arc, Mutex};
+    use std::task::{Context, Poll};
+    use tokio::io::AsyncWrite;
+
+    #[derive(Clone, Default)]
+    struct DeferredCommitWriter {
+        buffered: Vec<u8>,
+        committed: Arc<Mutex<Vec<u8>>>,
+    }
+
+    impl DeferredCommitWriter {
+        fn new(committed: Arc<Mutex<Vec<u8>>>) -> Self {
+            Self {
+                buffered: Vec::new(),
+                committed,
+            }
+        }
+    }
+
+    impl AsyncWrite for DeferredCommitWriter {
+        fn poll_write(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::io::Result<usize>> {
+            self.buffered.extend_from_slice(buf);
+            Poll::Ready(Ok(buf.len()))
+        }
+
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_shutdown(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            let buffered = std::mem::take(&mut self.buffered);
+            let mut committed = self.committed.lock().unwrap();
+            committed.extend_from_slice(&buffered);
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    #[tokio::test]
+    async fn encode_shutdowns_writers_after_small_shards() {
+        let committed = Arc::new(Mutex::new(Vec::new()));
+        let writer = DeferredCommitWriter::new(committed.clone());
+        let mut writers = vec![Some(BitrotWriterWrapper::new(
+            CustomWriter::new_tokio_writer(writer),
+            16,
+            HashAlgorithm::HighwayHash256S,
+        ))];
+
+        let erasure = Arc::new(Erasure::new(1, 0, 16));
+        let reader = tokio::io::BufReader::new(std::io::Cursor::new(b"small payload".to_vec()));
+        let (_reader, written) = erasure.encode(reader, &mut writers, 1).await.unwrap();
+
+        assert_eq!(written, b"small payload".len());
+        assert!(!committed.lock().unwrap().is_empty());
+    }
+}

crates/ecstore/src/global.rs

@@ -46,6 +46,7 @@ lazy_static! {
     pub static ref GLOBAL_IsDistErasure: RwLock<bool> = RwLock::new(false);
     pub static ref GLOBAL_IsErasureSD: RwLock<bool> = RwLock::new(false);
     pub static ref GLOBAL_LOCAL_DISK_MAP: Arc<RwLock<HashMap<String, Option<DiskStore>>>> = Arc::new(RwLock::new(HashMap::new()));
+    pub static ref GLOBAL_LOCAL_DISK_ID_MAP: Arc<RwLock<HashMap<Uuid, String>>> = Arc::new(RwLock::new(HashMap::new()));
     pub static ref GLOBAL_LOCAL_DISK_SET_DRIVES: Arc<RwLock<TypeLocalDiskSetDrives>> = Arc::new(RwLock::new(Vec::new()));
     pub static ref GLOBAL_Endpoints: OnceLock<EndpointServerPools> = OnceLock::new();
     pub static ref GLOBAL_RootDiskThreshold: RwLock<u64> = RwLock::new(0);
@@ -153,6 +154,10 @@ pub fn get_global_endpoints_opt() -> Option<EndpointServerPools> {
     GLOBAL_Endpoints.get().cloned()
 }
 
+pub async fn is_first_cluster_node_local() -> bool {
+    get_global_endpoints().first_local()
+}
+
 pub fn get_global_tier_config_mgr() -> Arc<RwLock<TierConfigMgr>> {
     GLOBAL_TierConfigMgr.clone()
 }

crates/ecstore/src/metrics_realtime.rs

@@ -14,8 +14,11 @@
 
 use crate::{admin_server_info::get_local_server_property, new_object_layer_fn, store_api::StorageAPI};
 use chrono::Utc;
-use rustfs_common::{GLOBAL_LOCAL_NODE_NAME, GLOBAL_RUSTFS_ADDR, heal_channel::DriveState, metrics::global_metrics};
-use rustfs_madmin::metrics::{DiskIOStats, DiskMetric, RealtimeMetrics};
+use rustfs_common::{
+    GLOBAL_LOCAL_NODE_NAME, GLOBAL_RUSTFS_ADDR, heal_channel::DriveState, internode_metrics::global_internode_metrics,
+    metrics::global_metrics,
+};
+use rustfs_madmin::metrics::{DiskIOStats, DiskMetric, NetDevLine, NetMetrics, RPCMetrics, RealtimeMetrics};
 use rustfs_utils::os::get_drive_stats;
 use serde::{Deserialize, Serialize};
 use std::collections::{HashMap, HashSet};
@@ -120,13 +123,50 @@ pub async fn collect_local_metrics(types: MetricType, opts: &CollectMetricsOpts)
 
     // if types.contains(&MetricType::SITE_RESYNC) {}
 
-    // if types.contains(&MetricType::NET) {}
+    if types.contains(&MetricType::NET) {
+        let snapshot = global_internode_metrics().snapshot();
+        real_time_metrics.aggregated.net = Some(NetMetrics {
+            collected_at: Utc::now(),
+            interface_name: "internode".to_string(),
+            net_stats: NetDevLine {
+                name: "internode".to_string(),
+                rx_bytes: snapshot.recv_bytes_total,
+                tx_bytes: snapshot.sent_bytes_total,
+                ..Default::default()
+            },
+        });
+    }
 
     // if types.contains(&MetricType::MEM) {}
 
     // if types.contains(&MetricType::CPU) {}
 
-    // if types.contains(&MetricType::RPC) {}
+    if types.contains(&MetricType::RPC) {
+        let collected_at = Utc::now();
+        let snapshot = global_internode_metrics().snapshot();
+        let last_connect_time =
+            chrono::DateTime::<Utc>::from_timestamp_millis(snapshot.last_dial_unix_millis as i64).unwrap_or(collected_at);
+
+        real_time_metrics.aggregated.rpc = Some(RPCMetrics {
+            collected_at,
+            connected: i32::from(snapshot.last_dial_unix_millis > 0),
+            reconnect_count: snapshot.dial_errors_total.min(i32::MAX as u64) as i32,
+            disconnected: 0,
+            outgoing_streams: 0,
+            incoming_streams: 0,
+            outgoing_bytes: snapshot.sent_bytes_total.min(i64::MAX as u64) as i64,
+            incoming_bytes: snapshot.recv_bytes_total.min(i64::MAX as u64) as i64,
+            outgoing_messages: snapshot.outgoing_requests_total.min(i64::MAX as u64) as i64,
+            incoming_messages: snapshot.incoming_requests_total.min(i64::MAX as u64) as i64,
+            out_queue: 0,
+            last_pong_time: collected_at,
+            last_ping_ms: snapshot.dial_avg_time_nanos as f64 / 1_000_000.0,
+            max_ping_dur_ms: snapshot.dial_avg_time_nanos as f64 / 1_000_000.0,
+            last_connect_time,
+            by_destination: None,
+            by_caller: None,
+        });
+    }
 
     real_time_metrics
         .by_host
@@ -211,7 +251,9 @@ async fn collect_local_disks_metrics(disks: &HashSet<String>) -> HashMap<String,
 
 #[cfg(test)]
 mod test {
-    use super::MetricType;
+    use super::*;
+    use rustfs_common::internode_metrics::global_internode_metrics;
+    use std::time::Duration;
 
     #[test]
     fn tes_types() {
@@ -229,4 +271,30 @@ mod test {
         let disk = MetricType::new(1 << 1);
         assert!(disk.contains(&MetricType::DISK));
     }
+
+    #[tokio::test]
+    async fn collect_local_metrics_reports_internode_net_and_rpc() {
+        let metrics = global_internode_metrics();
+        metrics.reset_for_test();
+        metrics.record_sent_bytes(128);
+        metrics.record_recv_bytes(64);
+        metrics.record_outgoing_request();
+        metrics.record_incoming_request();
+        metrics.record_dial_result(Duration::from_millis(4), true);
+
+        let realtime = collect_local_metrics(MetricType::NET, &CollectMetricsOpts::default()).await;
+        let net = realtime.aggregated.net.expect("net metrics");
+        assert_eq!(net.net_stats.tx_bytes, 128);
+        assert_eq!(net.net_stats.rx_bytes, 64);
+
+        let realtime = collect_local_metrics(MetricType::RPC, &CollectMetricsOpts::default()).await;
+        let rpc = realtime.aggregated.rpc.expect("rpc metrics");
+        assert_eq!(rpc.outgoing_bytes, 128);
+        assert_eq!(rpc.incoming_bytes, 64);
+        assert_eq!(rpc.outgoing_messages, 1);
+        assert_eq!(rpc.incoming_messages, 1);
+        assert!(rpc.last_ping_ms > 0.0);
+
+        metrics.reset_for_test();
+    }
 }

crates/ecstore/src/rpc/remote_disk.rs

@@ -23,6 +23,7 @@ use crate::disk::{
 };
 use crate::disk::{disk_store::DiskHealthTracker, error::DiskError, local::ScanGuard};
 use crate::rpc::client::{TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client};
+use crate::set_disk::DEFAULT_READ_BUFFER_SIZE;
 use crate::{
     disk::error::{Error, Result},
     rpc::build_auth_headers,
@@ -40,7 +41,9 @@ use rustfs_protos::proto_gen::node_service::{
     node_service_client::NodeServiceClient,
 };
 use rustfs_rio::{HttpReader, HttpWriter};
+use serde::{Serialize, de::DeserializeOwned};
 use std::{
+    io::Cursor,
     path::PathBuf,
     sync::{
         Arc,
@@ -49,12 +52,36 @@ use std::{
     time::Duration,
 };
 use tokio::time;
-use tokio::{io::AsyncWrite, net::TcpStream, time::timeout};
+use tokio::{
+    io::{self, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt},
+    net::TcpStream,
+    time::timeout,
+};
 use tokio_util::sync::CancellationToken;
 use tonic::{Request, service::interceptor::InterceptedService, transport::Channel};
 use tracing::{debug, info, warn};
 use uuid::Uuid;
 
+async fn copy_stream_with_buffer<R, W>(reader: &mut R, writer: &mut W, buffer_size: usize) -> io::Result<u64>
+where
+    R: AsyncRead + Unpin,
+    W: AsyncWrite + Unpin,
+{
+    let mut copied = 0_u64;
+    let mut buffer = vec![0_u8; buffer_size];
+
+    loop {
+        let bytes_read = reader.read(&mut buffer).await?;
+        if bytes_read == 0 {
+            writer.flush().await?;
+            return Ok(copied);
+        }
+
+        writer.write_all(&buffer[..bytes_read]).await?;
+        copied += bytes_read as u64;
+    }
+}
+
 #[derive(Debug)]
 pub struct RemoteDisk {
     pub id: Mutex<Option<Uuid>>,
@@ -259,6 +286,27 @@ impl RemoteDisk {
             .await
             .map_err(|err| Error::other(format!("can not get client, err: {err}")))
     }
+
+    async fn disk_ref(&self) -> String {
+        (*self.id.lock().await)
+            .map(|id| id.to_string())
+            .unwrap_or_else(|| self.endpoint.to_string())
+    }
+}
+
+fn encode_msgpack<T: Serialize>(value: &T) -> Result<Vec<u8>> {
+    let mut serializer = rmp_serde::Serializer::new(Vec::new());
+    value.serialize(&mut serializer)?;
+    Ok(serializer.into_inner())
+}
+
+fn decode_msgpack_or_json<T: DeserializeOwned>(binary: &[u8], json: &str) -> Result<T> {
+    if !binary.is_empty() {
+        let mut deserializer = rmp_serde::Deserializer::new(Cursor::new(binary));
+        return T::deserialize(&mut deserializer).map_err(Error::from);
+    }
+
+    serde_json::from_str(json).map_err(Error::from)
 }
 
 // TODO: all api need to handle errors
@@ -707,18 +755,21 @@ impl DiskAPI for RemoteDisk {
     async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> {
         info!("write_metadata {}/{}", volume, path);
         let file_info = serde_json::to_string(&fi)?;
+        let file_info_bin = encode_msgpack(&fi)?;
 
         self.execute_with_timeout(
             || async {
+                let disk = self.disk_ref().await;
                 let mut client = self
                     .get_client()
                     .await
                     .map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
                 let request = Request::new(WriteMetadataRequest {
-                    disk: self.endpoint.to_string(),
+                    disk,
                     volume: volume.to_string(),
                     path: path.to_string(),
                     file_info: file_info.clone(),
+                    file_info_bin: file_info_bin.clone(),
                 });
 
                 let response = client.write_metadata(request).await?.into_inner();
@@ -735,6 +786,7 @@ impl DiskAPI for RemoteDisk {
     }
 
     async fn read_metadata(&self, volume: &str, path: &str) -> Result<Bytes> {
+        let disk = self.disk_ref().await;
         let mut client = self
             .get_client()
             .await
@@ -742,7 +794,7 @@ impl DiskAPI for RemoteDisk {
         let request = Request::new(ReadMetadataRequest {
             volume: volume.to_string(),
             path: path.to_string(),
-            disk: self.endpoint.to_string(),
+            disk,
         });
 
         let response = client.read_metadata(request).await?.into_inner();
@@ -759,19 +811,24 @@ impl DiskAPI for RemoteDisk {
         info!("update_metadata");
         let file_info = serde_json::to_string(&fi)?;
         let opts_str = serde_json::to_string(&opts)?;
+        let file_info_bin = encode_msgpack(&fi)?;
+        let opts_bin = encode_msgpack(opts)?;
 
         self.execute_with_timeout(
             || async {
+                let disk = self.disk_ref().await;
                 let mut client = self
                     .get_client()
                     .await
                     .map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
                 let request = Request::new(UpdateMetadataRequest {
-                    disk: self.endpoint.to_string(),
+                    disk,
                     volume: volume.to_string(),
                     path: path.to_string(),
                     file_info: file_info.clone(),
                     opts: opts_str.clone(),
+                    file_info_bin: file_info_bin.clone(),
+                    opts_bin: opts_bin.clone(),
                 });
 
                 let response = client.update_metadata(request).await?.into_inner();
@@ -798,19 +855,22 @@ impl DiskAPI for RemoteDisk {
     ) -> Result<FileInfo> {
         info!("read_version");
         let opts_str = serde_json::to_string(opts)?;
+        let opts_bin = encode_msgpack(opts)?;
 
         self.execute_with_timeout(
             || async {
+                let disk = self.disk_ref().await;
                 let mut client = self
                     .get_client()
                     .await
                     .map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
                 let request = Request::new(ReadVersionRequest {
-                    disk: self.endpoint.to_string(),
+                    disk,
                     volume: volume.to_string(),
                     path: path.to_string(),
                     version_id: version_id.to_string(),
                     opts: opts_str.clone(),
+                    opts_bin: opts_bin.clone(),
                 });
 
                 let response = client.read_version(request).await?.into_inner();
@@ -819,7 +879,7 @@ impl DiskAPI for RemoteDisk {
                     return Err(response.error.unwrap_or_default().into());
                 }
 
-                let file_info = serde_json::from_str::<FileInfo>(&response.file_info)?;
+                let file_info = decode_msgpack_or_json::<FileInfo>(&response.file_info_bin, &response.file_info)?;
 
                 Ok(file_info)
             },
@@ -834,12 +894,13 @@ impl DiskAPI for RemoteDisk {
 
         self.execute_with_timeout(
             || async {
+                let disk = self.disk_ref().await;
                 let mut client = self
                     .get_client()
                     .await
                     .map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
                 let request = Request::new(ReadXlRequest {
-                    disk: self.endpoint.to_string(),
+                    disk,
                     volume: volume.to_string(),
                     path: path.to_string(),
                     read_data,
@@ -851,7 +912,7 @@ impl DiskAPI for RemoteDisk {
                     return Err(response.error.unwrap_or_default().into());
                 }
 
-                let raw_file_info = serde_json::from_str::<RawFileInfo>(&response.raw_file_info)?;
+                let raw_file_info = decode_msgpack_or_json::<RawFileInfo>(&response.raw_file_info_bin, &response.raw_file_info)?;
 
                 Ok(raw_file_info)
             },
@@ -909,13 +970,14 @@ impl DiskAPI for RemoteDisk {
         if self.health.is_faulty() {
             return Err(DiskError::FaultyDisk);
         }
+        let disk = self.disk_ref().await;
 
         let mut client = self
             .get_client()
             .await
             .map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
         let request = Request::new(ListDirRequest {
-            disk: self.endpoint.to_string(),
+            disk,
             volume: volume.to_string(),
             dir_path: dir_path.to_string(),
             count,
@@ -937,12 +999,9 @@ impl DiskAPI for RemoteDisk {
         if self.health.is_faulty() {
             return Err(DiskError::FaultyDisk);
         }
+        let disk = self.disk_ref().await;
 
-        let url = format!(
-            "{}/rustfs/rpc/walk_dir?disk={}",
-            self.endpoint.grid_host(),
-            urlencoding::encode(self.endpoint.to_string().as_str()),
-        );
+        let url = format!("{}/rustfs/rpc/walk_dir?disk={}", self.endpoint.grid_host(), urlencoding::encode(&disk),);
 
         let opts = serde_json::to_vec(&opts)?;
 
@@ -952,33 +1011,14 @@ impl DiskAPI for RemoteDisk {
 
         let mut reader = HttpReader::new(url, Method::GET, headers, Some(opts)).await?;
 
-        tokio::io::copy(&mut reader, wr).await?;
+        copy_stream_with_buffer(&mut reader, wr, DEFAULT_READ_BUFFER_SIZE).await?;
 
         Ok(())
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
     async fn read_file(&self, volume: &str, path: &str) -> Result<FileReader> {
-        info!("read_file {}/{}", volume, path);
-
-        if self.health.is_faulty() {
-            return Err(DiskError::FaultyDisk);
-        }
-
-        let url = format!(
-            "{}/rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}",
-            self.endpoint.grid_host(),
-            urlencoding::encode(self.endpoint.to_string().as_str()),
-            urlencoding::encode(volume),
-            urlencoding::encode(path),
-            0,
-            0
-        );
-
-        let mut headers = HeaderMap::new();
-        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
-        build_auth_headers(&url, &Method::GET, &mut headers);
-        Ok(Box::new(HttpReader::new(url, Method::GET, headers, None).await?))
+        self.read_file_stream(volume, path, 0, 0).await
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
@@ -995,11 +1035,12 @@ impl DiskAPI for RemoteDisk {
         if self.health.is_faulty() {
             return Err(DiskError::FaultyDisk);
         }
+        let disk = self.disk_ref().await;
 
         let url = format!(
             "{}/rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}",
             self.endpoint.grid_host(),
-            urlencoding::encode(self.endpoint.to_string().as_str()),
+            urlencoding::encode(&disk),
             urlencoding::encode(volume),
             urlencoding::encode(path),
             offset,
@@ -1019,11 +1060,12 @@ impl DiskAPI for RemoteDisk {
         if self.health.is_faulty() {
             return Err(DiskError::FaultyDisk);
         }
+        let disk = self.disk_ref().await;
 
         let url = format!(
             "{}/rustfs/rpc/put_file_stream?disk={}&volume={}&path={}&append={}&size={}",
             self.endpoint.grid_host(),
-            urlencoding::encode(self.endpoint.to_string().as_str()),
+            urlencoding::encode(&disk),
             urlencoding::encode(volume),
             urlencoding::encode(path),
             true,
@@ -1049,11 +1091,12 @@ impl DiskAPI for RemoteDisk {
         if self.health.is_faulty() {
             return Err(DiskError::FaultyDisk);
         }
+        let disk = self.disk_ref().await;
 
         let url = format!(
             "{}/rustfs/rpc/put_file_stream?disk={}&volume={}&path={}&append={}&size={}",
             self.endpoint.grid_host(),
-            urlencoding::encode(self.endpoint.to_string().as_str()),
+            urlencoding::encode(&disk),
             urlencoding::encode(volume),
             urlencoding::encode(path),
             false,
@@ -1261,13 +1304,16 @@ impl DiskAPI for RemoteDisk {
         self.execute_with_timeout(
             || async {
                 let read_multiple_req = serde_json::to_string(&req)?;
+                let read_multiple_req_bin = encode_msgpack(&req)?;
+                let disk = self.disk_ref().await;
                 let mut client = self
                     .get_client()
                     .await
                     .map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
                 let request = Request::new(ReadMultipleRequest {
-                    disk: self.endpoint.to_string(),
+                    disk,
                     read_multiple_req,
+                    read_multiple_req_bin,
                 });
 
                 let response = client.read_multiple(request).await?.into_inner();
@@ -1276,11 +1322,19 @@ impl DiskAPI for RemoteDisk {
                     return Err(response.error.unwrap_or_default().into());
                 }
 
-                let read_multiple_resps = response
-                    .read_multiple_resps
-                    .into_iter()
-                    .filter_map(|json_str| serde_json::from_str::<ReadMultipleResp>(&json_str).ok())
-                    .collect();
+                let read_multiple_resps = if !response.read_multiple_resps_bin.is_empty() {
+                    response
+                        .read_multiple_resps_bin
+                        .into_iter()
+                        .filter_map(|buf| decode_msgpack_or_json::<ReadMultipleResp>(&buf, "").ok())
+                        .collect()
+                } else {
+                    response
+                        .read_multiple_resps
+                        .into_iter()
+                        .filter_map(|json_str| serde_json::from_str::<ReadMultipleResp>(&json_str).ok())
+                        .collect()
+                };
 
                 Ok(read_multiple_resps)
             },
@@ -1295,12 +1349,13 @@ impl DiskAPI for RemoteDisk {
 
         self.execute_with_timeout(
             || async {
+                let disk = self.disk_ref().await;
                 let mut client = self
                     .get_client()
                     .await
                     .map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
                 let request = Request::new(WriteAllRequest {
-                    disk: self.endpoint.to_string(),
+                    disk,
                     volume: volume.to_string(),
                     path: path.to_string(),
                     data,
@@ -1325,12 +1380,13 @@ impl DiskAPI for RemoteDisk {
 
         self.execute_with_timeout(
             || async {
+                let disk = self.disk_ref().await;
                 let mut client = self
                     .get_client()
                     .await
                     .map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
                 let request = Request::new(ReadAllRequest {
-                    disk: self.endpoint.to_string(),
+                    disk,
                     volume: volume.to_string(),
                     path: path.to_string(),
                 });
@@ -1386,6 +1442,7 @@ impl DiskAPI for RemoteDisk {
 mod tests {
     use super::*;
     use std::sync::Once;
+    use tokio::io::duplex;
     use tokio::net::TcpListener;
     use tracing::Level;
     use uuid::Uuid;
@@ -1543,6 +1600,24 @@ mod tests {
         assert!(!remote_disk.is_online().await);
     }
 
+    #[tokio::test]
+    async fn test_copy_stream_with_buffer_copies_full_payload() {
+        let payload = b"walk-dir-stream".repeat(1024);
+        let expected = payload.clone();
+        let (mut write_half, mut read_half) = duplex(128);
+
+        let copy_task = tokio::spawn(async move {
+            let mut cursor = Cursor::new(payload);
+            copy_stream_with_buffer(&mut cursor, &mut write_half, 4 * 1024).await.unwrap();
+        });
+
+        let mut copied = Vec::new();
+        read_half.read_to_end(&mut copied).await.unwrap();
+        copy_task.await.unwrap();
+
+        assert_eq!(copied, expected);
+    }
+
     #[tokio::test]
     async fn test_remote_disk_disk_id() {
         let url = url::Url::parse("http://remote-server:9000").unwrap();
@@ -1579,6 +1654,30 @@ mod tests {
         assert!(cleared_id.is_none());
     }
 
+    #[tokio::test]
+    async fn test_remote_disk_ref_prefers_disk_id() {
+        let url = url::Url::parse("http://remote-server:9000").unwrap();
+        let endpoint = Endpoint {
+            url,
+            is_local: false,
+            pool_idx: 0,
+            set_idx: 0,
+            disk_idx: 0,
+        };
+        let disk_option = DiskOption {
+            cleanup: false,
+            health_check: false,
+        };
+
+        let remote_disk = RemoteDisk::new(&endpoint, &disk_option).await.unwrap();
+        assert_eq!(remote_disk.disk_ref().await, endpoint.to_string());
+
+        let disk_id = Uuid::new_v4();
+        remote_disk.set_disk_id(Some(disk_id)).await.unwrap();
+
+        assert_eq!(remote_disk.disk_ref().await, disk_id.to_string());
+    }
+
     #[tokio::test]
     async fn test_remote_disk_endpoints_with_different_schemes() {
         let test_cases = vec![

crates/ecstore/src/rpc/remote_locker.rs

@@ -52,6 +52,7 @@ impl RemoteClient {
             metadata: LockMetadata::default(),
             priority: LockPriority::Normal,
             deadlock_detection: false,
+            suppress_contention_logs: false,
         }
     }
 

crates/ecstore/src/store.rs

@@ -142,8 +142,8 @@ mod rebalance;
 
 use peer::init_local_peer;
 pub use peer::{
-    all_local_disk, all_local_disk_path, find_local_disk, get_disk_infos, get_disk_via_endpoint, has_space_for, init_local_disks,
-    init_lock_clients,
+    all_local_disk, all_local_disk_path, find_local_disk, find_local_disk_by_ref, get_disk_infos, get_disk_via_endpoint,
+    has_space_for, init_local_disks, init_lock_clients,
 };
 
 #[derive(Debug)]

crates/ecstore/src/store/init.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use super::*;
+use crate::global::is_first_cluster_node_local;
 
 impl ECStore {
     #[allow(clippy::new_ret_no_self)]
@@ -203,6 +204,7 @@ impl ECStore {
         let mut meta = PoolMeta::default();
         meta.load(self.pools[0].clone(), self.pools.clone()).await?;
         let update = meta.validate(self.pools.clone())?;
+        let should_persist_pool_meta = is_first_cluster_node_local().await;
 
         if !update {
             {
@@ -211,7 +213,9 @@ impl ECStore {
             }
         } else {
             let new_meta = PoolMeta::new(&self.pools, &meta);
-            new_meta.save(self.pools.clone()).await?;
+            if should_persist_pool_meta {
+                new_meta.save(self.pools.clone()).await?;
+            }
             {
                 let mut pool_meta = self.pool_meta.write().await;
                 *pool_meta = new_meta;

crates/ecstore/src/store/peer.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use super::*;
+use crate::global::GLOBAL_LOCAL_DISK_ID_MAP;
 
 pub async fn find_local_disk(disk_path: &String) -> Option<DiskStore> {
     let disk_map = GLOBAL_LOCAL_DISK_MAP.read().await;
@@ -24,6 +25,19 @@ pub async fn find_local_disk(disk_path: &String) -> Option<DiskStore> {
     }
 }
 
+pub async fn find_local_disk_by_ref(disk_ref: &str) -> Option<DiskStore> {
+    if let Some(disk) = find_local_disk(&disk_ref.to_string()).await {
+        return Some(disk);
+    }
+
+    let Ok(disk_id) = Uuid::parse_str(disk_ref) else {
+        return None;
+    };
+
+    let disk_path = GLOBAL_LOCAL_DISK_ID_MAP.read().await.get(&disk_id).cloned()?;
+    find_local_disk(&disk_path).await
+}
+
 pub async fn get_disk_via_endpoint(endpoint: &Endpoint) -> Option<DiskStore> {
     let global_set_drives = GLOBAL_LOCAL_DISK_SET_DRIVES.read().await;
     if global_set_drives.is_empty() {

crates/ecstore/src/tier/tier.rs

@@ -49,6 +49,7 @@ use crate::{
     StorageAPI,
     config::com::{CONFIG_PREFIX, read_config},
     disk::{MIGRATING_META_BUCKET, RUSTFS_META_BUCKET},
+    global::{get_global_endpoints, is_first_cluster_node_local},
     store::ECStore,
     store_api::{ObjectIO as _, ObjectOptions, PutObjReader},
 };
@@ -1118,7 +1119,15 @@ async fn load_tier_config(api: Arc<ECStore>) -> std::result::Result<TierConfigMg
                 }
                 Err(legacy_err) if is_err_config_not_found(&legacy_err) => {
                     warn!("config not found, start to init");
-                    new_and_save_tiering_config(api).await.map_err(io::Error::other)
+                    if is_first_cluster_node_local().await {
+                        new_and_save_tiering_config(api).await.map_err(io::Error::other)
+                    } else {
+                        Ok(TierConfigMgr {
+                            driver_cache: HashMap::new(),
+                            tiers: HashMap::new(),
+                            last_refreshed_at: OffsetDateTime::now_utc(),
+                        })
+                    }
                 }
                 Err(legacy_err) => Err(io::Error::other(legacy_err)),
             }
@@ -1166,7 +1175,14 @@ async fn write_tier_config_to_rustfs<S: StorageAPI>(api: Arc<S>, path: &str, dat
 pub async fn try_migrate_tiering_config<S: StorageAPI>(api: Arc<S>) {
     let target_path = tier_config_path(TIER_CONFIG_FILE);
     if api
-        .get_object_info(RUSTFS_META_BUCKET, &target_path, &ObjectOptions::default())
+        .get_object_info(
+            RUSTFS_META_BUCKET,
+            &target_path,
+            &ObjectOptions {
+                no_lock: true,
+                ..Default::default()
+            },
+        )
         .await
         .is_ok()
     {

crates/iam/src/manager.rs

@@ -25,6 +25,7 @@ use crate::{
 };
 use futures::future::join_all;
 use rustfs_credentials::{Credentials, EMBEDDED_POLICY_TYPE, INHERITED_POLICY_TYPE, get_global_action_cred};
+use rustfs_ecstore::global::is_first_cluster_node_local;
 use rustfs_madmin::{AccountStatus, AddOrUpdateUserReq, GroupDesc};
 use rustfs_policy::{
     arn::ARN,
@@ -303,6 +304,10 @@ where
             return Ok(());
         }
 
+        if !is_first_cluster_node_local().await {
+            return Ok(());
+        }
+
         self.api.save_iam_config(IAMFormat::new_version_1(), path).await
     }
     pub async fn get_user(&self, access_key: &str) -> Option<UserIdentity> {

crates/iam/src/store/object.rs

@@ -25,7 +25,7 @@ use rustfs_ecstore::store_api::{ListOperations as _, ObjectInfoOrErr, WalkOption
 use rustfs_ecstore::{
     config::{
         RUSTFS_CONFIG_PREFIX,
-        com::{delete_config, read_config, read_config_with_metadata, save_config},
+        com::{delete_config, read_config, read_config_no_lock, read_config_with_metadata, save_config},
     },
     store::ECStore,
     store_api::{ObjectInfo, ObjectOptions},
@@ -391,7 +391,7 @@ impl ObjectStore {
         // If it doesn't exist, the system bucket or metadata is not ready.
         let probe_path = format!("{}/format.json", *IAM_CONFIG_PREFIX);
 
-        match read_config(self.object_api.clone(), &probe_path).await {
+        match read_config_no_lock(self.object_api.clone(), &probe_path).await {
             Ok(_) => Ok(()),
             Err(rustfs_ecstore::error::StorageError::ConfigNotFound) => Err(Error::other(format!(
                 "Storage metadata not ready: probe object '{}' not found (expected IAM config to be initialized)",

crates/lock/src/distributed_lock.rs

@@ -230,7 +230,21 @@ impl DistributedLock {
         } else {
             // Check if it's a timeout or quorum failure
             if let Some(error_msg) = &resp.error {
-                warn!("acquire_lock_quorum error: {}", error_msg);
+                if request.suppress_contention_logs {
+                    tracing::debug!(
+                        resource = %request.resource,
+                        owner = %request.owner,
+                        "acquire_lock_quorum contention: {}",
+                        error_msg
+                    );
+                } else {
+                    warn!(
+                        resource = %request.resource,
+                        owner = %request.owner,
+                        "acquire_lock_quorum error: {}",
+                        error_msg
+                    );
+                }
                 if error_msg.contains("quorum") {
                     // This is a quorum failure - return appropriate error
                     // Extract achieved count from error message or use individual_locks.len()
@@ -266,6 +280,21 @@ impl DistributedLock {
         self.acquire_guard(&req).await
     }
 
+    /// Convenience: acquire exclusive lock with expected contention logs suppressed
+    pub async fn lock_guard_quiet(
+        &self,
+        resource: ObjectKey,
+        owner: &str,
+        timeout: Duration,
+        ttl: Duration,
+    ) -> Result<Option<DistributedLockGuard>> {
+        let req = LockRequest::new(resource, LockType::Exclusive, owner)
+            .with_acquire_timeout(timeout)
+            .with_ttl(ttl)
+            .with_suppress_contention_logs(true);
+        self.acquire_guard(&req).await
+    }
+
     /// Convenience: acquire shared lock as a guard
     pub async fn rlock_guard(
         &self,
@@ -307,11 +336,24 @@ impl DistributedLock {
                             individual_locks.push((lock_info.id.clone(), self.clients[idx].clone()));
                         }
                     } else {
-                        tracing::warn!(
-                            "Failed to acquire lock on client from response: {}, error: {}",
-                            idx,
-                            resp.error.unwrap_or_else(|| "unknown error".to_string())
-                        );
+                        let error = resp.error.unwrap_or_else(|| "unknown error".to_string());
+                        if request.suppress_contention_logs {
+                            tracing::debug!(
+                                resource = %request.resource,
+                                owner = %request.owner,
+                                "Failed to acquire lock on client from response: {}, error: {}",
+                                idx,
+                                error
+                            );
+                        } else {
+                            tracing::warn!(
+                                resource = %request.resource,
+                                owner = %request.owner,
+                                "Failed to acquire lock on client from response: {}, error: {}",
+                                idx,
+                                error
+                            );
+                        }
                     }
                 }
                 Err(e) => {

crates/lock/src/namespace/mod.rs

@@ -58,6 +58,16 @@ impl NamespaceLockWrapper {
         self.lock.get_write_lock(self.resource.clone(), &self.owner, timeout).await
     }
 
+    /// Acquire write lock with expected contention logs suppressed
+    pub async fn get_write_lock_quiet(
+        &self,
+        timeout: Duration,
+    ) -> std::result::Result<NamespaceLockGuard, crate::error::LockError> {
+        self.lock
+            .get_write_lock_quiet(self.resource.clone(), &self.owner, timeout)
+            .await
+    }
+
     /// Acquire read lock (shared lock) with timeout
     /// Returns the guard if acquisition succeeds, or an error if it fails
     pub async fn get_read_lock(&self, timeout: Duration) -> std::result::Result<NamespaceLockGuard, crate::error::LockError> {
@@ -237,6 +247,29 @@ impl NamespaceLock {
         }
     }
 
+    /// Acquire write lock while suppressing expected contention warnings
+    pub async fn get_write_lock_quiet(
+        &self,
+        resource: ObjectKey,
+        owner: &str,
+        timeout: Duration,
+    ) -> std::result::Result<NamespaceLockGuard, crate::error::LockError> {
+        let ttl = crate::fast_lock::DEFAULT_LOCK_TIMEOUT;
+        let resource_str = format!("{}", resource);
+        match self {
+            Self::Distributed(lock) => match lock.lock_guard_quiet(resource, owner, timeout, ttl).await {
+                Ok(Some(guard)) => Ok(NamespaceLockGuard::Standard(guard)),
+                Ok(None) => Err(crate::error::LockError::timeout(resource_str, timeout)),
+                Err(e) => Err(e),
+            },
+            Self::Local(lock) => match lock.lock_guard(resource, owner, timeout, ttl).await {
+                Ok(Some(guard)) => Ok(NamespaceLockGuard::Fast(guard)),
+                Ok(None) => Err(crate::error::LockError::timeout(resource_str, timeout)),
+                Err(e) => Err(e),
+            },
+        }
+    }
+
     /// Acquire read lock (shared lock) with timeout
     /// Returns the guard if acquisition succeeds, or an error if it fails
     pub async fn get_read_lock(

crates/lock/src/types.rs

@@ -225,6 +225,8 @@ pub struct LockRequest {
     pub priority: LockPriority,
     /// Deadlock detection
     pub deadlock_detection: bool,
+    /// Suppress warning logs for expected contention failures
+    pub suppress_contention_logs: bool,
 }
 
 impl LockRequest {
@@ -240,6 +242,7 @@ impl LockRequest {
             metadata: LockMetadata::default(),
             priority: LockPriority::default(),
             deadlock_detection: false,
+            suppress_contention_logs: false,
         }
     }
 
@@ -272,6 +275,12 @@ impl LockRequest {
         self.deadlock_detection = enabled;
         self
     }
+
+    /// Suppress warning logs for expected contention failures
+    pub fn with_suppress_contention_logs(mut self, enabled: bool) -> Self {
+        self.suppress_contention_logs = enabled;
+        self
+    }
 }
 
 /// Lock response structure

crates/notify/src/registry.rs

@@ -293,6 +293,12 @@ impl TargetRegistry {
                 }
             }
 
+            if &new_config == config {
+                info!("Notification target configuration unchanged, skip persisting server config");
+                info!(count = successful_targets.len(), "All target processing completed");
+                return Ok(successful_targets);
+            }
+
             let store = match rustfs_ecstore::global::new_object_layer_fn() {
                 Some(s) => s,
                 None => {

crates/protos/src/generated/proto_gen/node_service.rs

@@ -467,6 +467,10 @@ pub struct UpdateMetadataRequest {
     pub file_info: ::prost::alloc::string::String,
     #[prost(string, tag = "5")]
     pub opts: ::prost::alloc::string::String,
+    #[prost(bytes = "vec", tag = "6")]
+    pub file_info_bin: ::prost::alloc::vec::Vec<u8>,
+    #[prost(bytes = "vec", tag = "7")]
+    pub opts_bin: ::prost::alloc::vec::Vec<u8>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct UpdateMetadataResponse {
@@ -486,6 +490,8 @@ pub struct WriteMetadataRequest {
     pub path: ::prost::alloc::string::String,
     #[prost(string, tag = "4")]
     pub file_info: ::prost::alloc::string::String,
+    #[prost(bytes = "vec", tag = "5")]
+    pub file_info_bin: ::prost::alloc::vec::Vec<u8>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct WriteMetadataResponse {
@@ -506,6 +512,8 @@ pub struct ReadVersionRequest {
     pub version_id: ::prost::alloc::string::String,
     #[prost(string, tag = "5")]
     pub opts: ::prost::alloc::string::String,
+    #[prost(bytes = "vec", tag = "6")]
+    pub opts_bin: ::prost::alloc::vec::Vec<u8>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ReadVersionResponse {
@@ -515,6 +523,8 @@ pub struct ReadVersionResponse {
     pub file_info: ::prost::alloc::string::String,
     #[prost(message, optional, tag = "3")]
     pub error: ::core::option::Option<Error>,
+    #[prost(bytes = "vec", tag = "4")]
+    pub file_info_bin: ::prost::alloc::vec::Vec<u8>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ReadXlRequest {
@@ -535,6 +545,8 @@ pub struct ReadXlResponse {
     pub raw_file_info: ::prost::alloc::string::String,
     #[prost(message, optional, tag = "3")]
     pub error: ::core::option::Option<Error>,
+    #[prost(bytes = "vec", tag = "4")]
+    pub raw_file_info_bin: ::prost::alloc::vec::Vec<u8>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct DeleteVersionRequest {
@@ -586,6 +598,8 @@ pub struct ReadMultipleRequest {
     pub disk: ::prost::alloc::string::String,
     #[prost(string, tag = "2")]
     pub read_multiple_req: ::prost::alloc::string::String,
+    #[prost(bytes = "vec", tag = "3")]
+    pub read_multiple_req_bin: ::prost::alloc::vec::Vec<u8>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ReadMultipleResponse {
@@ -595,6 +609,8 @@ pub struct ReadMultipleResponse {
     pub read_multiple_resps: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
     #[prost(message, optional, tag = "3")]
     pub error: ::core::option::Option<Error>,
+    #[prost(bytes = "vec", repeated, tag = "4")]
+    pub read_multiple_resps_bin: ::prost::alloc::vec::Vec<::prost::alloc::vec::Vec<u8>>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct DeleteVolumeRequest {

crates/protos/src/lib.rs

@@ -16,8 +16,13 @@
 mod generated;
 
 use proto_gen::node_service::node_service_client::NodeServiceClient;
-use rustfs_common::{GLOBAL_CONN_MAP, GLOBAL_MTLS_IDENTITY, GLOBAL_ROOT_CERT, evict_connection};
-use std::{error::Error, time::Duration};
+use rustfs_common::{
+    GLOBAL_CONN_MAP, GLOBAL_MTLS_IDENTITY, GLOBAL_ROOT_CERT, evict_connection, internode_metrics::global_internode_metrics,
+};
+use std::{
+    error::Error,
+    time::{Duration, Instant},
+};
 use tonic::{
     Request, Status,
     service::interceptor::InterceptedService,
@@ -65,6 +70,7 @@ const RUSTFS_HTTPS_PREFIX: &str = "https://";
 /// - Overall RPC timeout of 30s (reduced from 60s)
 pub async fn create_new_channel(addr: &str) -> Result<Channel, Box<dyn Error>> {
     debug!("Creating new gRPC channel to: {}", addr);
+    let dial_started_at = Instant::now();
 
     let mut connector = Endpoint::from_shared(addr.to_string())?
         // Fast connection timeout for dead peer detection
@@ -119,7 +125,16 @@ pub async fn create_new_channel(addr: &str) -> Result<Channel, Box<dyn Error>> {
         }
     }
 
-    let channel = connector.connect().await?;
+    let channel = match connector.connect().await {
+        Ok(channel) => {
+            global_internode_metrics().record_dial_result(dial_started_at.elapsed(), true);
+            channel
+        }
+        Err(err) => {
+            global_internode_metrics().record_dial_result(dial_started_at.elapsed(), false);
+            return Err(err.into());
+        }
+    };
 
     // Cache the new connection
     {

crates/protos/src/node.proto

@@ -331,6 +331,8 @@ message UpdateMetadataRequest {
   string path = 3;
   string file_info = 4;
   string opts = 5;
+  bytes file_info_bin = 6;
+  bytes opts_bin = 7;
 }
 
 message UpdateMetadataResponse {
@@ -343,6 +345,7 @@ message WriteMetadataRequest {
   string volume = 2;
   string path = 3;
   string file_info = 4;
+  bytes file_info_bin = 5;
 }
 
 message WriteMetadataResponse {
@@ -356,12 +359,14 @@ message ReadVersionRequest {
   string path = 3;
   string version_id = 4;
   string opts = 5;
+  bytes opts_bin = 6;
 }
 
 message ReadVersionResponse {
   bool success = 1;
   string file_info = 2;
   optional Error error = 3;
+  bytes file_info_bin = 4;
 }
 
 message ReadXLRequest {
@@ -375,6 +380,7 @@ message ReadXLResponse {
   bool success = 1;
   string raw_file_info = 2;
   optional Error error = 3;
+  bytes raw_file_info_bin = 4;
 }
 
 message DeleteVersionRequest {
@@ -408,12 +414,14 @@ message DeleteVersionsResponse {
 message ReadMultipleRequest {
   string disk = 1;
   string read_multiple_req = 2;
+  bytes read_multiple_req_bin = 3;
 }
 
 message ReadMultipleResponse {
   bool success = 1;
   repeated string read_multiple_resps = 2;
   optional Error error = 3;
+  repeated bytes read_multiple_resps_bin = 4;
 }
 
 message DeleteVolumeRequest {

crates/rio/Cargo.toml

@@ -42,6 +42,7 @@ tokio-util.workspace = true
 faster-hex.workspace = true
 futures.workspace = true
 rustfs-config = { workspace = true, features = ["constants"] }
+rustfs-common.workspace = true
 rustfs-utils = { workspace = true, features = ["io", "hash", "compress", "tls"] }
 serde_json.workspace = true
 md-5 = { workspace = true }
@@ -55,3 +56,5 @@ hex-simd.workspace = true
 
 [dev-dependencies]
 tokio-test = { workspace = true }
+axum = { workspace = true }
+http-body-util = { workspace = true }

crates/rio/src/http_reader.rs

@@ -13,13 +13,14 @@
 // limitations under the License.
 
 use crate::{EtagResolvable, HashReaderDetector, HashReaderMut};
-use bytes::Bytes;
+use bytes::{Bytes, BytesMut};
 use futures::{Stream, TryStreamExt as _};
 use http::HeaderMap;
 use pin_project_lite::pin_project;
 use reqwest::{Certificate, Client, Identity, Method, RequestBuilder};
+use rustfs_common::internode_metrics::global_internode_metrics;
 use rustfs_utils::get_env_opt_str;
-use std::error::Error as _;
+use std::io::IoSlice;
 use std::io::{self, Error};
 use std::ops::Not as _;
 use std::pin::Pin;
@@ -28,6 +29,7 @@ use std::task::{Context, Poll};
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio::sync::mpsc;
 use tokio_util::io::StreamReader;
+use tokio_util::sync::PollSender;
 use tracing::error;
 
 /// Get the TLS path from the RUSTFS_TLS_PATH environment variable.
@@ -111,24 +113,12 @@ fn get_http_client() -> Client {
     CLIENT.clone()
 }
 
-static HTTP_DEBUG_LOG: bool = false;
-#[inline(always)]
-fn http_debug_log(args: std::fmt::Arguments) {
-    if HTTP_DEBUG_LOG {
-        println!("{args}");
-    }
-}
-macro_rules! http_log {
-    ($($arg:tt)*) => {
-        http_debug_log(format_args!($($arg)*));
-    };
-}
-
 pin_project! {
     pub struct HttpReader {
         url:String,
         method: Method,
         headers: HeaderMap,
+        track_internode_metrics: bool,
         #[pin]
         inner: StreamReader<Pin<Box<dyn Stream<Item=std::io::Result<Bytes>>+Send+Sync>>, Bytes>,
     }
@@ -147,53 +137,47 @@ impl HttpReader {
         body: Option<Vec<u8>>,
         _read_buf_size: usize,
     ) -> io::Result<Self> {
-        // http_log!(
-        //     "[HttpReader::with_capacity] url: {url}, method: {method:?}, headers: {headers:?}, buf_size: {}",
-        //     _read_buf_size
-        // );
-        // First, check if the connection is available (HEAD)
-        let client = get_http_client();
-        let head_resp = client.head(&url).headers(headers.clone()).send().await;
-        match head_resp {
-            Ok(resp) => {
-                http_log!("[HttpReader::new] HEAD status: {}", resp.status());
-                if !resp.status().is_success() {
-                    return Err(Error::other(format!("HEAD failed: url: {}, status {}", url, resp.status())));
-                }
-            }
-            Err(e) => {
-                http_log!("[HttpReader::new] HEAD error: {e}");
-                return Err(Error::other(e.source().map(|s| s.to_string()).unwrap_or_else(|| e.to_string())));
-            }
-        }
-
+        let track_internode_metrics = is_internode_rpc_url(&url);
         let client = get_http_client();
         let mut request: RequestBuilder = client.request(method.clone(), url.clone()).headers(headers.clone());
         if let Some(body) = body {
             request = request.body(body);
         }
 
-        let resp = request
-            .send()
-            .await
-            .map_err(|e| Error::other(format!("HttpReader HTTP request error: {e}")))?;
+        let resp = request.send().await.map_err(|e| {
+            if track_internode_metrics {
+                global_internode_metrics().record_error();
+            }
+            Error::other(format!("HttpReader HTTP request error: {e}"))
+        })?;
 
         if resp.status().is_success().not() {
+            if track_internode_metrics {
+                global_internode_metrics().record_error();
+            }
             return Err(Error::other(format!(
                 "HttpReader HTTP request failed with non-200 status {}",
                 resp.status()
             )));
         }
 
-        let stream = resp
-            .bytes_stream()
-            .map_err(|e| Error::other(format!("HttpReader stream error: {e}")));
+        if track_internode_metrics {
+            global_internode_metrics().record_outgoing_request();
+        }
+
+        let stream = resp.bytes_stream().map_err(move |e| {
+            if track_internode_metrics {
+                global_internode_metrics().record_error();
+            }
+            Error::other(format!("HttpReader stream error: {e}"))
+        });
 
         Ok(Self {
             inner: StreamReader::new(Box::pin(stream)),
             url,
             method,
             headers,
+            track_internode_metrics,
         })
     }
     pub fn url(&self) -> &str {
@@ -209,14 +193,17 @@ impl HttpReader {
 
 impl AsyncRead for HttpReader {
     fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
-        // http_log!(
-        //     "[HttpReader::poll_read] url: {}, method: {:?}, buf.remaining: {}",
-        //     self.url,
-        //     self.method,
-        //     buf.remaining()
-        // );
-        // Read from the inner stream
-        Pin::new(&mut self.inner).poll_read(cx, buf)
+        let filled_before = buf.filled().len();
+        match Pin::new(&mut self.inner).poll_read(cx, buf) {
+            Poll::Ready(Ok(())) => {
+                let bytes_read = buf.filled().len().saturating_sub(filled_before);
+                if self.track_internode_metrics && bytes_read > 0 {
+                    global_internode_metrics().record_recv_bytes(bytes_read);
+                }
+                Poll::Ready(Ok(()))
+            }
+            other => other,
+        }
     }
 }
 
@@ -241,6 +228,7 @@ impl HashReaderDetector for HttpReader {
 
 struct ReceiverStream {
     receiver: mpsc::Receiver<Option<Bytes>>,
+    track_internode_metrics: bool,
 }
 
 impl Stream for ReceiverStream {
@@ -262,7 +250,12 @@ impl Stream for ReceiverStream {
         //     }
         // }
         match poll {
-            Poll::Ready(Some(Some(bytes))) => Poll::Ready(Some(Ok(bytes))),
+            Poll::Ready(Some(Some(bytes))) => {
+                if self.track_internode_metrics {
+                    global_internode_metrics().record_sent_bytes(bytes.len());
+                }
+                Poll::Ready(Some(Ok(bytes)))
+            }
             Poll::Ready(Some(None)) => Poll::Ready(None), // Sender shutdown
             Poll::Ready(None) => Poll::Ready(None),
             Poll::Pending => Poll::Pending,
@@ -276,13 +269,17 @@ pin_project! {
         method: Method,
         headers: HeaderMap,
         err_rx: tokio::sync::oneshot::Receiver<std::io::Error>,
-        sender: tokio::sync::mpsc::Sender<Option<Bytes>>,
+        sender: PollSender<Option<Bytes>>,
         handle: tokio::task::JoinHandle<std::io::Result<()>>,
+        pending_chunk: BytesMut,
         finish:bool,
 
     }
 }
 
+const HTTP_WRITER_CHANNEL_CAPACITY: usize = 8;
+const HTTP_WRITER_BUFFER_SIZE: usize = 1024 * 1024;
+
 impl HttpWriter {
     /// Create a new HttpWriter for the given URL. The HTTP request is performed in the background.
     pub async fn new(url: String, method: Method, headers: HeaderMap) -> io::Result<Self> {
@@ -290,28 +287,16 @@ impl HttpWriter {
         let url_clone = url.clone();
         let method_clone = method.clone();
         let headers_clone = headers.clone();
+        let track_internode_metrics = is_internode_rpc_url(&url);
 
-        // First, try to write empty data to check if writable
-        let client = get_http_client();
-        let resp = client.put(&url).headers(headers.clone()).body(Vec::new()).send().await;
-        match resp {
-            Ok(resp) => {
-                // http_log!("[HttpWriter::new] empty PUT status: {}", resp.status());
-                if !resp.status().is_success() {
-                    return Err(Error::other(format!("Empty PUT failed: status {}", resp.status())));
-                }
-            }
-            Err(e) => {
-                // http_log!("[HttpWriter::new] empty PUT error: {e}");
-                return Err(Error::other(format!("Empty PUT failed: {e}")));
-            }
-        }
-
-        let (sender, receiver) = tokio::sync::mpsc::channel::<Option<Bytes>>(8);
+        let (sender, receiver) = tokio::sync::mpsc::channel::<Option<Bytes>>(HTTP_WRITER_CHANNEL_CAPACITY);
         let (err_tx, err_rx) = tokio::sync::oneshot::channel::<io::Error>();
 
         let handle = tokio::spawn(async move {
-            let stream = ReceiverStream { receiver };
+            let stream = ReceiverStream {
+                receiver,
+                track_internode_metrics,
+            };
             let body = reqwest::Body::wrap_stream(stream);
             // http_log!(
             //     "[HttpWriter::spawn] sending HTTP request: url={url_clone}, method={method_clone:?}, headers={headers_clone:?}"
@@ -330,6 +315,9 @@ impl HttpWriter {
                 Ok(resp) => {
                     // http_log!("[HttpWriter::spawn] got response: status={}", resp.status());
                     if !resp.status().is_success() {
+                        if track_internode_metrics {
+                            global_internode_metrics().record_error();
+                        }
                         let _ = err_tx.send(Error::other(format!(
                             "HttpWriter HTTP request failed with non-200 status {}",
                             resp.status()
@@ -338,6 +326,9 @@ impl HttpWriter {
                     }
                 }
                 Err(e) => {
+                    if track_internode_metrics {
+                        global_internode_metrics().record_error();
+                    }
                     // http_log!("[HttpWriter::spawn] HTTP request error: {e}");
                     let _ = err_tx.send(Error::other(format!("HTTP request failed: {e}")));
                     return Err(Error::other(format!("HTTP request failed: {e}")));
@@ -349,13 +340,17 @@ impl HttpWriter {
         });
 
         // http_log!("[HttpWriter::new] connection established successfully");
+        if track_internode_metrics {
+            global_internode_metrics().record_outgoing_request();
+        }
         Ok(Self {
             url,
             method,
             headers,
             err_rx,
-            sender,
+            sender: PollSender::new(sender),
             handle,
+            pending_chunk: BytesMut::with_capacity(HTTP_WRITER_BUFFER_SIZE),
             finish: false,
         })
     }
@@ -373,8 +368,40 @@ impl HttpWriter {
     }
 }
 
+fn is_internode_rpc_url(url: &str) -> bool {
+    url.contains("/rustfs/rpc/")
+}
+
+fn poll_send_error_to_io<T>(err: tokio_util::sync::PollSendError<T>, context: &str) -> io::Error {
+    Error::other(format!("{context}: {err}"))
+}
+
+fn send_error_to_io<T>(err: tokio_util::sync::PollSendError<T>, context: &str) -> io::Error {
+    Error::other(format!("{context}: {err}"))
+}
+
+impl HttpWriter {
+    fn poll_send_pending_chunk(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        if self.pending_chunk.is_empty() {
+            return Poll::Ready(Ok(()));
+        }
+
+        match self.sender.poll_reserve(cx) {
+            Poll::Ready(Ok(())) => {
+                let chunk = self.pending_chunk.split().freeze();
+                self.sender
+                    .send_item(Some(chunk))
+                    .map_err(|e| send_error_to_io(e, "HttpWriter send error"))?;
+                Poll::Ready(Ok(()))
+            }
+            Poll::Ready(Err(e)) => Poll::Ready(Err(poll_send_error_to_io(e, "HttpWriter send error"))),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+}
+
 impl AsyncWrite for HttpWriter {
-    fn poll_write(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
+    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
         // http_log!(
         //     "[HttpWriter::poll_write] url: {}, method: {:?}, buf.len: {}",
         //     self.url,
@@ -385,26 +412,104 @@ impl AsyncWrite for HttpWriter {
             return Poll::Ready(Err(e));
         }
 
-        self.sender
-            .try_send(Some(Bytes::copy_from_slice(buf)))
-            .map_err(|e| Error::other(format!("HttpWriter send error: {e}")))?;
+        let this = self.as_mut().get_mut();
+
+        if this.pending_chunk.len() >= HTTP_WRITER_BUFFER_SIZE {
+            match this.poll_send_pending_chunk(cx) {
+                Poll::Ready(Ok(())) => {}
+                Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
+                Poll::Pending => return Poll::Pending,
+            }
+        }
+
+        if buf.len() >= HTTP_WRITER_BUFFER_SIZE && this.pending_chunk.is_empty() {
+            match this.sender.poll_reserve(cx) {
+                Poll::Ready(Ok(())) => {
+                    this.sender
+                        .send_item(Some(Bytes::copy_from_slice(buf)))
+                        .map_err(|e| send_error_to_io(e, "HttpWriter send error"))?;
+                    return Poll::Ready(Ok(buf.len()));
+                }
+                Poll::Ready(Err(err)) => return Poll::Ready(Err(poll_send_error_to_io(err, "HttpWriter send error"))),
+                Poll::Pending => return Poll::Pending,
+            }
+        }
+
+        this.pending_chunk.extend_from_slice(buf);
 
         Poll::Ready(Ok(buf.len()))
     }
 
-    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
-        Poll::Ready(Ok(()))
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        self.as_mut().get_mut().poll_send_pending_chunk(cx)
     }
 
-    fn poll_shutdown(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+    fn poll_write_vectored(mut self: Pin<&mut Self>, cx: &mut Context<'_>, bufs: &[IoSlice<'_>]) -> Poll<io::Result<usize>> {
+        if let Ok(e) = Pin::new(&mut self.err_rx).try_recv() {
+            return Poll::Ready(Err(e));
+        }
+
+        let this = self.as_mut().get_mut();
+
+        if this.pending_chunk.len() >= HTTP_WRITER_BUFFER_SIZE {
+            match this.poll_send_pending_chunk(cx) {
+                Poll::Ready(Ok(())) => {}
+                Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
+                Poll::Pending => return Poll::Pending,
+            }
+        }
+
+        let total_len = bufs.iter().map(|buf| buf.len()).sum::<usize>();
+        if total_len == 0 {
+            return Poll::Ready(Ok(0));
+        }
+
+        if bufs.len() == 1 && this.pending_chunk.is_empty() && total_len >= HTTP_WRITER_BUFFER_SIZE {
+            match this.sender.poll_reserve(cx) {
+                Poll::Ready(Ok(())) => {
+                    this.sender
+                        .send_item(Some(Bytes::copy_from_slice(bufs[0].as_ref())))
+                        .map_err(|e| send_error_to_io(e, "HttpWriter send error"))?;
+                    return Poll::Ready(Ok(total_len));
+                }
+                Poll::Ready(Err(err)) => return Poll::Ready(Err(poll_send_error_to_io(err, "HttpWriter send error"))),
+                Poll::Pending => return Poll::Pending,
+            }
+        }
+
+        for buf in bufs {
+            this.pending_chunk.extend_from_slice(buf);
+        }
+
+        Poll::Ready(Ok(total_len))
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        true
+    }
+
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
         // let url = self.url.clone();
         // let method = self.method.clone();
 
+        match self.as_mut().get_mut().poll_send_pending_chunk(cx) {
+            Poll::Ready(Ok(())) => {}
+            Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
+            Poll::Pending => return Poll::Pending,
+        }
+
         if !self.finish {
             // http_log!("[HttpWriter::poll_shutdown] url: {}, method: {:?}", url, method);
-            self.sender
-                .try_send(None)
-                .map_err(|e| Error::other(format!("HttpWriter shutdown error: {e}")))?;
+            let this = self.as_mut().get_mut();
+            match this.sender.poll_reserve(cx) {
+                Poll::Ready(Ok(())) => {
+                    this.sender
+                        .send_item(None)
+                        .map_err(|e| send_error_to_io(e, "HttpWriter shutdown error"))?;
+                }
+                Poll::Ready(Err(err)) => return Poll::Ready(Err(poll_send_error_to_io(err, "HttpWriter shutdown error"))),
+                Poll::Pending => return Poll::Pending,
+            }
             // http_log!(
             //     "[HttpWriter::poll_shutdown] sent shutdown signal to HTTP request, url: {}, method: {:?}",
             //     url,
@@ -415,7 +520,7 @@ impl AsyncWrite for HttpWriter {
         }
         // Wait for the HTTP request to complete
         use futures::FutureExt;
-        match Pin::new(&mut self.get_mut().handle).poll_unpin(_cx) {
+        match Pin::new(&mut self.get_mut().handle).poll_unpin(cx) {
             Poll::Ready(Ok(_)) => {
                 // http_log!(
                 //     "[HttpWriter::poll_shutdown] HTTP request finished successfully, url: {}, method: {:?}",
@@ -437,77 +542,126 @@ impl AsyncWrite for HttpWriter {
     }
 }
 
-// #[cfg(test)]
-// mod tests {
-//     use super::*;
-//     use reqwest::Method;
-//     use std::vec;
-//     use tokio::io::{AsyncReadExt, AsyncWriteExt};
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use axum::{Router, body::Body, extract::State, http::StatusCode, response::IntoResponse, routing::get};
+    use http_body_util::BodyExt as _;
+    use std::io::IoSlice;
+    use std::sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    };
+    use tokio::{
+        io::{AsyncReadExt, AsyncWriteExt},
+        net::TcpListener,
+        sync::Mutex,
+    };
 
-//     #[tokio::test]
-//     async fn test_http_writer_err() {
-//         // Use a real local server for integration, or mockito for unit test
-//         // Here, we use the Go test server at 127.0.0.1:8081 (scripts/testfile.go)
-//         let url = "http://127.0.0.1:8081/testfile".to_string();
-//         let data = vec![42u8; 8];
+    #[derive(Clone, Default)]
+    struct TestState {
+        head_count: Arc<AtomicUsize>,
+        get_count: Arc<AtomicUsize>,
+        put_count: Arc<AtomicUsize>,
+        put_bodies: Arc<Mutex<Vec<Vec<u8>>>>,
+    }
 
-//         // Write
-//         // Add header X-Deny-Write = 1 to simulate non-writable situation
-//         let mut headers = HeaderMap::new();
-//         headers.insert("X-Deny-Write", "1".parse().unwrap());
-//         // Here we use PUT method
-//         let writer_result = HttpWriter::new(url.clone(), Method::PUT, headers).await;
-//         match writer_result {
-//             Ok(mut writer) => {
-//                 // If creation succeeds, write should fail
-//                 let write_result = writer.write_all(&data).await;
-//                 assert!(write_result.is_err(), "write_all should fail when server denies write");
-//                 if let Err(e) = write_result {
-//                     println!("write_all error: {e}");
-//                 }
-//                 let shutdown_result = writer.shutdown().await;
-//                 if let Err(e) = shutdown_result {
-//                     println!("shutdown error: {e}");
-//                 }
-//             }
-//             Err(e) => {
-//                 // Direct construction failure is also acceptable
-//                 println!("HttpWriter::new error: {e}");
-//                 assert!(
-//                     e.to_string().contains("Empty PUT failed") || e.to_string().contains("Forbidden"),
-//                     "unexpected error: {e}"
-//                 );
-//                 return;
-//             }
-//         }
-//         // Should not reach here
-//         panic!("HttpWriter should not allow writing when server denies write");
-//     }
+    async fn get_stream(State(state): State<TestState>) -> impl IntoResponse {
+        state.get_count.fetch_add(1, Ordering::SeqCst);
+        (StatusCode::OK, Body::from("hello"))
+    }
 
-//     #[tokio::test]
-//     async fn test_http_writer_and_reader_ok() {
-//         // Use local Go test server
-//         let url = "http://127.0.0.1:8081/testfile".to_string();
-//         let data = vec![99u8; 512 * 1024]; // 512KB of data
+    async fn reject_head(State(state): State<TestState>) -> impl IntoResponse {
+        state.head_count.fetch_add(1, Ordering::SeqCst);
+        StatusCode::METHOD_NOT_ALLOWED
+    }
 
-//         // Write (without X-Deny-Write)
-//         let headers = HeaderMap::new();
-//         let mut writer = HttpWriter::new(url.clone(), Method::PUT, headers).await.unwrap();
-//         writer.write_all(&data).await.unwrap();
-//         writer.shutdown().await.unwrap();
+    async fn accept_put(State(state): State<TestState>, body: Body) -> impl IntoResponse {
+        state.put_count.fetch_add(1, Ordering::SeqCst);
+        let bytes = body.collect().await.unwrap().to_bytes();
+        state.put_bodies.lock().await.push(bytes.to_vec());
+        StatusCode::OK
+    }
 
-//         http_log!("Wrote {} bytes to {} (ok case)", data.len(), url);
+    async fn start_test_server(state: TestState) -> (String, tokio::task::JoinHandle<()>) {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let app = Router::new()
+            .route("/stream", get(get_stream).head(reject_head).put(accept_put))
+            .with_state(state);
 
-//         // Read back
-//         let mut reader = HttpReader::with_capacity(url.clone(), Method::GET, HeaderMap::new(), 8192)
-//             .await
-//             .unwrap();
-//         let mut buf = Vec::new();
-//         reader.read_to_end(&mut buf).await.unwrap();
-//         assert_eq!(buf, data);
+        let handle = tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
 
-//         // println!("Read {} bytes from {} (ok case)", buf.len(), url);
-//         // tokio::time::sleep(std::time::Duration::from_secs(2)).await; // Wait for server to process
-//         // println!("[test_http_writer_and_reader_ok] completed successfully");
-//     }
-// }
+        (format!("http://{addr}/stream"), handle)
+    }
+
+    #[tokio::test]
+    async fn http_reader_does_not_send_preflight_head() {
+        let state = TestState::default();
+        let (url, handle) = start_test_server(state.clone()).await;
+
+        let mut reader = HttpReader::new(url, Method::GET, HeaderMap::new(), None).await.unwrap();
+        let mut buf = Vec::new();
+        reader.read_to_end(&mut buf).await.unwrap();
+
+        assert_eq!(buf, b"hello");
+        assert_eq!(state.head_count.load(Ordering::SeqCst), 0);
+        assert_eq!(state.get_count.load(Ordering::SeqCst), 1);
+
+        handle.abort();
+    }
+
+    #[tokio::test]
+    async fn http_writer_does_not_send_empty_preflight_put() {
+        let state = TestState::default();
+        let (url, handle) = start_test_server(state.clone()).await;
+
+        let mut writer = HttpWriter::new(url, Method::PUT, HeaderMap::new()).await.unwrap();
+        writer.write_all(b"payload").await.unwrap();
+        writer.shutdown().await.unwrap();
+
+        assert_eq!(state.put_count.load(Ordering::SeqCst), 1);
+        assert_eq!(state.put_bodies.lock().await.as_slice(), &[b"payload".to_vec()]);
+
+        handle.abort();
+    }
+
+    #[tokio::test]
+    async fn http_writer_handles_many_small_writes() {
+        let state = TestState::default();
+        let (url, handle) = start_test_server(state.clone()).await;
+
+        let mut writer = HttpWriter::new(url, Method::PUT, HeaderMap::new()).await.unwrap();
+        let chunk = b"0123456789abcdef";
+        let mut expected = Vec::new();
+        for _ in 0..256 {
+            writer.write_all(chunk).await.unwrap();
+            expected.extend_from_slice(chunk);
+        }
+        writer.shutdown().await.unwrap();
+
+        assert_eq!(state.put_count.load(Ordering::SeqCst), 1);
+        assert_eq!(state.put_bodies.lock().await.as_slice(), &[expected]);
+
+        handle.abort();
+    }
+
+    #[tokio::test]
+    async fn http_writer_supports_vectored_writes() {
+        let state = TestState::default();
+        let (url, handle) = start_test_server(state.clone()).await;
+
+        let mut writer = HttpWriter::new(url, Method::PUT, HeaderMap::new()).await.unwrap();
+        let bufs = [IoSlice::new(b"hello "), IoSlice::new(b"world")];
+        let written = writer.write_vectored(&bufs).await.unwrap();
+        assert_eq!(written, 11);
+        writer.shutdown().await.unwrap();
+
+        assert_eq!(state.put_count.load(Ordering::SeqCst), 1);
+        assert_eq!(state.put_bodies.lock().await.as_slice(), &[b"hello world".to_vec()]);
+
+        handle.abort();
+    }
+}

crates/rio/src/writer.rs

@@ -20,7 +20,7 @@ use crate::HttpWriter;
 
 pub enum Writer {
     Cursor(Cursor<Vec<u8>>),
-    Http(HttpWriter),
+    Http(Box<HttpWriter>),
     Other(Box<dyn AsyncWrite + Unpin + Send + Sync>),
 }
 
@@ -38,7 +38,7 @@ impl Writer {
     }
 
     pub fn from_http(w: HttpWriter) -> Self {
-        Writer::Http(w)
+        Writer::Http(Box::new(w))
     }
 
     pub fn into_cursor_inner(self) -> Option<Vec<u8>> {
@@ -56,14 +56,14 @@ impl Writer {
     }
     pub fn as_http(&mut self) -> Option<&mut HttpWriter> {
         match self {
-            Writer::Http(w) => Some(w),
+            Writer::Http(w) => Some(w.as_mut()),
             _ => None,
         }
     }
 
     pub fn into_http(self) -> Option<HttpWriter> {
         match self {
-            Writer::Http(w) => Some(w),
+            Writer::Http(w) => Some(*w),
             _ => None,
         }
     }

crates/scanner/src/scanner.rs

@@ -71,6 +71,16 @@ fn randomized_cycle_delay_for(interval: Duration) -> Duration {
     delay.max(Duration::from_secs(1))
 }
 
+fn initial_scanner_delay() -> Duration {
+    initial_scanner_delay_for(scanner_start_delay_secs())
+}
+
+fn initial_scanner_delay_for(start_delay_secs: Option<u64>) -> Duration {
+    start_delay_secs
+        .map(|secs| randomized_cycle_delay_for(Duration::from_secs(secs)))
+        .unwrap_or_else(|| Duration::from_secs(rand::random::<u64>() % 5))
+}
+
 pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
     // Force init global sleeper so config is read once at startup.
     let _ = &*SCANNER_SLEEPER;
@@ -78,7 +88,7 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
     let ctx_clone = ctx.clone();
     let storeapi_clone = storeapi.clone();
     tokio::spawn(async move {
-        let sleep_time = Duration::from_secs(rand::random::<u64>() % 5);
+        let sleep_time = initial_scanner_delay();
         tokio::time::sleep(sleep_time).await;
 
         loop {
@@ -246,7 +256,7 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc<ECStore>
 pub async fn run_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) -> Result<(), ScannerError> {
     // Acquire leader lock (write lock) to ensure only one scanner runs
     let _guard = match storeapi.new_ns_lock(RUSTFS_META_BUCKET, "leader.lock").await {
-        Ok(ns_lock) => match ns_lock.get_write_lock(get_lock_acquire_timeout()).await {
+        Ok(ns_lock) => match ns_lock.get_write_lock_quiet(get_lock_acquire_timeout()).await {
             Ok(guard) => {
                 debug!("run_data_scanner: acquired leader write lock");
                 guard
@@ -358,6 +368,14 @@ mod tests {
         assert!(delay <= Duration::from_secs(132));
     }
 
+    #[test]
+    #[serial]
+    fn test_initial_scanner_delay_uses_configured_start_delay() {
+        let delay = initial_scanner_delay_for(Some(120));
+        assert!(delay >= Duration::from_secs(108));
+        assert!(delay <= Duration::from_secs(132));
+    }
+
     #[test]
     #[serial]
     fn test_randomized_cycle_delay_handles_small_start_delay() {

rustfs/src/admin/mod.rs

@@ -16,7 +16,6 @@ mod auth;
 pub mod console;
 pub mod handlers;
 pub mod router;
-mod rpc;
 pub mod utils;
 
 #[cfg(test)]
@@ -28,7 +27,6 @@ use handlers::{
     bucket_meta, heal, health, kms, oidc, pools, profile_admin, quota, rebalance, replication, sts, system, tier, user,
 };
 use router::{AdminOperation, S3Router};
-use rpc::register_rpc_route;
 use s3s::route::S3Route;
 
 /// Create admin router
@@ -44,7 +42,6 @@ pub fn make_admin_route(console_enabled: bool) -> std::io::Result<impl S3Route>
     health::register_health_route(&mut r)?;
     sts::register_admin_auth_route(&mut r)?;
 
-    register_rpc_route(&mut r)?;
     user::register_user_route(&mut r)?;
     system::register_system_route(&mut r)?;
     pools::register_pool_route(&mut r)?;

rustfs/src/admin/route_registration_test.rs

@@ -15,7 +15,6 @@
 use crate::admin::{
     handlers::{bucket_meta, heal, health, kms, pools, profile_admin, quota, rebalance, replication, sts, system, tier, user},
     router::{AdminOperation, S3Router},
-    rpc,
 };
 use crate::server::{ADMIN_PREFIX, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH};
 use hyper::Method;
@@ -54,8 +53,6 @@ fn test_register_routes_cover_representative_admin_paths() {
     replication::register_replication_route(&mut router).expect("register replication route");
     profile_admin::register_profiling_route(&mut router).expect("register profile route");
     kms::register_kms_route(&mut router).expect("register kms route");
-    rpc::register_rpc_route(&mut router).expect("register rpc route");
-
     assert_route(&router, Method::GET, HEALTH_PREFIX);
     assert_route(&router, Method::HEAD, HEALTH_PREFIX);
     assert_route(&router, Method::GET, HEALTH_READY_PATH);
@@ -117,8 +114,11 @@ fn test_register_routes_cover_representative_admin_paths() {
     assert_route(&router, Method::POST, &admin_path("/v3/kms/keys"));
     assert_route(&router, Method::GET, &admin_path("/v3/kms/keys"));
     assert_route(&router, Method::GET, &admin_path("/v3/kms/keys/test-key"));
-    assert_route(&router, Method::GET, "/rustfs/rpc/read_file_stream");
-    assert_route(&router, Method::HEAD, "/rustfs/rpc/read_file_stream");
+
+    assert!(
+        !router.contains_route(Method::GET, "/rustfs/rpc/read_file_stream"),
+        "internode rpc routes should no longer be registered inside the admin router"
+    );
 }
 
 #[test]
@@ -160,9 +160,8 @@ fn test_admin_alias_paths_match_existing_admin_routes() {
 }
 
 #[test]
-fn test_phase5_admin_info_and_rpc_read_file_contract() {
+fn test_phase5_admin_info_contract() {
     let system_src = include_str!("handlers/system.rs");
-    let rpc_src = include_str!("rpc.rs");
 
     let server_info_impl_marker = "impl Operation for ServerInfoHandler";
     let server_info_impl_start = system_src
@@ -175,24 +174,4 @@ fn test_phase5_admin_info_and_rpc_read_file_contract() {
             && server_info_impl_block.contains("execute_query_server_info(QueryServerInfoRequest { include_pools: true })"),
         "admin server info path must be served through DefaultAdminUsecase::execute_query_server_info"
     );
-
-    let register_route_marker = "pub fn register_rpc_route";
-    let register_route_start = rpc_src
-        .find(register_route_marker)
-        .expect("Expected register_rpc_route in rpc.rs");
-    let register_route_block = &rpc_src[register_route_start..];
-
-    let read_file_marker = "pub struct ReadFile {}";
-    let read_file_start = rpc_src.find(read_file_marker).expect("Expected ReadFile operation in rpc.rs");
-    let read_file_block = &rpc_src[read_file_start..];
-
-    assert!(
-        register_route_block.contains("format!(\"{}{}\", RPC_PREFIX, \"/read_file_stream\")"),
-        "rpc read_file_stream route path must remain registered with RPC_PREFIX"
-    );
-
-    assert!(
-        read_file_block.contains(".read_file_stream(&query.volume, &query.path, query.offset, query.length)"),
-        "rpc read_file_stream route must remain wired to disk.read_file_stream"
-    );
 }

rustfs/src/admin/router.rs

@@ -14,9 +14,7 @@
 
 use crate::admin::console::{is_console_path, make_console_server};
 use crate::admin::handlers::oidc::is_oidc_path;
-use crate::server::{
-    ADMIN_PREFIX, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, RPC_PREFIX,
-};
+use crate::server::{ADMIN_PREFIX, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH};
 use hyper::HeaderMap;
 use hyper::Method;
 use hyper::StatusCode;
@@ -24,7 +22,6 @@ use hyper::Uri;
 use hyper::http::Extensions;
 use matchit::Params;
 use matchit::Router;
-use rustfs_ecstore::rpc::verify_rpc_signature;
 use s3s::Body;
 use s3s::S3Request;
 use s3s::S3Response;
@@ -33,7 +30,6 @@ use s3s::header;
 use s3s::route::S3Route;
 use s3s::s3_error;
 use tower::Service;
-use tracing::error;
 
 pub struct S3Router<T> {
     router: Router<T>,
@@ -140,7 +136,7 @@ where
             return true;
         }
 
-        is_admin_path(path) || path.starts_with(RPC_PREFIX) || is_console_path(path)
+        is_admin_path(path) || is_console_path(path)
     }
 
     // check_access before call
@@ -168,18 +164,6 @@ where
             return Ok(());
         }
 
-        // Check RPC signature verification
-        if req.uri.path().starts_with(RPC_PREFIX) {
-            // Skip signature verification for HEAD requests (health checks)
-            if req.method != Method::HEAD {
-                verify_rpc_signature(&req.uri.to_string(), &req.method, &req.headers).map_err(|e| {
-                    error!("RPC signature verification failed: {}", e);
-                    s3_error!(AccessDenied, "{}", e)
-                })?;
-            }
-            return Ok(());
-        }
-
         // Allow unauthenticated STS requests to POST / (AssumeRoleWithWebIdentity
         // doesn't use SigV4 — the JWT token in the request body is the authentication).
         // The handler dispatches on the Action parameter: AssumeRole will reject if

rustfs/src/admin/rpc.rs

@@ -1,218 +0,0 @@
-// 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::admin::router::{AdminOperation, Operation, S3Router};
-use crate::server::RPC_PREFIX;
-use futures::StreamExt;
-use http::StatusCode;
-use hyper::Method;
-use matchit::Params;
-use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
-use rustfs_ecstore::disk::DiskAPI;
-use rustfs_ecstore::disk::WalkDirOptions;
-use rustfs_ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE;
-use rustfs_ecstore::store::find_local_disk;
-use rustfs_utils::net::bytes_stream;
-use s3s::Body;
-use s3s::S3Request;
-use s3s::S3Response;
-use s3s::S3Result;
-use s3s::dto::StreamingBlob;
-use s3s::s3_error;
-use serde_urlencoded::from_bytes;
-use tokio::io::AsyncWriteExt;
-use tokio_util::io::ReaderStream;
-use tracing::warn;
-
-pub fn register_rpc_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
-    r.insert(
-        Method::GET,
-        format!("{}{}", RPC_PREFIX, "/read_file_stream").as_str(),
-        AdminOperation(&ReadFile {}),
-    )?;
-
-    r.insert(
-        Method::HEAD,
-        format!("{}{}", RPC_PREFIX, "/read_file_stream").as_str(),
-        AdminOperation(&ReadFile {}),
-    )?;
-
-    r.insert(
-        Method::PUT,
-        format!("{}{}", RPC_PREFIX, "/put_file_stream").as_str(),
-        AdminOperation(&PutFile {}),
-    )?;
-
-    r.insert(
-        Method::GET,
-        format!("{}{}", RPC_PREFIX, "/walk_dir").as_str(),
-        AdminOperation(&WalkDir {}),
-    )?;
-
-    r.insert(
-        Method::HEAD,
-        format!("{}{}", RPC_PREFIX, "/walk_dir").as_str(),
-        AdminOperation(&WalkDir {}),
-    )?;
-
-    Ok(())
-}
-
-// /rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}"
-#[derive(Debug, Default, serde::Deserialize)]
-pub struct ReadFileQuery {
-    disk: String,
-    volume: String,
-    path: String,
-    offset: usize,
-    length: usize,
-}
-pub struct ReadFile {}
-#[async_trait::async_trait]
-impl Operation for ReadFile {
-    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        if req.method == Method::HEAD {
-            return Ok(S3Response::new((StatusCode::OK, Body::empty())));
-        }
-        let query = {
-            if let Some(query) = req.uri.query() {
-                let input: ReadFileQuery =
-                    from_bytes(query.as_bytes()).map_err(|e| s3_error!(InvalidArgument, "get query failed1 {:?}", e))?;
-                input
-            } else {
-                ReadFileQuery::default()
-            }
-        };
-
-        let Some(disk) = find_local_disk(&query.disk).await else {
-            return Err(s3_error!(InvalidArgument, "disk not found"));
-        };
-
-        let file = disk
-            .read_file_stream(&query.volume, &query.path, query.offset, query.length)
-            .await
-            .map_err(|e| s3_error!(InternalError, "read file err {}", e))?;
-
-        Ok(S3Response::new((
-            StatusCode::OK,
-            Body::from(StreamingBlob::wrap(bytes_stream(
-                ReaderStream::with_capacity(file, DEFAULT_READ_BUFFER_SIZE),
-                query.length,
-            ))),
-        )))
-    }
-}
-
-#[derive(Debug, Default, serde::Deserialize)]
-pub struct WalkDirQuery {
-    disk: String,
-}
-
-pub struct WalkDir {}
-
-#[async_trait::async_trait]
-impl Operation for WalkDir {
-    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        if req.method == Method::HEAD {
-            return Ok(S3Response::new((StatusCode::OK, Body::empty())));
-        }
-
-        let query = {
-            if let Some(query) = req.uri.query() {
-                let input: WalkDirQuery =
-                    from_bytes(query.as_bytes()).map_err(|e| s3_error!(InvalidArgument, "get query failed1 {:?}", e))?;
-                input
-            } else {
-                WalkDirQuery::default()
-            }
-        };
-
-        let mut input = req.input;
-        let body = match input.store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE).await {
-            Ok(b) => b,
-            Err(e) => {
-                warn!("get body failed, e: {:?}", e);
-                return Err(s3_error!(InvalidRequest, "RPC request body too large or failed to read"));
-            }
-        };
-
-        // let body_bytes = decrypt_data(input_cred.secret_key.expose().as_bytes(), &body)
-        //     .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidArgument, format!("decrypt_data err {}", e)))?;
-
-        let args: WalkDirOptions =
-            serde_json::from_slice(&body).map_err(|e| s3_error!(InternalError, "unmarshal body err {}", e))?;
-        let Some(disk) = find_local_disk(&query.disk).await else {
-            return Err(s3_error!(InvalidArgument, "disk not found"));
-        };
-
-        let (rd, mut wd) = tokio::io::duplex(DEFAULT_READ_BUFFER_SIZE);
-
-        tokio::spawn(async move {
-            if let Err(e) = disk.walk_dir(args, &mut wd).await {
-                warn!("walk dir err {}", e);
-            }
-        });
-
-        let body = Body::from(StreamingBlob::wrap(ReaderStream::with_capacity(rd, DEFAULT_READ_BUFFER_SIZE)));
-        Ok(S3Response::new((StatusCode::OK, body)))
-    }
-}
-
-// /rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}"
-#[derive(Debug, Default, serde::Deserialize)]
-pub struct PutFileQuery {
-    disk: String,
-    volume: String,
-    path: String,
-    append: bool,
-    size: i64,
-}
-pub struct PutFile {}
-#[async_trait::async_trait]
-impl Operation for PutFile {
-    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        let query = {
-            if let Some(query) = req.uri.query() {
-                let input: PutFileQuery =
-                    from_bytes(query.as_bytes()).map_err(|e| s3_error!(InvalidArgument, "get query failed1 {:?}", e))?;
-                input
-            } else {
-                PutFileQuery::default()
-            }
-        };
-
-        let Some(disk) = find_local_disk(&query.disk).await else {
-            return Err(s3_error!(InvalidArgument, "disk not found"));
-        };
-
-        let mut file = if query.append {
-            disk.append_file(&query.volume, &query.path)
-                .await
-                .map_err(|e| s3_error!(InternalError, "append file err {}", e))?
-        } else {
-            disk.create_file("", &query.volume, &query.path, query.size)
-                .await
-                .map_err(|e| s3_error!(InternalError, "read file err {}", e))?
-        };
-
-        let mut body = req.input;
-        while let Some(item) = body.next().await {
-            let bytes = item.map_err(|e| s3_error!(InternalError, "body stream err {}", e))?;
-            let result = file.write_all(&bytes).await;
-            result.map_err(|e| s3_error!(InternalError, "write file err {}", e))?;
-        }
-
-        Ok(S3Response::new((StatusCode::OK, Body::empty())))
-    }
-}

rustfs/src/main.rs

@@ -530,14 +530,17 @@ async fn run(config: config::Config) -> Result<()> {
         "Background services configuration: scanner={}, heal={}", enable_scanner, enable_heal
     );
 
-    // Initialize heal manager and scanner based on environment variables
+    // Scanner depends on the heal channel/manager, so scanner implies heal.
     if enable_heal || enable_scanner {
         let heal_storage = Arc::new(ECStoreHealStorage::new(store.clone()));
-
         init_heal_manager(heal_storage, None).await?;
+    }
 
+    if enable_scanner {
         init_data_scanner(ctx.clone(), store.clone()).await;
-    } else {
+    }
+
+    if !enable_heal && !enable_scanner {
         info!(target: "rustfs::main::run","Both scanner and heal are disabled, skipping AHM service initialization");
     }
 
@@ -623,20 +626,24 @@ async fn handle_shutdown(
     let enable_scanner = get_env_bool_with_aliases(ENV_SCANNER_ENABLED, &[ENV_SCANNER_ENABLED_DEPRECATED], true);
     let enable_heal = get_env_bool_with_aliases(ENV_HEAL_ENABLED, &[ENV_HEAL_ENABLED_DEPRECATED], true);
 
-    // Stop background services based on what was enabled
-    if enable_scanner || enable_heal {
+    // Stop background services based on what was enabled.
+    if enable_scanner {
         info!(
             target: "rustfs::main::handle_shutdown",
-            "Stopping background services (data scanner and auto heal)..."
+            "Stopping background services (data scanner)..."
         );
         shutdown_background_services();
+    }
 
+    if enable_heal || enable_scanner {
         info!(
             target: "rustfs::main::handle_shutdown",
             "Stopping AHM services..."
         );
         shutdown_ahm_services();
-    } else {
+    }
+
+    if !enable_scanner && !enable_heal {
         info!(
             target: "rustfs::main::handle_shutdown",
             "Background services were disabled, skipping AHM shutdown"

rustfs/src/server/http.rs

@@ -24,6 +24,7 @@ use crate::server::{
     layer::{AdminChunkedContentLengthCompatLayer, ConditionalCorsLayer, ObjectAttributesEtagFixLayer, RedirectLayer},
 };
 use crate::storage;
+use crate::storage::rpc::InternodeRpcService;
 use crate::storage::tonic_service::make_server;
 use bytes::Bytes;
 use http::{HeaderMap, Method, Request as HttpRequest, Response};
@@ -592,6 +593,7 @@ fn process_connection(
         let http_service = SwiftService::new(true, None, s3_service);
         #[cfg(not(feature = "swift"))]
         let http_service = s3_service;
+        let http_service = InternodeRpcService::new(http_service);
 
         let service = hybrid(http_service, rpc_service);
 

rustfs/src/storage/rpc/disk.rs

@@ -13,6 +13,26 @@
 // limitations under the License.
 
 use super::*;
+use serde::de::DeserializeOwned;
+use std::io::Cursor;
+
+fn decode_msgpack_or_json<T: DeserializeOwned>(binary: &[u8], json: &str, value_name: &str) -> std::result::Result<T, DiskError> {
+    if !binary.is_empty() {
+        let mut deserializer = rmp_serde::Deserializer::new(Cursor::new(binary));
+        return T::deserialize(&mut deserializer)
+            .map_err(|err| DiskError::other(format!("decode {value_name} msgpack failed: {err}")));
+    }
+
+    serde_json::from_str(json).map_err(|err| DiskError::other(format!("decode {value_name} failed: {err}")))
+}
+
+fn encode_msgpack<T: serde::Serialize>(value: &T, value_name: &str) -> std::result::Result<Vec<u8>, DiskError> {
+    let mut serializer = rmp_serde::Serializer::new(Vec::new());
+    value
+        .serialize(&mut serializer)
+        .map_err(|err| DiskError::other(format!("encode {value_name} msgpack failed: {err}")))?;
+    Ok(serializer.into_inner())
+}
 
 impl NodeService {
     pub(super) async fn handle_disk_info(&self, request: Request<DiskInfoRequest>) -> Result<Response<DiskInfoResponse>, Status> {
@@ -86,32 +106,44 @@ impl NodeService {
     ) -> Result<Response<ReadMultipleResponse>, Status> {
         let request = request.into_inner();
         if let Some(disk) = self.find_disk(&request.disk).await {
-            let read_multiple_req = match serde_json::from_str::<ReadMultipleReq>(&request.read_multiple_req) {
+            let read_multiple_req = match decode_msgpack_or_json::<ReadMultipleReq>(
+                &request.read_multiple_req_bin,
+                &request.read_multiple_req,
+                "ReadMultipleReq",
+            ) {
                 Ok(read_multiple_req) => read_multiple_req,
                 Err(err) => {
                     return Ok(Response::new(ReadMultipleResponse {
                         success: false,
                         read_multiple_resps: Vec::new(),
+                        read_multiple_resps_bin: Vec::new(),
                         error: Some(DiskError::other(format!("decode ReadMultipleReq failed: {err}")).into()),
                     }));
                 }
             };
             match disk.read_multiple(read_multiple_req).await {
                 Ok(read_multiple_resps) => {
-                    let read_multiple_resps = read_multiple_resps
+                    let read_multiple_resps: Vec<String> = read_multiple_resps
                         .into_iter()
                         .filter_map(|read_multiple_resp| serde_json::to_string(&read_multiple_resp).ok())
                         .collect();
+                    let read_multiple_resps_bin = read_multiple_resps
+                        .iter()
+                        .filter_map(|json_str| serde_json::from_str::<ReadMultipleResp>(json_str).ok())
+                        .filter_map(|resp| encode_msgpack(&resp, "ReadMultipleResp").ok())
+                        .collect();
 
                     Ok(Response::new(ReadMultipleResponse {
                         success: true,
                         read_multiple_resps,
+                        read_multiple_resps_bin,
                         error: None,
                     }))
                 }
                 Err(err) => Ok(Response::new(ReadMultipleResponse {
                     success: false,
                     read_multiple_resps: Vec::new(),
+                    read_multiple_resps_bin: Vec::new(),
                     error: Some(err.into()),
                 })),
             }
@@ -119,6 +151,7 @@ impl NodeService {
             Ok(Response::new(ReadMultipleResponse {
                 success: false,
                 read_multiple_resps: Vec::new(),
+                read_multiple_resps_bin: Vec::new(),
                 error: Some(DiskError::other("can not find disk".to_string()).into()),
             }))
         }
@@ -239,21 +272,34 @@ impl NodeService {
         let request = request.into_inner();
         if let Some(disk) = self.find_disk(&request.disk).await {
             match disk.read_xl(&request.volume, &request.path, request.read_data).await {
-                Ok(raw_file_info) => match serde_json::to_string(&raw_file_info) {
-                    Ok(raw_file_info) => Ok(Response::new(ReadXlResponse {
-                        success: true,
-                        raw_file_info,
-                        error: None,
-                    })),
-                    Err(err) => Ok(Response::new(ReadXlResponse {
-                        success: false,
-                        raw_file_info: String::new(),
-                        error: Some(DiskError::other(format!("encode data failed: {err}")).into()),
-                    })),
-                },
+                Ok(raw_file_info) => {
+                    let raw_file_info_json = serde_json::to_string(&raw_file_info);
+                    let raw_file_info_bin = encode_msgpack(&raw_file_info, "RawFileInfo");
+                    match (raw_file_info_json, raw_file_info_bin) {
+                        (Ok(raw_file_info), Ok(raw_file_info_bin)) => Ok(Response::new(ReadXlResponse {
+                            success: true,
+                            raw_file_info,
+                            raw_file_info_bin,
+                            error: None,
+                        })),
+                        (Err(err), _) => Ok(Response::new(ReadXlResponse {
+                            success: false,
+                            raw_file_info: String::new(),
+                            raw_file_info_bin: Vec::new(),
+                            error: Some(DiskError::other(format!("encode data failed: {err}")).into()),
+                        })),
+                        (_, Err(err)) => Ok(Response::new(ReadXlResponse {
+                            success: false,
+                            raw_file_info: String::new(),
+                            raw_file_info_bin: Vec::new(),
+                            error: Some(DiskError::other(format!("encode data failed: {err}")).into()),
+                        })),
+                    }
+                }
                 Err(err) => Ok(Response::new(ReadXlResponse {
                     success: false,
                     raw_file_info: String::new(),
+                    raw_file_info_bin: Vec::new(),
                     error: Some(err.into()),
                 })),
             }
@@ -261,6 +307,7 @@ impl NodeService {
             Ok(Response::new(ReadXlResponse {
                 success: false,
                 raw_file_info: String::new(),
+                raw_file_info_bin: Vec::new(),
                 error: Some(DiskError::other("can not find disk".to_string()).into()),
             }))
         }
@@ -272,12 +319,13 @@ impl NodeService {
     ) -> Result<Response<ReadVersionResponse>, Status> {
         let request = request.into_inner();
         if let Some(disk) = self.find_disk(&request.disk).await {
-            let opts = match serde_json::from_str::<ReadOptions>(&request.opts) {
+            let opts = match decode_msgpack_or_json::<ReadOptions>(&request.opts_bin, &request.opts, "ReadOptions") {
                 Ok(options) => options,
                 Err(err) => {
                     return Ok(Response::new(ReadVersionResponse {
                         success: false,
                         file_info: String::new(),
+                        file_info_bin: Vec::new(),
                         error: Some(DiskError::other(format!("decode ReadOptions failed: {err}")).into()),
                     }));
                 }
@@ -286,21 +334,34 @@ impl NodeService {
                 .read_version("", &request.volume, &request.path, &request.version_id, &opts)
                 .await
             {
-                Ok(file_info) => match serde_json::to_string(&file_info) {
-                    Ok(file_info) => Ok(Response::new(ReadVersionResponse {
-                        success: true,
-                        file_info,
-                        error: None,
-                    })),
-                    Err(err) => Ok(Response::new(ReadVersionResponse {
-                        success: false,
-                        file_info: String::new(),
-                        error: Some(DiskError::other(format!("encode data failed: {err}")).into()),
-                    })),
-                },
+                Ok(file_info) => {
+                    let file_info_json = serde_json::to_string(&file_info);
+                    let file_info_bin = encode_msgpack(&file_info, "FileInfo");
+                    match (file_info_json, file_info_bin) {
+                        (Ok(file_info), Ok(file_info_bin)) => Ok(Response::new(ReadVersionResponse {
+                            success: true,
+                            file_info,
+                            file_info_bin,
+                            error: None,
+                        })),
+                        (Err(err), _) => Ok(Response::new(ReadVersionResponse {
+                            success: false,
+                            file_info: String::new(),
+                            file_info_bin: Vec::new(),
+                            error: Some(DiskError::other(format!("encode data failed: {err}")).into()),
+                        })),
+                        (_, Err(err)) => Ok(Response::new(ReadVersionResponse {
+                            success: false,
+                            file_info: String::new(),
+                            file_info_bin: Vec::new(),
+                            error: Some(DiskError::other(format!("encode data failed: {err}")).into()),
+                        })),
+                    }
+                }
                 Err(err) => Ok(Response::new(ReadVersionResponse {
                     success: false,
                     file_info: String::new(),
+                    file_info_bin: Vec::new(),
                     error: Some(err.into()),
                 })),
             }
@@ -308,6 +369,7 @@ impl NodeService {
             Ok(Response::new(ReadVersionResponse {
                 success: false,
                 file_info: String::new(),
+                file_info_bin: Vec::new(),
                 error: Some(DiskError::other("can not find disk".to_string()).into()),
             }))
         }
@@ -319,7 +381,7 @@ impl NodeService {
     ) -> Result<Response<WriteMetadataResponse>, Status> {
         let request = request.into_inner();
         if let Some(disk) = self.find_disk(&request.disk).await {
-            let file_info = match serde_json::from_str::<FileInfo>(&request.file_info) {
+            let file_info = match decode_msgpack_or_json::<FileInfo>(&request.file_info_bin, &request.file_info, "FileInfo") {
                 Ok(file_info) => file_info,
                 Err(err) => {
                     return Ok(Response::new(WriteMetadataResponse {
@@ -352,7 +414,7 @@ impl NodeService {
     ) -> Result<Response<UpdateMetadataResponse>, Status> {
         let request = request.into_inner();
         if let Some(disk) = self.find_disk(&request.disk).await {
-            let file_info = match serde_json::from_str::<FileInfo>(&request.file_info) {
+            let file_info = match decode_msgpack_or_json::<FileInfo>(&request.file_info_bin, &request.file_info, "FileInfo") {
                 Ok(file_info) => file_info,
                 Err(err) => {
                     return Ok(Response::new(UpdateMetadataResponse {
@@ -361,7 +423,8 @@ impl NodeService {
                     }));
                 }
             };
-            let opts = match serde_json::from_str::<UpdateMetadataOpts>(&request.opts) {
+            let opts = match decode_msgpack_or_json::<UpdateMetadataOpts>(&request.opts_bin, &request.opts, "UpdateMetadataOpts")
+            {
                 Ok(opts) => opts,
                 Err(err) => {
                     return Ok(Response::new(UpdateMetadataResponse {
@@ -910,3 +973,42 @@ impl NodeService {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde::{Deserialize, Serialize};
+
+    #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
+    struct SamplePayload {
+        name: String,
+        count: u32,
+    }
+
+    #[test]
+    fn decode_msgpack_or_json_prefers_binary_payload() {
+        let payload = SamplePayload {
+            name: "rustfs".to_string(),
+            count: 3,
+        };
+
+        let binary = encode_msgpack(&payload, "SamplePayload").unwrap();
+        let decoded =
+            decode_msgpack_or_json::<SamplePayload>(&binary, r#"{"name":"ignored","count":1}"#, "SamplePayload").unwrap();
+
+        assert_eq!(decoded, payload);
+    }
+
+    #[test]
+    fn decode_msgpack_or_json_falls_back_to_json() {
+        let decoded = decode_msgpack_or_json::<SamplePayload>(&[], r#"{"name":"compat","count":7}"#, "SamplePayload").unwrap();
+
+        assert_eq!(
+            decoded,
+            SamplePayload {
+                name: "compat".to_string(),
+                count: 7,
+            }
+        );
+    }
+}

rustfs/src/storage/rpc/http_service.rs

@@ -0,0 +1,404 @@
+// 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::server::RPC_PREFIX;
+use bytes::{Bytes, BytesMut};
+use futures_util::TryStreamExt;
+use http::{HeaderMap, Method, Request, Response, StatusCode, Uri};
+use http_body_util::{BodyExt, Limited};
+use hyper::body::Incoming;
+use rustfs_common::internode_metrics::global_internode_metrics;
+use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
+use rustfs_ecstore::disk::{DiskAPI, WalkDirOptions};
+use rustfs_ecstore::rpc::verify_rpc_signature;
+use rustfs_ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE;
+use rustfs_ecstore::store::find_local_disk_by_ref;
+use rustfs_utils::net::bytes_stream;
+use s3s::Body;
+use s3s::dto::StreamingBlob;
+use serde::de::DeserializeOwned;
+use serde_urlencoded::from_bytes;
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::io::{self, AsyncWriteExt};
+use tokio_util::io::ReaderStream;
+use tower::Service;
+use tracing::warn;
+
+type BoxError = Box<dyn std::error::Error + Send + Sync + 'static>;
+type RpcErrorResponse = Box<Response<Body>>;
+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";
+
+#[derive(Clone)]
+pub struct InternodeRpcService<S> {
+    inner: S,
+}
+
+impl<S> InternodeRpcService<S> {
+    pub fn new(inner: S) -> Self {
+        Self { inner }
+    }
+}
+
+#[derive(Debug, Default, serde::Deserialize)]
+struct ReadFileQuery {
+    disk: String,
+    volume: String,
+    path: String,
+    offset: usize,
+    length: usize,
+}
+
+#[derive(Debug, Default, serde::Deserialize)]
+struct WalkDirQuery {
+    disk: String,
+}
+
+#[derive(Debug, Default, serde::Deserialize)]
+struct PutFileQuery {
+    disk: String,
+    volume: String,
+    path: String,
+    append: bool,
+    size: i64,
+}
+
+impl<S> Service<Request<Incoming>> for InternodeRpcService<S>
+where
+    S: Service<Request<Incoming>, Response = Response<Body>> + Clone + Send + 'static,
+    S::Future: Send + 'static,
+    S::Error: Into<BoxError> + Send + 'static,
+{
+    type Response = Response<Body>;
+    type Error = S::Error;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, req: Request<Incoming>) -> Self::Future {
+        if !is_internode_rpc_path(req.uri().path()) {
+            let mut inner = self.inner.clone();
+            return Box::pin(async move { inner.call(req).await });
+        }
+
+        Box::pin(async move { Ok(handle_internode_rpc(req).await) })
+    }
+}
+
+fn is_internode_rpc_path(path: &str) -> bool {
+    path.starts_with(RPC_PREFIX)
+}
+
+async fn handle_internode_rpc(req: Request<Incoming>) -> Response<Body> {
+    if let Err(response) = verify_internode_rpc_signature(req.uri(), req.method(), req.headers()) {
+        global_internode_metrics().record_error();
+        return *response;
+    }
+
+    let method = req.method().clone();
+    let path = req.uri().path();
+
+    let response = match (method, path) {
+        (Method::GET, READ_FILE_STREAM_PATH) | (Method::HEAD, READ_FILE_STREAM_PATH) => handle_read_file(req).await,
+        (Method::GET, WALK_DIR_PATH) | (Method::HEAD, WALK_DIR_PATH) => handle_walk_dir(req).await,
+        (Method::PUT, PUT_FILE_STREAM_PATH) => handle_put_file(req).await,
+        _ => response_with_status(StatusCode::NOT_FOUND, "internode rpc route not found"),
+    };
+
+    if !response.status().is_success() {
+        global_internode_metrics().record_error();
+    }
+
+    response
+}
+
+fn verify_internode_rpc_signature(uri: &Uri, method: &Method, headers: &HeaderMap) -> Result<(), RpcErrorResponse> {
+    if method == Method::HEAD {
+        return Ok(());
+    }
+
+    verify_rpc_signature(&uri.to_string(), method, headers).map_err(|e| {
+        Box::new(response_with_status(
+            StatusCode::FORBIDDEN,
+            format!("rpc signature verification failed: {e}"),
+        ))
+    })
+}
+
+async fn handle_read_file(req: Request<Incoming>) -> Response<Body> {
+    if req.method() == Method::HEAD {
+        return empty_ok();
+    }
+
+    let query = match parse_query::<ReadFileQuery>(&req) {
+        Ok(query) => query,
+        Err(response) => return *response,
+    };
+
+    let Some(disk) = find_local_disk_by_ref(&query.disk).await else {
+        return response_with_status(StatusCode::BAD_REQUEST, "disk not found");
+    };
+
+    let file = match disk
+        .read_file_stream(&query.volume, &query.path, query.offset, query.length)
+        .await
+    {
+        Ok(file) => file,
+        Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("read file err {e}")),
+    };
+
+    global_internode_metrics().record_incoming_request();
+    let stream = read_file_body_stream(file, query.length);
+
+    Response::builder()
+        .status(StatusCode::OK)
+        .body(Body::from(StreamingBlob::wrap(stream)))
+        .expect("failed to build read file stream response")
+}
+
+fn read_file_body_stream<R>(reader: R, length: usize) -> Pin<Box<dyn futures::Stream<Item = io::Result<Bytes>> + Send + Sync>>
+where
+    R: tokio::io::AsyncRead + Unpin + Send + Sync + 'static,
+{
+    let metrics = global_internode_metrics().clone();
+    let stream = ReaderStream::with_capacity(reader, DEFAULT_READ_BUFFER_SIZE).map_ok(move |bytes| {
+        metrics.record_sent_bytes(bytes.len());
+        bytes
+    });
+
+    if length == 0 {
+        Box::pin(stream)
+    } else {
+        Box::pin(bytes_stream(stream, length))
+    }
+}
+
+async fn handle_walk_dir(req: Request<Incoming>) -> Response<Body> {
+    if req.method() == Method::HEAD {
+        return empty_ok();
+    }
+
+    let query = match parse_query::<WalkDirQuery>(&req) {
+        Ok(query) => query,
+        Err(response) => return *response,
+    };
+
+    let Some(disk) = find_local_disk_by_ref(&query.disk).await else {
+        return response_with_status(StatusCode::BAD_REQUEST, "disk not found");
+    };
+
+    let body = match Limited::new(req.into_body(), MAX_ADMIN_REQUEST_BODY_SIZE).collect().await {
+        Ok(body) => body.to_bytes(),
+        Err(e) => return response_with_status(StatusCode::PAYLOAD_TOO_LARGE, format!("read body err {e}")),
+    };
+
+    let args: WalkDirOptions = match serde_json::from_slice(&body) {
+        Ok(args) => args,
+        Err(e) => return response_with_status(StatusCode::BAD_REQUEST, format!("unmarshal body err {e}")),
+    };
+
+    let (rd, mut wd) = tokio::io::duplex(DEFAULT_READ_BUFFER_SIZE);
+    tokio::spawn(async move {
+        if let Err(e) = disk.walk_dir(args, &mut wd).await {
+            warn!("walk dir err {}", e);
+        }
+    });
+
+    global_internode_metrics().record_incoming_request();
+    let metrics = global_internode_metrics().clone();
+    let stream = ReaderStream::with_capacity(rd, DEFAULT_READ_BUFFER_SIZE).map_ok(move |bytes| {
+        metrics.record_sent_bytes(bytes.len());
+        bytes
+    });
+
+    Response::builder()
+        .status(StatusCode::OK)
+        .body(Body::from(StreamingBlob::wrap(stream)))
+        .expect("failed to build walk dir response")
+}
+
+async fn handle_put_file(req: Request<Incoming>) -> Response<Body> {
+    let query = match parse_query::<PutFileQuery>(&req) {
+        Ok(query) => query,
+        Err(response) => return *response,
+    };
+
+    let Some(disk) = find_local_disk_by_ref(&query.disk).await else {
+        return response_with_status(StatusCode::BAD_REQUEST, "disk not found");
+    };
+
+    let mut file = if query.append {
+        match disk.append_file(&query.volume, &query.path).await {
+            Ok(file) => file,
+            Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("append file err {e}")),
+        }
+    } else {
+        match disk.create_file("", &query.volume, &query.path, query.size).await {
+            Ok(file) => file,
+            Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("create file err {e}")),
+        }
+    };
+
+    let copied = match write_body_chunks_to_writer(req.into_body().into_data_stream(), &mut file).await {
+        Ok(copied) => copied,
+        Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("write file err {e}")),
+    };
+
+    global_internode_metrics().record_incoming_request();
+    global_internode_metrics().record_recv_bytes(copied as usize);
+
+    if let Err(e) = file.flush().await {
+        return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("write file err {e}"));
+    }
+
+    empty_ok()
+}
+
+async fn write_body_chunks_to_writer<S, E, W>(body: S, writer: &mut W) -> io::Result<u64>
+where
+    S: futures::TryStream<Ok = Bytes, Error = E> + Unpin,
+    E: Into<BoxError>,
+    W: tokio::io::AsyncWrite + Unpin,
+{
+    let mut body = body;
+    let mut copied = 0_u64;
+    let mut pending = BytesMut::with_capacity(DEFAULT_READ_BUFFER_SIZE);
+
+    while let Some(bytes) = body.try_next().await.map_err(io::Error::other)? {
+        copied += bytes.len() as u64;
+        pending.extend_from_slice(&bytes);
+
+        if pending.len() >= DEFAULT_READ_BUFFER_SIZE {
+            writer.write_all(&pending).await?;
+            pending.clear();
+        }
+    }
+
+    if !pending.is_empty() {
+        writer.write_all(&pending).await?;
+    }
+
+    Ok(copied)
+}
+
+fn parse_query<T>(req: &Request<Incoming>) -> Result<T, RpcErrorResponse>
+where
+    T: DeserializeOwned + Default,
+{
+    match req.uri().query() {
+        Some(query) => from_bytes(query.as_bytes())
+            .map_err(|e| Box::new(response_with_status(StatusCode::BAD_REQUEST, format!("get query failed {e}")))),
+        None => Ok(T::default()),
+    }
+}
+
+fn empty_ok() -> Response<Body> {
+    Response::builder()
+        .status(StatusCode::OK)
+        .body(Body::empty())
+        .expect("failed to build empty ok response")
+}
+
+fn response_with_status(status: StatusCode, message: impl Into<String>) -> Response<Body> {
+    Response::builder()
+        .status(status)
+        .header(http::header::CONTENT_TYPE, "text/plain; charset=utf-8")
+        .body(Body::from(Bytes::from(message.into())))
+        .expect("failed to build rpc error response")
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    use tokio_stream::StreamExt;
+    use tokio_stream::iter;
+
+    #[test]
+    fn internode_rpc_path_matches_rpc_prefix() {
+        assert!(is_internode_rpc_path("/rustfs/rpc/read_file_stream"));
+        assert!(is_internode_rpc_path("/rustfs/rpc/walk_dir"));
+        assert!(!is_internode_rpc_path("/rustfs/admin/v3/info"));
+    }
+
+    #[test]
+    fn rpc_head_signature_verification_is_skipped() {
+        let uri: Uri = READ_FILE_STREAM_PATH.parse().expect("uri");
+        let headers = HeaderMap::new();
+        assert!(verify_internode_rpc_signature(&uri, &Method::HEAD, &headers).is_ok());
+    }
+
+    #[test]
+    fn rpc_get_request_requires_signature() {
+        let uri: Uri = READ_FILE_STREAM_PATH.parse().expect("uri");
+        let headers = HeaderMap::new();
+        let response = verify_internode_rpc_signature(&uri, &Method::GET, &headers).expect_err("response");
+        assert_eq!(response.status(), StatusCode::FORBIDDEN);
+    }
+
+    #[tokio::test]
+    async fn write_body_chunks_to_writer_streams_all_chunks() {
+        let (mut reader, mut writer) = tokio::io::duplex(64);
+        let body = iter(vec![
+            Ok::<Bytes, io::Error>(Bytes::from_static(b"hello ")),
+            Ok(Bytes::from_static(b"world")),
+        ]);
+
+        let copied = write_body_chunks_to_writer(body, &mut writer).await.expect("copy succeeds");
+        drop(writer);
+
+        let mut out = Vec::new();
+        reader.read_to_end(&mut out).await.expect("read succeeds");
+
+        assert_eq!(copied, 11);
+        assert_eq!(out, b"hello world");
+    }
+
+    #[tokio::test]
+    async fn read_file_body_stream_keeps_full_stream_when_length_is_zero() {
+        let (reader, mut writer) = tokio::io::duplex(64);
+        tokio::spawn(async move {
+            writer.write_all(b"hello world").await.expect("write succeeds");
+        });
+
+        let mut stream = read_file_body_stream(reader, 0);
+        let mut out = Vec::new();
+        while let Some(chunk) = stream.next().await {
+            out.extend_from_slice(&chunk.expect("chunk succeeds"));
+        }
+
+        assert_eq!(out, b"hello world");
+    }
+
+    #[tokio::test]
+    async fn read_file_body_stream_truncates_to_requested_length() {
+        let (reader, mut writer) = tokio::io::duplex(64);
+        tokio::spawn(async move {
+            writer.write_all(b"hello world").await.expect("write succeeds");
+        });
+
+        let mut stream = read_file_body_stream(reader, 5);
+        let mut out = Vec::new();
+        while let Some(chunk) = stream.next().await {
+            out.extend_from_slice(&chunk.expect("chunk succeeds"));
+        }
+
+        assert_eq!(out, b"hello");
+    }
+}

rustfs/src/storage/rpc/mod.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub mod http_service;
 pub mod node_service;
 
+pub use http_service::InternodeRpcService;
 pub use node_service::{NodeService, make_server};

rustfs/src/storage/rpc/node_service.rs

@@ -21,14 +21,14 @@ use rustfs_ecstore::{
     admin_server_info::get_local_server_property,
     bucket::{metadata::load_bucket_metadata, metadata_sys},
     disk::{
-        DeleteOptions, DiskAPI, DiskInfoOptions, DiskStore, FileInfoVersions, ReadMultipleReq, ReadOptions, UpdateMetadataOpts,
-        error::DiskError,
+        DeleteOptions, DiskAPI, DiskInfoOptions, DiskStore, FileInfoVersions, ReadMultipleReq, ReadMultipleResp, ReadOptions,
+        UpdateMetadataOpts, error::DiskError,
     },
     get_global_lock_client,
     metrics_realtime::{CollectMetricsOpts, MetricType, collect_local_metrics},
     new_object_layer_fn,
     rpc::{LocalPeerS3Client, PeerS3Client},
-    store::{all_local_disk_path, find_local_disk},
+    store::{all_local_disk_path, find_local_disk_by_ref},
     store_api::{BucketOptions, DeleteBucketOptions, MakeBucketOptions, StorageAPI},
 };
 use rustfs_filemeta::{FileInfo, MetacacheReader};
@@ -74,8 +74,8 @@ pub fn make_server() -> NodeService {
 }
 
 impl NodeService {
-    async fn find_disk(&self, disk_path: &String) -> Option<DiskStore> {
-        find_local_disk(disk_path).await
+    async fn find_disk(&self, disk_path: &str) -> Option<DiskStore> {
+        find_local_disk_by_ref(disk_path).await
     }
 
     async fn all_disk(&self) -> Vec<String> {
@@ -1353,6 +1353,8 @@ mod tests {
             path: "test-path".to_string(),
             file_info: "{}".to_string(),
             opts: "{}".to_string(),
+            file_info_bin: Vec::new(),
+            opts_bin: Vec::new(),
         });
 
         let response = service.update_metadata(request).await;
@@ -1373,6 +1375,8 @@ mod tests {
             path: "test-path".to_string(),
             file_info: "invalid json".to_string(),
             opts: "{}".to_string(),
+            file_info_bin: Vec::new(),
+            opts_bin: Vec::new(),
         });
 
         let response = service.update_metadata(request).await;
@@ -1393,6 +1397,8 @@ mod tests {
             path: "test-path".to_string(),
             file_info: "{}".to_string(),
             opts: "invalid json".to_string(),
+            file_info_bin: Vec::new(),
+            opts_bin: Vec::new(),
         });
 
         let response = service.update_metadata(request).await;
@@ -1412,6 +1418,7 @@ mod tests {
             volume: "test-volume".to_string(),
             path: "test-path".to_string(),
             file_info: "{}".to_string(),
+            file_info_bin: Vec::new(),
         });
 
         let response = service.write_metadata(request).await;
@@ -1431,6 +1438,7 @@ mod tests {
             volume: "test-volume".to_string(),
             path: "test-path".to_string(),
             file_info: "invalid json".to_string(),
+            file_info_bin: Vec::new(),
         });
 
         let response = service.write_metadata(request).await;
@@ -1451,6 +1459,7 @@ mod tests {
             path: "test-path".to_string(),
             version_id: "version1".to_string(),
             opts: "{}".to_string(),
+            opts_bin: Vec::new(),
         });
 
         let response = service.read_version(request).await;
@@ -1472,6 +1481,7 @@ mod tests {
             path: "test-path".to_string(),
             version_id: "version1".to_string(),
             opts: "invalid json".to_string(),
+            opts_bin: Vec::new(),
         });
 
         let response = service.read_version(request).await;
@@ -1629,6 +1639,7 @@ mod tests {
         let request = Request::new(ReadMultipleRequest {
             disk: "invalid-disk-path".to_string(),
             read_multiple_req: "{}".to_string(),
+            read_multiple_req_bin: Vec::new(),
         });
 
         let response = service.read_multiple(request).await;
@@ -1647,6 +1658,7 @@ mod tests {
         let request = Request::new(ReadMultipleRequest {
             disk: "invalid-disk-path".to_string(),
             read_multiple_req: "invalid json".to_string(),
+            read_multiple_req_bin: Vec::new(),
         });
 
         let response = service.read_multiple(request).await;
@@ -2195,7 +2207,7 @@ mod tests {
     #[tokio::test]
     async fn test_find_disk_method() {
         let service = create_test_node_service();
-        let disk = service.find_disk(&"non-existent-disk".to_string()).await;
+        let disk = service.find_disk("non-existent-disk").await;
         // Should return None for non-existent disk
         assert!(disk.is_none());
     }