fix: getting deleted blobs returns errors

Frando · Frando · commit ea298630eaa0 · 2025-07-03T10:50:03.000+02: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/store/fs.rs b/src/store/fs.rs
@@ -305,6 +305,26 @@ impl HashContext {
         Ok(())
     }
 
+    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 +959,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 +1020,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(cmd.hash).await {
         Ok(handle) => {
             if let Err(cause) = export_bao_impl(cmd.inner, &mut cmd.tx, handle).await {
                 cmd.tx
@@ -1010,9 +1030,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();
         }
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
@@ -23,7 +23,7 @@ use bao_tree::{
         mixed::{traverse_ranges_validated, EncodedItem, ReadBytesAt},
         outboard::PreOrderMemOutboard,
         sync::{Outboard, ReadAt, WriteAt},
-        BaoContentItem, Leaf,
+        BaoContentItem, EncodeError, Leaf,
     },
     BaoTree, ChunkNum, ChunkRanges, TreeNode,
 };
@@ -193,8 +193,8 @@ impl Actor {
                 tx,
                 ..
             }) => {
-                let entry = self.get_or_create_entry(hash);
-                self.spawn(export_bao(entry, ranges, tx));
+                let entry = self.state.data.get(&hash).cloned();
+                self.spawn(export_bao(entry, ranges, tx))
             }
             Command::ExportPath(cmd) => {
                 let entry = self.state.data.get(&cmd.hash).cloned();
@@ -371,8 +371,8 @@ impl Actor {
                 cmd.tx.send(Ok(())).await.ok();
             }
             Command::ExportRanges(cmd) => {
-                let entry = self.get_or_create_entry(cmd.hash);
-                self.spawn(export_ranges(cmd, entry.clone()));
+                let entry = self.state.data.get(&cmd.hash).cloned();
+                self.spawn(export_ranges(cmd, entry));
             }
             Command::SyncDb(SyncDbMsg { tx, .. }) => {
                 tx.send(Ok(())).await.ok();
@@ -501,7 +501,12 @@ async fn handle_batch_impl(cmd: BatchMsg, id: Scope, scope: &Arc<TempTagScope>)
     Ok(())
 }
 
-async fn export_ranges(mut cmd: ExportRangesMsg, entry: BaoFileHandle) {
+async fn export_ranges(mut cmd: ExportRangesMsg, entry: Option<BaoFileHandle>) {
+    let Some(entry) = entry else {
+        let err = io::Error::new(io::ErrorKind::NotFound, "hash not found");
+        cmd.tx.send(ExportRangesItem::Error(err.into())).await.ok();
+        return;
+    };
     if let Err(cause) = export_ranges_impl(cmd.inner, &mut cmd.tx, entry).await {
         cmd.tx
             .send(ExportRangesItem::Error(cause.into()))
@@ -624,12 +629,18 @@ async fn import_bao(
     tx.send(Ok(())).await.ok();
 }
 
-#[instrument(skip_all, fields(hash = %entry.hash.fmt_short()))]
+#[instrument(skip_all, fields(hash = tracing::field::Empty))]
 async fn export_bao(
-    entry: BaoFileHandle,
+    entry: Option<BaoFileHandle>,
     ranges: ChunkRanges,
     mut sender: mpsc::Sender<EncodedItem>,
 ) {
+    let Some(entry) = entry else {
+        let err = EncodeError::Io(io::Error::new(io::ErrorKind::NotFound, "hash not found"));
+        sender.send(err.into()).await.ok();
+        return;
+    };
+    tracing::Span::current().record("hash", tracing::field::display(entry.hash));
     let data = entry.data_reader();
     let outboard = entry.outboard_reader();
     let tx = BaoTreeSender::new(&mut sender);

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`	`[`