diff --git a/Cargo.lock b/Cargo.lock
index ba80533f4..b48dce6bb 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -146,7 +146,7 @@ version = "0.16.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ceb7c683b2f8f40970b70e39ff8be514c95b96fcb9c4af87e1ed2cb2e10801a0"
 dependencies = [
- "digest 0.10.7",
+ "digest",
  "lazy_static",
  "libflate",
  "log",
@@ -982,13 +982,14 @@ dependencies = [
  "arroyo-rpc",
  "arroyo-types",
  "async-trait",
+ "aws-config 1.5.4",
+ "aws-credential-types",
  "bytes",
  "futures",
  "object_store",
  "once_cell",
  "rand 0.8.5",
  "regex",
- "rusoto_core",
  "thiserror",
  "tokio",
  "tokio-util",
@@ -1089,8 +1090,6 @@ dependencies = [
  "async-ffi",
  "async-stream",
  "async-trait",
- "aws-config 0.51.0",
- "aws-sdk-kinesis",
  "base64 0.21.7",
  "bincode",
  "bytes",
@@ -1105,7 +1104,7 @@ dependencies = [
  "itertools 0.12.1",
  "lazy_static",
  "local-ip-address",
- "md-5 0.10.6",
+ "md-5",
  "memchr",
  "object_store",
  "once_cell",
@@ -1118,12 +1117,10 @@ dependencies = [
  "regex",
  "regress 0.6.0",
  "reqwest",
- "rusoto_core",
- "rusoto_s3",
  "serde",
  "serde_json",
  "serde_json_path",
- "sha2 0.10.8",
+ "sha2",
  "stacker",
  "test-case",
  "tokio",
@@ -1532,22 +1529,22 @@ dependencies = [
 
 [[package]]
 name = "aws-config"
-version = "1.4.0"
+version = "1.5.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "40ddbfb5db93d62521f47b3f223da0884a2f02741ff54cb9cda192a0e73ba08b"
+checksum = "caf6cfe2881cb1fcbba9ae946fb9a6480d3b7a714ca84c74925014a89ef3387a"
 dependencies = [
  "aws-credential-types",
  "aws-runtime",
- "aws-sdk-sso 1.24.0",
+ "aws-sdk-sso 1.36.0",
  "aws-sdk-ssooidc",
- "aws-sdk-sts 1.24.0",
+ "aws-sdk-sts 1.36.0",
  "aws-smithy-async 1.2.1",
- "aws-smithy-http 0.60.8",
+ "aws-smithy-http 0.60.9",
  "aws-smithy-json 0.60.7",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
- "aws-smithy-types 1.1.9",
- "aws-types 1.2.1",
+ "aws-smithy-types 1.2.0",
+ "aws-types 1.3.3",
  "bytes",
  "fastrand 2.1.0",
  "hex",
@@ -1569,7 +1566,7 @@ checksum = "e16838e6c9e12125face1c1eff1343c75e3ff540de98ff7ebd61874a89bcfeb9"
 dependencies = [
  "aws-smithy-async 1.2.1",
  "aws-smithy-runtime-api",
- "aws-smithy-types 1.1.9",
+ "aws-smithy-types 1.2.0",
  "zeroize",
 ]
 
@@ -1607,17 +1604,17 @@ dependencies = [
 
 [[package]]
 name = "aws-runtime"
-version = "1.2.2"
+version = "1.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "75588e7ee5e8496eed939adac2035a6dbab9f7eb2acdd9ab2d31856dab6f3955"
+checksum = "87c5f920ffd1e0526ec9e70e50bf444db50b204395a0fa7016bbf9e31ea1698f"
 dependencies = [
  "aws-credential-types",
- "aws-sigv4 1.2.1",
+ "aws-sigv4 1.2.3",
  "aws-smithy-async 1.2.1",
- "aws-smithy-http 0.60.8",
+ "aws-smithy-http 0.60.9",
  "aws-smithy-runtime-api",
- "aws-smithy-types 1.1.9",
- "aws-types 1.2.1",
+ "aws-smithy-types 1.2.0",
+ "aws-types 1.3.3",
  "bytes",
  "fastrand 2.1.0",
  "http 0.2.12",
@@ -1637,12 +1634,12 @@ dependencies = [
  "aws-credential-types",
  "aws-runtime",
  "aws-smithy-async 1.2.1",
- "aws-smithy-http 0.60.8",
+ "aws-smithy-http 0.60.9",
  "aws-smithy-json 0.60.7",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
- "aws-smithy-types 1.1.9",
- "aws-types 1.2.1",
+ "aws-smithy-types 1.2.0",
+ "aws-types 1.3.3",
  "bytes",
  "fastrand 2.1.0",
  "http 0.2.12",
@@ -1697,19 +1694,19 @@ dependencies = [
 
 [[package]]
 name = "aws-sdk-sso"
-version = "1.24.0"
+version = "1.36.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5fd6a9b38fe2dcaa2422b42e2c667112872d03a83522533f8b4165fd2d9d4bd1"
+checksum = "6acca681c53374bf1d9af0e317a41d12a44902ca0f2d1e10e5cb5bb98ed74f35"
 dependencies = [
  "aws-credential-types",
  "aws-runtime",
  "aws-smithy-async 1.2.1",
- "aws-smithy-http 0.60.8",
+ "aws-smithy-http 0.60.9",
  "aws-smithy-json 0.60.7",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
- "aws-smithy-types 1.1.9",
- "aws-types 1.2.1",
+ "aws-smithy-types 1.2.0",
+ "aws-types 1.3.3",
  "bytes",
  "http 0.2.12",
  "once_cell",
@@ -1719,19 +1716,19 @@ dependencies = [
 
 [[package]]
 name = "aws-sdk-ssooidc"
-version = "1.24.0"
+version = "1.37.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b3f573392f21553835024164b795506363132b581a7192c27f23e7e8970c51a6"
+checksum = "b79c6bdfe612503a526059c05c9ccccbf6bd9530b003673cb863e547fd7c0c9a"
 dependencies = [
  "aws-credential-types",
  "aws-runtime",
  "aws-smithy-async 1.2.1",
- "aws-smithy-http 0.60.8",
+ "aws-smithy-http 0.60.9",
  "aws-smithy-json 0.60.7",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
- "aws-smithy-types 1.1.9",
- "aws-types 1.2.1",
+ "aws-smithy-types 1.2.0",
+ "aws-types 1.3.3",
  "bytes",
  "http 0.2.12",
  "once_cell",
@@ -1763,21 +1760,21 @@ dependencies = [
 
 [[package]]
 name = "aws-sdk-sts"
-version = "1.24.0"
+version = "1.36.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7207ca62206c93e470457babf0512d7b6b97170fdba929a2a2f5682f9f68407c"
+checksum = "32e6ecdb2bd756f3b2383e6f0588dc10a4e65f5d551e70a56e0bfe0c884673ce"
 dependencies = [
  "aws-credential-types",
  "aws-runtime",
  "aws-smithy-async 1.2.1",
- "aws-smithy-http 0.60.8",
+ "aws-smithy-http 0.60.9",
  "aws-smithy-json 0.60.7",
  "aws-smithy-query 0.60.7",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
- "aws-smithy-types 1.1.9",
+ "aws-smithy-types 1.2.0",
  "aws-smithy-xml 0.60.8",
- "aws-types 1.2.1",
+ "aws-types 1.3.3",
  "http 0.2.12",
  "once_cell",
  "regex-lite",
@@ -1817,23 +1814,23 @@ dependencies = [
 
 [[package]]
 name = "aws-sigv4"
-version = "1.2.1"
+version = "1.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "58b56f1cbe6fd4d0c2573df72868f20ab1c125ca9c9dbce17927a463433a2e57"
+checksum = "5df1b0fa6be58efe9d4ccc257df0a53b89cd8909e86591a13ca54817c87517be"
 dependencies = [
  "aws-credential-types",
- "aws-smithy-http 0.60.8",
+ "aws-smithy-http 0.60.9",
  "aws-smithy-runtime-api",
- "aws-smithy-types 1.1.9",
+ "aws-smithy-types 1.2.0",
  "bytes",
  "form_urlencoded",
  "hex",
- "hmac 0.12.1",
+ "hmac",
  "http 0.2.12",
  "http 1.1.0",
  "once_cell",
  "percent-encoding",
- "sha2 0.10.8",
+ "sha2",
  "time",
  "tracing",
 ]
@@ -1907,12 +1904,12 @@ dependencies = [
 
 [[package]]
 name = "aws-smithy-http"
-version = "0.60.8"
+version = "0.60.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4a7de001a1b9a25601016d8057ea16e31a45fdca3751304c8edf4ad72e706c08"
+checksum = "d9cd0ae3d97daa0a2bf377a4d8e8e1362cae590c4a1aad0d40058ebca18eb91e"
 dependencies = [
  "aws-smithy-runtime-api",
- "aws-smithy-types 1.1.9",
+ "aws-smithy-types 1.2.0",
  "bytes",
  "bytes-utils",
  "futures-core",
@@ -1955,7 +1952,7 @@ version = "0.60.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4683df9469ef09468dad3473d129960119a0d3593617542b7d52086c8486f2d6"
 dependencies = [
- "aws-smithy-types 1.1.9",
+ "aws-smithy-types 1.2.0",
 ]
 
 [[package]]
@@ -1974,26 +1971,27 @@ version = "0.60.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f2fbd61ceb3fe8a1cb7352e42689cec5335833cd9f94103a61e98f9bb61c64bb"
 dependencies = [
- "aws-smithy-types 1.1.9",
+ "aws-smithy-types 1.2.0",
  "urlencoding",
 ]
 
 [[package]]
 name = "aws-smithy-runtime"
-version = "1.5.0"
+version = "1.6.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c9ac79e9f3a4d576f3cd4a470a0275b138d9e7b11b1cd514a6858ae0a79dd5bb"
+checksum = "ce87155eba55e11768b8c1afa607f3e864ae82f03caf63258b37455b0ad02537"
 dependencies = [
  "aws-smithy-async 1.2.1",
- "aws-smithy-http 0.60.8",
+ "aws-smithy-http 0.60.9",
  "aws-smithy-runtime-api",
- "aws-smithy-types 1.1.9",
+ "aws-smithy-types 1.2.0",
  "bytes",
  "fastrand 2.1.0",
  "h2",
  "http 0.2.12",
  "http-body 0.4.6",
  "http-body 1.0.0",
+ "httparse",
  "hyper 0.14.28",
  "hyper-rustls 0.24.2",
  "once_cell",
@@ -2006,12 +2004,12 @@ dependencies = [
 
 [[package]]
 name = "aws-smithy-runtime-api"
-version = "1.6.0"
+version = "1.7.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "04ec42c2f5c0e7796a2848dde4d9f3bf8ce12ccbb3d5aa40c52fa0cdd61a1c47"
+checksum = "30819352ed0a04ecf6a2f3477e344d2d1ba33d43e0f09ad9047c12e0d923616f"
 dependencies = [
  "aws-smithy-async 1.2.1",
- "aws-smithy-types 1.1.9",
+ "aws-smithy-types 1.2.0",
  "bytes",
  "http 0.2.12",
  "http 1.1.0",
@@ -2035,9 +2033,9 @@ dependencies = [
 
 [[package]]
 name = "aws-smithy-types"
-version = "1.1.9"
+version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "baf98d97bba6ddaba180f1b1147e202d8fe04940403a95a3f826c790f931bbd1"
+checksum = "cfe321a6b21f5d8eabd0ade9c55d3d0335f3c3157fc2b3e87f05f34b539e4df5"
 dependencies = [
  "base64-simd",
  "bytes",
@@ -2095,15 +2093,14 @@ dependencies = [
 
 [[package]]
 name = "aws-types"
-version = "1.2.1"
+version = "1.3.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a807d90cd50a969b3d95e4e7ad1491fcae13c6e83948d8728363ecc09d66343a"
+checksum = "5221b91b3e441e6675310829fd8984801b772cb1546ef6c0e54dec9f1ac13fef"
 dependencies = [
  "aws-credential-types",
  "aws-smithy-async 1.2.1",
  "aws-smithy-runtime-api",
- "aws-smithy-types 1.1.9",
- "http 0.2.12",
+ "aws-smithy-types 1.2.0",
  "rustc_version",
  "tracing",
 ]
@@ -2313,7 +2310,7 @@ version = "0.10.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "46502ad458c9a52b69d4d4d32775c788b7a1b85e8bc9d482d92250fc0e3f8efe"
 dependencies = [
- "digest 0.10.7",
+ "digest",
 ]
 
 [[package]]
@@ -2329,15 +2326,6 @@ dependencies = [
  "constant_time_eq",
 ]
 
-[[package]]
-name = "block-buffer"
-version = "0.9.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4152116fd6e9dadb291ae18fc1ec3575ed6d84c29642d97890f4b4a3417297e4"
-dependencies = [
- "generic-array",
-]
-
 [[package]]
 name = "block-buffer"
 version = "0.10.4"
@@ -2876,16 +2864,6 @@ dependencies = [
  "typenum",
 ]
 
-[[package]]
-name = "crypto-mac"
-version = "0.11.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "25fab6889090c8133f3deb8f73ba3c65a7f456f66436fc012a1b1e272b1e103e"
-dependencies = [
- "generic-array",
- "subtle",
-]
-
 [[package]]
 name = "csv"
 version = "1.3.0"
@@ -2932,7 +2910,7 @@ dependencies = [
  "cfg-if",
  "cpufeatures",
  "curve25519-dalek-derive",
- "digest 0.10.7",
+ "digest",
  "fiat-crypto",
  "platforms",
  "rustc_version",
@@ -3177,9 +3155,9 @@ dependencies = [
  "hex",
  "itertools 0.12.1",
  "log",
- "md-5 0.10.6",
+ "md-5",
  "regex",
- "sha2 0.10.8",
+ "sha2",
  "unicode-segmentation",
  "uuid",
 ]
@@ -3201,9 +3179,9 @@ dependencies = [
  "hex",
  "itertools 0.12.1",
  "log",
- "md-5 0.10.6",
+ "md-5",
  "regex",
- "sha2 0.10.8",
+ "sha2",
  "unicode-segmentation",
  "uuid",
 ]
@@ -3289,12 +3267,12 @@ dependencies = [
  "indexmap 2.2.6",
  "itertools 0.12.1",
  "log",
- "md-5 0.10.6",
+ "md-5",
  "paste",
  "petgraph",
  "rand 0.8.5",
  "regex",
- "sha2 0.10.8",
+ "sha2",
  "unicode-segmentation",
 ]
 
@@ -3457,10 +3435,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f7c15dda219ed925c28e2387712f9a77ed7253a80b53fce59849067bc0918eb2"
 dependencies = [
  "async-trait",
- "aws-config 1.4.0",
+ "aws-config 1.5.4",
  "aws-credential-types",
  "aws-sdk-dynamodb",
- "aws-sdk-sts 1.24.0",
+ "aws-sdk-sts 1.36.0",
  "aws-smithy-runtime-api",
  "backoff",
  "bytes",
@@ -3614,22 +3592,13 @@ dependencies = [
  "syn 1.0.109",
 ]
 
-[[package]]
-name = "digest"
-version = "0.9.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d3dd60d1080a57a05ab032377049e0591415d2b31afd7028356dbf3cc6dcb066"
-dependencies = [
- "generic-array",
-]
-
 [[package]]
 name = "digest"
 version = "0.10.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
 dependencies = [
- "block-buffer 0.10.4",
+ "block-buffer",
  "const-oid",
  "crypto-common",
  "subtle",
@@ -3644,16 +3613,6 @@ dependencies = [
  "dirs-sys",
 ]
 
-[[package]]
-name = "dirs-next"
-version = "2.0.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b98cf8ebf19c3d1b223e151f99a4f9f0690dca41414773390fc824184ac833e1"
-dependencies = [
- "cfg-if",
- "dirs-sys-next",
-]
-
 [[package]]
 name = "dirs-sys"
 version = "0.4.1"
@@ -3666,17 +3625,6 @@ dependencies = [
  "windows-sys 0.48.0",
 ]
 
-[[package]]
-name = "dirs-sys-next"
-version = "0.1.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4ebda144c4fe02d1f7ea1a7d9641b6fc6b580adcfa024ae48797ecdeb6825b4d"
-dependencies = [
- "libc",
- "redox_users",
- "winapi",
-]
-
 [[package]]
 name = "dlopen2"
 version = "0.7.0"
@@ -3731,7 +3679,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ee27f32b5c5292967d2d4a9d7f1e0b0aed2c15daded5a60300e4abb9d8020bca"
 dependencies = [
  "der 0.7.9",
- "digest 0.10.7",
+ "digest",
  "elliptic-curve",
  "rfc6979",
  "signature 2.2.0",
@@ -3765,7 +3713,7 @@ checksum = "4a3daa8e81a3963a60642bcc1f90a670680bd4a77535faa384e9d1c79d620871"
 dependencies = [
  "curve25519-dalek",
  "ed25519",
- "sha2 0.10.8",
+ "sha2",
  "signature 2.2.0",
  "subtle",
 ]
@@ -3796,7 +3744,7 @@ checksum = "b5e6043086bf7973472e0c7dff2142ea0b680d30e18d9cc40f267efbf222bd47"
 dependencies = [
  "base16ct",
  "crypto-bigint",
- "digest 0.10.7",
+ "digest",
  "ff",
  "generic-array",
  "group",
@@ -4710,17 +4658,7 @@ version = "0.12.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7b5f8eb2ad728638ea2c7d47a21db23b7b58a72ed6a38256b8a1849f15fbbdf7"
 dependencies = [
- "hmac 0.12.1",
-]
-
-[[package]]
-name = "hmac"
-version = "0.11.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2a2a2320eb7ec0ebe8da8f744d7812d9fc4cb4d09344ac01898dbcb6a20ae69b"
-dependencies = [
- "crypto-mac",
- "digest 0.9.0",
+ "hmac",
 ]
 
 [[package]]
@@ -4729,7 +4667,7 @@ version = "0.12.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e"
 dependencies = [
- "digest 0.10.7",
+ "digest",
 ]
 
 [[package]]
@@ -4744,7 +4682,7 @@ version = "1.1.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3688e69b38018fec1557254f64c8dc2cc8ec502890182f395dbb0aa997aa5735"
 dependencies = [
- "digest 0.10.7",
+ "digest",
 ]
 
 [[package]]
@@ -4753,7 +4691,7 @@ version = "1.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e4ce1f4656bae589a3fab938f9f09bf58645b7ed01a2c5f8a3c238e01a4ef78a"
 dependencies = [
- "digest 0.10.7",
+ "digest",
 ]
 
 [[package]]
@@ -5246,7 +5184,7 @@ dependencies = [
  "ecdsa",
  "elliptic-curve",
  "once_cell",
- "sha2 0.10.8",
+ "sha2",
  "signature 2.2.0",
 ]
 
@@ -5645,17 +5583,6 @@ version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94"
 
-[[package]]
-name = "md-5"
-version = "0.9.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7b5a279bb9607f9f53c22d496eade00d138d1bdcccd07d74650387cf94942a15"
-dependencies = [
- "block-buffer 0.9.0",
- "digest 0.9.0",
- "opaque-debug",
-]
-
 [[package]]
 name = "md-5"
 version = "0.10.6"
@@ -5663,7 +5590,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d89e7ee0cfbedfc4da3340218492196241d89eefb6dab27de5df917a6d2e78cf"
 dependencies = [
  "cfg-if",
- "digest 0.10.7",
+ "digest",
 ]
 
 [[package]]
@@ -6032,7 +5959,7 @@ dependencies = [
  "humantime",
  "hyper 0.14.28",
  "itertools 0.12.1",
- "md-5 0.10.6",
+ "md-5",
  "parking_lot 0.12.2",
  "percent-encoding",
  "quick-xml",
@@ -6055,12 +5982,6 @@ version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3fdb12b2476b595f9358c5161aa467c2438859caa136dec86c26fdd2efe17b92"
 
-[[package]]
-name = "opaque-debug"
-version = "0.3.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381"
-
 [[package]]
 name = "openapiv3"
 version = "2.0.0"
@@ -6187,7 +6108,7 @@ dependencies = [
  "ecdsa",
  "elliptic-curve",
  "primeorder",
- "sha2 0.10.8",
+ "sha2",
 ]
 
 [[package]]
@@ -6199,7 +6120,7 @@ dependencies = [
  "ecdsa",
  "elliptic-curve",
  "primeorder",
- "sha2 0.10.8",
+ "sha2",
 ]
 
 [[package]]
@@ -6415,7 +6336,7 @@ checksum = "d7a240022f37c361ec1878d646fc5b7d7c4d28d5946e1a80ad5a7a4f4ca0bdcd"
 dependencies = [
  "once_cell",
  "pest",
- "sha2 0.10.8",
+ "sha2",
 ]
 
 [[package]]
@@ -6638,11 +6559,11 @@ dependencies = [
  "byteorder",
  "bytes",
  "fallible-iterator 0.2.0",
- "hmac 0.12.1",
- "md-5 0.10.6",
+ "hmac",
+ "md-5",
  "memchr",
  "rand 0.8.5",
- "sha2 0.10.8",
+ "sha2",
  "stringprep",
 ]
 
@@ -7370,7 +7291,7 @@ version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8dd2a808d456c4a54e300a23e9f5a67e122c3024119acbfd73e3bf664491cb2"
 dependencies = [
- "hmac 0.12.1",
+ "hmac",
  "subtle",
 ]
 
@@ -7427,7 +7348,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "094052d5470cbcef561cb848a7209968c9f12dfa6d668f4bca048ac5de51099c"
 dependencies = [
  "byteorder",
- "digest 0.10.7",
+ "digest",
  "num-bigint-dig",
  "num-integer",
  "num-iter",
@@ -7489,88 +7410,6 @@ dependencies = [
  "url",
 ]
 
-[[package]]
-name = "rusoto_core"
-version = "0.48.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1db30db44ea73551326269adcf7a2169428a054f14faf9e1768f2163494f2fa2"
-dependencies = [
- "async-trait",
- "base64 0.13.1",
- "bytes",
- "crc32fast",
- "futures",
- "http 0.2.12",
- "hyper 0.14.28",
- "hyper-tls",
- "lazy_static",
- "log",
- "rusoto_credential",
- "rusoto_signature",
- "rustc_version",
- "serde",
- "serde_json",
- "tokio",
- "xml-rs",
-]
-
-[[package]]
-name = "rusoto_credential"
-version = "0.48.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ee0a6c13db5aad6047b6a44ef023dbbc21a056b6dab5be3b79ce4283d5c02d05"
-dependencies = [
- "async-trait",
- "chrono",
- "dirs-next",
- "futures",
- "hyper 0.14.28",
- "serde",
- "serde_json",
- "shlex",
- "tokio",
- "zeroize",
-]
-
-[[package]]
-name = "rusoto_s3"
-version = "0.48.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7aae4677183411f6b0b412d66194ef5403293917d66e70ab118f07cc24c5b14d"
-dependencies = [
- "async-trait",
- "bytes",
- "futures",
- "rusoto_core",
- "xml-rs",
-]
-
-[[package]]
-name = "rusoto_signature"
-version = "0.48.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a5ae95491c8b4847931e291b151127eccd6ff8ca13f33603eb3d0035ecb05272"
-dependencies = [
- "base64 0.13.1",
- "bytes",
- "chrono",
- "digest 0.9.0",
- "futures",
- "hex",
- "hmac 0.11.0",
- "http 0.2.12",
- "hyper 0.14.28",
- "log",
- "md-5 0.9.1",
- "percent-encoding",
- "pin-project-lite",
- "rusoto_credential",
- "rustc_version",
- "serde",
- "sha2 0.9.9",
- "tokio",
-]
-
 [[package]]
 name = "rusqlite"
 version = "0.31.0"
@@ -7643,7 +7482,7 @@ version = "7.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9d38ff6bf570dc3bb7100fce9f7b60c33fa71d80e88da3f2580df4ff2bdded74"
 dependencies = [
- "sha2 0.10.8",
+ "sha2",
  "walkdir",
 ]
 
@@ -7653,7 +7492,7 @@ version = "8.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "86f69089032567ffff4eada41c573fc43ff466c7db7c5688b2e7969584345581"
 dependencies = [
- "sha2 0.10.8",
+ "sha2",
  "walkdir",
 ]
 
@@ -8139,7 +7978,7 @@ checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba"
 dependencies = [
  "cfg-if",
  "cpufeatures",
- "digest 0.10.7",
+ "digest",
 ]
 
 [[package]]
@@ -8148,19 +7987,6 @@ version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ae1a47186c03a32177042e55dbc5fd5aee900b8e0069a8d70fba96a9375cd012"
 
-[[package]]
-name = "sha2"
-version = "0.9.9"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4d58a1e1bf39749807d89cf2d98ac2dfa0ff1cb3faa38fbb64dd88ac8013d800"
-dependencies = [
- "block-buffer 0.9.0",
- "cfg-if",
- "cpufeatures",
- "digest 0.9.0",
- "opaque-debug",
-]
-
 [[package]]
 name = "sha2"
 version = "0.10.8"
@@ -8169,7 +7995,7 @@ checksum = "793db75ad2bcafc3ffa7c68b215fee268f537982cd901d132f89c6343f3a3dc8"
 dependencies = [
  "cfg-if",
  "cpufeatures",
- "digest 0.10.7",
+ "digest",
 ]
 
 [[package]]
@@ -8200,12 +8026,6 @@ dependencies = [
  "dirs",
 ]
 
-[[package]]
-name = "shlex"
-version = "1.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64"
-
 [[package]]
 name = "signal-hook-registry"
 version = "1.4.2"
@@ -8233,7 +8053,7 @@ version = "1.6.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "74233d3b3b2f6d4b006dc19dee745e73e2a6bfb6f93607cd3b02bd5b00797d7c"
 dependencies = [
- "digest 0.10.7",
+ "digest",
  "rand_core 0.6.4",
 ]
 
@@ -8243,7 +8063,7 @@ version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de"
 dependencies = [
- "digest 0.10.7",
+ "digest",
  "rand_core 0.6.4",
 ]
 
@@ -10024,12 +9844,6 @@ dependencies = [
  "web-sys",
 ]
 
-[[package]]
-name = "xml-rs"
-version = "0.8.20"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "791978798f0597cfc70478424c2b4fdc2b7a8024aaff78497ef00f24ef674193"
-
 [[package]]
 name = "xmlparser"
 version = "0.13.6"
diff --git a/Cargo.toml b/Cargo.toml
index cf4841a3b..1b72c95f0 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -51,6 +51,7 @@ deltalake = { version = "0.17.3" }
 cornucopia = { version = "0.9.0" }
 cornucopia_async = {version = "0.6.0"}
 deadpool-postgres = "0.12"
+
 [profile.release]
 debug = 1
 
diff --git a/crates/arroyo-server-common/src/lib.rs b/crates/arroyo-server-common/src/lib.rs
index cc533cc7a..8686b4e4f 100644
--- a/crates/arroyo-server-common/src/lib.rs
+++ b/crates/arroyo-server-common/src/lib.rs
@@ -61,7 +61,9 @@ pub fn init_logging_with_filter(_name: &str, filter: EnvFilter) -> WorkerGuard {
         eprintln!("Failed to initialize log tracer {:?}", e);
     }
 
-    let filter = filter.add_directive("refinery_core=warn".parse().unwrap());
+    let filter = filter
+        .add_directive("refinery_core=warn".parse().unwrap())
+        .add_directive("aws_config::profile::credentials=warn".parse().unwrap());
 
     let (nonblocking, guard) = tracing_appender::non_blocking(std::io::stderr());
 
diff --git a/crates/arroyo-state/src/lib.rs b/crates/arroyo-state/src/lib.rs
index 0149dc1da..1b1fc23f3 100644
--- a/crates/arroyo-state/src/lib.rs
+++ b/crates/arroyo-state/src/lib.rs
@@ -1,4 +1,4 @@
-use anyhow::Result;
+use anyhow::{Context, Result};
 use arrow_array::RecordBatch;
 use arroyo_rpc::grpc::rpc::{
     CheckpointMetadata, ExpiringKeyedTimeTableConfig, GlobalKeyedTableConfig,
@@ -9,12 +9,15 @@ use async_trait::async_trait;
 use bincode::config::Configuration;
 use bincode::{Decode, Encode};
 
+use arroyo_rpc::config::config;
 use arroyo_rpc::df::ArroyoSchema;
+use arroyo_storage::StorageProvider;
 use prost::Message;
 use std::collections::hash_map::DefaultHasher;
 use std::collections::HashMap;
 use std::hash::{Hash, Hasher};
 use std::ops::RangeInclusive;
+use std::sync::Arc;
 use std::time::{Duration, SystemTime};
 
 pub mod checkpoint_state;
@@ -160,3 +163,25 @@ pub fn hash_key<K: Hash>(key: &K) -> u64 {
     key.hash(&mut hasher);
     hasher.finish()
 }
+
+static STORAGE_PROVIDER: tokio::sync::OnceCell<Arc<StorageProvider>> =
+    tokio::sync::OnceCell::const_new();
+
+pub(crate) async fn get_storage_provider() -> Result<&'static Arc<StorageProvider>> {
+    // TODO: this should be encoded in the config so that the controller doesn't need
+    // to be synchronized with the workers
+
+    STORAGE_PROVIDER
+        .get_or_try_init(|| async {
+            let storage_url = &config().checkpoint_url;
+
+            StorageProvider::for_url(storage_url)
+                .await
+                .context(format!(
+                    "failed to construct checkpoint backend for URL {}",
+                    storage_url
+                ))
+                .map(Arc::new)
+        })
+        .await
+}
diff --git a/crates/arroyo-state/src/parquet.rs b/crates/arroyo-state/src/parquet.rs
index ef4312385..4b7c94974 100644
--- a/crates/arroyo-state/src/parquet.rs
+++ b/crates/arroyo-state/src/parquet.rs
@@ -1,12 +1,11 @@
 use crate::tables::expiring_time_key_map::ExpiringTimeKeyTable;
 use crate::tables::global_keyed_map::GlobalKeyedTable;
 use crate::tables::{CompactionConfig, ErasedTable};
-use crate::BackingStore;
-use anyhow::{bail, Context, Result};
+use crate::{get_storage_provider, BackingStore};
+use anyhow::{bail, Result};
 use arroyo_rpc::grpc::rpc::{
     CheckpointMetadata, OperatorCheckpointMetadata, TableCheckpointMetadata,
 };
-use arroyo_storage::StorageProvider;
 use futures::stream::FuturesUnordered;
 use futures::StreamExt;
 
@@ -23,17 +22,6 @@ use tracing::{debug, info};
 pub const FULL_KEY_RANGE: RangeInclusive<u64> = 0..=u64::MAX;
 pub const GENERATIONS_TO_COMPACT: u32 = 1; // only compact generation 0 files
 
-async fn get_storage_provider() -> anyhow::Result<StorageProvider> {
-    // TODO: this should be encoded in the config so that the controller doesn't need
-    // to be synchronized with the workers
-    let storage_url = &config().checkpoint_url;
-
-    StorageProvider::for_url(storage_url).await.context(format!(
-        "failed to construct checkpoint backend for URL {}",
-        storage_url
-    ))
-}
-
 pub struct ParquetBackend;
 
 fn base_path(job_id: &str, epoch: u32) -> String {
@@ -178,11 +166,11 @@ impl ParquetBackend {
             Self::load_operator_metadata(&job_id, &operator_id, epoch)
                 .await?
                 .expect("expect operator metadata to still be present");
-        let storage_provider = Arc::new(get_storage_provider().await?);
+        let storage_provider = get_storage_provider().await?;
         let compaction_config = CompactionConfig {
-            storage_provider,
             compact_generations: vec![0].into_iter().collect(),
             min_compaction_epochs: min_files_to_compact,
+            storage_provider: Arc::clone(storage_provider),
         };
         let operator_metadata = operator_checkpoint_metadata.operator_metadata.unwrap();
 
diff --git a/crates/arroyo-state/src/tables/table_manager.rs b/crates/arroyo-state/src/tables/table_manager.rs
index 39d98f495..7a5c8bf0d 100644
--- a/crates/arroyo-state/src/tables/table_manager.rs
+++ b/crates/arroyo-state/src/tables/table_manager.rs
@@ -21,7 +21,7 @@ use tokio::sync::{
 use arroyo_rpc::config::config;
 use tracing::{debug, error, info, warn};
 
-use crate::{tables::global_keyed_map::GlobalKeyedTable, StateMessage};
+use crate::{get_storage_provider, tables::global_keyed_map::GlobalKeyedTable, StateMessage};
 use crate::{CheckpointMessage, TableData};
 
 use super::expiring_time_key_map::{
@@ -225,20 +225,6 @@ impl BackendWriter {
     }
 }
 
-async fn get_storage_provider() -> anyhow::Result<StorageProviderRef> {
-    // TODO: this should be encoded in the config so that the controller doesn't need
-    // to be synchronized with the workers
-
-    Ok(Arc::new(
-        StorageProvider::for_url(&config().checkpoint_url)
-            .await
-            .context(format!(
-                "failed to construct checkpoint backend for URL {}",
-                config().checkpoint_url
-            ))?,
-    ))
-}
-
 impl TableManager {
     pub async fn new(
         task_info: TaskInfoRef,
@@ -320,7 +306,7 @@ impl TableManager {
             tables,
             writer,
             task_info,
-            storage,
+            storage: Arc::clone(storage),
             caches: HashMap::new(),
         })
     }
diff --git a/crates/arroyo-storage/Cargo.toml b/crates/arroyo-storage/Cargo.toml
index e3542ea39..8b8664e39 100644
--- a/crates/arroyo-storage/Cargo.toml
+++ b/crates/arroyo-storage/Cargo.toml
@@ -12,10 +12,9 @@ arroyo-types = { path = "../arroyo-types" }
 arroyo-rpc = { path = "../arroyo-rpc" }
 bytes = "1.4.0"
 tracing = "0.1"
-# used only for getting local AWS credentials; can be removed once we have a
-# better way to do this
-rusoto_core = "0.48.0"
 
+aws-credential-types = "1.2.0"
+aws-config = { version = "1.5.4" }
 rand = "0.8"
 object_store = {workspace = true, features = ["aws", "gcp"]}
 regex = "1.9.5"
diff --git a/crates/arroyo-storage/src/aws.rs b/crates/arroyo-storage/src/aws.rs
index 9adbe3d59..e5bba4d90 100644
--- a/crates/arroyo-storage/src/aws.rs
+++ b/crates/arroyo-storage/src/aws.rs
@@ -1,14 +1,11 @@
-use std::sync::Arc;
-
-use object_store::{aws::AwsCredential, CredentialProvider};
-use rusoto_core::credential::{
-    AutoRefreshingProvider, ChainProvider, ProfileProvider, ProvideAwsCredentials,
-};
-
 use crate::StorageError;
+use aws_config::BehaviorVersion;
+use aws_credential_types::provider::ProvideCredentials;
+use object_store::{aws::AwsCredential, CredentialProvider};
+use std::sync::Arc;
 
 pub struct ArroyoCredentialProvider {
-    provider: AutoRefreshingProvider<ChainProvider>,
+    provider: aws_credential_types::provider::SharedCredentialsProvider,
 }
 
 impl std::fmt::Debug for ArroyoCredentialProvider {
@@ -18,38 +15,47 @@ impl std::fmt::Debug for ArroyoCredentialProvider {
 }
 
 impl ArroyoCredentialProvider {
-    pub fn try_new() -> Result<Self, StorageError> {
-        let inner: AutoRefreshingProvider<ChainProvider> =
-            AutoRefreshingProvider::new(ChainProvider::new())
-                .map_err(|e| StorageError::CredentialsError(e.to_string()))?;
-
-        Ok(Self { provider: inner })
+    pub async fn try_new() -> Result<Self, StorageError> {
+        let config = aws_config::defaults(BehaviorVersion::latest()).load().await;
+
+        let credentials = config
+            .credentials_provider()
+            .ok_or_else(|| {
+                StorageError::CredentialsError(
+                    "Unable to load S3 credentials from environment".to_string(),
+                )
+            })?
+            .clone();
+
+        Ok(Self {
+            provider: credentials,
+        })
     }
 
     pub async fn default_region() -> Option<String> {
-        ProfileProvider::region().ok()?
+        aws_config::defaults(BehaviorVersion::latest())
+            .load()
+            .await
+            .region()
+            .map(|r| r.to_string())
     }
 }
 
 #[async_trait::async_trait]
 impl CredentialProvider for ArroyoCredentialProvider {
-    #[doc = " The type of credential returned by this provider"]
     type Credential = AwsCredential;
 
-    /// Return a credential
     async fn get_credential(&self) -> object_store::Result<Arc<Self::Credential>> {
-        let credentials =
-            self.provider
-                .credentials()
-                .await
-                .map_err(|err| object_store::Error::Generic {
-                    store: "s3",
-                    source: Box::new(err),
-                })?;
+        let creds = self.provider.provide_credentials().await.map_err(|e| {
+            object_store::Error::Generic {
+                store: "S3",
+                source: Box::new(e),
+            }
+        })?;
         Ok(Arc::new(AwsCredential {
-            key_id: credentials.aws_access_key_id().to_string(),
-            secret_key: credentials.aws_secret_access_key().to_string(),
-            token: credentials.token().clone(),
+            key_id: creds.access_key_id().to_string(),
+            secret_key: creds.secret_access_key().to_string(),
+            token: creds.session_token().map(ToString::to_string),
         }))
     }
 }
diff --git a/crates/arroyo-storage/src/lib.rs b/crates/arroyo-storage/src/lib.rs
index 1e046466b..a466a9095 100644
--- a/crates/arroyo-storage/src/lib.rs
+++ b/crates/arroyo-storage/src/lib.rs
@@ -16,12 +16,9 @@ use object_store::multipart::PartId;
 use object_store::path::Path;
 use object_store::{aws::AmazonS3Builder, local::LocalFileSystem, ObjectStore};
 use object_store::{CredentialProvider, MultipartId};
-use once_cell::sync::Lazy;
 use regex::{Captures, Regex};
-use std::time::{Duration, Instant};
 use thiserror::Error;
-use tokio::sync::RwLock;
-use tracing::{debug, error, trace};
+use tracing::{debug, error};
 
 mod aws;
 
@@ -296,23 +293,11 @@ fn last<I: Sized, const COUNT: usize>(opts: [Option<I>; COUNT]) -> Option<I> {
 }
 
 pub async fn get_current_credentials() -> Result<Arc<AwsCredential>, StorageError> {
-    let provider = ArroyoCredentialProvider::try_new()?;
+    let provider = ArroyoCredentialProvider::try_new().await?;
     let credentials = provider.get_credential().await?;
     Ok(credentials)
 }
 
-static OBJECT_STORE_CACHE: Lazy<RwLock<HashMap<String, CacheEntry<Arc<dyn ObjectStore>>>>> =
-    Lazy::new(Default::default);
-
-struct CacheEntry<T> {
-    value: T,
-    inserted_at: Instant,
-}
-
-// The bearer token should last for 3600 seconds,
-// but regenerating it every 5 minutes to avoid token expiry
-const GCS_CACHE_TTL: Duration = Duration::from_secs(5 * 60);
-
 impl StorageProvider {
     pub async fn for_url(url: &str) -> Result<Self, StorageError> {
         Self::for_url_with_options(url, HashMap::new()).await
@@ -360,11 +345,6 @@ impl StorageProvider {
         Ok(key.clone())
     }
 
-    pub async fn url_exists(url: &str) -> Result<bool, StorageError> {
-        let provider = Self::for_url(url).await?;
-        provider.exists("").await
-    }
-
     async fn construct_s3(
         mut config: S3Config,
         options: HashMap<String, String>,
@@ -386,7 +366,7 @@ impl StorageProvider {
 
         if !aws_key_manually_set {
             let credentials: Arc<ArroyoCredentialProvider> =
-                Arc::new(ArroyoCredentialProvider::try_new()?);
+                Arc::new(ArroyoCredentialProvider::try_new().await?);
             builder = builder.with_credentials(credentials);
         }
 
@@ -444,45 +424,6 @@ impl StorageProvider {
         })
     }
 
-    async fn get_or_create_object_store(
-        builder: GoogleCloudStorageBuilder,
-        bucket: &str,
-    ) -> Result<Arc<dyn ObjectStore>, StorageError> {
-        let mut cache = OBJECT_STORE_CACHE.write().await;
-
-        if let Some(entry) = cache.get(bucket) {
-            if entry.inserted_at.elapsed() < GCS_CACHE_TTL {
-                trace!(
-                    "Cache hit - using cached object store for bucket {}",
-                    bucket
-                );
-                return Ok(entry.value.clone());
-            } else {
-                debug!(
-                    "Cache expired - constructing new object store for bucket {}",
-                    bucket
-                );
-            }
-        } else {
-            debug!(
-                "Cache miss - constructing new object store for bucket {}",
-                bucket
-            );
-        }
-
-        let new_store = Arc::new(builder.build().map_err(Into::<StorageError>::into)?);
-
-        cache.insert(
-            bucket.to_string(),
-            CacheEntry {
-                value: new_store.clone(),
-                inserted_at: Instant::now(),
-            },
-        );
-
-        Ok(new_store)
-    }
-
     async fn construct_gcs(config: GCSConfig) -> Result<Self, StorageError> {
         let mut builder = GoogleCloudStorageBuilder::from_env().with_bucket_name(&config.bucket);
 
@@ -498,7 +439,7 @@ impl StorageProvider {
 
         let object_store_base_url = format!("https://{}.storage.googleapis.com", config.bucket);
 
-        let object_store = Self::get_or_create_object_store(builder, &config.bucket).await?;
+        let object_store = Arc::new(builder.build()?);
 
         Ok(Self {
             config: BackendConfig::GCS(config),
diff --git a/crates/arroyo-worker/Cargo.toml b/crates/arroyo-worker/Cargo.toml
index 246d21a11..2df2dbcaa 100644
--- a/crates/arroyo-worker/Cargo.toml
+++ b/crates/arroyo-worker/Cargo.toml
@@ -52,11 +52,7 @@ parquet = { workspace = true, features = ["async"]}
 arrow-array = { workspace = true}
 arrow-json = { workspace = true }
 
-aws-sdk-kinesis = { version = "0.21", default-features = false, features = ["rt-tokio", "native-tls"] }
-aws-config = { version = "0.51", default-features = false, features = ["rt-tokio", "native-tls"] }
 uuid = {version = "1.4.1", features = ["v4"]}
-rusoto_core = "0.48.0"
-rusoto_s3 = "0.48.0"
 
 tonic = { workspace = true }
 prost = "0.12"