Partial refactoring of structured cache implementation - WIP

7 files changed, 158 insertions, 50 deletions

diff --git a/zenserver-test/zenserver-test.cpp b/zenserver-test/zenserver-test.cpp
index cb4ff06da..b9fa96ca0 100644
--- a/zenserver-test/zenserver-test.cpp
+++ b/zenserver-test/zenserver-test.cpp
@@ -7,6 +7,7 @@
 #include <zencore/compactbinarypackage.h>
 #include <zencore/except.h>
 #include <zencore/filesystem.h>
+#include <zencore/fmtutils.h>
 #include <zencore/iohash.h>
 #include <zencore/string.h>
 #include <zencore/thread.h>
@@ -1068,6 +1069,7 @@ TEST_CASE("project.basic")
 				 zen::NiceRate(RequestCount, (uint32_t)Elapsed, "req"));
 }
 
+#	if 0  // this is extremely WIP
 TEST_CASE("project.pipe")
 {
 	using namespace std::literals;
@@ -1088,5 +1090,49 @@ TEST_CASE("project.pipe")
 
 	zen::CbObject Response = LocalClient.MessageTransaction(Cbow.Save());
 }
+#	endif
+
+TEST_CASE("z$.basic")
+{
+	using namespace std::literals;
+
+	std::filesystem::path TestDir = TestEnv.CreateNewTestDir();
+
+	const uint16_t PortNumber = 13337;
+
+	ZenServerInstance Instance1(TestEnv);
+	Instance1.SetTestDir(TestDir);
+	Instance1.SpawnServer(PortNumber);
+	Instance1.WaitUntilReady();
+
+	auto BaseUri = "http://localhost:{}/z$"_format(PortNumber);
+
+	// Populate with some simple data
+
+	for (int i = 0; i < 100; ++i)
+	{
+		zen::CbObjectWriter Cbo;
+		Cbo << "index" << i;
+
+		zen::MemoryOutStream MemOut;
+		zen::BinaryWriter	 Writer{MemOut};
+		Cbo.Save(Writer);
+
+		zen::IoHash Key = zen::IoHash::HashMemory(&i, sizeof i);
+
+		cpr::Response Result = cpr::Put(cpr::Url{"{}/{}/{}"_format(BaseUri, "test", Key)},
+										cpr::Body{(const char*)MemOut.Data(), MemOut.Size()},
+										cpr::Header{{"Content-Type", "application/x-ue-cb"}});
+	}
+
+	// Retrieve data
+
+	for (int i = 0; i < 100; ++i)
+	{
+		zen::IoHash Key = zen::IoHash::HashMemory(&i, sizeof i);
+
+		cpr::Response Result = cpr::Get(cpr::Url{"{}/{}/{}"_format(BaseUri, "test", Key)});
+	}
+}
 
 #endif
diff --git a/zenserver/cache/cachestore.cpp b/zenserver/cache/cachestore.cpp
index 87c89c6a8..e0dd57902 100644
--- a/zenserver/cache/cachestore.cpp
+++ b/zenserver/cache/cachestore.cpp
@@ -567,7 +567,7 @@ ZenCacheStore::~ZenCacheStore()
 }
 
 bool
-ZenCacheStore::Get(std::string_view InBucket, const zen::IoHash& HashKey, CacheValue& OutValue)
+ZenCacheStore::Get(std::string_view InBucket, const zen::IoHash& HashKey, ZenCacheValue& OutValue)
 {
 	bool Ok = m_MemLayer.Get(InBucket, HashKey, OutValue);
 
@@ -587,7 +587,7 @@ ZenCacheStore::Get(std::string_view InBucket, const zen::IoHash& HashKey, CacheV
 }
 
 void
-ZenCacheStore::Put(std::string_view InBucket, const zen::IoHash& HashKey, const CacheValue& Value)
+ZenCacheStore::Put(std::string_view InBucket, const zen::IoHash& HashKey, const ZenCacheValue& Value)
 {
 	m_MemLayer.Put(InBucket, HashKey, Value);
 	m_DiskLayer.Put(InBucket, HashKey, Value);
@@ -604,7 +604,7 @@ ZenCacheMemoryLayer::~ZenCacheMemoryLayer()
 }
 
 bool
-ZenCacheMemoryLayer::Get(std::string_view InBucket, const zen::IoHash& HashKey, CacheValue& OutValue)
+ZenCacheMemoryLayer::Get(std::string_view InBucket, const zen::IoHash& HashKey, ZenCacheValue& OutValue)
 {
 	CacheBucket* Bucket = nullptr;
 
@@ -628,7 +628,7 @@ ZenCacheMemoryLayer::Get(std::string_view InBucket, const zen::IoHash& HashKey,
 }
 
 void
-ZenCacheMemoryLayer::Put(std::string_view InBucket, const zen::IoHash& HashKey, const CacheValue& Value)
+ZenCacheMemoryLayer::Put(std::string_view InBucket, const zen::IoHash& HashKey, const ZenCacheValue& Value)
 {
 	CacheBucket* Bucket = nullptr;
 
@@ -658,7 +658,7 @@ ZenCacheMemoryLayer::Put(std::string_view InBucket, const zen::IoHash& HashKey,
 }
 
 bool
-ZenCacheMemoryLayer::CacheBucket::Get(const zen::IoHash& HashKey, CacheValue& OutValue)
+ZenCacheMemoryLayer::CacheBucket::Get(const zen::IoHash& HashKey, ZenCacheValue& OutValue)
 {
 	RwLock::SharedLockScope _(m_bucketLock);
 
@@ -675,7 +675,7 @@ ZenCacheMemoryLayer::CacheBucket::Get(const zen::IoHash& HashKey, CacheValue& Ou
 }
 
 void
-ZenCacheMemoryLayer::CacheBucket::Put(const zen::IoHash& HashKey, const CacheValue& Value)
+ZenCacheMemoryLayer::CacheBucket::Put(const zen::IoHash& HashKey, const ZenCacheValue& Value)
 {
 	RwLock::ExclusiveLockScope _(m_bucketLock);
 
@@ -882,11 +882,11 @@ struct ZenCacheDiskLayer::CacheBucket
 
 	void OpenOrCreate(std::filesystem::path BucketDir);
 
-	bool Get(const zen::IoHash& HashKey, CacheValue& OutValue);
-	void Put(const zen::IoHash& HashKey, const CacheValue& Value);
+	bool Get(const zen::IoHash& HashKey, ZenCacheValue& OutValue);
+	void Put(const zen::IoHash& HashKey, const ZenCacheValue& Value);
 	void Flush();
 
-	void PutLargeObject(const zen::IoHash& HashKey, const CacheValue& Value);
+	void PutLargeObject(const zen::IoHash& HashKey, const ZenCacheValue& Value);
 
 	inline bool IsOk() const { return m_Ok; }
 
@@ -1011,7 +1011,7 @@ ZenCacheDiskLayer::CacheBucket::BuildPath(zen::WideStringBuilderBase& Path, cons
 }
 
 bool
-ZenCacheDiskLayer::CacheBucket::Get(const zen::IoHash& HashKey, CacheValue& OutValue)
+ZenCacheDiskLayer::CacheBucket::Get(const zen::IoHash& HashKey, ZenCacheValue& OutValue)
 {
 	if (!m_Ok)
 		return false;
@@ -1047,7 +1047,7 @@ ZenCacheDiskLayer::CacheBucket::Get(const zen::IoHash& HashKey, CacheValue& OutV
 }
 
 void
-ZenCacheDiskLayer::CacheBucket::Put(const zen::IoHash& HashKey, const CacheValue& Value)
+ZenCacheDiskLayer::CacheBucket::Put(const zen::IoHash& HashKey, const ZenCacheValue& Value)
 {
 	if (!m_Ok)
 		return;
@@ -1094,7 +1094,7 @@ ZenCacheDiskLayer::CacheBucket::Flush()
 }
 
 void
-ZenCacheDiskLayer::CacheBucket::PutLargeObject(const zen::IoHash& HashKey, const CacheValue& Value)
+ZenCacheDiskLayer::CacheBucket::PutLargeObject(const zen::IoHash& HashKey, const ZenCacheValue& Value)
 {
 	zen::WideStringBuilder<128> dataFilePath;
 
@@ -1125,7 +1125,7 @@ ZenCacheDiskLayer::ZenCacheDiskLayer(CasStore& Cas, const std::filesystem::path&
 ZenCacheDiskLayer::~ZenCacheDiskLayer() = default;
 
 bool
-ZenCacheDiskLayer::Get(std::string_view InBucket, const zen::IoHash& HashKey, CacheValue& OutValue)
+ZenCacheDiskLayer::Get(std::string_view InBucket, const zen::IoHash& HashKey, ZenCacheValue& OutValue)
 {
 	CacheBucket* Bucket = nullptr;
 
@@ -1161,7 +1161,7 @@ ZenCacheDiskLayer::Get(std::string_view InBucket, const zen::IoHash& HashKey, Ca
 }
 
 void
-ZenCacheDiskLayer::Put(std::string_view InBucket, const zen::IoHash& HashKey, const CacheValue& Value)
+ZenCacheDiskLayer::Put(std::string_view InBucket, const zen::IoHash& HashKey, const ZenCacheValue& Value)
 {
 	CacheBucket* Bucket = nullptr;
 
diff --git a/zenserver/cache/cachestore.h b/zenserver/cache/cachestore.h
index 61a8061be..41e47d87d 100644
--- a/zenserver/cache/cachestore.h
+++ b/zenserver/cache/cachestore.h
@@ -21,7 +21,6 @@ class CasStore;
 struct CacheValue
 {
 	zen::IoBuffer Value;
-	bool		  IsCompactBinary = false;
 };
 
 /******************************************************************************
@@ -97,18 +96,24 @@ private:
 
   Cache store for UE5. Restricts keys to "{bucket}/{hash}" pairs where the hash
   is 40 (hex) chars in size. Values may be opaque blobs or structured objects
-  which can in turn contain references to other objects.
+  which can in turn contain references to other objects (or blobs).
 
 ******************************************************************************/
 
+struct ZenCacheValue
+{
+	zen::IoBuffer Value;
+	bool		  IsCompactBinary = false;
+};
+
 class ZenCacheMemoryLayer
 {
 public:
 	ZenCacheMemoryLayer();
 	~ZenCacheMemoryLayer();
 
-	bool Get(std::string_view Bucket, const zen::IoHash& HashKey, CacheValue& OutValue);
-	void Put(std::string_view Bucket, const zen::IoHash& HashKey, const CacheValue& Value);
+	bool Get(std::string_view Bucket, const zen::IoHash& HashKey, ZenCacheValue& OutValue);
+	void Put(std::string_view Bucket, const zen::IoHash& HashKey, const ZenCacheValue& Value);
 
 private:
 	struct CacheBucket
@@ -116,8 +121,8 @@ private:
 		zen::RwLock															m_bucketLock;
 		std::unordered_map<zen::IoHash, zen::IoBuffer, zen::IoHash::Hasher> m_cacheMap;
 
-		bool Get(const zen::IoHash& HashKey, CacheValue& OutValue);
-		void Put(const zen::IoHash& HashKey, const CacheValue& Value);
+		bool Get(const zen::IoHash& HashKey, ZenCacheValue& OutValue);
+		void Put(const zen::IoHash& HashKey, const ZenCacheValue& Value);
 	};
 
 	zen::RwLock									 m_Lock;
@@ -130,8 +135,8 @@ public:
 	ZenCacheDiskLayer(zen::CasStore& Cas, const std::filesystem::path& RootDir);
 	~ZenCacheDiskLayer();
 
-	bool Get(std::string_view Bucket, const zen::IoHash& HashKey, CacheValue& OutValue);
-	void Put(std::string_view Bucket, const zen::IoHash& HashKey, const CacheValue& Value);
+	bool Get(std::string_view Bucket, const zen::IoHash& HashKey, ZenCacheValue& OutValue);
+	void Put(std::string_view Bucket, const zen::IoHash& HashKey, const ZenCacheValue& Value);
 
 	void Flush();
 
@@ -153,8 +158,8 @@ public:
 	ZenCacheStore(zen::CasStore& Cas, const std::filesystem::path& RootDir);
 	~ZenCacheStore();
 
-	virtual bool Get(std::string_view Bucket, const zen::IoHash& HashKey, CacheValue& OutValue);
-	virtual void Put(std::string_view Bucket, const zen::IoHash& HashKey, const CacheValue& Value);
+	virtual bool Get(std::string_view Bucket, const zen::IoHash& HashKey, ZenCacheValue& OutValue);
+	virtual void Put(std::string_view Bucket, const zen::IoHash& HashKey, const ZenCacheValue& Value);
 
 private:
 	std::filesystem::path m_RootDir;
diff --git a/zenserver/cache/structuredcache.cpp b/zenserver/cache/structuredcache.cpp
index e90f838da..9e771a715 100644
--- a/zenserver/cache/structuredcache.cpp
+++ b/zenserver/cache/structuredcache.cpp
@@ -23,11 +23,13 @@ HttpStructuredCacheService::HttpStructuredCacheService(std::filesystem::path Roo
 {
 	spdlog::info("initializing structured cache at '{}'", RootPath);
 
+#if 0
 	m_Cloud = new CloudCacheClient("https://jupiter.devtools.epicgames.com"sv,
 								   "ue4.ddc"sv /* namespace */,
 								   "https://epicgames.okta.com/oauth2/auso645ojjWVdRI3d0x7/v1/token"sv /* provider */,
 								   "0oao91lrhqPiAlaGD0x7"sv /* client id */,
 								   "-GBWjjenhCgOwhxL5yBKNJECVIoDPH0MK4RDuN7d"sv /* oauth secret */);
+#endif
 }
 
 HttpStructuredCacheService::~HttpStructuredCacheService()
@@ -58,8 +60,8 @@ HttpStructuredCacheService::HandleRequest(zen::HttpServerRequest& Request)
 		case kHead:
 		case kGet:
 			{
-				CacheValue Value;
-				bool	   Success = m_CacheStore.Get(Ref.BucketSegment, Ref.HashKey, /* out */ Value);
+				ZenCacheValue Value;
+				bool		  Success = m_CacheStore.Get(Ref.BucketSegment, Ref.HashKey, /* out */ Value);
 
 				if (!Success)
 				{
@@ -84,7 +86,7 @@ HttpStructuredCacheService::HandleRequest(zen::HttpServerRequest& Request)
 			{
 				if (zen::IoBuffer Body = Request.ReadPayload())
 				{
-					CacheValue Value;
+					ZenCacheValue Value;
 					Value.Value = Body;
 
 					HttpContentType ContentType = Request.RequestContentType();
@@ -115,13 +117,21 @@ HttpStructuredCacheService::HandleRequest(zen::HttpServerRequest& Request)
 
 						zen::Stopwatch Timer;
 
-						if (Session.Put(Ref.BucketSegment, Ref.HashKey))
+						try
 						{
-							spdlog::debug("upstream PUT succeeded after {:5}! {}", zen::NiceTimeSpanMs(Timer.getElapsedTimeMs()), Ref.HashKey);
+							Session.Put(Ref.BucketSegment, Ref.HashKey, Value);
+							spdlog::debug("upstream PUT ({}) succeeded after {:5}!",
+										  Ref.HashKey,
+										  zen::NiceTimeSpanMs(Timer.getElapsedTimeMs()));
 						}
-						else
+						catch (std::exception& e)
 						{
-							spdlog::debug("upstream PUT failed after {:5}! {}", zen::NiceTimeSpanMs(Timer.getElapsedTimeMs()), Ref.HashKey);
+							spdlog::debug("upstream PUT ({}) failed after {:5}: '{}'",
+										  Ref.HashKey,
+										  zen::NiceTimeSpanMs(Timer.getElapsedTimeMs()),
+										  e.what());
+
+							throw;
 						}
 					}
 
@@ -146,22 +156,50 @@ HttpStructuredCacheService::HandleRequest(zen::HttpServerRequest& Request)
 HttpStructuredCacheService::ValidateUri(zen::HttpServerRequest& Request, CacheRef& OutRef)
 {
 	std::string_view			Key				  = Request.RelativeUri();
-	std::string_view::size_type BucketSplitOffset = Key.find_last_of('/');
+	std::string_view::size_type BucketSplitOffset = Key.find_first_of('/');
 
 	if (BucketSplitOffset == std::string_view::npos)
 	{
 		return false;
 	}
 
-	OutRef.BucketSegment		 = Key.substr(0, BucketSplitOffset);
-	std::string_view HashSegment = Key.substr(BucketSplitOffset + 1);
+	OutRef.BucketSegment = Key.substr(0, BucketSplitOffset);
+
+	std::string_view HashSegment;
+	std::string_view PayloadSegment;
+
+	std::string_view::size_type PayloadSplitOffset = Key.find_last_of('/');
+
+	// We know there is a slash so no need to check for npos return
+
+	if (PayloadSplitOffset == BucketSplitOffset)
+	{
+		// Basic cache record lookup
+		HashSegment = Key.substr(BucketSplitOffset + 1);
+	}
+	else
+	{
+		// Cache record + payload lookup
+		HashSegment	   = Key.substr(BucketSplitOffset + 1, PayloadSplitOffset - BucketSplitOffset - 1);
+		PayloadSegment = Key.substr(PayloadSplitOffset + 1);
+	}
 
 	if (HashSegment.size() != (2 * sizeof OutRef.HashKey.Hash))
 	{
 		return false;
 	}
 
-	bool IsOk = zen::ParseHexBytes(HashSegment.data(), HashSegment.size(), OutRef.HashKey.Hash);
+	if (!PayloadSegment.empty() && PayloadSegment.size() != 24)
+	{
+		OutRef.PayloadId = zen::Oid::FromHexString(PayloadSegment);
+
+		if (!OutRef.PayloadId)
+		{
+			return false;
+		}
+	}
+
+	const bool IsOk = zen::ParseHexBytes(HashSegment.data(), HashSegment.size(), OutRef.HashKey.Hash);
 
 	if (!IsOk)
 	{
@@ -170,5 +208,4 @@ HttpStructuredCacheService::ValidateUri(zen::HttpServerRequest& Request, CacheRe
 
 	return true;
 }
-
 }  // namespace zen
diff --git a/zenserver/cache/structuredcache.h b/zenserver/cache/structuredcache.h
index 48b56128f..ead9644f5 100644
--- a/zenserver/cache/structuredcache.h
+++ b/zenserver/cache/structuredcache.h
@@ -15,13 +15,13 @@ class CloudCacheClient;
 /**
  * New-style cache service. Imposes constraints on keys, supports blobs and
  * structured values
- * 
+ *
  * The storage strategy is as follows:
- * 
- * - 
- * 
+ *
  * -
- * 
+ *
+ * -
+ *
  */
 
 class HttpStructuredCacheService : public zen::HttpService
@@ -39,6 +39,7 @@ private:
 	{
 		std::string BucketSegment;
 		IoHash		HashKey;
+		Oid			PayloadId;
 	};
 
 	[[nodiscard]] bool ValidateUri(zen::HttpServerRequest& Request, CacheRef& OutRef);
diff --git a/zenserver/upstream/jupiter.cpp b/zenserver/upstream/jupiter.cpp
index 363ebcd61..319a6f781 100644
--- a/zenserver/upstream/jupiter.cpp
+++ b/zenserver/upstream/jupiter.cpp
@@ -2,6 +2,8 @@
 
 #include "jupiter.h"
 
+#include "cache/cachestore.h"
+
 #include <fmt/format.h>
 #include <zencore/compactbinary.h>
 #include <zencore/iobuffer.h>
@@ -110,23 +112,38 @@ CloudCacheSession::Put(std::string_view BucketId, std::string_view Key, IoBuffer
 }
 
 void
-CloudCacheSession::Put(std::string_view BucketId, std::string_view Key, CbObjectView Data)
+CloudCacheSession::Put(std::string_view BucketId, const IoHash& Key, ZenCacheValue Data)
 {
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl();
-	Uri << "/api/v1/c/ddc/" << m_CacheClient->Namespace() << "/" << BucketId << "/" TESTING_PREFIX << Key;
+	Uri << "/api/v1/c/ddc/" << m_CacheClient->Namespace() << "/" << BucketId << "/" TESTING_PREFIX << Key.ToHexString();
 
 	auto& Session = m_SessionState->Session;
 
-	IoHash	   Hash		= Data.GetHash();
-	MemoryView DataView = Data.GetView();
-
 	std::string Auth;
 	m_CacheClient->AcquireAccessToken(Auth);
 	Session.SetOption(cpr::Url{Uri.c_str()});
-	Session.SetOption(
-		cpr::Header{{"Authorization", Auth}, {"X-Jupiter-IoHash", Hash.ToHexString()}, {"Content-Type", "application/x-ue-cb"}});
-	Session.SetOption(cpr::Body{(const char*)DataView.GetData(), DataView.GetSize()});
+
+	if (Data.IsCompactBinary)
+	{
+		CbObjectView	 Cbo(Data.Value.Data());
+		const IoHash	 Hash	  = Cbo.GetHash();
+		const MemoryView DataView = Cbo.GetView();
+
+		Session.SetOption(
+			cpr::Header{{"Authorization", Auth}, {"X-Jupiter-IoHash", Hash.ToHexString()}, {"Content-Type", "application/x-ue-cb"}});
+
+		Session.SetOption(cpr::Body{(const char*)DataView.GetData(), DataView.GetSize()});
+	}
+	else
+	{
+		const IoHash Hash = IoHash::HashMemory(Data.Value.Data(), Data.Value.Size());
+
+		Session.SetOption(
+			cpr::Header{{"Authorization", Auth}, {"X-Jupiter-IoHash", Hash.ToHexString()}, {"Content-Type", "application/x-ue-cb"}});
+
+		Session.SetOption(cpr::Body{(const char*)Data.Value.Data(), Data.Value.Size()});
+	}
 
 	cpr::Response Response = Session.Put();
 
diff --git a/zenserver/upstream/jupiter.h b/zenserver/upstream/jupiter.h
index a0dbefa3c..2ed458142 100644
--- a/zenserver/upstream/jupiter.h
+++ b/zenserver/upstream/jupiter.h
@@ -11,6 +11,8 @@
 #include <memory>
 #include <vector>
 
+struct ZenCacheValue;
+
 namespace zen {
 namespace detail {
 	struct CloudCacheSessionState;
@@ -56,7 +58,7 @@ public:
 
 	// Structured cache operations
 	IoBuffer Get(std::string_view BucketId, const IoHash& Key);
-	void	 Put(std::string_view BucketId, std::string_view Key, CbObjectView Data);
+	void	 Put(std::string_view BucketId, const IoHash& Key, ZenCacheValue Data);
 
 	std::vector<IoHash> Filter(std::string_view BucketId, const std::vector<IoHash>& ChunkHashes);