mapping/
blob.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::Extents;
6use crate::reader::{BlockService, read_aligned_range};
7use delivery_blob::DataBuffer;
8use delivery_blob::compression::{CompressionInfo, StreamingDecompressor};
9use fuchsia_sync::Mutex;
10use std::cmp::min;
11use std::collections::HashMap;
12use std::ops::{ControlFlow, Range};
13use std::sync::Arc;
14
15/// A mapped blob containing extents and decompression metadata.
16pub struct Blob {
17    extents: Extents,
18    uncompressed_size: u64,
19    compression_info: Option<Arc<CompressionInfo>>,
20}
21
22impl Blob {
23    pub fn new(
24        extents: Extents,
25        uncompressed_size: u64,
26        compression_info: Option<CompressionInfo>,
27    ) -> Self {
28        Self { extents, uncompressed_size, compression_info: compression_info.map(Arc::new) }
29    }
30
31    /// Returns the extents mapping logical offsets to device offsets.
32    pub fn extents(&self) -> &Extents {
33        &self.extents
34    }
35
36    /// Returns the uncompressed size of the blob in bytes.
37    pub fn uncompressed_size(&self) -> u64 {
38        self.uncompressed_size
39    }
40
41    /// Returns decompression metadata if the blob is compressed.
42    pub fn compression_info(&self) -> Option<&CompressionInfo> {
43        self.compression_info.as_deref()
44    }
45
46    /// Streams and decodes the specified uncompressed `range` into the provided `dest_buf`.
47    ///
48    /// For uncompressed blobs, both `range.start` and `range.end` must be multiples of
49    /// `BLOCK_SIZE`. For compressed blobs, `range.start` must be a multiple of the compression
50    /// chunk size, and `range.end` must either be a multiple of the chunk size or equal to
51    /// `uncompressed_size`.
52    pub fn read_range(
53        &self,
54        range: Range<u64>,
55        service: &(impl BlockService + ?Sized),
56        mut dest_buf: impl DataBuffer,
57    ) {
58        if range.is_empty() {
59            return;
60        }
61
62        match &self.compression_info {
63            None => {
64                let mut current_offset = range.start;
65                let uncompressed_size = self.uncompressed_size;
66
67                read_aligned_range(&self.extents, range, service, move |res| {
68                    let buffer = match res {
69                        Ok(buf) => buf,
70                        Err(_) => {
71                            return ControlFlow::Break(());
72                        }
73                    };
74                    let valid_len =
75                        min(buffer.len() as u64, uncompressed_size.saturating_sub(current_offset))
76                            as usize;
77                    if valid_len > 0 {
78                        dest_buf
79                            .mut_ptr_slice()
80                            .subslice_mut(0..valid_len)
81                            .copy_from_ptr_slice(buffer.as_ptr_slice().subslice(0..valid_len));
82                        if dest_buf.commit(valid_len).is_err() {
83                            return ControlFlow::Break(());
84                        }
85                    }
86                    current_offset += buffer.len() as u64;
87                    ControlFlow::Continue(())
88                });
89            }
90            Some(info) => {
91                let info = Arc::clone(info);
92                let Ok((mut decompressor, aligned_range)) =
93                    StreamingDecompressor::new(info, range, self.uncompressed_size, dest_buf)
94                else {
95                    // The range must be out of range. This should be handled when `dest_buf`
96                    // is dropped.
97                    return;
98                };
99
100                read_aligned_range(&self.extents, aligned_range, service, move |res| {
101                    let buffer = match res {
102                        Ok(buf) => buf,
103                        Err(_) => {
104                            return ControlFlow::Break(());
105                        }
106                    };
107                    if decompressor.push(buffer.as_ptr_slice()).is_err() {
108                        return ControlFlow::Break(());
109                    }
110                    ControlFlow::Continue(())
111                });
112            }
113        }
114    }
115}
116
117/// A thread-safe registry of active [`Blob`] instances indexed by their Zircon pager port key.
118#[derive(Default)]
119pub struct Blobs {
120    map: Mutex<HashMap<u64, Arc<Blob>>>,
121}
122
123impl Blobs {
124    /// Creates a new empty blob registry.
125    pub fn new() -> Self {
126        Self::default()
127    }
128
129    /// Inserts a blob into the registry under `key`, returning the previous blob if one existed.
130    pub fn insert(&self, key: u64, blob: Arc<Blob>) -> Option<Arc<Blob>> {
131        self.map.lock().insert(key, blob)
132    }
133
134    /// Retrieves a cloned handle to the blob registered under `key`, or `None` if not present.
135    pub fn get(&self, key: u64) -> Option<Arc<Blob>> {
136        self.map.lock().get(&key).cloned()
137    }
138
139    /// Removes and returns the blob registered under `key`, or `None` if not present.
140    pub fn remove(&self, key: u64) -> Option<Arc<Blob>> {
141        self.map.lock().remove(&key)
142    }
143
144    /// Returns the number of blobs in the registry.
145    pub fn len(&self) -> usize {
146        self.map.lock().len()
147    }
148
149    /// Returns `true` if the registry contains no blobs.
150    pub fn is_empty(&self) -> bool {
151        self.map.lock().is_empty()
152    }
153
154    /// Removes all blobs from the registry.
155    pub fn clear(&self) {
156        self.map.lock().clear();
157    }
158}
159
160#[cfg(test)]
161mod tests {
162    use super::*;
163    use crate::reader::tests::FakeBlockService;
164    use crate::testing::TestVecBuffer;
165    use crate::{BLOCK_SIZE, Extent};
166    use anyhow::Error;
167    use delivery_blob::compression::{ChunkedArchiveOptions, CompressionAlgorithm};
168    use std::sync::Arc;
169
170    #[test]
171    fn test_read_range_uncompressed() {
172        let block_count = 8;
173        let mut expected_data = vec![0u8; (block_count as u64 * BLOCK_SIZE) as usize];
174        for (i, byte) in expected_data.iter_mut().enumerate() {
175            *byte = (i % 255) as u8;
176        }
177        let service = FakeBlockService::new(expected_data.clone());
178
179        let extents = Extents::encode_extents(&[Extent::new(0..(8 * BLOCK_SIZE), Some(0))]);
180        let extents = Extents::from_encoded(&extents).unwrap();
181        let blob = Arc::new(Blob::new(extents, 8 * BLOCK_SIZE, None));
182
183        let (dest_buf, rx) = TestVecBuffer::new(expected_data.len());
184        blob.read_range(0..(8 * BLOCK_SIZE), &service, dest_buf);
185
186        assert_eq!(rx.commits(), vec![(0, (8 * BLOCK_SIZE) as usize)]);
187        assert_eq!(rx.output(), expected_data);
188    }
189
190    #[test]
191    fn test_read_range_compressed_zstd() {
192        let uncompressed_size = 32768 * 2 + 1024;
193        let mut uncompressed_data = vec![0u8; uncompressed_size];
194        for (i, byte) in uncompressed_data.iter_mut().enumerate() {
195            *byte = ((i * 7) % 255) as u8;
196        }
197
198        let options =
199            ChunkedArchiveOptions::V3 { compression_algorithm: CompressionAlgorithm::Zstd };
200        let archive =
201            delivery_blob::compression::ChunkedArchive::new(&uncompressed_data, options).unwrap();
202
203        let mut compressed_offsets = vec![0];
204        let mut compressed_data = vec![];
205        for chunk in archive.chunks() {
206            compressed_data.extend_from_slice(&chunk.compressed_data);
207            compressed_offsets.push(compressed_data.len() as u64);
208        }
209        compressed_offsets.pop();
210
211        let chunk_size = archive.chunk_size();
212        let stored_size = compressed_data.len() as u64;
213        let stored_blocks = stored_size.div_ceil(BLOCK_SIZE);
214        let mut device_data = vec![0u8; (stored_blocks * BLOCK_SIZE) as usize];
215        device_data[..compressed_data.len()].copy_from_slice(&compressed_data);
216        let service = FakeBlockService::new(device_data);
217
218        let extents =
219            Extents::encode_extents(&[Extent::new(0..(stored_blocks * BLOCK_SIZE), Some(0))]);
220        let extents = Extents::from_encoded(&extents).unwrap();
221        let compression_info = CompressionInfo::new(
222            chunk_size as u64,
223            stored_size,
224            &compressed_offsets,
225            CompressionAlgorithm::Zstd,
226        )
227        .unwrap();
228        let blob = Arc::new(Blob::new(extents, uncompressed_size as u64, Some(compression_info)));
229
230        let dest_alloc_size = uncompressed_size.next_multiple_of(chunk_size);
231        let (dest_buf, rx) = TestVecBuffer::new(dest_alloc_size);
232        blob.read_range(0..(uncompressed_size as u64), &service, dest_buf);
233
234        assert_eq!(
235            rx.commits(),
236            vec![(0, chunk_size), (chunk_size as u64, chunk_size), (chunk_size as u64 * 2, 1024)]
237        );
238        assert_eq!(&rx.output()[..uncompressed_size], &uncompressed_data[..]);
239    }
240
241    #[test]
242    fn test_read_range_compressed_lz4_split_across_buffers() {
243        let uncompressed_size = 32768 * 2;
244        let mut uncompressed_data = vec![0u8; uncompressed_size];
245        for (i, byte) in uncompressed_data.iter_mut().enumerate() {
246            *byte = ((i * 13) % 255) as u8;
247        }
248
249        let options =
250            ChunkedArchiveOptions::V3 { compression_algorithm: CompressionAlgorithm::Lz4 };
251        let archive =
252            delivery_blob::compression::ChunkedArchive::new(&uncompressed_data, options).unwrap();
253
254        let mut compressed_offsets = vec![0];
255        let mut compressed_data = vec![];
256        for chunk in archive.chunks() {
257            compressed_data.extend_from_slice(&chunk.compressed_data);
258            compressed_offsets.push(compressed_data.len() as u64);
259        }
260        compressed_offsets.pop();
261
262        let chunk_size = archive.chunk_size();
263        let stored_size = compressed_data.len() as u64;
264        let stored_blocks = stored_size.div_ceil(BLOCK_SIZE);
265        let mut device_data = vec![0u8; (stored_blocks * BLOCK_SIZE) as usize];
266        device_data[..compressed_data.len()].copy_from_slice(&compressed_data);
267
268        // Force a small block allocation limit (e.g. 4096 bytes) so that read_aligned_range
269        // splits the compressed chunks across multiple consecutive OwnedBuffers!
270        let service = FakeBlockService::new_with_cap(device_data, Some(4096));
271
272        let extents =
273            Extents::encode_extents(&[Extent::new(0..(stored_blocks * BLOCK_SIZE), Some(0))]);
274        let extents = Extents::from_encoded(&extents).unwrap();
275        let compression_info = CompressionInfo::new(
276            chunk_size as u64,
277            stored_size,
278            &compressed_offsets,
279            CompressionAlgorithm::Lz4,
280        )
281        .unwrap();
282        let blob = Arc::new(Blob::new(extents, uncompressed_size as u64, Some(compression_info)));
283
284        let (dest_buf, rx) = TestVecBuffer::new(uncompressed_size);
285        blob.read_range(0..(uncompressed_size as u64), &service, dest_buf);
286
287        assert_eq!(rx.commits(), vec![(0, chunk_size), (chunk_size as u64, chunk_size)]);
288        assert_eq!(rx.output(), uncompressed_data);
289    }
290
291    #[test]
292    fn test_read_range_invalid_range_noop() {
293        let service = FakeBlockService::new(vec![0u8; 8192]);
294        let extents = Extents::encode_extents(&[Extent::new(0..8192, Some(0))]);
295        let extents = Extents::from_encoded(&extents).unwrap();
296        let blob = Arc::new(Blob::new(extents, 8192, None));
297
298        let (dest_buf, rx) = TestVecBuffer::new_with_offset(0, 4096);
299        // start >= end should be a no-op returning Ok(())
300        blob.read_range(4096..4096, &service, dest_buf);
301        assert_eq!(rx.commits().len(), 0);
302    }
303
304    #[test]
305    fn test_blob_getters() {
306        let extents_raw = Extents::encode_extents(&[Extent::new(0..8192, Some(0))]);
307        let extents = Extents::from_encoded(&extents_raw).unwrap();
308        let uncompressed_size = 8192u64;
309
310        let blob_uncompressed = Blob::new(extents, uncompressed_size, None);
311        assert_eq!(blob_uncompressed.uncompressed_size(), 8192);
312        assert!(blob_uncompressed.compression_info().is_none());
313
314        let compression_info =
315            CompressionInfo::new(32768, 4096, &[0], CompressionAlgorithm::Zstd).unwrap();
316        let blob_compressed = Blob::new(
317            Extents::from_encoded(&extents_raw).unwrap(),
318            uncompressed_size,
319            Some(compression_info),
320        );
321        assert!(blob_compressed.compression_info().is_some());
322    }
323
324    #[test]
325    fn test_read_range_block_service_error_returns_err() {
326        struct FailingBlockService;
327        impl BlockService for FailingBlockService {
328            fn allocate_buffer(&self, max_len: usize) -> storage_device::buffer::OwnedBuffer {
329                FakeBlockService::new(vec![0u8; max_len]).allocate_buffer(max_len)
330            }
331            fn read_blocks(
332                &self,
333                _device_offset: u64,
334                _dest_buffer: storage_device::buffer::OwnedBuffer,
335                _on_complete: Box<
336                    dyn FnOnce(Result<storage_device::buffer::OwnedBuffer, Error>) + Send,
337                >,
338            ) -> Result<(), Error> {
339                Err(anyhow::anyhow!("block read failure"))
340            }
341        }
342
343        let extents = Extents::encode_extents(&[Extent::new(0..8192, Some(0))]);
344        let extents = Extents::from_encoded(&extents).unwrap();
345        let blob = Blob::new(extents, 8192, None);
346
347        let (dest_buf, rx) = TestVecBuffer::new(8192);
348
349        blob.read_range(0..8192, &FailingBlockService, dest_buf);
350        assert_eq!(rx.commits().len(), 0);
351    }
352
353    #[test]
354    fn test_read_range_uncompressed_multi_chunk() {
355        let block_count = 4;
356        let mut expected_data = vec![0u8; (block_count as u64 * BLOCK_SIZE) as usize];
357        for (i, byte) in expected_data.iter_mut().enumerate() {
358            *byte = ((i * 11) % 255) as u8;
359        }
360        // Force capping to 4096 bytes per buffer allocation so read_range processes
361        // 4 separate chunks.
362        let service = FakeBlockService::new_with_cap(expected_data.clone(), Some(4096));
363
364        let extents =
365            Extents::encode_extents(&[Extent::new(0..(block_count * BLOCK_SIZE), Some(0))]);
366        let extents = Extents::from_encoded(&extents).unwrap();
367        let blob = Blob::new(extents, block_count * BLOCK_SIZE, None);
368
369        let (dest_buf, rx) = TestVecBuffer::new(expected_data.len());
370        blob.read_range(0..(block_count * BLOCK_SIZE), &service, dest_buf);
371
372        assert_eq!(rx.commits().len(), 4);
373        assert_eq!(rx.output(), expected_data);
374    }
375
376    #[test]
377    fn test_read_range_uncompressed_unaligned_uncompressed_size() {
378        let uncompressed_size = 5000u64;
379        let mut expected_data = vec![0u8; 8192];
380        for (i, byte) in expected_data.iter_mut().enumerate() {
381            *byte = (i % 251) as u8;
382        }
383        let service = FakeBlockService::new(expected_data.clone());
384
385        let extents = Extents::encode_extents(&[Extent::new(0..8192, Some(0))]);
386        let extents = Extents::from_encoded(&extents).unwrap();
387        let blob = Blob::new(extents, uncompressed_size, None);
388
389        let (dest_buf, rx) = TestVecBuffer::new(8192);
390        blob.read_range(0..8192, &service, dest_buf);
391
392        assert_eq!(rx.commits(), vec![(0, 5000)]);
393        assert_eq!(&rx.output()[..5000], &expected_data[..5000]);
394    }
395
396    #[test]
397    fn test_read_range_compressed_tail_chunk_only() {
398        let uncompressed_size = 32768 * 2 + 1024;
399        let mut uncompressed_data = vec![0u8; uncompressed_size];
400        for (i, byte) in uncompressed_data.iter_mut().enumerate() {
401            *byte = ((i * 7) % 251) as u8;
402        }
403
404        let options =
405            ChunkedArchiveOptions::V3 { compression_algorithm: CompressionAlgorithm::Zstd };
406        let archive =
407            delivery_blob::compression::ChunkedArchive::new(&uncompressed_data, options).unwrap();
408
409        let mut compressed_offsets = vec![0];
410        let mut compressed_data = vec![];
411        for chunk in archive.chunks() {
412            compressed_data.extend_from_slice(&chunk.compressed_data);
413            compressed_offsets.push(compressed_data.len() as u64);
414        }
415        compressed_offsets.pop();
416
417        let chunk_size = archive.chunk_size();
418        let stored_size = compressed_data.len() as u64;
419        let stored_blocks = stored_size.div_ceil(BLOCK_SIZE);
420        let mut device_data = vec![0u8; (stored_blocks * BLOCK_SIZE) as usize];
421        device_data[..compressed_data.len()].copy_from_slice(&compressed_data);
422        let service = FakeBlockService::new(device_data);
423
424        let extents =
425            Extents::encode_extents(&[Extent::new(0..(stored_blocks * BLOCK_SIZE), Some(0))]);
426        let extents = Extents::from_encoded(&extents).unwrap();
427        let compression_info = CompressionInfo::new(
428            chunk_size as u64,
429            stored_size,
430            &compressed_offsets,
431            CompressionAlgorithm::Zstd,
432        )
433        .unwrap();
434        let blob = Blob::new(extents, uncompressed_size as u64, Some(compression_info));
435
436        let tail_start = chunk_size as u64 * 2;
437        let (dest_buf, rx) = TestVecBuffer::new_with_offset(32768, tail_start);
438        blob.read_range(tail_start..(uncompressed_size as u64), &service, dest_buf);
439
440        assert_eq!(rx.commits(), vec![(tail_start, 1024)]);
441        assert_eq!(&rx.output()[..1024], &uncompressed_data[65536..]);
442    }
443
444    #[test]
445    fn test_read_range_compressed_partial_final_chunk_zero_tail() {
446        let uncompressed_size = 32768 + 1024;
447        let mut uncompressed_data = vec![0u8; uncompressed_size];
448        for (i, byte) in uncompressed_data.iter_mut().enumerate() {
449            *byte = ((i * 13) % 251) as u8;
450        }
451
452        let options =
453            ChunkedArchiveOptions::V3 { compression_algorithm: CompressionAlgorithm::Zstd };
454        let archive =
455            delivery_blob::compression::ChunkedArchive::new(&uncompressed_data, options).unwrap();
456
457        let mut compressed_offsets = vec![0];
458        let mut compressed_data = vec![];
459        for chunk in archive.chunks() {
460            compressed_data.extend_from_slice(&chunk.compressed_data);
461            compressed_offsets.push(compressed_data.len() as u64);
462        }
463        compressed_offsets.pop();
464
465        let chunk_size = archive.chunk_size();
466        let stored_size = compressed_data.len() as u64;
467        let stored_blocks = stored_size.div_ceil(BLOCK_SIZE);
468        let mut device_data = vec![0u8; (stored_blocks * BLOCK_SIZE) as usize];
469        device_data[..compressed_data.len()].copy_from_slice(&compressed_data);
470        let service = FakeBlockService::new(device_data);
471
472        let extents =
473            Extents::encode_extents(&[Extent::new(0..(stored_blocks * BLOCK_SIZE), Some(0))]);
474        let extents = Extents::from_encoded(&extents).unwrap();
475        let compression_info = CompressionInfo::new(
476            chunk_size as u64,
477            stored_size,
478            &compressed_offsets,
479            CompressionAlgorithm::Zstd,
480        )
481        .unwrap();
482        let blob = Blob::new(extents, uncompressed_size as u64, Some(compression_info));
483
484        // Pre-fill destination buffer with 0xFF bytes to verify tail zeroing
485        let (mut dest_buf, rx) = TestVecBuffer::new(65536);
486        dest_buf.data.fill(0xFF);
487        blob.read_range(0..(uncompressed_size as u64), &service, dest_buf);
488
489        assert_eq!(rx.commits(), vec![(0, chunk_size), (chunk_size as u64, 1024)]);
490        assert_eq!(&rx.output()[..uncompressed_size], &uncompressed_data[..]);
491        assert_eq!(&rx.output()[uncompressed_size..65536], &[0u8; 31744]);
492    }
493
494    #[test]
495    fn test_blobs_registry() {
496        let extents = Extents::encode_extents(&[Extent::new(0..4096, Some(0))]);
497        let extents = Extents::from_encoded(&extents).unwrap();
498        let blob = Arc::new(Blob::new(extents, 4096, None));
499        let blobs = Blobs::new();
500
501        assert!(blobs.is_empty());
502        assert_eq!(blobs.len(), 0);
503        assert!(blobs.get(100).is_none());
504
505        blobs.insert(100, blob.clone());
506        assert_eq!(blobs.len(), 1);
507        assert!(!blobs.is_empty());
508        assert!(blobs.get(100).is_some());
509
510        blobs.remove(100);
511        assert!(blobs.is_empty());
512    }
513}
mapping/blob.rs

mapping/
blob.rs
