fix: Don't apply blocking layer when service support blocking (#3050)

* fix: Don't apply blocking layer when service support blocking

Signed-off-by: Xuanwo <[email protected]>

* Update core/src/layers/blocking.rs

Co-authored-by: Suyan <[email protected]>

* Fix build

Signed-off-by: Xuanwo <[email protected]>

* Fix typo

Signed-off-by: Xuanwo <[email protected]>

* Fix multipart upload

Signed-off-by: Xuanwo <[email protected]>

* Fix clippy

Signed-off-by: Xuanwo <[email protected]>

---------

Signed-off-by: Xuanwo <[email protected]>
Co-authored-by: Suyan <[email protected]>

core/src/layers/blocking.rs

@@ -25,13 +25,11 @@ use crate::raw::oio::ReadExt;
 use crate::raw::*;
 use crate::*;
 
-/// Add blocking API support for every operations.
+/// Add blocking API support for non-blocking services.
 ///
-/// # Blocking API
+/// # Notes
 ///
-/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
-///
-/// Tracking issue: #2678
+/// Please only enable this layer when the underlying service does not support blocking.
 #[derive(Debug, Clone)]
 pub struct BlockingLayer {
     handle: Handle,

core/src/layers/retry.rs

@@ -998,7 +998,7 @@ impl<R: oio::Write, I: RetryInterceptor> oio::Write for RetryWrapper<R, I> {
                             &err,
                             dur,
                             &[
-                                ("operation", WriteOperation::Abort.into_static()),
+                                ("operation", WriteOperation::Close.into_static()),
                                 ("path", &self.path),
                             ],
                         );

core/src/raw/oio/write/multipart_upload_write.rs

@@ -195,6 +195,7 @@ where
                     let (w, part) = ready!(fut.as_mut().poll(cx));
                     self.state = State::Idle(Some(w));
                     self.parts.push(part?);
+
                     // Replace the cache when last write succeeded
                     let size = bs.remaining();
                     let cb = oio::ChunkedBytes::from_vec(bs.vectored_bytes(size));
@@ -264,8 +265,11 @@ where
                 State::Close(fut) => {
                     let (w, res) = futures::ready!(fut.as_mut().poll(cx));
                     self.state = State::Idle(Some(w));
+                    // We should check res first before clean up cache.
+                    res?;
+
                     self.cache = None;
-                    return Poll::Ready(res);
+                    return Poll::Ready(Ok(()));
                 }
                 State::Init(_) => unreachable!(
                     "MultipartUploadWriter must not go into State::Init during poll_close"

core/src/raw/oio/write/range_write.rs

@@ -274,8 +274,11 @@ impl<W: RangeWrite> oio::Write for RangeWriter<W> {
                 State::Abort(fut) => {
                     let (w, res) = ready!(fut.poll_unpin(cx));
                     self.state = State::Idle(Some(w));
+                    // We should check res first before clean up cache.
+                    res?;
+
                     self.buffer = None;
-                    return Poll::Ready(res);
+                    return Poll::Ready(Ok(()));
                 }
             }
         }

core/tests/behavior/utils.rs

@@ -82,14 +82,17 @@ pub fn init_service<B: Builder>() -> Option<Operator> {
         op.layer(ChaosLayer::new(0.1))
     };
 
-    let _guard = RUNTIME.enter();
-    let op = op
-        .layer(BlockingLayer::create().expect("blocking layer must be created"))
+    let mut op = op
         .layer(LoggingLayer::default().with_backtrace_output(true))
         .layer(TimeoutLayer::new())
         .layer(RetryLayer::new())
         .finish();
 
+    if !op.info().full_capability().blocking {
+        let _guard = RUNTIME.enter();
+        op = op.layer(BlockingLayer::create().expect("blocking layer must be created"))
+    }
+
     Some(op)
 }