remove zenutil dependency in zenremotestore (#547)

* remove dependency to zenutil/workerpools.h from remoteprojectstore.cpp
* remove dependency to zenutil/workerpools.h from buildstoragecache.cpp
* remove unneded include
* move jupiter helpers to zenremotestore
* move parallelwork to zencore
* remove zenutil dependency from zenremotestore
* clean up test project dependencies - use indirect dependencies

5 files changed, 277 insertions, 1 deletions

diff --git a/src/zenremotestore/include/zenremotestore/builds/buildstoragecache.h b/src/zenremotestore/include/zenremotestore/builds/buildstoragecache.h
index 2e8024915..e30270848 100644
--- a/src/zenremotestore/include/zenremotestore/builds/buildstoragecache.h
+++ b/src/zenremotestore/include/zenremotestore/builds/buildstoragecache.h
@@ -11,6 +11,7 @@
 namespace zen {
 
 class HttpClient;
+class WorkerThreadPool;
 
 class BuildStorageCache
 {
@@ -56,7 +57,7 @@ std::unique_ptr<BuildStorageCache> CreateZenBuildStorageCache(HttpClient&					 H
 															  std::string_view				 Namespace,
 															  std::string_view				 Bucket,
 															  const std::filesystem::path&	 TempFolderPath,
-															  bool							 BoostBackgroundThreadCount);
+															  WorkerThreadPool&				 BackgroundWorkerPool);
 
 struct ZenCacheEndpointTestResult
 {
diff --git a/src/zenremotestore/include/zenremotestore/jupiter/jupiterclient.h b/src/zenremotestore/include/zenremotestore/jupiter/jupiterclient.h
new file mode 100644
index 000000000..8a51bd60a
--- /dev/null
+++ b/src/zenremotestore/include/zenremotestore/jupiter/jupiterclient.h
@@ -0,0 +1,56 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <zenbase/refcount.h>
+#include <zencore/logging.h>
+#include <zenhttp/httpclient.h>
+
+#include <chrono>
+
+namespace zen {
+
+class IoBuffer;
+
+struct JupiterClientOptions
+{
+	std::string_view		  Name;
+	std::string_view		  ServiceUrl;
+	std::string_view		  DdcNamespace;
+	std::string_view		  BlobStoreNamespace;
+	std::string_view		  ComputeCluster;
+	std::chrono::milliseconds ConnectTimeout{5000};
+	std::chrono::milliseconds Timeout{};
+	bool					  AssumeHttp2 = false;
+	bool					  AllowResume = false;
+	uint8_t					  RetryCount  = 0;
+};
+
+/**
+ * Jupiter upstream cache client
+ */
+class JupiterClient : public RefCounted
+{
+public:
+	JupiterClient(const JupiterClientOptions& Options, std::function<HttpClientAccessToken()>&& TokenProvider);
+	~JupiterClient();
+
+	std::string_view DefaultDdcNamespace() const { return m_DefaultDdcNamespace; }
+	std::string_view DefaultBlobStoreNamespace() const { return m_DefaultBlobStoreNamespace; }
+	std::string_view ComputeCluster() const { return m_ComputeCluster; }
+	std::string_view ServiceUrl() const { return m_HttpClient.GetBaseUri(); }
+
+	LoggerRef	Logger() { return m_Log; }
+	HttpClient& Client() { return m_HttpClient; }
+
+private:
+	LoggerRef		  m_Log;
+	const std::string m_DefaultDdcNamespace;
+	const std::string m_DefaultBlobStoreNamespace;
+	const std::string m_ComputeCluster;
+	HttpClient		  m_HttpClient;
+
+	friend class JupiterSession;
+};
+
+}  // namespace zen
diff --git a/src/zenremotestore/include/zenremotestore/jupiter/jupiterhost.h b/src/zenremotestore/include/zenremotestore/jupiter/jupiterhost.h
new file mode 100644
index 000000000..3bbc700b8
--- /dev/null
+++ b/src/zenremotestore/include/zenremotestore/jupiter/jupiterhost.h
@@ -0,0 +1,35 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <string>
+#include <string_view>
+#include <vector>
+
+namespace zen {
+
+struct HttpClientSettings;
+
+struct JupiterServerDiscovery
+{
+	struct EndPoint
+	{
+		std::string Name;
+		std::string BaseUrl;
+		bool		AssumeHttp2 = false;
+	};
+	std::vector<EndPoint> ServerEndPoints;
+	std::vector<EndPoint> CacheEndPoints;
+};
+
+JupiterServerDiscovery DiscoverJupiterEndpoints(std::string_view Host, const HttpClientSettings& ClientSettings);
+
+struct JupiterEndpointTestResult
+{
+	bool		Success = false;
+	std::string FailureReason;
+};
+
+JupiterEndpointTestResult TestJupiterEndpoint(std::string_view BaseUrl, const bool AssumeHttp2);
+
+}  // namespace zen
diff --git a/src/zenremotestore/include/zenremotestore/jupiter/jupitersession.h b/src/zenremotestore/include/zenremotestore/jupiter/jupitersession.h
new file mode 100644
index 000000000..b79790f25
--- /dev/null
+++ b/src/zenremotestore/include/zenremotestore/jupiter/jupitersession.h
@@ -0,0 +1,179 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <zencore/iohash.h>
+#include <zencore/logging.h>
+#include <zenhttp/httpclient.h>
+
+#include <set>
+
+namespace zen {
+
+class IoBuffer;
+
+struct JupiterResult
+{
+	IoBuffer	Response;
+	uint64_t	SentBytes{};
+	uint64_t	ReceivedBytes{};
+	double		ElapsedSeconds{};
+	int32_t		ErrorCode{};
+	std::string Reason;
+	bool		Success = false;
+};
+
+struct PutRefResult : JupiterResult
+{
+	std::vector<IoHash> Needs;
+	IoHash				RawHash;
+};
+
+struct FinalizeRefResult : JupiterResult
+{
+	std::vector<IoHash> Needs;
+};
+
+struct JupiterExistsResult : JupiterResult
+{
+	std::set<IoHash> Needs;
+};
+
+struct GetObjectReferencesResult : JupiterResult
+{
+	std::set<IoHash> References;
+};
+
+struct PutBuildPartResult : JupiterResult
+{
+	std::vector<IoHash> Needs;
+	IoHash				RawHash;
+};
+
+struct FinalizeBuildPartResult : JupiterResult
+{
+	std::vector<IoHash> Needs;
+};
+
+/**
+ * Context for performing Jupiter operations
+ *
+ * Maintains an HTTP connection so that subsequent operations don't need to go
+ * through the whole connection setup process
+ *
+ */
+class JupiterSession
+{
+public:
+	JupiterSession(LoggerRef InLog, HttpClient& InHttpClient, bool AllowRedirect);
+	~JupiterSession();
+
+	JupiterResult Authenticate();
+
+	JupiterResult GetRef(std::string_view Namespace, std::string_view BucketId, const IoHash& Key, ZenContentType RefType);
+	JupiterResult GetBlob(std::string_view Namespace, const IoHash& Key);
+	JupiterResult GetCompressedBlob(std::string_view Namespace, const IoHash& Key, std::filesystem::path TempFolderPath = {});
+	JupiterResult GetObject(std::string_view Namespace, const IoHash& Key);
+	JupiterResult GetInlineBlob(std::string_view	  Namespace,
+								std::string_view	  BucketId,
+								const IoHash&		  Key,
+								IoHash&				  OutPayloadHash,
+								std::filesystem::path TempFolderPath = {});
+
+	PutRefResult  PutRef(std::string_view Namespace, std::string_view BucketId, const IoHash& Key, IoBuffer Ref, ZenContentType RefType);
+	JupiterResult PutBlob(std::string_view Namespace, const IoHash& Key, IoBuffer Blob);
+	JupiterResult PutCompressedBlob(std::string_view Namespace, const IoHash& Key, IoBuffer Blob);
+	JupiterResult PutCompressedBlob(std::string_view Namespace, const IoHash& Key, const CompositeBuffer& Blob);
+	JupiterResult PutObject(std::string_view Namespace, const IoHash& Key, IoBuffer Object);
+
+	FinalizeRefResult FinalizeRef(std::string_view Namespace, std::string_view BucketId, const IoHash& Key, const IoHash& RefHah);
+
+	JupiterResult RefExists(std::string_view Namespace, std::string_view BucketId, const IoHash& Key);
+
+	GetObjectReferencesResult GetObjectReferences(std::string_view Namespace, const IoHash& Key);
+
+	JupiterResult BlobExists(std::string_view Namespace, const IoHash& Key);
+	JupiterResult CompressedBlobExists(std::string_view Namespace, const IoHash& Key);
+	JupiterResult ObjectExists(std::string_view Namespace, const IoHash& Key);
+
+	JupiterExistsResult BlobExists(std::string_view Namespace, const std::set<IoHash>& Keys);
+	JupiterExistsResult CompressedBlobExists(std::string_view Namespace, const std::set<IoHash>& Keys);
+	JupiterExistsResult ObjectExists(std::string_view Namespace, const std::set<IoHash>& Keys);
+
+	std::vector<IoHash> Filter(std::string_view Namespace, std::string_view BucketId, const std::vector<IoHash>& ChunkHashes);
+
+	JupiterResult	   ListBuildNamespaces();
+	JupiterResult	   ListBuildBuckets(std::string_view Namespace);
+	JupiterResult	   ListBuilds(std::string_view Namespace, std::string_view BucketId, const IoBuffer& Payload);
+	JupiterResult	   PutBuild(std::string_view Namespace, std::string_view BucketId, const Oid& BuildId, const IoBuffer& Payload);
+	JupiterResult	   GetBuild(std::string_view Namespace, std::string_view BucketId, const Oid& BuildId);
+	JupiterResult	   FinalizeBuild(std::string_view Namespace, std::string_view BucketId, const Oid& BuildId);
+	PutBuildPartResult PutBuildPart(std::string_view Namespace,
+									std::string_view BucketId,
+									const Oid&		 BuildId,
+									const Oid&		 PartId,
+									std::string_view PartName,
+									const IoBuffer&	 Payload);
+	JupiterResult	   GetBuildPart(std::string_view Namespace, std::string_view BucketId, const Oid& BuildId, const Oid& PartId);
+	JupiterResult	   PutBuildBlob(std::string_view	   Namespace,
+									std::string_view	   BucketId,
+									const Oid&			   BuildId,
+									const IoHash&		   Hash,
+									ZenContentType		   ContentType,
+									const CompositeBuffer& Payload);
+	JupiterResult	   GetBuildBlob(std::string_view	  Namespace,
+									std::string_view	  BucketId,
+									const Oid&			  BuildId,
+									const IoHash&		  Hash,
+									std::filesystem::path TempFolderPath,
+									uint64_t			  Offset = 0,
+									uint64_t			  Size	 = (uint64_t)-1);
+
+	JupiterResult			PutMultipartBuildBlob(std::string_view												  Namespace,
+												  std::string_view												  BucketId,
+												  const Oid&													  BuildId,
+												  const IoHash&													  Hash,
+												  ZenContentType												  ContentType,
+												  uint64_t														  PayloadSize,
+												  std::function<IoBuffer(uint64_t Offset, uint64_t Size)>&&		  Transmitter,
+												  std::vector<std::function<JupiterResult(bool& OutIsComplete)>>& OutWorkItems);
+	JupiterResult			GetMultipartBuildBlob(std::string_view												Namespace,
+												  std::string_view												BucketId,
+												  const Oid&													BuildId,
+												  const IoHash&													Hash,
+												  uint64_t														ChunkSize,
+												  std::function<void(uint64_t Offset, const IoBuffer& Chunk)>&& OnReceive,
+												  std::function<void()>&&										OnComplete,
+												  std::vector<std::function<JupiterResult()>>&					OutWorkItems);
+	JupiterResult			PutBlockMetadata(std::string_view Namespace,
+											 std::string_view BucketId,
+											 const Oid&		  BuildId,
+											 const IoHash&	  Hash,
+											 const IoBuffer&  Payload);
+	FinalizeBuildPartResult FinalizeBuildPart(std::string_view Namespace,
+											  std::string_view BucketId,
+											  const Oid&	   BuildId,
+											  const Oid&	   PartId,
+											  const IoHash&	   RawHash);
+	JupiterResult			FindBlocks(std::string_view Namespace, std::string_view BucketId, const Oid& BuildId, uint64_t MaxBlockCount);
+	JupiterResult			GetBlockMetadata(std::string_view Namespace, std::string_view BucketId, const Oid& BuildId, IoBuffer Payload);
+
+	JupiterResult PutBuildPartStats(std::string_view Namespace,
+									std::string_view BucketId,
+									const Oid&		 BuildId,
+									const Oid&		 BuildPartId,
+									IoBuffer		 Payload);
+
+private:
+	inline LoggerRef Log() { return m_Log; }
+
+	JupiterResult CacheTypeExists(std::string_view Namespace, std::string_view TypeId, const IoHash& Key);
+
+	JupiterExistsResult CacheTypeExists(std::string_view Namespace, std::string_view TypeId, const std::set<IoHash>& Keys);
+
+	LoggerRef	m_Log;
+	HttpClient& m_HttpClient;
+	const bool	m_AllowRedirect = false;
+};
+
+}  // namespace zen
diff --git a/src/zenremotestore/include/zenremotestore/projectstore/remoteprojectstore.h b/src/zenremotestore/include/zenremotestore/projectstore/remoteprojectstore.h
index 7e5af5e6b..fbcdde955 100644
--- a/src/zenremotestore/include/zenremotestore/projectstore/remoteprojectstore.h
+++ b/src/zenremotestore/include/zenremotestore/projectstore/remoteprojectstore.h
@@ -126,6 +126,7 @@ RemoteProjectStore::LoadContainerResult BuildContainer(
 	CidStore&																	 ChunkStore,
 	ProjectStore::Project&														 Project,
 	ProjectStore::Oplog&														 Oplog,
+	WorkerThreadPool&															 WorkerPool,
 	size_t																		 MaxBlockSize,
 	size_t																		 MaxChunksPerBlock,
 	size_t																		 MaxChunkEmbedSize,
@@ -153,6 +154,8 @@ RemoteProjectStore::Result SaveOplog(CidStore&				ChunkStore,
 									 RemoteProjectStore&	RemoteStore,
 									 ProjectStore::Project& Project,
 									 ProjectStore::Oplog&	Oplog,
+									 WorkerThreadPool&		NetworkWorkerPool,
+									 WorkerThreadPool&		WorkerPool,
 									 size_t					MaxBlockSize,
 									 size_t					MaxChunksPerBlock,
 									 size_t					MaxChunkEmbedSize,
@@ -165,6 +168,8 @@ RemoteProjectStore::Result SaveOplog(CidStore&				ChunkStore,
 RemoteProjectStore::Result LoadOplog(CidStore&			  ChunkStore,
 									 RemoteProjectStore&  RemoteStore,
 									 ProjectStore::Oplog& Oplog,
+									 WorkerThreadPool&	  NetworkWorkerPool,
+									 WorkerThreadPool&	  WorkerPool,
 									 bool				  ForceDownload,
 									 bool				  IgnoreMissingAttachments,
 									 bool				  CleanOplog,