(opendal) branch main updated: feat(copy): apply timeout to copy operations (#7594)

[xu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

Xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git


The following commit(s) were added to refs/heads/main by this push:
     new dc200b3c2 feat(copy): apply timeout to copy operations (#7594)
dc200b3c2 is described below

commit dc200b3c25a637f1e105cd7642c321c8fca6fd4b
Author: dentiny <de...@gmail.com>
AuthorDate: Fri May 22 07:47:48 2026 -0700

    feat(copy): apply timeout to copy operations (#7594)
    
    * feat(copy): apply timeout to copy operations
    
    * fmt
---
 core/layers/timeout/src/lib.rs | 53 +++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 52 insertions(+), 1 deletion(-)

diff --git a/core/layers/timeout/src/lib.rs b/core/layers/timeout/src/lib.rs
index 9a24ae718..004c6d8c2 100644
--- a/core/layers/timeout/src/lib.rs
+++ b/core/layers/timeout/src/lib.rs
@@ -213,7 +213,7 @@ impl<A: Access> LayeredAccess for TimeoutAccessor<A> {
     type Writer = TimeoutWrapper<A::Writer>;
     type Lister = TimeoutWrapper<A::Lister>;
     type Deleter = TimeoutWrapper<A::Deleter>;
-    type Copier = A::Copier;
+    type Copier = TimeoutWrapper<A::Copier>;
 
     fn inner(&self) -> &Self::Inner {
         &self.inner
@@ -248,6 +248,7 @@ impl<A: Access> LayeredAccess for TimeoutAccessor<A> {
             self.inner.copy(from, to, args, opts.clone()),
         )
         .await
+        .map(|(rp, c)| (rp, TimeoutWrapper::new(c, self.io_timeout)))
     }
 
     async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
@@ -369,6 +370,18 @@ impl<R: oio::Delete> oio::Delete for TimeoutWrapper<R> {
     }
 }
 
+impl<C: oio::Copy> oio::Copy for TimeoutWrapper<C> {
+    async fn next(&mut self) -> Result<Option<usize>> {
+        let fut = self.inner.next();
+        Self::io_timeout(self.timeout, Operation::Copy.into_static(), fut).await
+    }
+
+    async fn abort(&mut self) -> Result<()> {
+        let fut = self.inner.abort();
+        Self::io_timeout(self.timeout, Operation::Copy.into_static(), fut).await
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::future::pending;
@@ -415,6 +428,16 @@ mod tests {
         async fn list(&self, _: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
             Ok((RpList::default(), Box::new(MockLister)))
         }
+
+        async fn copy(
+            &self,
+            _: &str,
+            _: &str,
+            _: OpCopy,
+            _: OpCopier,
+        ) -> Result<(RpCopy, Self::Copier)> {
+            Ok((RpCopy::default(), Box::new(MockCopier)))
+        }
     }
 
     #[derive(Debug, Clone, Default)]
@@ -448,6 +471,19 @@ mod tests {
         }
     }
 
+    #[derive(Debug, Clone, Default)]
+    struct MockCopier;
+
+    impl oio::Copy for MockCopier {
+        fn next(&mut self) -> impl Future<Output = Result<Option<usize>>> {
+            pending()
+        }
+
+        fn abort(&mut self) -> impl Future<Output = Result<()>> {
+            pending()
+        }
+    }
+
     #[tokio::test]
     async fn test_operation_timeout() {
         let srv = MockService;
@@ -513,6 +549,21 @@ mod tests {
         assert!(err.to_string().contains("timeout"));
     }
 
+    #[tokio::test]
+    async fn test_copy_io_timeout() {
+        use oio::Copy;
+
+        let acc = TimeoutLayer::default()
+            .with_io_timeout(Duration::from_millis(100))
+            .layer(MockService);
+        let (_, mut copier) = Access::copy(&acc, "f", "t", OpCopy::default(), OpCopier::default())
+            .await
+            .unwrap();
+
+        let err = copier.next().await.unwrap_err();
+        assert!(err.to_string().contains("timeout"));
+    }
+
     #[tokio::test]
     async fn test_list_timeout_raw() {
         use oio::List;