[PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s opened a new pull request, #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932

   ## Which issue does this PR close?
   
   <!--
   We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
   -->
   
   - Closes #2931.
   
   ## What changes are included in this PR?
   
   <!--
   Provide a summary of the modifications in this PR. List the main changes such as new features, bug fixes, refactoring, or any other updates.
   -->
   
   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?
   
   <!--
   Specify what test covers (unit test, integration test, etc.).
   
   If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
   -->
   
   Yes


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#issuecomment-5194607023

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


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#issuecomment-5539307322

   apache/opendal#8030 has been included as part of opendal-core 0.59.0, so Azure support is now possible. Unfortunately, 0.59.0 doesn't compile due to a packaging issue so we will have to wait for a patch to upgrade the dependency.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r4185342996


##########
crates/iceberg/src/io/file_io.rs:
##########
@@ -123,14 +132,28 @@ impl FileIO {
     ///
     /// All storage configuration properties are included in the serialized representation. These
     /// properties may contain credentials or other sensitive values, so the returned bytes must be
-    /// protected in transit and at rest by the application embedding this crate.
+    /// protected in transit and at rest by the application embedding this crate. A serialized
+    /// credential provider may likewise carry catalog authentication and vended credentials.
     ///
     /// Storage factories are serialized through [`typetag`](https://docs.rs/typetag). Third-party
     /// factories must use `#[typetag::serde]` on their [`StorageFactory`] implementation.
+    ///
+    /// A credential provider is serialized as the
+    /// [`StorageCredentialProviderFactory`] returned by
+    /// [`StorageCredentialProvider::factory`], and rebuilt on deserialization. Serialization fails
+    /// when the provider cannot be rebuilt in another process; use
+    /// [`FileIO::without_credential_provider`] to serialize without it.
     pub fn serialize_all(&self) -> Result<Vec<u8>> {
+        let credential_provider = self
+            .credential_provider
+            .as_ref()
+            .map(|provider| provider.factory())
+            .transpose()?;

Review Comment:
   Sounds good, I'm happy with emitting a warning instead



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3689802681


##########
crates/storage/opendal/src/lib.rs:
##########
@@ -553,6 +611,12 @@ impl Storage for OpenDalStorage {
         let mut deleters: HashMap<String, opendal::Deleter> = HashMap::new();
 
         while let Some(path) = paths.next().await {
+            if self.uses_dynamic_credentials(&path) {
+                let (op, relative_path) = self.create_operator(&path)?;
+                op.delete(relative_path).await.map_err(from_opendal_error)?;
+                continue;
+            }

Review Comment:
   This will probably need to be looked at again later since we forgo batching for paths with a vended credential provider. We would probably have to batch by path-prefix to make this work later. Happy to do this later in this PR or in a new PR



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3689786418


##########
crates/storage/opendal/src/lib.rs:
##########
@@ -310,9 +332,18 @@ impl OpenDalStorage {
                 }
             }
             #[cfg(feature = "opendal-gcs")]
-            OpenDalStorage::Gcs { config } => {
-                let operator = gcs_config_build(config, path)?;
-                let prefix = format!("gs://{}/", operator.info().name());
+            OpenDalStorage::Gcs {
+                config,
+                credential_provider,
+            } => {
+                let operator = gcs_config_build(config, credential_provider, path)?;
+                let url = url::Url::parse(path).map_err(|e| {
+                    Error::new(
+                        ErrorKind::DataInvalid,
+                        format!("Invalid gcs url: {path}: {e}"),
+                    )
+                })?;

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



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3735919517


##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,1203 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Refresh of vended storage credentials against a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache, and
+//! the returned [`StorageCredential`] enum lets the storage adapter enforce the
+//! expected backend-specific type. This preserves Java's per-cloud refresh
+//! policy while supporting mixed-cloud tables through a resolving FileIO.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant. To
+//! add a backend, first add its credential type to Iceberg's storage API and
+//! teach the storage adapter to consume it. Then write its `parse_*` function,
+//! add a `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+    AWS_REFRESH_CREDENTIALS_ENABLED, AWS_REFRESH_CREDENTIALS_ENDPOINT,
+    GCS_REFRESH_CREDENTIALS_ENABLED, GCS_REFRESH_CREDENTIALS_ENDPOINT, GCS_TOKEN,
+    GCS_TOKEN_EXPIRES_AT, GcsCredential, S3_ACCESS_KEY_ID, S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN,
+    S3_SESSION_TOKEN_EXPIRES_AT_MS, S3Credential, StorageCredential, StorageCredentialKind,
+    StorageCredentialProvider,
+};
+use iceberg::{Error, ErrorKind, Result};
+use rand::Rng;
+use reqwest::{Method, StatusCode, Url};
+use tokio::sync::Mutex;
+
+use crate::REST_CATALOG_PROP_SCAN_PLAN_ID;
+use crate::client::{HttpClient, deserialize_unexpected_catalog_error};
+use crate::types::LoadCredentialsResponse;
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// 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 {
+    /// Location URL schemes this backend serves.
+    schemes: &'static [&'static str],
+    /// Table property naming the refresh endpoint (absolute or catalog-relative).
+    endpoint_key: &'static str,
+    /// Table property to opt out; refresh is enabled unless this is `"false"`.
+    enabled_key: &'static str,
+    /// Whether to jitter successful prefetch times like AWS `CachedSupplier`.
+    jitter_prefetch: bool,
+    /// Parse a complete credential from catalog-supplied properties.
+    parse_credential:
+        fn(config: &HashMap<String, String>, prefix: Option<String>) -> Result<StorageCredential>,
+}
+
+impl CloudRefresh {
+    /// S3 / AWS
+    const AWS: Self = Self {
+        schemes: &["s3", "s3a", "s3n"],
+        endpoint_key: AWS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: AWS_REFRESH_CREDENTIALS_ENABLED,
+        jitter_prefetch: true,
+        parse_credential: parse_s3_credential,
+    };
+    /// Google Cloud Storage
+    const GCP: Self = Self {
+        schemes: &["gs", "gcs"],
+        endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED,
+        jitter_prefetch: false,
+        parse_credential: parse_gcs_credential,
+    };
+    // TODO: Azure (ADLS) is not yet supported: opendal 0.57's Azdls builder exposes no
+    // custom credential-provider hook, and reqsign's SAS-token credential has no
+    // expiry, so reqsign-based refresh isn't possible.
+
+    /// Backends with refresh support
+    const SUPPORTED: &[Self] = &[Self::AWS, Self::GCP];
+
+    /// The backend that serves `location`, by its URL scheme, or `None` if no
+    /// supported backend matches (in which case static credentials are used
+    /// as-is, as before).
+    fn for_location(location: &str) -> Option<&'static Self> {
+        Self::SUPPORTED
+            .iter()
+            .find(|cloud| cloud.matches_location(location))
+    }
+
+    fn matches_location(&self, location: &str) -> bool {
+        scheme_of(location).is_some_and(|scheme| self.schemes.contains(&scheme.as_str()))
+    }
+}
+
+/// Re-fetch a credential once it is within this window of expiry, so a fresh
+/// token is in hand before the object store would reject the old one.
+const REFRESH_BUFFER: Duration = Duration::from_mins(5);
+
+/// AWS keeps at least one minute between its jittered prefetch time and expiry.
+const MIN_REFRESH_BUFFER: Duration = Duration::from_mins(1);
+
+/// Initial ceiling for failure backoff. Equal jitter chooses from half this
+/// value through the full value.
+const INITIAL_FAILURE_BACKOFF: Duration = Duration::from_secs(1);
+
+/// Maximum failure backoff while a cached credential remains usable.
+const MAX_FAILURE_BACKOFF: Duration = Duration::from_secs(30);
+
+/// A cached vended credential and its refresh schedule.
+#[derive(Clone)]
+struct CachedEntry {
+    credential: StorageCredential,
+    /// When this entry becomes eligible for prefetch. `None` means it does not
+    /// expire and therefore never needs proactive refresh.
+    refresh_at: Option<SystemTime>,
+}
+
+impl CachedEntry {
+    fn new(credential: StorageCredential, jitter_prefetch: bool) -> Self {
+        let refresh_at = credential
+            .expires_at
+            .map(|expires_at| prefetch_time(expires_at, jitter_prefetch));
+        Self {
+            credential,
+            refresh_at,
+        }
+    }
+
+    /// Seed entries that are already inside the nominal five-minute window are
+    /// immediately due. Otherwise AWS applies the same jitter as it does to a
+    /// freshly fetched value.
+    fn seed(credential: StorageCredential, jitter_prefetch: bool) -> Self {
+        let due = credential.expires_at.is_some_and(|expires_at| {
+            SystemTime::now()
+                .checked_add(REFRESH_BUFFER)
+                .is_none_or(|refresh_boundary| refresh_boundary >= expires_at)
+        });
+        let mut entry = Self::new(credential, jitter_prefetch);
+        if due {
+            entry.refresh_at = Some(UNIX_EPOCH);
+        }
+        entry
+    }
+
+    fn is_fresh(&self, now: SystemTime) -> bool {
+        self.refresh_at.is_none_or(|refresh_at| now < refresh_at)
+    }
+
+    fn is_unexpired(&self, now: SystemTime) -> bool {
+        self.credential
+            .expires_at
+            .is_none_or(|expires_at| now < expires_at)
+    }
+}
+
+/// Cached credentials plus failure-backoff state.
+struct CacheState {
+    entries: Vec<CachedEntry>,
+    consecutive_failures: u32,
+    retry_not_before: Option<Instant>,
+}
+
+struct ConfiguredCloud {
+    cloud: &'static CloudRefresh,
+    endpoint: String,
+    cache: Mutex<CacheState>,
+    /// Only one caller fetches at a time. The cache lock is deliberately
+    /// separate so other callers can keep using an unexpired credential while
+    /// the refresh is in flight.
+    refresh: Mutex<()>,
+}
+
+/// Fetches and refreshes vended credentials from a REST catalog's table
+/// credentials endpoint.
+///
+/// Each cloud cache is seeded with the credential from the initial table
+/// properties (when complete) and re-fetched from its endpoint as it
+/// nears expiry.
+pub(crate) struct RestVendedCredentialProvider {
+    client: Arc<HttpClient>,
+    /// Optional scan-plan identifier.
+    plan_id: Option<String>,
+    /// Independently configured endpoint and cache for each backing cloud.
+    clouds: Vec<ConfiguredCloud>,
+}
+
+impl RestVendedCredentialProvider {
+    fn new(client: Arc<HttpClient>, plan_id: Option<String>, clouds: Vec<ConfiguredCloud>) -> Self {
+        Self {
+            client,
+            plan_id,
+            clouds,
+        }
+    }
+
+    fn configured_cloud_for_location(&self, location: &str) -> Option<&ConfiguredCloud> {
+        self.clouds
+            .iter()
+            .find(|configured| configured.cloud.matches_location(location))
+    }
+
+    /// Fetch fresh credentials from the catalog's credentials endpoint.
+    async fn fetch(&self, configured: &ConfiguredCloud) -> Result<Vec<CachedEntry>> {
+        let mut request = self.client.request(Method::GET, &configured.endpoint);
+        if let Some(plan_id) = &self.plan_id {
+            request = request.query(&[("planId", plan_id)]);
+        }
+        let request = request.build()?;
+        let response = self.client.query_catalog(request).await?;
+
+        match response.status() {
+            StatusCode::OK => {
+                let parsed: LoadCredentialsResponse = response.json().await?;
+                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()

Review Comment:
   `fetch()` now returns a `ParsedCredentials` which splits between the successful creds and errors



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3689705006


##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -1048,7 +1074,14 @@ impl Catalog for RestCatalog {
             "Metadata location missing in `register_table` response!",
         ))?;
 
-        let file_io = self.load_file_io(Some(metadata_location), None).await?;
+        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?;

Review Comment:
   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



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3689769620


##########
crates/storage/opendal/src/gcs.rs:
##########
@@ -45,7 +48,9 @@ pub(crate) fn gcs_config_parse(mut m: HashMap<String, String>) -> Result<GcsConf
         cfg.endpoint = Some(endpoint);
     }
 
-    if m.remove(GCS_NO_AUTH).is_some() {
+    if let Some(no_auth) = m.remove(GCS_NO_AUTH)
+        && is_truthy(no_auth.to_lowercase().as_str())
+    {

Review Comment:
   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`



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#issuecomment-5599752751

   0.59.1 was released, so Azure support should now be complete. This PR should be fully ready for review


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3735905025


##########
crates/storage/opendal/src/lib.rs:
##########
@@ -384,6 +415,27 @@ impl OpenDalStorage {
         }
     }
 
+    /// Returns whether `path` uses a path-scoped dynamic credential provider.
+    ///
+    /// Such paths cannot share an OpenDAL deleter unless the provider can expose
+    /// the credential scope that applies to each path. Process them independently
+    /// so bulk deletion remains bounded without crossing credential boundaries.
+    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 {

Review Comment:
   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



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3735944694


##########
crates/catalog/rest/src/client.rs:
##########
@@ -71,6 +75,24 @@ impl HttpClient {
         })
     }
 
+    /// Create a client for table-scoped resources while reusing this client's
+    /// underlying connection pool.
+    ///
+    /// A load-table response may supply a table token or `header.*` values that
+    /// must be used for subsequent table requests such as credential refresh.
+    pub(crate) fn for_table(
+        &self,
+        catalog_uri: &str,
+        props: HashMap<String, String>,
+    ) -> Result<Self> {
+        let cfg = RestCatalogConfig::builder()
+            .uri(catalog_uri.to_string())
+            .props(props)
+            .client(Some(self.client.clone()))
+            .build();
+        Self::new(&cfg)
+    }

Review Comment:
   Good idea, I missed this. The cached token was made shareable and `for_table()` constructs a table client manually instead of through `HttpClient::new()`



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["laskoviymishka (via GitHub)" <gi...@apache.org>]

laskoviymishka commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r4326462087


##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,3675 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Vended storage credentials from a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache. Like
+//! Java's `StorageCredential`, the returned [`StorageCredential`] holds the
+//! backend's storage properties, which the storage adapter turns into its own
+//! credential. This preserves Java's per-cloud credential selection and
+//! prefetch policies while supporting mixed-cloud tables through a resolving
+//! FileIO. Unlike Java's scheduled refresh, which permanently stops
+//! after a failed fetch, transient failures are retried here with jittered
+//! exponential backoff while an unexpired credential remains available.
+//!
+//! Like Java's `VendedCredentialsProvider`, a provider is rebuilt from the
+//! FileIO properties after [`FileIO`](iceberg::io::FileIO) serialization and
+//! connects to the catalog lazily in the receiving process. This requires
+//! catalog authentication that can be rebuilt from `rest.auth.type`.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant. To
+//! add a backend, teach its storage adapter to read its credential properties
+//! from a [`StorageCredential`]. Then write its `parse_*` function, add a
+//! `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+    ADLS_REFRESH_CREDENTIALS_ENABLED, ADLS_REFRESH_CREDENTIALS_ENDPOINT,
+    ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX, ADLS_SAS_TOKEN_PREFIX, AWS_REFRESH_CREDENTIALS_ENABLED,
+    AWS_REFRESH_CREDENTIALS_ENDPOINT, GCS_REFRESH_CREDENTIALS_ENABLED,
+    GCS_REFRESH_CREDENTIALS_ENDPOINT, GCS_TOKEN, GCS_TOKEN_EXPIRES_AT, S3_ACCESS_KEY_ID,
+    S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN, S3_SESSION_TOKEN_EXPIRES_AT_MS, StorageConfig,
+    StorageCredential, StorageCredentialProvider, StorageCredentialProviderFactory,
+};
+use iceberg::{Error, ErrorKind, Result, TableIdent};
+use rand::Rng;
+use reqwest::{Method, Url};
+use serde::{Deserialize, Serialize};
+use tokio::sync::{Mutex, OnceCell};
+
+use crate::auth::{AUTH_TYPE_NONE, AuthManager, load_auth_manager, static_token_session};
+use crate::catalog::{
+    REST_CATALOG_PROP_AUTH_TYPE, REST_CATALOG_PROP_REFERENCED_BY, REST_CATALOG_PROP_SCAN_PLAN_ID,
+    RestCatalogConfig,
+};
+use crate::client::{HttpClient, unexpected_catalog_error_without_body};
+use crate::request::HttpRequest;
+use crate::types::LoadCredentialsResponse;
+
+type CredentialParser =
+    fn(config: &HashMap<String, String>, prefix: String) -> Result<VendedCredential>;
+type KeyedSeedCredentialParser =
+    fn(config: &HashMap<String, String>) -> HashMap<String, VendedCredential>;
+type KeyedSeedPathResolver = fn(path: &str) -> Result<KeyedSeedPath>;
+
+/// A credential vended by the catalog, with the expiry its config declares.
+#[derive(Clone)]
+struct VendedCredential {
+    credential: StorageCredential,
+    expires_at: SystemTime,
+}
+
+enum SeedStrategy {
+    /// One credential stored in the backend's flat properties, for every
+    /// location with one of the scheme prefixes, like Java's root clients.
+    Flat { prefixes: &'static [&'static str] },
+    /// Credentials selected by a backend-specific key derived from each path.
+    Keyed(KeyedSeedStrategy),
+}
+
+struct KeyedSeedStrategy {
+    parse_credentials: KeyedSeedCredentialParser,
+    resolve_path: KeyedSeedPathResolver,
+}
+
+struct KeyedSeedPath {
+    /// Seed keys for the path, most specific first.
+    keys: Vec<String>,
+    scope: String,
+}
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// 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 {
+    /// Location URL schemes this backend serves.
+    schemes: &'static [&'static str],
+    /// Table property naming the refresh endpoint (absolute or catalog-relative).
+    endpoint_key: &'static str,
+    /// Table property controlling refresh; only missing or case-insensitive `"true"` enables it.
+    enabled_key: &'static str,
+    /// Table property that must be present for refresh to apply, as in Java,
+    /// where it decides whether the backend uses vended credentials at all.
+    required_key: Option<&'static str>,
+    /// Whether to jitter successful prefetch times like AWS `CachedSupplier`.
+    jitter_prefetch: bool,
+    /// Parse a complete credential from catalog-supplied properties.
+    parse_credential: CredentialParser,
+    /// How initial credentials in the table properties are selected.
+    seed_strategy: SeedStrategy,
+}
+
+impl CloudRefresh {
+    /// S3 / AWS
+    const AWS: Self = Self {
+        schemes: &["s3", "s3a", "s3n"],
+        endpoint_key: AWS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: AWS_REFRESH_CREDENTIALS_ENABLED,
+        required_key: None,
+        jitter_prefetch: true,
+        parse_credential: parse_s3_credential,
+        // `s3` also covers `s3a` and `s3n` locations.
+        seed_strategy: SeedStrategy::Flat { prefixes: &["s3"] },
+    };
+    /// Google Cloud Storage
+    const GCP: Self = Self {
+        schemes: &["gs", "gcs"],
+        endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED,
+        // Java's GCS client only refreshes a vended OAuth2 token; without
+        // one it uses Google's default credentials.
+        required_key: Some(GCS_TOKEN),
+        jitter_prefetch: false,
+        parse_credential: parse_gcs_credential,
+        seed_strategy: SeedStrategy::Flat {
+            prefixes: &["gs", "gcs"],
+        },
+    };
+    /// Azure Data Lake Storage
+    const AZURE: Self = Self {
+        schemes: &["abfs", "abfss", "wasb", "wasbs"],
+        endpoint_key: ADLS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: ADLS_REFRESH_CREDENTIALS_ENABLED,
+        required_key: None,
+        jitter_prefetch: false,
+        parse_credential: parse_azdls_credential,
+        seed_strategy: SeedStrategy::Keyed(KeyedSeedStrategy {
+            parse_credentials: parse_azdls_account_seeds,
+            resolve_path: resolve_azdls_seed_path,
+        }),
+    };
+
+    /// Backends with refresh support
+    const SUPPORTED: &[Self] = &[Self::AWS, Self::GCP, Self::AZURE];
+
+    fn matches_location(&self, location: &str) -> bool {
+        self.schemes
+            .iter()
+            .any(|scheme| location.eq_ignore_ascii_case(scheme))
+            || scheme_of(location).is_some_and(|scheme| self.schemes.contains(&scheme.as_str()))
+    }
+}
+
+/// Re-fetch a credential once it is within this window of expiry, so a fresh
+/// token is in hand before the object store would reject the old one.
+const REFRESH_BUFFER: Duration = Duration::from_mins(5);
+
+/// AWS keeps at least one minute between its jittered prefetch time and expiry.
+const MIN_REFRESH_BUFFER: Duration = Duration::from_mins(1);
+
+/// Initial ceiling for failure backoff. Equal jitter chooses from half this
+/// value through the full value.
+const INITIAL_FAILURE_BACKOFF: Duration = Duration::from_secs(1);
+
+/// Maximum delay between failed refresh attempts.
+const MAX_FAILURE_BACKOFF: Duration = Duration::from_secs(30);
+
+/// A cached vended credential and its refresh schedule.
+#[derive(Clone)]
+struct CachedEntry {
+    credential: StorageCredential,
+    expires_at: SystemTime,
+    /// When this entry becomes eligible for prefetch.
+    refresh_at: SystemTime,
+}
+
+impl CachedEntry {
+    fn new(vended: VendedCredential, jitter_prefetch: bool) -> Self {
+        let VendedCredential {
+            credential,
+            expires_at,
+        } = vended;
+        Self {
+            credential,
+            expires_at,
+            refresh_at: prefetch_time(SystemTime::now(), expires_at, jitter_prefetch),
+        }
+    }
+
+    /// Seed entries that are already inside the nominal five-minute window are
+    /// immediately due. Otherwise AWS applies the same jitter as it does to a
+    /// freshly fetched value.
+    fn seed(vended: VendedCredential, jitter_prefetch: bool) -> Self {
+        let due = SystemTime::now()
+            .checked_add(REFRESH_BUFFER)
+            .is_none_or(|refresh_boundary| refresh_boundary >= vended.expires_at);
+        let mut entry = Self::new(vended, jitter_prefetch);
+        if due {
+            entry.refresh_at = UNIX_EPOCH;
+        }
+        entry
+    }
+
+    fn is_fresh(&self, now: SystemTime) -> bool {
+        now < self.refresh_at
+    }
+
+    fn is_unexpired(&self, now: SystemTime) -> bool {
+        now < self.expires_at
+    }
+}
+
+/// Parsed entries from one successful credentials response.
+struct ParsedCredentials {
+    entries: Vec<CachedEntry>,
+    errors: Vec<CredentialError>,
+}
+
+struct CredentialError {
+    prefix: String,
+    error: Error,
+}
+
+/// Cached credentials plus failure-backoff state.
+///
+/// A fetch returns the credentials for every prefix of the table, so a path
+/// the last response did not cover would not be covered by an immediate
+/// re-fetch either. Backoff is therefore shared by the whole cloud.
+struct CacheState {
+    entries: Vec<CachedEntry>,
+    /// Prefixes whose credential in the last successful response was invalid.
+    /// Paths under them must not use a broader credential: the catalog scoped
+    /// them more tightly.
+    failed_prefixes: HashSet<String>,
+    consecutive_failures: u32,
+    retry_not_before: Option<Instant>,
+    /// Why the last refresh failed, reported while refresh is backed off.
+    last_failure: Option<String>,
+}
+
+impl CacheState {
+    fn new(entries: Vec<CachedEntry>) -> Self {
+        Self {
+            entries,
+            failed_prefixes: HashSet::new(),
+            consecutive_failures: 0,
+            retry_not_before: None,
+            last_failure: None,
+        }
+    }
+
+    /// The index of the unexpired credential with the longest prefix covering
+    /// `path`.
+    fn unexpired_match(&self, path: &str, now: SystemTime) -> Option<usize> {
+        self.entries
+            .iter()
+            .enumerate()
+            .filter(|(_, entry)| entry.is_unexpired(now) && entry.credential.covers(path))

Review Comment:
   This drops expired entries before it picks the longest prefix. So once a narrow `s3://bucket/table` credential expires while a broader entry is still valid, `cache_decision` serves the broader one and never refreshes the narrow prefix. The catalog scoped that prefix separately for a reason. After an idle gap longer than the narrow credential's lifetime we'd sign with a credential that may carry different permissions, and the caller gets a 403 with no refresh attempted.
   
   I'd pick the longest covering prefix whether or not it has expired, and return `Refresh` if it has. Java's `clientForStoragePath` uses the same longest-prefix rule without the fallback. `unexpired_match_selects_longest_unexpired_prefix` currently pins the fallback as intended, so that test flips.



##########
crates/catalog/rest/src/types.rs:
##########
@@ -244,6 +255,27 @@ pub struct StorageCredential {
     pub config: HashMap<String, String>,
 }
 
+impl std::fmt::Debug for StorageCredential {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("StorageCredential")
+            .field("prefix", &self.prefix)
+            .field("config_keys", &self.config.keys().collect::<Vec<_>>())
+            .finish_non_exhaustive()
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "kebab-case")]
+/// Response from the table credentials endpoint
+/// (`GET /v1/{prefix}/namespaces/{namespace}/tables/{table}/credentials`).
+///
+/// Returns freshly vended storage credentials so clients can refresh temporary
+/// credentials before they expire.
+pub struct LoadCredentialsResponse {

Review Comment:
   This becomes part of the crate's public API, with a `pub` field and five derives. Its only consumer is the private `credential` module, which only needs `Deserialize`. I'd make it `pub(crate)` and drop the extra derives. Once it ships, we're committed to keeping its shape stable.



##########
crates/iceberg/src/io/file_io.rs:
##########
@@ -123,14 +146,38 @@ impl FileIO {
     ///
     /// All storage configuration properties are included in the serialized representation. These
     /// properties may contain credentials or other sensitive values, so the returned bytes must be
-    /// protected in transit and at rest by the application embedding this crate.
+    /// protected in transit and at rest by the application embedding this crate. A serialized
+    /// credential provider may likewise carry catalog authentication and vended credentials.
     ///
     /// Storage factories are serialized through [`typetag`](https://docs.rs/typetag). Third-party
     /// factories must use `#[typetag::serde]` on their [`StorageFactory`] implementation.
+    ///
+    /// A credential provider is serialized as the
+    /// [`StorageCredentialProviderFactory`] returned by
+    /// [`StorageCredentialProvider::factory`], and rebuilt on deserialization. A provider that
+    /// cannot be rebuilt in another process is left out with a warning, as with
+    /// [`FileIO::without_credential_provider`]: the deserialized `FileIO` uses the credentials in
+    /// its configuration, which are not refreshed. Storage factories that hold their own
+    /// credential sources, such as a custom AWS credential loader, still fail to serialize, as
+    /// leaving them out would leave no credentials at all.
     pub fn serialize_all(&self) -> Result<Vec<u8>> {
+        let credential_provider = self.credential_provider.as_ref().and_then(|provider| {
+            provider
+                .factory()
+                .and_then(|factory| Ok(serde_json::to_value(factory)?))
+                .inspect_err(|error| {
+                    static WARNED: std::sync::Once = std::sync::Once::new();

Review Comment:
   Following up on @mbutrovich's thread about warning instead of erroring: a process-wide `Once` means that only the first `FileIO` to lose its provider gets a warning. On a long-lived worker, every later one downgrades silently, and the first sign of trouble is a 403 hours later. I'd dedupe per factory type or per error instead of with one static. Better still, add a way for callers to check that the provider survived, such as `FileIO::has_credential_provider()`.



##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,3675 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Vended storage credentials from a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache. Like
+//! Java's `StorageCredential`, the returned [`StorageCredential`] holds the
+//! backend's storage properties, which the storage adapter turns into its own
+//! credential. This preserves Java's per-cloud credential selection and
+//! prefetch policies while supporting mixed-cloud tables through a resolving
+//! FileIO. Unlike Java's scheduled refresh, which permanently stops
+//! after a failed fetch, transient failures are retried here with jittered
+//! exponential backoff while an unexpired credential remains available.
+//!
+//! Like Java's `VendedCredentialsProvider`, a provider is rebuilt from the
+//! FileIO properties after [`FileIO`](iceberg::io::FileIO) serialization and
+//! connects to the catalog lazily in the receiving process. This requires
+//! catalog authentication that can be rebuilt from `rest.auth.type`.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant. To
+//! add a backend, teach its storage adapter to read its credential properties
+//! from a [`StorageCredential`]. Then write its `parse_*` function, add a
+//! `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+    ADLS_REFRESH_CREDENTIALS_ENABLED, ADLS_REFRESH_CREDENTIALS_ENDPOINT,
+    ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX, ADLS_SAS_TOKEN_PREFIX, AWS_REFRESH_CREDENTIALS_ENABLED,
+    AWS_REFRESH_CREDENTIALS_ENDPOINT, GCS_REFRESH_CREDENTIALS_ENABLED,
+    GCS_REFRESH_CREDENTIALS_ENDPOINT, GCS_TOKEN, GCS_TOKEN_EXPIRES_AT, S3_ACCESS_KEY_ID,
+    S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN, S3_SESSION_TOKEN_EXPIRES_AT_MS, StorageConfig,
+    StorageCredential, StorageCredentialProvider, StorageCredentialProviderFactory,
+};
+use iceberg::{Error, ErrorKind, Result, TableIdent};
+use rand::Rng;
+use reqwest::{Method, Url};
+use serde::{Deserialize, Serialize};
+use tokio::sync::{Mutex, OnceCell};
+
+use crate::auth::{AUTH_TYPE_NONE, AuthManager, load_auth_manager, static_token_session};
+use crate::catalog::{
+    REST_CATALOG_PROP_AUTH_TYPE, REST_CATALOG_PROP_REFERENCED_BY, REST_CATALOG_PROP_SCAN_PLAN_ID,
+    RestCatalogConfig,
+};
+use crate::client::{HttpClient, unexpected_catalog_error_without_body};
+use crate::request::HttpRequest;
+use crate::types::LoadCredentialsResponse;
+
+type CredentialParser =
+    fn(config: &HashMap<String, String>, prefix: String) -> Result<VendedCredential>;
+type KeyedSeedCredentialParser =
+    fn(config: &HashMap<String, String>) -> HashMap<String, VendedCredential>;
+type KeyedSeedPathResolver = fn(path: &str) -> Result<KeyedSeedPath>;
+
+/// A credential vended by the catalog, with the expiry its config declares.
+#[derive(Clone)]
+struct VendedCredential {
+    credential: StorageCredential,
+    expires_at: SystemTime,
+}
+
+enum SeedStrategy {
+    /// One credential stored in the backend's flat properties, for every
+    /// location with one of the scheme prefixes, like Java's root clients.
+    Flat { prefixes: &'static [&'static str] },
+    /// Credentials selected by a backend-specific key derived from each path.
+    Keyed(KeyedSeedStrategy),
+}
+
+struct KeyedSeedStrategy {
+    parse_credentials: KeyedSeedCredentialParser,
+    resolve_path: KeyedSeedPathResolver,
+}
+
+struct KeyedSeedPath {
+    /// Seed keys for the path, most specific first.
+    keys: Vec<String>,
+    scope: String,
+}
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// 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 {
+    /// Location URL schemes this backend serves.
+    schemes: &'static [&'static str],
+    /// Table property naming the refresh endpoint (absolute or catalog-relative).
+    endpoint_key: &'static str,
+    /// Table property controlling refresh; only missing or case-insensitive `"true"` enables it.
+    enabled_key: &'static str,
+    /// Table property that must be present for refresh to apply, as in Java,
+    /// where it decides whether the backend uses vended credentials at all.
+    required_key: Option<&'static str>,
+    /// Whether to jitter successful prefetch times like AWS `CachedSupplier`.
+    jitter_prefetch: bool,
+    /// Parse a complete credential from catalog-supplied properties.
+    parse_credential: CredentialParser,
+    /// How initial credentials in the table properties are selected.
+    seed_strategy: SeedStrategy,
+}
+
+impl CloudRefresh {
+    /// S3 / AWS
+    const AWS: Self = Self {
+        schemes: &["s3", "s3a", "s3n"],
+        endpoint_key: AWS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: AWS_REFRESH_CREDENTIALS_ENABLED,
+        required_key: None,
+        jitter_prefetch: true,
+        parse_credential: parse_s3_credential,
+        // `s3` also covers `s3a` and `s3n` locations.
+        seed_strategy: SeedStrategy::Flat { prefixes: &["s3"] },
+    };
+    /// Google Cloud Storage
+    const GCP: Self = Self {
+        schemes: &["gs", "gcs"],
+        endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED,
+        // Java's GCS client only refreshes a vended OAuth2 token; without
+        // one it uses Google's default credentials.
+        required_key: Some(GCS_TOKEN),
+        jitter_prefetch: false,
+        parse_credential: parse_gcs_credential,
+        seed_strategy: SeedStrategy::Flat {
+            prefixes: &["gs", "gcs"],
+        },
+    };
+    /// Azure Data Lake Storage
+    const AZURE: Self = Self {
+        schemes: &["abfs", "abfss", "wasb", "wasbs"],
+        endpoint_key: ADLS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: ADLS_REFRESH_CREDENTIALS_ENABLED,
+        required_key: None,
+        jitter_prefetch: false,
+        parse_credential: parse_azdls_credential,
+        seed_strategy: SeedStrategy::Keyed(KeyedSeedStrategy {
+            parse_credentials: parse_azdls_account_seeds,
+            resolve_path: resolve_azdls_seed_path,
+        }),
+    };
+
+    /// Backends with refresh support
+    const SUPPORTED: &[Self] = &[Self::AWS, Self::GCP, Self::AZURE];
+
+    fn matches_location(&self, location: &str) -> bool {
+        self.schemes
+            .iter()
+            .any(|scheme| location.eq_ignore_ascii_case(scheme))
+            || scheme_of(location).is_some_and(|scheme| self.schemes.contains(&scheme.as_str()))
+    }
+}
+
+/// Re-fetch a credential once it is within this window of expiry, so a fresh
+/// token is in hand before the object store would reject the old one.
+const REFRESH_BUFFER: Duration = Duration::from_mins(5);
+
+/// AWS keeps at least one minute between its jittered prefetch time and expiry.
+const MIN_REFRESH_BUFFER: Duration = Duration::from_mins(1);
+
+/// Initial ceiling for failure backoff. Equal jitter chooses from half this
+/// value through the full value.
+const INITIAL_FAILURE_BACKOFF: Duration = Duration::from_secs(1);
+
+/// Maximum delay between failed refresh attempts.
+const MAX_FAILURE_BACKOFF: Duration = Duration::from_secs(30);
+
+/// A cached vended credential and its refresh schedule.
+#[derive(Clone)]
+struct CachedEntry {
+    credential: StorageCredential,
+    expires_at: SystemTime,
+    /// When this entry becomes eligible for prefetch.
+    refresh_at: SystemTime,
+}
+
+impl CachedEntry {
+    fn new(vended: VendedCredential, jitter_prefetch: bool) -> Self {
+        let VendedCredential {
+            credential,
+            expires_at,
+        } = vended;
+        Self {
+            credential,
+            expires_at,
+            refresh_at: prefetch_time(SystemTime::now(), expires_at, jitter_prefetch),
+        }
+    }
+
+    /// Seed entries that are already inside the nominal five-minute window are
+    /// immediately due. Otherwise AWS applies the same jitter as it does to a
+    /// freshly fetched value.
+    fn seed(vended: VendedCredential, jitter_prefetch: bool) -> Self {
+        let due = SystemTime::now()
+            .checked_add(REFRESH_BUFFER)
+            .is_none_or(|refresh_boundary| refresh_boundary >= vended.expires_at);
+        let mut entry = Self::new(vended, jitter_prefetch);
+        if due {
+            entry.refresh_at = UNIX_EPOCH;
+        }
+        entry
+    }
+
+    fn is_fresh(&self, now: SystemTime) -> bool {
+        now < self.refresh_at
+    }
+
+    fn is_unexpired(&self, now: SystemTime) -> bool {
+        now < self.expires_at
+    }
+}
+
+/// Parsed entries from one successful credentials response.
+struct ParsedCredentials {
+    entries: Vec<CachedEntry>,
+    errors: Vec<CredentialError>,
+}
+
+struct CredentialError {
+    prefix: String,
+    error: Error,
+}
+
+/// Cached credentials plus failure-backoff state.
+///
+/// A fetch returns the credentials for every prefix of the table, so a path
+/// the last response did not cover would not be covered by an immediate
+/// re-fetch either. Backoff is therefore shared by the whole cloud.
+struct CacheState {
+    entries: Vec<CachedEntry>,
+    /// Prefixes whose credential in the last successful response was invalid.
+    /// Paths under them must not use a broader credential: the catalog scoped
+    /// them more tightly.
+    failed_prefixes: HashSet<String>,
+    consecutive_failures: u32,
+    retry_not_before: Option<Instant>,
+    /// Why the last refresh failed, reported while refresh is backed off.
+    last_failure: Option<String>,
+}
+
+impl CacheState {
+    fn new(entries: Vec<CachedEntry>) -> Self {
+        Self {
+            entries,
+            failed_prefixes: HashSet::new(),
+            consecutive_failures: 0,
+            retry_not_before: None,
+            last_failure: None,
+        }
+    }
+
+    /// The index of the unexpired credential with the longest prefix covering
+    /// `path`.
+    fn unexpired_match(&self, path: &str, now: SystemTime) -> Option<usize> {
+        self.entries
+            .iter()
+            .enumerate()
+            .filter(|(_, entry)| entry.is_unexpired(now) && entry.credential.covers(path))
+            .max_by_key(|(_, entry)| entry.credential.prefix().len())
+            .map(|(index, _)| index)
+    }
+
+    /// Whether a failed prefix covering `path` is narrower than `credential`'s.
+    fn masks(&self, path: &str, credential: Option<&StorageCredential>) -> bool {
+        let prefix_len = credential.map(|credential| credential.prefix().len());
+        self.failed_prefixes.iter().any(|failed| {
+            prefix_covers(failed, path) && prefix_len.is_none_or(|len| failed.len() > len)
+        })
+    }
+
+    /// Cache `entry`, replacing any entry with the same prefix, and return its
+    /// index.
+    fn insert(&mut self, entry: CachedEntry) -> usize {
+        let prefix = entry.credential.prefix().to_owned();
+        self.entries
+            .retain(|cached| cached.credential.prefix() != prefix);
+        self.entries.push(entry);
+        self.entries.len() - 1
+    }
+
+    fn record_success(&mut self) {
+        self.consecutive_failures = 0;
+        self.retry_not_before = None;
+        self.last_failure = None;
+    }
+
+    fn record_failure(&mut self, error: &Error) {
+        self.consecutive_failures = self.consecutive_failures.saturating_add(1);
+        self.retry_not_before =
+            Instant::now().checked_add(failure_backoff(self.consecutive_failures));
+        self.last_failure = Some(error.to_string());
+    }
+
+    /// Replace cached credentials with the fetched ones, prefix by prefix,
+    /// and record the prefixes whose credential was invalid in this response.
+    ///
+    /// An unexpired cached credential survives when the response carries no
+    /// valid replacement for its prefix, so an absent or malformed entry never
+    /// evicts a usable credential. Returns the prefixes the response replaced.
+    fn merge(&mut self, fetched: Vec<CachedEntry>, errors: &[CredentialError]) -> HashSet<String> {
+        let fetched_prefixes = fetched
+            .iter()
+            .map(|entry| entry.credential.prefix().to_owned())
+            .collect::<HashSet<_>>();
+        // A response names every prefix of the table, so a prefix it omits is
+        // no longer scoped separately: a fetched credential that covers it
+        // replaces its cached one. A prefix whose credential is invalid keeps
+        // its cached one.
+        let failed_prefixes = errors
+            .iter()
+            .map(|error| error.prefix.clone())
+            .collect::<HashSet<_>>();
+        self.entries.retain(|entry| {
+            let prefix = entry.credential.prefix();
+            failed_prefixes.contains(prefix)
+                || !fetched
+                    .iter()
+                    .any(|fetched| fetched.credential.covers(prefix))
+        });
+        self.entries.extend(fetched);
+        self.failed_prefixes = failed_prefixes;
+        fetched_prefixes
+    }
+}
+
+struct ConfiguredCloud {
+    cloud: &'static CloudRefresh,
+    endpoint: String,
+    cache: Mutex<CacheState>,
+    /// Initial property credentials whose scope cannot be represented by a
+    /// single leading URI prefix, keyed according to the cloud strategy.
+    keyed_seeds: HashMap<String, CachedEntry>,
+    /// Only one caller fetches at a time. The cache lock is deliberately
+    /// separate so other callers can keep using an unexpired credential while
+    /// the refresh is in flight.
+    refresh: Mutex<()>,
+}
+
+impl ConfiguredCloud {
+    fn new(
+        cloud: &'static CloudRefresh,
+        endpoint: String,
+        entries: Vec<CachedEntry>,
+        keyed_seeds: HashMap<String, CachedEntry>,
+    ) -> Self {
+        Self {
+            cloud,
+            endpoint,
+            cache: Mutex::new(CacheState::new(entries)),
+            keyed_seeds,
+            refresh: Mutex::new(()),
+        }
+    }
+
+    /// Configure `cloud` from the FileIO properties, or `None` when they
+    /// advertise no enabled refresh endpoint for it.
+    fn configure(
+        cloud: &'static CloudRefresh,
+        base_uri: &str,
+        props: &HashMap<String, String>,
+    ) -> Option<Self> {
+        let enabled = props
+            .get(cloud.enabled_key)
+            .is_none_or(|value| value.eq_ignore_ascii_case("true"))
+            && cloud
+                .required_key
+                .is_none_or(|key| props.get(key).is_some_and(|value| !value.is_empty()));
+        let endpoint = props
+            .get(cloud.endpoint_key)
+            .filter(|endpoint| enabled && !endpoint.is_empty())
+            .map(|endpoint| resolve_endpoint(base_uri, endpoint))?;
+
+        let mut entries = Vec::new();
+        let mut keyed_seeds = HashMap::new();
+        match &cloud.seed_strategy {
+            SeedStrategy::Flat { prefixes } => {
+                entries.extend(prefixes.iter().filter_map(|prefix| {
+                    (cloud.parse_credential)(props, prefix.to_string())
+                        .ok()
+                        .map(|vended| CachedEntry::seed(vended, cloud.jitter_prefetch))
+                }));
+            }
+            SeedStrategy::Keyed(strategy) => {
+                keyed_seeds = (strategy.parse_credentials)(props)
+                    .into_iter()
+                    .map(|(key, credential)| {
+                        (key, CachedEntry::seed(credential, cloud.jitter_prefetch))
+                    })
+                    .collect();
+            }
+        }
+        Some(Self::new(cloud, endpoint, entries, keyed_seeds))
+    }
+
+    fn keyed_seed_for_path(&self, path: &str) -> Result<Option<CachedEntry>> {
+        let SeedStrategy::Keyed(strategy) = &self.cloud.seed_strategy else {
+            return Ok(None);
+        };
+        let resolved = (strategy.resolve_path)(path)?;
+        let Some(seed) = resolved
+            .keys
+            .iter()
+            .find_map(|key| self.keyed_seeds.get(key))
+        else {
+            return Ok(None);
+        };
+        let credential = StorageCredential::new(resolved.scope, seed.credential.config().clone());
+        // The scope is normalized, e.g. with a lowercase scheme, so it may not
+        // cover the path as written; the path then fetches its credential.
+        Ok(credential.covers(path).then_some(CachedEntry {
+            credential,
+            expires_at: seed.expires_at,
+            refresh_at: seed.refresh_at,
+        }))
+    }
+}
+
+/// Serializable recipe for rebuilding a [`RestVendedCredentialProvider`] from
+/// the FileIO properties, mirroring how Java rebuilds its provider on workers.
+#[derive(Clone, Serialize, Deserialize)]
+pub(crate) struct RestVendedCredentialProviderFactory {
+    /// Catalog URI, used to resolve relative refresh endpoints.
+    catalog_uri: String,
+    table: TableIdent,
+    /// The `rest.auth.type` the catalog resolved, so a rebuilt provider uses
+    /// the same auth manager even when the merged FileIO properties would
+    /// infer another one.
+    auth_type: String,
+    /// Whether `auth_type` was inferred rather than configured. Like Java,
+    /// a table `token` then authenticates a catalog without auth.
+    auth_type_inferred: bool,
+    /// The auth-related part of the unmerged config returned by the table
+    /// endpoint, from which the auth manager derives a table session. Keeping
+    /// it separate prevents local FileIO overrides from masking table auth.
+    table_auth: HashMap<String, String>,
+}
+
+/// Table config keys that may override authentication in a table session.
+const TABLE_AUTH_KEYS: &[&str] = &["token"];
+
+impl RestVendedCredentialProviderFactory {
+    pub(crate) fn new(
+        catalog_uri: impl Into<String>,
+        table: TableIdent,
+        auth_type: impl Into<String>,
+        auth_type_inferred: bool,
+        table_config: &HashMap<String, String>,
+    ) -> Self {
+        Self {
+            catalog_uri: catalog_uri.into(),
+            table,
+            auth_type: auth_type.into(),
+            auth_type_inferred,
+            table_auth: table_config
+                .iter()
+                .filter(|(key, _)| TABLE_AUTH_KEYS.contains(&key.as_str()))
+                .map(|(key, value)| (key.clone(), value.clone()))
+                .collect(),
+        }
+    }
+
+    /// The table token that authenticates credential requests of a catalog
+    /// without auth. Java's credential providers infer OAuth2 from a `token`
+    /// unless `rest.auth.type` is configured.
+    fn bearer_table_token(&self) -> Option<&str> {
+        (self.auth_type == AUTH_TYPE_NONE && self.auth_type_inferred)
+            .then(|| self.table_auth.get("token").map(String::as_str))
+            .flatten()
+    }
+
+    /// Connect to the catalog from the FileIO properties, as the catalog would.
+    async fn connect(&self, props: &HashMap<String, String>) -> Result<HttpClient> {
+        let mut config_props = props.clone();
+        config_props.insert(
+            REST_CATALOG_PROP_AUTH_TYPE.to_string(),
+            self.auth_type.clone(),
+        );
+        let config = RestCatalogConfig::builder()
+            .uri(self.catalog_uri.clone())
+            .props(config_props)
+            .build();
+        let auth_manager = load_auth_manager(&config)?;
+        let client = HttpClient::new(&config)?;
+        let session = auth_manager
+            .catalog_session(&client.without_auth_session(), &config.auth_props())
+            .await?;
+        table_client(
+            &client.with_auth_session(session),
+            auth_manager.as_ref(),
+            &self.table,
+            props,
+            &self.table_auth,
+            self.bearer_table_token(),
+        )
+        .await
+    }
+}
+
+impl std::fmt::Debug for RestVendedCredentialProviderFactory {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("RestVendedCredentialProviderFactory")
+            .field("catalog_uri", &self.catalog_uri)
+            .field("table", &self.table)
+            .finish_non_exhaustive()
+    }
+}
+
+#[typetag::serde(name = "RestVendedCredentialProviderFactory")]
+impl StorageCredentialProviderFactory for RestVendedCredentialProviderFactory {
+    fn build(&self, config: &StorageConfig) -> Result<Arc<dyn StorageCredentialProvider>> {
+        let provider =
+            RestVendedCredentialProvider::configure(self, config.props()).ok_or_else(|| {
+                Error::new(
+                    ErrorKind::DataInvalid,
+                    "FileIO configuration no longer configures vended credentials",
+                )
+            })?;
+        Ok(Arc::new(RestVendedCredentialProvider {
+            factory: Some(self.clone()),
+            ..provider
+        }))
+    }
+}
+
+/// Derive the table-scoped client used for credential requests.
+///
+/// A `bearer_table_token` authenticates instead of a session from the auth
+/// manager.
+async fn table_client(
+    catalog_client: &HttpClient,
+    auth_manager: &dyn AuthManager,
+    table: &TableIdent,
+    props: &HashMap<String, String>,
+    table_config: &HashMap<String, String>,
+    bearer_table_token: Option<&str>,
+) -> Result<HttpClient> {
+    let session = match bearer_table_token {
+        Some(token) => static_token_session(token),
+        None => {
+            auth_manager
+                .table_session(
+                    &catalog_client.without_auth_session(),
+                    table,
+                    table_config,
+                    catalog_client.auth_session(),
+                )
+                .await?
+        }
+    };
+    catalog_client.for_table(props, session)
+}
+
+/// Serves vended credentials for a table, refreshing them from the REST
+/// catalog's table credentials endpoint.
+///
+/// Each cloud cache is seeded with the credentials from the initial
+/// table properties (when complete) and re-fetched from its endpoint as they
+/// near expiry.
+pub(crate) struct RestVendedCredentialProvider {
+    /// Table-scoped catalog client. Set when the catalog builds the provider,
+    /// and connected on first refresh after deserialization.
+    client: OnceCell<HttpClient>,
+    /// Rebuilds this provider in another process. `None` when the catalog
+    /// authentication cannot be rebuilt from properties.
+    factory: Option<RestVendedCredentialProviderFactory>,
+    /// Effective FileIO properties, which supply catalog connection settings.
+    props: HashMap<String, String>,
+    /// Query parameters sent with every credentials request.
+    query_params: Vec<(&'static str, String)>,
+    /// Independently configured endpoint and cache for each backing cloud.
+    clouds: Vec<ConfiguredCloud>,
+}
+
+impl RestVendedCredentialProvider {
+    /// Configure a provider without a catalog connection, or `None` when no
+    /// supported cloud advertises an enabled refresh endpoint.
+    fn configure(
+        factory: &RestVendedCredentialProviderFactory,
+        props: &HashMap<String, String>,
+    ) -> Option<Self> {
+        let clouds = CloudRefresh::SUPPORTED
+            .iter()
+            .filter_map(|cloud| ConfiguredCloud::configure(cloud, &factory.catalog_uri, props))
+            .collect::<Vec<_>>();
+        (!clouds.is_empty()).then(|| Self {
+            client: OnceCell::new(),
+            factory: None,
+            props: props.clone(),
+            query_params: credentials_query_params(props),
+            clouds,
+        })
+    }
+
+    fn configured_cloud_for_location(&self, location: &str) -> Option<&ConfiguredCloud> {
+        self.clouds
+            .iter()
+            .find(|configured| configured.cloud.matches_location(location))
+    }
+
+    async fn client(&self) -> Result<&HttpClient> {
+        self.client
+            .get_or_try_init(|| async {
+                let factory = self.factory.as_ref().ok_or_else(|| {
+                    Error::new(
+                        ErrorKind::Unexpected,
+                        "vended credential provider has no catalog connection",
+                    )
+                })?;
+                factory.connect(&self.props).await
+            })
+            .await
+    }
+
+    /// Fetch fresh credentials from the catalog's credentials endpoint.
+    async fn fetch(&self, configured: &ConfiguredCloud) -> Result<ParsedCredentials> {
+        let cloud = configured.cloud;
+        let client = self.client().await?;
+        let url = credentials_url(&configured.endpoint, &self.query_params)?;
+        let request = HttpRequest::build(client.request(Method::GET, url))?;
+        let response = client.query_catalog(request).await?;
+
+        if !response.status().is_success() {
+            return Err(unexpected_catalog_error_without_body(
+                response,
+                client.disable_header_redaction(),
+            ));
+        }
+
+        // Credential responses contain secrets, and serde's errors quote the
+        // values they reject, so only the error's category and position are
+        // reported.
+        let parsed: LoadCredentialsResponse =
+            serde_json::from_slice(response.body()).map_err(|error| {
+                Error::new(
+                    ErrorKind::Unexpected,
+                    format!(
+                        "failed to parse vended credential response: {:?} error at line {}, \
+                         column {}",
+                        error.classify(),
+                        error.line(),
+                        error.column()
+                    ),
+                )
+            })?;
+        let now = SystemTime::now();
+        let mut entries = Vec::new();
+        let mut errors = Vec::new();
+        for credential in parsed
+            .storage_credentials
+            .into_iter()
+            .filter(|credential| cloud.matches_location(&credential.prefix))
+        {
+            let prefix = credential.prefix;
+            let parsed = (cloud.parse_credential)(&credential.config, prefix.clone())
+                .map(|vended| CachedEntry::new(vended, cloud.jitter_prefetch))
+                .and_then(|entry| {
+                    if entry.is_unexpired(now) {
+                        Ok(entry)
+                    } else {
+                        Err(Error::new(
+                            ErrorKind::DataInvalid,
+                            "invalid vended credential: credential is already expired",
+                        ))
+                    }
+                });
+            match parsed {
+                Ok(entry) => entries.push(entry),
+                Err(error) => errors.push(CredentialError { prefix, error }),
+            }
+        }
+        Ok(ParsedCredentials { entries, errors })
+    }
+
+    async fn refresh_credential(
+        &self,
+        configured: &ConfiguredCloud,
+        path: &str,
+        fallback: Option<CachedEntry>,
+    ) -> Result<StorageCredential> {
+        let fetched = self.fetch(configured).await;
+        let mut cache = configured.cache.lock().await;
+        let now = SystemTime::now();
+        cache.entries.retain(|entry| entry.is_unexpired(now));
+
+        let (fetched_prefixes, failure) = match fetched {
+            Ok(ParsedCredentials { entries, errors }) => {
+                let fetched_prefixes = cache.merge(entries, &errors);
+                // The most specific error for `path`.
+                let failure = errors
+                    .into_iter()
+                    .filter(|error| prefix_covers(&error.prefix, path))
+                    .max_by_key(|error| error.prefix.len())
+                    .map(|error| error.error);
+                (Some(fetched_prefixes), failure)
+            }
+            Err(error) => (None, Some(error)),
+        };
+
+        let selected = cache.unexpired_match(path, now);
+        // A keyed seed is only cached once served as the fallback.
+        let seed = selected
+            .is_none()
+            .then(|| {
+                fallback
+                    .clone()
+                    .filter(|fallback| fallback.is_unexpired(now))
+            })
+            .flatten();
+        let candidate = selected
+            .map(|index| &cache.entries[index].credential)

Review Comment:
   `cache.entries[index]` appears six times between here and line 795, plus once in `cache_decision`. It's correct today because nothing changes `entries` between `unexpired_match` and those uses. But `insert` does a `retain` and `merge` reorders the vector, so one reordered edit later turns this into a panic on the credential path. I'd use `get`/`get_mut` and return an error, or have `unexpired_match` return the entry itself instead of an index.



##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,3675 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Vended storage credentials from a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache. Like
+//! Java's `StorageCredential`, the returned [`StorageCredential`] holds the
+//! backend's storage properties, which the storage adapter turns into its own
+//! credential. This preserves Java's per-cloud credential selection and
+//! prefetch policies while supporting mixed-cloud tables through a resolving
+//! FileIO. Unlike Java's scheduled refresh, which permanently stops
+//! after a failed fetch, transient failures are retried here with jittered
+//! exponential backoff while an unexpired credential remains available.
+//!
+//! Like Java's `VendedCredentialsProvider`, a provider is rebuilt from the
+//! FileIO properties after [`FileIO`](iceberg::io::FileIO) serialization and
+//! connects to the catalog lazily in the receiving process. This requires
+//! catalog authentication that can be rebuilt from `rest.auth.type`.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant. To
+//! add a backend, teach its storage adapter to read its credential properties
+//! from a [`StorageCredential`]. Then write its `parse_*` function, add a
+//! `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+    ADLS_REFRESH_CREDENTIALS_ENABLED, ADLS_REFRESH_CREDENTIALS_ENDPOINT,
+    ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX, ADLS_SAS_TOKEN_PREFIX, AWS_REFRESH_CREDENTIALS_ENABLED,
+    AWS_REFRESH_CREDENTIALS_ENDPOINT, GCS_REFRESH_CREDENTIALS_ENABLED,
+    GCS_REFRESH_CREDENTIALS_ENDPOINT, GCS_TOKEN, GCS_TOKEN_EXPIRES_AT, S3_ACCESS_KEY_ID,
+    S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN, S3_SESSION_TOKEN_EXPIRES_AT_MS, StorageConfig,
+    StorageCredential, StorageCredentialProvider, StorageCredentialProviderFactory,
+};
+use iceberg::{Error, ErrorKind, Result, TableIdent};
+use rand::Rng;
+use reqwest::{Method, Url};
+use serde::{Deserialize, Serialize};
+use tokio::sync::{Mutex, OnceCell};
+
+use crate::auth::{AUTH_TYPE_NONE, AuthManager, load_auth_manager, static_token_session};
+use crate::catalog::{
+    REST_CATALOG_PROP_AUTH_TYPE, REST_CATALOG_PROP_REFERENCED_BY, REST_CATALOG_PROP_SCAN_PLAN_ID,
+    RestCatalogConfig,
+};
+use crate::client::{HttpClient, unexpected_catalog_error_without_body};
+use crate::request::HttpRequest;
+use crate::types::LoadCredentialsResponse;
+
+type CredentialParser =
+    fn(config: &HashMap<String, String>, prefix: String) -> Result<VendedCredential>;
+type KeyedSeedCredentialParser =
+    fn(config: &HashMap<String, String>) -> HashMap<String, VendedCredential>;
+type KeyedSeedPathResolver = fn(path: &str) -> Result<KeyedSeedPath>;
+
+/// A credential vended by the catalog, with the expiry its config declares.
+#[derive(Clone)]
+struct VendedCredential {
+    credential: StorageCredential,
+    expires_at: SystemTime,
+}
+
+enum SeedStrategy {
+    /// One credential stored in the backend's flat properties, for every
+    /// location with one of the scheme prefixes, like Java's root clients.
+    Flat { prefixes: &'static [&'static str] },
+    /// Credentials selected by a backend-specific key derived from each path.
+    Keyed(KeyedSeedStrategy),
+}
+
+struct KeyedSeedStrategy {
+    parse_credentials: KeyedSeedCredentialParser,
+    resolve_path: KeyedSeedPathResolver,
+}
+
+struct KeyedSeedPath {
+    /// Seed keys for the path, most specific first.
+    keys: Vec<String>,
+    scope: String,
+}
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// 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 {
+    /// Location URL schemes this backend serves.
+    schemes: &'static [&'static str],
+    /// Table property naming the refresh endpoint (absolute or catalog-relative).
+    endpoint_key: &'static str,
+    /// Table property controlling refresh; only missing or case-insensitive `"true"` enables it.
+    enabled_key: &'static str,
+    /// Table property that must be present for refresh to apply, as in Java,
+    /// where it decides whether the backend uses vended credentials at all.
+    required_key: Option<&'static str>,
+    /// Whether to jitter successful prefetch times like AWS `CachedSupplier`.
+    jitter_prefetch: bool,
+    /// Parse a complete credential from catalog-supplied properties.
+    parse_credential: CredentialParser,
+    /// How initial credentials in the table properties are selected.
+    seed_strategy: SeedStrategy,
+}
+
+impl CloudRefresh {
+    /// S3 / AWS
+    const AWS: Self = Self {
+        schemes: &["s3", "s3a", "s3n"],
+        endpoint_key: AWS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: AWS_REFRESH_CREDENTIALS_ENABLED,
+        required_key: None,
+        jitter_prefetch: true,
+        parse_credential: parse_s3_credential,
+        // `s3` also covers `s3a` and `s3n` locations.
+        seed_strategy: SeedStrategy::Flat { prefixes: &["s3"] },
+    };
+    /// Google Cloud Storage
+    const GCP: Self = Self {
+        schemes: &["gs", "gcs"],
+        endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED,
+        // Java's GCS client only refreshes a vended OAuth2 token; without
+        // one it uses Google's default credentials.
+        required_key: Some(GCS_TOKEN),
+        jitter_prefetch: false,
+        parse_credential: parse_gcs_credential,
+        seed_strategy: SeedStrategy::Flat {
+            prefixes: &["gs", "gcs"],
+        },
+    };
+    /// Azure Data Lake Storage
+    const AZURE: Self = Self {
+        schemes: &["abfs", "abfss", "wasb", "wasbs"],
+        endpoint_key: ADLS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: ADLS_REFRESH_CREDENTIALS_ENABLED,
+        required_key: None,
+        jitter_prefetch: false,
+        parse_credential: parse_azdls_credential,
+        seed_strategy: SeedStrategy::Keyed(KeyedSeedStrategy {
+            parse_credentials: parse_azdls_account_seeds,
+            resolve_path: resolve_azdls_seed_path,
+        }),
+    };
+
+    /// Backends with refresh support
+    const SUPPORTED: &[Self] = &[Self::AWS, Self::GCP, Self::AZURE];
+
+    fn matches_location(&self, location: &str) -> bool {
+        self.schemes
+            .iter()
+            .any(|scheme| location.eq_ignore_ascii_case(scheme))
+            || scheme_of(location).is_some_and(|scheme| self.schemes.contains(&scheme.as_str()))
+    }
+}
+
+/// Re-fetch a credential once it is within this window of expiry, so a fresh
+/// token is in hand before the object store would reject the old one.
+const REFRESH_BUFFER: Duration = Duration::from_mins(5);
+
+/// AWS keeps at least one minute between its jittered prefetch time and expiry.
+const MIN_REFRESH_BUFFER: Duration = Duration::from_mins(1);
+
+/// Initial ceiling for failure backoff. Equal jitter chooses from half this
+/// value through the full value.
+const INITIAL_FAILURE_BACKOFF: Duration = Duration::from_secs(1);
+
+/// Maximum delay between failed refresh attempts.
+const MAX_FAILURE_BACKOFF: Duration = Duration::from_secs(30);
+
+/// A cached vended credential and its refresh schedule.
+#[derive(Clone)]
+struct CachedEntry {
+    credential: StorageCredential,
+    expires_at: SystemTime,
+    /// When this entry becomes eligible for prefetch.
+    refresh_at: SystemTime,
+}
+
+impl CachedEntry {
+    fn new(vended: VendedCredential, jitter_prefetch: bool) -> Self {
+        let VendedCredential {
+            credential,
+            expires_at,
+        } = vended;
+        Self {
+            credential,
+            expires_at,
+            refresh_at: prefetch_time(SystemTime::now(), expires_at, jitter_prefetch),
+        }
+    }
+
+    /// Seed entries that are already inside the nominal five-minute window are
+    /// immediately due. Otherwise AWS applies the same jitter as it does to a
+    /// freshly fetched value.
+    fn seed(vended: VendedCredential, jitter_prefetch: bool) -> Self {
+        let due = SystemTime::now()
+            .checked_add(REFRESH_BUFFER)
+            .is_none_or(|refresh_boundary| refresh_boundary >= vended.expires_at);
+        let mut entry = Self::new(vended, jitter_prefetch);
+        if due {
+            entry.refresh_at = UNIX_EPOCH;
+        }
+        entry
+    }
+
+    fn is_fresh(&self, now: SystemTime) -> bool {
+        now < self.refresh_at
+    }
+
+    fn is_unexpired(&self, now: SystemTime) -> bool {
+        now < self.expires_at
+    }
+}
+
+/// Parsed entries from one successful credentials response.
+struct ParsedCredentials {
+    entries: Vec<CachedEntry>,
+    errors: Vec<CredentialError>,
+}
+
+struct CredentialError {
+    prefix: String,
+    error: Error,
+}
+
+/// Cached credentials plus failure-backoff state.
+///
+/// A fetch returns the credentials for every prefix of the table, so a path
+/// the last response did not cover would not be covered by an immediate
+/// re-fetch either. Backoff is therefore shared by the whole cloud.
+struct CacheState {
+    entries: Vec<CachedEntry>,
+    /// Prefixes whose credential in the last successful response was invalid.
+    /// Paths under them must not use a broader credential: the catalog scoped
+    /// them more tightly.
+    failed_prefixes: HashSet<String>,
+    consecutive_failures: u32,
+    retry_not_before: Option<Instant>,
+    /// Why the last refresh failed, reported while refresh is backed off.
+    last_failure: Option<String>,
+}
+
+impl CacheState {
+    fn new(entries: Vec<CachedEntry>) -> Self {
+        Self {
+            entries,
+            failed_prefixes: HashSet::new(),
+            consecutive_failures: 0,
+            retry_not_before: None,
+            last_failure: None,
+        }
+    }
+
+    /// The index of the unexpired credential with the longest prefix covering
+    /// `path`.
+    fn unexpired_match(&self, path: &str, now: SystemTime) -> Option<usize> {
+        self.entries
+            .iter()
+            .enumerate()
+            .filter(|(_, entry)| entry.is_unexpired(now) && entry.credential.covers(path))
+            .max_by_key(|(_, entry)| entry.credential.prefix().len())
+            .map(|(index, _)| index)
+    }
+
+    /// Whether a failed prefix covering `path` is narrower than `credential`'s.
+    fn masks(&self, path: &str, credential: Option<&StorageCredential>) -> bool {
+        let prefix_len = credential.map(|credential| credential.prefix().len());
+        self.failed_prefixes.iter().any(|failed| {
+            prefix_covers(failed, path) && prefix_len.is_none_or(|len| failed.len() > len)
+        })
+    }
+
+    /// Cache `entry`, replacing any entry with the same prefix, and return its
+    /// index.
+    fn insert(&mut self, entry: CachedEntry) -> usize {
+        let prefix = entry.credential.prefix().to_owned();
+        self.entries
+            .retain(|cached| cached.credential.prefix() != prefix);
+        self.entries.push(entry);
+        self.entries.len() - 1
+    }
+
+    fn record_success(&mut self) {
+        self.consecutive_failures = 0;
+        self.retry_not_before = None;
+        self.last_failure = None;
+    }
+
+    fn record_failure(&mut self, error: &Error) {
+        self.consecutive_failures = self.consecutive_failures.saturating_add(1);
+        self.retry_not_before =
+            Instant::now().checked_add(failure_backoff(self.consecutive_failures));
+        self.last_failure = Some(error.to_string());
+    }
+
+    /// Replace cached credentials with the fetched ones, prefix by prefix,
+    /// and record the prefixes whose credential was invalid in this response.
+    ///
+    /// An unexpired cached credential survives when the response carries no
+    /// valid replacement for its prefix, so an absent or malformed entry never
+    /// evicts a usable credential. Returns the prefixes the response replaced.
+    fn merge(&mut self, fetched: Vec<CachedEntry>, errors: &[CredentialError]) -> HashSet<String> {
+        let fetched_prefixes = fetched
+            .iter()
+            .map(|entry| entry.credential.prefix().to_owned())
+            .collect::<HashSet<_>>();
+        // A response names every prefix of the table, so a prefix it omits is
+        // no longer scoped separately: a fetched credential that covers it
+        // replaces its cached one. A prefix whose credential is invalid keeps
+        // its cached one.
+        let failed_prefixes = errors
+            .iter()
+            .map(|error| error.prefix.clone())
+            .collect::<HashSet<_>>();
+        self.entries.retain(|entry| {
+            let prefix = entry.credential.prefix();
+            failed_prefixes.contains(prefix)
+                || !fetched
+                    .iter()
+                    .any(|fetched| fetched.credential.covers(prefix))
+        });
+        self.entries.extend(fetched);
+        self.failed_prefixes = failed_prefixes;
+        fetched_prefixes
+    }
+}
+
+struct ConfiguredCloud {
+    cloud: &'static CloudRefresh,
+    endpoint: String,
+    cache: Mutex<CacheState>,
+    /// Initial property credentials whose scope cannot be represented by a
+    /// single leading URI prefix, keyed according to the cloud strategy.
+    keyed_seeds: HashMap<String, CachedEntry>,
+    /// Only one caller fetches at a time. The cache lock is deliberately
+    /// separate so other callers can keep using an unexpired credential while
+    /// the refresh is in flight.
+    refresh: Mutex<()>,
+}
+
+impl ConfiguredCloud {
+    fn new(
+        cloud: &'static CloudRefresh,
+        endpoint: String,
+        entries: Vec<CachedEntry>,
+        keyed_seeds: HashMap<String, CachedEntry>,
+    ) -> Self {
+        Self {
+            cloud,
+            endpoint,
+            cache: Mutex::new(CacheState::new(entries)),
+            keyed_seeds,
+            refresh: Mutex::new(()),
+        }
+    }
+
+    /// Configure `cloud` from the FileIO properties, or `None` when they
+    /// advertise no enabled refresh endpoint for it.
+    fn configure(
+        cloud: &'static CloudRefresh,
+        base_uri: &str,
+        props: &HashMap<String, String>,
+    ) -> Option<Self> {
+        let enabled = props
+            .get(cloud.enabled_key)
+            .is_none_or(|value| value.eq_ignore_ascii_case("true"))
+            && cloud
+                .required_key
+                .is_none_or(|key| props.get(key).is_some_and(|value| !value.is_empty()));
+        let endpoint = props
+            .get(cloud.endpoint_key)
+            .filter(|endpoint| enabled && !endpoint.is_empty())
+            .map(|endpoint| resolve_endpoint(base_uri, endpoint))?;
+
+        let mut entries = Vec::new();
+        let mut keyed_seeds = HashMap::new();
+        match &cloud.seed_strategy {
+            SeedStrategy::Flat { prefixes } => {
+                entries.extend(prefixes.iter().filter_map(|prefix| {
+                    (cloud.parse_credential)(props, prefix.to_string())
+                        .ok()

Review Comment:
   A seed that fails `parse_s3_credential` is dropped here with no trace. The vended chain replaces the whole S3 credential chain in `s3_config_build`, so the static `s3.access-key-id`/`s3.secret-access-key` from the same config stop being used too. Take a catalog that advertises `client.refresh-credentials-endpoint` but sends keys without `s3.session-token-expires-at-ms`. That works on main today. Here the first file op goes to the endpoint, and if that call fails, every S3 op fails.
   
   I agree with not falling back to ambient AWS credentials. But the keys in `config` aren't ambient: the catalog vended them. I'd keep them as a seed with no expiry, which the trait doc already allows. At the very least I'd `tracing::warn!` the parse error so the unexpected network fetch has an explanation.



##########
crates/storage/opendal/src/azdls.rs:
##########
@@ -226,16 +372,24 @@ pub(crate) struct AzureStoragePath {
 }
 
 impl AzureStoragePath {
+    /// The host of the path, e.g. `account.dfs.core.windows.net`.
+    fn host(&self) -> String {
+        let service = match self.scheme {
+            AzureStorageScheme::Abfs | AzureStorageScheme::Abfss => "dfs",
+            AzureStorageScheme::Wasb | AzureStorageScheme::Wasbs => "blob",
+        };
+        format!("{}.{service}.{}", self.account_name, self.endpoint_suffix)
+    }
+
     /// Converts the AzureStoragePath into a full endpoint URL.
     ///
     /// This is possible because the path is fully qualified.
+    ///
+    /// Like Iceberg Java, the endpoint always uses TLS, also for the
+    /// non-secure schemes: SAS tokens are query parameters and must not be
+    /// sent over a plaintext connection unless the user configures one.
     fn as_endpoint(&self) -> String {
-        format!(
-            "{}://{}.dfs.{}",
-            self.scheme.as_http_scheme(),
-            self.account_name,
-            self.endpoint_suffix
-        )
+        format!("https://{}.dfs.{}", self.account_name, self.endpoint_suffix)

Review Comment:
   This switches `abfs://`/`wasb://` from `http` to `https` for everyone, and `requires_tls` changes which configured endpoints are accepted. Keeping SAS tokens off plaintext connections and matching Java are good reasons. But it's a user-visible change unrelated to refresh, so I'd move it into the ADLS piece and call it out in the changelog.



##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,3675 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Vended storage credentials from a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache. Like
+//! Java's `StorageCredential`, the returned [`StorageCredential`] holds the
+//! backend's storage properties, which the storage adapter turns into its own
+//! credential. This preserves Java's per-cloud credential selection and
+//! prefetch policies while supporting mixed-cloud tables through a resolving
+//! FileIO. Unlike Java's scheduled refresh, which permanently stops
+//! after a failed fetch, transient failures are retried here with jittered
+//! exponential backoff while an unexpired credential remains available.
+//!
+//! Like Java's `VendedCredentialsProvider`, a provider is rebuilt from the
+//! FileIO properties after [`FileIO`](iceberg::io::FileIO) serialization and
+//! connects to the catalog lazily in the receiving process. This requires
+//! catalog authentication that can be rebuilt from `rest.auth.type`.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant. To
+//! add a backend, teach its storage adapter to read its credential properties
+//! from a [`StorageCredential`]. Then write its `parse_*` function, add a
+//! `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+    ADLS_REFRESH_CREDENTIALS_ENABLED, ADLS_REFRESH_CREDENTIALS_ENDPOINT,
+    ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX, ADLS_SAS_TOKEN_PREFIX, AWS_REFRESH_CREDENTIALS_ENABLED,
+    AWS_REFRESH_CREDENTIALS_ENDPOINT, GCS_REFRESH_CREDENTIALS_ENABLED,
+    GCS_REFRESH_CREDENTIALS_ENDPOINT, GCS_TOKEN, GCS_TOKEN_EXPIRES_AT, S3_ACCESS_KEY_ID,
+    S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN, S3_SESSION_TOKEN_EXPIRES_AT_MS, StorageConfig,
+    StorageCredential, StorageCredentialProvider, StorageCredentialProviderFactory,
+};
+use iceberg::{Error, ErrorKind, Result, TableIdent};
+use rand::Rng;
+use reqwest::{Method, Url};
+use serde::{Deserialize, Serialize};
+use tokio::sync::{Mutex, OnceCell};
+
+use crate::auth::{AUTH_TYPE_NONE, AuthManager, load_auth_manager, static_token_session};
+use crate::catalog::{
+    REST_CATALOG_PROP_AUTH_TYPE, REST_CATALOG_PROP_REFERENCED_BY, REST_CATALOG_PROP_SCAN_PLAN_ID,
+    RestCatalogConfig,
+};
+use crate::client::{HttpClient, unexpected_catalog_error_without_body};
+use crate::request::HttpRequest;
+use crate::types::LoadCredentialsResponse;
+
+type CredentialParser =
+    fn(config: &HashMap<String, String>, prefix: String) -> Result<VendedCredential>;
+type KeyedSeedCredentialParser =
+    fn(config: &HashMap<String, String>) -> HashMap<String, VendedCredential>;
+type KeyedSeedPathResolver = fn(path: &str) -> Result<KeyedSeedPath>;
+
+/// A credential vended by the catalog, with the expiry its config declares.
+#[derive(Clone)]
+struct VendedCredential {
+    credential: StorageCredential,
+    expires_at: SystemTime,
+}
+
+enum SeedStrategy {
+    /// One credential stored in the backend's flat properties, for every
+    /// location with one of the scheme prefixes, like Java's root clients.
+    Flat { prefixes: &'static [&'static str] },
+    /// Credentials selected by a backend-specific key derived from each path.
+    Keyed(KeyedSeedStrategy),
+}
+
+struct KeyedSeedStrategy {
+    parse_credentials: KeyedSeedCredentialParser,
+    resolve_path: KeyedSeedPathResolver,
+}
+
+struct KeyedSeedPath {
+    /// Seed keys for the path, most specific first.
+    keys: Vec<String>,
+    scope: String,
+}
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// 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 {
+    /// Location URL schemes this backend serves.
+    schemes: &'static [&'static str],
+    /// Table property naming the refresh endpoint (absolute or catalog-relative).
+    endpoint_key: &'static str,
+    /// Table property controlling refresh; only missing or case-insensitive `"true"` enables it.
+    enabled_key: &'static str,
+    /// Table property that must be present for refresh to apply, as in Java,
+    /// where it decides whether the backend uses vended credentials at all.
+    required_key: Option<&'static str>,
+    /// Whether to jitter successful prefetch times like AWS `CachedSupplier`.
+    jitter_prefetch: bool,
+    /// Parse a complete credential from catalog-supplied properties.
+    parse_credential: CredentialParser,
+    /// How initial credentials in the table properties are selected.
+    seed_strategy: SeedStrategy,
+}
+
+impl CloudRefresh {
+    /// S3 / AWS
+    const AWS: Self = Self {
+        schemes: &["s3", "s3a", "s3n"],
+        endpoint_key: AWS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: AWS_REFRESH_CREDENTIALS_ENABLED,
+        required_key: None,
+        jitter_prefetch: true,
+        parse_credential: parse_s3_credential,
+        // `s3` also covers `s3a` and `s3n` locations.
+        seed_strategy: SeedStrategy::Flat { prefixes: &["s3"] },
+    };
+    /// Google Cloud Storage
+    const GCP: Self = Self {
+        schemes: &["gs", "gcs"],
+        endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED,
+        // Java's GCS client only refreshes a vended OAuth2 token; without
+        // one it uses Google's default credentials.
+        required_key: Some(GCS_TOKEN),
+        jitter_prefetch: false,
+        parse_credential: parse_gcs_credential,
+        seed_strategy: SeedStrategy::Flat {
+            prefixes: &["gs", "gcs"],
+        },
+    };
+    /// Azure Data Lake Storage
+    const AZURE: Self = Self {
+        schemes: &["abfs", "abfss", "wasb", "wasbs"],
+        endpoint_key: ADLS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: ADLS_REFRESH_CREDENTIALS_ENABLED,
+        required_key: None,
+        jitter_prefetch: false,
+        parse_credential: parse_azdls_credential,
+        seed_strategy: SeedStrategy::Keyed(KeyedSeedStrategy {
+            parse_credentials: parse_azdls_account_seeds,
+            resolve_path: resolve_azdls_seed_path,
+        }),
+    };
+
+    /// Backends with refresh support
+    const SUPPORTED: &[Self] = &[Self::AWS, Self::GCP, Self::AZURE];
+
+    fn matches_location(&self, location: &str) -> bool {
+        self.schemes
+            .iter()
+            .any(|scheme| location.eq_ignore_ascii_case(scheme))
+            || scheme_of(location).is_some_and(|scheme| self.schemes.contains(&scheme.as_str()))
+    }
+}
+
+/// Re-fetch a credential once it is within this window of expiry, so a fresh
+/// token is in hand before the object store would reject the old one.
+const REFRESH_BUFFER: Duration = Duration::from_mins(5);
+
+/// AWS keeps at least one minute between its jittered prefetch time and expiry.
+const MIN_REFRESH_BUFFER: Duration = Duration::from_mins(1);
+
+/// Initial ceiling for failure backoff. Equal jitter chooses from half this
+/// value through the full value.
+const INITIAL_FAILURE_BACKOFF: Duration = Duration::from_secs(1);
+
+/// Maximum delay between failed refresh attempts.
+const MAX_FAILURE_BACKOFF: Duration = Duration::from_secs(30);
+
+/// A cached vended credential and its refresh schedule.
+#[derive(Clone)]
+struct CachedEntry {
+    credential: StorageCredential,
+    expires_at: SystemTime,
+    /// When this entry becomes eligible for prefetch.
+    refresh_at: SystemTime,
+}
+
+impl CachedEntry {
+    fn new(vended: VendedCredential, jitter_prefetch: bool) -> Self {
+        let VendedCredential {
+            credential,
+            expires_at,
+        } = vended;
+        Self {
+            credential,
+            expires_at,
+            refresh_at: prefetch_time(SystemTime::now(), expires_at, jitter_prefetch),
+        }
+    }
+
+    /// Seed entries that are already inside the nominal five-minute window are
+    /// immediately due. Otherwise AWS applies the same jitter as it does to a
+    /// freshly fetched value.
+    fn seed(vended: VendedCredential, jitter_prefetch: bool) -> Self {
+        let due = SystemTime::now()
+            .checked_add(REFRESH_BUFFER)
+            .is_none_or(|refresh_boundary| refresh_boundary >= vended.expires_at);
+        let mut entry = Self::new(vended, jitter_prefetch);
+        if due {
+            entry.refresh_at = UNIX_EPOCH;
+        }
+        entry
+    }
+
+    fn is_fresh(&self, now: SystemTime) -> bool {
+        now < self.refresh_at
+    }
+
+    fn is_unexpired(&self, now: SystemTime) -> bool {
+        now < self.expires_at
+    }
+}
+
+/// Parsed entries from one successful credentials response.
+struct ParsedCredentials {
+    entries: Vec<CachedEntry>,
+    errors: Vec<CredentialError>,
+}
+
+struct CredentialError {
+    prefix: String,
+    error: Error,
+}
+
+/// Cached credentials plus failure-backoff state.
+///
+/// A fetch returns the credentials for every prefix of the table, so a path
+/// the last response did not cover would not be covered by an immediate
+/// re-fetch either. Backoff is therefore shared by the whole cloud.
+struct CacheState {
+    entries: Vec<CachedEntry>,
+    /// Prefixes whose credential in the last successful response was invalid.
+    /// Paths under them must not use a broader credential: the catalog scoped
+    /// them more tightly.
+    failed_prefixes: HashSet<String>,
+    consecutive_failures: u32,
+    retry_not_before: Option<Instant>,
+    /// Why the last refresh failed, reported while refresh is backed off.
+    last_failure: Option<String>,
+}
+
+impl CacheState {
+    fn new(entries: Vec<CachedEntry>) -> Self {
+        Self {
+            entries,
+            failed_prefixes: HashSet::new(),
+            consecutive_failures: 0,
+            retry_not_before: None,
+            last_failure: None,
+        }
+    }
+
+    /// The index of the unexpired credential with the longest prefix covering
+    /// `path`.
+    fn unexpired_match(&self, path: &str, now: SystemTime) -> Option<usize> {
+        self.entries
+            .iter()
+            .enumerate()
+            .filter(|(_, entry)| entry.is_unexpired(now) && entry.credential.covers(path))
+            .max_by_key(|(_, entry)| entry.credential.prefix().len())
+            .map(|(index, _)| index)
+    }
+
+    /// Whether a failed prefix covering `path` is narrower than `credential`'s.
+    fn masks(&self, path: &str, credential: Option<&StorageCredential>) -> bool {
+        let prefix_len = credential.map(|credential| credential.prefix().len());
+        self.failed_prefixes.iter().any(|failed| {
+            prefix_covers(failed, path) && prefix_len.is_none_or(|len| failed.len() > len)
+        })
+    }
+
+    /// Cache `entry`, replacing any entry with the same prefix, and return its
+    /// index.
+    fn insert(&mut self, entry: CachedEntry) -> usize {
+        let prefix = entry.credential.prefix().to_owned();
+        self.entries
+            .retain(|cached| cached.credential.prefix() != prefix);
+        self.entries.push(entry);
+        self.entries.len() - 1
+    }
+
+    fn record_success(&mut self) {
+        self.consecutive_failures = 0;
+        self.retry_not_before = None;
+        self.last_failure = None;
+    }
+
+    fn record_failure(&mut self, error: &Error) {
+        self.consecutive_failures = self.consecutive_failures.saturating_add(1);
+        self.retry_not_before =
+            Instant::now().checked_add(failure_backoff(self.consecutive_failures));
+        self.last_failure = Some(error.to_string());
+    }
+
+    /// Replace cached credentials with the fetched ones, prefix by prefix,
+    /// and record the prefixes whose credential was invalid in this response.
+    ///
+    /// An unexpired cached credential survives when the response carries no
+    /// valid replacement for its prefix, so an absent or malformed entry never
+    /// evicts a usable credential. Returns the prefixes the response replaced.
+    fn merge(&mut self, fetched: Vec<CachedEntry>, errors: &[CredentialError]) -> HashSet<String> {
+        let fetched_prefixes = fetched
+            .iter()
+            .map(|entry| entry.credential.prefix().to_owned())
+            .collect::<HashSet<_>>();
+        // A response names every prefix of the table, so a prefix it omits is
+        // no longer scoped separately: a fetched credential that covers it
+        // replaces its cached one. A prefix whose credential is invalid keeps
+        // its cached one.
+        let failed_prefixes = errors
+            .iter()
+            .map(|error| error.prefix.clone())
+            .collect::<HashSet<_>>();
+        self.entries.retain(|entry| {
+            let prefix = entry.credential.prefix();
+            failed_prefixes.contains(prefix)
+                || !fetched
+                    .iter()
+                    .any(|fetched| fetched.credential.covers(prefix))
+        });
+        self.entries.extend(fetched);
+        self.failed_prefixes = failed_prefixes;
+        fetched_prefixes
+    }
+}
+
+struct ConfiguredCloud {
+    cloud: &'static CloudRefresh,
+    endpoint: String,
+    cache: Mutex<CacheState>,
+    /// Initial property credentials whose scope cannot be represented by a
+    /// single leading URI prefix, keyed according to the cloud strategy.
+    keyed_seeds: HashMap<String, CachedEntry>,
+    /// Only one caller fetches at a time. The cache lock is deliberately
+    /// separate so other callers can keep using an unexpired credential while
+    /// the refresh is in flight.
+    refresh: Mutex<()>,
+}
+
+impl ConfiguredCloud {
+    fn new(
+        cloud: &'static CloudRefresh,
+        endpoint: String,
+        entries: Vec<CachedEntry>,
+        keyed_seeds: HashMap<String, CachedEntry>,
+    ) -> Self {
+        Self {
+            cloud,
+            endpoint,
+            cache: Mutex::new(CacheState::new(entries)),
+            keyed_seeds,
+            refresh: Mutex::new(()),
+        }
+    }
+
+    /// Configure `cloud` from the FileIO properties, or `None` when they
+    /// advertise no enabled refresh endpoint for it.
+    fn configure(
+        cloud: &'static CloudRefresh,
+        base_uri: &str,
+        props: &HashMap<String, String>,
+    ) -> Option<Self> {
+        let enabled = props
+            .get(cloud.enabled_key)
+            .is_none_or(|value| value.eq_ignore_ascii_case("true"))
+            && cloud
+                .required_key
+                .is_none_or(|key| props.get(key).is_some_and(|value| !value.is_empty()));
+        let endpoint = props
+            .get(cloud.endpoint_key)
+            .filter(|endpoint| enabled && !endpoint.is_empty())
+            .map(|endpoint| resolve_endpoint(base_uri, endpoint))?;
+
+        let mut entries = Vec::new();
+        let mut keyed_seeds = HashMap::new();
+        match &cloud.seed_strategy {
+            SeedStrategy::Flat { prefixes } => {
+                entries.extend(prefixes.iter().filter_map(|prefix| {
+                    (cloud.parse_credential)(props, prefix.to_string())
+                        .ok()
+                        .map(|vended| CachedEntry::seed(vended, cloud.jitter_prefetch))
+                }));
+            }
+            SeedStrategy::Keyed(strategy) => {
+                keyed_seeds = (strategy.parse_credentials)(props)
+                    .into_iter()
+                    .map(|(key, credential)| {
+                        (key, CachedEntry::seed(credential, cloud.jitter_prefetch))
+                    })
+                    .collect();
+            }
+        }
+        Some(Self::new(cloud, endpoint, entries, keyed_seeds))
+    }
+
+    fn keyed_seed_for_path(&self, path: &str) -> Result<Option<CachedEntry>> {
+        let SeedStrategy::Keyed(strategy) = &self.cloud.seed_strategy else {
+            return Ok(None);
+        };
+        let resolved = (strategy.resolve_path)(path)?;
+        let Some(seed) = resolved
+            .keys
+            .iter()
+            .find_map(|key| self.keyed_seeds.get(key))
+        else {
+            return Ok(None);
+        };
+        let credential = StorageCredential::new(resolved.scope, seed.credential.config().clone());
+        // The scope is normalized, e.g. with a lowercase scheme, so it may not
+        // cover the path as written; the path then fetches its credential.
+        Ok(credential.covers(path).then_some(CachedEntry {
+            credential,
+            expires_at: seed.expires_at,
+            refresh_at: seed.refresh_at,
+        }))
+    }
+}
+
+/// Serializable recipe for rebuilding a [`RestVendedCredentialProvider`] from
+/// the FileIO properties, mirroring how Java rebuilds its provider on workers.
+#[derive(Clone, Serialize, Deserialize)]
+pub(crate) struct RestVendedCredentialProviderFactory {
+    /// Catalog URI, used to resolve relative refresh endpoints.
+    catalog_uri: String,
+    table: TableIdent,
+    /// The `rest.auth.type` the catalog resolved, so a rebuilt provider uses
+    /// the same auth manager even when the merged FileIO properties would
+    /// infer another one.
+    auth_type: String,
+    /// Whether `auth_type` was inferred rather than configured. Like Java,
+    /// a table `token` then authenticates a catalog without auth.
+    auth_type_inferred: bool,
+    /// The auth-related part of the unmerged config returned by the table
+    /// endpoint, from which the auth manager derives a table session. Keeping
+    /// it separate prevents local FileIO overrides from masking table auth.
+    table_auth: HashMap<String, String>,
+}
+
+/// Table config keys that may override authentication in a table session.
+const TABLE_AUTH_KEYS: &[&str] = &["token"];
+
+impl RestVendedCredentialProviderFactory {
+    pub(crate) fn new(
+        catalog_uri: impl Into<String>,
+        table: TableIdent,
+        auth_type: impl Into<String>,
+        auth_type_inferred: bool,
+        table_config: &HashMap<String, String>,
+    ) -> Self {
+        Self {
+            catalog_uri: catalog_uri.into(),
+            table,
+            auth_type: auth_type.into(),
+            auth_type_inferred,
+            table_auth: table_config
+                .iter()
+                .filter(|(key, _)| TABLE_AUTH_KEYS.contains(&key.as_str()))
+                .map(|(key, value)| (key.clone(), value.clone()))
+                .collect(),
+        }
+    }
+
+    /// The table token that authenticates credential requests of a catalog
+    /// without auth. Java's credential providers infer OAuth2 from a `token`
+    /// unless `rest.auth.type` is configured.
+    fn bearer_table_token(&self) -> Option<&str> {
+        (self.auth_type == AUTH_TYPE_NONE && self.auth_type_inferred)
+            .then(|| self.table_auth.get("token").map(String::as_str))
+            .flatten()
+    }
+
+    /// Connect to the catalog from the FileIO properties, as the catalog would.
+    async fn connect(&self, props: &HashMap<String, String>) -> Result<HttpClient> {
+        let mut config_props = props.clone();
+        config_props.insert(
+            REST_CATALOG_PROP_AUTH_TYPE.to_string(),
+            self.auth_type.clone(),
+        );
+        let config = RestCatalogConfig::builder()
+            .uri(self.catalog_uri.clone())
+            .props(config_props)
+            .build();
+        let auth_manager = load_auth_manager(&config)?;
+        let client = HttpClient::new(&config)?;
+        let session = auth_manager
+            .catalog_session(&client.without_auth_session(), &config.auth_props())
+            .await?;
+        table_client(
+            &client.with_auth_session(session),
+            auth_manager.as_ref(),
+            &self.table,
+            props,
+            &self.table_auth,
+            self.bearer_table_token(),
+        )
+        .await
+    }
+}
+
+impl std::fmt::Debug for RestVendedCredentialProviderFactory {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("RestVendedCredentialProviderFactory")
+            .field("catalog_uri", &self.catalog_uri)
+            .field("table", &self.table)
+            .finish_non_exhaustive()
+    }
+}
+
+#[typetag::serde(name = "RestVendedCredentialProviderFactory")]
+impl StorageCredentialProviderFactory for RestVendedCredentialProviderFactory {
+    fn build(&self, config: &StorageConfig) -> Result<Arc<dyn StorageCredentialProvider>> {
+        let provider =
+            RestVendedCredentialProvider::configure(self, config.props()).ok_or_else(|| {
+                Error::new(
+                    ErrorKind::DataInvalid,
+                    "FileIO configuration no longer configures vended credentials",
+                )
+            })?;
+        Ok(Arc::new(RestVendedCredentialProvider {
+            factory: Some(self.clone()),
+            ..provider
+        }))
+    }
+}
+
+/// Derive the table-scoped client used for credential requests.
+///
+/// A `bearer_table_token` authenticates instead of a session from the auth
+/// manager.
+async fn table_client(
+    catalog_client: &HttpClient,
+    auth_manager: &dyn AuthManager,
+    table: &TableIdent,
+    props: &HashMap<String, String>,
+    table_config: &HashMap<String, String>,
+    bearer_table_token: Option<&str>,
+) -> Result<HttpClient> {
+    let session = match bearer_table_token {
+        Some(token) => static_token_session(token),
+        None => {
+            auth_manager
+                .table_session(
+                    &catalog_client.without_auth_session(),
+                    table,
+                    table_config,
+                    catalog_client.auth_session(),
+                )
+                .await?
+        }
+    };
+    catalog_client.for_table(props, session)
+}
+
+/// Serves vended credentials for a table, refreshing them from the REST
+/// catalog's table credentials endpoint.
+///
+/// Each cloud cache is seeded with the credentials from the initial
+/// table properties (when complete) and re-fetched from its endpoint as they
+/// near expiry.
+pub(crate) struct RestVendedCredentialProvider {
+    /// Table-scoped catalog client. Set when the catalog builds the provider,
+    /// and connected on first refresh after deserialization.
+    client: OnceCell<HttpClient>,
+    /// Rebuilds this provider in another process. `None` when the catalog
+    /// authentication cannot be rebuilt from properties.
+    factory: Option<RestVendedCredentialProviderFactory>,
+    /// Effective FileIO properties, which supply catalog connection settings.
+    props: HashMap<String, String>,
+    /// Query parameters sent with every credentials request.
+    query_params: Vec<(&'static str, String)>,
+    /// Independently configured endpoint and cache for each backing cloud.
+    clouds: Vec<ConfiguredCloud>,
+}
+
+impl RestVendedCredentialProvider {
+    /// Configure a provider without a catalog connection, or `None` when no
+    /// supported cloud advertises an enabled refresh endpoint.
+    fn configure(
+        factory: &RestVendedCredentialProviderFactory,
+        props: &HashMap<String, String>,
+    ) -> Option<Self> {
+        let clouds = CloudRefresh::SUPPORTED
+            .iter()
+            .filter_map(|cloud| ConfiguredCloud::configure(cloud, &factory.catalog_uri, props))
+            .collect::<Vec<_>>();
+        (!clouds.is_empty()).then(|| Self {
+            client: OnceCell::new(),
+            factory: None,
+            props: props.clone(),
+            query_params: credentials_query_params(props),
+            clouds,
+        })
+    }
+
+    fn configured_cloud_for_location(&self, location: &str) -> Option<&ConfiguredCloud> {
+        self.clouds
+            .iter()
+            .find(|configured| configured.cloud.matches_location(location))
+    }
+
+    async fn client(&self) -> Result<&HttpClient> {
+        self.client
+            .get_or_try_init(|| async {
+                let factory = self.factory.as_ref().ok_or_else(|| {
+                    Error::new(
+                        ErrorKind::Unexpected,
+                        "vended credential provider has no catalog connection",
+                    )
+                })?;
+                factory.connect(&self.props).await
+            })
+            .await
+    }
+
+    /// Fetch fresh credentials from the catalog's credentials endpoint.
+    async fn fetch(&self, configured: &ConfiguredCloud) -> Result<ParsedCredentials> {
+        let cloud = configured.cloud;
+        let client = self.client().await?;
+        let url = credentials_url(&configured.endpoint, &self.query_params)?;
+        let request = HttpRequest::build(client.request(Method::GET, url))?;
+        let response = client.query_catalog(request).await?;

Review Comment:
   `fetch` has no timeout of its own, and as far as I can see the shared client doesn't set one either. Say the catalog accepts the connection and then stalls. The caller holding the single-flight guard (line 893) never returns. Every caller without a usable credential queues on `refresh.lock().await` behind it. No failure is ever recorded, so backoff never starts. Callers with a cached credential keep working until it expires, then they hang too, which is the long-running-writer case this PR is meant to fix.
   
   I'd wrap the fetch in `tokio::time::timeout`, record a timeout as a refresh failure, and add a test against an endpoint that never answers.



##########
crates/storage/opendal/src/utils.rs:
##########
@@ -27,3 +28,331 @@ pub(crate) fn from_opendal_error(e: opendal::Error) -> iceberg::Error {
     )
     .with_source(e)
 }
+
+/// The non-empty value of `key` in a vended credential's config.
+#[cfg(any(
+    feature = "opendal-s3",
+    feature = "opendal-gcs",
+    feature = "opendal-azdls"
+))]
+pub(crate) fn required_credential_property<'a>(
+    config: &'a std::collections::HashMap<String, String>,
+    key: &str,
+) -> reqsign_core::Result<&'a str> {
+    config
+        .get(key)
+        .map(String::as_str)
+        .filter(|value| !value.is_empty())
+        .ok_or_else(|| {
+            reqsign_core::Error::unexpected(format!("vended credential is missing {key}"))
+        })
+}
+
+/// The latest epoch millisecond `reqsign` timestamps can represent,
+/// `9999-12-30T22:00:00Z`.
+#[cfg(any(
+    feature = "opendal-s3",
+    feature = "opendal-gcs",
+    feature = "opendal-azdls"
+))]
+const MAX_TIMESTAMP_MILLIS: i64 = 253_402_207_200_000;
+
+/// The epoch-millisecond expiry under `key` in a vended credential's config,
+/// if present, as the `reqsign` timestamp backend credentials use. Later
+/// expiries, such as `Long.MAX_VALUE` for a credential that never expires,
+/// are clamped to the latest representable timestamp.
+#[cfg(any(
+    feature = "opendal-s3",
+    feature = "opendal-gcs",
+    feature = "opendal-azdls"
+))]
+pub(crate) fn credential_expiry(
+    config: &std::collections::HashMap<String, String>,
+    key: &str,
+) -> reqsign_core::Result<Option<reqsign_core::time::Timestamp>> {
+    config
+        .get(key)
+        .filter(|value| !value.is_empty())
+        .map(|value| {
+            value
+                .parse::<i64>()
+                .ok()
+                .and_then(|millis| {
+                    reqsign_core::time::Timestamp::from_millisecond(
+                        millis.min(MAX_TIMESTAMP_MILLIS),
+                    )
+                    .ok()
+                })
+                .ok_or_else(|| {
+                    reqsign_core::Error::unexpected(format!(
+                        "vended credential has an invalid {key}"
+                    ))
+                })
+        })
+        .transpose()
+}
+
+/// Validate that a provider's credential covers the location for which the
+/// backend requested it.
+#[cfg(any(
+    feature = "opendal-s3",
+    feature = "opendal-gcs",
+    feature = "opendal-azdls"
+))]
+pub(crate) fn validate_credential_prefix(
+    location: &str,
+    credential: &iceberg::io::StorageCredential,
+) -> reqsign_core::Result<()> {
+    if credential.covers(location) {
+        Ok(())
+    } else {
+        Err(reqsign_core::Error::unexpected(uncovered_location_message(
+            location, credential,
+        )))
+    }
+}
+
+/// The error message for a vended credential that does not cover `location`.
+pub(crate) fn uncovered_location_message(
+    location: &str,
+    credential: &iceberg::io::StorageCredential,
+) -> String {
+    format!(
+        "vended credential prefix {:?} does not cover storage location {location:?}",
+        credential.prefix()
+    )
+}
+
+/// The root of the storage location containing `path`, e.g. `s3://bucket/`.
+pub(crate) fn storage_root(path: &str) -> iceberg::Result<String> {
+    let url = url::Url::parse(path)?;
+    Ok(format!("{}://{}/", url.scheme(), url.authority()))
+}
+
+/// Loads vended credentials for one storage location on behalf of a backend's
+/// `reqsign` credential provider.
+#[cfg(any(
+    feature = "opendal-s3",
+    feature = "opendal-gcs",
+    feature = "opendal-azdls"
+))]
+pub(crate) struct VendedCredentialSource {
+    provider: std::sync::Arc<dyn iceberg::io::StorageCredentialProvider>,
+    /// Location handed to the provider: the path an operator serves, or the
+    /// scope shared by a bulk-delete batch.
+    location: String,
+}
+
+#[cfg(any(
+    feature = "opendal-s3",
+    feature = "opendal-gcs",
+    feature = "opendal-azdls"
+))]
+impl VendedCredentialSource {
+    pub(crate) fn new(
+        provider: std::sync::Arc<dyn iceberg::io::StorageCredentialProvider>,
+        location: String,
+    ) -> Self {
+        Self { provider, location }
+    }
+
+    /// Load a credential covering the location and convert its config with
+    /// `extract` into the backend's credential.
+    ///
+    /// Errors are logged here: reqsign's credential chain replaces them with an
+    /// error that only says no credential could be loaded, and reports the
+    /// cause only through the `log` crate.
+    pub(crate) async fn load<T>(
+        &self,
+        backend: &str,
+        extract: impl FnOnce(&std::collections::HashMap<String, String>) -> reqsign_core::Result<T>,
+    ) -> reqsign_core::Result<T> {
+        self.try_load(backend, extract).await.inspect_err(|error| {
+            tracing::warn!(

Review Comment:
   While refresh is backed off, `load_credential` fails immediately and every object-store request ends up here, so a busy scan logs thousands of identical warnings per second. I'd move the `warn!` into the provider's `record_failure`, which fires once per failed refresh, and drop this one to `debug`.



##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,179 @@ pub trait StorageFactory: Debug + Send + Sync {
     /// A `Result` containing an `Arc<dyn Storage>` on success, or an error
     /// if the storage could not be created.
     fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>>;
+
+    /// Build a new Storage instance, optionally supplying a credential provider
+    /// that the backend can call to obtain and refresh short-lived credentials.
+    ///
+    /// Backends that cannot use the provider ignore it and use the credentials
+    /// in `config`, as they would without one. The default does exactly that.
+    #[allow(unused_variables)]
+    fn build_with_credential_provider(

Review Comment:
   Ignoring the provider by default is the right call, since it removes the wiring-order hazard @mbutrovich described. The cost is that a third-party `StorageFactory` keeps the seeded credentials and silently loses refresh, which shows up as a 403 when they expire. I'd `tracing::warn!` here when `credential_provider` is `Some`; this runs once per storage build, not per request. I'd also rename it to `_credential_provider` instead of using `#[allow(unused_variables)]`.



##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,3675 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Vended storage credentials from a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache. Like
+//! Java's `StorageCredential`, the returned [`StorageCredential`] holds the
+//! backend's storage properties, which the storage adapter turns into its own
+//! credential. This preserves Java's per-cloud credential selection and
+//! prefetch policies while supporting mixed-cloud tables through a resolving
+//! FileIO. Unlike Java's scheduled refresh, which permanently stops
+//! after a failed fetch, transient failures are retried here with jittered
+//! exponential backoff while an unexpired credential remains available.
+//!
+//! Like Java's `VendedCredentialsProvider`, a provider is rebuilt from the
+//! FileIO properties after [`FileIO`](iceberg::io::FileIO) serialization and
+//! connects to the catalog lazily in the receiving process. This requires
+//! catalog authentication that can be rebuilt from `rest.auth.type`.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant. To
+//! add a backend, teach its storage adapter to read its credential properties
+//! from a [`StorageCredential`]. Then write its `parse_*` function, add a
+//! `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+    ADLS_REFRESH_CREDENTIALS_ENABLED, ADLS_REFRESH_CREDENTIALS_ENDPOINT,
+    ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX, ADLS_SAS_TOKEN_PREFIX, AWS_REFRESH_CREDENTIALS_ENABLED,
+    AWS_REFRESH_CREDENTIALS_ENDPOINT, GCS_REFRESH_CREDENTIALS_ENABLED,
+    GCS_REFRESH_CREDENTIALS_ENDPOINT, GCS_TOKEN, GCS_TOKEN_EXPIRES_AT, S3_ACCESS_KEY_ID,
+    S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN, S3_SESSION_TOKEN_EXPIRES_AT_MS, StorageConfig,
+    StorageCredential, StorageCredentialProvider, StorageCredentialProviderFactory,
+};
+use iceberg::{Error, ErrorKind, Result, TableIdent};
+use rand::Rng;
+use reqwest::{Method, Url};
+use serde::{Deserialize, Serialize};
+use tokio::sync::{Mutex, OnceCell};
+
+use crate::auth::{AUTH_TYPE_NONE, AuthManager, load_auth_manager, static_token_session};
+use crate::catalog::{
+    REST_CATALOG_PROP_AUTH_TYPE, REST_CATALOG_PROP_REFERENCED_BY, REST_CATALOG_PROP_SCAN_PLAN_ID,
+    RestCatalogConfig,
+};
+use crate::client::{HttpClient, unexpected_catalog_error_without_body};
+use crate::request::HttpRequest;
+use crate::types::LoadCredentialsResponse;
+
+type CredentialParser =
+    fn(config: &HashMap<String, String>, prefix: String) -> Result<VendedCredential>;
+type KeyedSeedCredentialParser =
+    fn(config: &HashMap<String, String>) -> HashMap<String, VendedCredential>;
+type KeyedSeedPathResolver = fn(path: &str) -> Result<KeyedSeedPath>;
+
+/// A credential vended by the catalog, with the expiry its config declares.
+#[derive(Clone)]
+struct VendedCredential {
+    credential: StorageCredential,
+    expires_at: SystemTime,
+}
+
+enum SeedStrategy {
+    /// One credential stored in the backend's flat properties, for every
+    /// location with one of the scheme prefixes, like Java's root clients.
+    Flat { prefixes: &'static [&'static str] },
+    /// Credentials selected by a backend-specific key derived from each path.
+    Keyed(KeyedSeedStrategy),
+}
+
+struct KeyedSeedStrategy {
+    parse_credentials: KeyedSeedCredentialParser,
+    resolve_path: KeyedSeedPathResolver,
+}
+
+struct KeyedSeedPath {
+    /// Seed keys for the path, most specific first.
+    keys: Vec<String>,
+    scope: String,
+}
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// 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 {
+    /// Location URL schemes this backend serves.
+    schemes: &'static [&'static str],
+    /// Table property naming the refresh endpoint (absolute or catalog-relative).
+    endpoint_key: &'static str,
+    /// Table property controlling refresh; only missing or case-insensitive `"true"` enables it.
+    enabled_key: &'static str,
+    /// Table property that must be present for refresh to apply, as in Java,
+    /// where it decides whether the backend uses vended credentials at all.
+    required_key: Option<&'static str>,
+    /// Whether to jitter successful prefetch times like AWS `CachedSupplier`.
+    jitter_prefetch: bool,
+    /// Parse a complete credential from catalog-supplied properties.
+    parse_credential: CredentialParser,
+    /// How initial credentials in the table properties are selected.
+    seed_strategy: SeedStrategy,
+}
+
+impl CloudRefresh {
+    /// S3 / AWS
+    const AWS: Self = Self {
+        schemes: &["s3", "s3a", "s3n"],
+        endpoint_key: AWS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: AWS_REFRESH_CREDENTIALS_ENABLED,
+        required_key: None,
+        jitter_prefetch: true,
+        parse_credential: parse_s3_credential,
+        // `s3` also covers `s3a` and `s3n` locations.
+        seed_strategy: SeedStrategy::Flat { prefixes: &["s3"] },
+    };
+    /// Google Cloud Storage
+    const GCP: Self = Self {
+        schemes: &["gs", "gcs"],
+        endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED,
+        // Java's GCS client only refreshes a vended OAuth2 token; without
+        // one it uses Google's default credentials.
+        required_key: Some(GCS_TOKEN),
+        jitter_prefetch: false,
+        parse_credential: parse_gcs_credential,
+        seed_strategy: SeedStrategy::Flat {
+            prefixes: &["gs", "gcs"],
+        },
+    };
+    /// Azure Data Lake Storage
+    const AZURE: Self = Self {
+        schemes: &["abfs", "abfss", "wasb", "wasbs"],
+        endpoint_key: ADLS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: ADLS_REFRESH_CREDENTIALS_ENABLED,
+        required_key: None,
+        jitter_prefetch: false,
+        parse_credential: parse_azdls_credential,
+        seed_strategy: SeedStrategy::Keyed(KeyedSeedStrategy {
+            parse_credentials: parse_azdls_account_seeds,
+            resolve_path: resolve_azdls_seed_path,
+        }),
+    };
+
+    /// Backends with refresh support
+    const SUPPORTED: &[Self] = &[Self::AWS, Self::GCP, Self::AZURE];
+
+    fn matches_location(&self, location: &str) -> bool {
+        self.schemes
+            .iter()
+            .any(|scheme| location.eq_ignore_ascii_case(scheme))
+            || scheme_of(location).is_some_and(|scheme| self.schemes.contains(&scheme.as_str()))
+    }
+}
+
+/// Re-fetch a credential once it is within this window of expiry, so a fresh
+/// token is in hand before the object store would reject the old one.
+const REFRESH_BUFFER: Duration = Duration::from_mins(5);
+
+/// AWS keeps at least one minute between its jittered prefetch time and expiry.
+const MIN_REFRESH_BUFFER: Duration = Duration::from_mins(1);
+
+/// Initial ceiling for failure backoff. Equal jitter chooses from half this
+/// value through the full value.
+const INITIAL_FAILURE_BACKOFF: Duration = Duration::from_secs(1);
+
+/// Maximum delay between failed refresh attempts.
+const MAX_FAILURE_BACKOFF: Duration = Duration::from_secs(30);
+
+/// A cached vended credential and its refresh schedule.
+#[derive(Clone)]
+struct CachedEntry {
+    credential: StorageCredential,
+    expires_at: SystemTime,
+    /// When this entry becomes eligible for prefetch.
+    refresh_at: SystemTime,
+}
+
+impl CachedEntry {
+    fn new(vended: VendedCredential, jitter_prefetch: bool) -> Self {
+        let VendedCredential {
+            credential,
+            expires_at,
+        } = vended;
+        Self {
+            credential,
+            expires_at,
+            refresh_at: prefetch_time(SystemTime::now(), expires_at, jitter_prefetch),
+        }
+    }
+
+    /// Seed entries that are already inside the nominal five-minute window are
+    /// immediately due. Otherwise AWS applies the same jitter as it does to a
+    /// freshly fetched value.
+    fn seed(vended: VendedCredential, jitter_prefetch: bool) -> Self {
+        let due = SystemTime::now()
+            .checked_add(REFRESH_BUFFER)
+            .is_none_or(|refresh_boundary| refresh_boundary >= vended.expires_at);
+        let mut entry = Self::new(vended, jitter_prefetch);
+        if due {
+            entry.refresh_at = UNIX_EPOCH;
+        }
+        entry
+    }
+
+    fn is_fresh(&self, now: SystemTime) -> bool {
+        now < self.refresh_at
+    }
+
+    fn is_unexpired(&self, now: SystemTime) -> bool {
+        now < self.expires_at
+    }
+}
+
+/// Parsed entries from one successful credentials response.
+struct ParsedCredentials {
+    entries: Vec<CachedEntry>,
+    errors: Vec<CredentialError>,
+}
+
+struct CredentialError {
+    prefix: String,
+    error: Error,
+}
+
+/// Cached credentials plus failure-backoff state.
+///
+/// A fetch returns the credentials for every prefix of the table, so a path
+/// the last response did not cover would not be covered by an immediate
+/// re-fetch either. Backoff is therefore shared by the whole cloud.
+struct CacheState {
+    entries: Vec<CachedEntry>,
+    /// Prefixes whose credential in the last successful response was invalid.
+    /// Paths under them must not use a broader credential: the catalog scoped
+    /// them more tightly.
+    failed_prefixes: HashSet<String>,
+    consecutive_failures: u32,
+    retry_not_before: Option<Instant>,
+    /// Why the last refresh failed, reported while refresh is backed off.
+    last_failure: Option<String>,
+}
+
+impl CacheState {
+    fn new(entries: Vec<CachedEntry>) -> Self {
+        Self {
+            entries,
+            failed_prefixes: HashSet::new(),
+            consecutive_failures: 0,
+            retry_not_before: None,
+            last_failure: None,
+        }
+    }
+
+    /// The index of the unexpired credential with the longest prefix covering
+    /// `path`.
+    fn unexpired_match(&self, path: &str, now: SystemTime) -> Option<usize> {
+        self.entries
+            .iter()
+            .enumerate()
+            .filter(|(_, entry)| entry.is_unexpired(now) && entry.credential.covers(path))
+            .max_by_key(|(_, entry)| entry.credential.prefix().len())
+            .map(|(index, _)| index)
+    }
+
+    /// Whether a failed prefix covering `path` is narrower than `credential`'s.
+    fn masks(&self, path: &str, credential: Option<&StorageCredential>) -> bool {
+        let prefix_len = credential.map(|credential| credential.prefix().len());
+        self.failed_prefixes.iter().any(|failed| {
+            prefix_covers(failed, path) && prefix_len.is_none_or(|len| failed.len() > len)
+        })
+    }
+
+    /// Cache `entry`, replacing any entry with the same prefix, and return its
+    /// index.
+    fn insert(&mut self, entry: CachedEntry) -> usize {
+        let prefix = entry.credential.prefix().to_owned();
+        self.entries
+            .retain(|cached| cached.credential.prefix() != prefix);
+        self.entries.push(entry);
+        self.entries.len() - 1
+    }
+
+    fn record_success(&mut self) {
+        self.consecutive_failures = 0;
+        self.retry_not_before = None;
+        self.last_failure = None;
+    }
+
+    fn record_failure(&mut self, error: &Error) {
+        self.consecutive_failures = self.consecutive_failures.saturating_add(1);
+        self.retry_not_before =
+            Instant::now().checked_add(failure_backoff(self.consecutive_failures));
+        self.last_failure = Some(error.to_string());
+    }
+
+    /// Replace cached credentials with the fetched ones, prefix by prefix,
+    /// and record the prefixes whose credential was invalid in this response.
+    ///
+    /// An unexpired cached credential survives when the response carries no
+    /// valid replacement for its prefix, so an absent or malformed entry never
+    /// evicts a usable credential. Returns the prefixes the response replaced.
+    fn merge(&mut self, fetched: Vec<CachedEntry>, errors: &[CredentialError]) -> HashSet<String> {
+        let fetched_prefixes = fetched
+            .iter()
+            .map(|entry| entry.credential.prefix().to_owned())
+            .collect::<HashSet<_>>();
+        // A response names every prefix of the table, so a prefix it omits is
+        // no longer scoped separately: a fetched credential that covers it
+        // replaces its cached one. A prefix whose credential is invalid keeps
+        // its cached one.
+        let failed_prefixes = errors
+            .iter()
+            .map(|error| error.prefix.clone())
+            .collect::<HashSet<_>>();
+        self.entries.retain(|entry| {
+            let prefix = entry.credential.prefix();
+            failed_prefixes.contains(prefix)
+                || !fetched
+                    .iter()
+                    .any(|fetched| fetched.credential.covers(prefix))
+        });
+        self.entries.extend(fetched);
+        self.failed_prefixes = failed_prefixes;
+        fetched_prefixes
+    }
+}
+
+struct ConfiguredCloud {
+    cloud: &'static CloudRefresh,
+    endpoint: String,
+    cache: Mutex<CacheState>,
+    /// Initial property credentials whose scope cannot be represented by a
+    /// single leading URI prefix, keyed according to the cloud strategy.
+    keyed_seeds: HashMap<String, CachedEntry>,
+    /// Only one caller fetches at a time. The cache lock is deliberately
+    /// separate so other callers can keep using an unexpired credential while
+    /// the refresh is in flight.
+    refresh: Mutex<()>,
+}
+
+impl ConfiguredCloud {
+    fn new(
+        cloud: &'static CloudRefresh,
+        endpoint: String,
+        entries: Vec<CachedEntry>,
+        keyed_seeds: HashMap<String, CachedEntry>,
+    ) -> Self {
+        Self {
+            cloud,
+            endpoint,
+            cache: Mutex::new(CacheState::new(entries)),
+            keyed_seeds,
+            refresh: Mutex::new(()),
+        }
+    }
+
+    /// Configure `cloud` from the FileIO properties, or `None` when they
+    /// advertise no enabled refresh endpoint for it.
+    fn configure(
+        cloud: &'static CloudRefresh,
+        base_uri: &str,
+        props: &HashMap<String, String>,
+    ) -> Option<Self> {
+        let enabled = props
+            .get(cloud.enabled_key)
+            .is_none_or(|value| value.eq_ignore_ascii_case("true"))
+            && cloud
+                .required_key
+                .is_none_or(|key| props.get(key).is_some_and(|value| !value.is_empty()));
+        let endpoint = props
+            .get(cloud.endpoint_key)
+            .filter(|endpoint| enabled && !endpoint.is_empty())
+            .map(|endpoint| resolve_endpoint(base_uri, endpoint))?;
+
+        let mut entries = Vec::new();
+        let mut keyed_seeds = HashMap::new();
+        match &cloud.seed_strategy {
+            SeedStrategy::Flat { prefixes } => {
+                entries.extend(prefixes.iter().filter_map(|prefix| {
+                    (cloud.parse_credential)(props, prefix.to_string())
+                        .ok()
+                        .map(|vended| CachedEntry::seed(vended, cloud.jitter_prefetch))
+                }));
+            }
+            SeedStrategy::Keyed(strategy) => {
+                keyed_seeds = (strategy.parse_credentials)(props)
+                    .into_iter()
+                    .map(|(key, credential)| {
+                        (key, CachedEntry::seed(credential, cloud.jitter_prefetch))
+                    })
+                    .collect();
+            }
+        }
+        Some(Self::new(cloud, endpoint, entries, keyed_seeds))
+    }
+
+    fn keyed_seed_for_path(&self, path: &str) -> Result<Option<CachedEntry>> {
+        let SeedStrategy::Keyed(strategy) = &self.cloud.seed_strategy else {
+            return Ok(None);
+        };
+        let resolved = (strategy.resolve_path)(path)?;
+        let Some(seed) = resolved
+            .keys
+            .iter()
+            .find_map(|key| self.keyed_seeds.get(key))
+        else {
+            return Ok(None);
+        };
+        let credential = StorageCredential::new(resolved.scope, seed.credential.config().clone());
+        // The scope is normalized, e.g. with a lowercase scheme, so it may not
+        // cover the path as written; the path then fetches its credential.
+        Ok(credential.covers(path).then_some(CachedEntry {
+            credential,
+            expires_at: seed.expires_at,
+            refresh_at: seed.refresh_at,
+        }))
+    }
+}
+
+/// Serializable recipe for rebuilding a [`RestVendedCredentialProvider`] from
+/// the FileIO properties, mirroring how Java rebuilds its provider on workers.
+#[derive(Clone, Serialize, Deserialize)]
+pub(crate) struct RestVendedCredentialProviderFactory {
+    /// Catalog URI, used to resolve relative refresh endpoints.
+    catalog_uri: String,
+    table: TableIdent,
+    /// The `rest.auth.type` the catalog resolved, so a rebuilt provider uses
+    /// the same auth manager even when the merged FileIO properties would
+    /// infer another one.
+    auth_type: String,
+    /// Whether `auth_type` was inferred rather than configured. Like Java,
+    /// a table `token` then authenticates a catalog without auth.
+    auth_type_inferred: bool,
+    /// The auth-related part of the unmerged config returned by the table
+    /// endpoint, from which the auth manager derives a table session. Keeping
+    /// it separate prevents local FileIO overrides from masking table auth.
+    table_auth: HashMap<String, String>,
+}
+
+/// Table config keys that may override authentication in a table session.
+const TABLE_AUTH_KEYS: &[&str] = &["token"];
+
+impl RestVendedCredentialProviderFactory {
+    pub(crate) fn new(
+        catalog_uri: impl Into<String>,
+        table: TableIdent,
+        auth_type: impl Into<String>,
+        auth_type_inferred: bool,
+        table_config: &HashMap<String, String>,
+    ) -> Self {
+        Self {
+            catalog_uri: catalog_uri.into(),
+            table,
+            auth_type: auth_type.into(),
+            auth_type_inferred,
+            table_auth: table_config
+                .iter()
+                .filter(|(key, _)| TABLE_AUTH_KEYS.contains(&key.as_str()))
+                .map(|(key, value)| (key.clone(), value.clone()))
+                .collect(),
+        }
+    }
+
+    /// The table token that authenticates credential requests of a catalog
+    /// without auth. Java's credential providers infer OAuth2 from a `token`
+    /// unless `rest.auth.type` is configured.
+    fn bearer_table_token(&self) -> Option<&str> {
+        (self.auth_type == AUTH_TYPE_NONE && self.auth_type_inferred)
+            .then(|| self.table_auth.get("token").map(String::as_str))
+            .flatten()
+    }
+
+    /// Connect to the catalog from the FileIO properties, as the catalog would.
+    async fn connect(&self, props: &HashMap<String, String>) -> Result<HttpClient> {
+        let mut config_props = props.clone();
+        config_props.insert(
+            REST_CATALOG_PROP_AUTH_TYPE.to_string(),
+            self.auth_type.clone(),
+        );
+        let config = RestCatalogConfig::builder()
+            .uri(self.catalog_uri.clone())
+            .props(config_props)
+            .build();
+        let auth_manager = load_auth_manager(&config)?;
+        let client = HttpClient::new(&config)?;
+        let session = auth_manager
+            .catalog_session(&client.without_auth_session(), &config.auth_props())
+            .await?;
+        table_client(
+            &client.with_auth_session(session),
+            auth_manager.as_ref(),
+            &self.table,
+            props,
+            &self.table_auth,
+            self.bearer_table_token(),
+        )
+        .await
+    }
+}
+
+impl std::fmt::Debug for RestVendedCredentialProviderFactory {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("RestVendedCredentialProviderFactory")
+            .field("catalog_uri", &self.catalog_uri)
+            .field("table", &self.table)
+            .finish_non_exhaustive()
+    }
+}
+
+#[typetag::serde(name = "RestVendedCredentialProviderFactory")]
+impl StorageCredentialProviderFactory for RestVendedCredentialProviderFactory {
+    fn build(&self, config: &StorageConfig) -> Result<Arc<dyn StorageCredentialProvider>> {
+        let provider =
+            RestVendedCredentialProvider::configure(self, config.props()).ok_or_else(|| {
+                Error::new(
+                    ErrorKind::DataInvalid,
+                    "FileIO configuration no longer configures vended credentials",
+                )
+            })?;
+        Ok(Arc::new(RestVendedCredentialProvider {
+            factory: Some(self.clone()),
+            ..provider
+        }))
+    }
+}
+
+/// Derive the table-scoped client used for credential requests.
+///
+/// A `bearer_table_token` authenticates instead of a session from the auth
+/// manager.
+async fn table_client(
+    catalog_client: &HttpClient,
+    auth_manager: &dyn AuthManager,
+    table: &TableIdent,
+    props: &HashMap<String, String>,
+    table_config: &HashMap<String, String>,
+    bearer_table_token: Option<&str>,
+) -> Result<HttpClient> {
+    let session = match bearer_table_token {
+        Some(token) => static_token_session(token),
+        None => {
+            auth_manager
+                .table_session(
+                    &catalog_client.without_auth_session(),
+                    table,
+                    table_config,
+                    catalog_client.auth_session(),
+                )
+                .await?
+        }
+    };
+    catalog_client.for_table(props, session)
+}
+
+/// Serves vended credentials for a table, refreshing them from the REST
+/// catalog's table credentials endpoint.
+///
+/// Each cloud cache is seeded with the credentials from the initial
+/// table properties (when complete) and re-fetched from its endpoint as they
+/// near expiry.
+pub(crate) struct RestVendedCredentialProvider {
+    /// Table-scoped catalog client. Set when the catalog builds the provider,
+    /// and connected on first refresh after deserialization.
+    client: OnceCell<HttpClient>,
+    /// Rebuilds this provider in another process. `None` when the catalog
+    /// authentication cannot be rebuilt from properties.
+    factory: Option<RestVendedCredentialProviderFactory>,
+    /// Effective FileIO properties, which supply catalog connection settings.
+    props: HashMap<String, String>,
+    /// Query parameters sent with every credentials request.
+    query_params: Vec<(&'static str, String)>,
+    /// Independently configured endpoint and cache for each backing cloud.
+    clouds: Vec<ConfiguredCloud>,
+}
+
+impl RestVendedCredentialProvider {
+    /// Configure a provider without a catalog connection, or `None` when no
+    /// supported cloud advertises an enabled refresh endpoint.
+    fn configure(
+        factory: &RestVendedCredentialProviderFactory,
+        props: &HashMap<String, String>,
+    ) -> Option<Self> {
+        let clouds = CloudRefresh::SUPPORTED
+            .iter()
+            .filter_map(|cloud| ConfiguredCloud::configure(cloud, &factory.catalog_uri, props))
+            .collect::<Vec<_>>();
+        (!clouds.is_empty()).then(|| Self {
+            client: OnceCell::new(),
+            factory: None,
+            props: props.clone(),
+            query_params: credentials_query_params(props),
+            clouds,
+        })
+    }
+
+    fn configured_cloud_for_location(&self, location: &str) -> Option<&ConfiguredCloud> {
+        self.clouds
+            .iter()
+            .find(|configured| configured.cloud.matches_location(location))
+    }
+
+    async fn client(&self) -> Result<&HttpClient> {
+        self.client
+            .get_or_try_init(|| async {
+                let factory = self.factory.as_ref().ok_or_else(|| {
+                    Error::new(
+                        ErrorKind::Unexpected,
+                        "vended credential provider has no catalog connection",
+                    )
+                })?;
+                factory.connect(&self.props).await
+            })
+            .await
+    }
+
+    /// Fetch fresh credentials from the catalog's credentials endpoint.
+    async fn fetch(&self, configured: &ConfiguredCloud) -> Result<ParsedCredentials> {
+        let cloud = configured.cloud;
+        let client = self.client().await?;
+        let url = credentials_url(&configured.endpoint, &self.query_params)?;
+        let request = HttpRequest::build(client.request(Method::GET, url))?;
+        let response = client.query_catalog(request).await?;
+
+        if !response.status().is_success() {
+            return Err(unexpected_catalog_error_without_body(
+                response,
+                client.disable_header_redaction(),
+            ));
+        }
+
+        // Credential responses contain secrets, and serde's errors quote the
+        // values they reject, so only the error's category and position are
+        // reported.
+        let parsed: LoadCredentialsResponse =
+            serde_json::from_slice(response.body()).map_err(|error| {
+                Error::new(
+                    ErrorKind::Unexpected,
+                    format!(
+                        "failed to parse vended credential response: {:?} error at line {}, \
+                         column {}",
+                        error.classify(),
+                        error.line(),
+                        error.column()
+                    ),
+                )
+            })?;
+        let now = SystemTime::now();
+        let mut entries = Vec::new();
+        let mut errors = Vec::new();
+        for credential in parsed
+            .storage_credentials
+            .into_iter()
+            .filter(|credential| cloud.matches_location(&credential.prefix))
+        {
+            let prefix = credential.prefix;
+            let parsed = (cloud.parse_credential)(&credential.config, prefix.clone())
+                .map(|vended| CachedEntry::new(vended, cloud.jitter_prefetch))
+                .and_then(|entry| {
+                    if entry.is_unexpired(now) {
+                        Ok(entry)
+                    } else {
+                        Err(Error::new(
+                            ErrorKind::DataInvalid,
+                            "invalid vended credential: credential is already expired",
+                        ))
+                    }
+                });
+            match parsed {
+                Ok(entry) => entries.push(entry),
+                Err(error) => errors.push(CredentialError { prefix, error }),
+            }
+        }
+        Ok(ParsedCredentials { entries, errors })
+    }
+
+    async fn refresh_credential(
+        &self,
+        configured: &ConfiguredCloud,
+        path: &str,
+        fallback: Option<CachedEntry>,
+    ) -> Result<StorageCredential> {
+        let fetched = self.fetch(configured).await;
+        let mut cache = configured.cache.lock().await;
+        let now = SystemTime::now();
+        cache.entries.retain(|entry| entry.is_unexpired(now));
+
+        let (fetched_prefixes, failure) = match fetched {
+            Ok(ParsedCredentials { entries, errors }) => {
+                let fetched_prefixes = cache.merge(entries, &errors);
+                // The most specific error for `path`.
+                let failure = errors
+                    .into_iter()
+                    .filter(|error| prefix_covers(&error.prefix, path))
+                    .max_by_key(|error| error.prefix.len())
+                    .map(|error| error.error);
+                (Some(fetched_prefixes), failure)
+            }
+            Err(error) => (None, Some(error)),
+        };
+
+        let selected = cache.unexpired_match(path, now);
+        // A keyed seed is only cached once served as the fallback.
+        let seed = selected
+            .is_none()
+            .then(|| {
+                fallback
+                    .clone()
+                    .filter(|fallback| fallback.is_unexpired(now))
+            })
+            .flatten();
+        let candidate = selected
+            .map(|index| &cache.entries[index].credential)
+            .or(seed.as_ref().map(|seed| &seed.credential));
+        if cache.masks(path, candidate) {
+            let error = failure.unwrap_or_else(|| masked_credential_error(path));
+            cache.record_failure(&error);
+            return Err(error);
+        }
+
+        let refreshed = selected.filter(|&index| {
+            fetched_prefixes
+                .as_ref()
+                .is_some_and(|fetched| fetched.contains(cache.entries[index].credential.prefix()))
+        });
+        if let Some(index) = refreshed {
+            let entry = &mut cache.entries[index];
+            // A catalog may return the same credential until shortly before
+            // it expires. Treat one that is no newer like no answer, so the
+            // checks stay few as expiry approaches.
+            if fallback
+                .as_ref()
+                .is_some_and(|previous| entry.expires_at <= previous.expires_at)
+            {
+                entry.refresh_at = recheck_time(now, entry.expires_at);
+            }
+            cache.record_success();
+            return Ok(cache.entries[index].credential.clone());
+        }
+
+        let selected = selected.or_else(|| seed.map(|seed| cache.insert(seed)));
+        let Some(index) = selected else {
+            let error = failure.unwrap_or_else(|| {
+                Error::new(
+                    ErrorKind::Unexpected,
+                    format!("no unexpired vended credential matches storage location: {path}"),
+                )
+            });
+            cache.record_failure(&error);

Review Comment:
   Recording a cloud-wide failure here treats "this path isn't vended" the same as "the catalog is failing". Once an endpoint is configured, `supports_path` (line 862) returns true for every `s3*` location. So a read outside the table's prefixes, like an external file in another bucket, triggers a fetch, ends up here, and backs off refresh for the whole cloud for up to 30s. During that window, a covered path whose credential has just expired gets the backoff error instead of a refresh.
   
   When the fetch succeeded and nothing covers `path`, I'd return the error without touching `consecutive_failures`/`retry_not_before`. Backoff should only apply to failed fetches and failed parses.



##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,179 @@ pub trait StorageFactory: Debug + Send + Sync {
     /// A `Result` containing an `Arc<dyn Storage>` on success, or an error
     /// if the storage could not be created.
     fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>>;
+
+    /// Build a new Storage instance, optionally supplying a credential provider
+    /// that the backend can call to obtain and refresh short-lived credentials.
+    ///
+    /// Backends that cannot use the provider ignore it and use the credentials
+    /// in `config`, as they would without one. The default does exactly that.
+    #[allow(unused_variables)]
+    fn build_with_credential_provider(
+        &self,
+        config: &StorageConfig,
+        credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
+    ) -> Result<Arc<dyn Storage>> {
+        self.build(config)
+    }
+}
+
+/// Supplies fresh, backend-specific storage credentials on demand.
+///
+/// A catalog that vends temporary credentials implements this trait so that
+/// storage backends can re-fetch credentials as they approach expiry instead
+/// of failing once the initial token's TTL runs out.
+///
+/// # Debug output
+///
+/// [`FileIO`](crate::io::FileIO) and storage implementations print the
+/// provider in their `Debug` output, so the provider's `Debug` implementation
+/// must not expose credentials or other secrets.
+///
+/// # Caching
+///
+/// [`load_credential`](Self::load_credential) may be called very frequently.
+/// Implementations must cache internally and only re-fetch when the current
+/// credential is at or near expiry; otherwise every object-store request could
+/// trigger a call back to the catalog.
+#[async_trait]
+pub trait StorageCredentialProvider: Debug + Send + Sync {
+    /// Return whether this provider supplies credentials for `path`.
+    ///
+    /// Backends replace their own credential chain with the provider only for
+    /// paths it supports, so a provider must return `false` for the storage it
+    /// does not serve, such as other schemes behind a resolving storage.
+    fn supports_path(&self, path: &str) -> bool;
+
+    /// Load a fresh credential for the storage location identified by `path`.
+    ///
+    /// `path` is an absolute location: usually the file being accessed, such
+    /// as `s3://bucket/warehouse/db/table/data/file.parquet`, but for a bulk
+    /// delete the location shared by a batch, which may be a storage root
+    /// (`s3://bucket/`) or the prefix of a previously returned credential
+    /// (`s3://bucket/warehouse/db/table`). Providers that vend distinct
+    /// credentials per location prefix use it to select the most specific
+    /// match, which must [cover](StorageCredential::covers) `path`.
+    ///
+    /// Backends read the credential's expiry from its config, such as
+    /// `s3.session-token-expires-at-ms`, and load a new one before it expires.
+    /// A credential without an expiry is used for as long as the backend lives.
+    async fn load_credential(&self, path: &str) -> Result<StorageCredential>;

Review Comment:
   The trait doc says this "may be called very frequently", yet each call returns an owned `StorageCredential`. So every file op, per-path delete key and signer refresh clones the config map and its secrets. The trait is new and public, so I'd return `Arc<StorageCredential>` now. Changing it later would break every implementor.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#issuecomment-5215939691

   First PR available here: https://github.com/apache/iceberg-rust/pull/2976


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3735912891


##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,1203 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Refresh of vended storage credentials against a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache, and
+//! the returned [`StorageCredential`] enum lets the storage adapter enforce the
+//! expected backend-specific type. This preserves Java's per-cloud refresh
+//! policy while supporting mixed-cloud tables through a resolving FileIO.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant. To
+//! add a backend, first add its credential type to Iceberg's storage API and
+//! teach the storage adapter to consume it. Then write its `parse_*` function,
+//! add a `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+    AWS_REFRESH_CREDENTIALS_ENABLED, AWS_REFRESH_CREDENTIALS_ENDPOINT,
+    GCS_REFRESH_CREDENTIALS_ENABLED, GCS_REFRESH_CREDENTIALS_ENDPOINT, GCS_TOKEN,
+    GCS_TOKEN_EXPIRES_AT, GcsCredential, S3_ACCESS_KEY_ID, S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN,
+    S3_SESSION_TOKEN_EXPIRES_AT_MS, S3Credential, StorageCredential, StorageCredentialKind,
+    StorageCredentialProvider,
+};
+use iceberg::{Error, ErrorKind, Result};
+use rand::Rng;
+use reqwest::{Method, StatusCode, Url};
+use tokio::sync::Mutex;
+
+use crate::REST_CATALOG_PROP_SCAN_PLAN_ID;
+use crate::client::{HttpClient, deserialize_unexpected_catalog_error};
+use crate::types::LoadCredentialsResponse;
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// 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 {
+    /// Location URL schemes this backend serves.
+    schemes: &'static [&'static str],
+    /// Table property naming the refresh endpoint (absolute or catalog-relative).
+    endpoint_key: &'static str,
+    /// Table property to opt out; refresh is enabled unless this is `"false"`.
+    enabled_key: &'static str,
+    /// Whether to jitter successful prefetch times like AWS `CachedSupplier`.
+    jitter_prefetch: bool,
+    /// Parse a complete credential from catalog-supplied properties.
+    parse_credential:
+        fn(config: &HashMap<String, String>, prefix: Option<String>) -> Result<StorageCredential>,
+}
+
+impl CloudRefresh {
+    /// S3 / AWS
+    const AWS: Self = Self {
+        schemes: &["s3", "s3a", "s3n"],
+        endpoint_key: AWS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: AWS_REFRESH_CREDENTIALS_ENABLED,
+        jitter_prefetch: true,
+        parse_credential: parse_s3_credential,
+    };
+    /// Google Cloud Storage
+    const GCP: Self = Self {
+        schemes: &["gs", "gcs"],
+        endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED,
+        jitter_prefetch: false,
+        parse_credential: parse_gcs_credential,
+    };
+    // TODO: Azure (ADLS) is not yet supported: opendal 0.57's Azdls builder exposes no
+    // custom credential-provider hook, and reqsign's SAS-token credential has no
+    // expiry, so reqsign-based refresh isn't possible.
+
+    /// Backends with refresh support
+    const SUPPORTED: &[Self] = &[Self::AWS, Self::GCP];
+
+    /// The backend that serves `location`, by its URL scheme, or `None` if no
+    /// supported backend matches (in which case static credentials are used
+    /// as-is, as before).
+    fn for_location(location: &str) -> Option<&'static Self> {
+        Self::SUPPORTED
+            .iter()
+            .find(|cloud| cloud.matches_location(location))
+    }
+
+    fn matches_location(&self, location: &str) -> bool {
+        scheme_of(location).is_some_and(|scheme| self.schemes.contains(&scheme.as_str()))
+    }
+}
+
+/// Re-fetch a credential once it is within this window of expiry, so a fresh
+/// token is in hand before the object store would reject the old one.
+const REFRESH_BUFFER: Duration = Duration::from_mins(5);
+
+/// AWS keeps at least one minute between its jittered prefetch time and expiry.
+const MIN_REFRESH_BUFFER: Duration = Duration::from_mins(1);
+
+/// Initial ceiling for failure backoff. Equal jitter chooses from half this
+/// value through the full value.
+const INITIAL_FAILURE_BACKOFF: Duration = Duration::from_secs(1);
+
+/// Maximum failure backoff while a cached credential remains usable.
+const MAX_FAILURE_BACKOFF: Duration = Duration::from_secs(30);
+
+/// A cached vended credential and its refresh schedule.
+#[derive(Clone)]
+struct CachedEntry {
+    credential: StorageCredential,
+    /// When this entry becomes eligible for prefetch. `None` means it does not
+    /// expire and therefore never needs proactive refresh.
+    refresh_at: Option<SystemTime>,
+}
+
+impl CachedEntry {
+    fn new(credential: StorageCredential, jitter_prefetch: bool) -> Self {
+        let refresh_at = credential
+            .expires_at
+            .map(|expires_at| prefetch_time(expires_at, jitter_prefetch));
+        Self {
+            credential,
+            refresh_at,
+        }
+    }
+
+    /// Seed entries that are already inside the nominal five-minute window are
+    /// immediately due. Otherwise AWS applies the same jitter as it does to a
+    /// freshly fetched value.
+    fn seed(credential: StorageCredential, jitter_prefetch: bool) -> Self {
+        let due = credential.expires_at.is_some_and(|expires_at| {
+            SystemTime::now()
+                .checked_add(REFRESH_BUFFER)
+                .is_none_or(|refresh_boundary| refresh_boundary >= expires_at)
+        });
+        let mut entry = Self::new(credential, jitter_prefetch);
+        if due {
+            entry.refresh_at = Some(UNIX_EPOCH);
+        }
+        entry
+    }
+
+    fn is_fresh(&self, now: SystemTime) -> bool {
+        self.refresh_at.is_none_or(|refresh_at| now < refresh_at)
+    }
+
+    fn is_unexpired(&self, now: SystemTime) -> bool {
+        self.credential
+            .expires_at
+            .is_none_or(|expires_at| now < expires_at)
+    }
+}
+
+/// Cached credentials plus failure-backoff state.
+struct CacheState {
+    entries: Vec<CachedEntry>,
+    consecutive_failures: u32,
+    retry_not_before: Option<Instant>,
+}
+
+struct ConfiguredCloud {
+    cloud: &'static CloudRefresh,
+    endpoint: String,
+    cache: Mutex<CacheState>,
+    /// Only one caller fetches at a time. The cache lock is deliberately
+    /// separate so other callers can keep using an unexpired credential while
+    /// the refresh is in flight.
+    refresh: Mutex<()>,
+}
+
+/// Fetches and refreshes vended credentials from a REST catalog's table
+/// credentials endpoint.
+///
+/// Each cloud cache is seeded with the credential from the initial table
+/// properties (when complete) and re-fetched from its endpoint as it
+/// nears expiry.
+pub(crate) struct RestVendedCredentialProvider {
+    client: Arc<HttpClient>,
+    /// Optional scan-plan identifier.
+    plan_id: Option<String>,
+    /// Independently configured endpoint and cache for each backing cloud.
+    clouds: Vec<ConfiguredCloud>,
+}
+
+impl RestVendedCredentialProvider {
+    fn new(client: Arc<HttpClient>, plan_id: Option<String>, clouds: Vec<ConfiguredCloud>) -> Self {
+        Self {
+            client,
+            plan_id,
+            clouds,
+        }
+    }
+
+    fn configured_cloud_for_location(&self, location: &str) -> Option<&ConfiguredCloud> {
+        self.clouds
+            .iter()
+            .find(|configured| configured.cloud.matches_location(location))
+    }
+
+    /// Fetch fresh credentials from the catalog's credentials endpoint.
+    async fn fetch(&self, configured: &ConfiguredCloud) -> Result<Vec<CachedEntry>> {
+        let mut request = self.client.request(Method::GET, &configured.endpoint);
+        if let Some(plan_id) = &self.plan_id {
+            request = request.query(&[("planId", plan_id)]);
+        }
+        let request = request.build()?;
+        let response = self.client.query_catalog(request).await?;
+
+        match response.status() {
+            StatusCode::OK => {
+                let parsed: LoadCredentialsResponse = response.json().await?;
+                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()
+            }
+            _ => Err(deserialize_unexpected_catalog_error(
+                response,
+                self.client.disable_header_redaction(),
+            )
+            .await),
+        }
+    }
+
+    async fn refresh_credential(
+        &self,
+        configured: &ConfiguredCloud,
+        path: &str,
+        fallback: Option<CachedEntry>,
+    ) -> Result<StorageCredential> {
+        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)
+            }
+        }

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



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#issuecomment-5717908692

   Hey @mbutrovich @CTTY could I get a review here please? Thanks


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3735924461


##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,107 @@ pub trait StorageFactory: Debug + Send + Sync {
     /// A `Result` containing an `Arc<dyn Storage>` on success, or an error
     /// if the storage could not be created.
     fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>>;
+
+    /// Build a new Storage instance, optionally supplying a credential provider
+    /// that the backend can call to obtain and refresh short-lived credentials.
+    fn build_with_credentials(
+        &self,
+        config: &StorageConfig,
+        credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
+    ) -> Result<Arc<dyn Storage>> {
+        if credential_provider.is_some() {
+            return Err(Error::new(
+                ErrorKind::FeatureUnsupported,
+                "Storage factory does not support refreshable credential providers",
+            ));
+        }
+
+        self.build(config)
+    }
+}
+
+/// Supplies fresh, backend-specific storage credentials on demand.
+///
+/// A catalog that vends temporary credentials implements this trait so that
+/// storage backends can re-fetch credentials as they approach expiry instead
+/// of failing once the initial token's TTL runs out.
+///
+/// # Caching
+///
+/// [`load_credential`](Self::load_credential) may be called very frequently —
+/// the S3 backend, for example, rebuilds its operator (and therefore its
+/// signer) on every file operation. Implementations must cache internally and
+/// only re-fetch when the current credential is at or near expiry; otherwise
+/// every object-store request would trigger a call back to the catalog.
+#[async_trait]
+pub trait StorageCredentialProvider: Debug + Send + Sync {
+    /// Return whether this provider has refresh configuration for `path`.
+    ///
+    /// Backends use this before replacing their normal credential chain. The
+    /// default is `true` for single-backend providers; multi-backend providers
+    /// should return `false` for schemes they do not configure.
+    fn supports_path(&self, _path: &str) -> bool {
+        true
+    }
+
+    /// Load a fresh credential for the storage location identified by `path`.
+    ///
+    /// `path` is the absolute location being accessed (e.g.
+    /// `s3://bucket/warehouse/db/table/...`). Providers that vend distinct
+    /// credentials per location prefix use it to select the most specific
+    /// match. When the selected credential has a declared
+    /// [`StorageCredential::prefix`], it must cover `path`.
+    async fn load_credential(&self, path: &str) -> Result<StorageCredential>;
+}
+
+/// A vended storage credential together with its scope and expiry.
+#[derive(Clone, Debug)]
+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,
+}

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



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3689693249


##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -54,6 +54,8 @@ pub const REST_CATALOG_PROP_URI: &str = "uri";
 pub const REST_CATALOG_PROP_WAREHOUSE: &str = "warehouse";
 /// 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";

Review Comment:
   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



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3689726524


##########
crates/catalog/rest/src/client.rs:
##########
@@ -278,29 +298,30 @@ pub(crate) async fn deserialize_catalog_response<R: DeserializeOwned>(
     let bytes = response.bytes().await?;
 
     serde_json::from_slice::<R>(&bytes).map_err(|e| {
+        // Successful REST responses can contain OAuth tokens and delegated
+        // storage credentials. Never copy an unparsable response into an error.
         Error::new(
             ErrorKind::Unexpected,
             "Failed to parse response from rest catalog server",
         )
-        .with_context("json", String::from_utf8_lossy(&bytes))
         .with_source(e)
     })
 }
 
-/// Headers that contain sensitive information and should be excluded from logs.
-const SENSITIVE_HEADERS: &[&str] = &[
-    "authorization",
-    "proxy-authorization",
-    "set-cookie",
-    "cookie",
-    "x-api-key",
-    "x-auth-token",
-];
-
-/// Returns true if the header name is considered sensitive.
+/// Returns true if the header may carry a secret.
 fn is_sensitive_header(name: &str) -> bool {
     let name_lower = name.to_lowercase();
-    SENSITIVE_HEADERS.iter().any(|h| name_lower == *h)
+    [
+        "auth",
+        "token",
+        "secret",
+        "key",
+        "password",
+        "cookie",
+        "credential",
+    ]

Review Comment:
   Not sure if we want to do it like this since I guess there could technically be other headers that would be made sensitive. The main change I wanted to do was also make `x-client-secret` and `x-client-credential` sensitive. I think Java redacts every header anyway, so I'm not sure if this really matters



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r4185331338


##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -1470,7 +1478,7 @@ impl SessionCatalog for RestSessionCatalog {
         };
 
         let file_io = self
-            .load_file_io(Some(&response.metadata_location), None)
+            .load_file_io(commit.identifier(), Some(&response.metadata_location), None)

Review Comment:
   Initially I did reuse the FileIO while I was drafting the PR, but noticed that #2651 already does this via `Table::with_file_io` in `do_commit`



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["mbutrovich (via GitHub)" <gi...@apache.org>]

mbutrovich commented on PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#issuecomment-5194842205

   > 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!


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#issuecomment-5341808796

   Hey @CTTY, here's the full PR to see the broader picture. Happy if you wish to review the full PR if it's easier compared to splitting it.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3689827205


##########
crates/storage/opendal/src/utils.rs:
##########
@@ -27,3 +27,25 @@ pub(crate) fn from_opendal_error(e: opendal::Error) -> iceberg::Error {
     )
     .with_source(e)
 }
+
+/// Convert a [`SystemTime`](std::time::SystemTime) credential expiry into the
+/// `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(

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



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3690419391


##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,1193 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Refresh of vended storage credentials against a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache, and
+//! the returned [`StorageCredential`] enum lets the storage adapter enforce the
+//! expected backend-specific type. This preserves Java's per-cloud refresh
+//! policy while supporting mixed-cloud tables through a resolving FileIO.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant. To
+//! add a backend, first add its credential type to Iceberg's storage API and
+//! teach the storage adapter to consume it. Then write its `parse_*` function,
+//! add a `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+    AWS_REFRESH_CREDENTIALS_ENABLED, AWS_REFRESH_CREDENTIALS_ENDPOINT,
+    GCS_REFRESH_CREDENTIALS_ENABLED, GCS_REFRESH_CREDENTIALS_ENDPOINT, GCS_TOKEN,
+    GCS_TOKEN_EXPIRES_AT, GcsCredential, S3_ACCESS_KEY_ID, S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN,
+    S3_SESSION_TOKEN_EXPIRES_AT_MS, S3Credential, StorageCredential, StorageCredentialKind,
+    StorageCredentialProvider,
+};
+use iceberg::{Error, ErrorKind, Result};
+use rand::Rng;
+use reqwest::{Method, StatusCode, Url};
+use tokio::sync::Mutex;
+
+use crate::REST_CATALOG_PROP_SCAN_PLAN_ID;
+use crate::client::{HttpClient, deserialize_unexpected_catalog_error};
+use crate::types::LoadCredentialsResponse;
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// 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 {
+    /// Location URL schemes this backend serves.
+    schemes: &'static [&'static str],
+    /// Table property naming the refresh endpoint (absolute or catalog-relative).
+    endpoint_key: &'static str,
+    /// Table property to opt out; refresh is enabled unless this is `"false"`.
+    enabled_key: &'static str,
+    /// Whether to jitter successful prefetch times like AWS `CachedSupplier`.
+    jitter_prefetch: bool,
+    /// Parse a complete credential from catalog-supplied properties.
+    parse_credential: fn(&HashMap<String, String>) -> Result<StorageCredential>,
+}
+
+impl CloudRefresh {
+    /// S3 / AWS
+    const AWS: Self = Self {
+        schemes: &["s3", "s3a", "s3n"],
+        endpoint_key: AWS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: AWS_REFRESH_CREDENTIALS_ENABLED,
+        jitter_prefetch: true,
+        parse_credential: parse_s3_credential,
+    };
+    /// Google Cloud Storage
+    const GCP: Self = Self {
+        schemes: &["gs", "gcs"],
+        endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED,
+        jitter_prefetch: false,
+        parse_credential: parse_gcs_credential,
+    };
+    // TODO: Azure (ADLS) is not yet supported: opendal 0.57's Azdls builder exposes no
+    // custom credential-provider hook, and reqsign's SAS-token credential has no
+    // expiry, so reqsign-based refresh isn't possible.
+
+    /// Backends with refresh support
+    const SUPPORTED: &[Self] = &[Self::AWS, Self::GCP];
+
+    /// The backend that serves `location`, by its URL scheme, or `None` if no
+    /// supported backend matches (in which case static credentials are used
+    /// as-is, as before).
+    fn for_location(location: &str) -> Option<&'static Self> {
+        let scheme = scheme_of(location)?;
+        Self::SUPPORTED
+            .iter()
+            .find(|cloud| cloud.schemes.contains(&scheme.as_str()))
+    }
+
+    fn matches_credential_prefix(&self, prefix: &str) -> bool {
+        scheme_of(prefix).is_some_and(|scheme| self.schemes.contains(&scheme.as_str()))
+    }
+}
+
+/// Re-fetch a credential once it is within this window of expiry, so a fresh
+/// token is in hand before the object store would reject the old one.
+const REFRESH_BUFFER: Duration = Duration::from_mins(5);
+
+/// AWS keeps at least one minute between its jittered prefetch time and expiry.
+const MIN_REFRESH_BUFFER: Duration = Duration::from_mins(1);
+
+/// Initial ceiling for failure backoff. Equal jitter chooses from half this
+/// value through the full value.
+const INITIAL_FAILURE_BACKOFF: Duration = Duration::from_secs(1);
+
+/// Maximum failure backoff while a cached credential remains usable.
+const MAX_FAILURE_BACKOFF: Duration = Duration::from_secs(30);
+
+/// A vended credential paired with the storage-location prefix it applies to.
+#[derive(Clone)]
+struct CachedEntry {
+    prefix: String,
+    credential: StorageCredential,
+    /// When this entry becomes eligible for prefetch. `None` means it does not
+    /// expire and therefore never needs proactive refresh.
+    refresh_at: Option<SystemTime>,
+}
+
+impl CachedEntry {
+    fn new(prefix: String, credential: StorageCredential, jitter_prefetch: bool) -> Self {
+        let refresh_at = credential
+            .expires_at
+            .map(|expires_at| prefetch_time(expires_at, jitter_prefetch));
+        Self {
+            prefix,
+            credential,
+            refresh_at,
+        }
+    }
+
+    /// Seed entries that are already inside the nominal five-minute window are
+    /// immediately due. Otherwise AWS applies the same jitter as it does to a
+    /// freshly fetched value.
+    fn seed(prefix: String, credential: StorageCredential, jitter_prefetch: bool) -> Self {
+        let due = credential.expires_at.is_some_and(|expires_at| {
+            SystemTime::now()
+                .checked_add(REFRESH_BUFFER)
+                .is_none_or(|refresh_boundary| refresh_boundary >= expires_at)
+        });
+        let mut entry = Self::new(prefix, credential, jitter_prefetch);
+        if due {
+            entry.refresh_at = Some(UNIX_EPOCH);
+        }
+        entry
+    }
+
+    fn is_fresh(&self, now: SystemTime) -> bool {
+        self.refresh_at.is_none_or(|refresh_at| now < refresh_at)
+    }
+
+    fn is_unexpired(&self, now: SystemTime) -> bool {
+        self.credential
+            .expires_at
+            .is_none_or(|expires_at| now < expires_at)
+    }
+}
+
+/// Cached credentials plus failure-backoff state.
+struct CacheState {
+    entries: Vec<CachedEntry>,
+    consecutive_failures: u32,
+    retry_not_before: Option<Instant>,
+}
+
+struct ConfiguredCloud {
+    cloud: &'static CloudRefresh,
+    endpoint: String,
+    cache: Mutex<CacheState>,
+    /// Only one caller fetches at a time. The cache lock is deliberately
+    /// separate so other callers can keep using an unexpired credential while
+    /// the refresh is in flight.
+    refresh: Mutex<()>,
+}
+
+/// Fetches and refreshes vended credentials from a REST catalog's table
+/// credentials endpoint.
+///
+/// Each cloud cache is seeded with the credential from the initial table
+/// properties (when complete) and re-fetched from its endpoint as it
+/// nears expiry.
+pub(crate) struct RestVendedCredentialProvider {
+    client: Arc<HttpClient>,
+    /// Optional scan-plan identifier.
+    plan_id: Option<String>,
+    /// Independently configured endpoint and cache for each backing cloud.
+    clouds: Vec<ConfiguredCloud>,
+}
+
+impl RestVendedCredentialProvider {
+    fn new(client: Arc<HttpClient>, plan_id: Option<String>, clouds: Vec<ConfiguredCloud>) -> Self {
+        Self {
+            client,
+            plan_id,
+            clouds,
+        }
+    }
+
+    /// Fetch fresh credentials from the catalog's credentials endpoint.
+    async fn fetch(&self, configured: &ConfiguredCloud) -> Result<Vec<CachedEntry>> {
+        let mut request = self.client.request(Method::GET, &configured.endpoint);
+        if let Some(plan_id) = &self.plan_id {
+            request = request.query(&[("planId", plan_id)]);
+        }
+        let request = request.build()?;
+        let response = self.client.query_catalog(request).await?;
+
+        match response.status() {
+            StatusCode::OK => {
+                let parsed: LoadCredentialsResponse = response.json().await?;
+                parsed
+                    .storage_credentials
+                    .into_iter()
+                    .filter(|sc| configured.cloud.matches_credential_prefix(&sc.prefix))
+                    .map(|sc| {
+                        (configured.cloud.parse_credential)(&sc.config).map(|credential| {
+                            CachedEntry::new(
+                                sc.prefix,
+                                credential,
+                                configured.cloud.jitter_prefetch,
+                            )
+                        })
+                    })
+                    .collect()
+            }
+            _ => Err(deserialize_unexpected_catalog_error(
+                response,
+                self.client.disable_header_redaction(),
+            )
+            .await),
+        }
+    }
+
+    async fn refresh_credential(
+        &self,
+        configured: &ConfiguredCloud,
+        path: &str,
+        fallback: Option<CachedEntry>,
+    ) -> Result<StorageCredential> {
+        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)
+            }
+        }
+    }
+}
+
+impl std::fmt::Debug for RestVendedCredentialProvider {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("RestVendedCredentialProvider")
+            .field("configured_clouds", &self.clouds.len())
+            .finish_non_exhaustive()
+    }
+}
+
+enum CacheDecision {
+    Use(StorageCredential),
+    Refresh(Option<CachedEntry>),
+}
+
+async fn cache_decision(configured: &ConfiguredCloud, path: &str) -> CacheDecision {
+    let cache = configured.cache.lock().await;
+    let current = longest_prefix_match(&cache.entries, path).cloned();
+    let now = SystemTime::now();
+
+    if let Some(entry) = current.as_ref().filter(|entry| entry.is_fresh(now)) {
+        return CacheDecision::Use(entry.credential.clone());
+    }
+
+    if cache
+        .retry_not_before
+        .is_some_and(|retry_at| Instant::now() < retry_at)
+        && let Some(entry) = current.as_ref().filter(|entry| entry.is_unexpired(now))
+    {
+        return CacheDecision::Use(entry.credential.clone());
+    }
+
+    CacheDecision::Refresh(current)
+}
+
+#[async_trait]
+impl StorageCredentialProvider for RestVendedCredentialProvider {
+    fn supports_path(&self, path: &str) -> bool {
+        CloudRefresh::for_location(path).is_some_and(|cloud| {
+            self.clouds
+                .iter()
+                .any(|configured| configured.cloud.endpoint_key == cloud.endpoint_key)
+        })
+    }
+
+    async fn load_credential(&self, path: &str) -> Result<StorageCredential> {
+        let cloud = CloudRefresh::for_location(path).ok_or_else(|| {
+            Error::new(
+                ErrorKind::FeatureUnsupported,
+                format!("no credential refresh implementation for storage location: {path}"),
+            )
+        })?;
+        let configured = self
+            .clouds
+            .iter()
+            .find(|configured| configured.cloud.endpoint_key == cloud.endpoint_key)
+            .ok_or_else(|| {
+                Error::new(
+                    ErrorKind::FeatureUnsupported,
+                    format!("credential refresh is not configured for storage location: {path}"),
+                )
+            })?;
+
+        let current = match cache_decision(configured, path).await {
+            CacheDecision::Use(credential) => return Ok(credential),
+            CacheDecision::Refresh(current) => current,
+        };
+
+        // One caller refreshes, while concurrent callers immediately keep using the
+        // unexpired cached credential. With no usable credential, callers wait
+        // for the in-flight refresh instead.
+        let usable = current
+            .as_ref()
+            .filter(|entry| entry.is_unexpired(SystemTime::now()));
+        let _refresh_guard = if let Some(entry) = usable {
+            match configured.refresh.try_lock() {
+                Ok(guard) => guard,
+                Err(_) => return Ok(entry.credential.clone()),
+            }
+        } else {
+            configured.refresh.lock().await
+        };
+
+        // Another caller may have completed a refresh between our cache check
+        // and acquiring the single-flight guard.
+        let current = match cache_decision(configured, path).await {
+            CacheDecision::Use(credential) => return Ok(credential),
+            CacheDecision::Refresh(current) => current,
+        };
+
+        self.refresh_credential(configured, path, current).await
+    }
+}
+
+/// Select the credential whose prefix is the longest match for `path`.
+fn longest_prefix_match<'a>(entries: &'a [CachedEntry], path: &str) -> Option<&'a CachedEntry> {
+    entries
+        .iter()
+        .filter(|entry| path.starts_with(&entry.prefix))
+        .max_by_key(|entry| entry.prefix.len())
+}
+
+/// Compute a successful credential's prefetch time.
+fn prefetch_time(expires_at: SystemTime, jitter: bool) -> SystemTime {
+    let base = expires_at.checked_sub(REFRESH_BUFFER).unwrap_or(UNIX_EPOCH);
+    if !jitter {
+        return base;
+    }
+
+    let jitter_window = REFRESH_BUFFER.saturating_sub(MIN_REFRESH_BUFFER);
+    let jitter_millis = rand::rng().random_range(0..jitter_window.as_millis() as u64);
+    base.checked_add(Duration::from_millis(jitter_millis))
+        .unwrap_or(base)
+}
+
+/// Equal-jitter exponential backoff. The random lower half avoids both hot
+/// retry loops and synchronized retries across clients.
+fn failure_backoff(consecutive_failures: u32) -> Duration {
+    let exponent = consecutive_failures.saturating_sub(1).min(5);
+    let ceiling = INITIAL_FAILURE_BACKOFF
+        .checked_mul(1 << exponent)
+        .unwrap_or(MAX_FAILURE_BACKOFF)
+        .min(MAX_FAILURE_BACKOFF);
+    let ceiling_millis = ceiling.as_millis() as u64;
+    let floor_millis = ceiling_millis / 2;
+    Duration::from_millis(rand::rng().random_range(floor_millis..=ceiling_millis))
+}
+
+/// Build a credential provider from a table's properties,
+/// or `None` when no supported cloud advertises an enabled refresh endpoint.
+///
+/// `base_uri` is the catalog URI, used to resolve a relative endpoint.
+pub(crate) fn build_vended_credential_provider(
+    client: Arc<HttpClient>,
+    base_uri: &str,
+    props: &HashMap<String, String>,
+) -> Result<Option<Arc<dyn StorageCredentialProvider>>> {
+    let clouds = CloudRefresh::SUPPORTED
+        .iter()
+        .filter_map(|cloud| {
+            // Refresh is enabled by default and invalid booleans disable refresh.
+            let enabled = props
+                .get(cloud.enabled_key)
+                .is_none_or(|value| value.parse().unwrap_or(false));
+            if !enabled {
+                return None;
+            }
+
+            let endpoint = resolve_endpoint(
+                base_uri,
+                props
+                    .get(cloud.endpoint_key)
+                    .filter(|value| !value.is_empty())?,
+            );
+            let entries = (cloud.parse_credential)(props)
+                .ok()
+                .map(|credential| {
+                    vec![CachedEntry::seed(
+                        // Flat table properties carry no prefix. This cache is
+                        // already cloud-specific, so an empty prefix is safe.
+                        String::new(),
+                        credential,
+                        cloud.jitter_prefetch,
+                    )]
+                })
+                .unwrap_or_default();
+
+            Some(ConfiguredCloud {
+                cloud,
+                endpoint,
+                cache: Mutex::new(CacheState {
+                    entries,
+                    consecutive_failures: 0,
+                    retry_not_before: None,
+                }),
+                refresh: Mutex::new(()),
+            })
+        })
+        .collect::<Vec<_>>();
+
+    if clouds.is_empty() {
+        return Ok(None);
+    }
+
+    let table_client = Arc::new(client.for_table(base_uri, props.clone())?);
+    let plan_id = props.get(REST_CATALOG_PROP_SCAN_PLAN_ID).cloned();
+    Ok(Some(Arc::new(RestVendedCredentialProvider::new(
+        table_client,
+        plan_id,
+        clouds,
+    ))))
+}
+
+/// Resolve a possibly-relative refresh endpoint against the catalog base URI.
+fn resolve_endpoint(base_uri: &str, endpoint: &str) -> String {

Review Comment:
   The behaviour here was done to match Java



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["mbutrovich (via GitHub)" <gi...@apache.org>]

mbutrovich commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3723653871


##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,103 @@ pub trait StorageFactory: Debug + Send + Sync {
     /// A `Result` containing an `Arc<dyn Storage>` on success, or an error
     /// if the storage could not be created.
     fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>>;
+
+    /// Build a new Storage instance, optionally supplying a credential provider
+    /// that the backend can call to obtain and refresh short-lived credentials.
+    fn build_with_credentials(
+        &self,
+        config: &StorageConfig,
+        credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
+    ) -> Result<Arc<dyn Storage>> {
+        if credential_provider.is_some() {
+            return Err(Error::new(
+                ErrorKind::FeatureUnsupported,
+                "Storage factory does not support refreshable credential providers",
+            ));
+        }
+
+        self.build(config)
+    }
+}
+
+/// Supplies fresh, backend-specific storage credentials on demand.
+///
+/// A catalog that vends temporary credentials implements this trait so that
+/// storage backends can re-fetch credentials as they approach expiry instead
+/// of failing once the initial token's TTL runs out.
+///
+/// # Caching
+///
+/// [`load_credential`](Self::load_credential) may be called very frequently —
+/// the S3 backend, for example, rebuilds its operator (and therefore its
+/// signer) on every file operation. Implementations must cache internally and
+/// only re-fetch when the current credential is at or near expiry; otherwise
+/// every object-store request would trigger a call back to the catalog.
+#[async_trait]
+pub trait StorageCredentialProvider: Debug + Send + Sync {

Review Comment:
   The trait-based design seems fine, Java not having an equivalent is not a reason to avoid one here. See above for the separate, concrete question on the same code (pub fields, no constructor).



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#issuecomment-5215447916

   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.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3690259055


##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,1193 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Refresh of vended storage credentials against a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache, and
+//! the returned [`StorageCredential`] enum lets the storage adapter enforce the
+//! expected backend-specific type. This preserves Java's per-cloud refresh
+//! policy while supporting mixed-cloud tables through a resolving FileIO.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant. To
+//! add a backend, first add its credential type to Iceberg's storage API and
+//! teach the storage adapter to consume it. Then write its `parse_*` function,
+//! add a `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+    AWS_REFRESH_CREDENTIALS_ENABLED, AWS_REFRESH_CREDENTIALS_ENDPOINT,
+    GCS_REFRESH_CREDENTIALS_ENABLED, GCS_REFRESH_CREDENTIALS_ENDPOINT, GCS_TOKEN,
+    GCS_TOKEN_EXPIRES_AT, GcsCredential, S3_ACCESS_KEY_ID, S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN,
+    S3_SESSION_TOKEN_EXPIRES_AT_MS, S3Credential, StorageCredential, StorageCredentialKind,
+    StorageCredentialProvider,
+};
+use iceberg::{Error, ErrorKind, Result};
+use rand::Rng;
+use reqwest::{Method, StatusCode, Url};
+use tokio::sync::Mutex;
+
+use crate::REST_CATALOG_PROP_SCAN_PLAN_ID;
+use crate::client::{HttpClient, deserialize_unexpected_catalog_error};
+use crate::types::LoadCredentialsResponse;
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// 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 {
+    /// Location URL schemes this backend serves.
+    schemes: &'static [&'static str],
+    /// Table property naming the refresh endpoint (absolute or catalog-relative).
+    endpoint_key: &'static str,
+    /// Table property to opt out; refresh is enabled unless this is `"false"`.
+    enabled_key: &'static str,
+    /// Whether to jitter successful prefetch times like AWS `CachedSupplier`.
+    jitter_prefetch: bool,
+    /// Parse a complete credential from catalog-supplied properties.
+    parse_credential: fn(&HashMap<String, String>) -> Result<StorageCredential>,
+}
+
+impl CloudRefresh {
+    /// S3 / AWS
+    const AWS: Self = Self {
+        schemes: &["s3", "s3a", "s3n"],
+        endpoint_key: AWS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: AWS_REFRESH_CREDENTIALS_ENABLED,
+        jitter_prefetch: true,
+        parse_credential: parse_s3_credential,
+    };
+    /// Google Cloud Storage
+    const GCP: Self = Self {
+        schemes: &["gs", "gcs"],
+        endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED,
+        jitter_prefetch: false,

Review Comment:
   GCP doesn't have jitter for Java as it's SDK dependent, which only AWS does it appears. I'm happy to discuss more about the retry strategy here in order to make it consistent (which Java doesn't have at the moment)



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3689840020


##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,1193 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Refresh of vended storage credentials against a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache, and
+//! the returned [`StorageCredential`] enum lets the storage adapter enforce the
+//! expected backend-specific type. This preserves Java's per-cloud refresh
+//! policy while supporting mixed-cloud tables through a resolving FileIO.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant. To
+//! add a backend, first add its credential type to Iceberg's storage API and
+//! teach the storage adapter to consume it. Then write its `parse_*` function,
+//! add a `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+    AWS_REFRESH_CREDENTIALS_ENABLED, AWS_REFRESH_CREDENTIALS_ENDPOINT,
+    GCS_REFRESH_CREDENTIALS_ENABLED, GCS_REFRESH_CREDENTIALS_ENDPOINT, GCS_TOKEN,
+    GCS_TOKEN_EXPIRES_AT, GcsCredential, S3_ACCESS_KEY_ID, S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN,
+    S3_SESSION_TOKEN_EXPIRES_AT_MS, S3Credential, StorageCredential, StorageCredentialKind,
+    StorageCredentialProvider,
+};
+use iceberg::{Error, ErrorKind, Result};
+use rand::Rng;
+use reqwest::{Method, StatusCode, Url};
+use tokio::sync::Mutex;
+
+use crate::REST_CATALOG_PROP_SCAN_PLAN_ID;
+use crate::client::{HttpClient, deserialize_unexpected_catalog_error};
+use crate::types::LoadCredentialsResponse;
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// 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 {

Review Comment:
   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



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3689733836


##########
crates/catalog/rest/Cargo.toml:
##########
@@ -35,6 +35,7 @@ chrono = { workspace = true }
 http = { workspace = true }
 iceberg = { workspace = true }
 itertools = { workspace = true }
+rand = { workspace = true }

Review Comment:
   ~Not sure if we're okay with adding a new dep~ `rand` was already a workspace dep. It's only added for the jittering functionality when retrying



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r4185310758


##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -879,9 +873,29 @@ impl RestSessionCatalog {
                 )
             })?;
 
-        let file_io = FileIOBuilder::new(factory).with_props(props).build();
+        // If the catalog vends refreshable credentials for this table's storage,
+        // attach a provider so the backend re-fetches them before they expire.
+        // Only catalog authentication resolved from the properties can be
+        // rebuilt after FileIO serialization.
+        let credential_provider = build_vended_credential_provider(
+            &client.http_client,
+            client.auth_manager.as_ref(),
+            RestVendedCredentialProviderFactory::new(
+                &client.config.uri,
+                table.clone(),
+                table_config.unwrap_or_default(),
+            ),
+            &props,
+            self.auth_manager.is_none(),
+        )

Review Comment:
   Indeed, this PR depends on #2651 for storage-credentials and won't consume them itself. After that other PR lands, its per-prefix storages will be passed the provider with optional seeding from storage-credentials.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#issuecomment-5570244331

   @CTTY this now includes Azure support too, but as I mentioned before OpenDAL's latest release is broken so checks are failing. Hopefully 0.59.1 should release soon.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3689754362


##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,103 @@ pub trait StorageFactory: Debug + Send + Sync {
     /// A `Result` containing an `Arc<dyn Storage>` on success, or an error
     /// if the storage could not be created.
     fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>>;
+
+    /// Build a new Storage instance, optionally supplying a credential provider
+    /// that the backend can call to obtain and refresh short-lived credentials.
+    fn build_with_credentials(
+        &self,
+        config: &StorageConfig,
+        credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
+    ) -> Result<Arc<dyn Storage>> {
+        if credential_provider.is_some() {
+            return Err(Error::new(
+                ErrorKind::FeatureUnsupported,
+                "Storage factory does not support refreshable credential providers",
+            ));
+        }
+
+        self.build(config)
+    }
+}
+
+/// Supplies fresh, backend-specific storage credentials on demand.
+///
+/// A catalog that vends temporary credentials implements this trait so that
+/// storage backends can re-fetch credentials as they approach expiry instead
+/// of failing once the initial token's TTL runs out.
+///
+/// # Caching
+///
+/// [`load_credential`](Self::load_credential) may be called very frequently —
+/// the S3 backend, for example, rebuilds its operator (and therefore its
+/// signer) on every file operation. Implementations must cache internally and
+/// only re-fetch when the current credential is at or near expiry; otherwise
+/// every object-store request would trigger a call back to the catalog.
+#[async_trait]
+pub trait StorageCredentialProvider: Debug + Send + Sync {

Review Comment:
   This is probably one of the most important additions in the PR but it doesn't really follow Java since Java doesn't have an interface over vended credential providers. I think it's probably fine and nicer like this, especially since OpenDAL abstracts everything away quite nicely



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3689733836


##########
crates/catalog/rest/Cargo.toml:
##########
@@ -35,6 +35,7 @@ chrono = { workspace = true }
 http = { workspace = true }
 iceberg = { workspace = true }
 itertools = { workspace = true }
+rand = { workspace = true }

Review Comment:
   Not sure if we're okay with adding a new dep. It's only added for the jittering functionality when retrying



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["mbutrovich (via GitHub)" <gi...@apache.org>]

mbutrovich commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r3723620788


##########
crates/storage/opendal/src/lib.rs:
##########
@@ -384,6 +415,27 @@ impl OpenDalStorage {
         }
     }
 
+    /// Returns whether `path` uses a path-scoped dynamic credential provider.
+    ///
+    /// Such paths cannot share an OpenDAL deleter unless the provider can expose
+    /// the credential scope that applies to each path. Process them independently
+    /// so bulk deletion remains bounded without crossing credential boundaries.
+    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 {

Review Comment:
   `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?



##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,1203 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Refresh of vended storage credentials against a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache, and
+//! the returned [`StorageCredential`] enum lets the storage adapter enforce the
+//! expected backend-specific type. This preserves Java's per-cloud refresh
+//! policy while supporting mixed-cloud tables through a resolving FileIO.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant. To
+//! add a backend, first add its credential type to Iceberg's storage API and
+//! teach the storage adapter to consume it. Then write its `parse_*` function,
+//! add a `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+    AWS_REFRESH_CREDENTIALS_ENABLED, AWS_REFRESH_CREDENTIALS_ENDPOINT,
+    GCS_REFRESH_CREDENTIALS_ENABLED, GCS_REFRESH_CREDENTIALS_ENDPOINT, GCS_TOKEN,
+    GCS_TOKEN_EXPIRES_AT, GcsCredential, S3_ACCESS_KEY_ID, S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN,
+    S3_SESSION_TOKEN_EXPIRES_AT_MS, S3Credential, StorageCredential, StorageCredentialKind,
+    StorageCredentialProvider,
+};
+use iceberg::{Error, ErrorKind, Result};
+use rand::Rng;
+use reqwest::{Method, StatusCode, Url};
+use tokio::sync::Mutex;
+
+use crate::REST_CATALOG_PROP_SCAN_PLAN_ID;
+use crate::client::{HttpClient, deserialize_unexpected_catalog_error};
+use crate::types::LoadCredentialsResponse;
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// 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 {
+    /// Location URL schemes this backend serves.
+    schemes: &'static [&'static str],
+    /// Table property naming the refresh endpoint (absolute or catalog-relative).
+    endpoint_key: &'static str,
+    /// Table property to opt out; refresh is enabled unless this is `"false"`.
+    enabled_key: &'static str,
+    /// Whether to jitter successful prefetch times like AWS `CachedSupplier`.
+    jitter_prefetch: bool,
+    /// Parse a complete credential from catalog-supplied properties.
+    parse_credential:
+        fn(config: &HashMap<String, String>, prefix: Option<String>) -> Result<StorageCredential>,
+}
+
+impl CloudRefresh {
+    /// S3 / AWS
+    const AWS: Self = Self {
+        schemes: &["s3", "s3a", "s3n"],
+        endpoint_key: AWS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: AWS_REFRESH_CREDENTIALS_ENABLED,
+        jitter_prefetch: true,
+        parse_credential: parse_s3_credential,
+    };
+    /// Google Cloud Storage
+    const GCP: Self = Self {
+        schemes: &["gs", "gcs"],
+        endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED,
+        jitter_prefetch: false,
+        parse_credential: parse_gcs_credential,
+    };
+    // TODO: Azure (ADLS) is not yet supported: opendal 0.57's Azdls builder exposes no
+    // custom credential-provider hook, and reqsign's SAS-token credential has no
+    // expiry, so reqsign-based refresh isn't possible.
+
+    /// Backends with refresh support
+    const SUPPORTED: &[Self] = &[Self::AWS, Self::GCP];
+
+    /// The backend that serves `location`, by its URL scheme, or `None` if no
+    /// supported backend matches (in which case static credentials are used
+    /// as-is, as before).
+    fn for_location(location: &str) -> Option<&'static Self> {
+        Self::SUPPORTED
+            .iter()
+            .find(|cloud| cloud.matches_location(location))
+    }
+
+    fn matches_location(&self, location: &str) -> bool {
+        scheme_of(location).is_some_and(|scheme| self.schemes.contains(&scheme.as_str()))
+    }
+}
+
+/// Re-fetch a credential once it is within this window of expiry, so a fresh
+/// token is in hand before the object store would reject the old one.
+const REFRESH_BUFFER: Duration = Duration::from_mins(5);
+
+/// AWS keeps at least one minute between its jittered prefetch time and expiry.
+const MIN_REFRESH_BUFFER: Duration = Duration::from_mins(1);
+
+/// Initial ceiling for failure backoff. Equal jitter chooses from half this
+/// value through the full value.
+const INITIAL_FAILURE_BACKOFF: Duration = Duration::from_secs(1);
+
+/// Maximum failure backoff while a cached credential remains usable.
+const MAX_FAILURE_BACKOFF: Duration = Duration::from_secs(30);
+
+/// A cached vended credential and its refresh schedule.
+#[derive(Clone)]
+struct CachedEntry {
+    credential: StorageCredential,
+    /// When this entry becomes eligible for prefetch. `None` means it does not
+    /// expire and therefore never needs proactive refresh.
+    refresh_at: Option<SystemTime>,
+}
+
+impl CachedEntry {
+    fn new(credential: StorageCredential, jitter_prefetch: bool) -> Self {
+        let refresh_at = credential
+            .expires_at
+            .map(|expires_at| prefetch_time(expires_at, jitter_prefetch));
+        Self {
+            credential,
+            refresh_at,
+        }
+    }
+
+    /// Seed entries that are already inside the nominal five-minute window are
+    /// immediately due. Otherwise AWS applies the same jitter as it does to a
+    /// freshly fetched value.
+    fn seed(credential: StorageCredential, jitter_prefetch: bool) -> Self {
+        let due = credential.expires_at.is_some_and(|expires_at| {
+            SystemTime::now()
+                .checked_add(REFRESH_BUFFER)
+                .is_none_or(|refresh_boundary| refresh_boundary >= expires_at)
+        });
+        let mut entry = Self::new(credential, jitter_prefetch);
+        if due {
+            entry.refresh_at = Some(UNIX_EPOCH);
+        }
+        entry
+    }
+
+    fn is_fresh(&self, now: SystemTime) -> bool {
+        self.refresh_at.is_none_or(|refresh_at| now < refresh_at)
+    }
+
+    fn is_unexpired(&self, now: SystemTime) -> bool {
+        self.credential
+            .expires_at
+            .is_none_or(|expires_at| now < expires_at)
+    }
+}
+
+/// Cached credentials plus failure-backoff state.
+struct CacheState {
+    entries: Vec<CachedEntry>,
+    consecutive_failures: u32,
+    retry_not_before: Option<Instant>,
+}
+
+struct ConfiguredCloud {
+    cloud: &'static CloudRefresh,
+    endpoint: String,
+    cache: Mutex<CacheState>,
+    /// Only one caller fetches at a time. The cache lock is deliberately
+    /// separate so other callers can keep using an unexpired credential while
+    /// the refresh is in flight.
+    refresh: Mutex<()>,
+}
+
+/// Fetches and refreshes vended credentials from a REST catalog's table
+/// credentials endpoint.
+///
+/// Each cloud cache is seeded with the credential from the initial table
+/// properties (when complete) and re-fetched from its endpoint as it
+/// nears expiry.
+pub(crate) struct RestVendedCredentialProvider {
+    client: Arc<HttpClient>,
+    /// Optional scan-plan identifier.
+    plan_id: Option<String>,
+    /// Independently configured endpoint and cache for each backing cloud.
+    clouds: Vec<ConfiguredCloud>,
+}
+
+impl RestVendedCredentialProvider {
+    fn new(client: Arc<HttpClient>, plan_id: Option<String>, clouds: Vec<ConfiguredCloud>) -> Self {
+        Self {
+            client,
+            plan_id,
+            clouds,
+        }
+    }
+
+    fn configured_cloud_for_location(&self, location: &str) -> Option<&ConfiguredCloud> {
+        self.clouds
+            .iter()
+            .find(|configured| configured.cloud.matches_location(location))
+    }
+
+    /// Fetch fresh credentials from the catalog's credentials endpoint.
+    async fn fetch(&self, configured: &ConfiguredCloud) -> Result<Vec<CachedEntry>> {
+        let mut request = self.client.request(Method::GET, &configured.endpoint);
+        if let Some(plan_id) = &self.plan_id {
+            request = request.query(&[("planId", plan_id)]);
+        }
+        let request = request.build()?;
+        let response = self.client.query_catalog(request).await?;
+
+        match response.status() {
+            StatusCode::OK => {
+                let parsed: LoadCredentialsResponse = response.json().await?;
+                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()
+            }
+            _ => Err(deserialize_unexpected_catalog_error(
+                response,
+                self.client.disable_header_redaction(),
+            )
+            .await),
+        }
+    }
+
+    async fn refresh_credential(
+        &self,
+        configured: &ConfiguredCloud,
+        path: &str,
+        fallback: Option<CachedEntry>,
+    ) -> Result<StorageCredential> {
+        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)
+            }
+        }

Review Comment:
   `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?



##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,1203 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Refresh of vended storage credentials against a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache, and
+//! the returned [`StorageCredential`] enum lets the storage adapter enforce the
+//! expected backend-specific type. This preserves Java's per-cloud refresh
+//! policy while supporting mixed-cloud tables through a resolving FileIO.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant. To
+//! add a backend, first add its credential type to Iceberg's storage API and
+//! teach the storage adapter to consume it. Then write its `parse_*` function,
+//! add a `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+    AWS_REFRESH_CREDENTIALS_ENABLED, AWS_REFRESH_CREDENTIALS_ENDPOINT,
+    GCS_REFRESH_CREDENTIALS_ENABLED, GCS_REFRESH_CREDENTIALS_ENDPOINT, GCS_TOKEN,
+    GCS_TOKEN_EXPIRES_AT, GcsCredential, S3_ACCESS_KEY_ID, S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN,
+    S3_SESSION_TOKEN_EXPIRES_AT_MS, S3Credential, StorageCredential, StorageCredentialKind,
+    StorageCredentialProvider,
+};
+use iceberg::{Error, ErrorKind, Result};
+use rand::Rng;
+use reqwest::{Method, StatusCode, Url};
+use tokio::sync::Mutex;
+
+use crate::REST_CATALOG_PROP_SCAN_PLAN_ID;
+use crate::client::{HttpClient, deserialize_unexpected_catalog_error};
+use crate::types::LoadCredentialsResponse;
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// 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 {
+    /// Location URL schemes this backend serves.
+    schemes: &'static [&'static str],
+    /// Table property naming the refresh endpoint (absolute or catalog-relative).
+    endpoint_key: &'static str,
+    /// Table property to opt out; refresh is enabled unless this is `"false"`.
+    enabled_key: &'static str,
+    /// Whether to jitter successful prefetch times like AWS `CachedSupplier`.
+    jitter_prefetch: bool,
+    /// Parse a complete credential from catalog-supplied properties.
+    parse_credential:
+        fn(config: &HashMap<String, String>, prefix: Option<String>) -> Result<StorageCredential>,
+}
+
+impl CloudRefresh {
+    /// S3 / AWS
+    const AWS: Self = Self {
+        schemes: &["s3", "s3a", "s3n"],
+        endpoint_key: AWS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: AWS_REFRESH_CREDENTIALS_ENABLED,
+        jitter_prefetch: true,
+        parse_credential: parse_s3_credential,
+    };
+    /// Google Cloud Storage
+    const GCP: Self = Self {
+        schemes: &["gs", "gcs"],
+        endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED,
+        jitter_prefetch: false,
+        parse_credential: parse_gcs_credential,
+    };
+    // TODO: Azure (ADLS) is not yet supported: opendal 0.57's Azdls builder exposes no
+    // custom credential-provider hook, and reqsign's SAS-token credential has no
+    // expiry, so reqsign-based refresh isn't possible.
+
+    /// Backends with refresh support
+    const SUPPORTED: &[Self] = &[Self::AWS, Self::GCP];
+
+    /// The backend that serves `location`, by its URL scheme, or `None` if no
+    /// supported backend matches (in which case static credentials are used
+    /// as-is, as before).
+    fn for_location(location: &str) -> Option<&'static Self> {
+        Self::SUPPORTED
+            .iter()
+            .find(|cloud| cloud.matches_location(location))
+    }
+
+    fn matches_location(&self, location: &str) -> bool {
+        scheme_of(location).is_some_and(|scheme| self.schemes.contains(&scheme.as_str()))
+    }
+}
+
+/// Re-fetch a credential once it is within this window of expiry, so a fresh
+/// token is in hand before the object store would reject the old one.
+const REFRESH_BUFFER: Duration = Duration::from_mins(5);
+
+/// AWS keeps at least one minute between its jittered prefetch time and expiry.
+const MIN_REFRESH_BUFFER: Duration = Duration::from_mins(1);
+
+/// Initial ceiling for failure backoff. Equal jitter chooses from half this
+/// value through the full value.
+const INITIAL_FAILURE_BACKOFF: Duration = Duration::from_secs(1);
+
+/// Maximum failure backoff while a cached credential remains usable.
+const MAX_FAILURE_BACKOFF: Duration = Duration::from_secs(30);
+
+/// A cached vended credential and its refresh schedule.
+#[derive(Clone)]
+struct CachedEntry {
+    credential: StorageCredential,
+    /// When this entry becomes eligible for prefetch. `None` means it does not
+    /// expire and therefore never needs proactive refresh.
+    refresh_at: Option<SystemTime>,
+}
+
+impl CachedEntry {
+    fn new(credential: StorageCredential, jitter_prefetch: bool) -> Self {
+        let refresh_at = credential
+            .expires_at
+            .map(|expires_at| prefetch_time(expires_at, jitter_prefetch));
+        Self {
+            credential,
+            refresh_at,
+        }
+    }
+
+    /// Seed entries that are already inside the nominal five-minute window are
+    /// immediately due. Otherwise AWS applies the same jitter as it does to a
+    /// freshly fetched value.
+    fn seed(credential: StorageCredential, jitter_prefetch: bool) -> Self {
+        let due = credential.expires_at.is_some_and(|expires_at| {
+            SystemTime::now()
+                .checked_add(REFRESH_BUFFER)
+                .is_none_or(|refresh_boundary| refresh_boundary >= expires_at)
+        });
+        let mut entry = Self::new(credential, jitter_prefetch);
+        if due {
+            entry.refresh_at = Some(UNIX_EPOCH);
+        }
+        entry
+    }
+
+    fn is_fresh(&self, now: SystemTime) -> bool {
+        self.refresh_at.is_none_or(|refresh_at| now < refresh_at)
+    }
+
+    fn is_unexpired(&self, now: SystemTime) -> bool {
+        self.credential
+            .expires_at
+            .is_none_or(|expires_at| now < expires_at)
+    }
+}
+
+/// Cached credentials plus failure-backoff state.
+struct CacheState {
+    entries: Vec<CachedEntry>,
+    consecutive_failures: u32,
+    retry_not_before: Option<Instant>,
+}
+
+struct ConfiguredCloud {
+    cloud: &'static CloudRefresh,
+    endpoint: String,
+    cache: Mutex<CacheState>,
+    /// Only one caller fetches at a time. The cache lock is deliberately
+    /// separate so other callers can keep using an unexpired credential while
+    /// the refresh is in flight.
+    refresh: Mutex<()>,
+}
+
+/// Fetches and refreshes vended credentials from a REST catalog's table
+/// credentials endpoint.
+///
+/// Each cloud cache is seeded with the credential from the initial table
+/// properties (when complete) and re-fetched from its endpoint as it
+/// nears expiry.
+pub(crate) struct RestVendedCredentialProvider {
+    client: Arc<HttpClient>,
+    /// Optional scan-plan identifier.
+    plan_id: Option<String>,
+    /// Independently configured endpoint and cache for each backing cloud.
+    clouds: Vec<ConfiguredCloud>,
+}
+
+impl RestVendedCredentialProvider {
+    fn new(client: Arc<HttpClient>, plan_id: Option<String>, clouds: Vec<ConfiguredCloud>) -> Self {
+        Self {
+            client,
+            plan_id,
+            clouds,
+        }
+    }
+
+    fn configured_cloud_for_location(&self, location: &str) -> Option<&ConfiguredCloud> {
+        self.clouds
+            .iter()
+            .find(|configured| configured.cloud.matches_location(location))
+    }
+
+    /// Fetch fresh credentials from the catalog's credentials endpoint.
+    async fn fetch(&self, configured: &ConfiguredCloud) -> Result<Vec<CachedEntry>> {
+        let mut request = self.client.request(Method::GET, &configured.endpoint);
+        if let Some(plan_id) = &self.plan_id {
+            request = request.query(&[("planId", plan_id)]);
+        }
+        let request = request.build()?;
+        let response = self.client.query_catalog(request).await?;
+
+        match response.status() {
+            StatusCode::OK => {
+                let parsed: LoadCredentialsResponse = response.json().await?;
+                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()

Review Comment:
   `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?



##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,107 @@ pub trait StorageFactory: Debug + Send + Sync {
     /// A `Result` containing an `Arc<dyn Storage>` on success, or an error
     /// if the storage could not be created.
     fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>>;
+
+    /// Build a new Storage instance, optionally supplying a credential provider
+    /// that the backend can call to obtain and refresh short-lived credentials.
+    fn build_with_credentials(
+        &self,
+        config: &StorageConfig,
+        credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
+    ) -> Result<Arc<dyn Storage>> {
+        if credential_provider.is_some() {
+            return Err(Error::new(
+                ErrorKind::FeatureUnsupported,
+                "Storage factory does not support refreshable credential providers",
+            ));
+        }
+
+        self.build(config)
+    }
+}
+
+/// Supplies fresh, backend-specific storage credentials on demand.
+///
+/// A catalog that vends temporary credentials implements this trait so that
+/// storage backends can re-fetch credentials as they approach expiry instead
+/// of failing once the initial token's TTL runs out.
+///
+/// # Caching
+///
+/// [`load_credential`](Self::load_credential) may be called very frequently —
+/// the S3 backend, for example, rebuilds its operator (and therefore its
+/// signer) on every file operation. Implementations must cache internally and
+/// only re-fetch when the current credential is at or near expiry; otherwise
+/// every object-store request would trigger a call back to the catalog.
+#[async_trait]
+pub trait StorageCredentialProvider: Debug + Send + Sync {
+    /// Return whether this provider has refresh configuration for `path`.
+    ///
+    /// Backends use this before replacing their normal credential chain. The
+    /// default is `true` for single-backend providers; multi-backend providers
+    /// should return `false` for schemes they do not configure.
+    fn supports_path(&self, _path: &str) -> bool {
+        true
+    }
+
+    /// Load a fresh credential for the storage location identified by `path`.
+    ///
+    /// `path` is the absolute location being accessed (e.g.
+    /// `s3://bucket/warehouse/db/table/...`). Providers that vend distinct
+    /// credentials per location prefix use it to select the most specific
+    /// match. When the selected credential has a declared
+    /// [`StorageCredential::prefix`], it must cover `path`.
+    async fn load_credential(&self, path: &str) -> Result<StorageCredential>;
+}
+
+/// A vended storage credential together with its scope and expiry.
+#[derive(Clone, Debug)]
+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,
+}

Review Comment:
   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?



##########
crates/catalog/rest/src/client.rs:
##########
@@ -278,29 +298,30 @@ pub(crate) async fn deserialize_catalog_response<R: DeserializeOwned>(
     let bytes = response.bytes().await?;
 
     serde_json::from_slice::<R>(&bytes).map_err(|e| {
+        // Successful REST responses can contain OAuth tokens and delegated
+        // storage credentials. Never copy an unparsable response into an error.
         Error::new(
             ErrorKind::Unexpected,
             "Failed to parse response from rest catalog server",
         )
-        .with_context("json", String::from_utf8_lossy(&bytes))
         .with_source(e)
     })
 }
 
-/// Headers that contain sensitive information and should be excluded from logs.
-const SENSITIVE_HEADERS: &[&str] = &[
-    "authorization",
-    "proxy-authorization",
-    "set-cookie",
-    "cookie",
-    "x-api-key",
-    "x-auth-token",
-];
-
-/// Returns true if the header name is considered sensitive.
+/// Returns true if the header may carry a secret.
 fn is_sensitive_header(name: &str) -> bool {
     let name_lower = name.to_lowercase();
-    SENSITIVE_HEADERS.iter().any(|h| name_lower == *h)
+    [
+        "auth",
+        "token",
+        "secret",
+        "key",
+        "password",
+        "cookie",
+        "credential",
+    ]

Review Comment:
   Given your own note that Java redacts every header regardless of sensitivity, that seems like the simpler and more conservative choice here too. It avoids maintaining a keyword list that can both over-match (a header like `x-auth-region` gets redacted for no reason) and under-match (a secret header whose name does not contain any of the chosen substrings). Suggest switching to redact-everything-by-default unless there is a concrete case where seeing an unredacted non-sensitive header in error logs matters.



##########
crates/storage/opendal/src/lib.rs:
##########
@@ -553,6 +611,12 @@ impl Storage for OpenDalStorage {
         let mut deleters: HashMap<String, opendal::Deleter> = HashMap::new();
 
         while let Some(path) = paths.next().await {
+            if self.uses_dynamic_credentials(&path) {
+                let (op, relative_path) = self.create_operator(&path)?;
+                op.delete(relative_path).await.map_err(from_opendal_error)?;
+                continue;
+            }

Review Comment:
   Agree this should be fixed in this PR rather than deferred. See finding 1 above for the specific fix and the added detail on `create_operator` also being rebuilt per file.



##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,1193 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Refresh of vended storage credentials against a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache, and
+//! the returned [`StorageCredential`] enum lets the storage adapter enforce the
+//! expected backend-specific type. This preserves Java's per-cloud refresh
+//! policy while supporting mixed-cloud tables through a resolving FileIO.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant. To
+//! add a backend, first add its credential type to Iceberg's storage API and
+//! teach the storage adapter to consume it. Then write its `parse_*` function,
+//! add a `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+    AWS_REFRESH_CREDENTIALS_ENABLED, AWS_REFRESH_CREDENTIALS_ENDPOINT,
+    GCS_REFRESH_CREDENTIALS_ENABLED, GCS_REFRESH_CREDENTIALS_ENDPOINT, GCS_TOKEN,
+    GCS_TOKEN_EXPIRES_AT, GcsCredential, S3_ACCESS_KEY_ID, S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN,
+    S3_SESSION_TOKEN_EXPIRES_AT_MS, S3Credential, StorageCredential, StorageCredentialKind,
+    StorageCredentialProvider,
+};
+use iceberg::{Error, ErrorKind, Result};
+use rand::Rng;
+use reqwest::{Method, StatusCode, Url};
+use tokio::sync::Mutex;
+
+use crate::REST_CATALOG_PROP_SCAN_PLAN_ID;
+use crate::client::{HttpClient, deserialize_unexpected_catalog_error};
+use crate::types::LoadCredentialsResponse;
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// 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 {
+    /// Location URL schemes this backend serves.
+    schemes: &'static [&'static str],
+    /// Table property naming the refresh endpoint (absolute or catalog-relative).
+    endpoint_key: &'static str,
+    /// Table property to opt out; refresh is enabled unless this is `"false"`.
+    enabled_key: &'static str,
+    /// Whether to jitter successful prefetch times like AWS `CachedSupplier`.
+    jitter_prefetch: bool,
+    /// Parse a complete credential from catalog-supplied properties.
+    parse_credential: fn(&HashMap<String, String>) -> Result<StorageCredential>,
+}
+
+impl CloudRefresh {
+    /// S3 / AWS
+    const AWS: Self = Self {
+        schemes: &["s3", "s3a", "s3n"],
+        endpoint_key: AWS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: AWS_REFRESH_CREDENTIALS_ENABLED,
+        jitter_prefetch: true,
+        parse_credential: parse_s3_credential,
+    };
+    /// Google Cloud Storage
+    const GCP: Self = Self {
+        schemes: &["gs", "gcs"],
+        endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED,
+        jitter_prefetch: false,

Review Comment:
   Checked this against Java directly. Java's scheduled refresh (`S3FileIO`/`GCSFileIO.refreshStorageCredentials()`) never retries after a failed fetch at all, it logs a warning and background refresh permanently stops for that `FileIO` instance until something else rebuilds the client map. The jittered backoff-and-retry here is already strictly better than Java's behavior, not something that needs to move toward parity. Might be worth saying so explicitly in the module doc comment, since it currently frames the design as "preserving Java's per-cloud refresh policy," and a reader comparing against Java will notice the retry model does not match 1:1.



##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -541,9 +556,20 @@ impl RestCatalog {
                 )
             })?;
 
-        let file_io = FileIOBuilder::new(factory).with_props(props).build();
+        // If the catalog vends refreshable credentials for this table's storage,
+        // attach a provider so the backend re-fetches them before they expire.
+        let credential_provider = crate::credential::build_vended_credential_provider(

Review Comment:
   The file already has a `use crate::client::{...}` / `use crate::types::{...}` block at the top; this call could join it as `use crate::credential::build_vended_credential_provider;` instead of qualifying the path inline at the call site.



##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,103 @@ pub trait StorageFactory: Debug + Send + Sync {
     /// A `Result` containing an `Arc<dyn Storage>` on success, or an error
     /// if the storage could not be created.
     fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>>;
+
+    /// Build a new Storage instance, optionally supplying a credential provider
+    /// that the backend can call to obtain and refresh short-lived credentials.
+    fn build_with_credentials(
+        &self,
+        config: &StorageConfig,
+        credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
+    ) -> Result<Arc<dyn Storage>> {
+        if credential_provider.is_some() {
+            return Err(Error::new(
+                ErrorKind::FeatureUnsupported,
+                "Storage factory does not support refreshable credential providers",
+            ));
+        }
+
+        self.build(config)
+    }
+}
+
+/// Supplies fresh, backend-specific storage credentials on demand.
+///
+/// A catalog that vends temporary credentials implements this trait so that
+/// storage backends can re-fetch credentials as they approach expiry instead
+/// of failing once the initial token's TTL runs out.
+///
+/// # Caching
+///
+/// [`load_credential`](Self::load_credential) may be called very frequently —
+/// the S3 backend, for example, rebuilds its operator (and therefore its
+/// signer) on every file operation. Implementations must cache internally and
+/// only re-fetch when the current credential is at or near expiry; otherwise
+/// every object-store request would trigger a call back to the catalog.
+#[async_trait]
+pub trait StorageCredentialProvider: Debug + Send + Sync {

Review Comment:
   The trait-based design seems fine, Java not having an equivalent is not a reason to avoid one here. See finding 5 above for the separate, concrete question on the same code (pub fields, no constructor).



##########
crates/catalog/rest/src/credential.rs:
##########
@@ -0,0 +1,1203 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Refresh of vended storage credentials against a REST catalog.
+//!
+//! A REST catalog can vend short-lived storage credentials whose lifetime the
+//! client does not control. [`RestVendedCredentialProvider`] implements the
+//! core [`StorageCredentialProvider`] trait so storage backends re-fetch those
+//! credentials from the catalog's table credentials endpoint before they
+//! expire, keeping long-running jobs authenticated instead of failing with a
+//! `403` once the initial token's TTL elapses.
+//!
+//! Unlike the Java client, which has one provider per cloud SDK, this is a
+//! single backend-agnostic provider with an independent endpoint and cache for
+//! each configured cloud. The path being accessed selects the cloud cache, and
+//! the returned [`StorageCredential`] enum lets the storage adapter enforce the
+//! expected backend-specific type. This preserves Java's per-cloud refresh
+//! policy while supporting mixed-cloud tables through a resolving FileIO.
+//!
+//! # Adding a cloud
+//!
+//! The refresh policy for each cloud lives in one [`CloudRefresh`] constant. To
+//! add a backend, first add its credential type to Iceberg's storage API and
+//! teach the storage adapter to consume it. Then write its `parse_*` function,
+//! add a `CloudRefresh` constant, and list it in [`CloudRefresh::SUPPORTED`].
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use async_trait::async_trait;
+use iceberg::io::{
+    AWS_REFRESH_CREDENTIALS_ENABLED, AWS_REFRESH_CREDENTIALS_ENDPOINT,
+    GCS_REFRESH_CREDENTIALS_ENABLED, GCS_REFRESH_CREDENTIALS_ENDPOINT, GCS_TOKEN,
+    GCS_TOKEN_EXPIRES_AT, GcsCredential, S3_ACCESS_KEY_ID, S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN,
+    S3_SESSION_TOKEN_EXPIRES_AT_MS, S3Credential, StorageCredential, StorageCredentialKind,
+    StorageCredentialProvider,
+};
+use iceberg::{Error, ErrorKind, Result};
+use rand::Rng;
+use reqwest::{Method, StatusCode, Url};
+use tokio::sync::Mutex;
+
+use crate::REST_CATALOG_PROP_SCAN_PLAN_ID;
+use crate::client::{HttpClient, deserialize_unexpected_catalog_error};
+use crate::types::LoadCredentialsResponse;
+
+/// Cloud-specific details regarding vended-credential refresh.
+///
+/// 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 {
+    /// Location URL schemes this backend serves.
+    schemes: &'static [&'static str],
+    /// Table property naming the refresh endpoint (absolute or catalog-relative).
+    endpoint_key: &'static str,
+    /// Table property to opt out; refresh is enabled unless this is `"false"`.
+    enabled_key: &'static str,
+    /// Whether to jitter successful prefetch times like AWS `CachedSupplier`.
+    jitter_prefetch: bool,
+    /// Parse a complete credential from catalog-supplied properties.
+    parse_credential:
+        fn(config: &HashMap<String, String>, prefix: Option<String>) -> Result<StorageCredential>,
+}
+
+impl CloudRefresh {
+    /// S3 / AWS
+    const AWS: Self = Self {
+        schemes: &["s3", "s3a", "s3n"],
+        endpoint_key: AWS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: AWS_REFRESH_CREDENTIALS_ENABLED,
+        jitter_prefetch: true,
+        parse_credential: parse_s3_credential,
+    };
+    /// Google Cloud Storage
+    const GCP: Self = Self {
+        schemes: &["gs", "gcs"],
+        endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT,
+        enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED,
+        jitter_prefetch: false,
+        parse_credential: parse_gcs_credential,
+    };
+    // TODO: Azure (ADLS) is not yet supported: opendal 0.57's Azdls builder exposes no
+    // custom credential-provider hook, and reqsign's SAS-token credential has no
+    // expiry, so reqsign-based refresh isn't possible.
+
+    /// Backends with refresh support
+    const SUPPORTED: &[Self] = &[Self::AWS, Self::GCP];
+
+    /// The backend that serves `location`, by its URL scheme, or `None` if no
+    /// supported backend matches (in which case static credentials are used
+    /// as-is, as before).
+    fn for_location(location: &str) -> Option<&'static Self> {
+        Self::SUPPORTED
+            .iter()
+            .find(|cloud| cloud.matches_location(location))
+    }
+
+    fn matches_location(&self, location: &str) -> bool {
+        scheme_of(location).is_some_and(|scheme| self.schemes.contains(&scheme.as_str()))
+    }
+}
+
+/// Re-fetch a credential once it is within this window of expiry, so a fresh
+/// token is in hand before the object store would reject the old one.
+const REFRESH_BUFFER: Duration = Duration::from_mins(5);
+
+/// AWS keeps at least one minute between its jittered prefetch time and expiry.
+const MIN_REFRESH_BUFFER: Duration = Duration::from_mins(1);
+
+/// Initial ceiling for failure backoff. Equal jitter chooses from half this
+/// value through the full value.
+const INITIAL_FAILURE_BACKOFF: Duration = Duration::from_secs(1);
+
+/// Maximum failure backoff while a cached credential remains usable.
+const MAX_FAILURE_BACKOFF: Duration = Duration::from_secs(30);
+
+/// A cached vended credential and its refresh schedule.
+#[derive(Clone)]
+struct CachedEntry {
+    credential: StorageCredential,
+    /// When this entry becomes eligible for prefetch. `None` means it does not
+    /// expire and therefore never needs proactive refresh.
+    refresh_at: Option<SystemTime>,
+}
+
+impl CachedEntry {
+    fn new(credential: StorageCredential, jitter_prefetch: bool) -> Self {
+        let refresh_at = credential
+            .expires_at
+            .map(|expires_at| prefetch_time(expires_at, jitter_prefetch));
+        Self {
+            credential,
+            refresh_at,
+        }
+    }
+
+    /// Seed entries that are already inside the nominal five-minute window are
+    /// immediately due. Otherwise AWS applies the same jitter as it does to a
+    /// freshly fetched value.
+    fn seed(credential: StorageCredential, jitter_prefetch: bool) -> Self {
+        let due = credential.expires_at.is_some_and(|expires_at| {
+            SystemTime::now()
+                .checked_add(REFRESH_BUFFER)
+                .is_none_or(|refresh_boundary| refresh_boundary >= expires_at)
+        });
+        let mut entry = Self::new(credential, jitter_prefetch);
+        if due {
+            entry.refresh_at = Some(UNIX_EPOCH);
+        }
+        entry
+    }
+
+    fn is_fresh(&self, now: SystemTime) -> bool {
+        self.refresh_at.is_none_or(|refresh_at| now < refresh_at)
+    }
+
+    fn is_unexpired(&self, now: SystemTime) -> bool {
+        self.credential
+            .expires_at
+            .is_none_or(|expires_at| now < expires_at)
+    }
+}
+
+/// Cached credentials plus failure-backoff state.
+struct CacheState {
+    entries: Vec<CachedEntry>,
+    consecutive_failures: u32,
+    retry_not_before: Option<Instant>,
+}
+
+struct ConfiguredCloud {
+    cloud: &'static CloudRefresh,
+    endpoint: String,
+    cache: Mutex<CacheState>,
+    /// Only one caller fetches at a time. The cache lock is deliberately
+    /// separate so other callers can keep using an unexpired credential while
+    /// the refresh is in flight.
+    refresh: Mutex<()>,
+}
+
+/// Fetches and refreshes vended credentials from a REST catalog's table
+/// credentials endpoint.
+///
+/// Each cloud cache is seeded with the credential from the initial table
+/// properties (when complete) and re-fetched from its endpoint as it
+/// nears expiry.
+pub(crate) struct RestVendedCredentialProvider {
+    client: Arc<HttpClient>,
+    /// Optional scan-plan identifier.
+    plan_id: Option<String>,
+    /// Independently configured endpoint and cache for each backing cloud.
+    clouds: Vec<ConfiguredCloud>,
+}
+
+impl RestVendedCredentialProvider {
+    fn new(client: Arc<HttpClient>, plan_id: Option<String>, clouds: Vec<ConfiguredCloud>) -> Self {
+        Self {
+            client,
+            plan_id,
+            clouds,
+        }
+    }
+
+    fn configured_cloud_for_location(&self, location: &str) -> Option<&ConfiguredCloud> {
+        self.clouds
+            .iter()
+            .find(|configured| configured.cloud.matches_location(location))
+    }
+
+    /// Fetch fresh credentials from the catalog's credentials endpoint.
+    async fn fetch(&self, configured: &ConfiguredCloud) -> Result<Vec<CachedEntry>> {
+        let mut request = self.client.request(Method::GET, &configured.endpoint);
+        if let Some(plan_id) = &self.plan_id {
+            request = request.query(&[("planId", plan_id)]);
+        }
+        let request = request.build()?;
+        let response = self.client.query_catalog(request).await?;
+
+        match response.status() {
+            StatusCode::OK => {
+                let parsed: LoadCredentialsResponse = response.json().await?;
+                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()
+            }
+            _ => Err(deserialize_unexpected_catalog_error(
+                response,
+                self.client.disable_header_redaction(),
+            )
+            .await),
+        }
+    }
+
+    async fn refresh_credential(
+        &self,
+        configured: &ConfiguredCloud,
+        path: &str,
+        fallback: Option<CachedEntry>,
+    ) -> Result<StorageCredential> {
+        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)
+            }
+        }
+    }
+}
+
+impl std::fmt::Debug for RestVendedCredentialProvider {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("RestVendedCredentialProvider")
+            .field("configured_clouds", &self.clouds.len())
+            .finish_non_exhaustive()
+    }
+}
+
+enum CacheDecision {
+    Use(StorageCredential),
+    Refresh(Option<CachedEntry>),
+}
+
+async fn cache_decision(configured: &ConfiguredCloud, path: &str) -> CacheDecision {
+    let cache = configured.cache.lock().await;
+    let current = longest_prefix_match(&cache.entries, path).cloned();
+    let now = SystemTime::now();
+
+    if let Some(entry) = current.as_ref().filter(|entry| entry.is_fresh(now)) {
+        return CacheDecision::Use(entry.credential.clone());
+    }
+
+    if cache
+        .retry_not_before
+        .is_some_and(|retry_at| Instant::now() < retry_at)
+        && let Some(entry) = current.as_ref().filter(|entry| entry.is_unexpired(now))
+    {
+        return CacheDecision::Use(entry.credential.clone());
+    }
+
+    CacheDecision::Refresh(current)
+}
+
+#[async_trait]
+impl StorageCredentialProvider for RestVendedCredentialProvider {
+    fn supports_path(&self, path: &str) -> bool {
+        self.configured_cloud_for_location(path).is_some()
+    }
+
+    async fn load_credential(&self, path: &str) -> Result<StorageCredential> {
+        if CloudRefresh::for_location(path).is_none() {
+            return Err(Error::new(
+                ErrorKind::FeatureUnsupported,
+                format!("no credential refresh implementation for storage location: {path}"),
+            ));
+        }
+        let configured = self.configured_cloud_for_location(path).ok_or_else(|| {
+            Error::new(
+                ErrorKind::FeatureUnsupported,
+                format!("credential refresh is not configured for storage location: {path}"),
+            )
+        })?;
+
+        let current = match cache_decision(configured, path).await {
+            CacheDecision::Use(credential) => return Ok(credential),
+            CacheDecision::Refresh(current) => current,
+        };
+
+        // One caller refreshes, while concurrent callers immediately keep using the
+        // unexpired cached credential. With no usable credential, callers wait
+        // for the in-flight refresh instead.
+        let usable = current
+            .as_ref()
+            .filter(|entry| entry.is_unexpired(SystemTime::now()));
+        let _refresh_guard = if let Some(entry) = usable {
+            match configured.refresh.try_lock() {
+                Ok(guard) => guard,
+                Err(_) => return Ok(entry.credential.clone()),
+            }
+        } else {
+            configured.refresh.lock().await
+        };
+
+        // Another caller may have completed a refresh between our cache check
+        // and acquiring the single-flight guard.
+        let current = match cache_decision(configured, path).await {
+            CacheDecision::Use(credential) => return Ok(credential),
+            CacheDecision::Refresh(current) => current,
+        };
+
+        self.refresh_credential(configured, path, current).await
+    }
+}
+
+/// Select the credential whose prefix is the longest match for `path`.
+fn longest_prefix_match<'a>(entries: &'a [CachedEntry], path: &str) -> Option<&'a CachedEntry> {
+    entries
+        .iter()
+        .filter(|entry| {
+            entry
+                .credential
+                .prefix
+                .as_deref()
+                .is_none_or(|prefix| path.starts_with(prefix))
+        })
+        .max_by_key(|entry| entry.credential.prefix.as_deref().map_or(0, str::len))
+}
+
+/// Compute a successful credential's prefetch time.
+fn prefetch_time(expires_at: SystemTime, jitter: bool) -> SystemTime {
+    let base = expires_at.checked_sub(REFRESH_BUFFER).unwrap_or(UNIX_EPOCH);
+    if !jitter {
+        return base;
+    }
+
+    let jitter_window = REFRESH_BUFFER.saturating_sub(MIN_REFRESH_BUFFER);
+    let jitter_millis = rand::rng().random_range(0..jitter_window.as_millis() as u64);
+    base.checked_add(Duration::from_millis(jitter_millis))
+        .unwrap_or(base)
+}
+
+/// Equal-jitter exponential backoff. The random lower half avoids both hot
+/// retry loops and synchronized retries across clients.
+fn failure_backoff(consecutive_failures: u32) -> Duration {
+    let exponent = consecutive_failures.saturating_sub(1).min(5);
+    let ceiling = INITIAL_FAILURE_BACKOFF
+        .checked_mul(1 << exponent)
+        .unwrap_or(MAX_FAILURE_BACKOFF)
+        .min(MAX_FAILURE_BACKOFF);
+    let ceiling_millis = ceiling.as_millis() as u64;
+    let floor_millis = ceiling_millis / 2;
+    Duration::from_millis(rand::rng().random_range(floor_millis..=ceiling_millis))
+}
+
+/// Build a credential provider from a table's properties,
+/// or `None` when no supported cloud advertises an enabled refresh endpoint.
+///
+/// `base_uri` is the catalog URI, used to resolve a relative endpoint.
+pub(crate) fn build_vended_credential_provider(
+    client: Arc<HttpClient>,
+    base_uri: &str,
+    props: &HashMap<String, String>,
+) -> Result<Option<Arc<dyn StorageCredentialProvider>>> {
+    let clouds = CloudRefresh::SUPPORTED
+        .iter()
+        .filter_map(|cloud| {
+            // Refresh is enabled by default and invalid booleans disable refresh.
+            let enabled = props
+                .get(cloud.enabled_key)
+                .is_none_or(|value| value.parse().unwrap_or(false));

Review Comment:
   `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.



##########
crates/catalog/rest/src/client.rs:
##########
@@ -71,6 +75,24 @@ impl HttpClient {
         })
     }
 
+    /// Create a client for table-scoped resources while reusing this client's
+    /// underlying connection pool.
+    ///
+    /// A load-table response may supply a table token or `header.*` values that
+    /// must be used for subsequent table requests such as credential refresh.
+    pub(crate) fn for_table(
+        &self,
+        catalog_uri: &str,
+        props: HashMap<String, String>,
+    ) -> Result<Self> {
+        let cfg = RestCatalogConfig::builder()
+            .uri(catalog_uri.to_string())
+            .props(props)
+            .client(Some(self.client.clone()))
+            .build();
+        Self::new(&cfg)
+    }

Review Comment:
   `for_table` builds a brand-new `HttpClient` via `HttpClient::new(&cfg)`, which starts `token: Mutex::new(cfg.token())`. `cfg.token()` reads the `token` property directly — it does not inherit the parent client's already-cached bearer token. When the catalog authenticates via OAuth client-credentials (`credential` prop, no static `token` prop), the table-scoped client has no seeded token and calls `exchange_credential_for_token()` itself on first use. Since `build_vended_credential_provider` (and therefore `for_table`) is constructed fresh inside `load_file_io`, and `load_file_io` runs on every `load_table`/`create_table`/`register_table`, every one of those calls does its own OAuth token exchange, independent of whatever token the parent `RestContext.client` already holds. For a query engine that reloads tables frequently, that's an extra round trip to the OAuth endpoint per load. Is reusing the parent's cached token (when the table doesn't vend its own) worth doing here?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r4185367533


##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,358 @@ pub trait StorageFactory: Debug + Send + Sync {
     /// A `Result` containing an `Arc<dyn Storage>` on success, or an error
     /// if the storage could not be created.
     fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>>;
+
+    /// Build a new Storage instance, optionally supplying a credential provider
+    /// that the backend can call to obtain and refresh short-lived credentials.
+    ///
+    /// Backends that cannot use the provider ignore it and use the credentials
+    /// in `config`, as they would without one. The default does exactly that.
+    #[allow(unused_variables)]
+    fn build_with_credentials(
+        &self,
+        config: &StorageConfig,
+        credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
+    ) -> Result<Arc<dyn Storage>> {
+        self.build(config)
+    }
+}
+
+/// Supplies fresh, backend-specific storage credentials on demand.
+///
+/// A catalog that vends temporary credentials implements this trait so that
+/// storage backends can re-fetch credentials as they approach expiry instead
+/// of failing once the initial token's TTL runs out.
+///
+/// # Caching
+///
+/// [`load_credential`](Self::load_credential) may be called very frequently.
+/// Implementations must cache internally and only re-fetch when the current
+/// credential is at or near expiry; otherwise every object-store request could
+/// trigger a call back to the catalog.
+#[async_trait]
+pub trait StorageCredentialProvider: Debug + Send + Sync {
+    /// Return whether this provider has refresh configuration for `path`.
+    ///
+    /// Backends use this before replacing their normal credential chain. The
+    /// default is `true` for single-backend providers; multi-backend providers
+    /// should return `false` for schemes they do not configure.
+    fn supports_path(&self, _path: &str) -> bool {
+        true
+    }
+
+    /// Load a fresh credential for the storage location identified by `path`.
+    ///
+    /// `path` is the absolute location being accessed (e.g.
+    /// `s3://bucket/warehouse/db/table/...`). Providers that vend distinct
+    /// credentials per location prefix use it to select the most specific
+    /// match. When the selected credential has a declared
+    /// [`StorageCredential::prefix`], it must [cover](StorageCredential::covers) `path`.
+    async fn load_credential(&self, path: &str) -> Result<StorageCredential>;
+
+    /// Return a factory that rebuilds an equivalent provider in another process.
+    ///
+    /// [`FileIO::serialize_all`](crate::io::FileIO::serialize_all) serializes this
+    /// factory in place of the provider. The default reports that the provider
+    /// cannot be serialized.
+    fn factory(&self) -> Result<Arc<dyn StorageCredentialProviderFactory>> {
+        Err(Error::new(
+            ErrorKind::FeatureUnsupported,
+            "storage credential provider cannot be serialized",
+        ))
+    }
+}
+
+/// Serializable recipe that rebuilds a [`StorageCredentialProvider`] after
+/// [`FileIO`](crate::io::FileIO) deserialization.
+///
+/// Factories are serialized through [`typetag`](https://docs.rs/typetag), so
+/// implementations must use `#[typetag::serde]`, and the receiving binary must
+/// link the concrete implementation.
+#[typetag::serde(tag = "type")]
+pub trait StorageCredentialProviderFactory: Debug + Send + Sync {
+    /// Build a provider for a `FileIO` with the given storage configuration.
+    fn build(&self, config: &StorageConfig) -> Result<Arc<dyn StorageCredentialProvider>>;
+}
+
+/// A vended storage credential together with its scope and expiry.
+#[derive(Clone, Debug)]
+pub struct StorageCredential {
+    /// Storage-location prefix this credential is scoped to. `None` represents a
+    /// credential without a declared scope, sourced from flat storage properties.
+    prefix: Option<String>,
+    /// The backend-specific credential material.
+    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.
+    expires_at: Option<SystemTime>,
+}
+
+impl StorageCredential {
+    /// Create a storage credential with no declared scope or expiration.
+    pub fn new(kind: StorageCredentialKind) -> Self {
+        Self {
+            prefix: None,
+            kind,
+            expires_at: None,
+        }
+    }
+
+    /// Set the storage-location prefix this credential is scoped to.
+    pub fn with_prefix(mut self, prefix: impl Into<String>) -> Self {
+        self.prefix = Some(prefix.into());
+        self
+    }
+
+    /// Set when this credential expires.
+    pub fn with_expiration(mut self, expires_at: SystemTime) -> Self {
+        self.expires_at = Some(expires_at);
+        self
+    }
+
+    /// Return the storage-location prefix this credential is scoped to.
+    pub fn prefix(&self) -> Option<&str> {
+        self.prefix.as_deref()
+    }
+
+    /// Return whether this credential applies to `location`.
+    ///
+    /// A credential without a prefix covers every location. Otherwise the
+    /// prefix must match whole path segments of `location`, and scheme
+    /// aliases (`s3a`/`s3n` for `s3`, `gcs` for `gs`, and the plain-text
+    /// Azure schemes for their TLS variants) are treated as equal. A prefix
+    /// that is only a scheme, such as `s3`, covers every location with that
+    /// scheme.
+    pub fn covers(&self, location: &str) -> bool {
+        self.prefix
+            .as_deref()
+            .is_none_or(|prefix| storage_prefix_covers(prefix, location))
+    }
+
+    /// Return the backend-specific credential material.
+    pub fn kind(&self) -> &StorageCredentialKind {
+        &self.kind
+    }
+
+    /// Consume this credential and return its backend-specific material.
+    pub fn into_kind(self) -> StorageCredentialKind {
+        self.kind
+    }
+
+    /// Return when this credential expires.
+    pub fn expires_at(&self) -> Option<SystemTime> {
+        self.expires_at
+    }
+}
+
+/// Return whether the storage-location `prefix` covers `location`, with the
+/// matching rules of [`StorageCredential::covers`].
+pub fn storage_prefix_covers(prefix: &str, location: &str) -> bool {
+    let Some((location_scheme, location_rest)) = location.split_once("://") else {
+        return false;
+    };
+    let Some((prefix_scheme, prefix_rest)) = prefix.split_once("://") else {
+        return !prefix.is_empty() && canonical_scheme(prefix) == canonical_scheme(location_scheme);
+    };
+
+    canonical_scheme(prefix_scheme) == canonical_scheme(location_scheme)
+        && location_rest
+            .strip_prefix(prefix_rest)
+            .is_some_and(|remainder| {
+                prefix_rest.is_empty()
+                    || prefix_rest.ends_with('/')
+                    || remainder.is_empty()
+                    || remainder.starts_with('/')
+            })
+}

Review Comment:
   This was intentional as whilst the spec doesn't mention segment boundaries, I thought it was in the spirit of it. If we want to follow Java's semantics (which is a good enough reason by itself), then I'd be happy to change this.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["zakariya-s (via GitHub)" <gi...@apache.org>]

zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r4185397272


##########
crates/storage/opendal/src/lib.rs:
##########
@@ -218,6 +233,7 @@ fn default_memory_operator() -> Operator {
 
 /// OpenDAL-based storage implementation.
 #[derive(Clone, Debug, Serialize, Deserialize)]
+#[non_exhaustive]

Review Comment:
   Yes will do in the relevant split PR



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [PR] feat(rest): Support refreshing vended storage credentials [iceberg-rust]

["mbutrovich (via GitHub)" <gi...@apache.org>]

mbutrovich commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r4136989403


##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -879,9 +873,29 @@ impl RestSessionCatalog {
                 )
             })?;
 
-        let file_io = FileIOBuilder::new(factory).with_props(props).build();
+        // If the catalog vends refreshable credentials for this table's storage,
+        // attach a provider so the backend re-fetches them before they expire.
+        // Only catalog authentication resolved from the properties can be
+        // rebuilt after FileIO serialization.
+        let credential_provider = build_vended_credential_provider(
+            &client.http_client,
+            client.auth_manager.as_ref(),
+            RestVendedCredentialProviderFactory::new(
+                &client.config.uri,
+                table.clone(),
+                table_config.unwrap_or_default(),
+            ),
+            &props,
+            self.auth_manager.is_none(),
+        )

Review Comment:
   `load_file_io` seeds the provider only from the flat `config` properties. At the head commit nothing reads `LoadTableResult::storage_credentials` (`grep -rn storage_credentials crates` only finds `types.rs`). The spec says "Clients must first check whether the respective credentials exist in the `storage-credentials` field before checking the `config` for credentials" ([`LoadTableResult`](https://github.com/apache/iceberg/blob/48330b8dacab6662242d252b39c8444190979bb2/open-api/rest-catalog-open-api.yaml#L4249)). The provider already caches credentials per prefix, so should the initial `storage-credentials` seed that cache? As I read it, for a catalog that vends only through `storage-credentials`, the first file access has to fetch from the refresh endpoint. With refresh disabled, the table gets no credentials at all.
   
   #2651 also routes each path to the credential with the longest matching prefix, built from `storage-credentials`, but it uses a different mechanism (`FileIOBuilder::with_prefixed_props`). How do you see the two PRs fitting together? It would help to settle on one prefix-routing path before either lands.



##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,358 @@ pub trait StorageFactory: Debug + Send + Sync {
     /// A `Result` containing an `Arc<dyn Storage>` on success, or an error
     /// if the storage could not be created.
     fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>>;
+
+    /// Build a new Storage instance, optionally supplying a credential provider
+    /// that the backend can call to obtain and refresh short-lived credentials.
+    ///
+    /// Backends that cannot use the provider ignore it and use the credentials
+    /// in `config`, as they would without one. The default does exactly that.
+    #[allow(unused_variables)]
+    fn build_with_credentials(
+        &self,
+        config: &StorageConfig,
+        credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
+    ) -> Result<Arc<dyn Storage>> {
+        self.build(config)
+    }
+}
+
+/// Supplies fresh, backend-specific storage credentials on demand.
+///
+/// A catalog that vends temporary credentials implements this trait so that
+/// storage backends can re-fetch credentials as they approach expiry instead
+/// of failing once the initial token's TTL runs out.
+///
+/// # Caching
+///
+/// [`load_credential`](Self::load_credential) may be called very frequently.
+/// Implementations must cache internally and only re-fetch when the current
+/// credential is at or near expiry; otherwise every object-store request could
+/// trigger a call back to the catalog.
+#[async_trait]
+pub trait StorageCredentialProvider: Debug + Send + Sync {
+    /// Return whether this provider has refresh configuration for `path`.
+    ///
+    /// Backends use this before replacing their normal credential chain. The
+    /// default is `true` for single-backend providers; multi-backend providers
+    /// should return `false` for schemes they do not configure.
+    fn supports_path(&self, _path: &str) -> bool {
+        true
+    }
+
+    /// Load a fresh credential for the storage location identified by `path`.
+    ///
+    /// `path` is the absolute location being accessed (e.g.
+    /// `s3://bucket/warehouse/db/table/...`). Providers that vend distinct
+    /// credentials per location prefix use it to select the most specific
+    /// match. When the selected credential has a declared
+    /// [`StorageCredential::prefix`], it must [cover](StorageCredential::covers) `path`.
+    async fn load_credential(&self, path: &str) -> Result<StorageCredential>;
+
+    /// Return a factory that rebuilds an equivalent provider in another process.
+    ///
+    /// [`FileIO::serialize_all`](crate::io::FileIO::serialize_all) serializes this
+    /// factory in place of the provider. The default reports that the provider
+    /// cannot be serialized.
+    fn factory(&self) -> Result<Arc<dyn StorageCredentialProviderFactory>> {
+        Err(Error::new(
+            ErrorKind::FeatureUnsupported,
+            "storage credential provider cannot be serialized",
+        ))
+    }
+}
+
+/// Serializable recipe that rebuilds a [`StorageCredentialProvider`] after
+/// [`FileIO`](crate::io::FileIO) deserialization.
+///
+/// Factories are serialized through [`typetag`](https://docs.rs/typetag), so
+/// implementations must use `#[typetag::serde]`, and the receiving binary must
+/// link the concrete implementation.
+#[typetag::serde(tag = "type")]
+pub trait StorageCredentialProviderFactory: Debug + Send + Sync {
+    /// Build a provider for a `FileIO` with the given storage configuration.
+    fn build(&self, config: &StorageConfig) -> Result<Arc<dyn StorageCredentialProvider>>;
+}
+
+/// A vended storage credential together with its scope and expiry.
+#[derive(Clone, Debug)]
+pub struct StorageCredential {
+    /// Storage-location prefix this credential is scoped to. `None` represents a
+    /// credential without a declared scope, sourced from flat storage properties.
+    prefix: Option<String>,
+    /// The backend-specific credential material.
+    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.
+    expires_at: Option<SystemTime>,
+}
+
+impl StorageCredential {
+    /// Create a storage credential with no declared scope or expiration.
+    pub fn new(kind: StorageCredentialKind) -> Self {
+        Self {
+            prefix: None,
+            kind,
+            expires_at: None,
+        }
+    }
+
+    /// Set the storage-location prefix this credential is scoped to.
+    pub fn with_prefix(mut self, prefix: impl Into<String>) -> Self {
+        self.prefix = Some(prefix.into());
+        self
+    }
+
+    /// Set when this credential expires.
+    pub fn with_expiration(mut self, expires_at: SystemTime) -> Self {
+        self.expires_at = Some(expires_at);
+        self
+    }
+
+    /// Return the storage-location prefix this credential is scoped to.
+    pub fn prefix(&self) -> Option<&str> {
+        self.prefix.as_deref()
+    }
+
+    /// Return whether this credential applies to `location`.
+    ///
+    /// A credential without a prefix covers every location. Otherwise the
+    /// prefix must match whole path segments of `location`, and scheme
+    /// aliases (`s3a`/`s3n` for `s3`, `gcs` for `gs`, and the plain-text
+    /// Azure schemes for their TLS variants) are treated as equal. A prefix
+    /// that is only a scheme, such as `s3`, covers every location with that
+    /// scheme.
+    pub fn covers(&self, location: &str) -> bool {
+        self.prefix
+            .as_deref()
+            .is_none_or(|prefix| storage_prefix_covers(prefix, location))
+    }
+
+    /// Return the backend-specific credential material.
+    pub fn kind(&self) -> &StorageCredentialKind {
+        &self.kind
+    }
+
+    /// Consume this credential and return its backend-specific material.
+    pub fn into_kind(self) -> StorageCredentialKind {
+        self.kind
+    }
+
+    /// Return when this credential expires.
+    pub fn expires_at(&self) -> Option<SystemTime> {
+        self.expires_at
+    }
+}
+
+/// Return whether the storage-location `prefix` covers `location`, with the
+/// matching rules of [`StorageCredential::covers`].
+pub fn storage_prefix_covers(prefix: &str, location: &str) -> bool {
+    let Some((location_scheme, location_rest)) = location.split_once("://") else {
+        return false;
+    };
+    let Some((prefix_scheme, prefix_rest)) = prefix.split_once("://") else {
+        return !prefix.is_empty() && canonical_scheme(prefix) == canonical_scheme(location_scheme);
+    };
+
+    canonical_scheme(prefix_scheme) == canonical_scheme(location_scheme)
+        && location_rest
+            .strip_prefix(prefix_rest)
+            .is_some_and(|remainder| {
+                prefix_rest.is_empty()
+                    || prefix_rest.ends_with('/')
+                    || remainder.is_empty()
+                    || remainder.starts_with('/')
+            })
+}

Review Comment:
   `storage_prefix_covers` only matches whole path segments, so a credential for `s3://bucket/tab` does not cover `s3://bucket/table/file.parquet`. The spec defines `prefix` as "a storage location prefix where the credential is relevant" and says clients should select the longest one ([`StorageCredential`](https://github.com/apache/iceberg/blob/48330b8dacab6662242d252b39c8444190979bb2/open-api/rest-catalog-open-api.yaml#L3843-L3846)). It doesn't say whether a prefix has to end on a segment boundary. Java settles that with a plain string `startsWith` in [`S3FileIO.clientForStoragePath`](https://github.com/apache/iceberg/blob/48330b8dacab6662242d252b39c8444190979bb2/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java#L387-L398), and it doesn't treat `s3a` and `s3` as the same scheme. With this code, a catalog that vends a prefix Java would accept gets a "does not cover" error here. Should this follow Java's matching? If the stricter rule is intentional, could the doc comment say 
 why?



##########
crates/storage/opendal/src/lib.rs:
##########
@@ -399,17 +483,77 @@ impl OpenDalStorage {
     ///
     /// For most backends the URL host (bucket name) is sufficient. For HF the host
     /// encodes the repo type, not the repo identity, so a more specific key is used.
-    fn batch_key_for_path(&self, path: &str) -> String {
+    #[allow(unreachable_patterns)]
+    fn batch_key_for_path(&self, path: &str) -> Result<String> {
         match self {
             #[cfg(feature = "opendal-hf")]
-            OpenDalStorage::Hf { .. } => hf_batch_key(path),
-            _ => url::Url::parse(path)
+            OpenDalStorage::Hf { .. } => Ok(hf_batch_key(path)),
+            #[cfg(feature = "opendal-azdls")]
+            OpenDalStorage::Azdls { .. } => azdls_batch_key(path),
+            _ => Ok(url::Url::parse(path)
                 .ok()
                 .and_then(|u| u.host_str().map(|s| s.to_string()))
-                .unwrap_or_default(),
+                .unwrap_or_default()),
         }
     }
 
+    /// Return the dynamic credential provider that serves `path`, if any.
+    fn credential_provider_for_path(
+        &self,
+        path: &str,
+    ) -> Option<&Arc<dyn StorageCredentialProvider>> {
+        let provider: &Arc<dyn StorageCredentialProvider> = (match self {
+            #[cfg(feature = "opendal-s3")]
+            OpenDalStorage::S3 {
+                customized_credential_load: None,
+                credential_provider: Some(provider),
+                ..
+            } => Some(provider),
+            #[cfg(feature = "opendal-gcs")]
+            OpenDalStorage::Gcs {
+                credential_provider: Some(provider),
+                ..
+            } => Some(provider),
+            #[cfg(feature = "opendal-azdls")]
+            OpenDalStorage::Azdls {
+                credential_provider: Some(provider),
+                ..
+            } => Some(provider),
+            _ => None,
+        })?;
+        provider.supports_path(path).then_some(provider)
+    }
+
+    /// Returns a key that keeps bulk deletes within one operator and credential
+    /// scope. Loading the credential is normally a cache hit and avoids rebuilding
+    /// an operator for every path while preventing a batch from crossing prefixes.
+    async fn delete_batch_key_for_path(&self, path: &str) -> Result<DeleteBatchKey> {
+        let credential_location = match self.credential_provider_for_path(path) {
+            Some(provider) => {
+                let credential = provider.load_credential(path).await?;
+                if !credential.covers(path) {
+                    return Err(Error::new(
+                        ErrorKind::DataInvalid,
+                        format!(
+                            "vended credential prefix {:?} does not cover storage location {path:?}",
+                            credential.prefix()
+                        ),
+                    ));
+                }
+                Some(match credential.prefix() {
+                    Some(prefix) => prefix.to_string(),
+                    None => utils::storage_root(path)?,
+                })

Review Comment:
   When the credential has no prefix, the batch looks up credentials for the storage root (`s3://bucket/`) for as long as its `Deleter` lives. What happens with a catalog that returns the initial credential in `config`, with no prefix, and returns prefix-scoped entries from the refresh endpoint? I tried this against `RestVendedCredentialProvider` at the head commit. I seeded an unscoped credential that expires in 2 seconds and a refresh endpoint that returns a credential for `s3://bucket/table`. After the seed expired, `load_credential("s3://bucket/table/data/f.parquet")` returned the refreshed credential, but `load_credential("s3://bucket/")` failed with `no unexpired vended credential matches storage location: s3://bucket/`. The deleters are closed at the end of `delete_stream`, so a stream that runs past the seed's expiry would fail on flush even though every path in it is covered. Before expiry, each root lookup inside the refresh window also fetches, finds no entry for `s3://buc
 ket/`, and calls `record_failure`, which backs off refresh for the whole AWS cache.
   
   One option is to flush the root-keyed deleter as soon as a path in the stream resolves to a prefixed credential, because that means the unscoped credential is being replaced while it is still valid. Could you also add a test for an unscoped seed followed by a scoped refresh?



##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -1470,7 +1478,7 @@ impl SessionCatalog for RestSessionCatalog {
         };
 
         let file_io = self
-            .load_file_io(Some(&response.metadata_location), None)
+            .load_file_io(commit.identifier(), Some(&response.metadata_location), None)

Review Comment:
   After a commit, the table returned here gets a `FileIO` built from the catalog properties only (`table_config` is `None`). So it has neither the vended credentials nor a refresh provider. `main` already behaves this way, but #2931 lists long-running writes as a motivation. A writer that keeps using the table returned by a commit loses refresh at that point. Should this reuse the `FileIO` from the table being committed? If that belongs in separate work, could you open a tracking issue and link it from #2931?



##########
crates/iceberg/src/io/file_io.rs:
##########
@@ -123,14 +132,28 @@ impl FileIO {
     ///
     /// All storage configuration properties are included in the serialized representation. These
     /// properties may contain credentials or other sensitive values, so the returned bytes must be
-    /// protected in transit and at rest by the application embedding this crate.
+    /// protected in transit and at rest by the application embedding this crate. A serialized
+    /// credential provider may likewise carry catalog authentication and vended credentials.
     ///
     /// Storage factories are serialized through [`typetag`](https://docs.rs/typetag). Third-party
     /// factories must use `#[typetag::serde]` on their [`StorageFactory`] implementation.
+    ///
+    /// A credential provider is serialized as the
+    /// [`StorageCredentialProviderFactory`] returned by
+    /// [`StorageCredentialProvider::factory`], and rebuilt on deserialization. Serialization fails
+    /// when the provider cannot be rebuilt in another process; use
+    /// [`FileIO::without_credential_provider`] to serialize without it.
     pub fn serialize_all(&self) -> Result<Vec<u8>> {
+        let credential_provider = self
+            .credential_provider
+            .as_ref()
+            .map(|provider| provider.factory())
+            .transpose()?;

Review Comment:
   With an injected `AuthManager`, a table whose config has a refresh endpoint now gets a `FileIO` that `serialize_all()` rejects with `FeatureUnsupported`. On `main` the same `FileIO` serializes, and `test_injected_auth_manager_file_io_serializes_without_provider_only` checks the new behavior. A caller that ships `FileIO` to workers would start failing after upgrading and would need a code change (`without_credential_provider`) to recover. Should `serialize_all` drop a provider that can't be rebuilt and serialize the static credentials, as `main` does, perhaps with a `tracing::warn!`? If failing is the behavior we want, could the PR description call it out as a behavior change?



##########
crates/storage/opendal/src/lib.rs:
##########
@@ -218,6 +233,7 @@ fn default_memory_operator() -> Operator {
 
 /// OpenDAL-based storage implementation.
 #[derive(Clone, Debug, Serialize, Deserialize)]
+#[non_exhaustive]

Review Comment:
   The new public fields on the `OpenDalStorage` variants (`credential_provider`, `sas_tokens`) and `#[non_exhaustive]` on the enum and its variants break downstream code that constructs or exhaustively matches `OpenDalStorage`. The repo marks breaking PRs with `!` in the title, as in #2838 (`feat!(rest): ...`). Could the title and description say this is breaking, or, once split, the PR that carries this change?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org