wgtmac commented on code in PR #892:
URL: https://github.com/apache/iceberg-cpp/pull/892#discussion_r4254289889
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -235,14 +306,60 @@ class ArrowS3FileIO final : public FileIO, public
SupportsStorageCredentials {
void InstallCredentials(std::vector<StorageCredential>& storage_credentials,
DelegatesByPrefix& delegates);
+ /// \brief Whether the installed credentials are close enough to expiring to
+ /// be replaced, and no backoff is in effect.
+ ///
+ /// Callers must hold `mutex_`, at least shared.
+ bool RefreshDue() const;
+
+ /// \brief Whether the installed credentials have already stopped being
valid.
+ ///
+ /// Callers must hold `mutex_`, at least shared.
+ bool Expired() const;
+
+ /// \brief When the next refresh attempt becomes allowed after a failure.
+ ///
+ /// Never past the point the credentials stop being valid.
+ ///
+ /// Callers must hold `mutex_`, at least shared.
+ std::chrono::steady_clock::time_point BackoffUntil() const;
+
+ /// \brief Replace the credentials once they are close to expiring.
+ ///
+ /// Called before each handle is created; a handle keeps the delegate it was
+ /// built from, so I/O on an open one is not re-checked. A failure keeps the
Review Comment:
Can we separate credential refresh from filesystem replacement? Existing
handles and streams keep the old token, even after FileIO refreshes. A
refreshable provider on the S3 client, as in Java, would cover those requests
and avoid rebuilding clients for token-only changes.
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -291,9 +407,131 @@ Result<ArrowS3FileIO::DelegatesByPrefix>
ArrowS3FileIO::BuildDelegates(
void ArrowS3FileIO::InstallCredentials(
std::vector<StorageCredential>& storage_credentials, DelegatesByPrefix&
delegates) {
file_io_by_prefix_.swap(delegates);
+ expires_at_ = EarliestExpiry(storage_credentials);
+ retry_refresh_at_ = {};
+ ++credential_generation_;
storage_credentials_.swap(storage_credentials);
}
+bool ArrowS3FileIO::RefreshDue() const {
+ return expires_at_.has_value() &&
+ std::chrono::system_clock::now() + kRefreshLeadTime >= *expires_at_ &&
+ std::chrono::steady_clock::now() >= retry_refresh_at_;
+}
+
+bool ArrowS3FileIO::Expired() const {
+ return expires_at_.has_value() && std::chrono::system_clock::now() >=
*expires_at_;
+}
+
+std::chrono::steady_clock::time_point ArrowS3FileIO::BackoffUntil() const {
+ auto delay =
+
std::chrono::duration_cast<std::chrono::milliseconds>(kRefreshRetryBackoff);
+ if (expires_at_.has_value()) {
+ const auto remaining =
std::chrono::duration_cast<std::chrono::milliseconds>(
+ *expires_at_ - std::chrono::system_clock::now());
+ // Worth retrying before they run out; once they have, faster retries only
+ // hammer a catalog that is already failing.
+ if (remaining > std::chrono::milliseconds::zero()) {
+ delay = std::clamp(
+ remaining,
+
std::chrono::duration_cast<std::chrono::milliseconds>(kMinRefreshRetryBackoff),
+ delay);
+ }
+ }
+ return std::chrono::steady_clock::now() + delay;
+}
+
+void ArrowS3FileIO::MaybeRefreshCredentials() {
+ {
+ // Cheap pre-check, so the common case costs one shared lock and no more.
+ std::shared_lock lock(mutex_);
+ if (!provider_ || !RefreshDue()) {
+ return;
+ }
+ }
+
+ std::unique_lock refresh_lock(refresh_mutex_, std::defer_lock);
+ if (!refresh_lock.try_lock()) {
+ // Another operation is already fetching; normally just use what we have.
+ {
+ std::shared_lock lock(mutex_);
+ if (!Expired()) {
+ return;
+ }
+ }
+ // Expired credentials leave nothing to proceed with, so wait instead.
+ if (!refresh_lock.try_lock_for(kExpiredCredentialWait)) {
+ return;
Review Comment:
If this wait times out, the caller still uses credentials we know are
expired. A failed refresh does the same. Return an error after expiry instead
of creating a handle with unusable credentials.
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -291,9 +407,131 @@ Result<ArrowS3FileIO::DelegatesByPrefix>
ArrowS3FileIO::BuildDelegates(
void ArrowS3FileIO::InstallCredentials(
std::vector<StorageCredential>& storage_credentials, DelegatesByPrefix&
delegates) {
file_io_by_prefix_.swap(delegates);
+ expires_at_ = EarliestExpiry(storage_credentials);
+ retry_refresh_at_ = {};
+ ++credential_generation_;
storage_credentials_.swap(storage_credentials);
}
+bool ArrowS3FileIO::RefreshDue() const {
+ return expires_at_.has_value() &&
+ std::chrono::system_clock::now() + kRefreshLeadTime >= *expires_at_ &&
+ std::chrono::steady_clock::now() >= retry_refresh_at_;
+}
+
+bool ArrowS3FileIO::Expired() const {
+ return expires_at_.has_value() && std::chrono::system_clock::now() >=
*expires_at_;
+}
+
+std::chrono::steady_clock::time_point ArrowS3FileIO::BackoffUntil() const {
+ auto delay =
+
std::chrono::duration_cast<std::chrono::milliseconds>(kRefreshRetryBackoff);
+ if (expires_at_.has_value()) {
+ const auto remaining =
std::chrono::duration_cast<std::chrono::milliseconds>(
+ *expires_at_ - std::chrono::system_clock::now());
+ // Worth retrying before they run out; once they have, faster retries only
+ // hammer a catalog that is already failing.
+ if (remaining > std::chrono::milliseconds::zero()) {
+ delay = std::clamp(
+ remaining,
+
std::chrono::duration_cast<std::chrono::milliseconds>(kMinRefreshRetryBackoff),
+ delay);
+ }
+ }
+ return std::chrono::steady_clock::now() + delay;
+}
+
+void ArrowS3FileIO::MaybeRefreshCredentials() {
+ {
+ // Cheap pre-check, so the common case costs one shared lock and no more.
+ std::shared_lock lock(mutex_);
+ if (!provider_ || !RefreshDue()) {
+ return;
+ }
+ }
+
+ std::unique_lock refresh_lock(refresh_mutex_, std::defer_lock);
+ if (!refresh_lock.try_lock()) {
+ // Another operation is already fetching; normally just use what we have.
+ {
+ std::shared_lock lock(mutex_);
+ if (!Expired()) {
+ return;
+ }
+ }
+ // Expired credentials leave nothing to proceed with, so wait instead.
+ if (!refresh_lock.try_lock_for(kExpiredCredentialWait)) {
+ return;
+ }
+ }
+ // Snapshot the provider and credentials' generation together.
+ std::shared_ptr<StorageCredentialProvider> provider;
+ uint64_t generation = 0;
+ {
+ // Whoever held the lock may also have just finished, leaving nothing to
do.
+ std::shared_lock lock(mutex_);
+ if (!provider_ || !RefreshDue()) {
+ return;
+ }
+ provider = provider_;
+ generation = credential_generation_;
+ }
+
+ // Outside `mutex_`: both are slow and must not block readers.
+ Status status;
+ DelegatesByPrefix delegates;
+ auto refreshed = provider->Load();
+ if (refreshed.has_value()) {
+ auto built = BuildDelegates(*refreshed);
+ if (!built.has_value()) {
+ status = std::unexpected(built.error());
+ } else if (built->empty()) {
+ // Installing this would drop working credentials for whatever ambient
+ // identity the AWS chain finds. Java refuses an empty list too.
+ status = NotFound("Refreshed credentials contain no S3-compatible
prefix");
+ } else {
+ delegates = std::move(built).value();
+ }
+ } else {
+ status = std::unexpected(refreshed.error());
+ }
+
+ std::unique_lock lock(mutex_);
+ // Credentials installed meanwhile supersede this refresh: what it fetched is
+ // by now the older set.
+ const bool superseded = credential_generation_ != generation;
+ if (!status.has_value()) {
+ // Reported either way, so a failing catalog stays visible.
+ if (superseded) {
+ ICEBERG_LOG_WARN(
+ "Failed to refresh vended storage credentials ({}); they have since
been "
+ "replaced",
+ status.error().message);
+ return;
+ }
+ retry_refresh_at_ = BackoffUntil();
+ ICEBERG_LOG_WARN(
+ "Failed to refresh vended storage credentials ({}); keeping the
current "
+ "ones and retrying in {}ms",
+ status.error().message,
+ std::chrono::duration_cast<std::chrono::milliseconds>(
+ retry_refresh_at_ - std::chrono::steady_clock::now())
+ .count());
+ return;
+ }
+ if (superseded) {
+ return;
+ }
+
+ // `delegates`/`*refreshed` take the retired generation; both are declared
+ // before the lock, so it destructs only after the lock releases.
+ InstallCredentials(*refreshed, delegates);
Review Comment:
This installs a returned token even if it is already expired or its expiry
is missing or invalid. It can replace a still-valid token and then suppress
retries for 30 seconds. Validate temporary credentials before installing them
and keep the old set on failure.
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -291,9 +407,131 @@ Result<ArrowS3FileIO::DelegatesByPrefix>
ArrowS3FileIO::BuildDelegates(
void ArrowS3FileIO::InstallCredentials(
std::vector<StorageCredential>& storage_credentials, DelegatesByPrefix&
delegates) {
file_io_by_prefix_.swap(delegates);
+ expires_at_ = EarliestExpiry(storage_credentials);
+ retry_refresh_at_ = {};
+ ++credential_generation_;
storage_credentials_.swap(storage_credentials);
}
+bool ArrowS3FileIO::RefreshDue() const {
+ return expires_at_.has_value() &&
+ std::chrono::system_clock::now() + kRefreshLeadTime >= *expires_at_ &&
+ std::chrono::steady_clock::now() >= retry_refresh_at_;
+}
+
+bool ArrowS3FileIO::Expired() const {
+ return expires_at_.has_value() && std::chrono::system_clock::now() >=
*expires_at_;
+}
+
+std::chrono::steady_clock::time_point ArrowS3FileIO::BackoffUntil() const {
+ auto delay =
+
std::chrono::duration_cast<std::chrono::milliseconds>(kRefreshRetryBackoff);
+ if (expires_at_.has_value()) {
+ const auto remaining =
std::chrono::duration_cast<std::chrono::milliseconds>(
+ *expires_at_ - std::chrono::system_clock::now());
+ // Worth retrying before they run out; once they have, faster retries only
+ // hammer a catalog that is already failing.
+ if (remaining > std::chrono::milliseconds::zero()) {
+ delay = std::clamp(
+ remaining,
+
std::chrono::duration_cast<std::chrono::milliseconds>(kMinRefreshRetryBackoff),
+ delay);
+ }
+ }
+ return std::chrono::steady_clock::now() + delay;
+}
+
+void ArrowS3FileIO::MaybeRefreshCredentials() {
+ {
+ // Cheap pre-check, so the common case costs one shared lock and no more.
+ std::shared_lock lock(mutex_);
+ if (!provider_ || !RefreshDue()) {
+ return;
+ }
+ }
+
+ std::unique_lock refresh_lock(refresh_mutex_, std::defer_lock);
+ if (!refresh_lock.try_lock()) {
+ // Another operation is already fetching; normally just use what we have.
+ {
+ std::shared_lock lock(mutex_);
+ if (!Expired()) {
+ return;
+ }
+ }
+ // Expired credentials leave nothing to proceed with, so wait instead.
+ if (!refresh_lock.try_lock_for(kExpiredCredentialWait)) {
+ return;
+ }
+ }
+ // Snapshot the provider and credentials' generation together.
+ std::shared_ptr<StorageCredentialProvider> provider;
+ uint64_t generation = 0;
+ {
+ // Whoever held the lock may also have just finished, leaving nothing to
do.
+ std::shared_lock lock(mutex_);
+ if (!provider_ || !RefreshDue()) {
+ return;
+ }
+ provider = provider_;
+ generation = credential_generation_;
+ }
+
+ // Outside `mutex_`: both are slow and must not block readers.
+ Status status;
+ DelegatesByPrefix delegates;
+ auto refreshed = provider->Load();
Review Comment:
The 10-second limit only covers waiting for the refresh lock. The thread
running `Load()` can block indefinitely in the REST request, even while the old
credentials are still valid. Add a timeout to the refresh request itself.
--
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]
