plusplusjiajia commented on code in PR #892:
URL: https://github.com/apache/iceberg-cpp/pull/892#discussion_r4270569469
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -430,53 +738,250 @@ Result<ArrowS3FileIO::DelegatesByPrefix>
ArrowS3FileIO::BuildDelegates(
void ArrowS3FileIO::InstallCredentials(
std::vector<StorageCredential>& storage_credentials, DelegatesByPrefix&
delegates) {
file_io_by_prefix_.swap(delegates);
+ expiry_ = ExpiryOf(storage_credentials);
+ retry_refresh_at_ = {};
+ request_refresh_at_ = {};
+ ++credential_generation_;
storage_credentials_.swap(storage_credentials);
}
-std::shared_ptr<ArrowFileSystemFileIO> ArrowS3FileIO::MatchDelegate(
+bool ArrowS3FileIO::RefreshDue() const {
+ return (expiry_.unknown ||
+ (expiry_.expires_at.has_value() &&
+ std::chrono::system_clock::now() + kRefreshLeadTime >=
*expiry_.expires_at)) &&
+ std::chrono::steady_clock::now() >= retry_refresh_at_;
+}
+
+bool ArrowS3FileIO::Expired() const {
+ return expiry_.expires_at.has_value() &&
+ std::chrono::system_clock::now() >= *expiry_.expires_at;
+}
+
+std::chrono::steady_clock::time_point ArrowS3FileIO::BackoffUntil() const {
+ auto delay =
+
std::chrono::duration_cast<std::chrono::milliseconds>(kRefreshRetryBackoff);
+ if (expiry_.expires_at.has_value()) {
+ const auto remaining =
std::chrono::duration_cast<std::chrono::milliseconds>(
+ *expiry_.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(bool from_request) {
+ const auto due = [&] {
+ return provider_ && RefreshDue() &&
+ (!from_request || std::chrono::steady_clock::now() >=
request_refresh_at_);
+ };
+ {
+ // Cheap pre-check, so the common case costs one shared lock and no more.
+ std::shared_lock lock(mutex_);
+ if (!due()) {
+ 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.
+ if (from_request) {
+ return;
Review Comment:
Thanks! Done. A refresh now only rotates keys and never builds clients, so a
request with expired keys waits up to 10s for one in flight. If that fails, the
provider can't fail the request (the SDK sends empty keys unsigned), so S3
rejects the expired keys with `ExpiredToken`; new handles and deletes fail with
`TokenExpired`.
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -430,53 +738,250 @@ Result<ArrowS3FileIO::DelegatesByPrefix>
ArrowS3FileIO::BuildDelegates(
void ArrowS3FileIO::InstallCredentials(
std::vector<StorageCredential>& storage_credentials, DelegatesByPrefix&
delegates) {
file_io_by_prefix_.swap(delegates);
+ expiry_ = ExpiryOf(storage_credentials);
+ retry_refresh_at_ = {};
+ request_refresh_at_ = {};
+ ++credential_generation_;
storage_credentials_.swap(storage_credentials);
}
-std::shared_ptr<ArrowFileSystemFileIO> ArrowS3FileIO::MatchDelegate(
+bool ArrowS3FileIO::RefreshDue() const {
+ return (expiry_.unknown ||
+ (expiry_.expires_at.has_value() &&
+ std::chrono::system_clock::now() + kRefreshLeadTime >=
*expiry_.expires_at)) &&
+ std::chrono::steady_clock::now() >= retry_refresh_at_;
+}
+
+bool ArrowS3FileIO::Expired() const {
+ return expiry_.expires_at.has_value() &&
+ std::chrono::system_clock::now() >= *expiry_.expires_at;
+}
+
+std::chrono::steady_clock::time_point ArrowS3FileIO::BackoffUntil() const {
+ auto delay =
+
std::chrono::duration_cast<std::chrono::milliseconds>(kRefreshRetryBackoff);
+ if (expiry_.expires_at.has_value()) {
+ const auto remaining =
std::chrono::duration_cast<std::chrono::milliseconds>(
+ *expiry_.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(bool from_request) {
+ const auto due = [&] {
+ return provider_ && RefreshDue() &&
+ (!from_request || std::chrono::steady_clock::now() >=
request_refresh_at_);
+ };
+ {
+ // Cheap pre-check, so the common case costs one shared lock and no more.
+ std::shared_lock lock(mutex_);
+ if (!due()) {
+ 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.
+ if (from_request) {
+ return;
+ }
+ {
+ 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, credentials and their generation together.
+ std::shared_ptr<StorageCredentialProvider> provider;
+ uint64_t generation = 0;
+ std::vector<StorageCredential> current;
+ DelegatesByPrefix current_delegates;
+ {
+ // Whoever held the lock may also have just finished, leaving nothing to
do.
+ std::shared_lock lock(mutex_);
+ if (!due()) {
+ return;
+ }
+ provider = provider_;
+ generation = credential_generation_;
+ current = storage_credentials_;
+ current_delegates = file_io_by_prefix_;
+ }
+
+ // Outside `mutex_`: both are slow and must not block readers.
+ Status status;
+ DelegatesByPrefix delegates;
+ bool needs_clients = false;
+ auto refreshed = provider->Load();
+ if (refreshed.has_value()) {
+ status = ValidateRefreshed(*refreshed);
+ } else {
+ status = ::iceberg::unexpected<Error>(refreshed.error());
+ }
+ if (status.has_value()) {
+ if (auto same_clients = WithRefreshedKeys(current, current_delegates,
*refreshed);
+ same_clients.has_value()) {
+ delegates = std::move(same_clients).value();
+ } else if (from_request) {
+ needs_clients = true;
+ } else if (auto built = BuildDelegates(*refreshed); built.has_value()) {
+ delegates = std::move(built).value();
+ } else {
+ status = ::iceberg::unexpected<Error>(built.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;
+ }
+ if (needs_clients) {
+ // Fetched again by the next file operation, so it is not stale by then.
+ request_refresh_at_ = std::chrono::steady_clock::now() +
kRefreshRetryBackoff;
+ ICEBERG_LOG_DEBUG(
+ "Refreshed vended storage credentials need new S3 clients; leaving
them to "
+ "the next file operation");
+ return;
+ }
+
+ // `delegates`/`*refreshed` take the retired generation; both are declared
+ // before the lock, so it destructs only after the lock releases.
+ InstallCredentials(*refreshed, delegates);
+ if (RefreshDue()) {
+ // Tokens shorter-lived than the lead time come back due again at once.
+ retry_refresh_at_ = BackoffUntil();
+ }
+}
+
+std::optional<ArrowS3FileIO::DelegatesByPrefix>
ArrowS3FileIO::WithRefreshedKeys(
+ const std::vector<StorageCredential>& current, const DelegatesByPrefix&
delegates,
+ const std::vector<StorageCredential>& refreshed) {
+ // BuildDelegates made one delegate per S3 credential, in order.
+ const auto current_s3 = S3CredentialsOf(current);
+ const auto refreshed_s3 = S3CredentialsOf(refreshed);
+ if (current_s3.size() != delegates.size() || refreshed_s3.size() !=
delegates.size()) {
+ return std::nullopt;
+ }
+ auto result = delegates;
+ for (size_t i = 0; i < result.size(); ++i) {
+ if (!result[i].vended_keys || !DiffersOnlyInKeys(*current_s3[i],
*refreshed_s3[i])) {
Review Comment:
Thanks! Done. Refreshed credentials are now matched by canonical prefix, so
a reordered response is a key-only refresh. One that adds or drops a prefix or
changes a setting is rejected.
##########
src/iceberg/catalog/rest/http_client.cc:
##########
@@ -144,6 +144,10 @@ Status HandleFailureResponse(const cpr::Response& response,
if (IsSuccessful(response.status_code)) {
return {};
}
+ // No response to hand the error handler.
+ if (response.error.code == cpr::ErrorCode::OPERATION_TIMEDOUT) {
Review Comment:
Thanks! Done. `Get` checks transport errors first, so a timeout or a broken
transfer after a 2xx is an IOError. I left POST as is: a commit may already
have succeeded, and an error would clean up its files.
##########
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:
Update: a refresh now only rotates keys, so the caveat above no longer
applies.
--
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]
