Merge remote-tracking branch 'origin/main' into de/cache-with-block-store

8 files changed, 446 insertions, 58 deletions

diff --git a/zencore/compactbinary.cpp b/zencore/compactbinary.cpp
index ffc1da10c..a51253989 100644
--- a/zencore/compactbinary.cpp
+++ b/zencore/compactbinary.cpp
@@ -3,6 +3,7 @@
 #include "zencore/compactbinary.h"
 
 #include <zencore/base64.h>
+#include <zencore/compactbinarybuilder.h>
 #include <zencore/compactbinaryvalidation.h>
 #include <zencore/compactbinaryvalue.h>
 #include <zencore/compress.h>
@@ -22,10 +23,9 @@
 #	include <time.h>
 #endif
 
-#if ZEN_WITH_TESTS
-#	include <json11.hpp>
-#	include <zencore/compactbinarybuilder.h>
-#endif
+ZEN_THIRD_PARTY_INCLUDES_START
+#include <json11.hpp>
+ZEN_THIRD_PARTY_INCLUDES_END
 
 namespace zen {
 
@@ -1715,6 +1715,225 @@ CompactBinaryToJson(const CbArrayView& Array, StringBuilderBase& Builder)
 
 //////////////////////////////////////////////////////////////////////////
 
+class CbJsonReader
+{
+public:
+	static CbFieldIterator Read(std::string_view JsonText, std::string& Error)
+	{
+		using namespace json11;
+
+		const Json Json = Json::parse(std::string(JsonText), Error);
+
+		if (Error.empty())
+		{
+			CbWriter Writer;
+			if (ReadField(Writer, Json, std::string_view(), Error))
+			{
+				return Writer.Save();
+			}
+		}
+
+		return CbFieldIterator();
+	}
+
+private:
+	static bool ReadField(CbWriter& Writer, const json11::Json& Json, const std::string_view FieldName, std::string& Error)
+	{
+		using namespace json11;
+
+		switch (Json.type())
+		{
+			case Json::Type::OBJECT:
+				{
+					if (FieldName.empty())
+					{
+						Writer.BeginObject();
+					}
+					else
+					{
+						Writer.BeginObject(FieldName);
+					}
+
+					for (const auto& Kv : Json.object_items())
+					{
+						const std::string&	Name = Kv.first;
+						const json11::Json& Item = Kv.second;
+
+						if (ReadField(Writer, Item, Name, Error) == false)
+						{
+							return false;
+						}
+					}
+
+					Writer.EndObject();
+				}
+				break;
+			case Json::Type::ARRAY:
+				{
+					if (FieldName.empty())
+					{
+						Writer.BeginArray();
+					}
+					else
+					{
+						Writer.BeginArray(FieldName);
+					}
+
+					for (const json11::Json& Item : Json.array_items())
+					{
+						if (ReadField(Writer, Item, std::string_view(), Error) == false)
+						{
+							return false;
+						}
+					}
+
+					Writer.EndArray();
+				}
+				break;
+			case Json::Type::NUL:
+				{
+					if (FieldName.empty())
+					{
+						Writer.AddNull();
+					}
+					else
+					{
+						Writer.AddNull(FieldName);
+					}
+				}
+				break;
+			case Json::Type::BOOL:
+				{
+					if (FieldName.empty())
+					{
+						Writer.AddBool(Json.bool_value());
+					}
+					else
+					{
+						Writer.AddBool(FieldName, Json.bool_value());
+					}
+				}
+				break;
+			case Json::Type::NUMBER:
+				{
+					if (FieldName.empty())
+					{
+						Writer.AddFloat(Json.number_value());
+					}
+					else
+					{
+						Writer.AddFloat(FieldName, Json.number_value());
+					}
+				}
+				break;
+			case Json::Type::STRING:
+				{
+					Oid Id;
+					if (TryParseObjectId(Json.string_value(), Id))
+					{
+						if (FieldName.empty())
+						{
+							Writer.AddObjectId(Id);
+						}
+						else
+						{
+							Writer.AddObjectId(FieldName, Id);
+						}
+
+						return true;
+					}
+
+					IoHash Hash;
+					if (TryParseIoHash(Json.string_value(), Hash))
+					{
+						if (FieldName.empty())
+						{
+							Writer.AddHash(Hash);
+						}
+						else
+						{
+							Writer.AddHash(FieldName, Hash);
+						}
+
+						return true;
+					}
+
+					if (FieldName.empty())
+					{
+						Writer.AddString(Json.string_value());
+					}
+					else
+					{
+						Writer.AddString(FieldName, Json.string_value());
+					}
+				}
+				break;
+			default:
+				break;
+		}
+
+		return true;
+	}
+
+	static constexpr AsciiSet HexCharSet = AsciiSet("0123456789abcdefABCDEF");
+
+	static bool TryParseObjectId(std::string_view Str, Oid& Id)
+	{
+		using namespace std::literals;
+
+		if (Str.size() == Oid::StringLength && AsciiSet::HasOnly(Str, HexCharSet))
+		{
+			Id = Oid::FromHexString(Str);
+			return true;
+		}
+
+		if (Str.starts_with("0x"sv))
+		{
+			return TryParseObjectId(Str.substr(2), Id);
+		}
+
+		return false;
+	}
+
+	static bool TryParseIoHash(std::string_view Str, IoHash& Hash)
+	{
+		using namespace std::literals;
+
+		if (Str.size() == IoHash::StringLength && AsciiSet::HasOnly(Str, HexCharSet))
+		{
+			Hash = IoHash::FromHexString(Str);
+			return true;
+		}
+
+		if (Str.starts_with("0x"sv))
+		{
+			return TryParseIoHash(Str.substr(2), Hash);
+		}
+
+		return false;
+	}
+};
+
+CbFieldIterator
+LoadCompactBinaryFromJson(std::string_view Json, std::string& Error)
+{
+	if (Json.empty() == false)
+	{
+		return CbJsonReader::Read(Json, Error);
+	}
+
+	return CbFieldIterator();
+}
+
+CbFieldIterator
+LoadCompactBinaryFromJson(std::string_view Json)
+{
+	std::string Error;
+	return LoadCompactBinaryFromJson(Json, Error);
+}
+
+//////////////////////////////////////////////////////////////////////////
+
 #if ZEN_WITH_TESTS
 void
 uson_forcelink()
@@ -1970,6 +2189,107 @@ TEST_CASE("uson.datetime")
 		CHECK_EQ(D72.GetSecond(), 10);
 	}
 }
+
+TEST_CASE("json.uson")
+{
+	using namespace std::literals;
+	using namespace json11;
+
+	SUBCASE("empty")
+	{
+		CbFieldIterator It = LoadCompactBinaryFromJson(""sv);
+		CHECK(It.HasValue() == false);
+	}
+
+	SUBCASE("object")
+	{
+		const Json JsonObject = Json::object{{"Null", nullptr},
+											 {"String", "Value1"},
+											 {"Bool", true},
+											 {"Number", 46.2},
+											 {"Array", Json::array{1, 2, 3}},
+											 {"Object",
+											  Json::object{
+												  {"String", "Value2"},
+											  }}};
+
+		CbObject Cb = LoadCompactBinaryFromJson(JsonObject.dump()).AsObject();
+
+		CHECK(Cb["Null"].IsNull());
+		CHECK(Cb["String"].AsString() == "Value1"sv);
+		CHECK(Cb["Bool"].AsBool());
+		CHECK(Cb["Number"].AsDouble() == 46.2);
+		CHECK(Cb["Object"].IsObject());
+		CbObjectView Object = Cb["Object"].AsObjectView();
+		CHECK(Object["String"].AsString() == "Value2"sv);
+	}
+
+	SUBCASE("array")
+	{
+		const Json JsonArray = Json::array{42, 43, 44};
+		CbArray	   Cb		 = LoadCompactBinaryFromJson(JsonArray.dump()).AsArray();
+
+		auto It = Cb.CreateIterator();
+		CHECK((*It).AsDouble() == 42);
+		It++;
+		CHECK((*It).AsDouble() == 43);
+		It++;
+		CHECK((*It).AsDouble() == 44);
+	}
+
+	SUBCASE("objectid")
+	{
+		const Oid& Id = Oid::NewOid();
+
+		StringBuilder<64> Sb;
+		Id.ToString(Sb);
+
+		Json	 JsonObject = Json::object{{"value", Sb.ToString()}};
+		CbObject Cb			= LoadCompactBinaryFromJson(JsonObject.dump()).AsObject();
+
+		CHECK(Cb["value"sv].IsObjectId());
+		CHECK(Cb["value"sv].AsObjectId() == Id);
+
+		Sb.Reset();
+		Sb << "0x";
+		Id.ToString(Sb);
+
+		JsonObject = Json::object{{"value", Sb.ToString()}};
+		Cb		   = LoadCompactBinaryFromJson(JsonObject.dump()).AsObject();
+
+		CHECK(Cb["value"sv].IsObjectId());
+		CHECK(Cb["value"sv].AsObjectId() == Id);
+	}
+
+	SUBCASE("iohash")
+	{
+		const uint8_t Data[] = {
+			1,
+			2,
+			3,
+			4,
+			5,
+			6,
+			7,
+			8,
+			9,
+		};
+
+		const IoHash Hash = IoHash::HashBuffer(Data, sizeof(Data));
+
+		Json	 JsonObject = Json::object{{"value", Hash.ToHexString()}};
+		CbObject Cb			= LoadCompactBinaryFromJson(JsonObject.dump()).AsObject();
+
+		CHECK(Cb["value"sv].IsHash());
+		CHECK(Cb["value"sv].AsHash() == Hash);
+
+		JsonObject = Json::object{{"value", "0x" + Hash.ToHexString()}};
+		Cb		   = LoadCompactBinaryFromJson(JsonObject.dump()).AsObject();
+
+		CHECK(Cb["value"sv].IsHash());
+		CHECK(Cb["value"sv].AsHash() == Hash);
+	}
+}
 #endif
 
 }  // namespace zen
diff --git a/zencore/compactbinarybuilder.cpp b/zencore/compactbinarybuilder.cpp
index 5111504e1..1d2ba45df 100644
--- a/zencore/compactbinarybuilder.cpp
+++ b/zencore/compactbinarybuilder.cpp
@@ -436,9 +436,10 @@ CbWriter::AddNull()
 void
 CbWriter::AddBinary(const void* const Value, const uint64_t Size)
 {
+	const size_t SizeByteCount = MeasureVarUInt(Size);
+	Data.reserve(Data.size() + 1 + SizeByteCount + Size);
 	BeginField();
-	const uint32_t SizeByteCount = MeasureVarUInt(Size);
-	const int64_t  SizeOffset	 = Data.size();
+	const size_t SizeOffset = Data.size();
 	Data.resize(Data.size() + SizeByteCount);
 	WriteVarUInt(Size, Data.data() + SizeOffset);
 	Data.insert(Data.end(), static_cast<const uint8_t*>(Value), static_cast<const uint8_t*>(Value) + Size);
diff --git a/zencore/include/zencore/compactbinary.h b/zencore/include/zencore/compactbinary.h
index 19f1597dc..eba4a1694 100644
--- a/zencore/include/zencore/compactbinary.h
+++ b/zencore/include/zencore/compactbinary.h
@@ -1405,6 +1405,12 @@ ZENCORE_API CbObject LoadCompactBinaryObject(const IoBuffer& Payload);
 ZENCORE_API CbObject LoadCompactBinaryObject(CompressedBuffer&& Payload);
 ZENCORE_API CbObject LoadCompactBinaryObject(const CompressedBuffer& Payload);
 
+/**
+ * Load a compact binary from JSON.
+ */
+ZENCORE_API CbFieldIterator LoadCompactBinaryFromJson(std::string_view Json, std::string& Error);
+ZENCORE_API CbFieldIterator LoadCompactBinaryFromJson(std::string_view Json);
+
 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
 /**
diff --git a/zencore/include/zencore/string.h b/zencore/include/zencore/string.h
index 027730063..012ee73ee 100644
--- a/zencore/include/zencore/string.h
+++ b/zencore/include/zencore/string.h
@@ -999,7 +999,7 @@ public:
 	static constexpr bool HasOnly(const StringType& Str, AsciiSet Set)
 	{
 		auto End = Str.data() + Str.size();
-		return FindFirst<EInclude::Members>(Set, GetData(Str), End) == End;
+		return FindFirst<EInclude::Members>(Set, Str.data(), End) == End;
 	}
 
 private:
diff --git a/zencore/iobuffer.cpp b/zencore/iobuffer.cpp
index 8a3ab8427..c069aa0f1 100644
--- a/zencore/iobuffer.cpp
+++ b/zencore/iobuffer.cpp
@@ -226,7 +226,15 @@ IoBufferExtendedCore::~IoBufferExtendedCore()
 	m_DataPtr = nullptr;
 }
 
-static RwLock g_MappingLock;
+static RwLock g_MappingLock[0x40];
+
+static RwLock&
+MappingLockForInstance(const IoBufferExtendedCore* instance)
+{
+	intptr_t base		= (intptr_t)instance;
+	size_t	 lock_index = ((base >> 8) ^ (base >> 16)) & 0x3f;
+	return g_MappingLock[lock_index];
+}
 
 void
 IoBufferExtendedCore::Materialize() const
@@ -237,7 +245,7 @@ IoBufferExtendedCore::Materialize() const
 	if (m_Flags.load(std::memory_order_acquire) & kIsMaterialized)
 		return;
 
-	RwLock::ExclusiveLockScope _(g_MappingLock);
+	RwLock::ExclusiveLockScope _(MappingLockForInstance(this));
 
 	// Someone could have gotten here first
 	// We can use memory_order_relaxed on this load because the mutex has already provided the fence
diff --git a/zenserver/cache/structuredcache.cpp b/zenserver/cache/structuredcache.cpp
index 8ae531720..8daf08bff 100644
--- a/zenserver/cache/structuredcache.cpp
+++ b/zenserver/cache/structuredcache.cpp
@@ -650,42 +650,52 @@ HttpStructuredCacheService::HandleCacheValueRequest(HttpServerRequest& Request,
 void
 HttpStructuredCacheService::HandleGetCacheValue(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromURL)
 {
+	Stopwatch Timer;
+
 	IoBuffer	Value			= m_CidStore.FindChunkByCid(Ref.ValueContentId);
 	bool		InUpstreamCache = false;
 	CachePolicy Policy			= PolicyFromURL;
-	const bool	QueryUpstream	= !Value && EnumHasAllFlags(Policy, CachePolicy::QueryRemote);
-
-	if (QueryUpstream)
 	{
-		if (auto UpstreamResult = m_UpstreamCache.GetCacheValue({Ref.BucketSegment, Ref.HashKey}, Ref.ValueContentId);
-			UpstreamResult.Success)
+		const bool QueryUpstream = !Value && EnumHasAllFlags(Policy, CachePolicy::QueryRemote);
+
+		if (QueryUpstream)
 		{
-			if (CompressedBuffer Compressed = CompressedBuffer::FromCompressed(SharedBuffer(UpstreamResult.Value)))
-			{
-				m_CidStore.AddChunk(Compressed);
-				InUpstreamCache = true;
-			}
-			else
+			if (auto UpstreamResult = m_UpstreamCache.GetCacheValue({Ref.BucketSegment, Ref.HashKey}, Ref.ValueContentId);
+				UpstreamResult.Success)
 			{
-				ZEN_WARN("got uncompressed upstream cache value");
+				if (CompressedBuffer Compressed = CompressedBuffer::FromCompressed(SharedBuffer(UpstreamResult.Value)))
+				{
+					m_CidStore.AddChunk(Compressed);
+					InUpstreamCache = true;
+				}
+				else
+				{
+					ZEN_WARN("got uncompressed upstream cache value");
+				}
 			}
 		}
 	}
 
 	if (!Value)
 	{
-		ZEN_DEBUG("MISS - '{}/{}/{}' '{}'", Ref.BucketSegment, Ref.HashKey, Ref.ValueContentId, ToString(Request.AcceptContentType()));
+		ZEN_DEBUG("MISS - '{}/{}/{}' '{}' in {}",
+				  Ref.BucketSegment,
+				  Ref.HashKey,
+				  Ref.ValueContentId,
+				  ToString(Request.AcceptContentType()),
+				  NiceLatencyNs(Timer.GetElapsedTimeUs() * 1000));
 		m_CacheStats.MissCount++;
 		return Request.WriteResponse(HttpResponseCode::NotFound);
 	}
 
-	ZEN_DEBUG("HIT - '{}/{}/{}' {} '{}' ({})",
+	ZEN_DEBUG("HIT - '{}/{}/{}' {} '{}' ({}) in {}",
 			  Ref.BucketSegment,
 			  Ref.HashKey,
 			  Ref.ValueContentId,
 			  NiceBytes(Value.Size()),
 			  ToString(Value.GetContentType()),
-			  InUpstreamCache ? "UPSTREAM" : "LOCAL");
+			  InUpstreamCache ? "UPSTREAM" : "LOCAL",
+			  NiceLatencyNs(Timer.GetElapsedTimeUs() * 1000));
 
 	m_CacheStats.HitCount++;
 	if (InUpstreamCache)
@@ -709,6 +719,8 @@ HttpStructuredCacheService::HandlePutCacheValue(zen::HttpServerRequest& Request,
 	// Note: Individual cacherecord values are not propagated upstream until a valid cache record has been stored
 	ZEN_UNUSED(PolicyFromURL);
 
+	Stopwatch Timer;
+
 	IoBuffer Body = Request.ReadPayload();
 
 	if (!Body || Body.Size() == 0)
@@ -734,13 +746,14 @@ HttpStructuredCacheService::HandlePutCacheValue(zen::HttpServerRequest& Request,
 
 	CidStore::InsertResult Result = m_CidStore.AddChunk(Compressed);
 
-	ZEN_DEBUG("PUT - '{}/{}/{}' {} '{}' ({})",
+	ZEN_DEBUG("PUT - '{}/{}/{}' {} '{}' ({}) in {}",
 			  Ref.BucketSegment,
 			  Ref.HashKey,
 			  Ref.ValueContentId,
 			  NiceBytes(Body.Size()),
 			  ToString(Body.GetContentType()),
-			  Result.New ? "NEW" : "OLD");
+			  Result.New ? "NEW" : "OLD",
+			  NiceLatencyNs(Timer.GetElapsedTimeUs() * 1000));
 
 	const HttpResponseCode ResponseCode = Result.New ? HttpResponseCode::Created : HttpResponseCode::OK;
 
@@ -1444,8 +1457,12 @@ HttpStructuredCacheService::HandleRpcGetCacheValues(zen::HttpServerRequest& Http
 
 	ZEN_ASSERT(RpcRequest["Method"sv].AsString() == "GetCacheValues"sv);
 
+	std::vector<size_t> RemoteRequestIndexes;
+
 	for (CbFieldView RequestField : Params["Requests"sv])
 	{
+		Stopwatch Timer;
+
 		RequestData& Request	   = Requests.emplace_back();
 		CbObjectView RequestObject = RequestField.AsObjectView();
 		CbObjectView KeyObject	   = RequestObject["Key"sv].AsObjectView();
@@ -1463,46 +1480,28 @@ HttpStructuredCacheService::HandleRpcGetCacheValues(zen::HttpServerRequest& Http
 		CachePolicy		  Policy = Request.Policy;
 		CompressedBuffer& Result = Request.Result;
 
-		ZenCacheValue	 CacheValue;
-		std::string_view Source;
+		ZenCacheValue CacheValue;
 		if (EnumHasAllFlags(Policy, CachePolicy::QueryLocal))
 		{
 			if (m_CacheStore.Get(Key.Bucket, Key.Hash, CacheValue) && IsCompressedBinary(CacheValue.Value.GetContentType()))
 			{
 				Result = CompressedBuffer::FromCompressed(SharedBuffer(CacheValue.Value));
-				if (Result)
-				{
-					Source = "LOCAL"sv;
-				}
-			}
-		}
-		if (!Result && EnumHasAllFlags(Policy, CachePolicy::QueryRemote))
-		{
-			GetUpstreamCacheResult UpstreamResult =
-				m_UpstreamCache.GetCacheRecord({Key.Bucket, Key.Hash}, ZenContentType::kCompressedBinary);
-			if (UpstreamResult.Success && IsCompressedBinary(UpstreamResult.Value.GetContentType()))
-			{
-				Result = CompressedBuffer::FromCompressed(SharedBuffer(UpstreamResult.Value));
-				if (Result)
-				{
-					UpstreamResult.Value.SetContentType(ZenContentType::kCompressedBinary);
-					Source = "UPSTREAM"sv;
-					// TODO: Respect the StoreLocal flag once we have upstream existence-only checks. For now the requirement
-					// that we copy data from upstream even when SkipData and !StoreLocal are true means that it is too expensive
-					// for us to keep the data only on the upstream server.
-					// if (EnumHasAllFlags(Policy, CachePolicy::StoreLocal))
-					{
-						m_CacheStore.Put(Key.Bucket, Key.Hash, ZenCacheValue{UpstreamResult.Value});
-					}
-				}
 			}
 		}
-
 		if (Result)
 		{
-			ZEN_DEBUG("GETCACHEVALUES HIT - '{}/{}' {} ({})", Key.Bucket, Key.Hash, NiceBytes(Result.GetCompressed().GetSize()), Source);
+			ZEN_DEBUG("GETCACHEVALUES HIT - '{}/{}' {} ({}) in {}",
+					  Key.Bucket,
+					  Key.Hash,
+					  NiceBytes(Result.GetCompressed().GetSize()),
+					  "LOCAL"sv,
+					  NiceLatencyNs(Timer.GetElapsedTimeUs() * 1000));
 			m_CacheStats.HitCount++;
 		}
+		else if (EnumHasAllFlags(Policy, CachePolicy::QueryRemote))
+		{
+			RemoteRequestIndexes.push_back(Requests.size() - 1);
+		}
 		else if (!EnumHasAnyFlags(Policy, CachePolicy::Query))
 		{
 			// If they requested no query, do not record this as a miss
@@ -1510,10 +1509,65 @@ HttpStructuredCacheService::HandleRpcGetCacheValues(zen::HttpServerRequest& Http
 		}
 		else
 		{
-			ZEN_DEBUG("GETCACHEVALUES MISS - '{}/{}'", Key.Bucket, Key.Hash);
+			ZEN_DEBUG("GETCACHEVALUES MISS - '{}/{}' ({}) in {}",
+					  Key.Bucket,
+					  Key.Hash,
+					  "LOCAL"sv,
+					  NiceLatencyNs(Timer.GetElapsedTimeUs() * 1000));
 			m_CacheStats.MissCount++;
 		}
 	}
+
+	if (!RemoteRequestIndexes.empty())
+	{
+		std::vector<CacheChunkRequest>	RequestedRecordsData;
+		std::vector<CacheChunkRequest*> CacheChunkRequests;
+		RequestedRecordsData.reserve(RemoteRequestIndexes.size());
+		CacheChunkRequests.reserve(RemoteRequestIndexes.size());
+		for (size_t Index : RemoteRequestIndexes)
+		{
+			RequestData& Request = Requests[Index];
+			RequestedRecordsData.push_back({Request.Key.Bucket, Request.Key.Hash});
+			CacheChunkRequests.push_back(&RequestedRecordsData.back());
+		}
+		Stopwatch Timer;
+		m_UpstreamCache.GetCacheValues(
+			CacheChunkRequests,
+			[this, &RequestedRecordsData, &Requests, &RemoteRequestIndexes, &Timer](CacheValueGetCompleteParams&& Params) {
+				CacheChunkRequest& ChunkRequest = Params.Request;
+				if (Params.Value)
+				{
+					size_t		 RequestOffset = std::distance(RequestedRecordsData.data(), &ChunkRequest);
+					size_t		 RequestIndex  = RemoteRequestIndexes[RequestOffset];
+					RequestData& Request	   = Requests[RequestIndex];
+					Request.Result			   = CompressedBuffer::FromCompressed(SharedBuffer(Params.Value));
+					if (Request.Result && IsCompressedBinary(Params.Value.GetContentType()))
+					{
+						// TODO: Respect the StoreLocal flag once we have upstream existence-only checks. For now the requirement
+						// that we copy data from upstream even when SkipData and !StoreLocal are true means that it is too expensive
+						// for us to keep the data only on the upstream server.
+						// if (EnumHasAllFlags(Policy, CachePolicy::StoreLocal))
+						m_CacheStore.Put(Request.Key.Bucket, Request.Key.Hash, ZenCacheValue{Params.Value});
+						ZEN_DEBUG("GETCACHEVALUES HIT - '{}/{}' {} ({}) in {}",
+								  ChunkRequest.Key.Bucket,
+								  ChunkRequest.Key.Hash,
+								  NiceBytes(Request.Result.GetCompressed().GetSize()),
+								  "UPSTREAM"sv,
+								  NiceLatencyNs(Timer.GetElapsedTimeUs() * 1000));
+						m_CacheStats.HitCount++;
+						m_CacheStats.UpstreamHitCount++;
+						return;
+					}
+				}
+				ZEN_DEBUG("GETCACHEVALUES MISS - '{}/{}' ({}) in {}",
+						  ChunkRequest.Key.Bucket,
+						  ChunkRequest.Key.Hash,
+						  "UPSTREAM"sv,
+						  NiceLatencyNs(Timer.GetElapsedTimeUs() * 1000));
+				m_CacheStats.MissCount++;
+			});
+	}
+
 	if (Requests.empty())
 	{
 		return HttpRequest.WriteResponse(HttpResponseCode::BadRequest);
diff --git a/zenserver/cache/structuredcachestore.cpp b/zenserver/cache/structuredcachestore.cpp
index 3ba4e6b05..53a479edb 100644
--- a/zenserver/cache/structuredcachestore.cpp
+++ b/zenserver/cache/structuredcachestore.cpp
@@ -2562,7 +2562,6 @@ ZenCacheDiskLayer::Flush()
 
 	{
 		RwLock::SharedLockScope _(m_Lock);
-
 		Buckets.reserve(m_Buckets.size());
 		for (auto& Kv : m_Buckets)
 		{
diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp
index da0743f0a..dba80faa9 100644
--- a/zenserver/upstream/upstreamcache.cpp
+++ b/zenserver/upstream/upstreamcache.cpp
@@ -1451,7 +1451,7 @@ public:
 
 	virtual void EnqueueUpstream(UpstreamCacheRecord CacheRecord) override
 	{
-		if (m_RunState.IsRunning && m_Options.WriteUpstream)
+		if (m_RunState.IsRunning && m_Options.WriteUpstream && m_Endpoints.size() > 0)
 		{
 			if (!m_UpstreamThreads.empty())
 			{