wgtmac commented on code in PR #892:
URL: https://github.com/apache/iceberg-cpp/pull/892#discussion_r4268160956
##########
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:
This also skips the refresh when the token has expired. Another request on
an open stream then signs with that expired token. Can we wait for a
request-safe credential refresh, or fail explicitly instead of returning
expired keys?
##########
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:
Can we match these credentials by prefix instead of array position? The
server may return the same prefixes in a different order. That makes a key-only
refresh look like a client change, so an existing stream keeps its old token.
##########
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:
Check the transport error before accepting the HTTP status. A server can
send 200 headers and then stall during the body. CPR keeps status 200 when the
transfer times out, so this returns success instead of an IOError.
--
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]
