1#![deny(missing_docs)]
6#![allow(clippy::let_unit_value)]
7
8use anyhow::{Context as _, Error, anyhow};
11use delivery_blob::{CompressionMode, Type1Blob};
12use fidl::endpoints::ClientEnd;
13use fidl_fuchsia_fs_startup::{CreateOptions, MountOptions};
14use fuchsia_merkle::Hash;
15use std::borrow::Cow;
16use std::collections::BTreeSet;
17use {fidl_fuchsia_fxfs as ffxfs, fidl_fuchsia_io as fio};
18
19const RAMDISK_BLOCK_SIZE: u64 = 512;
20static FXFS_BLOB_VOLUME_NAME: &str = "blob";
21
22#[cfg(test)]
23mod test;
24
25#[derive(Debug, Clone)]
27pub struct BlobInfo {
28 merkle: Hash,
29 contents: Cow<'static, [u8]>,
30}
31
32impl<B> From<B> for BlobInfo
33where
34 B: Into<Cow<'static, [u8]>>,
35{
36 fn from(bytes: B) -> Self {
37 let bytes = bytes.into();
38 Self { merkle: fuchsia_merkle::root_from_slice(&bytes), contents: bytes }
39 }
40}
41
42pub struct BlobfsRamdiskBuilder {
44 ramdisk: Option<SuppliedRamdisk>,
45 blobs: Vec<BlobInfo>,
46 implementation: Implementation,
47}
48
49enum SuppliedRamdisk {
50 Formatted(FormattedRamdisk),
51 Unformatted(Ramdisk),
52}
53
54#[derive(Debug, Clone, Copy, PartialEq, Eq)]
55pub enum Implementation {
57 CppBlobfs,
59 Fxblob,
61}
62
63impl Implementation {
64 pub fn from_env() -> Self {
70 match env!("FXFS_BLOB") {
71 "true" => Self::Fxblob,
72 "false" => Self::CppBlobfs,
73 other => panic!("unexpected value for env var 'FXFS_BLOB': {other}"),
74 }
75 }
76}
77
78impl BlobfsRamdiskBuilder {
79 fn new() -> Self {
80 Self { ramdisk: None, blobs: vec![], implementation: Implementation::CppBlobfs }
81 }
82
83 pub fn formatted_ramdisk(self, ramdisk: FormattedRamdisk) -> Self {
85 Self { ramdisk: Some(SuppliedRamdisk::Formatted(ramdisk)), ..self }
86 }
87
88 pub fn ramdisk(self, ramdisk: Ramdisk) -> Self {
90 Self { ramdisk: Some(SuppliedRamdisk::Unformatted(ramdisk)), ..self }
91 }
92
93 pub fn with_blob(mut self, blob: impl Into<BlobInfo>) -> Self {
95 self.blobs.push(blob.into());
96 self
97 }
98
99 pub fn cpp_blobfs(self) -> Self {
102 Self { implementation: Implementation::CppBlobfs, ..self }
103 }
104
105 pub fn fxblob(self) -> Self {
108 Self { implementation: Implementation::Fxblob, ..self }
109 }
110
111 pub fn implementation(self, implementation: Implementation) -> Self {
113 Self { implementation, ..self }
114 }
115
116 pub fn impl_from_env(self) -> Self {
118 self.implementation(Implementation::from_env())
119 }
120
121 pub async fn start(self) -> Result<BlobfsRamdisk, Error> {
123 let Self { ramdisk, blobs, implementation } = self;
124 let (ramdisk, needs_format) = match ramdisk {
125 Some(SuppliedRamdisk::Formatted(FormattedRamdisk(ramdisk))) => (ramdisk, false),
126 Some(SuppliedRamdisk::Unformatted(ramdisk)) => (ramdisk, true),
127 None => (Ramdisk::start().await.context("creating backing ramdisk for blobfs")?, true),
128 };
129
130 let ramdisk_controller = ramdisk.client.open_controller()?;
131
132 let mut fs = match implementation {
134 Implementation::CppBlobfs => fs_management::filesystem::Filesystem::new(
135 ramdisk_controller,
136 fs_management::Blobfs { ..fs_management::Blobfs::dynamic_child() },
137 ),
138 Implementation::Fxblob => fs_management::filesystem::Filesystem::new(
139 ramdisk_controller,
140 fs_management::Fxfs::default(),
141 ),
142 };
143 if needs_format {
144 let () = fs.format().await.context("formatting ramdisk")?;
145 }
146
147 let fs = match implementation {
148 Implementation::CppBlobfs => ServingFilesystem::SingleVolume(
149 fs.serve().await.context("serving single volume filesystem")?,
150 ),
151 Implementation::Fxblob => {
152 let fs =
153 fs.serve_multi_volume().await.context("serving multi volume filesystem")?;
154 let volume = if needs_format {
155 fs.create_volume(
156 FXFS_BLOB_VOLUME_NAME,
157 CreateOptions::default(),
158 MountOptions { as_blob: Some(true), ..MountOptions::default() },
159 )
160 .await
161 .context("creating blob volume")?
162 } else {
163 fs.open_volume(
164 FXFS_BLOB_VOLUME_NAME,
165 MountOptions { as_blob: Some(true), ..MountOptions::default() },
166 )
167 .await
168 .context("opening blob volume")?
169 };
170 ServingFilesystem::MultiVolume(fs, volume)
171 }
172 };
173
174 let blobfs = BlobfsRamdisk { backing_ramdisk: FormattedRamdisk(ramdisk), fs };
175
176 if !blobs.is_empty() {
178 let mut present_blobs = blobfs.list_blobs()?;
179
180 for blob in blobs {
181 if present_blobs.contains(&blob.merkle) {
182 continue;
183 }
184 blobfs
185 .write_blob(blob.merkle, &blob.contents)
186 .await
187 .context(format!("writing {}", blob.merkle))?;
188 present_blobs.insert(blob.merkle);
189 }
190 }
191
192 Ok(blobfs)
193 }
194}
195
196pub struct BlobfsRamdisk {
198 backing_ramdisk: FormattedRamdisk,
199 fs: ServingFilesystem,
200}
201
202enum ServingFilesystem {
207 SingleVolume(fs_management::filesystem::ServingSingleVolumeFilesystem),
208 MultiVolume(
209 fs_management::filesystem::ServingMultiVolumeFilesystem,
210 fs_management::filesystem::ServingVolume,
211 ),
212}
213
214impl ServingFilesystem {
215 async fn shutdown(self) -> Result<(), Error> {
216 match self {
217 Self::SingleVolume(fs) => fs.shutdown().await.context("shutting down single volume"),
218 Self::MultiVolume(fs, _volume) => {
219 fs.shutdown().await.context("shutting down multi volume")
220 }
221 }
222 }
223
224 fn exposed_dir(&self) -> Result<&fio::DirectoryProxy, Error> {
225 match self {
226 Self::SingleVolume(fs) => Ok(fs.exposed_dir()),
227 Self::MultiVolume(_fs, volume) => Ok(volume.exposed_dir()),
228 }
229 }
230
231 fn blob_dir_name(&self) -> &'static str {
233 match self {
234 Self::SingleVolume(_) => "blob-exec",
235 Self::MultiVolume(_, _) => "root",
236 }
237 }
238
239 fn svc_dir(&self) -> Result<fio::DirectoryProxy, Error> {
240 match self {
241 Self::SingleVolume(_) => Ok(fuchsia_fs::directory::open_directory_async(
242 self.exposed_dir()?,
243 ".",
244 fio::PERM_READABLE,
245 )
246 .context("opening svc dir")?),
247 Self::MultiVolume(_, _) => Ok(fuchsia_fs::directory::open_directory_async(
248 self.exposed_dir()?,
249 "svc",
250 fio::PERM_READABLE,
251 )
252 .context("opening svc dir")?),
253 }
254 }
255
256 fn blob_creator_proxy(&self) -> Result<ffxfs::BlobCreatorProxy, Error> {
257 fuchsia_component::client::connect_to_protocol_at_dir_root::<ffxfs::BlobCreatorMarker>(
258 &self.svc_dir()?,
259 )
260 .context("connecting to fuchsia.fxfs.BlobCreator")
261 }
262
263 fn blob_reader_proxy(&self) -> Result<ffxfs::BlobReaderProxy, Error> {
264 fuchsia_component::client::connect_to_protocol_at_dir_root::<ffxfs::BlobReaderMarker>(
265 &self.svc_dir()?,
266 )
267 .context("connecting to fuchsia.fxfs.BlobReader")
268 }
269
270 fn overwrite_configuration_proxy(
271 &self,
272 ) -> Result<fidl_fuchsia_storage_blobfs::OverwriteConfigurationProxy, Error> {
273 fuchsia_component::client::connect_to_protocol_at_dir_root::<
274 fidl_fuchsia_storage_blobfs::OverwriteConfigurationMarker,
275 >(&self.svc_dir()?)
276 .context("connecting to fuchsia.storage.blobfs.OverwriteConfiguration")
277 }
278
279 fn implementation(&self) -> Implementation {
280 match self {
281 Self::SingleVolume(_) => Implementation::CppBlobfs,
282 Self::MultiVolume(_, _) => Implementation::Fxblob,
283 }
284 }
285}
286
287impl BlobfsRamdisk {
288 pub fn builder() -> BlobfsRamdiskBuilder {
290 BlobfsRamdiskBuilder::new()
291 }
292
293 pub async fn start() -> Result<Self, Error> {
295 Self::builder().start().await
296 }
297
298 pub fn client(&self) -> blobfs::Client {
304 blobfs::Client::new(
305 self.root_dir_proxy().unwrap(),
306 Some(self.blob_creator_proxy().unwrap()),
307 self.blob_reader_proxy().unwrap(),
308 None,
309 )
310 .unwrap()
311 }
312
313 pub fn root_dir_handle(&self) -> Result<ClientEnd<fio::DirectoryMarker>, Error> {
315 let (root_clone, server_end) = zx::Channel::create();
316 self.fs.exposed_dir()?.open(
317 self.fs.blob_dir_name(),
318 fio::PERM_READABLE | fio::Flags::PERM_INHERIT_WRITE | fio::Flags::PERM_EXECUTE,
319 &Default::default(),
320 server_end,
321 )?;
322 Ok(root_clone.into())
323 }
324
325 pub fn root_dir_proxy(&self) -> Result<fio::DirectoryProxy, Error> {
327 Ok(self.root_dir_handle()?.into_proxy())
328 }
329
330 pub fn root_dir(&self) -> Result<openat::Dir, Error> {
332 use std::os::fd::{FromRawFd as _, IntoRawFd as _, OwnedFd};
333
334 let fd: OwnedFd =
335 fdio::create_fd(self.root_dir_handle()?.into()).context("failed to create fd")?;
336
337 unsafe { Ok(openat::Dir::from_raw_fd(fd.into_raw_fd())) }
342 }
343
344 pub async fn into_builder(self) -> Result<BlobfsRamdiskBuilder, Error> {
347 let implementation = self.fs.implementation();
348 let ramdisk = self.unmount().await?;
349 Ok(Self::builder().formatted_ramdisk(ramdisk).implementation(implementation))
350 }
351
352 pub async fn unmount(self) -> Result<FormattedRamdisk, Error> {
354 self.fs.shutdown().await?;
355 Ok(self.backing_ramdisk)
356 }
357
358 pub async fn stop(self) -> Result<(), Error> {
360 self.unmount().await?.stop().await
361 }
362
363 pub fn list_blobs(&self) -> Result<BTreeSet<Hash>, Error> {
365 self.root_dir()?
366 .list_dir(".")?
367 .map(|entry| {
368 Ok(entry?
369 .file_name()
370 .to_str()
371 .ok_or_else(|| anyhow!("expected valid utf-8"))?
372 .parse()?)
373 })
374 .collect()
375 }
376
377 pub async fn add_blob_from(
379 &self,
380 merkle: Hash,
381 mut source: impl std::io::Read,
382 ) -> Result<(), Error> {
383 let mut bytes = vec![];
384 source.read_to_end(&mut bytes)?;
385 self.write_blob(merkle, &bytes).await
386 }
387
388 pub async fn write_blob(&self, merkle: Hash, bytes: &[u8]) -> Result<(), Error> {
391 let compressed_data = Type1Blob::generate(bytes, CompressionMode::Attempt);
392 let blob_creator = self.blob_creator_proxy()?;
393 let writer_client_end = match blob_creator.create(&merkle.into(), false).await? {
394 Ok(writer_client_end) => writer_client_end,
395 Err(ffxfs::CreateBlobError::AlreadyExists) => {
396 return Ok(());
397 }
398 Err(e) => {
399 return Err(anyhow!("create blob error {:?}", e));
400 }
401 };
402 let writer = writer_client_end.into_proxy();
403 let mut blob_writer = blob_writer::BlobWriter::create(writer, compressed_data.len() as u64)
404 .await
405 .context("failed to create BlobWriter")?;
406 blob_writer.write(&compressed_data).await?;
407 Ok(())
408 }
409
410 pub fn svc_dir(&self) -> Result<fio::DirectoryProxy, Error> {
414 self.fs.svc_dir()
415 }
416
417 pub fn blob_creator_proxy(&self) -> Result<ffxfs::BlobCreatorProxy, Error> {
419 self.fs.blob_creator_proxy()
420 }
421
422 pub fn blob_reader_proxy(&self) -> Result<ffxfs::BlobReaderProxy, Error> {
424 self.fs.blob_reader_proxy()
425 }
426
427 pub fn overwrite_configuration_proxy(
429 &self,
430 ) -> Result<fidl_fuchsia_storage_blobfs::OverwriteConfigurationProxy, Error> {
431 self.fs.overwrite_configuration_proxy()
432 }
433}
434
435pub struct RamdiskBuilder {
437 block_count: u64,
438}
439
440impl RamdiskBuilder {
441 fn new() -> Self {
442 Self { block_count: 1 << 20 }
443 }
444
445 pub fn block_count(mut self, block_count: u64) -> Self {
447 self.block_count = block_count;
448 self
449 }
450
451 pub async fn start(self) -> Result<Ramdisk, Error> {
453 let client = ramdevice_client::RamdiskClient::builder(RAMDISK_BLOCK_SIZE, self.block_count);
454 let client = client.build().await?;
455 Ok(Ramdisk { client })
456 }
457
458 pub async fn into_blobfs_builder(self) -> Result<BlobfsRamdiskBuilder, Error> {
460 Ok(BlobfsRamdiskBuilder::new().ramdisk(self.start().await?))
461 }
462}
463
464pub struct Ramdisk {
466 client: ramdevice_client::RamdiskClient,
467}
468
469impl Ramdisk {
472 pub fn builder() -> RamdiskBuilder {
474 RamdiskBuilder::new()
475 }
476
477 pub async fn start() -> Result<Self, Error> {
480 Self::builder().start().await
481 }
482
483 pub async fn stop(self) -> Result<(), Error> {
485 self.client.destroy().await
486 }
487}
488
489pub struct FormattedRamdisk(Ramdisk);
491
492impl std::ops::Deref for FormattedRamdisk {
494 type Target = Ramdisk;
495 fn deref(&self) -> &Self::Target {
496 &self.0
497 }
498}
499
500impl FormattedRamdisk {
501 pub async fn stop(self) -> Result<(), Error> {
503 self.0.stop().await
504 }
505}
506
507#[cfg(test)]
508mod tests {
509 use super::*;
510 use test_case::test_case;
511
512 #[test_case(Implementation::CppBlobfs; "blobfs")]
513 #[test_case(Implementation::Fxblob; "fxblob")]
514 #[fuchsia_async::run_singlethreaded(test)]
515 async fn clean_start_and_stop(implementation: Implementation) {
516 let blobfs = BlobfsRamdisk::builder().implementation(implementation).start().await.unwrap();
517
518 let proxy = blobfs.root_dir_proxy().unwrap();
519 drop(proxy);
520
521 blobfs.stop().await.unwrap();
522 }
523
524 #[test_case(Implementation::CppBlobfs; "blobfs")]
525 #[test_case(Implementation::Fxblob; "fxblob")]
526 #[fuchsia_async::run_singlethreaded(test)]
527 async fn clean_start_contains_no_blobs(implementation: Implementation) {
528 let blobfs = BlobfsRamdisk::builder().implementation(implementation).start().await.unwrap();
529
530 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::new());
531
532 blobfs.stop().await.unwrap();
533 }
534
535 #[test]
536 fn blob_info_conversions() {
537 let a = BlobInfo::from(&b"static slice"[..]);
538 let b = BlobInfo::from(b"owned vec".to_vec());
539 let c = BlobInfo::from(Cow::from(&b"cow"[..]));
540 assert_ne!(a.merkle, b.merkle);
541 assert_ne!(b.merkle, c.merkle);
542 assert_eq!(a.merkle, fuchsia_merkle::root_from_slice(b"static slice"));
543
544 let _ = BlobfsRamdisk::builder()
546 .with_blob(&b"static slice"[..])
547 .with_blob(b"owned vec".to_vec())
548 .with_blob(Cow::from(&b"cow"[..]));
549 }
550
551 #[test_case(Implementation::CppBlobfs; "blobfs")]
552 #[test_case(Implementation::Fxblob; "fxblob")]
553 #[fuchsia_async::run_singlethreaded(test)]
554 async fn with_blob_ignores_duplicates(implementation: Implementation) {
555 let blob = BlobInfo::from(&b"duplicate"[..]);
556
557 let blobfs = BlobfsRamdisk::builder()
558 .implementation(implementation)
559 .with_blob(blob.clone())
560 .with_blob(blob.clone())
561 .start()
562 .await
563 .unwrap();
564 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([blob.merkle]));
565
566 let blobfs =
567 blobfs.into_builder().await.unwrap().with_blob(blob.clone()).start().await.unwrap();
568 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([blob.merkle]));
569 }
570
571 #[test_case(Implementation::CppBlobfs; "blobfs")]
572 #[test_case(Implementation::Fxblob; "fxblob")]
573 #[fuchsia_async::run_singlethreaded(test)]
574 async fn build_with_two_blobs(implementation: Implementation) {
575 let blobfs = BlobfsRamdisk::builder()
576 .implementation(implementation)
577 .with_blob(&b"blob 1"[..])
578 .with_blob(&b"blob 2"[..])
579 .start()
580 .await
581 .unwrap();
582
583 let expected = BTreeSet::from([
584 fuchsia_merkle::root_from_slice(b"blob 1"),
585 fuchsia_merkle::root_from_slice(b"blob 2"),
586 ]);
587 assert_eq!(expected.len(), 2);
588 assert_eq!(blobfs.list_blobs().unwrap(), expected);
589
590 blobfs.stop().await.unwrap();
591 }
592
593 #[test_case(Implementation::CppBlobfs; "blobfs")]
594 #[test_case(Implementation::Fxblob; "fxblob")]
595 #[fuchsia_async::run_singlethreaded(test)]
596 async fn remount(implementation: Implementation) {
597 let blobfs = BlobfsRamdisk::builder()
598 .implementation(implementation)
599 .with_blob(&b"test"[..])
600 .start()
601 .await
602 .unwrap();
603 let blobs = blobfs.list_blobs().unwrap();
604
605 let blobfs = blobfs.into_builder().await.unwrap().start().await.unwrap();
606
607 assert_eq!(blobs, blobfs.list_blobs().unwrap());
608
609 blobfs.stop().await.unwrap();
610 }
611
612 #[test_case(Implementation::CppBlobfs; "blobfs")]
613 #[test_case(Implementation::Fxblob; "fxblob")]
614 #[fuchsia_async::run_singlethreaded(test)]
615 async fn blob_appears_in_readdir(implementation: Implementation) {
616 let blobfs = BlobfsRamdisk::builder().implementation(implementation).start().await.unwrap();
617
618 let data = b"Hello blobfs!";
619 let hello_merkle = fuchsia_merkle::root_from_slice(data);
620 blobfs.write_blob(hello_merkle, data).await.unwrap();
621 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([hello_merkle]));
622
623 blobfs.stop().await.unwrap();
624 }
625
626 #[fuchsia_async::run_singlethreaded(test)]
627 async fn ramdisk_builder_sets_block_count() {
628 for block_count in [1, 2, 3, 16] {
629 let ramdisk = Ramdisk::builder().block_count(block_count).start().await.unwrap();
630 let client_end = ramdisk.client.open().unwrap();
631 let proxy = client_end.into_proxy();
632 let info = proxy.get_info().await.unwrap().unwrap();
633 assert_eq!(info.block_count, block_count);
634 }
635 }
636
637 #[test_case(Implementation::CppBlobfs; "blobfs")]
638 #[test_case(Implementation::Fxblob; "fxblob")]
639 #[fuchsia_async::run_singlethreaded(test)]
640 async fn ramdisk_into_blobfs_formats_ramdisk(implementation: Implementation) {
641 let _: BlobfsRamdisk = Ramdisk::builder()
642 .into_blobfs_builder()
643 .await
644 .unwrap()
645 .implementation(implementation)
646 .start()
647 .await
648 .unwrap();
649 }
650
651 #[test_case(Implementation::CppBlobfs; "blobfs")]
652 #[test_case(Implementation::Fxblob; "fxblob")]
653 #[fuchsia_async::run_singlethreaded(test)]
654 async fn read_and_write(implementation: Implementation) {
655 let blobfs = BlobfsRamdisk::builder().implementation(implementation).start().await.unwrap();
656
657 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([]));
658 let data = "Hello blobfs!".as_bytes();
659 let merkle = fuchsia_merkle::root_from_slice(data);
660 blobfs.write_blob(merkle, data).await.unwrap();
661
662 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([merkle]));
663
664 blobfs.stop().await.unwrap();
665 }
666
667 #[test_case(Implementation::CppBlobfs; "blobfs")]
668 #[test_case(Implementation::Fxblob; "fxblob")]
669 #[fuchsia_async::run_singlethreaded(test)]
670 async fn blob_creator_api(implementation: Implementation) {
671 let blobfs = BlobfsRamdisk::builder().implementation(implementation).start().await.unwrap();
672 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([]));
673
674 let bytes = &[1u8; 40];
675 let hash = fuchsia_merkle::root_from_slice(bytes);
676 let compressed_data = Type1Blob::generate(bytes, CompressionMode::Always);
677
678 let blob_creator = blobfs.blob_creator_proxy().unwrap();
679 let blob_writer = blob_creator.create(&hash, false).await.unwrap().unwrap();
680 let mut blob_writer =
681 blob_writer::BlobWriter::create(blob_writer.into_proxy(), compressed_data.len() as u64)
682 .await
683 .unwrap();
684 let () = blob_writer.write(&compressed_data).await.unwrap();
685
686 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([hash]));
687
688 blobfs.stop().await.unwrap();
689 }
690
691 #[test_case(Implementation::CppBlobfs; "blobfs")]
692 #[test_case(Implementation::Fxblob; "fxblob")]
693 #[fuchsia_async::run_singlethreaded(test)]
694 async fn blob_reader_api(implementation: Implementation) {
695 let data = "Hello blobfs!".as_bytes();
696 let hash = fuchsia_merkle::root_from_slice(data);
697 let blobfs = BlobfsRamdisk::builder()
698 .implementation(implementation)
699 .with_blob(data)
700 .start()
701 .await
702 .unwrap();
703
704 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([hash]));
705
706 let blob_reader = blobfs.blob_reader_proxy().unwrap();
707 let vmo = blob_reader.get_vmo(&hash.into()).await.unwrap().unwrap();
708 let mut buf = vec![0; vmo.get_content_size().unwrap() as usize];
709 let () = vmo.read(&mut buf, 0).unwrap();
710 assert_eq!(buf, data);
711
712 blobfs.stop().await.unwrap();
713 }
714}