1#![deny(missing_docs)]
6#![allow(clippy::let_unit_value)]
7
8use anyhow::{anyhow, Context as _, Error};
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::from_slice(&bytes).root(), 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()?.into_proxy();
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 mut fs =
153 fs.serve_multi_volume().await.context("serving multi volume filesystem")?;
154 if needs_format {
155 let _: &mut fs_management::filesystem::ServingVolume = fs
156 .create_volume(
157 FXFS_BLOB_VOLUME_NAME,
158 CreateOptions::default(),
159 MountOptions { as_blob: Some(true), ..MountOptions::default() },
160 )
161 .await
162 .context("creating blob volume")?;
163 } else {
164 let _: &mut fs_management::filesystem::ServingVolume = fs
165 .open_volume(
166 FXFS_BLOB_VOLUME_NAME,
167 MountOptions { as_blob: Some(true), ..MountOptions::default() },
168 )
169 .await
170 .context("opening blob volume")?;
171 }
172 ServingFilesystem::MultiVolume(fs)
173 }
174 };
175
176 let blobfs = BlobfsRamdisk { backing_ramdisk: FormattedRamdisk(ramdisk), fs };
177
178 if !blobs.is_empty() {
180 let mut present_blobs = blobfs.list_blobs()?;
181
182 for blob in blobs {
183 if present_blobs.contains(&blob.merkle) {
184 continue;
185 }
186 blobfs
187 .write_blob(blob.merkle, &blob.contents)
188 .await
189 .context(format!("writing {}", blob.merkle))?;
190 present_blobs.insert(blob.merkle);
191 }
192 }
193
194 Ok(blobfs)
195 }
196}
197
198pub struct BlobfsRamdisk {
200 backing_ramdisk: FormattedRamdisk,
201 fs: ServingFilesystem,
202}
203
204enum ServingFilesystem {
209 SingleVolume(fs_management::filesystem::ServingSingleVolumeFilesystem),
210 MultiVolume(fs_management::filesystem::ServingMultiVolumeFilesystem),
211}
212
213impl ServingFilesystem {
214 async fn shutdown(self) -> Result<(), Error> {
215 match self {
216 Self::SingleVolume(fs) => fs.shutdown().await.context("shutting down single volume"),
217 Self::MultiVolume(fs) => fs.shutdown().await.context("shutting down multi volume"),
218 }
219 }
220
221 fn exposed_dir(&self) -> Result<&fio::DirectoryProxy, Error> {
222 match self {
223 Self::SingleVolume(fs) => Ok(fs.exposed_dir()),
224 Self::MultiVolume(fs) => Ok(fs
225 .volume(FXFS_BLOB_VOLUME_NAME)
226 .ok_or(anyhow!("missing blob volume"))?
227 .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<Option<fio::DirectoryProxy>, Error> {
241 match self {
242 Self::SingleVolume(_) => Ok(Some(
243 fuchsia_fs::directory::open_directory_async(
244 self.exposed_dir()?,
245 ".",
246 fio::PERM_READABLE,
247 )
248 .context("opening svc dir")?,
249 )),
250 Self::MultiVolume(_) => Ok(Some(
251 fuchsia_fs::directory::open_directory_async(
252 self.exposed_dir()?,
253 "svc",
254 fio::PERM_READABLE,
255 )
256 .context("opening svc dir")?,
257 )),
258 }
259 }
260
261 fn blob_creator_proxy(&self) -> Result<Option<ffxfs::BlobCreatorProxy>, Error> {
263 Ok(match self.svc_dir()? {
264 Some(d) => Some(
265 fuchsia_component::client::connect_to_protocol_at_dir_root::<
266 ffxfs::BlobCreatorMarker,
267 >(&d)
268 .context("connecting to fuchsia.fxfs.BlobCreator")?,
269 ),
270 None => None,
271 })
272 }
273
274 fn blob_reader_proxy(&self) -> Result<Option<ffxfs::BlobReaderProxy>, Error> {
276 Ok(match self.svc_dir()? {
277 Some(d) => {
278 Some(
279 fuchsia_component::client::connect_to_protocol_at_dir_root::<
280 ffxfs::BlobReaderMarker,
281 >(&d)
282 .context("connecting to fuchsia.fxfs.BlobReader")?,
283 )
284 }
285 None => None,
286 })
287 }
288
289 fn implementation(&self) -> Implementation {
290 match self {
291 Self::SingleVolume(_) => Implementation::CppBlobfs,
292 Self::MultiVolume(_) => Implementation::Fxblob,
293 }
294 }
295}
296
297impl BlobfsRamdisk {
298 pub fn builder() -> BlobfsRamdiskBuilder {
300 BlobfsRamdiskBuilder::new()
301 }
302
303 pub async fn start() -> Result<Self, Error> {
305 Self::builder().start().await
306 }
307
308 pub fn client(&self) -> blobfs::Client {
314 blobfs::Client::new(
315 self.root_dir_proxy().unwrap(),
316 self.blob_creator_proxy().unwrap(),
317 self.blob_reader_proxy().unwrap(),
318 None,
319 )
320 .unwrap()
321 }
322
323 pub fn root_dir_handle(&self) -> Result<ClientEnd<fio::DirectoryMarker>, Error> {
325 let (root_clone, server_end) = zx::Channel::create();
326 self.fs.exposed_dir()?.open(
327 self.fs.blob_dir_name(),
328 fio::PERM_READABLE | fio::Flags::PERM_INHERIT_WRITE | fio::Flags::PERM_EXECUTE,
329 &Default::default(),
330 server_end,
331 )?;
332 Ok(root_clone.into())
333 }
334
335 pub fn root_dir_proxy(&self) -> Result<fio::DirectoryProxy, Error> {
337 Ok(self.root_dir_handle()?.into_proxy())
338 }
339
340 pub fn root_dir(&self) -> Result<openat::Dir, Error> {
342 use std::os::fd::{FromRawFd as _, IntoRawFd as _, OwnedFd};
343
344 let fd: OwnedFd =
345 fdio::create_fd(self.root_dir_handle()?.into()).context("failed to create fd")?;
346
347 unsafe { Ok(openat::Dir::from_raw_fd(fd.into_raw_fd())) }
352 }
353
354 pub async fn into_builder(self) -> Result<BlobfsRamdiskBuilder, Error> {
357 let implementation = self.fs.implementation();
358 let ramdisk = self.unmount().await?;
359 Ok(Self::builder().formatted_ramdisk(ramdisk).implementation(implementation))
360 }
361
362 pub async fn unmount(self) -> Result<FormattedRamdisk, Error> {
364 self.fs.shutdown().await?;
365 Ok(self.backing_ramdisk)
366 }
367
368 pub async fn stop(self) -> Result<(), Error> {
370 self.unmount().await?.stop().await
371 }
372
373 pub fn list_blobs(&self) -> Result<BTreeSet<Hash>, Error> {
375 self.root_dir()?
376 .list_dir(".")?
377 .map(|entry| {
378 Ok(entry?
379 .file_name()
380 .to_str()
381 .ok_or_else(|| anyhow!("expected valid utf-8"))?
382 .parse()?)
383 })
384 .collect()
385 }
386
387 pub async fn add_blob_from(
389 &self,
390 merkle: Hash,
391 mut source: impl std::io::Read,
392 ) -> Result<(), Error> {
393 let mut bytes = vec![];
394 source.read_to_end(&mut bytes)?;
395 self.write_blob(merkle, &bytes).await
396 }
397
398 pub async fn write_blob(&self, merkle: Hash, bytes: &[u8]) -> Result<(), Error> {
401 let compressed_data = Type1Blob::generate(bytes, CompressionMode::Attempt);
402 let blob_creator = self
403 .blob_creator_proxy()?
404 .ok_or_else(|| anyhow!("The filesystem does not expose the BlobCreator service"))?;
405 let writer_client_end = match blob_creator.create(&merkle.into(), false).await? {
406 Ok(writer_client_end) => writer_client_end,
407 Err(ffxfs::CreateBlobError::AlreadyExists) => {
408 return Ok(());
409 }
410 Err(e) => {
411 return Err(anyhow!("create blob error {:?}", e));
412 }
413 };
414 let writer = writer_client_end.into_proxy();
415 let mut blob_writer = blob_writer::BlobWriter::create(writer, compressed_data.len() as u64)
416 .await
417 .context("failed to create BlobWriter")?;
418 blob_writer.write(&compressed_data).await?;
419 Ok(())
420 }
421
422 pub fn svc_dir(&self) -> Result<Option<fio::DirectoryProxy>, Error> {
427 self.fs.svc_dir()
428 }
429
430 pub fn blob_creator_proxy(&self) -> Result<Option<ffxfs::BlobCreatorProxy>, Error> {
433 self.fs.blob_creator_proxy()
434 }
435
436 pub fn blob_reader_proxy(&self) -> Result<Option<ffxfs::BlobReaderProxy>, Error> {
439 self.fs.blob_reader_proxy()
440 }
441}
442
443pub struct RamdiskBuilder {
445 block_count: u64,
446}
447
448impl RamdiskBuilder {
449 fn new() -> Self {
450 Self { block_count: 1 << 20 }
451 }
452
453 pub fn block_count(mut self, block_count: u64) -> Self {
455 self.block_count = block_count;
456 self
457 }
458
459 pub async fn start(self) -> Result<Ramdisk, Error> {
461 let client = ramdevice_client::RamdiskClient::builder(RAMDISK_BLOCK_SIZE, self.block_count);
462 let client = client.build().await?;
463 Ok(Ramdisk { client })
464 }
465
466 pub async fn into_blobfs_builder(self) -> Result<BlobfsRamdiskBuilder, Error> {
468 Ok(BlobfsRamdiskBuilder::new().ramdisk(self.start().await?))
469 }
470}
471
472pub struct Ramdisk {
474 client: ramdevice_client::RamdiskClient,
475}
476
477impl Ramdisk {
480 pub fn builder() -> RamdiskBuilder {
482 RamdiskBuilder::new()
483 }
484
485 pub async fn start() -> Result<Self, Error> {
488 Self::builder().start().await
489 }
490
491 pub async fn stop(self) -> Result<(), Error> {
493 self.client.destroy().await
494 }
495}
496
497pub struct FormattedRamdisk(Ramdisk);
499
500impl std::ops::Deref for FormattedRamdisk {
502 type Target = Ramdisk;
503 fn deref(&self) -> &Self::Target {
504 &self.0
505 }
506}
507
508impl FormattedRamdisk {
509 pub async fn stop(self) -> Result<(), Error> {
511 self.0.stop().await
512 }
513}
514
515#[cfg(test)]
516mod tests {
517 use super::*;
518 use delivery_blob::delivery_blob_path;
519 use std::io::Write as _;
520
521 #[fuchsia_async::run_singlethreaded(test)]
522 async fn clean_start_and_stop() {
523 let blobfs = BlobfsRamdisk::start().await.unwrap();
524
525 let proxy = blobfs.root_dir_proxy().unwrap();
526 drop(proxy);
527
528 blobfs.stop().await.unwrap();
529 }
530
531 #[fuchsia_async::run_singlethreaded(test)]
532 async fn clean_start_contains_no_blobs() {
533 let blobfs = BlobfsRamdisk::start().await.unwrap();
534
535 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::new());
536
537 blobfs.stop().await.unwrap();
538 }
539
540 #[test]
541 fn blob_info_conversions() {
542 let a = BlobInfo::from(&b"static slice"[..]);
543 let b = BlobInfo::from(b"owned vec".to_vec());
544 let c = BlobInfo::from(Cow::from(&b"cow"[..]));
545 assert_ne!(a.merkle, b.merkle);
546 assert_ne!(b.merkle, c.merkle);
547 assert_eq!(a.merkle, fuchsia_merkle::from_slice(&b"static slice"[..]).root());
548
549 let _ = BlobfsRamdisk::builder()
551 .with_blob(&b"static slice"[..])
552 .with_blob(b"owned vec".to_vec())
553 .with_blob(Cow::from(&b"cow"[..]));
554 }
555
556 #[fuchsia_async::run_singlethreaded(test)]
557 async fn with_blob_ignores_duplicates() {
558 let blob = BlobInfo::from(&b"duplicate"[..]);
559
560 let blobfs = BlobfsRamdisk::builder()
561 .with_blob(blob.clone())
562 .with_blob(blob.clone())
563 .start()
564 .await
565 .unwrap();
566 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([blob.merkle]));
567
568 let blobfs =
569 blobfs.into_builder().await.unwrap().with_blob(blob.clone()).start().await.unwrap();
570 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([blob.merkle]));
571 }
572
573 #[fuchsia_async::run_singlethreaded(test)]
574 async fn build_with_two_blobs() {
575 let blobfs = BlobfsRamdisk::builder()
576 .with_blob(&b"blob 1"[..])
577 .with_blob(&b"blob 2"[..])
578 .start()
579 .await
580 .unwrap();
581
582 let expected = BTreeSet::from([
583 fuchsia_merkle::from_slice(&b"blob 1"[..]).root(),
584 fuchsia_merkle::from_slice(&b"blob 2"[..]).root(),
585 ]);
586 assert_eq!(expected.len(), 2);
587 assert_eq!(blobfs.list_blobs().unwrap(), expected);
588
589 blobfs.stop().await.unwrap();
590 }
591
592 #[fuchsia_async::run_singlethreaded(test)]
593 async fn blobfs_remount() {
594 let blobfs =
595 BlobfsRamdisk::builder().cpp_blobfs().with_blob(&b"test"[..]).start().await.unwrap();
596 let blobs = blobfs.list_blobs().unwrap();
597
598 let blobfs = blobfs.into_builder().await.unwrap().start().await.unwrap();
599
600 assert_eq!(blobs, blobfs.list_blobs().unwrap());
601
602 blobfs.stop().await.unwrap();
603 }
604
605 #[fuchsia_async::run_singlethreaded(test)]
606 async fn fxblob_remount() {
607 let blobfs =
608 BlobfsRamdisk::builder().fxblob().with_blob(&b"test"[..]).start().await.unwrap();
609 let blobs = blobfs.list_blobs().unwrap();
610
611 let blobfs = blobfs.into_builder().await.unwrap().start().await.unwrap();
612
613 assert_eq!(blobs, blobfs.list_blobs().unwrap());
614
615 blobfs.stop().await.unwrap();
616 }
617
618 #[fuchsia_async::run_singlethreaded(test)]
619 async fn blob_appears_in_readdir() {
620 let blobfs = BlobfsRamdisk::start().await.unwrap();
621 let root = blobfs.root_dir().unwrap();
622
623 let hello_merkle = write_blob(&root, "Hello blobfs!".as_bytes());
624 assert_eq!(list_blobs(&root), vec![hello_merkle]);
625
626 drop(root);
627 blobfs.stop().await.unwrap();
628 }
629
630 #[allow(clippy::zero_prefixed_literal)]
632 fn write_blob(dir: &openat::Dir, payload: &[u8]) -> String {
633 let merkle = fuchsia_merkle::from_slice(payload).root().to_string();
634 let compressed_data = Type1Blob::generate(payload, CompressionMode::Always);
635 let mut f = dir.new_file(delivery_blob_path(&merkle), 0600).unwrap();
636 f.set_len(compressed_data.len() as u64).unwrap();
637 f.write_all(&compressed_data).unwrap();
638
639 merkle
640 }
641
642 fn list_blobs(dir: &openat::Dir) -> Vec<String> {
644 dir.list_dir(".")
645 .unwrap()
646 .map(|entry| entry.unwrap().file_name().to_owned().into_string().unwrap())
647 .collect()
648 }
649
650 #[fuchsia_async::run_singlethreaded(test)]
651 async fn ramdisk_builder_sets_block_count() {
652 for block_count in [1, 2, 3, 16] {
653 let ramdisk = Ramdisk::builder().block_count(block_count).start().await.unwrap();
654 let client_end = ramdisk.client.open().unwrap();
655 let proxy = client_end.into_proxy();
656 let info = proxy.get_info().await.unwrap().unwrap();
657 assert_eq!(info.block_count, block_count);
658 }
659 }
660
661 #[fuchsia_async::run_singlethreaded(test)]
662 async fn ramdisk_into_blobfs_formats_ramdisk() {
663 let _: BlobfsRamdisk =
664 Ramdisk::builder().into_blobfs_builder().await.unwrap().start().await.unwrap();
665 }
666
667 #[fuchsia_async::run_singlethreaded(test)]
668 async fn blobfs_supports_blob_creator_api() {
669 let blobfs = BlobfsRamdisk::builder().cpp_blobfs().start().await.unwrap();
670
671 assert!(blobfs.blob_creator_proxy().unwrap().is_some());
672
673 blobfs.stop().await.unwrap();
674 }
675
676 #[fuchsia_async::run_singlethreaded(test)]
677 async fn blobfs_supports_blob_reader_api() {
678 let blobfs = BlobfsRamdisk::builder().cpp_blobfs().start().await.unwrap();
679
680 assert!(blobfs.blob_reader_proxy().unwrap().is_some());
681
682 blobfs.stop().await.unwrap();
683 }
684
685 #[fuchsia_async::run_singlethreaded(test)]
686 async fn fxblob_read_and_write() {
687 let blobfs = BlobfsRamdisk::builder().fxblob().start().await.unwrap();
688 let root = blobfs.root_dir().unwrap();
689
690 assert_eq!(list_blobs(&root), Vec::<String>::new());
691 let data = "Hello blobfs!".as_bytes();
692 let merkle = fuchsia_merkle::from_slice(data).root();
693 blobfs.write_blob(merkle, data).await.unwrap();
694
695 assert_eq!(list_blobs(&root), vec![merkle.to_string()]);
696
697 drop(root);
698 blobfs.stop().await.unwrap();
699 }
700
701 #[fuchsia_async::run_singlethreaded(test)]
702 async fn fxblob_blob_creator_api() {
703 let blobfs = BlobfsRamdisk::builder().fxblob().start().await.unwrap();
704 let root = blobfs.root_dir().unwrap();
705 assert_eq!(list_blobs(&root), Vec::<String>::new());
706
707 let bytes = [1u8; 40];
708 let hash = fuchsia_merkle::from_slice(&bytes).root();
709 let compressed_data = Type1Blob::generate(&bytes, CompressionMode::Always);
710
711 let blob_creator = blobfs.blob_creator_proxy().unwrap().unwrap();
712 let blob_writer = blob_creator.create(&hash, false).await.unwrap().unwrap();
713 let mut blob_writer =
714 blob_writer::BlobWriter::create(blob_writer.into_proxy(), compressed_data.len() as u64)
715 .await
716 .unwrap();
717 let () = blob_writer.write(&compressed_data).await.unwrap();
718
719 assert_eq!(list_blobs(&root), vec![hash.to_string()]);
720
721 drop(root);
722 blobfs.stop().await.unwrap();
723 }
724
725 #[fuchsia_async::run_singlethreaded(test)]
726 async fn fxblob_blob_reader_api() {
727 let data = "Hello blobfs!".as_bytes();
728 let hash = fuchsia_merkle::from_slice(data).root();
729 let blobfs = BlobfsRamdisk::builder().fxblob().with_blob(data).start().await.unwrap();
730
731 let root = blobfs.root_dir().unwrap();
732 assert_eq!(list_blobs(&root), vec![hash.to_string()]);
733
734 let blob_reader = blobfs.blob_reader_proxy().unwrap().unwrap();
735 let vmo = blob_reader.get_vmo(&hash.into()).await.unwrap().unwrap();
736 let mut buf = vec![0; vmo.get_content_size().unwrap() as usize];
737 let () = vmo.read(&mut buf, 0).unwrap();
738 assert_eq!(buf, data);
739
740 drop(root);
741 blobfs.stop().await.unwrap();
742 }
743}