address review: SigV4 session lifecycle via Make(), per-table sessions, Base64 wording, CI toolchain

Author: plusplusjiajia

ci/scripts/build_iceberg.sh

@@ -60,6 +60,11 @@ if is_windows; then
     CMAKE_ARGS+=("-DCMAKE_TOOLCHAIN_FILE=C:/vcpkg/scripts/buildsystems/vcpkg.cmake")
     CMAKE_ARGS+=("-DCMAKE_BUILD_TYPE=Release")
 else
+    # Pass an externally provided toolchain (e.g. vcpkg for the SigV4 job)
+    # explicitly instead of relying on CMake >= 3.21 reading the env var.
+    if [[ -n "${CMAKE_TOOLCHAIN_FILE:-}" ]]; then
+        CMAKE_ARGS+=("-DCMAKE_TOOLCHAIN_FILE=${CMAKE_TOOLCHAIN_FILE}")
+    fi
     CMAKE_ARGS+=("-DCMAKE_BUILD_TYPE=Debug")
 fi
 

meson.options

@@ -44,6 +44,8 @@ option(
     value: 'disabled',
 )
 
+# Resolves a system-installed AWS SDK via its CMake config; the bundled-AWS
+# path (ICEBERG_AWSSDK_SOURCE=BUNDLED) is CMake-only.
 option(
     'sigv4',
     type: 'feature',

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

@@ -172,11 +172,7 @@ SigV4AuthSession::SigV4AuthSession(
       signer_(std::make_unique<RestSigV4Signer>(
           credentials_provider_, signing_name_.c_str(), signing_region_.c_str())) {}
 
-SigV4AuthSession::~SigV4AuthSession() {
-  if (owns_sdk_registration_) {
-    AwsSdkLifecycle::Instance().UnregisterSession();
-  }
-}
+SigV4AuthSession::~SigV4AuthSession() { AwsSdkLifecycle::Instance().UnregisterSession(); }
 
 Result<HttpRequest> SigV4AuthSession::Authenticate(const HttpRequest& request) {
   ICEBERG_ASSIGN_OR_RAISE(auto delegate_request, delegate_->Authenticate(request));
@@ -198,8 +194,12 @@ Result<HttpRequest> SigV4AuthSession::Authenticate(const HttpRequest& request) {
     aws_request->SetHeaderValue(Aws::String(name.c_str()), Aws::String(value.c_str()));
   }
 
-  // Empty body: hex EMPTY_BODY_SHA256 (Java parity workaround for the signer
-  // computing an invalid checksum on empty bodies). Non-empty: Base64.
+  // Java parity: for non-empty bodies the signed x-amz-content-sha256 header
+  // carries Base64(SHA256(body)) — matching the Java client's
+  // SignerChecksumParams behavior — while the canonical request's payload hash
+  // line remains lowercase hex per SigV4. Empty bodies use the hex
+  // EMPTY_BODY_SHA256 constant (workaround for the signer computing an invalid
+  // checksum on empty bodies).
   if (delegate_request.body.empty()) {
     aws_request->SetHeaderValue("x-amz-content-sha256", Aws::String(kEmptyBodySha256));
   } else {
@@ -411,11 +411,23 @@ class SessionSlot {
 
 }  // namespace
 
+Result<std::shared_ptr<SigV4AuthSession>> SigV4AuthSession::Make(
+    std::shared_ptr<AuthSession> delegate, std::string signing_region,
+    std::string signing_name,
+    std::shared_ptr<Aws::Auth::AWSCredentialsProvider> credentials_provider) {
+  ICEBERG_ASSIGN_OR_RAISE(auto slot, SessionSlot::Reserve());
+  auto session = std::shared_ptr<SigV4AuthSession>(
+      new SigV4AuthSession(std::move(delegate), std::move(signing_region),
+                           std::move(signing_name), std::move(credentials_provider)));
+  // The session's destructor now owns the unregister.
+  slot.Release();
+  return session;
+}
+
 Result<std::shared_ptr<AuthSession>> SigV4AuthManager::WrapSession(
     std::shared_ptr<AuthSession> delegate_session,
     const std::unordered_map<std::string, std::string>& properties,
     std::shared_ptr<Aws::Auth::AWSCredentialsProvider> reuse_credentials) {
-  ICEBERG_ASSIGN_OR_RAISE(auto slot, SessionSlot::Reserve());
   ICEBERG_ASSIGN_OR_RAISE(auto region, ResolveSigningRegion(properties));
   auto service = ResolveSigningName(properties);
 
@@ -441,12 +453,9 @@ Result<std::shared_ptr<AuthSession>> SigV4AuthManager::WrapSession(
                          AuthProperties::kSigV4SecretAccessKey +
                          "' or configure the AWS credentials chain"});
   }
-  auto session =
-      std::make_shared<SigV4AuthSession>(std::move(delegate_session), std::move(region),
-                                         std::move(service), std::move(credentials));
-  // The reserved slot's unregister responsibility now belongs to the session.
-  session->owns_sdk_registration_ = true;
-  slot.Release();
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto session, SigV4AuthSession::Make(std::move(delegate_session), std::move(region),
+                                           std::move(service), std::move(credentials)));
   return session;
 }
 

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

@@ -58,7 +58,11 @@ class ICEBERG_REST_EXPORT SigV4AuthSession : public AuthSession {
   /// Prefix prepended to relocated headers that conflict with SigV4-signed headers.
   static constexpr std::string_view kRelocatedHeaderPrefix = "Original-";
 
-  SigV4AuthSession(
+  /// \brief Creates a session registered with the AWS SDK lifecycle.
+  ///
+  /// Fails if the SDK is not initialized. Every session owns its lifecycle
+  /// registration and unregisters on destruction.
+  static Result<std::shared_ptr<SigV4AuthSession>> Make(
       std::shared_ptr<AuthSession> delegate, std::string signing_region,
       std::string signing_name,
       std::shared_ptr<Aws::Auth::AWSCredentialsProvider> credentials_provider);
@@ -78,17 +82,16 @@ class ICEBERG_REST_EXPORT SigV4AuthSession : public AuthSession {
   }
 
  private:
-  // WrapSession() reserves an AwsSdkLifecycle slot and transfers it here.
-  friend class SigV4AuthManager;
+  SigV4AuthSession(
+      std::shared_ptr<AuthSession> delegate, std::string signing_region,
+      std::string signing_name,
+      std::shared_ptr<Aws::Auth::AWSCredentialsProvider> credentials_provider);
 
   std::shared_ptr<AuthSession> delegate_;
   std::string signing_region_;
   std::string signing_name_;
   std::shared_ptr<Aws::Auth::AWSCredentialsProvider> credentials_provider_;
   std::unique_ptr<Aws::Client::AWSAuthV4Signer> signer_;
-  // Only WrapSession()-created sessions registered a slot; directly-constructed
-  // ones (e.g. tests) must not unregister and underflow the count.
-  bool owns_sdk_registration_ = false;
 };
 
 /// \brief An AuthManager that produces SigV4AuthSession instances.

src/iceberg/catalog/rest/rest_catalog.cc

@@ -118,14 +118,56 @@ Result<bool> CaptureNoSuchNamespace(const auto& status) {
   return CaptureNoSuchObject(status, ErrorKind::kNoSuchNamespace);
 }
 
+std::string TableSessionKey(const TableIdentifier& identifier) {
+  std::string key;
+  for (const auto& level : identifier.ns.levels) {
+    key += level;
+    key += '\x1f';
+  }
+  key += identifier.name;
+  return key;
+}
+
 }  // namespace
 
 RestCatalog::~RestCatalog() {
+  for (auto& [key, session] : table_sessions_) {
+    if (session) {
+      std::ignore = session->Close();
+    }
+  }
   if (catalog_session_) {
     std::ignore = catalog_session_->Close();
   }
 }
 
+Status RestCatalog::RememberTableSession(
+    const TableIdentifier& identifier,
+    const std::unordered_map<std::string, std::string>& config) {
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto session, auth_manager_->TableSession(identifier, config, catalog_session_));
+  if (session == catalog_session_) {
+    return {};
+  }
+  std::shared_ptr<auth::AuthSession> replaced;
+  {
+    std::lock_guard<std::mutex> lock(table_sessions_mutex_);
+    auto& slot = table_sessions_[TableSessionKey(identifier)];
+    replaced = std::exchange(slot, std::move(session));
+  }
+  if (replaced) {
+    std::ignore = replaced->Close();
+  }
+  return {};
+}
+
+std::shared_ptr<auth::AuthSession> RestCatalog::SessionFor(
+    const TableIdentifier& identifier) {
+  std::lock_guard<std::mutex> lock(table_sessions_mutex_);
+  auto it = table_sessions_.find(TableSessionKey(identifier));
+  return it != table_sessions_.end() ? it->second : catalog_session_;
+}
+
 Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
     const RestCatalogProperties& config) {
   ICEBERG_ASSIGN_OR_RAISE(auto uri, config.Uri());
@@ -355,7 +397,9 @@ Result<LoadTableResult> RestCatalog::CreateTableInternal(
                     *catalog_session_));
 
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
-  return LoadTableResultFromJson(json);
+  ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFromJson(json));
+  ICEBERG_RETURN_UNEXPECTED(RememberTableSession(identifier, load_result.config));
+  return load_result;
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::CreateTable(
@@ -391,7 +435,7 @@ Result<std::shared_ptr<Table>> RestCatalog::UpdateTable(
   ICEBERG_ASSIGN_OR_RAISE(
       const auto response,
       client_->Post(path, json_request, /*headers=*/{}, *TableErrorHandler::Instance(),
-                    *catalog_session_));
+                    *SessionFor(identifier)));
 
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
   ICEBERG_ASSIGN_OR_RAISE(auto commit_response, CommitTableResponseFromJson(json));
@@ -479,6 +523,7 @@ Result<std::shared_ptr<Table>> RestCatalog::LoadTable(const TableIdentifier& ide
   ICEBERG_ASSIGN_OR_RAISE(const auto body, LoadTableInternal(identifier));
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(body));
   ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFromJson(json));
+  ICEBERG_RETURN_UNEXPECTED(RememberTableSession(identifier, load_result.config));
   /// FIXME: support per-table FileIO creation
   return Table::Make(identifier, std::move(load_result.metadata),
                      std::move(load_result.metadata_location), file_io_,
@@ -503,6 +548,7 @@ Result<std::shared_ptr<Table>> RestCatalog::RegisterTable(
 
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
   ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFromJson(json));
+  ICEBERG_RETURN_UNEXPECTED(RememberTableSession(identifier, load_result.config));
   return Table::Make(identifier, std::move(load_result.metadata),
                      std::move(load_result.metadata_location), file_io_,
                      shared_from_this());

src/iceberg/catalog/rest/rest_catalog.h

@@ -20,7 +20,9 @@
 #pragma once
 
 #include <memory>
+#include <mutex>
 #include <string>
+#include <unordered_map>
 #include <unordered_set>
 
 #include "iceberg/catalog.h"
@@ -109,6 +111,14 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog,
 
   Result<std::string> LoadTableInternal(const TableIdentifier& identifier) const;
 
+  /// \brief Derives and caches a per-table auth session from a table response
+  /// `config`, per the REST spec (`config.token` must be used for table requests).
+  Status RememberTableSession(const TableIdentifier& identifier,
+                              const std::unordered_map<std::string, std::string>& config);
+
+  /// \brief Returns the cached per-table session, or the catalog session.
+  std::shared_ptr<auth::AuthSession> SessionFor(const TableIdentifier& identifier);
+
   Result<LoadTableResult> CreateTableInternal(
       const TableIdentifier& identifier, const std::shared_ptr<Schema>& schema,
       const std::shared_ptr<PartitionSpec>& spec, const std::shared_ptr<SortOrder>& order,
@@ -124,6 +134,8 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog,
   std::unique_ptr<auth::AuthManager> auth_manager_;
   std::shared_ptr<auth::AuthSession> catalog_session_;
   SnapshotMode snapshot_mode_;
+  std::mutex table_sessions_mutex_;
+  std::unordered_map<std::string, std::shared_ptr<auth::AuthSession>> table_sessions_;
 };
 
 }  // namespace iceberg::rest

src/iceberg/test/sigv4_auth_test.cc

@@ -72,28 +72,23 @@ TEST_F(SigV4AuthTest, LifecycleFinalizeRefusesWhileSessionsAlive) {
   EXPECT_TRUE(IsAwsSdkInitialized());
 }
 
-TEST_F(SigV4AuthTest, DirectlyConstructedSessionDoesNotCorruptLifecycleCount) {
-  // A directly-constructed session never registered with AwsSdkLifecycle, so
-  // destroying it must not decrement the count. Otherwise it underflows, and a
-  // later real session wraps it back to zero, letting FinalizeAwsSdk() shut the
-  // SDK down while a session is still alive.
+TEST_F(SigV4AuthTest, SessionRegistrationBalancesLifecycleCount) {
   {
     auto delegate = AuthSession::MakeDefault({});
     auto credentials = std::make_shared<Aws::Auth::SimpleAWSCredentialsProvider>(
         Aws::Auth::AWSCredentials("id", "secret"));
-    auto direct = std::make_shared<SigV4AuthSession>(delegate, "us-east-1", "execute-api",
-                                                     credentials);
-  }  // destroyed here — must leave the lifecycle count untouched
+    auto session_result =
+        SigV4AuthSession::Make(delegate, "us-east-1", "execute-api", credentials);
+    ASSERT_THAT(session_result, IsOk());
+    EXPECT_THAT(FinalizeAwsSdk(), IsError(ErrorKind::kInvalid));
+  }
 
   auto properties = MakeSigV4Properties();
   auto manager_result = AuthManagers::Load("test-catalog", properties);
   ASSERT_THAT(manager_result, IsOk());
   auto session_result = manager_result.value()->CatalogSession(client_, properties);
   ASSERT_THAT(session_result, IsOk());
 
-  // Exactly one live (registered) session, so Finalize must refuse. With the
-  // underflow bug the count would have wrapped and Finalize could wrongly
-  // succeed and shut the SDK down.
   EXPECT_THAT(FinalizeAwsSdk(), IsError(ErrorKind::kInvalid));
   EXPECT_TRUE(IsAwsSdkInitialized());
 }
@@ -338,7 +333,8 @@ TEST_F(SigV4AuthTest, AuthenticateWithBodyDetailedHeaders) {
   ASSERT_NE(auth_it, headers.end());
   EXPECT_TRUE(auth_it->second.starts_with("AWS4-HMAC-SHA256 Credential="));
 
-  // x-amz-content-sha256 should be Base64-encoded body SHA256 (matching Java)
+  // Java parity: the x-amz-content-sha256 header is Base64(SHA256(body)) for
+  // non-empty bodies; the canonical request payload hash stays lowercase hex.
   auto sha_it = headers.find("x-amz-content-sha256");
   ASSERT_NE(sha_it, headers.end());
   EXPECT_NE(sha_it->second, SigV4AuthSession::kEmptyBodySha256);
@@ -390,8 +386,10 @@ TEST_F(SigV4AuthTest, ConflictingSigV4HeadersRelocated) {
   auto credentials =
       std::make_shared<Aws::Auth::SimpleAWSCredentialsProvider>(Aws::Auth::AWSCredentials(
           "AKIAIOSFODNN7EXAMPLE", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"));
-  auto session = std::make_shared<SigV4AuthSession>(delegate, "us-east-1", "execute-api",
-                                                    credentials);
+  auto session_result =
+      SigV4AuthSession::Make(delegate, "us-east-1", "execute-api", credentials);
+  ASSERT_THAT(session_result, IsOk());
+  auto session = session_result.value();
 
   HttpRequest request{.method = HttpMethod::kGet, .url = "http://localhost:8080/path"};
   auto auth_result = session->Authenticate(request);
@@ -418,11 +416,11 @@ TEST_F(SigV4AuthTest, SessionCloseDelegatesToInner) {
   auto delegate = AuthSession::MakeDefault({});
   auto credentials = std::make_shared<Aws::Auth::SimpleAWSCredentialsProvider>(
       Aws::Auth::AWSCredentials("id", "secret"));
-  auto session = std::make_shared<SigV4AuthSession>(delegate, "us-east-1", "execute-api",
-                                                    credentials);
+  auto session_result =
+      SigV4AuthSession::Make(delegate, "us-east-1", "execute-api", credentials);
+  ASSERT_THAT(session_result, IsOk());
 
-  // Close should succeed without error
-  EXPECT_THAT(session->Close(), IsOk());
+  EXPECT_THAT(session_result.value()->Close(), IsOk());
 }
 
 TEST_F(SigV4AuthTest, CreateCustomDelegateNone) {