feat(auth): implement AuthSessionCache for OAuth2 child sessions

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

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 =

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.

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