Added better async support

# Conflicts:
#	Cargo.lock
#	common/src/archive.rs
#	common/src/header.rs
#	common/src/lib.rs
#	common/src/store.rs
#	server/Cargo.toml
#	server/src/main.rs

Author: CEbbinghaus

.editorconfig

@@ -0,0 +1,5 @@
+root=true
+
+[*.rs]
+indent_style = tab
+indent_size = 4

Cargo.lock

@@ -7,8 +7,8 @@ use common::{
         RawEntryData, HEADER,
     },
     object_body::Object as OtherObject,
-    read_header_and_body, read_header_from_file, read_header_from_slice, read_object_into_headers,
-    Hash, Header, Mode, ObjectType, BLOB_KEY, INDEX_KEY, TREE_KEY,
+    read_header_and_body, read_header_from_file, read_header_from_slice,
+    read_object_into_headers_sync, Hash, Header, Mode, ObjectType, BLOB_KEY, INDEX_KEY, TREE_KEY,
 };
 use sha2::{Digest, Sha512};
 use std::{
@@ -919,7 +919,7 @@ fn pack_archive(
 
     let mut headers: HashMap<Hash, Header> = HashMap::new();
 
-    read_object_into_headers(cache, &mut headers, &index.tree)?;
+    read_object_into_headers_sync(cache, &mut headers, &index.tree)?;
 
     //TODO: Surely there is an algorithm to more efficiently lay out this data
     let mut i = 0;

client/src/main.rs

@@ -5,6 +5,7 @@ edition = "2021"
 
 [dependencies]
 anyhow = "1.0.100"
+bytes = "1.10.1"
 chrono = "0.4.41"
 flate2 = { version = "1.1.5", features = ["zlib-rs"] }
 futures = "0.3.31"

common/Cargo.toml

@@ -7,11 +7,14 @@ use std::{
 };
 
 use anyhow::{anyhow, Error};
+use futures::AsyncReadExt;
 use sha2::{Digest, Sha512};
 
 use crate::{
     object_body::{Index, Object},
-    pipe, Hash,
+    pipe,
+    store::Store,
+    Hash,
 };
 
 pub const HEADER: [u8; 4] = [b'a', b'r', b'x', b'a'];
@@ -156,6 +159,34 @@ where
     }
 }
 
+// /// Create a new `Body` from a [`Stream`].
+// ///
+// /// [`Stream`]: https://docs.rs/futures-core/latest/futures_core/stream/trait.Stream.html
+// pub fn from_stream<S>(stream: S) -> Self
+// where
+//     S: TryStream + Send + 'static,
+//     S::Ok: Into<Bytes>,
+//     S::Error: Into<BoxError>,
+// {
+//     Self::new(StreamBody {
+//         stream: SyncWrapper::new(stream),
+//     })
+// }
+
+// impl<T> Stream for Archive<T>
+// where
+//     T: ArchiveEntryData + Unpin,
+// {
+//     type Item = Result<Bytes, Error>;
+
+//     fn poll_next(
+//         self: std::pin::Pin<&mut Self>,
+//         cx: &mut std::task::Context<'_>,
+//     ) -> std::task::Poll<Option<Self::Item>> {
+
+//     }
+// }
+
 pub struct ArchiveHeaderEntry {
     pub hash: Hash,
     pub index: u64,
@@ -173,7 +204,6 @@ impl ArchiveEntryData for RawEntryData {
         self.0
     }
 }
-
 pub struct ReaderEntryData<T>(T)
 where
     T: Read;
@@ -211,6 +241,23 @@ impl ArchiveEntryData for FileEntryData {
     }
 }
 
+pub struct StoreEntryData {
+    pub store: Store,
+    pub hash: Hash,
+}
+
+impl<'a> ArchiveEntryData for StoreEntryData {
+    fn turn_into_vec(self) -> Vec<u8> {
+        let mut object = futures::executor::block_on(self.store.get_object(&self.hash))
+            .expect("Object to be available in store");
+
+        let mut data: Vec<u8> = Vec::new();
+        futures::executor::block_on(object.read_to_end(&mut data)).expect("Reading to work");
+
+        data
+    }
+}
+
 pub struct ArchiveBody<T>
 where
     T: ArchiveEntryData,

common/src/archive.rs

@@ -6,12 +6,14 @@ use std::{
     str::from_utf8,
 };
 
+use futures::AsyncReadExt;
+
 pub use crate::constants::{BLOB_KEY, INDEX_KEY, TREE_KEY};
 pub use crate::hash::Hash;
 pub use crate::header::Header;
 pub use crate::object::Object;
-use crate::object_body::Object as ObjectTrait;
 pub use crate::primitives::{Mode, ObjectType};
+use crate::{object_body::Object as ObjectTrait, store::Store};
 
 pub mod archive;
 mod constants;
@@ -59,42 +61,89 @@ pub fn read_header_from_file(reader: &mut BufReader<File>) -> Option<Header> {
     read_header_from_slice(&vec[..vec.len() - 1])
 }
 
-pub fn read_object_into_headers(
-    cache: &PathBuf,
+pub async fn read_object_into_headers(
+    store: &Store,
     headers: &mut HashMap<Hash, Header>,
     object_hash: &Hash,
 ) -> anyhow::Result<()> {
-    let object_path = object_hash.get_path(&cache);
-    assert!(object_path.exists());
+    let mut stack = vec![object_hash.clone()];
 
-    // We assume that if our hash already exists, We probably have already collected all children
-    if headers.contains_key(object_hash) {
-        return Ok(());
-    }
+    while let Some(current_hash) = stack.pop() {
+        if headers.contains_key(&current_hash) {
+            continue;
+        }
 
-    let file = File::open(object_path).expect("file to exist");
-    let mut reader = BufReader::new(file);
-    let mut data = Vec::new();
-    let bytes_read = reader
-        .read_until(0, &mut data)
-        .expect("File to be readable");
+        let mut object = store.get_object(&current_hash).await?;
 
-    let header =
-        read_header_from_slice(&data[..bytes_read - 1]).expect("File to be correctly formatted");
-    assert!(header.object_type != ObjectType::Index);
+        if object.header.object_type == ObjectType::Index {
+            return Err(anyhow::anyhow!(
+                "Indexes cannot exist within a tree. Likely a hash collision 😳"
+            ));
+        }
+
+        headers.insert(current_hash.clone(), object.header.clone());
+
+        if object.header.object_type == ObjectType::Blob {
+            continue;
+        }
 
-    headers.insert(object_hash.clone(), header.clone());
-    if header.object_type == ObjectType::Blob {
-        return Ok(());
+        let mut data = Vec::new();
+        let bytes_read = object.read_to_end(&mut data).await?;
+
+        assert!(
+            bytes_read as u64 == object.header.size,
+            "Read size must match header size"
+        );
+
+        let tree = crate::object_body::Tree::from_data(&data);
+
+        for entry in &tree.contents {
+            stack.push(entry.hash.clone());
+        }
     }
 
-    reader.read_to_end(&mut data)?;
+    Ok(())
+}
+
+pub fn read_object_into_headers_sync(
+    cache: &PathBuf,
+    headers: &mut HashMap<Hash, Header>,
+    object_hash: &Hash,
+) -> anyhow::Result<()> {
+    let mut stack = vec![object_hash.clone()];
 
-    // println!("Reading Tree {object_hash}");
-    let tree = crate::object_body::Tree::from_data(&data[bytes_read..]);
+    while let Some(current_hash) = stack.pop() {
+        if headers.contains_key(&current_hash) {
+            continue;
+        }
+
+        let object_path = current_hash.get_path(cache);
+        let file = File::open(object_path)?;
+        let mut reader = BufReader::new(file);
+        let mut data = Vec::new();
+        let bytes_read = reader.read_until(0, &mut data)?;
 
-    for entry in &tree.contents {
-        read_object_into_headers(cache, headers, &entry.hash)?;
+        let header = read_header_from_slice(&data[..bytes_read - 1])
+            .ok_or_else(|| anyhow::anyhow!("Invalid header"))?;
+
+        if header.object_type == ObjectType::Index {
+            return Err(anyhow::anyhow!("Indexes cannot exist within a tree"));
+        }
+
+        headers.insert(current_hash.clone(), header.clone());
+
+        if header.object_type == ObjectType::Blob {
+            continue;
+        }
+
+        data.clear();
+        reader.read_to_end(&mut data)?;
+
+        let tree = crate::object_body::Tree::from_data(&data);
+
+        for entry in &tree.contents {
+            stack.push(entry.hash.clone());
+        }
     }
 
     Ok(())

common/src/lib.rs

@@ -3,10 +3,13 @@ use anyhow::Result;
 use futures::io::copy;
 use futures::AsyncReadExt;
 use futures::{AsyncBufRead, AsyncRead, AsyncWriteExt};
-use opendal::{FuturesAsyncReader, Operator};
+use opendal::{Builder, FuturesAsyncReader, Operator};
 
-pub struct StoreObject<T> {
-    header: Header,
+pub struct StoreObject<T>
+where
+    T: AsyncBufRead + AsyncRead + Unpin,
+{
+    pub header: Header,
     body: T,
 }
 
@@ -65,8 +68,8 @@ where
         cx: &mut std::task::Context<'_>,
         buf: &mut [u8],
     ) -> std::task::Poll<std::io::Result<usize>> {
-        let body = unsafe { self.map_unchecked_mut(|s| &mut s.body) };
-        body.poll_read(cx, buf)
+        let this = self.get_mut();
+        std::pin::Pin::new(&mut this.body).poll_read(cx, buf)
     }
 }
 
@@ -78,13 +81,13 @@ where
         self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
     ) -> std::task::Poll<std::io::Result<&[u8]>> {
-        let body = unsafe { self.map_unchecked_mut(|s| &mut s.body) };
-        body.poll_fill_buf(cx)
+        let this = self.get_mut();
+        std::pin::Pin::new(&mut this.body).poll_fill_buf(cx)
     }
 
     fn consume(self: std::pin::Pin<&mut Self>, amt: usize) {
-        let body = unsafe { self.map_unchecked_mut(|s| &mut s.body) };
-        body.consume(amt);
+        let this = self.get_mut();
+        std::pin::Pin::new(&mut this.body).consume(amt);
     }
 }
 
@@ -98,6 +101,10 @@ impl Store {
         Self { operator }
     }
 
+    pub fn from_builder(builder: impl Builder) -> Result<Self> {
+        Ok(Self::new(Operator::new(builder)?.finish()))
+    }
+
     pub async fn exists(&self, hash: &Hash) -> Result<bool> {
         Ok(self.operator.exists(hash.as_str()).await?)
     }

common/src/store.rs

@@ -7,7 +7,8 @@ edition = "2021"
 axum = { version = "0.8.4", features = ["macros"] }
 clap = { version = "4.5.51", features = ["derive"] }
 common = { path = "../common" }
-futures = "0.3"
+futures = "0.3.31"
+futures-core = "0.3.31"
 http-body = "1.0.1"
 lazy_static = "1.5.0"
 opendal = { version = "0.54.1", features = ["services-fs", "services-s3"] }