From 2123b7a57dd1cf7f68d5649d1897affeadefefcd Mon Sep 17 00:00:00 2001 From: Arjan Bal Date: Wed, 19 Aug 2026 15:54:41 +0530 Subject: [PATCH 1/7] protobuf changes for server --- grpc-protobuf/Cargo.toml | 1 + grpc-protobuf/src/server/bidi.rs | 101 +++++++++++++++ grpc-protobuf/src/server/client_streaming.rs | 111 ++++++++++++++++ grpc-protobuf/src/server/mod.rs | 129 +++++++++++++++++++ grpc-protobuf/src/server/server_streaming.rs | 114 ++++++++++++++++ grpc-protobuf/src/server/unary.rs | 127 ++++++++++++++++++ grpc-protobuf/src/status.rs | 55 ++++++++ grpc-protobuf/src/trailers_conv.rs | 37 +++--- grpc/src/lib.rs | 3 + grpc/src/status.rs | 1 - grpc/src/status/server_status.rs | 90 ------------- 11 files changed, 662 insertions(+), 107 deletions(-) create mode 100644 grpc-protobuf/src/server/bidi.rs create mode 100644 grpc-protobuf/src/server/client_streaming.rs create mode 100644 grpc-protobuf/src/server/mod.rs create mode 100644 grpc-protobuf/src/server/server_streaming.rs create mode 100644 grpc-protobuf/src/server/unary.rs delete mode 100644 grpc/src/status/server_status.rs diff --git a/grpc-protobuf/Cargo.toml b/grpc-protobuf/Cargo.toml index 3dc636497..62de99750 100644 --- a/grpc-protobuf/Cargo.toml +++ b/grpc-protobuf/Cargo.toml @@ -25,6 +25,7 @@ bytes = "1.11.1" grpc = { version = "0.10.0", path = "../grpc" } protobuf = "4.35.1-release" protobuf-well-known-types = "4.35.1-release" +trait-variant = "0.1" [dev-dependencies] diff --git a/grpc-protobuf/src/server/bidi.rs b/grpc-protobuf/src/server/bidi.rs new file mode 100644 index 000000000..db18f235d --- /dev/null +++ b/grpc-protobuf/src/server/bidi.rs @@ -0,0 +1,101 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +use grpc::async_trait; +use grpc::server::BoxedRecvStream; +use grpc::server::CallOptions; +use grpc::server::DynHandle; +use grpc::server::DynSendStream; +use grpc::server::RequestHeaders; +use grpc::server::Trailers; +use protobuf::ClearAndParse; +use protobuf::Message; +use protobuf::MutProxied; +use protobuf::Proxied; +use protobuf::Serialize; + +use crate::ServerStatus; +use crate::server::GrpcStreamingRequest; +use crate::server::GrpcStreamingResponse; +use crate::trailers_conv::trailers_from_status; + +/// A bidirectional-streaming RPC method handler on the server. +/// +/// Implementations receive a stream of request messages and send a stream of +/// response messages to the client. +#[trait_variant::make(Send)] +pub trait BidiStreamingMethod: Sync + 'static { + /// The protobuf request message type. + type Request: Message + Default; + /// The protobuf response message type. + type Response: Message + Default; + + /// Handles a bidirectional-streaming RPC call. + /// + /// Receives incoming `requests` from the client and uses `responses` to + /// stream response messages back to the client, returning a [`ServerStatus`] + /// when the handler has completed. + async fn call( + &self, + requests: GrpcStreamingRequest, + responses: GrpcStreamingResponse<'_, Self::Response>, + ) -> ServerStatus; +} + +/// An adapter that wraps a [`BidiStreamingMethod`] to handle incoming +/// bidirectional-streaming RPCs. +pub struct BidiStreamingAdapter { + method: M, +} + +impl BidiStreamingAdapter { + /// Creates a new [`BidiStreamingAdapter`] wrapping the given `method`. + pub fn new(method: M) -> Self { + Self { method } + } +} + +#[async_trait] +impl DynHandle for BidiStreamingAdapter +where + M: BidiStreamingMethod, + for<'a> ::Mut<'a>: ClearAndParse + Send + Sync, + for<'a> ::View<'a>: Serialize + Send + Sync, +{ + async fn dyn_handle( + &self, + _headers: RequestHeaders, + _options: CallOptions, + tx: &mut dyn DynSendStream, + rx: BoxedRecvStream, + ) -> Trailers { + // The request stream owns `rx`; the response sink borrows `tx`. They + // are independent, so a handler can freely interleave receives and + // sends. + let requests = GrpcStreamingRequest::new(rx); + let responses = GrpcStreamingResponse::new(tx); + let status = self.method.call(requests, responses).await; + trailers_from_status(status) + } +} diff --git a/grpc-protobuf/src/server/client_streaming.rs b/grpc-protobuf/src/server/client_streaming.rs new file mode 100644 index 000000000..a9c9b7b52 --- /dev/null +++ b/grpc-protobuf/src/server/client_streaming.rs @@ -0,0 +1,111 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +use grpc::async_trait; +use grpc::server::BoxedRecvStream; +use grpc::server::CallOptions; +use grpc::server::DynHandle; +use grpc::server::DynSendStream; +use grpc::server::RequestHeaders; +use grpc::server::ResponseStreamItem; +use grpc::server::SendOptions; +use grpc::server::Trailers; +use protobuf::AsMut; +use protobuf::ClearAndParse; +use protobuf::Message; +use protobuf::MutProxied; +use protobuf::Proxied; +use protobuf::Serialize; + +use crate::ProtoSendMessage; +use crate::ServerStatus; +use crate::server::GrpcStreamingRequest; +use crate::trailers_conv::trailers_from_status; + +/// A client-streaming RPC method handler on the server. +/// +/// Implementations receive a stream of request messages from the client and +/// populate a single response message. +#[trait_variant::make(Send)] +pub trait ClientStreamingMethod: Sync + 'static { + /// The protobuf request message type. + type Request: Message + Default; + /// The protobuf response message type. + type Response: Message + Default; + + /// Handles a client-streaming RPC call. + /// + /// Receives a stream of incoming `requests` from the client and populates + /// the `response` message, returning a [`ServerStatus`] to indicate success + /// or failure. + async fn call( + &self, + requests: GrpcStreamingRequest, + response: ::Mut<'_>, + ) -> ServerStatus; +} + +/// An adapter that wraps a [`ClientStreamingMethod`] to handle incoming +/// client-streaming RPCs. +pub struct ClientStreamingAdapter { + method: M, +} + +impl ClientStreamingAdapter { + /// Creates a new [`ClientStreamingAdapter`] wrapping the given `method`. + pub fn new(method: M) -> Self { + Self { method } + } +} + +#[async_trait] +impl DynHandle for ClientStreamingAdapter +where + M: ClientStreamingMethod, + for<'a> ::Mut<'a>: ClearAndParse + Send + Sync, + for<'a> ::View<'a>: Serialize + Send + Sync, +{ + async fn dyn_handle( + &self, + _headers: RequestHeaders, + _options: CallOptions, + tx: &mut dyn DynSendStream, + rx: BoxedRecvStream, + ) -> Trailers { + let requests = GrpcStreamingRequest::new(rx); + let mut resp = ::default(); + let status = self.method.call(requests, resp.as_mut()).await; + + if status.is_ok() { + let send = ProtoSendMessage::from_view(&resp); + let mut options = SendOptions::default(); + options.final_msg = true; + let _ = tx + .dyn_send(ResponseStreamItem::Message(&send), options) + .await; + } + + trailers_from_status(status) + } +} diff --git a/grpc-protobuf/src/server/mod.rs b/grpc-protobuf/src/server/mod.rs new file mode 100644 index 000000000..88a8349eb --- /dev/null +++ b/grpc-protobuf/src/server/mod.rs @@ -0,0 +1,129 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +use std::marker::PhantomData; + +use grpc::server::BoxedRecvStream; +use grpc::server::DynRecvStream; +use grpc::server::DynSendStream; +use grpc::server::ResponseStreamItem; +use grpc::server::SendOptions; +use protobuf::AsMut; +use protobuf::AsView; +use protobuf::Message; +use protobuf::MessageMut; +use protobuf::MessageView; + +use crate::ProtoRecvMessage; +use crate::ProtoSendMessage; + +pub(crate) mod bidi; +pub(crate) mod client_streaming; +pub(crate) mod server_streaming; +pub(crate) mod unary; + +pub use bidi::*; +pub use client_streaming::*; +pub use server_streaming::*; +pub use unary::*; + +/// Allows receiving streaming RPC protobuf request messages on the server. +pub struct GrpcStreamingRequest { + rx: BoxedRecvStream, + _phantom: PhantomData, +} + +impl GrpcStreamingRequest +where + M: Message, + for<'b> M::Mut<'b>: MessageMut<'b>, +{ + /// Creates a new [`GrpcStreamingRequest`]. + pub(crate) fn new(rx: BoxedRecvStream) -> Self { + Self { + rx, + _phantom: PhantomData, + } + } + + /// Receives the next request message from the stream into `req`. + /// + /// Returns `Some(Ok(()))` on success, `Some(Err(()))` if the stream + /// encountered an error, or `None` if the client has closed the stream. + pub async fn recv_into( + &mut self, + req: &mut impl AsMut, + ) -> Option> { + let mut res_view = ProtoRecvMessage::from_mut(req); + self.rx.dyn_next(&mut res_view).await + } + + /// Receives the next request message from the stream. + /// + /// Returns `Some(Ok(msg))` on success, `Some(Err(()))` if the stream + /// encountered an error, or `None` if the client has closed the stream. + pub async fn recv(&mut self) -> Option> { + let mut req = M::default(); + match self.recv_into(&mut req).await { + Some(Ok(())) => Some(Ok(req)), + Some(Err(())) => Some(Err(())), + None => None, + } + } +} + +/// Allows sending streaming RPC protobuf response messages from the server. +pub struct GrpcStreamingResponse<'a, M> { + tx: &'a mut dyn DynSendStream, + _phantom: PhantomData, +} + +impl<'a, M> GrpcStreamingResponse<'a, M> +where + M: Message, + for<'b> M::View<'b>: MessageView<'b>, +{ + pub(crate) fn new(tx: &'a mut dyn DynSendStream) -> Self { + Self { + tx, + _phantom: PhantomData, + } + } + + /// Sends a response message on the stream. + /// + /// Will block if flow control does not allow sending the message. Returns + /// an error if the stream has ended or been cancelled. + /// + /// Note: success does *not* indicate successful receipt of the response by + /// the client; it only indicates that the stream has not yet terminated. + pub async fn send(&mut self, resp: &impl AsView) -> Result<(), ()> { + self.tx + .dyn_send( + ResponseStreamItem::Message(&ProtoSendMessage::from_view(resp)), + SendOptions::default(), + ) + .await + } +} diff --git a/grpc-protobuf/src/server/server_streaming.rs b/grpc-protobuf/src/server/server_streaming.rs new file mode 100644 index 000000000..41c747aa5 --- /dev/null +++ b/grpc-protobuf/src/server/server_streaming.rs @@ -0,0 +1,114 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +use grpc::async_trait; +use grpc::server::BoxedRecvStream; +use grpc::server::CallOptions; +use grpc::server::DynHandle; +use grpc::server::DynRecvStream; +use grpc::server::DynSendStream; +use grpc::server::RequestHeaders; +use grpc::server::Trailers; +use protobuf::AsView; +use protobuf::ClearAndParse; +use protobuf::Message; +use protobuf::MutProxied; +use protobuf::Proxied; +use protobuf::Serialize; + +use crate::ProtoRecvMessage; +use crate::ServerStatus; +use crate::ServerStatusError; +use crate::StatusCodeError; +use crate::server::GrpcStreamingResponse; +use crate::trailers_conv::trailers_from_status; + +/// A server-streaming RPC method handler on the server. +/// +/// Implementations receive a single request message and send a stream of +/// response messages to the client. +#[trait_variant::make(Send)] +pub trait ServerStreamingMethod: Sync + 'static { + /// The protobuf request message type. + type Request: Message + Default; + /// The protobuf response message type. + type Response: Message + Default; + + /// Handles a server-streaming RPC call. + /// + /// Receives a view of the incoming `request` message and uses `responses` + /// to stream response messages back to the client, returning a + /// [`ServerStatus`] when the handler has completed. + async fn call( + &self, + request: ::View<'_>, + responses: GrpcStreamingResponse<'_, Self::Response>, + ) -> ServerStatus; +} + +/// An adapter that wraps a [`ServerStreamingMethod`] to handle incoming +/// server-streaming RPCs. +pub struct ServerStreamingAdapter { + method: M, +} + +impl ServerStreamingAdapter { + /// Creates a new [`ServerStreamingAdapter`] wrapping the given `method`. + pub fn new(method: M) -> Self { + Self { method } + } +} + +#[async_trait] +impl DynHandle for ServerStreamingAdapter +where + M: ServerStreamingMethod, + for<'a> ::Mut<'a>: ClearAndParse + Send + Sync, + for<'a> ::View<'a>: Serialize + Send + Sync, +{ + async fn dyn_handle( + &self, + _headers: RequestHeaders, + _options: CallOptions, + tx: &mut dyn DynSendStream, + mut rx: BoxedRecvStream, + ) -> Trailers { + let mut req = ::default(); + + if rx + .dyn_next(&mut ProtoRecvMessage::from_mut(&mut req)) + .await + .is_none_or(|res| res.is_err()) + { + return trailers_from_status(Err(ServerStatusError::new( + StatusCodeError::Internal, + "client did not send a request message", + ))); + } + + let responses = GrpcStreamingResponse::new(tx); + let status = self.method.call(req.as_view(), responses).await; + trailers_from_status(status) + } +} diff --git a/grpc-protobuf/src/server/unary.rs b/grpc-protobuf/src/server/unary.rs new file mode 100644 index 000000000..eb3d3161f --- /dev/null +++ b/grpc-protobuf/src/server/unary.rs @@ -0,0 +1,127 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +use grpc::async_trait; +use grpc::server::BoxedRecvStream; +use grpc::server::CallOptions; +use grpc::server::DynHandle; +use grpc::server::DynRecvStream; +use grpc::server::DynSendStream; +use grpc::server::RequestHeaders; +use grpc::server::ResponseStreamItem; +use grpc::server::SendOptions; +use grpc::server::Trailers; +use protobuf::AsMut; +use protobuf::AsView; +use protobuf::ClearAndParse; +use protobuf::Message; +use protobuf::MutProxied; +use protobuf::Proxied; +use protobuf::Serialize; + +use crate::ProtoRecvMessage; +use crate::ProtoSendMessage; +use crate::ServerStatus; +use crate::ServerStatusError; +use crate::StatusCodeError; +use crate::trailers_conv::trailers_from_status; + +/// A unary RPC method handler on the server. +/// +/// Implementations receive a single request message and populate a single +/// response message. +#[trait_variant::make(Send)] +pub trait UnaryMethod: Sync + 'static { + /// The protobuf request message type. + type Request: Message + Default; + /// The protobuf response message type. + type Response: Message + Default; + + /// Handles a unary RPC call. + /// + /// Receives a view of the incoming `request` message and populates the + /// `response` message, returning a [`ServerStatus`] to indicate success + /// or failure. + async fn call( + &self, + request: ::View<'_>, + response: ::Mut<'_>, + ) -> ServerStatus; +} + +/// An adapter that wraps a [`UnaryMethod`] to handle incoming unary RPCs. +pub struct UnaryAdapter { + method: M, +} + +impl UnaryAdapter { + /// Creates a new [`UnaryAdapter`] wrapping the given `method`. + pub fn new(method: M) -> Self { + Self { method } + } +} + +#[async_trait] +impl DynHandle for UnaryAdapter +where + M: UnaryMethod, + for<'a> ::Mut<'a>: ClearAndParse + Send + Sync, + for<'a> ::View<'a>: Serialize + Send + Sync, +{ + async fn dyn_handle( + &self, + _headers: RequestHeaders, + _options: CallOptions, + tx: &mut dyn DynSendStream, + mut rx: BoxedRecvStream, + ) -> Trailers { + let mut req = ::default(); + + if rx + .dyn_next(&mut ProtoRecvMessage::from_mut(&mut req)) + .await + .is_none_or(|res| res.is_err()) + { + return trailers_from_status(Err(ServerStatusError::new( + StatusCodeError::Internal, + "client did not send a request message", + ))); + } + + let mut resp = ::default(); + let status = self.method.call(req.as_view(), resp.as_mut()).await; + + if status.is_ok() { + let send = ProtoSendMessage::from_view(&resp); + let mut options = SendOptions::default(); + options.final_msg = true; + + let _ = tx + .dyn_send(ResponseStreamItem::Message(&send), options) + .await; + } + + trailers_from_status(status) + } +} diff --git a/grpc-protobuf/src/status.rs b/grpc-protobuf/src/status.rs index 95900a065..ab290a16c 100644 --- a/grpc-protobuf/src/status.rs +++ b/grpc-protobuf/src/status.rs @@ -162,6 +162,12 @@ pub type StatusOr = Result; /// Represents either a failing gRPC status or a successful result. This is expected to be replaced /// with absl::Status when it becomes available. pub type Status = StatusOr<()>; +/// Represents either a failing gRPC status or a successful result containing +/// `T`. +pub type ServerStatusOr = Result; +/// Represents either a failing gRPC status or a successful result produced by +/// a server handler. +pub type ServerStatus = ServerStatusOr<()>; /// Represents a gRPC status. This is expected to be replaced with absl::StatusError when it becomes /// available. @@ -217,6 +223,55 @@ impl StatusError { } } +/// Represents a gRPC error status on the server. +#[derive(Debug, Clone)] +pub struct ServerStatusError(StatusError); + +impl std::ops::Deref for ServerStatusError { + type Target = StatusError; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +impl ServerStatusError { + /// Creates a new [`ServerStatusError`] with the given code and message. + pub fn new(code: StatusCodeError, message: impl Into) -> Self { + ServerStatusError(StatusError::new(code, message)) + } + + /// Creates a new [`ServerStatusError`] from a [`StatusError`]. + pub fn from_status(status: StatusError) -> Self { + ServerStatusError(status) + } + + /// Returns the [`StatusCodeError`] of this [`ServerStatusError`]. + pub fn code(&self) -> StatusCodeError { + self.0.code() + } + + /// Returns the message of this [`ServerStatusError`]. + pub fn message(&self) -> &str { + self.0.message() + } + + /// Gets the value for `type_url`. + pub fn get_payload<'a>(&'a self, type_url: &[u8]) -> Option<&'a [u8]> { + self.0.get_payload(type_url) + } + + /// Sets the value for `type_url`. + pub fn set_payload(&mut self, type_url: &[u8], payload: &[u8]) { + self.0.set_payload(type_url, payload); + } + + /// Converts the [`ServerStatusError`] to a [`StatusError`] for client responses. + pub(crate) fn into_status(self) -> StatusError { + self.0 + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/grpc-protobuf/src/trailers_conv.rs b/grpc-protobuf/src/trailers_conv.rs index b1dac4e7e..85de38ee5 100644 --- a/grpc-protobuf/src/trailers_conv.rs +++ b/grpc-protobuf/src/trailers_conv.rs @@ -23,6 +23,7 @@ */ use grpc::client::Trailers; +use grpc::server::Trailers as ServerTrailers; use protobuf::Parse; use protobuf::Serialize; use protobuf_well_known_types::Any; @@ -101,11 +102,11 @@ fn parse_rpc_status(buf: &[u8]) -> StatusOr { } /// Converts the status to trailers and inserts grpc-status-details-bin into the metadata. -#[allow(dead_code)] -pub(crate) fn trailers_from_status(s: Status) -> Trailers { +pub(crate) fn trailers_from_status(s: ServerStatus) -> ServerTrailers { match s { - Ok(()) => Trailers::new(Ok(())), - Err(status_err) => { + Ok(()) => ServerTrailers::new(Ok(())), + Err(server_status_err) => { + let status_err = server_status_err.into_status(); let has_payloads = status_err.has_payloads(); let (code, message, payloads) = status_err.into_parts(); let mut m = grpc::metadata::MetadataMap::new(); @@ -113,7 +114,7 @@ pub(crate) fn trailers_from_status(s: Status) -> Trailers { let code_i32 = code as i32; let bytes = match encode_rpc_status(code_i32, &message, payloads) { Ok(bytes) => bytes, - Err(err) => return Trailers::new(Err(err)), + Err(err) => return ServerTrailers::new(Err(err)), }; m.insert_bin( "grpc-status-details-bin", @@ -123,7 +124,7 @@ pub(crate) fn trailers_from_status(s: Status) -> Trailers { ); } let grpc_code = grpc::StatusCodeError::from(code as i32); - Trailers::new(Err(grpc::StatusError::new(grpc_code, message))).with_metadata(m) + ServerTrailers::new(Err(grpc::StatusError::new(grpc_code, message))).with_metadata(m) } } } @@ -163,7 +164,7 @@ mod tests { #[test] fn test_trailers_from_status_details_copied_to_grpc_status() { - let mut err = StatusError::new(StatusCodeError::NotFound, "not found detail"); + let mut err = ServerStatusError::new(StatusCodeError::NotFound, "not found detail"); err.set_payload(b"type.googleapis.com/test", b"hello world"); let trailers = trailers_from_status(Err(err)); @@ -192,11 +193,11 @@ mod tests { #[test] fn test_trailers_from_status_empty_payload_skips_metadata() { - let err = StatusError::new( + let err = ServerStatusError::new( StatusCodeError::NotFound, "Resource missing without details", ); - let status_or: Status = Err(err); + let status_or: ServerStatus = Err(err); let trailers = trailers_from_status(status_or); assert!(trailers.status().is_err()); @@ -210,12 +211,12 @@ mod tests { #[test] fn test_roundtrip_payload() { - let mut og_err = StatusError::new(StatusCodeError::NotFound, "not found detail"); + let mut og_err = ServerStatusError::new(StatusCodeError::NotFound, "not found detail"); og_err.set_payload(b"type.googleapis.com/foo", b"hello"); og_err.set_payload(b"type.googleapis.com/bar", b"world"); let trailers = trailers_from_status(Err(og_err.clone())); - let rt_err = status_from_trailers(trailers).unwrap_err(); + let rt_err = status_from_trailers(client_trailers(trailers)).unwrap_err(); assert_eq!(rt_err.code(), og_err.code()); assert_eq!(rt_err.message(), og_err.message()); assert_eq!( @@ -230,13 +231,13 @@ mod tests { #[test] fn test_roundtrip_invalid_utf8_dropped() { - let mut og_err = StatusError::new(StatusCodeError::NotFound, "not found detail"); + let mut og_err = ServerStatusError::new(StatusCodeError::NotFound, "not found detail"); og_err.set_payload(b"type.googleapis.com/foo", b"world"); og_err.set_payload(b"type.googleapis.com/bar\x80", b"ain't gonna work"); og_err.set_payload(b"type.googleapis.com/bar", b"hello"); let trailers = trailers_from_status(Err(og_err.clone())); - let rt_err = status_from_trailers(trailers).unwrap_err(); + let rt_err = status_from_trailers(client_trailers(trailers)).unwrap_err(); assert_eq!(rt_err.code(), og_err.code()); assert_eq!(rt_err.message(), og_err.message()); // Other payloads are preserved @@ -254,10 +255,10 @@ mod tests { #[test] fn test_roundtrip_no_payload() { - let og_err = StatusError::new(StatusCodeError::NotFound, "not found detail"); + let og_err = ServerStatusError::new(StatusCodeError::NotFound, "not found detail"); let trailers = trailers_from_status(Err(og_err.clone())); - let rt_err = status_from_trailers(trailers).unwrap_err(); + let rt_err = status_from_trailers(client_trailers(trailers)).unwrap_err(); assert_eq!(rt_err.code(), og_err.code()); assert_eq!(rt_err.message(), og_err.message()); assert!(!rt_err.has_payloads()); @@ -266,7 +267,7 @@ mod tests { #[test] fn test_roundtrip_ok() { let trailers = trailers_from_status(Ok(())); - let status = status_from_trailers(trailers); + let status = status_from_trailers(client_trailers(trailers)); assert!(status.is_ok()); } @@ -425,4 +426,8 @@ mod tests { .contains("Failed to parse grpc-status-details-bin:") ); } + + fn client_trailers(trailers: ServerTrailers) -> Trailers { + Trailers::new(trailers.status().clone()).with_metadata(trailers.metadata().clone()) + } } diff --git a/grpc/src/lib.rs b/grpc/src/lib.rs index e5002f999..cbf61e007 100644 --- a/grpc/src/lib.rs +++ b/grpc/src/lib.rs @@ -67,6 +67,9 @@ mod status; pub use status::Result; pub use status::StatusCodeError; pub use status::StatusError; +/// A re-export of [`async-trait`](https://docs.rs/async-trait) for use with +/// codegen. +pub use tonic::async_trait; #[cfg(feature = "__unstable")] #[doc(hidden)] diff --git a/grpc/src/status.rs b/grpc/src/status.rs index fb30106a6..a41d55bed 100644 --- a/grpc/src/status.rs +++ b/grpc/src/status.rs @@ -22,7 +22,6 @@ * */ -mod server_status; mod status_code; pub use status_code::StatusCodeError; diff --git a/grpc/src/status/server_status.rs b/grpc/src/status/server_status.rs deleted file mode 100644 index 839da9581..000000000 --- a/grpc/src/status/server_status.rs +++ /dev/null @@ -1,90 +0,0 @@ -/* - * - * Copyright 2025 gRPC authors. - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to - * deal in the Software without restriction, including without limitation the - * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or - * sell copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in - * all copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING - * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS - * IN THE SOFTWARE. - * - */ - -use crate::status::StatusError; -use crate::status::status_code::StatusCodeError; - -/// Represents a gRPC status on the server. -/// -/// This is a separate type from [`StatusError`] to prevent accidental conversion and -/// leaking of sensitive information from the server to the client. -#[derive(Debug, Clone)] -pub struct ServerStatusErr(StatusError); - -impl std::ops::Deref for ServerStatusErr { - type Target = StatusError; - - fn deref(&self) -> &Self::Target { - &self.0 - } -} - -impl ServerStatusErr { - /// Create a new [`ServerStatusErr`] with the given code and message. - pub fn new(code: StatusCodeError, message: impl Into) -> Self { - ServerStatusErr(StatusError::new(code, message)) - } - - /// Create a new [`ServerStatusErr`] from a [`StatusError`]. - pub fn from_status(status: StatusError) -> Self { - ServerStatusErr(status) - } - - /// Converts the [`ServerStatusErr`] to a [`StatusError`] for client responses. - pub(crate) fn into_status(self) -> StatusError { - self.0 - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_server_status_new() { - let status = ServerStatusErr::new(StatusCodeError::Internal, "not ok"); - assert_eq!(status.code(), StatusCodeError::Internal); - assert_eq!(status.message(), "not ok"); - } - - #[test] - fn test_server_status_deref() { - let status = ServerStatusErr::new(StatusCodeError::FailedPrecondition, "x"); - assert_eq!(status.code(), StatusCodeError::FailedPrecondition); - } - - #[test] - fn test_server_status_from_status() { - let status = StatusError::new(StatusCodeError::DeadlineExceeded, "DE"); - let server_status = ServerStatusErr::from_status(status); - assert_eq!(server_status.code(), StatusCodeError::DeadlineExceeded); - } - - #[test] - fn test_server_status_into_status() { - let server_status = ServerStatusErr::new(StatusCodeError::DataLoss, "DL"); - let status = server_status.into_status(); - assert_eq!(status.code(), StatusCodeError::DataLoss); - } -} From 61d8f5dfc5ceda14a208fff4c5fef2fb003cca08 Mon Sep 17 00:00:00 2001 From: Arjan Bal Date: Wed, 19 Aug 2026 16:44:27 +0530 Subject: [PATCH 2/7] export server mod --- grpc/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/grpc/src/lib.rs b/grpc/src/lib.rs index cbf61e007..9f9483e6c 100644 --- a/grpc/src/lib.rs +++ b/grpc/src/lib.rs @@ -55,13 +55,13 @@ pub(crate) mod codec; pub mod core; pub mod credentials; pub mod metadata; +pub mod server; mod byte_str; mod inmemory; mod macros; mod rt; mod send_future; -mod server; mod status; pub use status::Result; From e347a27b3af032f418eaf56b287c046404f256d2 Mon Sep 17 00:00:00 2001 From: Arjan Bal Date: Wed, 19 Aug 2026 16:56:17 +0530 Subject: [PATCH 3/7] external types --- grpc/Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/grpc/Cargo.toml b/grpc/Cargo.toml index 3ae204c2d..f4fb78801 100644 --- a/grpc/Cargo.toml +++ b/grpc/Cargo.toml @@ -16,9 +16,9 @@ edition = "2024" [package.metadata.cargo_check_external_types] allowed_external_types = [ + "async_trait::async_trait", "bytes::*", "tonic::*", - "futures_core::stream::Stream", "tokio::sync::oneshot::*", ] From 5fd7f289ae3976fd53c628c0f7669623917c2eb5 Mon Sep 17 00:00:00 2001 From: Arjan Bal Date: Wed, 19 Aug 2026 17:09:51 +0530 Subject: [PATCH 4/7] fix docs, use crate async_trait --- grpc/src/client/mod.rs | 2 +- grpc/src/client/name_resolution/dns/test.rs | 2 +- grpc/src/client/name_resolution/proxy_resolver.rs | 2 +- grpc/src/client/subchannel.rs | 2 +- grpc/src/client/transport/http_connect/mod.rs | 2 +- grpc/src/client/transport/mod.rs | 2 +- grpc/src/client/transport/tonic/test.rs | 2 +- grpc/src/credentials/call.rs | 2 +- grpc/src/credentials/client.rs | 4 ++-- grpc/src/credentials/dyn_wrapper.rs | 2 +- grpc/src/credentials/local.rs | 2 +- grpc/src/credentials/mod.rs | 2 +- grpc/src/credentials/rustls/client/mod.rs | 2 +- grpc/src/credentials/rustls/server/mod.rs | 2 +- grpc/src/rt/mod.rs | 4 ++-- grpc/src/rt/tokio/hickory_resolver.rs | 2 +- grpc/src/rt/tokio/mod.rs | 6 +++--- grpc/src/server/mod.rs | 4 +--- 18 files changed, 22 insertions(+), 24 deletions(-) diff --git a/grpc/src/client/mod.rs b/grpc/src/client/mod.rs index fb6c626ff..d74606811 100644 --- a/grpc/src/client/mod.rs +++ b/grpc/src/client/mod.rs @@ -51,7 +51,7 @@ use std::fmt::Display; use std::time::Instant; -use tonic::async_trait; +use crate::async_trait; use crate::core::RecvMessage; use crate::core::SendMessage; diff --git a/grpc/src/client/name_resolution/dns/test.rs b/grpc/src/client/name_resolution/dns/test.rs index f50616673..b9072c264 100644 --- a/grpc/src/client/name_resolution/dns/test.rs +++ b/grpc/src/client/name_resolution/dns/test.rs @@ -228,7 +228,7 @@ struct FakeDns { lookup_result: Result, String>, } -#[tonic::async_trait] +#[crate::async_trait] impl rt::DnsResolver for FakeDns { async fn lookup_host_name(&self, _: &str) -> Result, String> { tokio::time::sleep(self.latency).await; diff --git a/grpc/src/client/name_resolution/proxy_resolver.rs b/grpc/src/client/name_resolution/proxy_resolver.rs index 11f61a185..2dc043bbb 100644 --- a/grpc/src/client/name_resolution/proxy_resolver.rs +++ b/grpc/src/client/name_resolution/proxy_resolver.rs @@ -282,7 +282,7 @@ mod tests { lookup_result: Result, String>, } - #[tonic::async_trait] + #[crate::async_trait] impl rt::DnsResolver for FakeDns { async fn lookup_host_name(&self, _: &str) -> Result, String> { self.lookup_result.clone() diff --git a/grpc/src/client/subchannel.rs b/grpc/src/client/subchannel.rs index 2f5af89e7..5f79051d6 100644 --- a/grpc/src/client/subchannel.rs +++ b/grpc/src/client/subchannel.rs @@ -33,9 +33,9 @@ use std::sync::Weak; use std::time::Duration; use std::time::Instant; +use crate::async_trait; use tokio::sync::Notify; use tokio::sync::oneshot; -use tonic::async_trait; use crate::StatusCodeError; use crate::StatusError; diff --git a/grpc/src/client/transport/http_connect/mod.rs b/grpc/src/client/transport/http_connect/mod.rs index deac88919..d62791889 100644 --- a/grpc/src/client/transport/http_connect/mod.rs +++ b/grpc/src/client/transport/http_connect/mod.rs @@ -24,10 +24,10 @@ use std::sync::Arc; +use crate::async_trait; use bytes::Bytes; use tokio::io::AsyncReadExt; use tokio::io::AsyncWriteExt; -use tonic::async_trait; use crate::client::transport::ProxyOptions; use crate::client::transport::http_connect::rewind::Rewind; diff --git a/grpc/src/client/transport/mod.rs b/grpc/src/client/transport/mod.rs index a918959ca..dbea81d24 100644 --- a/grpc/src/client/transport/mod.rs +++ b/grpc/src/client/transport/mod.rs @@ -44,7 +44,7 @@ mod registry; #[cfg(feature = "_runtime-tokio")] pub(crate) mod tonic; -use ::tonic::async_trait; +use crate::async_trait; pub(crate) use registry::GLOBAL_TRANSPORT_REGISTRY; pub(crate) use registry::TransportRegistry; use tokio::sync::oneshot; diff --git a/grpc/src/client/transport/tonic/test.rs b/grpc/src/client/transport/tonic/test.rs index 847da4a72..920be7e85 100644 --- a/grpc/src/client/transport/tonic/test.rs +++ b/grpc/src/client/transport/tonic/test.rs @@ -30,6 +30,7 @@ use std::sync::Arc; use std::sync::Once; use std::time::Duration; +use crate::async_trait; use bytes::Buf; use bytes::Bytes; use h2::Reason; @@ -47,7 +48,6 @@ use tokio_stream::wrappers::ReceiverStream; use tokio_stream::wrappers::TcpListenerStream; use tonic::Response; use tonic::Status as TonicStatus; -use tonic::async_trait; use tonic::metadata::MetadataMap as TonicMetadata; use tonic::metadata::MetadataValue as TonicMetadataValue; use tonic::transport::Server; diff --git a/grpc/src/credentials/call.rs b/grpc/src/credentials/call.rs index 1145085e7..20a1e69c8 100644 --- a/grpc/src/credentials/call.rs +++ b/grpc/src/credentials/call.rs @@ -27,7 +27,7 @@ use std::fmt::Debug; use std::sync::Arc; -use tonic::async_trait; +use crate::async_trait; use crate::StatusError; use crate::attributes::Attributes; diff --git a/grpc/src/credentials/client.rs b/grpc/src/credentials/client.rs index d15550cf3..eb844aaa0 100644 --- a/grpc/src/credentials/client.rs +++ b/grpc/src/credentials/client.rs @@ -25,7 +25,7 @@ use std::fmt::Debug; use std::sync::Arc; -use tonic::async_trait; +use crate::async_trait; use crate::attributes::Attributes; use crate::credentials::ChannelCredentials; @@ -144,8 +144,8 @@ impl ChannelCredentials for CompositeChannelCredentials; pub type BoxEndpoint = Box; /// A server-side listening socket that yields incoming connections. -#[tonic::async_trait] +#[crate::async_trait] pub(crate) trait EndpointListener: Send + Sync + 'static { /// Accepts the next incoming connection. async fn accept(&self) -> Result, String>; @@ -125,7 +125,7 @@ pub trait TaskHandle: Send + Sync { } /// A trait for asynchronous DNS resolution. -#[tonic::async_trait] +#[crate::async_trait] pub trait DnsResolver: Send + Sync { /// Resolve an address async fn lookup_host_name(&self, name: &str) -> Result, String>; diff --git a/grpc/src/rt/tokio/hickory_resolver.rs b/grpc/src/rt/tokio/hickory_resolver.rs index 859b3cfab..b98d22414 100644 --- a/grpc/src/rt/tokio/hickory_resolver.rs +++ b/grpc/src/rt/tokio/hickory_resolver.rs @@ -43,7 +43,7 @@ pub(super) struct DnsResolver { resolver: hickory_resolver::TokioResolver, } -#[tonic::async_trait] +#[crate::async_trait] impl rt::DnsResolver for DnsResolver { async fn lookup_host_name(&self, name: &str) -> Result, String> { let response = self diff --git a/grpc/src/rt/tokio/mod.rs b/grpc/src/rt/tokio/mod.rs index cdf4cc7ec..9c92e55a6 100644 --- a/grpc/src/rt/tokio/mod.rs +++ b/grpc/src/rt/tokio/mod.rs @@ -52,7 +52,7 @@ struct TokioDefaultDnsResolver { _priv: (), } -#[tonic::async_trait] +#[crate::async_trait] impl DnsResolver for TokioDefaultDnsResolver { async fn lookup_host_name(&self, name: &str) -> Result, String> { let name_with_port = match name.parse::() { @@ -230,7 +230,7 @@ struct TokioTcpListener { listener: tokio::net::TcpListener, } -#[tonic::async_trait] +#[crate::async_trait] impl super::EndpointListener for TokioTcpListener { async fn accept(&self) -> Result, String> { let (stream, _addr) = self.listener.accept().await.map_err(|e| e.to_string())?; @@ -254,7 +254,7 @@ struct TokioUnixListener { } #[cfg(unix)] -#[tonic::async_trait] +#[crate::async_trait] impl super::EndpointListener for TokioUnixListener { async fn accept(&self) -> Result, String> { use crate::client::name_resolution::UNIX_NETWORK_TYPE; diff --git a/grpc/src/server/mod.rs b/grpc/src/server/mod.rs index 2088d3ca3..ed0ce83ee 100644 --- a/grpc/src/server/mod.rs +++ b/grpc/src/server/mod.rs @@ -37,7 +37,6 @@ //! //! # Additional Types //! -//! - **[`Call`]:** Represents an incoming RPC accepted by a [`Listener`]. //! - **[`SendStream`] / [`RecvStream`]:** Represent the sending and receiving //! sides of a server-side RPC. //! - **[`RequestHeaders`]:** Represents gRPC headers sent by the client to @@ -49,8 +48,7 @@ use std::future::Future; use std::pin::Pin; use std::sync::Arc; -use tonic::async_trait; - +use crate::async_trait; use crate::core::RecvMessage; use crate::core::SendMessage; use crate::metadata::MetadataMap; From 9f5a0bb8424dc4d481fefb323485e471ee98a72d Mon Sep 17 00:00:00 2001 From: Arjan Bal Date: Wed, 19 Aug 2026 17:23:59 +0530 Subject: [PATCH 5/7] export server mod --- grpc-protobuf/src/lib.rs | 1 + grpc-protobuf/src/server/mod.rs | 8 ++++---- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/grpc-protobuf/src/lib.rs b/grpc-protobuf/src/lib.rs index 70ffbd1f5..d2c2ca127 100644 --- a/grpc-protobuf/src/lib.rs +++ b/grpc-protobuf/src/lib.rs @@ -59,6 +59,7 @@ use protobuf::Proxied; use protobuf::Serialize; mod client; +pub mod server; mod status; mod trailers_conv; pub use client::bidi::*; diff --git a/grpc-protobuf/src/server/mod.rs b/grpc-protobuf/src/server/mod.rs index 88a8349eb..f7edbb633 100644 --- a/grpc-protobuf/src/server/mod.rs +++ b/grpc-protobuf/src/server/mod.rs @@ -38,10 +38,10 @@ use protobuf::MessageView; use crate::ProtoRecvMessage; use crate::ProtoSendMessage; -pub(crate) mod bidi; -pub(crate) mod client_streaming; -pub(crate) mod server_streaming; -pub(crate) mod unary; +mod bidi; +mod client_streaming; +mod server_streaming; +mod unary; pub use bidi::*; pub use client_streaming::*; From 6dbc44d9f46f251a8324ddd6378866c99cc9f26a Mon Sep 17 00:00:00 2001 From: Arjan Bal Date: Wed, 19 Aug 2026 18:35:10 +0530 Subject: [PATCH 6/7] fix lifetime workaround --- grpc-protobuf/src/lib.rs | 58 ++++++++++++++++++++ grpc-protobuf/src/server/bidi.rs | 4 +- grpc-protobuf/src/server/client_streaming.rs | 7 ++- grpc-protobuf/src/server/mod.rs | 3 +- grpc-protobuf/src/server/server_streaming.rs | 6 +- grpc-protobuf/src/server/unary.rs | 10 +++- grpc/src/inmemory/mod.rs | 3 +- grpc/src/send_future.rs | 1 + grpc/src/server/mod.rs | 30 ++-------- 9 files changed, 83 insertions(+), 39 deletions(-) diff --git a/grpc-protobuf/src/lib.rs b/grpc-protobuf/src/lib.rs index d2c2ca127..fcdaed19f 100644 --- a/grpc-protobuf/src/lib.rs +++ b/grpc-protobuf/src/lib.rs @@ -147,3 +147,61 @@ impl<'a, M: Message> MessageType for ProtoRecvMessage<'a, M> { mod private { pub struct Internal; } + +/// A helper trait to enforce and explicitly bound a [`Future`] as [`Send`]. +/// +/// This trait provides a mechanism to work around specific Rust compiler +/// limitations and bugs where the compiler's borrow checker or drop analysis +/// conservatively concludes that an `async` block is `!Send` (not safe to send +/// across threads), +/// even when it logically should be. +/// +/// # Problem Context +/// +/// As detailed in issues [#64552], [#102211], and [#96865], there are scenarios +/// where: +/// * An `async` function captures a reference to a type that is `!Sync`. +/// * A variable is dropped before an `.await` point, but the compiler's liveness +/// analysis incorrectly believes it is held across the await. +/// * Complex control flow confuses the auto-trait deduction for `Send`. +/// +/// These scenarios often result in obscure error messages when trying to spawn +/// the future on an executor (like `tokio::spawn`), claiming the future is not +/// `Send`. +/// +/// # The Solution +/// +/// The `make_send()` method acts as an identity function (a no-op at runtime) but +/// performs two critical compile-time tasks: +/// +/// 1. **Explicit Assertion:** It requires `Self` to implement `Send` at the +/// call site. This moves the error message from the deep internals of an +/// executor's spawn function to the specific line where the future is created, +/// making debugging significantly easier. +/// 2. **Type Erasure / Coercion:** By returning `impl Future<...> + Send`, it +/// creates an opaque type boundary. This can sometimes help the compiler's +/// trait solver "lock in" the `Send` guarantee and disregard phantom lifetime +/// issues that might otherwise propagate. +/// +/// [#64552]: https://github.com/rust-lang/rust/issues/64552 +/// [#102211]: https://github.com/rust-lang/rust/issues/102211 +/// [#96865]: https://github.com/rust-lang/rust/issues/96865 +/// [`Future`]: core::future::Future +/// [`Send`]: core::marker::Send +// TODO: delete this type once MSRV is v1.92. +trait SendFuture: Future { + /// Consumes the future and returns it as an opaque type that is guaranteed + /// to be [`Send`]. + /// + /// This is a zero-cost abstraction (it simply returns `self`) used primarily + /// to help the compiler resolve auto-traits or to produce better error + /// diagnostics. + fn make_send(self) -> impl Future + Send + where + Self: Sized + Send, + { + self + } +} + +impl SendFuture for T {} diff --git a/grpc-protobuf/src/server/bidi.rs b/grpc-protobuf/src/server/bidi.rs index db18f235d..8cae039d7 100644 --- a/grpc-protobuf/src/server/bidi.rs +++ b/grpc-protobuf/src/server/bidi.rs @@ -23,9 +23,9 @@ */ use grpc::async_trait; -use grpc::server::BoxedRecvStream; use grpc::server::CallOptions; use grpc::server::DynHandle; +use grpc::server::DynRecvStream; use grpc::server::DynSendStream; use grpc::server::RequestHeaders; use grpc::server::Trailers; @@ -88,7 +88,7 @@ where _headers: RequestHeaders, _options: CallOptions, tx: &mut dyn DynSendStream, - rx: BoxedRecvStream, + rx: Box, ) -> Trailers { // The request stream owns `rx`; the response sink borrows `tx`. They // are independent, so a handler can freely interleave receives and diff --git a/grpc-protobuf/src/server/client_streaming.rs b/grpc-protobuf/src/server/client_streaming.rs index a9c9b7b52..7d4b8d1a0 100644 --- a/grpc-protobuf/src/server/client_streaming.rs +++ b/grpc-protobuf/src/server/client_streaming.rs @@ -23,9 +23,9 @@ */ use grpc::async_trait; -use grpc::server::BoxedRecvStream; use grpc::server::CallOptions; use grpc::server::DynHandle; +use grpc::server::DynRecvStream; use grpc::server::DynSendStream; use grpc::server::RequestHeaders; use grpc::server::ResponseStreamItem; @@ -39,6 +39,7 @@ use protobuf::Proxied; use protobuf::Serialize; use crate::ProtoSendMessage; +use crate::SendFuture; use crate::ServerStatus; use crate::server::GrpcStreamingRequest; use crate::trailers_conv::trailers_from_status; @@ -91,11 +92,11 @@ where _headers: RequestHeaders, _options: CallOptions, tx: &mut dyn DynSendStream, - rx: BoxedRecvStream, + rx: Box, ) -> Trailers { let requests = GrpcStreamingRequest::new(rx); let mut resp = ::default(); - let status = self.method.call(requests, resp.as_mut()).await; + let status = self.method.call(requests, resp.as_mut()).make_send().await; if status.is_ok() { let send = ProtoSendMessage::from_view(&resp); diff --git a/grpc-protobuf/src/server/mod.rs b/grpc-protobuf/src/server/mod.rs index f7edbb633..7ba7cd44b 100644 --- a/grpc-protobuf/src/server/mod.rs +++ b/grpc-protobuf/src/server/mod.rs @@ -24,7 +24,6 @@ use std::marker::PhantomData; -use grpc::server::BoxedRecvStream; use grpc::server::DynRecvStream; use grpc::server::DynSendStream; use grpc::server::ResponseStreamItem; @@ -54,6 +53,8 @@ pub struct GrpcStreamingRequest { _phantom: PhantomData, } +type BoxedRecvStream = Box; + impl GrpcStreamingRequest where M: Message, diff --git a/grpc-protobuf/src/server/server_streaming.rs b/grpc-protobuf/src/server/server_streaming.rs index 41c747aa5..ae3080ac0 100644 --- a/grpc-protobuf/src/server/server_streaming.rs +++ b/grpc-protobuf/src/server/server_streaming.rs @@ -23,7 +23,6 @@ */ use grpc::async_trait; -use grpc::server::BoxedRecvStream; use grpc::server::CallOptions; use grpc::server::DynHandle; use grpc::server::DynRecvStream; @@ -38,6 +37,7 @@ use protobuf::Proxied; use protobuf::Serialize; use crate::ProtoRecvMessage; +use crate::SendFuture; use crate::ServerStatus; use crate::ServerStatusError; use crate::StatusCodeError; @@ -92,7 +92,7 @@ where _headers: RequestHeaders, _options: CallOptions, tx: &mut dyn DynSendStream, - mut rx: BoxedRecvStream, + mut rx: Box, ) -> Trailers { let mut req = ::default(); @@ -108,7 +108,7 @@ where } let responses = GrpcStreamingResponse::new(tx); - let status = self.method.call(req.as_view(), responses).await; + let status = self.method.call(req.as_view(), responses).make_send().await; trailers_from_status(status) } } diff --git a/grpc-protobuf/src/server/unary.rs b/grpc-protobuf/src/server/unary.rs index eb3d3161f..f12c65ee9 100644 --- a/grpc-protobuf/src/server/unary.rs +++ b/grpc-protobuf/src/server/unary.rs @@ -23,7 +23,6 @@ */ use grpc::async_trait; -use grpc::server::BoxedRecvStream; use grpc::server::CallOptions; use grpc::server::DynHandle; use grpc::server::DynRecvStream; @@ -42,6 +41,7 @@ use protobuf::Serialize; use crate::ProtoRecvMessage; use crate::ProtoSendMessage; +use crate::SendFuture; use crate::ServerStatus; use crate::ServerStatusError; use crate::StatusCodeError; @@ -94,7 +94,7 @@ where _headers: RequestHeaders, _options: CallOptions, tx: &mut dyn DynSendStream, - mut rx: BoxedRecvStream, + mut rx: Box, ) -> Trailers { let mut req = ::default(); @@ -110,7 +110,11 @@ where } let mut resp = ::default(); - let status = self.method.call(req.as_view(), resp.as_mut()).await; + let status = self + .method + .call(req.as_view(), resp.as_mut()) + .make_send() + .await; if status.is_ok() { let send = ProtoSendMessage::from_view(&resp); diff --git a/grpc/src/inmemory/mod.rs b/grpc/src/inmemory/mod.rs index 8aed3475a..224fc11ed 100644 --- a/grpc/src/inmemory/mod.rs +++ b/grpc/src/inmemory/mod.rs @@ -70,7 +70,6 @@ use crate::core::SendMessage; use crate::credentials::SecurityInfo; use crate::credentials::SecurityLevel; use crate::rt::GrpcRuntime; -use crate::server::BoxedRecvStream; use crate::server::DynHandle; use crate::server::GracefulConnection; use crate::server::Listener; @@ -234,7 +233,7 @@ impl ServerTransport for InMemoryServerCall { _token: crate::private::Internal, ) -> InMemoryServingConnection { let mut send = InMemoryServerSendStream { tx: self.resp_tx }; - let recv = BoxedRecvStream(Box::new(InMemoryServerRecvStream { rx: self.req_rx })); + let recv = Box::new(InMemoryServerRecvStream { rx: self.req_rx }); let options = crate::server::CallOptions::default(); let trailers_tx = self.trailer_tx; diff --git a/grpc/src/send_future.rs b/grpc/src/send_future.rs index 14a850107..70c537f4c 100644 --- a/grpc/src/send_future.rs +++ b/grpc/src/send_future.rs @@ -64,6 +64,7 @@ use core::future::Future; /// [#96865]: https://github.com/rust-lang/rust/issues/96865 /// [`Future`]: core::future::Future /// [`Send`]: core::marker::Send +// TODO: delete this type once MSRV is v1.92. pub trait SendFuture: Future { /// Consumes the future and returns it as an opaque type that is guaranteed /// to be [`Send`]. diff --git a/grpc/src/server/mod.rs b/grpc/src/server/mod.rs index ed0ce83ee..e2401ddfe 100644 --- a/grpc/src/server/mod.rs +++ b/grpc/src/server/mod.rs @@ -53,6 +53,7 @@ use crate::core::RecvMessage; use crate::core::SendMessage; use crate::metadata::MetadataMap; use crate::rt::GrpcRuntime; +use crate::send_future::SendFuture; pub(crate) mod interceptor; @@ -309,30 +310,11 @@ impl DynHandle for T { mut tx: &mut dyn DynSendStream, rx: BoxedRecvStream, ) -> Trailers { - self.handle(headers, options, &mut tx, rx).await + self.handle(headers, options, &mut tx, rx).make_send().await } } -// TODO: delete this type which is only needed pre-rust v1.92 due to a bug -// handling lifetimes: -// -// error: implementation of `server::RecvStream` is not general enough -// --> grpc/src/server/mod.rs:108:5 -// | -// 108 | async fn dyn_handle( -// | ^^^^^ implementation of `server::RecvStream` is not general enough -// | -// = note: `Box<(dyn server::DynRecvStream + '0)>` must implement `server::RecvStream`, for any lifetime `'0`... -// = note: ...but `server::RecvStream` is actually implemented for the type `Box<(dyn server::DynRecvStream + 'static)>` -#[doc(hidden)] -pub struct BoxedRecvStream(pub Box); - -// Implement RecvStream for the wrapper instead of the Box directly -impl RecvStream for BoxedRecvStream { - async fn next(&mut self, msg: &mut dyn RecvMessage) -> Option> { - self.0.dyn_next(msg).await - } -} +pub(crate) type BoxedRecvStream = Box; /// An item in a response stream from the server's view. /// @@ -357,9 +339,7 @@ impl Handle for DynHandleWrapper { tx: &mut impl SendStream, rx: impl RecvStream + 'static, ) -> Trailers { - self.0 - .dyn_handle(headers, options, tx, BoxedRecvStream(Box::new(rx))) - .await + self.0.dyn_handle(headers, options, tx, Box::new(rx)).await } } /// Represents the sending side of a server stream. See `ResponseStream` @@ -945,7 +925,7 @@ mod tests { ) -> Self::Connection { let inner = Box::pin(async move { let mut tx = NopSendStream; - let rx = BoxedRecvStream(Box::new(NopRecvStream)); + let rx = Box::new(NopRecvStream); let _ = handler .dyn_handle(RequestHeaders::new(), CallOptions::new(), &mut tx, rx) .await; From 6ca5f3a40c072c0678dfa8b772362e801cfe184b Mon Sep 17 00:00:00 2001 From: Arjan Bal Date: Wed, 19 Aug 2026 18:53:43 +0530 Subject: [PATCH 7/7] remove redundant bounds --- grpc-protobuf/src/server/bidi.rs | 10 ++-------- grpc-protobuf/src/server/client_streaming.rs | 9 ++------- grpc-protobuf/src/server/mod.rs | 4 ---- grpc-protobuf/src/server/server_streaming.rs | 9 ++------- grpc-protobuf/src/server/unary.rs | 8 ++------ 5 files changed, 8 insertions(+), 32 deletions(-) diff --git a/grpc-protobuf/src/server/bidi.rs b/grpc-protobuf/src/server/bidi.rs index 8cae039d7..da9a5feb5 100644 --- a/grpc-protobuf/src/server/bidi.rs +++ b/grpc-protobuf/src/server/bidi.rs @@ -29,11 +29,7 @@ use grpc::server::DynRecvStream; use grpc::server::DynSendStream; use grpc::server::RequestHeaders; use grpc::server::Trailers; -use protobuf::ClearAndParse; use protobuf::Message; -use protobuf::MutProxied; -use protobuf::Proxied; -use protobuf::Serialize; use crate::ServerStatus; use crate::server::GrpcStreamingRequest; @@ -47,9 +43,9 @@ use crate::trailers_conv::trailers_from_status; #[trait_variant::make(Send)] pub trait BidiStreamingMethod: Sync + 'static { /// The protobuf request message type. - type Request: Message + Default; + type Request: Message; /// The protobuf response message type. - type Response: Message + Default; + type Response: Message; /// Handles a bidirectional-streaming RPC call. /// @@ -80,8 +76,6 @@ impl BidiStreamingAdapter { impl DynHandle for BidiStreamingAdapter where M: BidiStreamingMethod, - for<'a> ::Mut<'a>: ClearAndParse + Send + Sync, - for<'a> ::View<'a>: Serialize + Send + Sync, { async fn dyn_handle( &self, diff --git a/grpc-protobuf/src/server/client_streaming.rs b/grpc-protobuf/src/server/client_streaming.rs index 7d4b8d1a0..ac5be526b 100644 --- a/grpc-protobuf/src/server/client_streaming.rs +++ b/grpc-protobuf/src/server/client_streaming.rs @@ -32,11 +32,8 @@ use grpc::server::ResponseStreamItem; use grpc::server::SendOptions; use grpc::server::Trailers; use protobuf::AsMut; -use protobuf::ClearAndParse; use protobuf::Message; use protobuf::MutProxied; -use protobuf::Proxied; -use protobuf::Serialize; use crate::ProtoSendMessage; use crate::SendFuture; @@ -51,9 +48,9 @@ use crate::trailers_conv::trailers_from_status; #[trait_variant::make(Send)] pub trait ClientStreamingMethod: Sync + 'static { /// The protobuf request message type. - type Request: Message + Default; + type Request: Message; /// The protobuf response message type. - type Response: Message + Default; + type Response: Message; /// Handles a client-streaming RPC call. /// @@ -84,8 +81,6 @@ impl ClientStreamingAdapter { impl DynHandle for ClientStreamingAdapter where M: ClientStreamingMethod, - for<'a> ::Mut<'a>: ClearAndParse + Send + Sync, - for<'a> ::View<'a>: Serialize + Send + Sync, { async fn dyn_handle( &self, diff --git a/grpc-protobuf/src/server/mod.rs b/grpc-protobuf/src/server/mod.rs index 7ba7cd44b..8e704e4ae 100644 --- a/grpc-protobuf/src/server/mod.rs +++ b/grpc-protobuf/src/server/mod.rs @@ -31,8 +31,6 @@ use grpc::server::SendOptions; use protobuf::AsMut; use protobuf::AsView; use protobuf::Message; -use protobuf::MessageMut; -use protobuf::MessageView; use crate::ProtoRecvMessage; use crate::ProtoSendMessage; @@ -58,7 +56,6 @@ type BoxedRecvStream = Box; impl GrpcStreamingRequest where M: Message, - for<'b> M::Mut<'b>: MessageMut<'b>, { /// Creates a new [`GrpcStreamingRequest`]. pub(crate) fn new(rx: BoxedRecvStream) -> Self { @@ -103,7 +100,6 @@ pub struct GrpcStreamingResponse<'a, M> { impl<'a, M> GrpcStreamingResponse<'a, M> where M: Message, - for<'b> M::View<'b>: MessageView<'b>, { pub(crate) fn new(tx: &'a mut dyn DynSendStream) -> Self { Self { diff --git a/grpc-protobuf/src/server/server_streaming.rs b/grpc-protobuf/src/server/server_streaming.rs index ae3080ac0..0fd802c97 100644 --- a/grpc-protobuf/src/server/server_streaming.rs +++ b/grpc-protobuf/src/server/server_streaming.rs @@ -30,11 +30,8 @@ use grpc::server::DynSendStream; use grpc::server::RequestHeaders; use grpc::server::Trailers; use protobuf::AsView; -use protobuf::ClearAndParse; use protobuf::Message; -use protobuf::MutProxied; use protobuf::Proxied; -use protobuf::Serialize; use crate::ProtoRecvMessage; use crate::SendFuture; @@ -51,9 +48,9 @@ use crate::trailers_conv::trailers_from_status; #[trait_variant::make(Send)] pub trait ServerStreamingMethod: Sync + 'static { /// The protobuf request message type. - type Request: Message + Default; + type Request: Message; /// The protobuf response message type. - type Response: Message + Default; + type Response: Message; /// Handles a server-streaming RPC call. /// @@ -84,8 +81,6 @@ impl ServerStreamingAdapter { impl DynHandle for ServerStreamingAdapter where M: ServerStreamingMethod, - for<'a> ::Mut<'a>: ClearAndParse + Send + Sync, - for<'a> ::View<'a>: Serialize + Send + Sync, { async fn dyn_handle( &self, diff --git a/grpc-protobuf/src/server/unary.rs b/grpc-protobuf/src/server/unary.rs index f12c65ee9..1a40c8ead 100644 --- a/grpc-protobuf/src/server/unary.rs +++ b/grpc-protobuf/src/server/unary.rs @@ -33,11 +33,9 @@ use grpc::server::SendOptions; use grpc::server::Trailers; use protobuf::AsMut; use protobuf::AsView; -use protobuf::ClearAndParse; use protobuf::Message; use protobuf::MutProxied; use protobuf::Proxied; -use protobuf::Serialize; use crate::ProtoRecvMessage; use crate::ProtoSendMessage; @@ -54,9 +52,9 @@ use crate::trailers_conv::trailers_from_status; #[trait_variant::make(Send)] pub trait UnaryMethod: Sync + 'static { /// The protobuf request message type. - type Request: Message + Default; + type Request: Message; /// The protobuf response message type. - type Response: Message + Default; + type Response: Message; /// Handles a unary RPC call. /// @@ -86,8 +84,6 @@ impl UnaryAdapter { impl DynHandle for UnaryAdapter where M: UnaryMethod, - for<'a> ::Mut<'a>: ClearAndParse + Send + Sync, - for<'a> ::View<'a>: Serialize + Send + Sync, { async fn dyn_handle( &self,