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, 220 insertions, 32 deletions

diff --git a/src/zenutil/cloud/sigv4.cpp b/src/zenutil/cloud/sigv4.cpp
index 055ccb2ad..34bd7f5f3 100644
--- a/src/zenutil/cloud/sigv4.cpp
+++ b/src/zenutil/cloud/sigv4.cpp
@@ -53,6 +53,62 @@ ComputeSha256(const void* Data, size_t Size)
 	return Result;
 }
 
+Sha256Stream::Sha256Stream()
+{
+	EVP_MD_CTX* Ctx = EVP_MD_CTX_new();
+	ZEN_ASSERT(Ctx != nullptr);
+	int Rc = EVP_DigestInit_ex(Ctx, EVP_sha256(), nullptr);
+	ZEN_ASSERT(Rc == 1);
+	m_Ctx = Ctx;
+}
+
+Sha256Stream::~Sha256Stream()
+{
+	if (m_Ctx)
+	{
+		EVP_MD_CTX_free(static_cast<EVP_MD_CTX*>(m_Ctx));
+		m_Ctx = nullptr;
+	}
+}
+
+Sha256Stream::Sha256Stream(Sha256Stream&& Other) noexcept : m_Ctx(Other.m_Ctx)
+{
+	Other.m_Ctx = nullptr;
+}
+
+Sha256Stream&
+Sha256Stream::operator=(Sha256Stream&& Other) noexcept
+{
+	if (this != &Other)
+	{
+		if (m_Ctx)
+		{
+			EVP_MD_CTX_free(static_cast<EVP_MD_CTX*>(m_Ctx));
+		}
+		m_Ctx		= Other.m_Ctx;
+		Other.m_Ctx = nullptr;
+	}
+	return *this;
+}
+
+void
+Sha256Stream::Update(const void* Data, size_t Size)
+{
+	int Rc = EVP_DigestUpdate(static_cast<EVP_MD_CTX*>(m_Ctx), Data, Size);
+	ZEN_ASSERT(Rc == 1);
+}
+
+Sha256Digest
+Sha256Stream::Finalize()
+{
+	Sha256Digest Result;
+	unsigned int Len = 0;
+	int			 Rc	 = EVP_DigestFinal_ex(static_cast<EVP_MD_CTX*>(m_Ctx), Result.data(), &Len);
+	ZEN_ASSERT(Rc == 1);
+	ZEN_ASSERT(Len == 32);
+	return Result;
+}
+
 Sha256Digest
 ComputeHmacSha256(const void* Key, size_t KeySize, const void* Data, size_t DataSize)
 {
@@ -171,6 +227,62 @@ ComputeSha256(const void* Data, size_t Size)
 	return BcryptHash(GetBcryptHandles().Sha256, Data, Size);
 }
 
+Sha256Stream::Sha256Stream()
+{
+	BCRYPT_HASH_HANDLE Handle = nullptr;
+	NTSTATUS		   Status = BCryptCreateHash(GetBcryptHandles().Sha256, &Handle, nullptr, 0, nullptr, 0, 0);
+	ZEN_ASSERT(NT_SUCCESS(Status));
+	m_Ctx = Handle;
+}
+
+Sha256Stream::~Sha256Stream()
+{
+	if (m_Ctx)
+	{
+		BCryptDestroyHash(static_cast<BCRYPT_HASH_HANDLE>(m_Ctx));
+		m_Ctx = nullptr;
+	}
+}
+
+Sha256Stream::Sha256Stream(Sha256Stream&& Other) noexcept : m_Ctx(Other.m_Ctx)
+{
+	Other.m_Ctx = nullptr;
+}
+
+Sha256Stream&
+Sha256Stream::operator=(Sha256Stream&& Other) noexcept
+{
+	if (this != &Other)
+	{
+		if (m_Ctx)
+		{
+			BCryptDestroyHash(static_cast<BCRYPT_HASH_HANDLE>(m_Ctx));
+		}
+		m_Ctx		= Other.m_Ctx;
+		Other.m_Ctx = nullptr;
+	}
+	return *this;
+}
+
+void
+Sha256Stream::Update(const void* Data, size_t Size)
+{
+	NTSTATUS Status = BCryptHashData(static_cast<BCRYPT_HASH_HANDLE>(m_Ctx),
+									 reinterpret_cast<PUCHAR>(const_cast<void*>(Data)),
+									 static_cast<ULONG>(Size),
+									 0);
+	ZEN_ASSERT(NT_SUCCESS(Status));
+}
+
+Sha256Digest
+Sha256Stream::Finalize()
+{
+	Sha256Digest Result;
+	NTSTATUS	 Status = BCryptFinishHash(static_cast<BCRYPT_HASH_HANDLE>(m_Ctx), Result.data(), static_cast<ULONG>(Result.size()), 0);
+	ZEN_ASSERT(NT_SUCCESS(Status));
+	return Result;
+}
+
 Sha256Digest
 ComputeHmacSha256(const void* Key, size_t KeySize, const void* Data, size_t DataSize)
 {
@@ -251,6 +363,8 @@ GetAmzTimestamp()
 std::string
 AwsUriEncode(std::string_view Input, bool EncodeSlash)
 {
+	static constexpr char kHex[] = "0123456789ABCDEF";
+
 	ExtendableStringBuilder<256> Result;
 	for (char C : Input)
 	{
@@ -264,7 +378,11 @@ AwsUriEncode(std::string_view Input, bool EncodeSlash)
 		}
 		else
 		{
-			Result.Append(fmt::format("%{:02X}", static_cast<unsigned char>(C)));
+			// Hand-rolled hex encode; avoids per-char fmt::format std::string alloc
+			// inside a loop that runs once per non-unreserved input character.
+			const uint8_t Byte		 = static_cast<uint8_t>(C);
+			const char	  Encoded[3] = {'%', kHex[Byte >> 4], kHex[Byte & 0x0F]};
+			Result.Append(std::string_view(Encoded, 3));
 		}
 	}
 	return std::string(Result.ToView());
@@ -313,18 +431,10 @@ SignRequestV4(const SigV4Credentials&								  Credentials,
 
 	std::string DateStamp = GetDateStamp(Result.AmzDate);
 
-	// Step 1: Create canonical request
-	// CanonicalRequest =
-	//   HTTPRequestMethod + '\n' +
-	//   CanonicalURI + '\n' +
-	//   CanonicalQueryString + '\n' +
-	//   CanonicalHeaders + '\n' +
-	//   SignedHeaders + '\n' +
-	//   HexEncode(Hash(RequestPayload))
-
+	// Step 1: canonical request.
 	std::string CanonicalUri = AwsUriEncode(Url, false);
 
-	// Build canonical headers and signed headers (headers must be sorted by lowercase name)
+	// Headers assumed pre-sorted by lowercase name.
 	ExtendableStringBuilder<512> CanonicalHeadersSb;
 	ExtendableStringBuilder<256> SignedHeadersSb;
 
@@ -342,30 +452,45 @@ SignRequestV4(const SigV4Credentials&								  Credentials,
 		SignedHeadersSb.Append(Headers[i].first);
 	}
 
-	std::string SignedHeaders = std::string(SignedHeadersSb.ToView());
-
-	std::string CanonicalRequest = fmt::format("{}\n{}\n{}\n{}\n{}\n{}",
-											   Method,
-											   CanonicalUri,
-											   CanonicalQueryString,
-											   CanonicalHeadersSb.ToView(),
-											   SignedHeaders,
-											   PayloadHash);
-
-	// Step 2: Create the string to sign
-	std::string CredentialScope = fmt::format("{}/{}/{}/aws4_request", DateStamp, Region, Service);
+	std::string_view SignedHeaders = SignedHeadersSb.ToView();
+
+	ExtendableStringBuilder<2048> CanonicalRequestSb;
+	CanonicalRequestSb.Append(Method);
+	CanonicalRequestSb.Append('\n');
+	CanonicalRequestSb.Append(CanonicalUri);
+	CanonicalRequestSb.Append('\n');
+	CanonicalRequestSb.Append(CanonicalQueryString);
+	CanonicalRequestSb.Append('\n');
+	CanonicalRequestSb.Append(CanonicalHeadersSb.ToView());
+	CanonicalRequestSb.Append('\n');
+	CanonicalRequestSb.Append(SignedHeaders);
+	CanonicalRequestSb.Append('\n');
+	CanonicalRequestSb.Append(PayloadHash);
+	std::string_view CanonicalRequest = CanonicalRequestSb.ToView();
+
+	// Step 2: string-to-sign.
+	ExtendableStringBuilder<128> CredentialScopeSb;
+	CredentialScopeSb.Append(DateStamp);
+	CredentialScopeSb.Append('/');
+	CredentialScopeSb.Append(Region);
+	CredentialScopeSb.Append('/');
+	CredentialScopeSb.Append(Service);
+	CredentialScopeSb.Append("/aws4_request");
+	std::string_view CredentialScope = CredentialScopeSb.ToView();
 
 	Sha256Digest CanonicalRequestHash = ComputeSha256(CanonicalRequest);
 	std::string	 CanonicalRequestHex  = Sha256ToHex(CanonicalRequestHash);
 
-	std::string StringToSign = fmt::format("AWS4-HMAC-SHA256\n{}\n{}\n{}", Result.AmzDate, CredentialScope, CanonicalRequestHex);
-
-	// Step 3: Calculate the signing key
-	// kDate    = HMAC("AWS4" + SecretKey, DateStamp)
-	// kRegion  = HMAC(kDate, Region)
-	// kService = HMAC(kRegion, Service)
-	// kSigning = HMAC(kService, "aws4_request")
+	ExtendableStringBuilder<256> StringToSignSb;
+	StringToSignSb.Append("AWS4-HMAC-SHA256\n");
+	StringToSignSb.Append(Result.AmzDate);
+	StringToSignSb.Append('\n');
+	StringToSignSb.Append(CredentialScope);
+	StringToSignSb.Append('\n');
+	StringToSignSb.Append(CanonicalRequestHex);
+	std::string_view StringToSign = StringToSignSb.ToView();
 
+	// Step 3: derive signing key (kDate -> kRegion -> kService -> kSigning HMAC chain).
 	Sha256Digest DerivedSigningKey;
 	if (!SigningKeyPtr)
 	{
@@ -380,11 +505,11 @@ SignRequestV4(const SigV4Credentials&								  Credentials,
 		SigningKeyPtr			= &DerivedSigningKey;
 	}
 
-	// Step 4: Calculate the signature
+	// Step 4: signature.
 	Sha256Digest Signature	  = ComputeHmacSha256(*SigningKeyPtr, StringToSign);
 	std::string	 SignatureHex = Sha256ToHex(Signature);
 
-	// Step 5: Build the Authorization header
+	// Step 5: Authorization header.
 	Result.Authorization = fmt::format("AWS4-HMAC-SHA256 Credential={}/{}, SignedHeaders={}, Signature={}",
 									   Credentials.AccessKeyId,
 									   CredentialScope,
@@ -489,6 +614,69 @@ TEST_CASE("sigv4.sha256")
 	CHECK(HelloHex == "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824");
 }
 
+TEST_CASE("sigv4.sha256stream.matches_oneshot")
+{
+	// Empty input.
+	{
+		Sha256Stream S;
+		Sha256Digest Streamed = S.Finalize();
+		Sha256Digest OneShot  = ComputeSha256("", 0);
+		CHECK(Streamed == OneShot);
+	}
+
+	// Single update.
+	{
+		Sha256Stream S;
+		S.Update("hello", 5);
+		Sha256Digest Streamed = S.Finalize();
+		Sha256Digest OneShot  = ComputeSha256("hello");
+		CHECK(Streamed == OneShot);
+	}
+
+	// Multi-chunk update; result must equal one-shot over concatenated bytes.
+	{
+		const std::string Whole = "the quick brown fox jumps over the lazy dog";
+		Sha256Stream	  S;
+		S.Update(Whole.data(), 4);
+		S.Update(Whole.data() + 4, 16);
+		S.Update(Whole.data() + 20, Whole.size() - 20);
+		Sha256Digest Streamed = S.Finalize();
+		Sha256Digest OneShot  = ComputeSha256(Whole.data(), Whole.size());
+		CHECK(Streamed == OneShot);
+	}
+
+	// Large input fed in 256 KiB chunks (matches PutMedium hash pass shape).
+	{
+		std::vector<uint8_t> Big(1u * 1024u * 1024u + 7u);
+		for (size_t I = 0; I < Big.size(); ++I)
+		{
+			Big[I] = static_cast<uint8_t>((I * 31u + 11u) & 0xFF);
+		}
+		Sha256Stream	 S;
+		constexpr size_t kChunk = 256u * 1024u;
+		size_t			 Off	= 0;
+		while (Off < Big.size())
+		{
+			const size_t Take = std::min(kChunk, Big.size() - Off);
+			S.Update(Big.data() + Off, Take);
+			Off += Take;
+		}
+		Sha256Digest Streamed = S.Finalize();
+		Sha256Digest OneShot  = ComputeSha256(Big.data(), Big.size());
+		CHECK(Streamed == OneShot);
+	}
+
+	// Move construct + finalize on the moved-to instance.
+	{
+		Sha256Stream A;
+		A.Update("abc", 3);
+		Sha256Stream B(std::move(A));
+		B.Update("def", 3);
+		Sha256Digest Result = B.Finalize();
+		CHECK(Result == ComputeSha256("abcdef"));
+	}
+}
+
 TEST_CASE("sigv4.hmac_sha256")
 {
 	// RFC 4231 Test Case 2