Add transport compression for UploadPath on server

server/Cargo.toml

@@ -44,6 +44,7 @@ humantime = "2.1.0"
 humantime-serde = "1.1.1"
 itoa = "=1.0.5"
 maybe-owned = "0.3.4"
+pin-project = "1.0.12"
 rand = "0.8.5"
 regex = "1.7.1"
 ryu = "1.0.13"
@@ -68,6 +69,8 @@ features = [
 	"xz",
 	"zstd",
 	"brotli",
+	"deflate",
+	"gzip"
 ]
 
 [dependencies.sea-orm]

server/src/api/v1/upload_path.rs

@@ -29,6 +29,7 @@ use tracing::instrument;
 use uuid::Uuid;
 
 use crate::config::CompressionType;
+use crate::decompression::StreamingDecompressor;
 use crate::error::{ErrorKind, ServerError, ServerResult};
 use crate::narinfo::Compression;
 use crate::{RequestState, State};
@@ -120,10 +121,17 @@ pub(crate) async fn upload_path(
     headers: HeaderMap,
     stream: BodyStream,
 ) -> ServerResult<Json<UploadPathResult>> {
-    let mut stream = StreamReader::new(
+    let compression_format = headers
+        .get("Content-Encoding")
+        .and_then(|e| e.to_str().ok())
+        .unwrap_or("");
+
+    let stream = StreamReader::new(
         stream.map(|r| r.map_err(|e| io::Error::new(io::ErrorKind::Other, e.to_string()))),
     );
 
+    let mut stream = StreamingDecompressor::new_unbuffered(stream, compression_format)?;
+
     let upload_info: UploadPathNarInfo = {
         if let Some(preamble_size_bytes) = headers.get(ATTIC_NAR_INFO_PREAMBLE_SIZE) {
             // Read from the beginning of the PUT body

server/src/decompression.rs

@@ -0,0 +1,85 @@
+//! Module for implementing streaming decompression across multiple
+//! algorithms
+
+use std::{
+    io,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use anyhow::anyhow;
+use async_compression::tokio::bufread::{
+    BrotliDecoder, DeflateDecoder, GzipDecoder, XzDecoder, ZstdDecoder,
+};
+use pin_project::pin_project;
+use tokio::io::{AsyncBufRead, AsyncRead, BufReader, ReadBuf};
+
+use crate::error::{ErrorKind, ServerResult};
+
+/// A streaming multi-codec decompressor
+#[pin_project(project = SDProj)]
+pub enum StreamingDecompressor<S: AsyncBufRead> {
+    /// None decompression
+    None(#[pin] S),
+    /// Brotli decompression
+    Brotli(#[pin] BrotliDecoder<S>),
+    /// Deflate decompression
+    Deflate(#[pin] DeflateDecoder<S>),
+    /// Gzip decompression
+    Gzip(#[pin] GzipDecoder<S>),
+    /// XZ decompression
+    Xz(#[pin] XzDecoder<S>),
+    /// Zstd decompression
+    Zstd(#[pin] ZstdDecoder<S>),
+}
+
+impl<S: AsyncBufRead> StreamingDecompressor<S> {
+    /// Creates a new streaming decompressor from a buffered stream and compression type.
+    ///
+    /// An empty string or "identity" corresponds to no decompression.
+    ///
+    /// # Errors
+    /// This function will return an error if the compression type is invalid
+    pub fn new(inner: S, kind: &str) -> ServerResult<Self> {
+        match kind {
+            "" | "identity" => Ok(Self::None(inner)),
+            "br" => Ok(Self::Brotli(BrotliDecoder::new(inner))),
+            "deflate" => Ok(Self::Deflate(DeflateDecoder::new(inner))),
+            "gzip" => Ok(Self::Gzip(GzipDecoder::new(inner))),
+            "xz" => Ok(Self::Xz(XzDecoder::new(inner))),
+            "zstd" => Ok(Self::Zstd(ZstdDecoder::new(inner))),
+            _ => Err(ErrorKind::RequestError(anyhow!(
+                "{} is unsupported transport compression",
+                kind
+            ))
+            .into()),
+        }
+    }
+}
+
+impl<U: AsyncRead> StreamingDecompressor<BufReader<U>> {
+    /// Creates a new streaming decompressor from an unbuffered stream and compression type.
+    ///
+    /// # Errors
+    /// This function will return an error if the compression type is invalid
+    pub fn new_unbuffered(inner: U, kind: &str) -> ServerResult<Self> {
+        Self::new(BufReader::new(inner), kind)
+    }
+}
+
+impl<S: AsyncBufRead> AsyncRead for StreamingDecompressor<S> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        match self.project() {
+            SDProj::None(i) => i.poll_read(cx, buf),
+            SDProj::Brotli(i) => i.poll_read(cx, buf),
+            SDProj::Deflate(i) => i.poll_read(cx, buf),
+            SDProj::Gzip(i) => i.poll_read(cx, buf),
+            SDProj::Xz(i) => i.poll_read(cx, buf),
+            SDProj::Zstd(i) => i.poll_read(cx, buf),
+        }
+    }
+}

server/src/lib.rs

@@ -18,6 +18,7 @@ mod api;
 mod chunking;
 pub mod config;
 pub mod database;
+mod decompression;
 pub mod error;
 pub mod gc;
 mod middleware;