delivery_blob/compression/
compression_info.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
5use crate::compression::{ChunkedArchiveError, CompressionAlgorithm, ThreadLocalDecompressor};
6use std::borrow::Borrow;
7use std::ops::Range;
8use storage_ptr_slice::{MutPtrByteSlice, PtrByteSlice};
9
10/// Trait for destination buffers where uncompressed or decompressed blob data is written.
11pub trait DataBuffer: Send + 'static {
12    /// Returns the logical byte range for this buffer.
13    fn range(&self) -> Range<u64>;
14
15    /// Returns a raw pointer slice to the remaining uncommitted memory in this allocation.
16    fn mut_ptr_slice(&mut self) -> MutPtrByteSlice<'_>;
17
18    /// Incrementally commits `size` bytes of data within this allocation, advancing the start
19    /// of the remaining memory returned by subsequent calls to [`mut_ptr_slice`].
20    fn commit(&mut self, size: usize) -> Result<(), ChunkedArchiveError>;
21}
22
23pub struct CompressionInfo {
24    chunk_size: u64,
25    compressed_size: u64,
26    // The chunked compression format stores 0 as the first offset but it's not stored here. Not
27    // storing the 0 avoids the allocation for blobs smaller than the chunk size.
28    small_offsets: Box<[u32]>,
29    large_offsets: Box<[u64]>,
30    decompressor: ThreadLocalDecompressor,
31}
32
33impl CompressionInfo {
34    pub fn new(
35        chunk_size: u64,
36        compressed_size: u64,
37        offsets: &[u64],
38        compression_algorithm: CompressionAlgorithm,
39    ) -> Result<Self, ChunkedArchiveError> {
40        let decompressor = compression_algorithm.thread_local_decompressor();
41        if chunk_size == 0 {
42            return Err(ChunkedArchiveError::IntegrityError);
43        } else if offsets.is_empty() || *offsets.first().unwrap() != 0 {
44            // There should always be at least 1 offset and the first offset must always be 0.
45            return Err(ChunkedArchiveError::IntegrityError);
46        } else if !offsets.array_windows().all(|[a, b]| a < b) {
47            // The offsets must be in ascending order.
48            return Err(ChunkedArchiveError::IntegrityError);
49        } else if offsets.len() == 1 {
50            // Simple case where the blob is smaller than the chunk size so only the 0 offset is
51            // present. The 0 isn't stored so no allocation is necessary.
52            Ok(Self {
53                chunk_size,
54                compressed_size,
55                small_offsets: Box::default(),
56                large_offsets: Box::default(),
57                decompressor,
58            })
59        } else if *offsets.last().unwrap() <= u32::MAX as u64 {
60            // Check the last index first since most compressed blobs are going to be smaller
61            // than 4GiB making all offsets small.
62            Ok(Self {
63                chunk_size,
64                compressed_size,
65                small_offsets: offsets[1..].iter().map(|x| *x as u32).collect(),
66                large_offsets: Box::default(),
67                decompressor,
68            })
69        } else {
70            // The partition point is the index of the first compressed offset that's > u32::MAX.
71            let partition_point = offsets.partition_point(|&x| x <= u32::MAX as u64);
72            Ok(Self {
73                chunk_size,
74                compressed_size,
75                small_offsets: offsets[1..partition_point].iter().map(|x| *x as u32).collect(),
76                large_offsets: offsets[partition_point..].into(),
77                decompressor,
78            })
79        }
80    }
81
82    /// Returns the chunk size for this compressed blob.
83    pub fn chunk_size(&self) -> u64 {
84        self.chunk_size
85    }
86
87    /// Returns the total compressed size of this blob.
88    pub fn compressed_size(&self) -> u64 {
89        self.compressed_size
90    }
91
92    /// Returns the compressed range for the specified uncompressed range.
93    pub fn compressed_range_for_uncompressed_range(
94        &self,
95        range: &Range<u64>,
96    ) -> Result<Range<u64>, ChunkedArchiveError> {
97        if range.start % self.chunk_size != 0 || range.start >= range.end {
98            return Err(ChunkedArchiveError::IntegrityError);
99        }
100
101        let start_chunk_index = (range.start / self.chunk_size) as usize;
102        let start_offset = self
103            .compressed_offset_for_chunk_index(start_chunk_index)
104            .ok_or(ChunkedArchiveError::OutOfRange)?;
105
106        // The end of the range may not be aligned to the chunk size for the last chunk.
107        let end_chunk_index = range.end.div_ceil(self.chunk_size) as usize;
108        let end_offset = match self.compressed_offset_for_chunk_index(end_chunk_index) {
109            None => self.compressed_size,
110            Some(offset) => {
111                // This isn't the last chunk so the end must be aligned.
112                if !range.end.is_multiple_of(self.chunk_size) {
113                    return Err(ChunkedArchiveError::IntegrityError);
114                }
115                // `CompressionInfo::new` validates that all of the offsets are ascending.
116                offset
117            }
118        };
119
120        Ok(start_offset..end_offset)
121    }
122
123    fn compressed_offset_for_chunk_index(&self, chunk_index: usize) -> Option<u64> {
124        if chunk_index == 0 {
125            Some(0)
126        } else if chunk_index - 1 < self.small_offsets.len() {
127            Some(self.small_offsets[chunk_index - 1] as u64)
128        } else if chunk_index - 1 - self.small_offsets.len() < self.large_offsets.len() {
129            Some(self.large_offsets[chunk_index - 1 - self.small_offsets.len()])
130        } else {
131            None
132        }
133    }
134
135    /// Decompress the bytes of `src` into `dst`.
136    ///   - `src` is allowed to span multiple chunks.
137    ///   - `dst` must have the exact size of the uncompressed bytes.
138    ///   - `dst_start_offset` is the location of the uncompressed bytes within the blob and must be
139    ///     chunk aligned. This is necessary for determining the chunk boundaries in `src`.
140    pub fn decompress<'a>(
141        &self,
142        src: impl Into<PtrByteSlice<'a>>,
143        mut dst: &mut [u8],
144        dst_start_offset: u64,
145    ) -> Result<(), ChunkedArchiveError> {
146        let mut src = src.into();
147        if dst_start_offset % self.chunk_size != 0 {
148            return Err(ChunkedArchiveError::IntegrityError);
149        }
150
151        let start_chunk_index = (dst_start_offset / self.chunk_size) as usize;
152        let chunk_count = dst.len().div_ceil(self.chunk_size as usize);
153        let mut start_offset = self
154            .compressed_offset_for_chunk_index(start_chunk_index)
155            .ok_or(ChunkedArchiveError::IntegrityError)?;
156
157        // Decompress each chunk individually.
158        for chunk_index in start_chunk_index..(start_chunk_index + chunk_count) {
159            match self.compressed_offset_for_chunk_index(chunk_index + 1) {
160                Some(end_offset) => {
161                    let len = (end_offset - start_offset) as usize;
162                    if len > src.len() {
163                        return Err(ChunkedArchiveError::IntegrityError);
164                    }
165                    let (to_decompress, src_remaining) = src.split_at(len);
166                    let (to_decompress_into, dst_remaining) = dst
167                        .split_at_mut_checked(self.chunk_size as usize)
168                        .ok_or(ChunkedArchiveError::IntegrityError)?;
169
170                    let decompressed_bytes = self.decompressor.decompress_into(
171                        to_decompress,
172                        to_decompress_into,
173                        chunk_index,
174                    )?;
175                    if decompressed_bytes != to_decompress_into.len() {
176                        return Err(ChunkedArchiveError::IntegrityError);
177                    }
178                    src = src_remaining;
179                    dst = dst_remaining;
180                    start_offset = end_offset;
181                }
182                None => {
183                    let decompressed_bytes =
184                        self.decompressor.decompress_into(src, dst, chunk_index)?;
185                    if decompressed_bytes != dst.len() {
186                        return Err(ChunkedArchiveError::IntegrityError);
187                    }
188                }
189            }
190        }
191
192        Ok(())
193    }
194}
195
196/// Stateful streaming decompressor that receives compressed block buffers
197/// and decompresses complete chunks into `page_request`.
198pub struct StreamingDecompressor<C, B> {
199    /// Reference or owned container for blob compression metadata.
200    info: C,
201
202    /// Target destination implementing `DataBuffer`.
203    data_buffer: B,
204
205    /// Uncompressed logical byte range remaining to be decompressed.
206    range: Range<u64>,
207
208    /// Total uncompressed size of the blob.
209    uncompressed_size: u64,
210
211    /// Index of the chunk currently being decompressed.
212    chunk_index: usize,
213
214    /// Accumulates compressed bytes for chunks that straddle buffer boundaries.
215    accumulator: Vec<u8>,
216
217    /// The current compressed device byte offset expected for incoming buffers.
218    current_compressed_offset: u64,
219
220    /// Indicates if an error occurred during decompression. Once `true`, `push` fuses and returns
221    /// error.
222    failed: bool,
223}
224
225impl<C: Borrow<CompressionInfo>, B: DataBuffer> StreamingDecompressor<C, B> {
226    /// Creates a new streaming decompressor for `data_buffer`.
227    ///
228    /// Accepts any container `info` implementing `Borrow<CompressionInfo>` (e.g.
229    /// `&CompressionInfo` or `Arc<CompressionInfo>`).
230    /// Returns the decompressor and the block-aligned compressed byte range (`Range<u64>`) to
231    /// read from storage.
232    ///
233    /// `data_buffer.range().start` must be chunk aligned (a multiple of `chunk_size`).
234    pub fn new(
235        info: C,
236        uncompressed_size: u64,
237        data_buffer: B,
238    ) -> Result<(Self, Range<u64>), ChunkedArchiveError> {
239        const BLOCK_SIZE: u64 = 4096;
240        let range = data_buffer.range();
241        let compressed = info.borrow().compressed_range_for_uncompressed_range(&range)?;
242        let aligned = (compressed.start / BLOCK_SIZE) * BLOCK_SIZE
243            ..compressed.end.next_multiple_of(BLOCK_SIZE);
244
245        let chunk_size = info.borrow().chunk_size();
246        assert_eq!(range.start % chunk_size, 0, "range.start must be chunk aligned");
247        let chunk_index = (range.start / chunk_size) as usize;
248
249        let decompressor = StreamingDecompressor {
250            info,
251            data_buffer,
252            range,
253            uncompressed_size,
254            chunk_index,
255            accumulator: Vec::new(),
256            current_compressed_offset: aligned.start,
257            failed: false,
258        };
259
260        Ok((decompressor, aligned))
261    }
262
263    /// Pushes a newly read compressed block buffer slice and decompresses any complete chunks.
264    /// Fuses on error: if an error occurs or has previously occurred, returns `Err`.
265    pub fn push<'b>(
266        &mut self,
267        buffer_slice: impl Into<PtrByteSlice<'b>>,
268    ) -> Result<(), ChunkedArchiveError> {
269        let buffer_slice = buffer_slice.into();
270        if self.failed {
271            return Err(ChunkedArchiveError::IntegrityError);
272        }
273
274        if self.range.is_empty() {
275            return Ok(());
276        }
277
278        let buffer = self.current_compressed_offset
279            ..self.current_compressed_offset + buffer_slice.len() as u64;
280        self.current_compressed_offset = buffer.end;
281
282        let info = self.info.borrow();
283        let chunk_size = info.chunk_size();
284
285        while self.range.start < self.range.end {
286            let chunk_start = info
287                .compressed_offset_for_chunk_index(self.chunk_index)
288                .ok_or(ChunkedArchiveError::OutOfRange)?;
289            let chunk_end = info
290                .compressed_offset_for_chunk_index(self.chunk_index + 1)
291                .unwrap_or_else(|| info.compressed_size());
292            let chunk = chunk_start..chunk_end;
293
294            let decompress_chunk = |compressed_src: PtrByteSlice<'_>,
295                                    data_buffer: &mut B|
296             -> Result<(), ChunkedArchiveError> {
297                let buffer_len =
298                    std::cmp::min(data_buffer.mut_ptr_slice().len(), chunk_size as usize);
299                let mut dest_buffer = data_buffer.mut_ptr_slice().subslice_mut(0..buffer_len);
300                let remaining = (self.uncompressed_size.saturating_sub(self.range.start)) as usize;
301                let chunk_uncompressed_len = if remaining < chunk_size as usize {
302                    // Zero the block tail if this partial final chunk is smaller than chunk_size.
303                    let (head, mut tail) = dest_buffer.split_at_mut(remaining);
304                    tail.fill(0);
305                    dest_buffer = head;
306                    remaining
307                } else {
308                    chunk_size as usize
309                };
310
311                // SAFETY: `data_buffer` is exclusively held by this decompressor, and
312                // `dest_buffer` points to uncommitted remaining memory of length
313                // `chunk_uncompressed_len`.
314                let dst_slice = unsafe { &mut *dest_buffer.as_raw_mut_slice_ptr() };
315
316                let decompressed_bytes = info.decompressor.decompress_into(
317                    compressed_src,
318                    dst_slice,
319                    self.chunk_index,
320                )?;
321                if decompressed_bytes != chunk_uncompressed_len {
322                    return Err(ChunkedArchiveError::IntegrityError);
323                }
324                data_buffer.commit(buffer_len)?;
325                Ok(())
326            };
327
328            if chunk.start < buffer.start {
329                // Chunk started in a previous buffer; accumulate remainder and decompress if
330                // complete.
331                assert!(!self.accumulator.is_empty());
332                if chunk.end <= buffer.end {
333                    let needed = (chunk.end - buffer.start) as usize;
334                    buffer_slice.subslice(0..needed).append_to(&mut self.accumulator);
335                    let src = self.accumulator.as_slice().into();
336                    if let Err(e) = decompress_chunk(src, &mut self.data_buffer) {
337                        self.failed = true;
338                        return Err(e);
339                    }
340                    self.accumulator.clear();
341                    self.range.start += chunk_size;
342                    self.chunk_index += 1;
343                    continue;
344                } else {
345                    buffer_slice.append_to(&mut self.accumulator);
346                    break;
347                }
348            } else if chunk.end <= buffer.end {
349                // Chunk is fully contained in current buffer.
350                let rel_start = (chunk.start - buffer.start) as usize;
351                let rel_end = (chunk.end - buffer.start) as usize;
352                let compressed_slice = buffer_slice.subslice(rel_start..rel_end);
353
354                if let Err(e) = decompress_chunk(compressed_slice, &mut self.data_buffer) {
355                    self.failed = true;
356                    return Err(e);
357                }
358                self.range.start += chunk_size;
359                self.chunk_index += 1;
360            } else {
361                // Chunk extends past current buffer; accumulate prefix and await next buffer.
362                let rel_start = (chunk.start - buffer.start) as usize;
363                buffer_slice
364                    .subslice(rel_start..buffer_slice.len())
365                    .append_to(&mut self.accumulator);
366                break;
367            }
368        }
369        Ok(())
370    }
371}
372
373#[cfg(test)]
374mod tests {
375    use super::*;
376
377    #[test]
378    fn test_compression_info_new_small_and_large_offsets() {
379        let info = CompressionInfo::new(4096, 50, &[0], CompressionAlgorithm::Zstd).unwrap();
380        assert_eq!(info.chunk_size(), 4096);
381        assert_eq!(info.compressed_size(), 50);
382        assert_eq!(info.compressed_offset_for_chunk_index(0), Some(0));
383        assert_eq!(info.compressed_offset_for_chunk_index(1), None);
384
385        let info =
386            CompressionInfo::new(4096, 350, &[0, 100, 250], CompressionAlgorithm::Zstd).unwrap();
387        assert_eq!(info.compressed_offset_for_chunk_index(0), Some(0));
388        assert_eq!(info.compressed_offset_for_chunk_index(1), Some(100));
389        assert_eq!(info.compressed_offset_for_chunk_index(2), Some(250));
390        assert_eq!(info.compressed_offset_for_chunk_index(3), None);
391
392        let large_val = u32::MAX as u64 + 1000;
393        let info = CompressionInfo::new(
394            4096,
395            large_val + 500,
396            &[0, 500, large_val],
397            CompressionAlgorithm::Zstd,
398        )
399        .unwrap();
400        assert_eq!(info.compressed_offset_for_chunk_index(0), Some(0));
401        assert_eq!(info.compressed_offset_for_chunk_index(1), Some(500));
402        assert_eq!(info.compressed_offset_for_chunk_index(2), Some(large_val));
403        assert_eq!(info.compressed_offset_for_chunk_index(3), None);
404    }
405
406    #[test]
407    fn test_compressed_range_for_uncompressed_range() {
408        let info = CompressionInfo::new(4096, 500, &[0, 100, 250, 400], CompressionAlgorithm::Zstd)
409            .unwrap();
410        let range = info.compressed_range_for_uncompressed_range(&(0..4096)).unwrap();
411        assert_eq!(range, 0..100);
412
413        let range = info.compressed_range_for_uncompressed_range(&(4096..12288)).unwrap();
414        assert_eq!(range, 100..400);
415
416        let range = info.compressed_range_for_uncompressed_range(&(4096..16384)).unwrap();
417        assert_eq!(range, 100..500);
418    }
419
420    #[test]
421    fn test_compression_info_offsets_must_start_with_zero() {
422        assert!(CompressionInfo::new(4096, 100, &[], CompressionAlgorithm::Zstd).is_err());
423        assert!(CompressionInfo::new(4096, 100, &[1], CompressionAlgorithm::Zstd).is_err());
424        assert!(CompressionInfo::new(4096, 100, &[0], CompressionAlgorithm::Zstd).is_ok());
425    }
426
427    #[test]
428    fn test_compression_info_offsets_must_be_sorted() {
429        assert!(CompressionInfo::new(4096, 100, &[0, 1, 2], CompressionAlgorithm::Zstd).is_ok());
430        assert!(CompressionInfo::new(4096, 100, &[0, 2, 1], CompressionAlgorithm::Zstd).is_err());
431        assert!(CompressionInfo::new(4096, 100, &[0, 1, 1], CompressionAlgorithm::Zstd).is_err());
432    }
433
434    #[test]
435    fn test_compression_info_splitting_offsets() {
436        const MAX_SMALL_OFFSET: u64 = u32::MAX as u64;
437        let compression_info =
438            CompressionInfo::new(4096, 100, &[0], CompressionAlgorithm::Zstd).unwrap();
439        assert!(compression_info.small_offsets.is_empty());
440        assert!(compression_info.large_offsets.is_empty());
441
442        let compression_info =
443            CompressionInfo::new(4096, 20, &[0, 10], CompressionAlgorithm::Zstd).unwrap();
444        assert_eq!(&*compression_info.small_offsets, &[10]);
445        assert!(compression_info.large_offsets.is_empty());
446
447        let compression_info =
448            CompressionInfo::new(4096, 40, &[0, 10, 20, 30], CompressionAlgorithm::Zstd).unwrap();
449        assert_eq!(&*compression_info.small_offsets, &[10, 20, 30]);
450        assert!(compression_info.large_offsets.is_empty());
451
452        let compression_info = CompressionInfo::new(
453            4096,
454            MAX_SMALL_OFFSET,
455            &[0, MAX_SMALL_OFFSET - 1],
456            CompressionAlgorithm::Zstd,
457        )
458        .unwrap();
459        assert_eq!(&*compression_info.small_offsets, &[u32::MAX - 1]);
460        assert!(compression_info.large_offsets.is_empty());
461
462        let compression_info = CompressionInfo::new(
463            4096,
464            MAX_SMALL_OFFSET + 1,
465            &[0, MAX_SMALL_OFFSET],
466            CompressionAlgorithm::Zstd,
467        )
468        .unwrap();
469        assert_eq!(&*compression_info.small_offsets, &[u32::MAX]);
470        assert!(compression_info.large_offsets.is_empty());
471
472        let compression_info = CompressionInfo::new(
473            4096,
474            MAX_SMALL_OFFSET + 2,
475            &[0, MAX_SMALL_OFFSET + 1],
476            CompressionAlgorithm::Zstd,
477        )
478        .unwrap();
479        assert!(compression_info.small_offsets.is_empty());
480        assert_eq!(&*compression_info.large_offsets, &[MAX_SMALL_OFFSET + 1]);
481
482        let compression_info = CompressionInfo::new(
483            4096,
484            MAX_SMALL_OFFSET + 2,
485            &[0, MAX_SMALL_OFFSET - 1, MAX_SMALL_OFFSET, MAX_SMALL_OFFSET + 1],
486            CompressionAlgorithm::Zstd,
487        )
488        .unwrap();
489        assert_eq!(&*compression_info.small_offsets, &[u32::MAX - 1, u32::MAX]);
490        assert_eq!(&*compression_info.large_offsets, &[MAX_SMALL_OFFSET + 1]);
491
492        let compression_info = CompressionInfo::new(
493            4096,
494            MAX_SMALL_OFFSET + 20,
495            &[0, MAX_SMALL_OFFSET + 10],
496            CompressionAlgorithm::Zstd,
497        )
498        .unwrap();
499        assert!(compression_info.small_offsets.is_empty());
500        assert_eq!(&*compression_info.large_offsets, &[MAX_SMALL_OFFSET + 10]);
501
502        let compression_info = CompressionInfo::new(
503            4096,
504            MAX_SMALL_OFFSET + 30,
505            &[0, MAX_SMALL_OFFSET + 10, MAX_SMALL_OFFSET + 20],
506            CompressionAlgorithm::Zstd,
507        )
508        .unwrap();
509        assert!(compression_info.small_offsets.is_empty());
510        assert_eq!(
511            &*compression_info.large_offsets,
512            &[MAX_SMALL_OFFSET + 10, MAX_SMALL_OFFSET + 20]
513        );
514    }
515
516    struct TestBuffer {
517        data: Vec<u8>,
518        range: Range<u64>,
519        committed: usize,
520    }
521
522    impl TestBuffer {
523        fn new(size: usize) -> Self {
524            Self { data: vec![0u8; size], range: 0..size as u64, committed: 0 }
525        }
526
527        fn new_with_range(range: Range<u64>) -> Self {
528            let size = (range.end - range.start) as usize;
529            Self { data: vec![0u8; size], range, committed: 0 }
530        }
531    }
532
533    impl DataBuffer for TestBuffer {
534        fn range(&self) -> Range<u64> {
535            self.range.clone()
536        }
537
538        fn mut_ptr_slice(&mut self) -> MutPtrByteSlice<'_> {
539            let slice = &mut self.data[self.committed..];
540            unsafe { MutPtrByteSlice::new(slice as *mut [u8]) }
541        }
542
543        fn commit(&mut self, size: usize) -> Result<(), ChunkedArchiveError> {
544            self.committed += size;
545            Ok(())
546        }
547    }
548
549    #[test]
550    fn test_streaming_decompressor_single_buffer() {
551        let uncompressed_data: Vec<u8> = (0..32768).map(|i| (i % 251) as u8).collect();
552        let options = crate::compression::ChunkedArchiveOptions::V3 {
553            compression_algorithm: CompressionAlgorithm::Zstd,
554        };
555        let archive = crate::compression::ChunkedArchive::new(&uncompressed_data, options).unwrap();
556
557        let mut compressed_offsets = vec![0];
558        let mut compressed_data = vec![];
559        for chunk in archive.chunks() {
560            compressed_data.extend_from_slice(&chunk.compressed_data);
561            compressed_offsets.push(compressed_data.len() as u64);
562        }
563        compressed_offsets.pop();
564
565        let info = CompressionInfo::new(
566            archive.chunk_size() as u64,
567            compressed_data.len() as u64,
568            &compressed_offsets,
569            CompressionAlgorithm::Zstd,
570        )
571        .unwrap();
572
573        let buf = TestBuffer::new(32768);
574        let (mut decompressor, aligned) = StreamingDecompressor::new(&info, 32768, buf).unwrap();
575        assert_eq!(aligned, 0..4096);
576
577        decompressor.push(&compressed_data).unwrap();
578        assert_eq!(&decompressor.data_buffer.data[..32768], &uncompressed_data[..]);
579    }
580
581    #[test]
582    fn test_streaming_decompressor_straddled_buffers() {
583        let uncompressed_data: Vec<u8> = (0..65536).map(|i| (i % 251) as u8).collect();
584        let options = crate::compression::ChunkedArchiveOptions::V3 {
585            compression_algorithm: CompressionAlgorithm::Zstd,
586        };
587        let archive = crate::compression::ChunkedArchive::new(&uncompressed_data, options).unwrap();
588
589        let mut compressed_offsets = vec![0];
590        let mut compressed_data = vec![];
591        for chunk in archive.chunks() {
592            compressed_data.extend_from_slice(&chunk.compressed_data);
593            compressed_offsets.push(compressed_data.len() as u64);
594        }
595        compressed_offsets.pop();
596
597        let info = CompressionInfo::new(
598            archive.chunk_size() as u64,
599            compressed_data.len() as u64,
600            &compressed_offsets,
601            CompressionAlgorithm::Zstd,
602        )
603        .unwrap();
604
605        let buf = TestBuffer::new(65536);
606        let (mut decompressor, _) = StreamingDecompressor::new(&info, 65536, buf).unwrap();
607
608        for slice in compressed_data.chunks(10) {
609            decompressor.push(slice).unwrap();
610        }
611        assert_eq!(&decompressor.data_buffer.data[..65536], &uncompressed_data[..]);
612    }
613
614    #[test]
615    fn test_streaming_decompressor_partial_last_chunk_zero_tail() {
616        let uncompressed_size = 32768 + 1024;
617        let uncompressed_data: Vec<u8> = (0..uncompressed_size).map(|i| (i % 251) as u8).collect();
618
619        let options = crate::compression::ChunkedArchiveOptions::V3 {
620            compression_algorithm: CompressionAlgorithm::Zstd,
621        };
622        let archive = crate::compression::ChunkedArchive::new(&uncompressed_data, options).unwrap();
623
624        let mut compressed_offsets = vec![0];
625        let mut compressed_data = vec![];
626        for chunk in archive.chunks() {
627            compressed_data.extend_from_slice(&chunk.compressed_data);
628            compressed_offsets.push(compressed_data.len() as u64);
629        }
630        compressed_offsets.pop();
631
632        let info = CompressionInfo::new(
633            archive.chunk_size() as u64,
634            compressed_data.len() as u64,
635            &compressed_offsets,
636            CompressionAlgorithm::Zstd,
637        )
638        .unwrap();
639
640        let mut buf = TestBuffer::new_with_range(0..uncompressed_size as u64);
641        buf.data.resize(65536, 0);
642        buf.data.fill(0xFF);
643
644        let (mut decompressor, _) =
645            StreamingDecompressor::new(&info, uncompressed_size as u64, buf).unwrap();
646        decompressor.push(&compressed_data).unwrap();
647
648        assert_eq!(&decompressor.data_buffer.data[..uncompressed_size], &uncompressed_data[..]);
649        assert_eq!(&decompressor.data_buffer.data[uncompressed_size..65536], &[0u8; 31744]);
650    }
651
652    #[test]
653    fn test_streaming_decompressor_unaligned_start_returns_err() {
654        let info = CompressionInfo::new(4096, 500, &[0], CompressionAlgorithm::Zstd).unwrap();
655        let buf = TestBuffer::new_with_range(100..4096);
656        assert!(StreamingDecompressor::new(&info, 4096, buf).is_err());
657    }
658
659    #[test]
660    fn test_streaming_decompressor_fused_error() {
661        let info = CompressionInfo::new(4096, 500, &[0], CompressionAlgorithm::Zstd).unwrap();
662        let buf = TestBuffer::new(4096);
663        let (mut decompressor, _) = StreamingDecompressor::new(&info, 4096, buf).unwrap();
664
665        let invalid_compressed_data = vec![0xFFu8; 4096];
666        assert!(decompressor.push(&invalid_compressed_data).is_err());
667        assert!(decompressor.push(&invalid_compressed_data).is_err());
668    }
669}
delivery_blob/compression/compression_info.rs

delivery_blob/compression/
compression_info.rs
