add read method to interface

Author: seanaye

src/store/fs.rs

@@ -450,6 +450,11 @@ pub trait Persistence: Clone {
     /// return the size of the file in bytes if it can be found/read
     /// otherwise return a [Self::Err]
     fn size(&self, path: &Path) -> impl Future<Output = Result<u64, Self::Err>>;
+
+    /// read the contents of the file at the path
+    /// returning the bytes of the file in the success case
+    /// and [Self::Err] in the error case
+    fn read(&self, path: &Path) -> impl Future<Output = Result<Vec<u8>, Self::Err>>;
 }
 
 /// A persistence layer that writes to the local file system
@@ -463,6 +468,11 @@ impl Persistence for FileSystemPersistence {
         let res = std::fs::metadata(path).map(|m| m.len());
         async move { res }
     }
+
+    fn read(&self, path: &Path) -> impl Future<Output = Result<Vec<u8>, Self::Err>> {
+        let res = std::fs::read(path);
+        async move { res }
+    }
 }
 
 impl ImportSource {
@@ -1117,7 +1127,7 @@ where
                     // we don't know if the data will be inlined since we don't
                     // have the inline options here. But still for such a small file
                     // it does not seem worth it do to the temp file ceremony.
-                    let data = std::fs::read(&path)?;
+                    let data = Handle::current().block_on(self.fs.read(&path))?;
                     ImportSource::Memory(data.into())
                 } else {
                     let temp_path = self.temp_file_name();
@@ -1216,23 +1226,24 @@ impl<T> Drop for StoreInner<T> {
     }
 }
 
-struct ActorState {
+struct ActorState<T> {
     handles: BTreeMap<Hash, BaoFileHandleWeak>,
     protected: BTreeSet<Hash>,
     temp: Arc<RwLock<TempCounterMap>>,
     msgs_rx: async_channel::Receiver<ActorMessage>,
     create_options: Arc<BaoFileConfig>,
     options: Options,
     rt: tokio::runtime::Handle,
+    fs: T,
 }
 
 /// The actor for the redb store.
 ///
 /// It is split into the database and the rest of the state to allow for split
 /// borrows in the message handlers.
-struct Actor {
+struct Actor<T = FileSystemPersistence> {
     db: redb::Database,
-    state: ActorState,
+    state: ActorState<T>,
 }
 
 /// Error type for message handler functions of the redb actor.
@@ -1583,12 +1594,13 @@ pub(super) async fn gc_sweep_task(
     Ok(())
 }
 
-impl Actor {
-    fn new(
+impl<T> Actor<T> {
+    fn new_with_backend(
         path: &Path,
         options: Options,
         temp: Arc<RwLock<TempCounterMap>>,
         rt: tokio::runtime::Handle,
+        fs: T,
     ) -> ActorResult<(Self, async_channel::Sender<ActorMessage>)> {
         let db = match redb::Database::create(path) {
             Ok(db) => db,
@@ -1632,11 +1644,23 @@ impl Actor {
                     options,
                     create_options: Arc::new(create_options),
                     rt,
+                    fs,
                 },
             },
             tx,
         ))
     }
+}
+
+impl Actor {
+    fn new(
+        path: &Path,
+        options: Options,
+        temp: Arc<RwLock<TempCounterMap>>,
+        rt: tokio::runtime::Handle,
+    ) -> ActorResult<(Self, async_channel::Sender<ActorMessage>)> {
+        Self::new_with_backend(path, options, temp, rt, FileSystemPersistence)
+    }
 
     async fn run_batched(mut self) -> ActorResult<()> {
         let mut msgs = PeekableFlumeReceiver::new(self.state.msgs_rx.clone());
@@ -1720,7 +1744,11 @@ impl Actor {
     }
 }
 
-impl ActorState {
+impl<T> ActorState<T>
+where
+    T: Persistence,
+    ActorError: From<T::Err>,
+{
     fn entry_status(
         &mut self,
         tables: &impl ReadableTables,
@@ -1911,7 +1939,8 @@ impl ActorState {
                         "reading external data to inline it: {}",
                         external_path.display()
                     );
-                    let data = Bytes::from(std::fs::read(&external_path)?);
+                    let data =
+                        Bytes::from(Handle::current().block_on(self.fs.read(&external_path))?);
                     DataLocation::Inline(data)
                 } else {
                     DataLocation::External(vec![external_path], data_size)
@@ -2164,7 +2193,7 @@ impl ActorState {
                                 // inline
                                 if size <= self.options.inline.max_data_inlined {
                                     let path = self.options.path.owned_data_path(&hash);
-                                    let data = std::fs::read(&path)?;
+                                    let data = Handle::current().block_on(self.fs.read(&path))?;
                                     tables.delete_after_commit.insert(hash, [BaoFilePart::Data]);
                                     tables.inline_data.insert(hash, data.as_slice())?;
                                     (DataLocation::Inline(()), size, true)
@@ -2199,7 +2228,7 @@ impl ActorState {
                                 if outboard_size <= self.options.inline.max_outboard_inlined =>
                             {
                                 let path = self.options.path.owned_outboard_path(&hash);
-                                let outboard = std::fs::read(&path)?;
+                                let outboard = Handle::current().block_on(self.fs.read(&path))?;
                                 tables
                                     .delete_after_commit
                                     .insert(hash, [BaoFilePart::Outboard]);

src/store/fs/test_support.rs

@@ -150,7 +150,7 @@ pub(crate) struct EntryStateResponse {
     pub db: Option<EntryState<Vec<u8>>>,
 }
 
-impl ActorState {
+impl<T> ActorState<T> {
     pub(super) fn get_full_entry_state(
         &mut self,
         tables: &impl ReadableTables,

src/store/fs/validate.rs

@@ -12,7 +12,7 @@ use crate::{
     util::progress::BoxedProgressSender,
 };
 
-impl ActorState {
+impl<T> ActorState<T> {
     //! This performs a full consistency check. Eventually it will also validate
     //! file content again, but that part is not yet implemented.
     //!