skip upstream logic early if we have no upstream endpoints (#359)

* Skip upstream logic early if we have not upstream endpoints
* make cache store logging of CbObjects async
* changelog

1 files changed, 30 insertions, 14 deletions

diff --git a/src/zenserver/cache/httpstructuredcache.cpp b/src/zenserver/cache/httpstructuredcache.cpp
index 7c4097d4f..0120f3599 100644
--- a/src/zenserver/cache/httpstructuredcache.cpp
+++ b/src/zenserver/cache/httpstructuredcache.cpp
@@ -888,6 +888,8 @@ HttpStructuredCacheService::HandleGetCacheRecord(HttpServerRequest& Request, con
 		return Request.WriteResponse(HttpResponseCode::OK);
 	}
 
+	const bool HasUpstream = m_UpstreamCache.IsActive();
+
 	CacheRequestContext RequestContext = {.SessionId = Request.SessionId(), .RequestId = Request.RequestId()};
 	Stopwatch			Timer;
 
@@ -973,7 +975,7 @@ HttpStructuredCacheService::HandleGetCacheRecord(HttpServerRequest& Request, con
 			return Request.WriteResponse(HttpResponseCode::OK, ClientResultValue.Value.GetContentType(), ClientResultValue.Value);
 		}
 	}
-	else if (!EnumHasAllFlags(PolicyFromUrl, CachePolicy::QueryRemote))
+	else if (!HasUpstream || !EnumHasAllFlags(PolicyFromUrl, CachePolicy::QueryRemote))
 	{
 		ZEN_DEBUG("GETCACHERECORD MISS - '{}/{}/{}' '{}' in {}",
 				  Ref.Namespace,
@@ -1212,6 +1214,8 @@ HttpStructuredCacheService::HandlePutCacheRecord(HttpServerRequest& Request, con
 
 	CacheRequestContext RequestContext = {.SessionId = Request.SessionId(), .RequestId = Request.RequestId()};
 
+	const bool HasUpstream = m_UpstreamCache.IsActive();
+
 	Stopwatch Timer;
 
 	if (ContentType == HttpContentType::kBinary || ContentType == HttpContentType::kCompressedBinary)
@@ -1237,7 +1241,7 @@ HttpStructuredCacheService::HandlePutCacheRecord(HttpServerRequest& Request, con
 						 Ref.HashKey,
 						 {.Value = Body, .RawSize = RawSize, .RawHash = RawHash});
 
-		if (EnumHasAllFlags(PolicyFromUrl, CachePolicy::StoreRemote))
+		if (HasUpstream && EnumHasAllFlags(PolicyFromUrl, CachePolicy::StoreRemote))
 		{
 			m_UpstreamCache.EnqueueUpstream({.Type = ContentType, .Namespace = Ref.Namespace, .Key = {Ref.BucketSegment, Ref.HashKey}});
 		}
@@ -1294,7 +1298,7 @@ HttpStructuredCacheService::HandlePutCacheRecord(HttpServerRequest& Request, con
 		const bool IsPartialRecord = TotalCount != static_cast<int32_t>(ValidAttachments.size());
 
 		CachePolicy Policy = PolicyFromUrl;
-		if (EnumHasAllFlags(Policy, CachePolicy::StoreRemote) && !IsPartialRecord)
+		if (HasUpstream && EnumHasAllFlags(Policy, CachePolicy::StoreRemote) && !IsPartialRecord)
 		{
 			m_UpstreamCache.EnqueueUpstream({.Type			  = ZenContentType::kCbObject,
 											 .Namespace		  = Ref.Namespace,
@@ -1390,7 +1394,7 @@ HttpStructuredCacheService::HandlePutCacheRecord(HttpServerRequest& Request, con
 
 		const bool IsPartialRecord = Count.Valid != Count.Total;
 
-		if (EnumHasAllFlags(Policy, CachePolicy::StoreRemote) && !IsPartialRecord)
+		if (HasUpstream && EnumHasAllFlags(Policy, CachePolicy::StoreRemote) && !IsPartialRecord)
 		{
 			m_UpstreamCache.EnqueueUpstream({.Type			  = ZenContentType::kCbPackage,
 											 .Namespace		  = Ref.Namespace,
@@ -1431,8 +1435,10 @@ HttpStructuredCacheService::HandleGetCacheChunk(HttpServerRequest& Request, cons
 	IoBuffer					Value  = m_CidStore.FindChunkByCid(Ref.ValueContentId);
 	const UpstreamEndpointInfo* Source = nullptr;
 	CachePolicy					Policy = PolicyFromUrl;
+
+	const bool HasUpstream = m_UpstreamCache.IsActive();
 	{
-		const bool QueryUpstream = !Value && EnumHasAllFlags(Policy, CachePolicy::QueryRemote);
+		const bool QueryUpstream = HasUpstream && !Value && EnumHasAllFlags(Policy, CachePolicy::QueryRemote);
 
 		if (QueryUpstream)
 		{
@@ -1835,6 +1841,8 @@ HttpStructuredCacheService::PutCacheRecord(PutRequestData& Request, const CbPack
 	ValidAttachments.reserve(NumAttachments);
 	AttachmentsToStoreLocally.reserve(NumAttachments);
 
+	const bool HasUpstream = m_UpstreamCache.IsActive();
+
 	Stopwatch Timer;
 
 	Request.RecordObject.IterateAttachments(
@@ -1901,7 +1909,7 @@ HttpStructuredCacheService::PutCacheRecord(PutRequestData& Request, const CbPack
 
 	const bool IsPartialRecord = Count.Valid != Count.Total;
 
-	if (EnumHasAllFlags(Request.Policy.GetRecordPolicy(), CachePolicy::StoreRemote) && !IsPartialRecord)
+	if (HasUpstream && EnumHasAllFlags(Request.Policy.GetRecordPolicy(), CachePolicy::StoreRemote) && !IsPartialRecord)
 	{
 		m_UpstreamCache.EnqueueUpstream({.Type			  = ZenContentType::kCbPackage,
 										 .Namespace		  = Request.Namespace,
@@ -1948,6 +1956,9 @@ HttpStructuredCacheService::HandleRpcGetCacheRecords(const CacheRequestContext&
 	{
 		return CbPackage{};
 	}
+
+	const bool HasUpstream = m_UpstreamCache.IsActive();
+
 	std::vector<RecordRequestData> Requests;
 	std::vector<size_t>			   UpstreamIndexes;
 	CbArrayView					   RequestsArray = Params["Requests"sv].AsArrayView();
@@ -2062,8 +2073,8 @@ HttpStructuredCacheService::HandleRpcGetCacheRecords(const CacheRequestContext&
 		}
 		if (!Request.Complete)
 		{
-			bool NeedUpstreamRecord =
-				!Request.RecordObject && !FoundLocalInvalid && EnumHasAllFlags(Policy.GetRecordPolicy(), CachePolicy::QueryRemote);
+			bool NeedUpstreamRecord = HasUpstream && !Request.RecordObject && !FoundLocalInvalid &&
+									  EnumHasAllFlags(Policy.GetRecordPolicy(), CachePolicy::QueryRemote);
 			if (NeedUpstreamRecord || NeedUpstreamAttachment)
 			{
 				UpstreamIndexes.push_back(Requests.size() - 1);
@@ -2265,6 +2276,8 @@ HttpStructuredCacheService::HandleRpcPutCacheValues(const CacheRequestContext& C
 	{
 		return CbPackage{};
 	}
+	const bool HasUpstream = m_UpstreamCache.IsActive();
+
 	std::vector<bool> Results;
 	for (CbFieldView RequestField : Params["Requests"sv])
 	{
@@ -2329,7 +2342,7 @@ HttpStructuredCacheService::HandleRpcPutCacheValues(const CacheRequestContext& C
 		// We do not search the Upstream. No data in a put means the caller is probing for whether they need to do a heavy put.
 		// If it doesn't exist locally they should do the heavy put rather than having us fetch it from upstream.
 
-		if (Succeeded && EnumHasAllFlags(Policy, CachePolicy::StoreRemote))
+		if (HasUpstream && Succeeded && EnumHasAllFlags(Policy, CachePolicy::StoreRemote))
 		{
 			m_UpstreamCache.EnqueueUpstream({.Type = ZenContentType::kCompressedBinary, .Namespace = *Namespace, .Key = Key});
 		}
@@ -2387,6 +2400,8 @@ HttpStructuredCacheService::HandleRpcGetCacheValues(const CacheRequestContext& C
 
 	std::vector<size_t> RemoteRequestIndexes;
 
+	const bool HasUpstream = m_UpstreamCache.IsActive();
+
 	for (CbFieldView RequestField : Params["Requests"sv])
 	{
 		Stopwatch Timer;
@@ -2428,7 +2443,7 @@ HttpStructuredCacheService::HandleRpcGetCacheValues(const CacheRequestContext& C
 					  NiceLatencyNs(Timer.GetElapsedTimeUs() * 1000));
 			m_CacheStats.HitCount++;
 		}
-		else if (EnumHasAllFlags(Policy, CachePolicy::QueryRemote))
+		else if (HasUpstream && EnumHasAllFlags(Policy, CachePolicy::QueryRemote))
 		{
 			RemoteRequestIndexes.push_back(Requests.size() - 1);
 		}
@@ -2748,6 +2763,7 @@ HttpStructuredCacheService::GetLocalCacheRecords(const CacheRequestContext&
 												 std::vector<CacheChunkRequest*>&			OutUpstreamChunks)
 {
 	using namespace cache::detail;
+	const bool HasUpstream = m_UpstreamCache.IsActive();
 
 	std::vector<CacheKeyRequest*> UpstreamRecordRequests;
 	for (size_t RecordIndex = 0; RecordIndex < Records.size(); ++RecordIndex)
@@ -2768,7 +2784,7 @@ HttpStructuredCacheService::GetLocalCacheRecords(const CacheRequestContext&
 					Record.CacheValue = std::move(CacheValue.Value);
 				}
 			}
-			if (!Record.Exists && EnumHasAllFlags(Record.DownstreamPolicy, CachePolicy::QueryRemote))
+			if (HasUpstream && !Record.Exists && EnumHasAllFlags(Record.DownstreamPolicy, CachePolicy::QueryRemote))
 			{
 				RecordKey.Policy = CacheRecordPolicy(ConvertToUpstream(Record.DownstreamPolicy));
 				UpstreamRecordRequests.push_back(&RecordKey);
@@ -2806,7 +2822,6 @@ HttpStructuredCacheService::GetLocalCacheRecords(const CacheRequestContext&
 		m_UpstreamCache.GetCacheRecords(Namespace, UpstreamRecordRequests, std::move(OnCacheRecordGetComplete));
 	}
 
-	std::vector<CacheChunkRequest*> UpstreamPayloadRequests;
 	for (ChunkRequest* Request : RecordRequests)
 	{
 		Stopwatch Timer;
@@ -2883,7 +2898,7 @@ HttpStructuredCacheService::GetLocalCacheRecords(const CacheRequestContext&
 					}
 				}
 			}
-			if (!Request->Exists && EnumHasAllFlags(Request->DownstreamPolicy, CachePolicy::QueryRemote))
+			if (HasUpstream && !Request->Exists && EnumHasAllFlags(Request->DownstreamPolicy, CachePolicy::QueryRemote))
 			{
 				Request->Key->Policy = ConvertToUpstream(Request->DownstreamPolicy);
 				OutUpstreamChunks.push_back(Request->Key);
@@ -2900,6 +2915,7 @@ HttpStructuredCacheService::GetLocalCacheValues(const CacheRequestContext&
 												std::vector<CacheChunkRequest*>&		   OutUpstreamChunks)
 {
 	using namespace cache::detail;
+	const bool HasUpstream = m_UpstreamCache.IsActive();
 
 	for (ChunkRequest* Request : ValueRequests)
 	{
@@ -2922,7 +2938,7 @@ HttpStructuredCacheService::GetLocalCacheValues(const CacheRequestContext&
 				}
 			}
 		}
-		if (!Request->Exists && EnumHasAllFlags(Request->DownstreamPolicy, CachePolicy::QueryRemote))
+		if (HasUpstream && !Request->Exists && EnumHasAllFlags(Request->DownstreamPolicy, CachePolicy::QueryRemote))
 		{
 			if (EnumHasAllFlags(Request->DownstreamPolicy, CachePolicy::StoreLocal))
 			{