Ensure upstream put propagation includes overwrite

When changing the default limit-overwrite behavior, a unit test surfaced a bug where an put of data with overwrite cache policy would not get propagated via zen's built-in upstream mechanism with a matching overwrite cache policy to the upstream.  This change ensures that it does and leaves the unit test configured to exercise this scenario.

4 files changed, 39 insertions, 5 deletions

diff --git a/src/zenserver/storage/cache/httpstructuredcache.cpp b/src/zenserver/storage/cache/httpstructuredcache.cpp
index 03d2140a1..72f29d14e 100644
--- a/src/zenserver/storage/cache/httpstructuredcache.cpp
+++ b/src/zenserver/storage/cache/httpstructuredcache.cpp
@@ -1268,7 +1268,8 @@ HttpStructuredCacheService::HandlePutCacheRecord(HttpServerRequest& Request, con
 
 		if (HasUpstream && EnumHasAllFlags(PolicyFromUrl, CachePolicy::StoreRemote))
 		{
-			m_UpstreamCache.EnqueueUpstream({.Type = ContentType, .Namespace = Ref.Namespace, .Key = {Ref.BucketSegment, Ref.HashKey}});
+			m_UpstreamCache.EnqueueUpstream(
+				{.Type = ContentType, .Namespace = Ref.Namespace, .Key = {Ref.BucketSegment, Ref.HashKey}, .Overwrite = Overwrite});
 		}
 
 		ZEN_DEBUG("PUTCACHERECORD - '{}/{}/{}' {} '{}' in {}",
@@ -1347,7 +1348,8 @@ HttpStructuredCacheService::HandlePutCacheRecord(HttpServerRequest& Request, con
 			m_UpstreamCache.EnqueueUpstream({.Type			  = ZenContentType::kCbObject,
 											 .Namespace		  = Ref.Namespace,
 											 .Key			  = {Ref.BucketSegment, Ref.HashKey},
-											 .ValueContentIds = std::move(ValidAttachments)});
+											 .ValueContentIds = std::move(ValidAttachments),
+											 .Overwrite		  = Overwrite});
 		}
 
 		Request.WriteResponse(HttpResponseCode::Created);
@@ -1465,7 +1467,8 @@ HttpStructuredCacheService::HandlePutCacheRecord(HttpServerRequest& Request, con
 			m_UpstreamCache.EnqueueUpstream({.Type			  = ZenContentType::kCbPackage,
 											 .Namespace		  = Ref.Namespace,
 											 .Key			  = {Ref.BucketSegment, Ref.HashKey},
-											 .ValueContentIds = std::move(ValidAttachments)});
+											 .ValueContentIds = std::move(ValidAttachments),
+											 .Overwrite		  = Overwrite});
 		}
 
 		Request.WriteResponse(HttpResponseCode::Created);
diff --git a/src/zenserver/storage/upstream/upstreamcache.cpp b/src/zenserver/storage/upstream/upstreamcache.cpp
index 938a1a011..b26c57414 100644
--- a/src/zenserver/storage/upstream/upstreamcache.cpp
+++ b/src/zenserver/storage/upstream/upstreamcache.cpp
@@ -554,6 +554,7 @@ namespace detail {
 						Result								= Session.PutRef(BlobStoreNamespace,
 												 CacheRecord.Key.Bucket,
 												 CacheRecord.Key.Hash,
+												 CacheRecord.Overwrite,
 												 RecordValue,
 												 ZenContentType::kBinary);
 					}
@@ -585,6 +586,7 @@ namespace detail {
 						Session,
 						CacheRecord.Namespace,
 						CacheRecord.Key,
+						CacheRecord.Overwrite,
 						ReferencingObject.Save().GetBuffer().AsIoBuffer(),
 						MaxAttempts,
 						[&](const IoHash& ValueContentId, IoBuffer& OutBuffer, std::string& OutReason) {
@@ -605,6 +607,7 @@ namespace detail {
 						Session,
 						CacheRecord.Namespace,
 						CacheRecord.Key,
+						CacheRecord.Overwrite,
 						RecordValue,
 						MaxAttempts,
 						[&](const IoHash& ValueContentId, IoBuffer& OutBuffer, std::string& OutReason) {
@@ -651,6 +654,7 @@ namespace detail {
 			JupiterSession&																					 Session,
 			std::string_view																				 Namespace,
 			const CacheKey&																					 Key,
+			bool																							 Overwrite,
 			IoBuffer																						 ObjectBuffer,
 			const int32_t																					 MaxAttempts,
 			std::function<bool(const IoHash& ValueContentId, IoBuffer& OutBuffer, std::string& OutReason)>&& BlobFetchFn)
@@ -692,7 +696,7 @@ namespace detail {
 			PutRefResult RefResult;
 			for (int32_t Attempt = 0; Attempt < MaxAttempts && !RefResult.Success; Attempt++)
 			{
-				RefResult = Session.PutRef(BlobStoreNamespace, Key.Bucket, Key.Hash, ObjectBuffer, ZenContentType::kCbObject);
+				RefResult = Session.PutRef(BlobStoreNamespace, Key.Bucket, Key.Hash, Overwrite, ObjectBuffer, ZenContentType::kCbObject);
 			}
 
 			m_Status.SetFromErrorCode(RefResult.ErrorCode, RefResult.Reason);
@@ -1302,6 +1306,7 @@ namespace detail {
 						Result = Session.PutCacheRecord(CacheRecord.Namespace,
 														CacheRecord.Key.Bucket,
 														CacheRecord.Key.Hash,
+														CacheRecord.Overwrite ? CachePolicy::Store : CachePolicy::Default,
 														PackagePayload,
 														CacheRecord.Type);
 					}
@@ -1329,7 +1334,14 @@ namespace detail {
 
 					BatchWriter.BeginObject("Params"sv);
 					{
-						// DefaultPolicy unspecified and expected to be Default
+						if (CacheRecord.Overwrite)
+						{
+							BatchWriter << "DefaultPolicy"sv << WriteToString<128>(CachePolicy::Store).ToView();
+						}
+						else
+						{
+							// DefaultPolicy unspecified and expected to be Default
+						}
 
 						BatchWriter << "Namespace"sv << CacheRecord.Namespace;
 
@@ -1376,6 +1388,7 @@ namespace detail {
 							Result = Session.PutCacheValue(CacheRecord.Namespace,
 														   CacheRecord.Key.Bucket,
 														   CacheRecord.Key.Hash,
+														   CacheRecord.Overwrite ? CachePolicy::Store : CachePolicy::Default,
 														   CacheRecord.ValueContentIds[Idx],
 														   Values[Idx]);
 						}
@@ -1400,6 +1413,7 @@ namespace detail {
 						Result = Session.PutCacheRecord(CacheRecord.Namespace,
 														CacheRecord.Key.Bucket,
 														CacheRecord.Key.Hash,
+														CacheRecord.Overwrite ? CachePolicy::Store : CachePolicy::Default,
 														RecordValue,
 														CacheRecord.Type);
 					}
diff --git a/src/zenserver/storage/upstream/zen.cpp b/src/zenserver/storage/upstream/zen.cpp
index 25fd3a3bb..423c9039c 100644
--- a/src/zenserver/storage/upstream/zen.cpp
+++ b/src/zenserver/storage/upstream/zen.cpp
@@ -127,6 +127,7 @@ ZenCacheResult
 ZenStructuredCacheSession::PutCacheRecord(std::string_view Namespace,
 										  std::string_view BucketId,
 										  const IoHash&	   Key,
+										  CachePolicy	   Policy,
 										  IoBuffer		   Value,
 										  ZenContentType   Type)
 {
@@ -140,6 +141,12 @@ ZenStructuredCacheSession::PutCacheRecord(std::string_view Namespace,
 	}
 	Uri << BucketId << "/" << Key.ToHexString();
 
+	if (Policy != CachePolicy::Default)
+	{
+		Uri << "?Policy=";
+		Uri << Policy;
+	}
+
 	Value.SetContentType(Type);
 
 	HttpClient::Response Response = Http.Put(Uri, Value);
@@ -159,6 +166,7 @@ ZenCacheResult
 ZenStructuredCacheSession::PutCacheValue(std::string_view Namespace,
 										 std::string_view BucketId,
 										 const IoHash&	  Key,
+										 CachePolicy	  Policy,
 										 const IoHash&	  ValueContentId,
 										 IoBuffer		  Payload)
 {
@@ -172,6 +180,12 @@ ZenStructuredCacheSession::PutCacheValue(std::string_view Namespace,
 	}
 	Uri << BucketId << "/" << Key.ToHexString() << "/" << ValueContentId.ToHexString();
 
+	if (Policy != CachePolicy::Default)
+	{
+		Uri << "?Policy=";
+		Uri << Policy;
+	}
+
 	Payload.SetContentType(HttpContentType::kCompressedBinary);
 
 	HttpClient::Response Response = Http.Put(Uri, Payload);
diff --git a/src/zenserver/storage/upstream/zen.h b/src/zenserver/storage/upstream/zen.h
index 6321b46b1..54fb063c2 100644
--- a/src/zenserver/storage/upstream/zen.h
+++ b/src/zenserver/storage/upstream/zen.h
@@ -8,6 +8,7 @@
 #include <zencore/memoryview.h>
 #include <zencore/uid.h>
 #include <zencore/zencore.h>
+#include <zenstore/cache/cachepolicy.h>
 
 #include <chrono>
 
@@ -59,11 +60,13 @@ public:
 	ZenCacheResult PutCacheRecord(std::string_view Namespace,
 								  std::string_view BucketId,
 								  const IoHash&	   Key,
+								  CachePolicy	   Policy,
 								  IoBuffer		   Value,
 								  ZenContentType   Type);
 	ZenCacheResult PutCacheValue(std::string_view Namespace,
 								 std::string_view BucketId,
 								 const IoHash&	  Key,
+								 CachePolicy	  Policy,
 								 const IoHash&	  ValueContentId,
 								 IoBuffer		  Payload);
 	ZenCacheResult InvokeRpc(const CbObjectView& Request);