Merge remote-tracking branch 'origin/main' into de/zen-service-command

1 files changed, 823 insertions, 5 deletions

diff --git a/src/zenserver-test/zenserver-test.cpp b/src/zenserver-test/zenserver-test.cpp
index 78d8ab68a..77ed87cb1 100644
--- a/src/zenserver-test/zenserver-test.cpp
+++ b/src/zenserver-test/zenserver-test.cpp
@@ -13,6 +13,7 @@
 #include <zencore/iohash.h>
 #include <zencore/logging.h>
 #include <zencore/memoryview.h>
+#include <zencore/scopeguard.h>
 #include <zencore/stream.h>
 #include <zencore/string.h>
 #include <zencore/testutils.h>
@@ -22,6 +23,7 @@
 #include <zenhttp/httpclient.h>
 #include <zenhttp/packageformat.h>
 #include <zenhttp/zenhttp.h>
+#include <zenutil/buildstoragecache.h>
 #include <zenutil/cache/cache.h>
 #include <zenutil/cache/cacherequests.h>
 #include <zenutil/chunkrequests.h>
@@ -37,6 +39,7 @@
 
 ZEN_THIRD_PARTY_INCLUDES_START
 #include <cpr/cpr.h>
+#include <tsl/robin_set.h>
 #undef GetObject
 ZEN_THIRD_PARTY_INCLUDES_END
 
@@ -554,9 +557,13 @@ namespace utils {
 			return ZenConfig{.DataDir = TestEnv.CreateNewTestDir(), .Port = Port, .Args = std::move(Args)};
 		}
 
-		static ZenConfig NewWithUpstream(uint16_t Port, uint16_t UpstreamPort)
+		static ZenConfig NewWithUpstream(uint16_t Port, uint16_t UpstreamPort, std::string Args = "")
 		{
-			return New(Port, fmt::format("--debug --upstream-thread-count=0 --upstream-zen-url=http://localhost:{}", UpstreamPort));
+			return New(Port,
+					   fmt::format("{}{}--debug --upstream-thread-count=0 --upstream-zen-url=http://localhost:{}",
+								   Args,
+								   Args.length() > 0 ? " " : "",
+								   UpstreamPort));
 		}
 
 		static ZenConfig NewWithThreadedUpstreams(uint16_t NewPort, std::span<uint16_t> UpstreamPorts, bool Debug)
@@ -1302,8 +1309,10 @@ TEST_CASE("zcache.rpc")
 												std::string_view Namespace,
 												std::string_view Bucket,
 												size_t			 Num,
-												size_t			 PayloadSize = 1024,
-												size_t			 KeyOffset	 = 1) -> std::vector<CacheKey> {
+												size_t			 PayloadSize		= 1024,
+												size_t			 KeyOffset			= 1,
+												CachePolicy		 PutPolicy			= CachePolicy::Default,
+												std::vector<CbPackage>* OutPackages = nullptr) -> std::vector<CacheKey> {
 		std::vector<zen::CacheKey> OutKeys;
 
 		for (uint32_t Key = 1; Key <= Num; ++Key)
@@ -1313,7 +1322,7 @@ TEST_CASE("zcache.rpc")
 			const zen::CacheKey CacheKey   = zen::CacheKey::Create(Bucket, KeyHash);
 
 			cacherequests::PutCacheRecordsRequest Request = {.AcceptMagic = kCbPkgMagic, .Namespace = std::string(Namespace)};
-			AppendCacheRecord(Request, CacheKey, PayloadSize, CachePolicy::Default);
+			AppendCacheRecord(Request, CacheKey, PayloadSize, PutPolicy);
 			OutKeys.push_back(CacheKey);
 
 			CbPackage Package;
@@ -1325,6 +1334,10 @@ TEST_CASE("zcache.rpc")
 											 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});
 
 			CHECK(Result.status_code == 200);
+			if (OutPackages)
+			{
+				OutPackages->emplace_back(std::move(Package));
+			}
 		}
 
 		return OutKeys;
@@ -1503,6 +1516,339 @@ TEST_CASE("zcache.rpc")
 		}
 	}
 
+	SUBCASE("policy - 'QueryLocal' on put allows overwrite with differing value when not limiting overwrites")
+	{
+		using namespace utils;
+
+		ZenConfig		  UpstreamCfg = ZenConfig::New(TestEnv.GetNewPortNumber());
+		ZenServerInstance UpstreamServer(TestEnv);
+		SpawnServer(UpstreamServer, UpstreamCfg);
+
+		ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(TestEnv.GetNewPortNumber(), UpstreamCfg.Port);
+		ZenServerInstance LocalServer(TestEnv);
+		SpawnServer(LocalServer, LocalCfg);
+
+		size_t					   PayloadSize = 1024;
+		std::string_view		   Namespace("ue4.ddc"sv);
+		std::string_view		   Bucket("mastodon"sv);
+		const size_t			   NumRecords = 4;
+		std::vector<zen::CacheKey> Keys		  = PutCacheRecords(LocalCfg.BaseUri, Namespace, Bucket, NumRecords, PayloadSize);
+
+		for (const zen::CacheKey& CacheKey : Keys)
+		{
+			cacherequests::PutCacheRecordsRequest Request = {.AcceptMagic = kCbPkgMagic, .Namespace = std::string(Namespace)};
+			AppendCacheRecord(Request, CacheKey, PayloadSize * 2, CachePolicy::Default);
+
+			CbPackage Package;
+			CHECK(Request.Format(Package));
+
+			IoBuffer	  Body	 = FormatPackageMessageBuffer(Package).Flatten().AsIoBuffer();
+			cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", LocalCfg.BaseUri)},
+											 cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}, {"Accept", "application/x-ue-cbpkg"}},
+											 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});
+
+			CHECK(Result.status_code == 200);
+			cacherequests::PutCacheRecordsResult ParsedResult;
+			CbPackage Response = ParsePackageMessage(zen::IoBuffer(zen::IoBuffer::Wrap, Result.text.data(), Result.text.size()));
+			CHECK(!Response.IsNull());
+			CHECK(ParsedResult.Parse(Response));
+			for (bool ResponseSuccess : ParsedResult.Success)
+			{
+				CHECK(ResponseSuccess);
+			}
+		}
+
+		auto CheckRecordCorrectness = [&](const ZenConfig& Cfg) {
+			CachePolicy			 Policy = (CachePolicy::QueryLocal | CachePolicy::QueryRemote);
+			GetCacheRecordResult Result = GetCacheRecords(Cfg.BaseUri, "ue4.ddc"sv, Keys, Policy);
+
+			CHECK(Result.Result.Results.size() == Keys.size());
+
+			for (size_t Index = 0; const std::optional<cacherequests::GetCacheRecordResult>& Record : Result.Result.Results)
+			{
+				CHECK(Record);
+				const CacheKey& ExpectedKey = Keys[Index++];
+				CHECK(Record->Key == ExpectedKey);
+				for (const cacherequests::GetCacheRecordResultValue& Value : Record->Values)
+				{
+					CHECK(Value.RawSize == PayloadSize * 2);
+				}
+			}
+		};
+
+		// Check that the records are present and overwritten in the local server
+		CheckRecordCorrectness(LocalCfg);
+
+		// Check that the records are present and overwritten in the upstream server
+		CheckRecordCorrectness(UpstreamCfg);
+	}
+
+	SUBCASE("policy - 'QueryLocal' on put denies overwrite with differing value when limiting overwrites")
+	{
+		using namespace utils;
+
+		ZenConfig		  UpstreamCfg = ZenConfig::New(TestEnv.GetNewPortNumber());
+		ZenServerInstance UpstreamServer(TestEnv);
+		SpawnServer(UpstreamServer, UpstreamCfg);
+
+		ZenConfig LocalCfg = ZenConfig::NewWithUpstream(TestEnv.GetNewPortNumber(), UpstreamCfg.Port, "--cache-bucket-limit-overwrites");
+		ZenServerInstance LocalServer(TestEnv);
+		SpawnServer(LocalServer, LocalCfg);
+
+		size_t					   PayloadSize = 1024;
+		std::string_view		   Namespace("ue4.ddc"sv);
+		std::string_view		   Bucket("mastodon"sv);
+		const size_t			   NumRecords = 4;
+		std::vector<zen::CacheKey> Keys		  = PutCacheRecords(LocalCfg.BaseUri, Namespace, Bucket, NumRecords, PayloadSize);
+
+		for (const zen::CacheKey& CacheKey : Keys)
+		{
+			cacherequests::PutCacheRecordsRequest Request = {.AcceptMagic = kCbPkgMagic, .Namespace = std::string(Namespace)};
+			AppendCacheRecord(Request, CacheKey, PayloadSize * 2, CachePolicy::Default);
+
+			CbPackage Package;
+			CHECK(Request.Format(Package));
+
+			IoBuffer	  Body	 = FormatPackageMessageBuffer(Package).Flatten().AsIoBuffer();
+			cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", LocalCfg.BaseUri)},
+											 cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}, {"Accept", "application/x-ue-cbpkg"}},
+											 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});
+
+			CHECK(Result.status_code == 200);
+			cacherequests::PutCacheRecordsResult ParsedResult;
+			CbPackage Response = ParsePackageMessage(zen::IoBuffer(zen::IoBuffer::Wrap, Result.text.data(), Result.text.size()));
+			CHECK(!Response.IsNull());
+			CHECK(ParsedResult.Parse(Response));
+			for (bool ResponseSuccess : ParsedResult.Success)
+			{
+				CHECK(!ResponseSuccess);
+			}
+		}
+
+		auto CheckRecordCorrectness = [&](const ZenConfig& Cfg) {
+			CachePolicy			 Policy = (CachePolicy::QueryLocal | CachePolicy::QueryRemote);
+			GetCacheRecordResult Result = GetCacheRecords(Cfg.BaseUri, "ue4.ddc"sv, Keys, Policy);
+
+			CHECK(Result.Result.Results.size() == Keys.size());
+
+			for (size_t Index = 0; const std::optional<cacherequests::GetCacheRecordResult>& Record : Result.Result.Results)
+			{
+				CHECK(Record);
+				const CacheKey& ExpectedKey = Keys[Index++];
+				CHECK(Record->Key == ExpectedKey);
+				for (const cacherequests::GetCacheRecordResultValue& Value : Record->Values)
+				{
+					CHECK(Value.RawSize == PayloadSize);
+				}
+			}
+		};
+
+		// Check that the records are present and not overwritten in the local server
+		CheckRecordCorrectness(LocalCfg);
+
+		// Check that the records are present and not overwritten in the upstream server
+		CheckRecordCorrectness(UpstreamCfg);
+	}
+
+	SUBCASE("policy - no 'QueryLocal' on put allows overwrite with differing value when limiting overwrites")
+	{
+		using namespace utils;
+
+		ZenConfig		  UpstreamCfg = ZenConfig::New(TestEnv.GetNewPortNumber());
+		ZenServerInstance UpstreamServer(TestEnv);
+		SpawnServer(UpstreamServer, UpstreamCfg);
+
+		ZenConfig LocalCfg = ZenConfig::NewWithUpstream(TestEnv.GetNewPortNumber(), UpstreamCfg.Port, "--cache-bucket-limit-overwrites");
+		ZenServerInstance LocalServer(TestEnv);
+		SpawnServer(LocalServer, LocalCfg);
+
+		size_t					   PayloadSize = 1024;
+		std::string_view		   Namespace("ue4.ddc"sv);
+		std::string_view		   Bucket("mastodon"sv);
+		const size_t			   NumRecords = 4;
+		std::vector<zen::CacheKey> Keys		  = PutCacheRecords(LocalCfg.BaseUri, Namespace, Bucket, NumRecords, PayloadSize);
+
+		for (const zen::CacheKey& CacheKey : Keys)
+		{
+			cacherequests::PutCacheRecordsRequest Request = {.AcceptMagic = kCbPkgMagic, .Namespace = std::string(Namespace)};
+			AppendCacheRecord(Request, CacheKey, PayloadSize * 2, CachePolicy::Store);
+
+			CbPackage Package;
+			CHECK(Request.Format(Package));
+
+			IoBuffer	  Body	 = FormatPackageMessageBuffer(Package).Flatten().AsIoBuffer();
+			cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", LocalCfg.BaseUri)},
+											 cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}, {"Accept", "application/x-ue-cbpkg"}},
+											 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});
+
+			CHECK(Result.status_code == 200);
+			cacherequests::PutCacheRecordsResult ParsedResult;
+			CbPackage Response = ParsePackageMessage(zen::IoBuffer(zen::IoBuffer::Wrap, Result.text.data(), Result.text.size()));
+			CHECK(!Response.IsNull());
+			CHECK(ParsedResult.Parse(Response));
+			for (bool ResponseSuccess : ParsedResult.Success)
+			{
+				CHECK(ResponseSuccess);
+			}
+		}
+
+		auto CheckRecordCorrectness = [&](const ZenConfig& Cfg) {
+			CachePolicy			 Policy = (CachePolicy::QueryLocal | CachePolicy::QueryRemote);
+			GetCacheRecordResult Result = GetCacheRecords(Cfg.BaseUri, "ue4.ddc"sv, Keys, Policy);
+
+			CHECK(Result.Result.Results.size() == Keys.size());
+
+			for (size_t Index = 0; const std::optional<cacherequests::GetCacheRecordResult>& Record : Result.Result.Results)
+			{
+				CHECK(Record);
+				const CacheKey& ExpectedKey = Keys[Index++];
+				CHECK(Record->Key == ExpectedKey);
+				for (const cacherequests::GetCacheRecordResultValue& Value : Record->Values)
+				{
+					CHECK(Value.RawSize == PayloadSize * 2);
+				}
+			}
+		};
+
+		// Check that the records are present and overwritten in the local server
+		CheckRecordCorrectness(LocalCfg);
+
+		// Check that the records are present and overwritten in the upstream server
+		CheckRecordCorrectness(UpstreamCfg);
+	}
+
+	SUBCASE("policy - 'QueryLocal' on put allows overwrite with equivalent value when limiting overwrites")
+	{
+		using namespace utils;
+
+		ZenConfig		  UpstreamCfg = ZenConfig::New(TestEnv.GetNewPortNumber());
+		ZenServerInstance UpstreamServer(TestEnv);
+		SpawnServer(UpstreamServer, UpstreamCfg);
+
+		ZenConfig LocalCfg = ZenConfig::NewWithUpstream(TestEnv.GetNewPortNumber(), UpstreamCfg.Port, "--cache-bucket-limit-overwrites");
+		ZenServerInstance LocalServer(TestEnv);
+		SpawnServer(LocalServer, LocalCfg);
+
+		size_t					   PayloadSize = 1024;
+		std::string_view		   Namespace("ue4.ddc"sv);
+		std::string_view		   Bucket("mastodon"sv);
+		const size_t			   NumRecords = 4;
+		std::vector<CbPackage>	   Packages;
+		std::vector<zen::CacheKey> Keys =
+			PutCacheRecords(LocalCfg.BaseUri, Namespace, Bucket, NumRecords, PayloadSize, 1, CachePolicy::Default, &Packages);
+
+		for (const CbPackage& Package : Packages)
+		{
+			IoBuffer	  Body	 = FormatPackageMessageBuffer(Package).Flatten().AsIoBuffer();
+			cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", LocalCfg.BaseUri)},
+											 cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}, {"Accept", "application/x-ue-cbpkg"}},
+											 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});
+
+			CHECK(Result.status_code == 200);
+			cacherequests::PutCacheRecordsResult ParsedResult;
+			CbPackage Response = ParsePackageMessage(zen::IoBuffer(zen::IoBuffer::Wrap, Result.text.data(), Result.text.size()));
+			CHECK(!Response.IsNull());
+			CHECK(ParsedResult.Parse(Response));
+			for (bool ResponseSuccess : ParsedResult.Success)
+			{
+				CHECK(ResponseSuccess);
+			}
+		}
+
+		auto CheckRecordCorrectness = [&](const ZenConfig& Cfg) {
+			CachePolicy			 Policy = (CachePolicy::QueryLocal | CachePolicy::QueryRemote);
+			GetCacheRecordResult Result = GetCacheRecords(Cfg.BaseUri, "ue4.ddc"sv, Keys, Policy);
+
+			CHECK(Result.Result.Results.size() == Keys.size());
+
+			for (size_t Index = 0; const std::optional<cacherequests::GetCacheRecordResult>& Record : Result.Result.Results)
+			{
+				CHECK(Record);
+				const CacheKey& ExpectedKey = Keys[Index++];
+				CHECK(Record->Key == ExpectedKey);
+				for (const cacherequests::GetCacheRecordResultValue& Value : Record->Values)
+				{
+					CHECK(Value.RawSize == PayloadSize);
+				}
+			}
+		};
+
+		// Check that the records are present and unchanged in the local server
+		CheckRecordCorrectness(LocalCfg);
+
+		// Check that the records are present and unchanged in the upstream server
+		CheckRecordCorrectness(UpstreamCfg);
+	}
+
+	// TODO: Propagation for rejected PUTs
+	// SUBCASE("policy - 'QueryLocal' on put denies overwrite with differing value when limiting overwrites but allows propagation to
+	// upstream")
+	// {
+	// 	using namespace utils;
+
+	// 	ZenConfig		  UpstreamCfg = ZenConfig::New(TestEnv.GetNewPortNumber());
+	// 	ZenServerInstance UpstreamServer(TestEnv);
+	// 	SpawnServer(UpstreamServer, UpstreamCfg);
+
+	// 	ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(TestEnv.GetNewPortNumber(), UpstreamCfg.Port,
+	// "--cache-bucket-limit-overwrites"); 	ZenServerInstance LocalServer(TestEnv); 	SpawnServer(LocalServer, LocalCfg);
+
+	// 	size_t					   PayloadSize = 1024;
+	// 	std::string_view		   Namespace("ue4.ddc"sv);
+	// 	std::string_view		   Bucket("mastodon"sv);
+	// 	const size_t			   NumRecords = 4;
+	// 	std::vector<zen::CacheKey> Keys		  = PutCacheRecords(LocalCfg.BaseUri, Namespace, Bucket, NumRecords, PayloadSize, 1,
+	// CachePolicy::Local);
+
+	// 	for (const zen::CacheKey& CacheKey : Keys)
+	// 	{
+	// 		cacherequests::PutCacheRecordsRequest Request = {.AcceptMagic = kCbPkgMagic, .Namespace = std::string(Namespace)};
+	// 		AppendCacheRecord(Request, CacheKey, PayloadSize * 2, CachePolicy::Default);
+
+	// 		CbPackage Package;
+	// 		CHECK(Request.Format(Package));
+
+	// 		IoBuffer	  Body	 = FormatPackageMessageBuffer(Package).Flatten().AsIoBuffer();
+	// 		cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", LocalCfg.BaseUri)},
+	// 										 cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}, {"Accept", "application/x-ue-cbpkg"}},
+	// 										 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});
+
+	// 		CHECK(Result.status_code == 200);
+	// 		cacherequests::PutCacheRecordsResult ParsedResult;
+	// 		CbPackage Response = ParsePackageMessage(zen::IoBuffer(zen::IoBuffer::Wrap, Result.text.data(), Result.text.size()));
+	// 		CHECK(!Response.IsNull());
+	// 		CHECK(ParsedResult.Parse(Response));
+	// 		for (bool ResponseSuccess : ParsedResult.Success)
+	// 		{
+	// 			CHECK(!ResponseSuccess);
+	// 		}
+	// 	}
+
+	// 	auto CheckRecordCorrectness = [&](const ZenConfig& Cfg, size_t ExpectedPayloadSize) {
+	// 		CachePolicy			 Policy = (CachePolicy::QueryLocal | CachePolicy::QueryRemote);
+	// 		GetCacheRecordResult Result = GetCacheRecords(Cfg.BaseUri, "ue4.ddc"sv, Keys, Policy);
+
+	// 		CHECK(Result.Result.Results.size() == Keys.size());
+
+	// 		for (size_t Index = 0; const std::optional<cacherequests::GetCacheRecordResult>& Record : Result.Result.Results)
+	// 		{
+	// 			CHECK(Record);
+	// 			const CacheKey& ExpectedKey = Keys[Index++];
+	// 			CHECK(Record->Key == ExpectedKey);
+	// 			for (const cacherequests::GetCacheRecordResultValue& Value : Record->Values)
+	// 			{
+	// 				CHECK(Value.RawSize == ExpectedPayloadSize);
+	// 			}
+	// 		}
+	// 	};
+
+	// 	// Check that the records are present and not overwritten in the local server
+	// 	CheckRecordCorrectness(LocalCfg, PayloadSize);
+
+	// 	// Check that the records are present and are the newer size in the upstream server
+	// 	CheckRecordCorrectness(UpstreamCfg, PayloadSize*2);
+	// }
+
 	SUBCASE("RpcAcceptOptions")
 	{
 		using namespace utils;
@@ -3817,6 +4163,478 @@ TEST_CASE("workspaces.share")
 	CHECK(Client.Get(fmt::format("/ws/{}", WorkspaceId)).StatusCode == HttpResponseCode::NotFound);
 }
 
+TEST_CASE("buildstore.blobs")
+{
+	std::filesystem::path SystemRootPath = TestEnv.CreateNewTestDir();
+	auto				  _				 = MakeGuard([&SystemRootPath]() { DeleteDirectories(SystemRootPath); });
+
+	std::string_view Namespace = "ns"sv;
+	std::string_view Bucket	   = "bkt"sv;
+	Oid				 BuildId   = Oid::NewOid();
+
+	std::vector<IoHash> CompressedBlobsHashes;
+	{
+		ZenServerInstance Instance(TestEnv);
+
+		const uint16_t PortNumber =
+			Instance.SpawnServerAndWaitUntilReady(fmt::format("--buildstore-enabled --system-dir {}", SystemRootPath));
+		CHECK(PortNumber != 0);
+
+		HttpClient Client(Instance.GetBaseUri() + "/builds/");
+
+		for (size_t I = 0; I < 5; I++)
+		{
+			IoBuffer		 Blob			= CreateSemiRandomBlob(4711 + I * 7);
+			CompressedBuffer CompressedBlob = CompressedBuffer::Compress(SharedBuffer(std::move(Blob)));
+			CompressedBlobsHashes.push_back(CompressedBlob.DecodeRawHash());
+			IoBuffer Payload = std::move(CompressedBlob).GetCompressed().Flatten().AsIoBuffer();
+			Payload.SetContentType(ZenContentType::kCompressedBinary);
+
+			HttpClient::Response Result =
+				Client.Put(fmt::format("{}/{}/{}/blobs/{}", Namespace, Bucket, BuildId, CompressedBlobsHashes.back()), Payload);
+			CHECK(Result);
+		}
+
+		for (const IoHash& RawHash : CompressedBlobsHashes)
+		{
+			HttpClient::Response Result = Client.Get(fmt::format("{}/{}/{}/blobs/{}", Namespace, Bucket, BuildId, RawHash),
+													 HttpClient::Accept(ZenContentType::kCompressedBinary));
+			CHECK(Result);
+			IoBuffer Payload = Result.ResponsePayload;
+			CHECK(Payload.GetContentType() == ZenContentType::kCompressedBinary);
+			IoHash			 VerifyRawHash;
+			uint64_t		 VerifyRawSize;
+			CompressedBuffer CompressedBlob =
+				CompressedBuffer::FromCompressed(SharedBuffer(std::move(Payload)), VerifyRawHash, VerifyRawSize);
+			CHECK(CompressedBlob);
+			CHECK(VerifyRawHash == RawHash);
+			IoBuffer Decompressed = CompressedBlob.Decompress().AsIoBuffer();
+			CHECK(IoHash::HashBuffer(Decompressed) == RawHash);
+		}
+	}
+	{
+		ZenServerInstance Instance(TestEnv);
+
+		const uint16_t PortNumber =
+			Instance.SpawnServerAndWaitUntilReady(fmt::format("--buildstore-enabled --system-dir {}", SystemRootPath));
+		CHECK(PortNumber != 0);
+
+		HttpClient Client(Instance.GetBaseUri() + "/builds/");
+
+		for (const IoHash& RawHash : CompressedBlobsHashes)
+		{
+			HttpClient::Response Result = Client.Get(fmt::format("{}/{}/{}/blobs/{}", Namespace, Bucket, BuildId, RawHash),
+													 HttpClient::Accept(ZenContentType::kCompressedBinary));
+			CHECK(Result);
+			IoBuffer Payload = Result.ResponsePayload;
+			CHECK(Payload.GetContentType() == ZenContentType::kCompressedBinary);
+			IoHash			 VerifyRawHash;
+			uint64_t		 VerifyRawSize;
+			CompressedBuffer CompressedBlob =
+				CompressedBuffer::FromCompressed(SharedBuffer(std::move(Payload)), VerifyRawHash, VerifyRawSize);
+			CHECK(CompressedBlob);
+			CHECK(VerifyRawHash == RawHash);
+			IoBuffer Decompressed = CompressedBlob.Decompress().AsIoBuffer();
+			CHECK(IoHash::HashBuffer(Decompressed) == RawHash);
+		}
+
+		for (size_t I = 0; I < 5; I++)
+		{
+			IoBuffer		 Blob			= CreateSemiRandomBlob(5713 + I * 7);
+			CompressedBuffer CompressedBlob = CompressedBuffer::Compress(SharedBuffer(std::move(Blob)));
+			CompressedBlobsHashes.push_back(CompressedBlob.DecodeRawHash());
+			IoBuffer Payload = std::move(CompressedBlob).GetCompressed().Flatten().AsIoBuffer();
+			Payload.SetContentType(ZenContentType::kCompressedBinary);
+
+			HttpClient::Response Result =
+				Client.Put(fmt::format("{}/{}/{}/blobs/{}", Namespace, Bucket, BuildId, CompressedBlobsHashes.back()), Payload);
+			CHECK(Result);
+		}
+	}
+	{
+		ZenServerInstance Instance(TestEnv);
+
+		const uint16_t PortNumber =
+			Instance.SpawnServerAndWaitUntilReady(fmt::format("--buildstore-enabled --system-dir {}", SystemRootPath));
+		CHECK(PortNumber != 0);
+
+		HttpClient Client(Instance.GetBaseUri() + "/builds/");
+
+		for (const IoHash& RawHash : CompressedBlobsHashes)
+		{
+			HttpClient::Response Result = Client.Get(fmt::format("{}/{}/{}/blobs/{}", Namespace, Bucket, BuildId, RawHash),
+													 HttpClient::Accept(ZenContentType::kCompressedBinary));
+			CHECK(Result);
+			IoBuffer Payload = Result.ResponsePayload;
+			CHECK(Payload.GetContentType() == ZenContentType::kCompressedBinary);
+			IoHash			 VerifyRawHash;
+			uint64_t		 VerifyRawSize;
+			CompressedBuffer CompressedBlob =
+				CompressedBuffer::FromCompressed(SharedBuffer(std::move(Payload)), VerifyRawHash, VerifyRawSize);
+			CHECK(CompressedBlob);
+			CHECK(VerifyRawHash == RawHash);
+			IoBuffer Decompressed = CompressedBlob.Decompress().AsIoBuffer();
+			CHECK(IoHash::HashBuffer(Decompressed) == RawHash);
+		}
+	}
+}
+
+namespace {
+	CbObject MakeMetadata(const IoHash& BlobHash, const std::vector<std::pair<std::string, std::string>>& KeyValues)
+	{
+		CbObjectWriter Writer;
+		Writer.AddHash("rawHash"sv, BlobHash);
+		Writer.BeginObject("values");
+		{
+			for (const auto& V : KeyValues)
+			{
+				Writer.AddString(V.first, V.second);
+			}
+		}
+		Writer.EndObject();	 // values
+		return Writer.Save();
+	};
+
+}  // namespace
+
+TEST_CASE("buildstore.metadata")
+{
+	std::filesystem::path SystemRootPath = TestEnv.CreateNewTestDir();
+	auto				  _				 = MakeGuard([&SystemRootPath]() { DeleteDirectories(SystemRootPath); });
+
+	std::string_view Namespace = "ns"sv;
+	std::string_view Bucket	   = "bkt"sv;
+	Oid				 BuildId   = Oid::NewOid();
+
+	std::vector<IoHash>	  BlobHashes;
+	std::vector<CbObject> Metadatas;
+	std::vector<IoHash>	  MetadataHashes;
+
+	auto GetMetadatas =
+		[](HttpClient& Client, std::string_view Namespace, std::string_view Bucket, const Oid& BuildId, std::vector<IoHash> BlobHashes) {
+			CbObjectWriter Request;
+
+			Request.BeginArray("blobHashes"sv);
+			for (const IoHash& BlobHash : BlobHashes)
+			{
+				Request.AddHash(BlobHash);
+			}
+			Request.EndArray();
+
+			IoBuffer Payload = Request.Save().GetBuffer().AsIoBuffer();
+			Payload.SetContentType(ZenContentType::kCbObject);
+
+			HttpClient::Response Result = Client.Post(fmt::format("{}/{}/{}/blobs/getBlobMetadata", Namespace, Bucket, BuildId),
+													  Payload,
+													  HttpClient::Accept(ZenContentType::kCbObject));
+			CHECK(Result);
+
+			std::vector<CbObject> ResultMetadatas;
+
+			CbPackage ResponsePackage = ParsePackageMessage(Result.ResponsePayload);
+			CbObject  ResponseObject  = ResponsePackage.GetObject();
+
+			CbArrayView BlobHashArray  = ResponseObject["blobHashes"sv].AsArrayView();
+			CbArrayView MetadatasArray = ResponseObject["metadatas"sv].AsArrayView();
+			ResultMetadatas.reserve(MetadatasArray.Num());
+			auto BlobHashesIt	 = BlobHashes.begin();
+			auto BlobHashArrayIt = begin(BlobHashArray);
+			auto MetadataArrayIt = begin(MetadatasArray);
+			while (MetadataArrayIt != end(MetadatasArray))
+			{
+				const IoHash BlobHash = (*BlobHashArrayIt).AsHash();
+				while (BlobHash != *BlobHashesIt)
+				{
+					ZEN_ASSERT(BlobHashesIt != BlobHashes.end());
+					BlobHashesIt++;
+				}
+
+				ZEN_ASSERT(BlobHash == *BlobHashesIt);
+
+				const IoHash		MetaHash	   = (*MetadataArrayIt).AsAttachment();
+				const CbAttachment* MetaAttachment = ResponsePackage.FindAttachment(MetaHash);
+				ZEN_ASSERT(MetaAttachment);
+
+				CbObject Metadata = MetaAttachment->AsObject();
+				ResultMetadatas.emplace_back(std::move(Metadata));
+
+				BlobHashArrayIt++;
+				MetadataArrayIt++;
+				BlobHashesIt++;
+			}
+			return ResultMetadatas;
+		};
+
+	{
+		ZenServerInstance Instance(TestEnv);
+
+		const uint16_t PortNumber =
+			Instance.SpawnServerAndWaitUntilReady(fmt::format("--buildstore-enabled --system-dir {}", SystemRootPath));
+		CHECK(PortNumber != 0);
+
+		HttpClient Client(Instance.GetBaseUri() + "/builds/");
+
+		const size_t BlobCount = 5;
+
+		for (size_t I = 0; I < BlobCount; I++)
+		{
+			BlobHashes.push_back(IoHash::HashBuffer(&I, sizeof(I)));
+			Metadatas.push_back(MakeMetadata(BlobHashes.back(), {{"index", fmt::format("{}", I)}}));
+			MetadataHashes.push_back(IoHash::HashBuffer(Metadatas.back().GetBuffer().AsIoBuffer()));
+		}
+
+		{
+			CbPackage							   RequestPackage;
+			std::vector<CbAttachment>			   Attachments;
+			tsl::robin_set<IoHash, IoHash::Hasher> AttachmentHashes;
+			Attachments.reserve(BlobCount);
+			AttachmentHashes.reserve(BlobCount);
+			{
+				CbObjectWriter RequestWriter;
+				RequestWriter.BeginArray("blobHashes");
+				for (size_t BlockHashIndex = 0; BlockHashIndex < BlobHashes.size(); BlockHashIndex++)
+				{
+					RequestWriter.AddHash(BlobHashes[BlockHashIndex]);
+				}
+				RequestWriter.EndArray();  // blobHashes
+
+				RequestWriter.BeginArray("metadatas");
+				for (size_t BlockHashIndex = 0; BlockHashIndex < BlobHashes.size(); BlockHashIndex++)
+				{
+					const IoHash ObjectHash = Metadatas[BlockHashIndex].GetHash();
+					RequestWriter.AddBinaryAttachment(ObjectHash);
+					if (!AttachmentHashes.contains(ObjectHash))
+					{
+						Attachments.push_back(CbAttachment(Metadatas[BlockHashIndex], ObjectHash));
+						AttachmentHashes.insert(ObjectHash);
+					}
+				}
+
+				RequestWriter.EndArray();  // metadatas
+
+				RequestPackage.SetObject(RequestWriter.Save());
+			}
+			RequestPackage.AddAttachments(Attachments);
+
+			CompositeBuffer RpcRequestBuffer = FormatPackageMessageBuffer(RequestPackage);
+
+			HttpClient::Response Result = Client.Post(fmt::format("{}/{}/{}/blobs/putBlobMetadata", Namespace, Bucket, BuildId),
+													  RpcRequestBuffer,
+													  ZenContentType::kCbPackage);
+			CHECK(Result);
+		}
+
+		{
+			std::vector<CbObject> ResultMetadatas = GetMetadatas(Client, Namespace, Bucket, BuildId, BlobHashes);
+
+			for (size_t Index = 0; Index < MetadataHashes.size(); Index++)
+			{
+				const IoHash& ExpectedHash = MetadataHashes[Index];
+				IoHash		  Hash		   = IoHash::HashBuffer(ResultMetadatas[Index].GetBuffer().AsIoBuffer());
+				CHECK_EQ(ExpectedHash, Hash);
+			}
+		}
+	}
+	{
+		ZenServerInstance Instance(TestEnv);
+
+		const uint16_t PortNumber =
+			Instance.SpawnServerAndWaitUntilReady(fmt::format("--buildstore-enabled --system-dir {}", SystemRootPath));
+		CHECK(PortNumber != 0);
+
+		HttpClient Client(Instance.GetBaseUri() + "/builds/");
+
+		std::vector<CbObject> ResultMetadatas = GetMetadatas(Client, Namespace, Bucket, BuildId, BlobHashes);
+
+		for (size_t Index = 0; Index < MetadataHashes.size(); Index++)
+		{
+			const IoHash& ExpectedHash = MetadataHashes[Index];
+			IoHash		  Hash		   = IoHash::HashBuffer(ResultMetadatas[Index].GetBuffer().AsIoBuffer());
+			CHECK_EQ(ExpectedHash, Hash);
+		}
+	}
+}
+
+TEST_CASE("buildstore.cache")
+{
+	std::filesystem::path SystemRootPath = TestEnv.CreateNewTestDir();
+	std::filesystem::path TempDir		 = TestEnv.CreateNewTestDir();
+	auto				  _				 = MakeGuard([&SystemRootPath, &TempDir]() {
+		  DeleteDirectories(SystemRootPath);
+		  DeleteDirectories(TempDir);
+	  });
+
+	std::string_view Namespace = "ns"sv;
+	std::string_view Bucket	   = "bkt"sv;
+	Oid				 BuildId   = Oid::NewOid();
+
+	std::vector<IoHash>	  BlobHashes;
+	std::vector<CbObject> Metadatas;
+	std::vector<IoHash>	  MetadataHashes;
+
+	const size_t BlobCount = 5;
+	{
+		ZenServerInstance Instance(TestEnv);
+
+		const uint16_t PortNumber =
+			Instance.SpawnServerAndWaitUntilReady(fmt::format("--buildstore-enabled --system-dir {}", SystemRootPath));
+		CHECK(PortNumber != 0);
+
+		HttpClient Client(Instance.GetBaseUri());
+
+		BuildStorageCache::Statistics	   Stats;
+		std::unique_ptr<BuildStorageCache> Cache(CreateZenBuildStorageCache(Client, Stats, Namespace, Bucket, TempDir, false));
+
+		{
+			IoHash											 NoneBlob	= IoHash::HashBuffer("data", 4);
+			std::vector<BuildStorageCache::BlobExistsResult> NoneExists = Cache->BlobsExists(BuildId, std::vector<IoHash>{NoneBlob});
+			CHECK(NoneExists.size() == 1);
+			CHECK(!NoneExists[0].HasBody);
+			CHECK(!NoneExists[0].HasMetadata);
+		}
+
+		for (size_t I = 0; I < BlobCount; I++)
+		{
+			IoBuffer		 Blob			= CreateSemiRandomBlob(4711 + I * 7);
+			CompressedBuffer CompressedBlob = CompressedBuffer::Compress(SharedBuffer(std::move(Blob)));
+			BlobHashes.push_back(CompressedBlob.DecodeRawHash());
+			Cache->PutBuildBlob(BuildId, BlobHashes.back(), ZenContentType::kCompressedBinary, CompressedBlob.GetCompressed());
+		}
+
+		Cache->Flush(500);
+		Cache = CreateZenBuildStorageCache(Client, Stats, Namespace, Bucket, TempDir, false);
+
+		{
+			std::vector<BuildStorageCache::BlobExistsResult> Exists = Cache->BlobsExists(BuildId, BlobHashes);
+			CHECK(Exists.size() == BlobHashes.size());
+			for (size_t I = 0; I < BlobCount; I++)
+			{
+				CHECK(Exists[I].HasBody);
+				CHECK(!Exists[I].HasMetadata);
+			}
+
+			std::vector<CbObject> FetchedMetadatas = Cache->GetBlobMetadatas(BuildId, BlobHashes);
+			CHECK_EQ(0, FetchedMetadatas.size());
+		}
+
+		{
+			for (size_t I = 0; I < BlobCount; I++)
+			{
+				IoBuffer BuildBlob = Cache->GetBuildBlob(BuildId, BlobHashes[I]);
+				CHECK(BuildBlob);
+				CHECK_EQ(BlobHashes[I],
+						 IoHash::HashBuffer(CompressedBuffer::FromCompressedNoValidate(std::move(BuildBlob)).Decompress().AsIoBuffer()));
+			}
+		}
+
+		{
+			for (size_t I = 0; I < BlobCount; I++)
+			{
+				CbObject Metadata = MakeMetadata(BlobHashes[I],
+												 {{"key", fmt::format("{}", I)},
+												  {"key_plus_one", fmt::format("{}", I + 1)},
+												  {"block_hash", fmt::format("{}", BlobHashes[I])}});
+				Metadatas.push_back(Metadata);
+				MetadataHashes.push_back(IoHash::HashBuffer(Metadata.GetBuffer().AsIoBuffer()));
+			}
+			Cache->PutBlobMetadatas(BuildId, BlobHashes, Metadatas);
+		}
+
+		Cache->Flush(500);
+		Cache = CreateZenBuildStorageCache(Client, Stats, Namespace, Bucket, TempDir, false);
+
+		{
+			std::vector<BuildStorageCache::BlobExistsResult> Exists = Cache->BlobsExists(BuildId, BlobHashes);
+			CHECK(Exists.size() == BlobHashes.size());
+			for (size_t I = 0; I < BlobCount; I++)
+			{
+				CHECK(Exists[I].HasBody);
+				CHECK(Exists[I].HasMetadata);
+			}
+
+			std::vector<CbObject> FetchedMetadatas = Cache->GetBlobMetadatas(BuildId, BlobHashes);
+			CHECK_EQ(BlobCount, FetchedMetadatas.size());
+
+			for (size_t I = 0; I < BlobCount; I++)
+			{
+				CHECK_EQ(MetadataHashes[I], IoHash::HashBuffer(FetchedMetadatas[I].GetBuffer().AsIoBuffer()));
+			}
+		}
+
+		for (size_t I = 0; I < BlobCount; I++)
+		{
+			IoBuffer		 Blob			= CreateSemiRandomBlob(4711 + I * 7);
+			CompressedBuffer CompressedBlob = CompressedBuffer::Compress(SharedBuffer(std::move(Blob)));
+			BlobHashes.push_back(CompressedBlob.DecodeRawHash());
+			Cache->PutBuildBlob(BuildId, BlobHashes.back(), ZenContentType::kCompressedBinary, CompressedBlob.GetCompressed());
+		}
+
+		Cache->Flush(500);
+		Cache = CreateZenBuildStorageCache(Client, Stats, Namespace, Bucket, TempDir, false);
+
+		{
+			std::vector<BuildStorageCache::BlobExistsResult> Exists = Cache->BlobsExists(BuildId, BlobHashes);
+			CHECK(Exists.size() == BlobHashes.size());
+			for (size_t I = 0; I < BlobCount * 2; I++)
+			{
+				CHECK(Exists[I].HasBody);
+				CHECK_EQ(I < BlobCount, Exists[I].HasMetadata);
+			}
+
+			std::vector<CbObject> MetaDatas = Cache->GetBlobMetadatas(BuildId, BlobHashes);
+			CHECK_EQ(BlobCount, MetaDatas.size());
+
+			std::vector<CbObject> FetchedMetadatas = Cache->GetBlobMetadatas(BuildId, BlobHashes);
+			CHECK_EQ(BlobCount, FetchedMetadatas.size());
+
+			for (size_t I = 0; I < BlobCount; I++)
+			{
+				CHECK_EQ(MetadataHashes[I], IoHash::HashBuffer(FetchedMetadatas[I].GetBuffer().AsIoBuffer()));
+			}
+		}
+	}
+
+	{
+		ZenServerInstance Instance(TestEnv);
+
+		const uint16_t PortNumber =
+			Instance.SpawnServerAndWaitUntilReady(fmt::format("--buildstore-enabled --system-dir {}", SystemRootPath));
+		CHECK(PortNumber != 0);
+
+		HttpClient Client(Instance.GetBaseUri());
+
+		BuildStorageCache::Statistics	   Stats;
+		std::unique_ptr<BuildStorageCache> Cache(CreateZenBuildStorageCache(Client, Stats, Namespace, Bucket, TempDir, false));
+
+		std::vector<BuildStorageCache::BlobExistsResult> Exists = Cache->BlobsExists(BuildId, BlobHashes);
+		CHECK(Exists.size() == BlobHashes.size());
+		for (size_t I = 0; I < BlobCount * 2; I++)
+		{
+			CHECK(Exists[I].HasBody);
+			CHECK_EQ(I < BlobCount, Exists[I].HasMetadata);
+		}
+
+		for (size_t I = 0; I < BlobCount * 2; I++)
+		{
+			IoBuffer BuildBlob = Cache->GetBuildBlob(BuildId, BlobHashes[I]);
+			CHECK(BuildBlob);
+			CHECK_EQ(BlobHashes[I],
+					 IoHash::HashBuffer(CompressedBuffer::FromCompressedNoValidate(std::move(BuildBlob)).Decompress().AsIoBuffer()));
+		}
+
+		std::vector<CbObject> MetaDatas = Cache->GetBlobMetadatas(BuildId, BlobHashes);
+		CHECK_EQ(BlobCount, MetaDatas.size());
+
+		std::vector<CbObject> FetchedMetadatas = Cache->GetBlobMetadatas(BuildId, BlobHashes);
+		CHECK_EQ(BlobCount, FetchedMetadatas.size());
+
+		for (size_t I = 0; I < BlobCount; I++)
+		{
+			CHECK_EQ(MetadataHashes[I], IoHash::HashBuffer(FetchedMetadatas[I].GetBuffer().AsIoBuffer()));
+		}
+	}
+}
+
 #	if 0
 TEST_CASE("lifetime.owner")
 {