This is an automated email from the ASF dual-hosted git repository.
wgtmac pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 9c478f75 feat(auth): implement AuthSessionCache for OAuth2 child
sessions (#985)
9c478f75 is described below
commit 9c478f7571e89ef6388c4fa776d7a4e5f448cbac
Author: lishuxu <[email protected]>
AuthorDate: Wed Nov 4 16:43:28 2026 +0800
feat(auth): implement AuthSessionCache for OAuth2 child sessions (#985)
Cache contextual and table-scoped OAuth2 child sessions so that context
credentials do not trigger a token request on every catalog call (#962).
AuthSessionCache:
- Expire entries after auth.session-timeout-ms without access (default
1 hour), and when the token of an OAuth2 session expires.
- Load each key once under concurrent Get() calls, with the loader
running outside the lock. Failed loads are not cached.
- Hand every removed session to the removal listener exactly once. A
session loaded after Close() is removed immediately.
- Sweep expired entries periodically on the auth refresh executor, every
session timeout clamped to [1s, 60s]. Without an executor, entries are
reclaimed when accessed again or on Close().
OAuth2Manager:
- Cache contextual sessions by session_id and table sessions by token.
- Authenticate child token requests with the parent session's headers.
- Credential child sessions follow token-refresh-enabled and, with
refresh enabled, require the auth executor. Token and exchange child
sessions do not refresh until refresh via token exchange is supported.
- Token child sessions no longer carry the parent's credential.
OAuth2Session:
- Add StopRefreshing(), a non-blocking stop used on cache removal.
- Add ExpiresAt() and IsRefreshable(), and include the session headers
in OAuth2Info().
---
src/iceberg/catalog/rest/CMakeLists.txt | 1 +
src/iceberg/catalog/rest/auth/auth_manager.cc | 293 +++++++++--
src/iceberg/catalog/rest/auth/auth_properties.h | 8 +
src/iceberg/catalog/rest/auth/auth_session.h | 1 +
.../catalog/rest/auth/auth_session_cache.cc | 320 +++++++++++
.../rest/auth/auth_session_cache_internal.h | 161 ++++++
.../catalog/rest/auth/auth_session_internal.h | 49 +-
src/iceberg/catalog/session_context.h | 3 +
src/iceberg/test/CMakeLists.txt | 1 +
src/iceberg/test/auth_manager_test.cc | 319 +++++++++++
src/iceberg/test/auth_session_cache_test.cc | 582 +++++++++++++++++++++
src/iceberg/test/rest_catalog_integration_test.cc | 7 +
12 files changed, 1705 insertions(+), 40 deletions(-)
diff --git a/src/iceberg/catalog/rest/CMakeLists.txt
b/src/iceberg/catalog/rest/CMakeLists.txt
index a8146d46..65c03724 100644
--- a/src/iceberg/catalog/rest/CMakeLists.txt
+++ b/src/iceberg/catalog/rest/CMakeLists.txt
@@ -22,6 +22,7 @@ set(ICEBERG_REST_SOURCES
auth/auth_managers.cc
auth/auth_properties.cc
auth/auth_session.cc
+ auth/auth_session_cache.cc
auth/oauth2_util.cc
auth/sigv4_manager.cc
catalog_properties.cc
diff --git a/src/iceberg/catalog/rest/auth/auth_manager.cc
b/src/iceberg/catalog/rest/auth/auth_manager.cc
index df5d7304..0fad64ec 100644
--- a/src/iceberg/catalog/rest/auth/auth_manager.cc
+++ b/src/iceberg/catalog/rest/auth/auth_manager.cc
@@ -21,18 +21,23 @@
#include <array>
#include <chrono>
+#include <mutex>
#include <optional>
+#include <string>
#include <string_view>
#include <utility>
#include "iceberg/catalog/rest/auth/auth_manager_internal.h"
#include "iceberg/catalog/rest/auth/auth_properties.h"
#include "iceberg/catalog/rest/auth/auth_session.h"
+#include "iceberg/catalog/rest/auth/auth_session_cache_internal.h"
#include "iceberg/catalog/rest/auth/auth_session_internal.h"
#include "iceberg/catalog/rest/auth/oauth2_util.h"
#include "iceberg/catalog/session_context.h"
+#include "iceberg/logging/log_macros.h"
#include "iceberg/util/base64.h"
#include "iceberg/util/macros.h"
+#include "iceberg/util/string_util.h"
namespace iceberg::rest::auth {
@@ -57,6 +62,127 @@ std::optional<std::pair<std::string, std::string>>
FindPreferredTypedToken(
return std::nullopt;
}
+/// \brief Credential used to create an OAuth2 child session.
+struct ChildCredential {
+ enum class Kind { kToken, kCredential, kTokenExchange };
+
+ Kind kind;
+ std::string value;
+ /// Subject token type; only set for kTokenExchange.
+ std::string token_type;
+};
+
+/// \brief Select the credential for a child session.
+///
+/// The order matches Java: a bearer token, then a client credential (only when
+/// `allow_credential` is true), then a typed token in kTokenPreferenceOrder.
+std::optional<ChildCredential> SelectChildCredential(
+ const std::unordered_map<std::string, std::string>& credentials,
+ bool allow_credential) {
+ if (auto token_it = credentials.find(AuthProperties::kToken.key());
+ token_it != credentials.end()) {
+ return ChildCredential{.kind = ChildCredential::Kind::kToken,
+ .value = token_it->second};
+ }
+ if (allow_credential) {
+ if (auto credential_it =
credentials.find(AuthProperties::kCredential.key());
+ credential_it != credentials.end()) {
+ return ChildCredential{.kind = ChildCredential::Kind::kCredential,
+ .value = credential_it->second};
+ }
+ }
+ if (auto typed_token = FindPreferredTypedToken(credentials);
typed_token.has_value()) {
+ return ChildCredential{.kind = ChildCredential::Kind::kTokenExchange,
+ .value = std::move(typed_token->second),
+ .token_type = std::move(typed_token->first)};
+ }
+ return std::nullopt;
+}
+
+/// \brief Actor token sent with a token exchange.
+struct ActorToken {
+ std::string token;
+ std::string token_type;
+};
+
+/// \brief The parent session's token, sent as the actor of a token exchange,
or
+/// std::nullopt if the parent has no token.
+std::optional<ActorToken> ExchangeActor(const OAuth2SessionInfo& parent_info) {
+ if (parent_info.token.empty()) {
+ return std::nullopt;
+ }
+ return ActorToken{.token = parent_info.token,
+ .token_type = parent_info.issued_token_type};
+}
+
+void AppendCacheKeyPart(std::string& key, std::string_view part) {
+ key += std::to_string(part.size());
+ key += ':';
+ key += part;
+}
+
+/// \brief Build the cache key of a table session.
+///
+/// The key covers every input that determines how the session authenticates,
so
+/// two table sessions are shared only when they would send the same
credentials.
+/// With P(s) = "<byte length of s>:<s>", the formats are:
+///
+/// bearer token: "tbl:" + P("token") + P(token)
+/// token exchange: "tbl:" + P("exchange") + P(subject type) + P(subject)
+/// + P(actor type) + P(actor)
+///
+/// A bearer token is sent as-is, so its key does not depend on the parent. An
+/// exchanged token also depends on the actor, the parent's token; the actor
parts
+/// are empty when the parent has no token, matching the exchange request. The
+/// length prefix keeps the key unambiguous although token types contain ':'.
+///
+/// For example, bearer token "abc" gives "tbl:5:token3:abc", and JWT "eyJ.a.b"
+/// exchanged under the parent access token "tok" gives (on one line):
+///
+/// tbl:8:exchange36:urn:ietf:params:oauth:token-type:jwt7:eyJ.a.b
+/// 45:urn:ietf:params:oauth:token-type:access_token3:tok
+///
+/// An empty token gives an empty key, which disables caching.
+std::string TableCacheKey(const ChildCredential& credential,
+ const OAuth2SessionInfo& parent_info) {
+ if (credential.value.empty()) {
+ return {};
+ }
+ std::string key = "tbl:";
+ switch (credential.kind) {
+ case ChildCredential::Kind::kToken:
+ AppendCacheKeyPart(key, "token");
+ AppendCacheKeyPart(key, credential.value);
+ return key;
+ case ChildCredential::Kind::kTokenExchange: {
+ auto actor = ExchangeActor(parent_info);
+ AppendCacheKeyPart(key, "exchange");
+ AppendCacheKeyPart(key, credential.token_type);
+ AppendCacheKeyPart(key, credential.value);
+ AppendCacheKeyPart(key, actor.has_value() ? actor->token_type : "");
+ AppendCacheKeyPart(key, actor.has_value() ? actor->token : "");
+ return key;
+ }
+ case ChildCredential::Kind::kCredential:
+ // Table sessions never use a client credential.
+ return {};
+ }
+ std::unreachable();
+}
+
+/// \brief Parse auth.session-timeout-ms, returning an error for a malformed or
+/// negative value rather than throwing as AuthProperties::Get() would.
+Result<std::chrono::milliseconds> ParseSessionTimeout(const AuthProperties&
config) {
+ const auto& entry = AuthProperties::kSessionTimeoutMs;
+ int64_t timeout_ms = entry.value();
+ if (auto it = config.configs().find(entry.key()); it !=
config.configs().end()) {
+ ICEBERG_ASSIGN_OR_RAISE(timeout_ms,
StringUtils::ParseNumber<int64_t>(it->second));
+ }
+ ICEBERG_PRECHECK(timeout_ms >= 0, "Invalid value for {}: {} (must not be
negative)",
+ entry.key(), timeout_ms);
+ return std::chrono::milliseconds(timeout_ms);
+}
+
std::unordered_map<std::string, std::string> FilterTableSessionProperties(
const std::unordered_map<std::string, std::string>& properties) {
std::unordered_map<std::string, std::string> filtered;
@@ -180,6 +306,30 @@ class OAuth2Manager : public AuthManager {
ICEBERG_PRECHECK(shared_client != nullptr,
"OAuth2 catalog session HTTP client must not be null");
refresh_client_ = std::move(shared_client);
+
+ // Initialize catalog-level configuration
+ keep_refreshed_ = config.keep_refreshed();
+ ICEBERG_ASSIGN_OR_RAISE(session_timeout_, ParseSessionTimeout(config));
+
+ // Create session cache
+ ICEBERG_ASSIGN_OR_RAISE(
+ session_cache_,
+ internal::AuthSessionCache::Make(
+ session_timeout_, [](std::shared_ptr<AuthSession> session) {
+ if (auto oauth2 =
+
std::dynamic_pointer_cast<internal::OAuth2Session>(session)) {
+ oauth2->StopRefreshing();
+ }
+ }));
+
+ // Reclaim idle sessions in the background on the refresh executor.
Without an
+ // executor, the cache sweeps from Get() instead.
+ if (refresh_executor_ != nullptr) {
+ ICEBERG_RETURN_UNEXPECTED(session_cache_->StartPeriodicSweep(
+ *refresh_executor_,
+ internal::AuthSessionCache::DefaultSweepInterval(session_timeout_)));
+ }
+
// Reuse the token response and start time from the init phase.
if (auth_response_.has_value()) {
return MakeSession(*auth_response_, config, config.keep_refreshed(),
start_time_);
@@ -212,21 +362,35 @@ class OAuth2Manager : public AuthManager {
Result<std::shared_ptr<AuthSession>> ContextualSession(
const SessionContext& context, std::shared_ptr<AuthSession> parent)
override {
- // TODO(lishuxu): Add child-session caching and refresh, matching Java
- // AuthSessionCache.
- return MaybeCreateChildSession(context.credentials,
/*allow_credential=*/true,
- std::move(parent));
+ auto credential =
+ SelectChildCredential(context.credentials, /*allow_credential=*/true);
+ if (!credential.has_value()) {
+ return parent;
+ }
+ ICEBERG_ASSIGN_OR_RAISE(auto parent_info, ParentInfo(*parent));
+ // Contextual sessions are cached by session ID, whatever credential they
use.
+ std::string cache_key = context.session_id.empty() ? "" : "ctx:" +
context.session_id;
+ return GetOrCreateChildSession(*credential, parent_info, cache_key);
}
Result<std::shared_ptr<AuthSession>> TableSession(
[[maybe_unused]] const TableIdentifier& table,
const std::unordered_map<std::string, std::string>& properties,
std::shared_ptr<AuthSession> parent) override {
- return MaybeCreateChildSession(FilterTableSessionProperties(properties),
- /*allow_credential=*/false,
std::move(parent));
+ auto credential =
SelectChildCredential(FilterTableSessionProperties(properties),
+ /*allow_credential=*/false);
+ if (!credential.has_value()) {
+ return parent;
+ }
+ ICEBERG_ASSIGN_OR_RAISE(auto parent_info, ParentInfo(*parent));
+ return GetOrCreateChildSession(*credential, parent_info,
+ TableCacheKey(*credential, parent_info));
}
Status Close() override {
+ if (session_cache_) {
+ session_cache_->Close();
+ }
refresh_client_.reset();
return {};
}
@@ -263,54 +427,99 @@ class OAuth2Manager : public AuthManager {
token_request_started_at,
refresh_executor_);
}
- Result<std::shared_ptr<AuthSession>> MaybeCreateChildSession(
- const std::unordered_map<std::string, std::string>& credentials,
- bool allow_credential, std::shared_ptr<AuthSession> parent) {
- auto token_it = credentials.find(AuthProperties::kToken.key());
- auto credential_it = credentials.find(AuthProperties::kCredential.key());
- auto typed_token = FindPreferredTypedToken(credentials);
- if (token_it == credentials.end() &&
- (!allow_credential || credential_it == credentials.end()) &&
- !typed_token.has_value()) {
- return parent;
- }
-
+ Result<OAuth2SessionInfo> ParentInfo(const AuthSession& parent) const {
ICEBERG_PRECHECK(refresh_client_ != nullptr,
"OAuth2 catalog session must be initialized before child
sessions");
- auto parent_info = parent->OAuth2Info();
+ auto parent_info = parent.OAuth2Info();
ICEBERG_PRECHECK(parent_info.has_value(),
"OAuth2 child session requires OAuth2 parent metadata");
+ return std::move(*parent_info);
+ }
- if (token_it != credentials.end()) {
- ICEBERG_ASSIGN_OR_RAISE(auto config,
- ChildConfig(*parent_info,
parent_info->credential));
- return MakeSession(AccessTokenResponse(token_it->second), config,
- /*keep_refreshed=*/false);
+ Result<std::shared_ptr<AuthSession>> GetOrCreateChildSession(
+ const ChildCredential& credential, const OAuth2SessionInfo& parent_info,
+ const std::string& cache_key) {
+ // Use cache if we have a valid cache_key
+ if (!cache_key.empty() && session_cache_) {
+ return session_cache_->Get(
+ cache_key, [&]() -> Result<std::shared_ptr<AuthSession>> {
+ return CreateChildSessionUncached(credential, parent_info);
+ });
}
- if (allow_credential && credential_it != credentials.end()) {
- ICEBERG_ASSIGN_OR_RAISE(auto config,
- ChildConfig(*parent_info,
credential_it->second));
- ICEBERG_ASSIGN_OR_RAISE(auto response,
- OAuth2Util::FetchToken(*refresh_client_,
*parent, config));
- return MakeSession(response, config, /*keep_refreshed=*/false);
+ // No cache, create session directly
+ return CreateChildSessionUncached(credential, parent_info);
+ }
+
+ Result<std::shared_ptr<AuthSession>> CreateChildSessionUncached(
+ const ChildCredential& credential, const OAuth2SessionInfo& parent_info)
{
+ switch (credential.kind) {
+ case ChildCredential::Kind::kToken:
+ return CreateTokenChildSession(credential.value, parent_info);
+ case ChildCredential::Kind::kCredential:
+ return CreateCredentialChildSession(credential.value, parent_info);
+ case ChildCredential::Kind::kTokenExchange:
+ return CreateTokenExchangeChildSession(credential.value,
credential.token_type,
+ parent_info);
}
+ std::unreachable();
+ }
+ Result<std::shared_ptr<AuthSession>> CreateTokenChildSession(
+ const std::string& token, const OAuth2SessionInfo& parent_info) {
+ // Token child session: don't include parent credential in config.
+ // Token-based sessions are not refreshed; when token exchange is enabled,
+ // refresh will use the token itself rather than the parent credential.
+ auto properties = parent_info.optional_oauth_params;
+ properties[AuthProperties::kScope.key()] = parent_info.scope;
+ properties[AuthProperties::kOAuth2ServerUri.key()] =
parent_info.oauth2_server_uri;
+ ICEBERG_ASSIGN_OR_RAISE(auto config,
AuthProperties::FromProperties(properties));
+ return MakeSession(AccessTokenResponse(token), config,
/*keep_refreshed=*/false);
+ }
+
+ Result<std::shared_ptr<AuthSession>> CreateCredentialChildSession(
+ const std::string& credential, const OAuth2SessionInfo& parent_info) {
+ ICEBERG_ASSIGN_OR_RAISE(auto config, ChildConfig(parent_info, credential));
+ // Use parent session headers for FetchToken authentication
+ auto temp_parent = AuthSession::MakeDefault(parent_info.headers);
+ auto token_request_started_at = std::chrono::steady_clock::now();
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto response, OAuth2Util::FetchToken(*refresh_client_, *temp_parent,
config));
+ // Credential child sessions follow the catalog's keep_refreshed, but
refreshing
+ // needs an executor. Without one, the child keeps its initial token.
+ if (keep_refreshed_ && refresh_executor_ == nullptr) {
+ std::call_once(no_refresh_executor_warning_, [] {
+ ICEBERG_LOG_WARN(
+ "OAuth2 credential child sessions will not refresh their tokens
because "
+ "no auth executor was provided");
+ });
+ }
+ return MakeSession(response, config, keep_refreshed_ && refresh_executor_
!= nullptr,
+ token_request_started_at);
+ }
+
+ Result<std::shared_ptr<AuthSession>> CreateTokenExchangeChildSession(
+ const std::string& subject_token, const std::string& subject_token_type,
+ const OAuth2SessionInfo& parent_info) {
std::optional<std::string> actor_token;
std::optional<std::string> actor_token_type;
- if (!parent_info->token.empty()) {
- actor_token = parent_info->token;
- actor_token_type = parent_info->issued_token_type;
+ if (auto actor = ExchangeActor(parent_info); actor.has_value()) {
+ actor_token = std::move(actor->token);
+ actor_token_type = std::move(actor->token_type);
}
+ // Use parent session headers for ExchangeToken authentication
+ auto temp_parent = AuthSession::MakeDefault(parent_info.headers);
+ auto token_request_started_at = std::chrono::steady_clock::now();
ICEBERG_ASSIGN_OR_RAISE(
auto response,
- OAuth2Util::ExchangeToken(*refresh_client_, *parent, {},
typed_token->second,
- typed_token->first, actor_token,
actor_token_type,
- parent_info->scope,
parent_info->oauth2_server_uri,
- parent_info->optional_oauth_params));
+ OAuth2Util::ExchangeToken(*refresh_client_, *temp_parent, {},
subject_token,
+ subject_token_type, actor_token,
actor_token_type,
+ parent_info.scope,
parent_info.oauth2_server_uri,
+ parent_info.optional_oauth_params));
ICEBERG_ASSIGN_OR_RAISE(auto config,
- ChildConfig(*parent_info,
parent_info->credential));
- return MakeSession(response, config, /*keep_refreshed=*/false);
+ ChildConfig(parent_info, parent_info.credential));
+ return MakeSession(response, config, /*keep_refreshed=*/false,
+ token_request_started_at);
}
/// Token response and start time captured by InitSession.
@@ -318,6 +527,12 @@ class OAuth2Manager : public AuthManager {
std::optional<std::chrono::steady_clock::time_point> start_time_;
std::shared_ptr<HttpClient> refresh_client_;
Executor* refresh_executor_ = nullptr;
+
+ // Catalog-level configuration initialized in CatalogSession
+ bool keep_refreshed_ = true;
+ std::chrono::milliseconds session_timeout_{3'600'000};
+ std::shared_ptr<internal::AuthSessionCache> session_cache_;
+ std::once_flag no_refresh_executor_warning_;
};
Result<std::unique_ptr<AuthManager>> MakeOAuth2Manager(
diff --git a/src/iceberg/catalog/rest/auth/auth_properties.h
b/src/iceberg/catalog/rest/auth/auth_properties.h
index 8784194c..0cf9be5e 100644
--- a/src/iceberg/catalog/rest/auth/auth_properties.h
+++ b/src/iceberg/catalog/rest/auth/auth_properties.h
@@ -78,10 +78,18 @@ class ICEBERG_REST_EXPORT AuthProperties : public
ConfigBase<AuthProperties> {
inline static Entry<std::string> kOAuth2ServerUri{"oauth2-server-uri",
"v1/oauth/tokens"};
inline static Entry<bool> kKeepRefreshed{"token-refresh-enabled", true};
+
+ /// Whether to use token exchange for refreshing OAuth2 sessions.
+ /// NOTE: Not yet implemented; the value is ignored.
inline static Entry<bool> kExchangeEnabled{"token-exchange-enabled", true};
+
inline static Entry<std::string> kAudience{"audience", ""};
inline static Entry<std::string> kResource{"resource", ""};
+ /// Session cache timeout in milliseconds. Sessions will be eligible for
eviction
+ /// after this duration of inactivity. Default is 1 hour (3,600,000 ms).
+ inline static Entry<int64_t> kSessionTimeoutMs{"auth.session-timeout-ms",
3'600'000};
+
// ---- OAuth2 token type constants ----
inline static const std::string kAccessTokenType =
diff --git a/src/iceberg/catalog/rest/auth/auth_session.h
b/src/iceberg/catalog/rest/auth/auth_session.h
index 7e330706..393418e2 100644
--- a/src/iceberg/catalog/rest/auth/auth_session.h
+++ b/src/iceberg/catalog/rest/auth/auth_session.h
@@ -43,6 +43,7 @@ struct ICEBERG_REST_EXPORT OAuth2SessionInfo {
std::string scope;
std::string oauth2_server_uri;
std::unordered_map<std::string, std::string> optional_oauth_params;
+ std::unordered_map<std::string, std::string> headers;
};
/// \brief An authentication session that can authenticate outgoing HTTP
requests.
diff --git a/src/iceberg/catalog/rest/auth/auth_session_cache.cc
b/src/iceberg/catalog/rest/auth/auth_session_cache.cc
new file mode 100644
index 00000000..aa62a8ac
--- /dev/null
+++ b/src/iceberg/catalog/rest/auth/auth_session_cache.cc
@@ -0,0 +1,320 @@
+/*
+ * 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.
+ */
+
+#include <algorithm>
+#include <exception>
+#include <utility>
+
+#include "iceberg/catalog/rest/auth/auth_session_cache_internal.h"
+#include "iceberg/catalog/rest/auth/auth_session_internal.h"
+#include "iceberg/logging/log_macros.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest::auth::internal {
+
+namespace {
+
+// Bounds of DefaultSweepInterval(): sweeping at least once a minute reclaims
idle
+// sessions promptly, and at most once a second keeps a zero timeout from
sweeping
+// continuously.
+constexpr auto kMinSweepInterval = std::chrono::seconds(1);
+constexpr auto kMaxSweepInterval = std::chrono::minutes(1);
+
+// The loader may call user-provided code such as an HttpClient, which can
throw.
+// Turning exceptions into errors lets Get() always clean up the loading slot.
+Result<std::shared_ptr<AuthSession>> RunLoader(const AuthSessionCache::Loader&
loader) {
+ try {
+ return loader();
+ } catch (const std::exception& e) {
+ return UnknownError("Auth session loader threw: {}", e.what());
+ } catch (...) {
+ return UnknownError("Auth session loader threw a non-standard exception");
+ }
+}
+
+} // namespace
+
+Result<std::shared_ptr<AuthSessionCache>> AuthSessionCache::Make(
+ std::chrono::milliseconds session_timeout, RemovalListener on_removal,
Clock clock) {
+ ICEBERG_PRECHECK(session_timeout >= std::chrono::milliseconds::zero(),
+ "Auth session timeout must not be negative: {} ms",
+ session_timeout.count());
+ ICEBERG_PRECHECK(clock != nullptr, "Auth session cache clock must not be
null");
+ return std::shared_ptr<AuthSessionCache>(
+ new AuthSessionCache(session_timeout, std::move(on_removal),
std::move(clock)));
+}
+
+std::chrono::milliseconds AuthSessionCache::DefaultSweepInterval(
+ std::chrono::milliseconds session_timeout) {
+ return std::clamp<std::chrono::milliseconds>(session_timeout,
kMinSweepInterval,
+ kMaxSweepInterval);
+}
+
+AuthSessionCache::AuthSessionCache(std::chrono::milliseconds session_timeout,
+ RemovalListener on_removal, Clock clock)
+ : session_timeout_(session_timeout),
+ on_removal_(std::move(on_removal)),
+ clock_(std::move(clock)) {}
+
+AuthSessionCache::~AuthSessionCache() { Close(); }
+
+Result<std::shared_ptr<AuthSession>> AuthSessionCache::Get(const std::string&
key,
+ const Loader&
loader) {
+ // Phase 1: Look up cache (under lock)
+ std::shared_ptr<Slot> slot;
+ std::vector<std::shared_ptr<AuthSession>> removed;
+ {
+ std::unique_lock lock(mutex_);
+ while (true) {
+ if (closed_) {
+ return Invalid("Auth session cache is closed");
+ }
+ auto it = slots_.find(key);
+ if (it == slots_.end()) {
+ break;
+ }
+ auto existing = it->second;
+ if (existing->loading) {
+ // Wait for another thread to finish loading. Loop again after waking
+ // because a failed load removes the slot.
+ load_finished_.wait(lock, [&] { return closed_ || !existing->loading;
});
+ continue;
+ }
+ auto now = clock_();
+ if (!IsExpired(*existing, now)) {
+ TouchLocked(it, now);
+ return existing->session;
+ }
+ EraseLocked(it, removed);
+ break;
+ }
+ if (!sweep_scheduler_.has_value()) {
+ // Without a periodic sweep, keys that are never accessed again would
stay
+ // until Close(). Reclaim idle entries before adding one, so the cache
grows
+ // with recently used keys rather than with every key ever used.
+ EvictIdleLocked(clock_(), removed);
+ }
+ // Create loading slot. Other threads calling Get(key) will wait.
+ slot = std::make_shared<Slot>();
+ slots_.emplace(key, slot);
+ }
+
+ // Phase 2: Call loader (lock released)
+ NotifyRemoved(removed);
+ auto result = RunLoader(loader);
+ if (result.has_value() && result.value() == nullptr) {
+ result = Invalid("Auth session loader returned a null session");
+ }
+
+ // Phase 3: Update cache (re-acquire lock)
+ std::shared_ptr<AuthSession> orphan;
+ {
+ std::lock_guard lock(mutex_);
+ slot->loading = false;
+ auto it = slots_.find(key);
+ bool current = it != slots_.end() && it->second == slot;
+
+ if (!result.has_value()) {
+ // Failed loads are not cached. Remove the slot so next Get() can retry;
a
+ // loading slot is not in access_order_ yet.
+ if (current) {
+ slots_.erase(it);
+ }
+ } else if (closed_ || !current) {
+ // Cache closed during load, or slot replaced. Mark as orphan.
+ orphan = result.value();
+ } else {
+ slot->session = result.value();
+ TouchLocked(it, clock_());
+ }
+ }
+ load_finished_.notify_all();
+
+ if (orphan != nullptr) {
+ if (on_removal_) {
+ on_removal_(orphan);
+ }
+ return Invalid("Auth session cache is closed");
+ }
+ return result;
+}
+
+void AuthSessionCache::Sweep() {
+ std::vector<std::shared_ptr<AuthSession>> removed;
+ {
+ std::lock_guard lock(mutex_);
+ EvictIdleLocked(clock_(), removed);
+ }
+ NotifyRemoved(removed);
+}
+
+void AuthSessionCache::TouchLocked(SlotMap::iterator it,
+ std::chrono::steady_clock::time_point now) {
+ // Every last_access is read from the clock under the lock and its entry
moved to
+ // the back, so access_order_ stays sorted by last access.
+ Slot& slot = *it->second;
+ slot.last_access = now;
+ if (slot.access_pos.has_value()) {
+ access_order_.splice(access_order_.end(), access_order_, *slot.access_pos);
+ } else {
+ slot.access_pos = access_order_.insert(access_order_.end(),
+ AccessEntry{.key = &it->first,
.slot = &slot});
+ }
+}
+
+void AuthSessionCache::EraseLocked(SlotMap::iterator it,
+ std::vector<std::shared_ptr<AuthSession>>&
removed) {
+ Slot& slot = *it->second;
+ if (slot.access_pos.has_value()) {
+ access_order_.erase(*slot.access_pos);
+ }
+ removed.push_back(std::move(slot.session));
+ slots_.erase(it);
+}
+
+void AuthSessionCache::EvictIdleLocked(
+ std::chrono::steady_clock::time_point now,
+ std::vector<std::shared_ptr<AuthSession>>& removed) {
+ // Every entry after the first one that is not idle was accessed more
recently.
+ while (!access_order_.empty()) {
+ const AccessEntry& oldest = access_order_.front();
+ if (now - oldest.slot->last_access < session_timeout_) {
+ break;
+ }
+ EraseLocked(slots_.find(*oldest.key), removed);
+ }
+}
+
+Status AuthSessionCache::StartPeriodicSweep(Executor& executor,
+ std::chrono::milliseconds
interval) {
+ ICEBERG_PRECHECK(interval > std::chrono::milliseconds::zero(),
+ "Auth session sweep interval must be positive: {} ms",
+ interval.count());
+ std::lock_guard lock(mutex_);
+ if (closed_ || sweep_scheduler_.has_value()) {
+ return {};
+ }
+ sweep_scheduler_.emplace(executor);
+ sweep_interval_ = interval;
+ ScheduleSweepLocked();
+ return {};
+}
+
+void AuthSessionCache::Close() {
+ std::vector<std::shared_ptr<AuthSession>> removed;
+ {
+ std::lock_guard lock(mutex_);
+ if (closed_) {
+ return;
+ }
+ closed_ = true;
+ for (auto& [key, slot] : slots_) {
+ // Sessions still loading are handed to the listener when their load
ends.
+ if (!slot->loading) {
+ removed.push_back(std::move(slot->session));
+ }
+ }
+ slots_.clear();
+ access_order_.clear();
+ }
+ load_finished_.notify_all();
+ // Shut down outside the lock: Shutdown() waits for an in-progress
submission,
+ // whose callback may take the lock. No sweep is scheduled once closed_ is
set.
+ if (sweep_scheduler_.has_value()) {
+ sweep_scheduler_->Shutdown();
+ }
+ NotifyRemoved(removed);
+}
+
+size_t AuthSessionCache::size() const {
+ std::lock_guard lock(mutex_);
+ return slots_.size();
+}
+
+bool AuthSessionCache::IsExpired(const Slot& slot,
+ std::chrono::steady_clock::time_point now)
const {
+ // Check 1: Idle timeout (expireAfterAccess semantics)
+ if (now - slot.last_access >= session_timeout_) {
+ return true;
+ }
+
+ // Check 2: Token expiration, for sessions that will not refresh
+ // This covers sessions without refresh and sessions whose refresh gave up
after
+ // repeated failures. Such a session is useless once its token expires, and
+ // evicting it is harmless since stopping its refresh is a no-op; the next
Get()
+ // loads a new one. A session that is still refreshing is left alone: its
refresh
+ // may be in flight, and evicting it would stop the refresh for callers that
+ // still hold it.
+ auto oauth2 =
std::dynamic_pointer_cast<internal::OAuth2Session>(slot.session);
+ if (oauth2 && !oauth2->IsRefreshable()) {
+ auto expires_at = oauth2->ExpiresAt();
+ return expires_at.has_value() && now >= *expires_at;
+ }
+
+ return false;
+}
+
+void AuthSessionCache::ScheduleSweepLocked() {
+ // Scheduling under the cache lock ensures Close() either cancels this task
or
+ // sees closed_ and keeps it from being scheduled.
+ std::weak_ptr<AuthSessionCache> weak_self = weak_from_this();
+ sweep_scheduler_->Schedule(
+ sweep_interval_, ExecutorTask([weak_self] {
+ if (auto self = weak_self.lock()) {
+ self->RunScheduledSweep();
+ }
+ }),
+ ::iceberg::internal::TaskSubmissionCallback([weak_self](Status status) {
+ if (status.has_value()) {
+ return;
+ }
+ // The sweep did not run; keep sweeping so idle sessions are still
reclaimed.
+ ICEBERG_LOG_WARN("Failed to submit auth session cache sweep: {}",
+ status.error().message);
+ if (auto self = weak_self.lock()) {
+ self->ScheduleNextSweep();
+ }
+ }));
+}
+
+void AuthSessionCache::ScheduleNextSweep() {
+ std::lock_guard lock(mutex_);
+ if (!closed_) {
+ ScheduleSweepLocked();
+ }
+}
+
+void AuthSessionCache::RunScheduledSweep() {
+ Sweep();
+ ScheduleNextSweep();
+}
+
+void AuthSessionCache::NotifyRemoved(
+ std::vector<std::shared_ptr<AuthSession>>& sessions) const {
+ if (!on_removal_) {
+ return;
+ }
+ for (auto& session : sessions) {
+ if (session != nullptr) {
+ on_removal_(session);
+ }
+ }
+}
+
+} // namespace iceberg::rest::auth::internal
diff --git a/src/iceberg/catalog/rest/auth/auth_session_cache_internal.h
b/src/iceberg/catalog/rest/auth/auth_session_cache_internal.h
new file mode 100644
index 00000000..aee88147
--- /dev/null
+++ b/src/iceberg/catalog/rest/auth/auth_session_cache_internal.h
@@ -0,0 +1,161 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <chrono>
+#include <condition_variable>
+#include <cstddef>
+#include <functional>
+#include <list>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "iceberg/catalog/rest/iceberg_rest_export.h"
+#include "iceberg/catalog/rest/type_fwd.h"
+#include "iceberg/result.h"
+#include "iceberg/util/executor.h"
+#include "iceberg/util/scheduled_executor_internal.h"
+
+/// \file iceberg/catalog/rest/auth/auth_session_cache_internal.h
+/// \brief Cache of child authentication sessions, keyed by caller identity.
+
+namespace iceberg::rest::auth::internal {
+
+/// \brief Caches child auth sessions and expires them after a period of
inactivity.
+///
+/// Entries expire after `session_timeout` without access. OAuth2 sessions that
+/// will not refresh also expire once their token does; such an entry is
replaced
+/// when its key is next accessed. Each session that leaves the cache is
handed to
+/// the removal listener exactly once.
+///
+/// Idle entries are reclaimed in access order, least recently used first, so
the
+/// cost is proportional to the number of entries removed. Sweep() does this,
and
+/// StartPeriodicSweep() runs it in the background; without a periodic sweep,
Get()
+/// does it before adding an entry.
+///
+/// Thread safety: all public methods are thread-safe. Loaders and the removal
+/// listener are always invoked without holding the cache lock.
+class ICEBERG_REST_EXPORT AuthSessionCache
+ : public std::enable_shared_from_this<AuthSessionCache> {
+ public:
+ using Clock = std::function<std::chrono::steady_clock::time_point()>;
+ using Loader = std::function<Result<std::shared_ptr<AuthSession>>()>;
+ /// Called once for every session that leaves the cache. Must not block or
throw.
+ using RemovalListener = std::function<void(const
std::shared_ptr<AuthSession>&)>;
+
+ /// \brief Create a cache.
+ ///
+ /// \param session_timeout Idle time after which an entry expires; must not
be
+ /// negative. Zero disables caching in effect.
+ /// \param on_removal Listener invoked for each session that leaves the
cache.
+ /// \param clock Time source; tests can inject a fake clock.
+ static Result<std::shared_ptr<AuthSessionCache>> Make(
+ std::chrono::milliseconds session_timeout, RemovalListener on_removal,
+ Clock clock = std::chrono::steady_clock::now);
+
+ ~AuthSessionCache();
+
+ AuthSessionCache(const AuthSessionCache&) = delete;
+ AuthSessionCache& operator=(const AuthSessionCache&) = delete;
+
+ /// \brief Suggested StartPeriodicSweep() interval for `session_timeout`: the
+ /// timeout clamped to [1s, 60s].
+ static std::chrono::milliseconds DefaultSweepInterval(
+ std::chrono::milliseconds session_timeout);
+
+ /// \brief Return the cached session for `key`, loading it on a miss.
+ ///
+ /// Concurrent calls for the same key run `loader` at most once; the others
wait
+ /// for its result. Failed loads are not cached; an exception thrown by
`loader`
+ /// is returned as an UnknownError. Returns an error once the cache
+ /// is closed; a session loaded after Close() is handed to the removal
listener.
+ Result<std::shared_ptr<AuthSession>> Get(const std::string& key, const
Loader& loader);
+
+ /// \brief Remove entries idle for at least the session timeout and notify
the
+ /// removal listener. Entries whose token expired are replaced on their next
+ /// access instead.
+ void Sweep();
+
+ /// \brief Run Sweep() on `executor` every `interval` until Close().
+ ///
+ /// `executor` is borrowed and must outlive this cache and its submitted
tasks.
+ /// `interval` must be positive. Calling this more than once, or after
Close(),
+ /// has no effect.
+ Status StartPeriodicSweep(Executor& executor, std::chrono::milliseconds
interval);
+
+ /// \brief Stop periodic sweeping and remove every entry. Idempotent.
+ void Close();
+
+ /// \brief Number of entries, including loads in progress.
+ size_t size() const;
+
+ private:
+ struct Slot;
+
+ /// One element per loaded slot. `key` points at the key stored in slots_,
which
+ /// stays valid until that element is erased, even when the map rehashes.
+ struct AccessEntry {
+ const std::string* key;
+ Slot* slot;
+ };
+ using AccessOrder = std::list<AccessEntry>;
+
+ struct Slot {
+ bool loading = true;
+ std::shared_ptr<AuthSession> session;
+ std::chrono::steady_clock::time_point last_access;
+ /// Position in access_order_; set once the load succeeds.
+ std::optional<AccessOrder::iterator> access_pos;
+ };
+ using SlotMap = std::unordered_map<std::string, std::shared_ptr<Slot>>;
+
+ AuthSessionCache(std::chrono::milliseconds session_timeout, RemovalListener
on_removal,
+ Clock clock);
+
+ bool IsExpired(const Slot& slot, std::chrono::steady_clock::time_point now)
const;
+ void TouchLocked(SlotMap::iterator it, std::chrono::steady_clock::time_point
now);
+ void EraseLocked(SlotMap::iterator it,
+ std::vector<std::shared_ptr<AuthSession>>& removed);
+ void EvictIdleLocked(std::chrono::steady_clock::time_point now,
+ std::vector<std::shared_ptr<AuthSession>>& removed);
+ void ScheduleSweepLocked();
+ void ScheduleNextSweep();
+ void RunScheduledSweep();
+ void NotifyRemoved(std::vector<std::shared_ptr<AuthSession>>& sessions)
const;
+
+ const std::chrono::milliseconds session_timeout_;
+ const RemovalListener on_removal_;
+ const Clock clock_;
+
+ mutable std::mutex mutex_;
+ std::condition_variable load_finished_;
+ SlotMap slots_;
+ /// Loaded slots ordered by last access, least recent first.
+ AccessOrder access_order_;
+ bool closed_ = false;
+ std::optional<::iceberg::internal::ScheduledExecutor> sweep_scheduler_;
+ std::chrono::milliseconds sweep_interval_{0};
+};
+
+} // namespace iceberg::rest::auth::internal
diff --git a/src/iceberg/catalog/rest/auth/auth_session_internal.h
b/src/iceberg/catalog/rest/auth/auth_session_internal.h
index fa2f7ede..a4454aaf 100644
--- a/src/iceberg/catalog/rest/auth/auth_session_internal.h
+++ b/src/iceberg/catalog/rest/auth/auth_session_internal.h
@@ -102,11 +102,46 @@ class OAuth2Session final : public AuthSession,
.scope = config_.scope,
.oauth2_server_uri = config_.token_endpoint,
.optional_oauth_params = config_.optional_oauth_params,
+ .headers = headers_,
};
}
Status Close() override { return CloseImpl(); }
+ /// \brief Stop scheduling token refreshes without waiting.
+ ///
+ /// Unlike Close(), this does not wait for an in-flight refresh to finish,
so it
+ /// is safe to call from any thread, including refresh executor workers. The
+ /// session keeps its current token and remains usable; an in-flight refresh
may
+ /// still update the token but will not schedule further refreshes.
+ void StopRefreshing() {
+ {
+ std::lock_guard lock(refresh_mutex_);
+ closed_ = true;
+ }
+ if (scheduler_) {
+ scheduler_->Shutdown();
+ }
+ }
+
+ /// \brief Expiration time of the current token, or std::nullopt if unknown.
+ std::optional<std::chrono::steady_clock::time_point> ExpiresAt() const {
+ std::shared_lock lock(mutex_);
+ if (expires_at_ == std::chrono::steady_clock::time_point{}) {
+ return std::nullopt;
+ }
+ return expires_at_;
+ }
+
+ /// \brief Whether this session will keep refreshing its token.
+ ///
+ /// False if refresh is disabled, was stopped by StopRefreshing() or
Close(), or
+ /// gave up after repeated failures.
+ bool IsRefreshable() const {
+ std::lock_guard lock(refresh_mutex_);
+ return scheduler_.has_value() && !closed_ && !refresh_stopped_;
+ }
+
~OAuth2Session() override { std::ignore = CloseImpl(); }
private:
@@ -221,11 +256,18 @@ class OAuth2Session final : public AuthSession,
});
if (!result.has_value()) {
ICEBERG_LOG_WARN("Failed to refresh OAuth2 token: {}",
result.error().message);
+ MarkRefreshStopped();
return std::nullopt;
}
return *result;
}
+ // Record that no further refresh will be scheduled after a failure.
+ void MarkRefreshStopped() {
+ std::lock_guard lock(refresh_mutex_);
+ refresh_stopped_ = true;
+ }
+
Result<std::optional<std::chrono::milliseconds>> RefreshToken() {
{
std::lock_guard lock(refresh_mutex_);
@@ -276,6 +318,9 @@ class OAuth2Session final : public AuthSession,
ICEBERG_LOG_WARN(
"Stopping OAuth2 token refresh after repeated submission
failures: {}",
status.error().message);
+ if (auto self = weak_self.lock()) {
+ self->MarkRefreshStopped();
+ }
return;
}
if (auto self = weak_self.lock()) {
@@ -313,7 +358,9 @@ class OAuth2Session final : public AuthSession,
AuthProperties refresh_properties_;
std::shared_ptr<HttpClient> client_;
bool closed_ = false;
- std::mutex refresh_mutex_;
+ // Set when refresh gives up after repeated failures; guarded by
refresh_mutex_.
+ bool refresh_stopped_ = false;
+ mutable std::mutex refresh_mutex_;
std::condition_variable refresh_cv_;
int active_refresh_count_ = 0;
std::optional<::iceberg::internal::ScheduledExecutor> scheduler_;
diff --git a/src/iceberg/catalog/session_context.h
b/src/iceberg/catalog/session_context.h
index 19eb8fc5..1a6bc819 100644
--- a/src/iceberg/catalog/session_context.h
+++ b/src/iceberg/catalog/session_context.h
@@ -41,6 +41,9 @@ namespace iceberg {
/// silently overriding credentials.
struct ICEBERG_EXPORT SessionContext {
/// Unique session identifier. Explicit contexts must provide a non-empty ID.
+ ///
+ /// A session ID identifies a stable caller context. Its identity and
credentials
+ /// must not change while the ID is in use.
std::string session_id;
/// Caller identity. Empty means no identity was supplied. This value is
diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt
index 2513ff17..b2fd4f9b 100644
--- a/src/iceberg/test/CMakeLists.txt
+++ b/src/iceberg/test/CMakeLists.txt
@@ -345,6 +345,7 @@ if(ICEBERG_BUILD_REST)
add_rest_iceberg_test(rest_catalog_test
SOURCES
auth_manager_test.cc
+ auth_session_cache_test.cc
catalog_properties_test.cc
error_handlers_test.cc
endpoint_test.cc
diff --git a/src/iceberg/test/auth_manager_test.cc
b/src/iceberg/test/auth_manager_test.cc
index 0bae4306..076df633 100644
--- a/src/iceberg/test/auth_manager_test.cc
+++ b/src/iceberg/test/auth_manager_test.cc
@@ -359,6 +359,35 @@ TEST_F(AuthManagerTest, OAuth2StaticToken) {
"catalog-resource");
}
+TEST_F(AuthManagerTest, OAuth2ZeroSessionTimeoutCreatesCatalogSession) {
+ std::unordered_map<std::string, std::string> properties = {
+ {AuthProperties::kAuthType, "oauth2"},
+ {AuthProperties::kToken.key(), "my-static-token"},
+ {AuthProperties::kSessionTimeoutMs.key(), "0"},
+ };
+
+ ICEBERG_UNWRAP_OR_FAIL(auto manager, AuthManagers::Load("test-catalog",
properties));
+ ICEBERG_UNWRAP_OR_FAIL(auto session, manager->CatalogSession(client_,
properties));
+
+ EXPECT_NE(session, nullptr);
+ EXPECT_THAT(manager->Close(), IsOk());
+}
+
+TEST_F(AuthManagerTest, OAuth2InvalidSessionTimeoutReturnsError) {
+ for (const std::string value : {"1h", "abc", "-1"}) {
+ std::unordered_map<std::string, std::string> properties = {
+ {AuthProperties::kAuthType, "oauth2"},
+ {AuthProperties::kToken.key(), "my-static-token"},
+ {AuthProperties::kSessionTimeoutMs.key(), value},
+ };
+
+ ICEBERG_UNWRAP_OR_FAIL(auto manager, AuthManagers::Load("test-catalog",
properties));
+ EXPECT_THAT(manager->CatalogSession(client_, properties),
+ IsError(ErrorKind::kInvalidArgument))
+ << value;
+ }
+}
+
// Verifies OAuth2 type is inferred from token property
TEST_F(AuthManagerTest, OAuth2InferredFromToken) {
std::unordered_map<std::string, std::string> properties = {
@@ -483,6 +512,29 @@ TEST_F(AuthManagerTest, OAuth2TableTokenCreatesChild) {
EXPECT_TRUE(info->credential.empty());
}
+TEST_F(AuthManagerTest, OAuth2TableTokenSessionIsCachedByToken) {
+ std::unordered_map<std::string, std::string> properties = {
+ {AuthProperties::kAuthType, "oauth2"},
+ };
+ ICEBERG_UNWRAP_OR_FAIL(auto manager, AuthManagers::Load("test-catalog",
properties));
+ ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client_,
properties));
+ TableIdentifier table{.ns = Namespace{{"db"}}, .name = "table"};
+
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto first,
+ manager->TableSession(table, {{AuthProperties::kToken.key(),
"token-a"}}, parent));
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto second,
+ manager->TableSession(table, {{AuthProperties::kToken.key(),
"token-a"}}, parent));
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto other,
+ manager->TableSession(table, {{AuthProperties::kToken.key(),
"token-b"}}, parent));
+
+ EXPECT_EQ(first, second);
+ EXPECT_NE(first, other);
+ EXPECT_THAT(manager->Close(), IsOk());
+}
+
TEST_F(AuthManagerTest, OAuth2TableIgnoresCredential) {
std::unordered_map<std::string, std::string> properties = {
{AuthProperties::kAuthType, "oauth2"},
@@ -804,18 +856,75 @@ TEST(OAuth2SessionTest, StopsAfterHttpRetryBudget) {
});
auto client = std::make_shared<RefreshHttpClient>();
ICEBERG_UNWRAP_OR_FAIL(auto session, RefreshingSession(client, executor));
+ auto oauth2 = std::dynamic_pointer_cast<internal::OAuth2Session>(session);
+ ASSERT_NE(oauth2, nullptr);
ASSERT_EQ(task.wait_for(std::chrono::seconds(5)), std::future_status::ready);
+ EXPECT_TRUE(oauth2->IsRefreshable());
test::FakeRetryEnvironment fake_retry;
ScopedRetryTestHooks hooks(fake_retry.hooks());
std::move(task.get())();
EXPECT_EQ(client->requests, 6);
EXPECT_THAT(fake_retry.observed_base_delays_ms(),
::testing::ElementsAre(100, 200, 400, 800, 1600));
+ // No further refresh is scheduled, so the session no longer counts as
refreshable.
+ EXPECT_FALSE(oauth2->IsRefreshable());
ICEBERG_UNWRAP_OR_FAIL(auto request, session->Authenticate({}));
EXPECT_EQ(request.headers.at("Authorization"), "Bearer initial");
ASSERT_THAT(session->Close(), IsOk());
}
+TEST(OAuth2SessionTest, StopRefreshingKeepsSessionUsable) {
+ MockExecutor executor;
+ EXPECT_CALL(executor, Submit).Times(0);
+ OAuthTokenResponse token_response{
+ .access_token = "token",
+ .token_type = "bearer",
+ .expires_in_secs = 3600,
+ };
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto session,
+ internal::OAuth2Session::Make(token_response,
+ {.token_endpoint =
"http://localhost/oauth/tokens";,
+ .client_id = "client_id",
+ .client_secret = "client_secret",
+ .scope = "catalog",
+ .refresh_executor = &executor},
+ std::make_shared<HttpClient>(),
std::nullopt));
+ EXPECT_TRUE(session->IsRefreshable());
+
+ session->StopRefreshing();
+ session->StopRefreshing();
+ EXPECT_FALSE(session->IsRefreshable());
+
+ ICEBERG_UNWRAP_OR_FAIL(auto request, session->Authenticate({}));
+ EXPECT_EQ(request.headers.at("Authorization"), "Bearer token");
+ EXPECT_THAT(session->Close(), IsOk());
+}
+
+TEST(OAuth2SessionTest, ExpiresAtReflectsTokenLifetime) {
+ auto make_session = [](std::optional<int64_t> expires_in_secs) {
+ OAuthTokenResponse token_response{
+ .access_token = "opaque-token",
+ .token_type = "bearer",
+ .expires_in_secs = expires_in_secs,
+ };
+ return internal::OAuth2Session::Make(
+ token_response, {.token_endpoint = "http://localhost/oauth/tokens"},
+ std::make_shared<HttpClient>(), std::nullopt);
+ };
+
+ auto before = std::chrono::steady_clock::now();
+ ICEBERG_UNWRAP_OR_FAIL(auto expiring, make_session(60));
+ auto after = std::chrono::steady_clock::now();
+ auto expires_at = expiring->ExpiresAt();
+ ASSERT_TRUE(expires_at.has_value());
+ EXPECT_GE(*expires_at, before + std::chrono::seconds(60));
+ EXPECT_LE(*expires_at, after + std::chrono::seconds(60));
+
+ ICEBERG_UNWRAP_OR_FAIL(auto unknown, make_session(std::nullopt));
+ EXPECT_FALSE(unknown->ExpiresAt().has_value());
+}
+
TEST(OAuth2SessionTest, InitTokenExpirationUsesRequestStartTime) {
OAuthTokenResponse token_response{
.access_token = "opaque-token",
@@ -847,4 +956,214 @@ TEST(OAuth2SessionTest,
JwtExpirationTakesPriorityOverExpiresIn) {
EXPECT_EQ(*expiration,
std::chrono::steady_clock::time_point(std::chrono::seconds(70)));
}
+namespace {
+
+/// Creates a credential child session from a static-token catalog whose token
+/// endpoint always succeeds.
+Result<std::shared_ptr<internal::OAuth2Session>> CredentialChildSession(
+ AuthManager& manager) {
+ auto client = std::make_shared<RefreshHttpClient>();
+ client->response = []() -> Result<HttpResponse> {
+ return HttpResponse(
+ 200,
R"({"access_token":"child-token","token_type":"bearer","expires_in":3600})");
+ };
+ std::unordered_map<std::string, std::string> properties = {
+ {AuthProperties::kAuthType, "oauth2"},
+ {AuthProperties::kToken.key(), "catalog-token"},
+ };
+ ICEBERG_ASSIGN_OR_RAISE(auto parent, manager.CatalogSession(client,
properties));
+ SessionContext context{
+ .session_id = "tenant-a",
+ .credentials = {{AuthProperties::kCredential.key(), "client:secret"}},
+ };
+ ICEBERG_ASSIGN_OR_RAISE(auto child, manager.ContextualSession(context,
parent));
+ auto oauth2 = std::dynamic_pointer_cast<internal::OAuth2Session>(child);
+ ICEBERG_PRECHECK(oauth2 != nullptr, "child session is not an OAuth2
session");
+ return oauth2;
+}
+
+} // namespace
+
+TEST(OAuth2ChildSessionTest, CredentialChildRefreshesWithExecutor) {
+ // The refresh is due long after this test ends, so nothing is submitted.
+ MockExecutor executor;
+ EXPECT_CALL(executor, Submit).Times(0);
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto manager,
+ AuthManagers::Load("test-catalog", {{AuthProperties::kAuthType,
"oauth2"}}));
+ manager->SetAuthRefreshExecutor(&executor);
+
+ ICEBERG_UNWRAP_OR_FAIL(auto child, CredentialChildSession(*manager));
+
+ EXPECT_TRUE(child->IsRefreshable());
+ ICEBERG_UNWRAP_OR_FAIL(auto request, child->Authenticate({}));
+ EXPECT_EQ(request.headers.at("Authorization"), "Bearer child-token");
+ EXPECT_THAT(manager->Close(), IsOk());
+}
+
+TEST(OAuth2ChildSessionTest, CredentialChildWithoutExecutorDoesNotRefresh) {
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto manager,
+ AuthManagers::Load("test-catalog", {{AuthProperties::kAuthType,
"oauth2"}}));
+
+ ICEBERG_UNWRAP_OR_FAIL(auto child, CredentialChildSession(*manager));
+
+ EXPECT_FALSE(child->IsRefreshable());
+ ICEBERG_UNWRAP_OR_FAIL(auto request, child->Authenticate({}));
+ EXPECT_EQ(request.headers.at("Authorization"), "Bearer child-token");
+ EXPECT_THAT(manager->Close(), IsOk());
+}
+
+TEST(OAuth2ChildSessionTest, ContextSessionIsReloadedAfterRefreshGivesUp) {
+ MockExecutor executor;
+ std::promise<ExecutorTask> queued;
+ auto task = queued.get_future();
+ EXPECT_CALL(executor, Submit).WillOnce([&](ExecutorTask work) -> Status {
+ queued.set_value(std::move(work));
+ return {};
+ });
+ // The token endpoint first issues an already expired token, then fails
until it
+ // recovers.
+ auto client = std::make_shared<RefreshHttpClient>();
+ std::atomic<bool> recovered = false;
+ client->response = [&]() -> Result<HttpResponse> {
+ if (client->requests == 1) {
+ return HttpResponse(
+ 200,
R"({"access_token":"expired","token_type":"bearer","expires_in":0})");
+ }
+ if (!recovered) {
+ return IOError("token endpoint unavailable");
+ }
+ return HttpResponse(
+ 200,
R"({"access_token":"fresh","token_type":"bearer","expires_in":3600})");
+ };
+ std::unordered_map<std::string, std::string> properties = {
+ {AuthProperties::kAuthType, "oauth2"},
+ {AuthProperties::kToken.key(), "catalog-token"},
+ };
+ ICEBERG_UNWRAP_OR_FAIL(auto manager, AuthManagers::Load("test-catalog",
properties));
+ manager->SetAuthRefreshExecutor(&executor);
+ ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client,
properties));
+ SessionContext context{
+ .session_id = "tenant-a",
+ .credentials = {{AuthProperties::kCredential.key(), "client:secret"}},
+ };
+
+ ICEBERG_UNWRAP_OR_FAIL(auto first, manager->ContextualSession(context,
parent));
+ ASSERT_EQ(task.wait_for(std::chrono::seconds(5)), std::future_status::ready);
+ // While its refresh is pending, the expired session stays cached.
+ ICEBERG_UNWRAP_OR_FAIL(auto pending, manager->ContextualSession(context,
parent));
+ EXPECT_EQ(pending, first);
+
+ {
+ test::FakeRetryEnvironment fake_retry;
+ ScopedRetryTestHooks hooks(fake_retry.hooks());
+ std::move(task.get())(); // every refresh attempt fails
+ }
+ recovered = true;
+
+ // The refresh gave up and the token has expired, so a new session is loaded.
+ ICEBERG_UNWRAP_OR_FAIL(auto reloaded, manager->ContextualSession(context,
parent));
+ EXPECT_NE(reloaded, first);
+ ICEBERG_UNWRAP_OR_FAIL(auto request, reloaded->Authenticate({}));
+ EXPECT_EQ(request.headers.at("Authorization"), "Bearer fresh");
+ EXPECT_THAT(manager->Close(), IsOk());
+}
+
+namespace {
+
+/// Answers token exchanges with a token that records the subject and actor.
+class ExchangeHttpClient : public HttpClient {
+ public:
+ Result<HttpResponse> PostForm(const std::string&,
+ const std::unordered_map<std::string,
std::string>& form,
+ const std::unordered_map<std::string,
std::string>&,
+ const ErrorHandler&, AuthSession&) override {
+ ++requests;
+ auto actor = form.find("actor_token");
+ std::string token = "sub=" + form.at("subject_token") + ",act=" +
+ (actor == form.end() ? std::string("none") :
actor->second);
+ return HttpResponse(200, R"({"access_token":")" + token +
+
R"(","token_type":"bearer","issued_token_type":")" +
+ AuthProperties::kAccessTokenType + R"("})");
+ }
+ std::atomic<int> requests{0};
+};
+
+class TableSessionCacheKeyTest : public ::testing::Test {
+ protected:
+ void SetUp() override {
+ ICEBERG_UNWRAP_OR_FAIL(
+ manager_,
+ AuthManagers::Load("test-catalog", {{AuthProperties::kAuthType,
"oauth2"}}));
+ ICEBERG_UNWRAP_OR_FAIL(
+ catalog_, manager_->CatalogSession(
+ client_, {{AuthProperties::kAuthType, "oauth2"},
+ {AuthProperties::kToken.key(),
"catalog-token"}}));
+ }
+
+ void TearDown() override { EXPECT_THAT(manager_->Close(), IsOk()); }
+
+ /// A contextual session holding `token`, used as a table session parent.
+ Result<std::shared_ptr<AuthSession>> Parent(const std::string& session_id,
+ const std::string& token) {
+ return manager_->ContextualSession(
+ {.session_id = session_id,
+ .credentials = {{AuthProperties::kToken.key(), token}}},
+ catalog_);
+ }
+
+ Result<std::shared_ptr<AuthSession>> Table(
+ const std::unordered_map<std::string, std::string>& config,
+ std::shared_ptr<AuthSession> parent) {
+ return manager_->TableSession(table_, config, std::move(parent));
+ }
+
+ static std::string Authorization(AuthSession& session) {
+ auto request = session.Authenticate({});
+ return request.has_value() ? request->headers.at("Authorization") : "";
+ }
+
+ std::shared_ptr<ExchangeHttpClient> client_ =
std::make_shared<ExchangeHttpClient>();
+ std::unique_ptr<AuthManager> manager_;
+ std::shared_ptr<AuthSession> catalog_;
+ TableIdentifier table_{.ns = Namespace{{"db"}}, .name = "orders"};
+};
+
+} // namespace
+
+TEST_F(TableSessionCacheKeyTest, ExchangedSessionsAreKeyedByActor) {
+ ICEBERG_UNWRAP_OR_FAIL(auto alice, Parent("alice", "alice-token"));
+ ICEBERG_UNWRAP_OR_FAIL(auto bob, Parent("bob", "bob-token"));
+ std::unordered_map<std::string, std::string> config = {
+ {AuthProperties::kJwtTokenType, "table-jwt"}};
+
+ ICEBERG_UNWRAP_OR_FAIL(auto alice_first, Table(config, alice));
+ ICEBERG_UNWRAP_OR_FAIL(auto alice_again, Table(config, alice));
+ ICEBERG_UNWRAP_OR_FAIL(auto bob_session, Table(config, bob));
+
+ EXPECT_EQ(alice_again, alice_first);
+ EXPECT_NE(bob_session, alice_first);
+ EXPECT_EQ(client_->requests, 2);
+ EXPECT_EQ(Authorization(*alice_first), "Bearer
sub=table-jwt,act=alice-token");
+ EXPECT_EQ(Authorization(*bob_session), "Bearer sub=table-jwt,act=bob-token");
+}
+
+TEST_F(TableSessionCacheKeyTest, SessionsAreKeyedByFlowAndTokenType) {
+ ICEBERG_UNWRAP_OR_FAIL(auto alice, Parent("alice", "alice-token"));
+
+ ICEBERG_UNWRAP_OR_FAIL(auto bearer,
+ Table({{AuthProperties::kToken.key(), "same-value"}},
alice));
+ ICEBERG_UNWRAP_OR_FAIL(auto jwt,
+ Table({{AuthProperties::kJwtTokenType,
"same-value"}}, alice));
+ ICEBERG_UNWRAP_OR_FAIL(auto saml2,
+ Table({{AuthProperties::kSaml2TokenType,
"same-value"}}, alice));
+
+ EXPECT_NE(jwt, bearer);
+ EXPECT_NE(saml2, jwt);
+ EXPECT_EQ(client_->requests, 2);
+ EXPECT_EQ(Authorization(*bearer), "Bearer same-value");
+ EXPECT_EQ(Authorization(*jwt), "Bearer sub=same-value,act=alice-token");
+}
+
} // namespace iceberg::rest::auth
diff --git a/src/iceberg/test/auth_session_cache_test.cc
b/src/iceberg/test/auth_session_cache_test.cc
new file mode 100644
index 00000000..074459e7
--- /dev/null
+++ b/src/iceberg/test/auth_session_cache_test.cc
@@ -0,0 +1,582 @@
+/*
+ * 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.
+ */
+
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <cstdint>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <set>
+#include <stdexcept>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include "iceberg/catalog/rest/auth/auth_session.h"
+#include "iceberg/catalog/rest/auth/auth_session_cache_internal.h"
+#include "iceberg/catalog/rest/auth/auth_session_internal.h"
+#include "iceberg/catalog/rest/http_client.h"
+#include "iceberg/test/executor.h"
+#include "iceberg/test/matchers.h"
+
+namespace iceberg::rest::auth {
+
+namespace {
+
+using internal::AuthSessionCache;
+using namespace std::chrono_literals;
+
+constexpr auto kTimeout = std::chrono::milliseconds(1000);
+constexpr auto kTokenLifetime = std::chrono::seconds(60);
+
+/// Starts at the real steady clock so it is comparable with
OAuth2Session::ExpiresAt().
+class FakeClock {
+ public:
+ std::chrono::steady_clock::time_point Now() const {
+ return
std::chrono::steady_clock::time_point(std::chrono::milliseconds(now_ms_));
+ }
+
+ AuthSessionCache::Clock AsClock() {
+ return [this] { return Now(); };
+ }
+
+ void Advance(std::chrono::milliseconds delta) { now_ms_ += delta.count(); }
+
+ private:
+ std::atomic<int64_t>
now_ms_{std::chrono::duration_cast<std::chrono::milliseconds>(
+
std::chrono::steady_clock::now().time_since_epoch())
+ .count()};
+};
+
+class RemovalRecorder {
+ public:
+ AuthSessionCache::RemovalListener Listener() {
+ return [this](const std::shared_ptr<AuthSession>& session) {
+ std::lock_guard lock(mutex_);
+ removed_.push_back(session);
+ };
+ }
+
+ std::vector<std::shared_ptr<AuthSession>> removed() const {
+ std::lock_guard lock(mutex_);
+ return removed_;
+ }
+
+ private:
+ mutable std::mutex mutex_;
+ std::vector<std::shared_ptr<AuthSession>> removed_;
+};
+
+/// One-shot latch used to hold a loader inside Get().
+class Gate {
+ public:
+ void Open() {
+ {
+ std::lock_guard lock(mutex_);
+ open_ = true;
+ }
+ cv_.notify_all();
+ }
+
+ void Wait() {
+ std::unique_lock lock(mutex_);
+ cv_.wait(lock, [this] { return open_; });
+ }
+
+ private:
+ std::mutex mutex_;
+ std::condition_variable cv_;
+ bool open_ = false;
+};
+
+AuthSessionCache::Loader CountingLoader(std::atomic<int>& calls) {
+ return [&calls]() -> Result<std::shared_ptr<AuthSession>> {
+ ++calls;
+ return AuthSession::MakeDefault({});
+ };
+}
+
+/// Loads an OAuth2 session whose token expires kTokenLifetime after `clock`
reads.
+/// The session refreshes on `refresh_executor` when it is not null.
+AuthSessionCache::Loader OAuth2Loader(std::atomic<int>& calls, const
FakeClock& clock,
+ Executor* refresh_executor) {
+ return [&calls, &clock, refresh_executor]() ->
Result<std::shared_ptr<AuthSession>> {
+ ++calls;
+ OAuthTokenResponse token_response{
+ .access_token = "opaque-token",
+ .token_type = "bearer",
+ .expires_in_secs = kTokenLifetime.count(),
+ };
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto session,
+ internal::OAuth2Session::Make(token_response,
+ {.token_endpoint =
"http://localhost/oauth/tokens";,
+ .refresh_executor = refresh_executor},
+ std::make_shared<HttpClient>(),
clock.Now()));
+ return session;
+ };
+}
+
+} // namespace
+
+class AuthSessionCacheTest : public ::testing::Test {
+ protected:
+ std::shared_ptr<AuthSessionCache> MakeCache(
+ std::chrono::milliseconds timeout = kTimeout) {
+ auto cache = AuthSessionCache::Make(timeout, recorder_.Listener(),
clock_.AsClock());
+ EXPECT_THAT(cache, IsOk());
+ return cache.value();
+ }
+
+ // Declared before any cache so they outlive it.
+ FakeClock clock_;
+ RemovalRecorder recorder_;
+};
+
+TEST_F(AuthSessionCacheTest, RejectsNegativeTimeout) {
+ EXPECT_THAT(AuthSessionCache::Make(-1ms, recorder_.Listener()),
+ IsError(ErrorKind::kInvalidArgument));
+}
+
+TEST_F(AuthSessionCacheTest, ReturnsCachedSessionOnHit) {
+ auto cache = MakeCache();
+ std::atomic<int> calls = 0;
+
+ ICEBERG_UNWRAP_OR_FAIL(auto first, cache->Get("tenant-a",
CountingLoader(calls)));
+ ICEBERG_UNWRAP_OR_FAIL(auto second, cache->Get("tenant-a",
CountingLoader(calls)));
+
+ EXPECT_EQ(first, second);
+ EXPECT_EQ(calls, 1);
+ EXPECT_EQ(cache->size(), 1);
+}
+
+TEST_F(AuthSessionCacheTest, KeysAreIsolated) {
+ auto cache = MakeCache();
+ std::atomic<int> calls = 0;
+
+ ICEBERG_UNWRAP_OR_FAIL(auto a, cache->Get("tenant-a",
CountingLoader(calls)));
+ ICEBERG_UNWRAP_OR_FAIL(auto b, cache->Get("tenant-b",
CountingLoader(calls)));
+
+ EXPECT_NE(a, b);
+ EXPECT_EQ(calls, 2);
+}
+
+TEST_F(AuthSessionCacheTest, FailedLoadIsNotCached) {
+ auto cache = MakeCache();
+ auto failing = []() -> Result<std::shared_ptr<AuthSession>> {
+ return AuthenticationFailed("token endpoint unavailable");
+ };
+ std::atomic<int> calls = 0;
+
+ EXPECT_THAT(cache->Get("tenant-a", failing),
IsError(ErrorKind::kAuthenticationFailed));
+ EXPECT_EQ(cache->size(), 0);
+ EXPECT_THAT(cache->Get("tenant-a", CountingLoader(calls)), IsOk());
+ EXPECT_EQ(calls, 1);
+}
+
+TEST_F(AuthSessionCacheTest, NullSessionIsRejected) {
+ auto cache = MakeCache();
+ auto null_loader = []() -> Result<std::shared_ptr<AuthSession>> { return
nullptr; };
+
+ EXPECT_THAT(cache->Get("tenant-a", null_loader),
IsError(ErrorKind::kInvalid));
+ EXPECT_EQ(cache->size(), 0);
+}
+
+TEST_F(AuthSessionCacheTest, ExpiresAfterIdleTimeout) {
+ auto cache = MakeCache();
+ std::atomic<int> calls = 0;
+ ICEBERG_UNWRAP_OR_FAIL(auto first, cache->Get("tenant-a",
CountingLoader(calls)));
+
+ // Each access resets the idle timer, so the entry outlives the timeout in
total.
+ clock_.Advance(kTimeout - 1ms);
+ ICEBERG_UNWRAP_OR_FAIL(auto hit, cache->Get("tenant-a",
CountingLoader(calls)));
+ clock_.Advance(kTimeout - 1ms);
+ ICEBERG_UNWRAP_OR_FAIL(auto still_hit, cache->Get("tenant-a",
CountingLoader(calls)));
+ EXPECT_EQ(hit, first);
+ EXPECT_EQ(still_hit, first);
+ EXPECT_EQ(calls, 1);
+
+ clock_.Advance(kTimeout);
+ ICEBERG_UNWRAP_OR_FAIL(auto reloaded, cache->Get("tenant-a",
CountingLoader(calls)));
+ EXPECT_NE(reloaded, first);
+ EXPECT_EQ(calls, 2);
+ EXPECT_THAT(recorder_.removed(), ::testing::ElementsAre(first));
+}
+
+TEST_F(AuthSessionCacheTest, SweepRemovesOnlyExpiredEntries) {
+ auto cache = MakeCache();
+ std::atomic<int> calls = 0;
+ ICEBERG_UNWRAP_OR_FAIL(auto idle, cache->Get("idle", CountingLoader(calls)));
+ ICEBERG_UNWRAP_OR_FAIL(auto active, cache->Get("active",
CountingLoader(calls)));
+
+ clock_.Advance(kTimeout - 1ms);
+ ASSERT_THAT(cache->Get("active", CountingLoader(calls)), IsOk());
+ clock_.Advance(1ms);
+ cache->Sweep();
+
+ EXPECT_EQ(cache->size(), 1);
+ EXPECT_THAT(recorder_.removed(), ::testing::ElementsAre(idle));
+ ICEBERG_UNWRAP_OR_FAIL(auto still_active, cache->Get("active",
CountingLoader(calls)));
+ EXPECT_EQ(still_active, active);
+}
+
+TEST_F(AuthSessionCacheTest, NonRefreshableSessionExpiresWithToken) {
+ auto cache = MakeCache(std::chrono::hours(1));
+ std::atomic<int> calls = 0;
+ auto loader = OAuth2Loader(calls, clock_, /*refresh_executor=*/nullptr);
+ ICEBERG_UNWRAP_OR_FAIL(auto first, cache->Get("tenant-a", loader));
+
+ clock_.Advance(kTokenLifetime - 1ms);
+ ICEBERG_UNWRAP_OR_FAIL(auto hit, cache->Get("tenant-a", loader));
+ EXPECT_EQ(hit, first);
+
+ // Sweep() only reclaims idle entries; the expired token is replaced on
access.
+ clock_.Advance(1ms);
+ cache->Sweep();
+ EXPECT_EQ(cache->size(), 1);
+ EXPECT_TRUE(recorder_.removed().empty());
+
+ ICEBERG_UNWRAP_OR_FAIL(auto reloaded, cache->Get("tenant-a", loader));
+ EXPECT_NE(reloaded, first);
+ EXPECT_EQ(calls, 2);
+ EXPECT_THAT(recorder_.removed(), ::testing::ElementsAre(first));
+}
+
+TEST_F(AuthSessionCacheTest, RefreshableSessionOutlivesToken) {
+ // Must outlive the cache. The refresh is due long after this test ends, so
+ // nothing is submitted.
+ test::MockExecutor executor;
+ EXPECT_CALL(executor, Submit).Times(0);
+ auto cache = MakeCache(std::chrono::hours(1));
+ std::atomic<int> calls = 0;
+ auto loader = OAuth2Loader(calls, clock_, &executor);
+ ICEBERG_UNWRAP_OR_FAIL(auto first, cache->Get("tenant-a", loader));
+
+ // Its refresh may still be in flight; eviction would stop it for other
holders.
+ clock_.Advance(kTokenLifetime + 1s);
+ cache->Sweep();
+ ICEBERG_UNWRAP_OR_FAIL(auto hit, cache->Get("tenant-a", loader));
+
+ EXPECT_EQ(hit, first);
+ EXPECT_EQ(calls, 1);
+ EXPECT_TRUE(recorder_.removed().empty());
+}
+
+TEST_F(AuthSessionCacheTest, GetEvictsIdleEntriesWithoutPeriodicSweep) {
+ auto cache = MakeCache();
+ std::atomic<int> calls = 0;
+ ICEBERG_UNWRAP_OR_FAIL(auto a, cache->Get("a", CountingLoader(calls)));
+ ICEBERG_UNWRAP_OR_FAIL(auto b, cache->Get("b", CountingLoader(calls)));
+ ICEBERG_UNWRAP_OR_FAIL(auto c, cache->Get("c", CountingLoader(calls)));
+
+ // Keys that are never accessed again are reclaimed when another key is
added.
+ clock_.Advance(kTimeout);
+ ASSERT_THAT(cache->Get("new", CountingLoader(calls)), IsOk());
+
+ EXPECT_EQ(cache->size(), 1);
+ EXPECT_THAT(recorder_.removed(), ::testing::UnorderedElementsAre(a, b, c));
+}
+
+TEST_F(AuthSessionCacheTest, EvictionFollowsAccessOrder) {
+ auto cache = MakeCache();
+ std::atomic<int> calls = 0;
+ ICEBERG_UNWRAP_OR_FAIL(auto a, cache->Get("a", CountingLoader(calls)));
+ ICEBERG_UNWRAP_OR_FAIL(auto b, cache->Get("b", CountingLoader(calls)));
+ ICEBERG_UNWRAP_OR_FAIL(auto c, cache->Get("c", CountingLoader(calls)));
+
+ // Accessing a moves it behind b and c, so only b and c become idle.
+ clock_.Advance(kTimeout / 2);
+ ASSERT_THAT(cache->Get("a", CountingLoader(calls)), IsOk());
+ clock_.Advance(kTimeout * 6 / 10);
+ cache->Sweep();
+
+ EXPECT_EQ(cache->size(), 1);
+ EXPECT_THAT(recorder_.removed(), ::testing::ElementsAre(b, c));
+ ICEBERG_UNWRAP_OR_FAIL(auto still_cached, cache->Get("a",
CountingLoader(calls)));
+ EXPECT_EQ(still_cached, a);
+ EXPECT_EQ(calls, 3);
+}
+
+TEST_F(AuthSessionCacheTest, GetDoesNotEvictWithPeriodicSweep) {
+ test::ThreadExecutor executor; // must outlive the cache
+ auto cache = MakeCache();
+ ASSERT_THAT(cache->StartPeriodicSweep(executor, std::chrono::hours(1)),
IsOk());
+ std::atomic<int> calls = 0;
+ ASSERT_THAT(cache->Get("a", CountingLoader(calls)), IsOk());
+
+ // Idle entries are left to the periodic sweep.
+ clock_.Advance(kTimeout * 2);
+ ASSERT_THAT(cache->Get("b", CountingLoader(calls)), IsOk());
+
+ EXPECT_EQ(cache->size(), 2);
+ EXPECT_TRUE(recorder_.removed().empty());
+ cache->Close();
+}
+
+TEST_F(AuthSessionCacheTest, ConcurrentGetsForSameKeyLoadOnce) {
+ auto cache = MakeCache();
+ std::atomic<int> calls = 0;
+ Gate entered;
+ Gate release;
+ auto blocking_loader = [&]() -> Result<std::shared_ptr<AuthSession>> {
+ ++calls;
+ entered.Open();
+ release.Wait();
+ return AuthSession::MakeDefault({});
+ };
+
+ constexpr int kThreads = 8;
+ std::vector<std::shared_ptr<AuthSession>> results(kThreads);
+ std::vector<std::thread> threads;
+ for (int i = 0; i < kThreads; ++i) {
+ threads.emplace_back([&, i] {
+ auto session = cache->Get("tenant-a", blocking_loader);
+ if (session.has_value()) {
+ results[i] = session.value();
+ }
+ });
+ }
+ entered.Wait();
+ std::this_thread::sleep_for(20ms); // let the other callers queue up
+ release.Open();
+ for (auto& thread : threads) {
+ thread.join();
+ }
+
+ EXPECT_EQ(calls, 1);
+ ASSERT_NE(results[0], nullptr);
+ for (const auto& session : results) {
+ EXPECT_EQ(session, results[0]);
+ }
+}
+
+TEST_F(AuthSessionCacheTest, WaitersRetryAfterFailedLoad) {
+ auto cache = MakeCache();
+ std::atomic<int> calls = 0;
+ Gate entered;
+ Gate release;
+ auto loader = [&]() -> Result<std::shared_ptr<AuthSession>> {
+ if (++calls == 1) {
+ entered.Open();
+ release.Wait();
+ return AuthenticationFailed("first attempt fails");
+ }
+ return AuthSession::MakeDefault({});
+ };
+
+ Result<std::shared_ptr<AuthSession>> first_result = nullptr;
+ std::thread first([&] { first_result = cache->Get("tenant-a", loader); });
+ entered.Wait();
+ Result<std::shared_ptr<AuthSession>> waiter_result = nullptr;
+ std::thread waiter([&] { waiter_result = cache->Get("tenant-a", loader); });
+ std::this_thread::sleep_for(20ms);
+ release.Open();
+ first.join();
+ waiter.join();
+
+ EXPECT_THAT(first_result, IsError(ErrorKind::kAuthenticationFailed));
+ EXPECT_THAT(waiter_result, IsOk());
+ EXPECT_EQ(calls, 2);
+ EXPECT_EQ(cache->size(), 1);
+}
+
+TEST_F(AuthSessionCacheTest, LoaderExceptionDoesNotBlockLaterGets) {
+ auto cache = MakeCache();
+ std::atomic<int> calls = 0;
+ Gate entered;
+ Gate release;
+ auto loader = [&]() -> Result<std::shared_ptr<AuthSession>> {
+ if (++calls == 1) {
+ entered.Open();
+ release.Wait();
+ throw std::runtime_error("HTTP library failure");
+ }
+ return AuthSession::MakeDefault({});
+ };
+
+ auto first =
+ std::async(std::launch::async, [&] { return cache->Get("tenant-a",
loader); });
+ entered.Wait();
+ auto waiter =
+ std::async(std::launch::async, [&] { return cache->Get("tenant-a",
loader); });
+ std::this_thread::sleep_for(20ms);
+ release.Open();
+
+ if (waiter.wait_for(5s) != std::future_status::ready) {
+ cache->Close(); // Wake the waiter so the test fails instead of hanging.
+ FAIL() << "a waiter stayed blocked after the loader threw";
+ }
+ auto first_result = first.get();
+ ASSERT_THAT(first_result, IsError(ErrorKind::kUnknownError));
+ EXPECT_THAT(first_result.error().message, ::testing::HasSubstr("HTTP library
failure"));
+ ICEBERG_UNWRAP_OR_FAIL(auto loaded, waiter.get());
+
+ ICEBERG_UNWRAP_OR_FAIL(auto hit, cache->Get("tenant-a", loader));
+ EXPECT_EQ(hit, loaded);
+ EXPECT_EQ(calls, 2);
+ EXPECT_EQ(cache->size(), 1);
+}
+
+TEST_F(AuthSessionCacheTest, CloseRemovesAllEntriesOnce) {
+ auto cache = MakeCache();
+ std::atomic<int> calls = 0;
+ ICEBERG_UNWRAP_OR_FAIL(auto a, cache->Get("tenant-a",
CountingLoader(calls)));
+ ICEBERG_UNWRAP_OR_FAIL(auto b, cache->Get("tenant-b",
CountingLoader(calls)));
+
+ cache->Close();
+ cache->Close();
+
+ EXPECT_EQ(cache->size(), 0);
+ EXPECT_THAT(recorder_.removed(), ::testing::UnorderedElementsAre(a, b));
+ EXPECT_THAT(cache->Get("tenant-a", CountingLoader(calls)),
+ IsError(ErrorKind::kInvalid));
+ EXPECT_EQ(calls, 2);
+}
+
+TEST_F(AuthSessionCacheTest, LoadFinishingAfterCloseIsRemovedAndWaitersFail) {
+ auto cache = MakeCache();
+ Gate entered;
+ Gate release;
+ std::shared_ptr<AuthSession> loaded;
+ auto loader = [&]() -> Result<std::shared_ptr<AuthSession>> {
+ entered.Open();
+ release.Wait();
+ loaded = AuthSession::MakeDefault({});
+ return loaded;
+ };
+
+ Result<std::shared_ptr<AuthSession>> loader_result = nullptr;
+ std::thread loading([&] { loader_result = cache->Get("tenant-a", loader); });
+ entered.Wait();
+ Result<std::shared_ptr<AuthSession>> waiter_result = nullptr;
+ std::thread waiter([&] { waiter_result = cache->Get("tenant-a", loader); });
+ std::this_thread::sleep_for(20ms);
+
+ cache->Close();
+ waiter.join(); // Close() wakes waiters before the load finishes.
+ EXPECT_THAT(waiter_result, IsError(ErrorKind::kInvalid));
+
+ release.Open();
+ loading.join();
+ EXPECT_THAT(loader_result, IsError(ErrorKind::kInvalid));
+ EXPECT_EQ(cache->size(), 0);
+ EXPECT_THAT(recorder_.removed(), ::testing::ElementsAre(loaded));
+}
+
+TEST_F(AuthSessionCacheTest, EverySessionIsRemovedExactlyOnceUnderContention) {
+ auto cache = MakeCache();
+ std::mutex loaded_mutex;
+ std::set<std::shared_ptr<AuthSession>> loaded;
+ auto loader = [&]() -> Result<std::shared_ptr<AuthSession>> {
+ auto session = AuthSession::MakeDefault({});
+ std::lock_guard lock(loaded_mutex);
+ loaded.insert(session);
+ return session;
+ };
+
+ std::atomic<bool> stop = false;
+ std::thread sweeper([&] {
+ while (!stop) {
+ cache->Sweep();
+ std::this_thread::yield();
+ }
+ });
+ std::vector<std::thread> workers;
+ for (int t = 0; t < 4; ++t) {
+ workers.emplace_back([&, t] {
+ for (int i = 0; i < 500; ++i) {
+ EXPECT_THAT(cache->Get("key-" + std::to_string((i + t) % 5), loader),
IsOk());
+ if (i % 10 == 0) {
+ clock_.Advance(kTimeout / 3);
+ }
+ }
+ });
+ }
+ for (auto& worker : workers) {
+ worker.join();
+ }
+ stop = true;
+ sweeper.join();
+ cache->Close();
+
+ auto removed = recorder_.removed();
+ std::set<std::shared_ptr<AuthSession>> unique_removed(removed.begin(),
removed.end());
+ EXPECT_EQ(unique_removed.size(), removed.size()) << "a session was removed
twice";
+ EXPECT_EQ(unique_removed, loaded);
+}
+
+TEST_F(AuthSessionCacheTest, PeriodicSweepRemovesExpiredEntries) {
+ test::ThreadExecutor executor; // must outlive the cache
+ auto cache = MakeCache();
+ std::atomic<int> calls = 0;
+ ICEBERG_UNWRAP_OR_FAIL(auto idle, cache->Get("idle", CountingLoader(calls)));
+ clock_.Advance(kTimeout);
+
+ ASSERT_THAT(cache->StartPeriodicSweep(executor, 10ms), IsOk());
+
+ // Sweep invokes the listener after erasing the entry, so wait for the
listener.
+ auto deadline = std::chrono::steady_clock::now() + 5s;
+ while (recorder_.removed().empty() && std::chrono::steady_clock::now() <
deadline) {
+ std::this_thread::sleep_for(5ms);
+ }
+ EXPECT_EQ(cache->size(), 0);
+ EXPECT_THAT(recorder_.removed(), ::testing::ElementsAre(idle));
+}
+
+TEST_F(AuthSessionCacheTest, PeriodicSweepContinuesAfterSubmissionFailure) {
+ test::ThreadExecutor executor(IOError("executor rejected the task"));
+ auto cache = MakeCache();
+ ASSERT_THAT(cache->StartPeriodicSweep(executor, 10ms), IsOk());
+
+ auto deadline = std::chrono::steady_clock::now() + 5s;
+ while (executor.submit_count() < 3 && std::chrono::steady_clock::now() <
deadline) {
+ std::this_thread::sleep_for(5ms);
+ }
+ EXPECT_GE(executor.submit_count(), 3);
+ cache->Close();
+}
+
+TEST_F(AuthSessionCacheTest, PeriodicSweepArgumentsAndLifecycle) {
+ test::ThreadExecutor executor;
+ auto cache = MakeCache();
+
+ EXPECT_THAT(cache->StartPeriodicSweep(executor, 0ms),
+ IsError(ErrorKind::kInvalidArgument));
+ EXPECT_THAT(cache->StartPeriodicSweep(executor, 10ms), IsOk());
+ EXPECT_THAT(cache->StartPeriodicSweep(executor, 10ms), IsOk()); // no-op
+
+ // Destroying the cache cancels the pending sweep.
+ cache.reset();
+ int submitted = executor.submit_count();
+ ASSERT_TRUE(test::WaitForSharedTimer(30ms));
+ EXPECT_EQ(executor.submit_count(), submitted);
+
+ auto closed = MakeCache();
+ closed->Close();
+ EXPECT_THAT(closed->StartPeriodicSweep(executor, 10ms), IsOk()); // no-op
+ ASSERT_TRUE(test::WaitForSharedTimer(30ms));
+ EXPECT_EQ(executor.submit_count(), submitted);
+}
+
+} // namespace iceberg::rest::auth
diff --git a/src/iceberg/test/rest_catalog_integration_test.cc
b/src/iceberg/test/rest_catalog_integration_test.cc
index c03c9c27..0e3ba6a3 100644
--- a/src/iceberg/test/rest_catalog_integration_test.cc
+++ b/src/iceberg/test/rest_catalog_integration_test.cc
@@ -313,6 +313,13 @@ TEST_F(RestCatalogIntegrationTest,
OAuthTableTypedTokenEndToEnd) {
auto info = child->OAuth2Info();
ASSERT_TRUE(info.has_value());
EXPECT_EQ(info->issued_token_type, auth::AuthProperties::kAccessTokenType);
+
+ // The same typed token reuses the cached session instead of exchanging
again.
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto cached,
+ manager->TableSession(
+ table, {{auth::AuthProperties::kJwtTokenType, "table-jwt-token"}},
parent));
+ EXPECT_EQ(cached, child);
}
TEST_F(RestCatalogIntegrationTest, OAuthTokenExchangeWithoutActorEndToEnd) {
