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

delivery_blob/compression/
compression_info.rs
