path: root/src/zenremotestore/builds/buildstoragecache.cpp

// Copyright Epic Games, Inc. All Rights Reserved.

#include <zenremotestore/builds/buildstoragecache.h>

#include <zencore/compactbinarybuilder.h>
#include <zencore/compactbinaryvalidation.h>
#include <zencore/fmtutils.h>
#include <zencore/scopeguard.h>
#include <zencore/timer.h>
#include <zencore/trace.h>
#include <zencore/workthreadpool.h>
#include <zenhttp/httpclient.h>
#include <zenhttp/packageformat.h>

ZEN_THIRD_PARTY_INCLUDES_START
#include <tsl/robin_map.h>
#include <tsl/robin_set.h>
ZEN_THIRD_PARTY_INCLUDES_END

namespace zen {

using namespace std::literals;

class ZenBuildStorageCache : public BuildStorageCache
{
public:
	explicit ZenBuildStorageCache(HttpClient&					 HttpClient,
								  BuildStorageCache::Statistics& Stats,
								  std::string_view				 Namespace,
								  std::string_view				 Bucket,
								  const std::filesystem::path&	 TempFolderPath,
								  WorkerThreadPool&				 BackgroundWorkerPool)
	: m_HttpClient(HttpClient)
	, m_Stats(Stats)
	, m_Namespace(Namespace.empty() ? "none" : Namespace)
	, m_Bucket(Bucket.empty() ? "none" : Bucket)
	, m_TempFolderPath(std::filesystem::path(TempFolderPath).make_preferred())
	, m_BackgroundWorkPool(BackgroundWorkerPool)
	, m_PendingBackgroundWorkCount(1)
	, m_CancelBackgroundWork(false)
	{
	}

	virtual ~ZenBuildStorageCache()
	{
		try
		{
			m_CancelBackgroundWork.store(true);
			if (!IsFlushed)
			{
				m_PendingBackgroundWorkCount.CountDown();
				m_PendingBackgroundWorkCount.Wait();
			}
		}
		catch (const std::exception& Ex)
		{
			ZEN_ERROR("~ZenBuildStorageCache() failed with: {}", Ex.what());
		}
	}

	void ScheduleBackgroundWork(std::function<void()>&& Work)
	{
		m_PendingBackgroundWorkCount.AddCount(1);
		try
		{
			m_BackgroundWorkPool.ScheduleWork(
				[this, Work = std::move(Work)]() {
					ZEN_TRACE_CPU("ZenBuildStorageCache::BackgroundWork");
					auto _ = MakeGuard([this]() { m_PendingBackgroundWorkCount.CountDown(); });
					if (!m_CancelBackgroundWork)
					{
						try
						{
							Work();
						}
						catch (const std::exception& Ex)
						{
							ZEN_ERROR("Failed executing background upload to build cache. Reason: {}", Ex.what());
						}
					}
				},
				WorkerThreadPool::EMode::EnableBacklog);
		}
		catch (const std::exception& Ex)
		{
			m_PendingBackgroundWorkCount.CountDown();
			ZEN_ERROR("Failed scheduling background upload to build cache. Reason: {}", Ex.what());
		}
	}

	virtual void PutBuildBlob(const Oid&			 BuildId,
							  const IoHash&			 RawHash,
							  ZenContentType		 ContentType,
							  const CompositeBuffer& Payload) override
	{
		ZEN_ASSERT(!IsFlushed);
		ZEN_ASSERT(ContentType == ZenContentType::kCompressedBinary);

		// Move all segments in Payload to be file handle based so if Payload is materialized it does not affect buffers in queue
		std::vector<SharedBuffer>	  FileBasedSegments;
		std::span<const SharedBuffer> Segments = Payload.GetSegments();
		FileBasedSegments.reserve(Segments.size());
		{
			tsl::robin_map<void*, std::filesystem::path> HandleToPath;
			for (const SharedBuffer& Segment : Segments)
			{
				std::filesystem::path FilePath;
				IoBufferFileReference Ref;
				if (Segment.AsIoBuffer().GetFileReference(Ref))
				{
					if (auto It = HandleToPath.find(Ref.FileHandle); It != HandleToPath.end())
					{
						FilePath = It->second;
					}
					else
					{
						std::error_code		  Ec;
						std::filesystem::path Path = PathFromHandle(Ref.FileHandle, Ec);
						if (!Ec && !Path.empty())
						{
							HandleToPath.insert_or_assign(Ref.FileHandle, Path);
							FilePath = std::move(Path);
						}
					}
				}

				if (!FilePath.empty())
				{
					IoBuffer BufferFromFile = IoBufferBuilder::MakeFromFile(FilePath, Ref.FileChunkOffset, Ref.FileChunkSize);
					if (BufferFromFile)
					{
						FileBasedSegments.push_back(SharedBuffer(std::move(BufferFromFile)));
					}
					else
					{
						FileBasedSegments.push_back(Segment);
					}
				}
				else
				{
					FileBasedSegments.push_back(Segment);
				}
			}
		}

		CompositeBuffer FilePayload(std::move(FileBasedSegments));

		ScheduleBackgroundWork([this, BuildId = Oid(BuildId), RawHash = IoHash(RawHash), ContentType, Payload = std::move(FilePayload)]() {
			ZEN_TRACE_CPU("ZenBuildStorageCache::PutBuildBlob");
			Stopwatch ExecutionTimer;
			auto	  _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });

			HttpClient::Response CacheResponse =
				m_HttpClient.Upload(fmt::format("/builds/{}/{}/{}/blobs/{}", m_Namespace, m_Bucket, BuildId, RawHash),
									Payload,
									ContentType);

			m_Stats.PutBlobCount++;
			m_Stats.PutBlobByteCount += Payload.GetSize();

			AddStatistic(CacheResponse);
			if (!CacheResponse.IsSuccess())
			{
				ZEN_DEBUG("Failed posting blob to cache: {}", CacheResponse.ErrorMessage(""sv));
			}
		});
	}

	virtual IoBuffer GetBuildBlob(const Oid& BuildId, const IoHash& RawHash, uint64_t RangeOffset, uint64_t RangeBytes) override
	{
		ZEN_TRACE_CPU("ZenBuildStorageCache::GetBuildBlob");

		Stopwatch ExecutionTimer;
		auto	  _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });

		HttpClient::KeyValueMap Headers;
		if (RangeOffset != 0 || RangeBytes != (uint64_t)-1)
		{
			Headers.Entries.insert({"Range", fmt::format("bytes={}-{}", RangeOffset, RangeOffset + RangeBytes - 1)});
		}
		CreateDirectories(m_TempFolderPath);
		HttpClient::Response CacheResponse =
			m_HttpClient.Download(fmt::format("/builds/{}/{}/{}/blobs/{}", m_Namespace, m_Bucket, BuildId, RawHash),
								  m_TempFolderPath,
								  Headers);
		AddStatistic(CacheResponse);
		if (CacheResponse.IsSuccess())
		{
			return CacheResponse.ResponsePayload;
		}
		return {};
	}

	virtual BuildBlobRanges GetBuildBlobRanges(const Oid&									  BuildId,
											   const IoHash&								  RawHash,
											   std::span<const std::pair<uint64_t, uint64_t>> Ranges) override
	{
		ZEN_TRACE_CPU("ZenBuildStorageCache::GetBuildBlobRanges");

		Stopwatch ExecutionTimer;
		auto	  _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });

		CbObjectWriter Writer;
		Writer.BeginArray("ranges"sv);
		{
			for (const std::pair<uint64_t, uint64_t>& Range : Ranges)
			{
				Writer.BeginObject();
				{
					Writer.AddInteger("offset"sv, Range.first);
					Writer.AddInteger("length"sv, Range.second);
				}
				Writer.EndObject();
			}
		}
		Writer.EndArray();	// ranges

		CreateDirectories(m_TempFolderPath);
		HttpClient::Response CacheResponse =
			m_HttpClient.Post(fmt::format("/builds/{}/{}/{}/blobs/{}", m_Namespace, m_Bucket, BuildId, RawHash),
							  Writer.Save(),
							  HttpClient::Accept(ZenContentType::kCbPackage));
		AddStatistic(CacheResponse);
		if (CacheResponse.IsSuccess())
		{
			CbPackage	 ResponsePackage = ParsePackageMessage(CacheResponse.ResponsePayload);
			CbObjectView ResponseObject	 = ResponsePackage.GetObject();

			CbArrayView RangeArray = ResponseObject["ranges"sv].AsArrayView();

			std::vector<std::pair<uint64_t, uint64_t>> ReceivedRanges;
			ReceivedRanges.reserve(RangeArray.Num());

			uint64_t OffsetInPayloadRanges = 0;

			for (CbFieldView View : RangeArray)
			{
				CbObjectView RangeView = View.AsObjectView();
				uint64_t	 Offset	   = RangeView["offset"sv].AsUInt64();
				uint64_t	 Length	   = RangeView["length"sv].AsUInt64();

				const std::pair<uint64_t, uint64_t>& Range = Ranges[ReceivedRanges.size()];

				if (Offset != Range.first || Length != Range.second)
				{
					return {};
				}
				ReceivedRanges.push_back(std::make_pair(OffsetInPayloadRanges, Length));
				OffsetInPayloadRanges += Length;
			}

			const CbAttachment* DataAttachment = ResponsePackage.FindAttachment(RawHash);
			if (DataAttachment)
			{
				SharedBuffer PayloadRanges = DataAttachment->AsBinary();
				return BuildBlobRanges{.PayloadBuffer = PayloadRanges.AsIoBuffer(), .Ranges = std::move(ReceivedRanges)};
			}
		}
		return {};
	}

	virtual void PutBlobMetadatas(const Oid& BuildId, std::span<const IoHash> BlobHashes, std::span<const CbObject> MetaDatas) override
	{
		ZEN_ASSERT(!IsFlushed);
		ScheduleBackgroundWork([this,
								BuildId		  = Oid(BuildId),
								BlobRawHashes = std::vector<IoHash>(BlobHashes.begin(), BlobHashes.end()),
								MetaDatas	  = std::vector<CbObject>(MetaDatas.begin(), MetaDatas.end())]() {
			ZEN_TRACE_CPU("ZenBuildStorageCache::PutBlobMetadatas");

			Stopwatch ExecutionTimer;
			auto	  _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });

			const uint64_t BlobCount = BlobRawHashes.size();

			CbPackage							   RequestPackage;
			std::vector<CbAttachment>			   Attachments;
			tsl::robin_set<IoHash, IoHash::Hasher> AttachmentHashes;
			Attachments.reserve(BlobCount);
			AttachmentHashes.reserve(BlobCount);
			{
				CbObjectWriter RequestWriter;
				RequestWriter.BeginArray("blobHashes");
				for (size_t BlockHashIndex = 0; BlockHashIndex < BlobRawHashes.size(); BlockHashIndex++)
				{
					RequestWriter.AddHash(BlobRawHashes[BlockHashIndex]);
				}
				RequestWriter.EndArray();  // blobHashes

				RequestWriter.BeginArray("metadatas");
				for (size_t BlockHashIndex = 0; BlockHashIndex < BlobRawHashes.size(); BlockHashIndex++)
				{
					const IoHash ObjectHash = MetaDatas[BlockHashIndex].GetHash();
					RequestWriter.AddBinaryAttachment(ObjectHash);
					if (!AttachmentHashes.contains(ObjectHash))
					{
						Attachments.push_back(CbAttachment(MetaDatas[BlockHashIndex], ObjectHash));
						AttachmentHashes.insert(ObjectHash);
					}
				}

				RequestWriter.EndArray();  // metadatas

				RequestPackage.SetObject(RequestWriter.Save());
			}
			RequestPackage.AddAttachments(Attachments);

			CompositeBuffer RpcRequestBuffer = FormatPackageMessageBuffer(RequestPackage);

			HttpClient::Response CacheResponse =
				m_HttpClient.Post(fmt::format("/builds/{}/{}/{}/blobs/putBlobMetadata", m_Namespace, m_Bucket, BuildId),
								  RpcRequestBuffer,
								  ZenContentType::kCbPackage);
			AddStatistic(CacheResponse);
			if (!CacheResponse.IsSuccess())
			{
				ZEN_DEBUG("Failed posting blob metadata to cache: {}", CacheResponse.ErrorMessage(""sv));
			}
		});
	}

	virtual std::vector<CbObject> GetBlobMetadatas(const Oid& BuildId, std::span<const IoHash> BlobHashes) override
	{
		ZEN_TRACE_CPU("ZenBuildStorageCache::GetBlobMetadatas");

		Stopwatch ExecutionTimer;
		auto	  _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });

		CbObjectWriter Request;

		Request.BeginArray("blobHashes"sv);
		for (const IoHash& BlobHash : BlobHashes)
		{
			Request.AddHash(BlobHash);
		}
		Request.EndArray();

		IoBuffer Payload = Request.Save().GetBuffer().AsIoBuffer();
		Payload.SetContentType(ZenContentType::kCbObject);

		HttpClient::Response Response =
			m_HttpClient.Post(fmt::format("/builds/{}/{}/{}/blobs/getBlobMetadata", m_Namespace, m_Bucket, BuildId),
							  Payload,
							  HttpClient::Accept(ZenContentType::kCbObject));
		AddStatistic(Response);
		if (Response.IsSuccess())
		{
			std::vector<CbObject> Result;

			CbPackage ResponsePackage = ParsePackageMessage(Response.ResponsePayload);
			CbObject  ResponseObject  = ResponsePackage.GetObject();

			CbArrayView BlobHashArray  = ResponseObject["blobHashes"sv].AsArrayView();
			CbArrayView MetadatasArray = ResponseObject["metadatas"sv].AsArrayView();
			Result.reserve(MetadatasArray.Num());
			auto BlobHashesIt	 = BlobHashes.begin();
			auto BlobHashArrayIt = begin(BlobHashArray);
			auto MetadataArrayIt = begin(MetadatasArray);
			while (MetadataArrayIt != end(MetadatasArray))
			{
				const IoHash BlobHash = (*BlobHashArrayIt).AsHash();
				while (BlobHash != *BlobHashesIt)
				{
					ZEN_ASSERT(BlobHashesIt != BlobHashes.end());
					BlobHashesIt++;
				}

				ZEN_ASSERT(BlobHash == *BlobHashesIt);

				const IoHash		MetaHash	   = (*MetadataArrayIt).AsAttachment();
				const CbAttachment* MetaAttachment = ResponsePackage.FindAttachment(MetaHash);
				ZEN_ASSERT(MetaAttachment);

				CbObject Metadata = MetaAttachment->AsObject();
				Result.emplace_back(std::move(Metadata));

				BlobHashArrayIt++;
				MetadataArrayIt++;
				BlobHashesIt++;
			}
			return Result;
		}
		return {};
	}

	virtual std::vector<BlobExistsResult> BlobsExists(const Oid& BuildId, std::span<const IoHash> BlobHashes) override
	{
		ZEN_TRACE_CPU("ZenBuildStorageCache::BlobsExists");

		Stopwatch ExecutionTimer;
		auto	  _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });

		CbObjectWriter Request;

		Request.BeginArray("blobHashes"sv);
		for (const IoHash& BlobHash : BlobHashes)
		{
			Request.AddHash(BlobHash);
		}
		Request.EndArray();

		IoBuffer Payload = Request.Save().GetBuffer().AsIoBuffer();
		Payload.SetContentType(ZenContentType::kCbObject);

		HttpClient::Response Response = m_HttpClient.Post(fmt::format("/builds/{}/{}/{}/blobs/exists", m_Namespace, m_Bucket, BuildId),
														  Payload,
														  HttpClient::Accept(ZenContentType::kCbObject));
		AddStatistic(Response);
		if (Response.IsSuccess())
		{
			CbObject ResponseObject = LoadCompactBinaryObject(Response.ResponsePayload);
			if (!ResponseObject)
			{
				throw std::runtime_error("BlobExists reponse is invalid, failed to load payload as compact binary object");
			}
			CbArrayView BlobsExistsArray = ResponseObject["blobExists"sv].AsArrayView();
			if (!BlobsExistsArray)
			{
				throw std::runtime_error("BlobExists reponse is invalid, 'blobExists' array is missing");
			}
			if (BlobsExistsArray.Num() != BlobHashes.size())
			{
				throw std::runtime_error(fmt::format("BlobExists reponse is invalid, 'blobExists' array contains {} entries, expected {}",
													 BlobsExistsArray.Num(),
													 BlobHashes.size()));
			}

			CbArrayView MetadatasExistsArray = ResponseObject["metadataExists"sv].AsArrayView();
			if (!MetadatasExistsArray)
			{
				throw std::runtime_error("BlobExists reponse is invalid, 'metadataExists' array is missing");
			}
			if (MetadatasExistsArray.Num() != BlobHashes.size())
			{
				throw std::runtime_error(
					fmt::format("BlobExists reponse is invalid, 'metadataExists' array contains {} entries, expected {}",
								MetadatasExistsArray.Num(),
								BlobHashes.size()));
			}

			std::vector<BlobExistsResult> Result;
			Result.reserve(BlobHashes.size());
			auto BlobExistsIt	  = begin(BlobsExistsArray);
			auto MetadataExistsIt = begin(MetadatasExistsArray);
			while (BlobExistsIt != end(BlobsExistsArray))
			{
				ZEN_ASSERT(MetadataExistsIt != end(MetadatasExistsArray));

				const bool HasBody	   = (*BlobExistsIt).AsBool();
				const bool HasMetadata = (*MetadataExistsIt).AsBool();

				Result.push_back({.HasBody = HasBody, .HasMetadata = HasMetadata});

				BlobExistsIt++;
				MetadataExistsIt++;
			}
			return Result;
		}
		return {};
	}

	virtual void Flush(int32_t UpdateIntervalMS, std::function<bool(intptr_t Remaining)>&& UpdateCallback) override
	{
		if (IsFlushed)
		{
			return;
		}
		if (!IsFlushed)
		{
			m_PendingBackgroundWorkCount.CountDown();
			IsFlushed = true;
		}
		if (m_PendingBackgroundWorkCount.Wait(100))
		{
			return;
		}
		while (true)
		{
			intptr_t Remaining = m_PendingBackgroundWorkCount.Remaining();
			if (!UpdateCallback(Remaining))
			{
				m_CancelBackgroundWork.store(true);
			}
			if (m_PendingBackgroundWorkCount.Wait(UpdateIntervalMS))
			{
				break;
			}
		}
		UpdateCallback(0);
	}

private:
	void AddStatistic(const HttpClient::Response& Result)
	{
		m_Stats.TotalBytesWritten += Result.UploadedBytes;
		m_Stats.TotalBytesRead += Result.DownloadedBytes;
		m_Stats.TotalRequestTimeUs += uint64_t(Result.ElapsedSeconds * 1000000.0);
		m_Stats.TotalRequestCount++;
		SetAtomicMax(m_Stats.PeakSentBytes, Result.UploadedBytes);
		SetAtomicMax(m_Stats.PeakReceivedBytes, Result.DownloadedBytes);
		if (Result.ElapsedSeconds > 0.0)
		{
			uint64_t BytesPerSec = uint64_t((Result.UploadedBytes + Result.DownloadedBytes) / Result.ElapsedSeconds);
			SetAtomicMax(m_Stats.PeakBytesPerSec, BytesPerSec);
		}
	}

	HttpClient&					   m_HttpClient;
	BuildStorageCache::Statistics& m_Stats;
	const std::string			   m_Namespace;
	const std::string			   m_Bucket;
	const std::filesystem::path	   m_TempFolderPath;
	bool						   IsFlushed = false;

	WorkerThreadPool& m_BackgroundWorkPool;
	Latch			  m_PendingBackgroundWorkCount;
	std::atomic<bool> m_CancelBackgroundWork;
};

std::unique_ptr<BuildStorageCache>
CreateZenBuildStorageCache(HttpClient&					  HttpClient,
						   BuildStorageCache::Statistics& Stats,
						   std::string_view				  Namespace,
						   std::string_view				  Bucket,
						   const std::filesystem::path&	  TempFolderPath,
						   WorkerThreadPool&			  BackgroundWorkerPool)
{
	return std::make_unique<ZenBuildStorageCache>(HttpClient, Stats, Namespace, Bucket, TempFolderPath, BackgroundWorkerPool);
}

#if ZEN_WITH_TESTS

class InMemoryBuildStorageCache : public BuildStorageCache
{
public:
	// MaxRangeSupported == 0 : no range requests are accepted, always return full blob
	// MaxRangeSupported == 1 : single range is supported, multi range returns full blob
	// MaxRangeSupported > 1 : multirange is supported up to MaxRangeSupported, more ranges returns empty blob (bad request)
	explicit InMemoryBuildStorageCache(uint64_t						  MaxRangeSupported,
									   BuildStorageCache::Statistics& Stats,
									   double						  LatencySec	= 0.0,
									   double						  DelayPerKBSec = 0.0)
	: m_MaxRangeSupported(MaxRangeSupported)
	, m_Stats(Stats)
	, m_LatencySec(LatencySec)
	, m_DelayPerKBSec(DelayPerKBSec)
	{
	}
	void PutBuildBlob(const Oid&, const IoHash& RawHash, ZenContentType, const CompositeBuffer& Payload) override
	{
		IoBuffer Buf = Payload.Flatten().AsIoBuffer();
		Buf.MakeOwned();
		const uint64_t SentBytes	 = Buf.Size();
		uint64_t	   ReceivedBytes = 0;
		SimulateLatency(SentBytes, 0);
		auto	  _ = MakeGuard([&]() { SimulateLatency(0, ReceivedBytes); });
		Stopwatch ExecutionTimer;
		auto	  __ = MakeGuard([&]() { AddStatistic(ExecutionTimer.GetElapsedTimeUs(), ReceivedBytes, SentBytes); });
		{
			std::lock_guard Lock(m_Mutex);
			m_Entries[RawHash] = std::move(Buf);
		}
		m_Stats.PutBlobCount.fetch_add(1);
		m_Stats.PutBlobByteCount.fetch_add(SentBytes);
	}

	IoBuffer GetBuildBlob(const Oid&, const IoHash& RawHash, uint64_t RangeOffset = 0, uint64_t RangeBytes = (uint64_t)-1) override
	{
		uint64_t SentBytes	   = 0;
		uint64_t ReceivedBytes = 0;
		SimulateLatency(SentBytes, 0);
		auto	  _ = MakeGuard([&]() { SimulateLatency(0, ReceivedBytes); });
		Stopwatch ExecutionTimer;
		auto	  __ = MakeGuard([&]() { AddStatistic(ExecutionTimer.GetElapsedTimeUs(), ReceivedBytes, SentBytes); });
		IoBuffer  FullPayload;
		{
			std::lock_guard Lock(m_Mutex);
			auto			It = m_Entries.find(RawHash);
			if (It == m_Entries.end())
			{
				return {};
			}
			FullPayload = It->second;
		}

		if (RangeOffset != 0 || RangeBytes != (uint64_t)-1)
		{
			if (m_MaxRangeSupported == 0)
			{
				ReceivedBytes = FullPayload.Size();
				return FullPayload;
			}
			else
			{
				ReceivedBytes = (RangeBytes == (uint64_t)-1) ? FullPayload.Size() - RangeOffset : RangeBytes;
				return IoBuffer(FullPayload, RangeOffset, RangeBytes);
			}
		}
		else
		{
			ReceivedBytes = FullPayload.Size();
			return FullPayload;
		}
	}

	BuildBlobRanges GetBuildBlobRanges(const Oid&, const IoHash& RawHash, std::span<const std::pair<uint64_t, uint64_t>> Ranges) override
	{
		ZEN_ASSERT(!Ranges.empty());
		uint64_t SentBytes	   = 0;
		uint64_t ReceivedBytes = 0;
		SimulateLatency(SentBytes, 0);
		auto	  _ = MakeGuard([&]() { SimulateLatency(0, ReceivedBytes); });
		Stopwatch ExecutionTimer;
		auto	  __ = MakeGuard([&]() { AddStatistic(ExecutionTimer.GetElapsedTimeUs(), ReceivedBytes, SentBytes); });
		if (m_MaxRangeSupported > 1 && Ranges.size() > m_MaxRangeSupported)
		{
			return {};
		}
		IoBuffer FullPayload;
		{
			std::lock_guard Lock(m_Mutex);
			auto			It = m_Entries.find(RawHash);
			if (It == m_Entries.end())
			{
				return {};
			}
			FullPayload = It->second;
		}

		if (Ranges.size() > m_MaxRangeSupported)
		{
			// An empty Ranges signals to the caller: "full buffer given, use it for all requested ranges".
			ReceivedBytes = FullPayload.Size();
			return {.PayloadBuffer = FullPayload};
		}
		else
		{
			uint64_t								   PayloadStart = Ranges.front().first;
			uint64_t								   PayloadSize	= Ranges.back().first + Ranges.back().second - PayloadStart;
			IoBuffer								   RangeBuffer	= IoBuffer(FullPayload, PayloadStart, PayloadSize);
			std::vector<std::pair<uint64_t, uint64_t>> PayloadRanges;
			PayloadRanges.reserve(Ranges.size());
			for (const std::pair<uint64_t, uint64_t>& Range : Ranges)
			{
				PayloadRanges.push_back(std::make_pair(Range.first - PayloadStart, Range.second));
			}
			ReceivedBytes = PayloadSize;
			return {.PayloadBuffer = RangeBuffer, .Ranges = std::move(PayloadRanges)};
		}
	}

	void PutBlobMetadatas(const Oid&, std::span<const IoHash>, std::span<const CbObject>) override {}

	std::vector<CbObject> GetBlobMetadatas(const Oid&, std::span<const IoHash> Hashes) override
	{
		return std::vector<CbObject>(Hashes.size());
	}

	std::vector<BlobExistsResult> BlobsExists(const Oid&, std::span<const IoHash> Hashes) override
	{
		std::lock_guard				  Lock(m_Mutex);
		std::vector<BlobExistsResult> Result;
		Result.reserve(Hashes.size());
		for (const IoHash& Hash : Hashes)
		{
			auto It = m_Entries.find(Hash);
			Result.push_back({.HasBody = (It != m_Entries.end() && It->second)});
		}
		return Result;
	}

	void Flush(int32_t, std::function<bool(intptr_t)>&&) override {}

private:
	void AddStatistic(uint64_t ElapsedTimeUs, uint64_t ReceivedBytes, uint64_t SentBytes)
	{
		m_Stats.TotalBytesWritten += SentBytes;
		m_Stats.TotalBytesRead += ReceivedBytes;
		m_Stats.TotalExecutionTimeUs += ElapsedTimeUs;
		m_Stats.TotalRequestCount++;
		SetAtomicMax(m_Stats.PeakSentBytes, SentBytes);
		SetAtomicMax(m_Stats.PeakReceivedBytes, ReceivedBytes);
		if (ElapsedTimeUs > 0)
		{
			SetAtomicMax(m_Stats.PeakBytesPerSec, (ReceivedBytes + SentBytes) * 1000000 / ElapsedTimeUs);
		}
	}

	void SimulateLatency(uint64_t SendBytes, uint64_t ReceiveBytes)
	{
		double SleepSec = m_LatencySec;
		if (m_DelayPerKBSec > 0.0)
		{
			SleepSec += m_DelayPerKBSec * (double(SendBytes + ReceiveBytes) / 1024u);
		}
		if (SleepSec > 0)
		{
			Sleep(int(SleepSec * 1000));
		}
	}

	uint64_t											 m_MaxRangeSupported = 0;
	BuildStorageCache::Statistics&						 m_Stats;
	const double										 m_LatencySec	 = 0.0;
	const double										 m_DelayPerKBSec = 0.0;
	std::mutex											 m_Mutex;
	std::unordered_map<IoHash, IoBuffer, IoHash::Hasher> m_Entries;
};

std::unique_ptr<BuildStorageCache>
CreateInMemoryBuildStorageCache(uint64_t MaxRangeSupported, BuildStorageCache::Statistics& Stats, double LatencySec, double DelayPerKBSec)
{
	return std::make_unique<InMemoryBuildStorageCache>(MaxRangeSupported, Stats, LatencySec, DelayPerKBSec);
}
#endif	// ZEN_WITH_TESTS

ZenCacheEndpointTestResult
TestZenCacheEndpoint(std::string_view BaseUrl, const bool AssumeHttp2, const bool HttpVerbose)
{
	HttpClientSettings	 TestClientSettings{.LogCategory	= "httpcacheclient",
											.ConnectTimeout = std::chrono::milliseconds{2000},
											.Timeout		= std::chrono::milliseconds{3000},
											.AssumeHttp2	= AssumeHttp2,
											.AllowResume	= true,
											.RetryCount		= 1,
											.Verbose		= HttpVerbose};
	HttpClient			 TestHttpClient(BaseUrl, TestClientSettings);
	HttpClient::Response TestResponse = TestHttpClient.Get("/status/builds");
	if (TestResponse.IsSuccess())
	{
		uint64_t MaxRangeCountPerRequest = 1;
		CbObject StatusResponse			 = TestResponse.AsObject();
		if (StatusResponse["ok"].AsBool())
		{
			MaxRangeCountPerRequest = StatusResponse["capabilities"].AsObjectView()["maxrangecountperrequest"].AsUInt64(1);

			LatencyTestResult LatencyResult = MeasureLatency(TestHttpClient, "/health");

			if (!LatencyResult.Success)
			{
				return {.Success = false, .FailureReason = LatencyResult.FailureReason};
			}

			return {.Success = true, .LatencySeconds = LatencyResult.LatencySeconds, .MaxRangeCountPerRequest = MaxRangeCountPerRequest};
		}
		else
		{
			return {.Success	   = false,
					.FailureReason = fmt::format("ZenCache endpoint {}/status/builds did not respond with \"ok\"", BaseUrl)};
		}
	}
	return {.Success = false, .FailureReason = TestResponse.ErrorMessage("")};
}

}  // namespace zen