Merge branch 'main' into entity-manager

rklaehn · rklaehn · commit 126d8343fbbb · 2025-07-08T10:19:09.000+03:00
diff --git a/src/api.rs b/src/api.rs
@@ -4,6 +4,7 @@
 //! with a remote store via rpc calls.
 use std::{io, net::SocketAddr, ops::Deref, sync::Arc};
 
+use bao_tree::io::EncodeError;
 use iroh::Endpoint;
 use irpc::rpc::{listen, Handler};
 use n0_snafu::SpanTrace;
@@ -211,6 +212,15 @@ impl std::error::Error for Error {
     }
 }
 
+impl From<EncodeError> for Error {
+    fn from(value: EncodeError) -> Self {
+        match value {
+            EncodeError::Io(cause) => Self::Io(cause),
+            _ => Self::other(value),
+        }
+    }
+}
+
 pub type Result<T> = std::result::Result<T, Error>;
 
 /// The main entry point for the store API.
diff --git a/src/api/blobs.rs b/src/api/blobs.rs
@@ -102,13 +102,21 @@ impl Blobs {
         })
     }
 
-    pub async fn delete_with_opts(&self, options: DeleteOptions) -> RequestResult<()> {
+    /// Delete a blob.
+    ///
+    /// This function is not public, because it does not work as expected when called manually,
+    /// because blobs are protected from deletion. This is only called from the gc task, which
+    /// clears the protections before.
+    ///
+    /// Users should rely only on garbage collection for blob deletion.
+    pub(crate) async fn delete_with_opts(&self, options: DeleteOptions) -> RequestResult<()> {
         trace!("{options:?}");
         self.client.rpc(options).await??;
         Ok(())
     }
 
-    pub async fn delete(
+    /// See [`Self::delete_with_opts`].
+    pub(crate) async fn delete(
         &self,
         hashes: impl IntoIterator<Item = impl Into<Hash>>,
     ) -> RequestResult<()> {
@@ -962,7 +970,6 @@ impl ExportBaoProgress {
         let mut data = Vec::new();
         let mut stream = self.into_byte_stream();
         while let Some(item) = stream.next().await {
-            println!("item: {item:?}");
             data.extend_from_slice(&item?);
         }
         Ok(data)
@@ -1088,7 +1095,7 @@ impl ExportBaoProgress {
             }
             EncodedItem::Leaf(leaf) => Some(Ok(leaf.data)),
             EncodedItem::Done => None,
-            EncodedItem::Error(cause) => Some(Err(super::Error::other(cause))),
+            EncodedItem::Error(cause) => Some(Err(cause.into())),
         })
     }
 
diff --git a/src/api/remote.rs b/src/api/remote.rs
@@ -10,7 +10,7 @@ use nested_enum_utils::common_fields;
 use ref_cast::RefCast;
 use snafu::{Backtrace, IntoError, Snafu};
 
-use super::blobs::Bitfield;
+use super::blobs::{Bitfield, ExportBaoOptions};
 use crate::{
     api::{blobs::WriteProgress, ApiClient},
     get::{fsm::DecodeError, BadRequestSnafu, GetError, GetResult, LocalFailureSnafu, Stats},
@@ -159,7 +159,7 @@ impl PushProgress {
 
 async fn just_result<S, R>(stream: S) -> Option<R>
 where
-    S: Stream,
+    S: Stream<Item: std::fmt::Debug>,
     R: TryFrom<S::Item>,
 {
     tokio::pin!(stream);
@@ -417,12 +417,17 @@ impl Remote {
         let root = request.hash;
         let bitfield = self.store().observe(root).await?;
         let children = if !request.ranges.is_blob() {
-            let bao = self.store().export_bao(root, bitfield.ranges.clone());
+            let opts = ExportBaoOptions {
+                hash: root,
+                ranges: bitfield.ranges.clone(),
+            };
+            let bao = self.store().export_bao_with_opts(opts, 32);
             let mut by_index = BTreeMap::new();
             let mut stream = bao.hashes_with_index();
             while let Some(item) = stream.next().await {
-                let (index, hash) = item?;
-                by_index.insert(index, hash);
+                if let Ok((index, hash)) = item {
+                    by_index.insert(index, hash);
+                }
             }
             let mut bitfields = BTreeMap::new();
             let mut hash_seq = BTreeMap::new();
diff --git a/src/store/fs.rs b/src/store/fs.rs
@@ -305,6 +305,34 @@ impl HashContext {
         Ok(())
     }
 
+    pub async fn get_maybe_create(&self, hash: Hash, create: bool) -> api::Result<BaoFileHandle> {
+        if create {
+            self.get_or_create(hash).await
+        } else {
+            self.get(hash).await
+        }
+    }
+
+    pub async fn get(&self, hash: Hash) -> api::Result<BaoFileHandle> {
+        if hash == Hash::EMPTY {
+            return Ok(self.ctx.empty.clone());
+        }
+        let res = self
+            .slot
+            .get_or_create(|| async {
+                let res = self.db().get(hash).await.map_err(io::Error::other)?;
+                let res = match res {
+                    Some(state) => open_bao_file(&hash, state, &self.ctx).await,
+                    None => Err(io::Error::new(io::ErrorKind::NotFound, "hash not found")),
+                };
+                Ok((res?, ()))
+            })
+            .await
+            .map_err(api::Error::from);
+        let (res, _) = res?;
+        Ok(res)
+    }
+
     pub async fn get_or_create(&self, hash: Hash) -> api::Result<BaoFileHandle> {
         if hash == Hash::EMPTY {
             return Ok(self.ctx.empty.clone());
@@ -939,7 +967,7 @@ async fn observe(cmd: ObserveMsg, ctx: HashContext) {
 
 #[instrument(skip_all, fields(hash = %cmd.hash_short()))]
 async fn export_ranges(mut cmd: ExportRangesMsg, ctx: HashContext) {
-    match ctx.get_or_create(cmd.hash).await {
+    match ctx.get(cmd.hash).await {
         Ok(handle) => {
             if let Err(cause) = export_ranges_impl(cmd.inner, &mut cmd.tx, handle).await {
                 cmd.tx
@@ -1000,7 +1028,7 @@ async fn export_ranges_impl(
 
 #[instrument(skip_all, fields(hash = %cmd.hash_short()))]
 async fn export_bao(mut cmd: ExportBaoMsg, ctx: HashContext) {
-    match ctx.get_or_create(cmd.hash).await {
+    match ctx.get_maybe_create(cmd.hash, false).await {
         Ok(handle) => {
             if let Err(cause) = export_bao_impl(cmd.inner, &mut cmd.tx, handle).await {
                 cmd.tx
@@ -1010,9 +1038,9 @@ async fn export_bao(mut cmd: ExportBaoMsg, ctx: HashContext) {
             }
         }
         Err(cause) => {
-            let cause = anyhow::anyhow!("failed to open file: {cause}");
+            let crate::api::Error::Io(cause) = cause;
             cmd.tx
-                .send(bao_tree::io::EncodeError::Io(io::Error::other(cause)).into())
+                .send(bao_tree::io::EncodeError::Io(cause).into())
                 .await
                 .ok();
         }
@@ -1024,7 +1052,7 @@ async fn export_bao_impl(
     tx: &mut mpsc::Sender<EncodedItem>,
     handle: BaoFileHandle,
 ) -> anyhow::Result<()> {
-    let ExportBaoRequest { ranges, hash } = cmd;
+    let ExportBaoRequest { ranges, hash, .. } = cmd;
     debug_assert!(handle.hash() == hash, "hash mismatch");
     let outboard = handle.outboard()?;
     let size = outboard.tree.size();
diff --git a/src/store/fs/gc.rs b/src/store/fs/gc.rs
@@ -189,14 +189,18 @@ pub async fn run_gc(store: Store, config: GcConfig) {
 
 #[cfg(test)]
 mod tests {
-    use std::path::Path;
+    use std::{
+        io::{self},
+        path::Path,
+    };
 
-    use bao_tree::ChunkNum;
+    use bao_tree::{io::EncodeError, ChunkNum};
+    use range_collections::RangeSet2;
     use testresult::TestResult;
 
     use super::*;
     use crate::{
-        api::{blobs::AddBytesOptions, Store},
+        api::{blobs::AddBytesOptions, ExportBaoError, RequestError, Store},
         hashseq::HashSeq,
         store::fs::{options::PathOptions, tests::create_n0_bao},
         BlobFormat,
@@ -326,4 +330,83 @@ mod tests {
         gc_smoke(&store).await?;
         Ok(())
     }
+
+    #[tokio::test]
+    async fn gc_check_deletion_fs() -> TestResult {
+        tracing_subscriber::fmt::try_init().ok();
+        let testdir = tempfile::tempdir()?;
+        let db_path = testdir.path().join("db");
+        let store = crate::store::fs::FsStore::load(&db_path).await?;
+        gc_check_deletion(&store).await
+    }
+
+    #[tokio::test]
+    async fn gc_check_deletion_mem() -> TestResult {
+        tracing_subscriber::fmt::try_init().ok();
+        let store = crate::store::mem::MemStore::default();
+        gc_check_deletion(&store).await
+    }
+
+    async fn gc_check_deletion(store: &Store) -> TestResult {
+        let temp_tag = store.add_bytes(b"foo".to_vec()).temp_tag().await?;
+        let hash = *temp_tag.hash();
+        assert_eq!(store.get_bytes(hash).await?.as_ref(), b"foo");
+        drop(temp_tag);
+        let mut live = HashSet::new();
+        gc_run_once(store, &mut live).await?;
+
+        // check that `get_bytes` returns an error.
+        let res = store.get_bytes(hash).await;
+        assert!(res.is_err());
+        assert!(matches!(
+            res,
+            Err(ExportBaoError::ExportBaoInner {
+                source: EncodeError::Io(cause),
+                ..
+            }) if cause.kind() == io::ErrorKind::NotFound
+        ));
+
+        // check that `export_ranges` returns an error.
+        let res = store
+            .export_ranges(hash, RangeSet2::all())
+            .concatenate()
+            .await;
+        assert!(res.is_err());
+        assert!(matches!(
+            res,
+            Err(RequestError::Inner{
+                source: crate::api::Error::Io(cause),
+                ..
+            }) if cause.kind() == io::ErrorKind::NotFound
+        ));
+
+        // check that `export_bao` returns an error.
+        let res = store
+            .export_bao(hash, ChunkRanges::all())
+            .bao_to_vec()
+            .await;
+        assert!(res.is_err());
+        println!("export_bao res {res:?}");
+        assert!(matches!(
+            res,
+            Err(RequestError::Inner{
+                source: crate::api::Error::Io(cause),
+                ..
+            }) if cause.kind() == io::ErrorKind::NotFound
+        ));
+
+        // check that `export` returns an error.
+        let target = tempfile::NamedTempFile::new()?;
+        let path = target.path();
+        let res = store.export(hash, path).await;
+        assert!(res.is_err());
+        assert!(matches!(
+            res,
+            Err(RequestError::Inner{
+                source: crate::api::Error::Io(cause),
+                ..
+            }) if cause.kind() == io::ErrorKind::NotFound
+        ));
+        Ok(())
+    }
 }
diff --git a/src/store/fs/meta.rs b/src/store/fs/meta.rs
@@ -463,6 +463,7 @@ impl Actor {
         } = cmd;
         for hash in hashes {
             if !force && protected.contains(&hash) {
+                trace!("delete {hash}: skip (protected)");
                 continue;
             }
             if let Some(entry) = tables.blobs.remove(hash).context(StorageSnafu)? {
@@ -471,6 +472,7 @@ impl Actor {
                         data_location,
                         outboard_location,
                     } => {
+                        trace!("delete {hash}: currently complete. will be deleted.");
                         match data_location {
                             DataLocation::Inline(_) => {
                                 tables.inline_data.remove(hash).context(StorageSnafu)?;
@@ -493,6 +495,7 @@ impl Actor {
                         }
                     }
                     EntryState::Partial { .. } => {
+                        trace!("delete {hash}: currently partial. will be deleted.");
                         tables.ftx.delete(
                             hash,
                             [
diff --git a/src/store/mem.rs b/src/store/mem.rs
diff --git a/src/store/readonly_mem.rs b/src/store/readonly_mem.rs

Original file line number	Diff line number	Diff line change
`@@ -463,6 +463,7 @@ impl Actor {`
`463`	`463`	`} = cmd;`
`464`	`464`	`for hash in hashes {`
`465`	`465`	`if !force && protected.contains(&hash) {`
	`466`	`+ trace!("delete {hash}: skip (protected)");`
`466`	`467`	`continue;`
`467`	`468`	`}`
`468`	`469`	`if let Some(entry) = tables.blobs.remove(hash).context(StorageSnafu)? {`
`@@ -471,6 +472,7 @@ impl Actor {`
`471`	`472`	`data_location,`
`472`	`473`	`outboard_location,`
`473`	`474`	`} => {`
	`475`	`+ trace!("delete {hash}: currently complete. will be deleted.");`
`474`	`476`	`match data_location {`
`475`	`477`	`DataLocation::Inline(_) => {`
`476`	`478`	`tables.inline_data.remove(hash).context(StorageSnafu)?;`
`@@ -493,6 +495,7 @@ impl Actor {`
`493`	`495`	`}`
`494`	`496`	`}`
`495`	`497`	`EntryState::Partial { .. } => {`
	`498`	`+ trace!("delete {hash}: currently partial. will be deleted.");`
`496`	`499`	`tables.ftx.delete(`
`497`	`500`	`hash,`
`498`	`501`	`[`