opendal_layer_timeout/
lib.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18#![doc = include_str!("../README.md")]
19#![cfg_attr(docsrs, feature(doc_cfg))]
20#![cfg_attr(docsrs, doc(auto_cfg))]
21#![deny(missing_docs)]
22use std::future::Future;
23use std::sync::Arc;
24use std::time::Duration;
25
26use opendal_core::raw::*;
27use opendal_core::*;
28
29/// `TimeoutLayer` adds deadlines to operations so slow or stalled work cannot
30/// hang indefinitely.
31///
32/// For example, a dead connection could stall a database query. `TimeoutLayer`
33/// interrupts the operation and returns an error that users can retry or report.
34///
35/// # Notes
36///
37/// `TimeoutLayer` applies two timeout budgets:
38///
39/// - `timeout` bounds control operations such as `stat`, `create_dir`, `rename`,
40///   and `presign`.
41/// - `io_timeout` bounds operations that open IO bodies, such as `read`, `write`,
42///   and `list`, and every method call on returned readers, writers, listers,
43///   deleters, and copiers.
44///
45/// # Default
46///
47/// - timeout: 60 seconds
48/// - io_timeout: 10 seconds
49///
50/// # Cancellation Safety
51///
52/// `TimeoutLayer` enforces deadlines by dropping the in-flight future when a
53/// timeout is reached. This can break lower layers that rely on a future being
54/// resolved to restore internal state.
55///
56/// For example, while using `TimeoutLayer` with `RetryLayer` at the same time,
57/// please make sure timeout layer is added before retry layer.
58///
59/// ```no_run
60/// # use std::time::Duration;
61/// #
62/// # use opendal_core::services;
63/// # use opendal_core::Operator;
64/// # use opendal_core::Result;
65/// # use opendal_layer_retry::RetryLayer;
66/// # use opendal_layer_timeout::TimeoutLayer;
67/// #
68/// # fn main() -> Result<()> {
69/// let op = Operator::new(services::Memory::default())?
70///     // This is fine: each retry attempt is timed out.
71///     .layer(TimeoutLayer::default().with_io_timeout(Duration::from_nanos(1)))
72///     .layer(RetryLayer::default())
73///     // This is wrong: timeout can drop RetryLayer's future before it restores body state.
74///     .layer(TimeoutLayer::default().with_io_timeout(Duration::from_nanos(1)));
75/// # Ok(())
76/// # }
77/// ```
78///
79/// # Examples
80///
81/// The following example creates a timeout layer with a 10-second timeout for
82/// control operations and a 3-second timeout for IO operations.
83///
84/// ```no_run
85/// # use std::time::Duration;
86/// #
87/// # use opendal_core::services;
88/// # use opendal_core::Operator;
89/// # use opendal_core::Result;
90/// # use opendal_layer_timeout::TimeoutLayer;
91/// #
92/// # fn main() -> Result<()> {
93/// let _ = Operator::new(services::Memory::default())?
94///     .layer(
95///         TimeoutLayer::default()
96///             .with_timeout(Duration::from_secs(10))
97///             .with_io_timeout(Duration::from_secs(3)),
98///     );
99/// # Ok(())
100/// # }
101/// ```
102///
103/// # Implementation Notes
104///
105/// `TimeoutLayer` uses [`tokio::time::timeout`] to bound service calls and IO
106/// body methods. It also supplies an executor timeout so concurrent block write
107/// and copy tasks can fail instead of waiting forever.
108///
109/// This introduces a small amount of overhead for IO operations, but it is needed
110/// to implement timeouts correctly. OpenDAL used to implement this as a
111/// zero-cost deadline check that only stored an [`Instant`] and compared it with
112/// the current time. However, that approach does not work for all cases.
113///
114/// For example, a user's TCP connection could enter the
115/// [Busy ESTAB](https://blog.cloudflare.com/when-tcp-sockets-refuse-to-die)
116/// state. In this state, the connection emits no IO events, so the runtime never
117/// polls the future again. The future hangs until Linux closes the connection
118/// after it reaches the
119/// [net.ipv4.tcp_retries2](https://man7.org/linux/man-pages/man7/tcp.7.html)
120/// limit.
121#[derive(Clone, Debug)]
122pub struct TimeoutLayer {
123    timeout: Duration,
124    io_timeout: Duration,
125}
126
127impl Default for TimeoutLayer {
128    fn default() -> Self {
129        Self {
130            timeout: Duration::from_secs(60),
131            io_timeout: Duration::from_secs(10),
132        }
133    }
134}
135
136impl TimeoutLayer {
137    /// Create a new [`TimeoutLayer`] with default settings.
138    pub fn new() -> Self {
139        Self::default()
140    }
141
142    /// Set the timeout for control operations.
143    ///
144    /// This timeout is for all non-io operations like `stat`, `delete`.
145    pub fn with_timeout(mut self, timeout: Duration) -> Self {
146        self.timeout = timeout;
147        self
148    }
149
150    /// Set the timeout for IO operations and body methods.
151    ///
152    /// This timeout is for all io operations like `read`, `Reader::read` and `Writer::write`.
153    pub fn with_io_timeout(mut self, timeout: Duration) -> Self {
154        self.io_timeout = timeout;
155        self
156    }
157}
158
159impl Layer for TimeoutLayer {
160    fn apply_service(&self, inner: Servicer) -> Servicer {
161        Arc::new(self.layer(inner))
162    }
163
164    fn apply_context(&self, _srv: Servicer, inner: OperationContext) -> OperationContext {
165        // Concurrent block IO paths read this timeout from the operation context's executor.
166        let executor = Executor::with(TimeoutExecutor::new(
167            inner.executor().clone().into_inner(),
168            self.io_timeout,
169        ));
170        inner.with_executor(executor)
171    }
172}
173
174impl TimeoutLayer {
175    fn layer(&self, inner: Servicer) -> TimeoutService {
176        TimeoutService {
177            inner,
178            timeout: self.timeout,
179            io_timeout: self.io_timeout,
180        }
181    }
182}
183
184#[doc(hidden)]
185#[derive(Debug)]
186pub struct TimeoutService {
187    inner: Servicer,
188    timeout: Duration,
189    io_timeout: Duration,
190}
191
192impl TimeoutService {
193    async fn timeout<F: Future<Output = Result<T>>, T>(&self, op: Operation, fut: F) -> Result<T> {
194        tokio::time::timeout(self.timeout, fut).await.map_err(|_| {
195            Error::new(ErrorKind::Unexpected, "operation timeout reached")
196                .with_operation(op)
197                .with_context("timeout", self.timeout.as_secs_f64().to_string())
198                .set_temporary()
199        })?
200    }
201}
202
203impl Service for TimeoutService {
204    type Reader = TimeoutWrapper<oio::Reader>;
205    type Writer = TimeoutWrapper<oio::Writer>;
206    type Lister = TimeoutWrapper<oio::Lister>;
207    type Deleter = TimeoutWrapper<oio::Deleter>;
208    type Copier = TimeoutWrapper<oio::Copier>;
209
210    fn info(&self) -> ServiceInfo {
211        self.inner.info()
212    }
213
214    fn capability(&self) -> Capability {
215        self.inner.capability()
216    }
217
218    async fn create_dir(
219        &self,
220        ctx: &OperationContext,
221        path: &str,
222        args: OpCreateDir,
223    ) -> Result<RpCreateDir> {
224        self.timeout(Operation::CreateDir, self.inner.create_dir(ctx, path, args))
225            .await
226    }
227
228    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
229        self.inner
230            .read(ctx, path, args)
231            .map(|r| TimeoutWrapper::new(r, self.io_timeout))
232    }
233
234    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
235        self.inner
236            .write(ctx, path, args)
237            .map(|r| TimeoutWrapper::new(r, self.io_timeout))
238    }
239
240    fn copy(
241        &self,
242        ctx: &OperationContext,
243        from: &str,
244        to: &str,
245        args: OpCopy,
246        opts: OpCopier,
247    ) -> Result<Self::Copier> {
248        self.inner
249            .copy(ctx, from, to, args, opts)
250            .map(|c| TimeoutWrapper::new(c, self.io_timeout))
251    }
252
253    async fn rename(
254        &self,
255        ctx: &OperationContext,
256        from: &str,
257        to: &str,
258        args: OpRename,
259    ) -> Result<RpRename> {
260        self.timeout(Operation::Rename, self.inner.rename(ctx, from, to, args))
261            .await
262    }
263
264    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
265        self.timeout(Operation::Stat, self.inner.stat(ctx, path, args))
266            .await
267    }
268
269    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
270        self.inner
271            .delete(ctx)
272            .map(|r| TimeoutWrapper::new(r, self.io_timeout))
273    }
274
275    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
276        self.inner
277            .list(ctx, path, args)
278            .map(|r| TimeoutWrapper::new(r, self.io_timeout))
279    }
280
281    async fn presign(
282        &self,
283        ctx: &OperationContext,
284        path: &str,
285        args: OpPresign,
286    ) -> Result<RpPresign> {
287        self.timeout(Operation::Presign, self.inner.presign(ctx, path, args))
288            .await
289    }
290}
291
292struct TimeoutExecutor {
293    exec: Arc<dyn Execute>,
294    timeout: Duration,
295}
296
297impl TimeoutExecutor {
298    fn new(exec: Arc<dyn Execute>, timeout: Duration) -> Self {
299        Self { exec, timeout }
300    }
301}
302
303impl Execute for TimeoutExecutor {
304    fn execute(&self, f: BoxedStaticFuture<()>) {
305        self.exec.execute(f)
306    }
307
308    fn timeout(&self) -> Option<BoxedStaticFuture<()>> {
309        Some(Box::pin(tokio::time::sleep(self.timeout)))
310    }
311}
312
313#[doc(hidden)]
314pub struct TimeoutWrapper<R> {
315    inner: R,
316
317    timeout: Duration,
318}
319
320impl<R> TimeoutWrapper<R> {
321    fn new(inner: R, timeout: Duration) -> Self {
322        Self { inner, timeout }
323    }
324
325    #[inline]
326    async fn io_timeout<F: Future<Output = Result<T>>, T>(
327        timeout: Duration,
328        op: &'static str,
329        fut: F,
330    ) -> Result<T> {
331        tokio::time::timeout(timeout, fut).await.map_err(|_| {
332            Error::new(ErrorKind::Unexpected, "io operation timeout reached")
333                .with_operation(op)
334                .with_context("timeout", timeout.as_secs_f64().to_string())
335                .set_temporary()
336        })?
337    }
338}
339
340impl<R: oio::ReadStream> oio::ReadStream for TimeoutWrapper<R> {
341    async fn read(&mut self) -> Result<Buffer> {
342        let fut = self.inner.read();
343        Self::io_timeout(self.timeout, Operation::Read.into_static(), fut).await
344    }
345}
346
347impl<R: oio::Read> oio::Read for TimeoutWrapper<R> {
348    async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
349        let (rp, stream) = Self::io_timeout(
350            self.timeout,
351            Operation::Read.into_static(),
352            self.inner.open(range),
353        )
354        .await?;
355        Ok((
356            rp,
357            Box::new(TimeoutWrapper::new(stream, self.timeout)) as Box<dyn oio::ReadStreamDyn>,
358        ))
359    }
360
361    async fn read(&self, range: BytesRange) -> Result<(RpRead, Buffer)> {
362        Self::io_timeout(
363            self.timeout,
364            Operation::Read.into_static(),
365            self.inner.read(range),
366        )
367        .await
368    }
369}
370
371impl<R: oio::Write> oio::Write for TimeoutWrapper<R> {
372    async fn write(&mut self, bs: Buffer) -> Result<()> {
373        let fut = self.inner.write(bs);
374        Self::io_timeout(self.timeout, Operation::Write.into_static(), fut).await
375    }
376
377    async fn close(&mut self) -> Result<Metadata> {
378        let fut = self.inner.close();
379        Self::io_timeout(self.timeout, Operation::Write.into_static(), fut).await
380    }
381
382    async fn abort(&mut self) -> Result<()> {
383        let fut = self.inner.abort();
384        Self::io_timeout(self.timeout, Operation::Write.into_static(), fut).await
385    }
386}
387
388impl<R: oio::List> oio::List for TimeoutWrapper<R> {
389    async fn next(&mut self) -> Result<Option<oio::Entry>> {
390        let fut = self.inner.next();
391        Self::io_timeout(self.timeout, Operation::List.into_static(), fut).await
392    }
393}
394
395impl<R: oio::Delete> oio::Delete for TimeoutWrapper<R> {
396    async fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
397        let fut = self.inner.delete(path, args);
398        Self::io_timeout(self.timeout, Operation::Delete.into_static(), fut).await
399    }
400
401    async fn close(&mut self) -> Result<()> {
402        let fut = self.inner.close();
403        Self::io_timeout(self.timeout, Operation::Delete.into_static(), fut).await
404    }
405}
406
407impl<C: oio::Copy> oio::Copy for TimeoutWrapper<C> {
408    async fn next(&mut self) -> Result<Option<usize>> {
409        let fut = self.inner.next();
410        Self::io_timeout(self.timeout, Operation::Copy.into_static(), fut).await
411    }
412
413    async fn close(&mut self) -> Result<Metadata> {
414        let fut = self.inner.close();
415        Self::io_timeout(self.timeout, Operation::Copy.into_static(), fut).await
416    }
417
418    async fn abort(&mut self) -> Result<()> {
419        let fut = self.inner.abort();
420        Self::io_timeout(self.timeout, Operation::Copy.into_static(), fut).await
421    }
422}
423
424#[cfg(test)]
425mod tests {
426    use std::future::pending;
427
428    use futures::StreamExt;
429    use tokio::time::timeout;
430
431    use super::*;
432
433    #[derive(Debug, Clone, Default)]
434    struct MockService;
435
436    impl Service for MockService {
437        type Reader = MockReader;
438        type Writer = ();
439        type Lister = MockLister;
440        type Deleter = MockDeleter;
441        type Copier = MockCopier;
442
443        fn info(&self) -> ServiceInfo {
444            ServiceInfo::with_scheme("mock")
445        }
446
447        fn capability(&self) -> Capability {
448            Capability {
449                read: true,
450                delete: true,
451                list: true,
452                copy: true,
453                ..Default::default()
454            }
455        }
456
457        async fn create_dir(
458            &self,
459            _: &OperationContext,
460            _: &str,
461            _: OpCreateDir,
462        ) -> Result<RpCreateDir> {
463            Err(Error::new(
464                ErrorKind::Unsupported,
465                "operation is not supported",
466            ))
467        }
468
469        async fn stat(&self, _: &OperationContext, _: &str, _: OpStat) -> Result<RpStat> {
470            Err(Error::new(
471                ErrorKind::Unsupported,
472                "operation is not supported",
473            ))
474        }
475
476        /// Return a reader whose operations never complete.
477        fn read(&self, _ctx: &OperationContext, _: &str, _: OpRead) -> Result<Self::Reader> {
478            Ok(MockReader)
479        }
480
481        fn write(&self, _ctx: &OperationContext, _: &str, _: OpWrite) -> Result<Self::Writer> {
482            Err(Error::new(
483                ErrorKind::Unsupported,
484                "operation is not supported",
485            ))
486        }
487
488        /// Return a deleter whose operations never complete.
489        fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
490            Ok(MockDeleter)
491        }
492
493        fn list(&self, _ctx: &OperationContext, _: &str, _: OpList) -> Result<Self::Lister> {
494            Ok(MockLister)
495        }
496
497        fn copy(
498            &self,
499            _: &OperationContext,
500            _: &str,
501            _: &str,
502            _: OpCopy,
503            _: OpCopier,
504        ) -> Result<Self::Copier> {
505            Ok(MockCopier)
506        }
507
508        async fn rename(
509            &self,
510            _: &OperationContext,
511            _: &str,
512            _: &str,
513            _: OpRename,
514        ) -> Result<RpRename> {
515            Err(Error::new(
516                ErrorKind::Unsupported,
517                "operation is not supported",
518            ))
519        }
520
521        async fn presign(&self, _: &OperationContext, _: &str, _: OpPresign) -> Result<RpPresign> {
522            Err(Error::new(
523                ErrorKind::Unsupported,
524                "operation is not supported",
525            ))
526        }
527    }
528
529    #[derive(Debug, Clone, Default)]
530    struct MockReader;
531
532    impl oio::Read for MockReader {
533        async fn open(&self, _: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
534            pending().await
535        }
536
537        async fn read(&self, _: BytesRange) -> Result<(RpRead, Buffer)> {
538            pending().await
539        }
540    }
541
542    #[derive(Debug, Clone, Default)]
543    struct MockLister;
544
545    impl oio::List for MockLister {
546        async fn next(&mut self) -> Result<Option<oio::Entry>> {
547            pending().await
548        }
549    }
550
551    #[derive(Debug, Clone, Default)]
552    struct MockDeleter;
553
554    impl oio::Delete for MockDeleter {
555        async fn delete(&mut self, _: &str, _: OpDelete) -> Result<()> {
556            pending().await
557        }
558
559        async fn close(&mut self) -> Result<()> {
560            Ok(())
561        }
562    }
563
564    #[derive(Debug, Clone, Default)]
565    struct MockCopier;
566
567    impl oio::Copy for MockCopier {
568        async fn next(&mut self) -> Result<Option<usize>> {
569            pending().await
570        }
571
572        async fn close(&mut self) -> Result<Metadata> {
573            pending().await
574        }
575
576        async fn abort(&mut self) -> Result<()> {
577            pending().await
578        }
579    }
580
581    #[tokio::test]
582    async fn test_delete_timeout() {
583        let srv = MockService;
584        let op = Operator::from_parts(OperationContext::default(), Arc::new(srv))
585            .layer(TimeoutLayer::default().with_io_timeout(Duration::from_secs(1)));
586
587        let fut = async {
588            let res = op.delete("test").await;
589            assert!(res.is_err());
590            let err = res.unwrap_err();
591            assert_eq!(err.kind(), ErrorKind::Unexpected);
592            assert!(err.to_string().contains("timeout"))
593        };
594
595        timeout(Duration::from_secs(2), fut)
596            .await
597            .expect("this test should not exceed 2 seconds")
598    }
599
600    #[tokio::test]
601    async fn test_io_timeout() {
602        let srv = MockService;
603        let op = Operator::from_parts(OperationContext::default(), Arc::new(srv))
604            .layer(TimeoutLayer::default().with_io_timeout(Duration::from_secs(1)));
605
606        let reader = op.reader("test").await.unwrap();
607
608        let res = reader.read(0..4).await;
609        assert!(res.is_err());
610        let err = res.unwrap_err();
611        assert_eq!(err.kind(), ErrorKind::Unexpected);
612        assert!(err.to_string().contains("timeout"))
613    }
614
615    #[tokio::test]
616    async fn test_list_timeout() {
617        let srv = MockService;
618        let op = Operator::from_parts(OperationContext::default(), Arc::new(srv)).layer(
619            TimeoutLayer::default()
620                .with_timeout(Duration::from_secs(1))
621                .with_io_timeout(Duration::from_secs(1)),
622        );
623
624        let mut lister = op.lister("test").await.unwrap();
625
626        let res = lister.next().await.unwrap();
627        assert!(res.is_err());
628        let err = res.unwrap_err();
629        assert_eq!(err.kind(), ErrorKind::Unexpected);
630        assert!(err.to_string().contains("timeout"))
631    }
632
633    #[tokio::test]
634    async fn test_delete_io_timeout() {
635        use oio::Delete;
636
637        let mut deleter = TimeoutWrapper::new(MockDeleter, Duration::from_secs(1));
638
639        let res = deleter.delete("test", OpDelete::default()).await;
640        assert!(res.is_err());
641        let err = res.unwrap_err();
642        assert_eq!(err.kind(), ErrorKind::Unexpected);
643        assert!(err.to_string().contains("timeout"));
644    }
645
646    #[tokio::test]
647    async fn test_copy_io_timeout() {
648        use oio::Copy;
649
650        let service = TimeoutLayer::default()
651            .with_io_timeout(Duration::from_millis(100))
652            .apply_service(Arc::new(MockService));
653        let ctx = OperationContext::new();
654        let mut copier = service
655            .copy(&ctx, "f", "t", OpCopy::default(), OpCopier::default())
656            .unwrap();
657
658        let err = copier.next().await.unwrap_err();
659        assert!(err.to_string().contains("timeout"));
660    }
661
662    #[tokio::test]
663    async fn test_list_timeout_raw() {
664        use oio::List;
665
666        let timeout_layer = TimeoutLayer::default()
667            .with_timeout(Duration::from_secs(1))
668            .with_io_timeout(Duration::from_secs(1));
669        let service = timeout_layer.apply_service(Arc::new(MockService));
670        let ctx = OperationContext::new();
671
672        let mut lister = service.list(&ctx, "test", OpList::default()).unwrap();
673
674        let res = lister.next().await;
675        assert!(res.is_err());
676        let err = res.unwrap_err();
677        assert_eq!(err.kind(), ErrorKind::Unexpected);
678        assert!(err.to_string().contains("timeout"));
679    }
680}
opendal_layer_timeout/lib.rs

opendal_layer_timeout/
lib.rs
