move store into store and add back the protocol and net_protocol

Author: rklaehn

Cargo.lock

@@ -13,7 +13,7 @@ quinn = { package = "iroh-quinn", version = "0.13.0" }
 n0-future = "0.1.2"
 range-collections = "0.4.5"
 redb = "2.4.0"
-smallvec = { version = "1", features = ["serde"] }
+smallvec = { version = "1", features = ["serde", "const_new"] }
 thiserror = "2.0.11"
 tokio = { version = "1.43.0", features = ["full"] }
 tokio-util = { version = "0.7.13", features = ["full"] }
@@ -28,6 +28,7 @@ chrono = "0.4.39"
 nested_enum_utils = "0.1.0"
 ref-cast = "1.0.24"
 arrayvec = "0.7.6"
+iroh = "0.33.0"
 
 [dev-dependencies]
 hex = "0.4.3"

Cargo.toml

@@ -5,7 +5,7 @@ use serde::{Deserialize, Serialize};
 use smallvec::SmallVec;
 
 use super::meta::{ActorError, ActorResult};
-use crate::{hash::DD, util::SliceInfoExt};
+use crate::util::SliceInfoExt;
 
 /// Location of the data.
 ///

src/fs/entry_state.rs

@@ -7,6 +7,8 @@ use bao_tree::blake3;
 use postcard::experimental::max_size::MaxSize;
 use serde::{de, Deserialize, Deserializer, Serialize, Serializer};
 
+use crate::store::util::DD;
+
 /// Hash type used throughout.
 #[derive(PartialEq, Eq, Copy, Clone, Hash)]
 pub struct Hash(blake3::Hash);
@@ -17,20 +19,6 @@ impl fmt::Debug for Hash {
     }
 }
 
-pub(crate) struct DD<T: fmt::Display>(T);
-
-impl<T: fmt::Display> From<T> for DD<T> {
-    fn from(value: T) -> Self {
-        DD(value)
-    }
-}
-
-impl<T: fmt::Display> fmt::Debug for DD<T> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        fmt::Display::fmt(&self.0, f)
-    }
-}
-
 impl Hash {
     /// The hash for the empty byte range (`b""`).
     pub const EMPTY: Hash = Hash::from_bytes([

src/hash.rs

@@ -0,0 +1,156 @@
+//! traits related to collections of blobs
+use std::{fmt::Debug, io};
+
+use bytes::Bytes;
+use iroh_io::{AsyncSliceReader, AsyncSliceReaderExt};
+
+use crate::Hash;
+
+/// A sequence of links, backed by a [`Bytes`] object.
+#[derive(Debug, Clone, derive_more::Into)]
+pub struct HashSeq(Bytes);
+
+impl FromIterator<Hash> for HashSeq {
+    fn from_iter<T: IntoIterator<Item = Hash>>(iter: T) -> Self {
+        let iter = iter.into_iter();
+        let (lower, _upper) = iter.size_hint();
+        let mut bytes = Vec::with_capacity(lower * 32);
+        for hash in iter {
+            bytes.extend_from_slice(hash.as_ref());
+        }
+        Self(bytes.into())
+    }
+}
+
+impl TryFrom<Bytes> for HashSeq {
+    type Error = anyhow::Error;
+
+    fn try_from(bytes: Bytes) -> Result<Self, Self::Error> {
+        Self::new(bytes).ok_or_else(|| anyhow::anyhow!("invalid hash sequence"))
+    }
+}
+
+impl IntoIterator for HashSeq {
+    type Item = Hash;
+    type IntoIter = HashSeqIter;
+
+    fn into_iter(self) -> Self::IntoIter {
+        HashSeqIter(self)
+    }
+}
+
+/// Stream over the hashes in a [`HashSeq`].
+///
+/// todo: make this wrap a reader instead of a [`HashSeq`].
+#[derive(Debug, Clone)]
+pub struct HashSeqStream(HashSeq);
+
+impl HashSeqStream {
+    /// Get the next hash in the sequence.
+    #[allow(clippy::should_implement_trait, clippy::unused_async)]
+    pub async fn next(&mut self) -> io::Result<Option<Hash>> {
+        Ok(self.0.pop_front())
+    }
+
+    /// Skip a number of hashes in the sequence.
+    #[allow(clippy::unused_async)]
+    pub async fn skip(&mut self, n: u64) -> io::Result<()> {
+        let ok = self.0.drop_front(n as usize);
+        if !ok {
+            Err(io::Error::new(
+                io::ErrorKind::UnexpectedEof,
+                "end of sequence",
+            ))
+        } else {
+            Ok(())
+        }
+    }
+}
+
+impl HashSeq {
+    /// Create a new sequence of hashes.
+    pub fn new(bytes: Bytes) -> Option<Self> {
+        if bytes.len() % 32 == 0 {
+            Some(Self(bytes))
+        } else {
+            None
+        }
+    }
+
+    fn drop_front(&mut self, n: usize) -> bool {
+        let start = n * 32;
+        if start > self.0.len() {
+            false
+        } else {
+            self.0 = self.0.slice(start..);
+            true
+        }
+    }
+
+    /// Iterate over the hashes in this sequence.
+    pub fn iter(&self) -> impl Iterator<Item = Hash> + '_ {
+        self.0.chunks_exact(32).map(|chunk| {
+            let hash: [u8; 32] = chunk.try_into().unwrap();
+            hash.into()
+        })
+    }
+
+    /// Get the number of hashes in this sequence.
+    pub fn len(&self) -> usize {
+        self.0.len() / 32
+    }
+
+    /// Check if this sequence is empty.
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
+
+    /// Get the hash at the given index.
+    pub fn get(&self, index: usize) -> Option<Hash> {
+        if index < self.len() {
+            let hash: [u8; 32] = self.0[index * 32..(index + 1) * 32].try_into().unwrap();
+            Some(hash.into())
+        } else {
+            None
+        }
+    }
+
+    /// Get and remove the first hash in this sequence.
+    pub fn pop_front(&mut self) -> Option<Hash> {
+        if self.is_empty() {
+            None
+        } else {
+            let hash = self.get(0).unwrap();
+            self.0 = self.0.slice(32..);
+            Some(hash)
+        }
+    }
+
+    /// Get the underlying bytes.
+    pub fn into_inner(self) -> Bytes {
+        self.0
+    }
+}
+
+/// Iterator over the hashes in a [`HashSeq`].
+#[derive(Debug, Clone)]
+pub struct HashSeqIter(HashSeq);
+
+impl Iterator for HashSeqIter {
+    type Item = Hash;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.0.pop_front()
+    }
+}
+
+/// Parse a sequence of hashes.
+pub async fn parse_hash_seq<'a, R: AsyncSliceReader + 'a>(
+    mut reader: R,
+) -> anyhow::Result<(HashSeqStream, u64)> {
+    let bytes = reader.read_to_end().await?;
+    let hashes = HashSeq::try_from(bytes)?;
+    let num_hashes = hashes.len() as u64;
+    let stream = HashSeqStream(hashes);
+    Ok((stream, num_hashes))
+}

src/hashseq.rs

@@ -1,32 +1,9 @@
-use bao_tree::BlockSize;
-mod api;
-mod bitfield;
-mod fs;
-mod hash;
-mod mem;
-mod proto;
-mod readonly_mem;
-mod test;
-mod util;
+pub mod store;
 
+mod hash;
 pub use hash::{BlobFormat, Hash, HashAndFormat};
-use ref_cast::RefCast;
-
-#[derive(Debug, Clone, ref_cast::RefCast)]
-#[repr(transparent)]
-pub struct Store {
-    sender: tokio::sync::mpsc::Sender<proto::Command>,
-}
-
-impl Store {
-    pub fn from_sender(sender: tokio::sync::mpsc::Sender<proto::Command>) -> Self {
-        Self { sender }
-    }
-
-    pub fn ref_from_sender(sender: &tokio::sync::mpsc::Sender<proto::Command>) -> &Self {
-        Store::ref_cast(sender)
-    }
-}
-
-/// Block size used by iroh, 2^4*1024 = 16KiB
-pub const IROH_BLOCK_SIZE: BlockSize = BlockSize::from_chunk_log(4);
+pub use store::IROH_BLOCK_SIZE;
+pub mod hashseq;
+pub mod net_protocol;
+pub mod protocol;
+pub mod provider;

src/lib.rs

@@ -0,0 +1,64 @@
+//! Adaptation of `iroh-blobs` as an `iroh` protocol.
+
+// TODO: reduce API surface and add documentation
+#![allow(missing_docs)]
+
+use std::{
+    collections::BTreeSet,
+    fmt::Debug,
+    ops::{Deref, DerefMut},
+    sync::Arc,
+};
+
+use anyhow::{bail, Result};
+use futures_lite::future::Boxed as BoxedFuture;
+use iroh::{endpoint::Connecting, protocol::ProtocolHandler, Endpoint, NodeAddr};
+use serde::{Deserialize, Serialize};
+use tracing::debug;
+
+use crate::{store::Store, BlobFormat, Hash};
+
+#[derive(Debug)]
+pub(crate) struct BlobsInner {
+    pub(crate) store: Store,
+    pub(crate) endpoint: Endpoint,
+}
+
+#[derive(Debug, Clone)]
+pub struct Blobs {
+    pub(crate) inner: Arc<BlobsInner>,
+}
+
+impl Blobs {
+    fn new(store: Store, endpoint: Endpoint) -> Self {
+        Self {
+            inner: Arc::new(BlobsInner { store, endpoint }),
+        }
+    }
+
+    pub fn store(&self) -> &Store {
+        &self.inner.store
+    }
+
+    pub fn endpoint(&self) -> &Endpoint {
+        &self.inner.endpoint
+    }
+}
+
+impl ProtocolHandler for Blobs {
+    fn accept(&self, conn: Connecting) -> BoxedFuture<Result<()>> {
+        let store = self.store().clone();
+
+        Box::pin(async move {
+            crate::provider::handle_connection(conn.await?, store).await;
+            Ok(())
+        })
+    }
+
+    fn shutdown(&self) -> BoxedFuture<()> {
+        let store = self.store().clone();
+        Box::pin(async move {
+            store.shutdown().await;
+        })
+    }
+}

src/net_protocol.rs

@@ -14,8 +14,7 @@ use serde::{Deserialize, Serialize};
 use tokio::sync::{mpsc, oneshot};
 
 use crate::{
-    bitfield::Bitfield,
-    hash::DD,
+    bitfield::Bitfield
     util::{observer::Observer, Tag},
     BlobFormat, HashAndFormat,
 };