V0.14.x restitch

Cargo.toml

@@ -9,11 +9,11 @@ members = [
   "tonic-reflection",
   "tonic-prost",
   "tonic-prost-build",
-  "tonic-web", # Non-published crates
+  "tonic-web",
   "examples",
   "codegen",
   "grpc",
-  "interop", # Tests
+  "interop",
   "tests/disable_comments",
   "tests/wellknown",
   "tests/wellknown-compiled",

tests/compression/src/compressing_request.rs

@@ -234,3 +234,71 @@ async fn client_mark_compressed_without_header_server_enabled(encoding: Compress
         "protocol error: received message with compressed-flag but no grpc-encoding was specified"
     );
 }
+
+util::parametrized_tests! {
+    limit_decoded_message_size,
+    zstd: CompressionEncoding::Zstd,
+    gzip: CompressionEncoding::Gzip,
+    deflate: CompressionEncoding::Deflate,
+}
+
+#[cfg(test)]
+async fn limit_decoded_message_size(encoding: CompressionEncoding) {
+    use prost::Message;
+
+    let under_limit_request = SomeData {
+        data: [0_u8; UNCOMPRESSED_MIN_BODY_SIZE].to_vec(),
+    };
+    let limit = under_limit_request.encoded_len();
+    let over_limit_request = SomeData {
+        data: [0_u8; 1 + UNCOMPRESSED_MIN_BODY_SIZE].to_vec(),
+    };
+
+    let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10);
+
+    let svc = test_server::TestServer::new(Svc::default())
+        .accept_compressed(encoding)
+        .max_decoding_message_size(limit);
+
+    let request_bytes_counter = Arc::new(AtomicUsize::new(0));
+
+    tokio::spawn({
+        let request_bytes_counter = request_bytes_counter.clone();
+        async move {
+            Server::builder()
+                .layer(
+                    ServiceBuilder::new()
+                        .layer(
+                            ServiceBuilder::new()
+                                .layer(measure_request_body_size_layer(request_bytes_counter))
+                                .into_inner(),
+                        )
+                        .into_inner(),
+                )
+                .add_service(svc)
+                .serve_with_incoming(tokio_stream::once(Ok::<_, std::io::Error>(server)))
+                .await
+                .unwrap();
+        }
+    });
+
+    let mut client =
+        test_client::TestClient::new(mock_io_channel(client).await).send_compressed(encoding);
+
+    for _ in 0..3 {
+        // compressed messages that are under or exactly at the limit are successful.
+        client
+            .compress_input_unary(under_limit_request.clone())
+            .await
+            .unwrap();
+        let bytes_sent = request_bytes_counter.load(SeqCst);
+        assert!(bytes_sent < UNCOMPRESSED_MIN_BODY_SIZE);
+
+        // compressed messages that are over the limit are fail with resource exhausted
+        let status = client
+            .compress_input_unary(over_limit_request.clone())
+            .await
+            .unwrap_err();
+        assert_eq!(status.code(), tonic::Code::ResourceExhausted);
+    }
+}

tests/compression/src/compressing_response.rs

@@ -488,3 +488,78 @@ async fn disabling_compression_on_response_from_client_stream(encoding: Compress
     let bytes_sent = response_bytes_counter.load(SeqCst);
     assert!(bytes_sent > UNCOMPRESSED_MIN_BODY_SIZE);
 }
+
+util::parametrized_tests! {
+    limit_decoded_message_size,
+    zstd: CompressionEncoding::Zstd,
+    gzip: CompressionEncoding::Gzip,
+    deflate: CompressionEncoding::Deflate,
+}
+
+#[cfg(test)]
+async fn limit_decoded_message_size(encoding: CompressionEncoding) {
+    use prost::Message;
+
+    let under_limit_request = SomeData {
+        data: [0_u8; UNCOMPRESSED_MIN_BODY_SIZE].to_vec(),
+    };
+    let limit = under_limit_request.encoded_len();
+
+    let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10);
+
+    let svc = test_server::TestServer::new(Svc::default()).send_compressed(encoding);
+
+    let response_bytes_counter = Arc::new(AtomicUsize::new(0));
+
+    tokio::spawn({
+        let response_bytes_counter = response_bytes_counter.clone();
+        async move {
+            Server::builder()
+                .layer(
+                    ServiceBuilder::new()
+                        .layer(MapResponseBodyLayer::new(move |body| {
+                            util::CountBytesBody {
+                                inner: body,
+                                counter: response_bytes_counter.clone(),
+                            }
+                        }))
+                        .into_inner(),
+                )
+                .add_service(svc)
+                .serve_with_incoming(tokio_stream::once(Ok::<_, std::io::Error>(server)))
+                .await
+                .unwrap();
+        }
+    });
+
+    let expected = match encoding {
+        CompressionEncoding::Gzip => "gzip",
+        CompressionEncoding::Zstd => "zstd",
+        CompressionEncoding::Deflate => "deflate",
+        _ => panic!("unexpected encoding {encoding:?}"),
+    };
+
+    // compressed messages that are under or exactly at the limit are successful.
+    let mut under_limit_client = test_client::TestClient::new(mock_io_channel(client).await)
+        .accept_compressed(encoding)
+        .max_decoding_message_size(limit);
+
+    for _ in 0..3 {
+        let res = under_limit_client.compress_output_unary(()).await.unwrap();
+        assert_eq!(res.metadata().get("grpc-encoding").unwrap(), expected);
+        let bytes_sent = response_bytes_counter.load(SeqCst);
+        assert!(bytes_sent < UNCOMPRESSED_MIN_BODY_SIZE);
+    }
+
+    // compressed messages that are over the limit are fail with resource exhausted
+    let mut over_limit_client = under_limit_client.max_decoding_message_size(limit - 1);
+
+    for _ in 0..3 {
+        let status = over_limit_client
+            .compress_output_unary(())
+            .await
+            .unwrap_err();
+
+        assert_eq!(status.code(), tonic::Code::ResourceExhausted);
+    }
+}

tonic-build/Cargo.toml

@@ -11,7 +11,7 @@ license = "MIT"
 name = "tonic-build"
 readme = "README.md"
 repository = "https://github.com/hyperium/tonic"
-version = "0.14.2"
+version = "0.14.3"
 rust-version = { workspace = true }
 
 [dependencies]

tonic-build/README.md

@@ -1,13 +1,9 @@
 # tonic-build
 
-Provides code generation for service stubs to use with tonic. For protobuf compilation via prost, use the `tonic-prost-build` crate.
+Provides code generation for service stubs to use with tonic. For protobuf compilation via prost, use the `tonic-prost-build` crate instead.
 
 # Feature flags
 
-- `cleanup-markdown`: Enables cleaning up documentation from the generated code.
-  Useful when documentation of the generated code fails `cargo test --doc` for example.
-  The `prost` feature must be enabled to use this feature.
-- `prost`: Enables usage of prost generator (enabled by default).
 - `transport`: Enables generation of `connect` method using `tonic::transport::Channel`
   (enabled by default).
 
@@ -110,7 +106,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 }
 ```
 
-Then you can reference the generated Rust like this this in your code:
+Then you can reference the generated Rust like this in your code:
 ```rust,ignore
 pub mod api {
     tonic::include_proto!("google.pubsub.v1");

tonic-build/src/lib.rs

@@ -5,7 +5,7 @@
 )]
 #![doc(issue_tracker_base_url = "https://github.com/hyperium/tonic/issues/")]
 #![doc(test(no_crate_inject, attr(deny(rust_2018_idioms))))]
-#![cfg_attr(docsrs, feature(doc_auto_cfg))]
+#![cfg_attr(docsrs, feature(doc_cfg))]
 
 use proc_macro2::{Delimiter, Group, Ident, Literal, Punct, Spacing, Span, TokenStream};
 use quote::TokenStreamExt;

tonic-health/Cargo.toml

@@ -11,7 +11,7 @@ license = "MIT"
 name = "tonic-health"
 readme = "README.md"
 repository = "https://github.com/hyperium/tonic"
-version = "0.14.2"
+version = "0.14.3"
 rust-version = { workspace = true }
 
 [dependencies]

tonic-health/src/lib.rs

@@ -11,7 +11,7 @@
 )]
 #![doc(issue_tracker_base_url = "https://github.com/hyperium/tonic/issues/")]
 #![doc(test(no_crate_inject, attr(deny(rust_2018_idioms))))]
-#![cfg_attr(docsrs, feature(doc_auto_cfg))]
+#![cfg_attr(docsrs, feature(doc_cfg))]
 
 use std::fmt::{Display, Formatter};
 

tonic-prost-build/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tonic-prost-build"
-version = "0.14.2"
+version = "0.14.3"
 authors = ["Lucio Franco <luciofranco14@gmail.com>"]
 edition = "2021"
 license = "MIT"

tonic-prost-build/README.md

@@ -20,7 +20,6 @@ fn main() {
 
 ## Features
 
-- `prost`: Enables prost-based protobuf code generation (enabled by default)
 - `transport`: Enables transport layer code generation
 - `cleanup-markdown`: Enables markdown cleanup in generated documentation
 

tonic-prost/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tonic-prost"
-version = "0.14.2"
+version = "0.14.3"
 authors = ["Lucio Franco <luciofranco14@gmail.com>"]
 edition = "2021"
 license = "MIT"

tonic-reflection/Cargo.toml

@@ -14,7 +14,7 @@ license = "MIT"
 name = "tonic-reflection"
 readme = "README.md"
 repository = "https://github.com/hyperium/tonic"
-version = "0.14.2"
+version = "0.14.3"
 rust-version = { workspace = true }
 
 [package.metadata.docs.rs]

tonic-reflection/src/lib.rs

@@ -5,7 +5,7 @@
 )]
 #![doc(issue_tracker_base_url = "https://github.com/hyperium/tonic/issues/")]
 #![doc(test(no_crate_inject, attr(deny(rust_2018_idioms))))]
-#![cfg_attr(docsrs, feature(doc_auto_cfg))]
+#![cfg_attr(docsrs, feature(doc_cfg))]
 
 mod generated {
     #![allow(unreachable_pub)]

tonic-types/Cargo.toml

@@ -14,7 +14,7 @@ license = "MIT"
 name = "tonic-types"
 readme = "README.md"
 repository = "https://github.com/hyperium/tonic"
-version = "0.14.2"
+version = "0.14.3"
 rust-version = { workspace = true }
 
 [dependencies]

tonic-web/Cargo.toml

@@ -11,7 +11,7 @@ license = "MIT"
 name = "tonic-web"
 readme = "README.md"
 repository = "https://github.com/hyperium/tonic"
-version = "0.14.2"
+version = "0.14.3"
 rust-version = { workspace = true }
 
 [dependencies]

tonic/Cargo.toml

@@ -15,7 +15,7 @@ keywords = ["rpc", "grpc", "async", "futures", "protobuf"]
 license = "MIT"
 readme = "../README.md"
 repository = "https://github.com/hyperium/tonic"
-version = "0.14.2"
+version = "0.14.3"
 rust-version = {workspace = true}
 exclude = ["benches-disabled"]
 
@@ -57,7 +57,7 @@ transport = ["server", "channel"]
 [dependencies]
 base64 = "0.22"
 bytes = "1.0"
-http = "1"
+http = "1.1.0"
 tracing = "0.1"
 
 http-body = "1"

tonic/src/codec/compression.rs

@@ -256,14 +256,17 @@ pub(crate) fn compress(
 pub(crate) fn decompress(
     settings: CompressionSettings,
     compressed_buf: &mut BytesMut,
-    out_buf: &mut BytesMut,
+    mut out_buf: bytes::buf::Limit<&mut BytesMut>,
     len: usize,
 ) -> Result<(), std::io::Error> {
     let buffer_growth_interval = settings.buffer_growth_interval;
     let estimate_decompressed_len = len * 2;
-    let capacity =
-        ((estimate_decompressed_len / buffer_growth_interval) + 1) * buffer_growth_interval;
-    out_buf.reserve(capacity);
+    let capacity = std::cmp::min(
+        bytes::buf::Limit::limit(&out_buf),
+        ((estimate_decompressed_len / buffer_growth_interval) + 1) * buffer_growth_interval,
+    );
+
+    out_buf.get_mut().reserve(capacity);
 
     #[cfg(any(feature = "gzip", feature = "deflate", feature = "zstd"))]
     let mut out_writer = out_buf.writer();

tonic/src/codec/decode.rs

@@ -211,16 +211,25 @@ impl StreamingInner {
 
             let decode_buf = if let Some(encoding) = compression {
                 self.decompress_buf.clear();
+                let limit = self
+                    .max_message_size
+                    .unwrap_or(DEFAULT_MAX_RECV_MESSAGE_SIZE);
+                let limited_out_buf = (&mut self.decompress_buf).limit(limit);
 
                 if let Err(err) = decompress(
                     CompressionSettings {
                         encoding,
                         buffer_growth_interval: buffer_settings.buffer_size,
                     },
                     &mut self.buf,
-                    &mut self.decompress_buf,
+                    limited_out_buf,
                     len,
                 ) {
+                    if matches!(err.kind(), std::io::ErrorKind::WriteZero) {
+                        return Err(Status::resource_exhausted(format!(
+                            "Error decompressing: size limit, of {limit} bytes, exceeded while decompressing message"
+                        )));
+                    }
                     let message = if let Direction::Response(status) = self.direction {
                         format!(
                             "Error decompressing: {err}, while receiving response with status: {status}"

tonic/src/codec/mod.rs

@@ -88,7 +88,7 @@ impl Default for BufferSettings {
     }
 }
 
-// Doc hidden because its used in tests in another crate but not part of the
+// Doc hidden because it's used in tests in another crate but not part of the
 // public api.
 #[doc(hidden)]
 pub const HEADER_SIZE: usize =
@@ -110,7 +110,7 @@ pub trait Codec {
 
     /// The encoder that can encode a message.
     type Encoder: Encoder<Item = Self::Encode, Error = Status> + Send + 'static;
-    /// The encoder that can decode a message.
+    /// The decoder that can decode a message.
     type Decoder: Decoder<Item = Self::Decode, Error = Status> + Send + 'static;
 
     /// Fetch the encoder.

tonic/src/lib.rs

@@ -97,7 +97,7 @@
 )]
 #![doc(issue_tracker_base_url = "https://github.com/hyperium/tonic/issues/")]
 #![doc(test(no_crate_inject, attr(deny(rust_2018_idioms))))]
-#![cfg_attr(docsrs, feature(doc_auto_cfg))]
+#![cfg_attr(docsrs, feature(doc_cfg))]
 
 pub mod body;
 pub mod client;