feat(core)!: implement write returns metadata

bindings/cpp/src/async.rs

@@ -103,6 +103,7 @@ unsafe fn operator_write(op: ffi::OperatorPtr, path: String, bs: Vec<u8>) -> Rus
             .0
             .write(&path, bs)
             .await
+            .map(|_| ())
             .map_err(|e| CxxAsyncException::new(e.to_string().into_boxed_str()))?)
     })
 }

bindings/cpp/src/lib.rs

@@ -124,7 +124,7 @@ impl Operator {
     // Safety: The bytes created from bs will be dropped after the function call.
     // So it's safe to declare its lifetime as 'static.
     fn write(&self, path: &str, bs: &'static [u8]) -> Result<()> {
-        Ok(self.0.write(path, bs)?)
+        Ok(self.0.write(path, bs).map(|_| ())?)
     }
 
     fn exists(&self, path: &str) -> Result<bool> {

bindings/dotnet/src/lib.rs

@@ -61,7 +61,7 @@ pub unsafe extern "C" fn blocking_operator_write(
     let op = &*(op);
     let path = std::ffi::CStr::from_ptr(path).to_str().unwrap();
     let content = std::ffi::CStr::from_ptr(content).to_str().unwrap();
-    op.write(path, content.to_owned()).unwrap()
+    op.write(path, content.to_owned()).map(|_| ()).unwrap()
 }
 
 /// # Safety

bindings/haskell/src/lib.rs

@@ -206,7 +206,7 @@ pub unsafe extern "C" fn blocking_write(
     let bytes = Vec::from_raw_parts(bytes as *mut u8, len, len);
 
     let res = match op.write(path_str, bytes.clone()) {
-        Ok(()) => FFIResult::ok(()),
+        Ok(_) => FFIResult::ok(()),
         Err(e) => FFIResult::err_with_source("Failed to write", e),
     };
 

bindings/java/src/async_operator.rs

@@ -139,7 +139,7 @@ fn intern_write(
 }
 
 async fn do_write(op: &mut Operator, path: String, content: Vec<u8>) -> Result<()> {
-    Ok(op.write(&path, content).await?)
+    Ok(op.write(&path, content).await.map(|_| ())?)
 }
 
 /// # Safety
@@ -182,7 +182,11 @@ fn intern_append(
 }
 
 async fn do_append(op: &mut Operator, path: String, content: Vec<u8>) -> Result<()> {
-    Ok(op.write_with(&path, content).append(true).await?)
+    Ok(op
+        .write_with(&path, content)
+        .append(true)
+        .await
+        .map(|_| ())?)
 }
 
 /// # Safety

bindings/java/src/operator.rs

@@ -104,7 +104,7 @@ fn intern_write(
 ) -> Result<()> {
     let path = jstring_to_string(env, &path)?;
     let content = env.convert_byte_array(content)?;
-    Ok(op.write(&path, content)?)
+    Ok(op.write(&path, content).map(|_| ())?)
 }
 
 /// # Safety

bindings/nodejs/src/lib.rs

@@ -269,7 +269,7 @@ impl Operator {
                 writer = writer.cache_control(cache_control);
             }
         }
-        writer.await.map_err(format_napi_error)
+        writer.await.map(|_| ()).map_err(format_napi_error)
     }
 
     //noinspection DuplicatedCode
@@ -371,7 +371,7 @@ impl Operator {
                 writer = writer.cache_control(cache_control);
             }
         }
-        writer.call().map_err(format_napi_error)
+        writer.call().map(|_| ()).map_err(format_napi_error)
     }
 
     /// Copy file according to given `from` and `to` path.
@@ -809,7 +809,7 @@ impl BlockingWriter {
     /// ```
     #[napi]
     pub unsafe fn close(&mut self) -> Result<()> {
-        self.0.close().map_err(format_napi_error)
+        self.0.close().map(|_| ()).map_err(format_napi_error)
     }
 }
 
@@ -855,7 +855,7 @@ impl Writer {
     /// ```
     #[napi]
     pub async unsafe fn close(&mut self) -> Result<()> {
-        self.0.close().await.map_err(format_napi_error)
+        self.0.close().await.map(|_| ()).map_err(format_napi_error)
     }
 }
 

bindings/ocaml/src/operator/mod.rs

@@ -91,7 +91,7 @@ pub fn blocking_write(
     path: String,
     bs: &'static [u8],
 ) -> Result<(), String> {
-    map_res_error(operator.0.write(path.as_str(), bs))
+    map_res_error(operator.0.write(path.as_str(), bs).map(|_| ()))
 }
 
 #[ocaml::func]

bindings/python/src/operator.rs

@@ -135,7 +135,7 @@ impl Operator {
             write = write.cache_control(cache_control);
         }
 
-        write.call().map_err(format_pyerr)
+        write.call().map(|_| ()).map_err(format_pyerr)
     }
 
     /// Get current path's metadata **without cache** directly.
@@ -351,7 +351,7 @@ impl AsyncOperator {
             if let Some(cache_control) = &kwargs.cache_control {
                 write = write.cache_control(cache_control);
             }
-            write.await.map_err(format_pyerr)
+            write.await.map(|_| ()).map_err(format_pyerr)
         })
     }
 

bindings/ruby/src/operator.rs

@@ -69,6 +69,7 @@ impl Operator {
         rb_self
             .0
             .write(&path, bs.to_bytes())
+            .map(|_| ())
             .map_err(|err| Error::new(ruby.exception_runtime_error(), err.to_string()))
     }
 

core/src/layers/async_backtrace.rs

@@ -179,7 +179,7 @@ impl<R: oio::Write> oio::Write for AsyncBacktraceWrapper<R> {
     }
 
     #[async_backtrace::framed]
-    async fn close(&mut self) -> Result<()> {
+    async fn close(&mut self) -> Result<Metadata> {
         self.inner.close().await
     }
 
@@ -194,7 +194,7 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for AsyncBacktraceWrapper<R> {
         self.inner.write(bs)
     }
 
-    fn close(&mut self) -> Result<()> {
+    fn close(&mut self) -> Result<Metadata> {
         self.inner.close()
     }
 }

core/src/layers/await_tree.rs

@@ -206,7 +206,7 @@ impl<R: oio::Write> oio::Write for AwaitTreeWrapper<R> {
             .instrument_await(format!("opendal::{}", Operation::WriterAbort.into_static()))
     }
 
-    fn close(&mut self) -> impl Future<Output = Result<()>> + MaybeSend {
+    fn close(&mut self) -> impl Future<Output = Result<Metadata>> + MaybeSend {
         self.inner
             .close()
             .instrument_await(format!("opendal::{}", Operation::WriterClose.into_static()))
@@ -218,7 +218,7 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for AwaitTreeWrapper<R> {
         self.inner.write(bs)
     }
 
-    fn close(&mut self) -> Result<()> {
+    fn close(&mut self) -> Result<Metadata> {
         self.inner.close()
     }
 }

core/src/layers/blocking.rs

@@ -289,7 +289,7 @@ impl<I: oio::Write + 'static> oio::BlockingWrite for BlockingWrapper<I> {
         self.handle.block_on(self.inner.write(bs))
     }
 
-    fn close(&mut self) -> Result<()> {
+    fn close(&mut self) -> Result<Metadata> {
         self.handle.block_on(self.inner.close())
     }
 }

core/src/layers/complete.rs

@@ -350,6 +350,8 @@ impl<A: Access> LayeredAccess for CompleteAccessor<A> {
         if cap.list && cap.write_can_empty {
             cap.create_dir = true;
         }
+        // write operations should always return content length
+        cap.write_has_content_length = true;
         meta.into()
     }
 
@@ -517,15 +519,17 @@ where
         w.write(bs).await
     }
 
-    async fn close(&mut self) -> Result<()> {
+    async fn close(&mut self) -> Result<Metadata> {
         let w = self.inner.as_mut().ok_or_else(|| {
             Error::new(ErrorKind::Unexpected, "writer has been closed or aborted")
         })?;
 
-        w.close().await?;
+        // we must return `Err` before setting inner to None; otherwise,
+        // we won't be able to retry `close` in `RetryLayer`.
+        let ret = w.close().await?;
         self.inner = None;
 
-        Ok(())
+        Ok(ret)
     }
 
     async fn abort(&mut self) -> Result<()> {
@@ -552,13 +556,14 @@ where
         w.write(bs)
     }
 
-    fn close(&mut self) -> Result<()> {
+    fn close(&mut self) -> Result<Metadata> {
         let w = self.inner.as_mut().ok_or_else(|| {
             Error::new(ErrorKind::Unexpected, "writer has been closed or aborted")
         })?;
 
-        w.close()?;
+        let ret = w.close()?;
         self.inner = None;
-        Ok(())
+
+        Ok(ret)
     }
 }

core/src/layers/concurrent_limit.rs

@@ -267,7 +267,7 @@ impl<R: oio::Write> oio::Write for ConcurrentLimitWrapper<R> {
         self.inner.write(bs).await
     }
 
-    async fn close(&mut self) -> Result<()> {
+    async fn close(&mut self) -> Result<Metadata> {
         self.inner.close().await
     }
 
@@ -281,7 +281,7 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for ConcurrentLimitWrapper<R> {
         self.inner.write(bs)
     }
 
-    fn close(&mut self) -> Result<()> {
+    fn close(&mut self) -> Result<Metadata> {
         self.inner.close()
     }
 }

core/src/layers/correctness_check.rs

@@ -383,8 +383,8 @@ mod tests {
             Ok(())
         }
 
-        async fn close(&mut self) -> Result<()> {
-            Ok(())
+        async fn close(&mut self) -> Result<Metadata> {
+            Ok(Metadata::default())
         }
 
         async fn abort(&mut self) -> Result<()> {

core/src/layers/dtrace.rs

@@ -382,14 +382,15 @@ impl<R: oio::Write> oio::Write for DtraceLayerWrapper<R> {
             })
     }
 
-    async fn close(&mut self) -> Result<()> {
+    async fn close(&mut self) -> Result<Metadata> {
         let c_path = CString::new(self.path.clone()).unwrap();
         probe_lazy!(opendal, writer_close_start, c_path.as_ptr());
         self.inner
             .close()
             .await
-            .map(|_| {
+            .map(|meta| {
                 probe_lazy!(opendal, writer_close_ok, c_path.as_ptr());
+                meta
             })
             .map_err(|err| {
                 probe_lazy!(opendal, writer_close_error, c_path.as_ptr());
@@ -413,13 +414,14 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for DtraceLayerWrapper<R> {
             })
     }
 
-    fn close(&mut self) -> Result<()> {
+    fn close(&mut self) -> Result<Metadata> {
         let c_path = CString::new(self.path.clone()).unwrap();
         probe_lazy!(opendal, blocking_writer_close_start, c_path.as_ptr());
         self.inner
             .close()
-            .map(|_| {
+            .map(|meta| {
                 probe_lazy!(opendal, blocking_writer_close_ok, c_path.as_ptr());
+                meta
             })
             .map_err(|err| {
                 probe_lazy!(opendal, blocking_writer_close_error, c_path.as_ptr());

core/src/layers/error_context.rs

@@ -375,7 +375,7 @@ impl<T: oio::Write> oio::Write for ErrorContextWrapper<T> {
             })
     }
 
-    async fn close(&mut self) -> Result<()> {
+    async fn close(&mut self) -> Result<Metadata> {
         self.inner.close().await.map_err(|err| {
             err.with_operation(Operation::WriterClose)
                 .with_context("service", self.scheme)
@@ -411,7 +411,7 @@ impl<T: oio::BlockingWrite> oio::BlockingWrite for ErrorContextWrapper<T> {
             })
     }
 
-    fn close(&mut self) -> Result<()> {
+    fn close(&mut self) -> Result<Metadata> {
         self.inner.close().map_err(|err| {
             err.with_operation(Operation::BlockingWriterClose)
                 .with_context("service", self.scheme)

core/src/layers/fastrace.rs

@@ -327,7 +327,7 @@ impl<R: oio::Write> oio::Write for FastraceWrapper<R> {
         self.inner.abort()
     }
 
-    fn close(&mut self) -> impl Future<Output = Result<()>> + MaybeSend {
+    fn close(&mut self) -> impl Future<Output = Result<Metadata>> + MaybeSend {
         let _g = self.span.set_local_parent();
         let _span = LocalSpan::enter_with_local_parent(Operation::WriterClose.into_static());
         self.inner.close()
@@ -342,7 +342,7 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for FastraceWrapper<R> {
         self.inner.write(bs)
     }
 
-    fn close(&mut self) -> Result<()> {
+    fn close(&mut self) -> Result<Metadata> {
         let _g = self.span.set_local_parent();
         let _span =
             LocalSpan::enter_with_local_parent(Operation::BlockingWriterClose.into_static());

core/src/layers/logging.rs

@@ -1047,7 +1047,7 @@ impl<W: oio::Write, I: LoggingInterceptor> oio::Write for LoggingWriter<W, I> {
         }
     }
 
-    async fn close(&mut self) -> Result<()> {
+    async fn close(&mut self) -> Result<Metadata> {
         self.logger.log(
             &self.info,
             Operation::WriterClose,
@@ -1057,15 +1057,15 @@ impl<W: oio::Write, I: LoggingInterceptor> oio::Write for LoggingWriter<W, I> {
         );
 
         match self.inner.close().await {
-            Ok(_) => {
+            Ok(meta) => {
                 self.logger.log(
                     &self.info,
                     Operation::WriterClose,
                     &[("path", &self.path), ("written", &self.written.to_string())],
                     "succeeded",
                     None,
                 );
-                Ok(())
+                Ok(meta)
             }
             Err(err) => {
                 self.logger.log(
@@ -1129,7 +1129,7 @@ impl<W: oio::BlockingWrite, I: LoggingInterceptor> oio::BlockingWrite for Loggin
         }
     }
 
-    fn close(&mut self) -> Result<()> {
+    fn close(&mut self) -> Result<Metadata> {
         self.logger.log(
             &self.info,
             Operation::BlockingWriterClose,
@@ -1139,15 +1139,15 @@ impl<W: oio::BlockingWrite, I: LoggingInterceptor> oio::BlockingWrite for Loggin
         );
 
         match self.inner.close() {
-            Ok(_) => {
+            Ok(meta) => {
                 self.logger.log(
                     &self.info,
                     Operation::BlockingWriterWrite,
                     &[("path", &self.path), ("written", &self.written.to_string())],
                     "succeeded",
                     None,
                 );
-                Ok(())
+                Ok(meta)
             }
             Err(err) => {
                 self.logger.log(