feat(rest): Support refreshing vended storage credentials by zakariya-s · Pull Request #2932 · apache/iceberg-rust

zakariya-s · 2026-07-30T21:32:27Z

Which issue does this PR close?

Closes Support refreshing vended storage credentials for REST catalog tables #2931.

What changes are included in this PR?

This PR adds support for refreshing short-lived storage credentials vended by REST catalogs:

Adds a backend-independent StorageCredentialProvider interface to FileIO
Implements a REST credential provider for AWS S3 and GCS refresh endpoints
Uses table-scoped tokens and header.* properties for refresh requests
Caches credentials independently by cloud and selects credentials using the longest matching storage prefix
Adds jittered failure backoff while a cached credential remains valid
Adapts refreshed credentials into the OpenDAL/reqsign S3 and GCS credential providers
Redacts credential-bearing configuration from Debug output

Azure credential refresh is not included because the current OpenDAL Azure backend does not expose the credential-provider and expiry hooks required for safe refresh.

Are these changes tested?

Yes

zakariya-s · 2026-07-31T10:13:47Z

 /// Disable header redaction in error logs (defaults to false for security)
 pub const REST_CATALOG_PROP_DISABLE_HEADER_REDACTION: &str = "disable-header-redaction";
+/// Identifier for a server-side scan plan associated with credential requests.
+pub const REST_CATALOG_PROP_SCAN_PLAN_ID: &str = "rest.scan.plan-id";


Server-side planning isn't supported yet so this will always be empty anyway. I don't think it's a problem to keep it until it eventually is supported

zakariya-s · 2026-07-31T10:16:09Z

+        let config = response
+            .config
+            .into_iter()
+            .chain(self.user_config.props.clone())
+            .collect();
+        let file_io = self
+            .load_file_io(Some(metadata_location), Some(config))
+            .await?;


Drive-by fix since this is inconsistent with create_table() and load_table()

https://github.com/zakariya-s/iceberg-rust/blob/26d7fcc6d299a8e814b6cceca480409b2c2cc3c9/crates/catalog/rest/src/catalog.rs#L875-L883

zakariya-s · 2026-07-31T10:28:23Z

+    if let Some(no_auth) = m.remove(GCS_NO_AUTH)
+        && is_truthy(no_auth.to_lowercase().as_str())
+    {


Drive-by fix since this looked pretty bad. AWS did this correctly, but GCS would disable this even if gcs.no-auth was set to true

zakariya-s · 2026-07-31T10:30:28Z

+                let url = url::Url::parse(path).map_err(|e| {
+                    Error::new(
+                        ErrorKind::DataInvalid,
+                        format!("Invalid gcs url: {path}: {e}"),
+                    )
+                })?;


S3 had this validation before above but GCS didn't, so another drive-by fix

zakariya-s · 2026-07-31T10:35:11Z

+/// `reqsign` [`Timestamp`](reqsign_core::time::Timestamp) used on backend
+/// credential types (e.g. `AwsCredential::expires_in`, `google::Token::expires_at`).
+#[cfg(any(feature = "opendal-s3", feature = "opendal-gcs"))]
+pub(crate) fn system_time_to_timestamp(


Didn't want to rely on directly on reqsign's Timestamp, but I'm happy to hear different opinions

zakariya-s · 2026-07-31T10:37:37Z

+/// It contains the location schemes it backs, the property keys it is configured
+/// with, and how to parse its credential. The generic provider stays free of
+/// any per-cloud knowledge.
+struct CloudRefresh {


This could also be made into a trait I guess, and we could split aws and gcp support into different modules, but I thought it was small enough to keep it as a struct and make consts for the different cloud providers

zakariya-s · 2026-08-05T16:41:47Z

Hi @mbutrovich! Would it be possible to get a first-round review of this PR when you have time please?

mbutrovich · 2026-08-05T17:04:39Z

Hi @mbutrovich! Would it be possible to get a first-round review of this PR when you have time please?

Yep, it's in my queue! Just slammed with review requests :(

Thanks for your patience!

mbutrovich

First pass, thanks for tackling this @zakariya-s. I have specific feedback, and also some ideas how we might break this up for other reviewers since this is a lot to review in one pass. A split along the layers already present in the diff looks fairly clean and, other than one ordering constraint, each piece is independently testable rather than a bare stub:

The Debug-redaction hardening (RestCatalogConfig, HttpClient, StorageConfig, LoadTableResult, StorageCredential, OpenDalResolvingStorage, plus the is_sensitive_header broadening) and the two small unrelated fixes riding along (the register_table config merge, the GCS no-auth truthy parsing). None of this depends on credential refresh existing, and it already has its own tests.
The StorageCredentialProvider trait and credential types in the iceberg crate, plus StorageFactory::build_with_credentials with its safe default (errors if a provider is supplied and the factory has not opted in). Covered by finding 5. No behavior change for existing backends.
OpenDAL S3 consuming the trait (the adapter, the anonymous-access guard, the delete_stream change). Testable on its own with a hand-rolled provider, same as the tests already in this diff, without needing anything from the REST side.
OpenDAL GCS consuming the trait, same shape as 3, independent of it.

Finding 1 (lost delete batching) sits in the shared uses_dynamic_credentials/delete_stream code in lib.rs rather than in s3.rs or gcs.rs specifically, so it isn't purely a 3-or-4 problem: whichever of the two lands first introduces that shared mechanism, and the other reuses it as-is, so the fix only needs to happen once.

The REST catalog fetch/cache/jitter/backoff logic (RestVendedCredentialProvider, CloudRefresh, resolve_endpoint, the table-scoped client for_table). Covered by findings 2, 3, 4, and the root cause of 6 (the fresh-HttpClient-per-table-scoped-provider behavior lives in client.rs/credential.rs, both part of this PR). This can be written and tested against the trait directly with mockito, same as the tests already in this diff, without touching OpenDAL at all.
Wiring RestCatalog::load_file_io to actually attach the provider (catalog.rs:561-565). Covered by finding 7 (the inline path at that call site) and the rest of finding 6 (this is the call site that turns "every load_file_io call re-runs the OAuth handshake" from a latent property of for_table into something that fires on every load_table/create_table/register_table).

This piece should land last on purpose, not just for tidiness: client.refresh-credentials-endpoint and gcs.oauth2.refresh-credentials-enabled are property keys the Java client and the REST spec already define, so a production catalog that's Java-interoperable may already be sending them today regardless of which client is asking. If the wiring lands before both the S3 and GCS sides can consume a provider, every existing rust client hitting such a catalog would start failing table loads on S3 or GCS the moment that PR merges, since the default build_with_credentials errors whenever a provider is supplied to a factory that hasn't opted in. So 3 and 4 both need to land before 6, even though 3, 4, and 5 are otherwise independent of each other and can be reviewed in any order.

#2931 is currently a single feature request rather than a tracking issue for a multi-PR stack. Worth turning it into one, or opening a separate tracking issue with a checklist for the pieces above, so reviewers can see the whole plan and where a given PR sits in it before reviewing any single piece.

mbutrovich · 2026-08-05T20:00:19Z

+    fn uses_dynamic_credentials(&self, path: &str) -> bool {
+        match self {
+            #[cfg(feature = "opendal-s3")]
+            OpenDalStorage::S3 {
+                credential_provider: Some(provider),
+                ..
+            } => provider.supports_path(path),
+            #[cfg(feature = "opendal-gcs")]
+            OpenDalStorage::Gcs {


crates/storage/opendal/src/lib.rs:423-431 (uses_dynamic_credentials) and :610-628 (delete_stream).

When a path is served by a credential provider, delete_stream skips the shared per-bucket Deleter and instead calls create_operator + a single op.delete(relative_path) per path, sequentially, inside the stream loop. The non-dynamic branch batches deletes through OpenDAL's Deleter (which can use bulk delete APIs); the dynamic branch does neither batching nor concurrency, and rebuilds the operator from scratch for every single file.

For expire_snapshots/purge on a table with vended-credential refresh enabled, this turns what would be a handful of batched multi-object delete calls into one HTTP round trip per file, plus an operator-construction cost per file. The code comment explains why deletes can't share a Deleter across different credential-prefix scopes (correctness: batch_key_for_path only groups by bucket, not by credential scope), but the fix taken forfeits batching entirely rather than partially, i.e. grouping deletes by (bucket, matched credential prefix) instead of just bucket would preserve batched delete within each credential-scope group. Was that considered?

Yeah I left this in the initial commit for sake of simplicity, we now batch by DeleteBatchKey which includes the credential scope so batching behaviour is now restored

mbutrovich · 2026-08-05T20:00:58Z

+        let refreshed = self.fetch(configured).await.and_then(|entries| {
+            let credential = longest_prefix_match(&entries, path)
+                .filter(|entry| entry.is_unexpired(SystemTime::now()))
+                .map(|entry| entry.credential.clone())
+                .ok_or_else(|| {
+                    Error::new(
+                        ErrorKind::Unexpected,
+                        format!("no unexpired vended credential matches storage location: {path}"),
+                    )
+                })?;
+            Ok((entries, credential))
+        });
+
+        match refreshed {
+            Ok((entries, credential)) => {
+                let mut cache = configured.cache.lock().await;
+                cache.entries = entries;
+                cache.consecutive_failures = 0;
+                cache.retry_not_before = None;
+                Ok(credential)
+            }
+            Err(fetch_error) => {
+                let mut cache = configured.cache.lock().await;
+                cache.consecutive_failures = cache.consecutive_failures.saturating_add(1);
+                cache.retry_not_before =
+                    Instant::now().checked_add(failure_backoff(cache.consecutive_failures));
+
+                // Graceful degradation: while the cached credential remains
+                // usable, serve it and retry after jittered backoff. Expired
+                // credentials are never served.
+                fallback
+                    .filter(|entry| entry.is_unexpired(SystemTime::now()))
+                    .map(|entry| entry.credential)
+                    .ok_or(fetch_error)
+            }
+        }


crates/catalog/rest/src/credential.rs:265-300 (refresh_credential), contrast with S3FileIO.refreshStorageCredentials()/GCSFileIO.refreshStorageCredentials() in Java.

Java's actual multi-prefix refresh (used by both S3FileIO and GCSFileIO, not the single-credential VendedCredentialsProvider/OAuth2RefreshCredentialsHandler used as an SDK credentials provider) is unconditional: on each scheduled refresh it fetches the credentials endpoint once, keeps every entry matching the cloud's root prefix, and replaces storageCredentials wholesale — no per-path filtering happens at refresh time at all.

The Rust refresh_credential instead does per-path filtering inline: it fetches all entries for a cloud, then immediately narrows to longest_prefix_match(&entries, path).filter(unexpired) for this specific call's path. If that narrowing yields nothing (no entry covers this path, or the covering entry happens to already be expired), the whole outcome is treated as Err and:

the freshly-fetched entries are never written to cache.entries (only the Ok((entries, credential)) branch at line 279-284 updates the cache) — so if the response contained entries for other prefixes (as the existing longest_prefix_match_ignores_freshness test exercises), they're thrown away even though a subsequent call for a different, valid path would have to fetch them all over again;

cache.consecutive_failures is incremented and retry_not_before backoff is armed (lines 288-290) even though the catalog responded successfully — it just didn't vend anything for this path.

Given Java's model of "cache everything the endpoint returns, unconditionally," was per-path filtering at refresh time (rather than only at cache-read time, where it already happens in cache_decision/longest_prefix_match) intentional here?

Good point that I missed, now all results are cached (minus the malformed or expired ones) and then it's filtered

mbutrovich · 2026-08-05T20:01:15Z

+                parsed
+                    .storage_credentials
+                    .into_iter()
+                    .filter(|sc| configured.cloud.matches_location(&sc.prefix))
+                    .map(|sc| {
+                        (configured.cloud.parse_credential)(&sc.config, Some(sc.prefix)).map(
+                            |credential| {
+                                CachedEntry::new(credential, configured.cloud.jitter_prefetch)
+                            },
+                        )
+                    })
+                    .collect()


collect() into Result<Vec<CachedEntry>> short-circuits on the first parse_credential error. If a server vends N credentials for one cloud and one entry is missing a required field, all N become unusable (and, per finding 2, this also counts as a "failure" for backoff purposes) rather than just the one bad entry. Java's equivalent (VendedCredentialsProvider.refreshCredential) only ever expects a single S3-prefixed entry and asserts on it directly, so there's no directly analogous "partial batch" behavior to compare against — but given this PR's own design supports N entries per cloud, is one bad entry meant to invalidate all the others?

fetch() now returns a ParsedCredentials which splits between the successful creds and errors

mbutrovich · 2026-08-05T20:01:42Z

+            let enabled = props
+                .get(cloud.enabled_key)
+                .is_none_or(|value| value.parse().unwrap_or(false));


str::parse::<bool>() only accepts the exact strings "true"/"false". Java's PropertyUtil.propertyAsBoolean uses Boolean.parseBoolean, which is case-insensitive for "true" ("True", "TRUE" all parse as true). A config value of client.refresh-credentials-enabled: "True" would enable refresh in Java but silently disable it in Rust (parse error → unwrap_or(false)). Low severity (fails closed either way), but worth a case-insensitive comparison to match Java's actual accepted input space.

mbutrovich · 2026-08-05T20:03:02Z

+pub struct StorageCredential {
+    /// Storage-location prefix this credential is scoped to. `None` represents a
+    /// credential without a declared scope, sourced from flat storage properties.
+    pub prefix: Option<String>,
+    /// The backend-specific credential material.
+    pub kind: StorageCredentialKind,
+    /// When the credential expires, if known. `None` means non-expiring and
+    /// backends treat such a credential as always valid and never refresh it.
+    pub expires_at: Option<SystemTime>,
+}
+
+/// Backend-specific credential material.
+#[derive(Clone, Debug)]
+pub enum StorageCredentialKind {
+    /// Amazon S3 credentials.
+    S3(S3Credential),
+    /// Google Cloud Storage credentials.
+    Gcs(GcsCredential),
+}
+
+/// Temporary Amazon S3 credentials.
+#[derive(Clone)]
+pub struct S3Credential {
+    /// AWS access key ID.
+    pub access_key_id: String,
+    /// AWS secret access key.
+    pub secret_access_key: String,
+    /// AWS session token, set for temporary (STS/vended) credentials.
+    pub session_token: Option<String>,
+}
+
+impl Debug for S3Credential {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("S3Credential").finish_non_exhaustive()
+    }
+}
+
+/// Temporary Google Cloud Storage credentials (an OAuth2 access token).
+#[derive(Clone)]
+pub struct GcsCredential {
+    /// OAuth2 bearer token used to access GCS.
+    pub token: String,
+}


These are new public types (in public-api.txt) that third-party StorageCredentialProvider implementors must construct by hand. Every field is pub, with no constructor. That's inconsistent with StorageConfig in the same module (crates/iceberg/src/io/storage/config/mod.rs:55-58), which keeps props private and exposes with_prop/from_props instead. Was a constructor considered, or is direct struct-literal construction intentional here?

Sorry this wasn't intentional, they're no longer public

…ential-refresh # Conflicts: # crates/storage/opendal/src/gcs.rs

zakariya-s · 2026-08-07T09:43:49Z

Thanks @mbutrovich for the first review! I've made some changes to address your comments and pushed them to this PR for posterity. I will follow your suggestion and split this PR and use a tracker issue.

zakariya-s · 2026-08-07T10:35:10Z

First PR available here: #2976

feat(rest): support refreshing vended storage credentials

992bc10

zakariya-s mentioned this pull request Jul 30, 2026

feat(rest): support vended storage credentials with per-prefix FileIO #2651

Open

chore: fix typo

26d7fcc