Value propagation fix - Read/Write ValueAPI as CompressedBinary type when writing to zen and horde upstreams. Return failure from HandleGetCacheRecord if the requested type does not match the cachetype. (#55)

* Fix bug with getting values PUT to Jupiter as CompressedBinary. When getting CompressedBinary records from Jupiter, they are expected to now be a record with a reference to the compact binary.  This has to be accounted for when performing upstream GETs.
* HandleGetCacheRecord: avoid crashing on invalid type, and avoid sending back data that doesn't match the AcceptType. 

4 files changed, 92 insertions, 39 deletions

diff --git a/zenserver-test/zenserver-test.cpp b/zenserver-test/zenserver-test.cpp
index 8ca16e560..293af7816 100644
--- a/zenserver-test/zenserver-test.cpp
+++ b/zenserver-test/zenserver-test.cpp
@@ -324,7 +324,7 @@ main(int argc, char** argv)
 	zen::logging::InitializeLogging();
 
 	spdlog::set_level(spdlog::level::debug);
-	spdlog::set_formatter(std::make_unique<::logging::full_test_formatter>("test", std::chrono::system_clock::now()));
+	spdlog::set_formatter(std::make_unique< ::logging::full_test_formatter>("test", std::chrono::system_clock::now()));
 
 	std::filesystem::path ProgramBaseDir = std::filesystem::path(argv[0]).parent_path();
 	std::filesystem::path TestBaseDir	 = ProgramBaseDir.parent_path().parent_path() / ".test";
@@ -797,7 +797,8 @@ TEST_CASE("zcache.basic")
 		{
 			zen::IoHash Key = zen::IoHash::HashBuffer(&i, sizeof i);
 
-			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}", BaseUri, "test", Key)});
+			cpr::Response Result =
+				cpr::Get(cpr::Url{fmt::format("{}/{}/{}", BaseUri, "test", Key)}, cpr::Header{{"Accept", "application/x-ue-cbpkg"}});
 
 			CHECK(Result.status_code == 200);
 		}
@@ -835,7 +836,8 @@ TEST_CASE("zcache.basic")
 		{
 			zen::IoHash Key = HashKey(i);
 
-			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}", BaseUri, "test", Key)});
+			cpr::Response Result =
+				cpr::Get(cpr::Url{fmt::format("{}/{}/{}", BaseUri, "test", Key)}, cpr::Header{{"Accept", "application/x-ue-cbpkg"}});
 
 			CHECK(Result.status_code == 200);
 		}
@@ -1348,7 +1350,7 @@ TEST_CASE("zcache.policy")
 		// Get record
 		{
 			cpr::Response Result = cpr::Get(cpr::Url{fmt::format("{}/{}/{}?Policy=Default,SkipData", Cfg.BaseUri, Bucket, Key)},
-											cpr::Header{{"Accept", "application/x-ue-cbobject"}});
+											cpr::Header{{"Accept", "application/x-ue-cb"}});
 			CHECK(IsHttpSuccessCode(Result.status_code));
 			IoBuffer Buffer(IoBuffer::Wrap, Result.text.c_str(), Result.text.size());
 			CbObject ResponseObject = zen::LoadCompactBinaryObject(Buffer);
@@ -2581,9 +2583,9 @@ public:
 	ZenServerInstance& GetInstance(int Index) { return *m_Instances[Index]; }
 
 private:
-	std::string										m_HelperId;
-	int												m_ServerCount = 0;
-	std::vector<std::unique_ptr<ZenServerInstance>> m_Instances;
+	std::string										 m_HelperId;
+	int												 m_ServerCount = 0;
+	std::vector<std::unique_ptr<ZenServerInstance> > m_Instances;
 };
 
 TEST_CASE("http.basics")
diff --git a/zenserver/cache/structuredcache.cpp b/zenserver/cache/structuredcache.cpp
index 25077d92a..8ae531720 100644
--- a/zenserver/cache/structuredcache.cpp
+++ b/zenserver/cache/structuredcache.cpp
@@ -227,44 +227,60 @@ HttpStructuredCacheService::HandleGetCacheRecord(zen::HttpServerRequest& Request
 
 	if (EnumHasAllFlags(PolicyFromURL, CachePolicy::QueryLocal) && m_CacheStore.Get(Ref.BucketSegment, Ref.HashKey, ClientResultValue))
 	{
-		Success = true;
+		Success					   = true;
+		ZenContentType ContentType = ClientResultValue.Value.GetContentType();
 
 		if (AcceptType == ZenContentType::kCbPackage)
 		{
-			CbPackage Package;
-			uint32_t  MissingCount = 0;
+			if (ContentType == ZenContentType::kCbObject)
+			{
+				CbPackage Package;
+				uint32_t  MissingCount = 0;
 
-			CbObjectView CacheRecord(ClientResultValue.Value.Data());
-			CacheRecord.IterateAttachments([this, &MissingCount, &Package, SkipData](CbFieldView AttachmentHash) {
-				if (SkipData)
-				{
-					MissingCount += m_CidStore.ContainsChunk(AttachmentHash.AsHash()) ? 0 : 1;
-				}
-				else
-				{
-					if (IoBuffer Chunk = m_CidStore.FindChunkByCid(AttachmentHash.AsHash()))
+				CbObjectView CacheRecord(ClientResultValue.Value.Data());
+				CacheRecord.IterateAttachments([this, &MissingCount, &Package, SkipData](CbFieldView AttachmentHash) {
+					if (SkipData)
 					{
-						Package.AddAttachment(CbAttachment(CompressedBuffer::FromCompressed(SharedBuffer(Chunk))));
+						if (!m_CidStore.ContainsChunk(AttachmentHash.AsHash()))
+						{
+							MissingCount++;
+						}
 					}
 					else
 					{
-						MissingCount++;
+						if (IoBuffer Chunk = m_CidStore.FindChunkByCid(AttachmentHash.AsHash()))
+						{
+							Package.AddAttachment(CbAttachment(CompressedBuffer::FromCompressed(SharedBuffer(Chunk))));
+						}
+						else
+						{
+							MissingCount++;
+						}
 					}
-				}
-			});
+				});
 
-			Success = MissingCount == 0 || PartialRecord;
+				Success = MissingCount == 0 || PartialRecord;
 
-			if (Success)
-			{
-				Package.SetObject(LoadCompactBinaryObject(ClientResultValue.Value));
+				if (Success)
+				{
+					Package.SetObject(LoadCompactBinaryObject(ClientResultValue.Value));
 
-				BinaryWriter MemStream;
-				Package.Save(MemStream);
+					BinaryWriter MemStream;
+					Package.Save(MemStream);
 
-				ClientResultValue.Value = IoBuffer(IoBuffer::Clone, MemStream.Data(), MemStream.Size());
-				ClientResultValue.Value.SetContentType(HttpContentType::kCbPackage);
+					ClientResultValue.Value = IoBuffer(IoBuffer::Clone, MemStream.Data(), MemStream.Size());
+					ClientResultValue.Value.SetContentType(HttpContentType::kCbPackage);
+				}
 			}
+			else
+			{
+				Success = false;
+			}
+		}
+		else if (AcceptType != ClientResultValue.Value.GetContentType() && AcceptType != ZenContentType::kUnknownContentType &&
+				 AcceptType != ZenContentType::kBinary)
+		{
+			Success = false;
 		}
 	}
 
@@ -277,13 +293,13 @@ HttpStructuredCacheService::HandleGetCacheRecord(zen::HttpServerRequest& Request
 				  ToString(ClientResultValue.Value.GetContentType()));
 
 		m_CacheStats.HitCount++;
-		if (SkipData && AcceptType == ZenContentType::kBinary)
+		if (SkipData && AcceptType != ZenContentType::kCbPackage && AcceptType != ZenContentType::kCbObject)
 		{
 			return Request.WriteResponse(HttpResponseCode::OK);
 		}
 		else
 		{
-			// Other types handled SkipData when constructing the ClientResultValue
+			// kCbPackage handled SkipData when constructing the ClientResultValue, kcbObject ignores SkipData
 			return Request.WriteResponse(HttpResponseCode::OK, ClientResultValue.Value.GetContentType(), ClientResultValue.Value);
 		}
 	}
@@ -1462,7 +1478,8 @@ HttpStructuredCacheService::HandleRpcGetCacheValues(zen::HttpServerRequest& Http
 		}
 		if (!Result && EnumHasAllFlags(Policy, CachePolicy::QueryRemote))
 		{
-			GetUpstreamCacheResult UpstreamResult = m_UpstreamCache.GetCacheRecord({Key.Bucket, Key.Hash}, ZenContentType::kBinary);
+			GetUpstreamCacheResult UpstreamResult =
+				m_UpstreamCache.GetCacheRecord({Key.Bucket, Key.Hash}, ZenContentType::kCompressedBinary);
 			if (UpstreamResult.Success && IsCompressedBinary(UpstreamResult.Value.GetContentType()))
 			{
 				Result = CompressedBuffer::FromCompressed(SharedBuffer(UpstreamResult.Value));
diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp
index 5e0f6a297..da0743f0a 100644
--- a/zenserver/upstream/upstreamcache.cpp
+++ b/zenserver/upstream/upstreamcache.cpp
@@ -177,6 +177,43 @@ namespace detail {
 				{
 					Result = Session.GetDerivedData(CacheKey.Bucket, CacheKey.Hash);
 				}
+				else if (Type == ZenContentType::kCompressedBinary)
+				{
+					Result = Session.GetRef(CacheKey.Bucket, CacheKey.Hash, ZenContentType::kCbObject);
+
+					if (Result.Success)
+					{
+						const CbValidateError ValidationResult = ValidateCompactBinary(Result.Response, CbValidateMode::All);
+						if (Result.Success = ValidationResult == CbValidateError::None; Result.Success)
+						{
+							CbObject CacheRecord = LoadCompactBinaryObject(Result.Response);
+							IoBuffer ContentBuffer;
+							int		 NumAttachments = 0;
+
+							CacheRecord.IterateAttachments(
+								[&Session, &Result, &ContentBuffer, &NumAttachments](CbFieldView AttachmentHash) {
+									CloudCacheResult AttachmentResult = Session.GetCompressedBlob(AttachmentHash.AsHash());
+									Result.Bytes += AttachmentResult.Bytes;
+									Result.ElapsedSeconds += AttachmentResult.ElapsedSeconds;
+									Result.ErrorCode = AttachmentResult.ErrorCode;
+
+									if (CompressedBuffer Chunk = CompressedBuffer::FromCompressed(SharedBuffer(AttachmentResult.Response)))
+									{
+										Result.Response = AttachmentResult.Response;
+										++NumAttachments;
+									}
+									else
+									{
+										Result.Success = false;
+									}
+								});
+							if (NumAttachments != 1)
+							{
+								Result.Success = false;
+							}
+						}
+					}
+				}
 				else
 				{
 					const ZenContentType AcceptType = Type == ZenContentType::kCbPackage ? ZenContentType::kCbObject : Type;
diff --git a/zenserver/upstream/zen.cpp b/zenserver/upstream/zen.cpp
index 8c26bdf8f..1ac4afe5c 100644
--- a/zenserver/upstream/zen.cpp
+++ b/zenserver/upstream/zen.cpp
@@ -8,6 +8,7 @@
 #include <zencore/fmtutils.h>
 #include <zencore/session.h>
 #include <zencore/stream.h>
+#include <zenhttp/httpcommon.h>
 #include <zenhttp/httpshared.h>
 
 #include "cache/structuredcachestore.h"
@@ -415,11 +416,7 @@ ZenStructuredCacheSession::GetCacheRecord(std::string_view BucketId, const IoHas
 	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
-	Session.SetHeader(cpr::Header{{"Accept",
-								   Type == ZenContentType::kCbPackage  ? "application/x-ue-cbpkg"
-								   : Type == ZenContentType::kCbObject ? "application/x-ue-cb"
-																	   : "application/octet-stream"}});
-
+	Session.SetHeader(cpr::Header{{"Accept", std::string{MapContentTypeToString(Type)}}});
 	cpr::Response Response = Session.Get();
 	ZEN_DEBUG("GET {}", Response);