ci: Add behavior test for b2

.github/services/b2/b2/action.yml

@@ -0,0 +1,32 @@
+# 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: b2
+description: 'Behavior test for B2'
+
+runs:
+  using: "composite"
+  steps:
+    - name: Setup
+      uses: 1password/load-secrets-action@v1
+      with:
+        export-env: true
+      env:
+        OPENDAL_B2_BUCKET: op://services/b2/bucket
+        OPENDAL_B2_BUCKET_ID: op://services/b2/bucket_id
+        OPENDAL_B2_APPLICATION_KEY_ID: op://services/b2/application_key_id
+        OPENDAL_B2_APPLICATION_KEY: op://services/b2/application_key

bindings/java/Cargo.toml

@@ -88,6 +88,7 @@ services-all = [
   "services-libsql",
   "services-swift",
   "services-alluxio",
+  "services-b2",
 ]
 
 # Default services provided by opendal.
@@ -109,6 +110,7 @@ services-webhdfs = ["opendal/services-webhdfs"]
 # Optional services provided by opendal.
 services-alluxio = ["opendal/services-alluxio"]
 services-azfile = ["opendal/services-azfile"]
+services-b2 = ["opendal/services-b2"]
 services-cacache = ["opendal/services-cacache"]
 services-dashmap = ["opendal/services-dashmap"]
 services-dropbox = ["opendal/services-dropbox"]

bindings/nodejs/Cargo.toml

@@ -83,6 +83,7 @@ services-all = [
   "services-sqlite",
   "services-libsql",
   "services-alluxio",
+  "services-b2",
 ]
 
 # Default services provided by opendal.
@@ -104,6 +105,7 @@ services-webhdfs = ["opendal/services-webhdfs"]
 # Optional services provided by opendal.
 services-alluxio = ["opendal/services-alluxio"]
 services-azfile = ["opendal/services-azfile"]
+services-b2 = ["opendal/services-b2"]
 services-cacache = ["opendal/services-cacache"]
 services-dashmap = ["opendal/services-dashmap"]
 services-dropbox = ["opendal/services-dropbox"]

bindings/python/Cargo.toml

@@ -82,6 +82,7 @@ services-all = [
   "services-sqlite",
   "services-libsql",
   "services-alluxio",
+  "services-b2",
 ]
 
 # Default services provided by opendal.
@@ -103,6 +104,7 @@ services-webhdfs = ["opendal/services-webhdfs"]
 # Optional services provided by opendal.
 services-alluxio = ["opendal/services-alluxio"]
 services-azfile = ["opendal/services-azfile"]
+services-b2 = ["opendal/services-b2"]
 services-cacache = ["opendal/services-cacache"]
 services-dashmap = ["opendal/services-dashmap"]
 services-dropbox = ["opendal/services-dropbox"]

core/Cargo.toml

@@ -257,7 +257,9 @@ mini-moka = { version = "0.10", optional = true }
 minitrace = { version = "0.6", optional = true }
 moka = { version = "0.12", optional = true, features = ["future", "sync"] }
 mongodb = { version = "2.7.0", optional = true, features = ["tokio-runtime"] }
-mysql_async = { version = "0.32.2", default-features = false, features = ["default-rustls"], optional = true }
+mysql_async = { version = "0.32.2", default-features = false, features = [
+  "default-rustls",
+], optional = true }
 once_cell = "1"
 openssh = { version = "0.10.0", optional = true }
 openssh-sftp-client = { version = "0.14.0", optional = true, features = [

core/src/services/b2/core.rs

@@ -89,7 +89,7 @@ impl B2Core {
 
         {
             let mut signer = self.signer.write().await;
-            let req = Request::get("https://api.backblazeb2.com/b2api/v3/b2_authorize_account")
+            let req = Request::get("https://api.backblazeb2.com/b2api/v2/b2_authorize_account")
                 .header(
                     header::AUTHORIZATION,
                     format_authorization_by_basic(
@@ -110,8 +110,8 @@ impl B2Core {
                         .map_err(new_json_deserialize_error)?;
                     signer.auth_info = AuthInfo {
                         authorization_token: token.authorization_token.clone(),
-                        api_url: token.api_info.storage_api.api_url.clone(),
-                        download_url: token.api_info.storage_api.download_url.clone(),
+                        api_url: token.api_url.clone(),
+                        download_url: token.download_url.clone(),
                         // This authorization token is valid for at most 24 hours.
                         expires_in: Utc::now() + chrono::Duration::hours(20),
                     };
@@ -149,7 +149,7 @@ impl B2Core {
 
         let range = args.range();
         if !range.is_full() {
-            req = req.header(http::header::RANGE, range.to_header());
+            req = req.header(header::RANGE, range.to_header());
         }
 
         let req = req
@@ -210,7 +210,7 @@ impl B2Core {
 
         let range = args.range();
         if !range.is_full() {
-            req = req.header(http::header::RANGE, range.to_header());
+            req = req.header(header::RANGE, range.to_header());
         }
         let body = GetDownloadAuthorizationRequest {
             bucket_id: self.bucket_id.clone(),
@@ -444,7 +444,6 @@ impl B2Core {
         }
 
         if let Some(start_after) = start_after {
-            let start_after = build_abs_path(&self.root, &start_after);
             url.push_str(&format!(
                 "&startFileName={}",
                 percent_encode_path(&start_after)
@@ -586,18 +585,6 @@ pub struct AuthorizeAccountResponse {
     /// An authorization token to use with all calls, other than b2_authorize_account, that need an Authorization header. This authorization token is valid for at most 24 hours.
     /// So we should call b2_authorize_account every 24 hours.
     pub authorization_token: String,
-    pub api_info: ApiInfo,
-}
-
-#[derive(Debug, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub struct ApiInfo {
-    pub storage_api: StorageApi,
-}
-
-#[derive(Debug, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub struct StorageApi {
     pub api_url: String,
     pub download_url: String,
 }

core/src/services/b2/error.rs

@@ -44,6 +44,7 @@ pub async fn parse_error(resp: Response<IncomingAsyncBody>) -> Result<Error> {
         304 | 412 => (ErrorKind::ConditionNotMatch, false),
         // Service b2 could return 403, show the authorization error
         401 => (ErrorKind::PermissionDenied, true),
+        429 => (ErrorKind::RateLimited, true),
         500 | 502 | 503 | 504 => (ErrorKind::Unexpected, true),
         _ => (ErrorKind::Unexpected, false),
     };

core/src/services/b2/lister.rs

@@ -67,7 +67,13 @@ impl oio::PageList for B2Lister {
                 Some(&self.path),
                 self.delimiter,
                 self.limit,
-                self.start_after.clone(),
+                if ctx.token.is_empty() {
+                    self.start_after
+                        .as_ref()
+                        .map(|v| build_abs_path(&self.core.root, v))
+                } else {
+                    Some(ctx.token.clone())
+                },
             )
             .await?;
 
@@ -80,7 +86,11 @@ impl oio::PageList for B2Lister {
         let output: ListFileNamesResponse =
             serde_json::from_reader(bs.reader()).map_err(new_json_deserialize_error)?;
 
-        ctx.done = output.next_file_name.is_none();
+        if let Some(token) = output.next_file_name {
+            ctx.token = token;
+        } else {
+            ctx.done = true;
+        }
 
         for file in output.files {
             if let Some(start_after) = self.start_after.clone() {