Merge branch 'main' into non-elevated-asio

1 files changed, 199 insertions, 141 deletions

diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp
index 091406db3..9d3ed2f94 100644
--- a/zenserver/upstream/upstreamcache.cpp
+++ b/zenserver/upstream/upstreamcache.cpp
@@ -18,6 +18,7 @@
 #include <zenstore/cas.h>
 #include <zenstore/cidstore.h>
 
+#include "cache/structuredcache.h"
 #include "cache/structuredcachestore.h"
 #include "diag/logging.h"
 
@@ -215,21 +216,16 @@ namespace detail {
 			}
 		}
 
-		virtual GetUpstreamCacheResult GetCacheRecords(std::span<CacheKey>		  CacheKeys,
-													   std::span<size_t>		  KeyIndex,
-													   const CacheRecordPolicy&	  Policy,
-													   OnCacheRecordGetComplete&& OnComplete) override
+		virtual GetUpstreamCacheResult GetCacheRecords(std::span<CacheKeyRequest*> Requests, OnCacheRecordGetComplete&& OnComplete) override
 		{
 			ZEN_TRACE_CPU("Upstream::Horde::GetCacheRecords");
 
-			ZEN_UNUSED(Policy);
-
 			CloudCacheSession	   Session(m_Client);
 			GetUpstreamCacheResult Result;
 
-			for (size_t Index : KeyIndex)
+			for (CacheKeyRequest* Request : Requests)
 			{
-				const CacheKey& CacheKey = CacheKeys[Index];
+				const CacheKey& CacheKey = Request->Key;
 				CbPackage		Package;
 				CbObject		Record;
 
@@ -264,20 +260,20 @@ namespace detail {
 					}
 				}
 
-				OnComplete({.Key = CacheKey, .KeyIndex = Index, .Record = Record, .Package = Package});
+				OnComplete({.Request = *Request, .Record = Record, .Package = Package});
 			}
 
 			return Result;
 		}
 
-		virtual GetUpstreamCacheResult GetCachePayload(const CacheKey&, const IoHash& PayloadId) override
+		virtual GetUpstreamCacheResult GetCacheValue(const CacheKey&, const IoHash& ValueContentId) override
 		{
-			ZEN_TRACE_CPU("Upstream::Horde::GetSingleCachePayload");
+			ZEN_TRACE_CPU("Upstream::Horde::GetSingleCacheValue");
 
 			try
 			{
 				CloudCacheSession	   Session(m_Client);
-				const CloudCacheResult Result = Session.GetCompressedBlob(PayloadId);
+				const CloudCacheResult Result = Session.GetCompressedBlob(ValueContentId);
 
 				m_Status.SetFromErrorCode(Result.ErrorCode, Result.Reason);
 
@@ -301,20 +297,20 @@ namespace detail {
 			}
 		}
 
-		virtual GetUpstreamCacheResult GetCachePayloads(std::span<CacheChunkRequest> CacheChunkRequests,
-														std::span<size_t>			 RequestIndex,
-														OnCachePayloadGetComplete&&	 OnComplete) override final
+		virtual GetUpstreamCacheResult GetCacheValues(std::span<CacheChunkRequest*> CacheChunkRequests,
+													  OnCacheValueGetComplete&&		OnComplete) override final
 		{
-			ZEN_TRACE_CPU("Upstream::Horde::GetCachePayloads");
+			ZEN_TRACE_CPU("Upstream::Horde::GetCacheValues");
 
 			CloudCacheSession	   Session(m_Client);
 			GetUpstreamCacheResult Result;
 
-			for (size_t Index : RequestIndex)
+			for (CacheChunkRequest* RequestPtr : CacheChunkRequests)
 			{
-				const CacheChunkRequest& Request = CacheChunkRequests[Index];
-				IoBuffer				 Payload;
+				CacheChunkRequest& Request = *RequestPtr;
+				IoBuffer		   Payload;
 
+				CompressedBuffer Compressed;
 				if (!Result.Error)
 				{
 					const CloudCacheResult BlobResult = Session.GetCompressedBlob(Request.ChunkId);
@@ -323,9 +319,23 @@ namespace detail {
 					AppendResult(BlobResult, Result);
 
 					m_Status.SetFromErrorCode(BlobResult.ErrorCode, BlobResult.Reason);
+					if (Payload && IsCompressedBinary(Payload.GetContentType()))
+					{
+						Compressed = CompressedBuffer::FromCompressed(SharedBuffer(Payload));
+					}
 				}
 
-				OnComplete({.Request = Request, .RequestIndex = Index, .Payload = Payload});
+				if (Compressed)
+				{
+					OnComplete({.Request = Request,
+								.RawHash = IoHash::FromBLAKE3(Compressed.GetRawHash()),
+								.RawSize = Compressed.GetRawSize(),
+								.Value	 = Payload});
+				}
+				else
+				{
+					OnComplete({.Request = Request, .RawHash = IoHash::Zero, .RawSize = 0, .Value = IoBuffer()});
+				}
 			}
 
 			return Result;
@@ -333,11 +343,11 @@ namespace detail {
 
 		virtual PutUpstreamCacheResult PutCacheRecord(const UpstreamCacheRecord& CacheRecord,
 													  IoBuffer					 RecordValue,
-													  std::span<IoBuffer const>	 Payloads) override
+													  std::span<IoBuffer const>	 Values) override
 		{
 			ZEN_TRACE_CPU("Upstream::Horde::PutCacheRecord");
 
-			ZEN_ASSERT(CacheRecord.PayloadIds.size() == Payloads.size());
+			ZEN_ASSERT(CacheRecord.ValueContentIds.size() == Values.size());
 			const int32_t MaxAttempts = 3;
 
 			try
@@ -371,30 +381,31 @@ namespace detail {
 					int64_t TotalBytes			= 0ull;
 					double	TotalElapsedSeconds = 0.0;
 
-					const auto PutBlobs = [&](std::span<IoHash> PayloadIds, std::string& OutReason) -> bool {
-						for (const IoHash& PayloadId : PayloadIds)
+					const auto PutBlobs = [&](std::span<IoHash> ValueContentIds, std::string& OutReason) -> bool {
+						for (const IoHash& ValueContentId : ValueContentIds)
 						{
-							const auto It = std::find(std::begin(CacheRecord.PayloadIds), std::end(CacheRecord.PayloadIds), PayloadId);
+							const auto It =
+								std::find(std::begin(CacheRecord.ValueContentIds), std::end(CacheRecord.ValueContentIds), ValueContentId);
 
-							if (It == std::end(CacheRecord.PayloadIds))
+							if (It == std::end(CacheRecord.ValueContentIds))
 							{
-								OutReason = fmt::format("value '{}' MISSING from local cache", PayloadId);
+								OutReason = fmt::format("value '{}' MISSING from local cache", ValueContentId);
 								return false;
 							}
 
-							const size_t Idx = std::distance(std::begin(CacheRecord.PayloadIds), It);
+							const size_t Idx = std::distance(std::begin(CacheRecord.ValueContentIds), It);
 
 							CloudCacheResult BlobResult;
 							for (int32_t Attempt = 0; Attempt < MaxAttempts && !BlobResult.Success; Attempt++)
 							{
-								BlobResult = Session.PutCompressedBlob(CacheRecord.PayloadIds[Idx], Payloads[Idx]);
+								BlobResult = Session.PutCompressedBlob(CacheRecord.ValueContentIds[Idx], Values[Idx]);
 							}
 
 							m_Status.SetFromErrorCode(BlobResult.ErrorCode, BlobResult.Reason);
 
 							if (!BlobResult.Success)
 							{
-								OutReason = fmt::format("upload value '{}' FAILED, reason '{}'", PayloadId, BlobResult.Reason);
+								OutReason = fmt::format("upload value '{}' FAILED, reason '{}'", ValueContentId, BlobResult.Reason);
 								return false;
 							}
 
@@ -619,15 +630,10 @@ namespace detail {
 			}
 		}
 
-		virtual GetUpstreamCacheResult GetCacheRecords(std::span<CacheKey>		  CacheKeys,
-													   std::span<size_t>		  KeyIndex,
-													   const CacheRecordPolicy&	  Policy,
-													   OnCacheRecordGetComplete&& OnComplete) override
+		virtual GetUpstreamCacheResult GetCacheRecords(std::span<CacheKeyRequest*> Requests, OnCacheRecordGetComplete&& OnComplete) override
 		{
 			ZEN_TRACE_CPU("Upstream::Zen::GetCacheRecords");
-
-			std::vector<size_t> IndexMap;
-			IndexMap.reserve(KeyIndex.size());
+			ZEN_ASSERT(Requests.size() > 0);
 
 			CbObjectWriter BatchRequest;
 			BatchRequest << "Method"sv
@@ -635,21 +641,30 @@ namespace detail {
 
 			BatchRequest.BeginObject("Params"sv);
 			{
-				BatchRequest.BeginArray("CacheKeys"sv);
-				for (size_t Index : KeyIndex)
-				{
-					const CacheKey& Key = CacheKeys[Index];
-					IndexMap.push_back(Index);
+				CachePolicy DefaultPolicy = Requests[0]->Policy.GetRecordPolicy();
+				BatchRequest << "DefaultPolicy"sv << WriteToString<128>(DefaultPolicy);
 
+				BatchRequest.BeginArray("Requests"sv);
+				for (CacheKeyRequest* Request : Requests)
+				{
 					BatchRequest.BeginObject();
-					BatchRequest << "Bucket"sv << Key.Bucket;
-					BatchRequest << "Hash"sv << Key.Hash;
+					{
+						const CacheKey& Key = Request->Key;
+						BatchRequest.BeginObject("Key"sv);
+						{
+							BatchRequest << "Bucket"sv << Key.Bucket;
+							BatchRequest << "Hash"sv << Key.Hash;
+						}
+						BatchRequest.EndObject();
+						if (!Request->Policy.IsUniform() || Request->Policy.GetRecordPolicy() != DefaultPolicy)
+						{
+							BatchRequest.SetName("Policy"sv);
+							Request->Policy.Save(BatchRequest);
+						}
+					}
 					BatchRequest.EndObject();
 				}
 				BatchRequest.EndArray();
-
-				BatchRequest.SetName("Policy"sv);
-				Policy.Save(BatchRequest);
 			}
 			BatchRequest.EndObject();
 
@@ -667,32 +682,40 @@ namespace detail {
 			{
 				if (BatchResponse.TryLoad(Result.Response))
 				{
-					for (size_t LocalIndex = 0; CbFieldView Record : BatchResponse.GetObject()["Result"sv])
+					CbArrayView Results = BatchResponse.GetObject()["Result"sv].AsArrayView();
+					if (Results.Num() != Requests.size())
 					{
-						const size_t Index = IndexMap[LocalIndex++];
-						OnComplete({.Key = CacheKeys[Index], .KeyIndex = Index, .Record = Record.AsObjectView(), .Package = BatchResponse});
+						ZEN_WARN("Upstream::Zen::GetCacheRecords invalid number of Requests from Upstream.");
 					}
+					else
+					{
+						for (size_t Index = 0; CbFieldView Record : Results)
+						{
+							CacheKeyRequest* Request = Requests[Index++];
+							OnComplete({.Request = *Request, .Record = Record.AsObjectView(), .Package = BatchResponse});
+						}
 
-					return {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = true};
+						return {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = true};
+					}
 				}
 			}
 
-			for (size_t Index : KeyIndex)
+			for (CacheKeyRequest* Request : Requests)
 			{
-				OnComplete({.Key = CacheKeys[Index], .KeyIndex = Index, .Record = CbObjectView(), .Package = CbPackage()});
+				OnComplete({.Request = *Request, .Record = CbObjectView(), .Package = CbPackage()});
 			}
 
 			return {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}};
 		}
 
-		virtual GetUpstreamCacheResult GetCachePayload(const CacheKey& CacheKey, const IoHash& PayloadId) override
+		virtual GetUpstreamCacheResult GetCacheValue(const CacheKey& CacheKey, const IoHash& ValueContentId) override
 		{
-			ZEN_TRACE_CPU("Upstream::Zen::GetSingleCachePayload");
+			ZEN_TRACE_CPU("Upstream::Zen::GetSingleCacheValue");
 
 			try
 			{
 				ZenStructuredCacheSession Session(*m_Client);
-				const ZenCacheResult	  Result = Session.GetCachePayload(CacheKey.Bucket, CacheKey.Hash, PayloadId);
+				const ZenCacheResult	  Result = Session.GetCacheValue(CacheKey.Bucket, CacheKey.Hash, ValueContentId);
 
 				m_Status.SetFromErrorCode(Result.ErrorCode, Result.Reason);
 
@@ -716,27 +739,28 @@ namespace detail {
 			}
 		}
 
-		virtual GetUpstreamCacheResult GetCachePayloads(std::span<CacheChunkRequest> CacheChunkRequests,
-														std::span<size_t>			 RequestIndex,
-														OnCachePayloadGetComplete&&	 OnComplete) override final
+		virtual GetUpstreamCacheResult GetCacheValues(std::span<CacheChunkRequest*> CacheChunkRequests,
+													  OnCacheValueGetComplete&&		OnComplete) override final
 		{
-			ZEN_TRACE_CPU("Upstream::Zen::GetCachePayloads");
-
-			std::vector<size_t> IndexMap;
-			IndexMap.reserve(RequestIndex.size());
+			ZEN_TRACE_CPU("Upstream::Zen::GetCacheValues");
+			ZEN_ASSERT(!CacheChunkRequests.empty());
 
 			CbObjectWriter BatchRequest;
 			BatchRequest << "Method"sv
-						 << "GetCacheValues";
+						 << "GetCacheChunks";
 
+#if BACKWARDS_COMPATABILITY_JAN2022
+			BatchRequest.AddInteger("MethodVersion"sv, 1);
+#endif
 			BatchRequest.BeginObject("Params"sv);
 			{
+				CachePolicy DefaultPolicy = CacheChunkRequests[0]->Policy;
+				BatchRequest << "DefaultPolicy"sv << WriteToString<128>(DefaultPolicy).ToView();
 				BatchRequest.BeginArray("ChunkRequests"sv);
 				{
-					for (size_t Index : RequestIndex)
+					for (CacheChunkRequest* RequestPtr : CacheChunkRequests)
 					{
-						const CacheChunkRequest& Request = CacheChunkRequests[Index];
-						IndexMap.push_back(Index);
+						const CacheChunkRequest& Request = *RequestPtr;
 
 						BatchRequest.BeginObject();
 						{
@@ -744,11 +768,26 @@ namespace detail {
 							BatchRequest << "Bucket"sv << Request.Key.Bucket;
 							BatchRequest << "Hash"sv << Request.Key.Hash;
 							BatchRequest.EndObject();
-							BatchRequest.AddObjectId("ValueId"sv, Request.PayloadId);
-							BatchRequest << "ChunkId"sv << Request.ChunkId;
-							BatchRequest << "RawOffset"sv << Request.RawOffset;
-							BatchRequest << "RawSize"sv << Request.RawSize;
-							BatchRequest << "Policy"sv << WriteToString<128>(Request.Policy).ToView();
+							if (Request.ValueId)
+							{
+								BatchRequest.AddObjectId("ValueId"sv, Request.ValueId);
+							}
+							if (Request.ChunkId != Request.ChunkId.Zero)
+							{
+								BatchRequest << "ChunkId"sv << Request.ChunkId;
+							}
+							if (Request.RawOffset != 0)
+							{
+								BatchRequest << "RawOffset"sv << Request.RawOffset;
+							}
+							if (Request.RawSize != UINT64_MAX)
+							{
+								BatchRequest << "RawSize"sv << Request.RawSize;
+							}
+							if (Request.Policy != DefaultPolicy)
+							{
+								BatchRequest << "Policy"sv << WriteToString<128>(Request.Policy).ToView();
+							}
 						}
 						BatchRequest.EndObject();
 					}
@@ -771,29 +810,56 @@ namespace detail {
 			{
 				if (BatchResponse.TryLoad(Result.Response))
 				{
-					for (size_t LocalIndex = 0; CbFieldView AttachmentHash : BatchResponse.GetObject()["Result"sv])
+					CbArrayView Results = BatchResponse.GetObject()["Result"sv].AsArrayView();
+					if (CacheChunkRequests.size() != Results.Num())
 					{
-						const size_t Index = IndexMap[LocalIndex++];
-						IoBuffer	 Payload;
-
-						if (const CbAttachment* Attachment = BatchResponse.FindAttachment(AttachmentHash.AsHash()))
+						ZEN_WARN("Upstream::Zen::GetCacheValues invalid number of Requests from Upstream.");
+					}
+					else
+					{
+						for (size_t RequestIndex = 0; CbFieldView ChunkField : Results)
 						{
-							if (const CompressedBuffer& Compressed = Attachment->AsCompressedBinary())
+							CacheChunkRequest& Request	   = *CacheChunkRequests[RequestIndex++];
+							CbObjectView	   ChunkObject = ChunkField.AsObjectView();
+							IoHash			   RawHash	   = ChunkObject["RawHash"sv].AsHash();
+							IoBuffer		   Payload;
+							uint64_t		   RawSize = 0;
+							if (RawHash != IoHash::Zero)
 							{
-								Payload = Compressed.GetCompressed().Flatten().AsIoBuffer();
+								bool				Success	   = false;
+								const CbAttachment* Attachment = BatchResponse.FindAttachment(RawHash);
+								if (Attachment)
+								{
+									if (const CompressedBuffer& Compressed = Attachment->AsCompressedBinary())
+									{
+										Payload = Compressed.GetCompressed().Flatten().AsIoBuffer();
+										Payload.SetContentType(ZenContentType::kCompressedBinary);
+										RawSize = Compressed.GetRawSize();
+										Success = true;
+									}
+								}
+								if (!Success)
+								{
+									CbFieldView RawSizeField = ChunkObject["RawSize"sv];
+									RawSize					 = RawSizeField.AsUInt64();
+									Success					 = !RawSizeField.HasError();
+								}
+								if (!Success)
+								{
+									RawHash = IoHash::Zero;
+								}
 							}
+							OnComplete({.Request = Request, .RawHash = RawHash, .RawSize = RawSize, .Value = std::move(Payload)});
 						}
 
-						OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Payload = std::move(Payload)});
+						return {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = true};
 					}
-
-					return {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = true};
 				}
 			}
 
-			for (size_t Index : RequestIndex)
+			for (CacheChunkRequest* RequestPtr : CacheChunkRequests)
 			{
-				OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Payload = IoBuffer()});
+				OnComplete({.Request = *RequestPtr, .RawHash = IoHash::Zero, .RawSize = 0, .Value = IoBuffer()});
 			}
 
 			return {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}};
@@ -801,11 +867,11 @@ namespace detail {
 
 		virtual PutUpstreamCacheResult PutCacheRecord(const UpstreamCacheRecord& CacheRecord,
 													  IoBuffer					 RecordValue,
-													  std::span<IoBuffer const>	 Payloads) override
+													  std::span<IoBuffer const>	 Values) override
 		{
 			ZEN_TRACE_CPU("Upstream::Zen::PutCacheRecord");
 
-			ZEN_ASSERT(CacheRecord.PayloadIds.size() == Payloads.size());
+			ZEN_ASSERT(CacheRecord.ValueContentIds.size() == Values.size());
 			const int32_t MaxAttempts = 3;
 
 			try
@@ -820,9 +886,9 @@ namespace detail {
 					CbPackage Package;
 					Package.SetObject(CbObject(SharedBuffer(RecordValue)));
 
-					for (const IoBuffer& Payload : Payloads)
+					for (const IoBuffer& Value : Values)
 					{
-						if (CompressedBuffer AttachmentBuffer = CompressedBuffer::FromCompressed(SharedBuffer(Payload)))
+						if (CompressedBuffer AttachmentBuffer = CompressedBuffer::FromCompressed(SharedBuffer(Value)))
 						{
 							Package.AddAttachment(CbAttachment(AttachmentBuffer));
 						}
@@ -848,15 +914,15 @@ namespace detail {
 				}
 				else
 				{
-					for (size_t Idx = 0, Count = Payloads.size(); Idx < Count; Idx++)
+					for (size_t Idx = 0, Count = Values.size(); Idx < Count; Idx++)
 					{
 						Result.Success = false;
 						for (uint32_t Attempt = 0; Attempt < MaxAttempts && !Result.Success; Attempt++)
 						{
-							Result = Session.PutCachePayload(CacheRecord.Key.Bucket,
-															 CacheRecord.Key.Hash,
-															 CacheRecord.PayloadIds[Idx],
-															 Payloads[Idx]);
+							Result = Session.PutCacheValue(CacheRecord.Key.Bucket,
+														   CacheRecord.Key.Hash,
+														   CacheRecord.ValueContentIds[Idx],
+														   Values[Idx]);
 						}
 
 						m_Status.SetFromErrorCode(Result.ErrorCode, Result.Reason);
@@ -866,7 +932,7 @@ namespace detail {
 
 						if (!Result.Success)
 						{
-							return {.Reason			= "Failed to upload payload",
+							return {.Reason			= "Failed to upload value",
 									.Bytes			= TotalBytes,
 									.ElapsedSeconds = TotalElapsedSeconds,
 									.Success		= false};
@@ -1044,21 +1110,16 @@ public:
 		return {};
 	}
 
-	virtual void GetCacheRecords(std::span<CacheKey>		CacheKeys,
-								 std::span<size_t>			KeyIndex,
-								 const CacheRecordPolicy&	DownstreamPolicy,
-								 OnCacheRecordGetComplete&& OnComplete) override final
+	virtual void GetCacheRecords(std::span<CacheKeyRequest*> Requests, OnCacheRecordGetComplete&& OnComplete) override final
 	{
 		ZEN_TRACE_CPU("Upstream::GetCacheRecords");
 
 		std::shared_lock<std::shared_mutex> _(m_EndpointsMutex);
 
-		std::vector<size_t> RemainingKeys(KeyIndex.begin(), KeyIndex.end());
+		std::vector<CacheKeyRequest*> RemainingKeys(Requests.begin(), Requests.end());
 
 		if (m_Options.ReadUpstream)
 		{
-			CacheRecordPolicy UpstreamPolicy = DownstreamPolicy.ConvertToUpstream();
-
 			for (auto& Endpoint : m_Endpoints)
 			{
 				if (RemainingKeys.empty())
@@ -1071,25 +1132,24 @@ public:
 					continue;
 				}
 
-				UpstreamEndpointStats& Stats = Endpoint->Stats();
-				std::vector<size_t>	   Missing;
-				GetUpstreamCacheResult Result;
+				UpstreamEndpointStats&		  Stats = Endpoint->Stats();
+				std::vector<CacheKeyRequest*> Missing;
+				GetUpstreamCacheResult		  Result;
 				{
 					metrics::OperationTiming::Scope Scope(Stats.CacheGetRequestTiming);
 
-					Result =
-						Endpoint->GetCacheRecords(CacheKeys, RemainingKeys, UpstreamPolicy, [&](CacheRecordGetCompleteParams&& Params) {
-							if (Params.Record)
-							{
-								OnComplete(std::forward<CacheRecordGetCompleteParams>(Params));
+					Result = Endpoint->GetCacheRecords(RemainingKeys, [&](CacheRecordGetCompleteParams&& Params) {
+						if (Params.Record)
+						{
+							OnComplete(std::forward<CacheRecordGetCompleteParams>(Params));
 
-								Stats.CacheHitCount.Increment(1);
-							}
-							else
-							{
-								Missing.push_back(Params.KeyIndex);
-							}
-						});
+							Stats.CacheHitCount.Increment(1);
+						}
+						else
+						{
+							Missing.push_back(&Params.Request);
+						}
+					});
 				}
 
 				Stats.CacheGetCount.Increment(int64_t(RemainingKeys.size()));
@@ -1109,21 +1169,19 @@ public:
 			}
 		}
 
-		for (size_t Index : RemainingKeys)
+		for (CacheKeyRequest* Request : RemainingKeys)
 		{
-			OnComplete({.Key = CacheKeys[Index], .KeyIndex = Index, .Record = CbObjectView(), .Package = CbPackage()});
+			OnComplete({.Request = *Request, .Record = CbObjectView(), .Package = CbPackage()});
 		}
 	}
 
-	virtual void GetCachePayloads(std::span<CacheChunkRequest> CacheChunkRequests,
-								  std::span<size_t>			   RequestIndex,
-								  OnCachePayloadGetComplete&&  OnComplete) override final
+	virtual void GetCacheValues(std::span<CacheChunkRequest*> CacheChunkRequests, OnCacheValueGetComplete&& OnComplete) override final
 	{
-		ZEN_TRACE_CPU("Upstream::GetCachePayloads");
+		ZEN_TRACE_CPU("Upstream::GetCacheValues");
 
 		std::shared_lock<std::shared_mutex> _(m_EndpointsMutex);
 
-		std::vector<size_t> RemainingKeys(RequestIndex.begin(), RequestIndex.end());
+		std::vector<CacheChunkRequest*> RemainingKeys(CacheChunkRequests.begin(), CacheChunkRequests.end());
 
 		if (m_Options.ReadUpstream)
 		{
@@ -1139,22 +1197,22 @@ public:
 					continue;
 				}
 
-				UpstreamEndpointStats& Stats = Endpoint->Stats();
-				std::vector<size_t>	   Missing;
-				GetUpstreamCacheResult Result;
+				UpstreamEndpointStats&			Stats = Endpoint->Stats();
+				std::vector<CacheChunkRequest*> Missing;
+				GetUpstreamCacheResult			Result;
 				{
 					metrics::OperationTiming::Scope Scope(Endpoint->Stats().CacheGetRequestTiming);
 
-					Result = Endpoint->GetCachePayloads(CacheChunkRequests, RemainingKeys, [&](CachePayloadGetCompleteParams&& Params) {
-						if (Params.Payload)
+					Result = Endpoint->GetCacheValues(RemainingKeys, [&](CacheValueGetCompleteParams&& Params) {
+						if (Params.RawHash != Params.RawHash.Zero)
 						{
-							OnComplete(std::forward<CachePayloadGetCompleteParams>(Params));
+							OnComplete(std::forward<CacheValueGetCompleteParams>(Params));
 
 							Stats.CacheHitCount.Increment(1);
 						}
 						else
 						{
-							Missing.push_back(Params.RequestIndex);
+							Missing.push_back(&Params.Request);
 						}
 					});
 				}
@@ -1166,7 +1224,7 @@ public:
 				{
 					Stats.CacheErrorCount.Increment(1);
 
-					ZEN_ERROR("get cache payloads(s) (rpc) FAILED, endpoint '{}', reason '{}', error code '{}'",
+					ZEN_ERROR("get cache values(s) (rpc) FAILED, endpoint '{}', reason '{}', error code '{}'",
 							  Endpoint->GetEndpointInfo().Url,
 							  Result.Error.Reason,
 							  Result.Error.ErrorCode);
@@ -1176,15 +1234,15 @@ public:
 			}
 		}
 
-		for (size_t Index : RemainingKeys)
+		for (CacheChunkRequest* RequestPtr : CacheChunkRequests)
 		{
-			OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Payload = IoBuffer()});
+			OnComplete({.Request = *RequestPtr, .RawHash = IoHash::Zero, .RawSize = 0, .Value = IoBuffer()});
 		}
 	}
 
-	virtual GetUpstreamCacheResult GetCachePayload(const CacheKey& CacheKey, const IoHash& PayloadId) override
+	virtual GetUpstreamCacheResult GetCacheValue(const CacheKey& CacheKey, const IoHash& ValueContentId) override
 	{
-		ZEN_TRACE_CPU("Upstream::GetCachePayload");
+		ZEN_TRACE_CPU("Upstream::GetCacheValue");
 
 		if (m_Options.ReadUpstream)
 		{
@@ -1200,7 +1258,7 @@ public:
 
 				{
 					metrics::OperationTiming::Scope Scope(Stats.CacheGetRequestTiming);
-					Result = Endpoint->GetCachePayload(CacheKey, PayloadId);
+					Result = Endpoint->GetCacheValue(CacheKey, ValueContentId);
 				}
 
 				Stats.CacheGetCount.Increment(1);
@@ -1217,7 +1275,7 @@ public:
 				{
 					Stats.CacheErrorCount.Increment(1);
 
-					ZEN_ERROR("get cache payload FAILED, endpoint '{}', reason '{}', error code '{}'",
+					ZEN_ERROR("get cache value FAILED, endpoint '{}', reason '{}', error code '{}'",
 							  Endpoint->GetEndpointInfo().Url,
 							  Result.Error.Reason,
 							  Result.Error.ErrorCode);
@@ -1302,18 +1360,18 @@ private:
 			return;
 		}
 
-		for (const IoHash& PayloadId : CacheRecord.PayloadIds)
+		for (const IoHash& ValueContentId : CacheRecord.ValueContentIds)
 		{
-			if (IoBuffer Payload = m_CidStore.FindChunkByCid(PayloadId))
+			if (IoBuffer Payload = m_CidStore.FindChunkByCid(ValueContentId))
 			{
 				Payloads.push_back(Payload);
 			}
 			else
 			{
-				ZEN_WARN("process upstream FAILED, '{}/{}/{}', payload doesn't exist in CAS",
+				ZEN_WARN("process upstream FAILED, '{}/{}/{}', ValueContentId doesn't exist in CAS",
 						 CacheRecord.Key.Bucket,
 						 CacheRecord.Key.Hash,
-						 PayloadId);
+						 ValueContentId);
 				return;
 			}
 		}