hub async s3 client (#1024)

- Feature: `AsyncHttpClient` adds cancellable request tokens, streaming GET to a file (`AsyncDownload`), zero-copy chunk-callback GET (`AsyncStream`), pull-mode body source for streaming `AsyncPut`, retry layer mirroring the synchronous client, and a submit-side in-flight cap (`HttpClientSettings::MaxConcurrentRequests`) so hub-scale fanout against a single host cannot stall queued handles into curl's connect-timeout window
- Feature: Hub hydration can route S3 transfers through a non-blocking `AsyncHttpClient` (curl_multi + asio) backed by a single io thread; hydrate and dehydrate now pipeline requests instead of blocking worker threads
  - `--hub-hydration-async-enabled` (Lua: `hub.hydration.async.enabled`, default true)
  - `--hub-hydration-async-max-concurrent-requests` (Lua: `hub.hydration.async.maxconcurrentrequests`, default `clamp(cpu*4, 128, 512)`)
- Feature: Hub provision/deprovision/obliterate now run as two phases on separate worker pools so per-module hydration cannot starve child-process spawn/despawn (and vice versa)
  - New `--hub-instance-spawn-threads` (Lua: `hub.instance.spawnthreads`, default `clamp(cpu/8, 4, 16)`) drives child-process spawn/despawn
  - `--hub-instance-provision-threads` (Lua: `hub.instance.provisionthreads`) now drives per-module hydrate/dehydrate scheduling only; default changed from `max(cpu/4, 2)` to `clamp(cpu/8, 4, 12)`
  - `--hub-hydration-threads` (Lua: `hub.hydration.threads`) now controls per-file workers inside a single hydrate/dehydrate; default changed from `max(cpu/4, 2)` to `clamp(cpu/8, 4, 12)`
- Feature: `AsyncHttpClient` owns its `asio::io_context` and one io thread by default; the `(BaseUri, io_context&)` constructor is preserved for callers that want to share an externally-driven `io_context` across clients (caller MUST keep the loop running until the client destructs)
- Feature: `Hub::Configuration` C++ struct fields renamed (`OptionalProvisionWorkerPool`/`OptionalHydrationWorkerPool` -> `OptionalProvisionPool`/`OptionalSpawnPool`/`OptionalHydrationPool`). Embedders constructing `Hub` directly must update field names; provision and spawn pools must both be set or both null (asserted at construction).
- Bugfix: `S3Client` signing-key cache no longer returns stale signatures after IMDS-rotated credentials change `AccessKeyId`; cache is now keyed on `(DateStamp, AccessKeyId)`

1 files changed, 149 insertions, 0 deletions

diff --git a/src/zenutil/cloud/s3requestbuilder.cpp b/src/zenutil/cloud/s3requestbuilder.cpp
new file mode 100644
index 000000000..7bf200308
--- /dev/null
+++ b/src/zenutil/cloud/s3requestbuilder.cpp
@@ -0,0 +1,149 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include <zenutil/cloud/s3requestbuilder.h>
+
+#include <zencore/fmtutils.h>
+
+#include <algorithm>
+
+namespace zen {
+
+S3RequestBuilder::S3RequestBuilder(std::string Region, std::string BucketName, std::string Endpoint, bool PathStyle)
+: m_Region(std::move(Region))
+, m_BucketName(std::move(BucketName))
+, m_Endpoint(DeriveEndpoint(Endpoint, m_Region, m_BucketName, PathStyle))
+, m_Host(HostFromEndpoint(m_Endpoint))
+, m_PathStyle(PathStyle)
+{
+}
+
+std::string
+S3RequestBuilder::DeriveEndpoint(std::string_view Endpoint, std::string_view Region, std::string_view BucketName, bool PathStyle)
+{
+	if (!Endpoint.empty())
+	{
+		return std::string(Endpoint);
+	}
+	if (PathStyle)
+	{
+		return fmt::format("https://s3.{}.amazonaws.com", Region);
+	}
+	return fmt::format("https://{}.s3.{}.amazonaws.com", BucketName, Region);
+}
+
+std::string
+S3RequestBuilder::HostFromEndpoint(std::string_view Endpoint)
+{
+	std::string_view Ep = Endpoint;
+	if (size_t Pos = Ep.find("://"); Pos != std::string_view::npos)
+	{
+		Ep = Ep.substr(Pos + 3);
+	}
+	if (!Ep.empty() && Ep.back() == '/')
+	{
+		Ep = Ep.substr(0, Ep.size() - 1);
+	}
+	return std::string(Ep);
+}
+
+std::string
+S3RequestBuilder::KeyToPath(std::string_view Key) const
+{
+	if (m_PathStyle)
+	{
+		return fmt::format("/{}/{}", m_BucketName, Key);
+	}
+	return fmt::format("/{}", Key);
+}
+
+std::string
+S3RequestBuilder::BucketRootPath() const
+{
+	if (m_PathStyle)
+	{
+		return fmt::format("/{}/", m_BucketName);
+	}
+	return "/";
+}
+
+// Cached for the (DateStamp, AccessKeyId) tuple. DateStamp rolls over at UTC
+// midnight; concurrent callers around the rollover may each compute a fresh
+// key once before the cache settles on the new day's value. Harmless extra
+// HMACs, no signature corruption.
+Sha256Digest
+S3RequestBuilder::GetSigningKey(std::string_view DateStamp, const SigV4Credentials& Credentials)
+{
+	{
+		RwLock::SharedLockScope _(m_SigningKeyLock);
+		if (m_CachedDateStamp == DateStamp && m_CachedAccessKeyId == Credentials.AccessKeyId)
+		{
+			return m_CachedSigningKey;
+		}
+	}
+
+	RwLock::ExclusiveLockScope _(m_SigningKeyLock);
+	if (m_CachedDateStamp == DateStamp && m_CachedAccessKeyId == Credentials.AccessKeyId)
+	{
+		return m_CachedSigningKey;
+	}
+
+	std::string	 SecretPrefix = fmt::format("AWS4{}", Credentials.SecretAccessKey);
+	Sha256Digest DateKey	  = ComputeHmacSha256(SecretPrefix.data(), SecretPrefix.size(), DateStamp.data(), DateStamp.size());
+	SecureZeroSecret(SecretPrefix.data(), SecretPrefix.size());
+
+	Sha256Digest RegionKey	= ComputeHmacSha256(DateKey, m_Region);
+	Sha256Digest ServiceKey = ComputeHmacSha256(RegionKey, "s3");
+	m_CachedSigningKey		= ComputeHmacSha256(ServiceKey, "aws4_request");
+	m_CachedDateStamp		= std::string(DateStamp);
+	m_CachedAccessKeyId		= Credentials.AccessKeyId;
+
+	return m_CachedSigningKey;
+}
+
+HttpClient::KeyValueMap
+S3RequestBuilder::SignRequest(const SigV4Credentials&							   Credentials,
+							  std::string_view									   Method,
+							  std::string_view									   Path,
+							  std::string_view									   CanonicalQueryString,
+							  std::string_view									   PayloadHash,
+							  std::span<const std::pair<std::string, std::string>> ExtraSignedHeaders)
+{
+	std::string AmzDate = GetAmzTimestamp();
+
+	std::vector<std::pair<std::string, std::string>> HeadersToSign;
+	HeadersToSign.reserve(4 + ExtraSignedHeaders.size());
+	HeadersToSign.emplace_back("host", m_Host);
+	HeadersToSign.emplace_back("x-amz-content-sha256", std::string(PayloadHash));
+	HeadersToSign.emplace_back("x-amz-date", AmzDate);
+	if (!Credentials.SessionToken.empty())
+	{
+		HeadersToSign.emplace_back("x-amz-security-token", Credentials.SessionToken);
+	}
+	for (const auto& [K, V] : ExtraSignedHeaders)
+	{
+		HeadersToSign.emplace_back(K, V);
+	}
+	std::sort(HeadersToSign.begin(), HeadersToSign.end());
+
+	std::string_view DateStamp(AmzDate.data(), 8);
+	Sha256Digest	 SigningKey = GetSigningKey(DateStamp, Credentials);
+
+	SigV4SignedHeaders Signed =
+		SignRequestV4(Credentials, Method, Path, CanonicalQueryString, m_Region, "s3", AmzDate, HeadersToSign, PayloadHash, &SigningKey);
+
+	HttpClient::KeyValueMap Result;
+	Result->emplace("Authorization", std::move(Signed.Authorization));
+	Result->emplace("x-amz-date", std::move(Signed.AmzDate));
+	Result->emplace("x-amz-content-sha256", std::move(Signed.PayloadHash));
+	if (!Credentials.SessionToken.empty())
+	{
+		Result->emplace("x-amz-security-token", Credentials.SessionToken);
+	}
+	for (const auto& [K, V] : ExtraSignedHeaders)
+	{
+		Result->emplace(K, V);
+	}
+	return Result;
+}
+
+}  // namespace zen