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_bin: file_info_bin.clone(),
+                    file_info,
+                    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(),
-                    opts: opts_str.clone(),
-                    file_info_bin: file_info_bin.clone(),
-                    opts_bin: opts_bin.clone(),
+                    file_info,
+                    opts: opts_str,
+                    file_info_bin: file_info_bin.into(),
+                    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_bin: opts_bin.clone(),
+                    opts: opts_str,
+                    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;
-use core::mem;
-
-extern crate flatbuffers;
-use self::flatbuffers::{EndianScalar, Follow};
+extern crate alloc;
 
 #[allow(unused_imports, dead_code)]
 pub mod models {
 
-    use core::cmp::Ordering;
-    use core::mem;
-
-    extern crate flatbuffers;
-    use self::flatbuffers::{EndianScalar, Follow};
+    extern crate alloc;
 
     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>(
-            _fbb: &'mut_bldr mut flatbuffers::FlatBufferBuilder<'bldr, A>,
+        pub fn create<'bldr: 'args, 'args: 'mut_bldr, 'mut_bldr, A: ::flatbuffers::Allocator + 'bldr>(
+            _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> {
-            use self::flatbuffers::Verifiable;
+        fn run_verifier(v: &mut ::flatbuffers::Verifier, pos: usize) -> Result<(), ::flatbuffers::InvalidFlatbuffer> {
             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> {
-        fbb_: &'b mut flatbuffers::FlatBufferBuilder<'a, A>,
-        start_: flatbuffers::WIPOffset<flatbuffers::TableUnfinishedWIPOffset>,
+    pub struct PingBodyBuilder<'a: 'b, 'b, A: ::flatbuffers::Allocator + 'a> {
+        fbb_: &'b mut ::flatbuffers::FlatBufferBuilder<'a, A>,
+        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<'_> {
-        fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+    impl ::core::fmt::Debug for PingBody<'_> {
+        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")]
-    pub file_info_bin: ::prost::alloc::vec::Vec<u8>,
-    #[prost(bytes = "vec", tag = "7")]
-    pub opts_bin: ::prost::alloc::vec::Vec<u8>,
+    #[prost(bytes = "bytes", tag = "6")]
+    pub file_info_bin: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "7")]
+    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")]
-    pub file_info_bin: ::prost::alloc::vec::Vec<u8>,
+    #[prost(bytes = "bytes", tag = "5")]
+    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")]
-    pub opts_bin: ::prost::alloc::vec::Vec<u8>,
+    #[prost(bytes = "bytes", tag = "6")]
+    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")]
-    pub file_info_bin: ::prost::alloc::vec::Vec<u8>,
+    #[prost(bytes = "bytes", tag = "4")]
+    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")]
-    pub raw_file_info_bin: ::prost::alloc::vec::Vec<u8>,
+    #[prost(bytes = "bytes", tag = "4")]
+    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")]
-    pub read_multiple_req_bin: ::prost::alloc::vec::Vec<u8>,
+    #[prost(bytes = "bytes", tag = "3")]
+    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")]
-    pub read_multiple_resps_bin: ::prost::alloc::vec::Vec<::prost::alloc::vec::Vec<u8>>,
+    #[prost(bytes = "bytes", repeated, tag = "4")]
+    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/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(),
-            opts_bin: Vec::new(),
+            file_info_bin: Vec::new().into(),
+            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(),
-            opts_bin: Vec::new(),
+            file_info_bin: Vec::new().into(),
+            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(),
-            opts_bin: Vec::new(),
+            file_info_bin: Vec::new().into(),
+            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;