hub s3 hydrate improvements (#902)

- Feature: Added `--hub-hydration-target-config` option to specify the hydration target via a JSON config file (mutually exclusive with `--hub-hydration-target-spec`); supports `file` and `s3` types with structured settings
  ```json
  {
    "type": "file",
    "settings": {
      "path": "/path/to/hydration/storage"
    }
  }
  ```
  ```json
  {
    "type": "s3",
    "settings": {
      "uri": "s3://bucket[/prefix]",
      "region": "us-east-1",
      "endpoint": "http://localhost:9000",
      "path-style": true
    }
  }
  ```
- Improvement: Hub hydration dehydration skips the `.sentry-native` directory
- Bugfix: Fixed `MakeSafeAbsolutePathInPlace` when a UNC prefix is present but path uses mixed delimiters

1 files changed, 310 insertions, 89 deletions

diff --git a/src/zenserver/hub/hydration.cpp b/src/zenserver/hub/hydration.cpp
index 541127590..ed16bfe56 100644
--- a/src/zenserver/hub/hydration.cpp
+++ b/src/zenserver/hub/hydration.cpp
@@ -10,6 +10,7 @@
 #include <zencore/fmtutils.h>
 #include <zencore/logging.h>
 #include <zencore/system.h>
+#include <zencore/timer.h>
 #include <zenutil/cloud/imdscredentials.h>
 #include <zenutil/cloud/s3client.h>
 
@@ -60,6 +61,7 @@ namespace {
 ///////////////////////////////////////////////////////////////////////////
 
 constexpr std::string_view FileHydratorPrefix = "file://";
+constexpr std::string_view FileHydratorType	  = "file";
 
 struct FileHydrator : public HydrationStrategyBase
 {
@@ -77,7 +79,21 @@ FileHydrator::Configure(const HydrationConfig& Config)
 {
 	m_Config = Config;
 
-	std::filesystem::path ConfigPath(Utf8ToWide(m_Config.TargetSpecification.substr(FileHydratorPrefix.length())));
+	std::filesystem::path ConfigPath;
+	if (!m_Config.TargetSpecification.empty())
+	{
+		ConfigPath = Utf8ToWide(m_Config.TargetSpecification.substr(FileHydratorPrefix.length()));
+	}
+	else
+	{
+		CbObjectView	 Settings = m_Config.Options["settings"].AsObjectView();
+		std::string_view Path	  = Settings["path"].AsString();
+		if (Path.empty())
+		{
+			throw zen::runtime_error("Hydration config 'file' type requires 'settings.path'");
+		}
+		ConfigPath = Utf8ToWide(std::string(Path));
+	}
 	MakeSafeAbsolutePathInPlace(ConfigPath);
 
 	if (!std::filesystem::exists(ConfigPath))
@@ -95,6 +111,8 @@ FileHydrator::Hydrate()
 {
 	ZEN_INFO("Hydrating state from '{}' to '{}'", m_StorageModuleRootDir, m_Config.ServerStateDir);
 
+	Stopwatch Timer;
+
 	// Ensure target is clean
 	ZEN_DEBUG("Wiping server state at '{}'", m_Config.ServerStateDir);
 	const bool ForceRemoveReadOnlyFiles = true;
@@ -120,6 +138,10 @@ FileHydrator::Hydrate()
 		ZEN_DEBUG("Cleaning server state '{}'", m_Config.ServerStateDir);
 		CleanDirectory(m_Config.ServerStateDir, ForceRemoveReadOnlyFiles);
 	}
+	else
+	{
+		ZEN_INFO("Hydration complete in {}", NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
+	}
 }
 
 void
@@ -127,6 +149,8 @@ FileHydrator::Dehydrate()
 {
 	ZEN_INFO("Dehydrating state from '{}' to '{}'", m_Config.ServerStateDir, m_StorageModuleRootDir);
 
+	Stopwatch Timer;
+
 	const std::filesystem::path TargetDir = m_StorageModuleRootDir;
 
 	// Ensure target is clean. This could be replaced with an atomic copy at a later date
@@ -141,7 +165,23 @@ FileHydrator::Dehydrate()
 	try
 	{
 		ZEN_DEBUG("Copying '{}' to '{}'", m_Config.ServerStateDir, TargetDir);
-		CopyTree(m_Config.ServerStateDir, TargetDir, {.EnableClone = true});
+		for (const std::filesystem::directory_entry& Entry : std::filesystem::directory_iterator(m_Config.ServerStateDir))
+		{
+			if (Entry.path().filename() == ".sentry-native")
+			{
+				continue;
+			}
+			std::filesystem::path Dest = TargetDir / Entry.path().filename();
+			if (Entry.is_directory())
+			{
+				CreateDirectories(Dest);
+				CopyTree(Entry.path(), Dest, {.EnableClone = true});
+			}
+			else
+			{
+				CopyFile(Entry.path(), Dest, {.EnableClone = true});
+			}
+		}
 	}
 	catch (std::exception& Ex)
 	{
@@ -159,11 +199,17 @@ FileHydrator::Dehydrate()
 
 	ZEN_DEBUG("Wiping server state '{}'", m_Config.ServerStateDir);
 	CleanDirectory(m_Config.ServerStateDir, ForceRemoveReadOnlyFiles);
+
+	if (CopySuccess)
+	{
+		ZEN_INFO("Dehydration complete in {}", NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
+	}
 }
 
 ///////////////////////////////////////////////////////////////////////////
 
 constexpr std::string_view S3HydratorPrefix = "s3://";
+constexpr std::string_view S3HydratorType	= "s3";
 
 struct S3Hydrator : public HydrationStrategyBase
 {
@@ -182,6 +228,8 @@ private:
 	std::string					m_Region;
 	SigV4Credentials			m_Credentials;
 	Ref<ImdsCredentialProvider> m_CredentialProvider;
+
+	static constexpr uint64_t MultipartChunkSize = 8 * 1024 * 1024;
 };
 
 void
@@ -189,8 +237,23 @@ S3Hydrator::Configure(const HydrationConfig& Config)
 {
 	m_Config = Config;
 
-	std::string_view Spec = m_Config.TargetSpecification;
-	Spec.remove_prefix(S3HydratorPrefix.size());
+	CbObjectView	 Settings = m_Config.Options["settings"].AsObjectView();
+	std::string_view Spec;
+	if (!m_Config.TargetSpecification.empty())
+	{
+		Spec = m_Config.TargetSpecification;
+		Spec.remove_prefix(S3HydratorPrefix.size());
+	}
+	else
+	{
+		std::string_view Uri = Settings["uri"].AsString();
+		if (Uri.empty())
+		{
+			throw zen::runtime_error("Hydration config 's3' type requires 'settings.uri'");
+		}
+		Spec = Uri;
+		Spec.remove_prefix(S3HydratorPrefix.size());
+	}
 
 	size_t		SlashPos   = Spec.find('/');
 	std::string UserPrefix = SlashPos != std::string_view::npos ? std::string(Spec.substr(SlashPos + 1)) : std::string{};
@@ -199,7 +262,11 @@ S3Hydrator::Configure(const HydrationConfig& Config)
 
 	ZEN_ASSERT(!m_Bucket.empty());
 
-	std::string Region = GetEnvVariable("AWS_DEFAULT_REGION");
+	std::string Region = std::string(Settings["region"].AsString());
+	if (Region.empty())
+	{
+		Region = GetEnvVariable("AWS_DEFAULT_REGION");
+	}
 	if (Region.empty())
 	{
 		Region = GetEnvVariable("AWS_REGION");
@@ -230,10 +297,12 @@ S3Hydrator::CreateS3Client() const
 	Options.BucketName = m_Bucket;
 	Options.Region	   = m_Region;
 
-	if (!m_Config.S3Endpoint.empty())
+	CbObjectView	 Settings = m_Config.Options["settings"].AsObjectView();
+	std::string_view Endpoint = Settings["endpoint"].AsString();
+	if (!Endpoint.empty())
 	{
-		Options.Endpoint  = m_Config.S3Endpoint;
-		Options.PathStyle = m_Config.S3PathStyle;
+		Options.Endpoint  = std::string(Endpoint);
+		Options.PathStyle = Settings["path-style"].AsBool();
 	}
 
 	if (m_CredentialProvider)
@@ -245,6 +314,8 @@ S3Hydrator::CreateS3Client() const
 		Options.Credentials = m_Credentials;
 	}
 
+	Options.HttpSettings.MaximumInMemoryDownloadSize = 16u * 1024u;
+
 	return S3Client(Options);
 }
 
@@ -275,11 +346,11 @@ S3Hydrator::Dehydrate()
 
 	try
 	{
-		S3Client							  Client	  = CreateS3Client();
-		std::string							  FolderName  = BuildTimestampFolderName();
-		uint64_t							  TotalBytes  = 0;
-		uint32_t							  FileCount	  = 0;
-		std::chrono::steady_clock::time_point UploadStart = std::chrono::steady_clock::now();
+		S3Client	Client	   = CreateS3Client();
+		std::string FolderName = BuildTimestampFolderName();
+		uint64_t	TotalBytes = 0;
+		uint32_t	FileCount  = 0;
+		Stopwatch	Timer;
 
 		DirectoryContent DirContent;
 		GetDirectoryContent(m_Config.ServerStateDir, DirectoryContentFlags::IncludeFiles | DirectoryContentFlags::Recursive, DirContent);
@@ -295,13 +366,20 @@ S3Hydrator::Dehydrate()
 					AbsPath.string(),
 					m_Config.ServerStateDir.string());
 			}
+			if (*RelPath.begin() == ".sentry-native")
+			{
+				continue;
+			}
 			std::string Key = MakeObjectKey(FolderName, RelPath);
 
 			BasicFile File(AbsPath, BasicFile::Mode::kRead);
 			uint64_t  FileSize = File.FileSize();
 
-			S3Result UploadResult =
-				Client.PutObjectMultipart(Key, FileSize, [&File](uint64_t Offset, uint64_t Size) { return File.ReadRange(Offset, Size); });
+			S3Result UploadResult = Client.PutObjectMultipart(
+				Key,
+				FileSize,
+				[&File](uint64_t Offset, uint64_t Size) { return File.ReadRange(Offset, Size); },
+				MultipartChunkSize);
 			if (!UploadResult.IsSuccess())
 			{
 				throw zen::runtime_error("Failed to upload '{}' to S3: {}", Key, UploadResult.Error);
@@ -312,8 +390,7 @@ S3Hydrator::Dehydrate()
 		}
 
 		// Write current-state.json
-		int64_t UploadDurationMs =
-			std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - UploadStart).count();
+		uint64_t UploadDurationMs = Timer.GetElapsedTimeMs();
 
 		UtcTime		Now			  = UtcTime::Now();
 		std::string UploadTimeUtc = fmt::format("{:04d}-{:02d}-{:02d}T{:02d}:{:02d}:{:02d}.{:03d}Z",
@@ -346,7 +423,7 @@ S3Hydrator::Dehydrate()
 			throw zen::runtime_error("Failed to write current-state.json to '{}': {}", MetaKey, MetaUploadResult.Error);
 		}
 
-		ZEN_INFO("Dehydration complete: {} files, {} bytes, {} ms", FileCount, TotalBytes, UploadDurationMs);
+		ZEN_INFO("Dehydration complete: {} files, {}, {}", FileCount, NiceBytes(TotalBytes), NiceTimeSpanMs(UploadDurationMs));
 	}
 	catch (std::exception& Ex)
 	{
@@ -361,6 +438,7 @@ S3Hydrator::Hydrate()
 {
 	ZEN_INFO("Hydrating state from s3://{}/{} to '{}'", m_Bucket, m_KeyPrefix, m_Config.ServerStateDir);
 
+	Stopwatch  Timer;
 	const bool ForceRemoveReadOnlyFiles = true;
 
 	// Clean temp dir before starting in case of leftover state from a previous failed hydration
@@ -374,19 +452,17 @@ S3Hydrator::Hydrate()
 		S3Client	Client	= CreateS3Client();
 		std::string MetaKey = m_KeyPrefix + "/current-state.json";
 
-		S3HeadObjectResult HeadResult = Client.HeadObject(MetaKey);
-		if (HeadResult.Status == HeadObjectResult::NotFound)
-		{
-			throw zen::runtime_error("No state found in S3 at '{}'", MetaKey);
-		}
-		if (!HeadResult.IsSuccess())
-		{
-			throw zen::runtime_error("Failed to check for state in S3 at '{}': {}", MetaKey, HeadResult.Error);
-		}
-
 		S3GetObjectResult MetaResult = Client.GetObject(MetaKey);
 		if (!MetaResult.IsSuccess())
 		{
+			if (MetaResult.Error == S3GetObjectResult::NotFoundErrorText)
+			{
+				ZEN_INFO("No state found in S3 at {}", MetaKey);
+
+				ZEN_DEBUG("Wiping server state '{}'", m_Config.ServerStateDir);
+				CleanDirectory(m_Config.ServerStateDir, ForceRemoveReadOnlyFiles);
+				return;
+			}
 			throw zen::runtime_error("Failed to read current-state.json from '{}': {}", MetaKey, MetaResult.Error);
 		}
 
@@ -426,17 +502,17 @@ S3Hydrator::Hydrate()
 			std::filesystem::path DestPath = MakeSafeAbsolutePath(m_Config.TempDir / std::filesystem::path(RelKey));
 			CreateDirectories(DestPath.parent_path());
 
-			BasicFile DestFile(DestPath, BasicFile::Mode::kTruncate);
-			DestFile.SetFileSize(Obj.Size);
-
-			if (Obj.Size > 0)
+			if (Obj.Size > MultipartChunkSize)
 			{
+				BasicFile DestFile(DestPath, BasicFile::Mode::kTruncate);
+				DestFile.SetFileSize(Obj.Size);
+
 				BasicFileWriter Writer(DestFile, 64 * 1024);
 
 				uint64_t Offset = 0;
 				while (Offset < Obj.Size)
 				{
-					uint64_t		  ChunkSize = std::min<uint64_t>(8 * 1024 * 1024, Obj.Size - Offset);
+					uint64_t		  ChunkSize = std::min<uint64_t>(MultipartChunkSize, Obj.Size - Offset);
 					S3GetObjectResult Chunk		= Client.GetObjectRange(Obj.Key, Offset, ChunkSize);
 					if (!Chunk.IsSuccess())
 					{
@@ -453,6 +529,34 @@ S3Hydrator::Hydrate()
 
 				Writer.Flush();
 			}
+			else
+			{
+				S3GetObjectResult Chunk = Client.GetObject(Obj.Key, m_Config.TempDir);
+				if (!Chunk.IsSuccess())
+				{
+					throw zen::runtime_error("Failed to download '{}' from S3: {}", Obj.Key, Chunk.Error);
+				}
+
+				if (IoBufferFileReference FileRef; Chunk.Content.GetFileReference(FileRef))
+				{
+					std::error_code		  Ec;
+					std::filesystem::path ChunkPath = PathFromHandle(FileRef.FileHandle, Ec);
+					if (Ec)
+					{
+						WriteFile(DestPath, Chunk.Content);
+					}
+					else
+					{
+						Chunk.Content.SetDeleteOnClose(false);
+						Chunk.Content = {};
+						RenameFile(ChunkPath, DestPath, Ec);
+					}
+				}
+				else
+				{
+					WriteFile(DestPath, Chunk.Content);
+				}
+			}
 		}
 
 		// Downloaded successfully - swap into ServerStateDir
@@ -465,19 +569,20 @@ S3Hydrator::Hydrate()
 			std::mismatch(m_Config.TempDir.begin(), m_Config.TempDir.end(), m_Config.ServerStateDir.begin(), m_Config.ServerStateDir.end());
 		if (ItTmp != m_Config.TempDir.begin())
 		{
-			// Fast path: atomic renames - no data copying needed
-			for (const std::filesystem::directory_entry& Entry : std::filesystem::directory_iterator(m_Config.TempDir))
+			DirectoryContent DirContent;
+			GetDirectoryContent(m_Config.TempDir, DirectoryContentFlags::IncludeFiles | DirectoryContentFlags::IncludeDirs, DirContent);
+
+			for (const std::filesystem::path& AbsPath : DirContent.Directories)
 			{
-				std::filesystem::path Dest = MakeSafeAbsolutePath(m_Config.ServerStateDir / Entry.path().filename());
-				if (Entry.is_directory())
-				{
-					RenameDirectory(Entry.path(), Dest);
-				}
-				else
-				{
-					RenameFile(Entry.path(), Dest);
-				}
+				std::filesystem::path Dest = MakeSafeAbsolutePath(m_Config.ServerStateDir / AbsPath.filename());
+				RenameDirectory(AbsPath, Dest);
+			}
+			for (const std::filesystem::path& AbsPath : DirContent.Files)
+			{
+				std::filesystem::path Dest = MakeSafeAbsolutePath(m_Config.ServerStateDir / AbsPath.filename());
+				RenameFile(AbsPath, Dest);
 			}
+
 			ZEN_DEBUG("Cleaning temp dir '{}'", m_Config.TempDir);
 			CleanDirectory(m_Config.TempDir, ForceRemoveReadOnlyFiles);
 		}
@@ -491,7 +596,7 @@ S3Hydrator::Hydrate()
 			CleanDirectory(m_Config.TempDir, ForceRemoveReadOnlyFiles);
 		}
 
-		ZEN_INFO("Hydration complete from folder '{}'", FolderName);
+		ZEN_INFO("Hydration complete from folder '{}' in {}", FolderName, NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
 	}
 	catch (std::exception& Ex)
 	{
@@ -513,19 +618,41 @@ S3Hydrator::Hydrate()
 std::unique_ptr<HydrationStrategyBase>
 CreateHydrator(const HydrationConfig& Config)
 {
-	if (StrCaseCompare(Config.TargetSpecification.substr(0, FileHydratorPrefix.length()), FileHydratorPrefix) == 0)
+	if (!Config.TargetSpecification.empty())
+	{
+		if (StrCaseCompare(Config.TargetSpecification.substr(0, FileHydratorPrefix.length()), FileHydratorPrefix) == 0)
+		{
+			std::unique_ptr<HydrationStrategyBase> Hydrator = std::make_unique<FileHydrator>();
+			Hydrator->Configure(Config);
+			return Hydrator;
+		}
+		if (StrCaseCompare(Config.TargetSpecification.substr(0, S3HydratorPrefix.length()), S3HydratorPrefix) == 0)
+		{
+			std::unique_ptr<HydrationStrategyBase> Hydrator = std::make_unique<S3Hydrator>();
+			Hydrator->Configure(Config);
+			return Hydrator;
+		}
+		throw std::runtime_error(fmt::format("Unknown hydration strategy: {}", Config.TargetSpecification));
+	}
+
+	std::string_view Type = Config.Options["type"].AsString();
+	if (Type == FileHydratorType)
 	{
 		std::unique_ptr<HydrationStrategyBase> Hydrator = std::make_unique<FileHydrator>();
 		Hydrator->Configure(Config);
 		return Hydrator;
 	}
-	if (StrCaseCompare(Config.TargetSpecification.substr(0, S3HydratorPrefix.length()), S3HydratorPrefix) == 0)
+	if (Type == S3HydratorType)
 	{
 		std::unique_ptr<HydrationStrategyBase> Hydrator = std::make_unique<S3Hydrator>();
 		Hydrator->Configure(Config);
 		return Hydrator;
 	}
-	throw std::runtime_error(fmt::format("Unknown hydration strategy: {}", Config.TargetSpecification));
+	if (!Type.empty())
+	{
+		throw zen::runtime_error("Unknown hydration target type '{}'", Type);
+	}
+	throw zen::runtime_error("No hydration target configured");
 }
 
 #if ZEN_WITH_TESTS
@@ -607,6 +734,12 @@ namespace {
 		AddFile("file_a.bin", CreateSemiRandomBlob(1024));
 		AddFile("subdir/file_b.bin", CreateSemiRandomBlob(2048));
 		AddFile("subdir/nested/file_c.bin", CreateSemiRandomBlob(512));
+		AddFile("subdir/nested/file_d.bin", CreateSemiRandomBlob(512));
+		AddFile("subdir/nested/file_e.bin", CreateSemiRandomBlob(512));
+		AddFile("subdir/nested/file_f.bin", CreateSemiRandomBlob(512));
+		AddFile("subdir/nested/medium.bulk", CreateSemiRandomBlob(256u * 1024u));
+		AddFile("subdir/nested/big.bulk", CreateSemiRandomBlob(512u * 1024u));
+		AddFile("subdir/nested/huge.bulk", CreateSemiRandomBlob(9u * 1024u * 1024u));
 
 		return Files;
 	}
@@ -844,12 +977,16 @@ TEST_CASE("hydration.s3.dehydrate_hydrate")
 	auto			  TestFiles = CreateTestTree(ServerStateDir);
 
 	HydrationConfig Config;
-	Config.ServerStateDir	   = ServerStateDir;
-	Config.TempDir			   = HydrationTemp;
-	Config.ModuleId			   = ModuleId;
-	Config.TargetSpecification = "s3://zen-hydration-test";
-	Config.S3Endpoint		   = Minio.Endpoint();
-	Config.S3PathStyle		   = true;
+	Config.ServerStateDir = ServerStateDir;
+	Config.TempDir		  = HydrationTemp;
+	Config.ModuleId		  = ModuleId;
+	std::string ConfigJson =
+		fmt::format(R"({{"type":"s3","settings":{{"uri":"s3://zen-hydration-test","endpoint":"{}","path-style":true}}}})",
+					Minio.Endpoint());
+	std::string		ParseError;
+	CbFieldIterator Root = LoadCompactBinaryFromJson(ConfigJson, ParseError);
+	ZEN_ASSERT(ParseError.empty() && Root.IsObject());
+	Config.Options = std::move(Root).AsObject();
 
 	// Dehydrate: upload server state to MinIO
 	{
@@ -902,12 +1039,18 @@ TEST_CASE("hydration.s3.current_state_json_selects_latest_folder")
 	const std::string ModuleId = "s3test_folder_select";
 
 	HydrationConfig Config;
-	Config.ServerStateDir	   = ServerStateDir;
-	Config.TempDir			   = HydrationTemp;
-	Config.ModuleId			   = ModuleId;
-	Config.TargetSpecification = "s3://zen-hydration-test";
-	Config.S3Endpoint		   = Minio.Endpoint();
-	Config.S3PathStyle		   = true;
+	Config.ServerStateDir = ServerStateDir;
+	Config.TempDir		  = HydrationTemp;
+	Config.ModuleId		  = ModuleId;
+	{
+		std::string ConfigJson =
+			fmt::format(R"({{"type":"s3","settings":{{"uri":"s3://zen-hydration-test","endpoint":"{}","path-style":true}}}})",
+						Minio.Endpoint());
+		std::string		ParseError;
+		CbFieldIterator Root = LoadCompactBinaryFromJson(ConfigJson, ParseError);
+		ZEN_ASSERT(ParseError.empty() && Root.IsObject());
+		Config.Options = std::move(Root).AsObject();
+	}
 
 	// v1: dehydrate without a marker file
 	CreateTestTree(ServerStateDir);
@@ -972,13 +1115,19 @@ TEST_CASE("hydration.s3.module_isolation")
 		CreateDirectories(TempPath);
 
 		ModuleData Data;
-		Data.Config.ServerStateDir		= StateDir;
-		Data.Config.TempDir				= TempPath;
-		Data.Config.ModuleId			= ModuleId;
-		Data.Config.TargetSpecification = "s3://zen-hydration-test";
-		Data.Config.S3Endpoint			= Minio.Endpoint();
-		Data.Config.S3PathStyle			= true;
-		Data.Files						= CreateTestTree(StateDir);
+		Data.Config.ServerStateDir = StateDir;
+		Data.Config.TempDir		   = TempPath;
+		Data.Config.ModuleId	   = ModuleId;
+		{
+			std::string ConfigJson =
+				fmt::format(R"({{"type":"s3","settings":{{"uri":"s3://zen-hydration-test","endpoint":"{}","path-style":true}}}})",
+							Minio.Endpoint());
+			std::string		ParseError;
+			CbFieldIterator Root = LoadCompactBinaryFromJson(ConfigJson, ParseError);
+			ZEN_ASSERT(ParseError.empty() && Root.IsObject());
+			Data.Config.Options = std::move(Root).AsObject();
+		}
+		Data.Files = CreateTestTree(StateDir);
 
 		std::unique_ptr<HydrationStrategyBase> Hydrator = CreateHydrator(Data.Config);
 		Hydrator->Dehydrate();
@@ -1015,7 +1164,8 @@ TEST_CASE("hydration.s3.concurrent")
 	ScopedEnvVar EnvAccessKey("AWS_ACCESS_KEY_ID", Minio.RootUser());
 	ScopedEnvVar EnvSecretKey("AWS_SECRET_ACCESS_KEY", Minio.RootPassword());
 
-	constexpr int kModuleCount = 4;
+	constexpr int kModuleCount = 16;
+	constexpr int kThreadCount = 4;
 
 	ScopedTemporaryDirectory TempDir;
 
@@ -1034,18 +1184,24 @@ TEST_CASE("hydration.s3.concurrent")
 		CreateDirectories(StateDir);
 		CreateDirectories(TempPath);
 
-		Modules[I].Config.ServerStateDir	  = StateDir;
-		Modules[I].Config.TempDir			  = TempPath;
-		Modules[I].Config.ModuleId			  = ModuleId;
-		Modules[I].Config.TargetSpecification = "s3://zen-hydration-test";
-		Modules[I].Config.S3Endpoint		  = Minio.Endpoint();
-		Modules[I].Config.S3PathStyle		  = true;
-		Modules[I].Files					  = CreateTestTree(StateDir);
+		Modules[I].Config.ServerStateDir = StateDir;
+		Modules[I].Config.TempDir		 = TempPath;
+		Modules[I].Config.ModuleId		 = ModuleId;
+		{
+			std::string ConfigJson =
+				fmt::format(R"({{"type":"s3","settings":{{"uri":"s3://zen-hydration-test","endpoint":"{}","path-style":true}}}})",
+							Minio.Endpoint());
+			std::string		ParseError;
+			CbFieldIterator Root = LoadCompactBinaryFromJson(ConfigJson, ParseError);
+			ZEN_ASSERT(ParseError.empty() && Root.IsObject());
+			Modules[I].Config.Options = std::move(Root).AsObject();
+		}
+		Modules[I].Files = CreateTestTree(StateDir);
 	}
 
 	// Concurrent dehydrate
 	{
-		WorkerThreadPool  Pool(kModuleCount, "hydration_s3_dehy");
+		WorkerThreadPool  Pool(kThreadCount, "hydration_s3_dehy");
 		std::atomic<bool> AbortFlag{false};
 		std::atomic<bool> PauseFlag{false};
 		ParallelWork	  Work(AbortFlag, PauseFlag, WorkerThreadPool::EMode::EnableBacklog);
@@ -1063,7 +1219,7 @@ TEST_CASE("hydration.s3.concurrent")
 
 	// Concurrent hydrate
 	{
-		WorkerThreadPool  Pool(kModuleCount, "hydration_s3_hy");
+		WorkerThreadPool  Pool(kThreadCount, "hydration_s3_hy");
 		std::atomic<bool> AbortFlag{false};
 		std::atomic<bool> PauseFlag{false};
 		ParallelWork	  Work(AbortFlag, PauseFlag, WorkerThreadPool::EMode::EnableBacklog);
@@ -1116,12 +1272,18 @@ TEST_CASE("hydration.s3.no_prior_state")
 	WriteFile(ServerStateDir / "stale.bin", CreateSemiRandomBlob(256));
 
 	HydrationConfig Config;
-	Config.ServerStateDir	   = ServerStateDir;
-	Config.TempDir			   = HydrationTemp;
-	Config.ModuleId			   = "s3test_no_prior";
-	Config.TargetSpecification = "s3://zen-hydration-test";
-	Config.S3Endpoint		   = Minio.Endpoint();
-	Config.S3PathStyle		   = true;
+	Config.ServerStateDir = ServerStateDir;
+	Config.TempDir		  = HydrationTemp;
+	Config.ModuleId		  = "s3test_no_prior";
+	{
+		std::string ConfigJson =
+			fmt::format(R"({{"type":"s3","settings":{{"uri":"s3://zen-hydration-test","endpoint":"{}","path-style":true}}}})",
+						Minio.Endpoint());
+		std::string		ParseError;
+		CbFieldIterator Root = LoadCompactBinaryFromJson(ConfigJson, ParseError);
+		ZEN_ASSERT(ParseError.empty() && Root.IsObject());
+		Config.Options = std::move(Root).AsObject();
+	}
 
 	std::unique_ptr<HydrationStrategyBase> Hydrator = CreateHydrator(Config);
 	Hydrator->Hydrate();
@@ -1159,12 +1321,71 @@ TEST_CASE("hydration.s3.path_prefix")
 	std::vector<std::pair<std::filesystem::path, IoBuffer>> TestFiles = CreateTestTree(ServerStateDir);
 
 	HydrationConfig Config;
-	Config.ServerStateDir	   = ServerStateDir;
-	Config.TempDir			   = HydrationTemp;
-	Config.ModuleId			   = "s3test_prefix";
-	Config.TargetSpecification = "s3://zen-hydration-test/team/project";
-	Config.S3Endpoint		   = Minio.Endpoint();
-	Config.S3PathStyle		   = true;
+	Config.ServerStateDir = ServerStateDir;
+	Config.TempDir		  = HydrationTemp;
+	Config.ModuleId		  = "s3test_prefix";
+	{
+		std::string ConfigJson =
+			fmt::format(R"({{"type":"s3","settings":{{"uri":"s3://zen-hydration-test/team/project","endpoint":"{}","path-style":true}}}})",
+						Minio.Endpoint());
+		std::string		ParseError;
+		CbFieldIterator Root = LoadCompactBinaryFromJson(ConfigJson, ParseError);
+		ZEN_ASSERT(ParseError.empty() && Root.IsObject());
+		Config.Options = std::move(Root).AsObject();
+	}
+
+	{
+		std::unique_ptr<HydrationStrategyBase> Hydrator = CreateHydrator(Config);
+		Hydrator->Dehydrate();
+	}
+
+	CleanDirectory(ServerStateDir, true);
+
+	{
+		std::unique_ptr<HydrationStrategyBase> Hydrator = CreateHydrator(Config);
+		Hydrator->Hydrate();
+	}
+
+	VerifyTree(ServerStateDir, TestFiles);
+}
+
+TEST_CASE("hydration.s3.options_region_override")
+{
+	// Verify that 'region' in Options["settings"] takes precedence over AWS_DEFAULT_REGION env var.
+	// AWS_DEFAULT_REGION is set to a bogus value; hydration must succeed using the region from Options.
+
+	MinioProcessOptions MinioOpts;
+	MinioOpts.Port = 19016;
+	MinioProcess Minio(MinioOpts);
+	Minio.SpawnMinioServer();
+	Minio.CreateBucket("zen-hydration-test");
+
+	ScopedEnvVar EnvAccessKey("AWS_ACCESS_KEY_ID", Minio.RootUser());
+	ScopedEnvVar EnvSecretKey("AWS_SECRET_ACCESS_KEY", Minio.RootPassword());
+	ScopedEnvVar EnvRegion("AWS_DEFAULT_REGION", "wrong-region");
+
+	ScopedTemporaryDirectory TempDir;
+
+	std::filesystem::path ServerStateDir = TempDir.Path() / "server_state";
+	std::filesystem::path HydrationTemp	 = TempDir.Path() / "hydration_temp";
+	CreateDirectories(ServerStateDir);
+	CreateDirectories(HydrationTemp);
+
+	auto TestFiles = CreateTestTree(ServerStateDir);
+
+	HydrationConfig Config;
+	Config.ServerStateDir = ServerStateDir;
+	Config.TempDir		  = HydrationTemp;
+	Config.ModuleId		  = "s3test_region_override";
+	{
+		std::string ConfigJson = fmt::format(
+			R"({{"type":"s3","settings":{{"uri":"s3://zen-hydration-test","endpoint":"{}","path-style":true,"region":"us-east-1"}}}})",
+			Minio.Endpoint());
+		std::string		ParseError;
+		CbFieldIterator Root = LoadCompactBinaryFromJson(ConfigJson, ParseError);
+		ZEN_ASSERT(ParseError.empty() && Root.IsObject());
+		Config.Options = std::move(Root).AsObject();
+	}
 
 	{
 		std::unique_ptr<HydrationStrategyBase> Hydrator = CreateHydrator(Config);