refactor(auth): refresh OAuth2 tokens on a caller-provided executor

src/iceberg/CMakeLists.txt

@@ -155,6 +155,7 @@ set(ICEBERG_SOURCES
     util/murmurhash3_internal.cc
     util/property_util.cc
     util/retry_util.cc
+    util/scheduled_executor.cc
     util/snapshot_util.cc
     util/string_util.cc
     util/struct_like_set.cc

src/iceberg/catalog/rest/CMakeLists.txt

@@ -24,7 +24,6 @@ set(ICEBERG_REST_SOURCES
     auth/auth_session.cc
     auth/oauth2_util.cc
     auth/sigv4_manager.cc
-    auth/token_refresh_scheduler.cc
     catalog_properties.cc
     endpoint.cc
     error_handlers.cc

src/iceberg/catalog/rest/auth/auth_manager.cc

@@ -140,6 +140,10 @@ Result<std::unique_ptr<AuthManager>> MakeBasicAuthManager(
 /// \brief OAuth2 authentication manager.
 class OAuth2Manager : public AuthManager {
  public:
+  void SetAuthRefreshExecutor(Executor* executor) override {
+    refresh_executor_ = executor;
+  }
+
   Result<std::shared_ptr<AuthSession>> InitSession(
       std::shared_ptr<HttpClient> init_client,
       const std::unordered_map<std::string, std::string>& properties) override {
@@ -178,10 +182,7 @@ class OAuth2Manager : public AuthManager {
     refresh_client_ = std::move(shared_client);
     // Reuse the token response and start time from the init phase.
     if (auth_response_.has_value()) {
-      return internal::MakeOAuth2Session(
-          *auth_response_, config.oauth2_server_uri(), config.client_id(),
-          config.client_secret(), config.scope(), config.keep_refreshed(),
-          config.optional_oauth_params(), refresh_client_, start_time_);
+      return MakeSession(*auth_response_, config, config.keep_refreshed(), start_time_);
     }
 
     // TODO(lishuxu): Honor token-refresh-enabled for catalog bearer tokens, matching
@@ -192,10 +193,7 @@ class OAuth2Manager : public AuthManager {
           .token_type = "bearer",
           .issued_token_type = AuthProperties::kAccessTokenType,
       };
-      return AuthSession::MakeOAuth2(token_response, config.oauth2_server_uri(),
-                                     config.client_id(), config.client_secret(),
-                                     config.scope(), /*keep_refreshed=*/false,
-                                     config.optional_oauth_params(), refresh_client_);
+      return MakeSession(token_response, config, /*keep_refreshed=*/false);
     }
 
     // Fetch a new token using client_credentials grant.
@@ -206,10 +204,7 @@ class OAuth2Manager : public AuthManager {
       ICEBERG_ASSIGN_OR_RAISE(
           token_response,
           OAuth2Util::FetchToken(*refresh_client_, *base_session, config));
-      return AuthSession::MakeOAuth2(token_response, config.oauth2_server_uri(),
-                                     config.client_id(), config.client_secret(),
-                                     config.scope(), config.keep_refreshed(),
-                                     config.optional_oauth_params(), refresh_client_);
+      return MakeSession(token_response, config, config.keep_refreshed());
     }
 
     return MakeSession(AccessTokenResponse(""), config, /*keep_refreshed=*/false);
@@ -256,13 +251,16 @@ class OAuth2Manager : public AuthManager {
 
   Result<std::shared_ptr<AuthSession>> MakeSession(
       const OAuthTokenResponse& token_response, const AuthProperties& config,
-      bool keep_refreshed) const {
+      bool keep_refreshed,
+      std::optional<std::chrono::steady_clock::time_point> token_request_started_at =
+          std::nullopt) const {
     ICEBERG_PRECHECK(refresh_client_ != nullptr,
                      "OAuth2 catalog session must be initialized before child sessions");
-    return AuthSession::MakeOAuth2(token_response, config.oauth2_server_uri(),
-                                   config.client_id(), config.client_secret(),
-                                   config.scope(), keep_refreshed,
-                                   config.optional_oauth_params(), refresh_client_);
+    return internal::MakeOAuth2Session(token_response, config.oauth2_server_uri(),
+                                       config.client_id(), config.client_secret(),
+                                       config.scope(), keep_refreshed,
+                                       config.optional_oauth_params(), refresh_client_,
+                                       token_request_started_at, refresh_executor_);
   }
 
   Result<std::shared_ptr<AuthSession>> MaybeCreateChildSession(
@@ -319,6 +317,7 @@ class OAuth2Manager : public AuthManager {
   std::optional<OAuthTokenResponse> auth_response_;
   std::optional<std::chrono::steady_clock::time_point> start_time_;
   std::shared_ptr<HttpClient> refresh_client_;
+  Executor* refresh_executor_ = nullptr;
 };
 
 Result<std::unique_ptr<AuthManager>> MakeOAuth2Manager(

src/iceberg/catalog/rest/auth/auth_manager.h

@@ -97,6 +97,10 @@ class ICEBERG_REST_EXPORT AuthManager {
   ///
   /// \return Status of the close operation.
   virtual Status Close() { return {}; }
+
+  /// \brief Set the borrowed refresh executor before InitSession.
+  /// It must outlive this manager and its sessions. Wrappers must forward this call.
+  virtual void SetAuthRefreshExecutor(Executor* /*executor*/) {}
 };
 
 }  // namespace iceberg::rest::auth

src/iceberg/catalog/rest/auth/auth_session.cc

@@ -19,18 +19,13 @@
 
 #include "iceberg/catalog/rest/auth/auth_session.h"
 
-#include <algorithm>
 #include <chrono>
-#include <condition_variable>
 #include <memory>
-#include <mutex>
-#include <shared_mutex>
 #include <utility>
 
 #include "iceberg/catalog/rest/auth/auth_properties.h"
 #include "iceberg/catalog/rest/auth/auth_session_internal.h"
 #include "iceberg/catalog/rest/auth/oauth2_util.h"
-#include "iceberg/catalog/rest/auth/token_refresh_scheduler.h"
 #include "iceberg/catalog/rest/http_client.h"
 #include "iceberg/util/macros.h"
 
@@ -67,10 +62,10 @@ Result<std::shared_ptr<AuthSession>> AuthSession::MakeOAuth2(
     const std::string& client_id, const std::string& client_secret,
     const std::string& scope, bool keep_refreshed,
     const std::unordered_map<std::string, std::string>& optional_oauth_params,
-    std::shared_ptr<HttpClient> client) {
+    std::shared_ptr<HttpClient> client, Executor* auth_executor) {
   return internal::MakeOAuth2Session(
       initial_token, token_endpoint, client_id, client_secret, scope, keep_refreshed,
-      optional_oauth_params, std::move(client), std::nullopt);
+      optional_oauth_params, std::move(client), std::nullopt, auth_executor);
 }
 
 Result<std::shared_ptr<AuthSession>> internal::MakeOAuth2Session(
@@ -79,14 +74,19 @@ Result<std::shared_ptr<AuthSession>> internal::MakeOAuth2Session(
     const std::string& scope, bool keep_refreshed,
     const std::unordered_map<std::string, std::string>& optional_oauth_params,
     std::shared_ptr<HttpClient> client,
-    std::optional<std::chrono::steady_clock::time_point> token_request_started_at) {
+    std::optional<std::chrono::steady_clock::time_point> token_request_started_at,
+    Executor* refresh_executor) {
+  ICEBERG_PRECHECK(!keep_refreshed || refresh_executor != nullptr,
+                   "OAuth2 token refresh is enabled but no auth executor was provided; "
+                   "pass an executor or set {}=false",
+                   AuthProperties::kKeepRefreshed.key());
   internal::OAuth2Session::Config config{
       .token_endpoint = token_endpoint,
       .client_id = client_id,
       .client_secret = client_secret,
       .scope = scope,
       .optional_oauth_params = optional_oauth_params,
-      .keep_refreshed = keep_refreshed,
+      .refresh_executor = keep_refreshed ? refresh_executor : nullptr,
   };
   ICEBERG_ASSIGN_OR_RAISE(auto session, internal::OAuth2Session::Make(
                                             initial_token, std::move(config),

src/iceberg/catalog/rest/auth/auth_session.h

@@ -28,6 +28,7 @@
 #include "iceberg/catalog/rest/iceberg_rest_export.h"
 #include "iceberg/catalog/rest/type_fwd.h"
 #include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
 
 /// \file iceberg/catalog/rest/auth/auth_session.h
 /// \brief Authentication session interface for REST catalog.
@@ -68,13 +69,8 @@ class ICEBERG_REST_EXPORT AuthSession {
   /// \brief Return OAuth2 metadata when this is an OAuth2 session.
   virtual std::optional<OAuth2SessionInfo> OAuth2Info() const { return std::nullopt; }
 
-  /// \brief Close the session and release any resources.
-  ///
-  /// This method is called when the session is no longer needed. For stateful
-  /// sessions (e.g., OAuth2 with token refresh), this should stop any background
-  /// threads and release resources.
-  ///
-  /// \return Status indicating success or failure of closing the session.
+  /// \brief Stop background refresh and release session resources.
+  /// \return Status of the close operation.
   virtual Status Close() { return {}; }
 
   /// \brief Create a session with static headers.
@@ -90,10 +86,8 @@ class ICEBERG_REST_EXPORT AuthSession {
 
   /// \brief Create an OAuth2 session with automatic token refresh.
   ///
-  /// This factory method creates a session that holds an access token and,
-  /// when keep_refreshed is enabled, schedules background refresh based on
-  /// token expiration. Authenticate() uses the latest cached Authorization
-  /// header and does not perform a synchronous token refresh.
+  /// Refreshes expiring tokens in the background when keep_refreshed is true.
+  /// Authenticate() uses the cached token without refreshing it.
   ///
   /// \param initial_token The initial token response from FetchToken().
   /// \param token_endpoint Full URL of the OAuth2 token endpoint for refresh.
@@ -104,13 +98,16 @@ class ICEBERG_REST_EXPORT AuthSession {
   /// \param optional_oauth_params Optional OAuth params (audience, resource) for refresh.
   /// \param client HTTP client for making refresh requests. The session retains
   ///        ownership of the client.
-  /// \return A new session that manages token lifecycle automatically.
+  /// \param auth_executor Borrowed refresh executor; required if keep_refreshed is true.
+  /// Ignored otherwise. Must enqueue asynchronously and outlive the session and its
+  /// tasks.
+  /// \return The session, or InvalidArgument if refresh is enabled without an executor.
   static Result<std::shared_ptr<AuthSession>> MakeOAuth2(
       const OAuthTokenResponse& initial_token, const std::string& token_endpoint,
       const std::string& client_id, const std::string& client_secret,
       const std::string& scope, bool keep_refreshed,
       const std::unordered_map<std::string, std::string>& optional_oauth_params,
-      std::shared_ptr<HttpClient> client);
+      std::shared_ptr<HttpClient> client, Executor* auth_executor = nullptr);
 };
 
 }  // namespace iceberg::rest::auth