send payloads as duplicated handles (#240)

* send payloads as duplicated handles if requestor provides process id and allows local file references.
* linux/macos fixes
* tests
* fix access rights when duplicating handle
* fix closing of duplicated handles on error
* cleanup
* changelog

1 files changed, 157 insertions, 6 deletions

diff --git a/zenserver-test/zenserver-test.cpp b/zenserver-test/zenserver-test.cpp
index c1d2e8838..079f70984 100644
--- a/zenserver-test/zenserver-test.cpp
+++ b/zenserver-test/zenserver-test.cpp
@@ -1334,11 +1334,16 @@ TEST_CASE("zcache.rpc")
 								CachePolicy							   RecordPolicy) {
 		std::vector<uint8_t> Data;
 		Data.resize(PayloadSize);
-		for (size_t Idx = 0; Idx < PayloadSize; ++Idx)
+		uint32_t  DataSeed = *reinterpret_cast<const uint32_t*>(&CacheKey.Hash.Hash[0]);
+		uint16_t* DataPtr  = reinterpret_cast<uint16_t*>(Data.data());
+		for (size_t Idx = 0; Idx < PayloadSize / 2; ++Idx)
 		{
-			Data[Idx] = Idx % 255;
+			DataPtr[Idx] = static_cast<uint16_t>((Idx + DataSeed) % 0xffffu);
+		}
+		if (PayloadSize & 1)
+		{
+			Data[PayloadSize - 1] = static_cast<uint8_t>((PayloadSize - 1) & 0xff);
 		}
-
 		CompressedBuffer Value = zen::CompressedBuffer::Compress(SharedBuffer::MakeView(Data.data(), Data.size()));
 		Request.Requests.push_back({.Key = CacheKey, .Values = {{.Id = Oid::NewOid(), .Body = std::move(Value)}}, .Policy = RecordPolicy});
 	};
@@ -1347,13 +1352,14 @@ TEST_CASE("zcache.rpc")
 												std::string_view Namespace,
 												std::string_view Bucket,
 												size_t			 Num,
-												size_t			 PayloadSize = 1024) -> std::vector<CacheKey> {
+												size_t			 PayloadSize = 1024,
+												size_t			 KeyOffset	 = 1) -> std::vector<CacheKey> {
 		std::vector<zen::CacheKey> OutKeys;
 
 		for (uint32_t Key = 1; Key <= Num; ++Key)
 		{
 			zen::IoHash KeyHash;
-			((uint32_t*)(KeyHash.Hash))[0] = Key;
+			((uint32_t*)(KeyHash.Hash))[0] = gsl::narrow<uint32_t>(KeyOffset + Key);
 			const zen::CacheKey CacheKey   = zen::CacheKey::Create(Bucket, KeyHash);
 
 			cacherequests::PutCacheRecordsRequest Request = {.AcceptMagic = kCbPkgMagic, .Namespace = std::string(Namespace)};
@@ -1384,8 +1390,12 @@ TEST_CASE("zcache.rpc")
 	auto GetCacheRecords = [](std::string_view		   BaseUri,
 							  std::string_view		   Namespace,
 							  std::span<zen::CacheKey> Keys,
-							  zen::CachePolicy		   Policy) -> GetCacheRecordResult {
+							  zen::CachePolicy		   Policy,
+							  zen::RpcAcceptOptions	   AcceptOptions = zen::RpcAcceptOptions::kNone,
+							  int					   Pid			 = 0) -> GetCacheRecordResult {
 		cacherequests::GetCacheRecordsRequest Request = {.AcceptMagic	= kCbPkgMagic,
+														 .AcceptOptions = static_cast<uint16_t>(AcceptOptions),
+														 .ProcessPid	= Pid,
 														 .DefaultPolicy = Policy,
 														 .Namespace		= std::string(Namespace)};
 		for (const CacheKey& Key : Keys)
@@ -1547,6 +1557,147 @@ TEST_CASE("zcache.rpc")
 			CHECK(Record->Key == ExpectedKey);
 		}
 	}
+
+	SUBCASE("RpcAcceptOptions")
+	{
+		using namespace utils;
+
+		std::filesystem::path TestDir	 = TestEnv.CreateNewTestDir();
+		const uint16_t		  PortNumber = 13337;
+		const auto			  BaseUri	 = fmt::format("http://localhost:{}/z$", PortNumber);
+
+		ZenServerInstance Inst(TestEnv);
+		Inst.SetTestDir(TestDir);
+		Inst.SpawnServer(PortNumber);
+		Inst.WaitUntilReady();
+
+		std::vector<zen::CacheKey> SmallKeys = PutCacheRecords(BaseUri, "ue4.ddc"sv, "mastodon"sv, 4, 1024);
+		std::vector<zen::CacheKey> LargeKeys = PutCacheRecords(BaseUri, "ue4.ddc"sv, "mastodon"sv, 4, 1024 * 1024 * 16, SmallKeys.size());
+
+		std::vector<zen::CacheKey> Keys(SmallKeys.begin(), SmallKeys.end());
+		Keys.insert(Keys.end(), LargeKeys.begin(), LargeKeys.end());
+
+		{
+			GetCacheRecordResult Result = GetCacheRecords(BaseUri, "ue4.ddc"sv, Keys, CachePolicy::Default);
+
+			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)
+				{
+					const IoBuffer&		  Body = Value.Body.GetCompressed().Flatten().AsIoBuffer();
+					IoBufferFileReference Ref;
+					bool				  IsFileRef = Body.GetFileReference(Ref);
+					CHECK(!IsFileRef);
+				}
+			}
+		}
+
+		// File path, but only for large files
+		{
+			GetCacheRecordResult Result =
+				GetCacheRecords(BaseUri, "ue4.ddc"sv, Keys, CachePolicy::Default, RpcAcceptOptions::kAllowLocalReferences);
+
+			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)
+				{
+					const IoBuffer&		  Body = Value.Body.GetCompressed().Flatten().AsIoBuffer();
+					IoBufferFileReference Ref;
+					bool				  IsFileRef = Body.GetFileReference(Ref);
+					CHECK(IsFileRef == (Body.Size() > 1024));
+				}
+			}
+		}
+
+		// File path, for all files
+		{
+			GetCacheRecordResult Result =
+				GetCacheRecords(BaseUri,
+								"ue4.ddc"sv,
+								Keys,
+								CachePolicy::Default,
+								RpcAcceptOptions::kAllowLocalReferences | RpcAcceptOptions::kAllowPartialLocalReferences);
+
+			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)
+				{
+					const IoBuffer&		  Body = Value.Body.GetCompressed().Flatten().AsIoBuffer();
+					IoBufferFileReference Ref;
+					bool				  IsFileRef = Body.GetFileReference(Ref);
+					CHECK(IsFileRef);
+				}
+			}
+		}
+
+		// File handle, but only for large files
+		{
+			GetCacheRecordResult Result = GetCacheRecords(BaseUri,
+														  "ue4.ddc"sv,
+														  Keys,
+														  CachePolicy::Default,
+														  RpcAcceptOptions::kAllowLocalReferences,
+														  GetCurrentProcessId());
+
+			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)
+				{
+					const IoBuffer&		  Body = Value.Body.GetCompressed().Flatten().AsIoBuffer();
+					IoBufferFileReference Ref;
+					bool				  IsFileRef = Body.GetFileReference(Ref);
+					CHECK(IsFileRef == (Body.Size() > 1024));
+				}
+			}
+		}
+
+		// File handle, for all files
+		{
+			GetCacheRecordResult Result =
+				GetCacheRecords(BaseUri,
+								"ue4.ddc"sv,
+								Keys,
+								CachePolicy::Default,
+								RpcAcceptOptions::kAllowLocalReferences | RpcAcceptOptions::kAllowPartialLocalReferences,
+								GetCurrentProcessId());
+
+			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)
+				{
+					const IoBuffer&		  Body = Value.Body.GetCompressed().Flatten().AsIoBuffer();
+					IoBufferFileReference Ref;
+					bool				  IsFileRef = Body.GetFileReference(Ref);
+					CHECK(IsFileRef);
+				}
+			}
+		}
+	}
 }
 
 TEST_CASE("zcache.failing.upstream")