SS-369 Refresh catalog-vended storage credentials (#38475)

### Motivation

Catalog-vended credentials expire, typically within the hour, and
nothing
refreshed them. `loadTable` hands their access keys to the FileIO once
and
OpenDAL keeps signing with them until they stop working, at which point
the
sink fails and only recovers by restarting the dataflow. Any sink
running
longer than one credential lifetime hits this.

### Description

Adds `VendedCredentialLoader`, a `ProvideCredential` implementation that
re-fetches from the catalog's `loadCredentials` endpoint and hands the
result
to OpenDAL's S3 credential chain.

The loader caches with its own deadline rather than relying on reqsign's
cache. OpenDAL rebuilds its `Operator` for every file operation, so the
`Signer` that holds reqsign's cached credential never survives a single
call;
without an internal cache this would be one catalog round trip per
parquet
write and per metadata read. The deadline comes from
`s3.session-token-expires-at-ms` where the catalog reports one,
refreshing
ahead of expiry, and from a short fixed interval where it does not. That
interval is a constant for now, with a TODO to make it a dyncfg once we
know
what real catalogs report.

On a 401 or 403 the loader invalidates the catalog token so the next
attempt
mints a fresh one, since nothing else on the storage path re-mints it.

Two supporting changes:

- Materialize now constructs the OAuth2 provider itself instead of
passing a
`credential` catalog property, so a single token object serves both
catalog
  requests and credential refreshes. The two cannot coexist: the catalog
  client rejects a custom authenticator combined with that property.
- `connect` takes the table the handle will be used against, because the
credentials endpoint is table-scoped. Callers that only prove
reachability
  (connection validation, sink purification) pass `None`.

Installing a loader hands it sole responsibility for S3 credentials,
since
OpenDAL replaces its entire provider chain and discards the static keys
parsed from the vended properties. It is therefore installed only when
the
connection asked for delegation and a table is known; every other case
keeps
the existing static-property path.

### Verification

`cargo check` and `cargo clippy` are clean. Not yet exercised against a
live
catalog: the refresh path needs a sink run long enough to cross a
credential
lifetime, and which branch of the deadline logic applies depends on
whether
the catalog reports an expiry.

Author: patrickwwbutler

src/sql/src/pure.rs

@@ -625,8 +625,10 @@ async fn purify_create_sink(
             // Now that we've validated the sink's storage creds (if they exist)
             // we _could_ use them to build a complete Iceberg client (both catalog and storage).
             // TODO(kynan): Actually use those sink-specific creds here instead of ignoring them.
+            // Purification only proves the catalog is reachable, so it needs no table-scoped
+            // storage credentials.
             let _catalog = connection
-                .connect(storage_configuration, InTask::No)
+                .connect(storage_configuration, InTask::No, None)
                 .await
                 .map_err(|e| IcebergSinkPurificationError::CatalogError(Arc::new(e)))?;
         }

src/storage-types/src/connections.rs

@@ -24,15 +24,17 @@ use aws_sigv4::sign::v4;
 // Aliased to avoid colliding with `mz_ccsr::tls::Identity`.
 use aws_smithy_runtime_api::client::identity::Identity as AwsIdentity;
 use base64::Engine;
-use http::{HeaderName, HeaderValue};
+use http::{HeaderMap, HeaderName, HeaderValue};
 use iceberg::Catalog;
 use iceberg::CatalogBuilder;
+use iceberg::TableIdent;
 use iceberg::io::{
     GCS_CREDENTIALS_JSON, GCS_DISABLE_CONFIG_LOAD, GCS_DISABLE_VM_METADATA, GCS_USER_PROJECT,
     S3_ACCESS_KEY_ID, S3_DISABLE_EC2_METADATA, S3_REGION, S3_SECRET_ACCESS_KEY,
 };
 use iceberg_catalog_rest::{
-    REST_CATALOG_PROP_URI, REST_CATALOG_PROP_WAREHOUSE, RequestAuthenticator, RestCatalogBuilder,
+    OAuth2TokenProvider, REST_CATALOG_PROP_URI, REST_CATALOG_PROP_WAREHOUSE, RequestAuthenticator,
+    RestCatalogBuilder, TokenProvider,
 };
 use iceberg_storage_opendal::{
     AwsCredential, CustomAwsCredentialLoader, OpenDalStorageFactory, ProvideCredential,
@@ -89,13 +91,17 @@ use crate::errors::{ContextCreationError, CsrConnectError};
 
 pub mod aws;
 pub mod gcp;
+mod iceberg_credentials;
 pub mod inline;
 pub mod string_or_secret;
 
-const REST_CATALOG_PROP_SCOPE: &str = "scope";
-const REST_CATALOG_PROP_CREDENTIAL: &str = "credential";
-/// Overrides the OAuth2 token endpoint. Spelled `uri` because that is the property name the
-/// Iceberg REST clients agree on, even though the SQL option says `URL`.
+/// The OAuth2 form field naming the scopes a token is requested for.
+///
+/// Materialize drives the OAuth2 exchange itself rather than through the `credential`,
+/// `oauth2-server-uri`, and `scope` catalog properties, so that one token object serves both
+/// catalog requests and storage-credential refreshes.
+const OAUTH2_PARAM_SCOPE: &str = "scope";
+
 const REST_CATALOG_PROP_OAUTH2_SERVER_URI: &str = "oauth2-server-uri";
 /// Requests catalog-vended storage credentials. `iceberg-rust` turns `header.*` catalog
 /// properties into headers on every REST request, the same way the Iceberg Java client
@@ -793,18 +799,27 @@ impl<C: ConnectionAccess> IcebergCatalogConnection<C> {
 }
 
 impl IcebergCatalogConnection<InlinedConnection> {
+    /// Connects to the catalog.
+    ///
+    /// `table` names the table this handle will be used against. It is needed only to keep
+    /// catalog-vended storage credentials refreshed, which the REST specification scopes to a
+    /// single table. Passing `None` leaves the connection on whatever credentials the catalog
+    /// supplies at `loadTable` time, which expire.
     pub async fn connect(
         &self,
         storage_configuration: &StorageConfiguration,
         in_task: InTask,
+        table: Option<&TableIdent>,
     ) -> Result<Arc<dyn Catalog>, anyhow::Error> {
         match self.catalog {
             IcebergCatalogImpl::S3TablesRest(ref s3tables) => {
+                // S3 Tables signs every request with SigV4 off a refreshable AWS provider, so it
+                // has no vended credential to keep alive.
                 self.connect_s3tables(s3tables, storage_configuration, in_task)
                     .await
             }
             IcebergCatalogImpl::Rest(ref rest) => {
-                self.connect_rest(rest, storage_configuration, in_task)
+                self.connect_rest(rest, storage_configuration, in_task, table)
                     .await
             }
         }
@@ -972,6 +987,7 @@ impl IcebergCatalogConnection<InlinedConnection> {
         rest: &RestIcebergCatalog,
         storage_configuration: &StorageConfiguration,
         in_task: InTask,
+        table: Option<&TableIdent>,
     ) -> Result<Arc<dyn Catalog>, anyhow::Error> {
         let mut props = BTreeMap::from([(
             REST_CATALOG_PROP_URI.to_string(),
@@ -982,6 +998,10 @@ impl IcebergCatalogConnection<InlinedConnection> {
             props.insert(REST_CATALOG_PROP_WAREHOUSE.to_string(), warehouse.clone());
         }
 
+        // One client for catalog requests, OAuth token requests, and credential refreshes, so all
+        // three share a connection pool. `iceberg-rust` would otherwise default to its own.
+        let client = reqwest::Client::new();
+
         // Catalog auth is configured through a combination of `props` and `.with_authenticator(...)`,
         // which happen at different stages of the [`RestCatalogBuilder`] -> [`RestCatalog`]
         // construction pipeline.
@@ -998,7 +1018,6 @@ impl IcebergCatalogConnection<InlinedConnection> {
                     )
                     .await
                     .map_err(|e| anyhow!("failed to read Iceberg catalog credential: {e}"))?;
-                props.insert(REST_CATALOG_PROP_CREDENTIAL.to_string(), credential);
 
                 if let Some(server_url) = server_url {
                     // The OAuth2 exchange POSTs the catalog credential to this URL, so a URL
@@ -1031,9 +1050,63 @@ impl IcebergCatalogConnection<InlinedConnection> {
                     );
                 }
 
-                if let Some(scope) = scope {
-                    props.insert(REST_CATALOG_PROP_SCOPE.to_string(), scope.clone());
-                }
+                // Materialize builds an OAuth2 provider shared across both catalog requests
+                // and the vended credentials refresh below.
+                let token_endpoint = match server_url {
+                    Some(server_url) => server_url.clone(),
+                    // Matches `iceberg-rust`'s default when no `oauth2-server-uri` is configured.
+                    None => format!(
+                        "{}/v1/oauth/tokens",
+                        self.uri.as_str().trim_end_matches('/')
+                    ),
+                };
+                let (client_id, client_secret) = match credential.split_once(':') {
+                    Some((client_id, client_secret)) => {
+                        (Some(client_id.to_string()), client_secret.to_string())
+                    }
+                    None => (None, credential),
+                };
+                let oauth_params = BTreeMap::from([(
+                    OAUTH2_PARAM_SCOPE.to_string(),
+                    // The default `iceberg-rust` applies when the connection names no scope.
+                    scope.clone().unwrap_or_else(|| "catalog".to_string()),
+                )]);
+                let token: Arc<dyn TokenProvider> = Arc::new(OAuth2TokenProvider::new(
+                    client.clone(),
+                    client_id,
+                    client_secret,
+                    token_endpoint,
+                    // The token request needs none of the catalog's headers, and
+                    // `OAuth2TokenProvider` sets the form content type itself.
+                    HeaderMap::new(),
+                    oauth_params.into_iter().collect(),
+                ));
+
+                // Installing a loader hands it sole responsibility for S3 credentials: OpenDAL
+                // replaces its whole provider chain, including the static keys parsed out of the
+                // catalog's vended `storage-credentials` props. So only install one when the
+                // connection asked for delegation and we know which table to refresh, and let the
+                // static props serve every other case.
+                let customized_credential_load = match (&rest.access_delegation, table) {
+                    (Some(IcebergAccessDelegation::VendedCredentials), Some(table)) => {
+                        let endpoint = iceberg_credentials::table_credentials_endpoint(
+                            &self.uri,
+                            &client,
+                            &token,
+                            rest.warehouse.as_deref(),
+                            table,
+                        )
+                        .await?;
+                        Some(CustomAwsCredentialLoader::new(
+                            iceberg_credentials::VendedCredentialLoader::new(
+                                client.clone(),
+                                endpoint,
+                                Arc::clone(&token),
+                            ),
+                        ))
+                    }
+                    _ => None,
+                };
 
                 (
                     OpenDalStorageFactory::S3 {
@@ -1043,9 +1116,12 @@ impl IcebergCatalogConnection<InlinedConnection> {
                         // vends instead, it returns per-table `storage-credentials` that
                         // `iceberg-rust` wires into the same FileIO.
                         // N.B. This is not confirmed to work with other catalog & storage implementations.
-                        customized_credential_load: None,
+                        customized_credential_load,
                     },
-                    None,
+                    // NOTE: We construct our own OAuth authenticator for the Catalog client instead of using the one built in.
+                    // This means we ignore auth overrides from `/v1/config` (e.g. `oauth2-server-uri`).
+                    // This is okay because users can set these configs from Mz SQL.
+                    Some(iceberg_catalog_rest::BearerTokenAuthenticator::new(token)),
                 )
             }
             IcebergCatalogAuth::Gcp(gcp_connection_reference) => {
@@ -1089,8 +1165,9 @@ impl IcebergCatalogConnection<InlinedConnection> {
             );
         }
 
-        let mut catalog =
-            RestCatalogBuilder::default().with_storage_factory(Arc::new(storage_factory));
+        let mut catalog = RestCatalogBuilder::default()
+            .with_storage_factory(Arc::new(storage_factory))
+            .with_client(client);
         if let Some(auth) = custom_authenticator {
             catalog = catalog.with_authenticator(Arc::new(auth));
         }
@@ -1106,8 +1183,9 @@ impl IcebergCatalogConnection<InlinedConnection> {
         _id: CatalogItemId,
         storage_configuration: &StorageConfiguration,
     ) -> Result<(), ConnectionValidationError> {
+        // Validation only lists namespaces, so it needs no table-scoped credentials.
         let catalog = self
-            .connect(storage_configuration, InTask::No)
+            .connect(storage_configuration, InTask::No, None)
             .await
             .map_err(|e| {
                 ConnectionValidationError::Other(anyhow!("failed to connect to catalog: {e}"))