fix: preserve walk-dir internode metrics fallback (#5095)

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

crates/rio/src/http_reader.rs

@@ -1496,6 +1496,7 @@ mod tests {
     use axum::{Router, body::Body, extract::State, http::StatusCode, response::IntoResponse, routing::get};
     use futures::stream::{self, StreamExt as _};
     use http_body_util::BodyExt as _;
+    use rustfs_io_metrics::internode_metrics::global_internode_metrics;
     use std::io::{self, IoSlice};
     use std::sync::{
         Arc,
@@ -1741,6 +1742,7 @@ mod tests {
         let addr = listener.local_addr().expect("listener local address should be available");
         let app = Router::new()
             .route("/stream", get(get_stream).head(reject_head).put(accept_put))
+            .route(WALK_DIR_PATH, get(get_stream))
             .route("/reject-put", get(get_stream).put(reject_put))
             .route("/stall", get(get_stalling_stream))
             .route("/delayed-first", get(get_delayed_first_chunk))
@@ -1801,6 +1803,33 @@ mod tests {
         handle.abort();
     }
 
+    #[tokio::test]
+    async fn http_reader_records_walk_dir_recv_bytes() {
+        let state = TestState::default();
+        let Some((base_url, handle)) = start_test_server(state.clone()).await else {
+            return;
+        };
+        let metrics = global_internode_metrics();
+        let before = metrics.snapshot().recv_bytes_total;
+        let url = base_url.replace("/stream", WALK_DIR_PATH);
+
+        let mut reader = HttpReader::new(url, Method::GET, HeaderMap::new(), None)
+            .await
+            .expect("walk_dir reader should open");
+        let mut buf = Vec::new();
+        reader.read_to_end(&mut buf).await.expect("walk_dir body should read to EOF");
+        let after = metrics.snapshot().recv_bytes_total;
+
+        assert_eq!(buf, b"hello");
+        assert_eq!(state.get_count.load(Ordering::SeqCst), 1);
+        assert!(
+            after >= before.saturating_add(5),
+            "walk_dir HttpReader should record streamed bytes as internode recv bytes: before={before}, after={after}"
+        );
+
+        handle.abort();
+    }
+
     #[tokio::test]
     async fn http_reader_stall_timeout_triggers_after_progress_stops() {
         let state = TestState::default();

rustfs/src/storage/rpc/http_service.rs

@@ -840,6 +840,7 @@ mod tests {
     use http_body_util::BodyExt;
     use rustfs_io_metrics::internode_metrics::{
         INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_OPERATION_WALK_DIR,
+        global_internode_metrics,
     };
     use sha2::Digest as _;
     use tokio::io;
@@ -988,6 +989,30 @@ mod tests {
         assert_eq!(bytes, Bytes::from_static(b"complete walk data"));
     }
 
+    #[tokio::test]
+    async fn walk_dir_body_records_operation_sent_bytes() {
+        let metrics = global_internode_metrics();
+        let before = metrics.snapshot().sent_bytes_total;
+        let payload = Bytes::from_static(b"metered walk data");
+        let expected_len = u64::try_from(payload.len()).expect("test payload length should fit u64");
+        let body = walk_dir_response_body(true, move |mut writer| async move {
+            writer.write_all(&payload).await?;
+            Ok(())
+        });
+
+        let bytes = BodyExt::collect(body)
+            .await
+            .expect("successful completion should preserve the metered body")
+            .to_bytes();
+        let after = metrics.snapshot().sent_bytes_total;
+
+        assert_eq!(bytes, Bytes::from_static(b"metered walk data"));
+        assert!(
+            after >= before.saturating_add(expected_len),
+            "walk_dir response body should record streamed bytes as internode sent bytes: before={before}, after={after}, expected_delta={expected_len}"
+        );
+    }
+
     #[tokio::test]
     async fn walk_dir_completion_stream_surfaces_cancelled_producer() {
         let (completion_tx, completion_rx) = tokio::sync::oneshot::channel();

rustfs/src/storage/runtime_sources.rs

@@ -17,7 +17,10 @@ use crate::runtime_sources as root_runtime_sources;
 use crate::storage::storage_api::runtime_sources_consumer::ECStore;
 use rustfs_credentials::Credentials;
 use rustfs_iam::{error::Result as IamResult, store::object::ObjectStore, sys::IamSys};
-use rustfs_io_metrics::{PerformanceMetrics, internode_metrics::InternodeMetrics};
+use rustfs_io_metrics::{
+    PerformanceMetrics,
+    internode_metrics::{InternodeMetrics, global_internode_metrics},
+};
 use rustfs_kms::ObjectEncryptionService;
 use rustfs_lock::LockClient;
 use std::sync::Arc;
@@ -41,7 +44,7 @@ pub(crate) fn current_buffer_config() -> RustFSBufferConfig {
 }
 
 pub(crate) fn current_internode_metrics() -> Arc<InternodeMetrics> {
-    root_runtime_sources::current_internode_metrics().unwrap_or_else(|| Arc::new(InternodeMetrics::default()))
+    root_runtime_sources::current_internode_metrics().unwrap_or_else(|| global_internode_metrics().clone())
 }
 
 pub(crate) async fn current_local_node_name() -> String {