Replace api::Error::other with dedicated errors.

Author: rklaehn

src/api.rs

@@ -199,7 +199,7 @@ impl From<EncodeError> for Error {
     fn from(value: EncodeError) -> Self {
         match value {
             EncodeError::Io(cause) => Self::Io(cause),
-            _ => Self::other(value),
+            _ => Self::Io(io::Error::other(value)),
         }
     }
 }

src/api/blobs.rs

@@ -440,13 +440,13 @@ impl Blobs {
         reader
             .recv_exact(&mut size)
             .await
-            .map_err(super::Error::other)?;
+            .map_err(io::Error::from)?;
         let size = u64::from_le_bytes(size);
         let Some(size) = NonZeroU64::new(size) else {
             return if hash == Hash::EMPTY {
                 Ok(reader)
             } else {
-                Err(super::Error::other("invalid size for hash").into())
+                Err(io::Error::other("invalid size for hash").into())
             };
         };
         let tree = BaoTree::new(size.get(), IROH_BLOCK_SIZE);
@@ -651,7 +651,7 @@ impl<'a> AddProgress<'a> {
                 _ => {}
             }
         }
-        Err(super::Error::other("unexpected end of stream").into())
+        Err(io::Error::other("unexpected end of stream").into())
     }
 
     pub async fn with_named_tag(self, name: impl AsRef<[u8]>) -> RequestResult<HashAndFormat> {
@@ -704,7 +704,7 @@ impl IntoFuture for ObserveProgress {
             let mut rx = self.inner.await?;
             match rx.recv().await? {
                 Some(bitfield) => Ok(bitfield),
-                None => Err(super::Error::other("unexpected end of stream").into()),
+                None => Err(io::Error::other("unexpected end of stream").into()),
             }
         })
     }
@@ -726,7 +726,7 @@ impl ObserveProgress {
                 return Ok(item);
             }
         }
-        Err(super::Error::other("unexpected end of stream").into())
+        Err(io::Error::other("unexpected end of stream").into())
     }
 
     /// Returns an infinite stream of bitfields. The first bitfield is the
@@ -805,7 +805,7 @@ impl ExportProgress {
         if let Some(size) = size {
             Ok(size)
         } else {
-            Err(super::Error::other("unexpected end of stream").into())
+            Err(io::Error::other("unexpected end of stream").into())
         }
     }
 }

src/store/fs.rs

@@ -853,7 +853,7 @@ async fn handle_batch(cmd: BatchMsg, id: Scope, scope: Arc<TempTagScope>, ctx: A
 async fn handle_batch_impl(cmd: BatchMsg, id: Scope, scope: &Arc<TempTagScope>) -> api::Result<()> {
     let BatchMsg { tx, mut rx, .. } = cmd;
     trace!("created scope {}", id);
-    tx.send(id).await.map_err(api::Error::other)?;
+    tx.send(id).await?;
     while let Some(msg) = rx.recv().await? {
         match msg {
             BatchResponse::Drop(msg) => scope.on_drop(&msg),
@@ -1266,9 +1266,7 @@ async fn export_path_impl(
         MemOrFile::Mem(data) => data.len() as u64,
         MemOrFile::File((_, size)) => *size,
     };
-    tx.send(ExportProgressItem::Size(size))
-        .await
-        .map_err(api::Error::other)?;
+    tx.send(ExportProgressItem::Size(size)).await?;
     match data {
         MemOrFile::Mem(data) => {
             let mut target = fs::File::create(&target)?;
@@ -1324,9 +1322,7 @@ async fn export_path_impl(
             }
         },
     }
-    tx.send(ExportProgressItem::Done)
-        .await
-        .map_err(api::Error::other)?;
+    tx.send(ExportProgressItem::Done).await?;
     Ok(())
 }
 

src/store/fs/meta.rs

@@ -363,7 +363,7 @@ async fn handle_list_tags(msg: ListTagsMsg, tables: &impl ReadableTables) -> Act
                     res.push(crate::api::Result::Ok(info));
                 }
             }
-            Err(e) => res.push(Err(crate::api::Error::other(e))),
+            Err(e) => res.push(Err(api_error_from_storage_error(e))),
         }
     }
     tx.send(res).await.ok();
@@ -629,8 +629,12 @@ impl Actor {
             tx,
             ..
         } = cmd;
-        let res = tables.tags.insert(tag, value).map(|_| ());
-        tx.send(res.map_err(crate::api::Error::other)).await.ok();
+        let res = tables
+            .tags
+            .insert(tag, value)
+            .map_err(api_error_from_storage_error)
+            .map(|_| ());
+        tx.send(res).await.ok();
         Ok(())
     }
 
@@ -852,6 +856,13 @@ impl Actor {
     }
 }
 
+/// Convert a redb StorageError into an api::Error
+///
+/// This can't be a From instance because that would require exposing redb::StorageError in the public API.
+fn api_error_from_storage_error(e: redb::StorageError) -> api::Error {
+    api::Error::Io(io::Error::other(e))
+}
+
 #[derive(Debug)]
 struct DbWrapper(Option<Database>);
 
@@ -953,8 +964,12 @@ async fn list_blobs_impl(
     _cmd: ListRequest,
     tx: &mut mpsc::Sender<api::Result<Hash>>,
 ) -> api::Result<()> {
-    for item in snapshot.blobs.iter().map_err(api::Error::other)? {
-        let (k, _) = item.map_err(api::Error::other)?;
+    for item in snapshot
+        .blobs
+        .iter()
+        .map_err(api_error_from_storage_error)?
+    {
+        let (k, _) = item.map_err(api_error_from_storage_error)?;
         let k = k.value();
         tx.send(Ok(k)).await.ok();
     }

src/store/mem.rs

@@ -542,7 +542,7 @@ async fn handle_batch(cmd: BatchMsg, id: Scope, scope: Arc<TempTagScope>) -> Sco
 async fn handle_batch_impl(cmd: BatchMsg, id: Scope, scope: &Arc<TempTagScope>) -> api::Result<()> {
     let BatchMsg { tx, mut rx, .. } = cmd;
     trace!("created scope {}", id);
-    tx.send(id).await.map_err(api::Error::other)?;
+    tx.send(id).await?;
     while let Some(msg) = rx.recv().await? {
         match msg {
             BatchResponse::Drop(msg) => scope.on_drop(&msg),