refactor(hf): cache TranslatorConfig instead of sessions, serialize integration tests

- Cache Arc<TranslatorConfig> per read/write direction; create a fresh
  FileUploadSession/FileDownloadSession per operation so that finalize()
  on one write doesn't poison the next
- Factor repeated config construction into xet_config(token_type)
- Mark all write/delete integration tests with #[serial] to prevent
  412 "commit has happened since" races on the shared test repo
- Add serial_test dev-dependency

Author: kszucs

core/Cargo.lock

@@ -53,4 +53,5 @@ opendal-core = { path = "../../core", version = "0.55.0", features = [
   "reqwest-rustls-tls",
 ] }
 serde_json = { workspace = true }
+serial_test = "3"
 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }

core/services/hf/Cargo.toml

@@ -31,6 +31,7 @@ use xet_client::cas_client::auth::TokenRefresher;
 use xet_data::processing::FileDownloadSession;
 use xet_data::processing::FileUploadSession;
 use xet_data::processing::XetFileInfo;
+use xet_data::processing::configurations::TranslatorConfig;
 
 use super::error::parse_error;
 use super::uri::HfRepo;
@@ -225,8 +226,8 @@ pub struct HfCore {
     /// inspect headers on 302 responses.
     pub no_redirect_client: HttpClient,
 
-    download_session: Arc<OnceCell<Arc<FileDownloadSession>>>,
-    upload_session: Arc<OnceCell<Arc<FileUploadSession>>>,
+    upload_config: Arc<OnceCell<Arc<TranslatorConfig>>>,
+    download_config: Arc<OnceCell<Arc<TranslatorConfig>>>,
 }
 
 impl Debug for HfCore {
@@ -255,8 +256,8 @@ impl HfCore {
             token,
             endpoint,
             no_redirect_client,
-            download_session: Arc::new(OnceCell::new()),
-            upload_session: Arc::new(OnceCell::new()),
+            upload_config: Arc::new(OnceCell::new()),
+            download_config: Arc::new(OnceCell::new()),
         }
     }
 
@@ -351,58 +352,54 @@ impl HfCore {
         Ok(token)
     }
 
-    pub(super) async fn xet_upload_session(&self) -> Result<Arc<FileUploadSession>> {
-        self.upload_session
-            .get_or_try_init(|| async {
-                let cas_url = self.xet_token("write").await?.cas_url;
-                let refresher = Arc::new(XetTokenRefresher::new(self, "write"));
-                let config = xet_data::processing::data_client::default_config(
-                    cas_url,
-                    None,
-                    Some(refresher as Arc<dyn TokenRefresher>),
-                    None,
-                )
-                .map_err(|err| {
-                    Error::new(ErrorKind::Unexpected, "failed to create upload config")
-                        .set_source(err)
-                })?;
-                FileUploadSession::new(Arc::new(config))
-                    .await
-                    .map_err(|err| {
-                        Error::new(ErrorKind::Unexpected, "failed to create upload session")
-                            .set_source(err)
-                    })
-            })
+    async fn xet_config(&self, token_type: &'static str) -> Result<Arc<TranslatorConfig>> {
+        let cas_url = self.xet_token(token_type).await?.cas_url;
+        let refresher = Arc::new(XetTokenRefresher::new(self, token_type));
+        xet_data::processing::data_client::default_config(
+            cas_url,
+            None,
+            Some(refresher as Arc<dyn TokenRefresher>),
+            None,
+        )
+        .map(Arc::new)
+        .map_err(|err| {
+            Error::new(ErrorKind::Unexpected, "failed to create xet config").set_source(err)
+        })
+    }
+
+    async fn upload_config(&self) -> Result<Arc<TranslatorConfig>> {
+        let this = self.clone();
+        self.upload_config
+            .get_or_try_init(|| async move { this.xet_config("write").await })
             .await
             .map(Arc::clone)
     }
 
-    pub(super) async fn xet_download_session(&self) -> Result<Arc<FileDownloadSession>> {
-        self.download_session
-            .get_or_try_init(|| async {
-                let cas_url = self.xet_token("read").await?.cas_url;
-                let refresher = Arc::new(XetTokenRefresher::new(self, "read"));
-                let config = xet_data::processing::data_client::default_config(
-                    cas_url,
-                    None,
-                    Some(refresher as Arc<dyn TokenRefresher>),
-                    None,
-                )
-                .map_err(|err| {
-                    Error::new(ErrorKind::Unexpected, "failed to create download config")
-                        .set_source(err)
-                })?;
-                FileDownloadSession::new(Arc::new(config))
-                    .await
-                    .map_err(|err| {
-                        Error::new(ErrorKind::Unexpected, "failed to create download session")
-                            .set_source(err)
-                    })
-            })
+    async fn download_config(&self) -> Result<Arc<TranslatorConfig>> {
+        let this = self.clone();
+        self.download_config
+            .get_or_try_init(|| async move { this.xet_config("read").await })
             .await
             .map(Arc::clone)
     }
 
+    pub(super) async fn xet_upload_session(&self) -> Result<Arc<FileUploadSession>> {
+        FileUploadSession::new(self.upload_config().await?)
+            .await
+            .map_err(|err| {
+                Error::new(ErrorKind::Unexpected, "failed to create upload session").set_source(err)
+            })
+    }
+
+    pub(super) async fn xet_download_session(&self) -> Result<Arc<FileDownloadSession>> {
+        FileDownloadSession::new(self.download_config().await?)
+            .await
+            .map_err(|err| {
+                Error::new(ErrorKind::Unexpected, "failed to create download session")
+                    .set_source(err)
+            })
+    }
+
     /// Issue a HEAD request and extract XET file info (hash and size).
     ///
     /// Returns `None` if the `X-Xet-Hash` header is absent or empty.

core/services/hf/src/core.rs

@@ -63,9 +63,11 @@ impl oio::BatchDelete for HfDeleter {
 mod tests {
     use super::super::backend::test_utils::testing_operator;
     use opendal_core::*;
+    use serial_test::serial;
 
     #[tokio::test]
     #[ignore]
+    #[serial]
     async fn test_delete_nonexistent() {
         let op = testing_operator();
 
@@ -76,6 +78,7 @@ mod tests {
 
     #[tokio::test]
     #[ignore]
+    #[serial]
     async fn test_delete_then_read() {
         let op = testing_operator();
         let path = "tests/delete-then-read.txt";
@@ -95,6 +98,7 @@ mod tests {
 
     #[tokio::test]
     #[ignore]
+    #[serial]
     async fn test_batch_delete() {
         let op = testing_operator();
         let paths = [

core/services/hf/src/deleter.rs

@@ -95,11 +95,13 @@ impl oio::Write for HfWriter {
             }
             HfWriter::Xet { cleaner, .. } => {
                 // get_mut() required here: lock() would hold a MutexGuard across .await
-                let cleaner = cleaner.get_mut().expect("mutex poisoned")
-                    .as_mut().expect("cleaner missing");
+                let cleaner = cleaner
+                    .get_mut()
+                    .expect("mutex poisoned")
+                    .as_mut()
+                    .expect("cleaner missing");
                 cleaner.add_data(&bs.to_bytes()).await.map_err(|err| {
-                    Error::new(ErrorKind::Unexpected, "failed to write xet chunk")
-                        .set_source(err)
+                    Error::new(ErrorKind::Unexpected, "failed to write xet chunk").set_source(err)
                 })
             }
         }
@@ -136,8 +138,8 @@ impl oio::Write for HfWriter {
                     .expect("mutex poisoned")
                     .take()
                     .ok_or_else(|| {
-                    Error::new(ErrorKind::Unexpected, "xet writer already closed")
-                })?;
+                        Error::new(ErrorKind::Unexpected, "xet writer already closed")
+                    })?;
 
                 let (file_info, _metrics) = cleaner.finish().await.map_err(|err| {
                     Error::new(ErrorKind::Unexpected, "failed to finish xet upload").set_source(err)
@@ -211,6 +213,7 @@ mod tests {
     use super::super::backend::test_utils::testing_operator;
     use super::*;
     use base64::Engine;
+    use serial_test::serial;
 
     // --- Unit tests (no network required) ---
 
@@ -243,6 +246,7 @@ mod tests {
 
     #[tokio::test]
     #[ignore]
+    #[serial]
     async fn test_write_http() {
         let op = testing_operator();
         op.write("test-file.txt", b"Hello, HuggingFace!".as_slice())
@@ -252,6 +256,7 @@ mod tests {
 
     #[tokio::test]
     #[ignore]
+    #[serial]
     async fn test_write_http_with_content_type() {
         let op = testing_operator();
         op.write_with("test.json", br#"{"test": "data"}"#.as_slice())
@@ -262,6 +267,7 @@ mod tests {
 
     #[tokio::test]
     #[ignore]
+    #[serial]
     async fn test_write_xet() {
         let op = testing_operator();
         op.write("test-xet.bin", b"Binary data for XET test".as_slice())
@@ -273,6 +279,7 @@ mod tests {
     /// in commit) and verify the content roundtrips correctly.
     #[tokio::test]
     #[ignore]
+    #[serial]
     async fn test_write_regular_roundtrip() {
         let op = testing_operator();
         let path = "tests/regular-roundtrip.txt";
@@ -292,6 +299,7 @@ mod tests {
     /// Verify stat returns correct metadata after writing.
     #[tokio::test]
     #[ignore]
+    #[serial]
     async fn test_write_and_stat() {
         let op = testing_operator();
         let path = "tests/stat-after-write.txt";
@@ -310,6 +318,7 @@ mod tests {
     /// Overwriting an existing file should replace its content.
     #[tokio::test]
     #[ignore]
+    #[serial]
     async fn test_write_overwrite() {
         let op = testing_operator();
         let path = "tests/overwrite-test.txt";
@@ -331,6 +340,7 @@ mod tests {
     /// Full lifecycle: write → stat → read → delete → confirm gone.
     #[tokio::test]
     #[ignore]
+    #[serial]
     async fn test_write_delete_lifecycle() {
         let op = testing_operator();
         let path = "tests/lifecycle-test.txt";
@@ -351,6 +361,7 @@ mod tests {
     /// Write an empty (0-byte) file and verify roundtrip.
     #[tokio::test]
     #[ignore]
+    #[serial]
     async fn test_write_empty_file_roundtrip() {
         let op = testing_operator();
         let path = "tests/empty-file.txt";
@@ -372,6 +383,7 @@ mod tests {
     /// HuggingFace creates intermediate directories implicitly.
     #[tokio::test]
     #[ignore]
+    #[serial]
     async fn test_write_nested_directory() {
         let op = testing_operator();
         let path = "tests/deep/nested/dir/file.txt";
@@ -390,6 +402,7 @@ mod tests {
     /// Write a file with special characters in the path.
     #[tokio::test]
     #[ignore]
+    #[serial]
     async fn test_write_special_characters_in_path() {
         let op = testing_operator();
         let path = "tests/special chars (1).txt";
@@ -409,6 +422,7 @@ mod tests {
 
     #[tokio::test]
     #[ignore]
+    #[serial]
     async fn test_bucket_write() {
         let op = testing_bucket_operator();
         let path = "test-bucket-file.txt";
@@ -426,6 +440,7 @@ mod tests {
 
     #[tokio::test]
     #[ignore]
+    #[serial]
     async fn test_bucket_write_roundtrip() {
         let op = testing_bucket_operator();
         let path = "tests/bucket-roundtrip.bin";
@@ -446,6 +461,7 @@ mod tests {
 
     #[tokio::test]
     #[ignore]
+    #[serial]
     async fn test_bucket_overwrite() {
         let op = testing_bucket_operator();
         let path = "tests/bucket-overwrite.txt";