fix: Fix bugs that failed wasabi's integration tests (#2273)

Signed-off-by: Xuanwo <github@xuanwo.io>

.env.example

@@ -105,3 +105,10 @@ OPENDAL_VERCEL_ARTIFACTS_ACCESS_TOKEN=<token>
 # onedrive
 OPENDAL_ONEDRIVE_TEST=false
 OPENDAL_ONEDRIVE_ACCESS_TOKEN=<access_token>
+# wasabi
+OPENDAL_WASABI_TEST=false
+OPENDAL_WASABI_REGION=<region>
+OPENDAL_WASABI_BUCKET=<bucket>
+OPENDAL_WASABI_ENDPOINT=<endpoint>
+OPENDAL_WASABI_ACCESS_KEY_ID=<ak>
+OPENDAL_WASABI_SECRET_ACCESS_KEY=<sk>

.github/workflows/service_test_wasabi.yml

@@ -0,0 +1,68 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: Service Test Wasabi
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+      - main
+    paths:
+      - "core/src/**"
+      - "core/tests/**"
+      - "!core/src/docs/**"
+      - "!core/src/services/**"
+      - "core/src/services/wasabi/**"
+      - ".github/workflows/service_test_wasabi.yml"
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}-${{ github.event_name }}
+  cancel-in-progress: true
+
+jobs:
+  wasabi:
+    runs-on: ubuntu-latest
+    if: github.event_name == 'push' || !github.event.pull_request.head.repo.fork
+    steps:
+      - uses: actions/checkout@v3
+      - name: Setup Rust toolchain
+        uses: ./.github/actions/setup
+
+      - name: Load secret
+        id: op-load-secret
+        uses: 1password/load-secrets-action@v1
+        with:
+          export-env: true
+        env:
+          OP_SERVICE_ACCOUNT_TOKEN: ${{ secrets.OP_SERVICE_ACCOUNT_TOKEN }}
+          OPENDAL_WASABI_TEST: op://services/wasabi/test
+          OPENDAL_WASABI_REGION: op://services/wasabi/region
+          OPENDAL_WASABI_BUCKET: op://services/wasabi/bucket
+          OPENDAL_WASABI_ENDPOINT: op://services/wasabi/endpoint
+          OPENDAL_WASABI_ACCESS_KEY_ID: op://services/wasabi/access_key_id
+          OPENDAL_WASABI_SECRET_ACCESS_KEY: op://services/wasabi/secret_access_key
+
+      - name: Test
+        shell: bash
+        working-directory: core
+        run: cargo test wasabi --features=services-wasabi -- --show-output
+        env:
+          RUST_BACKTRACE: full
+          RUST_LOG: debug

core/src/services/wasabi/backend.rs

@@ -975,6 +975,13 @@ impl Accessor for WasabiBackend {
     }
 
     async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+        if args.content_length().is_none() {
+            return Err(Error::new(
+                ErrorKind::Unsupported,
+                "write without content length is not supported",
+            ));
+        }
+
         Ok((
             RpWrite::default(),
             WasabiWriter::new(self.core.clone(), args, path.to_string()),
@@ -1004,7 +1011,7 @@ impl Accessor for WasabiBackend {
             return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
         }
 
-        let resp = self.core.head_object(path, args.if_none_match()).await?;
+        let resp = self.core.head_object(path, &args).await?;
 
         let status = resp.status();
 
@@ -1038,7 +1045,7 @@ impl Accessor for WasabiBackend {
     async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
         // We will not send this request out, just for signing.
         let mut req = match args.operation() {
-            PresignOperation::Stat(v) => self.core.head_object_request(path, v.if_none_match())?,
+            PresignOperation::Stat(v) => self.core.head_object_request(path, v)?,
             PresignOperation::Read(v) => self.core.get_object_request(
                 path,
                 v.range(),
@@ -1153,7 +1160,7 @@ mod tests {
         let endpoint_cases = vec![
             Some("s3.wasabisys.com"),
             Some("https://s3.wasabisys.com"),
-            Some("https://s3.us-east-2.amazonaws.com"),
+            Some("https://s3.us-east-2.wasabisys.com"),
             None,
         ];
 

core/src/services/wasabi/core.rs

@@ -36,6 +36,7 @@ use reqsign::AwsV4Signer;
 use serde::Deserialize;
 use serde::Serialize;
 
+use crate::ops::OpStat;
 use crate::raw::*;
 use crate::*;
 
@@ -207,11 +208,7 @@ impl WasabiCore {
 }
 
 impl WasabiCore {
-    pub fn head_object_request(
-        &self,
-        path: &str,
-        if_none_match: Option<&str>,
-    ) -> Result<Request<AsyncBody>> {
+    pub fn head_object_request(&self, path: &str, args: &OpStat) -> Result<Request<AsyncBody>> {
         let p = build_abs_path(&self.root, path);
 
         let url = format!("{}/{}", self.endpoint, percent_encode_path(&p));
@@ -220,8 +217,12 @@ impl WasabiCore {
 
         req = self.insert_sse_headers(req, false);
 
-        if let Some(if_none_match) = if_none_match {
-            req = req.header(http::header::IF_NONE_MATCH, if_none_match);
+        if let Some(v) = args.if_match() {
+            req = req.header(http::header::IF_MATCH, v);
+        }
+
+        if let Some(v) = args.if_none_match() {
+            req = req.header(http::header::IF_NONE_MATCH, v);
         }
 
         let req = req
@@ -346,9 +347,9 @@ impl WasabiCore {
     pub async fn head_object(
         &self,
         path: &str,
-        if_none_match: Option<&str>,
+        args: &OpStat,
     ) -> Result<Response<IncomingAsyncBody>> {
-        let mut req = self.head_object_request(path, if_none_match)?;
+        let mut req = self.head_object_request(path, args)?;
 
         self.sign(&mut req).await?;
 

core/src/services/wasabi/error.rs

@@ -44,7 +44,9 @@ pub async fn parse_error(resp: Response<IncomingAsyncBody>) -> Result<Error> {
     let (mut kind, mut retryable) = match parts.status {
         StatusCode::NOT_FOUND => (ErrorKind::NotFound, false),
         StatusCode::FORBIDDEN => (ErrorKind::PermissionDenied, false),
-        StatusCode::PRECONDITION_FAILED => (ErrorKind::ConditionNotMatch, false),
+        StatusCode::PRECONDITION_FAILED | StatusCode::NOT_MODIFIED => {
+            (ErrorKind::ConditionNotMatch, false)
+        }
         StatusCode::INTERNAL_SERVER_ERROR
         | StatusCode::BAD_GATEWAY
         | StatusCode::SERVICE_UNAVAILABLE

core/src/services/wasabi/writer.rs

@@ -43,39 +43,24 @@ impl WasabiWriter {
 #[async_trait]
 impl oio::Write for WasabiWriter {
     async fn write(&mut self, bs: Bytes) -> Result<()> {
-        if self.op.content_length().unwrap_or_default() == bs.len() as u64 {
-            let resp = self
-                .core
-                .put_object(
-                    &self.path,
-                    Some(bs.len()),
-                    self.op.content_type(),
-                    self.op.content_disposition(),
-                    self.op.cache_control(),
-                    AsyncBody::Bytes(bs),
-                )
-                .await?;
+        let resp = self
+            .core
+            .put_object(
+                &self.path,
+                Some(bs.len()),
+                self.op.content_type(),
+                self.op.content_disposition(),
+                self.op.cache_control(),
+                AsyncBody::Bytes(bs),
+            )
+            .await?;
 
-            match resp.status() {
-                StatusCode::CREATED | StatusCode::OK => {
-                    resp.into_body().consume().await?;
-                    Ok(())
-                }
-                _ => Err(parse_error(resp).await?),
-            }
-        } else {
-            let resp = self
-                .core
-                .append_object(&self.path, Some(bs.len()), AsyncBody::Bytes(bs))
-                .await?;
-
-            match resp.status() {
-                StatusCode::CREATED | StatusCode::OK | StatusCode::NO_CONTENT => {
-                    resp.into_body().consume().await?;
-                    Ok(())
-                }
-                _ => Err(parse_error(resp).await?),
+        match resp.status() {
+            StatusCode::CREATED | StatusCode::OK => {
+                resp.into_body().consume().await?;
+                Ok(())
             }
+            _ => Err(parse_error(resp).await?),
         }
     }
 

core/tests/behavior/main.rs

@@ -139,6 +139,8 @@ behavior_tests!(Supabase);
 behavior_tests!(Sled);
 #[cfg(feature = "services-vercel-artifacts")]
 behavior_tests!(VercelArtifacts);
+#[cfg(feature = "services-wasabi")]
+behavior_tests!(Wasabi);
 #[cfg(feature = "services-webdav")]
 behavior_tests!(Webdav);
 #[cfg(feature = "services-webhdfs")]