fxfs_platform_testing/fuchsia/fxblob/
mapping_provider.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::fuchsia::errors::map_to_status;
6use crate::fuchsia::fxblob::blob::FxBlob;
7use crate::fuchsia::fxblob::directory::BlobDirectory;
8use crate::fuchsia::node::OpenedNode;
9use anyhow::Error;
10use fidl_fuchsia_storage_mapping as fmapping;
11use fuchsia_merkle::Hash;
12use futures::TryStreamExt;
13use fxfs::errors::FxfsError;
14use log::{error, warn};
15use mapping::{
16    Extents, MAPPING_VMO_SIZE, MappingCommand, PENDING_COMMANDS_CAPACITY, RawMappingCommand,
17};
18use std::collections::HashMap;
19use std::sync::Arc;
20use vmo_fifo::AsyncSender;
21
22/// BlobMappingProvider services requests to open mapping sessions over a given BlobDirectory.
23pub struct BlobMappingProvider {
24    blob_directory: Arc<BlobDirectory>,
25}
26
27impl BlobMappingProvider {
28    pub fn new(blob_directory: Arc<BlobDirectory>) -> Result<Self, Error> {
29        Ok(Self { blob_directory })
30    }
31
32    pub async fn handle_mapping_provider_requests(
33        self: Arc<Self>,
34        mut stream: fmapping::MappingProviderRequestStream,
35    ) {
36        while let Ok(Some(request)) = stream.try_next().await {
37            match request {
38                fmapping::MappingProviderRequest::OpenSession { session, responder } => {
39                    let vmo = match zx::Vmo::create(MAPPING_VMO_SIZE) {
40                        Ok(v) => v,
41                        Err(e) => {
42                            let _ = responder.send(Err(e.into_raw()));
43                            continue;
44                        }
45                    };
46                    let sender = match AsyncSender::<RawMappingCommand>::new(
47                        vmo,
48                        8,
49                        PENDING_COMMANDS_CAPACITY,
50                    ) {
51                        Ok(s) => s,
52                        Err(e) => {
53                            let _ = responder.send(Err(e.into_raw()));
54                            continue;
55                        }
56                    };
57                    let vmo_clone = match sender.vmo().duplicate_handle(zx::Rights::SAME_RIGHTS) {
58                        Ok(v) => v,
59                        Err(e) => {
60                            let _ = responder.send(Err(e.into_raw()));
61                            continue;
62                        }
63                    };
64                    if let Err(error) = responder.send(Ok(vmo_clone)) {
65                        error!(error:?; "Failed to send open session response");
66                    } else {
67                        let mapping_session =
68                            BlobMappingSession::new(self.blob_directory.clone(), sender);
69                        self.blob_directory.volume().scope().spawn(async move {
70                            mapping_session
71                                .handle_mapping_session_requests(session.into_stream())
72                                .await;
73                        });
74                    }
75                }
76                fmapping::MappingProviderRequest::_UnknownMethod { ordinal, .. } => {
77                    warn!(ordinal; "Unknown MappingProvider method");
78                }
79            }
80        }
81    }
82}
83
84/// `BlobMappingSession` services requests to open or close blobs within a `BlobDirectory`.
85/// Upon opening a blob, the blob's underlying extents (data and merkle blocks) are retrieved and
86/// then forwarded to client asynchronously via `sender`. The client provides a session-unique
87/// `key` for each opened blob, which can also be referenced during a `Close` request.
88pub struct BlobMappingSession {
89    blob_directory: Arc<BlobDirectory>,
90    sender: AsyncSender<RawMappingCommand>,
91    opened_blobs: HashMap<u64, OpenedNode<FxBlob>>,
92}
93
94impl BlobMappingSession {
95    pub fn new(blob_directory: Arc<BlobDirectory>, sender: AsyncSender<RawMappingCommand>) -> Self {
96        Self { blob_directory, sender, opened_blobs: HashMap::new() }
97    }
98
99    /// Retrieves the extent mappings for the blob and registers the blob in the mapping session.
100    /// Returns the uncompressed byte size of the blob.
101    async fn open_blob(&mut self, key: u64, hash: Hash) -> Result<u64, Error> {
102        let node = self.blob_directory.open_blob(&hash.into()).await?.ok_or(FxfsError::NotFound)?;
103
104        let extents = node.get_mapping_extents().await?;
105        let size = node.as_ref().byte_size();
106        let stored_size = node.as_ref().stored_size().await?;
107        let data_extents = Extents::try_new(&extents.data, 0)?;
108        let merkle_extents = Extents::try_new(&extents.merkle, 0)?;
109        let extent_count = data_extents.len() as u32;
110        let metadata_count = merkle_extents.len() as u32;
111
112        let allocation_size = (extent_count + metadata_count) as usize * std::mem::size_of::<u64>();
113
114        if allocation_size > 0 {
115            let mut payload = self.sender.reserve_payload(allocation_size).await?;
116            let offset_in_vmo = payload.offset();
117
118            for (mut chunk, val_res) in payload.data().chunks_mut(std::mem::size_of::<u64>()).zip(
119                Extents::encode_extents(&data_extents)
120                    .chain(Extents::encode_extents(&merkle_extents)),
121            ) {
122                chunk.copy_from_slice(&val_res.to_le_bytes());
123            }
124
125            let command = MappingCommand::Mappings {
126                key,
127                offset: offset_in_vmo as u32,
128                stored_size,
129                device_offset: 0,
130                metadata_count,
131                extent_count,
132                encrypted: false,
133            };
134
135            payload.commit(command.into()).await?;
136        }
137
138        self.opened_blobs.insert(key, node);
139
140        Ok(size)
141    }
142
143    /// Unregisters the blob mapping and signals the block driver to terminate tracking.
144    async fn close_blob(&mut self, key: u64) -> Result<(), Error> {
145        if self.opened_blobs.remove(&key).is_none() {
146            return Ok(());
147        }
148        self.sender.push(MappingCommand::CloseBlob { key }.into()).await?;
149        Ok(())
150    }
151
152    pub async fn handle_mapping_session_requests(
153        mut self,
154        mut stream: fmapping::MappingSessionRequestStream,
155    ) {
156        while let Ok(Some(request)) = stream.try_next().await {
157            match request {
158                fmapping::MappingSessionRequest::Open { key, identifier, responder } => {
159                    // We expect the identifier to be a Merkle Root Hash for the blob.
160                    if identifier.len() != fuchsia_hash::HASH_SIZE {
161                        responder.send(Err(zx::Status::INVALID_ARGS.into_raw())).unwrap_or_else(
162                            |error| warn!(error:?; "Failed to send mapping session response"),
163                        );
164                        continue;
165                    }
166                    let hash = Hash::from(<[u8; 32]>::try_from(identifier.as_slice()).unwrap());
167                    match self.open_blob(key, hash).await {
168                        Ok(size) => {
169                            responder.send(Ok(size)).unwrap_or_else(
170                                |error| warn!(error:?; "Failed to send mapping session response"),
171                            );
172                        }
173                        Err(error) => {
174                            error!(error:?; "Failed to open blob");
175                            responder.send(Err(map_to_status(error).into_raw())).unwrap_or_else(
176                                |error| warn!(error:?; "Failed to send mapping session response"),
177                            );
178                        }
179                    }
180                }
181                fmapping::MappingSessionRequest::Close { key, responder } => {
182                    let result = match self.close_blob(key).await {
183                        Ok(()) => Ok(()),
184                        Err(error) => {
185                            error!(error:?; "Failed to close blob");
186                            Err(map_to_status(error).into_raw())
187                        }
188                    };
189                    responder.send(result).unwrap_or_else(
190                        |error| warn!(error:?; "Failed to send mapping session response"),
191                    );
192                }
193                fmapping::MappingSessionRequest::GetProfile { name, responder } => {
194                    let result = match self.blob_directory.volume().get_profile_vmo(&name).await {
195                        Ok(vmo) => Ok(vmo),
196                        Err(error) => Err(map_to_status(error).into_raw()),
197                    };
198                    responder.send(result).unwrap_or_else(
199                        |error| warn!(error:?; "Failed to send mapping session response"),
200                    );
201                }
202                fmapping::MappingSessionRequest::WriteProfile { name, profile_vmo, responder } => {
203                    let result = match self
204                        .blob_directory
205                        .volume()
206                        .write_profile_vmo(&name, profile_vmo)
207                        .await
208                    {
209                        Ok(()) => Ok(()),
210                        Err(error) => {
211                            error!(error:?, name = name.as_str(); "Failed to write profile VMO");
212                            Err(map_to_status(error).into_raw())
213                        }
214                    };
215                    responder.send(result).unwrap_or_else(
216                        |error| warn!(error:?; "Failed to send mapping session response"),
217                    );
218                }
219                fmapping::MappingSessionRequest::_UnknownMethod { ordinal, .. } => {
220                    warn!(ordinal; "Unknown MappingSession method");
221                }
222            }
223        }
224    }
225}
226
227#[cfg(test)]
228mod tests {
229    use super::*;
230    use crate::fuchsia::fxblob::testing::{BlobFixture, new_blob_fixture, open_blob_fixture};
231    use crate::fuchsia::node::FxNode;
232    use crate::fuchsia::testing::TestFixture;
233    use blob_writer::BlobWriter;
234    use delivery_blob::{CompressionMode, Type1Blob};
235    use fidl_fuchsia_io::UnlinkOptions;
236    use fuchsia_async as fasync;
237    use futures::channel::oneshot;
238    use fxfs::lsm_tree::types::Item;
239    use fxfs::object_handle::ObjectHandle;
240    use fxfs::object_store::{
241        AttributeId, ExtentValue, HandleOptions, ObjectKey, ObjectStore, ObjectValue,
242    };
243    use storage_device::Device;
244    use storage_device::buffer::OwnedBuffer;
245    use storage_device::buffer_allocator::{BufferAllocator, BufferSource};
246    use vmo_fifo::Receiver;
247
248    #[fuchsia::test]
249    async fn test_blob_mapping_provider() {
250        let fixture = new_blob_fixture().await;
251        // Test with a large amount of non-compressible data to generate many extents
252        let data = vec![42; 300_000];
253        let hash = fixture.write_blob(&data, CompressionMode::Never).await;
254
255        let blob_dir = fixture
256            .volume()
257            .root()
258            .clone()
259            .as_node()
260            .into_any()
261            .downcast::<BlobDirectory>()
262            .expect("Failed to downcast root directory to BlobDirectory");
263
264        let node = blob_dir
265            .open_blob(&hash.into())
266            .await
267            .expect("Failed to open blob in Fxfs")
268            .expect("open_blob returned None instead of node");
269        let extents = node.get_mapping_extents().await.expect("Failed to retrieve extents");
270        let data_extents = extents.data;
271        let merkle_extents = extents.merkle;
272
273        // Un-dropped nodes pin the Blob as actively opened. The unmount routine in fixture.close()
274        // will wait forever for this blob to be fully closed, causing a test timeout.
275        drop(node);
276
277        let vmo = zx::Vmo::create(MAPPING_VMO_SIZE).unwrap();
278        let client_mapping = vmo.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap();
279        let sender =
280            AsyncSender::<RawMappingCommand>::new(vmo, 8, PENDING_COMMANDS_CAPACITY).unwrap();
281
282        let mut session = BlobMappingSession::new(blob_dir, sender);
283
284        let receiver_task = fasync::unblock(move || {
285            let mut receiver = Receiver::<RawMappingCommand>::new(client_mapping, 256)
286                .expect("Failed to create the Receiver wrapper");
287
288            // First Open Command
289            let cmd1_raw = receiver.peek().expect("peek failed");
290            let cmd1 = MappingCommand::try_from(*cmd1_raw).expect("try_from failed");
291
292            let (cmd1_offset, cmd1_extent_count, cmd1_metadata_count) = match cmd1 {
293                MappingCommand::Mappings {
294                    key,
295                    offset,
296                    stored_size: _,
297                    device_offset: _,
298                    metadata_count,
299                    extent_count,
300                    encrypted,
301                } => {
302                    assert_eq!(key, 1);
303                    assert_eq!(extent_count, data_extents.len() as u32);
304                    assert_eq!(metadata_count, merkle_extents.len() as u32);
305                    assert!(!encrypted);
306                    (offset, extent_count, metadata_count)
307                }
308                _ => panic!("Expected Mappings command"),
309            };
310
311            // Verify payload
312            let total_extents = cmd1_extent_count + cmd1_metadata_count;
313            let buffer = cmd1_raw.payload_slice(cmd1_offset, total_extents * 8).to_vec();
314
315            let data_extents_container = Extents::try_new(&data_extents, 0).unwrap();
316            let merkle_extents_container = Extents::try_new(&merkle_extents, 0).unwrap();
317            let mut expected_payload = Vec::new();
318            for val in Extents::encode_extents(&data_extents_container)
319                .chain(Extents::encode_extents(&merkle_extents_container))
320            {
321                expected_payload.extend_from_slice(&val.to_le_bytes());
322            }
323            assert_eq!(buffer, expected_payload);
324            cmd1_raw.pop().expect("Failed pop_commit");
325
326            let cmd2_raw = receiver.peek().expect("peek failed");
327            let cmd2 = MappingCommand::try_from(*cmd2_raw).expect("try_from failed");
328            match cmd2 {
329                MappingCommand::CloseBlob { key } => assert_eq!(key, 1),
330                _ => panic!("Expected CloseBlob command"),
331            };
332            cmd2_raw.pop().expect("Failed pop_commit");
333        });
334
335        let server_task = async move {
336            let key = 1;
337            let size = session.open_blob(key, hash).await.expect("open_blob failed");
338            assert_eq!(size, data.len() as u64);
339
340            session.close_blob(key).await.expect("close_blob failed on existing key");
341
342            std::mem::drop(session);
343        };
344
345        futures::join!(receiver_task, server_task);
346
347        fixture.close().await;
348    }
349
350    #[fuchsia::test]
351    async fn test_missing_blob() {
352        let fixture = new_blob_fixture().await;
353        let blob_dir = fixture
354            .volume()
355            .root()
356            .clone()
357            .as_node()
358            .into_any()
359            .downcast::<BlobDirectory>()
360            .expect("Failed to downcast");
361
362        let vmo = zx::Vmo::create(MAPPING_VMO_SIZE).unwrap();
363        let sender =
364            AsyncSender::<RawMappingCommand>::new(vmo, 8, PENDING_COMMANDS_CAPACITY).unwrap();
365        let mut session = BlobMappingSession::new(blob_dir, sender);
366        let hash = Hash::from([1u8; 32]);
367        session
368            .open_blob(1, hash)
369            .await
370            .expect_err("open_blob should fail with blob that doesn't exist");
371
372        std::mem::drop(session);
373        fixture.close().await;
374    }
375
376    #[fuchsia::test]
377    async fn test_invalid_key() {
378        let fixture = new_blob_fixture().await;
379        let blob_dir = fixture
380            .volume()
381            .root()
382            .clone()
383            .as_node()
384            .into_any()
385            .downcast::<BlobDirectory>()
386            .expect("Failed to downcast");
387
388        let vmo = zx::Vmo::create(MAPPING_VMO_SIZE).unwrap();
389        let sender =
390            AsyncSender::<RawMappingCommand>::new(vmo, 8, PENDING_COMMANDS_CAPACITY).unwrap();
391        let mut session = BlobMappingSession::new(blob_dir, sender);
392        session.close_blob(42).await.expect("close_blob should return Ok with invalid key");
393        std::mem::drop(session);
394
395        fixture.close().await;
396    }
397
398    // Verifies that unlinking an open blob does not deallocate its extents until all references
399    // are closed.
400    #[fuchsia::test]
401    async fn test_unlink_open_blob() {
402        let fixture = new_blob_fixture().await;
403        let data = vec![77u8; 8192];
404        let hash = fixture.write_blob(&data, CompressionMode::Never).await;
405        let hash_string = format!("{}", hash);
406
407        let object_id = fixture.get_blob_handle(&hash_string).await.object_id();
408
409        let blob_dir = fixture
410            .volume()
411            .root()
412            .clone()
413            .as_node()
414            .into_any()
415            .downcast::<BlobDirectory>()
416            .expect("Failed to downcast root directory to BlobDirectory");
417
418        let vmo = zx::Vmo::create(MAPPING_VMO_SIZE).unwrap();
419        let sender =
420            AsyncSender::<RawMappingCommand>::new(vmo, 8, PENDING_COMMANDS_CAPACITY).unwrap();
421        let mut session = BlobMappingSession::new(blob_dir, sender);
422
423        let key = 1;
424        session.open_blob(key, hash).await.expect("open_blob failed");
425
426        // Unlink the blob from the root directory.
427        fixture
428            .root()
429            .unlink(&hash_string, &UnlinkOptions::default())
430            .await
431            .expect("FIDL failed")
432            .expect("unlink failed");
433
434        // Flush the graveyard to trigger purging if the object has no open references.
435        fixture.fs().graveyard().flush().await;
436
437        // Since the blob is currently open in the mapping session, it must not be purged.
438        assert!(
439            ObjectStore::open_object(
440                fixture.volume().volume(),
441                object_id,
442                HandleOptions::default(),
443                None
444            )
445            .await
446            .is_ok(),
447            "Blob was purged from store while still open in BlobMappingSession"
448        );
449
450        // Close the blob in the mapping session.
451        session.close_blob(key).await.expect("close_blob failed");
452
453        // Flush the graveyard again now that the session has released the blob.
454        fixture.fs().graveyard().flush().await;
455
456        // Now the blob must be tombstoned and purged from the object store.
457        assert!(
458            ObjectStore::open_object(
459                fixture.volume().volume(),
460                object_id,
461                HandleOptions::default(),
462                None
463            )
464            .await
465            .is_err(),
466            "Blob should have been purged from store after close_blob"
467        );
468
469        drop(session);
470        fixture.close().await;
471    }
472
473    #[fuchsia::test]
474    async fn test_mapping_provider_and_mapping_session() {
475        let fixture = new_blob_fixture().await;
476        let data = vec![42; 8192];
477        let hash = fixture.write_blob(&data, CompressionMode::Never).await;
478
479        let blob_dir = fixture
480            .volume()
481            .root()
482            .clone()
483            .as_node()
484            .into_any()
485            .downcast::<BlobDirectory>()
486            .expect("Failed to downcast root directory to BlobDirectory");
487
488        let scope = blob_dir.volume().scope().clone();
489        let server = Arc::new(
490            BlobMappingProvider::new(blob_dir).expect("Failed to create BlobMappingProvider"),
491        );
492
493        // Spawn the mapping provider stream.
494        let (provider_proxy, provider_server_end) =
495            fidl::endpoints::create_proxy::<fmapping::MappingProviderMarker>();
496        scope.spawn(async move {
497            server.handle_mapping_provider_requests(provider_server_end.into_stream()).await;
498        });
499
500        // Open a mapping session from the mapping provider
501        let (session_proxy, session_server_end) =
502            fidl::endpoints::create_proxy::<fmapping::MappingSessionMarker>();
503        let _shared_vmo = provider_proxy
504            .open_session(session_server_end)
505            .await
506            .expect("open_session failed")
507            .expect("vmo returned an error");
508
509        let not_found_profile_err = session_proxy
510            .get_profile(".boot")
511            .await
512            .expect("get_profile wire call failed")
513            .expect_err("expected get_profile(.boot) to return NOT_FOUND");
514        assert_eq!(not_found_profile_err, zx::Status::NOT_FOUND.into_raw());
515
516        let profile_data = vec![0xABu8; 4096];
517        let write_vmo = zx::Vmo::create(profile_data.len() as u64).expect("create vmo failed");
518        write_vmo.write(&profile_data, 0).expect("vmo write failed");
519        session_proxy
520            .write_profile(".boot", write_vmo)
521            .await
522            .expect("write_profile wire call failed")
523            .expect("write_profile returned an error");
524
525        let read_vmo = session_proxy
526            .get_profile(".boot")
527            .await
528            .expect("get_profile wire call failed")
529            .expect("get_profile returned an error");
530        assert_eq!(
531            read_vmo.read_to_vec::<u8>(0, profile_data.len() as u64).expect("vmo read failed"),
532            profile_data
533        );
534
535        let id: [u8; 32] = hash.into();
536        let key = 1;
537        let size = session_proxy
538            .open(key, &id)
539            .await
540            .expect("open failed")
541            .expect("open explicitly returned an error");
542
543        assert_eq!(size, 8192);
544
545        session_proxy
546            .close(key)
547            .await
548            .expect("close failed")
549            .expect("close explicitly returned an error");
550
551        // Test some failures
552
553        // Sending an invalid length hash to open should return INVALID_ARGS
554        let bad_length_id = vec![1u8, 2, 3];
555        let invalid_args_err = session_proxy
556            .open(key, &bad_length_id)
557            .await
558            .expect("open wire call failed")
559            .unwrap_err();
560        assert_eq!(invalid_args_err, zx::Status::INVALID_ARGS.into_raw());
561
562        // Sending a valid length hash that does not exist should return NOT_FOUND
563        let not_found_id = [0u8; 32];
564        let not_found_err = session_proxy
565            .open(key, &not_found_id)
566            .await
567            .expect("open wire call failed")
568            .unwrap_err();
569        assert_eq!(not_found_err, zx::Status::NOT_FOUND.into_raw());
570
571        // Trying to close an invalid blob key currently always succeeds. The BlobMappingServer
572        // unconditionally passes the command down the FIFO queue to the block driver and doesn't
573        // explicitly track active connections.
574        session_proxy.close(123).await.expect("close wire call failed").expect("close failed");
575
576        fixture.close().await;
577    }
578
579    struct DeviceBlockService {
580        device: Arc<dyn Device>,
581    }
582
583    impl mapping::reader::BlockService for DeviceBlockService {
584        fn allocate_buffer(&self, max_len: usize) -> OwnedBuffer {
585            let block_size = self.device.block_size() as usize;
586            let nblocks = (std::cmp::max(max_len, block_size) + block_size - 1) / block_size;
587            let aligned_len = nblocks * block_size;
588            let pool_size = nblocks.next_power_of_two() * block_size;
589            let buffer_source = BufferSource::new(pool_size);
590            let allocator = BufferAllocator::new(block_size, buffer_source);
591            Arc::new(allocator).allocate_buffer_sync_owned(aligned_len)
592        }
593
594        fn read_blocks(
595            &self,
596            device_offset: u64,
597            mut dest_buffer: OwnedBuffer,
598            on_complete: Box<dyn FnOnce(Result<OwnedBuffer, anyhow::Error>) + Send>,
599        ) -> Result<(), anyhow::Error> {
600            let device = self.device.clone();
601            futures::executor::block_on(async move {
602                let res = device.read(device_offset, dest_buffer.as_mut()).await;
603                on_complete(res.map(|_| dest_buffer));
604            });
605            Ok(())
606        }
607    }
608
609    async fn run_blob_mapping_test(
610        fixture: &TestFixture,
611        test_data: &[u8],
612        mode: CompressionMode,
613        min_data_extents: usize,
614        min_merkle_extents: usize,
615    ) {
616        let hash = fixture.write_blob(test_data, mode).await;
617        run_blob_mapping_test_with_hash(
618            fixture,
619            hash,
620            test_data,
621            min_data_extents,
622            min_merkle_extents,
623        )
624        .await;
625    }
626
627    async fn write_blob_chunked(fx: &TestFixture, data: &[u8], mode: CompressionMode) -> Hash {
628        let hash = fuchsia_merkle::root_from_slice(data);
629        let compressed_data = Type1Blob::generate(data, mode);
630        let writer = fx.create_blob(&hash.into(), false).await.expect("create blob failed");
631        let vmo = writer
632            .get_vmo(compressed_data.len() as u64)
633            .await
634            .expect("transport error on get_vmo")
635            .expect("failed to get vmo");
636        let vmo_size = vmo.get_size().expect("failed to get vmo size");
637
638        // Write in small chunks (512 B) to allocate extents across transactions.
639        let chunk_size = 512;
640        let mut write_offset = 0u64;
641        let mut bytes_left = compressed_data.len() as u64;
642        while bytes_left > 0 {
643            let chunk_len = std::cmp::min(bytes_left, chunk_size);
644            vmo.write(
645                &compressed_data[write_offset as usize..(write_offset + chunk_len) as usize],
646                write_offset % vmo_size,
647            )
648            .expect("failed to write to vmo");
649            let _ = writer
650                .bytes_ready(chunk_len)
651                .await
652                .expect("transport error on bytes_ready")
653                .expect("failed to write data to vmo");
654            write_offset += chunk_len;
655            bytes_left -= chunk_len;
656        }
657        hash
658    }
659
660    async fn run_blob_mapping_test_with_hash(
661        fixture: &TestFixture,
662        hash: Hash,
663        uncompressed_data: &[u8],
664        min_data_extents: usize,
665        min_merkle_extents: usize,
666    ) {
667        let blob_dir = fixture
668            .volume()
669            .root()
670            .clone()
671            .as_node()
672            .into_any()
673            .downcast::<BlobDirectory>()
674            .expect("Failed to downcast root directory to BlobDirectory");
675
676        let scope = blob_dir.volume().scope().clone();
677        let server = Arc::new(
678            BlobMappingProvider::new(blob_dir).expect("Failed to create BlobMappingProvider"),
679        );
680
681        let (provider_proxy, provider_server_end) =
682            fidl::endpoints::create_proxy::<fmapping::MappingProviderMarker>();
683        scope.spawn(async move {
684            server.handle_mapping_provider_requests(provider_server_end.into_stream()).await;
685        });
686
687        let (session_proxy, session_server_end) =
688            fidl::endpoints::create_proxy::<fmapping::MappingSessionMarker>();
689        let mapping_vmo = provider_proxy
690            .open_session(session_server_end)
691            .await
692            .expect("open_session failed")
693            .expect("vmo returned an error");
694
695        let mut receiver =
696            vmo_fifo::Receiver::<RawMappingCommand>::new(mapping_vmo, PENDING_COMMANDS_CAPACITY)
697                .expect("Failed to create receiver");
698
699        let device = fixture.fs().device().clone();
700        let service: Arc<dyn mapping::reader::BlockService> =
701            Arc::new(DeviceBlockService { device });
702
703        let port = zx::Port::create();
704        let delivery_queue = zx::Vmo::create(mapping::DELIVERY_VMO_SIZE)
705            .expect("Failed to create delivery_queue VMO");
706        let pager = Arc::new(zx::Pager::create(zx::PagerOptions::empty()).unwrap());
707        let vmo_provider = Arc::new(blob_pager_and_verifier::TestVmoProvider::new(
708            pager.clone(),
709            delivery_queue.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap(),
710        ));
711        let delivery_receiver = vmo_fifo::Receiver::<mapping::RawDeliveryCommand>::new(
712            delivery_queue.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap(),
713            mapping::PENDING_DELIVERY_COMMANDS_CAPACITY,
714        )
715        .unwrap();
716        let _delivery_processor = blob_pager_and_verifier::DeliveryQueueProcessor::spawn(
717            delivery_receiver,
718            vmo_provider.clone(),
719            delivery_queue.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap(),
720        )
721        .unwrap();
722
723        let verifier = block_server::verifier::Verifier::new(delivery_queue);
724        let files = Arc::new(mapping::Files::new(
725            service,
726            verifier,
727            port.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap(),
728        ));
729
730        let id: [u8; 32] = hash.into();
731        let key = 1;
732        let blob_size = session_proxy
733            .open(key, &id)
734            .await
735            .expect("open failed")
736            .expect("open explicitly returned an error");
737
738        assert_eq!(blob_size as usize, uncompressed_data.len());
739
740        let msg = receiver.peek().expect("Failed to peek message");
741        if let Ok(MappingCommand::Mappings { extent_count, metadata_count, .. }) =
742            MappingCommand::try_from(*msg)
743        {
744            if (extent_count as usize) < min_data_extents
745                || (metadata_count as usize) < min_merkle_extents
746            {
747                panic!(
748                    "EXTENTS MISMATCH: extent_count = {}, metadata_count = {}, \
749                     required min_data = {}, min_merkle = {}",
750                    extent_count, metadata_count, min_data_extents, min_merkle_extents
751                );
752            }
753        }
754        mapping::process_mapping_command(&msg, &files).expect("process_mapping_command failed");
755        msg.pop().expect("pop failed");
756
757        let paged_vmo = pager.create_vmo(zx::VmoOptions::empty(), &port, key, blob_size).unwrap();
758        vmo_provider
759            .register_vmo(key, paged_vmo.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap());
760
761        let _pager_thread = files.spawn_pager_thread();
762
763        let (tx, rx) = oneshot::channel();
764        let len = uncompressed_data.len();
765        std::thread::spawn(move || {
766            let mut buf = vec![0u8; len];
767            paged_vmo.read(&mut buf, 0).expect("paged vmo read failed");
768            let _ = tx.send(buf);
769        });
770
771        let read_bytes = rx.await.unwrap();
772        assert_eq!(&read_bytes[..], &uncompressed_data[..]);
773
774        session_proxy.close(key).await.expect("close failed").expect("close error");
775        let msg = receiver.peek().expect("Failed to peek close message");
776        mapping::process_mapping_command(&msg, &files).expect("process_mapping_command failed");
777        msg.pop().expect("pop failed");
778    }
779
780    async fn fragment_free_space(
781        fixture: TestFixture,
782        block_size: usize,
783        anchor_size: Option<usize>,
784    ) -> TestFixture {
785        let mut data = vec![0u8; block_size];
786        let mut hashes = Vec::new();
787        let mut i = 0usize;
788        loop {
789            rand::fill(&mut data[..]);
790            data[..8].copy_from_slice(&(i as u64).to_le_bytes());
791            let hash = fuchsia_merkle::root_from_slice(&data);
792            let delivery_data = Type1Blob::generate(&data, CompressionMode::Never);
793            let writer = match fixture.create_blob(&hash.into(), false).await {
794                Ok(w) => w,
795                Err(_) => break,
796            };
797            if let Ok(mut blob_writer) =
798                BlobWriter::create(writer, delivery_data.len() as u64).await
799            {
800                if blob_writer.write(&delivery_data).await.is_err() {
801                    break;
802                }
803            } else {
804                break;
805            }
806            hashes.push(hash);
807            i += 1;
808        }
809
810        // Unlink every second small blob to create scattered free space holes.
811        let root = fixture.root();
812        for ix in (0..hashes.len()).step_by(2) {
813            let _ = root.unlink(&format!("{}", hashes[ix]), &UnlinkOptions::default()).await;
814        }
815
816        // Remount fixture so unlinked blobs are purged and blocks are freed to allocator.
817        let device = fixture.close().await;
818        let fixture = open_blob_fixture(device).await;
819
820        if let Some(size) = anchor_size {
821            let anchor_data = vec![0u8; size];
822            let _ = fixture.write_blob(&anchor_data, CompressionMode::Never).await;
823        }
824
825        fixture
826    }
827
828    #[fuchsia::test]
829    async fn test_fxfs_blob_mapping_uncompressed() {
830        let fixture = new_blob_fixture().await;
831        let test_data = vec![123u8; 8192];
832        run_blob_mapping_test(&fixture, &test_data, CompressionMode::Never, 1, 0).await;
833        fixture.close().await;
834    }
835
836    #[fuchsia::test]
837    async fn test_fxfs_blob_mapping_compressed() {
838        let fixture = new_blob_fixture().await;
839        // Generate pseudo-random compressible test data.
840        let mut test_data = vec![0u8; 16384];
841        for i in 0..test_data.len() {
842            test_data[i] = ((i / 64) % 256) as u8;
843        }
844        run_blob_mapping_test(&fixture, &test_data, CompressionMode::Always, 1, 0).await;
845        fixture.close().await;
846    }
847
848    #[fuchsia::test]
849    async fn test_fxfs_blob_mapping_fragmented_data() {
850        let fixture = new_blob_fixture().await;
851        let fixture = fragment_free_space(fixture, 32768, Some(1_000_000)).await;
852
853        let mut uncompressed_data = vec![0u8; 262_144];
854        rand::fill(&mut uncompressed_data[..]);
855        let hash1 = write_blob_chunked(&fixture, &uncompressed_data, CompressionMode::Never).await;
856        run_blob_mapping_test_with_hash(&fixture, hash1, &uncompressed_data, 2, 0).await;
857        fixture
858            .root()
859            .unlink(&format!("{}", hash1), &UnlinkOptions::default())
860            .await
861            .unwrap()
862            .unwrap();
863
864        let mut compressed_data = vec![0u8; 262_144];
865        rand::fill(&mut compressed_data[..]);
866        let hash2 = write_blob_chunked(&fixture, &compressed_data, CompressionMode::Always).await;
867        run_blob_mapping_test_with_hash(&fixture, hash2, &compressed_data, 2, 0).await;
868
869        fixture.close().await;
870    }
871
872    #[fuchsia::test]
873    async fn test_blob_mapping_provider_large_extent() {
874        let fixture = new_blob_fixture().await;
875        let data = vec![42; 8192];
876        let hash = fixture.write_blob(&data, CompressionMode::Never).await;
877
878        let blob_dir = fixture
879            .volume()
880            .root()
881            .clone()
882            .as_node()
883            .into_any()
884            .downcast::<BlobDirectory>()
885            .expect("Failed to downcast root directory to BlobDirectory");
886
887        let node = blob_dir
888            .open_blob(&hash.into())
889            .await
890            .expect("Failed to open blob in Fxfs")
891            .expect("open_blob returned None instead of node");
892        let object_id = node.object_id();
893        let original_extents =
894            node.get_mapping_extents().await.expect("Failed to retrieve extents");
895        assert_eq!(original_extents.data.len(), 1);
896        let start_dev = original_extents.data[0].device_offset().unwrap();
897        drop(node);
898
899        // Replace the blob's single 8 KiB data extent in the in-memory LSM tree with a single
900        // extent larger than MAX_REGULAR_EXTENT_BLOCKS (0x3fff_ffff blocks) so Extents::try_new
901        // splits it into 2 encoded entries.
902        let large_len = (0x3fff_ffffu64 + 5) * mapping::BLOCK_SIZE;
903        let key = ObjectKey::extent(object_id, AttributeId::DATA, 0..large_len);
904        let lower_bound = key.key_for_merge_into();
905        fixture.volume().volume().store().tree().merge_into(
906            Item::new(key, ObjectValue::Extent(ExtentValue::new_raw(start_dev, 0))),
907            &lower_bound,
908        );
909
910        let vmo = zx::Vmo::create(MAPPING_VMO_SIZE).unwrap();
911        let client_mapping = vmo.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap();
912        let sender =
913            AsyncSender::<RawMappingCommand>::new(vmo, 8, PENDING_COMMANDS_CAPACITY).unwrap();
914        let mut session = BlobMappingSession::new(blob_dir, sender);
915
916        let merkle_extents = original_extents.merkle;
917        let receiver_task = fasync::unblock(move || {
918            let mut receiver = Receiver::<RawMappingCommand>::new(client_mapping, 256)
919                .expect("Failed to create the Receiver wrapper");
920
921            let cmd1_raw = receiver.peek().expect("peek failed");
922            let cmd1 = MappingCommand::try_from(*cmd1_raw).expect("try_from failed");
923
924            let (cmd1_offset, cmd1_extent_count, cmd1_metadata_count) = match cmd1 {
925                MappingCommand::Mappings {
926                    key,
927                    offset,
928                    stored_size: _,
929                    device_offset: _,
930                    metadata_count,
931                    extent_count,
932                    encrypted,
933                } => {
934                    assert_eq!(key, 1);
935                    assert_eq!(extent_count, 2);
936                    assert_eq!(metadata_count, merkle_extents.len() as u32);
937                    assert!(!encrypted);
938                    (offset, extent_count, metadata_count)
939                }
940                _ => panic!("Expected Mappings command"),
941            };
942
943            let total_extents = cmd1_extent_count + cmd1_metadata_count;
944            let buffer = cmd1_raw.payload_slice(cmd1_offset, total_extents * 8).to_vec();
945
946            let expected_data_extents =
947                Extents::try_new([mapping::Extent::new(0..large_len, Some(start_dev))], 0).unwrap();
948            let expected_merkle_extents = Extents::try_new(&merkle_extents, 0).unwrap();
949            let mut expected_payload = Vec::new();
950            for val in Extents::encode_extents(&expected_data_extents)
951                .chain(Extents::encode_extents(&expected_merkle_extents))
952            {
953                expected_payload.extend_from_slice(&val.to_le_bytes());
954            }
955            assert_eq!(buffer, expected_payload);
956            cmd1_raw.pop().expect("Failed pop_commit");
957
958            let cmd2_raw = receiver.peek().expect("peek failed");
959            let cmd2 = MappingCommand::try_from(*cmd2_raw).expect("try_from failed");
960            match cmd2 {
961                MappingCommand::CloseBlob { key } => assert_eq!(key, 1),
962                _ => panic!("Expected CloseBlob command"),
963            };
964            cmd2_raw.pop().expect("Failed pop_commit");
965        });
966
967        let server_task = async move {
968            let key = 1;
969            let size = session.open_blob(key, hash).await.expect("open_blob failed");
970            assert_eq!(size, data.len() as u64);
971            session.close_blob(key).await.expect("close_blob failed");
972            std::mem::drop(session);
973        };
974
975        futures::join!(receiver_task, server_task);
976
977        fixture.close().await;
978    }
979}
fxfs_platform_testing/fuchsia/fxblob/mapping_provider.rs

fxfs_platform_testing/fuchsia/fxblob/
mapping_provider.rs
