chore(hf): more explicit error handling

Author: kszucs

core/services/hf/src/core.rs

@@ -382,7 +382,9 @@ impl HfCore {
             Some(refresher),
             "opendal/1.0".to_string(),
         )
-        .map_err(map_xet_error)
+        .map_err(|err| {
+            Error::new(ErrorKind::Unexpected, "failed to create xet client").set_source(err)
+        })
     }
 
     /// Issue a HEAD request and extract XET file info (hash and size).
@@ -719,10 +721,6 @@ mod tests {
     }
 }
 
-pub(super) fn map_xet_error(err: impl std::error::Error + Send + Sync + 'static) -> Error {
-    Error::new(ErrorKind::Unexpected, "xet operation failed").set_source(err)
-}
-
 fn build_reqwest(policy: reqwest::redirect::Policy) -> Result<reqwest::Client> {
     reqwest::Client::builder()
         .redirect(policy)

core/services/hf/src/reader.rs

@@ -23,11 +23,12 @@ use futures::StreamExt;
 use subxet::cas_types::FileRange;
 
 use super::core::HfCore;
-use super::core::map_xet_error;
 use super::uri::RepoType;
 use opendal_core::raw::*;
 use opendal_core::*;
+use subxet::cas_client::CasClientError;
 use subxet::data::XetFileInfo;
+use subxet::data::errors::DataProcessingError;
 use subxet::data::streaming::XetReader;
 
 pub enum HfReader {
@@ -99,7 +100,9 @@ impl HfReader {
 
         let reader = client
             .read(file_info.clone(), file_range, None, 256)
-            .map_err(map_xet_error)?;
+            .map_err(|err| {
+                Error::new(ErrorKind::Unexpected, "failed to create xet reader").set_source(err)
+            })?;
         Ok(Self::Xet(reader))
     }
 }
@@ -110,7 +113,20 @@ impl oio::Read for HfReader {
             Self::Http(body) => body.read().await,
             Self::Xet(stream) => match stream.next().await {
                 Some(Ok(bytes)) => Ok(Buffer::from(bytes)),
-                Some(Err(e)) => Err(map_xet_error(e)),
+                Some(Err(e)) => {
+                    let kind = match &e {
+                        DataProcessingError::CasClientError(
+                            CasClientError::FileNotFound(_) | CasClientError::XORBNotFound(_),
+                        )
+                        | DataProcessingError::HashNotFound => ErrorKind::NotFound,
+                        DataProcessingError::CasClientError(CasClientError::InvalidRange) => {
+                            ErrorKind::RangeNotSatisfied
+                        }
+                        _ => ErrorKind::Unexpected,
+                    };
+                    let err = Error::new(kind, "xet read error").set_source(e);
+                    Err(err)
+                }
                 None => Ok(Buffer::new()),
             },
         }

core/services/hf/src/writer.rs

@@ -20,10 +20,11 @@ use std::sync::Mutex;
 
 use base64::Engine;
 
-use super::core::{BucketOperation, CommitFile, HfCore, LfsFile, map_xet_error};
+use super::core::{BucketOperation, CommitFile, HfCore, LfsFile};
 use super::uri::RepoType;
 use opendal_core::raw::*;
 use opendal_core::*;
+use subxet::data::errors::DataProcessingError;
 use subxet::data::streaming::XetWriter;
 
 /// Writer that handles both regular (small) and XET (large) file uploads.
@@ -45,34 +46,33 @@ pub enum HfWriter {
 impl HfWriter {
     /// Create a new writer by determining the upload mode from the API.
     pub async fn try_new(core: Arc<HfCore>, path: String) -> Result<Self> {
-        // Buckets always use XET and don't have a preupload endpoint
-        if core.repo.repo_type == RepoType::Bucket {
+        // Buckets always use XET and don't have a preupload endpoint;
+        // other repo types check the preupload API.
+        let use_xet = core.repo.repo_type == RepoType::Bucket
+            || core.determine_upload_mode(&path).await? == "lfs";
+
+        let writer = if use_xet {
             let client = core.xet_client("write").await?;
-            let writer = client.write(None).await.map_err(map_xet_error)?;
-            return Ok(HfWriter::Xet {
+            let writer = client.write(None).await.map_err(|err| {
+                let kind = match &err {
+                    DataProcessingError::AuthError(_) => ErrorKind::PermissionDenied,
+                    _ => ErrorKind::Unexpected,
+                };
+                Error::new(kind, "failed to create xet writer").set_source(err)
+            })?;
+            HfWriter::Xet {
                 core,
                 path,
                 writer: Mutex::new(writer),
-            });
-        }
-
-        let mode_str = core.determine_upload_mode(&path).await?;
-
-        if mode_str == "lfs" {
-            let client = core.xet_client("write").await?;
-            let writer = client.write(None).await.map_err(map_xet_error)?;
-            return Ok(HfWriter::Xet {
+            }
+        } else {
+            HfWriter::Regular {
                 core,
                 path,
-                writer: Mutex::new(writer),
-            });
-        }
-
-        Ok(HfWriter::Regular {
-            core,
-            path,
-            buf: Vec::new(),
-        })
+                buf: Vec::new(),
+            }
+        };
+        Ok(writer)
     }
 
     fn prepare_commit_file(path: &str, body: &[u8]) -> CommitFile {
@@ -97,7 +97,10 @@ impl oio::Write for HfWriter {
                 .unwrap()
                 .write(bs.to_bytes())
                 .await
-                .map_err(map_xet_error),
+                .map_err(|err| {
+                    Error::new(ErrorKind::Unexpected, "failed to write chunk to xet stream")
+                        .set_source(err)
+                }),
         }
     }
 
@@ -122,12 +125,9 @@ impl oio::Write for HfWriter {
                 Ok(meta)
             }
             HfWriter::Xet { core, path, writer } => {
-                let file_info = writer
-                    .get_mut()
-                    .unwrap()
-                    .close()
-                    .await
-                    .map_err(map_xet_error)?;
+                let file_info = writer.get_mut().unwrap().close().await.map_err(|err| {
+                    Error::new(ErrorKind::Unexpected, "failed to close xet writer").set_source(err)
+                })?;
 
                 let meta = Metadata::default().with_content_length(file_info.file_size());