grpc-protobuf: add server codegen utils

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]
 

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::*;
@@ -146,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<Output = Self::Output> + Send
+    where
+        Self: Sized + Send,
+    {
+        self
+    }
+}
+
+impl<T: Future> SendFuture for T {}

grpc-protobuf/src/server/bidi.rs

@@ -0,0 +1,95 @@
+/*
+ *
+ * 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::CallOptions;
+use grpc::server::DynHandle;
+use grpc::server::DynRecvStream;
+use grpc::server::DynSendStream;
+use grpc::server::RequestHeaders;
+use grpc::server::Trailers;
+use protobuf::Message;
+
+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;
+    /// The protobuf response message type.
+    type Response: Message;
+
+    /// 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<Self::Request>,
+        responses: GrpcStreamingResponse<'_, Self::Response>,
+    ) -> ServerStatus;
+}
+
+/// An adapter that wraps a [`BidiStreamingMethod`] to handle incoming
+/// bidirectional-streaming RPCs.
+pub struct BidiStreamingAdapter<M: BidiStreamingMethod> {
+    method: M,
+}
+
+impl<M: BidiStreamingMethod> BidiStreamingAdapter<M> {
+    /// Creates a new [`BidiStreamingAdapter`] wrapping the given `method`.
+    pub fn new(method: M) -> Self {
+        Self { method }
+    }
+}
+
+#[async_trait]
+impl<M> DynHandle for BidiStreamingAdapter<M>
+where
+    M: BidiStreamingMethod,
+{
+    async fn dyn_handle(
+        &self,
+        _headers: RequestHeaders,
+        _options: CallOptions,
+        tx: &mut dyn DynSendStream,
+        rx: Box<dyn DynRecvStream>,
+    ) -> 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)
+    }
+}

grpc-protobuf/src/server/client_streaming.rs

@@ -0,0 +1,107 @@
+/*
+ *
+ * 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::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::Message;
+use protobuf::MutProxied;
+
+use crate::ProtoSendMessage;
+use crate::SendFuture;
+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;
+    /// The protobuf response message type.
+    type Response: Message;
+
+    /// 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<Self::Request>,
+        response: <Self::Response as MutProxied>::Mut<'_>,
+    ) -> ServerStatus;
+}
+
+/// An adapter that wraps a [`ClientStreamingMethod`] to handle incoming
+/// client-streaming RPCs.
+pub struct ClientStreamingAdapter<M: ClientStreamingMethod> {
+    method: M,
+}
+
+impl<M: ClientStreamingMethod> ClientStreamingAdapter<M> {
+    /// Creates a new [`ClientStreamingAdapter`] wrapping the given `method`.
+    pub fn new(method: M) -> Self {
+        Self { method }
+    }
+}
+
+#[async_trait]
+impl<M> DynHandle for ClientStreamingAdapter<M>
+where
+    M: ClientStreamingMethod,
+{
+    async fn dyn_handle(
+        &self,
+        _headers: RequestHeaders,
+        _options: CallOptions,
+        tx: &mut dyn DynSendStream,
+        rx: Box<dyn DynRecvStream>,
+    ) -> Trailers {
+        let requests = GrpcStreamingRequest::new(rx);
+        let mut resp = <M::Response as Default>::default();
+        let status = self.method.call(requests, resp.as_mut()).make_send().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)
+    }
+}