| // Copyright 2021 The Fuchsia Authors. All rights reserved. |
| // Use of this source code is governed by a BSD-style license that can be |
| // found in the LICENSE file. |
| |
| #![deny(missing_docs)] |
| |
| //! Typesafe wrappers around the /blob filesystem. |
| |
| use { |
| fidl::endpoints::{Proxy as _, ServerEnd}, |
| fidl_fuchsia_fxfs as ffxfs, fidl_fuchsia_io as fio, fidl_fuchsia_pkg as fpkg, |
| fuchsia_hash::{Hash, ParseHashError}, |
| fuchsia_zircon::{self as zx, AsHandleRef as _, Status}, |
| futures::{stream, StreamExt as _}, |
| std::collections::HashSet, |
| thiserror::Error, |
| tracing::{error, info, warn}, |
| vfs::{ |
| common::send_on_open_with_error, execution_scope::ExecutionScope, file::StreamIoConnection, |
| ObjectRequest, ObjectRequestRef, ProtocolsExt, ToObjectRequest as _, |
| }, |
| }; |
| |
| pub mod mock; |
| pub use mock::Mock; |
| |
| /// Blobfs client errors. |
| #[derive(Debug, Error)] |
| #[allow(missing_docs)] |
| pub enum BlobfsError { |
| #[error("while opening blobfs dir")] |
| OpenDir(#[from] fuchsia_fs::node::OpenError), |
| |
| #[error("while cloning the blobfs dir")] |
| CloneDir(#[from] fuchsia_fs::node::CloneError), |
| |
| #[error("while listing blobfs dir")] |
| ReadDir(#[source] fuchsia_fs::directory::EnumerateError), |
| |
| #[error("while deleting blob")] |
| Unlink(#[source] Status), |
| |
| #[error("while sync'ing")] |
| Sync(#[source] Status), |
| |
| #[error("while parsing blob merkle hash")] |
| ParseHash(#[from] ParseHashError), |
| |
| #[error("FIDL error")] |
| Fidl(#[from] fidl::Error), |
| |
| #[error("while connecting to fuchsia.fxfs/BlobCreator")] |
| ConnectToBlobCreator(#[source] anyhow::Error), |
| |
| #[error("while connecting to fuchsia.fxfs/BlobReader")] |
| ConnectToBlobReader(#[source] anyhow::Error), |
| |
| #[error("while connecting to fuchsia.kernel/VmexResource")] |
| ConnectToVmexResource(#[source] anyhow::Error), |
| |
| #[error("while setting the VmexResource")] |
| InitVmexResource(#[source] anyhow::Error), |
| } |
| |
| /// An error encountered while creating a blob |
| #[derive(Debug, Error)] |
| #[allow(missing_docs)] |
| pub enum CreateError { |
| #[error("the blob already exists or is being concurrently written")] |
| AlreadyExists, |
| |
| #[error("while creating the blob")] |
| Io(#[source] fuchsia_fs::node::OpenError), |
| |
| #[error("while converting the proxy into a client end")] |
| ConvertToClientEnd, |
| |
| #[error("FIDL error")] |
| Fidl(#[from] fidl::Error), |
| |
| #[error("while calling fuchsia.fxfs/BlobCreator.Create: {0:?}")] |
| BlobCreator(ffxfs::CreateBlobError), |
| } |
| |
| impl From<ffxfs::CreateBlobError> for CreateError { |
| fn from(e: ffxfs::CreateBlobError) -> Self { |
| match e { |
| ffxfs::CreateBlobError::AlreadyExists => CreateError::AlreadyExists, |
| e @ ffxfs::CreateBlobError::Internal => CreateError::BlobCreator(e), |
| } |
| } |
| } |
| |
| /// A builder for [`Client`] |
| #[derive(Default)] |
| pub struct ClientBuilder { |
| use_reader: Reader, |
| use_creator: bool, |
| readable: bool, |
| writable: bool, |
| executable: bool, |
| } |
| |
| #[derive(Default)] |
| enum Reader { |
| #[default] |
| DontUse, |
| Use { |
| use_vmex: bool, |
| }, |
| } |
| |
| impl ClientBuilder { |
| /// Opens the /blob directory in the component's namespace with readable, writable, and/or |
| /// executable flags. Connects to the fuchsia.fxfs.BlobCreator and BlobReader if requested. |
| /// Connects to and initializes the VmexResource if `use_vmex` is set. Returns a `Client`. |
| pub async fn build(self) -> Result<Client, BlobfsError> { |
| let mut flags = fio::OpenFlags::empty(); |
| if self.readable { |
| flags |= fio::OpenFlags::RIGHT_READABLE |
| } |
| if self.writable { |
| flags |= fio::OpenFlags::RIGHT_WRITABLE |
| } |
| if self.executable { |
| flags |= fio::OpenFlags::RIGHT_EXECUTABLE |
| } |
| let dir = fuchsia_fs::directory::open_in_namespace("/blob", flags)?; |
| let reader = match self.use_reader { |
| Reader::DontUse => None, |
| Reader::Use { use_vmex } => { |
| if use_vmex { |
| if let Ok(client) = fuchsia_component::client::connect_to_protocol::< |
| fidl_fuchsia_kernel::VmexResourceMarker, |
| >() { |
| if let Ok(vmex) = client.get().await { |
| info!("Got vmex resource"); |
| vmo_blob::init_vmex_resource(vmex) |
| .map_err(BlobfsError::InitVmexResource)?; |
| } |
| } |
| } |
| Some( |
| fuchsia_component::client::connect_to_protocol::<ffxfs::BlobReaderMarker>() |
| .map_err(BlobfsError::ConnectToBlobReader)?, |
| ) |
| } |
| }; |
| |
| let creator = if self.use_creator { |
| Some( |
| fuchsia_component::client::connect_to_protocol::<ffxfs::BlobCreatorMarker>() |
| .map_err(BlobfsError::ConnectToBlobCreator)?, |
| ) |
| } else { |
| None |
| }; |
| |
| Ok(Client { dir, creator, reader }) |
| } |
| |
| /// `Client` will connect to and use fuchsia.fxfs/BlobReader for reads. Sets the VmexResource |
| /// for `Client`. The VmexResource is used by `get_backing_memory` to mark blobs as executable. |
| pub fn use_reader(self) -> Self { |
| Self { use_reader: Reader::Use { use_vmex: true }, ..self } |
| } |
| |
| /// `Client` will connect to and use fuchsia.fxfs/BlobReader for reads. Does not set the |
| /// VmexResource. |
| pub fn use_reader_no_vmex(self) -> Self { |
| Self { use_reader: Reader::Use { use_vmex: false }, ..self } |
| } |
| |
| /// If set, `Client` will connect to and use fuchsia.fxfs/BlobCreator for writes. |
| pub fn use_creator(self) -> Self { |
| Self { use_creator: true, ..self } |
| } |
| |
| /// If set, `Client` will connect to /blob in the current component's namespace with |
| /// OPEN_RIGHT_READABLE. |
| pub fn readable(self) -> Self { |
| Self { readable: true, ..self } |
| } |
| |
| /// If set, `Client` will connect to /blob in the current component's namespace with |
| /// OPEN_RIGHT_WRITABLE. WRITABLE is needed so that `delete_blob` can call |
| /// fuchsia.io/Directory.Unlink. |
| pub fn writable(self) -> Self { |
| Self { writable: true, ..self } |
| } |
| |
| /// If set, `Client` will connect to /blob in the current component's namespace with |
| /// OPEN_RIGHT_EXECUTABLE. |
| pub fn executable(self) -> Self { |
| Self { executable: true, ..self } |
| } |
| } |
| |
| impl Client { |
| /// Create an empty `ClientBuilder` |
| pub fn builder() -> ClientBuilder { |
| Default::default() |
| } |
| } |
| /// Blobfs client |
| #[derive(Debug, Clone)] |
| pub struct Client { |
| dir: fio::DirectoryProxy, |
| creator: Option<ffxfs::BlobCreatorProxy>, |
| reader: Option<ffxfs::BlobReaderProxy>, |
| } |
| |
| impl Client { |
| /// Returns a client connected to the given blob directory, BlobCreatorProxy, and |
| /// BlobReaderProxy. If `vmex` is passed in, sets the VmexResource, which is used to mark blobs |
| /// as executable. If `creator` or `reader` is not supplied, writes or reads respectively will |
| /// be performed through the blob directory. |
| pub fn new( |
| dir: fio::DirectoryProxy, |
| creator: Option<ffxfs::BlobCreatorProxy>, |
| reader: Option<ffxfs::BlobReaderProxy>, |
| vmex: Option<zx::Resource>, |
| ) -> Result<Self, anyhow::Error> { |
| if let Some(vmex) = vmex { |
| vmo_blob::init_vmex_resource(vmex)?; |
| } |
| Ok(Self { dir, creator, reader }) |
| } |
| |
| /// Creates a new client backed by the returned request stream. This constructor should not be |
| /// used outside of tests. |
| /// |
| /// # Panics |
| /// |
| /// Panics on error |
| pub fn new_test() -> (Self, fio::DirectoryRequestStream) { |
| let (dir, stream) = |
| fidl::endpoints::create_proxy_and_stream::<fio::DirectoryMarker>().unwrap(); |
| |
| (Self { dir, creator: None, reader: None }, stream) |
| } |
| |
| /// Creates a new client backed by the returned mock. This constructor should not be used |
| /// outside of tests. |
| /// |
| /// # Panics |
| /// |
| /// Panics on error |
| pub fn new_mock() -> (Self, mock::Mock) { |
| let (dir, stream) = |
| fidl::endpoints::create_proxy_and_stream::<fio::DirectoryMarker>().unwrap(); |
| |
| (Self { dir, creator: None, reader: None }, mock::Mock { stream }) |
| } |
| |
| /// Open a blob for read. `scope` will only be used if the client was configured to use |
| /// fuchsia.fxfs.BlobReader. |
| pub fn open_blob_for_read( |
| &self, |
| blob: &Hash, |
| flags: fio::OpenFlags, |
| scope: ExecutionScope, |
| server_end: ServerEnd<fio::NodeMarker>, |
| ) -> Result<(), fidl::Error> { |
| let describe = flags.contains(fio::OpenFlags::DESCRIBE); |
| // Reject requests that attempt to open blobs as writable. |
| if flags.contains(fio::OpenFlags::RIGHT_WRITABLE) { |
| send_on_open_with_error(describe, server_end, zx::Status::ACCESS_DENIED); |
| return Ok(()); |
| } |
| // Reject requests that attempt to create new blobs. |
| if flags.intersects(fio::OpenFlags::CREATE | fio::OpenFlags::CREATE_IF_ABSENT) { |
| send_on_open_with_error(describe, server_end, zx::Status::NOT_SUPPORTED); |
| return Ok(()); |
| } |
| // Use blob reader protocol if available, otherwise fallback to fuchsia.io/Directory.Open. |
| return if let Some(reader) = &self.reader { |
| let object_request = flags.to_object_request(server_end); |
| let () = open_blob_with_reader(reader.clone(), *blob, scope, flags, object_request); |
| Ok(()) |
| } else { |
| self.dir.open(flags, fio::ModeType::empty(), &blob.to_string(), server_end) |
| }; |
| } |
| |
| /// Open a blob for read using open2. `scope` will only be used if the client was configured to |
| /// use fuchsia.fxfs.BlobReader. |
| pub fn open2_blob_for_read( |
| &self, |
| blob: &Hash, |
| protocols: fio::ConnectionProtocols, |
| scope: ExecutionScope, |
| object_request: ObjectRequestRef<'_>, |
| ) -> Result<(), zx::Status> { |
| // Reject requests that attempt to open blobs as writable. |
| if protocols.rights().is_some_and(|rights| rights.contains(fio::Operations::WRITE_BYTES)) { |
| return Err(zx::Status::ACCESS_DENIED); |
| } |
| // Reject requests that attempt to create new blobs. |
| if protocols.creation_mode() != vfs::CreationMode::Never { |
| return Err(zx::Status::NOT_SUPPORTED); |
| } |
| // Errors below will be communicated via the `object_request` channel. |
| let object_request = object_request.take(); |
| // Use blob reader protocol if available, otherwise fallback to fuchsia.io/Directory.Open2. |
| if let Some(reader) = &self.reader { |
| let () = open_blob_with_reader(reader.clone(), *blob, scope, protocols, object_request); |
| } else { |
| let _: Result<(), ()> = self |
| .dir |
| .open2(&blob.to_string(), &protocols, object_request.into_channel()) |
| .map_err(|fidl_error| warn!("Failed to open blob {:?}: {:?}", blob, fidl_error)); |
| } |
| Ok(()) |
| } |
| |
| /// Returns the list of known blobs in blobfs. |
| pub async fn list_known_blobs(&self) -> Result<HashSet<Hash>, BlobfsError> { |
| // fuchsia.io.Directory.ReadDirents uses a per-connection index into the array of |
| // directory entries. To prevent contention over this index by concurrent calls (either |
| // from concurrent calls to list_known_blobs on this object, or on clones of this object, |
| // or other clones of the DirectoryProxy this object was made from), create a new |
| // connection which will have its own index. |
| let private_connection = fuchsia_fs::directory::clone_no_describe(&self.dir, None)?; |
| fuchsia_fs::directory::readdir(&private_connection) |
| .await |
| .map_err(BlobfsError::ReadDir)? |
| .into_iter() |
| .filter(|entry| entry.kind == fuchsia_fs::directory::DirentKind::File) |
| .map(|entry| entry.name.parse().map_err(BlobfsError::ParseHash)) |
| .collect() |
| } |
| |
| /// Delete the blob with the given merkle hash. |
| pub async fn delete_blob(&self, blob: &Hash) -> Result<(), BlobfsError> { |
| self.dir |
| .unlink(&blob.to_string(), &fio::UnlinkOptions::default()) |
| .await? |
| .map_err(|s| BlobfsError::Unlink(Status::from_raw(s))) |
| } |
| |
| /// Open a new blob for write. |
| pub async fn open_blob_for_write(&self, blob: &Hash) -> Result<fpkg::BlobWriter, CreateError> { |
| Ok(if let Some(blob_creator) = &self.creator { |
| fpkg::BlobWriter::Writer(blob_creator.create(blob, false).await??) |
| } else { |
| fpkg::BlobWriter::File( |
| self.open_blob_proxy_from_dir_for_write(blob) |
| .await? |
| .into_channel() |
| .map_err(|_: fio::FileProxy| CreateError::ConvertToClientEnd)? |
| .into_zx_channel() |
| .into(), |
| ) |
| }) |
| } |
| |
| /// Open a new blob for write, unconditionally using the blob directory. |
| async fn open_blob_proxy_from_dir_for_write( |
| &self, |
| blob: &Hash, |
| ) -> Result<fio::FileProxy, CreateError> { |
| let flags = fio::OpenFlags::CREATE |
| | fio::OpenFlags::RIGHT_WRITABLE |
| | fio::OpenFlags::RIGHT_READABLE; |
| |
| let path = delivery_blob::delivery_blob_path(blob); |
| fuchsia_fs::directory::open_file(&self.dir, &path, flags).await.map_err(|e| match e { |
| fuchsia_fs::node::OpenError::OpenError(Status::ACCESS_DENIED) => { |
| CreateError::AlreadyExists |
| } |
| other => CreateError::Io(other), |
| }) |
| } |
| |
| /// Returns whether blobfs has a blob with the given hash. |
| pub async fn has_blob(&self, blob: &Hash) -> bool { |
| if let Some(reader) = &self.reader { |
| // TODO(https://fxbug.dev/295552228): Use faster API for determining blob presence. |
| matches!(reader.get_vmo(blob).await, Ok(Ok(_))) |
| } else { |
| let file = match fuchsia_fs::directory::open_file_no_describe( |
| &self.dir, |
| &blob.to_string(), |
| fio::OpenFlags::DESCRIBE | fio::OpenFlags::RIGHT_READABLE, |
| ) { |
| Ok(file) => file, |
| Err(_) => return false, |
| }; |
| |
| let mut events = file.take_event_stream(); |
| |
| let event = match events.next().await { |
| None => return false, |
| Some(event) => match event { |
| Err(_) => return false, |
| Ok(event) => match event { |
| fio::FileEvent::OnOpen_ { s, info } => { |
| if Status::from_raw(s) != Status::OK { |
| return false; |
| } |
| |
| match info { |
| Some(info) => match *info { |
| fio::NodeInfoDeprecated::File(fio::FileObject { |
| event: Some(event), |
| stream: _, // TODO(https://fxbug.dev/293606235): Use stream |
| }) => event, |
| _ => return false, |
| }, |
| _ => return false, |
| } |
| } |
| fio::FileEvent::OnRepresentation { payload } => match payload { |
| fio::Representation::File(fio::FileInfo { |
| observer: Some(event), |
| stream: _, // TODO(https://fxbug.dev/293606235): Use stream |
| .. |
| }) => event, |
| _ => return false, |
| }, |
| }, |
| }, |
| }; |
| |
| // Check that the USER_0 signal has been asserted on the file's event to make sure we |
| // return false on the edge case of the blob is current being written. |
| match event.wait_handle(zx::Signals::USER_0, zx::Time::INFINITE_PAST) { |
| Ok(_) => true, |
| Err(status) => { |
| if status != Status::TIMED_OUT { |
| warn!("blobfs: unknown error asserting blob existence: {}", status); |
| } |
| false |
| } |
| } |
| } |
| } |
| |
| /// Determines which of candidate blobs exist and are readable in blobfs, returning the |
| /// set difference of candidates and readable. |
| /// If provided, `all_known` should be a superset of all readable blobs in blobfs, i.e. |
| /// if a blob is readable it must be in `all_known`, but non-readable blobs may also be |
| /// included. |
| /// `all_known` is used to skip the expensive per-blob readable check for blobs that we are |
| /// sure are missing. |
| pub async fn filter_to_missing_blobs( |
| &self, |
| candidates: &HashSet<Hash>, |
| all_known: Option<&HashSet<Hash>>, |
| ) -> HashSet<Hash> { |
| // This heuristic was taken from pkgfs. We are not sure how useful it is or why it was |
| // added, however it is kept in out of an abundance of caution. We *suspect* the heuristic |
| // is a performance optimization. Without the heuristic, we would always have to open every |
| // candidate blob and see if it's readable, which may be expensive if there are many blobs. |
| // |
| // Note that if there are less than 20 blobs, we don't use the heuristic. This is because we |
| // assume there is a certain threshold of number of blobs in a package where it is faster to |
| // first do a readdir on blobfs to help rule out some blobs without having to open them. We |
| // assume this threshold is 20. The optimal threshold is likely different between pkg-cache |
| // and pkgfs, and, especially since this checks multiple blobs concurrently, we may not be |
| // getting any benefits from the heuristic anymore. |
| // |
| // If you wish to remove this heuristic or change the threshold, consider doing a trace on |
| // packages with varying numbers of blobs present/missing. |
| // TODO(https://fxbug.dev/42157763) re-evaluate filter_to_missing_blobs heuristic. |
| let all_known_storage; |
| let all_known = if let Some(all_known) = all_known { |
| Some(all_known) |
| } else { |
| if candidates.len() > 20 { |
| if let Some(all_known) = self.list_known_blobs().await.ok() { |
| all_known_storage = all_known; |
| Some(&all_known_storage) |
| } else { |
| None |
| } |
| } else { |
| None |
| } |
| }; |
| |
| stream::iter(candidates.clone()) |
| .map(move |blob| { |
| async move { |
| // We still need to check `has_blob()` even if the blob is in `all_known`, |
| // because it might not have been fully written yet. |
| if all_known.map(|blobs| blobs.contains(&blob)) == Some(false) |
| || !self.has_blob(&blob).await |
| { |
| Some(blob) |
| } else { |
| None |
| } |
| } |
| }) |
| .buffer_unordered(50) |
| .filter_map(|blob| async move { blob }) |
| .collect() |
| .await |
| } |
| |
| /// Call fuchsia.io/Node.Sync on the blobfs directory. |
| pub async fn sync(&self) -> Result<(), BlobfsError> { |
| self.dir.sync().await?.map_err(zx::Status::from_raw).map_err(BlobfsError::Sync) |
| } |
| } |
| |
| /// Spawns a task on `scope` to attempt opening `blob` via `reader`. Creates a file connection to |
| /// the blob using [`vmo_blob::VmoBlob`]. Errors will be sent via `object_request` asynchronously. |
| fn open_blob_with_reader<P: ProtocolsExt + Send>( |
| reader: ffxfs::BlobReaderProxy, |
| blob_hash: Hash, |
| scope: ExecutionScope, |
| protocols: P, |
| object_request: ObjectRequest, |
| ) { |
| object_request.spawn(&scope.clone(), move |object_request| { |
| Box::pin(async move { |
| let get_vmo_result = reader.get_vmo(&blob_hash.into()).await.map_err(|fidl_error| { |
| if let fidl::Error::ClientChannelClosed { status, .. } = fidl_error { |
| error!("Blob reader channel closed: {:?}", status); |
| status |
| } else { |
| error!("Transport error on get_vmo: {:?}", fidl_error); |
| zx::Status::INTERNAL |
| } |
| })?; |
| let vmo = get_vmo_result.map_err(zx::Status::from_raw)?; |
| let vmo_blob = vmo_blob::VmoBlob::new(vmo); |
| object_request.create_connection(scope, vmo_blob, protocols, StreamIoConnection::create) |
| }) |
| }); |
| } |
| |
| #[cfg(test)] |
| impl Client { |
| /// Constructs a new [`Client`] connected to the provided [`BlobfsRamdisk`]. Tests in this |
| /// crate should use this constructor rather than [`BlobfsRamdisk::client`], which returns |
| /// the non-cfg(test) build of this crate's [`blobfs::Client`]. While tests could use the |
| /// [`blobfs::Client`] returned by [`BlobfsRamdisk::client`], it will be a different type than |
| /// [`super::Client`], and the tests could not access its private members or any cfg(test) |
| /// specific functionality. |
| /// |
| /// # Panics |
| /// |
| /// Panics on error. |
| pub fn for_ramdisk(blobfs: &blobfs_ramdisk::BlobfsRamdisk) -> Self { |
| Self::new( |
| blobfs.root_dir_proxy().unwrap(), |
| blobfs.blob_creator_proxy().unwrap(), |
| blobfs.blob_reader_proxy().unwrap(), |
| None, |
| ) |
| .unwrap() |
| } |
| } |
| |
| #[cfg(test)] |
| #[allow(clippy::bool_assert_comparison)] |
| mod tests { |
| use { |
| super::*, assert_matches::assert_matches, blobfs_ramdisk::BlobfsRamdisk, |
| fuchsia_async as fasync, futures::stream::TryStreamExt, maplit::hashset, |
| std::io::Write as _, std::sync::Arc, |
| }; |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn list_known_blobs_empty() { |
| let blobfs = BlobfsRamdisk::start().await.unwrap(); |
| let client = Client::for_ramdisk(&blobfs); |
| |
| assert_eq!(client.list_known_blobs().await.unwrap(), HashSet::new()); |
| blobfs.stop().await.unwrap(); |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn list_known_blobs() { |
| let blobfs = BlobfsRamdisk::builder() |
| .with_blob(&b"blob 1"[..]) |
| .with_blob(&b"blob 2"[..]) |
| .start() |
| .await |
| .unwrap(); |
| let client = Client::for_ramdisk(&blobfs); |
| |
| let expected = blobfs.list_blobs().unwrap().into_iter().collect(); |
| assert_eq!(client.list_known_blobs().await.unwrap(), expected); |
| blobfs.stop().await.unwrap(); |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn delete_blob_and_then_list() { |
| let blobfs = BlobfsRamdisk::builder() |
| .with_blob(&b"blob 1"[..]) |
| .with_blob(&b"blob 2"[..]) |
| .start() |
| .await |
| .unwrap(); |
| let client = Client::for_ramdisk(&blobfs); |
| |
| let merkle = fuchsia_merkle::from_slice(&b"blob 1"[..]).root(); |
| assert_matches!(client.delete_blob(&merkle).await, Ok(())); |
| |
| let expected = hashset! {fuchsia_merkle::from_slice(&b"blob 2"[..]).root()}; |
| assert_eq!(client.list_known_blobs().await.unwrap(), expected); |
| blobfs.stop().await.unwrap(); |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn delete_non_existing_blob() { |
| let blobfs = BlobfsRamdisk::start().await.unwrap(); |
| let client = Client::for_ramdisk(&blobfs); |
| let blob_merkle = Hash::from([1; 32]); |
| |
| assert_matches!( |
| client.delete_blob(&blob_merkle).await, |
| Err(BlobfsError::Unlink(Status::NOT_FOUND)) |
| ); |
| blobfs.stop().await.unwrap(); |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn delete_blob_mock() { |
| let (client, mut stream) = Client::new_test(); |
| let blob_merkle = Hash::from([1; 32]); |
| fasync::Task::spawn(async move { |
| match stream.try_next().await.unwrap().unwrap() { |
| fio::DirectoryRequest::Unlink { name, responder, .. } => { |
| assert_eq!(name, blob_merkle.to_string()); |
| responder.send(Ok(())).unwrap(); |
| } |
| other => panic!("unexpected request: {other:?}"), |
| } |
| }) |
| .detach(); |
| |
| assert_matches!(client.delete_blob(&blob_merkle).await, Ok(())); |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn has_blob() { |
| let blobfs = BlobfsRamdisk::builder().with_blob(&b"blob 1"[..]).start().await.unwrap(); |
| let client = Client::for_ramdisk(&blobfs); |
| |
| assert_eq!(client.has_blob(&fuchsia_merkle::from_slice(&b"blob 1"[..]).root()).await, true); |
| assert_eq!(client.has_blob(&Hash::from([1; 32])).await, false); |
| |
| blobfs.stop().await.unwrap(); |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn has_blob_fxblob() { |
| let blobfs = |
| BlobfsRamdisk::builder().fxblob().with_blob(&b"blob 1"[..]).start().await.unwrap(); |
| let client = Client::for_ramdisk(&blobfs); |
| |
| assert!(client.has_blob(&fuchsia_merkle::from_slice(&b"blob 1"[..]).root()).await); |
| assert!(!client.has_blob(&Hash::from([1; 32])).await); |
| |
| blobfs.stop().await.unwrap(); |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn has_blob_return_false_if_blob_is_partially_written() { |
| let blobfs = BlobfsRamdisk::start().await.unwrap(); |
| let client = Client::for_ramdisk(&blobfs); |
| |
| let blob = [3; 1024]; |
| let hash = fuchsia_merkle::from_slice(&blob).root(); |
| |
| let mut file = blobfs.root_dir().unwrap().write_file(hash.to_string(), 0o777).unwrap(); |
| assert_eq!(client.has_blob(&hash).await, false); |
| file.set_len(blob.len() as u64).unwrap(); |
| assert_eq!(client.has_blob(&hash).await, false); |
| file.write_all(&blob[..512]).unwrap(); |
| assert_eq!(client.has_blob(&hash).await, false); |
| file.write_all(&blob[512..]).unwrap(); |
| assert_eq!(client.has_blob(&hash).await, true); |
| |
| blobfs.stop().await.unwrap(); |
| } |
| |
| async fn resize(blob: &fio::FileProxy, size: usize) { |
| let () = blob.resize(size as u64).await.unwrap().map_err(Status::from_raw).unwrap(); |
| } |
| |
| async fn write(blob: &fio::FileProxy, bytes: &[u8]) { |
| assert_eq!( |
| blob.write(bytes).await.unwrap().map_err(Status::from_raw).unwrap(), |
| bytes.len() as u64 |
| ); |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn write_delivery_blob() { |
| let blobfs = BlobfsRamdisk::start().await.unwrap(); |
| let client = Client::for_ramdisk(&blobfs); |
| |
| let content = [3; 1024]; |
| let hash = fuchsia_merkle::from_slice(&content).root(); |
| let delivery_content = |
| delivery_blob::Type1Blob::generate(&content, delivery_blob::CompressionMode::Always); |
| |
| let proxy = client.open_blob_proxy_from_dir_for_write(&hash).await.unwrap(); |
| |
| let () = resize(&proxy, delivery_content.len()).await; |
| let () = write(&proxy, &delivery_content).await; |
| |
| assert!(client.has_blob(&hash).await); |
| |
| blobfs.stop().await.unwrap(); |
| } |
| |
| /// Wrapper for a blob and its hash. This lets the tests retain ownership of the Blob, |
| /// which is important because it ensures blobfs will not close partially written blobs for the |
| /// duration of the test. |
| struct TestBlob { |
| _blob: fio::FileProxy, |
| hash: Hash, |
| } |
| |
| async fn open_blob_only(client: &Client, content: &[u8]) -> TestBlob { |
| let hash = fuchsia_merkle::from_slice(content).root(); |
| let _blob = client.open_blob_proxy_from_dir_for_write(&hash).await.unwrap(); |
| TestBlob { _blob, hash } |
| } |
| |
| async fn open_and_truncate_blob(client: &Client, content: &[u8]) -> TestBlob { |
| let hash = fuchsia_merkle::from_slice(content).root(); |
| let _blob = client.open_blob_proxy_from_dir_for_write(&hash).await.unwrap(); |
| let () = resize(&_blob, content.len()).await; |
| TestBlob { _blob, hash } |
| } |
| |
| async fn partially_write_blob(client: &Client, content: &[u8]) -> TestBlob { |
| let hash = fuchsia_merkle::from_slice(content).root(); |
| let _blob = client.open_blob_proxy_from_dir_for_write(&hash).await.unwrap(); |
| let content = delivery_blob::generate(delivery_blob::DeliveryBlobType::Type1, content); |
| let () = resize(&_blob, content.len()).await; |
| let () = write(&_blob, &content[..content.len() / 2]).await; |
| TestBlob { _blob, hash } |
| } |
| |
| async fn fully_write_blob(client: &Client, content: &[u8]) -> TestBlob { |
| let hash = fuchsia_merkle::from_slice(content).root(); |
| let _blob = client.open_blob_proxy_from_dir_for_write(&hash).await.unwrap(); |
| let content = delivery_blob::generate(delivery_blob::DeliveryBlobType::Type1, content); |
| let () = resize(&_blob, content.len()).await; |
| let () = write(&_blob, &content).await; |
| TestBlob { _blob, hash } |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn filter_to_missing_blobs_without_heuristic() { |
| let blobfs = BlobfsRamdisk::builder().start().await.unwrap(); |
| let client = Client::for_ramdisk(&blobfs); |
| |
| let missing_hash0 = Hash::from([0; 32]); |
| let missing_hash1 = Hash::from([1; 32]); |
| |
| let present_blob0 = fully_write_blob(&client, &[2; 1024]).await; |
| let present_blob1 = fully_write_blob(&client, &[3; 1024]).await; |
| |
| assert_eq!( |
| client |
| .filter_to_missing_blobs( |
| // Pass in <= 20 candidates so the heuristic is not used. |
| &hashset! { missing_hash0, missing_hash1, |
| present_blob0.hash, present_blob1.hash |
| }, |
| None |
| ) |
| .await, |
| hashset! { missing_hash0, missing_hash1 } |
| ); |
| |
| blobfs.stop().await.unwrap(); |
| } |
| |
| /// Similar to the above test, except also test that partially written blobs count as missing. |
| #[fasync::run_singlethreaded(test)] |
| async fn filter_to_missing_blobs_without_heuristic_and_with_partially_written_blobs() { |
| let blobfs = BlobfsRamdisk::builder().start().await.unwrap(); |
| let client = Client::for_ramdisk(&blobfs); |
| |
| // Some blobs are created (but not yet truncated). |
| let missing_blob0 = open_blob_only(&client, &[0; 1024]).await; |
| |
| // Some are truncated but not written. |
| let missing_blob1 = open_and_truncate_blob(&client, &[1; 1024]).await; |
| |
| // Some are partially written. |
| let missing_blob2 = partially_write_blob(&client, &[2; 1024]).await; |
| |
| // Some are fully written. |
| let present_blob = fully_write_blob(&client, &[3; 1024]).await; |
| |
| assert_eq!( |
| client |
| .filter_to_missing_blobs( |
| &hashset! { |
| missing_blob0.hash, |
| missing_blob1.hash, |
| missing_blob2.hash, |
| present_blob.hash |
| }, |
| None |
| ) |
| .await, |
| // All partially written blobs should count as missing. |
| hashset! { missing_blob0.hash, missing_blob1.hash, missing_blob2.hash } |
| ); |
| |
| blobfs.stop().await.unwrap(); |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn filter_to_missing_blobs_with_heuristic() { |
| let blobfs = BlobfsRamdisk::builder().start().await.unwrap(); |
| let client = Client::for_ramdisk(&blobfs); |
| |
| let missing_hash0 = Hash::from([0; 32]); |
| let missing_hash1 = Hash::from([1; 32]); |
| let missing_hash2 = Hash::from([2; 32]); |
| let missing_hash3 = Hash::from([3; 32]); |
| let missing_hash4 = Hash::from([4; 32]); |
| let missing_hash5 = Hash::from([5; 32]); |
| let missing_hash6 = Hash::from([6; 32]); |
| let missing_hash7 = Hash::from([7; 32]); |
| let missing_hash8 = Hash::from([8; 32]); |
| let missing_hash9 = Hash::from([9; 32]); |
| let missing_hash10 = Hash::from([10; 32]); |
| |
| let present_blob0 = fully_write_blob(&client, &[20; 1024]).await; |
| let present_blob1 = fully_write_blob(&client, &[21; 1024]).await; |
| let present_blob2 = fully_write_blob(&client, &[22; 1024]).await; |
| let present_blob3 = fully_write_blob(&client, &[23; 1024]).await; |
| let present_blob4 = fully_write_blob(&client, &[24; 1024]).await; |
| let present_blob5 = fully_write_blob(&client, &[25; 1024]).await; |
| let present_blob6 = fully_write_blob(&client, &[26; 1024]).await; |
| let present_blob7 = fully_write_blob(&client, &[27; 1024]).await; |
| let present_blob8 = fully_write_blob(&client, &[28; 1024]).await; |
| let present_blob9 = fully_write_blob(&client, &[29; 1024]).await; |
| let present_blob10 = fully_write_blob(&client, &[30; 1024]).await; |
| |
| assert_eq!( |
| client |
| .filter_to_missing_blobs( |
| // Pass in over 20 candidates to trigger the heuristic. |
| &hashset! { missing_hash0, missing_hash1, missing_hash2, missing_hash3, |
| missing_hash4, missing_hash5, missing_hash6, missing_hash7, missing_hash8, |
| missing_hash9, missing_hash10, present_blob0.hash, present_blob1.hash, |
| present_blob2.hash, present_blob3.hash, present_blob4.hash, |
| present_blob5.hash, present_blob6.hash, present_blob7.hash, |
| present_blob8.hash, present_blob9.hash, present_blob10.hash |
| }, |
| None |
| ) |
| .await, |
| hashset! { missing_hash0, missing_hash1, missing_hash2, missing_hash3, |
| missing_hash4, missing_hash5, missing_hash6, missing_hash7, missing_hash8, |
| missing_hash9, missing_hash10 |
| } |
| ); |
| |
| blobfs.stop().await.unwrap(); |
| } |
| |
| /// Similar to the above test, except also test that partially written blobs count as missing. |
| #[fasync::run_singlethreaded(test)] |
| async fn filter_to_missing_blobs_with_heuristic_and_with_partially_written_blobs() { |
| let blobfs = BlobfsRamdisk::builder().start().await.unwrap(); |
| let client = Client::for_ramdisk(&blobfs); |
| |
| // Some blobs are created (but not yet truncated). |
| let missing_blob0 = open_blob_only(&client, &[0; 1024]).await; |
| let missing_blob1 = open_blob_only(&client, &[1; 1024]).await; |
| let missing_blob2 = open_blob_only(&client, &[2; 1024]).await; |
| |
| // Some are truncated but not written. |
| let missing_blob3 = open_and_truncate_blob(&client, &[3; 1024]).await; |
| let missing_blob4 = open_and_truncate_blob(&client, &[4; 1024]).await; |
| let missing_blob5 = open_and_truncate_blob(&client, &[5; 1024]).await; |
| |
| // Some are partially written. |
| let missing_blob6 = partially_write_blob(&client, &[6; 1024]).await; |
| let missing_blob7 = partially_write_blob(&client, &[7; 1024]).await; |
| let missing_blob8 = partially_write_blob(&client, &[8; 1024]).await; |
| |
| // Some aren't even open. |
| let missing_hash9 = Hash::from([9; 32]); |
| let missing_hash10 = Hash::from([10; 32]); |
| |
| let present_blob0 = fully_write_blob(&client, &[20; 1024]).await; |
| let present_blob1 = fully_write_blob(&client, &[21; 1024]).await; |
| let present_blob2 = fully_write_blob(&client, &[22; 1024]).await; |
| let present_blob3 = fully_write_blob(&client, &[23; 1024]).await; |
| let present_blob4 = fully_write_blob(&client, &[24; 1024]).await; |
| let present_blob5 = fully_write_blob(&client, &[25; 1024]).await; |
| let present_blob6 = fully_write_blob(&client, &[26; 1024]).await; |
| let present_blob7 = fully_write_blob(&client, &[27; 1024]).await; |
| let present_blob8 = fully_write_blob(&client, &[28; 1024]).await; |
| let present_blob9 = fully_write_blob(&client, &[29; 1024]).await; |
| let present_blob10 = fully_write_blob(&client, &[30; 1024]).await; |
| |
| assert_eq!( |
| client |
| .filter_to_missing_blobs( |
| &hashset! { missing_blob0.hash, missing_blob1.hash, missing_blob2.hash, |
| missing_blob3.hash, missing_blob4.hash, missing_blob5.hash, |
| missing_blob6.hash, missing_blob7.hash, missing_blob8.hash, |
| missing_hash9, missing_hash10, present_blob0.hash, |
| present_blob1.hash, present_blob2.hash, present_blob3.hash, |
| present_blob4.hash, present_blob5.hash, present_blob6.hash, |
| present_blob7.hash, present_blob8.hash, present_blob9.hash, |
| present_blob10.hash |
| }, |
| None |
| ) |
| .await, |
| // All partially written blobs should count as missing. |
| hashset! { missing_blob0.hash, missing_blob1.hash, missing_blob2.hash, |
| missing_blob3.hash, missing_blob4.hash, missing_blob5.hash, missing_blob6.hash, |
| missing_blob7.hash, missing_blob8.hash, missing_hash9, missing_hash10 |
| } |
| ); |
| |
| blobfs.stop().await.unwrap(); |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn sync() { |
| let counter = Arc::new(std::sync::atomic::AtomicU32::new(0)); |
| let counter_clone = Arc::clone(&counter); |
| let (client, mut stream) = Client::new_test(); |
| fasync::Task::spawn(async move { |
| match stream.try_next().await.unwrap().unwrap() { |
| fio::DirectoryRequest::Sync { responder } => { |
| counter_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst); |
| responder.send(Ok(())).unwrap(); |
| } |
| other => panic!("unexpected request: {other:?}"), |
| } |
| }) |
| .detach(); |
| |
| assert_matches!(client.sync().await, Ok(())); |
| assert_eq!(counter.load(std::sync::atomic::Ordering::SeqCst), 1); |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn open_blob_for_write_uses_fxblob_if_configured() { |
| let (blob_creator, mut blob_creator_stream) = |
| fidl::endpoints::create_proxy_and_stream::<ffxfs::BlobCreatorMarker>().unwrap(); |
| let (blob_reader, _) = fidl::endpoints::create_proxy::<ffxfs::BlobReaderMarker>().unwrap(); |
| let client = Client::new( |
| fidl::endpoints::create_proxy::<fio::DirectoryMarker>().unwrap().0, |
| Some(blob_creator), |
| Some(blob_reader), |
| None, |
| ) |
| .unwrap(); |
| |
| fuchsia_async::Task::spawn(async move { |
| match blob_creator_stream.next().await.unwrap().unwrap() { |
| ffxfs::BlobCreatorRequest::Create { hash, allow_existing, responder } => { |
| assert_eq!(hash, [0; 32]); |
| assert!(!allow_existing); |
| let () = responder.send(Ok(fidl::endpoints::create_endpoints().0)).unwrap(); |
| } |
| } |
| }) |
| .detach(); |
| |
| assert_matches!( |
| client.open_blob_for_write(&[0; 32].into()).await, |
| Ok(fpkg::BlobWriter::Writer(_)) |
| ); |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn open_blob_for_write_fxblob_maps_already_exists() { |
| let (blob_creator, mut blob_creator_stream) = |
| fidl::endpoints::create_proxy_and_stream::<ffxfs::BlobCreatorMarker>().unwrap(); |
| let (blob_reader, _) = fidl::endpoints::create_proxy::<ffxfs::BlobReaderMarker>().unwrap(); |
| |
| let client = Client::new( |
| fidl::endpoints::create_proxy::<fio::DirectoryMarker>().unwrap().0, |
| Some(blob_creator), |
| Some(blob_reader), |
| None, |
| ) |
| .unwrap(); |
| |
| fuchsia_async::Task::spawn(async move { |
| match blob_creator_stream.next().await.unwrap().unwrap() { |
| ffxfs::BlobCreatorRequest::Create { hash, allow_existing, responder } => { |
| assert_eq!(hash, [0; 32]); |
| assert!(!allow_existing); |
| let () = responder.send(Err(ffxfs::CreateBlobError::AlreadyExists)).unwrap(); |
| } |
| } |
| }) |
| .detach(); |
| |
| assert_matches!( |
| client.open_blob_for_write(&[0; 32].into()).await, |
| Err(CreateError::AlreadyExists) |
| ); |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn concurrent_list_known_blobs_all_return_full_contents() { |
| use futures::StreamExt; |
| let blobfs = BlobfsRamdisk::builder().start().await.unwrap(); |
| let client = Client::for_ramdisk(&blobfs); |
| |
| // ReadDirents returns an 8,192 byte buffer, and each entry is 74 bytes [0] (including 64 |
| // bytes of filename), so use more than 110 entries to guarantee that listing all contents |
| // requires multiple ReadDirents calls. This isn't necessary to cause conflict, because |
| // each successful listing requires a call to Rewind as well, but it does make conflict |
| // more likely. |
| // [0] https://cs.opensource.google/fuchsia/fuchsia/+/main:sdk/fidl/fuchsia.io/directory.fidl;l=261;drc=9e84e19d3f42240c46d2b0c3c132c2f0b5a3343f |
| for i in 0..256u16 { |
| let _: TestBlob = fully_write_blob(&client, i.to_le_bytes().as_slice()).await; |
| } |
| |
| let () = futures::stream::iter(0..100) |
| .for_each_concurrent(None, |_| async { |
| assert_eq!(client.list_known_blobs().await.unwrap().len(), 256); |
| }) |
| .await; |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn filter_to_missing_uses_provided_all_known() { |
| let blobfs = BlobfsRamdisk::builder().start().await.unwrap(); |
| let client = Client::for_ramdisk(&blobfs); |
| let present_blob = fully_write_blob(&client, &[0; 1024]).await; |
| |
| // Even though actually present, the written blob will be reported as missing because the |
| // provided `all_known` is empty. |
| assert_eq!( |
| client |
| .filter_to_missing_blobs(&HashSet::from([present_blob.hash]), Some(&HashSet::new())) |
| .await, |
| HashSet::from([present_blob.hash]) |
| ); |
| } |
| } |
