delivery_blob/compression/
compression_algorithm.rs

1// Copyright 2026 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5//! Compression algorithms supported by chunked-compression and corresponding compressors and
6//! decompressors.
7//!
8//! The compressors and decompressors are enums rather than traits with multiple implementations
9//! because the enums are small and avoid the heap allocation of `Box<dyn Decompressor>`.
10
11use super::{FormatError, ZstdError};
12use crate::compression::ChunkedArchiveError;
13use std::mem::MaybeUninit;
14use std::ptr::NonNull;
15use storage_ptr_slice::PtrByteSlice;
16use zstd::zstd_safe::zstd_sys;
17
18thread_local! {
19    static ZSTD_COMPRESSOR: std::cell::RefCell<zstd::zstd_safe::CCtx<'static>> =
20        std::cell::RefCell::new({
21            let mut cctx = zstd::zstd_safe::CCtx::create();
22            cctx.set_parameter(zstd::zstd_safe::CParameter::ChecksumFlag(true)).unwrap();
23            cctx
24        });
25    static ZSTD_DECOMPRESSOR: std::cell::RefCell<RawDCtx> = {
26        // SAFETY: Creating a ZSTD decompression context does not access
27        // invalid memory or violate invariants.
28        let raw_ptr = unsafe { zstd_sys::ZSTD_createDCtx() };
29        let ptr = NonNull::new(raw_ptr).expect("ZSTD_createDCtx failed");
30        std::cell::RefCell::new(RawDCtx(ptr))
31    };
32}
33
34struct RawDCtx(NonNull<zstd_sys::ZSTD_DCtx>);
35impl Drop for RawDCtx {
36    fn drop(&mut self) {
37        // SAFETY: `self.0` is non-null and was allocated via `ZSTD_createDCtx`.
38        unsafe {
39            zstd_sys::ZSTD_freeDCtx(self.0.as_ptr());
40        }
41    }
42}
43
44/// Decompresses ZSTD using a pointer slice and a raw destination pointer slice.
45///
46/// # Safety
47///
48/// `dst` must point to valid memory allocated for writes of at least
49/// `dst.len()` bytes.
50unsafe fn zstd_decompress_ptr(
51    src: PtrByteSlice<'_>,
52    dst: *mut [MaybeUninit<u8>],
53) -> Result<usize, usize> {
54    ZSTD_DECOMPRESSOR.with_borrow_mut(|decompressor| {
55        let dctx = decompressor.0.as_ptr();
56        // SAFETY: `PtrByteSlice` guarantees `src` is valid for reads. `dst`
57        // points to valid memory allocated for writes of at least `dst.len()`
58        // bytes (ensured by caller safety preconditions).
59        let result = unsafe {
60            zstd_sys::ZSTD_decompressDCtx(
61                dctx,
62                dst as *mut std::os::raw::c_void,
63                dst.len(),
64                src.as_raw_slice_ptr() as *const std::os::raw::c_void,
65                src.len(),
66            )
67        };
68        // SAFETY: Checking an integer return code is pure inspection.
69        if unsafe { zstd_sys::ZSTD_isError(result) } != 0 { Err(result) } else { Ok(result) }
70    })
71}
72
73unsafe extern "C" {
74    fn LZ4_decompress_safe(
75        src: *const std::os::raw::c_char,
76        dst: *mut std::os::raw::c_char,
77        compressedSize: std::os::raw::c_int,
78        dstCapacity: std::os::raw::c_int,
79    ) -> std::os::raw::c_int;
80}
81
82/// Decompresses LZ4 using a pointer slice and a raw destination pointer slice.
83///
84/// # Safety
85///
86/// `dst` must point to valid memory allocated for writes of at least
87/// `dst.len()` bytes.
88unsafe fn lz4_decompress_ptr(
89    src: PtrByteSlice<'_>,
90    dst: *mut [MaybeUninit<u8>],
91) -> Result<usize, lz4::Error> {
92    if src.is_empty() {
93        return Ok(0);
94    }
95    if dst.len() == 0 {
96        return Err(lz4::Error::DecompressionFailed);
97    }
98    // SAFETY: `PtrByteSlice` guarantees `src` is valid for reads. `dst`
99    // points to valid memory allocated for writes of at least `dst.len()`
100    // bytes (ensured by caller safety preconditions).
101    let result = unsafe {
102        LZ4_decompress_safe(
103            src.as_raw_slice_ptr() as *const std::os::raw::c_char,
104            dst as *mut std::os::raw::c_char,
105            src.len().try_into().map_err(|_| lz4::Error::InputTooLarge)?,
106            dst.len().try_into().map_err(|_| lz4::Error::InputTooLarge)?,
107        )
108    };
109    if result < 0 { Err(lz4::Error::DecompressionFailed) } else { Ok(result as usize) }
110}
111
112/// The compression algorithm used to compress the chunks.
113#[derive(Copy, Clone, Debug, Eq, PartialEq)]
114#[repr(u8)]
115pub enum CompressionAlgorithm {
116    Zstd = 0,
117    Lz4 = 1,
118}
119
120impl CompressionAlgorithm {
121    /// Returns a decompressor that can decompress a chunk compressed with this compression
122    /// algorithm.
123    pub fn decompressor(&self) -> Decompressor {
124        match self {
125            Self::Zstd => Decompressor::Zstd,
126            Self::Lz4 => Decompressor::Lz4,
127        }
128    }
129
130    /// Returns a decompressor that can decompress a chunk compressed with this compression
131    /// algorithm. Some decompressors require a large state object that is expensive to create but
132    /// can be reused for many decompressions. A thread-local decompressor stores the state object
133    /// in a thread-local variable.
134    pub fn thread_local_decompressor(&self) -> ThreadLocalDecompressor {
135        match self {
136            Self::Zstd => ThreadLocalDecompressor::Zstd,
137            Self::Lz4 => ThreadLocalDecompressor::Lz4,
138        }
139    }
140}
141
142impl From<CompressionAlgorithm> for u8 {
143    fn from(value: CompressionAlgorithm) -> Self {
144        value as u8
145    }
146}
147
148impl TryFrom<u8> for CompressionAlgorithm {
149    type Error = ChunkedArchiveError;
150    fn try_from(value: u8) -> Result<Self, Self::Error> {
151        match value {
152            0 => Ok(CompressionAlgorithm::Zstd),
153            1 => Ok(CompressionAlgorithm::Lz4),
154            _ => Err(ChunkedArchiveError::IntegrityError),
155        }
156    }
157}
158
159/// A decompressor that is capable of decompressing chunks of a compressed archive.
160pub enum Decompressor {
161    Zstd,
162    Lz4,
163}
164
165impl Decompressor {
166    /// Decompresses a chunk of a chunked-compression archive.
167    pub fn decompress<'a>(
168        &mut self,
169        data: impl Into<PtrByteSlice<'a>>,
170        uncompressed_size: usize,
171        chunk_index: usize,
172    ) -> Result<Vec<u8>, ChunkedArchiveError> {
173        let src = data.into();
174        let mut buffer = Vec::with_capacity(uncompressed_size);
175        let dst = buffer.spare_capacity_mut() as *mut [MaybeUninit<u8>];
176        let len = match self {
177            Self::Zstd => {
178                // SAFETY: `dst` points to `uncompressed_size` bytes of capacity
179                // in `buffer`.
180                unsafe { zstd_decompress_ptr(src, dst) }.map_err(|code| {
181                    ChunkedArchiveError::DecompressionError {
182                        index: chunk_index,
183                        error: FormatError::Zstd(ZstdError(code)),
184                    }
185                })?
186            }
187            Self::Lz4 => {
188                // SAFETY: `dst` points to `uncompressed_size` bytes of capacity
189                // in `buffer`.
190                unsafe { lz4_decompress_ptr(src, dst) }.map_err(|e| {
191                    ChunkedArchiveError::DecompressionError {
192                        index: chunk_index,
193                        error: FormatError::Lz4(e),
194                    }
195                })?
196            }
197        };
198        // SAFETY: Decompression wrote `len` initialized bytes into `buffer`.
199        unsafe {
200            buffer.set_len(len);
201        }
202        Ok(buffer)
203    }
204
205    /// Decompresses a chunk of a chunked-compression archive into a pre-allocated buffer.
206    pub fn decompress_into<'a>(
207        &mut self,
208        data: impl Into<PtrByteSlice<'a>>,
209        destination: &mut [u8],
210        chunk_index: usize,
211    ) -> Result<usize, ChunkedArchiveError> {
212        let src = data.into();
213        let dst = destination as *mut [u8] as *mut [MaybeUninit<u8>];
214        match self {
215            Self::Zstd => {
216                // SAFETY: `dst` points to `destination.len()` bytes of valid
217                // memory in `destination`.
218                unsafe { zstd_decompress_ptr(src, dst) }.map_err(|code| {
219                    ChunkedArchiveError::DecompressionError {
220                        index: chunk_index,
221                        error: FormatError::Zstd(ZstdError(code)),
222                    }
223                })
224            }
225            Self::Lz4 => {
226                // SAFETY: `dst` points to `destination.len()` bytes of valid
227                // memory in `destination`.
228                unsafe { lz4_decompress_ptr(src, dst) }.map_err(|e| {
229                    ChunkedArchiveError::DecompressionError {
230                        index: chunk_index,
231                        error: FormatError::Lz4(e),
232                    }
233                })
234            }
235        }
236    }
237}
238
239#[derive(Copy, Clone)]
240/// A decompressor that uses thread-local storage to avoid reallocation of large state objects.
241pub enum ThreadLocalDecompressor {
242    Zstd,
243    Lz4,
244}
245
246impl ThreadLocalDecompressor {
247    /// Decompresses a chunk of a chunked-compression archive.
248    pub fn decompress<'a>(
249        &self,
250        data: impl Into<PtrByteSlice<'a>>,
251        uncompressed_size: usize,
252        chunk_index: usize,
253    ) -> Result<Vec<u8>, ChunkedArchiveError> {
254        let src = data.into();
255        let mut buffer = Vec::with_capacity(uncompressed_size);
256        let dst = buffer.spare_capacity_mut() as *mut [MaybeUninit<u8>];
257        let len = match self {
258            Self::Zstd => {
259                // SAFETY: `dst` points to `uncompressed_size` bytes of capacity
260                // in `buffer`.
261                unsafe { zstd_decompress_ptr(src, dst) }.map_err(|code| {
262                    ChunkedArchiveError::DecompressionError {
263                        index: chunk_index,
264                        error: FormatError::Zstd(ZstdError(code)),
265                    }
266                })?
267            }
268            Self::Lz4 => {
269                // SAFETY: `dst` points to `uncompressed_size` bytes of capacity
270                // in `buffer`.
271                unsafe { lz4_decompress_ptr(src, dst) }.map_err(|e| {
272                    ChunkedArchiveError::DecompressionError {
273                        index: chunk_index,
274                        error: FormatError::Lz4(e),
275                    }
276                })?
277            }
278        };
279        // SAFETY: Decompression wrote `len` initialized bytes into `buffer`.
280        unsafe {
281            buffer.set_len(len);
282        }
283        Ok(buffer)
284    }
285
286    /// Decompresses a chunk of a chunked-compression archive into a pre-allocated buffer.
287    pub fn decompress_into<'a>(
288        &self,
289        data: impl Into<PtrByteSlice<'a>>,
290        destination: &mut [u8],
291        chunk_index: usize,
292    ) -> Result<usize, ChunkedArchiveError> {
293        let src = data.into();
294        let dst = destination as *mut [u8] as *mut [MaybeUninit<u8>];
295        match self {
296            Self::Zstd => {
297                // SAFETY: `dst` points to `destination.len()` bytes of valid
298                // memory in `destination`.
299                unsafe { zstd_decompress_ptr(src, dst) }.map_err(|code| {
300                    ChunkedArchiveError::DecompressionError {
301                        index: chunk_index,
302                        error: FormatError::Zstd(ZstdError(code)),
303                    }
304                })
305            }
306            Self::Lz4 => {
307                // SAFETY: `dst` points to `destination.len()` bytes of valid
308                // memory in `destination`.
309                unsafe { lz4_decompress_ptr(src, dst) }.map_err(|e| {
310                    ChunkedArchiveError::DecompressionError {
311                        index: chunk_index,
312                        error: FormatError::Lz4(e),
313                    }
314                })
315            }
316        }
317    }
318}
319
320/// A compressor that is capable of compressing chunks of a chunked-compression archive.
321pub enum Compressor {
322    Zstd(zstd::zstd_safe::CCtx<'static>),
323    Lz4 { compression_level: lz4::HcCompressionLevel },
324}
325
326impl Compressor {
327    /// Compresses a chunk of a chunked-compression archive.
328    pub fn compress(
329        &mut self,
330        data: &[u8],
331        chunk_index: usize,
332    ) -> Result<Vec<u8>, ChunkedArchiveError> {
333        match self {
334            Self::Zstd(cctx) => {
335                let buffer_len = zstd::zstd_safe::compress_bound(data.len());
336                let mut buffer = Vec::with_capacity(buffer_len);
337                match cctx.compress2(&mut buffer, data) {
338                    Ok(_) => {
339                        buffer.shrink_to_fit();
340                        Ok(buffer)
341                    }
342                    Err(code) => Err(ChunkedArchiveError::CompressionError {
343                        index: chunk_index,
344                        error: FormatError::Zstd(ZstdError(code)),
345                    }),
346                }
347            }
348            Self::Lz4 { compression_level } => {
349                let mut buffer = lz4::compress_hc(data, *compression_level).map_err(|error| {
350                    ChunkedArchiveError::CompressionError {
351                        index: chunk_index,
352                        error: FormatError::Lz4(error),
353                    }
354                })?;
355                buffer.shrink_to_fit();
356                Ok(buffer)
357            }
358        }
359    }
360}
361
362#[derive(Copy, Clone)]
363/// A compressor that uses thread-local storage to avoid reallocation of large state objects.
364pub enum ThreadLocalCompressor {
365    Zstd { compression_level: i32 },
366    Lz4 { compression_level: lz4::HcCompressionLevel },
367}
368
369impl ThreadLocalCompressor {
370    /// Compresses a chunk of a chunked-compression archive.
371    pub fn compress(
372        &self,
373        data: &[u8],
374        chunk_index: usize,
375    ) -> Result<Vec<u8>, ChunkedArchiveError> {
376        match self {
377            Self::Zstd { compression_level } => ZSTD_COMPRESSOR.with_borrow_mut(|cctx| {
378                cctx.set_parameter(zstd::zstd_safe::CParameter::CompressionLevel(
379                    *compression_level,
380                ))
381                .expect("setting the compression level should never fail");
382                let buffer_len = zstd::zstd_safe::compress_bound(data.len());
383                let mut buffer = Vec::with_capacity(buffer_len);
384                match cctx.compress2(&mut buffer, data) {
385                    Ok(_) => {
386                        buffer.shrink_to_fit();
387                        Ok(buffer)
388                    }
389                    Err(code) => Err(ChunkedArchiveError::CompressionError {
390                        index: chunk_index,
391                        error: FormatError::Zstd(ZstdError(code)),
392                    }),
393                }
394            }),
395            Self::Lz4 { compression_level } => {
396                let mut buffer = lz4::compress_hc(data, *compression_level).map_err(|error| {
397                    ChunkedArchiveError::CompressionError {
398                        index: chunk_index,
399                        error: FormatError::Lz4(error),
400                    }
401                })?;
402                buffer.shrink_to_fit();
403                Ok(buffer)
404            }
405        }
406    }
407}
408
409#[cfg(test)]
410mod tests {
411    use super::*;
412    use crate::compression::ChunkedArchiveOptions;
413
414    const TEST_DATA: &[u8] = b"hello world this is some test data to compress and decompress";
415
416    #[test]
417    fn test_zstd_roundtrip() {
418        let options =
419            ChunkedArchiveOptions::V3 { compression_algorithm: CompressionAlgorithm::Zstd };
420        let mut compressor = options.compressor();
421        let compressed = compressor.compress(TEST_DATA, 0).unwrap();
422
423        let mut decompressor = CompressionAlgorithm::Zstd.decompressor();
424        let decompressed = decompressor.decompress(&compressed, TEST_DATA.len(), 0).unwrap();
425
426        assert_eq!(decompressed, TEST_DATA);
427    }
428
429    #[test]
430    fn test_lz4_roundtrip() {
431        let options =
432            ChunkedArchiveOptions::V3 { compression_algorithm: CompressionAlgorithm::Lz4 };
433        let mut compressor = options.compressor();
434        let compressed = compressor.compress(TEST_DATA, 0).unwrap();
435
436        let mut decompressor = CompressionAlgorithm::Lz4.decompressor();
437        let decompressed = decompressor.decompress(&compressed, TEST_DATA.len(), 0).unwrap();
438
439        assert_eq!(decompressed, TEST_DATA);
440    }
441
442    #[test]
443    fn test_thread_local_zstd_roundtrip() {
444        let options =
445            ChunkedArchiveOptions::V3 { compression_algorithm: CompressionAlgorithm::Zstd };
446        let compressor = options.thread_local_compressor();
447        let compressed = compressor.compress(TEST_DATA, 0).unwrap();
448
449        let decompressor = CompressionAlgorithm::Zstd.thread_local_decompressor();
450        let decompressed = decompressor.decompress(&compressed, TEST_DATA.len(), 0).unwrap();
451
452        assert_eq!(decompressed, TEST_DATA);
453    }
454
455    #[test]
456    fn test_thread_local_lz4_roundtrip() {
457        let options =
458            ChunkedArchiveOptions::V3 { compression_algorithm: CompressionAlgorithm::Lz4 };
459        let compressor = options.thread_local_compressor();
460        let compressed = compressor.compress(TEST_DATA, 0).unwrap();
461
462        let decompressor = CompressionAlgorithm::Lz4.thread_local_decompressor();
463        let decompressed = decompressor.decompress(&compressed, TEST_DATA.len(), 0).unwrap();
464
465        assert_eq!(decompressed, TEST_DATA);
466    }
467
468    #[test]
469    fn test_decompress_into() {
470        let options = ChunkedArchiveOptions::V2 {
471            minimum_chunk_size: 0,
472            chunk_alignment: 0,
473            compression_level: 1,
474        };
475        let mut compressor = options.compressor();
476        let compressed = compressor.compress(TEST_DATA, 0).unwrap();
477
478        let mut decompressor = CompressionAlgorithm::Zstd.decompressor();
479        let mut buffer = vec![0u8; TEST_DATA.len()];
480        let len = decompressor.decompress_into(&compressed, &mut buffer, 0).unwrap();
481
482        assert_eq!(len, TEST_DATA.len());
483        assert_eq!(buffer, TEST_DATA);
484    }
485
486    #[test]
487    fn test_algorithm_conversion() {
488        assert_eq!(u8::from(CompressionAlgorithm::Zstd), 0);
489        assert_eq!(u8::from(CompressionAlgorithm::Lz4), 1);
490
491        assert_eq!(CompressionAlgorithm::try_from(0).unwrap(), CompressionAlgorithm::Zstd);
492        assert_eq!(CompressionAlgorithm::try_from(1).unwrap(), CompressionAlgorithm::Lz4);
493        assert!(CompressionAlgorithm::try_from(2).is_err());
494    }
495
496    #[test]
497    fn test_decompress_into_ptr() {
498        for algorithm in [CompressionAlgorithm::Zstd, CompressionAlgorithm::Lz4] {
499            let options = ChunkedArchiveOptions::V3 { compression_algorithm: algorithm };
500            let compressor = options.thread_local_compressor();
501            let compressed = compressor.compress(TEST_DATA, 0).unwrap();
502
503            let decompressor = algorithm.thread_local_decompressor();
504            let mut buffer = vec![0u8; TEST_DATA.len()];
505            let len = decompressor
506                .decompress_into(PtrByteSlice::from(compressed.as_slice()), &mut buffer, 0)
507                .unwrap();
508
509            assert_eq!(len, TEST_DATA.len());
510            assert_eq!(buffer, TEST_DATA);
511        }
512    }
513}
delivery_blob/compression/compression_algorithm.rs

delivery_blob/compression/
compression_algorithm.rs
