refactor(hf): remove the option to disable xet in runtime

Author: kszucs

core/services/hf/src/backend.rs

@@ -121,21 +121,6 @@ impl HfBuilder {
         self
     }
 
-    /// Enable XET storage protocol for reads.
-    ///
-    /// When enabled, reads will check for XET-backed files and use the
-    /// XET protocol for downloading. Default is disabled.
-    pub fn enable_xet(mut self) -> Self {
-        self.config.xet = true;
-        self
-    }
-
-    /// Disable XET storage protocol for reads.
-    pub fn disable_xet(mut self) -> Self {
-        self.config.xet = false;
-        self
-    }
-
     /// Set the maximum number of retries for commit operations.
     ///
     /// Retries on commit conflicts (HTTP 412) and transient server
@@ -211,14 +196,13 @@ impl Builder for HfBuilder {
         debug!("backend max_retries: {}", max_retries);
 
         Ok(HfBackend {
-            core: Arc::new(HfCore::new(
+            core: Arc::new(HfCore::build(
                 info,
                 repo,
                 root,
                 token,
                 endpoint,
                 max_retries,
-                self.config.xet,
             )?),
         })
     }
@@ -318,31 +302,14 @@ pub(super) mod test_utils {
         finish_operator(op)
     }
 
-    pub fn testing_xet_operator() -> Operator {
-        let (repo_id, token) = testing_credentials();
-        let op = Operator::new(
-            HfBuilder::default()
-                .repo_type("dataset")
-                .repo_id(&repo_id)
-                .token(&token)
-                .enable_xet()
-                .max_retries(10),
-        )
-        .unwrap()
-        .finish();
-        finish_operator(op)
-    }
-
     /// Operator for a bucket requiring HF_OPENDAL_BUCKET and HF_OPENDAL_TOKEN.
-    /// Buckets always use XET for writes.
     pub fn testing_bucket_operator() -> Operator {
         let (repo_id, token) = testing_bucket_credentials();
         let op = Operator::new(
             HfBuilder::default()
                 .repo_type("bucket")
                 .repo_id(&repo_id)
                 .token(&token)
-                .enable_xet()
                 .max_retries(10),
         )
         .unwrap()
@@ -371,24 +338,11 @@ pub(super) mod test_utils {
         .finish();
         finish_operator(op)
     }
-
-    pub fn mbpp_xet_operator() -> Operator {
-        let mut builder = HfBuilder::default()
-            .repo_type("dataset")
-            .repo_id("google-research-datasets/mbpp")
-            .enable_xet();
-        if let Ok(token) = std::env::var("HF_OPENDAL_TOKEN") {
-            builder = builder.token(&token);
-        }
-        let op = Operator::new(builder).unwrap().finish();
-        finish_operator(op)
-    }
 }
 
 #[cfg(test)]
 mod tests {
     use super::test_utils::mbpp_operator;
-    use super::test_utils::mbpp_xet_operator;
     use super::*;
 
     #[test]
@@ -462,7 +416,7 @@ mod tests {
     #[tokio::test]
     #[ignore = "requires network access"]
     async fn test_read_parquet_xet() {
-        let op = mbpp_xet_operator();
+        let op = mbpp_operator();
         let path = "full/train-00000-of-00001.parquet";
 
         // Full read via XET and verify parquet magic at both ends

core/services/hf/src/config.rs

@@ -52,12 +52,6 @@ pub struct HfConfig {
     ///
     /// Default is "https://huggingface.co".
     pub endpoint: Option<String>,
-    /// Enable XET storage protocol for reads.
-    ///
-    /// When true and the `xet` feature is compiled in, reads will
-    /// check for XET-backed files and use the XET protocol for
-    /// downloading. Default is false.
-    pub xet: bool,
     /// Maximum number of retries for commit operations.
     ///
     /// Retries on commit conflicts (HTTP 412) and transient server
@@ -125,7 +119,6 @@ impl opendal_core::Configurator for HfConfig {
                 root: opts.get("root").cloned(),
                 token: opts.get("token").cloned(),
                 endpoint: opts.get("endpoint").cloned(),
-                xet: opts.get("xet").is_some_and(|v| v == "true"),
                 ..Default::default()
             })
         }

core/services/hf/src/core.rs

@@ -221,24 +221,18 @@ pub struct HfCore {
     pub endpoint: String,
     pub max_retries: usize,
 
-    // Whether XET storage protocol is enabled for reads. When true,
-    // reads will check for XET-backed files and use the XET protocol
-    // for downloading.
-    pub xet_enabled: bool,
-
     /// HTTP client with redirects disabled, used by XET probes to
     /// inspect headers on 302 responses.
     pub no_redirect_client: HttpClient,
 }
 
 impl Debug for HfCore {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        let mut s = f.debug_struct("HfCore");
-        s.field("repo", &self.repo)
+        f.debug_struct("HfCore")
+            .field("repo", &self.repo)
             .field("root", &self.root)
-            .field("endpoint", &self.endpoint);
-        s.field("xet_enabled", &self.xet_enabled);
-        s.finish_non_exhaustive()
+            .field("endpoint", &self.endpoint)
+            .finish_non_exhaustive()
     }
 }
 
@@ -250,33 +244,44 @@ impl HfCore {
         token: Option<String>,
         endpoint: String,
         max_retries: usize,
-        xet_enabled: bool,
+        no_redirect_client: HttpClient,
+    ) -> Self {
+        Self {
+            info,
+            repo,
+            root,
+            token,
+            endpoint,
+            max_retries,
+            no_redirect_client,
+        }
+    }
+
+    /// Build HfCore with dedicated reqwest HTTP clients.
+    ///
+    /// Uses separate clients for standard and no-redirect requests to
+    /// avoid "dispatch task is gone" errors with multiple tokio runtimes.
+    pub fn build(
+        info: Arc<AccessorInfo>,
+        repo: HfRepo,
+        root: String,
+        token: Option<String>,
+        endpoint: String,
+        max_retries: usize,
     ) -> Result<Self> {
-        // When xet is enabled at runtime, use dedicated reqwest clients instead
-        // of the global one. This avoids "dispatch task is gone" errors when
-        // multiple tokio runtimes exist (e.g. in tests) and ensures the
-        // no-redirect client shares the same runtime as the standard client.
-        // When xet is disabled, preserve whatever HTTP client is already set
-        // on `info` (important for mock-based unit tests).
-        let no_redirect_client = if xet_enabled {
-            let standard = HttpClient::with(build_reqwest(reqwest::redirect::Policy::default())?);
-            let no_redirect = HttpClient::with(build_reqwest(reqwest::redirect::Policy::none())?);
-            info.update_http_client(|_| standard);
-            no_redirect
-        } else {
-            info.http_client()
-        };
+        let standard = HttpClient::with(build_reqwest(reqwest::redirect::Policy::default())?);
+        let no_redirect = HttpClient::with(build_reqwest(reqwest::redirect::Policy::none())?);
+        info.update_http_client(|_| standard);
 
-        Ok(Self {
+        Ok(Self::new(
             info,
             repo,
             root,
             token,
             endpoint,
             max_retries,
-            xet_enabled,
-            no_redirect_client,
-        })
+            no_redirect,
+        ))
     }
 
     /// Build an authenticated HTTP request.
@@ -620,9 +625,8 @@ pub(crate) mod test_utils {
             None,
             endpoint.to_string(),
             3,
-            false,
-        )
-        .unwrap();
+            HttpClient::with(mock_client.clone()),
+        );
 
         (core, mock_client)
     }

core/services/hf/src/reader.rs

@@ -38,26 +38,19 @@ pub enum HfReader {
 impl HfReader {
     /// Create a reader, automatically choosing between XET and HTTP.
     ///
-    /// Buckets always use XET. For other repo types, when XET is enabled
-    /// a HEAD request probes for the `X-Xet-Hash` header. Files stored on
-    /// XET are downloaded via the CAS protocol; all others fall back to HTTP GET.
+    /// Buckets always use XET. For other repo types, a HEAD request
+    /// probes for the `X-Xet-Hash` header. Files stored on XET are
+    /// downloaded via the CAS protocol; all others fall back to HTTP GET.
     pub async fn try_new(core: &HfCore, path: &str, range: BytesRange) -> Result<Self> {
-        if core.xet_enabled {
-            // Buckets always use XET
-            if core.repo.repo_type == RepoType::Bucket {
-                if let Some(xet_file) = core.maybe_xet_file(path).await? {
-                    return Self::try_new_xet(core, &xet_file, range).await;
-                }
-                return Err(Error::new(
-                    ErrorKind::Unexpected,
-                    "bucket file is missing XET metadata",
-                ));
-            }
+        if let Some(xet_file) = core.maybe_xet_file(path).await? {
+            return Self::try_new_xet(core, &xet_file, range).await;
+        }
 
-            // For other repos, probe for XET
-            if let Some(xet_file) = core.maybe_xet_file(path).await? {
-                return Self::try_new_xet(core, &xet_file, range).await;
-            }
+        if core.repo.repo_type == RepoType::Bucket {
+            return Err(Error::new(
+                ErrorKind::Unexpected,
+                "bucket file is missing XET metadata",
+            ));
         }
 
         Self::try_new_http(core, path, range).await
@@ -126,7 +119,6 @@ impl oio::Read for HfReader {
 
 #[cfg(test)]
 mod tests {
-    use super::super::backend::test_utils::mbpp_xet_operator;
     use super::super::backend::test_utils::{gpt2_operator, mbpp_operator};
 
     /// Parquet magic bytes: "PAR1"
@@ -155,7 +147,7 @@ mod tests {
     #[tokio::test]
     #[ignore = "requires network access"]
     async fn test_read_xet_parquet() {
-        let op = mbpp_xet_operator();
+        let op = mbpp_operator();
         let data = op
             .read("full/train-00000-of-00001.parquet")
             .await
@@ -169,7 +161,7 @@ mod tests {
     #[tokio::test]
     #[ignore = "requires network access"]
     async fn test_read_xet_range() {
-        let op = mbpp_xet_operator();
+        let op = mbpp_operator();
         let data = op
             .read_with("full/train-00000-of-00001.parquet")
             .range(0..4)

core/services/hf/src/writer.rs

@@ -47,12 +47,6 @@ impl HfWriter {
     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 {
-            if !core.xet_enabled {
-                return Err(Error::new(
-                    ErrorKind::Unsupported,
-                    "buckets require XET to be enabled",
-                ));
-            }
             let client = core.xet_client("write").await?;
             let writer = client.write(None).await.map_err(map_xet_error)?;
             return Ok(HfWriter::Xet {
@@ -65,19 +59,13 @@ impl HfWriter {
         let mode_str = core.determine_upload_mode(&path).await?;
 
         if mode_str == "lfs" {
-            if core.xet_enabled {
-                let client = core.xet_client("write").await?;
-                let writer = client.write(None).await.map_err(map_xet_error)?;
-                return Ok(HfWriter::Xet {
-                    core,
-                    path,
-                    writer: Mutex::new(writer),
-                });
-            }
-            return Err(Error::new(
-                ErrorKind::Unsupported,
-                "file requires LFS; call enable_xet() on the builder for large file support",
-            ));
+            let client = core.xet_client("write").await?;
+            let writer = client.write(None).await.map_err(map_xet_error)?;
+            return Ok(HfWriter::Xet {
+                core,
+                path,
+                writer: Mutex::new(writer),
+            });
         }
 
         Ok(HfWriter::Regular {
@@ -191,8 +179,8 @@ impl oio::Write for HfWriter {
 
 #[cfg(test)]
 mod tests {
+    use super::super::backend::test_utils::testing_bucket_operator;
     use super::super::backend::test_utils::testing_operator;
-    use super::super::backend::test_utils::{testing_bucket_operator, testing_xet_operator};
     use super::*;
     use base64::Engine;
 
@@ -247,7 +235,7 @@ mod tests {
     #[tokio::test]
     #[ignore]
     async fn test_write_xet() {
-        let op = testing_xet_operator();
+        let op = testing_operator();
         op.write("test-xet.bin", b"Binary data for XET test".as_slice())
             .await
             .expect("xet write should succeed");