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 implementation(&self) -> Implementation {
271 match self {
272 Self::SingleVolume(_) => Implementation::CppBlobfs,
273 Self::MultiVolume(_, _) => Implementation::Fxblob,
274 }
275 }
276}
277
278impl BlobfsRamdisk {
279 pub fn builder() -> BlobfsRamdiskBuilder {
281 BlobfsRamdiskBuilder::new()
282 }
283
284 pub async fn start() -> Result<Self, Error> {
286 Self::builder().start().await
287 }
288
289 pub fn client(&self) -> blobfs::Client {
295 blobfs::Client::new(
296 self.root_dir_proxy().unwrap(),
297 Some(self.blob_creator_proxy().unwrap()),
298 self.blob_reader_proxy().unwrap(),
299 None,
300 )
301 .unwrap()
302 }
303
304 pub fn root_dir_handle(&self) -> Result<ClientEnd<fio::DirectoryMarker>, Error> {
306 let (root_clone, server_end) = zx::Channel::create();
307 self.fs.exposed_dir()?.open(
308 self.fs.blob_dir_name(),
309 fio::PERM_READABLE | fio::Flags::PERM_INHERIT_WRITE | fio::Flags::PERM_EXECUTE,
310 &Default::default(),
311 server_end,
312 )?;
313 Ok(root_clone.into())
314 }
315
316 pub fn root_dir_proxy(&self) -> Result<fio::DirectoryProxy, Error> {
318 Ok(self.root_dir_handle()?.into_proxy())
319 }
320
321 pub fn root_dir(&self) -> Result<openat::Dir, Error> {
323 use std::os::fd::{FromRawFd as _, IntoRawFd as _, OwnedFd};
324
325 let fd: OwnedFd =
326 fdio::create_fd(self.root_dir_handle()?.into()).context("failed to create fd")?;
327
328 unsafe { Ok(openat::Dir::from_raw_fd(fd.into_raw_fd())) }
333 }
334
335 pub async fn into_builder(self) -> Result<BlobfsRamdiskBuilder, Error> {
338 let implementation = self.fs.implementation();
339 let ramdisk = self.unmount().await?;
340 Ok(Self::builder().formatted_ramdisk(ramdisk).implementation(implementation))
341 }
342
343 pub async fn unmount(self) -> Result<FormattedRamdisk, Error> {
345 self.fs.shutdown().await?;
346 Ok(self.backing_ramdisk)
347 }
348
349 pub async fn stop(self) -> Result<(), Error> {
351 self.unmount().await?.stop().await
352 }
353
354 pub fn list_blobs(&self) -> Result<BTreeSet<Hash>, Error> {
356 self.root_dir()?
357 .list_dir(".")?
358 .map(|entry| {
359 Ok(entry?
360 .file_name()
361 .to_str()
362 .ok_or_else(|| anyhow!("expected valid utf-8"))?
363 .parse()?)
364 })
365 .collect()
366 }
367
368 pub async fn add_blob_from(
370 &self,
371 merkle: Hash,
372 mut source: impl std::io::Read,
373 ) -> Result<(), Error> {
374 let mut bytes = vec![];
375 source.read_to_end(&mut bytes)?;
376 self.write_blob(merkle, &bytes).await
377 }
378
379 pub async fn write_blob(&self, merkle: Hash, bytes: &[u8]) -> Result<(), Error> {
382 let compressed_data = Type1Blob::generate(bytes, CompressionMode::Attempt);
383 let blob_creator = self.blob_creator_proxy()?;
384 let writer_client_end = match blob_creator.create(&merkle.into(), false).await? {
385 Ok(writer_client_end) => writer_client_end,
386 Err(ffxfs::CreateBlobError::AlreadyExists) => {
387 return Ok(());
388 }
389 Err(e) => {
390 return Err(anyhow!("create blob error {:?}", e));
391 }
392 };
393 let writer = writer_client_end.into_proxy();
394 let mut blob_writer = blob_writer::BlobWriter::create(writer, compressed_data.len() as u64)
395 .await
396 .context("failed to create BlobWriter")?;
397 blob_writer.write(&compressed_data).await?;
398 Ok(())
399 }
400
401 pub fn svc_dir(&self) -> Result<fio::DirectoryProxy, Error> {
405 self.fs.svc_dir()
406 }
407
408 pub fn blob_creator_proxy(&self) -> Result<ffxfs::BlobCreatorProxy, Error> {
410 self.fs.blob_creator_proxy()
411 }
412
413 pub fn blob_reader_proxy(&self) -> Result<ffxfs::BlobReaderProxy, Error> {
415 self.fs.blob_reader_proxy()
416 }
417}
418
419pub struct RamdiskBuilder {
421 block_count: u64,
422}
423
424impl RamdiskBuilder {
425 fn new() -> Self {
426 Self { block_count: 1 << 20 }
427 }
428
429 pub fn block_count(mut self, block_count: u64) -> Self {
431 self.block_count = block_count;
432 self
433 }
434
435 pub async fn start(self) -> Result<Ramdisk, Error> {
437 let client = ramdevice_client::RamdiskClient::builder(RAMDISK_BLOCK_SIZE, self.block_count);
438 let client = client.build().await?;
439 Ok(Ramdisk { client })
440 }
441
442 pub async fn into_blobfs_builder(self) -> Result<BlobfsRamdiskBuilder, Error> {
444 Ok(BlobfsRamdiskBuilder::new().ramdisk(self.start().await?))
445 }
446}
447
448pub struct Ramdisk {
450 client: ramdevice_client::RamdiskClient,
451}
452
453impl Ramdisk {
456 pub fn builder() -> RamdiskBuilder {
458 RamdiskBuilder::new()
459 }
460
461 pub async fn start() -> Result<Self, Error> {
464 Self::builder().start().await
465 }
466
467 pub async fn stop(self) -> Result<(), Error> {
469 self.client.destroy().await
470 }
471}
472
473pub struct FormattedRamdisk(Ramdisk);
475
476impl std::ops::Deref for FormattedRamdisk {
478 type Target = Ramdisk;
479 fn deref(&self) -> &Self::Target {
480 &self.0
481 }
482}
483
484impl FormattedRamdisk {
485 pub async fn stop(self) -> Result<(), Error> {
487 self.0.stop().await
488 }
489}
490
491#[cfg(test)]
492mod tests {
493 use super::*;
494 use test_case::test_case;
495
496 #[test_case(Implementation::CppBlobfs; "blobfs")]
497 #[test_case(Implementation::Fxblob; "fxblob")]
498 #[fuchsia_async::run_singlethreaded(test)]
499 async fn clean_start_and_stop(implementation: Implementation) {
500 let blobfs = BlobfsRamdisk::builder().implementation(implementation).start().await.unwrap();
501
502 let proxy = blobfs.root_dir_proxy().unwrap();
503 drop(proxy);
504
505 blobfs.stop().await.unwrap();
506 }
507
508 #[test_case(Implementation::CppBlobfs; "blobfs")]
509 #[test_case(Implementation::Fxblob; "fxblob")]
510 #[fuchsia_async::run_singlethreaded(test)]
511 async fn clean_start_contains_no_blobs(implementation: Implementation) {
512 let blobfs = BlobfsRamdisk::builder().implementation(implementation).start().await.unwrap();
513
514 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::new());
515
516 blobfs.stop().await.unwrap();
517 }
518
519 #[test]
520 fn blob_info_conversions() {
521 let a = BlobInfo::from(&b"static slice"[..]);
522 let b = BlobInfo::from(b"owned vec".to_vec());
523 let c = BlobInfo::from(Cow::from(&b"cow"[..]));
524 assert_ne!(a.merkle, b.merkle);
525 assert_ne!(b.merkle, c.merkle);
526 assert_eq!(a.merkle, fuchsia_merkle::root_from_slice(b"static slice"));
527
528 let _ = BlobfsRamdisk::builder()
530 .with_blob(&b"static slice"[..])
531 .with_blob(b"owned vec".to_vec())
532 .with_blob(Cow::from(&b"cow"[..]));
533 }
534
535 #[test_case(Implementation::CppBlobfs; "blobfs")]
536 #[test_case(Implementation::Fxblob; "fxblob")]
537 #[fuchsia_async::run_singlethreaded(test)]
538 async fn with_blob_ignores_duplicates(implementation: Implementation) {
539 let blob = BlobInfo::from(&b"duplicate"[..]);
540
541 let blobfs = BlobfsRamdisk::builder()
542 .implementation(implementation)
543 .with_blob(blob.clone())
544 .with_blob(blob.clone())
545 .start()
546 .await
547 .unwrap();
548 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([blob.merkle]));
549
550 let blobfs =
551 blobfs.into_builder().await.unwrap().with_blob(blob.clone()).start().await.unwrap();
552 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([blob.merkle]));
553 }
554
555 #[test_case(Implementation::CppBlobfs; "blobfs")]
556 #[test_case(Implementation::Fxblob; "fxblob")]
557 #[fuchsia_async::run_singlethreaded(test)]
558 async fn build_with_two_blobs(implementation: Implementation) {
559 let blobfs = BlobfsRamdisk::builder()
560 .implementation(implementation)
561 .with_blob(&b"blob 1"[..])
562 .with_blob(&b"blob 2"[..])
563 .start()
564 .await
565 .unwrap();
566
567 let expected = BTreeSet::from([
568 fuchsia_merkle::root_from_slice(b"blob 1"),
569 fuchsia_merkle::root_from_slice(b"blob 2"),
570 ]);
571 assert_eq!(expected.len(), 2);
572 assert_eq!(blobfs.list_blobs().unwrap(), expected);
573
574 blobfs.stop().await.unwrap();
575 }
576
577 #[test_case(Implementation::CppBlobfs; "blobfs")]
578 #[test_case(Implementation::Fxblob; "fxblob")]
579 #[fuchsia_async::run_singlethreaded(test)]
580 async fn remount(implementation: Implementation) {
581 let blobfs = BlobfsRamdisk::builder()
582 .implementation(implementation)
583 .with_blob(&b"test"[..])
584 .start()
585 .await
586 .unwrap();
587 let blobs = blobfs.list_blobs().unwrap();
588
589 let blobfs = blobfs.into_builder().await.unwrap().start().await.unwrap();
590
591 assert_eq!(blobs, blobfs.list_blobs().unwrap());
592
593 blobfs.stop().await.unwrap();
594 }
595
596 #[test_case(Implementation::CppBlobfs; "blobfs")]
597 #[test_case(Implementation::Fxblob; "fxblob")]
598 #[fuchsia_async::run_singlethreaded(test)]
599 async fn blob_appears_in_readdir(implementation: Implementation) {
600 let blobfs = BlobfsRamdisk::builder().implementation(implementation).start().await.unwrap();
601
602 let data = b"Hello blobfs!";
603 let hello_merkle = fuchsia_merkle::root_from_slice(data);
604 blobfs.write_blob(hello_merkle, data).await.unwrap();
605 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([hello_merkle]));
606
607 blobfs.stop().await.unwrap();
608 }
609
610 #[fuchsia_async::run_singlethreaded(test)]
611 async fn ramdisk_builder_sets_block_count() {
612 for block_count in [1, 2, 3, 16] {
613 let ramdisk = Ramdisk::builder().block_count(block_count).start().await.unwrap();
614 let client_end = ramdisk.client.open().unwrap();
615 let proxy = client_end.into_proxy();
616 let info = proxy.get_info().await.unwrap().unwrap();
617 assert_eq!(info.block_count, block_count);
618 }
619 }
620
621 #[test_case(Implementation::CppBlobfs; "blobfs")]
622 #[test_case(Implementation::Fxblob; "fxblob")]
623 #[fuchsia_async::run_singlethreaded(test)]
624 async fn ramdisk_into_blobfs_formats_ramdisk(implementation: Implementation) {
625 let _: BlobfsRamdisk = Ramdisk::builder()
626 .into_blobfs_builder()
627 .await
628 .unwrap()
629 .implementation(implementation)
630 .start()
631 .await
632 .unwrap();
633 }
634
635 #[test_case(Implementation::CppBlobfs; "blobfs")]
636 #[test_case(Implementation::Fxblob; "fxblob")]
637 #[fuchsia_async::run_singlethreaded(test)]
638 async fn read_and_write(implementation: Implementation) {
639 let blobfs = BlobfsRamdisk::builder().implementation(implementation).start().await.unwrap();
640
641 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([]));
642 let data = "Hello blobfs!".as_bytes();
643 let merkle = fuchsia_merkle::root_from_slice(data);
644 blobfs.write_blob(merkle, data).await.unwrap();
645
646 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([merkle]));
647
648 blobfs.stop().await.unwrap();
649 }
650
651 #[test_case(Implementation::CppBlobfs; "blobfs")]
652 #[test_case(Implementation::Fxblob; "fxblob")]
653 #[fuchsia_async::run_singlethreaded(test)]
654 async fn blob_creator_api(implementation: Implementation) {
655 let blobfs = BlobfsRamdisk::builder().implementation(implementation).start().await.unwrap();
656 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([]));
657
658 let bytes = &[1u8; 40];
659 let hash = fuchsia_merkle::root_from_slice(bytes);
660 let compressed_data = Type1Blob::generate(bytes, CompressionMode::Always);
661
662 let blob_creator = blobfs.blob_creator_proxy().unwrap();
663 let blob_writer = blob_creator.create(&hash, false).await.unwrap().unwrap();
664 let mut blob_writer =
665 blob_writer::BlobWriter::create(blob_writer.into_proxy(), compressed_data.len() as u64)
666 .await
667 .unwrap();
668 let () = blob_writer.write(&compressed_data).await.unwrap();
669
670 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([hash]));
671
672 blobfs.stop().await.unwrap();
673 }
674
675 #[test_case(Implementation::CppBlobfs; "blobfs")]
676 #[test_case(Implementation::Fxblob; "fxblob")]
677 #[fuchsia_async::run_singlethreaded(test)]
678 async fn blob_reader_api(implementation: Implementation) {
679 let data = "Hello blobfs!".as_bytes();
680 let hash = fuchsia_merkle::root_from_slice(data);
681 let blobfs = BlobfsRamdisk::builder()
682 .implementation(implementation)
683 .with_blob(data)
684 .start()
685 .await
686 .unwrap();
687
688 assert_eq!(blobfs.list_blobs().unwrap(), BTreeSet::from([hash]));
689
690 let blob_reader = blobfs.blob_reader_proxy().unwrap();
691 let vmo = blob_reader.get_vmo(&hash.into()).await.unwrap().unwrap();
692 let mut buf = vec![0; vmo.get_content_size().unwrap() as usize];
693 let () = vmo.read(&mut buf, 0).unwrap();
694 assert_eq!(buf, data);
695
696 blobfs.stop().await.unwrap();
697 }
698}