WIP add fn open

Author: rklaehn

Cargo.lock

@@ -18,6 +18,7 @@ async-channel = "2.3.1"
 bao-tree = { version = "0.13", features = [
     "tokio_fsm",
     "validate",
+    "experimental-mixed",
 ], default-features = false }
 blake3 = { version = "1.4.5", package = "iroh-blake3" }
 bytes = { version = "1.7", features = ["serde"] }
@@ -77,6 +78,7 @@ walkdir = { version = "2.5.0", optional = true }
 # Examples
 console = { version = "0.15.8", optional = true }
 tracing-test = "0.2.5"
+positioned-io = "0.3.3"
 
 [dev-dependencies]
 http-body = "1.0"
@@ -189,3 +191,4 @@ incremental = false
 iroh = { git = "https://github.com/n0-computer/iroh.git", branch = "main" }
 iroh-base = { git = "https://github.com/n0-computer/iroh.git", branch = "main" }
 quic-rpc = { git = "https://github.com/n0-computer/quic-rpc", branch = "main" }
+bao-tree = { git = "https://github.com/n0-computer/bao-tree", branch = "read_and_seek" }

Cargo.toml

@@ -23,10 +23,10 @@ use proto::{
         AddPathRequest, AddPathResponse, AddStreamRequest, AddStreamResponse, AddStreamUpdate,
         BatchAddPathRequest, BatchAddPathResponse, BatchAddStreamRequest, BatchAddStreamResponse,
         BatchAddStreamUpdate, BatchCreateRequest, BatchCreateResponse, BatchCreateTempTagRequest,
-        BatchUpdate, BlobStatusRequest, BlobStatusResponse, ConsistencyCheckRequest,
-        CreateCollectionRequest, CreateCollectionResponse, DeleteRequest, DownloadResponse,
-        ExportRequest, ExportResponse, ListIncompleteRequest, ListRequest, ReadAtRequest,
-        ReadAtResponse, ValidateRequest,
+        BatchUpdate, BlobEntryInfoRequest, BlobStatusRequest, BlobStatusResponse,
+        ConsistencyCheckRequest, CreateCollectionRequest, CreateCollectionResponse, DeleteRequest,
+        DownloadResponse, ExportRequest, ExportResponse, ListIncompleteRequest, ListRequest,
+        ReadAtRequest, ReadAtResponse, ValidateRequest,
     },
     tags::{
         CreateRequest as TagsCreateRequest, DeleteRequest as TagDeleteRequest,
@@ -51,7 +51,9 @@ use crate::{
     },
     net_protocol::{BlobDownloadRequest, Blobs, BlobsInner},
     provider::{AddProgress, BatchAddPathProgress},
-    store::{ConsistencyCheckProgress, ImportProgress, MapEntry, ValidateProgress},
+    store::{
+        ConsistencyCheckProgress, EntryPathOrData, ImportProgress, MapEntry, ValidateProgress,
+    },
     util::{
         local_pool::LocalPoolHandle,
         progress::{AsyncChannelProgressSender, ProgressSender},
@@ -203,6 +205,7 @@ impl<D: crate::store::Store> Handler<D> {
                     .await
             }
             BatchCreateTempTag(msg) => chan.rpc(msg, self, Self::batch_create_temp_tag).await,
+            EntryInfo(msg) => chan.rpc(msg, self, Self::blob_entry_info).await,
         }
     }
 
@@ -309,6 +312,17 @@ impl<D: crate::store::Store> Handler<D> {
         Ok(())
     }
 
+    async fn blob_entry_info(
+        self,
+        msg: BlobEntryInfoRequest,
+    ) -> RpcResult<Option<EntryPathOrData>> {
+        Ok(self
+            .store()
+            .entry_path_or_data(msg.hash)
+            .await
+            .map_err(|e| RpcError::new(&e))?)
+    }
+
     fn blob_list_tags(self, msg: TagListRequest) -> impl Stream<Item = TagInfo> + Send + 'static {
         tracing::info!("blob_list_tags");
         let blobs = self;

src/rpc.rs

@@ -66,6 +66,10 @@ use std::{
 };
 
 use anyhow::{anyhow, Context as _, Result};
+use bao_tree::{
+    io::{baofile::BaoFile, outboard::PreOrderOutboard},
+    BaoTree,
+};
 use bytes::Bytes;
 use futures_lite::{Stream, StreamExt};
 use futures_util::SinkExt;
@@ -87,10 +91,10 @@ use crate::{
     format::collection::{Collection, SimpleStore},
     get::db::DownloadProgress as BytesDownloadProgress,
     net_protocol::BlobDownloadRequest,
-    rpc::proto::RpcService,
+    rpc::proto::{blobs::BlobEntryInfoRequest, RpcService},
     store::{BaoBlobSize, ConsistencyCheckProgress, ExportFormat, ExportMode, ValidateProgress},
-    util::SetTagOption,
-    BlobFormat, Hash, Tag,
+    util::{MemOrFile, SetTagOption},
+    BlobFormat, Hash, Tag, IROH_BLOCK_SIZE,
 };
 
 mod batch;
@@ -380,6 +384,31 @@ where
         ))
     }
 
+    /// Open a blob as an independent bao file
+    pub async fn open(&self, hash: Hash) -> Result<impl std::io::Read + std::io::Seek> {
+        let Some(info) = self.rpc.rpc(BlobEntryInfoRequest { hash }).await?? else {
+            return Err(anyhow!("Blob not found"));
+        };
+        let (data, size) = match info.data {
+            MemOrFile::Mem(data) => (MemOrFile::Mem(data.clone()), data.len() as u64),
+            MemOrFile::File((path, size)) => (MemOrFile::File(std::fs::File::open(path)?), size),
+        };
+        let outboard = match info.outboard {
+            MemOrFile::Mem(data) => MemOrFile::Mem(data.clone()),
+            MemOrFile::File(path) => MemOrFile::File(std::fs::File::open(path)?),
+        };
+        let file = BaoFile {
+            data,
+            outboard: PreOrderOutboard {
+                tree: BaoTree::new(size, IROH_BLOCK_SIZE),
+                root: hash.into(),
+                data: outboard,
+            },
+        };
+        let file = positioned_io::Cursor::new(file);
+        Ok(file)
+    }
+
     /// Export a blob from the internal blob store to a path on the node's filesystem.
     ///
     /// `destination` should be an writeable, absolute path on the local node's filesystem.

src/rpc/client/blobs.rs

@@ -15,8 +15,8 @@ use crate::{
     provider::{AddProgress, BatchAddPathProgress},
     rpc::client::blobs::{BlobInfo, BlobStatus, IncompleteBlobInfo, ReadAtLen, WrapOption},
     store::{
-        BaoBlobSize, ConsistencyCheckProgress, ExportFormat, ExportMode, ImportMode,
-        ValidateProgress,
+        BaoBlobSize, ConsistencyCheckProgress, EntryPathOrData, ExportFormat, ExportMode,
+        ImportMode, ValidateProgress,
     },
     util::SetTagOption,
     BlobFormat, Hash, HashAndFormat, Tag,
@@ -63,6 +63,8 @@ pub enum Request {
     BatchAddPath(BatchAddPathRequest),
     #[rpc(response = RpcResult<()>)]
     BatchCreateTempTag(BatchCreateTempTagRequest),
+    #[rpc(response = RpcResult<Option<EntryPathOrData>>)]
+    EntryInfo(BlobEntryInfoRequest),
 }
 
 #[allow(missing_docs)]
@@ -83,6 +85,7 @@ pub enum Response {
     BatchCreate(BatchCreateResponse),
     BatchAddStream(BatchAddStreamResponse),
     BatchAddPath(BatchAddPathResponse),
+    EntryInfo(RpcResult<Option<EntryPathOrData>>),
 }
 
 /// A request to the node to provide the data at the given path
@@ -313,6 +316,13 @@ pub struct BatchAddPathRequest {
     pub batch: BatchId,
 }
 
+/// Write a blob from a byte stream
+#[derive(Serialize, Deserialize, Debug)]
+pub struct BlobEntryInfoRequest {
+    /// The hash of the blob
+    pub hash: Hash,
+}
+
 /// Response to a batch add path request
 #[derive(Serialize, Deserialize, Debug)]
 pub struct BatchAddPathResponse(pub BatchAddPathProgress);

src/rpc/proto/blobs.rs

@@ -43,7 +43,7 @@ pub enum EntryStatus {
 }
 
 /// Get the path or data for an entry
-#[derive(Debug)]
+#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
 pub struct EntryPathOrData {
     /// The path to the data file or the inline data
     pub data: MemOrFile<Bytes, (PathBuf, u64)>,

src/store/traits.rs

@@ -1,19 +1,36 @@
 use std::{fs::File, io};
 
-use bao_tree::io::sync::{ReadAt, Size};
+use bao_tree::io::{
+    mixed::ReadBytesAt,
+    sync::{ReadAt, Size},
+};
 use bytes::Bytes;
+use serde::{Deserialize, Serialize};
 
 /// This is a general purpose Either, just like Result, except that the two cases
 /// are Mem for something that is in memory, and File for something that is somewhere
 /// external and only available via io.
-#[derive(Debug)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub enum MemOrFile<M, F> {
     /// We got it all in memory
     Mem(M),
     /// A file
     File(F),
 }
 
+impl<M, F> ReadBytesAt for MemOrFile<M, F>
+where
+    M: ReadBytesAt,
+    F: ReadBytesAt,
+{
+    fn read_bytes_at(&self, offset: u64, size: usize) -> io::Result<Bytes> {
+        match self {
+            MemOrFile::Mem(mem) => mem.read_bytes_at(offset, size),
+            MemOrFile::File(file) => file.read_bytes_at(offset, size),
+        }
+    }
+}
+
 /// Helper methods for a common way to use MemOrFile, where the memory part is something
 /// like a slice, and the file part is a tuple consisiting of path or file and size.
 impl<M, F> MemOrFile<M, (F, u64)>