Merge pull request #21 from rklaehn/persistence

feat(iroh): add persistence to a local iroh-blobs store

Author: cameronfyfe

Cargo.lock

@@ -5,13 +5,12 @@ resolver = "2"
 [workspace.dependencies]
 anyhow = "1"
 async-trait = "0.1"
-axum = "0.8"
+axum = { version = "0.8", features = ["macros"] }
 axum-extra = { version = "0.10", features = ["typed-header"] }
 azure_core = "0.19"
 azure_identity = "0.19"
 azure_storage = "0.19"
 azure_storage_blobs = "0.19"
-bao-tree = "0.16"
 blake3 = "1.5"
 bytes = "1.10.1"
 chrono = "0.4"
@@ -28,7 +27,6 @@ http-body-util = "0.1.3"
 iroh = "0.95"
 iroh-base = "0.95"
 iroh-blobs = "0.97"
-iroh-tickets = "0.2"
 itertools = "0.12"
 jsonwebtoken = "9"
 log = "0.4"
@@ -53,3 +51,4 @@ utoipa = { version = "5", features = ["axum_extras"] }
 utoipa-axum = "0.2"
 utoipa-swagger-ui = { version = "9", features = ["axum"] }
 uuid = { version = "1.18.1", features = ["serde", "v4"] }
+irpc = "0.11.0"

Cargo.toml

@@ -11,7 +11,6 @@ anyhow.workspace = true
 api-utils = { path = "../crates/api-utils" }
 async-trait.workspace = true
 axum.workspace = true
-bao-tree.workspace = true
 bytes.workspace = true
 chrono.workspace = true
 cid.workspace = true

core/Cargo.toml

@@ -1,4 +1,4 @@
-use std::pin::Pin;
+use std::{fmt::Debug, pin::Pin, sync::Arc};
 
 use anyhow::Result;
 use async_trait::async_trait;
@@ -39,7 +39,7 @@ impl std::fmt::Display for ProviderType {
 
 /// CID Route Provider (CRP) Trait
 #[async_trait]
-pub trait Crp: Send + Sync {
+pub trait Crp: Send + Sync + Debug {
     fn provider_id(&self) -> String;
     fn provider_type(&self) -> ProviderType;
     async fn reindex(&self, cx: &Context) -> Result<()>;
@@ -53,10 +53,31 @@ pub trait Crp: Send + Sync {
     }
 }
 
+#[async_trait]
+impl Crp for Arc<dyn Crp> {
+    fn provider_id(&self) -> String {
+        self.as_ref().provider_id()
+    }
+    fn provider_type(&self) -> ProviderType {
+        self.as_ref().provider_type()
+    }
+    async fn reindex(&self, cx: &Context) -> Result<()> {
+        self.as_ref().reindex(cx).await
+    }
+
+    fn capabilities<'a>(&'a self) -> CrpCapabilities<'a> {
+        self.as_ref().capabilities()
+    }
+
+    fn cid_filter(&self) -> CidFilter {
+        self.as_ref().cid_filter()
+    }
+}
+
 /// All capabilities a CRP may have represented as self-referential trait objects.
 pub struct CrpCapabilities<'a> {
     pub route_resolver: Option<&'a dyn RouteResolver>,
-    pub size_resolver: Option<&'a dyn SizeResolver>,
+    pub blob_writer: Option<&'a dyn BlobWriter>,
 }
 
 /// A RouteResolver can dereference a route, turning it into a stream of bytes, accepting
@@ -78,14 +99,19 @@ pub trait RouteResolver {
     >;
 }
 
-/// A SizeResolver can return the length in bytes of the blob a CID points at.
-/// This is useful both as a preflight check before downloading a CID,
-/// and as a fast means of checking if a CRP has the CID in the first place.
+/// A RouteResolver can dereference a route, turning it into a stream of bytes, accepting
+/// authentication data.
 #[async_trait]
-pub trait SizeResolver {
-    async fn get_size(
+pub trait BlobWriter: Send + Sync {
+    /// Puts a blob into the CRP, given optional authentication data, a CID, and the data bytes.
+    ///
+    /// Note that this assumes that the data fits in memory, which is probably the case for most
+    /// data that eqty wants to write. If this becomes a problem, we will add a second method that
+    /// takes a stream of bytes instead.
+    async fn put_blob(
         &self,
+        auth: Option<bytes::Bytes>,
         cid: &Cid,
-        auth: Vec<u8>,
-    ) -> Result<u64, Box<dyn std::error::Error + Send + Sync>>;
+        data: &[u8],
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>>;
 }

core/src/crp.rs

@@ -362,6 +362,7 @@ mod tests {
         Context,
     };
 
+    #[derive(Debug)]
     struct StubAzureProvider {}
 
     #[async_trait]
@@ -381,7 +382,7 @@ mod tests {
         fn capabilities(&self) -> CrpCapabilities<'_> {
             CrpCapabilities {
                 route_resolver: None,
-                size_resolver: None,
+                blob_writer: None,
             }
         }
 

core/src/db.rs

@@ -48,7 +48,7 @@ impl Crp for Container {
     fn capabilities<'a>(&'a self) -> CrpCapabilities<'a> {
         CrpCapabilities {
             route_resolver: Some(self),
-            size_resolver: None, // TODO
+            blob_writer: None, // TODO
         }
     }
 

crps/azure/src/container.rs

@@ -9,16 +9,16 @@ edition = "2021"
 
 anyhow.workspace = true
 async-trait.workspace = true
-bao-tree.workspace = true
 bytes.workspace = true
 cid.workspace = true
 cid-router-core = { path = "../../core" }
 futures.workspace = true
 iroh.workspace = true
 iroh-base.workspace = true
 iroh-blobs.workspace = true
-iroh-tickets.workspace = true
 n0-future.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 tokio.workspace = true
+irpc.workspace = true
+log.workspace = true

crps/iroh/Cargo.toml

@@ -1,65 +1,42 @@
-use std::{pin::Pin, str::FromStr};
+use std::{io, path::PathBuf, pin::Pin};
 
 use anyhow::Result;
 use async_trait::async_trait;
-use bao_tree::io::BaoContentItem;
 use bytes::Bytes;
+use cid::Cid;
 use cid_router_core::{
     cid_filter::{CidFilter, CodeFilter},
-    crp::{Crp, CrpCapabilities, ProviderType, RouteResolver},
+    crp::{BlobWriter, Crp, CrpCapabilities, ProviderType, RouteResolver},
     routes::Route,
     Context,
 };
-use futures::{Stream, StreamExt};
-use iroh::{Endpoint, EndpointAddr, EndpointId};
-use iroh_blobs::{get::request::GetBlobItem, ticket::BlobTicket, Hash};
+use futures::Stream;
+use iroh_blobs::Hash;
+use log::info;
 use serde::{Deserialize, Serialize};
-use serde_json::Value;
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct IrohCrp {
-    addr: EndpointAddr,
-    endpoint: Endpoint,
+    store: iroh_blobs::store::fs::FsStore,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct IrohCrpConfig {
-    pub node_addr_ref: IrohNodeAddrRef,
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-#[serde(rename_all = "snake_case")]
-pub enum IrohNodeAddrRef {
-    EndpointId(String),
-    EndpointTicket(String),
-    Ticket(String),
+    /// Path to the directory where blobs are stored
+    pub path: PathBuf,
 }
 
 impl IrohCrp {
-    pub async fn new_from_config(config: Value) -> Result<Self> {
-        let IrohCrpConfig { node_addr_ref } = serde_json::from_value(config)?;
-
-        let endpoint_addr = match node_addr_ref {
-            IrohNodeAddrRef::EndpointId(node_id) => {
-                let endpoint_id = EndpointId::from_str(&node_id)?;
-                EndpointAddr::from(endpoint_id)
-            }
-            IrohNodeAddrRef::EndpointTicket(ticket) => {
-                let ticket = iroh_tickets::endpoint::EndpointTicket::from_str(&ticket)?;
-                ticket.endpoint_addr().to_owned()
-            }
-            IrohNodeAddrRef::Ticket(ticket) => {
-                let ticket = BlobTicket::from_str(&ticket)?;
-                ticket.addr().clone()
-            }
+    pub async fn new_from_config(config: IrohCrpConfig) -> io::Result<Self> {
+        let path = if config.path.is_absolute() {
+            config.path
+        } else {
+            std::env::current_dir()?.join(config.path)
         };
-
-        let endpoint = Endpoint::bind().await?;
-
-        Ok(Self {
-            addr: endpoint_addr,
-            endpoint,
-        })
+        let store = iroh_blobs::store::fs::FsStore::load(path)
+            .await
+            .map_err(|e| io::Error::other(e))?;
+        Ok(Self { store })
     }
 }
 
@@ -75,13 +52,13 @@ impl Crp for IrohCrp {
 
     async fn reindex(&self, _cx: &Context) -> anyhow::Result<()> {
         // TODO: Implement reindexing logic
-        todo!();
+        Ok(())
     }
 
     fn capabilities<'a>(&'a self) -> CrpCapabilities<'a> {
         CrpCapabilities {
             route_resolver: Some(self),
-            size_resolver: None, // TODO
+            blob_writer: Some(self),
         }
     }
 
@@ -90,6 +67,24 @@ impl Crp for IrohCrp {
     }
 }
 
+#[async_trait]
+impl BlobWriter for IrohCrp {
+    async fn put_blob(
+        &self,
+        _auth: Option<Bytes>,
+        cid: &Cid,
+        data: &[u8],
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        let blobs = self.store.blobs().clone();
+        let data = Bytes::copy_from_slice(data);
+        if cid.hash().code() != 0x1e {
+            return Err("Unsupported CID hash code; only blake3 is supported".into());
+        }
+        blobs.add_bytes(data).with_tag().await.map_err(Box::new)?;
+        Ok(())
+    }
+}
+
 #[async_trait]
 impl RouteResolver for IrohCrp {
     async fn get_bytes(
@@ -105,39 +100,13 @@ impl RouteResolver for IrohCrp {
         >,
         Box<dyn std::error::Error + Send + Sync>,
     > {
-        let Self { addr, .. } = self;
+        info!("get_bytes for route: {:?}", route);
         let cid = route.cid;
-
         let hash = cid.hash().digest();
         let hash: [u8; 32] = hash.try_into()?;
         let hash = Hash::from_bytes(hash);
-
-        let conn = self
-            .endpoint
-            .connect(addr.clone(), iroh_blobs::ALPN)
-            .await?;
-
-        println!("get {:?} from {}", hash, addr.id.fmt_short());
-
-        let res = iroh_blobs::get::request::get_blob(conn, hash);
-        let res = res
-            .take_while(|item| n0_future::future::ready(!matches!(item, GetBlobItem::Done(_))))
-            .filter_map(|item| {
-                n0_future::future::ready(match item {
-                    GetBlobItem::Item(item) => match item {
-                        BaoContentItem::Leaf(leaf) => Some(Ok(leaf.data)),
-                        // TODO - I don't think this is right. returning None here
-                        // will likely end the stream prematurely
-                        BaoContentItem::Parent(_parent) => None,
-                    },
-                    // This is filtered out, only for compiler happiness
-                    GetBlobItem::Done(_stats) => None,
-                    GetBlobItem::Error(err) => Some(Err(
-                        Box::new(err) as Box<dyn std::error::Error + Send + Sync>
-                    )),
-                })
-            });
-
-        Ok(Box::pin(res))
+        let data = self.store.blobs().get_bytes(hash).await.map_err(Box::new)?;
+        let stream = futures::stream::once(async move { Ok(data) });
+        Ok(Box::pin(stream))
     }
 }

crps/iroh/src/iroh.rs

@@ -1,3 +1,3 @@
 mod iroh;
 
-pub use iroh::{IrohCrp, IrohCrpConfig, IrohNodeAddrRef};
+pub use iroh::{IrohCrp, IrohCrpConfig};

crps/iroh/src/lib.rs

@@ -12,10 +12,10 @@ api-utils = { path = "../crates/api-utils" }
 async-trait.workspace = true
 axum.workspace = true
 axum-extra.workspace = true
-bao-tree.workspace = true
 bytes.workspace = true
 chrono.workspace = true
 cid.workspace = true
+blake3.workspace = true
 cid-router-core = { path = "../core" }
 clap.workspace = true
 crp-azure = { path = "../crps/azure" }