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

delivery_blob/compression/
compression_info.rs
