From c0f1a6edcfd0315e565a9ac07c2e1f588494b486 Mon Sep 17 00:00:00 2001
From: Li Jiajia <plusplusjiajia@alibaba-inc.com>
Date: Mon, 24 Aug 2026 00:43:07 -0400
Subject: [PATCH 1/2] feat(rest): load table credentials from the
 LoadCredentials endpoint

---
 src/iceberg/catalog/rest/json_serde.cc        | 42 ++++++++---
 .../catalog/rest/json_serde_internal.h        |  4 ++
 src/iceberg/catalog/rest/rest_catalog.cc      | 70 +++++++++++++++++--
 src/iceberg/catalog/rest/rest_catalog.h       | 14 +++-
 src/iceberg/catalog/rest/rest_file_io.cc      |  8 ++-
 src/iceberg/catalog/rest/rest_file_io.h       |  6 +-
 src/iceberg/catalog/rest/types.h              | 15 ++++
 src/iceberg/file_io.h                         |  5 ++
 src/iceberg/resolving_file_io.cc              | 34 +++++++--
 src/iceberg/resolving_file_io.h               | 19 +++--
 src/iceberg/storage_credential.h              |  7 ++
 src/iceberg/test/resolving_file_io_test.cc    | 52 ++++++++++++++
 src/iceberg/test/rest_file_io_test.cc         | 34 +++++++++
 src/iceberg/test/rest_json_serde_test.cc      | 26 +++++++
 14 files changed, 307 insertions(+), 29 deletions(-)

diff --git a/src/iceberg/catalog/rest/json_serde.cc b/src/iceberg/catalog/rest/json_serde.cc
index 3ce753f18..098f2ba08 100644
--- a/src/iceberg/catalog/rest/json_serde.cc
+++ b/src/iceberg/catalog/rest/json_serde.cc
@@ -152,6 +152,25 @@ Result<StorageCredential> StorageCredentialFromJson(const nlohmann::json& json)
   return credential;
 }
 
+/// \brief Reads the optional `storage-credentials` array shared by the
+/// LoadTable and LoadCredentials responses.
+Result<std::vector<StorageCredential>> StorageCredentialsFromJson(
+    const nlohmann::json& json) {
+  std::vector<StorageCredential> credentials;
+  auto it = json.find(kStorageCredentials);
+  if (it == json.end() || it->is_null()) {
+    return credentials;
+  }
+  if (!it->is_array()) {
+    return JsonParseError("Cannot parse storage credentials from non-array");
+  }
+  for (const auto& entry : *it) {
+    ICEBERG_ASSIGN_OR_RAISE(auto credential, StorageCredentialFromJson(entry));
+    credentials.push_back(std::move(credential));
+  }
+  return credentials;
+}
+
 template <typename Value>
 Result<std::map<int32_t, Value>> KeyValueMapFromJson(const nlohmann::json& json,
                                                      std::string_view key) {
@@ -738,19 +757,24 @@ Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {
   ICEBERG_ASSIGN_OR_RAISE(result.metadata, TableMetadataFromJson(metadata_json));
   ICEBERG_ASSIGN_OR_RAISE(result.config,
                           GetJsonValueOrDefault<decltype(result.config)>(json, kConfig));
-  if (auto it = json.find(kStorageCredentials); it != json.end() && !it->is_null()) {
-    if (!it->is_array()) {
-      return JsonParseError("Cannot parse storage credentials from non-array");
-    }
-    for (const auto& entry : *it) {
-      ICEBERG_ASSIGN_OR_RAISE(auto cred, StorageCredentialFromJson(entry));
-      result.storage_credentials.push_back(std::move(cred));
-    }
-  }
+  ICEBERG_ASSIGN_OR_RAISE(result.storage_credentials, StorageCredentialsFromJson(json));
   ICEBERG_RETURN_UNEXPECTED(result.Validate());
   return result;
 }
 
+Result<LoadCredentialsResponse> LoadCredentialsResponseFromJson(
+    const nlohmann::json& json) {
+  // Required here, unlike in LoadTable: reading a malformed response as "no
+  // credentials" would look like a refresh that succeeded and dropped them.
+  if (auto it = json.find(kStorageCredentials); it == json.end() || it->is_null()) {
+    return JsonParseError("Missing '{}'", kStorageCredentials);
+  }
+  LoadCredentialsResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.storage_credentials, StorageCredentialsFromJson(json));
+  ICEBERG_RETURN_UNEXPECTED(response.Validate());
+  return response;
+}
+
 nlohmann::json ToJson(const ListNamespacesResponse& response) {
   nlohmann::json json;
   SetOptionalStringField(json, kNextPageToken, response.next_page_token);
diff --git a/src/iceberg/catalog/rest/json_serde_internal.h b/src/iceberg/catalog/rest/json_serde_internal.h
index 6e148e0d3..aedd2cd1a 100644
--- a/src/iceberg/catalog/rest/json_serde_internal.h
+++ b/src/iceberg/catalog/rest/json_serde_internal.h
@@ -71,6 +71,10 @@ template <>
 ICEBERG_REST_EXPORT Result<LoadTableResult> FromJson(const nlohmann::json& json);
 ICEBERG_REST_EXPORT Result<nlohmann::json> ToJson(const LoadTableResult& model);
 
+// Response-only model: a client never serializes it, so no ToJson.
+ICEBERG_REST_EXPORT Result<LoadCredentialsResponse> LoadCredentialsResponseFromJson(
+    const nlohmann::json& json);
+
 ICEBERG_REST_EXPORT Result<CreateTableRequest> CreateTableRequestFromJson(
     const nlohmann::json& json);
 template <>
diff --git a/src/iceberg/catalog/rest/rest_catalog.cc b/src/iceberg/catalog/rest/rest_catalog.cc
index 9635aebef..91c19d7b8 100644
--- a/src/iceberg/catalog/rest/rest_catalog.cc
+++ b/src/iceberg/catalog/rest/rest_catalog.cc
@@ -42,6 +42,7 @@
 #include "iceberg/catalog/rest/rest_util.h"
 #include "iceberg/catalog/rest/types.h"
 #include "iceberg/json_serde_internal.h"
+#include "iceberg/logging/log_macros.h"
 #include "iceberg/metrics/metrics_reporters.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/result.h"
@@ -508,12 +509,65 @@ Result<std::shared_ptr<auth::AuthSession>> RestCatalog::TableAuthSession(
                                      std::move(contextual_session));
 }
 
+StorageCredentialRefresher RestCatalog::MakeCredentialRefresher(
+    const TableIdentifier& identifier,
+    std::shared_ptr<auth::AuthSession> table_session) const {
+  if (!supported_endpoints_.contains(Endpoint::TableCredentials())) {
+    // Not an error, but it surfaces much later as credentials expiring.
+    ICEBERG_LOG_DEBUG(
+        "Catalog does not advertise {}; vended credentials for '{}' will not be "
+        "refreshed",
+        Endpoint::TableCredentials().ToString(), ToString(identifier));
+    return nullptr;
+  }
+  auto path = paths_->Credentials(identifier);
+  if (!path.has_value()) {
+    ICEBERG_LOG_WARN(
+        "Cannot build the credentials path for '{}' ({}); its vended credentials "
+        "will not be refreshed",
+        ToString(identifier), path.error().message);
+    return nullptr;
+  }
+  auto client = client_;
+  auto credentials_path = std::move(path.value());
+  auto session = std::move(table_session);
+  // The catalog's destructor closes the session, and a table's FileIO can
+  // outlive the table keeping the catalog alive. No cycle: the catalog's own
+  // FileIO never gets a refresher.
+  auto catalog = shared_from_this();
+  return [catalog, client, credentials_path,
+          session]() -> Result<std::vector<StorageCredential>> {
+    ICEBERG_ASSIGN_OR_RAISE(const auto response,
+                            client->Get(credentials_path, /*params=*/{}, /*headers=*/{},
+                                        *TableErrorHandler::Instance(), *session));
+    // Parse errors embed the offending input, and this body carries
+    // credentials; strip the message so it can never reach a log.
+    auto json = FromJsonString(response.body());
+    if (!json.has_value()) {
+      return JsonParseError("Malformed LoadCredentials response");
+    }
+    auto result = LoadCredentialsResponseFromJson(*json);
+    if (!result.has_value()) {
+      return std::unexpected<Error>(
+          {.kind = result.error().kind, .message = "Malformed LoadCredentials response"});
+    }
+    return std::move(result->storage_credentials);
+  };
+}
+
 Result<std::shared_ptr<FileIO>> RestCatalog::TableFileIO(
-    const SessionContext& /*context*/,
+    const SessionContext& /*context*/, const TableIdentifier& identifier,
     const std::unordered_map<std::string, std::string>& table_config,
-    const std::vector<StorageCredential>& storage_credentials) const {
+    const std::vector<StorageCredential>& storage_credentials,
+    std::shared_ptr<auth::AuthSession> table_session) const {
   if (!table_config.empty() || !storage_credentials.empty()) {
-    return MakeTableFileIO(config_.configs(), table_config, storage_credentials);
+    // Only vended credentials expire, so only they need a refresher.
+    StorageCredentialRefresher refresher;
+    if (!storage_credentials.empty()) {
+      refresher = MakeCredentialRefresher(identifier, std::move(table_session));
+    }
+    return MakeTableFileIO(config_.configs(), table_config, storage_credentials,
+                           std::move(refresher));
   }
 
   return file_io_;
@@ -772,11 +826,12 @@ Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
                           /*stage_create=*/true, *contextual_session));
   auto table_config = std::move(result.config);
   auto storage_credentials = std::move(result.storage_credentials);
-  ICEBERG_ASSIGN_OR_RAISE(auto table_io,
-                          TableFileIO(context, table_config, storage_credentials));
+  // Before the FileIO: refreshing its credentials reuses the table session.
   ICEBERG_ASSIGN_OR_RAISE(
       auto table_session,
       TableAuthSession(identifier, table_config, std::move(contextual_session)));
+  ICEBERG_ASSIGN_OR_RAISE(auto table_io, TableFileIO(context, identifier, table_config,
+                                                     storage_credentials, table_session));
   ICEBERG_ASSIGN_OR_RAISE(auto reporter, MakeTableReporter(identifier, table_session));
   auto table_catalog = std::make_shared<TableScopedCatalog>(
       shared_from_this(), context, identifier, table_config, std::move(table_session),
@@ -890,11 +945,12 @@ Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromLoadResult(
     std::shared_ptr<auth::AuthSession> contextual_session) {
   auto table_config = std::move(result.config);
   auto storage_credentials = std::move(result.storage_credentials);
-  ICEBERG_ASSIGN_OR_RAISE(auto table_io,
-                          TableFileIO(context, table_config, storage_credentials));
+  // Before the FileIO: refreshing its credentials reuses the table session.
   ICEBERG_ASSIGN_OR_RAISE(
       auto table_session,
       TableAuthSession(identifier, table_config, std::move(contextual_session)));
+  ICEBERG_ASSIGN_OR_RAISE(auto table_io, TableFileIO(context, identifier, table_config,
+                                                     storage_credentials, table_session));
   ICEBERG_ASSIGN_OR_RAISE(auto reporter, MakeTableReporter(identifier, table_session));
   auto table_catalog = std::make_shared<TableScopedCatalog>(
       shared_from_this(), context, identifier, table_config, table_session, table_io);
diff --git a/src/iceberg/catalog/rest/rest_catalog.h b/src/iceberg/catalog/rest/rest_catalog.h
index 65b0b5eab..bcb1be53a 100644
--- a/src/iceberg/catalog/rest/rest_catalog.h
+++ b/src/iceberg/catalog/rest/rest_catalog.h
@@ -85,9 +85,19 @@ class ICEBERG_REST_EXPORT RestCatalog final
       std::shared_ptr<auth::AuthSession> contextual_session);
 
   Result<std::shared_ptr<FileIO>> TableFileIO(
-      const SessionContext& context,
+      const SessionContext& context, const TableIdentifier& identifier,
       const std::unordered_map<std::string, std::string>& table_config,
-      const std::vector<StorageCredential>& storage_credentials) const;
+      const std::vector<StorageCredential>& storage_credentials,
+      std::shared_ptr<auth::AuthSession> table_session) const;
+
+  /// \brief Callback that reloads this table's vended credentials, or nullptr
+  /// when the catalog does not serve the LoadCredentials endpoint.
+  ///
+  /// Captures the catalog, so anything documented to outlive it -- the metrics
+  /// executor included -- must also outlive every FileIO holding a refresher.
+  StorageCredentialRefresher MakeCredentialRefresher(
+      const TableIdentifier& identifier,
+      std::shared_ptr<auth::AuthSession> table_session) const;
 
   Result<std::vector<Namespace>> ListNamespaces(const Namespace& ns,
                                                 auth::AuthSession& session) const;
diff --git a/src/iceberg/catalog/rest/rest_file_io.cc b/src/iceberg/catalog/rest/rest_file_io.cc
index cdc1204f1..2531008f3 100644
--- a/src/iceberg/catalog/rest/rest_file_io.cc
+++ b/src/iceberg/catalog/rest/rest_file_io.cc
@@ -21,6 +21,7 @@
 
 #include <string>
 #include <unordered_map>
+#include <utility>
 #include <vector>
 
 #include "iceberg/catalog/rest/types.h"
@@ -57,7 +58,8 @@ Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(const RestCatalogProperties& c
 Result<std::unique_ptr<FileIO>> MakeTableFileIO(
     const std::unordered_map<std::string, std::string>& catalog_config,
     const std::unordered_map<std::string, std::string>& table_config,
-    const std::vector<StorageCredential>& storage_credentials) {
+    const std::vector<StorageCredential>& storage_credentials,
+    StorageCredentialRefresher refresher) {
   const auto default_properties = MergeFileIOProperties(catalog_config, table_config);
   ICEBERG_ASSIGN_OR_RAISE(
       auto io, MakeCatalogFileIO(RestCatalogProperties::FromMap(default_properties)));
@@ -65,6 +67,10 @@ Result<std::unique_ptr<FileIO>> MakeTableFileIO(
   if (storage_credentials.empty()) {
     return io;
   } else if (auto* credentialed = io->AsSupportsStorageCredentials()) {
+    // First, so the FileIO never briefly holds credentials it cannot replace.
+    if (refresher) {
+      credentialed->SetCredentialRefresher(std::move(refresher));
+    }
     ICEBERG_RETURN_UNEXPECTED(credentialed->SetStorageCredentials(storage_credentials));
   } else {
     return NotSupported("Configured FileIO does not support vended storage credentials");
diff --git a/src/iceberg/catalog/rest/rest_file_io.h b/src/iceberg/catalog/rest/rest_file_io.h
index e2316c3e8..301bf62ff 100644
--- a/src/iceberg/catalog/rest/rest_file_io.h
+++ b/src/iceberg/catalog/rest/rest_file_io.h
@@ -41,9 +41,13 @@ ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(
     const RestCatalogProperties& config);
 
 /// \brief Build the configured table FileIO and apply storage credentials if present.
+///
+/// \param refresher Optional callback used to replace the vended credentials
+/// before they expire; ignored when the FileIO cannot tell when they do.
 ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeTableFileIO(
     const std::unordered_map<std::string, std::string>& catalog_config,
     const std::unordered_map<std::string, std::string>& table_config,
-    const std::vector<StorageCredential>& storage_credentials);
+    const std::vector<StorageCredential>& storage_credentials,
+    StorageCredentialRefresher refresher = nullptr);
 
 }  // namespace iceberg::rest
diff --git a/src/iceberg/catalog/rest/types.h b/src/iceberg/catalog/rest/types.h
index 20a59fa59..a47a8e851 100644
--- a/src/iceberg/catalog/rest/types.h
+++ b/src/iceberg/catalog/rest/types.h
@@ -209,6 +209,21 @@ using CreateTableResponse = LoadTableResult;
 /// \brief Alias of LoadTableResult used as the body of LoadTableResponse
 using LoadTableResponse = LoadTableResult;
 
+/// \brief Response body of the LoadCredentials API.
+struct ICEBERG_REST_EXPORT LoadCredentialsResponse {
+  std::vector<StorageCredential> storage_credentials;
+
+  /// \brief Validates the LoadCredentialsResponse.
+  Status Validate() const {
+    for (const auto& credential : storage_credentials) {
+      ICEBERG_RETURN_UNEXPECTED(credential.Validate());
+    }
+    return {};
+  }
+
+  bool operator==(const LoadCredentialsResponse& other) const = default;
+};
+
 /// \brief Response body for listing namespaces.
 struct ICEBERG_REST_EXPORT ListNamespacesResponse {
   PageToken next_page_token;
diff --git a/src/iceberg/file_io.h b/src/iceberg/file_io.h
index e22e5cf21..86ba316b9 100644
--- a/src/iceberg/file_io.h
+++ b/src/iceberg/file_io.h
@@ -198,6 +198,11 @@ class ICEBERG_EXPORT SupportsStorageCredentials {
   /// By value because a concurrent install may replace them. An implementation
   /// that delegates may report what was installed on it.
   virtual std::vector<StorageCredential> credentials() const = 0;
+
+  /// \brief Install a callback that re-fetches credentials before they expire.
+  ///
+  /// Ignored by implementations that cannot tell when theirs expire.
+  virtual void SetCredentialRefresher(StorageCredentialRefresher /*refresher*/) {}
 };
 
 }  // namespace iceberg
diff --git a/src/iceberg/resolving_file_io.cc b/src/iceberg/resolving_file_io.cc
index 91be67eaf..4ba7bafe8 100644
--- a/src/iceberg/resolving_file_io.cc
+++ b/src/iceberg/resolving_file_io.cc
@@ -44,12 +44,17 @@ Result<std::shared_ptr<FileIO>> ResolvingFileIO::FileIOForPath(
   // without static keys may wait on the EC2 metadata service), which would
   // stall every other operation. Forwards all credentials; each implementation
   // applies the prefixes it understands.
-  auto load = [&](const std::vector<StorageCredential>& credentials)
+  auto load = [&](const std::vector<StorageCredential>& credentials,
+                  const StorageCredentialRefresher& refresher)
       -> Result<std::shared_ptr<FileIO>> {
     ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<FileIO> io,
                             FileIORegistry::Load(name, properties_));
-    if (!credentials.empty()) {
-      if (auto* credentialed = io->AsSupportsStorageCredentials()) {
+    if (auto* credentialed = io->AsSupportsStorageCredentials()) {
+      // Before the credentials, so the delegate can always replace them.
+      if (refresher) {
+        credentialed->SetCredentialRefresher(refresher);
+      }
+      if (!credentials.empty()) {
         ICEBERG_RETURN_UNEXPECTED(credentialed->SetStorageCredentials(credentials));
       }
     }
@@ -59,6 +64,7 @@ Result<std::shared_ptr<FileIO>> ResolvingFileIO::FileIOForPath(
   while (true) {
     uint64_t generation = 0;
     std::vector<StorageCredential> credentials;
+    StorageCredentialRefresher refresher;
     {
       std::shared_lock lock(mutex_);
       if (const auto cached = io_by_name_.find(name); cached != io_by_name_.end()) {
@@ -66,13 +72,14 @@ Result<std::shared_ptr<FileIO>> ResolvingFileIO::FileIOForPath(
       }
       generation = credential_generation_;
       credentials = storage_credentials_;
+      refresher = refresher_;
     }
     // Declared before the lock, so a delegate that is not cached is torn down
     // only after the lock is released.
-    auto loaded = load(credentials);
+    auto loaded = load(credentials, refresher);
     std::unique_lock lock(mutex_);
     if (generation != credential_generation_) {
-      continue;  // Credentials were replaced mid-load; load again with them.
+      continue;  // Replaced mid-load; load again with what is installed now.
     }
     if (const auto cached = io_by_name_.find(name); cached != io_by_name_.end()) {
       return cached->second;
@@ -137,4 +144,21 @@ std::vector<StorageCredential> ResolvingFileIO::credentials() const {
   return storage_credentials_;
 }
 
+void ResolvingFileIO::SetCredentialRefresher(StorageCredentialRefresher refresher) {
+  // Drop the cached delegates so they are rebuilt with the refresher. Retired
+  // outside the lock: teardown can block, and the outgoing callback's captures
+  // must not destruct under `mutex_`.
+  decltype(io_by_name_) retired;
+  // Holds the incoming callback going in and the outgoing one coming out; a
+  // pure swap never destroys a target under the lock, which std::exchange's
+  // move is permitted to do.
+  StorageCredentialRefresher handoff = std::move(refresher);
+  {
+    std::unique_lock lock(mutex_);
+    refresher_.swap(handoff);
+    ++credential_generation_;
+    retired.swap(io_by_name_);
+  }
+}
+
 }  // namespace iceberg
diff --git a/src/iceberg/resolving_file_io.h b/src/iceberg/resolving_file_io.h
index 9a5cf047e..c192f8ada 100644
--- a/src/iceberg/resolving_file_io.h
+++ b/src/iceberg/resolving_file_io.h
@@ -40,8 +40,8 @@ namespace iceberg {
 
 /// \brief FileIO that resolves and caches implementations by registry name.
 ///
-/// Vended credentials are forwarded to every resolved implementation that
-/// supports them; each applies what it understands.
+/// Vended credentials and any refresher are forwarded to every resolved
+/// implementation that supports them; each applies what it understands.
 class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
                                              public SupportsStorageCredentials {
  public:
@@ -62,8 +62,18 @@ class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
   Status SetStorageCredentials(
       const std::vector<StorageCredential>& storage_credentials) override;
 
+  /// \brief Return the credentials installed on this resolver.
+  ///
+  /// Not necessarily the ones in use: a resolved implementation refreshes its
+  /// own without reporting back.
   std::vector<StorageCredential> credentials() const override;
 
+  /// \brief Install a refresher and rebuild cached delegates with it.
+  ///
+  /// A delegate dropped here is rebuilt from the credentials installed on this
+  /// resolver, which may be older than what it had refreshed itself to.
+  void SetCredentialRefresher(StorageCredentialRefresher refresher) override;
+
   SupportsStorageCredentials* AsSupportsStorageCredentials() override { return this; }
 
  private:
@@ -74,8 +84,9 @@ class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
   // Guards lazy resolution and credential state.
   mutable std::shared_mutex mutex_;
   std::vector<StorageCredential> storage_credentials_;
-  // Bumped by every credential install, so a delegate loaded from an older set
-  // never reaches the cache.
+  StorageCredentialRefresher refresher_;
+  // Bumped by every credential or refresher install, so a delegate loaded from
+  // an older state never reaches the cache.
   uint64_t credential_generation_ = 0;
   std::unordered_map<std::string, std::shared_ptr<FileIO>, StringHash, StringEqual>
       io_by_name_;
diff --git a/src/iceberg/storage_credential.h b/src/iceberg/storage_credential.h
index 746bad086..e5d8022d7 100644
--- a/src/iceberg/storage_credential.h
+++ b/src/iceberg/storage_credential.h
@@ -22,8 +22,10 @@
 /// \file iceberg/storage_credential.h
 /// \brief Define storage credential metadata.
 
+#include <functional>
 #include <string>
 #include <unordered_map>
+#include <vector>
 
 #include "iceberg/iceberg_export.h"
 #include "iceberg/result.h"
@@ -48,4 +50,9 @@ struct ICEBERG_EXPORT StorageCredential {
   bool operator==(const StorageCredential& other) const = default;
 };
 
+/// \brief Re-fetches the currently valid vended credentials, without the
+/// FileIO knowing how they are delivered. Returns the whole vended list.
+using StorageCredentialRefresher =
+    std::function<Result<std::vector<StorageCredential>>()>;
+
 }  // namespace iceberg
diff --git a/src/iceberg/test/resolving_file_io_test.cc b/src/iceberg/test/resolving_file_io_test.cc
index 2189b64a6..db1c0d6b0 100644
--- a/src/iceberg/test/resolving_file_io_test.cc
+++ b/src/iceberg/test/resolving_file_io_test.cc
@@ -67,10 +67,25 @@ class RecordingCredentialedFileIO : public RecordingFileIO,
 
   std::vector<StorageCredential> credentials() const override { return credentials_; }
 
+  void SetCredentialRefresher(StorageCredentialRefresher refresher) override {
+    refresher_ = std::move(refresher);
+  }
+
   SupportsStorageCredentials* AsSupportsStorageCredentials() override { return this; }
 
+  Status Refresh() {
+    if (!refresher_) {
+      return NotFound("no refresher installed");
+    }
+    ICEBERG_ASSIGN_OR_RAISE(auto refreshed, refresher_());
+    return SetStorageCredentials(refreshed);
+  }
+
+  bool has_refresher() const { return static_cast<bool>(refresher_); }
+
  private:
   std::vector<StorageCredential> credentials_;
+  StorageCredentialRefresher refresher_;
 };
 
 // File-scope recording state: registry factories are process-global, so they
@@ -318,4 +333,41 @@ TEST(ResolvingFileIOTest, LoadsWithoutTheLockAndDropsStaleDelegates) {
   EXPECT_EQ(last->credentials(), fresh);
 }
 
+TEST(ResolvingFileIOTest, ForwardsCredentialRefresherToResolvedImplementations) {
+  RegisterRecordingFileIOs();
+  ResolvingFileIO io({});
+
+  std::vector<StorageCredential> refreshed = {{.prefix = "s3", .config = {{"k2", "v2"}}}};
+  io.SetCredentialRefresher(
+      [&]() -> Result<std::vector<StorageCredential>> { return refreshed; });
+  EXPECT_THAT(io.SetStorageCredentials({{.prefix = "s3", .config = {{"k1", "v1"}}}}),
+              IsOk());
+
+  std::ignore = io.NewInputFile("s3://bucket/db/table/data/file.parquet");
+  ASSERT_NE(last_s3_io, nullptr);
+  ASSERT_TRUE(last_s3_io->has_refresher());
+  EXPECT_THAT(last_s3_io->Refresh(), IsOk());
+  EXPECT_EQ(last_s3_io->credentials(), refreshed);
+}
+
+TEST(ResolvingFileIOTest, RebuildsResolvedImplementationsForALaterRefresher) {
+  RegisterRecordingFileIOs();
+  ResolvingFileIO io({});
+
+  EXPECT_THAT(io.SetStorageCredentials({{.prefix = "s3", .config = {{"k1", "v1"}}}}),
+              IsOk());
+  std::ignore = io.NewInputFile("s3://bucket/db/table/data/file.parquet");
+  ASSERT_NE(last_s3_io, nullptr);
+  EXPECT_FALSE(last_s3_io->has_refresher());
+  EXPECT_EQ(s3_factory_calls, 1);
+
+  io.SetCredentialRefresher([]() -> Result<std::vector<StorageCredential>> {
+    return std::vector<StorageCredential>{};
+  });
+  std::ignore = io.NewInputFile("s3://bucket/db/table/data/other.parquet");
+  EXPECT_EQ(s3_factory_calls, 2);
+  ASSERT_NE(last_s3_io, nullptr);
+  EXPECT_TRUE(last_s3_io->has_refresher());
+}
+
 }  // namespace iceberg
diff --git a/src/iceberg/test/rest_file_io_test.cc b/src/iceberg/test/rest_file_io_test.cc
index 2fce9dacf..ed03f45d2 100644
--- a/src/iceberg/test/rest_file_io_test.cc
+++ b/src/iceberg/test/rest_file_io_test.cc
@@ -52,6 +52,7 @@ class MockFileIO : public FileIO {
 
 std::vector<StorageCredential> captured_storage_credentials;
 std::unordered_map<std::string, std::string> captured_file_io_properties;
+StorageCredentialRefresher captured_refresher;
 
 class MockCredentialedFileIO : public MockFileIO, public SupportsStorageCredentials {
  public:
@@ -65,6 +66,10 @@ class MockCredentialedFileIO : public MockFileIO, public SupportsStorageCredenti
     return captured_storage_credentials;
   }
 
+  void SetCredentialRefresher(StorageCredentialRefresher refresher) override {
+    captured_refresher = std::move(refresher);
+  }
+
   SupportsStorageCredentials* AsSupportsStorageCredentials() override { return this; }
 };
 
@@ -155,6 +160,35 @@ TEST(RestFileIOTest, TableFileIOMergesConfigAndCredentials) {
   EXPECT_EQ(credentialed->credentials(), captured_storage_credentials);
 }
 
+TEST(RestFileIOTest, InstallsCredentialRefresherOnlyForVendedCredentials) {
+  const std::string custom_impl = "rest-file-io-test-refresher";
+  FileIORegistry::Register(
+      custom_impl,
+      {.create = [](const std::unordered_map<std::string, std::string>& /*properties*/)
+           -> Result<std::unique_ptr<FileIO>> {
+        return std::make_unique<MockCredentialedFileIO>();
+      }});
+  const std::unordered_map<std::string, std::string> table_config{
+      {"io-impl", custom_impl}};
+
+  captured_refresher = nullptr;
+  std::vector<StorageCredential> refreshed = {{.prefix = "s3", .config = {{"k", "v2"}}}};
+  auto result = MakeTableFileIO(
+      {}, table_config, {{.prefix = "s3", .config = {{"k", "v1"}}}},
+      [&]() -> Result<std::vector<StorageCredential>> { return refreshed; });
+  ASSERT_THAT(result, IsOk());
+  ASSERT_TRUE(captured_refresher);
+  EXPECT_THAT(captured_refresher(), IsOk());
+  EXPECT_EQ(captured_refresher().value(), refreshed);
+
+  captured_refresher = nullptr;
+  result = MakeTableFileIO(
+      {}, table_config, /*storage_credentials=*/{},
+      [&]() -> Result<std::vector<StorageCredential>> { return refreshed; });
+  ASSERT_THAT(result, IsOk());
+  EXPECT_FALSE(captured_refresher);
+}
+
 TEST(RestFileIOTest, TableImplOverridesWarehouseScheme) {
   captured_file_io_properties.clear();
   FileIORegistry::Register(
diff --git a/src/iceberg/test/rest_json_serde_test.cc b/src/iceberg/test/rest_json_serde_test.cc
index ec41e4a66..70b760b48 100644
--- a/src/iceberg/test/rest_json_serde_test.cc
+++ b/src/iceberg/test/rest_json_serde_test.cc
@@ -1229,6 +1229,32 @@ INSTANTIATE_TEST_SUITE_P(
       return info.param.test_name;
     });
 
+TEST(LoadCredentialsResponseTest, ParsesVendedCredentials) {
+  auto json = nlohmann::json::parse(
+      R"({"storage-credentials":[{"prefix":"s3","config":{"s3.access-key-id":"AKIAtest"}}]})");
+  auto response = LoadCredentialsResponseFromJson(json);
+  ASSERT_THAT(response, IsOk());
+  EXPECT_EQ(response->storage_credentials,
+            (std::vector<StorageCredential>{
+                {.prefix = "s3", .config = {{"s3.access-key-id", "AKIAtest"}}}}));
+}
+
+TEST(LoadCredentialsResponseTest, RejectsResponseWithoutCredentials) {
+  for (std::string_view body : {"{}", R"({"storage-credentials": null})"}) {
+    auto response = LoadCredentialsResponseFromJson(nlohmann::json::parse(body));
+    EXPECT_THAT(response, IsError(ErrorKind::kJsonParseError));
+    EXPECT_THAT(response, HasErrorMessage("Missing 'storage-credentials'"));
+  }
+}
+
+TEST(LoadCredentialsResponseTest, RejectsNonArrayCredentials) {
+  auto response = LoadCredentialsResponseFromJson(
+      nlohmann::json::parse(R"({"storage-credentials":"oops"})"));
+  EXPECT_THAT(response, IsError(ErrorKind::kJsonParseError));
+  EXPECT_THAT(response,
+              HasErrorMessage("Cannot parse storage credentials from non-array"));
+}
+
 DECLARE_ROUNDTRIP_TEST(CommitTableRequest)
 
 INSTANTIATE_TEST_SUITE_P(

From c3bc7d3e69cb2baa2b303328c6889c304a7289e4 Mon Sep 17 00:00:00 2001
From: Gang Wu <ustcwg@gmail.com>
Date: Thu, 8 Oct 2026 23:13:54 +0800
Subject: [PATCH 2/2] refactor(rest): replace credential refresh callback

Replace the opaque refresh callback with a typed provider and two-phase FileIO credential initialization. Keep REST fetching encapsulated and preserve resolver delegate setup.
---
 src/iceberg/catalog/rest/rest_catalog.cc   | 71 ++++++++++++++--------
 src/iceberg/catalog/rest/rest_catalog.h    |  8 +--
 src/iceberg/catalog/rest/rest_file_io.cc   | 14 +++--
 src/iceberg/catalog/rest/rest_file_io.h    |  5 +-
 src/iceberg/file_io.h                      | 14 ++++-
 src/iceberg/resolving_file_io.cc           | 44 +++++++-------
 src/iceberg/resolving_file_io.h            | 17 +++---
 src/iceberg/storage_credential.h           | 15 +++--
 src/iceberg/test/resolving_file_io_test.cc | 59 +++++++++++-------
 src/iceberg/test/rest_file_io_test.cc      | 54 ++++++++++------
 10 files changed, 178 insertions(+), 123 deletions(-)

diff --git a/src/iceberg/catalog/rest/rest_catalog.cc b/src/iceberg/catalog/rest/rest_catalog.cc
index 91c19d7b8..e2aaf5f3e 100644
--- a/src/iceberg/catalog/rest/rest_catalog.cc
+++ b/src/iceberg/catalog/rest/rest_catalog.cc
@@ -20,6 +20,8 @@
 #include "iceberg/catalog/rest/rest_catalog.h"
 
 #include <memory>
+#include <mutex>
+#include <string>
 #include <string_view>
 #include <tuple>
 #include <unordered_map>
@@ -61,6 +63,41 @@ namespace iceberg::rest {
 
 namespace {
 
+class RestStorageCredentialProvider final : public StorageCredentialProvider {
+ public:
+  RestStorageCredentialProvider(std::shared_ptr<HttpClient> client,
+                                std::shared_ptr<auth::AuthSession> session,
+                                std::string path)
+      : client_(std::move(client)),
+        session_(std::move(session)),
+        path_(std::move(path)) {}
+
+  Result<std::vector<StorageCredential>> Load() override {
+    // The provider can be shared by several FileIOs; keep fetches serialized.
+    std::lock_guard lock(mutex_);
+    ICEBERG_ASSIGN_OR_RAISE(const auto response,
+                            client_->Get(path_, /*params=*/{}, /*headers=*/{},
+                                         *TableErrorHandler::Instance(), *session_));
+    // Parse errors can contain credential data; return a fixed message.
+    auto json = FromJsonString(response.body());
+    if (!json.has_value()) {
+      return JsonParseError("Malformed LoadCredentials response");
+    }
+    auto result = LoadCredentialsResponseFromJson(*json);
+    if (!result.has_value()) {
+      return std::unexpected<Error>(
+          {.kind = result.error().kind, .message = "Malformed LoadCredentials response"});
+    }
+    return std::move(result->storage_credentials);
+  }
+
+ private:
+  std::shared_ptr<HttpClient> client_;
+  std::shared_ptr<auth::AuthSession> session_;
+  std::string path_;
+  std::mutex mutex_;
+};
+
 /// \brief Get the default set of endpoints for backwards compatibility according to the
 /// iceberg rest spec.
 std::unordered_set<Endpoint> GetDefaultEndpoints() {
@@ -509,7 +546,7 @@ Result<std::shared_ptr<auth::AuthSession>> RestCatalog::TableAuthSession(
                                      std::move(contextual_session));
 }
 
-StorageCredentialRefresher RestCatalog::MakeCredentialRefresher(
+std::shared_ptr<StorageCredentialProvider> RestCatalog::MakeStorageCredentialProvider(
     const TableIdentifier& identifier,
     std::shared_ptr<auth::AuthSession> table_session) const {
   if (!supported_endpoints_.contains(Endpoint::TableCredentials())) {
@@ -531,28 +568,8 @@ StorageCredentialRefresher RestCatalog::MakeCredentialRefresher(
   auto client = client_;
   auto credentials_path = std::move(path.value());
   auto session = std::move(table_session);
-  // The catalog's destructor closes the session, and a table's FileIO can
-  // outlive the table keeping the catalog alive. No cycle: the catalog's own
-  // FileIO never gets a refresher.
-  auto catalog = shared_from_this();
-  return [catalog, client, credentials_path,
-          session]() -> Result<std::vector<StorageCredential>> {
-    ICEBERG_ASSIGN_OR_RAISE(const auto response,
-                            client->Get(credentials_path, /*params=*/{}, /*headers=*/{},
-                                        *TableErrorHandler::Instance(), *session));
-    // Parse errors embed the offending input, and this body carries
-    // credentials; strip the message so it can never reach a log.
-    auto json = FromJsonString(response.body());
-    if (!json.has_value()) {
-      return JsonParseError("Malformed LoadCredentials response");
-    }
-    auto result = LoadCredentialsResponseFromJson(*json);
-    if (!result.has_value()) {
-      return std::unexpected<Error>(
-          {.kind = result.error().kind, .message = "Malformed LoadCredentials response"});
-    }
-    return std::move(result->storage_credentials);
-  };
+  return std::make_shared<RestStorageCredentialProvider>(
+      std::move(client), std::move(session), std::move(credentials_path));
 }
 
 Result<std::shared_ptr<FileIO>> RestCatalog::TableFileIO(
@@ -561,13 +578,13 @@ Result<std::shared_ptr<FileIO>> RestCatalog::TableFileIO(
     const std::vector<StorageCredential>& storage_credentials,
     std::shared_ptr<auth::AuthSession> table_session) const {
   if (!table_config.empty() || !storage_credentials.empty()) {
-    // Only vended credentials expire, so only they need a refresher.
-    StorageCredentialRefresher refresher;
+    // Only vended credentials expire, so only they need a provider.
+    std::shared_ptr<StorageCredentialProvider> provider;
     if (!storage_credentials.empty()) {
-      refresher = MakeCredentialRefresher(identifier, std::move(table_session));
+      provider = MakeStorageCredentialProvider(identifier, std::move(table_session));
     }
     return MakeTableFileIO(config_.configs(), table_config, storage_credentials,
-                           std::move(refresher));
+                           std::move(provider));
   }
 
   return file_io_;
diff --git a/src/iceberg/catalog/rest/rest_catalog.h b/src/iceberg/catalog/rest/rest_catalog.h
index bcb1be53a..6f65e6d61 100644
--- a/src/iceberg/catalog/rest/rest_catalog.h
+++ b/src/iceberg/catalog/rest/rest_catalog.h
@@ -90,12 +90,10 @@ class ICEBERG_REST_EXPORT RestCatalog final
       const std::vector<StorageCredential>& storage_credentials,
       std::shared_ptr<auth::AuthSession> table_session) const;
 
-  /// \brief Callback that reloads this table's vended credentials, or nullptr
-  /// when the catalog does not serve the LoadCredentials endpoint.
+  /// \brief Build a provider for this table's vended credentials.
   ///
-  /// Captures the catalog, so anything documented to outlive it -- the metrics
-  /// executor included -- must also outlive every FileIO holding a refresher.
-  StorageCredentialRefresher MakeCredentialRefresher(
+  /// Returns nullptr when the catalog does not serve LoadCredentials.
+  std::shared_ptr<StorageCredentialProvider> MakeStorageCredentialProvider(
       const TableIdentifier& identifier,
       std::shared_ptr<auth::AuthSession> table_session) const;
 
diff --git a/src/iceberg/catalog/rest/rest_file_io.cc b/src/iceberg/catalog/rest/rest_file_io.cc
index 2531008f3..4cd102704 100644
--- a/src/iceberg/catalog/rest/rest_file_io.cc
+++ b/src/iceberg/catalog/rest/rest_file_io.cc
@@ -27,6 +27,7 @@
 #include "iceberg/catalog/rest/types.h"
 #include "iceberg/file_io.h"
 #include "iceberg/file_io_registry.h"
+#include "iceberg/logging/log_macros.h"
 #include "iceberg/resolving_file_io.h"
 #include "iceberg/util/macros.h"
 
@@ -59,7 +60,7 @@ Result<std::unique_ptr<FileIO>> MakeTableFileIO(
     const std::unordered_map<std::string, std::string>& catalog_config,
     const std::unordered_map<std::string, std::string>& table_config,
     const std::vector<StorageCredential>& storage_credentials,
-    StorageCredentialRefresher refresher) {
+    std::shared_ptr<StorageCredentialProvider> provider) {
   const auto default_properties = MergeFileIOProperties(catalog_config, table_config);
   ICEBERG_ASSIGN_OR_RAISE(
       auto io, MakeCatalogFileIO(RestCatalogProperties::FromMap(default_properties)));
@@ -68,10 +69,15 @@ Result<std::unique_ptr<FileIO>> MakeTableFileIO(
     return io;
   } else if (auto* credentialed = io->AsSupportsStorageCredentials()) {
     // First, so the FileIO never briefly holds credentials it cannot replace.
-    if (refresher) {
-      credentialed->SetCredentialRefresher(std::move(refresher));
+    auto status = credentialed->InitializeStorageCredentials(storage_credentials,
+                                                             std::move(provider));
+    if (!status && !storage_credentials.empty() &&
+        status.error().kind == ErrorKind::kNotSupported) {
+      ICEBERG_LOG_WARN("Configured FileIO cannot refresh vended storage credentials: {}",
+                       status.error().message);
+      status = credentialed->SetStorageCredentials(storage_credentials);
     }
-    ICEBERG_RETURN_UNEXPECTED(credentialed->SetStorageCredentials(storage_credentials));
+    ICEBERG_RETURN_UNEXPECTED(status);
   } else {
     return NotSupported("Configured FileIO does not support vended storage credentials");
   }
diff --git a/src/iceberg/catalog/rest/rest_file_io.h b/src/iceberg/catalog/rest/rest_file_io.h
index 301bf62ff..5eb9f97fa 100644
--- a/src/iceberg/catalog/rest/rest_file_io.h
+++ b/src/iceberg/catalog/rest/rest_file_io.h
@@ -41,13 +41,10 @@ ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(
     const RestCatalogProperties& config);
 
 /// \brief Build the configured table FileIO and apply storage credentials if present.
-///
-/// \param refresher Optional callback used to replace the vended credentials
-/// before they expire; ignored when the FileIO cannot tell when they do.
 ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeTableFileIO(
     const std::unordered_map<std::string, std::string>& catalog_config,
     const std::unordered_map<std::string, std::string>& table_config,
     const std::vector<StorageCredential>& storage_credentials,
-    StorageCredentialRefresher refresher = nullptr);
+    std::shared_ptr<StorageCredentialProvider> provider = nullptr);
 
 }  // namespace iceberg::rest
diff --git a/src/iceberg/file_io.h b/src/iceberg/file_io.h
index 86ba316b9..4dd96530d 100644
--- a/src/iceberg/file_io.h
+++ b/src/iceberg/file_io.h
@@ -29,6 +29,7 @@
 #include <span>
 #include <string>
 #include <string_view>
+#include <utility>
 #include <vector>
 
 #include "iceberg/iceberg_export.h"
@@ -199,10 +200,17 @@ class ICEBERG_EXPORT SupportsStorageCredentials {
   /// that delegates may report what was installed on it.
   virtual std::vector<StorageCredential> credentials() const = 0;
 
-  /// \brief Install a callback that re-fetches credentials before they expire.
+  /// \brief Install initial credentials and their refresh source before first use.
   ///
-  /// Ignored by implementations that cannot tell when theirs expire.
-  virtual void SetCredentialRefresher(StorageCredentialRefresher /*refresher*/) {}
+  /// The default implementation supports static credentials only.
+  virtual Status InitializeStorageCredentials(
+      const std::vector<StorageCredential>& storage_credentials,
+      std::shared_ptr<StorageCredentialProvider> provider) {
+    if (provider) {
+      return NotSupported("Credential refresh is not supported");
+    }
+    return SetStorageCredentials(storage_credentials);
+  }
 };
 
 }  // namespace iceberg
diff --git a/src/iceberg/resolving_file_io.cc b/src/iceberg/resolving_file_io.cc
index 4ba7bafe8..ddf19b6fe 100644
--- a/src/iceberg/resolving_file_io.cc
+++ b/src/iceberg/resolving_file_io.cc
@@ -24,6 +24,7 @@
 #include <utility>
 
 #include "iceberg/file_io_registry.h"
+#include "iceberg/logging/log_macros.h"
 #include "iceberg/util/location_util.h"
 #include "iceberg/util/macros.h"
 #include "iceberg/util/string_util.h"
@@ -45,17 +46,18 @@ Result<std::shared_ptr<FileIO>> ResolvingFileIO::FileIOForPath(
   // stall every other operation. Forwards all credentials; each implementation
   // applies the prefixes it understands.
   auto load = [&](const std::vector<StorageCredential>& credentials,
-                  const StorageCredentialRefresher& refresher)
+                  const std::shared_ptr<StorageCredentialProvider>& provider)
       -> Result<std::shared_ptr<FileIO>> {
     ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<FileIO> io,
                             FileIORegistry::Load(name, properties_));
     if (auto* credentialed = io->AsSupportsStorageCredentials()) {
-      // Before the credentials, so the delegate can always replace them.
-      if (refresher) {
-        credentialed->SetCredentialRefresher(refresher);
+      auto status = credentialed->InitializeStorageCredentials(credentials, provider);
+      if (!status && provider && status.error().kind == ErrorKind::kNotSupported) {
+        ICEBERG_LOG_WARN("FileIO '{}' cannot refresh vended storage credentials", name);
+        status = credentialed->SetStorageCredentials(credentials);
       }
-      if (!credentials.empty()) {
-        ICEBERG_RETURN_UNEXPECTED(credentialed->SetStorageCredentials(credentials));
+      if (!status) {
+        return std::unexpected(status.error());
       }
     }
     return io;
@@ -64,7 +66,7 @@ Result<std::shared_ptr<FileIO>> ResolvingFileIO::FileIOForPath(
   while (true) {
     uint64_t generation = 0;
     std::vector<StorageCredential> credentials;
-    StorageCredentialRefresher refresher;
+    std::shared_ptr<StorageCredentialProvider> provider;
     {
       std::shared_lock lock(mutex_);
       if (const auto cached = io_by_name_.find(name); cached != io_by_name_.end()) {
@@ -72,11 +74,11 @@ Result<std::shared_ptr<FileIO>> ResolvingFileIO::FileIOForPath(
       }
       generation = credential_generation_;
       credentials = storage_credentials_;
-      refresher = refresher_;
+      provider = provider_;
     }
     // Declared before the lock, so a delegate that is not cached is torn down
     // only after the lock is released.
-    auto loaded = load(credentials, refresher);
+    auto loaded = load(credentials, provider);
     std::unique_lock lock(mutex_);
     if (generation != credential_generation_) {
       continue;  // Replaced mid-load; load again with what is installed now.
@@ -144,21 +146,17 @@ std::vector<StorageCredential> ResolvingFileIO::credentials() const {
   return storage_credentials_;
 }
 
-void ResolvingFileIO::SetCredentialRefresher(StorageCredentialRefresher refresher) {
-  // Drop the cached delegates so they are rebuilt with the refresher. Retired
-  // outside the lock: teardown can block, and the outgoing callback's captures
-  // must not destruct under `mutex_`.
-  decltype(io_by_name_) retired;
-  // Holds the incoming callback going in and the outgoing one coming out; a
-  // pure swap never destroys a target under the lock, which std::exchange's
-  // move is permitted to do.
-  StorageCredentialRefresher handoff = std::move(refresher);
-  {
-    std::unique_lock lock(mutex_);
-    refresher_.swap(handoff);
-    ++credential_generation_;
-    retired.swap(io_by_name_);
+Status ResolvingFileIO::InitializeStorageCredentials(
+    const std::vector<StorageCredential>& storage_credentials,
+    std::shared_ptr<StorageCredentialProvider> provider) {
+  std::unique_lock lock(mutex_);
+  if (!io_by_name_.empty()) {
+    return InvalidArgument("Storage credentials must be initialized before first use");
   }
+  storage_credentials_ = storage_credentials;
+  provider_ = std::move(provider);
+  ++credential_generation_;
+  return {};
 }
 
 }  // namespace iceberg
diff --git a/src/iceberg/resolving_file_io.h b/src/iceberg/resolving_file_io.h
index c192f8ada..05e11eb8f 100644
--- a/src/iceberg/resolving_file_io.h
+++ b/src/iceberg/resolving_file_io.h
@@ -40,7 +40,7 @@ namespace iceberg {
 
 /// \brief FileIO that resolves and caches implementations by registry name.
 ///
-/// Vended credentials and any refresher are forwarded to every resolved
+/// Vended credentials and any credential provider are forwarded to every resolved
 /// implementation that supports them; each applies what it understands.
 class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
                                              public SupportsStorageCredentials {
@@ -68,11 +68,10 @@ class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
   /// own without reporting back.
   std::vector<StorageCredential> credentials() const override;
 
-  /// \brief Install a refresher and rebuild cached delegates with it.
-  ///
-  /// A delegate dropped here is rebuilt from the credentials installed on this
-  /// resolver, which may be older than what it had refreshed itself to.
-  void SetCredentialRefresher(StorageCredentialRefresher refresher) override;
+  /// \brief Install initial credentials before the first delegate is loaded.
+  Status InitializeStorageCredentials(
+      const std::vector<StorageCredential>& storage_credentials,
+      std::shared_ptr<StorageCredentialProvider> provider) override;
 
   SupportsStorageCredentials* AsSupportsStorageCredentials() override { return this; }
 
@@ -84,9 +83,9 @@ class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
   // Guards lazy resolution and credential state.
   mutable std::shared_mutex mutex_;
   std::vector<StorageCredential> storage_credentials_;
-  StorageCredentialRefresher refresher_;
-  // Bumped by every credential or refresher install, so a delegate loaded from
-  // an older state never reaches the cache.
+  std::shared_ptr<StorageCredentialProvider> provider_;
+  // Bumped by every credential install, so a delegate loaded from an older set
+  // never reaches the cache.
   uint64_t credential_generation_ = 0;
   std::unordered_map<std::string, std::shared_ptr<FileIO>, StringHash, StringEqual>
       io_by_name_;
diff --git a/src/iceberg/storage_credential.h b/src/iceberg/storage_credential.h
index e5d8022d7..d21099615 100644
--- a/src/iceberg/storage_credential.h
+++ b/src/iceberg/storage_credential.h
@@ -22,7 +22,6 @@
 /// \file iceberg/storage_credential.h
 /// \brief Define storage credential metadata.
 
-#include <functional>
 #include <string>
 #include <unordered_map>
 #include <vector>
@@ -50,9 +49,15 @@ struct ICEBERG_EXPORT StorageCredential {
   bool operator==(const StorageCredential& other) const = default;
 };
 
-/// \brief Re-fetches the currently valid vended credentials, without the
-/// FileIO knowing how they are delivered. Returns the whole vended list.
-using StorageCredentialRefresher =
-    std::function<Result<std::vector<StorageCredential>>()>;
+/// \brief Source for loading the current vended credentials.
+///
+/// Implementations must be safe for concurrent calls.
+class ICEBERG_EXPORT StorageCredentialProvider {
+ public:
+  virtual ~StorageCredentialProvider() = default;
+
+  /// \brief Load the current credentials. Returns the whole vended list.
+  virtual Result<std::vector<StorageCredential>> Load() = 0;
+};
 
 }  // namespace iceberg
diff --git a/src/iceberg/test/resolving_file_io_test.cc b/src/iceberg/test/resolving_file_io_test.cc
index db1c0d6b0..a3b194877 100644
--- a/src/iceberg/test/resolving_file_io_test.cc
+++ b/src/iceberg/test/resolving_file_io_test.cc
@@ -56,9 +56,28 @@ class RecordingFileIO : public FileIO {
   std::vector<std::vector<std::string>> deleted_batches;
 };
 
+class StaticStorageCredentialProvider : public StorageCredentialProvider {
+ public:
+  explicit StaticStorageCredentialProvider(std::vector<StorageCredential> credentials)
+      : credentials_(std::move(credentials)) {}
+
+  Result<std::vector<StorageCredential>> Load() override { return credentials_; }
+
+ private:
+  std::vector<StorageCredential> credentials_;
+};
+
 class RecordingCredentialedFileIO : public RecordingFileIO,
                                     public SupportsStorageCredentials {
  public:
+  Status InitializeStorageCredentials(
+      const std::vector<StorageCredential>& storage_credentials,
+      std::shared_ptr<StorageCredentialProvider> provider) override {
+    credentials_ = storage_credentials;
+    provider_ = std::move(provider);
+    return {};
+  }
+
   Status SetStorageCredentials(
       const std::vector<StorageCredential>& storage_credentials) override {
     credentials_ = storage_credentials;
@@ -67,25 +86,21 @@ class RecordingCredentialedFileIO : public RecordingFileIO,
 
   std::vector<StorageCredential> credentials() const override { return credentials_; }
 
-  void SetCredentialRefresher(StorageCredentialRefresher refresher) override {
-    refresher_ = std::move(refresher);
-  }
-
   SupportsStorageCredentials* AsSupportsStorageCredentials() override { return this; }
 
   Status Refresh() {
-    if (!refresher_) {
-      return NotFound("no refresher installed");
+    if (!provider_) {
+      return NotFound("no provider installed");
     }
-    ICEBERG_ASSIGN_OR_RAISE(auto refreshed, refresher_());
+    ICEBERG_ASSIGN_OR_RAISE(auto refreshed, provider_->Load());
     return SetStorageCredentials(refreshed);
   }
 
-  bool has_refresher() const { return static_cast<bool>(refresher_); }
+  bool has_provider() const { return provider_ != nullptr; }
 
  private:
   std::vector<StorageCredential> credentials_;
-  StorageCredentialRefresher refresher_;
+  std::shared_ptr<StorageCredentialProvider> provider_;
 };
 
 // File-scope recording state: registry factories are process-global, so they
@@ -333,24 +348,24 @@ TEST(ResolvingFileIOTest, LoadsWithoutTheLockAndDropsStaleDelegates) {
   EXPECT_EQ(last->credentials(), fresh);
 }
 
-TEST(ResolvingFileIOTest, ForwardsCredentialRefresherToResolvedImplementations) {
+TEST(ResolvingFileIOTest, ForwardsCredentialProviderToResolvedImplementations) {
   RegisterRecordingFileIOs();
   ResolvingFileIO io({});
 
   std::vector<StorageCredential> refreshed = {{.prefix = "s3", .config = {{"k2", "v2"}}}};
-  io.SetCredentialRefresher(
-      [&]() -> Result<std::vector<StorageCredential>> { return refreshed; });
-  EXPECT_THAT(io.SetStorageCredentials({{.prefix = "s3", .config = {{"k1", "v1"}}}}),
+  auto provider = std::make_shared<StaticStorageCredentialProvider>(refreshed);
+  EXPECT_THAT(io.InitializeStorageCredentials(
+                  {{.prefix = "s3", .config = {{"k1", "v1"}}}}, provider),
               IsOk());
 
   std::ignore = io.NewInputFile("s3://bucket/db/table/data/file.parquet");
   ASSERT_NE(last_s3_io, nullptr);
-  ASSERT_TRUE(last_s3_io->has_refresher());
+  ASSERT_TRUE(last_s3_io->has_provider());
   EXPECT_THAT(last_s3_io->Refresh(), IsOk());
   EXPECT_EQ(last_s3_io->credentials(), refreshed);
 }
 
-TEST(ResolvingFileIOTest, RebuildsResolvedImplementationsForALaterRefresher) {
+TEST(ResolvingFileIOTest, RejectsCredentialProviderAfterFirstUse) {
   RegisterRecordingFileIOs();
   ResolvingFileIO io({});
 
@@ -358,16 +373,14 @@ TEST(ResolvingFileIOTest, RebuildsResolvedImplementationsForALaterRefresher) {
               IsOk());
   std::ignore = io.NewInputFile("s3://bucket/db/table/data/file.parquet");
   ASSERT_NE(last_s3_io, nullptr);
-  EXPECT_FALSE(last_s3_io->has_refresher());
+  EXPECT_FALSE(last_s3_io->has_provider());
   EXPECT_EQ(s3_factory_calls, 1);
 
-  io.SetCredentialRefresher([]() -> Result<std::vector<StorageCredential>> {
-    return std::vector<StorageCredential>{};
-  });
-  std::ignore = io.NewInputFile("s3://bucket/db/table/data/other.parquet");
-  EXPECT_EQ(s3_factory_calls, 2);
-  ASSERT_NE(last_s3_io, nullptr);
-  EXPECT_TRUE(last_s3_io->has_refresher());
+  auto provider =
+      std::make_shared<StaticStorageCredentialProvider>(std::vector<StorageCredential>{});
+  EXPECT_THAT(io.InitializeStorageCredentials({}, provider),
+              IsError(ErrorKind::kInvalidArgument));
+  EXPECT_EQ(s3_factory_calls, 1);
 }
 
 }  // namespace iceberg
diff --git a/src/iceberg/test/rest_file_io_test.cc b/src/iceberg/test/rest_file_io_test.cc
index ed03f45d2..31a027ab4 100644
--- a/src/iceberg/test/rest_file_io_test.cc
+++ b/src/iceberg/test/rest_file_io_test.cc
@@ -50,12 +50,31 @@ class MockFileIO : public FileIO {
   Status DeleteFile(const std::string& /*file_location*/) override { return {}; }
 };
 
+class StaticStorageCredentialProvider : public StorageCredentialProvider {
+ public:
+  explicit StaticStorageCredentialProvider(std::vector<StorageCredential> credentials)
+      : credentials_(std::move(credentials)) {}
+
+  Result<std::vector<StorageCredential>> Load() override { return credentials_; }
+
+ private:
+  std::vector<StorageCredential> credentials_;
+};
+
 std::vector<StorageCredential> captured_storage_credentials;
 std::unordered_map<std::string, std::string> captured_file_io_properties;
-StorageCredentialRefresher captured_refresher;
+std::shared_ptr<StorageCredentialProvider> captured_provider;
 
 class MockCredentialedFileIO : public MockFileIO, public SupportsStorageCredentials {
  public:
+  Status InitializeStorageCredentials(
+      const std::vector<StorageCredential>& credentials,
+      std::shared_ptr<StorageCredentialProvider> provider) override {
+    captured_storage_credentials = credentials;
+    captured_provider = std::move(provider);
+    return {};
+  }
+
   Status SetStorageCredentials(
       const std::vector<StorageCredential>& credentials) override {
     captured_storage_credentials = credentials;
@@ -66,10 +85,6 @@ class MockCredentialedFileIO : public MockFileIO, public SupportsStorageCredenti
     return captured_storage_credentials;
   }
 
-  void SetCredentialRefresher(StorageCredentialRefresher refresher) override {
-    captured_refresher = std::move(refresher);
-  }
-
   SupportsStorageCredentials* AsSupportsStorageCredentials() override { return this; }
 };
 
@@ -160,8 +175,8 @@ TEST(RestFileIOTest, TableFileIOMergesConfigAndCredentials) {
   EXPECT_EQ(credentialed->credentials(), captured_storage_credentials);
 }
 
-TEST(RestFileIOTest, InstallsCredentialRefresherOnlyForVendedCredentials) {
-  const std::string custom_impl = "rest-file-io-test-refresher";
+TEST(RestFileIOTest, InstallsCredentialProviderOnlyForVendedCredentials) {
+  const std::string custom_impl = "rest-file-io-test-provider";
   FileIORegistry::Register(
       custom_impl,
       {.create = [](const std::unordered_map<std::string, std::string>& /*properties*/)
@@ -171,22 +186,21 @@ TEST(RestFileIOTest, InstallsCredentialRefresherOnlyForVendedCredentials) {
   const std::unordered_map<std::string, std::string> table_config{
       {"io-impl", custom_impl}};
 
-  captured_refresher = nullptr;
+  captured_provider = nullptr;
   std::vector<StorageCredential> refreshed = {{.prefix = "s3", .config = {{"k", "v2"}}}};
-  auto result = MakeTableFileIO(
-      {}, table_config, {{.prefix = "s3", .config = {{"k", "v1"}}}},
-      [&]() -> Result<std::vector<StorageCredential>> { return refreshed; });
+  auto provider = std::make_shared<StaticStorageCredentialProvider>(refreshed);
+  auto result = MakeTableFileIO({}, table_config,
+                                {{.prefix = "s3", .config = {{"k", "v1"}}}}, provider);
   ASSERT_THAT(result, IsOk());
-  ASSERT_TRUE(captured_refresher);
-  EXPECT_THAT(captured_refresher(), IsOk());
-  EXPECT_EQ(captured_refresher().value(), refreshed);
-
-  captured_refresher = nullptr;
-  result = MakeTableFileIO(
-      {}, table_config, /*storage_credentials=*/{},
-      [&]() -> Result<std::vector<StorageCredential>> { return refreshed; });
+  ASSERT_TRUE(captured_provider);
+  EXPECT_EQ(captured_provider, provider);
+  EXPECT_THAT(captured_provider->Load(), IsOk());
+  EXPECT_EQ(captured_provider->Load().value(), refreshed);
+
+  captured_provider = nullptr;
+  result = MakeTableFileIO({}, table_config, /*storage_credentials=*/{}, provider);
   ASSERT_THAT(result, IsOk());
-  EXPECT_FALSE(captured_refresher);
+  EXPECT_FALSE(captured_provider);
 }
 
 TEST(RestFileIOTest, TableImplOverridesWarehouseScheme) {