merge from main

1 files changed, 359 insertions, 43 deletions

diff --git a/zenserver-test/zenserver-test.cpp b/zenserver-test/zenserver-test.cpp
index 48600bb36..b3a4348f0 100644
--- a/zenserver-test/zenserver-test.cpp
+++ b/zenserver-test/zenserver-test.cpp
@@ -20,6 +20,7 @@
 #include <zenhttp/httpclient.h>
 #include <zenhttp/httpshared.h>
 #include <zenhttp/zenhttp.h>
+#include <zenutil/cache/cache.h>
 #include <zenutil/zenserverprocess.h>
 
 #if ZEN_USE_MIMALLOC
@@ -466,7 +467,9 @@ using namespace std::literals;
 class full_test_formatter final : public spdlog::formatter
 {
 public:
-	full_test_formatter(std::string_view LogId, std::chrono::time_point<std::chrono::system_clock> Epoch) : m_Epoch(Epoch), m_LogId(LogId) {}
+	full_test_formatter(std::string_view LogId, std::chrono::time_point<std::chrono::system_clock> Epoch) : m_Epoch(Epoch), m_LogId(LogId)
+	{
+	}
 
 	virtual std::unique_ptr<formatter> clone() const override { return std::make_unique<full_test_formatter>(m_LogId, m_Epoch); }
 
@@ -991,7 +994,7 @@ TEST_CASE("project.basic")
 			}
 		}
 
-		BaseUri << "/oplog/ps5";
+		BaseUri << "/oplog/foobar";
 
 		{
 			{
@@ -1007,7 +1010,7 @@ TEST_CASE("project.basic")
 
 				zen::CbObjectView ResponseObject = zen::CbFieldView(Response.text.data()).AsObjectView();
 
-				CHECK(ResponseObject["id"].AsString() == "ps5"sv);
+				CHECK(ResponseObject["id"].AsString() == "foobar"sv);
 				CHECK(ResponseObject["project"].AsString() == "test"sv);
 			}
 		}
@@ -1029,11 +1032,13 @@ TEST_CASE("project.basic")
 				"00010000"};
 			auto FileOid = zen::Oid::FromHexString(ChunkId);
 
+			std::filesystem::path ReliablePath = zen::GetRunningExecutablePath();
+
 			OpWriter.BeginArray("files");
 			OpWriter.BeginObject();
 			OpWriter << "id" << FileOid;
-			OpWriter << "clientpath" << __FILE__;
-			OpWriter << "serverpath" << __FILE__;
+			OpWriter << "clientpath" << ReliablePath.c_str();
+			OpWriter << "serverpath" << ReliablePath.c_str();
 			OpWriter.EndObject();
 			OpWriter.EndArray();
 
@@ -1108,6 +1113,45 @@ TEST_CASE("project.pipe")
 }
 #	endif
 
+namespace utils {
+
+	struct ZenConfig
+	{
+		std::filesystem::path DataDir;
+		uint16_t			  Port;
+		std::string			  BaseUri;
+		std::string			  Args;
+
+		static ZenConfig New(uint16_t Port = 13337, std::string Args = "")
+		{
+			return ZenConfig{.DataDir = TestEnv.CreateNewTestDir(),
+							 .Port	  = Port,
+							 .BaseUri = "http://localhost:{}/z$"_format(Port),
+							 .Args	  = std::move(Args)};
+		}
+
+		static ZenConfig NewWithUpstream(uint16_t UpstreamPort)
+		{
+			return New(13337, "--debug --upstream-thread-count=0 --upstream-zen-url=http://localhost:{}"_format(UpstreamPort));
+		}
+
+		void Spawn(ZenServerInstance& Inst)
+		{
+			Inst.SetTestDir(DataDir);
+			Inst.SpawnServer(Port, Args);
+			Inst.WaitUntilReady();
+		}
+	};
+
+	void SpawnServer(ZenServerInstance& Server, ZenConfig& Cfg)
+	{
+		Server.SetTestDir(Cfg.DataDir);
+		Server.SpawnServer(Cfg.Port, Cfg.Args);
+		Server.WaitUntilReady();
+	}
+
+}  // namespace utils
+
 TEST_CASE("zcache.basic")
 {
 	using namespace std::literals;
@@ -1414,34 +1458,7 @@ TEST_CASE("zcache.cbpackage")
 TEST_CASE("zcache.policy")
 {
 	using namespace std::literals;
-
-	struct ZenConfig
-	{
-		std::filesystem::path DataDir;
-		uint16_t			  Port;
-		std::string			  BaseUri;
-		std::string			  Args;
-
-		static ZenConfig New(uint16_t Port = 13337, std::string Args = "")
-		{
-			return ZenConfig{.DataDir = TestEnv.CreateNewTestDir(),
-							 .Port	  = Port,
-							 .BaseUri = "http://localhost:{}/z$"_format(Port),
-							 .Args	  = std::move(Args)};
-		}
-
-		static ZenConfig NewWithUpstream(uint16_t UpstreamPort)
-		{
-			return New(13337, "--upstream-thread-count=0 --upstream-zen-url=http://localhost:{}"_format(UpstreamPort));
-		}
-
-		void Spawn(ZenServerInstance& Inst)
-		{
-			Inst.SetTestDir(DataDir);
-			Inst.SpawnServer(Port, Args);
-			Inst.WaitUntilReady();
-		}
-	};
+	using namespace utils;
 
 	auto GenerateData = [](uint64_t Size, zen::IoHash& OutHash) -> zen::UniqueBuffer {
 		auto	 Buf  = zen::UniqueBuffer::Alloc(Size);
@@ -1705,14 +1722,15 @@ TEST_CASE("zcache.policy")
 
 		LocalCfg.Spawn(LocalInst);
 
-		zen::IoHash	   Key;
-		zen::IoHash	   PayloadId;
-		zen::CbPackage OriginalPackage = GeneratePackage(Key, PayloadId);
-		auto		   Buf			   = ToBuffer(OriginalPackage);
+		zen::IoHash Key;
+		zen::IoHash PayloadId;
 
 		// Store package locally
 		{
-			CHECK(OriginalPackage.GetAttachments().size() != 0);
+			zen::CbPackage Package = GeneratePackage(Key, PayloadId);
+			auto		   Buf	   = ToBuffer(Package);
+
+			CHECK(Package.GetAttachments().size() != 0);
 			cpr::Response Result = cpr::Put(cpr::Url{"{}/{}/{}"_format(LocalCfg.BaseUri, Bucket, Key)},
 											cpr::Body{(const char*)Buf.GetData(), Buf.GetSize()},
 											cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}});
@@ -1765,14 +1783,15 @@ TEST_CASE("zcache.policy")
 		UpstreamCfg.Spawn(UpstreamInst);
 		LocalCfg.Spawn(LocalInst);
 
-		zen::IoHash	   Key;
-		zen::IoHash	   PayloadId;
-		zen::CbPackage OriginalPackage = GeneratePackage(Key, PayloadId);
-		auto		   Buf			   = ToBuffer(OriginalPackage);
+		zen::IoHash Key;
+		zen::IoHash PayloadId;
 
 		// Store package upstream
 		{
-			CHECK(OriginalPackage.GetAttachments().size() != 0);
+			zen::CbPackage Package = GeneratePackage(Key, PayloadId);
+			auto		   Buf	   = ToBuffer(Package);
+
+			CHECK(Package.GetAttachments().size() != 0);
 			cpr::Response Result = cpr::Put(cpr::Url{"{}/{}/{}"_format(UpstreamCfg.BaseUri, Bucket, Key)},
 											cpr::Body{(const char*)Buf.GetData(), Buf.GetSize()},
 											cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}});
@@ -1888,6 +1907,303 @@ TEST_CASE("zcache.policy")
 	}
 }
 
+TEST_CASE("zcache.rpc")
+{
+	using namespace std::literals;
+
+	auto CreateCacheRecord = [](const zen::CacheKey& CacheKey, size_t PayloadSize) -> zen::CbPackage {
+		std::vector<uint8_t> Data;
+		Data.resize(PayloadSize);
+		for (size_t Idx = 0; Idx < PayloadSize; ++Idx)
+		{
+			Data[Idx] = Idx % 255;
+		}
+
+		zen::CbAttachment Attachment(zen::CompressedBuffer::Compress(SharedBuffer::MakeView(Data.data(), Data.size())));
+
+		zen::CbObjectWriter CacheRecord;
+		CacheRecord.BeginObject("CacheKey"sv);
+		CacheRecord << "Bucket"sv << CacheKey.Bucket << "Hash"sv << CacheKey.Hash;
+		CacheRecord.EndObject();
+		CacheRecord << "Data"sv << Attachment;
+
+		zen::CbPackage Package;
+		Package.SetObject(CacheRecord.Save());
+		Package.AddAttachment(Attachment);
+
+		return Package;
+	};
+
+	auto ToIoBuffer = [](zen::CbPackage Package) -> zen::IoBuffer {
+		zen::BinaryWriter MemStream;
+		Package.Save(MemStream);
+		return zen::IoBuffer(zen::IoBuffer::Clone, MemStream.Data(), MemStream.Size());
+	};
+
+	auto PutCacheRecords = [&CreateCacheRecord, &ToIoBuffer](std::string_view BaseUri,
+															 std::string_view Query,
+															 std::string_view Bucket,
+															 size_t			  Num,
+															 size_t			  PayloadSize = 1024) -> std::vector<CacheKey> {
+		std::vector<zen::CacheKey> OutKeys;
+
+		for (uint32_t Key = 1; Key <= Num; ++Key)
+		{
+			const zen::CacheKey CacheKey	= zen::CacheKey::Create(Bucket, zen::IoHash::HashBuffer(&Key, sizeof uint32_t));
+			CbPackage			CacheRecord = CreateCacheRecord(CacheKey, PayloadSize);
+
+			OutKeys.push_back(CacheKey);
+
+			IoBuffer Payload = ToIoBuffer(CacheRecord);
+
+			cpr::Response Result = cpr::Put(cpr::Url{"{}/{}/{}{}"_format(BaseUri, CacheKey.Bucket, CacheKey.Hash, Query)},
+											cpr::Body{(const char*)Payload.Data(), Payload.Size()},
+											cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}});
+
+			CHECK(Result.status_code == 201);
+		}
+
+		return OutKeys;
+	};
+
+	struct GetCacheRecordResult
+	{
+		zen::CbPackage				  Response;
+		std::vector<zen::CbFieldView> Records;
+		bool						  Success;
+	};
+
+	auto GetCacheRecords =
+		[](std::string_view BaseUri, std::span<zen::CacheKey> Keys, const zen::CacheRecordPolicy& Policy) -> GetCacheRecordResult {
+		using namespace zen;
+
+		CbObjectWriter Request;
+		Request << "Method"sv
+				<< "GetCacheRecords"sv;
+		Request.BeginObject("Params"sv);
+
+		Request.BeginArray("CacheKeys"sv);
+		for (const CacheKey& Key : Keys)
+		{
+			Request.BeginObject();
+			Request << "Bucket"sv << Key.Bucket << "Hash"sv << Key.Hash;
+			Request.EndObject();
+		}
+		Request.EndArray();
+
+		Request.BeginObject("Policy");
+		CacheRecordPolicy::Save(Policy, Request);
+		Request.EndObject();
+
+		Request.EndObject();
+
+		BinaryWriter Body;
+		Request.Save(Body);
+
+		cpr::Response Result = cpr::Post(cpr::Url{"{}/$rpc"_format(BaseUri)},
+										 cpr::Header{{"Content-Type", "application/x-ue-cb"}, {"Accept", "application/x-ue-cbpkg"}},
+										 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});
+
+		GetCacheRecordResult OutResult;
+
+		if (Result.status_code == 200)
+		{
+			CbPackage Response;
+			if (Response.TryLoad(zen::IoBuffer(zen::IoBuffer::Wrap, Result.text.data(), Result.text.size())))
+			{
+				OutResult.Response			= std::move(Response);
+				CbObjectView ResponseObject = OutResult.Response.GetObject();
+
+				for (CbFieldView RecordView : ResponseObject["Result"])
+				{
+					ExtendableStringBuilder<256> Tmp;
+					auto						 JSON = RecordView.AsObjectView().ToJson(Tmp).ToView();
+					OutResult.Records.push_back(RecordView);
+				}
+
+				OutResult.Success = true;
+			}
+		}
+
+		return OutResult;
+	};
+
+	auto LoadKey = [](zen::CbFieldView KeyView) -> zen::CacheKey {
+		if (zen::CbObjectView KeyObj = KeyView.AsObjectView())
+		{
+			return CacheKey::Create(KeyObj["Bucket"sv].AsString(), KeyObj["Hash"].AsHash());
+		}
+		return CacheKey::Empty;
+	};
+
+	SUBCASE("get cache records")
+	{
+		std::filesystem::path TestDir	 = TestEnv.CreateNewTestDir();
+		const uint16_t		  PortNumber = 13337;
+		const auto			  BaseUri	 = "http://localhost:{}/z$"_format(PortNumber);
+
+		ZenServerInstance Inst(TestEnv);
+		Inst.SetTestDir(TestDir);
+		Inst.SpawnServer(PortNumber);
+		Inst.WaitUntilReady();
+
+		CacheRecordPolicy		   Policy;
+		std::vector<zen::CacheKey> Keys	  = PutCacheRecords(BaseUri, ""sv, "mastodon"sv, 128);
+		GetCacheRecordResult	   Result = GetCacheRecords(BaseUri, Keys, Policy);
+
+		CHECK(Result.Records.size() == Keys.size());
+
+		for (size_t Index = 0; CbFieldView RecordView : Result.Records)
+		{
+			const CacheKey& ExpectedKey = Keys[Index++];
+
+			CbObjectView		RecordObj	   = RecordView.AsObjectView();
+			CbObjectView		KeyObj		   = RecordObj["CacheKey"sv].AsObjectView();
+			const CacheKey		Key			   = CacheKey::Create(KeyObj["Bucket"sv].AsString(), KeyObj["Hash"].AsHash());
+			const IoHash		AttachmentHash = RecordObj["Data"sv].AsHash();
+			const CbAttachment* Attachment	   = Result.Response.FindAttachment(AttachmentHash);
+
+			CHECK(Key == ExpectedKey);
+			CHECK(Attachment != nullptr);
+		}
+	}
+
+	SUBCASE("get missing cache records")
+	{
+		std::filesystem::path TestDir	 = TestEnv.CreateNewTestDir();
+		const uint16_t		  PortNumber = 13337;
+		const auto			  BaseUri	 = "http://localhost:{}/z$"_format(PortNumber);
+
+		ZenServerInstance Inst(TestEnv);
+		Inst.SetTestDir(TestDir);
+		Inst.SpawnServer(PortNumber);
+		Inst.WaitUntilReady();
+
+		CacheRecordPolicy		   Policy;
+		std::vector<zen::CacheKey> ExistingKeys = PutCacheRecords(BaseUri, ""sv, "mastodon"sv, 128);
+		std::vector<zen::CacheKey> Keys;
+
+		for (const zen::CacheKey& Key : ExistingKeys)
+		{
+			Keys.push_back(Key);
+			Keys.push_back(CacheKey::Create("missing"sv, IoHash::Zero));
+		}
+
+		GetCacheRecordResult Result = GetCacheRecords(BaseUri, Keys, Policy);
+
+		CHECK(Result.Records.size() == Keys.size());
+
+		size_t KeyIndex = 0;
+		for (size_t Index = 0; CbFieldView RecordView : Result.Records)
+		{
+			const bool Missing = Index++ % 2 != 0;
+
+			if (Missing)
+			{
+				CHECK(RecordView.IsNull());
+			}
+			else
+			{
+				const CacheKey&		ExpectedKey	   = ExistingKeys[KeyIndex++];
+				CbObjectView		RecordObj	   = RecordView.AsObjectView();
+				CbObjectView		KeyObj		   = RecordObj["CacheKey"sv].AsObjectView();
+				zen::CacheKey		Key			   = LoadKey(RecordObj["CacheKey"sv]);
+				const IoHash		AttachmentHash = RecordObj["Data"sv].AsHash();
+				const CbAttachment* Attachment	   = Result.Response.FindAttachment(AttachmentHash);
+
+				CHECK(Key == ExpectedKey);
+				CHECK(Attachment != nullptr);
+			}
+		}
+	}
+
+	SUBCASE("policy - 'SkipAttachments' does not return any record attachments")
+	{
+		std::filesystem::path TestDir	 = TestEnv.CreateNewTestDir();
+		const uint16_t		  PortNumber = 13337;
+		const auto			  BaseUri	 = "http://localhost:{}/z$"_format(PortNumber);
+
+		ZenServerInstance Inst(TestEnv);
+		Inst.SetTestDir(TestDir);
+		Inst.SpawnServer(PortNumber);
+		Inst.WaitUntilReady();
+
+		CacheRecordPolicy		   Policy(CachePolicy::QueryLocal | CachePolicy::SkipAttachments);
+		std::vector<zen::CacheKey> Keys	  = PutCacheRecords(BaseUri, ""sv, "mastodon"sv, 4);
+		GetCacheRecordResult	   Result = GetCacheRecords(BaseUri, Keys, Policy);
+
+		CHECK(Result.Records.size() == Keys.size());
+
+		std::span<const zen::CbAttachment> Attachments = Result.Response.GetAttachments();
+		CHECK(Attachments.empty());
+
+		for (size_t Index = 0; CbFieldView RecordView : Result.Records)
+		{
+			const CacheKey& ExpectedKey = Keys[Index++];
+
+			CbObjectView   RecordObj	  = RecordView.AsObjectView();
+			CbObjectView   KeyObj		  = RecordObj["CacheKey"sv].AsObjectView();
+			const CacheKey Key			  = CacheKey::Create(KeyObj["Bucket"sv].AsString(), KeyObj["Hash"].AsHash());
+			const IoHash   AttachmentHash = RecordObj["Data"sv].AsHash();
+
+			CHECK(Key == ExpectedKey);
+		}
+	}
+
+	SUBCASE("policy - 'QueryLocal' does not query upstream")
+	{
+		using namespace utils;
+
+		ZenConfig		  UpstreamCfg = ZenConfig::New(13338);
+		ZenServerInstance UpstreamServer(TestEnv);
+		ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(13338);
+		ZenServerInstance LocalServer(TestEnv);
+
+		SpawnServer(UpstreamServer, UpstreamCfg);
+		SpawnServer(LocalServer, LocalCfg);
+
+		std::vector<zen::CacheKey> Keys = PutCacheRecords(UpstreamCfg.BaseUri, ""sv, "mastodon"sv, 4);
+
+		CacheRecordPolicy	 Policy(CachePolicy::QueryLocal);
+		GetCacheRecordResult Result = GetCacheRecords(LocalCfg.BaseUri, Keys, Policy);
+
+		CHECK(Result.Records.size() == Keys.size());
+
+		for (CbFieldView RecordView : Result.Records)
+		{
+			CHECK(RecordView.IsNull());
+		}
+	}
+
+	SUBCASE("policy - 'QueryRemote' does query upstream")
+	{
+		using namespace utils;
+
+		ZenConfig		  UpstreamCfg = ZenConfig::New(13338);
+		ZenServerInstance UpstreamServer(TestEnv);
+		ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(13338);
+		ZenServerInstance LocalServer(TestEnv);
+
+		SpawnServer(UpstreamServer, UpstreamCfg);
+		SpawnServer(LocalServer, LocalCfg);
+
+		std::vector<zen::CacheKey> Keys = PutCacheRecords(UpstreamCfg.BaseUri, ""sv, "mastodon"sv, 4);
+
+		CacheRecordPolicy	 Policy(CachePolicy::QueryLocal | CachePolicy::QueryRemote);
+		GetCacheRecordResult Result = GetCacheRecords(LocalCfg.BaseUri, Keys, Policy);
+
+		CHECK(Result.Records.size() == Keys.size());
+
+		for (size_t Index = 0; CbFieldView RecordView : Result.Response.GetObject()["Result"sv])
+		{
+			const zen::CacheKey& ExpectedKey = Keys[Index++];
+			CbObjectView		 RecordObj	 = RecordView.AsObjectView();
+			zen::CacheKey		 Key		 = LoadKey(RecordObj["CacheKey"sv]);
+			CHECK(Key == ExpectedKey);
+		}
+	}
+}
+
 #	if ZEN_USE_EXEC
 
 struct RemoteExecutionRequest