2 files changed, 97 insertions, 38 deletions

diff --git a/src/zenserver/storage/cache/httpstructuredcache.cpp b/src/zenserver/storage/cache/httpstructuredcache.cpp
index 72f29d14e..4d3673e70 100644
--- a/src/zenserver/storage/cache/httpstructuredcache.cpp
+++ b/src/zenserver/storage/cache/httpstructuredcache.cpp
@@ -80,7 +80,8 @@ HttpStructuredCacheService::HttpStructuredCacheService(ZenCacheStore&		   InCach
 													   HttpStatusService&	   StatusService,
 													   UpstreamCache&		   UpstreamCache,
 													   const DiskWriteBlocker* InDiskWriteBlocker,
-													   OpenProcessCache&	   InOpenProcessCache)
+													   OpenProcessCache&	   InOpenProcessCache,
+													   const ILocalRefPolicy*  InLocalRefPolicy)
 : m_Log(logging::Get("cache"))
 , m_CacheStore(InCacheStore)
 , m_StatsService(StatsService)
@@ -90,6 +91,7 @@ HttpStructuredCacheService::HttpStructuredCacheService(ZenCacheStore&		   InCach
 , m_DiskWriteBlocker(InDiskWriteBlocker)
 , m_OpenProcessCache(InOpenProcessCache)
 , m_RpcHandler(m_Log, m_CacheStats, UpstreamCache, InCacheStore, InCidStore, InDiskWriteBlocker)
+, m_LocalRefPolicy(InLocalRefPolicy)
 {
 	m_StatsService.RegisterHandler("z$", *this);
 	m_StatusService.RegisterHandler("z$", *this);
@@ -114,6 +116,18 @@ HttpStructuredCacheService::BaseUri() const
 	return "/z$/";
 }
 
+bool
+HttpStructuredCacheService::AcceptsLocalFileReferences() const
+{
+	return true;
+}
+
+const ILocalRefPolicy*
+HttpStructuredCacheService::GetLocalRefPolicy() const
+{
+	return m_LocalRefPolicy;
+}
+
 void
 HttpStructuredCacheService::Flush()
 {
@@ -423,19 +437,22 @@ HttpStructuredCacheService::HandleRequest(HttpServerRequest& Request)
 
 		std::string RecordPath = UrlDecode(Params.GetValue("path"));
 
-		uint32_t ThreadCount = GetHardwareConcurrency();
+		const uint32_t HardwareConcurrency = GetHardwareConcurrency();
+		const uint32_t MaxThreadCount	   = std::max<uint32_t>(HardwareConcurrency, 16u);
+		uint32_t	   ThreadCount		   = HardwareConcurrency;
 		if (auto Param = Params.GetValue("thread_count"); Param.empty() == false)
 		{
 			if (auto Value = ParseInt<uint64_t>(Param))
 			{
-				ThreadCount = gsl::narrow<uint32_t>(Value.value());
+				ThreadCount = gsl::narrow_cast<uint32_t>(std::min<uint64_t>(Value.value(), MaxThreadCount));
 			}
 		}
+		ThreadCount = std::clamp<uint32_t>(ThreadCount, 1u, MaxThreadCount);
 
 		ZEN_INFO("initiating cache RPC replay using {} threads, from '{}'", ThreadCount, RecordPath);
 
 		std::unique_ptr<cache::IRpcRequestReplayer> Replayer(cache::MakeDiskRequestReplayer(RecordPath, false));
-		ReplayRequestRecorder(RequestContext, *Replayer, ThreadCount < 1 ? 1 : ThreadCount);
+		ReplayRequestRecorder(RequestContext, *Replayer, ThreadCount);
 
 		ZEN_INFO("cache RPC replay COMPLETED");
 
@@ -543,7 +560,7 @@ HttpStructuredCacheService::HandleCacheRequest(HttpServerRequest& Request)
 			break;
 		default:
 			m_CacheStats.BadRequestCount++;
-			break;
+			return Request.WriteResponse(HttpResponseCode::MethodNotAllowed);
 	}
 }
 
@@ -654,7 +671,7 @@ HttpStructuredCacheService::HandleCacheNamespaceRequest(HttpServerRequest& Reque
 					auto NewEnd = std::unique(AllAttachments.begin(), AllAttachments.end());
 					AllAttachments.erase(NewEnd, AllAttachments.end());
 
-					uint64_t AttachmentsSize = 0;
+					std::atomic<uint64_t> AttachmentsSize = 0;
 
 					m_CidStore.IterateChunks(
 						AllAttachments,
@@ -693,7 +710,8 @@ HttpStructuredCacheService::HandleCacheNamespaceRequest(HttpServerRequest& Reque
 			break;
 
 		default:
-			break;
+			m_CacheStats.BadRequestCount++;
+			return Request.WriteResponse(HttpResponseCode::MethodNotAllowed);
 	}
 }
 
@@ -746,7 +764,7 @@ HttpStructuredCacheService::HandleCacheBucketRequest(HttpServerRequest& Request,
 						ResponseWriter << "Size" << ValuesSize;
 						ResponseWriter << "AttachmentCount" << ContentStats.Attachments.size();
 
-						uint64_t AttachmentsSize = 0;
+						std::atomic<uint64_t> AttachmentsSize = 0;
 
 						WorkerThreadPool& WorkerPool = GetMediumWorkerPool(EWorkloadType::Background);
 
@@ -783,7 +801,8 @@ HttpStructuredCacheService::HandleCacheBucketRequest(HttpServerRequest& Request,
 			break;
 
 		default:
-			break;
+			m_CacheStats.BadRequestCount++;
+			return Request.WriteResponse(HttpResponseCode::MethodNotAllowed);
 	}
 }
 
@@ -802,7 +821,8 @@ HttpStructuredCacheService::HandleCacheRecordRequest(HttpServerRequest& Request,
 			break;
 
 		default:
-			break;
+			m_CacheStats.BadRequestCount++;
+			return Request.WriteResponse(HttpResponseCode::MethodNotAllowed);
 	}
 }
 
@@ -1202,8 +1222,6 @@ HttpStructuredCacheService::HandlePutCacheRecord(HttpServerRequest& Request, con
 	}
 
 	auto WriteFailureResponse = [&Request](const ZenCacheStore::PutResult& PutResult) {
-		ZEN_UNUSED(PutResult);
-
 		HttpResponseCode ResponseCode = HttpResponseCode::InternalServerError;
 		switch (PutResult.Status)
 		{
@@ -1217,7 +1235,7 @@ HttpStructuredCacheService::HandlePutCacheRecord(HttpServerRequest& Request, con
 
 		if (PutResult.Details)
 		{
-			Request.WriteResponse(ResponseCode, PutResult.Details);
+			return Request.WriteResponse(ResponseCode, PutResult.Details);
 		}
 		return Request.WriteResponse(ResponseCode);
 	};
@@ -1493,7 +1511,8 @@ HttpStructuredCacheService::HandleCacheChunkRequest(HttpServerRequest& Request,
 			HandlePutCacheChunk(Request, Ref, PolicyFromUrl);
 			break;
 		default:
-			break;
+			m_CacheStats.BadRequestCount++;
+			return Request.WriteResponse(HttpResponseCode::MethodNotAllowed);
 	}
 }
 
@@ -1828,10 +1847,23 @@ HttpStructuredCacheService::HandleRpcRequest(HttpServerRequest& Request, std::st
 }
 
 void
+HttpStructuredCacheService::HandleStatusRequest(HttpServerRequest& Request)
+{
+	CbObjectWriter Cbo;
+	Cbo << "ok" << true;
+	Request.WriteResponse(HttpResponseCode::OK, Cbo.Save());
+}
+
+void
 HttpStructuredCacheService::HandleStatsRequest(HttpServerRequest& Request)
 {
 	ZEN_MEMSCOPE(GetCacheHttpTag());
 
+	bool ShowCidStoreStats	 = Request.GetQueryParams().GetValue("cidstorestats") == "true";
+	bool ShowCacheStoreStats = Request.GetQueryParams().GetValue("cachestorestats") == "true";
+
+	// Full stats with optional detailed store/cid breakdowns
+
 	CbObjectWriter Cbo;
 
 	EmitSnapshot("requests", m_HttpRequests, Cbo);
@@ -1858,13 +1890,10 @@ HttpStructuredCacheService::HandleStatsRequest(HttpServerRequest& Request)
 	const CidStoreSize	 CidSize   = m_CidStore.TotalSize();
 	const CacheStoreSize CacheSize = m_CacheStore.TotalSize();
 
-	bool ShowCidStoreStats	 = Request.GetQueryParams().GetValue("cidstorestats") == "true";
-	bool ShowCacheStoreStats = Request.GetQueryParams().GetValue("cachestorestats") == "true";
-
-	CidStoreStats CidStoreStats = {};
+	CidStoreStats DetailedCidStoreStats = {};
 	if (ShowCidStoreStats)
 	{
-		CidStoreStats = m_CidStore.Stats();
+		DetailedCidStoreStats = m_CidStore.Stats();
 	}
 	ZenCacheStore::CacheStoreStats CacheStoreStats = {};
 	if (ShowCacheStoreStats)
@@ -1906,8 +1935,6 @@ HttpStructuredCacheService::HandleStatsRequest(HttpServerRequest& Request)
 		{
 			Cbo << "upstream_ratio" << (HitCount > 0 ? (double(UpstreamHitCount) / double(HitCount)) : 0.0);
 			Cbo << "upstream_hits" << m_CacheStats.UpstreamHitCount;
-			Cbo << "upstream_ratio" << (HitCount > 0 ? (double(UpstreamHitCount) / double(HitCount)) : 0.0);
-			Cbo << "upstream_ratio" << (HitCount > 0 ? (double(UpstreamHitCount) / double(HitCount)) : 0.0);
 		}
 
 		Cbo << "cidhits" << ChunkHitCount << "cidmisses" << ChunkMissCount << "cidwrites" << ChunkWriteCount;
@@ -2002,8 +2029,8 @@ HttpStructuredCacheService::HandleStatsRequest(HttpServerRequest& Request)
 			}
 			Cbo.EndObject();
 		}
-		Cbo.EndObject();
 	}
+	Cbo.EndObject();
 
 	if (m_UpstreamCache.IsActive())
 	{
@@ -2029,10 +2056,10 @@ HttpStructuredCacheService::HandleStatsRequest(HttpServerRequest& Request)
 		if (ShowCidStoreStats)
 		{
 			Cbo.BeginObject("store");
-			Cbo << "hits" << CidStoreStats.HitCount << "misses" << CidStoreStats.MissCount << "writes" << CidStoreStats.WriteCount;
-			EmitSnapshot("read", CidStoreStats.FindChunkOps, Cbo);
-			EmitSnapshot("write", CidStoreStats.AddChunkOps, Cbo);
-			//		EmitSnapshot("exists", CidStoreStats.ContainChunkOps, Cbo);
+			Cbo << "hits" << DetailedCidStoreStats.HitCount << "misses" << DetailedCidStoreStats.MissCount << "writes"
+				<< DetailedCidStoreStats.WriteCount;
+			EmitSnapshot("read", DetailedCidStoreStats.FindChunkOps, Cbo);
+			EmitSnapshot("write", DetailedCidStoreStats.AddChunkOps, Cbo);
 			Cbo.EndObject();
 		}
 	}
@@ -2041,12 +2068,38 @@ HttpStructuredCacheService::HandleStatsRequest(HttpServerRequest& Request)
 	Request.WriteResponse(HttpResponseCode::OK, Cbo.Save());
 }
 
-void
-HttpStructuredCacheService::HandleStatusRequest(HttpServerRequest& Request)
+CbObject
+HttpStructuredCacheService::CollectStats()
 {
+	ZEN_TRACE_CPU("HttpStructuredCacheService::Stats");
+	ZEN_MEMSCOPE(GetCacheHttpTag());
+
 	CbObjectWriter Cbo;
-	Cbo << "ok" << true;
-	Request.WriteResponse(HttpResponseCode::OK, Cbo.Save());
+
+	EmitSnapshot("requests", m_HttpRequests, Cbo);
+
+	const CacheStoreSize CacheSize = m_CacheStore.TotalSize();
+
+	Cbo.BeginObject("cache");
+	{
+		Cbo.BeginObject("size");
+		{
+			Cbo << "disk" << CacheSize.DiskSize;
+			Cbo << "memory" << CacheSize.MemorySize;
+		}
+		Cbo.EndObject();
+
+		Cbo << "hits" << m_CacheStats.HitCount << "misses" << m_CacheStats.MissCount;
+	}
+	Cbo.EndObject();
+
+	return Cbo.Save();
+}
+
+uint64_t
+HttpStructuredCacheService::GetActivityCounter()
+{
+	return m_HttpRequests.Count();
 }
 
 bool
diff --git a/src/zenserver/storage/cache/httpstructuredcache.h b/src/zenserver/storage/cache/httpstructuredcache.h
index 5a795c215..f606126d6 100644
--- a/src/zenserver/storage/cache/httpstructuredcache.h
+++ b/src/zenserver/storage/cache/httpstructuredcache.h
@@ -76,11 +76,14 @@ public:
 							   HttpStatusService&	   StatusService,
 							   UpstreamCache&		   UpstreamCache,
 							   const DiskWriteBlocker* InDiskWriteBlocker,
-							   OpenProcessCache&	   InOpenProcessCache);
+							   OpenProcessCache&	   InOpenProcessCache,
+							   const ILocalRefPolicy*  InLocalRefPolicy = nullptr);
 	~HttpStructuredCacheService();
 
-	virtual const char* BaseUri() const override;
-	virtual void		HandleRequest(HttpServerRequest& Request) override;
+	virtual const char*			   BaseUri() const override;
+	virtual void				   HandleRequest(HttpServerRequest& Request) override;
+	virtual bool				   AcceptsLocalFileReferences() const override;
+	virtual const ILocalRefPolicy* GetLocalRefPolicy() const override;
 
 	void Flush();
 
@@ -102,11 +105,13 @@ private:
 	void HandleRpcRequest(HttpServerRequest& Request, std::string_view UriNamespace);
 	void HandleDetailsRequest(HttpServerRequest& Request);
 
-	void		 HandleCacheRequest(HttpServerRequest& Request);
-	void		 HandleCacheNamespaceRequest(HttpServerRequest& Request, std::string_view Namespace);
-	void		 HandleCacheBucketRequest(HttpServerRequest& Request, std::string_view Namespace, std::string_view Bucket);
-	virtual void HandleStatsRequest(HttpServerRequest& Request) override;
-	virtual void HandleStatusRequest(HttpServerRequest& Request) override;
+	void			 HandleCacheRequest(HttpServerRequest& Request);
+	void			 HandleCacheNamespaceRequest(HttpServerRequest& Request, std::string_view Namespace);
+	void			 HandleCacheBucketRequest(HttpServerRequest& Request, std::string_view Namespace, std::string_view Bucket);
+	virtual void	 HandleStatusRequest(HttpServerRequest& Request) override;
+	virtual void	 HandleStatsRequest(HttpServerRequest& Request) override;
+	virtual CbObject CollectStats() override;
+	virtual uint64_t GetActivityCounter() override;
 
 	bool AreDiskWritesAllowed() const;
 
@@ -123,6 +128,7 @@ private:
 	const DiskWriteBlocker*	 m_DiskWriteBlocker = nullptr;
 	OpenProcessCache&		 m_OpenProcessCache;
 	CacheRpcHandler			 m_RpcHandler;
+	const ILocalRefPolicy*	 m_LocalRefPolicy = nullptr;
 
 	void ReplayRequestRecorder(const CacheRequestContext& Context, cache::IRpcRequestReplayer& Replayer, uint32_t ThreadCount);