Migrate S3 store to official AWS SDK (TraceMachina#369)

This is a breaking change.

- The S3 store now uses the aws-sdk for Rust to handle higher level
  interactions with the S3 API.
- It now also supports HTTP/2 via hyper/rustls.
- The concurrency limits have been removed and are now only configurable
  for multipart uploads.
- The default behavior is now HTTPS. To enable unencrypted
  communication, an `insecure_allow_http` flag has been added.

Fixes TraceMachina#131

Cargo.toml

@@ -103,6 +103,11 @@ members = [
 [workspace.dependencies]
 async-lock = "2.7.0"
 async-trait = "0.1.71"
+aws-config = "0.57.1"
+aws-sdk-config = "0.35.0"
+aws-sdk-s3 = { version = "0.35.0", features = ["rt-tokio", "test-util"] }
+aws-smithy-runtime = { version = "0.57.1", features = ["client", "connector-hyper-0-14-x"] }
+aws-smithy-types = "0.57.1"
 axum = "0.6.18"
 bincode = "1.3.3"
 blake3 = "1.4.1"
@@ -118,7 +123,8 @@ futures = "0.3.28"
 hashbrown = "0.14"
 hex = "0.4.3"
 http = "^0.2"
-hyper = "0.14.27"
+hyper = { version = "0.14.27", features = ["tcp", "client"] }
+hyper-rustls = { version = "0.24.2", features = ["webpki-tokio"] }
 json5 = "0.4.1"
 log = "0.4.19"
 lru = "0.10.1"
@@ -138,10 +144,6 @@ prost-types = "0.11.9"
 rand = "0.8.5"
 rcgen = "0.11.3"
 relative-path = "1.8.0"
-rusoto_core = "0.48.0"
-rusoto_mock = "=0.48.0"
-rusoto_s3 = "0.48.0"
-rusoto_signature = "0.48.0"
 rustls-pemfile = "1.0.3"
 scopeguard = "1.2.0"
 serde = "1.0.167"

cas/scheduler/grpc_scheduler.rs

@@ -78,7 +78,7 @@ impl GrpcScheduler {
             platform_property_managers: Mutex::new(HashMap::new()),
             jitter_fn,
             retry: config.retry.clone(),
-            retrier: Retrier::new(Box::new(|duration| Box::pin(sleep(duration)))),
+            retrier: Retrier::new(Arc::new(|duration| Box::pin(sleep(duration)))),
         })
     }
 

cas/store/BUILD

@@ -275,16 +275,15 @@ rust_library(
         "//util:common",
         "//util:error",
         "//util:retry",
-        "@crate_index//:bytes",
+        "@crate_index//:aws-config",
+        "@crate_index//:aws-sdk-s3",
+        "@crate_index//:aws-smithy-runtime",
         "@crate_index//:futures",
-        "@crate_index//:http",
-        "@crate_index//:once_cell",
+        "@crate_index//:hyper",
+        "@crate_index//:hyper-rustls",
         "@crate_index//:rand",
-        "@crate_index//:rusoto_core",
-        "@crate_index//:rusoto_s3",
-        "@crate_index//:rusoto_signature",
         "@crate_index//:tokio",
-        "@crate_index//:tokio-util",
+        "@crate_index//:tokio-stream",
     ],
 )
 
@@ -468,17 +467,14 @@ rust_test(
         ":s3_store",
         ":traits",
         "//config",
-        "//util:async_fixed_buffer",
         "//util:common",
         "//util:error",
-        "@crate_index//:futures",
+        "@crate_index//:aws-sdk-s3",
+        "@crate_index//:aws-smithy-runtime",
+        "@crate_index//:aws-smithy-types",
         "@crate_index//:http",
         "@crate_index//:pretty_assertions",
-        "@crate_index//:rusoto_core",
-        "@crate_index//:rusoto_mock",
-        "@crate_index//:rusoto_s3",
         "@crate_index//:tokio",
-        "@crate_index//:tokio-util",
     ],
 )
 

cas/store/default_store_factory.rs

@@ -46,7 +46,7 @@ pub fn store_factory<'a>(
     Box::pin(async move {
         let store: Arc<dyn Store> = match backend {
             StoreConfig::memory(config) => Arc::new(MemoryStore::new(config)),
-            StoreConfig::s3_store(config) => Arc::new(S3Store::new(config)?),
+            StoreConfig::s3_store(config) => Arc::new(S3Store::new(config).await?),
             StoreConfig::verify(config) => Arc::new(VerifyStore::new(
                 config,
                 store_factory(&config.backend, store_manager, None).await?,

cas/store/grpc_store.rs

@@ -108,7 +108,7 @@ impl GrpcStore {
             store_type: config.store_type,
             jitter_fn,
             retry: config.retry.to_owned(),
-            retrier: Retrier::new(Box::new(|duration| Box::pin(sleep(duration)))),
+            retrier: Retrier::new(Arc::new(|duration| Box::pin(sleep(duration)))),
         })
     }