Repository: ArroyoSystems/arroyo (language: rust)

Issue:
Allow to configure worker's IRSA through worker's ServiceAccount annotation
When it comes to connecting to an S3 bucket, the documentation and the blog suggest to either use `aws_access_key_id` / `aws_secret_access_key` or to grant permission to the Kubernetes node IAM role. 

I believe this misses a more granular permission mechanism and I suggest to add to possibility to define the annotations of the `ServiceAccount` the workers run with so that an IRSA can be linked, as already done for the controller's ServiceAccount.

An even more granular possibility would be to configure the used IRSA in the [backend configuration](https://doc.arroyo.dev/connectors/filesystem#additional-backend-configuration): either configure `storage.aws_secret_access_key` + `storage.aws_access_key_id` or (new) `storage.iam_role_for_service_account` (this means we know the namespace and the name of the `ServiceAccount` as it is required to create the IRSA).



Discussion / hints:

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index 0a2d3d287..18b32e626 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 0606476ec..72899cc1e 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 82625498a..7412b055b 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 8c0a90697..21035df98 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"


Test patch: