per namespace/project cas prep refactor (#470)

- Refactor so we can have more than one cas store for project store and cache.
- Refactor `UpstreamCacheClient` so it is not tied to a specific CidStore
- Refactor scrub to keep the GC interface ScrubStorage function separate from scrub accessor functions (renamed to Scrub).
- Refactor storage size to keep GC interface StorageSize function separate from size accessor functions (renamed to TotalSize)
- Refactor cache storage so  `ZenCacheDiskLayer::CacheBucket` implements GcStorage interface rather than `ZenCacheNamespace`

1 files changed, 37 insertions, 17 deletions

diff --git a/src/zenstore/cache/cacherpc.cpp b/src/zenstore/cache/cacherpc.cpp
index 5d9a68919..ff21d1ede 100644
--- a/src/zenstore/cache/cacherpc.cpp
+++ b/src/zenstore/cache/cacherpc.cpp
@@ -153,13 +153,13 @@ CacheRpcHandler::CacheRpcHandler(LoggerRef				 InLog,
 								 CacheStats&			 InCacheStats,
 								 UpstreamCacheClient&	 InUpstreamCache,
 								 ZenCacheStore&			 InCacheStore,
-								 CidStore&				 InCidStore,
+								 GetCidStoreFunc&&		 InGetCidStore,
 								 const DiskWriteBlocker* InDiskWriteBlocker)
 : m_Log(InLog)
 , m_CacheStats(InCacheStats)
 , m_UpstreamCache(InUpstreamCache)
 , m_CacheStore(InCacheStore)
-, m_CidStore(InCidStore)
+, m_GetCidStore(std::move(InGetCidStore))
 , m_DiskWriteBlocker(InDiskWriteBlocker)
 {
 }
@@ -174,6 +174,12 @@ CacheRpcHandler::AreDiskWritesAllowed() const
 	return (m_DiskWriteBlocker == nullptr || m_DiskWriteBlocker->AreDiskWritesAllowed());
 }
 
+CidStore&
+CacheRpcHandler::GetCidStore(std::string_view Namespace)
+{
+	return m_GetCidStore(Namespace);
+}
+
 CacheRpcHandler::RpcResponseCode
 CacheRpcHandler::HandleRpcRequest(const CacheRequestContext& Context,
 								  std::string_view			 UriNamespace,
@@ -381,9 +387,12 @@ CacheRpcHandler::PutCacheRecord(PutRequestData& Request, const CbPackage* Packag
 
 	Stopwatch Timer;
 
+	CidStore& ChunkStore = m_GetCidStore(Request.Namespace);
+
 	Request.RecordObject.IterateAttachments([this,
 											 &Request,
 											 Package,
+											 &ChunkStore,
 											 &WriteAttachmentBuffers,
 											 &WriteRawHashes,
 											 &ValidAttachments,
@@ -412,7 +421,7 @@ CacheRpcHandler::PutCacheRecord(PutRequestData& Request, const CbPackage* Packag
 				Count.Invalid++;
 			}
 		}
-		else if (m_CidStore.ContainsChunk(ValueHash))
+		else if (ChunkStore.ContainsChunk(ValueHash))
 		{
 			ValidAttachments.emplace_back(ValueHash);
 			Count.Valid++;
@@ -448,7 +457,7 @@ CacheRpcHandler::PutCacheRecord(PutRequestData& Request, const CbPackage* Packag
 
 	if (!WriteAttachmentBuffers.empty())
 	{
-		std::vector<CidStore::InsertResult> InsertResults = m_CidStore.AddChunks(WriteAttachmentBuffers, WriteRawHashes);
+		std::vector<CidStore::InsertResult> InsertResults = ChunkStore.AddChunks(WriteAttachmentBuffers, WriteRawHashes);
 		for (size_t Index = 0; Index < InsertResults.size(); Index++)
 		{
 			if (InsertResults[Index].New)
@@ -475,10 +484,12 @@ CacheRpcHandler::PutCacheRecord(PutRequestData& Request, const CbPackage* Packag
 
 	if (HasUpstream && EnumHasAllFlags(Request.Policy.GetRecordPolicy(), CachePolicy::StoreRemote) && !IsPartialRecord)
 	{
-		m_UpstreamCache.EnqueueUpstream({.Type			  = ZenContentType::kCbPackage,
-										 .Namespace		  = Request.Namespace,
-										 .Key			  = Request.Key,
-										 .ValueContentIds = std::move(ValidAttachments)});
+		m_UpstreamCache.EnqueueUpstream(
+			{.Type			  = ZenContentType::kCbPackage,
+			 .Namespace		  = Request.Namespace,
+			 .Key			  = Request.Key,
+			 .ValueContentIds = std::move(ValidAttachments)},
+			[ChunkStore = &ChunkStore](const IoHash& ValueHash) { return ChunkStore->FindChunkByCid(ValueHash); });
 	}
 	return PutStatus::Success;
 }
@@ -521,6 +532,8 @@ CacheRpcHandler::HandleRpcGetCacheRecords(const CacheRequestContext& Context, Cb
 		return CbPackage{};
 	}
 
+	CidStore& ChunkStore = m_GetCidStore(Namespace.value());
+
 	const bool HasUpstream = m_UpstreamCache.IsActive();
 
 	eastl::fixed_vector<RecordRequestData, 16> Requests;
@@ -620,7 +633,7 @@ CacheRpcHandler::HandleRpcGetCacheRecords(const CacheRequestContext& Context, Cb
 					}
 					else if (EnumHasAllFlags(ValuePolicy, CachePolicy::SkipData))
 					{
-						if (m_CidStore.ContainsChunk(Value.ContentId))
+						if (ChunkStore.ContainsChunk(Value.ContentId))
 						{
 							Value.Exists = true;
 						}
@@ -641,7 +654,7 @@ CacheRpcHandler::HandleRpcGetCacheRecords(const CacheRequestContext& Context, Cb
 					}
 					else
 					{
-						if (IoBuffer Chunk = m_CidStore.FindChunkByCid(Value.ContentId))
+						if (IoBuffer Chunk = ChunkStore.FindChunkByCid(Value.ContentId))
 						{
 							if (Chunk.GetSize() > 0)
 							{
@@ -665,7 +678,7 @@ CacheRpcHandler::HandleRpcGetCacheRecords(const CacheRequestContext& Context, Cb
 				}
 				if (!RequestValueIndexes.empty())
 				{
-					m_CidStore.IterateChunks(
+					ChunkStore.IterateChunks(
 						CidHashes,
 						[this, &Request, &RequestValueIndexes](size_t Index, const IoBuffer& Payload) -> bool {
 							try
@@ -758,7 +771,7 @@ CacheRpcHandler::HandleRpcGetCacheRecords(const CacheRequestContext& Context, Cb
 			}
 		}
 
-		const auto OnCacheRecordGetComplete = [this, Namespace, &ParseValues, Context](CacheRecordGetCompleteParams&& Params) {
+		const auto OnCacheRecordGetComplete = [this, Namespace, &ChunkStore, &ParseValues, Context](CacheRecordGetCompleteParams&& Params) {
 			if (!Params.Record)
 			{
 				return;
@@ -827,7 +840,7 @@ CacheRpcHandler::HandleRpcGetCacheRecords(const CacheRequestContext& Context, Cb
 							Value.Exists   = true;
 							if (StoreLocal)
 							{
-								m_CidStore.AddChunk(Compressed.GetCompressed().Flatten().AsIoBuffer(), Attachment->GetHash());
+								ChunkStore.AddChunk(Compressed.GetCompressed().Flatten().AsIoBuffer(), Attachment->GetHash());
 							}
 							if (!EnumHasAllFlags(ValuePolicy, CachePolicy::SkipData))
 							{
@@ -944,6 +957,8 @@ CacheRpcHandler::HandleRpcPutCacheValues(const CacheRequestContext& Context, con
 	const bool	HasUpstream	  = m_UpstreamCache.IsActive();
 	CbArrayView RequestsArray = Params["Requests"sv].AsArrayView();
 
+	CidStore& ChunkStore = m_GetCidStore(Namespace.value());
+
 	std::vector<ZenCacheStore::PutResult>			  BatchResults;
 	eastl::fixed_vector<size_t, 32>					  BatchResultIndexes;
 	eastl::fixed_vector<ZenCacheStore::PutResult, 32> Results;
@@ -1094,7 +1109,8 @@ CacheRpcHandler::HandleRpcPutCacheValues(const CacheRequestContext& Context, con
 		if ((Results[Index].Status == zen::PutStatus::Success) && UpstreamCacheKeys[Index] != CacheKey::Empty)
 		{
 			m_UpstreamCache.EnqueueUpstream(
-				{.Type = ZenContentType::kCompressedBinary, .Namespace = *Namespace, .Key = UpstreamCacheKeys[Index]});
+				{.Type = ZenContentType::kCompressedBinary, .Namespace = *Namespace, .Key = UpstreamCacheKeys[Index]},
+				[ChunkStore = &ChunkStore](const IoHash& ValueHash) { return ChunkStore->FindChunkByCid(ValueHash); });
 		}
 	}
 	{
@@ -1546,6 +1562,8 @@ CacheRpcHandler::GetLocalCacheRecords(const CacheRequestContext&				 Context,
 	using namespace cache::detail;
 	const bool HasUpstream = m_UpstreamCache.IsActive();
 
+	CidStore& ChunkStore = m_GetCidStore(Namespace);
+
 	// TODO: BatchGet records?
 	std::vector<CacheKeyRequest*> UpstreamRecordRequests;
 	for (size_t RecordIndex = 0; RecordIndex < Records.size(); ++RecordIndex)
@@ -1684,12 +1702,12 @@ CacheRpcHandler::GetLocalCacheRecords(const CacheRequestContext&				 Context,
 			{
 				if (EnumHasAllFlags(Request->DownstreamPolicy, CachePolicy::SkipData) && Request->RawSizeKnown)
 				{
-					if (m_CidStore.ContainsChunk(Request->Key->ChunkId))
+					if (ChunkStore.ContainsChunk(Request->Key->ChunkId))
 					{
 						Request->Exists = true;
 					}
 				}
-				else if (IoBuffer Payload = m_CidStore.FindChunkByCid(Request->Key->ChunkId))
+				else if (IoBuffer Payload = ChunkStore.FindChunkByCid(Request->Key->ChunkId))
 				{
 					if (!EnumHasAllFlags(Request->DownstreamPolicy, CachePolicy::SkipData))
 					{
@@ -1822,6 +1840,8 @@ CacheRpcHandler::GetUpstreamCacheChunks(const CacheRequestContext&				  Context,
 			return;
 		}
 
+		CidStore& ChunkStore = m_GetCidStore(Namespace);
+
 		CacheChunkRequest& Key			= Params.Request;
 		size_t			   RequestIndex = std::distance(RequestKeys.data(), &Key);
 		ChunkRequest&	   Request		= Requests[RequestIndex];
@@ -1841,7 +1861,7 @@ CacheRpcHandler::GetUpstreamCacheChunks(const CacheRequestContext&				  Context,
 				bool Overwrite = !EnumHasAllFlags(Request.DownstreamPolicy, CachePolicy::QueryLocal);
 				if (Request.IsRecordRequest)
 				{
-					m_CidStore.AddChunk(Params.Value, Params.RawHash);
+					ChunkStore.AddChunk(Params.Value, Params.RawHash);
 				}
 				else
 				{