refactor: update binary field types and conversions in RPC and protofiles (#2619)

Signed-off-by: 唐小鸭 <tangtang1251@qq.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: loverustfs <hello@rustfs.com>

crates/ecstore/src/rpc/remote_disk.rs

@@ -886,7 +886,7 @@ impl DiskAPI for RemoteDisk {
 
         self.execute_with_timeout_for_op(
             "write_metadata",
-            || async {
+            move || async move {
                 let disk = self.disk_ref().await;
                 let mut client = self
                     .get_client()
@@ -896,8 +896,8 @@ impl DiskAPI for RemoteDisk {
                     disk,
                     volume: volume.to_string(),
                     path: path.to_string(),
-                    file_info: file_info.clone(),
+                    file_info,
-                    file_info_bin: file_info_bin.clone(),
+                    file_info_bin: file_info_bin.into(),
                 });
 
                 let response = client.write_metadata(request).await?.into_inner();
@@ -951,7 +951,7 @@ impl DiskAPI for RemoteDisk {
 
         self.execute_with_timeout_for_op(
             "update_metadata",
-            || async {
+            move || async move {
                 let disk = self.disk_ref().await;
                 let mut client = self
                     .get_client()
@@ -961,10 +961,10 @@ impl DiskAPI for RemoteDisk {
                     disk,
                     volume: volume.to_string(),
                     path: path.to_string(),
-                    file_info: file_info.clone(),
+                    file_info,
-                    opts: opts_str.clone(),
+                    opts: opts_str,
-                    file_info_bin: file_info_bin.clone(),
+                    file_info_bin: file_info_bin.into(),
-                    opts_bin: opts_bin.clone(),
+                    opts_bin: opts_bin.into(),
                 });
 
                 let response = client.update_metadata(request).await?.into_inner();
@@ -994,7 +994,7 @@ impl DiskAPI for RemoteDisk {
         let opts_bin = encode_msgpack(opts)?;
 
         self.execute_with_timeout(
-            || async {
+            move || async {
                 let disk = self.disk_ref().await;
                 let mut client = self
                     .get_client()
@@ -1005,8 +1005,8 @@ impl DiskAPI for RemoteDisk {
                     volume: volume.to_string(),
                     path: path.to_string(),
                     version_id: version_id.to_string(),
-                    opts: opts_str.clone(),
+                    opts: opts_str,
-                    opts_bin: opts_bin.clone(),
+                    opts_bin: opts_bin.into(),
                 });
 
                 let response = client.read_version(request).await?.into_inner();
@@ -1480,7 +1480,7 @@ impl DiskAPI for RemoteDisk {
                 let request = Request::new(ReadMultipleRequest {
                     disk,
                     read_multiple_req,
-                    read_multiple_req_bin,
+                    read_multiple_req_bin: read_multiple_req_bin.into(),
                 });
 
                 let response = client.read_multiple(request).await?.into_inner();

crates/ecstore/tests/protobuf_bytes_regression_test.rs

@@ -0,0 +1,36 @@
+// Copyright (c) RustFS contributors
+// SPDX-License-Identifier: Apache-2.0
+
+use bytes::Bytes;
+use rustfs_protos::proto_gen::node_service::{
+    ReadMultipleRequest, ReadMultipleResponse, ReadVersionResponse, ReadXlResponse, UpdateMetadataRequest, WriteMetadataRequest,
+};
+
+fn expect_bytes(_: &Bytes) {}
+
+#[test]
+fn protobuf_bytes_fields_use_bytes_consistently() {
+    let update = UpdateMetadataRequest::default();
+    expect_bytes(&update.file_info_bin);
+    expect_bytes(&update.opts_bin);
+
+    let write = WriteMetadataRequest::default();
+    expect_bytes(&write.file_info_bin);
+
+    let version = ReadVersionResponse::default();
+    expect_bytes(&version.file_info_bin);
+
+    let read_xl = ReadXlResponse::default();
+    expect_bytes(&read_xl.raw_file_info_bin);
+
+    let read_multiple = ReadMultipleRequest::default();
+    expect_bytes(&read_multiple.read_multiple_req_bin);
+
+    let read_multiple_response = ReadMultipleResponse::default();
+    let first = read_multiple_response
+        .read_multiple_resps_bin
+        .first()
+        .cloned()
+        .unwrap_or_default();
+    expect_bytes(&first);
+}

crates/protos/src/generated/flatbuffers_generated/models.rs

@@ -1,55 +1,46 @@
 // automatically generated by the FlatBuffers compiler, do not modify
-
 // @generated
 
-use core::cmp::Ordering;
+extern crate alloc;
-use core::mem;
-
-extern crate flatbuffers;
-use self::flatbuffers::{EndianScalar, Follow};
 
 #[allow(unused_imports, dead_code)]
 pub mod models {
 
-    use core::cmp::Ordering;
+    extern crate alloc;
-    use core::mem;
-
-    extern crate flatbuffers;
-    use self::flatbuffers::{EndianScalar, Follow};
 
     pub enum PingBodyOffset {}
     #[derive(Copy, Clone, PartialEq)]
 
     pub struct PingBody<'a> {
-        pub _tab: flatbuffers::Table<'a>,
+        pub _tab: ::flatbuffers::Table<'a>,
     }
 
-    impl<'a> flatbuffers::Follow<'a> for PingBody<'a> {
+    impl<'a> ::flatbuffers::Follow<'a> for PingBody<'a> {
         type Inner = PingBody<'a>;
         #[inline]
         unsafe fn follow(buf: &'a [u8], loc: usize) -> Self::Inner {
             Self {
-                _tab: unsafe { flatbuffers::Table::new(buf, loc) },
+                _tab: unsafe { ::flatbuffers::Table::new(buf, loc) },
             }
         }
     }
 
     impl<'a> PingBody<'a> {
-        pub const VT_PAYLOAD: flatbuffers::VOffsetT = 4;
+        pub const VT_PAYLOAD: ::flatbuffers::VOffsetT = 4;
 
         pub const fn get_fully_qualified_name() -> &'static str {
             "models.PingBody"
         }
 
         #[inline]
-        pub unsafe fn init_from_table(table: flatbuffers::Table<'a>) -> Self {
+        pub unsafe fn init_from_table(table: ::flatbuffers::Table<'a>) -> Self {
             PingBody { _tab: table }
         }
         #[allow(unused_mut)]
-        pub fn create<'bldr: 'args, 'args: 'mut_bldr, 'mut_bldr, A: flatbuffers::Allocator + 'bldr>(
+        pub fn create<'bldr: 'args, 'args: 'mut_bldr, 'mut_bldr, A: ::flatbuffers::Allocator + 'bldr>(
-            _fbb: &'mut_bldr mut flatbuffers::FlatBufferBuilder<'bldr, A>,
+            _fbb: &'mut_bldr mut ::flatbuffers::FlatBufferBuilder<'bldr, A>,
             args: &'args PingBodyArgs<'args>,
-        ) -> flatbuffers::WIPOffset<PingBody<'bldr>> {
+        ) -> ::flatbuffers::WIPOffset<PingBody<'bldr>> {
             let mut builder = PingBodyBuilder::new(_fbb);
             if let Some(x) = args.payload {
                 builder.add_payload(x);
@@ -58,29 +49,28 @@ pub mod models {
         }
 
         #[inline]
-        pub fn payload(&self) -> Option<flatbuffers::Vector<'a, u8>> {
+        pub fn payload(&self) -> Option<::flatbuffers::Vector<'a, u8>> {
             // Safety:
             // Created from valid Table for this object
             // which contains a valid value in this slot
             unsafe {
                 self._tab
-                    .get::<flatbuffers::ForwardsUOffset<flatbuffers::Vector<'a, u8>>>(PingBody::VT_PAYLOAD, None)
+                    .get::<::flatbuffers::ForwardsUOffset<::flatbuffers::Vector<'a, u8>>>(PingBody::VT_PAYLOAD, None)
             }
         }
     }
 
-    impl flatbuffers::Verifiable for PingBody<'_> {
+    impl ::flatbuffers::Verifiable for PingBody<'_> {
         #[inline]
-        fn run_verifier(v: &mut flatbuffers::Verifier, pos: usize) -> Result<(), flatbuffers::InvalidFlatbuffer> {
+        fn run_verifier(v: &mut ::flatbuffers::Verifier, pos: usize) -> Result<(), ::flatbuffers::InvalidFlatbuffer> {
-            use self::flatbuffers::Verifiable;
             v.visit_table(pos)?
-                .visit_field::<flatbuffers::ForwardsUOffset<flatbuffers::Vector<'_, u8>>>("payload", Self::VT_PAYLOAD, false)?
+                .visit_field::<::flatbuffers::ForwardsUOffset<::flatbuffers::Vector<'_, u8>>>("payload", Self::VT_PAYLOAD, false)?
                 .finish();
             Ok(())
         }
     }
     pub struct PingBodyArgs<'a> {
-        pub payload: Option<flatbuffers::WIPOffset<flatbuffers::Vector<'a, u8>>>,
+        pub payload: Option<::flatbuffers::WIPOffset<::flatbuffers::Vector<'a, u8>>>,
     }
     impl<'a> Default for PingBodyArgs<'a> {
         #[inline]
@@ -89,18 +79,18 @@ pub mod models {
         }
     }
 
-    pub struct PingBodyBuilder<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> {
+    pub struct PingBodyBuilder<'a: 'b, 'b, A: ::flatbuffers::Allocator + 'a> {
-        fbb_: &'b mut flatbuffers::FlatBufferBuilder<'a, A>,
+        fbb_: &'b mut ::flatbuffers::FlatBufferBuilder<'a, A>,
-        start_: flatbuffers::WIPOffset<flatbuffers::TableUnfinishedWIPOffset>,
+        start_: ::flatbuffers::WIPOffset<::flatbuffers::TableUnfinishedWIPOffset>,
     }
-    impl<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> PingBodyBuilder<'a, 'b, A> {
+    impl<'a: 'b, 'b, A: ::flatbuffers::Allocator + 'a> PingBodyBuilder<'a, 'b, A> {
         #[inline]
-        pub fn add_payload(&mut self, payload: flatbuffers::WIPOffset<flatbuffers::Vector<'b, u8>>) {
+        pub fn add_payload(&mut self, payload: ::flatbuffers::WIPOffset<::flatbuffers::Vector<'b, u8>>) {
             self.fbb_
-                .push_slot_always::<flatbuffers::WIPOffset<_>>(PingBody::VT_PAYLOAD, payload);
+                .push_slot_always::<::flatbuffers::WIPOffset<_>>(PingBody::VT_PAYLOAD, payload);
         }
         #[inline]
-        pub fn new(_fbb: &'b mut flatbuffers::FlatBufferBuilder<'a, A>) -> PingBodyBuilder<'a, 'b, A> {
+        pub fn new(_fbb: &'b mut ::flatbuffers::FlatBufferBuilder<'a, A>) -> PingBodyBuilder<'a, 'b, A> {
             let start = _fbb.start_table();
             PingBodyBuilder {
                 fbb_: _fbb,
@@ -108,14 +98,14 @@ pub mod models {
             }
         }
         #[inline]
-        pub fn finish(self) -> flatbuffers::WIPOffset<PingBody<'a>> {
+        pub fn finish(self) -> ::flatbuffers::WIPOffset<PingBody<'a>> {
             let o = self.fbb_.end_table(self.start_);
-            flatbuffers::WIPOffset::new(o.value())
+            ::flatbuffers::WIPOffset::new(o.value())
         }
     }
 
-    impl core::fmt::Debug for PingBody<'_> {
+    impl ::core::fmt::Debug for PingBody<'_> {
-        fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        fn fmt(&self, f: &mut ::core::fmt::Formatter<'_>) -> ::core::fmt::Result {
             let mut ds = f.debug_struct("PingBody");
             ds.field("payload", &self.payload());
             ds.finish()

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

@@ -467,10 +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")]
+    #[prost(bytes = "bytes", tag = "6")]
-    pub file_info_bin: ::prost::alloc::vec::Vec<u8>,
+    pub file_info_bin: ::prost::bytes::Bytes,
-    #[prost(bytes = "vec", tag = "7")]
+    #[prost(bytes = "bytes", tag = "7")]
-    pub opts_bin: ::prost::alloc::vec::Vec<u8>,
+    pub opts_bin: ::prost::bytes::Bytes,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct UpdateMetadataResponse {
@@ -490,8 +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")]
+    #[prost(bytes = "bytes", tag = "5")]
-    pub file_info_bin: ::prost::alloc::vec::Vec<u8>,
+    pub file_info_bin: ::prost::bytes::Bytes,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct WriteMetadataResponse {
@@ -512,8 +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")]
+    #[prost(bytes = "bytes", tag = "6")]
-    pub opts_bin: ::prost::alloc::vec::Vec<u8>,
+    pub opts_bin: ::prost::bytes::Bytes,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ReadVersionResponse {
@@ -523,8 +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")]
+    #[prost(bytes = "bytes", tag = "4")]
-    pub file_info_bin: ::prost::alloc::vec::Vec<u8>,
+    pub file_info_bin: ::prost::bytes::Bytes,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ReadXlRequest {
@@ -545,8 +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")]
+    #[prost(bytes = "bytes", tag = "4")]
-    pub raw_file_info_bin: ::prost::alloc::vec::Vec<u8>,
+    pub raw_file_info_bin: ::prost::bytes::Bytes,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct DeleteVersionRequest {
@@ -598,8 +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")]
+    #[prost(bytes = "bytes", tag = "3")]
-    pub read_multiple_req_bin: ::prost::alloc::vec::Vec<u8>,
+    pub read_multiple_req_bin: ::prost::bytes::Bytes,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ReadMultipleResponse {
@@ -609,8 +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")]
+    #[prost(bytes = "bytes", repeated, tag = "4")]
-    pub read_multiple_resps_bin: ::prost::alloc::vec::Vec<::prost::alloc::vec::Vec<u8>>,
+    pub read_multiple_resps_bin: ::prost::alloc::vec::Vec<::prost::bytes::Bytes>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct DeleteVolumeRequest {
@@ -673,7 +673,7 @@ pub struct GenerallyLockResult {
     #[prost(string, optional, tag = "3")]
     pub lock_info: ::core::option::Option<::prost::alloc::string::String>,
 }
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+#[derive(Clone, PartialEq, ::prost::Message)]
 pub struct BatchGenerallyLockResponse {
     #[prost(message, repeated, tag = "1")]
     pub results: ::prost::alloc::vec::Vec<GenerallyLockResult>,
@@ -816,26 +816,6 @@ pub struct GetMetricsResponse {
     pub error_info: ::core::option::Option<::prost::alloc::string::String>,
 }
 #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct GetLiveEventsRequest {
-    #[prost(uint64, tag = "1")]
-    pub after_sequence: u64,
-    #[prost(uint32, tag = "2")]
-    pub limit: u32,
-}
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct GetLiveEventsResponse {
-    #[prost(bool, tag = "1")]
-    pub success: bool,
-    #[prost(bytes = "bytes", tag = "2")]
-    pub events: ::prost::bytes::Bytes,
-    #[prost(uint64, tag = "3")]
-    pub next_sequence: u64,
-    #[prost(bool, tag = "4")]
-    pub truncated: bool,
-    #[prost(string, optional, tag = "5")]
-    pub error_info: ::core::option::Option<::prost::alloc::string::String>,
-}
-#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct GetProcInfoRequest {}
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct GetProcInfoResponse {
@@ -1130,6 +1110,26 @@ pub struct LoadTransitionTierConfigResponse {
     #[prost(string, optional, tag = "2")]
     pub error_info: ::core::option::Option<::prost::alloc::string::String>,
 }
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct GetLiveEventsRequest {
+    #[prost(uint64, tag = "1")]
+    pub after_sequence: u64,
+    #[prost(uint32, tag = "2")]
+    pub limit: u32,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct GetLiveEventsResponse {
+    #[prost(bool, tag = "1")]
+    pub success: bool,
+    #[prost(bytes = "bytes", tag = "2")]
+    pub events: ::prost::bytes::Bytes,
+    #[prost(uint64, tag = "3")]
+    pub next_sequence: u64,
+    #[prost(bool, tag = "4")]
+    pub truncated: bool,
+    #[prost(string, optional, tag = "5")]
+    pub error_info: ::core::option::Option<::prost::alloc::string::String>,
+}
 /// Generated client implementations.
 pub mod node_service_client {
     #![allow(unused_variables, dead_code, missing_docs, clippy::wildcard_imports, clippy::let_unit_value)]
@@ -1991,21 +1991,6 @@ pub mod node_service_client {
                 .insert(GrpcMethod::new("node_service.NodeService", "GetMetrics"));
             self.inner.unary(req, path, codec).await
         }
-        pub async fn get_live_events(
-            &mut self,
-            request: impl tonic::IntoRequest<super::GetLiveEventsRequest>,
-        ) -> std::result::Result<tonic::Response<super::GetLiveEventsResponse>, tonic::Status> {
-            self.inner
-                .ready()
-                .await
-                .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
-            let codec = tonic_prost::ProstCodec::default();
-            let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/GetLiveEvents");
-            let mut req = request.into_request();
-            req.extensions_mut()
-                .insert(GrpcMethod::new("node_service.NodeService", "GetLiveEvents"));
-            self.inner.unary(req, path, codec).await
-        }
         pub async fn get_proc_info(
             &mut self,
             request: impl tonic::IntoRequest<super::GetProcInfoRequest>,
@@ -2383,6 +2368,21 @@ pub mod node_service_client {
                 .insert(GrpcMethod::new("node_service.NodeService", "LoadTransitionTierConfig"));
             self.inner.unary(req, path, codec).await
         }
+        pub async fn get_live_events(
+            &mut self,
+            request: impl tonic::IntoRequest<super::GetLiveEventsRequest>,
+        ) -> std::result::Result<tonic::Response<super::GetLiveEventsResponse>, tonic::Status> {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/GetLiveEvents");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("node_service.NodeService", "GetLiveEvents"));
+            self.inner.unary(req, path, codec).await
+        }
     }
 }
 /// Generated server implementations.
@@ -2614,10 +2614,6 @@ pub mod node_service_server {
             &self,
             request: tonic::Request<super::GetMetricsRequest>,
         ) -> std::result::Result<tonic::Response<super::GetMetricsResponse>, tonic::Status>;
-        async fn get_live_events(
-            &self,
-            request: tonic::Request<super::GetLiveEventsRequest>,
-        ) -> std::result::Result<tonic::Response<super::GetLiveEventsResponse>, tonic::Status>;
         async fn get_proc_info(
             &self,
             request: tonic::Request<super::GetProcInfoRequest>,
@@ -2720,6 +2716,10 @@ pub mod node_service_server {
             &self,
             request: tonic::Request<super::LoadTransitionTierConfigRequest>,
         ) -> std::result::Result<tonic::Response<super::LoadTransitionTierConfigResponse>, tonic::Status>;
+        async fn get_live_events(
+            &self,
+            request: tonic::Request<super::GetLiveEventsRequest>,
+        ) -> std::result::Result<tonic::Response<super::GetLiveEventsResponse>, tonic::Status>;
     }
     #[derive(Debug)]
     pub struct NodeServiceServer<T> {
@@ -4250,34 +4250,6 @@ pub mod node_service_server {
                     };
                     Box::pin(fut)
                 }
-                "/node_service.NodeService/GetLiveEvents" => {
-                    #[allow(non_camel_case_types)]
-                    struct GetLiveEventsSvc<T: NodeService>(pub Arc<T>);
-                    impl<T: NodeService> tonic::server::UnaryService<super::GetLiveEventsRequest> for GetLiveEventsSvc<T> {
-                        type Response = super::GetLiveEventsResponse;
-                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
-                        fn call(&mut self, request: tonic::Request<super::GetLiveEventsRequest>) -> Self::Future {
-                            let inner = Arc::clone(&self.0);
-                            let fut = async move { <T as NodeService>::get_live_events(&inner, request).await };
-                            Box::pin(fut)
-                        }
-                    }
-                    let accept_compression_encodings = self.accept_compression_encodings;
-                    let send_compression_encodings = self.send_compression_encodings;
-                    let max_decoding_message_size = self.max_decoding_message_size;
-                    let max_encoding_message_size = self.max_encoding_message_size;
-                    let inner = self.inner.clone();
-                    let fut = async move {
-                        let method = GetLiveEventsSvc(inner);
-                        let codec = tonic_prost::ProstCodec::default();
-                        let mut grpc = tonic::server::Grpc::new(codec)
-                            .apply_compression_config(accept_compression_encodings, send_compression_encodings)
-                            .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
-                        let res = grpc.unary(method, req).await;
-                        Ok(res)
-                    };
-                    Box::pin(fut)
-                }
                 "/node_service.NodeService/GetProcInfo" => {
                     #[allow(non_camel_case_types)]
                     struct GetProcInfoSvc<T: NodeService>(pub Arc<T>);
@@ -4980,6 +4952,34 @@ pub mod node_service_server {
                     };
                     Box::pin(fut)
                 }
+                "/node_service.NodeService/GetLiveEvents" => {
+                    #[allow(non_camel_case_types)]
+                    struct GetLiveEventsSvc<T: NodeService>(pub Arc<T>);
+                    impl<T: NodeService> tonic::server::UnaryService<super::GetLiveEventsRequest> for GetLiveEventsSvc<T> {
+                        type Response = super::GetLiveEventsResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(&mut self, request: tonic::Request<super::GetLiveEventsRequest>) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move { <T as NodeService>::get_live_events(&inner, request).await };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = GetLiveEventsSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(accept_compression_encodings, send_compression_encodings)
+                            .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
                 _ => Box::pin(async move {
                     let mut response = http::Response::new(tonic::body::Body::default());
                     let headers = response.headers_mut();

rustfs/src/server/http.rs

@@ -22,8 +22,8 @@ use crate::server::{
     compress::{CompressionConfig, PathAwareCompressionPredicate, PathCategoryInjectionLayer},
     hybrid::hybrid,
     layer::{
-        AdminChunkedContentLengthCompatLayer, ConditionalCorsLayer, ObjectAttributesEtagFixLayer, RedirectLayer,
+        AdminChunkedContentLengthCompatLayer, BodylessStatusFixLayer, ConditionalCorsLayer, ObjectAttributesEtagFixLayer,
-        RequestContextLayer, S3ErrorMessageCompatLayer,
+        RedirectLayer, RequestContextLayer, S3ErrorMessageCompatLayer,
     },
     tls_material::{TlsAcceptorHolder, TlsHandshakeFailureKind, TlsMaterialSnapshot, spawn_reload_loop},
 };
@@ -593,6 +593,7 @@ fn process_connection(
         // 15. ObjectAttributesEtagFixLayer           — ETag fix for GetObjectAttributes
         // 16. ConditionalCorsLayer                   — S3 API CORS
         // 17. RedirectLayer                          — console redirect (conditional)
+        // 18. BodylessStatusFixLayer                 — clears body for 1xx/204/205/304 responses
         // ─────────────────────────────────────────────────────────────
         let hybrid_service = ServiceBuilder::new()
             // NOTE: Both extension types are intentionally inserted to maintain compatibility:
@@ -735,6 +736,12 @@ fn process_connection(
             // Bucket-level CORS takes precedence when configured (handled in router.rs for OPTIONS, and in ecfs.rs for actual requests)
             .layer(ConditionalCorsLayer::new())
             .option_layer(if is_console { Some(RedirectLayer) } else { None })
+            // Must run first on responses: clear the body and remove
+            // Content-Length, Content-Type, and Transfer-Encoding for statuses
+            // that MUST NOT carry a body (1xx/204/304). Kept innermost so all
+            // other response-transforming layers see the already-bodyless
+            // response and so no layer (e.g. CORS) re-adds body headers afterward.
+            .layer(BodylessStatusFixLayer)
             .service(service);
 
         let hybrid_service = TowerToHyperService::new(hybrid_service);

rustfs/src/server/layer.rs

@@ -399,6 +399,91 @@ where
     }
 }
 
+/// Tower middleware that strips the body (and body-describing headers) from
+/// responses whose HTTP status code MUST NOT carry a body per RFC 9110 §6.4.1
+/// and §15 (1xx, 204, 205, 304).
+///
+/// The inner s3s layer serializes every `S3Error` — including 304 `NotModified`
+/// preconditions — as an XML body. Returning that body for a 304 is a protocol
+/// violation: hyper's HTTP/1.1 encoder forces the body to zero length but
+/// preserves the response, while the HTTP/2 path fills in `content-length`
+/// from the body's size hint and writes DATA frames after a HEADERS frame that
+/// should have carried END_STREAM. h2 clients (curl, browsers) and proxies see
+/// the malformed response as a connection-level failure — in the wild this
+/// surfaces as `GOAWAY error=0` on h2 and as an upstream-disconnect 5xx from
+/// reverse proxies like ngrok (`ERR_NGROK_3004`).
+#[derive(Clone)]
+pub struct BodylessStatusFixLayer;
+
+impl<S> Layer<S> for BodylessStatusFixLayer {
+    type Service = BodylessStatusFixService<S>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        BodylessStatusFixService { inner }
+    }
+}
+
+#[derive(Clone)]
+pub struct BodylessStatusFixService<S> {
+    inner: S,
+}
+
+impl<S, ReqBody, RestBody, GrpcBody> Service<HttpRequest<ReqBody>> for BodylessStatusFixService<S>
+where
+    S: Service<HttpRequest<ReqBody>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
+    S::Future: Send + 'static,
+    S::Error: Send + 'static,
+    ReqBody: Send + 'static,
+    RestBody: Body<Data = Bytes> + From<Bytes> + Send + 'static,
+    RestBody::Error: Into<S::Error> + Send + 'static,
+    GrpcBody: Send + 'static,
+{
+    type Response = Response<HybridBody<RestBody, GrpcBody>>;
+    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: HttpRequest<ReqBody>) -> Self::Future {
+        let mut inner = self.inner.clone();
+
+        Box::pin(async move {
+            let response = inner.call(req).await?;
+            let (mut parts, body) = response.into_parts();
+
+            if !is_bodyless_status(parts.status) {
+                return Ok(Response::from_parts(parts, body));
+            }
+
+            let response = match body {
+                HybridBody::Rest { .. } => {
+                    parts.headers.remove(http::header::CONTENT_LENGTH);
+                    parts.headers.remove(http::header::CONTENT_TYPE);
+                    parts.headers.remove(http::header::TRANSFER_ENCODING);
+                    Response::from_parts(
+                        parts,
+                        HybridBody::Rest {
+                            rest_body: RestBody::from(Bytes::new()),
+                        },
+                    )
+                }
+                HybridBody::Grpc { grpc_body } => Response::from_parts(parts, HybridBody::Grpc { grpc_body }),
+            };
+
+            Ok(response)
+        })
+    }
+}
+
+fn is_bodyless_status(status: StatusCode) -> bool {
+    status.is_informational()
+        || status == StatusCode::NO_CONTENT
+        || status == StatusCode::RESET_CONTENT
+        || status == StatusCode::NOT_MODIFIED
+}
+
 fn is_xml_response(headers: &HeaderMap) -> bool {
     let is_xml = headers
         .get(http::header::CONTENT_TYPE)
@@ -1052,6 +1137,138 @@ mod tests {
         assert!(response_headers.get(cors::response::ACCESS_CONTROL_MAX_AGE).is_none());
     }
 
+    mod bodyless_status_fix {
+        use super::*;
+        use crate::server::hybrid::HybridBody;
+        use http_body_util::Empty;
+
+        // The production service takes `Request<Incoming>`, but `Incoming` can't be
+        // constructed in unit tests. `BodylessStatusFixService` doesn't inspect the
+        // request body, so parameterising over an arbitrary `B` is safe here.
+        #[derive(Clone)]
+        struct FixedResponse {
+            status: StatusCode,
+            body: Bytes,
+            content_type: Option<&'static str>,
+        }
+
+        impl<B: Send + 'static> Service<Request<B>> for FixedResponse {
+            type Response = Response<HybridBody<Full<Bytes>, Empty<Bytes>>>;
+            type Error = Infallible;
+            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>> {
+                Poll::Ready(Ok(()))
+            }
+
+            fn call(&mut self, _req: Request<B>) -> Self::Future {
+                let this = self.clone();
+                Box::pin(async move {
+                    let body = this.body.clone();
+                    let len = body.len();
+                    let mut builder = Response::builder().status(this.status);
+                    builder = builder.header(http::header::CONTENT_LENGTH, len.to_string());
+                    if let Some(ct) = this.content_type {
+                        builder = builder.header(http::header::CONTENT_TYPE, ct);
+                    }
+                    builder = builder.header(http::header::ETAG, "\"abc123\"");
+                    Ok(builder
+                        .body(HybridBody::Rest {
+                            rest_body: Full::from(body),
+                        })
+                        .expect("build response"))
+                })
+            }
+        }
+
+        fn empty_request() -> Request<()> {
+            Request::builder().uri("/").body(()).expect("request")
+        }
+
+        async fn collect_body<B: Body<Data = Bytes>>(body: B) -> Bytes
+        where
+            B::Error: std::fmt::Debug,
+        {
+            BodyExt::collect(body).await.expect("collect body").to_bytes()
+        }
+
+        #[tokio::test]
+        async fn strips_body_and_content_headers_for_304() {
+            let mut svc = BodylessStatusFixLayer.layer(FixedResponse {
+                status: StatusCode::NOT_MODIFIED,
+                body: Bytes::from_static(b"<Error><Code>NotModified</Code></Error>"),
+                content_type: Some("application/xml"),
+            });
+
+            let res = svc.call(empty_request()).await.expect("service call");
+            let (parts, body) = res.into_parts();
+
+            assert_eq!(parts.status, StatusCode::NOT_MODIFIED);
+            assert!(parts.headers.get(http::header::CONTENT_LENGTH).is_none());
+            assert!(parts.headers.get(http::header::CONTENT_TYPE).is_none());
+            assert_eq!(parts.headers.get(http::header::ETAG).unwrap(), "\"abc123\"");
+
+            let bytes = collect_body(body).await;
+            assert!(bytes.is_empty(), "304 response body must be empty");
+        }
+
+        #[tokio::test]
+        async fn strips_body_for_204() {
+            let mut svc = BodylessStatusFixLayer.layer(FixedResponse {
+                status: StatusCode::NO_CONTENT,
+                body: Bytes::from_static(b"unexpected"),
+                content_type: None,
+            });
+
+            let res = svc.call(empty_request()).await.expect("service call");
+            let (parts, body) = res.into_parts();
+
+            assert_eq!(parts.status, StatusCode::NO_CONTENT);
+            assert!(parts.headers.get(http::header::CONTENT_LENGTH).is_none());
+
+            let bytes = collect_body(body).await;
+            assert!(bytes.is_empty());
+        }
+
+        #[tokio::test]
+        async fn preserves_body_for_200() {
+            let payload = Bytes::from_static(b"hello");
+            let mut svc = BodylessStatusFixLayer.layer(FixedResponse {
+                status: StatusCode::OK,
+                body: payload.clone(),
+                content_type: Some("text/plain"),
+            });
+
+            let res = svc.call(empty_request()).await.expect("service call");
+            let (parts, body) = res.into_parts();
+
+            assert_eq!(parts.status, StatusCode::OK);
+            assert_eq!(parts.headers.get(http::header::CONTENT_TYPE).unwrap(), "text/plain");
+            assert_eq!(
+                parts.headers.get(http::header::CONTENT_LENGTH).unwrap(),
+                payload.len().to_string().as_str()
+            );
+
+            let bytes = collect_body(body).await;
+            assert_eq!(bytes, payload);
+        }
+
+        #[test]
+        fn is_bodyless_status_matches_rfc9110_statuses() {
+            assert!(is_bodyless_status(StatusCode::CONTINUE));
+            assert!(is_bodyless_status(StatusCode::SWITCHING_PROTOCOLS));
+            assert!(is_bodyless_status(StatusCode::NO_CONTENT));
+            assert!(is_bodyless_status(StatusCode::RESET_CONTENT));
+            assert!(is_bodyless_status(StatusCode::NOT_MODIFIED));
+
+            assert!(!is_bodyless_status(StatusCode::OK));
+            assert!(!is_bodyless_status(StatusCode::PARTIAL_CONTENT));
+            assert!(!is_bodyless_status(StatusCode::NOT_FOUND));
+            assert!(!is_bodyless_status(StatusCode::PRECONDITION_FAILED));
+            assert!(!is_bodyless_status(StatusCode::INTERNAL_SERVER_ERROR));
+        }
+    }
+
     #[test]
     fn test_apply_bucket_cors_result_replaces_existing_cors_headers() {
         let mut response_headers = HeaderMap::new();

rustfs/src/storage/rpc/disk.rs

@@ -131,6 +131,7 @@ impl NodeService {
                         .iter()
                         .filter_map(|json_str| serde_json::from_str::<ReadMultipleResp>(json_str).ok())
                         .filter_map(|resp| encode_msgpack(&resp, "ReadMultipleResp").ok())
+                        .map(Into::into)
                         .collect();
 
                     Ok(Response::new(ReadMultipleResponse {
@@ -279,19 +280,19 @@ impl NodeService {
                         (Ok(raw_file_info), Ok(raw_file_info_bin)) => Ok(Response::new(ReadXlResponse {
                             success: true,
                             raw_file_info,
-                            raw_file_info_bin,
+                            raw_file_info_bin: raw_file_info_bin.into(),
                             error: None,
                         })),
                         (Err(err), _) => Ok(Response::new(ReadXlResponse {
                             success: false,
                             raw_file_info: String::new(),
-                            raw_file_info_bin: Vec::new(),
+                            raw_file_info_bin: Vec::new().into(),
                             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(),
+                            raw_file_info_bin: Vec::new().into(),
                             error: Some(DiskError::other(format!("encode data failed: {err}")).into()),
                         })),
                     }
@@ -299,7 +300,7 @@ impl NodeService {
                 Err(err) => Ok(Response::new(ReadXlResponse {
                     success: false,
                     raw_file_info: String::new(),
-                    raw_file_info_bin: Vec::new(),
+                    raw_file_info_bin: Vec::new().into(),
                     error: Some(err.into()),
                 })),
             }
@@ -307,7 +308,7 @@ impl NodeService {
             Ok(Response::new(ReadXlResponse {
                 success: false,
                 raw_file_info: String::new(),
-                raw_file_info_bin: Vec::new(),
+                raw_file_info_bin: Vec::new().into(),
                 error: Some(DiskError::other("can not find disk".to_string()).into()),
             }))
         }
@@ -325,7 +326,7 @@ impl NodeService {
                     return Ok(Response::new(ReadVersionResponse {
                         success: false,
                         file_info: String::new(),
-                        file_info_bin: Vec::new(),
+                        file_info_bin: Vec::new().into(),
                         error: Some(DiskError::other(format!("decode ReadOptions failed: {err}")).into()),
                     }));
                 }
@@ -341,19 +342,19 @@ impl NodeService {
                         (Ok(file_info), Ok(file_info_bin)) => Ok(Response::new(ReadVersionResponse {
                             success: true,
                             file_info,
-                            file_info_bin,
+                            file_info_bin: file_info_bin.into(),
                             error: None,
                         })),
                         (Err(err), _) => Ok(Response::new(ReadVersionResponse {
                             success: false,
                             file_info: String::new(),
-                            file_info_bin: Vec::new(),
+                            file_info_bin: Vec::new().into(),
                             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(),
+                            file_info_bin: Vec::new().into(),
                             error: Some(DiskError::other(format!("encode data failed: {err}")).into()),
                         })),
                     }
@@ -361,7 +362,7 @@ impl NodeService {
                 Err(err) => Ok(Response::new(ReadVersionResponse {
                     success: false,
                     file_info: String::new(),
-                    file_info_bin: Vec::new(),
+                    file_info_bin: Vec::new().into(),
                     error: Some(err.into()),
                 })),
             }
@@ -369,7 +370,7 @@ impl NodeService {
             Ok(Response::new(ReadVersionResponse {
                 success: false,
                 file_info: String::new(),
-                file_info_bin: Vec::new(),
+                file_info_bin: Vec::new().into(),
                 error: Some(DiskError::other("can not find disk".to_string()).into()),
             }))
         }

rustfs/src/storage/rpc/node_service.rs

@@ -1389,8 +1389,8 @@ mod tests {
             path: "test-path".to_string(),
             file_info: "{}".to_string(),
             opts: "{}".to_string(),
-            file_info_bin: Vec::new(),
+            file_info_bin: Vec::new().into(),
-            opts_bin: Vec::new(),
+            opts_bin: Vec::new().into(),
         });
 
         let response = service.update_metadata(request).await;
@@ -1411,8 +1411,8 @@ mod tests {
             path: "test-path".to_string(),
             file_info: "invalid json".to_string(),
             opts: "{}".to_string(),
-            file_info_bin: Vec::new(),
+            file_info_bin: Vec::new().into(),
-            opts_bin: Vec::new(),
+            opts_bin: Vec::new().into(),
         });
 
         let response = service.update_metadata(request).await;
@@ -1433,8 +1433,8 @@ mod tests {
             path: "test-path".to_string(),
             file_info: "{}".to_string(),
             opts: "invalid json".to_string(),
-            file_info_bin: Vec::new(),
+            file_info_bin: Vec::new().into(),
-            opts_bin: Vec::new(),
+            opts_bin: Vec::new().into(),
         });
 
         let response = service.update_metadata(request).await;
@@ -1454,7 +1454,7 @@ mod tests {
             volume: "test-volume".to_string(),
             path: "test-path".to_string(),
             file_info: "{}".to_string(),
-            file_info_bin: Vec::new(),
+            file_info_bin: Vec::new().into(),
         });
 
         let response = service.write_metadata(request).await;
@@ -1474,7 +1474,7 @@ mod tests {
             volume: "test-volume".to_string(),
             path: "test-path".to_string(),
             file_info: "invalid json".to_string(),
-            file_info_bin: Vec::new(),
+            file_info_bin: Vec::new().into(),
         });
 
         let response = service.write_metadata(request).await;
@@ -1495,7 +1495,7 @@ mod tests {
             path: "test-path".to_string(),
             version_id: "version1".to_string(),
             opts: "{}".to_string(),
-            opts_bin: Vec::new(),
+            opts_bin: Vec::new().into(),
         });
 
         let response = service.read_version(request).await;
@@ -1517,7 +1517,7 @@ mod tests {
             path: "test-path".to_string(),
             version_id: "version1".to_string(),
             opts: "invalid json".to_string(),
-            opts_bin: Vec::new(),
+            opts_bin: Vec::new().into(),
         });
 
         let response = service.read_version(request).await;
@@ -1675,7 +1675,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(),
+            read_multiple_req_bin: Vec::new().into(),
         });
 
         let response = service.read_multiple(request).await;
@@ -1694,7 +1694,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(),
+            read_multiple_req_bin: Vec::new().into(),
         });
 
         let response = service.read_multiple(request).await;