From 50d143375d3fb2d8ae3c35de8e41065c125bbb59 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 20 Oct 2026 11:46:27 +0000 Subject: [PATCH 1/7] feat: Support gzip compression in server and Rust client Adds Compression::Gzip to the shared types, so the server accepts and stores Content-Encoding: gzip. GCS (and S3-compatible) downloads now send Accept-Encoding: gzip to opt out of GCS decompressive transcoding, which would otherwise return decompressed bytes and ignore Range. The Rust client can compress, decompress, and size-bound gzip uploads. Refs FS-575 --- Cargo.lock | 4 + Cargo.toml | 1 + clients/rust/Cargo.toml | 3 +- clients/rust/src/get.rs | 65 +++++++++- clients/rust/src/many.rs | 52 +++++++- clients/rust/src/put.rs | 90 ++++++++------ clients/rust/tests/e2e.rs | 31 +++++ objectstore-service/Cargo.toml | 1 + objectstore-service/src/backend/gcs.rs | 114 +++++++++++++++++- .../src/backend/s3_compatible.rs | 9 +- objectstore-types/src/metadata.rs | 28 ++--- 11 files changed, 334 insertions(+), 64 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 6bcc5d59..ace31083 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -698,6 +698,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ce2548391e9c1929c21bf6aa2680af86fe4c1b33e6cea9ac1cfeec0bd11218cf" dependencies = [ "compression-core", + "flate2", + "memchr", "zstd", "zstd-safe", ] @@ -2823,6 +2825,7 @@ dependencies = [ "anyhow", "async-compression", "bytes", + "flate2", "futures-util", "http-body-util", "infer", @@ -2957,6 +2960,7 @@ dependencies = [ "blake3", "bytes", "chrono", + "flate2", "futures", "futures-util", "gcp_auth", diff --git a/Cargo.toml b/Cargo.toml index ebf06e3e..e914890f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -52,6 +52,7 @@ console = "0.16.3" ed25519-dalek = { version = "2.2.0" } elegant-departure = "0.3.2" figment = "0.10.19" +flate2 = "1.1.9" futures = "0.3.32" futures-util = "0.3.32" gcp_auth = "0.12.6" diff --git a/clients/rust/Cargo.toml b/clients/rust/Cargo.toml index 85c3bc6b..a9345fc0 100644 --- a/clients/rust/Cargo.toml +++ b/clients/rust/Cargo.toml @@ -14,7 +14,7 @@ publish = true features = ["resumable-upload-api"] [dependencies] -async-compression = { workspace = true, features = ["tokio", "zstd"] } +async-compression = { workspace = true, features = ["gzip", "tokio", "zstd"] } bytes = { workspace = true } futures-util = { workspace = true } infer = { workspace = true } @@ -33,6 +33,7 @@ zstd-safe = { workspace = true } [dev-dependencies] anyhow = { workspace = true } +flate2 = { workspace = true } futures-util = { workspace = true } http-body-util = { workspace = true } jsonwebtoken = { workspace = true } diff --git a/clients/rust/src/get.rs b/clients/rust/src/get.rs index 23459b54..42bfee02 100644 --- a/clients/rust/src/get.rs +++ b/clients/rust/src/get.rs @@ -1,6 +1,6 @@ use std::{fmt, io}; -use async_compression::tokio::bufread::ZstdDecoder; +use async_compression::tokio::bufread::{GzipDecoder, ZstdDecoder}; use bytes::BytesMut; use futures_util::{StreamExt, TryStreamExt}; use objectstore_types::metadata::{Compression, Metadata}; @@ -118,9 +118,9 @@ impl GetBuilder { } } -/// Wraps a stream in a zstd decompression layer. +/// Wraps a stream in a decompression layer matching the object's compression. /// -/// Decompresses if the metadata indicates zstd compression, `decompress` is `true`, +/// Decompresses if the metadata indicates compression, `decompress` is `true`, /// and the stored encoding is not listed in `accept_encoding`. When the stored encoding /// is in `accept_encoding`, the payload is returned compressed and `metadata.compression` /// is preserved. Clears `metadata.compression` when decompression is applied. @@ -134,6 +134,9 @@ pub(crate) fn maybe_decompress( .compression .is_some_and(|c| accept_encoding.contains(&c)); match (metadata.compression, decompress && !encoding_accepted) { + // Multipart uploads with compression, when each part is compressed individually, + // will consist of multiple concatenated zstd frames or gzip members. + // This allows the client to handle automatic decompression for these objects transparently. (Some(Compression::Zstd), true) => { metadata.compression = None; let mut decoder = ZstdDecoder::new(StreamReader::new(stream)); @@ -141,7 +144,13 @@ pub(crate) fn maybe_decompress( decoder.multiple_members(true); ReaderStream::new(decoder).boxed() } - _ => stream, + (Some(Compression::Gzip), true) => { + metadata.compression = None; + let mut decoder = GzipDecoder::new(StreamReader::new(stream)); + decoder.multiple_members(true); + ReaderStream::new(decoder).boxed() + } + (Some(_), false) | (None, _) => stream, } } @@ -242,6 +251,54 @@ mod tests { assert_eq!(metadata.compression, None); } + fn gzip_compress(data: &[u8]) -> Vec { + use std::io::Write as _; + let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default()); + encoder.write_all(data).unwrap(); + encoder.finish().unwrap() + } + + fn gzip_metadata() -> Metadata { + Metadata { + compression: Some(Compression::Gzip), + ..Default::default() + } + } + + #[tokio::test] + async fn gzip_decompresses() { + let payload = b"hello world"; + let stream = raw_stream(&gzip_compress(payload)); + let mut metadata = gzip_metadata(); + + let out = maybe_decompress(stream, &mut metadata, true, &[]); + assert_eq!(collect(out).await, payload); + assert_eq!(metadata.compression, None); + } + + #[tokio::test] + async fn gzip_accept_skips_decompression() { + let compressed = gzip_compress(b"hello world"); + let stream = raw_stream(&compressed); + let mut metadata = gzip_metadata(); + + let out = maybe_decompress(stream, &mut metadata, true, &[Compression::Gzip]); + assert_eq!(collect(out).await, compressed); + assert_eq!(metadata.compression, Some(Compression::Gzip)); + } + + #[tokio::test] + async fn gzip_concatenated_members_decompress() { + let mut compressed = gzip_compress(b"hello "); + compressed.extend(gzip_compress(b"world")); + let stream = raw_stream(&compressed); + let mut metadata = gzip_metadata(); + + let out = maybe_decompress(stream, &mut metadata, true, &[]); + assert_eq!(collect(out).await, b"hello world"); + assert_eq!(metadata.compression, None); + } + #[tokio::test] async fn zstd_concatenated_frames_decompress() { let payload1 = b"hello "; diff --git a/clients/rust/src/many.rs b/clients/rust/src/many.rs index 0f9b12a7..e06c05c9 100644 --- a/clients/rust/src/many.rs +++ b/clients/rust/src/many.rs @@ -531,6 +531,15 @@ async fn send_batch( Ok(results) } +/// Returns the worst-case size of `size` bytes after compressing them with `compression`. +fn compress_bound(compression: Compression, size: usize) -> usize { + match compression { + Compression::Zstd => zstd_safe::compress_bound(size), + // zlib's conservative `deflateBound`, plus 18 bytes of gzip header and trailer. + Compression::Gzip => size + (size >> 12) + (size >> 14) + (size >> 25) + 7 + 18, + } +} + fn classify_fail(key: Option, error: Error) -> Classified { Classified::Failed(OperationResult::Put( key.unwrap_or_else(|| "".to_owned()), @@ -567,9 +576,9 @@ async fn classify(op: BatchOperation) -> Classified { // Client-side compression happens while streaming the body, so the on-wire size is // only known up to its worst case. Precompressed bodies are sent as-is. let size = match (compression, size) { - (Some(CompressionMode::Compress(Compression::Zstd)), Some(size)) => { - usize::try_from(size).ok().map(zstd_safe::compress_bound) - } + (Some(CompressionMode::Compress(compression)), Some(size)) => usize::try_from(size) + .ok() + .map(|size| compress_bound(compression, size)), (Some(CompressionMode::Precompressed(_)) | None, Some(size)) => { usize::try_from(size).ok() } @@ -853,6 +862,43 @@ mod tests { ); } + #[tokio::test] + async fn gzip_put_uses_compress_bound() { + let size = 1_000_000; + let op = put_with_compression(size, CompressionMode::Compress(Compression::Gzip)); + core::assert_matches!( + classify(op).await, + Classified::Batchable(_, s) if s == compress_bound(Compression::Gzip, size) as u64 + ); + } + + #[tokio::test] + async fn gzip_compress_bound_holds_for_incompressible_data() { + use tokio::io::AsyncReadExt as _; + + // A cheap xorshift generator, so the payload does not compress. + let mut state = 0x2545_f491_4f6c_dd1du64; + let payload: Vec = (0..(MAX_BATCH_PART_SIZE as usize)) + .map(|_| { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + state as u8 + }) + .collect(); + + for size in [0, 1, 1_000, 65_536, 300_000, payload.len()] { + let mut encoder = async_compression::tokio::bufread::GzipEncoder::new(&payload[..size]); + let mut compressed = Vec::new(); + encoder.read_to_end(&mut compressed).await.unwrap(); + assert!( + compressed.len() <= compress_bound(Compression::Gzip, size), + "{} > bound for {size}", + compressed.len() + ); + } + } + #[tokio::test] async fn precompressed_put_uses_exact_size() { // A size that would exceed the limit once inflated by `compress_bound`, but is sent diff --git a/clients/rust/src/put.rs b/clients/rust/src/put.rs index 6e4e5ba8..e4f1b1d6 100644 --- a/clients/rust/src/put.rs +++ b/clients/rust/src/put.rs @@ -1,16 +1,17 @@ use std::fmt; use std::io::{self, Cursor}; use std::path::PathBuf; +use std::pin::Pin; use std::{borrow::Cow, collections::BTreeMap}; -use async_compression::tokio::bufread::ZstdEncoder; +use async_compression::tokio::bufread::{GzipEncoder, ZstdEncoder}; use bytes::Bytes; use futures_util::StreamExt; use objectstore_types::metadata::Metadata; use reqwest::Body; use serde::Deserialize; use tokio::fs::File; -use tokio::io::{AsyncRead, BufReader}; +use tokio::io::{AsyncBufRead, AsyncRead, BufReader}; use tokio_util::io::{ReaderStream, StreamReader}; pub use objectstore_types::metadata::{Compression, ExpirationPolicy}; @@ -273,43 +274,27 @@ pub(crate) async fn encode_body(body: PutBody, mode: Option) -> Some(CompressionMode::Precompressed(_)) | None => None, }; - Ok(match (compression, body) { - (Some(Compression::Zstd), PutBody::Buffer(bytes)) => { - let cursor = Cursor::new(bytes); - let encoder = ZstdEncoder::new(cursor); - let stream = ReaderStream::new(encoder); - Body::wrap_stream(stream) - } - (Some(Compression::Zstd), PutBody::Stream(stream)) => { - let stream = StreamReader::new(stream); - let encoder = ZstdEncoder::new(stream); - let stream = ReaderStream::new(encoder); - Body::wrap_stream(stream) - } - (Some(Compression::Zstd), PutBody::File(file)) => { - let reader = BufReader::new(file); - let encoder = ZstdEncoder::new(reader); - let stream = ReaderStream::new(encoder); - Body::wrap_stream(stream) - } - (Some(Compression::Zstd), PutBody::Path(file)) => { - let file = File::open(file).await?; - let reader = BufReader::new(file); - let encoder = ZstdEncoder::new(reader); - let stream = ReaderStream::new(encoder); - Body::wrap_stream(stream) - } - (None, PutBody::Buffer(bytes)) => bytes.into(), - (None, PutBody::Stream(stream)) => Body::wrap_stream(stream), - (None, PutBody::File(file)) => { - let stream = ReaderStream::new(file); - Body::wrap_stream(stream) - } - (None, PutBody::Path(path)) => { - let stream = ReaderStream::new(File::open(path).await?); - Body::wrap_stream(stream) - } - }) + let Some(compression) = compression else { + return Ok(match body { + PutBody::Buffer(bytes) => bytes.into(), + PutBody::Stream(stream) => Body::wrap_stream(stream), + PutBody::File(file) => Body::wrap_stream(ReaderStream::new(file)), + PutBody::Path(path) => Body::wrap_stream(ReaderStream::new(File::open(path).await?)), + }); + }; + + let reader: Pin> = match body { + PutBody::Buffer(bytes) => Box::pin(Cursor::new(bytes)), + PutBody::Stream(stream) => Box::pin(StreamReader::new(stream)), + PutBody::File(file) => Box::pin(BufReader::new(file)), + PutBody::Path(path) => Box::pin(BufReader::new(File::open(path).await?)), + }; + + let stream = match compression { + Compression::Zstd => ReaderStream::new(ZstdEncoder::new(reader)).boxed(), + Compression::Gzip => ReaderStream::new(GzipEncoder::new(reader)).boxed(), + }; + Ok(Body::wrap_stream(stream)) } // TODO: instead of a separate `send` method, it would be nice to just implement `IntoFuture`. @@ -339,6 +324,8 @@ impl PutBuilder { #[cfg(test)] mod tests { + use std::io::Read as _; + use futures_util::stream; use http_body_util::BodyExt as _; @@ -382,6 +369,31 @@ mod tests { ); } + #[tokio::test] + async fn gzip_compresses_all_body_kinds() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("payload"); + std::fs::write(&path, b"hello world").unwrap(); + + let bodies = [ + PutBody::Buffer(Bytes::from_static(b"hello world")), + stream_body(vec![b"hello ", b"world"]), + PutBody::File(File::open(&path).await.unwrap()), + PutBody::Path(path.clone()), + ]; + + for body in bodies { + let mode = Some(CompressionMode::Compress(Compression::Gzip)); + let encoded = collect(encode_body(body, mode).await.unwrap()).await; + + let mut decoded = Vec::new(); + flate2::read::GzDecoder::new(&encoded[..]) + .read_to_end(&mut decoded) + .unwrap(); + assert_eq!(decoded, b"hello world"); + } + } + #[tokio::test] async fn precompressed_buffer_is_forwarded_verbatim() { let compressed = zstd_compress(b"hello world"); diff --git a/clients/rust/tests/e2e.rs b/clients/rust/tests/e2e.rs index 96d8e763..0a81eb88 100644 --- a/clients/rust/tests/e2e.rs +++ b/clients/rust/tests/e2e.rs @@ -112,6 +112,37 @@ async fn uses_zstd_by_default() { assert_eq!(received, "oh hai!"); } +#[tokio::test] +async fn roundtrips_gzip() { + let server = test_server().await; + + let client = Client::builder(server.url("/")) + .token(test_token_generator()) + .build() + .unwrap(); + let usecase = Usecase::new("usecase").with_compression(Compression::Gzip); + let session = client.session(usecase.for_organization(12345)).unwrap(); + + let stored_id = session.put("oh hai!").send().await.unwrap().key; + + let request = session.get(&stored_id).decompress(false); + let response = request.send().await.unwrap().unwrap(); + assert_eq!(response.metadata.compression, Some(Compression::Gzip)); + + let received_compressed = response.payload().await.unwrap(); + let mut decompressed = Vec::new(); + std::io::Read::read_to_end( + &mut flate2::read::GzDecoder::new(&received_compressed[..]), + &mut decompressed, + ) + .unwrap(); + assert_eq!(decompressed, b"oh hai!"); + + let response = session.get(&stored_id).send().await.unwrap().unwrap(); + assert_eq!(response.metadata.compression, None); + assert_eq!(response.payload().await.unwrap(), "oh hai!"); +} + #[tokio::test] async fn stores_precompressed_payload_verbatim() { let server = test_server().await; diff --git a/objectstore-service/Cargo.toml b/objectstore-service/Cargo.toml index d70938d9..659455ac 100644 --- a/objectstore-service/Cargo.toml +++ b/objectstore-service/Cargo.toml @@ -49,6 +49,7 @@ uuid = { workspace = true, features = ["v7"] } storage-cogs = ["dep:objectstore-inventory-tracker"] [dev-dependencies] +flate2 = { workspace = true } futures = { workspace = true } objectstore-inventory-tracker = { workspace = true, features = ["test-utils"] } tokio = { workspace = true, features = ["test-util"] } diff --git a/objectstore-service/src/backend/gcs.rs b/objectstore-service/src/backend/gcs.rs index 25b3d944..8c09a9c7 100644 --- a/objectstore-service/src/backend/gcs.rs +++ b/objectstore-service/src/backend/gcs.rs @@ -983,7 +983,15 @@ impl Backend for GcsBackend { let payload_response = self .with_retry("get_payload", || async { - let mut req = self.request(Method::GET, download_url.clone()).await?; + // GCS transparently decompresses objects stored with `Content-Encoding: gzip` + // unless the request accepts gzip (decompressive transcoding). We always return + // payloads exactly as stored, so opt out of it. This also keeps range reads + // applying to the stored bytes. See: + // https://cloud.google.com/storage/docs/transcoding#decompressive_transcoding + let mut req = self + .request(Method::GET, download_url.clone()) + .await? + .header(header::ACCEPT_ENCODING, "gzip"); if let Some(r) = range { req = req.header(header::RANGE, r.to_header_value()); } @@ -1972,6 +1980,15 @@ mod tests { assert!(requests[1].contains("alt=media&ifGenerationMatch=1")); assert!(!requests[2].contains("alt=media")); assert!(requests[3].contains("alt=media&ifGenerationMatch=2")); + // Payload downloads opt out of GCS decompressive transcoding. + for request in [&requests[1], &requests[3]] { + assert!( + request + .to_ascii_lowercase() + .contains("\r\naccept-encoding: gzip\r\n"), + "missing accept-encoding: {request}" + ); + } server.join().unwrap(); Ok(()) @@ -2835,6 +2852,101 @@ mod tests { Ok(()) } + fn gzip_encode(data: &[u8]) -> Vec { + let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default()); + encoder.write_all(data).unwrap(); + encoder.finish().unwrap() + } + + /// GCS decompresses `Content-Encoding: gzip` objects on download unless the request accepts + /// gzip. The backend must return the payload exactly as stored. + #[tokio::test] + async fn test_gzip_payload_roundtrip() -> Result<()> { + use objectstore_types::metadata::Compression; + + let backend = create_test_backend().await?; + + let plaintext = b"hello, world (but compressed with gzip)"; + let compressed = gzip_encode(plaintext); + + let id = make_id(); + let metadata = Metadata { + content_type: "text/plain".into(), + compression: Some(Compression::Gzip), + ..Default::default() + }; + + backend + .put_object( + &id, + &metadata, + stream::single(compressed.clone()), + Timestamp::now(), + ) + .await?; + + let (meta, _, stream) = backend + .get_object(&id, Timestamp::now(), None) + .await? + .unwrap(); + let payload = stream::read_to_vec(stream).await?; + + assert_eq!(meta.compression, Some(Compression::Gzip)); + assert_eq!( + payload, compressed, + "Payload should be returned still compressed, not transcoded" + ); + + Ok(()) + } + + /// Range reads of gzip objects must apply to the stored (compressed) bytes, which requires + /// opting out of decompressive transcoding. + #[tokio::test] + async fn test_gzip_payload_range_read() -> Result<()> { + use objectstore_types::metadata::Compression; + + let backend = create_test_backend().await?; + + let plaintext = b"hello, world (but compressed with gzip)"; + let compressed = gzip_encode(plaintext); + + let id = make_id(); + let metadata = Metadata { + content_type: "text/plain".into(), + compression: Some(Compression::Gzip), + ..Default::default() + }; + + backend + .put_object( + &id, + &metadata, + stream::single(compressed.clone()), + Timestamp::now(), + ) + .await?; + + let (meta, content_range, stream) = backend + .get_object(&id, Timestamp::now(), Some(ByteRange::Bounded(2, 9))) + .await? + .unwrap(); + let payload = stream::read_to_vec(stream).await?; + + assert_eq!(meta.compression, Some(Compression::Gzip)); + assert_eq!( + content_range, + Some(ContentRange { + start: 2, + end: 9, + total: compressed.len() as u64, + }) + ); + assert_eq!(payload, &compressed[2..=9]); + + Ok(()) + } + #[tokio::test] async fn test_multipart_single_part() -> Result<()> { let backend = create_test_backend().await?; diff --git a/objectstore-service/src/backend/s3_compatible.rs b/objectstore-service/src/backend/s3_compatible.rs index d8c0e645..c6f65a4e 100644 --- a/objectstore-service/src/backend/s3_compatible.rs +++ b/objectstore-service/src/backend/s3_compatible.rs @@ -245,7 +245,14 @@ where ) -> Result, Response)>> { let object_url = self.object_url(id); - let mut builder = self.request(method, &object_url).await?; + // Some S3-compatible services, such as GCS, transparently decompress objects stored with + // `Content-Encoding: gzip` unless the request accepts gzip. We always return payloads + // exactly as stored, so opt out of it. See: + // https://cloud.google.com/storage/docs/transcoding#decompressive_transcoding + let mut builder = self + .request(method, &object_url) + .await? + .header(reqwest::header::ACCEPT_ENCODING, "gzip"); if let Some(r) = range { builder = builder.header(reqwest::header::RANGE, r.to_header_value()); } diff --git a/objectstore-types/src/metadata.rs b/objectstore-types/src/metadata.rs index f2515ae8..6f9e1a5e 100644 --- a/objectstore-types/src/metadata.rs +++ b/objectstore-types/src/metadata.rs @@ -283,16 +283,14 @@ impl FromStr for ExpirationPolicy { /// The compression algorithm applied to an object's payload. /// -/// Transmitted via the standard `Content-Encoding` HTTP header. Currently only -/// Zstandard (`zstd`) is supported. +/// Transmitted via the standard `Content-Encoding` HTTP header. Supported +/// algorithms are Zstandard (`zstd`) and `gzip`. #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] pub enum Compression { /// Compressed using `zstd`. Zstd, - // /// Compressed using `gzip`. - // Gzip, - // /// Compressed using `lz4`. - // Lz4, + /// Compressed using `gzip`. + Gzip, } impl Compression { @@ -300,8 +298,7 @@ impl Compression { pub fn as_str(&self) -> &str { match self { Compression::Zstd => "zstd", - // Compression::Gzip => "gzip", - // Compression::Lz4 => "lz4", + Compression::Gzip => "gzip", } } } @@ -318,8 +315,7 @@ impl FromStr for Compression { fn from_str(s: &str) -> Result { match s { "zstd" => Ok(Compression::Zstd), - // "gzip" => Compression::Gzip, - // "lz4" => Compression::Lz4, + "gzip" => Ok(Compression::Gzip), _ => Err(Error::Compression), } } @@ -1242,15 +1238,17 @@ mod tests { #[test] fn compression_display_roundtrip() { - let displayed = Compression::Zstd.to_string(); - assert_eq!(displayed, "zstd"); - let parsed: Compression = displayed.parse().unwrap(); - assert_eq!(parsed, Compression::Zstd); + for (compression, expected) in [(Compression::Zstd, "zstd"), (Compression::Gzip, "gzip")] { + let displayed = compression.to_string(); + assert_eq!(displayed, expected); + let parsed: Compression = displayed.parse().unwrap(); + assert_eq!(parsed, compression); + } } #[test] fn compression_parse_invalid() { - assert!(Compression::from_str("gzip").is_err()); + assert!(Compression::from_str("lz4").is_err()); assert!(Compression::from_str("").is_err()); } From 68b4aea72d79d30759f8efaee2c1602851f0d8df Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 20 Oct 2026 11:57:19 +0000 Subject: [PATCH 2/7] ref(s3): Don't send Accept-Encoding on S3-compatible reads S3 and SeaweedFS serve objects as stored regardless of Accept-Encoding, so only the GCS backend needs to opt out of transcoding. --- objectstore-service/src/backend/s3_compatible.rs | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/objectstore-service/src/backend/s3_compatible.rs b/objectstore-service/src/backend/s3_compatible.rs index c6f65a4e..d8c0e645 100644 --- a/objectstore-service/src/backend/s3_compatible.rs +++ b/objectstore-service/src/backend/s3_compatible.rs @@ -245,14 +245,7 @@ where ) -> Result, Response)>> { let object_url = self.object_url(id); - // Some S3-compatible services, such as GCS, transparently decompress objects stored with - // `Content-Encoding: gzip` unless the request accepts gzip. We always return payloads - // exactly as stored, so opt out of it. See: - // https://cloud.google.com/storage/docs/transcoding#decompressive_transcoding - let mut builder = self - .request(method, &object_url) - .await? - .header(reqwest::header::ACCEPT_ENCODING, "gzip"); + let mut builder = self.request(method, &object_url).await?; if let Some(r) = range { builder = builder.header(reqwest::header::RANGE, r.to_header_value()); } From cb68896bc0ca5c02151a0214faf0d43ea16dff8b Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 20 Oct 2026 13:10:26 +0000 Subject: [PATCH 3/7] feat(py-client): Support gzip compression Uploads can be compressed with gzip on the fly or spooled for resumable uploads, precompressed gzip payloads are accepted, and gzip downloads are transparently decompressed, including multi-member payloads from multipart uploads. Unsupported algorithms are rejected up front, including on Usecase. Refs FS-575 --- clients/python/README.md | 8 +- .../src/objectstore_client/_compression.py | 133 ++++++++++++++++++ .../src/objectstore_client/_resumable.py | 10 +- .../python/src/objectstore_client/client.py | 42 ++---- .../python/src/objectstore_client/metadata.py | 2 +- clients/python/tests/test_compression.py | 86 +++++++++++ clients/python/tests/test_e2e.py | 87 ++++++++++-- 7 files changed, 320 insertions(+), 48 deletions(-) create mode 100644 clients/python/src/objectstore_client/_compression.py create mode 100644 clients/python/tests/test_compression.py diff --git a/clients/python/README.md b/clients/python/README.md index 7cd45433..ae59ccd2 100644 --- a/clients/python/README.md +++ b/clients/python/README.md @@ -92,14 +92,18 @@ session.put(b"payload", origin="203.0.113.42") ### Compression Uploads are compressed with Zstd by default, and downloads are transparently decompressed. -Compression can be overridden per-upload: +The supported algorithms are `"zstd"` and `"gzip"`. Compression can be set per-usecase +or overridden per-upload: ```python +# compress all uploads of this usecase with gzip: +usecase = Usecase("my-usecase", compression="gzip") + # upload as-is and record no encoding: session.put(video_data, compress="none") # upload as-is, but record the encoding so that downloads still decompress: -session.put(zstd_data, precompressed="zstd") +session.put(gzip_data, precompressed="gzip") ``` ### Resumable Uploads diff --git a/clients/python/src/objectstore_client/_compression.py b/clients/python/src/objectstore_client/_compression.py new file mode 100644 index 00000000..a6126f3c --- /dev/null +++ b/clients/python/src/objectstore_client/_compression.py @@ -0,0 +1,133 @@ +"""Streaming compression and decompression for the supported algorithms.""" + +from __future__ import annotations + +import gzip +import io +import zlib +from typing import IO, cast + +import zstandard + +from objectstore_client.metadata import Compression +from objectstore_client.utils import _ZstdCompressionReaderWrapper + +SUPPORTED_COMPRESSIONS: tuple[Compression, ...] = ("zstd", "gzip") +"""Compression algorithms that the client can apply and transparently decode.""" + +_CHUNK_SIZE = 64 * 1024 +# `wbits` for `zlib` that produces and consumes the gzip container format. +_GZIP_WBITS = 16 + zlib.MAX_WBITS + + +def validate(compression: str | None, *, allow_none: bool = True) -> None: + """Raise `ValueError` if `compression` is not a supported algorithm. + + `None` is always accepted. `"none"` is accepted only if `allow_none` is set. + """ + if compression is None or (allow_none and compression == "none"): + return + if compression not in SUPPORTED_COMPRESSIONS: + raise ValueError(f"Invalid compression: {compression}") + + +def compressing_reader(source: IO[bytes], compression: Compression) -> IO[bytes]: + """Return a readable stream yielding `source` compressed with `compression`. + + The returned stream compresses lazily as it is read, reports the number of + compressed bytes produced via `tell()`, and supports `seek(0)` only until + reading starts. Closing it does not close `source`. + """ + if compression == "zstd": + reader = zstandard.ZstdCompressor().stream_reader(source, closefd=False) + return cast(IO[bytes], _ZstdCompressionReaderWrapper(reader)) + if compression == "gzip": + return cast(IO[bytes], _GzipCompressionReader(source)) + raise ValueError(f"Invalid compression: {compression}") + + +def compress_into( + source: IO[bytes], dest: IO[bytes], compression: Compression +) -> tuple[int, int]: + """Compress `source` into `dest`, returning the bytes read and written.""" + if compression == "zstd": + read, written = zstandard.ZstdCompressor().copy_stream(source, dest) + return read, written + if compression == "gzip": + compressor = zlib.compressobj(wbits=_GZIP_WBITS) + read = written = 0 + while chunk := source.read(_CHUNK_SIZE): + read += len(chunk) + written += dest.write(compressor.compress(chunk)) + written += dest.write(compressor.flush()) + return read, written + raise ValueError(f"Invalid compression: {compression}") + + +def decompressing_reader(stream: IO[bytes], compression: str) -> IO[bytes]: + """Return a readable stream decoding `stream` compressed with `compression`. + + Concatenated zstd frames and gzip members, as produced by multipart uploads + with individually compressed parts, are decoded as a single payload. + """ + if compression == "zstd": + dctx = zstandard.ZstdDecompressor() + return cast(IO[bytes], dctx.stream_reader(stream, read_across_frames=True)) + if compression == "gzip": + return cast(IO[bytes], gzip.GzipFile(fileobj=stream, mode="rb")) + raise NotImplementedError( + f"Transparent decoding of `{compression}` is not implemented" + ) + + +class _GzipCompressionReader(io.RawIOBase): + """A readable stream that gzip-compresses `source` as it is read. + + Like :class:`_ZstdCompressionReaderWrapper`, it allows `seek(0)` as long as + no data has been read yet, so that urllib3 can record the body position. + """ + + def __init__(self, source: IO[bytes]) -> None: + super().__init__() + self._source = source + self._compressor = zlib.compressobj(wbits=_GZIP_WBITS) + self._buffer = bytearray() + self._finished = False + self._position = 0 + + def readable(self) -> bool: + return True + + def read(self, size: int | None = -1, /) -> bytes: + if size is None or size < 0: + size = -1 + while not self._finished and (size < 0 or len(self._buffer) < size): + chunk = self._source.read(_CHUNK_SIZE) + if chunk: + self._buffer += self._compressor.compress(chunk) + else: + self._buffer += self._compressor.flush() + self._finished = True + + if size < 0: + size = len(self._buffer) + out = bytes(self._buffer[:size]) + del self._buffer[:size] + self._position += len(out) + return out + + def readall(self) -> bytes: + return self.read(-1) + + def readinto(self, buffer: memoryview | bytearray) -> int: # type: ignore[override] + data = self.read(len(buffer)) + buffer[: len(data)] = data + return len(data) + + def tell(self) -> int: + return self._position + + def seek(self, offset: int, whence: int = io.SEEK_SET, /) -> int: + if offset == 0 and whence == io.SEEK_SET and self._position == 0: + return 0 + raise OSError("Cannot seek in a compressed stream after reading has started") diff --git a/clients/python/src/objectstore_client/_resumable.py b/clients/python/src/objectstore_client/_resumable.py index 56aa0bd1..96076918 100644 --- a/clients/python/src/objectstore_client/_resumable.py +++ b/clients/python/src/objectstore_client/_resumable.py @@ -12,8 +12,8 @@ from urllib.parse import urlencode import urllib3 -import zstandard +from objectstore_client import _compression from objectstore_client.errors import RequestError, raise_for_status from objectstore_client.metadata import Compression, ExpirationPolicy from objectstore_client.metrics import measure_storage_operation @@ -236,7 +236,7 @@ def cancel(self) -> None: def prepare_body( contents: bytes | IO[bytes], source_size: int, - compress: bool, + compression: Compression | None, spool_dir: str | PathLike[str] | None, ) -> Iterator[tuple[IO[bytes], int, int] | None]: """Yield the body, source size, and compressed size (equal if uncompressed). @@ -248,7 +248,7 @@ def prepare_body( with ( BytesIO(contents) if isinstance(contents, bytes) else nullcontext(contents) ) as source: - if not compress: + if compression is None or compression == "none": yield source, source_size, source_size return @@ -259,8 +259,8 @@ def prepare_body( return with temporary as body: - source_size, compressed_size = zstandard.ZstdCompressor().copy_stream( - source, body + source_size, compressed_size = _compression.compress_into( + source, body, compression ) body.seek(0) yield body, source_size, compressed_size diff --git a/clients/python/src/objectstore_client/client.py b/clients/python/src/objectstore_client/client.py index f659ce5f..e83c16a6 100644 --- a/clients/python/src/objectstore_client/client.py +++ b/clients/python/src/objectstore_client/client.py @@ -12,10 +12,9 @@ from urllib.parse import urlparse import urllib3 -import zstandard from urllib3.connectionpool import HTTPConnectionPool -from objectstore_client import _resumable, presign, utils +from objectstore_client import _compression, _resumable, presign, utils from objectstore_client.auth import Permission, SecretKey, TokenProvider from objectstore_client.errors import ( ExpiryExtensionRejected, @@ -91,7 +90,7 @@ def __init__( """Create a usecase. Raises ``ValueError`` if the name violates the naming rules on ``Usecase``, - or if the ``resumable_threshold_bytes`` is invalid. + or if the ``compression`` or ``resumable_threshold_bytes`` is invalid. """ if not 3 <= len(name) <= 64: raise ValueError("usecase name must be 3–64 ASCII characters long") @@ -108,6 +107,7 @@ def __init__( 'resumable_threshold_bytes must be a non-negative integer or "none"' ) self._resumable_threshold_bytes = resumable_threshold_bytes + _compression.validate(compression) self._compression = compression self._expiration_policy = expiration_policy @@ -437,10 +437,8 @@ def put( """ if compress is not None and precompressed is not None: raise ValueError("Cannot pass both `compress` and `precompressed`") - if compress and compress not in ("none", "zstd"): - raise ValueError(f"Invalid compression: {compress}") - if precompressed and precompressed != "zstd": - raise ValueError(f"Invalid compression: {precompressed}") + _compression.validate(compress) + _compression.validate(precompressed, allow_none=False) if resumable_threshold_bytes is None: resumable_threshold_bytes = self._usecase._resumable_threshold_bytes @@ -486,13 +484,13 @@ def put( with _resumable.prepare_body( contents, source_size, - compress_with != "none", + compress_with, self._spool_dir, ) as prepared: if prepared is None: # Temporary-file creation failed before consuming the input. result_key, source_size, compressed_size = self._put_direct( - contents, key, headers, compress=compress_with != "none" + contents, key, headers, compression=compress_with ) else: body, source_size, compressed_size = prepared @@ -513,11 +511,11 @@ def put( # Send the prepared body without compressing it again. headers["Content-Length"] = str(compressed_size) result_key, _, _ = self._put_direct( - body, key, headers, compress=False + body, key, headers, compression="none" ) else: result_key, source_size, compressed_size = self._put_direct( - contents, key, headers, compress=compress_with != "none" + contents, key, headers, compression=compress_with ) if precompressed is None and source_size is not None: @@ -543,7 +541,7 @@ def _put_direct( contents: bytes | IO[bytes], key: str | None, headers: dict[str, str], - compress: bool, + compression: Compression, ) -> tuple[str, int | None, int | None]: """Return the key and final source/body positions when available.""" with ( @@ -552,12 +550,8 @@ def _put_direct( body = original_body retries = None # by default use the pool's value, set by the Client try: - if compress: - cctx = zstandard.ZstdCompressor() - body = cctx.stream_reader(original_body, closefd=False) - body = cast(IO[bytes], utils._ZstdCompressionReaderWrapper(body)) - - if compress: + if compression != "none": + body = _compression.compressing_reader(original_body, compression) # For on-the-fly compression, don't attempt read retries, # as the stream cannot be rewound after data has been consumed. pool_retries = self._pool.retries @@ -645,15 +639,8 @@ def get( ) decompressed = False if metadata.compression and decompress and not encoding_accepted: - if metadata.compression != "zstd": - raise NotImplementedError( - "Transparent decoding of anything but `zstd` is not " - "implemented yet" - ) - + stream = _compression.decompressing_reader(stream, metadata.compression) metadata.compression = None - dctx = zstandard.ZstdDecompressor() - stream = dctx.stream_reader(stream, read_across_frames=True) decompressed = True span.set_attribute("objectstore.decompressed", decompressed) @@ -933,8 +920,7 @@ def _create_upload( """ if object_length < 0: raise ValueError("Object length must not be negative") - if compression is not None and compression not in ("none", "zstd"): - raise ValueError(f"Invalid compression: {compression}") + _compression.validate(compression) headers = self._metadata_headers( compression=compression, diff --git a/clients/python/src/objectstore_client/metadata.py b/clients/python/src/objectstore_client/metadata.py index 42214f33..717fd1b2 100644 --- a/clients/python/src/objectstore_client/metadata.py +++ b/clients/python/src/objectstore_client/metadata.py @@ -9,7 +9,7 @@ from objectstore_client.utils import decode_header_value -Compression = Literal["zstd"] | Literal["none"] +Compression = Literal["zstd", "gzip"] | Literal["none"] HEADER_EXPIRATION = "x-sn-expiration" HEADER_TIME_CREATED = "x-sn-time-created" diff --git a/clients/python/tests/test_compression.py b/clients/python/tests/test_compression.py new file mode 100644 index 00000000..03ae08a8 --- /dev/null +++ b/clients/python/tests/test_compression.py @@ -0,0 +1,86 @@ +import gzip +import io +import random + +import pytest +import zstandard +from objectstore_client import _compression +from objectstore_client.metadata import Compression + +# Incompressible, so the output spans many internal chunks. +PAYLOAD = random.Random(0).randbytes(300_000) + b"tail" * 1000 + + +@pytest.mark.parametrize("compression", ["zstd", "gzip"]) +@pytest.mark.parametrize("read_size", [-1, 1, 4096, 100_000]) +def test_compressing_reader_roundtrip(compression: Compression, read_size: int) -> None: + source = io.BytesIO(PAYLOAD) + reader = _compression.compressing_reader(source, compression) + + chunks = [] + while chunk := reader.read(read_size): + chunks.append(chunk) + compressed = b"".join(chunks) + + assert reader.tell() == len(compressed) + assert source.tell() == len(PAYLOAD) + decoded = _compression.decompressing_reader(io.BytesIO(compressed), compression) + assert decoded.read() == PAYLOAD + + reader.close() + assert not source.closed + + +@pytest.mark.parametrize("compression", ["zstd", "gzip"]) +def test_compressing_reader_seeks_to_start_only_before_reading( + compression: Compression, +) -> None: + reader = _compression.compressing_reader(io.BytesIO(PAYLOAD), compression) + assert reader.seek(0) == 0 + reader.read(10) + with pytest.raises(OSError): + reader.seek(0) + + +@pytest.mark.parametrize("compression", ["zstd", "gzip"]) +def test_compress_into(compression: Compression) -> None: + dest = io.BytesIO() + read, written = _compression.compress_into(io.BytesIO(PAYLOAD), dest, compression) + + assert read == len(PAYLOAD) + assert written == len(dest.getvalue()) + dest.seek(0) + assert _compression.decompressing_reader(dest, compression).read() == PAYLOAD + + +def test_gzip_output_is_standard() -> None: + reader = _compression.compressing_reader(io.BytesIO(PAYLOAD), "gzip") + assert gzip.decompress(reader.read()) == PAYLOAD + + +@pytest.mark.parametrize( + "compression,compress", + [("zstd", zstandard.ZstdCompressor().compress), ("gzip", gzip.compress)], +) +def test_decompressing_reader_reads_concatenated_members( + compression: str, compress: object +) -> None: + assert callable(compress) + stream = io.BytesIO(compress(b"hello ") + compress(b"world")) + assert _compression.decompressing_reader(stream, compression).read() == ( + b"hello world" + ) + + +def test_decompressing_reader_rejects_unknown() -> None: + with pytest.raises(NotImplementedError): + _compression.decompressing_reader(io.BytesIO(b""), "lz4") + + +def test_validate() -> None: + for compression in ["zstd", "gzip", "none", None]: + _compression.validate(compression) + with pytest.raises(ValueError): + _compression.validate("none", allow_none=False) + with pytest.raises(ValueError): + _compression.validate("lz4") diff --git a/clients/python/tests/test_e2e.py b/clients/python/tests/test_e2e.py index 3e82abda..86b33620 100644 --- a/clients/python/tests/test_e2e.py +++ b/clients/python/tests/test_e2e.py @@ -1,3 +1,4 @@ +import gzip import os import shutil import signal @@ -12,6 +13,7 @@ from datetime import timedelta from io import BytesIO from pathlib import Path +from typing import cast from unittest.mock import Mock, patch import pytest @@ -33,7 +35,7 @@ ) from objectstore_client.auth import Permission, SecretKey from objectstore_client.errors import RequestError -from objectstore_client.metadata import TimeToLive +from objectstore_client.metadata import Compression, TimeToLive from objectstore_client.scope import Scope TEST_EDDSA_KID: str = "test_kid" @@ -477,6 +479,61 @@ def test_precompressed_is_stored_verbatim(server_url: str) -> None: assert retrieved.payload.read() == data +def test_full_cycle_gzip(server_url: str) -> None: + client = Client(server_url, token=TestSecretKey.get()) + session = client.session( + Usecase("test-usecase", compression="gzip"), org=42, project=1337 + ) + + object_key = session.put(b"test data" * 100) + + metadata = session.head(object_key) + assert metadata is not None + assert metadata.compression == "gzip" + + retrieved = session.get(object_key) + assert retrieved is not None + assert retrieved.metadata.compression is None + assert retrieved.payload.read() == b"test data" * 100 + + for raw in [ + session.get(object_key, decompress=False), + session.get(object_key, accept_encoding=["gzip"]), + ]: + assert raw is not None + assert raw.metadata.compression == "gzip" + assert gzip.decompress(raw.payload.read()) == b"test data" * 100 + + +def test_precompressed_gzip_is_stored_verbatim(server_url: str) -> None: + client = Client(server_url, token=TestSecretKey.get()) + session = client.session(Usecase("test-usecase"), org=42, project=1337) + + compressed = gzip.compress(b"test data") + object_key = session.put(compressed, precompressed="gzip") + + raw = session.get(object_key, decompress=False) + assert raw is not None + assert raw.metadata.compression == "gzip" + assert raw.payload.read() == compressed + + retrieved = session.get(object_key) + assert retrieved is not None + assert retrieved.payload.read() == b"test data" + + +def test_rejects_unsupported_compression(server_url: str) -> None: + client = Client(server_url, token=TestSecretKey.get()) + session = client.session(Usecase("test-usecase"), org=42, project=1337) + + with pytest.raises(ValueError, match="Invalid compression"): + session.put(b"test data", compress=cast(Compression, "lz4")) + with pytest.raises(ValueError, match="Invalid compression"): + session.put(b"test data", precompressed=cast(Compression, "lz4")) + with pytest.raises(ValueError, match="Invalid compression"): + Usecase("test-usecase", compression=cast(Compression, "lz4")) + + def test_precompressed_rejects_conflicting_arguments(server_url: str) -> None: client = Client(server_url, token=TestSecretKey.get()) session = client.session(Usecase("test-usecase"), org=42, project=1337) @@ -895,25 +952,31 @@ def test_put_stores_under_literal_key(server_url: str) -> None: @pytest.mark.parametrize("source_kind", ["bytes", "buffer", "stream"]) -@pytest.mark.parametrize("compression", ["none", "zstd", "precompressed"]) +@pytest.mark.parametrize( + "compression", + ["none", "zstd", "gzip", "precompressed-zstd", "precompressed-gzip"], +) @pytest.mark.parametrize("threshold", [1, 2048]) def test_automatic_resumable_upload( server_url: str, source_kind: str, compression: str, threshold: int ) -> None: + precompressed = compression.startswith("precompressed-") + algorithm = cast(Compression, compression.removeprefix("precompressed-")) session = Client(server_url, token=TestSecretKey.get()).session( Usecase( "test-usecase", - compression="zstd" if compression == "zstd" else "none", + compression="none" if precompressed else algorithm, expiration_policy=TimeToLive(timedelta(days=1)), resumable_threshold_bytes=threshold, ), org=42, ) contents = b"file contents\n" * 100 - precompressed = compression == "precompressed" - encoded = ( - zstandard.ZstdCompressor().compress(contents) if precompressed else contents - ) + encoded = contents + if algorithm == "zstd" and precompressed: + encoded = zstandard.ZstdCompressor().compress(contents) + elif algorithm == "gzip" and precompressed: + encoded = gzip.compress(contents) with ( BytesIO() if source_kind == "buffer" @@ -926,7 +989,7 @@ def test_automatic_resumable_upload( source.seek(len(b"skip this prefix")) key = session.put( source if source_kind in ("buffer", "stream") else encoded, - precompressed="zstd" if precompressed else None, + precompressed=algorithm if precompressed else None, content_type="text/plain", metadata={"source": "file"}, origin="203.0.113.42", @@ -936,7 +999,7 @@ def test_automatic_resumable_upload( assert not source.closed stored = session.head(key) assert stored is not None - assert stored.compression == ("zstd" if compression != "none" else None) + assert stored.compression == (algorithm if algorithm != "none" else None) assert stored.content_type == "text/plain" assert stored.filename == "example.txt" assert stored.origin == "203.0.113.42" @@ -971,13 +1034,13 @@ def test_resumable_creation_fallback( assert retrieved.payload.read() == b"payload" -@pytest.mark.parametrize("compress", [False, True]) -def test_nonseekable_upload(server_url: str, compress: bool) -> None: +@pytest.mark.parametrize("compress", ["none", "zstd", "gzip"]) +def test_nonseekable_upload(server_url: str, compress: Compression) -> None: session = Client(server_url, token=TestSecretKey.get()).session( Usecase("test-usecase", resumable_threshold_bytes=1) ) with UnrewindableStream(b"payload") as source: - key = session.put(source, compress="zstd" if compress else "none") + key = session.put(source, compress=compress) assert not source.closed retrieved = session.get(key) From 810ab32ab5a9c064208245f18cc1dc59648e42dd Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 20 Oct 2026 14:17:05 +0000 Subject: [PATCH 4/7] ref: Address review feedback on gzip compression - py-client: Replace the zstd-specific reader wrapper and the gzip reader with a single _CompressionReader driven by a compress/flush compressor. - py-client: Document why the gzip compressor sets wbits. - rust-client: Name the parts of the gzip size bound and drop its test. --- .../src/objectstore_client/_compression.py | 76 ++++--------------- .../python/src/objectstore_client/utils.py | 74 +++++++++++++----- clients/python/tests/test_utils.py | 9 +-- clients/rust/src/many.rs | 43 ++++------- clients/rust/tests/e2e.rs | 4 +- 5 files changed, 92 insertions(+), 114 deletions(-) diff --git a/clients/python/src/objectstore_client/_compression.py b/clients/python/src/objectstore_client/_compression.py index a6126f3c..8444a86b 100644 --- a/clients/python/src/objectstore_client/_compression.py +++ b/clients/python/src/objectstore_client/_compression.py @@ -3,21 +3,28 @@ from __future__ import annotations import gzip -import io import zlib from typing import IO, cast import zstandard from objectstore_client.metadata import Compression -from objectstore_client.utils import _ZstdCompressionReaderWrapper +from objectstore_client.utils import _CompressionReader, _Compressor SUPPORTED_COMPRESSIONS: tuple[Compression, ...] = ("zstd", "gzip") """Compression algorithms that the client can apply and transparently decode.""" _CHUNK_SIZE = 64 * 1024 -# `wbits` for `zlib` that produces and consumes the gzip container format. -_GZIP_WBITS = 16 + zlib.MAX_WBITS + + +def _gzip_compressor() -> _Compressor: + """Return a zlib compressor that emits the gzip format. + + By default, zlib wraps deflate data in the zlib format. Adding 16 to the window + size (`wbits`) selects the gzip header and trailer instead, which is what + `Content-Encoding: gzip` requires. See `zlib.compressobj`. + """ + return zlib.compressobj(wbits=16 + zlib.MAX_WBITS) def validate(compression: str | None, *, allow_none: bool = True) -> None: @@ -39,10 +46,10 @@ def compressing_reader(source: IO[bytes], compression: Compression) -> IO[bytes] reading starts. Closing it does not close `source`. """ if compression == "zstd": - reader = zstandard.ZstdCompressor().stream_reader(source, closefd=False) - return cast(IO[bytes], _ZstdCompressionReaderWrapper(reader)) + compressor = zstandard.ZstdCompressor().compressobj() + return cast(IO[bytes], _CompressionReader(source, compressor)) if compression == "gzip": - return cast(IO[bytes], _GzipCompressionReader(source)) + return cast(IO[bytes], _CompressionReader(source, _gzip_compressor())) raise ValueError(f"Invalid compression: {compression}") @@ -54,7 +61,7 @@ def compress_into( read, written = zstandard.ZstdCompressor().copy_stream(source, dest) return read, written if compression == "gzip": - compressor = zlib.compressobj(wbits=_GZIP_WBITS) + compressor = _gzip_compressor() read = written = 0 while chunk := source.read(_CHUNK_SIZE): read += len(chunk) @@ -78,56 +85,3 @@ def decompressing_reader(stream: IO[bytes], compression: str) -> IO[bytes]: raise NotImplementedError( f"Transparent decoding of `{compression}` is not implemented" ) - - -class _GzipCompressionReader(io.RawIOBase): - """A readable stream that gzip-compresses `source` as it is read. - - Like :class:`_ZstdCompressionReaderWrapper`, it allows `seek(0)` as long as - no data has been read yet, so that urllib3 can record the body position. - """ - - def __init__(self, source: IO[bytes]) -> None: - super().__init__() - self._source = source - self._compressor = zlib.compressobj(wbits=_GZIP_WBITS) - self._buffer = bytearray() - self._finished = False - self._position = 0 - - def readable(self) -> bool: - return True - - def read(self, size: int | None = -1, /) -> bytes: - if size is None or size < 0: - size = -1 - while not self._finished and (size < 0 or len(self._buffer) < size): - chunk = self._source.read(_CHUNK_SIZE) - if chunk: - self._buffer += self._compressor.compress(chunk) - else: - self._buffer += self._compressor.flush() - self._finished = True - - if size < 0: - size = len(self._buffer) - out = bytes(self._buffer[:size]) - del self._buffer[:size] - self._position += len(out) - return out - - def readall(self) -> bytes: - return self.read(-1) - - def readinto(self, buffer: memoryview | bytearray) -> int: # type: ignore[override] - data = self.read(len(buffer)) - buffer[: len(data)] = data - return len(data) - - def tell(self) -> int: - return self._position - - def seek(self, offset: int, whence: int = io.SEEK_SET, /) -> int: - if offset == 0 and whence == io.SEEK_SET and self._position == 0: - return 0 - raise OSError("Cannot seek in a compressed stream after reading has started") diff --git a/clients/python/src/objectstore_client/utils.py b/clients/python/src/objectstore_client/utils.py index 32d72941..a31e0c2a 100644 --- a/clients/python/src/objectstore_client/utils.py +++ b/clients/python/src/objectstore_client/utils.py @@ -1,11 +1,10 @@ from __future__ import annotations import io -from typing import IO, Any +from typing import IO, Protocol from urllib.parse import quote, unquote import filetype # type: ignore[import-untyped] -from zstandard import ZstdCompressionReader # Characters left unescaped when percent-encoding a URL path or query, expressed # in RFC 3986 terms. `urllib.parse.quote` already leaves the *unreserved* set @@ -128,26 +127,65 @@ def guess_mime_type(contents: bytes | IO[bytes]) -> str | None: return kind.mime if kind else None -class _ZstdCompressionReaderWrapper: +class _Compressor(Protocol): + """A streaming compression object, such as `zlib.compressobj()`.""" + + def compress(self, data: bytes, /) -> bytes: ... + + def flush(self) -> bytes: ... + + +class _CompressionReader(io.RawIOBase): """ - Wraps a `ZstdCompressionReader`, allowing `seek(0)` as long as no data has been read - yet. + A readable stream that compresses `source` with `compressor` as it is read. + + `tell()` reports the number of compressed bytes produced so far. `seek(0)` is + allowed as long as no data has been read yet, so that urllib3 can record the body + position. Closing the reader does not close `source`. """ - def __init__(self, inner: ZstdCompressionReader) -> None: - self._inner = inner + _CHUNK_SIZE = 64 * 1024 + + def __init__(self, source: IO[bytes], compressor: _Compressor) -> None: + super().__init__() + self._source = source + self._compressor = compressor + self._buffer = bytearray() + self._finished = False + self._position = 0 + + def readable(self) -> bool: + return True + + def read(self, size: int | None = -1, /) -> bytes: + if size is None or size < 0: + size = -1 + while not self._finished and (size < 0 or len(self._buffer) < size): + if chunk := self._source.read(self._CHUNK_SIZE): + self._buffer += self._compressor.compress(chunk) + else: + self._buffer += self._compressor.flush() + self._finished = True + + if size < 0: + size = len(self._buffer) + out = bytes(self._buffer[:size]) + del self._buffer[:size] + self._position += len(out) + return out + + def readall(self) -> bytes: + return self.read(-1) + + def readinto(self, buffer: memoryview | bytearray) -> int: # type: ignore[override] + data = self.read(len(buffer)) + buffer[: len(data)] = data + return len(data) + + def tell(self) -> int: + return self._position def seek(self, offset: int, whence: int = io.SEEK_SET, /) -> int: - current = self._inner.tell() - if current == offset == 0 and whence == io.SEEK_SET: + if offset == 0 and whence == io.SEEK_SET and self._position == 0: return 0 raise OSError("Cannot seek in a compressed stream after reading has started") - - def __getattr__(self, attr: str) -> Any: - return getattr(self._inner, attr) - - def __setattr__(self, name: str, value: object) -> None: - if name == "_inner": - object.__setattr__(self, name, value) - else: - setattr(self._inner, name, value) diff --git a/clients/python/tests/test_utils.py b/clients/python/tests/test_utils.py index ccff8c95..b7dea74e 100644 --- a/clients/python/tests/test_utils.py +++ b/clients/python/tests/test_utils.py @@ -3,17 +3,16 @@ import pytest import zstandard from objectstore_client.utils import ( - _ZstdCompressionReaderWrapper, + _CompressionReader, decode_header_value, encode_header_value, parse_accept_encoding, ) -def _make_wrapper(data: bytes = b"hello world") -> _ZstdCompressionReaderWrapper: - cctx = zstandard.ZstdCompressor() - reader = cctx.stream_reader(io.BytesIO(data)) - return _ZstdCompressionReaderWrapper(reader) +def _make_wrapper(data: bytes = b"hello world") -> _CompressionReader: + compressor = zstandard.ZstdCompressor().compressobj() + return _CompressionReader(io.BytesIO(data), compressor) def test_seek_zero_before_read() -> None: diff --git a/clients/rust/src/many.rs b/clients/rust/src/many.rs index e06c05c9..7153eaec 100644 --- a/clients/rust/src/many.rs +++ b/clients/rust/src/many.rs @@ -535,11 +535,23 @@ async fn send_batch( fn compress_bound(compression: Compression, size: usize) -> usize { match compression { Compression::Zstd => zstd_safe::compress_bound(size), - // zlib's conservative `deflateBound`, plus 18 bytes of gzip header and trailer. - Compression::Gzip => size + (size >> 12) + (size >> 14) + (size >> 25) + 7 + 18, + Compression::Gzip => deflate_bound(size) + GZIP_FRAMING_SIZE, } } +/// Size of the gzip header and trailer around the deflate stream (RFC 1952). +const GZIP_FRAMING_SIZE: usize = 18; + +/// Returns the worst-case size of a raw deflate stream for `size` input bytes. +/// +/// Neither `flate2` nor `miniz_oxide` expose a bound, so this mirrors zlib's +/// `compressBound` without its 6-byte zlib wrapper. Incompressible input is emitted +/// as stored blocks with a 5-byte header each, which this conservatively bounds at +/// 5 bytes per 16 KiB of input plus a constant for the final block. +fn deflate_bound(size: usize) -> usize { + size + (size >> 12) + (size >> 14) + (size >> 25) + 7 +} + fn classify_fail(key: Option, error: Error) -> Classified { Classified::Failed(OperationResult::Put( key.unwrap_or_else(|| "".to_owned()), @@ -872,33 +884,6 @@ mod tests { ); } - #[tokio::test] - async fn gzip_compress_bound_holds_for_incompressible_data() { - use tokio::io::AsyncReadExt as _; - - // A cheap xorshift generator, so the payload does not compress. - let mut state = 0x2545_f491_4f6c_dd1du64; - let payload: Vec = (0..(MAX_BATCH_PART_SIZE as usize)) - .map(|_| { - state ^= state << 13; - state ^= state >> 7; - state ^= state << 17; - state as u8 - }) - .collect(); - - for size in [0, 1, 1_000, 65_536, 300_000, payload.len()] { - let mut encoder = async_compression::tokio::bufread::GzipEncoder::new(&payload[..size]); - let mut compressed = Vec::new(); - encoder.read_to_end(&mut compressed).await.unwrap(); - assert!( - compressed.len() <= compress_bound(Compression::Gzip, size), - "{} > bound for {size}", - compressed.len() - ); - } - } - #[tokio::test] async fn precompressed_put_uses_exact_size() { // A size that would exceed the limit once inflated by `compress_bound`, but is sent diff --git a/clients/rust/tests/e2e.rs b/clients/rust/tests/e2e.rs index 0a81eb88..bfa1922f 100644 --- a/clients/rust/tests/e2e.rs +++ b/clients/rust/tests/e2e.rs @@ -120,7 +120,9 @@ async fn roundtrips_gzip() { .token(test_token_generator()) .build() .unwrap(); - let usecase = Usecase::new("usecase").with_compression(Compression::Gzip); + let usecase = Usecase::new("usecase") + .unwrap() + .with_compression(Compression::Gzip); let session = client.session(usecase.for_organization(12345)).unwrap(); let stored_id = session.put("oh hai!").send().await.unwrap().key; From d4ef03072c3abf9657e9fe88cbd979233b6f9f35 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Thu, 22 Oct 2026 14:41:25 +0200 Subject: [PATCH 5/7] fix(py-client): Handle gzip download stream ownership and seeking --- .../src/objectstore_client/_compression.py | 28 +++++++++++- clients/python/tests/test_compression.py | 27 ++++++++++++ clients/python/tests/test_upload.py | 44 +++++++++++++++++++ 3 files changed, 97 insertions(+), 2 deletions(-) diff --git a/clients/python/src/objectstore_client/_compression.py b/clients/python/src/objectstore_client/_compression.py index 8444a86b..d8f9267d 100644 --- a/clients/python/src/objectstore_client/_compression.py +++ b/clients/python/src/objectstore_client/_compression.py @@ -3,6 +3,7 @@ from __future__ import annotations import gzip +import io import zlib from typing import IO, cast @@ -17,6 +18,28 @@ _CHUNK_SIZE = 64 * 1024 +class _GzipReader(gzip.GzipFile): + """Decode gzip from a stream, closing the source with the reader.""" + + def __init__(self, source: IO[bytes]) -> None: + super().__init__(fileobj=source, mode="rb") + self._source = source + + def seekable(self) -> bool: + return False + + def seek(self, offset: int, whence: int = io.SEEK_SET) -> int: + raise io.UnsupportedOperation("gzip download stream is not seekable") + + def close(self) -> None: + if self.closed: + return + try: + super().close() + finally: + self._source.close() + + def _gzip_compressor() -> _Compressor: """Return a zlib compressor that emits the gzip format. @@ -75,13 +98,14 @@ def decompressing_reader(stream: IO[bytes], compression: str) -> IO[bytes]: """Return a readable stream decoding `stream` compressed with `compression`. Concatenated zstd frames and gzip members, as produced by multipart uploads - with individually compressed parts, are decoded as a single payload. + with individually compressed parts, are decoded as a single payload. The + returned reader is non-seekable and closes the source when closed. """ if compression == "zstd": dctx = zstandard.ZstdDecompressor() return cast(IO[bytes], dctx.stream_reader(stream, read_across_frames=True)) if compression == "gzip": - return cast(IO[bytes], gzip.GzipFile(fileobj=stream, mode="rb")) + return cast(IO[bytes], _GzipReader(stream)) raise NotImplementedError( f"Transparent decoding of `{compression}` is not implemented" ) diff --git a/clients/python/tests/test_compression.py b/clients/python/tests/test_compression.py index 03ae08a8..762d8136 100644 --- a/clients/python/tests/test_compression.py +++ b/clients/python/tests/test_compression.py @@ -1,8 +1,11 @@ import gzip import io import random +from typing import IO, cast +from unittest.mock import Mock import pytest +import urllib3 import zstandard from objectstore_client import _compression from objectstore_client.metadata import Compression @@ -72,6 +75,30 @@ def test_decompressing_reader_reads_concatenated_members( ) +def test_gzip_reader_closes_partially_read_http_response() -> None: + connection = Mock() + body = io.BytesIO(gzip.compress(PAYLOAD)) + response = urllib3.HTTPResponse( + body=body, + preload_content=False, + connection=connection, + ) + reader = _compression.decompressing_reader(cast(IO[bytes], response), "gzip") + + assert reader.read(10) == PAYLOAD[:10] + assert not response.closed + assert not reader.seekable() + with pytest.raises(io.UnsupportedOperation): + reader.seek(0) + + reader.close() + assert body.closed + connection.close.assert_called_once_with() + + reader.close() + connection.close.assert_called_once_with() + + def test_decompressing_reader_rejects_unknown() -> None: with pytest.raises(NotImplementedError): _compression.decompressing_reader(io.BytesIO(b""), "lz4") diff --git a/clients/python/tests/test_upload.py b/clients/python/tests/test_upload.py index 39e9c33b..7e0a78c6 100644 --- a/clients/python/tests/test_upload.py +++ b/clients/python/tests/test_upload.py @@ -1,11 +1,14 @@ +import gzip from io import BytesIO from typing import Any from unittest.mock import Mock import pytest import urllib3 +import zstandard from objectstore_client import Client, RequestError, Session, Usecase from objectstore_client._resumable import NoUploadProgress +from objectstore_client.metadata import Compression @pytest.fixture @@ -122,3 +125,44 @@ def test_sizing_must_restore_position( session.put(source) request.assert_not_called() + + +@pytest.mark.parametrize("compression", ["gzip", "zstd"]) +def test_put_downloaded_compressed_stream( + session: Session, monkeypatch: pytest.MonkeyPatch, compression: Compression +) -> None: + payload = b"downloaded data" * 100 + compressed = ( + gzip.compress(payload) + if compression == "gzip" + else zstandard.ZstdCompressor().compress(payload) + ) + download = urllib3.HTTPResponse( + body=BytesIO(compressed), + headers={"content-encoding": compression}, + preload_content=False, + decode_content=False, + ) + close_download = Mock(wraps=download.close) + monkeypatch.setattr(download, "close", close_download) + uploaded: list[bytes] = [] + + def request(method: str, _url: str, **kwargs: Any) -> urllib3.HTTPResponse: + if method == "GET": + return download + assert method == "PUT" + assert "Content-Length" not in kwargs["headers"] + uploaded.append(kwargs["body"].read()) + return urllib3.HTTPResponse(status=201, body=b'{"key":"copy"}') + + monkeypatch.setattr(session._pool, "request", request) + retrieved = session.get("original") + assert retrieved is not None + assert not retrieved.payload.seekable() + + assert session.put(retrieved.payload, key="copy") == "copy" + assert uploaded == [payload] + assert not retrieved.payload.closed + + retrieved.payload.close() + close_download.assert_called_once_with() From 23b7ad63b16f57479a960ef2ce61a3e5e21e3578 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Thu, 22 Oct 2026 14:47:42 +0200 Subject: [PATCH 6/7] ref: Consolidate gzip tests and simplify decoder comments --- .../src/objectstore_client/_compression.py | 5 ++- clients/rust/src/get.rs | 4 +-- objectstore-service/src/backend/gcs.rs | 34 ++----------------- 3 files changed, 5 insertions(+), 38 deletions(-) diff --git a/clients/python/src/objectstore_client/_compression.py b/clients/python/src/objectstore_client/_compression.py index d8f9267d..ffb7dbf2 100644 --- a/clients/python/src/objectstore_client/_compression.py +++ b/clients/python/src/objectstore_client/_compression.py @@ -97,9 +97,8 @@ def compress_into( def decompressing_reader(stream: IO[bytes], compression: str) -> IO[bytes]: """Return a readable stream decoding `stream` compressed with `compression`. - Concatenated zstd frames and gzip members, as produced by multipart uploads - with individually compressed parts, are decoded as a single payload. The - returned reader is non-seekable and closes the source when closed. + Concatenated zstd frames and gzip members are decoded as a single payload. + The returned reader is non-seekable and closes the source when closed. """ if compression == "zstd": dctx = zstandard.ZstdDecompressor() diff --git a/clients/rust/src/get.rs b/clients/rust/src/get.rs index 42bfee02..6f5cf86e 100644 --- a/clients/rust/src/get.rs +++ b/clients/rust/src/get.rs @@ -134,9 +134,7 @@ pub(crate) fn maybe_decompress( .compression .is_some_and(|c| accept_encoding.contains(&c)); match (metadata.compression, decompress && !encoding_accepted) { - // Multipart uploads with compression, when each part is compressed individually, - // will consist of multiple concatenated zstd frames or gzip members. - // This allows the client to handle automatic decompression for these objects transparently. + // Compressed payloads may contain concatenated zstd frames or gzip members. (Some(Compression::Zstd), true) => { metadata.compression = None; let mut decoder = ZstdDecoder::new(StreamReader::new(stream)); diff --git a/objectstore-service/src/backend/gcs.rs b/objectstore-service/src/backend/gcs.rs index 8c09a9c7..7df1b0dc 100644 --- a/objectstore-service/src/backend/gcs.rs +++ b/objectstore-service/src/backend/gcs.rs @@ -2859,9 +2859,9 @@ mod tests { } /// GCS decompresses `Content-Encoding: gzip` objects on download unless the request accepts - /// gzip. The backend must return the payload exactly as stored. + /// gzip. Full and range reads must return the stored compressed bytes. #[tokio::test] - async fn test_gzip_payload_roundtrip() -> Result<()> { + async fn test_gzip_payload_full_and_range_reads() -> Result<()> { use objectstore_types::metadata::Compression; let backend = create_test_backend().await?; @@ -2897,36 +2897,6 @@ mod tests { "Payload should be returned still compressed, not transcoded" ); - Ok(()) - } - - /// Range reads of gzip objects must apply to the stored (compressed) bytes, which requires - /// opting out of decompressive transcoding. - #[tokio::test] - async fn test_gzip_payload_range_read() -> Result<()> { - use objectstore_types::metadata::Compression; - - let backend = create_test_backend().await?; - - let plaintext = b"hello, world (but compressed with gzip)"; - let compressed = gzip_encode(plaintext); - - let id = make_id(); - let metadata = Metadata { - content_type: "text/plain".into(), - compression: Some(Compression::Gzip), - ..Default::default() - }; - - backend - .put_object( - &id, - &metadata, - stream::single(compressed.clone()), - Timestamp::now(), - ) - .await?; - let (meta, content_range, stream) = backend .get_object(&id, Timestamp::now(), Some(ByteRange::Bounded(2, 9))) .await? From 21b4ec5f0c17861f6bbecea97b4cc507af4d21ed Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Thu, 22 Oct 2026 17:24:29 +0200 Subject: [PATCH 7/7] ref: Address gzip client review feedback --- clients/python/README.md | 10 +-- .../src/objectstore_client/_compression.py | 87 ++++++++++++++++--- .../python/src/objectstore_client/client.py | 2 +- .../python/src/objectstore_client/utils.py | 67 +------------- clients/python/tests/test_compression.py | 44 ++++++++-- clients/python/tests/test_e2e.py | 30 ++++--- clients/python/tests/test_utils.py | 36 -------- clients/rust/src/get.rs | 10 +-- clients/rust/src/many.rs | 13 ++- clients/rust/src/put.rs | 35 ++++---- 10 files changed, 164 insertions(+), 170 deletions(-) diff --git a/clients/python/README.md b/clients/python/README.md index ae59ccd2..a5bd93f9 100644 --- a/clients/python/README.md +++ b/clients/python/README.md @@ -92,18 +92,18 @@ session.put(b"payload", origin="203.0.113.42") ### Compression Uploads are compressed with Zstd by default, and downloads are transparently decompressed. -The supported algorithms are `"zstd"` and `"gzip"`. Compression can be set per-usecase -or overridden per-upload: +The supported algorithms are `"zstd"` and `"gzip"`, or `"none"` to disable compression. +Compression can be set per-usecase or overridden per-upload: ```python -# compress all uploads of this usecase with gzip: -usecase = Usecase("my-usecase", compression="gzip") +# compress all uploads of this usecase with zstd (the default): +usecase = Usecase("my-usecase", compression="zstd") # upload as-is and record no encoding: session.put(video_data, compress="none") # upload as-is, but record the encoding so that downloads still decompress: -session.put(gzip_data, precompressed="gzip") +session.put(zstd_data, precompressed="zstd") ``` ### Resumable Uploads diff --git a/clients/python/src/objectstore_client/_compression.py b/clients/python/src/objectstore_client/_compression.py index ffb7dbf2..bc06a31e 100644 --- a/clients/python/src/objectstore_client/_compression.py +++ b/clients/python/src/objectstore_client/_compression.py @@ -4,18 +4,80 @@ import gzip import io +import shutil import zlib -from typing import IO, cast +from typing import IO, Protocol, cast import zstandard from objectstore_client.metadata import Compression -from objectstore_client.utils import _CompressionReader, _Compressor SUPPORTED_COMPRESSIONS: tuple[Compression, ...] = ("zstd", "gzip") """Compression algorithms that the client can apply and transparently decode.""" -_CHUNK_SIZE = 64 * 1024 + +class _Compressor(Protocol): + """A streaming compression object, such as `zlib.compressobj()`.""" + + def compress(self, data: bytes, /) -> bytes: ... + + def flush(self) -> bytes: ... + + +class _CompressionReader(io.RawIOBase): + """ + A readable stream that compresses `source` with `compressor` as it is read. + + `tell()` reports the number of compressed bytes produced so far. `seek(0)` is + allowed as long as no data has been read yet, so that urllib3 can record the body + position. Closing the reader does not close `source`. + """ + + _CHUNK_SIZE = 64 * 1024 + + def __init__(self, source: IO[bytes], compressor: _Compressor) -> None: + super().__init__() + self._source = source + self._compressor = compressor + self._buffer = bytearray() + self._finished = False + self._position = 0 + + def readable(self) -> bool: + return True + + def read(self, size: int | None = -1, /) -> bytes: + if size is None or size < 0: + size = -1 + while not self._finished and (size < 0 or len(self._buffer) < size): + if chunk := self._source.read(self._CHUNK_SIZE): + self._buffer += self._compressor.compress(chunk) + else: + self._buffer += self._compressor.flush() + self._finished = True + + if size < 0: + size = len(self._buffer) + out = bytes(self._buffer[:size]) + del self._buffer[:size] + self._position += len(out) + return out + + def readall(self) -> bytes: + return self.read(-1) + + def readinto(self, buffer: memoryview | bytearray) -> int: # type: ignore[override] + data = self.read(len(buffer)) + buffer[: len(data)] = data + return len(data) + + def tell(self) -> int: + return self._position + + def seek(self, offset: int, whence: int = io.SEEK_SET, /) -> int: + if offset == 0 and whence == io.SEEK_SET and self._position == 0: + return 0 + raise OSError("Cannot seek in a compressed stream after reading has started") class _GzipReader(gzip.GzipFile): @@ -79,18 +141,21 @@ def compressing_reader(source: IO[bytes], compression: Compression) -> IO[bytes] def compress_into( source: IO[bytes], dest: IO[bytes], compression: Compression ) -> tuple[int, int]: - """Compress `source` into `dest`, returning the bytes read and written.""" + """Compress `source` into `dest`, returning the bytes read and written. + + For gzip, `dest` must support `tell()` to measure the compressed size. + """ if compression == "zstd": read, written = zstandard.ZstdCompressor().copy_stream(source, dest) return read, written if compression == "gzip": - compressor = _gzip_compressor() - read = written = 0 - while chunk := source.read(_CHUNK_SIZE): - read += len(chunk) - written += dest.write(compressor.compress(chunk)) - written += dest.write(compressor.flush()) - return read, written + start = dest.tell() + with gzip.GzipFile( + filename="", fileobj=dest, mode="wb", compresslevel=6, mtime=0 + ) as writer: + shutil.copyfileobj(source, writer) + read = writer.tell() + return read, dest.tell() - start raise ValueError(f"Invalid compression: {compression}") diff --git a/clients/python/src/objectstore_client/client.py b/clients/python/src/objectstore_client/client.py index e83c16a6..cfdedcd6 100644 --- a/clients/python/src/objectstore_client/client.py +++ b/clients/python/src/objectstore_client/client.py @@ -90,7 +90,7 @@ def __init__( """Create a usecase. Raises ``ValueError`` if the name violates the naming rules on ``Usecase``, - or if the ``compression`` or ``resumable_threshold_bytes`` is invalid. + or if either ``compression`` or ``resumable_threshold_bytes`` is invalid. """ if not 3 <= len(name) <= 64: raise ValueError("usecase name must be 3–64 ASCII characters long") diff --git a/clients/python/src/objectstore_client/utils.py b/clients/python/src/objectstore_client/utils.py index a31e0c2a..1c43588f 100644 --- a/clients/python/src/objectstore_client/utils.py +++ b/clients/python/src/objectstore_client/utils.py @@ -1,7 +1,6 @@ from __future__ import annotations -import io -from typing import IO, Protocol +from typing import IO from urllib.parse import quote, unquote import filetype # type: ignore[import-untyped] @@ -125,67 +124,3 @@ def guess_mime_type(contents: bytes | IO[bytes]) -> str | None: kind = filetype.guess(header) return kind.mime if kind else None - - -class _Compressor(Protocol): - """A streaming compression object, such as `zlib.compressobj()`.""" - - def compress(self, data: bytes, /) -> bytes: ... - - def flush(self) -> bytes: ... - - -class _CompressionReader(io.RawIOBase): - """ - A readable stream that compresses `source` with `compressor` as it is read. - - `tell()` reports the number of compressed bytes produced so far. `seek(0)` is - allowed as long as no data has been read yet, so that urllib3 can record the body - position. Closing the reader does not close `source`. - """ - - _CHUNK_SIZE = 64 * 1024 - - def __init__(self, source: IO[bytes], compressor: _Compressor) -> None: - super().__init__() - self._source = source - self._compressor = compressor - self._buffer = bytearray() - self._finished = False - self._position = 0 - - def readable(self) -> bool: - return True - - def read(self, size: int | None = -1, /) -> bytes: - if size is None or size < 0: - size = -1 - while not self._finished and (size < 0 or len(self._buffer) < size): - if chunk := self._source.read(self._CHUNK_SIZE): - self._buffer += self._compressor.compress(chunk) - else: - self._buffer += self._compressor.flush() - self._finished = True - - if size < 0: - size = len(self._buffer) - out = bytes(self._buffer[:size]) - del self._buffer[:size] - self._position += len(out) - return out - - def readall(self) -> bytes: - return self.read(-1) - - def readinto(self, buffer: memoryview | bytearray) -> int: # type: ignore[override] - data = self.read(len(buffer)) - buffer[: len(data)] = data - return len(data) - - def tell(self) -> int: - return self._position - - def seek(self, offset: int, whence: int = io.SEEK_SET, /) -> int: - if offset == 0 and whence == io.SEEK_SET and self._position == 0: - return 0 - raise OSError("Cannot seek in a compressed stream after reading has started") diff --git a/clients/python/tests/test_compression.py b/clients/python/tests/test_compression.py index 762d8136..abdf1abb 100644 --- a/clients/python/tests/test_compression.py +++ b/clients/python/tests/test_compression.py @@ -14,6 +14,38 @@ PAYLOAD = random.Random(0).randbytes(300_000) + b"tail" * 1000 +def _make_wrapper(data: bytes = b"hello world") -> _compression._CompressionReader: + compressor = zstandard.ZstdCompressor().compressobj() + return _compression._CompressionReader(io.BytesIO(data), compressor) + + +def test_seek_zero_before_read() -> None: + wrapper = _make_wrapper() + assert wrapper.seek(0) == 0 + assert wrapper.seek(0) == 0 + + +def test_seek_nonzero_before_read() -> None: + wrapper = _make_wrapper() + with pytest.raises(OSError, match="Cannot seek"): + wrapper.seek(1) + + +def test_seek_zero_after_read() -> None: + wrapper = _make_wrapper() + wrapper.read(1) + with pytest.raises(OSError, match="Cannot seek"): + wrapper.seek(0) + + +def test_read_and_tell_forwarded() -> None: + wrapper = _make_wrapper() + assert wrapper.tell() == 0 + data = wrapper.read(4) + assert len(data) > 0 + assert wrapper.tell() > 0 + + @pytest.mark.parametrize("compression", ["zstd", "gzip"]) @pytest.mark.parametrize("read_size", [-1, 1, 4096, 100_000]) def test_compressing_reader_roundtrip(compression: Compression, read_size: int) -> None: @@ -35,7 +67,7 @@ def test_compressing_reader_roundtrip(compression: Compression, read_size: int) @pytest.mark.parametrize("compression", ["zstd", "gzip"]) -def test_compressing_reader_seeks_to_start_only_before_reading( +def test_compressing_reader_seek( compression: Compression, ) -> None: reader = _compression.compressing_reader(io.BytesIO(PAYLOAD), compression) @@ -56,7 +88,7 @@ def test_compress_into(compression: Compression) -> None: assert _compression.decompressing_reader(dest, compression).read() == PAYLOAD -def test_gzip_output_is_standard() -> None: +def test_gzip_format() -> None: reader = _compression.compressing_reader(io.BytesIO(PAYLOAD), "gzip") assert gzip.decompress(reader.read()) == PAYLOAD @@ -65,9 +97,7 @@ def test_gzip_output_is_standard() -> None: "compression,compress", [("zstd", zstandard.ZstdCompressor().compress), ("gzip", gzip.compress)], ) -def test_decompressing_reader_reads_concatenated_members( - compression: str, compress: object -) -> None: +def test_concatenated_members(compression: str, compress: object) -> None: assert callable(compress) stream = io.BytesIO(compress(b"hello ") + compress(b"world")) assert _compression.decompressing_reader(stream, compression).read() == ( @@ -75,7 +105,7 @@ def test_decompressing_reader_reads_concatenated_members( ) -def test_gzip_reader_closes_partially_read_http_response() -> None: +def test_gzip_partial_read_close() -> None: connection = Mock() body = io.BytesIO(gzip.compress(PAYLOAD)) response = urllib3.HTTPResponse( @@ -99,7 +129,7 @@ def test_gzip_reader_closes_partially_read_http_response() -> None: connection.close.assert_called_once_with() -def test_decompressing_reader_rejects_unknown() -> None: +def test_unknown_compression() -> None: with pytest.raises(NotImplementedError): _compression.decompressing_reader(io.BytesIO(b""), "lz4") diff --git a/clients/python/tests/test_e2e.py b/clients/python/tests/test_e2e.py index 86b33620..8987dfc4 100644 --- a/clients/python/tests/test_e2e.py +++ b/clients/python/tests/test_e2e.py @@ -505,7 +505,7 @@ def test_full_cycle_gzip(server_url: str) -> None: assert gzip.decompress(raw.payload.read()) == b"test data" * 100 -def test_precompressed_gzip_is_stored_verbatim(server_url: str) -> None: +def test_precompressed_gzip(server_url: str) -> None: client = Client(server_url, token=TestSecretKey.get()) session = client.session(Usecase("test-usecase"), org=42, project=1337) @@ -522,7 +522,7 @@ def test_precompressed_gzip_is_stored_verbatim(server_url: str) -> None: assert retrieved.payload.read() == b"test data" -def test_rejects_unsupported_compression(server_url: str) -> None: +def test_unsupported_compression(server_url: str) -> None: client = Client(server_url, token=TestSecretKey.get()) session = client.session(Usecase("test-usecase"), org=42, project=1337) @@ -952,20 +952,20 @@ def test_put_stores_under_literal_key(server_url: str) -> None: @pytest.mark.parametrize("source_kind", ["bytes", "buffer", "stream"]) -@pytest.mark.parametrize( - "compression", - ["none", "zstd", "gzip", "precompressed-zstd", "precompressed-gzip"], -) +@pytest.mark.parametrize("compression", ["none", "zstd", "gzip"]) +@pytest.mark.parametrize("precompressed", [False, True]) @pytest.mark.parametrize("threshold", [1, 2048]) def test_automatic_resumable_upload( - server_url: str, source_kind: str, compression: str, threshold: int + server_url: str, + source_kind: str, + compression: Compression, + precompressed: bool, + threshold: int, ) -> None: - precompressed = compression.startswith("precompressed-") - algorithm = cast(Compression, compression.removeprefix("precompressed-")) session = Client(server_url, token=TestSecretKey.get()).session( Usecase( "test-usecase", - compression="none" if precompressed else algorithm, + compression="none" if precompressed else compression, expiration_policy=TimeToLive(timedelta(days=1)), resumable_threshold_bytes=threshold, ), @@ -973,9 +973,9 @@ def test_automatic_resumable_upload( ) contents = b"file contents\n" * 100 encoded = contents - if algorithm == "zstd" and precompressed: + if compression == "zstd" and precompressed: encoded = zstandard.ZstdCompressor().compress(contents) - elif algorithm == "gzip" and precompressed: + elif compression == "gzip" and precompressed: encoded = gzip.compress(contents) with ( BytesIO() @@ -989,7 +989,9 @@ def test_automatic_resumable_upload( source.seek(len(b"skip this prefix")) key = session.put( source if source_kind in ("buffer", "stream") else encoded, - precompressed=algorithm if precompressed else None, + precompressed=compression + if precompressed and compression != "none" + else None, content_type="text/plain", metadata={"source": "file"}, origin="203.0.113.42", @@ -999,7 +1001,7 @@ def test_automatic_resumable_upload( assert not source.closed stored = session.head(key) assert stored is not None - assert stored.compression == (algorithm if algorithm != "none" else None) + assert stored.compression == (compression if compression != "none" else None) assert stored.content_type == "text/plain" assert stored.filename == "example.txt" assert stored.origin == "203.0.113.42" diff --git a/clients/python/tests/test_utils.py b/clients/python/tests/test_utils.py index b7dea74e..c6f78523 100644 --- a/clients/python/tests/test_utils.py +++ b/clients/python/tests/test_utils.py @@ -1,47 +1,11 @@ -import io - import pytest -import zstandard from objectstore_client.utils import ( - _CompressionReader, decode_header_value, encode_header_value, parse_accept_encoding, ) -def _make_wrapper(data: bytes = b"hello world") -> _CompressionReader: - compressor = zstandard.ZstdCompressor().compressobj() - return _CompressionReader(io.BytesIO(data), compressor) - - -def test_seek_zero_before_read() -> None: - wrapper = _make_wrapper() - assert wrapper.seek(0) == 0 - assert wrapper.seek(0) == 0 - - -def test_seek_nonzero_before_read() -> None: - wrapper = _make_wrapper() - with pytest.raises(OSError, match="Cannot seek"): - wrapper.seek(1) - - -def test_seek_zero_after_read() -> None: - wrapper = _make_wrapper() - wrapper.read(1) - with pytest.raises(OSError, match="Cannot seek"): - wrapper.seek(0) - - -def test_read_and_tell_forwarded() -> None: - wrapper = _make_wrapper() - assert wrapper.tell() == 0 - data = wrapper.read(4) - assert len(data) > 0 - assert wrapper.tell() > 0 - - def test_parse_accept_encoding_single() -> None: assert parse_accept_encoding("zstd") == ["zstd"] diff --git a/clients/rust/src/get.rs b/clients/rust/src/get.rs index 6f5cf86e..44085c14 100644 --- a/clients/rust/src/get.rs +++ b/clients/rust/src/get.rs @@ -118,7 +118,7 @@ impl GetBuilder { } } -/// Wraps a stream in a decompression layer matching the object's compression. +/// Wraps a stream in a decompression layer. /// /// Decompresses if the metadata indicates compression, `decompress` is `true`, /// and the stored encoding is not listed in `accept_encoding`. When the stored encoding @@ -134,7 +134,6 @@ pub(crate) fn maybe_decompress( .compression .is_some_and(|c| accept_encoding.contains(&c)); match (metadata.compression, decompress && !encoding_accepted) { - // Compressed payloads may contain concatenated zstd frames or gzip members. (Some(Compression::Zstd), true) => { metadata.compression = None; let mut decoder = ZstdDecoder::new(StreamReader::new(stream)); @@ -145,6 +144,7 @@ pub(crate) fn maybe_decompress( (Some(Compression::Gzip), true) => { metadata.compression = None; let mut decoder = GzipDecoder::new(StreamReader::new(stream)); + // Gzip streams may contain concatenated members (RFC 1952 §2.2). decoder.multiple_members(true); ReaderStream::new(decoder).boxed() } @@ -264,7 +264,7 @@ mod tests { } #[tokio::test] - async fn gzip_decompresses() { + async fn gzip_decompression() { let payload = b"hello world"; let stream = raw_stream(&gzip_compress(payload)); let mut metadata = gzip_metadata(); @@ -275,7 +275,7 @@ mod tests { } #[tokio::test] - async fn gzip_accept_skips_decompression() { + async fn gzip_accept_encoding() { let compressed = gzip_compress(b"hello world"); let stream = raw_stream(&compressed); let mut metadata = gzip_metadata(); @@ -286,7 +286,7 @@ mod tests { } #[tokio::test] - async fn gzip_concatenated_members_decompress() { + async fn gzip_concatenated_members() { let mut compressed = gzip_compress(b"hello "); compressed.extend(gzip_compress(b"world")); let stream = raw_stream(&compressed); diff --git a/clients/rust/src/many.rs b/clients/rust/src/many.rs index 7153eaec..751d02ba 100644 --- a/clients/rust/src/many.rs +++ b/clients/rust/src/many.rs @@ -543,12 +543,11 @@ fn compress_bound(compression: Compression, size: usize) -> usize { const GZIP_FRAMING_SIZE: usize = 18; /// Returns the worst-case size of a raw deflate stream for `size` input bytes. -/// -/// Neither `flate2` nor `miniz_oxide` expose a bound, so this mirrors zlib's -/// `compressBound` without its 6-byte zlib wrapper. Incompressible input is emitted -/// as stored blocks with a 5-byte header each, which this conservatively bounds at -/// 5 bytes per 16 KiB of input plus a constant for the final block. fn deflate_bound(size: usize) -> usize { + // Neither `flate2` nor `miniz_oxide` expose a bound, so this mirrors zlib's + // `compressBound` without its 6-byte zlib wrapper. Incompressible input is emitted + // as stored blocks with a 5-byte header each, which this conservatively bounds at + // 5 bytes per 16 KiB of input plus a constant for the final block. size + (size >> 12) + (size >> 14) + (size >> 25) + 7 } @@ -875,10 +874,10 @@ mod tests { } #[tokio::test] - async fn gzip_put_uses_compress_bound() { + async fn gzip_put_bound() { let size = 1_000_000; let op = put_with_compression(size, CompressionMode::Compress(Compression::Gzip)); - core::assert_matches!( + std::assert_matches!( classify(op).await, Classified::Batchable(_, s) if s == compress_bound(Compression::Gzip, size) as u64 ); diff --git a/clients/rust/src/put.rs b/clients/rust/src/put.rs index e4f1b1d6..2528ecc1 100644 --- a/clients/rust/src/put.rs +++ b/clients/rust/src/put.rs @@ -1,7 +1,6 @@ use std::fmt; use std::io::{self, Cursor}; use std::path::PathBuf; -use std::pin::Pin; use std::{borrow::Cow, collections::BTreeMap}; use async_compression::tokio::bufread::{GzipEncoder, ZstdEncoder}; @@ -274,27 +273,27 @@ pub(crate) async fn encode_body(body: PutBody, mode: Option) -> Some(CompressionMode::Precompressed(_)) | None => None, }; - let Some(compression) = compression else { - return Ok(match body { + Ok(match compression { + None => match body { PutBody::Buffer(bytes) => bytes.into(), PutBody::Stream(stream) => Body::wrap_stream(stream), PutBody::File(file) => Body::wrap_stream(ReaderStream::new(file)), PutBody::Path(path) => Body::wrap_stream(ReaderStream::new(File::open(path).await?)), - }); - }; - - let reader: Pin> = match body { - PutBody::Buffer(bytes) => Box::pin(Cursor::new(bytes)), - PutBody::Stream(stream) => Box::pin(StreamReader::new(stream)), - PutBody::File(file) => Box::pin(BufReader::new(file)), - PutBody::Path(path) => Box::pin(BufReader::new(File::open(path).await?)), - }; + }, + Some(compression) => match body { + PutBody::Buffer(bytes) => compress(Cursor::new(bytes), compression), + PutBody::Stream(stream) => compress(StreamReader::new(stream), compression), + PutBody::File(file) => compress(BufReader::new(file), compression), + PutBody::Path(path) => compress(BufReader::new(File::open(path).await?), compression), + }, + }) +} - let stream = match compression { - Compression::Zstd => ReaderStream::new(ZstdEncoder::new(reader)).boxed(), - Compression::Gzip => ReaderStream::new(GzipEncoder::new(reader)).boxed(), - }; - Ok(Body::wrap_stream(stream)) +fn compress(reader: impl AsyncBufRead + Send + 'static, compression: Compression) -> Body { + match compression { + Compression::Zstd => Body::wrap_stream(ReaderStream::new(ZstdEncoder::new(reader))), + Compression::Gzip => Body::wrap_stream(ReaderStream::new(GzipEncoder::new(reader))), + } } // TODO: instead of a separate `send` method, it would be nice to just implement `IntoFuture`. @@ -370,7 +369,7 @@ mod tests { } #[tokio::test] - async fn gzip_compresses_all_body_kinds() { + async fn gzip_body_kinds() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("payload"); std::fs::write(&path, b"hello world").unwrap();