chore: bump aws sdk

Cargo.toml

@@ -68,35 +68,38 @@ repository = "https://github.com/risingwavelabs/risingwave"
 
 [workspace.dependencies]
 await-tree = "0.1.1"
-aws-config = { version = "0.57", default-features = false, features = [
+aws-config = { version = "1", default-features = false, features = [
+  "behavior-version-latest",
   "rt-tokio",
   "rustls",
 ] }
-aws-credential-types = { version = "0.57", default-features = false, features = [
+aws-credential-types = { version = "1", default-features = false, features = [
   "hardcoded-credentials",
 ] }
-aws-sdk-kinesis = { version = "0.35", default-features = false, features = [
+aws-sdk-kinesis = { version = "1", default-features = false, features = [
   "rt-tokio",
   "rustls",
 ] }
-aws-sdk-s3 = { version = "0.35", default-features = false, features = [
+aws-sdk-s3 = { version = "1", default-features = false, features = [
   "rt-tokio",
   "rustls",
 ] }
-aws-sdk-ec2 = { version = "0.35", default-features = false, features = [
+aws-sdk-ec2 = { version = "1", default-features = false, features = [
   "rt-tokio",
   "rustls",
 ] }
-aws-sdk-sqs = { version = "0.35", default-features = false, features = [
+aws-sdk-sqs = { version = "1", default-features = false, features = [
   "rt-tokio",
   "rustls",
 ] }
-aws-smithy-http = "0.57"
-aws-smithy-runtime-api = "0.57"
-aws-smithy-runtime = "0.57"
-aws-smithy-types = { version = "0.57", default-features = false, features = ["hyper-0-14-x"] }
-aws-endpoint = "0.57"
-aws-types = "0.57"
+aws-smithy-http = "0.60"
+aws-smithy-runtime-api = "1"
+aws-smithy-runtime = "1"
+aws-smithy-types = { version = "1", default-features = false, features = [
+  "hyper-0-14-x",
+] }
+aws-endpoint = "0.60"
+aws-types = "1"
 etcd-client = { package = "madsim-etcd-client", version = "0.4" }
 futures-async-stream = "0.2.9"
 hytra = "0.1"

src/connector/src/source/filesystem/file_common.rs

@@ -32,7 +32,7 @@ impl From<&Object> for FsSplit {
         Self {
             name: value.key().unwrap().to_owned(),
             offset: 0,
-            size: value.size() as usize,
+            size: value.size().unwrap_or_default() as usize,
         }
     }
 }
@@ -81,7 +81,7 @@ impl From<&Object> for FsPageItem {
         let aws_ts = value.last_modified().unwrap();
         Self {
             name: value.key().unwrap().to_owned(),
-            size: value.size(),
+            size: value.size().unwrap_or_default(),
             timestamp: Timestamp::from_timestamp_uncheck(aws_ts.secs(), aws_ts.subsec_nanos()),
         }
     }

src/connector/src/source/filesystem/s3/source/reader.rs

@@ -24,7 +24,6 @@ use aws_smithy_runtime_api::client::result::SdkError;
 use aws_smithy_types::body::SdkBody;
 use aws_smithy_types::byte_stream::ByteStream;
 use futures_async_stream::try_stream;
-use hyper::Response;
 use io::StreamReader;
 use risingwave_common::error::RwError;
 use tokio::io::BufReader;
@@ -145,7 +144,10 @@ impl S3FileReader {
         bucket_name: &str,
         object_name: &str,
         start: usize,
-    ) -> std::result::Result<ByteStream, SdkError<GetObjectError, Response<SdkBody>>> {
+    ) -> std::result::Result<
+        ByteStream,
+        SdkError<GetObjectError, aws_smithy_runtime_api::http::Response<SdkBody>>,
+    > {
         let range = if start == 0 {
             None
         } else {

src/connector/src/source/filesystem/s3_v2/lister.rs

@@ -39,7 +39,7 @@ impl FsListInner for S3SplitEnumerator {
             .send()
             .await
             .map_err(|e| anyhow!(DisplayErrorContext(e)))?;
-        if res.is_truncated() {
+        if res.is_truncated().unwrap_or_default() {
             self.next_continuation_token = res.next_continuation_token.clone();
         } else {
             has_finished = true;

src/object_store/Cargo.toml

@@ -12,7 +12,7 @@ repository = { workspace = true }
 async-trait = "0.1"
 await-tree = { workspace = true }
 aws-config = { workspace = true }
-aws-sdk-s3 = { version = "0.3", package = "madsim-aws-sdk-s3" }
+aws-sdk-s3 = { version = "0.5", package = "madsim-aws-sdk-s3" }
 aws-smithy-http = { workspace = true }
 aws-smithy-runtime = { workspace = true }
 aws-smithy-runtime-api = { workspace = true }
@@ -22,7 +22,7 @@ crc32fast = "1.3.2"
 either = "1"
 fail = "0.5"
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
-hyper = "0.14"
+hyper = { version = "0.14", features = ["tcp", "client"] }
 hyper-rustls = { version = "0.24.2", features = ["webpki-roots"] }
 hyper-tls = "0.5.0"
 itertools = "0.12"

src/object_store/src/object/error.rs

@@ -21,7 +21,6 @@ use aws_sdk_s3::operation::get_object::GetObjectError;
 use aws_sdk_s3::operation::head_object::HeadObjectError;
 use aws_sdk_s3::primitives::ByteStreamError;
 use aws_smithy_types::body::SdkBody;
-use hyper::Response;
 use risingwave_common::error::BoxedError;
 use thiserror::Error;
 use tokio::sync::oneshot::error::RecvError;
@@ -84,15 +83,15 @@ impl ObjectError {
                 if let Some(aws_smithy_runtime_api::client::result::SdkError::ServiceError(err)) = e
                     .downcast_ref::<aws_smithy_runtime_api::client::result::SdkError<
                         GetObjectError,
-                        Response<SdkBody>,
+                        aws_smithy_runtime_api::http::Response<SdkBody>,
                     >>()
                 {
                     return matches!(err.err(), GetObjectError::NoSuchKey(_));
                 }
                 if let Some(aws_smithy_runtime_api::client::result::SdkError::ServiceError(err)) = e
                     .downcast_ref::<aws_smithy_runtime_api::client::result::SdkError<
                         HeadObjectError,
-                        Response<SdkBody>,
+                        aws_smithy_runtime_api::http::Response<SdkBody>,
                     >>()
                 {
                     return matches!(err.err(), HeadObjectError::NotFound(_));
@@ -113,11 +112,12 @@ impl ObjectError {
     }
 }
 
-impl<E> From<aws_sdk_s3::error::SdkError<E>> for ObjectError
+impl<E, R> From<aws_smithy_runtime_api::client::result::SdkError<E, R>> for ObjectError
 where
     E: std::error::Error + Sync + Send + 'static,
+    R: Send + Sync + 'static + std::fmt::Debug,
 {
-    fn from(e: aws_sdk_s3::error::SdkError<E>) -> Self {
+    fn from(e: aws_smithy_runtime_api::client::result::SdkError<E, R>) -> Self {
         ObjectErrorInner::S3(e.into()).into()
     }
 }

src/object_store/src/object/s3.rs

@@ -41,7 +41,7 @@ use either::Either;
 use fail::fail_point;
 use futures::future::{try_join_all, BoxFuture, FutureExt};
 use futures::{stream, Stream, StreamExt, TryStreamExt};
-use hyper::{Body, Response};
+use hyper::Body;
 use itertools::Itertools;
 use risingwave_common::config::ObjectStoreConfig;
 use risingwave_common::monitor::connection::monitor_connector;
@@ -420,7 +420,7 @@ impl ObjectStore for S3ObjectStore {
                 .last_modified()
                 .expect("last_modified required")
                 .as_secs_f64(),
-            total_size: resp.content_length as usize,
+            total_size: resp.content_length.unwrap_or_default() as usize,
         })
     }
 
@@ -645,20 +645,27 @@ impl S3ObjectStore {
 
         let s3_object_store_config = ObjectStoreConfig::default();
         #[cfg(madsim)]
-        let builder = aws_sdk_s3::config::Builder::new();
+        let builder = aws_sdk_s3::config::Builder::new().credentials_provider(
+            Credentials::from_keys(access_key_id, secret_access_key, None),
+        );
         #[cfg(not(madsim))]
-        let builder: aws_sdk_s3::config::Builder =
-            aws_sdk_s3::config::Builder::from(&aws_config::ConfigLoader::default().load().await)
-                .force_path_style(true)
-                .http_client(Self::new_http_client(&s3_object_store_config));
+        let builder = aws_sdk_s3::config::Builder::from(
+            &aws_config::ConfigLoader::default()
+                // FIXME: https://github.com/awslabs/aws-sdk-rust/issues/973
+                .credentials_provider(Credentials::from_keys(
+                    access_key_id,
+                    secret_access_key,
+                    None,
+                ))
+                .load()
+                .await,
+        )
+        .force_path_style(true)
+        .http_client(Self::new_http_client(&s3_object_store_config))
+        .behavior_version_latest();
         let config = builder
             .region(Region::new("custom"))
             .endpoint_url(format!("{}{}", endpoint_prefix, address))
-            .credentials_provider(Credentials::from_keys(
-                access_key_id,
-                secret_access_key,
-                None,
-            ))
             .build();
         let client = Client::from_conf(config);
 
@@ -808,7 +815,10 @@ impl S3ObjectStore {
 
     #[inline(always)]
     fn should_retry(
-        err: &Either<SdkError<GetObjectError, Response<SdkBody>>, ByteStreamError>,
+        err: &Either<
+            SdkError<GetObjectError, aws_smithy_runtime_api::http::Response<SdkBody>>,
+            ByteStreamError,
+        >,
     ) -> bool {
         match err {
             Either::Left(err) => {
@@ -845,14 +855,14 @@ struct S3ObjectIter {
     bucket: String,
     prefix: String,
     next_continuation_token: Option<String>,
-    is_truncated: bool,
+    is_truncated: Option<bool>,
     #[allow(clippy::type_complexity)]
     send_future: Option<
         BoxFuture<
             'static,
             Result<
-                (Vec<ObjectMetadata>, Option<String>, bool),
-                SdkError<ListObjectsV2Error, Response<SdkBody>>,
+                (Vec<ObjectMetadata>, Option<String>, Option<bool>),
+                SdkError<ListObjectsV2Error, aws_smithy_runtime_api::http::Response<SdkBody>>,
             >,
         >,
     >,
@@ -866,7 +876,7 @@ impl S3ObjectIter {
             bucket,
             prefix,
             next_continuation_token: None,
-            is_truncated: true,
+            is_truncated: Some(true),
             send_future: None,
         }
     }
@@ -894,7 +904,7 @@ impl Stream for S3ObjectIter {
                 }
             };
         }
-        if !self.is_truncated {
+        if !self.is_truncated.unwrap_or_default() {
             return Poll::Ready(None);
         }
         let mut request = self
@@ -917,7 +927,7 @@ impl Stream for S3ObjectIter {
                                 .last_modified()
                                 .map(|l| l.as_secs_f64())
                                 .unwrap_or(0f64),
-                            total_size: obj.size() as usize,
+                            total_size: obj.size().unwrap_or_default() as usize,
                         })
                         .collect_vec();
                     let is_truncated = r.is_truncated;
@@ -932,8 +942,20 @@ impl Stream for S3ObjectIter {
     }
 }
 
-impl From<Either<SdkError<GetObjectError, Response<SdkBody>>, ByteStreamError>> for ObjectError {
-    fn from(e: Either<SdkError<GetObjectError, Response<SdkBody>>, ByteStreamError>) -> Self {
+impl
+    From<
+        Either<
+            SdkError<GetObjectError, aws_smithy_runtime_api::http::Response<SdkBody>>,
+            ByteStreamError,
+        >,
+    > for ObjectError
+{
+    fn from(
+        e: Either<
+            SdkError<GetObjectError, aws_smithy_runtime_api::http::Response<SdkBody>>,
+            ByteStreamError,
+        >,
+    ) -> Self {
         match e {
             Either::Left(e) => e.into(),
             Either::Right(e) => e.into(),

src/tests/simulation/Cargo.toml

@@ -13,7 +13,7 @@ normal = ["serde"]
 [dependencies]
 anyhow = "1.0"
 async-trait = "0.1"
-aws-sdk-s3 = { version = "0.3", package = "madsim-aws-sdk-s3" }
+aws-sdk-s3 = { version = "0.5", package = "madsim-aws-sdk-s3" }
 cfg-or-panic = "0.2"
 clap = { version = "4", features = ["derive"] }
 console = "0.15"

src/workspace-hack/Cargo.toml

@@ -22,12 +22,12 @@ ahash = { version = "0.8" }
 allocator-api2 = { version = "0.2", default-features = false, features = ["alloc", "nightly"] }
 anyhow = { version = "1", features = ["backtrace"] }
 async-std = { version = "1", features = ["attributes", "tokio1"] }
-aws-credential-types = { version = "0.57", default-features = false, features = ["hardcoded-credentials"] }
-aws-runtime = { version = "0.57", default-features = false, features = ["event-stream", "sigv4a"] }
-aws-sdk-s3 = { version = "0.35" }
-aws-sigv4 = { version = "0.57", features = ["http0-compat", "sign-eventstream", "sigv4a"] }
-aws-smithy-runtime = { version = "0.57", default-features = false, features = ["client", "rt-tokio", "tls-rustls"] }
-aws-smithy-types = { version = "0.57", default-features = false, features = ["byte-stream-poll-next", "http-body-0-4-x", "hyper-0-14-x", "rt-tokio"] }
+aws-credential-types = { version = "1", default-features = false, features = ["hardcoded-credentials"] }
+aws-runtime = { version = "1", default-features = false, features = ["event-stream", "sigv4a"] }
+aws-sdk-s3 = { version = "1" }
+aws-sigv4 = { version = "1", features = ["http0-compat", "sign-eventstream", "sigv4a"] }
+aws-smithy-runtime = { version = "1", default-features = false, features = ["client", "rt-tokio", "tls-rustls"] }
+aws-smithy-types = { version = "1", default-features = false, features = ["byte-stream-poll-next", "http-body-0-4-x", "hyper-0-14-x", "rt-tokio"] }
 base64 = { version = "0.21", features = ["alloc"] }
 bit-vec = { version = "0.6" }
 bitflags = { version = "2", default-features = false, features = ["serde", "std"] }
@@ -39,6 +39,7 @@ clap_builder = { version = "4", default-features = false, features = ["cargo", "
 combine = { version = "4", features = ["tokio"] }
 crossbeam-epoch = { version = "0.9" }
 crossbeam-utils = { version = "0.8" }
+crypto-bigint = { version = "0.5", features = ["generic-array", "zeroize"] }
 deranged = { version = "0.3", default-features = false, features = ["powerfmt", "serde", "std"] }
 digest = { version = "0.10", features = ["mac", "oid", "std"] }
 either = { version = "1", features = ["serde"] }