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, 2 insertions, 76 deletions

diff --git a/src/zenserver/hub/storageserverinstance.cpp b/src/zenserver/hub/storageserverinstance.cpp
index 8d36e6a46..34860eb9b 100644
--- a/src/zenserver/hub/storageserverinstance.cpp
+++ b/src/zenserver/hub/storageserverinstance.cpp
@@ -2,8 +2,6 @@
 
 #include "storageserverinstance.h"
 
-#include "hydration.h"
-
 #include <zencore/assertfmt.h>
 #include <zencore/filesystem.h>
 #include <zencore/fmtutils.h>
@@ -14,12 +12,8 @@
 
 namespace zen {
 
-StorageServerInstance::StorageServerInstance(ZenServerEnvironment& RunEnvironment,
-											 HydrationBase&		   Hydration,
-											 const Configuration&  Config,
-											 std::string_view	   ModuleId)
-: m_Hydration(Hydration)
-, m_Config(Config)
+StorageServerInstance::StorageServerInstance(ZenServerEnvironment& RunEnvironment, const Configuration& Config, std::string_view ModuleId)
+: m_Config(Config)
 , m_ModuleId(ModuleId)
 , m_ServerInstance(RunEnvironment, ZenServerInstance::ServerMode::kStorageServer)
 {
@@ -138,7 +132,6 @@ StorageServerInstance::ProvisionLocked()
 	ZEN_INFO("Provisioning storage server instance for module '{}', at '{}'", m_ModuleId, m_Config.StateDir);
 	try
 	{
-		Hydrate();
 		SpawnServerProcess();
 	}
 	catch (const std::exception& Ex)
@@ -156,19 +149,6 @@ StorageServerInstance::DeprovisionLocked()
 {
 	ZEN_TRACE_CPU("StorageServerInstance::DeprovisionLocked");
 	ShutdownServerProcess();
-
-	// Crashed or Hibernated: process already dead; skip Shutdown.
-	// Dehydrate preserves instance state for future re-provisioning. Failure means saved state
-	// may be stale or absent, but the process is already dead so the slot can still be released.
-	// Swallow the exception and proceed with cleanup rather than leaving the module stuck.
-	try
-	{
-		Dehydrate();
-	}
-	catch (const std::exception& Ex)
-	{
-		ZEN_WARN("Dehydration of module {} failed during deprovisioning, current state not saved. Reason: {}", m_ModuleId, Ex.what());
-	}
 }
 
 void
@@ -176,12 +156,6 @@ StorageServerInstance::ObliterateLocked()
 {
 	ZEN_TRACE_CPU("StorageServerInstance::ObliterateLocked");
 	ShutdownServerProcess();
-
-	std::atomic<bool>					   AbortFlag{false};
-	std::atomic<bool>					   PauseFlag{false};
-	HydrationConfig						   Config	= MakeHydrationConfig(AbortFlag, PauseFlag);
-	std::unique_ptr<HydrationStrategyBase> Hydrator = m_Hydration.CreateHydrator(Config);
-	Hydrator->Obliterate();
 }
 
 void
@@ -218,54 +192,6 @@ StorageServerInstance::WakeLocked()
 	}
 }
 
-void
-StorageServerInstance::Hydrate()
-{
-	if (!m_Config.EnableHydration)
-	{
-		ZEN_INFO("Hydration disabled; skipping hydrate for module '{}'", m_ModuleId);
-		return;
-	}
-	ZEN_TRACE_CPU("StorageServerInstance::Hydrate");
-	std::atomic<bool>					   AbortFlag{false};
-	std::atomic<bool>					   PauseFlag{false};
-	HydrationConfig						   Config	= MakeHydrationConfig(AbortFlag, PauseFlag);
-	std::unique_ptr<HydrationStrategyBase> Hydrator = m_Hydration.CreateHydrator(Config);
-	m_HydrationState								= Hydrator->Hydrate();
-}
-
-void
-StorageServerInstance::Dehydrate()
-{
-	if (!m_Config.EnableDehydration)
-	{
-		ZEN_INFO("Dehydration disabled; skipping dehydrate for module '{}'", m_ModuleId);
-		return;
-	}
-	ZEN_TRACE_CPU("StorageServerInstance::Dehydrate");
-	std::atomic<bool>					   AbortFlag{false};
-	std::atomic<bool>					   PauseFlag{false};
-	HydrationConfig						   Config	= MakeHydrationConfig(AbortFlag, PauseFlag);
-	std::unique_ptr<HydrationStrategyBase> Hydrator = m_Hydration.CreateHydrator(Config);
-	Hydrator->Dehydrate(m_HydrationState);
-}
-
-HydrationConfig
-StorageServerInstance::MakeHydrationConfig(std::atomic<bool>& AbortFlag, std::atomic<bool>& PauseFlag)
-{
-	HydrationConfig Config{.ServerStateDir = m_Config.StateDir, .TempDir = m_Config.TempDir, .ModuleId = m_ModuleId};
-	if (m_Config.OptionalWorkerPool)
-	{
-		Config.Threading.emplace(
-			HydrationConfig::ThreadingOptions{.WorkerPool = m_Config.OptionalWorkerPool, .AbortFlag = &AbortFlag, .PauseFlag = &PauseFlag});
-	}
-	Config.PackEnabled		  = m_Config.HydrationPackEnabled;
-	Config.PackThresholdBytes = m_Config.HydrationPackThresholdBytes;
-	Config.MaxPackBytes		  = m_Config.HydrationMaxPackBytes;
-
-	return Config;
-}
-
 StorageServerInstance::SharedLockedPtr::SharedLockedPtr() : m_Lock(nullptr), m_Instance(nullptr)
 {
 }