Refactored upstream cache to better handle different states in prep for dynamic auth tokens.

1 files changed, 22 insertions, 29 deletions

diff --git a/zenserver/cache/structuredcache.cpp b/zenserver/cache/structuredcache.cpp
index 2675590e0..5918d5178 100644
--- a/zenserver/cache/structuredcache.cpp
+++ b/zenserver/cache/structuredcache.cpp
@@ -59,17 +59,17 @@ struct AttachmentCount
 
 //////////////////////////////////////////////////////////////////////////
 
-HttpStructuredCacheService::HttpStructuredCacheService(ZenCacheStore&				  InCacheStore,
-													   CidStore&					  InCidStore,
-													   HttpStatsService&			  StatsService,
-													   HttpStatusService&			  StatusService,
-													   std::unique_ptr<UpstreamCache> UpstreamCache)
+HttpStructuredCacheService::HttpStructuredCacheService(ZenCacheStore&	  InCacheStore,
+													   CidStore&		  InCidStore,
+													   HttpStatsService&  StatsService,
+													   HttpStatusService& StatusService,
+													   UpstreamCache&	  UpstreamCache)
 : m_Log(logging::Get("cache"))
 , m_CacheStore(InCacheStore)
 , m_StatsService(StatsService)
 , m_StatusService(StatusService)
 , m_CidStore(InCidStore)
-, m_UpstreamCache(std::move(UpstreamCache))
+, m_UpstreamCache(UpstreamCache)
 {
 	m_StatsService.RegisterHandler("z$", *this);
 	m_StatusService.RegisterHandler("z$", *this);
@@ -206,7 +206,7 @@ HttpStructuredCacheService::HandleGetCacheRecord(zen::HttpServerRequest& Request
 	const bool			 SkipData		 = (Policy & CachePolicy::SkipData) == CachePolicy::SkipData;
 	const bool			 SkipAttachments = (Policy & CachePolicy::SkipAttachments) == CachePolicy::SkipAttachments;
 	const bool			 PartialOnError	 = (Policy & CachePolicy::PartialOnError) == CachePolicy::PartialOnError;
-	const bool			 QueryUpstream	 = m_UpstreamCache && ((Policy & CachePolicy::QueryRemote) == CachePolicy::QueryRemote);
+	const bool			 QueryUpstream	 = (Policy & CachePolicy::QueryRemote) == CachePolicy::QueryRemote;
 
 	bool		  Success = false;
 	ZenCacheValue LocalCacheValue;
@@ -292,7 +292,7 @@ HttpStructuredCacheService::HandleGetCacheRecord(zen::HttpServerRequest& Request
 
 		metrics::OperationTiming::Scope $(m_UpstreamGetRequestTiming);
 
-		if (GetUpstreamCacheResult UpstreamResult = m_UpstreamCache->GetCacheRecord({Ref.BucketSegment, Ref.HashKey}, AcceptType);
+		if (GetUpstreamCacheResult UpstreamResult = m_UpstreamCache.GetCacheRecord({Ref.BucketSegment, Ref.HashKey}, AcceptType);
 			UpstreamResult.Success)
 		{
 			Success = true;
@@ -437,7 +437,7 @@ HttpStructuredCacheService::HandlePutCacheRecord(zen::HttpServerRequest& Request
 	}
 
 	const HttpContentType ContentType	= Request.RequestContentType();
-	const bool			  StoreUpstream = m_UpstreamCache && (CachePolicy::StoreRemote == (Policy & CachePolicy::StoreRemote));
+	const bool			  StoreUpstream = (Policy & CachePolicy::StoreRemote) == CachePolicy::StoreRemote;
 
 	Body.SetContentType(ContentType);
 
@@ -448,8 +448,7 @@ HttpStructuredCacheService::HandlePutCacheRecord(zen::HttpServerRequest& Request
 
 		if (StoreUpstream)
 		{
-			ZEN_ASSERT(m_UpstreamCache);
-			m_UpstreamCache->EnqueueUpstream({.Type = ZenContentType::kBinary, .Key = {Ref.BucketSegment, Ref.HashKey}});
+			m_UpstreamCache.EnqueueUpstream({.Type = ZenContentType::kBinary, .Key = {Ref.BucketSegment, Ref.HashKey}});
 		}
 
 		Request.WriteResponse(HttpResponseCode::Created);
@@ -492,8 +491,7 @@ HttpStructuredCacheService::HandlePutCacheRecord(zen::HttpServerRequest& Request
 
 		if (StoreUpstream && !IsPartialRecord)
 		{
-			ZEN_ASSERT(m_UpstreamCache);
-			m_UpstreamCache->EnqueueUpstream(
+			m_UpstreamCache.EnqueueUpstream(
 				{.Type = ZenContentType::kCbObject, .Key = {Ref.BucketSegment, Ref.HashKey}, .PayloadIds = std::move(ValidAttachments)});
 		}
 
@@ -574,8 +572,7 @@ HttpStructuredCacheService::HandlePutCacheRecord(zen::HttpServerRequest& Request
 
 		if (StoreUpstream && !IsPartialRecord)
 		{
-			ZEN_ASSERT(m_UpstreamCache);
-			m_UpstreamCache->EnqueueUpstream(
+			m_UpstreamCache.EnqueueUpstream(
 				{.Type = ZenContentType::kCbPackage, .Key = {Ref.BucketSegment, Ref.HashKey}, .PayloadIds = std::move(ValidAttachments)});
 		}
 
@@ -611,11 +608,11 @@ HttpStructuredCacheService::HandleGetCachePayload(zen::HttpServerRequest& Reques
 {
 	IoBuffer   Payload		   = m_CidStore.FindChunkByCid(Ref.PayloadId);
 	bool	   InUpstreamCache = false;
-	const bool QueryUpstream   = !Payload && m_UpstreamCache && (CachePolicy::QueryRemote == (Policy & CachePolicy::QueryRemote));
+	const bool QueryUpstream   = !Payload && (Policy & CachePolicy::QueryRemote) == CachePolicy::QueryRemote;
 
 	if (QueryUpstream)
 	{
-		if (auto UpstreamResult = m_UpstreamCache->GetCachePayload({Ref.BucketSegment, Ref.HashKey}, Ref.PayloadId); UpstreamResult.Success)
+		if (auto UpstreamResult = m_UpstreamCache.GetCachePayload({Ref.BucketSegment, Ref.HashKey}, Ref.PayloadId); UpstreamResult.Success)
 		{
 			if (CompressedBuffer Compressed = CompressedBuffer::FromCompressed(SharedBuffer(UpstreamResult.Value)))
 			{
@@ -825,7 +822,7 @@ HttpStructuredCacheService::HandleRpcGetCacheRecords(zen::HttpServerRequest& Req
 
 	const bool PartialOnError  = Policy.HasRecordPolicy(CachePolicy::PartialOnError);
 	const bool SkipAttachments = Policy.HasRecordPolicy(CachePolicy::SkipAttachments);
-	const bool QueryRemote	   = Policy.HasRecordPolicy(CachePolicy::QueryRemote) && m_UpstreamCache;
+	const bool QueryRemote	   = Policy.HasRecordPolicy(CachePolicy::QueryRemote);
 
 	for (CbFieldView KeyView : Params["CacheKeys"sv])
 	{
@@ -896,7 +893,7 @@ HttpStructuredCacheService::HandleRpcGetCacheRecords(zen::HttpServerRequest& Req
 		++KeyIndex;
 	}
 
-	if (!UpstreamRequests.empty() && m_UpstreamCache)
+	if (!UpstreamRequests.empty())
 	{
 		const auto OnCacheRecordGetComplete =
 			[this, &CacheValues, &RpcResponse, PartialOnError, SkipAttachments](CacheRecordGetCompleteParams&& Params) {
@@ -973,7 +970,7 @@ HttpStructuredCacheService::HandleRpcGetCacheRecords(zen::HttpServerRequest& Req
 				}
 			};
 
-		m_UpstreamCache->GetCacheRecords(CacheKeys, UpstreamRequests, Policy, std::move(OnCacheRecordGetComplete));
+		m_UpstreamCache.GetCacheRecords(CacheKeys, UpstreamRequests, Policy, std::move(OnCacheRecordGetComplete));
 	}
 
 	CbObjectWriter ResponseObject;
@@ -1156,7 +1153,7 @@ HttpStructuredCacheService::HandleRpcGetCachePayloads(zen::HttpServerRequest& Re
 		++RequestIndex;
 	}
 
-	if (!UpstreamRequests.empty() && m_UpstreamCache)
+	if (!UpstreamRequests.empty())
 	{
 		const auto OnCachePayloadGetComplete = [this, &Chunks](CachePayloadGetCompleteParams&& Params) {
 			if (CompressedBuffer Compressed = CompressedBuffer::FromCompressed(SharedBuffer(Params.Payload)))
@@ -1183,7 +1180,7 @@ HttpStructuredCacheService::HandleRpcGetCachePayloads(zen::HttpServerRequest& Re
 			}
 		};
 
-		m_UpstreamCache->GetCachePayloads(ChunkRequests, UpstreamRequests, std::move(OnCachePayloadGetComplete));
+		m_UpstreamCache.GetCachePayloads(ChunkRequests, UpstreamRequests, std::move(OnCachePayloadGetComplete));
 	}
 
 	CbPackage	   RpcResponse;
@@ -1242,13 +1239,9 @@ HttpStructuredCacheService::HandleStatsRequest(zen::HttpServerRequest& Request)
 	Cbo << "upstream_hits" << m_CacheStats.UpstreamHitCount;
 	Cbo << "upstream_ratio" << (HitCount > 0 ? (double(UpstreamHitCount) / double(HitCount)) : 0.0);
 	Cbo.EndObject();
-
-	if (m_UpstreamCache)
-	{
-		Cbo.BeginObject("upstream");
-		m_UpstreamCache->GetStatus(Cbo);
-		Cbo.EndObject();
-	}
+	Cbo.BeginObject("upstream");
+	m_UpstreamCache.GetStatus(Cbo);
+	Cbo.EndObject();
 
 	Cbo.BeginObject("cas");
 	Cbo.BeginObject("size");