zakariya-s commented on code in PR #2932: URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r4379274337
########## 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: only failed fetches and invalid credentials back off the cloud now -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
