feat(codec): add compression API, standard algorithms, and global registry

This change introduces the foundational compression abstractions and
implementations for the grpc server component.

- **Compression API (`grpc/src/codec/compression.rs`)**: Defines the
  `Compressor` and `Decompressor` traits for handling the compression
  and decompression of gRPC payloads via byte buffers.
- **Simplified Design Philosophy**: The API is intentionally kept
  simple. It deliberately avoids handling concerns like limiting input
  and output buffer sizes, instead delegating these safety boundaries to
  the higher-level gRPC business logic.
- **Performance Improvements**: By relying on direct buffer
  manipulation, this implementation avoids the hidden memory copying
  overhead present in the `tonic` crate (which utilizes 8KB intermediate
  buffers for I/O).
- **Standard Algorithms**: Implements `gzip`, `deflate`, and `zstd`
  compression logic in their respective modules. These are conditionally
  compiled based on feature flags.
- **Global Registry (`registry.rs`)**: Introduces a thread-safe
  `GlobalCompressionRegistry` (backed by `RwLock` and `Arc`) to manage
  registered compressors and decompressors. It handles safe concurrent
  access and automatically updates the broadcasted
  `grpc-accept-encoding` headers.
- **Module Integration**: Exposes the new `codec` and `compression`
  modules to the library's root structure.

Author: sauravzg

grpc/Cargo.toml

@@ -41,10 +41,15 @@ tls-rustls = [
     "dep:rustls-platform-verifier",
     "dep:rustls-webpki",
 ]
+deflate = ["dep:flate2"]
+gzip = ["dep:flate2"]
+zstd = ["dep:zstd"]
 
 [dependencies]
 base64 = "0.22"
 bytes = "1.10.1"
+flate2 = { version = "1.0", optional = true }
+zstd = { version = "0.13", optional = true }
 futures = { version = "0.3", default-features = false, optional = true }
 hickory-resolver = { version = "0.25.1", optional = true }
 http = "1.1.0"

grpc/src/codec.rs

@@ -0,0 +1 @@
+pub(crate) mod compression;

grpc/src/codec/compression.rs

@@ -0,0 +1,48 @@
+use std::io;
+
+use bytes::Buf;
+use bytes::BufMut;
+
+#[cfg(feature = "deflate")]
+mod deflate;
+#[cfg(feature = "gzip")]
+mod gzip;
+#[cfg(feature = "zstd")]
+mod zstd;
+
+pub(crate) mod registry;
+
+/// A trait for compressing outgoing gRPC payloads.
+pub trait Compressor: Send + Sync + 'static {
+    /// The canonical gRPC content coding name (e.g., "gzip").
+    fn name(&self) -> &'static str;
+
+    /// Compress data from `source` into `destination`.
+    ///
+    /// # Errors
+    ///
+    /// Returns an `io::Error` if compression fails. Implementations should gracefully
+    /// handle constrained `destination` buffers by returning an error rather than panicking
+    /// (e.g., by verifying `destination.remaining_mut()` is sufficient before writing).
+    fn compress(&self, source: &mut dyn Buf, destination: &mut dyn BufMut)
+    -> Result<(), io::Error>;
+}
+
+/// A trait for decompressing incoming gRPC payloads.
+pub trait Decompressor: Send + Sync + 'static {
+    /// The canonical gRPC content coding name (e.g., "gzip").
+    fn name(&self) -> &'static str;
+
+    /// Decompress data from `source` into `destination`.
+    ///
+    /// # Errors
+    ///
+    /// Returns an `io::Error` if decompression fails. Implementations should gracefully
+    /// handle constrained `destination` buffers by returning an error rather than panicking
+    /// (e.g., by verifying `destination.remaining_mut()` is sufficient before writing).
+    fn decompress(
+        &self,
+        source: &mut dyn Buf,
+        destination: &mut dyn BufMut,
+    ) -> Result<(), io::Error>;
+}

grpc/src/codec/compression/deflate.rs

@@ -0,0 +1,96 @@
+use std::io::Write;
+use std::io::{self};
+
+use bytes::Buf;
+use bytes::BufMut;
+use flate2::Compression as FlateCompression;
+use flate2::write::ZlibDecoder;
+use flate2::write::ZlibEncoder;
+
+use crate::codec::compression::Compressor;
+use crate::codec::compression::Decompressor;
+
+/// A deflate compression implementation.
+#[derive(Debug, Clone, Copy)]
+pub struct Deflate {
+    level: FlateCompression,
+}
+
+impl Deflate {
+    /// Creates a new deflate compression implementation.
+    pub fn new() -> Self {
+        Self {
+            level: FlateCompression::new(6),
+        }
+    }
+}
+
+impl Default for Deflate {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Compressor for Deflate {
+    fn name(&self) -> &'static str {
+        "deflate"
+    }
+
+    fn compress(
+        &self,
+        source: &mut dyn Buf,
+        destination: &mut dyn BufMut,
+    ) -> Result<(), io::Error> {
+        let mut encoder = ZlibEncoder::new(destination.writer(), self.level);
+        while source.has_remaining() {
+            let chunk = source.chunk();
+            encoder.write_all(chunk)?;
+            source.advance(chunk.len());
+        }
+        encoder.finish()?;
+        Ok(())
+    }
+}
+
+impl Decompressor for Deflate {
+    fn name(&self) -> &'static str {
+        "deflate"
+    }
+
+    fn decompress(
+        &self,
+        source: &mut dyn Buf,
+        destination: &mut dyn BufMut,
+    ) -> Result<(), io::Error> {
+        let mut decoder = ZlibDecoder::new(destination.writer());
+        while source.has_remaining() {
+            let chunk = source.chunk();
+            decoder.write_all(chunk)?;
+            source.advance(chunk.len());
+        }
+        decoder.finish()?;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use bytes::Bytes;
+
+    use super::*;
+
+    #[test]
+    fn deflate_compress_decompress() {
+        let compressor = Deflate::new();
+        let data = Bytes::from_static(b"hello world");
+        let mut compressed = Vec::new();
+        compressor
+            .compress(&mut data.clone(), &mut compressed)
+            .unwrap();
+        let mut decompressed = Vec::new();
+        compressor
+            .decompress(&mut compressed.as_slice(), &mut decompressed)
+            .unwrap();
+        assert_eq!(data, decompressed.as_slice());
+    }
+}

grpc/src/codec/compression/gzip.rs

@@ -0,0 +1,96 @@
+use std::io::Write;
+use std::io::{self};
+
+use bytes::Buf;
+use bytes::BufMut;
+use flate2::Compression as FlateCompression;
+use flate2::write::GzDecoder;
+use flate2::write::GzEncoder;
+
+use crate::codec::compression::Compressor;
+use crate::codec::compression::Decompressor;
+
+/// A gzip compression implementation.
+#[derive(Debug, Clone, Copy)]
+pub struct Gzip {
+    level: FlateCompression,
+}
+
+impl Gzip {
+    /// Creates a new gzip compression implementation.
+    pub fn new() -> Self {
+        Self {
+            level: FlateCompression::new(6),
+        }
+    }
+}
+
+impl Default for Gzip {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Compressor for Gzip {
+    fn name(&self) -> &'static str {
+        "gzip"
+    }
+
+    fn compress(
+        &self,
+        source: &mut dyn Buf,
+        destination: &mut dyn BufMut,
+    ) -> Result<(), io::Error> {
+        let mut encoder = GzEncoder::new(destination.writer(), self.level);
+        while source.has_remaining() {
+            let chunk = source.chunk();
+            encoder.write_all(chunk)?;
+            source.advance(chunk.len());
+        }
+        encoder.finish()?;
+        Ok(())
+    }
+}
+
+impl Decompressor for Gzip {
+    fn name(&self) -> &'static str {
+        "gzip"
+    }
+
+    fn decompress(
+        &self,
+        source: &mut dyn Buf,
+        destination: &mut dyn BufMut,
+    ) -> Result<(), io::Error> {
+        let mut decoder = GzDecoder::new(destination.writer());
+        while source.has_remaining() {
+            let chunk = source.chunk();
+            decoder.write_all(chunk)?;
+            source.advance(chunk.len());
+        }
+        decoder.finish()?;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use bytes::Bytes;
+
+    use super::*;
+
+    #[test]
+    fn gzip_compress_decompress() {
+        let compressor = Gzip::new();
+        let data = Bytes::from_static(b"hello world");
+        let mut compressed = Vec::new();
+        compressor
+            .compress(&mut data.clone(), &mut compressed)
+            .unwrap();
+        let mut decompressed = Vec::new();
+        compressor
+            .decompress(&mut compressed.as_slice(), &mut decompressed)
+            .unwrap();
+        assert_eq!(data, decompressed.as_slice());
+    }
+}