merge from main

8 files changed, 2610 insertions, 270 deletions

diff --git a/zenserver/upstream/jupiter.cpp b/zenserver/upstream/jupiter.cpp
index 556a2124d..9223ea0f4 100644
--- a/zenserver/upstream/jupiter.cpp
+++ b/zenserver/upstream/jupiter.cpp
@@ -46,13 +46,20 @@ namespace detail {
 
 		void InvalidateAccessToken() { AccessToken = {}; }
 
-		void Reset()
+		void Reset(std::chrono::milliseconds ConnectTimeout, std::chrono::milliseconds Timeout)
 		{
 			Session.SetBody({});
 			Session.SetHeader({});
+			Session.SetConnectTimeout(ConnectTimeout);
+			Session.SetTimeout(Timeout);
 			AccessToken = GetAccessToken();
 		}
 
+		cpr::Session& GetSession() { return Session; }
+
+	private:
+		friend class CloudCacheClient;
+
 		CloudCacheClient&	  OwnerClient;
 		CloudCacheAccessToken AccessToken;
 		cpr::Session		  Session;
@@ -87,12 +94,12 @@ CloudCacheSession::GetDerivedData(std::string_view BucketId, std::string_view Ke
 	}
 
 	ExtendableStringBuilder<256> Uri;
-	Uri << m_CacheClient->ServiceUrl() << "/api/v1/c/ddc/" << m_CacheClient->DdcNamespace() << "/" << BucketId << "/" << Key << ".raw";
+	Uri << m_CacheClient->ServiceUrl() << "/api/v1/c/ddc/" << m_CacheClient->DdcNamespace() << "/" << BucketId << "/" << Key;
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
-	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}});
+	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/octet-stream"}});
 
 	cpr::Response Response = Session.Get();
 	ZEN_DEBUG("GET {}", Response);
@@ -133,10 +140,11 @@ CloudCacheSession::GetRef(std::string_view BucketId, const IoHash& Key, ZenConte
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/refs/" << m_CacheClient->BlobStoreNamespace() << "/" << BucketId << "/"
 		<< Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", ContentType}});
+	Session.SetOption(cpr::Body{});
 
 	cpr::Response Response = Session.Get();
 	ZEN_DEBUG("GET {}", Response);
@@ -168,10 +176,11 @@ CloudCacheSession::GetBlob(const IoHash& Key)
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/octet-stream"}});
+	Session.SetOption(cpr::Body{});
 
 	cpr::Response Response = Session.Get();
 	ZEN_DEBUG("GET {}", Response);
@@ -180,9 +189,14 @@ CloudCacheSession::GetBlob(const IoHash& Key)
 	{
 		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
 	}
+	else if (!VerifyAccessToken(Response.status_code))
+	{
+		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
+	}
 
 	const bool	   Success = Response.status_code == 200;
-	const IoBuffer Buffer  = Success ? IoBufferBuilder::MakeCloneFromMemory(Response.text.data(), Response.text.size()) : IoBuffer();
+	const IoBuffer Buffer =
+		Success && Response.text.size() > 0 ? IoBufferBuilder::MakeCloneFromMemory(Response.text.data(), Response.text.size()) : IoBuffer();
 
 	return {.Response = Buffer, .Bytes = Response.downloaded_bytes, .ElapsedSeconds = Response.elapsed, .Success = Success};
 }
@@ -199,10 +213,11 @@ CloudCacheSession::GetCompressedBlob(const IoHash& Key)
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compressed-blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/x-ue-comp"}});
+	Session.SetOption(cpr::Body{});
 
 	cpr::Response Response = Session.Get();
 	ZEN_DEBUG("GET {}", Response);
@@ -211,6 +226,10 @@ CloudCacheSession::GetCompressedBlob(const IoHash& Key)
 	{
 		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
 	}
+	else if (!VerifyAccessToken(Response.status_code))
+	{
+		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
+	}
 
 	const bool	   Success = Response.status_code == 200;
 	const IoBuffer Buffer  = Success ? IoBufferBuilder::MakeCloneFromMemory(Response.text.data(), Response.text.size()) : IoBuffer();
@@ -230,10 +249,11 @@ CloudCacheSession::GetObject(const IoHash& Key)
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/objects/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/x-ue-cb"}});
+	Session.SetOption(cpr::Body{});
 
 	cpr::Response Response = Session.Get();
 	ZEN_DEBUG("GET {}", Response);
@@ -242,6 +262,10 @@ CloudCacheSession::GetObject(const IoHash& Key)
 	{
 		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
 	}
+	else if (!VerifyAccessToken(Response.status_code))
+	{
+		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
+	}
 
 	const bool	   Success = Response.status_code == 200;
 	const IoBuffer Buffer  = Success ? IoBufferBuilder::MakeCloneFromMemory(Response.text.data(), Response.text.size()) : IoBuffer();
@@ -263,7 +287,7 @@ CloudCacheSession::PutDerivedData(std::string_view BucketId, std::string_view Ke
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/c/ddc/" << m_CacheClient->DdcNamespace() << "/" << BucketId << "/" << Key;
 
-	auto& Session = m_SessionState->Session;
+	auto& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value},
@@ -314,7 +338,7 @@ CloudCacheSession::PutRef(std::string_view BucketId, const IoHash& Key, IoBuffer
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/refs/" << m_CacheClient->BlobStoreNamespace() << "/" << BucketId << "/"
 		<< Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(
@@ -377,12 +401,13 @@ CloudCacheSession::FinalizeRef(std::string_view BucketId, const IoHash& Key, con
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/refs/" << m_CacheClient->BlobStoreNamespace() << "/" << BucketId << "/"
 		<< Key.ToHexString() << "/finalize/" << RefHash.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value},
 								  {"X-Jupiter-IoHash", RefHash.ToHexString()},
 								  {"Content-Type", "application/x-ue-cb"}});
+	Session.SetBody(cpr::Body{});
 
 	cpr::Response Response = Session.Post();
 	ZEN_DEBUG("POST {}", Response);
@@ -436,7 +461,7 @@ CloudCacheSession::PutBlob(const IoHash& Key, IoBuffer Blob)
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/octet-stream"}});
@@ -471,7 +496,7 @@ CloudCacheSession::PutCompressedBlob(const IoHash& Key, IoBuffer Blob)
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compressed-blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/x-ue-comp"}});
@@ -506,7 +531,7 @@ CloudCacheSession::PutObject(const IoHash& Key, IoBuffer Object)
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/objects/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/x-ue-cb"}});
@@ -542,10 +567,11 @@ CloudCacheSession::RefExists(std::string_view BucketId, const IoHash& Key)
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/refs/" << m_CacheClient->BlobStoreNamespace() << "/" << BucketId << "/"
 		<< Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}});
+	Session.SetOption(cpr::Body{});
 
 	cpr::Response Response = Session.Head();
 	ZEN_DEBUG("HEAD {}", Response);
@@ -565,67 +591,41 @@ CloudCacheSession::RefExists(std::string_view BucketId, const IoHash& Key)
 CloudCacheResult
 CloudCacheSession::BlobExists(const IoHash& Key)
 {
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
-
-	ExtendableStringBuilder<256> Uri;
-	Uri << m_CacheClient->ServiceUrl() << "/api/v1/blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
-
-	cpr::Session& Session = m_SessionState->Session;
-
-	Session.SetOption(cpr::Url{Uri.c_str()});
-
-	cpr::Response Response = Session.Head();
-	ZEN_DEBUG("HEAD {}", Response);
-
-	if (Response.error)
-	{
-		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
-	}
-	else if (!VerifyAccessToken(Response.status_code))
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
-
-	return {.ElapsedSeconds = Response.elapsed, .Success = Response.status_code == 200};
+	return CacheTypeExists("blobs"sv, Key);
 }
 
 CloudCacheResult
 CloudCacheSession::CompressedBlobExists(const IoHash& Key)
 {
-	const CloudCacheAccessToken& AccessToken = GetAccessToken();
-	if (!AccessToken.IsValid())
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
-
-	ExtendableStringBuilder<256> Uri;
-	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compressed-blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
-
-	cpr::Session& Session = m_SessionState->Session;
+	return CacheTypeExists("compressed-blobs"sv, Key);
+}
 
-	Session.SetOption(cpr::Url{Uri.c_str()});
+CloudCacheResult
+CloudCacheSession::ObjectExists(const IoHash& Key)
+{
+	return CacheTypeExists("objects"sv, Key);
+}
 
-	cpr::Response Response = Session.Head();
-	ZEN_DEBUG("HEAD {}", Response);
+CloudCacheExistsResult
+CloudCacheSession::BlobExists(const std::set<IoHash>& Keys)
+{
+	return CacheTypeExists("blobs"sv, Keys);
+}
 
-	if (Response.error)
-	{
-		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
-	}
-	else if (!VerifyAccessToken(Response.status_code))
-	{
-		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
-	}
+CloudCacheExistsResult
+CloudCacheSession::CompressedBlobExists(const std::set<IoHash>& Keys)
+{
+	return CacheTypeExists("compressed-blobs"sv, Keys);
+}
 
-	return {.ElapsedSeconds = Response.elapsed, .Success = Response.status_code == 200};
+CloudCacheExistsResult
+CloudCacheSession::ObjectExists(const std::set<IoHash>& Keys)
+{
+	return CacheTypeExists("objects"sv, Keys);
 }
 
 CloudCacheResult
-CloudCacheSession::ObjectExists(const IoHash& Key)
+CloudCacheSession::PostComputeTasks(std::string_view ChannelId, IoBuffer TasksData)
 {
 	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 	if (!AccessToken.IsValid())
@@ -634,14 +634,16 @@ CloudCacheSession::ObjectExists(const IoHash& Key)
 	}
 
 	ExtendableStringBuilder<256> Uri;
-	Uri << m_CacheClient->ServiceUrl() << "/api/v1/objects/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
+	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compute/" << ChannelId;
 
-	cpr::Session& Session = m_SessionState->Session;
+	auto& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
+	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/x-ue-cb"}});
+	Session.SetBody(cpr::Body{(const char*)TasksData.Data(), TasksData.Size()});
 
-	cpr::Response Response = Session.Head();
-	ZEN_DEBUG("HEAD {}", Response);
+	cpr::Response Response = Session.Post();
+	ZEN_DEBUG("POST {}", Response);
 
 	if (Response.error)
 	{
@@ -656,7 +658,7 @@ CloudCacheSession::ObjectExists(const IoHash& Key)
 }
 
 CloudCacheResult
-CloudCacheSession::PostComputeTasks(std::string_view ChannelId, IoBuffer TasksData)
+CloudCacheSession::GetComputeUpdates(std::string_view ChannelId, const uint32_t WaitSeconds)
 {
 	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 	if (!AccessToken.IsValid())
@@ -665,13 +667,12 @@ CloudCacheSession::PostComputeTasks(std::string_view ChannelId, IoBuffer TasksDa
 	}
 
 	ExtendableStringBuilder<256> Uri;
-	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compute/" << ChannelId;
+	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compute/" << ChannelId << "/updates?wait=" << WaitSeconds;
 
-	auto& Session = m_SessionState->Session;
+	auto& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
-	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/x-ue-cb"}});
-	Session.SetBody(cpr::Body{(const char*)TasksData.Data(), TasksData.Size()});
+	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/x-ue-cb"}});
 
 	cpr::Response Response = Session.Post();
 	ZEN_DEBUG("POST {}", Response);
@@ -692,7 +693,7 @@ CloudCacheSession::PostComputeTasks(std::string_view ChannelId, IoBuffer TasksDa
 }
 
 CloudCacheResult
-CloudCacheSession::GetComputeUpdates(std::string_view ChannelId, const uint32_t WaitSeconds)
+CloudCacheSession::GetObjectTree(const IoHash& Key)
 {
 	const CloudCacheAccessToken& AccessToken = GetAccessToken();
 	if (!AccessToken.IsValid())
@@ -701,15 +702,15 @@ CloudCacheSession::GetComputeUpdates(std::string_view ChannelId, const uint32_t
 	}
 
 	ExtendableStringBuilder<256> Uri;
-	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compute/" << ChannelId << "/updates?wait=" << WaitSeconds;
+	Uri << m_CacheClient->ServiceUrl() << "/api/v1/objects/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString() << "/tree";
 
-	auto& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
-	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/x-ue-cb"}});
+	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/octet-stream"}});
 
-	cpr::Response Response = Session.Post();
-	ZEN_DEBUG("POST {}", Response);
+	cpr::Response Response = Session.Get();
+	ZEN_DEBUG("GET {}", Response);
 
 	if (Response.error)
 	{
@@ -755,6 +756,92 @@ CloudCacheSession::VerifyAccessToken(long StatusCode)
 	return true;
 }
 
+CloudCacheResult
+CloudCacheSession::CacheTypeExists(std::string_view TypeId, const IoHash& Key)
+{
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
+	if (!AccessToken.IsValid())
+	{
+		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
+	}
+
+	ExtendableStringBuilder<256> Uri;
+	Uri << m_CacheClient->ServiceUrl() << "/api/v1/" << TypeId << "/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
+
+	cpr::Session& Session = m_SessionState->GetSession();
+
+	Session.SetOption(cpr::Url{Uri.c_str()});
+	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}});
+
+	cpr::Response Response = Session.Head();
+	ZEN_DEBUG("HEAD {}", Response);
+
+	if (Response.error)
+	{
+		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message};
+	}
+	else if (!VerifyAccessToken(Response.status_code))
+	{
+		return {.ErrorCode = 401, .Reason = std::string("Invalid access token")};
+	}
+
+	return {.ElapsedSeconds = Response.elapsed, .Success = Response.status_code == 200};
+}
+
+CloudCacheExistsResult
+CloudCacheSession::CacheTypeExists(std::string_view TypeId, const std::set<IoHash>& Keys)
+{
+	const CloudCacheAccessToken& AccessToken = GetAccessToken();
+	if (!AccessToken.IsValid())
+	{
+		return {CloudCacheResult{.ErrorCode = 401, .Reason = std::string("Invalid access token")}};
+	}
+
+	ExtendableStringBuilder<256> Query;
+	for (const auto& Key : Keys)
+	{
+		Query << (Query.Size() != 0 ? "&id=" : "id=") << Key.ToHexString();
+	}
+
+	ExtendableStringBuilder<256> Uri;
+	Uri << m_CacheClient->ServiceUrl() << "/api/v1/" << TypeId << "/" << m_CacheClient->BlobStoreNamespace() << "/exists?" << Query;
+
+	cpr::Session& Session = m_SessionState->GetSession();
+
+	Session.SetOption(cpr::Url{Uri.c_str()});
+	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/x-ue-cb"}});
+
+	cpr::Response Response = Session.Post();
+	ZEN_DEBUG("POST {}", Response);
+
+	if (Response.error)
+	{
+		return {CloudCacheResult{.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = Response.error.message}};
+	}
+	else if (!VerifyAccessToken(Response.status_code))
+	{
+		return {CloudCacheResult{.ErrorCode = 401, .Reason = std::string("Invalid access token")}};
+	}
+
+	CloudCacheExistsResult Result{
+		CloudCacheResult{.Bytes = Response.downloaded_bytes, .ElapsedSeconds = Response.elapsed, .Success = Response.status_code == 200}};
+
+	if (Result.Success)
+	{
+		IoBuffer	   Buffer		  = IoBuffer(zen::IoBuffer::Wrap, Response.text.data(), Response.text.size());
+		const CbObject ExistsResponse = LoadCompactBinaryObject(Buffer);
+		for (auto& Item : ExistsResponse["id"sv])
+		{
+			if (Item.IsHash())
+			{
+				Result.Have.insert(Item.AsHash());
+			}
+		}
+	}
+
+	return Result;
+}
+
 //////////////////////////////////////////////////////////////////////////
 //
 // ServiceUrl:		https://jupiter.devtools.epicgames.com
@@ -772,7 +859,16 @@ CloudCacheClient::CloudCacheClient(const CloudCacheClientOptions& Options)
 , m_BlobStoreNamespace(Options.BlobStoreNamespace)
 , m_OAuthClientId(Options.OAuthClientId)
 , m_OAuthSecret(Options.OAuthSecret)
+, m_AccessToken(Options.AccessToken)
+, m_ConnectTimeout(Options.ConnectTimeout)
+, m_Timeout(Options.Timeout)
 {
+	if (!Options.AccessToken.empty())
+	{
+		// If an access token was provided, OAuth settings are not used.
+		return;
+	}
+
 	if (!Options.OAuthProvider.starts_with("http://"sv) && !Options.OAuthProvider.starts_with("https://"sv))
 	{
 		ZEN_WARN("bad provider specification: '{}' - must be fully qualified", Options.OAuthProvider);
@@ -822,6 +918,12 @@ CloudCacheClient::AcquireAccessToken()
 {
 	using namespace std::chrono;
 
+	// If an access token was provided, return it instead of querying OAuth
+	if (!m_AccessToken.empty())
+	{
+		return {m_AccessToken, steady_clock::time_point::max()};
+	}
+
 	ExtendableStringBuilder<128> OAuthFormData;
 	OAuthFormData << "client_id=" << m_OAuthClientId << "&scope=cache_access&grant_type=client_credentials&client_secret=" << m_OAuthSecret;
 
@@ -862,7 +964,7 @@ CloudCacheClient::AllocSessionState()
 		State = new detail::CloudCacheSessionState(*this);
 	}
 
-	State->Reset();
+	State->Reset(m_ConnectTimeout, m_Timeout);
 
 	return State;
 }
diff --git a/zenserver/upstream/jupiter.h b/zenserver/upstream/jupiter.h
index 9471ef64f..68c7361e0 100644
--- a/zenserver/upstream/jupiter.h
+++ b/zenserver/upstream/jupiter.h
@@ -12,6 +12,7 @@
 #include <chrono>
 #include <list>
 #include <memory>
+#include <set>
 #include <vector>
 
 struct ZenCacheValue;
@@ -64,6 +65,11 @@ struct FinalizeRefResult : CloudCacheResult
 	std::vector<IoHash> Needs;
 };
 
+struct CloudCacheExistsResult : CloudCacheResult
+{
+	std::set<IoHash> Have;
+};
+
 /**
  * Context for performing Jupiter operations
  *
@@ -95,12 +101,18 @@ public:
 	FinalizeRefResult FinalizeRef(std::string_view BucketId, const IoHash& Key, const IoHash& RefHah);
 
 	CloudCacheResult RefExists(std::string_view BucketId, const IoHash& Key);
+
 	CloudCacheResult BlobExists(const IoHash& Key);
 	CloudCacheResult CompressedBlobExists(const IoHash& Key);
 	CloudCacheResult ObjectExists(const IoHash& Key);
 
+	CloudCacheExistsResult BlobExists(const std::set<IoHash>& Keys);
+	CloudCacheExistsResult CompressedBlobExists(const std::set<IoHash>& Keys);
+	CloudCacheExistsResult ObjectExists(const std::set<IoHash>& Keys);
+
 	CloudCacheResult PostComputeTasks(std::string_view ChannelId, IoBuffer TasksData);
 	CloudCacheResult GetComputeUpdates(std::string_view ChannelId, const uint32_t WaitSeconds = 0);
+	CloudCacheResult GetObjectTree(const IoHash& Key);
 
 	std::vector<IoHash> Filter(std::string_view BucketId, const std::vector<IoHash>& ChunkHashes);
 
@@ -109,6 +121,10 @@ private:
 	const CloudCacheAccessToken& GetAccessToken();
 	bool						 VerifyAccessToken(long StatusCode);
 
+	CloudCacheResult CacheTypeExists(std::string_view TypeId, const IoHash& Key);
+
+	CloudCacheExistsResult CacheTypeExists(std::string_view TypeId, const std::set<IoHash>& Keys);
+
 	spdlog::logger&					m_Log;
 	RefPtr<CloudCacheClient>		m_CacheClient;
 	detail::CloudCacheSessionState* m_SessionState;
@@ -116,13 +132,16 @@ private:
 
 struct CloudCacheClientOptions
 {
-	std::string_view ServiceUrl;
-	std::string_view DdcNamespace;
-	std::string_view BlobStoreNamespace;
-	std::string_view OAuthProvider;
-	std::string_view OAuthClientId;
-	std::string_view OAuthSecret;
-	bool			 UseLegacyDdc = false;
+	std::string_view		  ServiceUrl;
+	std::string_view		  DdcNamespace;
+	std::string_view		  BlobStoreNamespace;
+	std::string_view		  OAuthProvider;
+	std::string_view		  OAuthClientId;
+	std::string_view		  OAuthSecret;
+	std::string_view		  AccessToken;
+	std::chrono::milliseconds ConnectTimeout{5000};
+	std::chrono::milliseconds Timeout{};
+	bool					  UseLegacyDdc = false;
 };
 
 /**
@@ -143,16 +162,19 @@ public:
 	spdlog::logger& Logger() { return m_Log; }
 
 private:
-	spdlog::logger& m_Log;
-	std::string		m_ServiceUrl;
-	std::string		m_OAuthDomain;
-	std::string		m_OAuthUriPath;
-	std::string		m_OAuthFullUri;
-	std::string		m_DdcNamespace;
-	std::string		m_BlobStoreNamespace;
-	std::string		m_OAuthClientId;
-	std::string		m_OAuthSecret;
-	bool			m_IsValid = false;
+	spdlog::logger&			  m_Log;
+	std::string				  m_ServiceUrl;
+	std::string				  m_OAuthDomain;
+	std::string				  m_OAuthUriPath;
+	std::string				  m_OAuthFullUri;
+	std::string				  m_DdcNamespace;
+	std::string				  m_BlobStoreNamespace;
+	std::string				  m_OAuthClientId;
+	std::string				  m_OAuthSecret;
+	std::string				  m_AccessToken;
+	std::chrono::milliseconds m_ConnectTimeout{};
+	std::chrono::milliseconds m_Timeout{};
+	bool					  m_IsValid = false;
 
 	RwLock									   m_SessionStateLock;
 	std::list<detail::CloudCacheSessionState*> m_SessionStateCache;
diff --git a/zenserver/upstream/upstreamapply.cpp b/zenserver/upstream/upstreamapply.cpp
new file mode 100644
index 000000000..651c2dcc8
--- /dev/null
+++ b/zenserver/upstream/upstreamapply.cpp
@@ -0,0 +1,1559 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include "upstreamapply.h"
+#include "jupiter.h"
+#include "zen.h"
+
+#include <zencore/blockingqueue.h>
+#include <zencore/compactbinary.h>
+#include <zencore/compactbinarybuilder.h>
+#include <zencore/compactbinarypackage.h>
+#include <zencore/compactbinaryvalidation.h>
+#include <zencore/compress.h>
+#include <zencore/fmtutils.h>
+#include <zencore/session.h>
+#include <zencore/stats.h>
+#include <zencore/stream.h>
+#include <zencore/timer.h>
+
+#include <zenstore/cas.h>
+#include <zenstore/cidstore.h>
+
+#include "cache/structuredcachestore.h"
+#include "diag/logging.h"
+
+#include <fmt/format.h>
+
+#include <algorithm>
+#include <atomic>
+#include <map>
+#include <set>
+#include <stack>
+#include <thread>
+#include <unordered_map>
+
+namespace zen {
+
+using namespace std::literals;
+
+namespace detail {
+
+	class HordeUpstreamApplyEndpoint final : public UpstreamApplyEndpoint
+	{
+	public:
+		HordeUpstreamApplyEndpoint(const CloudCacheClientOptions& Options, CasStore& CasStore, CidStore& CidStore)
+		: m_Log(logging::Get("upstream-apply"))
+		, m_CasStore(CasStore)
+		, m_CidStore(CidStore)
+		{
+			using namespace fmt::literals;
+			m_DisplayName = "Horde - '{}'"_format(Options.ServiceUrl);
+			m_Client	  = new CloudCacheClient(Options);
+			m_ChannelId	  = "zen-{}"_format(zen::GetSessionIdString());
+		}
+
+		virtual ~HordeUpstreamApplyEndpoint() = default;
+
+		virtual UpstreamEndpointHealth Initialize() override { return CheckHealth(); }
+
+		virtual bool IsHealthy() const override { return m_HealthOk.load(); }
+
+		virtual UpstreamEndpointHealth CheckHealth() override
+		{
+			try
+			{
+				CloudCacheSession Session(m_Client);
+				CloudCacheResult  Result = Session.Authenticate();
+
+				m_HealthOk = Result.ErrorCode == 0;
+
+				return {.Reason = std::move(Result.Reason), .Ok = Result.Success};
+			}
+			catch (std::exception& Err)
+			{
+				return {.Reason = Err.what(), .Ok = false};
+			}
+		}
+
+		virtual std::string_view DisplayName() const override { return m_DisplayName; }
+
+		virtual PostUpstreamApplyResult PostApply(const UpstreamApplyRecord& ApplyRecord) override
+		{
+			int64_t Bytes{};
+			double	ElapsedSeconds{};
+
+			try
+			{
+				UpstreamData UpstreamData;
+				if (!ProcessApplyKey(ApplyRecord, UpstreamData))
+				{
+					return {.Error{.ErrorCode = -1, .Reason = "Failed to generate task data"}};
+				}
+
+				{
+					std::scoped_lock Lock(m_TaskMutex);
+					if (m_PendingTasks.contains(UpstreamData.TaskId))
+					{
+						// Pending task is already queued, return success
+						return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Success = true};
+					}
+					m_PendingTasks[UpstreamData.TaskId] = ApplyRecord;
+				}
+
+				CloudCacheSession Session(m_Client);
+
+				{
+					CloudCacheResult Result = BatchPutBlobsIfMissing(Session, UpstreamData.Blobs);
+					Bytes += Result.Bytes;
+					ElapsedSeconds += Result.ElapsedSeconds;
+					if (!Result.Success)
+					{
+						return {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)},
+								.Bytes			= Bytes,
+								.ElapsedSeconds = ElapsedSeconds};
+					}
+					UpstreamData.Blobs.clear();
+				}
+
+				{
+					CloudCacheResult Result = BatchPutObjectsIfMissing(Session, UpstreamData.Objects);
+					Bytes += Result.Bytes;
+					ElapsedSeconds += Result.ElapsedSeconds;
+					if (!Result.Success)
+					{
+						return {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)},
+								.Bytes			= Bytes,
+								.ElapsedSeconds = ElapsedSeconds};
+					}
+					UpstreamData.Objects.clear();
+				}
+
+				CbObjectWriter Writer;
+				Writer.AddObjectAttachment("r"sv, UpstreamData.RequirementsId);
+				Writer.BeginArray("t"sv);
+				Writer.AddObjectAttachment(UpstreamData.TaskId);
+				Writer.EndArray();
+				IoBuffer TasksData = Writer.Save().GetBuffer().AsIoBuffer();
+
+				CloudCacheResult Result = Session.PostComputeTasks(m_ChannelId, TasksData);
+				Bytes += Result.Bytes;
+				ElapsedSeconds += Result.ElapsedSeconds;
+				if (!Result.Success)
+				{
+					{
+						std::scoped_lock Lock(m_TaskMutex);
+						m_PendingTasks.erase(UpstreamData.TaskId);
+					}
+
+					return {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)},
+							.Bytes			= Bytes,
+							.ElapsedSeconds = ElapsedSeconds};
+				}
+
+				return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Success = true};
+			}
+			catch (std::exception& Err)
+			{
+				m_HealthOk = false;
+				return {.Error{.ErrorCode = -1, .Reason = Err.what()}, .Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds};
+			}
+		}
+
+		[[nodiscard]] CloudCacheResult BatchPutBlobsIfMissing(CloudCacheSession& Session, const std::map<IoHash, IoBuffer>& Blobs)
+		{
+			if (Blobs.size() == 0)
+			{
+				return {.Success = true};
+			}
+
+			int64_t Bytes{};
+			double	ElapsedSeconds{};
+
+			// Batch check for missing blobs
+			std::set<IoHash> Keys;
+			for (const auto& It : Blobs)
+			{
+				Keys.insert(It.first);
+			}
+
+			CloudCacheExistsResult ExistsResult = Session.BlobExists(Keys);
+			ElapsedSeconds += ExistsResult.ElapsedSeconds;
+			if (ExistsResult.ErrorCode != 0)
+			{
+				return {.Bytes			= Bytes,
+						.ElapsedSeconds = ElapsedSeconds,
+						.ErrorCode		= ExistsResult.ErrorCode,
+						.Reason			= std::move(ExistsResult.Reason)};
+			}
+
+			// TODO: Batch upload missing blobs
+
+			for (const auto& It : Blobs)
+			{
+				if (ExistsResult.Have.contains(It.first))
+				{
+					continue;
+				}
+
+				CloudCacheResult Result = Session.PutBlob(It.first, It.second);
+				Bytes += Result.Bytes;
+				ElapsedSeconds += Result.ElapsedSeconds;
+				if (!Result.Success)
+				{
+					return {.Bytes			= Bytes,
+							.ElapsedSeconds = ElapsedSeconds,
+							.ErrorCode		= Result.ErrorCode,
+							.Reason			= std::move(Result.Reason)};
+				}
+			}
+
+			return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Success = true};
+		}
+
+		[[nodiscard]] CloudCacheResult BatchPutObjectsIfMissing(CloudCacheSession& Session, const std::map<IoHash, CbObject>& Objects)
+		{
+			if (Objects.size() == 0)
+			{
+				return {.Success = true};
+			}
+
+			int64_t Bytes{};
+			double	ElapsedSeconds{};
+
+			// Batch check for missing objects
+			std::set<IoHash> Keys;
+			for (const auto& It : Objects)
+			{
+				Keys.insert(It.first);
+			}
+
+			CloudCacheExistsResult ExistsResult = Session.ObjectExists(Keys);
+			ElapsedSeconds += ExistsResult.ElapsedSeconds;
+			if (ExistsResult.ErrorCode != 0)
+			{
+				return {.Bytes			= Bytes,
+						.ElapsedSeconds = ElapsedSeconds,
+						.ErrorCode		= ExistsResult.ErrorCode,
+						.Reason			= std::move(ExistsResult.Reason)};
+			}
+
+			// TODO: Batch upload missing objects
+
+			for (const auto& It : Objects)
+			{
+				if (ExistsResult.Have.contains(It.first))
+				{
+					continue;
+				}
+
+				CloudCacheResult Result = Session.PutObject(It.first, It.second.GetBuffer().AsIoBuffer());
+				Bytes += Result.Bytes;
+				ElapsedSeconds += Result.ElapsedSeconds;
+				if (!Result.Success)
+				{
+					return {.Bytes			= Bytes,
+							.ElapsedSeconds = ElapsedSeconds,
+							.ErrorCode		= Result.ErrorCode,
+							.Reason			= std::move(Result.Reason)};
+				}
+			}
+
+			return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Success = true};
+		}
+
+		enum class ComputeTaskState : int32_t
+		{
+			Queued	  = 0,
+			Executing = 1,
+			Complete  = 2,
+		};
+
+		enum class ComputeTaskOutcome : int32_t
+		{
+			Success		 = 0,
+			Failed		 = 1,
+			Cancelled	 = 2,
+			NoResult	 = 3,
+			Exipred		 = 4,
+			BlobNotFound = 5,
+			Exception	 = 6,
+		};
+
+		std::string_view ComputeTaskOutcomeToString(const ComputeTaskOutcome Outcome)
+		{
+			switch (Outcome)
+			{
+				case ComputeTaskOutcome::Success:
+					return "Success"sv;
+				case ComputeTaskOutcome::Failed:
+					return "Failed"sv;
+				case ComputeTaskOutcome::Cancelled:
+					return "Cancelled"sv;
+				case ComputeTaskOutcome::NoResult:
+					return "NoResult"sv;
+				case ComputeTaskOutcome::Exipred:
+					return "Exipred"sv;
+				case ComputeTaskOutcome::BlobNotFound:
+					return "BlobNotFound"sv;
+				case ComputeTaskOutcome::Exception:
+					return "Exception"sv;
+			};
+			return "Unknown"sv;
+		}
+
+		virtual GetUpstreamApplyUpdatesResult GetUpdates() override
+		{
+			int64_t				   Bytes{};
+			double				   ElapsedSeconds{};
+			UpstreamApplyCompleted CompletedTasks;
+
+			{
+				std::scoped_lock Lock(m_TaskMutex);
+				if (m_PendingTasks.empty())
+				{
+					// Nothing to do.
+					return {.Success = true};
+				}
+			}
+
+			try
+			{
+				CloudCacheSession Session(m_Client);
+
+				CloudCacheResult UpdatesResult = Session.GetComputeUpdates(m_ChannelId);
+				Bytes += UpdatesResult.Bytes;
+				ElapsedSeconds += UpdatesResult.ElapsedSeconds;
+				if (UpdatesResult.ErrorCode != 0)
+				{
+					return {.Error{.ErrorCode = UpdatesResult.ErrorCode, .Reason = std::move(UpdatesResult.Reason)},
+							.Bytes			= Bytes,
+							.ElapsedSeconds = ElapsedSeconds};
+				}
+
+				if (!UpdatesResult.Success)
+				{
+					return {.Error{.ErrorCode = -1, .Reason = "Failed get task updates"}, .Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds};
+				}
+
+				CbObject TaskStatus = LoadCompactBinaryObject(UpdatesResult.Response);
+
+				// zen::StringBuilder<4096> ObjStr;
+				// zen::CompactBinaryToJson(TaskStatus, ObjStr);
+
+				for (auto& It : TaskStatus["u"sv])
+				{
+					CbObjectView		   Status = It.AsObjectView();
+					const ComputeTaskState State  = (ComputeTaskState)Status["s"sv].AsInt32();
+
+					const std::string_view AgentId = TaskStatus["a"sv].AsString();
+					const std::string_view LeaseId = TaskStatus["l"sv].AsString();
+
+					// Only care about completed tasks
+					if (State != ComputeTaskState::Complete)
+					{
+						continue;
+					}
+
+					const IoHash TaskId = Status["h"sv].AsObjectAttachment();
+
+					IoHash WorkerId;
+					IoHash ActionId;
+
+					{
+						std::scoped_lock Lock(m_TaskMutex);
+						auto			 TaskIt = m_PendingTasks.find(TaskId);
+						if (TaskIt == m_PendingTasks.end())
+						{
+							continue;
+						}
+						WorkerId = TaskIt->second.WorkerDescriptor.GetHash();
+						ActionId = TaskIt->second.Action.GetHash();
+						m_PendingTasks.erase(TaskIt);
+					}
+
+					GetUpstreamApplyResult Result = ProcessTaskStatus(Status, Session);
+					Bytes += Result.Bytes;
+					ElapsedSeconds += Result.ElapsedSeconds;
+
+					CompletedTasks[WorkerId][ActionId] = std::move(Result);
+				}
+
+				return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Completed = std::move(CompletedTasks), .Success = true};
+			}
+			catch (std::exception& Err)
+			{
+				m_HealthOk = false;
+				return {
+					.Error{.ErrorCode = -1, .Reason = Err.what()},
+					.Bytes			= Bytes,
+					.ElapsedSeconds = ElapsedSeconds,
+					.Completed		= std::move(CompletedTasks),
+				};
+			}
+		}
+
+		virtual UpstreamApplyEndpointStats& Stats() override { return m_Stats; }
+
+	private:
+		spdlog::logger& Log() { return m_Log; }
+
+		CasStore&				   m_CasStore;
+		CidStore&				   m_CidStore;
+		spdlog::logger&			   m_Log;
+		std::string				   m_DisplayName;
+		RefPtr<CloudCacheClient>   m_Client;
+		UpstreamApplyEndpointStats m_Stats;
+		std::atomic_bool		   m_HealthOk{false};
+		std::string				   m_ChannelId;
+
+		std::mutex										m_TaskMutex;
+		std::unordered_map<IoHash, UpstreamApplyRecord> m_PendingTasks;
+
+		struct UpstreamData
+		{
+			std::map<IoHash, IoBuffer> Blobs;
+			std::map<IoHash, CbObject> Objects;
+			IoHash					   TaskId;
+			IoHash					   RequirementsId;
+		};
+
+		struct UpstreamDirectory
+		{
+			std::filesystem::path					 Path;
+			std::map<std::string, UpstreamDirectory> Directories;
+			std::set<std::string>					 Files;
+		};
+
+		[[nodiscard]] GetUpstreamApplyResult ProcessTaskStatus(const CbObjectView& TaskStatus, CloudCacheSession& Session)
+		{
+			try
+			{
+				const ComputeTaskOutcome Outcome = (ComputeTaskOutcome)TaskStatus["o"sv].AsInt32();
+
+				if (Outcome != ComputeTaskOutcome::Success)
+				{
+					using namespace fmt::literals;
+					const std::string_view Detail = TaskStatus["d"sv].AsString();
+					if (!Detail.empty())
+					{
+						return {.Error{.ErrorCode = -1,
+									   .Reason	  = "Task {}: {}"_format(ComputeTaskOutcomeToString(Outcome), std::string(Detail))}};
+					}
+					return {.Error{.ErrorCode = -1, .Reason = "Task {}"_format(ComputeTaskOutcomeToString(Outcome))}};
+				}
+
+				const IoHash		   TaskId	  = TaskStatus["h"sv].AsObjectAttachment();
+				const DateTime		   Time		  = TaskStatus["t"sv].AsDateTime();
+				const IoHash		   ResultHash = TaskStatus["r"sv].AsObjectAttachment();
+				const std::string_view AgentId	  = TaskStatus["a"sv].AsString();
+				const std::string_view LeaseId	  = TaskStatus["l"sv].AsString();
+
+				int64_t Bytes{};
+				double	ElapsedSeconds{};
+
+				// Get Result object and all Object Attachments + Binary Attachment IDs
+				CloudCacheResult ObjectTreeResult = Session.GetObjectTree(ResultHash);
+				Bytes += ObjectTreeResult.Bytes;
+				ElapsedSeconds += ObjectTreeResult.ElapsedSeconds;
+
+				if (ObjectTreeResult.ErrorCode != 0)
+				{
+					return {.Error{.ErrorCode = ObjectTreeResult.ErrorCode, .Reason = std::move(ObjectTreeResult.Reason)},
+							.Bytes			= Bytes,
+							.ElapsedSeconds = ElapsedSeconds};
+				}
+
+				if (!ObjectTreeResult.Success)
+				{
+					return {.Error{.ErrorCode = -1, .Reason = "Failed to get result object data"},
+							.Bytes			= Bytes,
+							.ElapsedSeconds = ElapsedSeconds};
+				}
+
+				std::map<IoHash, IoBuffer> TreeObjectData;
+				std::map<IoHash, IoBuffer> TreeBinaryData;
+
+				MemoryView ResponseView = ObjectTreeResult.Response;
+				while (ResponseView.GetSize() > 0)
+				{
+					CbFieldView Field = CbFieldView(ResponseView.GetData());
+					ResponseView += Field.GetSize();
+					if (Field.IsObjectAttachment())
+					{
+						const IoHash Hash = Field.AsObjectAttachment();
+						Field			  = CbFieldView(ResponseView.GetData());
+						ResponseView += Field.GetSize();
+						if (!Field.IsObject())	// No data
+						{
+							TreeObjectData[Hash] = {};
+							continue;
+						}
+						MemoryView FieldView = Field.AsObjectView().GetView();
+
+						TreeObjectData[Hash] = IoBuffer(IoBuffer::Wrap, FieldView.GetData(), FieldView.GetSize());
+					}
+					else if (Field.IsBinaryAttachment())
+					{
+						const IoHash Hash	 = Field.AsBinaryAttachment();
+						TreeBinaryData[Hash] = {};
+					}
+					else  // Unknown type
+					{
+					}
+				}
+
+				for (auto& It : TreeObjectData)
+				{
+					if (It.second.GetSize() == 0)
+					{
+						CloudCacheResult ObjectResult = Session.GetObject(It.first);
+						Bytes += ObjectTreeResult.Bytes;
+						ElapsedSeconds += ObjectTreeResult.ElapsedSeconds;
+						if (ObjectTreeResult.ErrorCode != 0)
+						{
+							return {.Error{.ErrorCode = ObjectResult.ErrorCode, .Reason = std::move(ObjectResult.Reason)},
+									.Bytes			= Bytes,
+									.ElapsedSeconds = ElapsedSeconds};
+						}
+
+						if (!ObjectResult.Success)
+						{
+							return {.Error{.ErrorCode = -1, .Reason = "Failed to get result object attachment data"},
+									.Bytes			= Bytes,
+									.ElapsedSeconds = ElapsedSeconds};
+						}
+						It.second = std::move(ObjectResult.Response);
+					}
+				}
+
+				for (auto& It : TreeBinaryData)
+				{
+					if (It.second.GetSize() == 0)
+					{
+						CloudCacheResult BlobResult = Session.GetBlob(It.first);
+						Bytes += ObjectTreeResult.Bytes;
+						ElapsedSeconds += ObjectTreeResult.ElapsedSeconds;
+						if (BlobResult.ErrorCode != 0)
+						{
+							return {.Error{.ErrorCode = BlobResult.ErrorCode, .Reason = std::move(BlobResult.Reason)},
+									.Bytes			= Bytes,
+									.ElapsedSeconds = ElapsedSeconds};
+						}
+
+						if (!BlobResult.Success)
+						{
+							return {.Error{.ErrorCode = -1, .Reason = "Failed to get result binary attachment data"},
+									.Bytes			= Bytes,
+									.ElapsedSeconds = ElapsedSeconds};
+						}
+						It.second = std::move(BlobResult.Response);
+					}
+				}
+
+				CbObject ResultObject = LoadCompactBinaryObject(TreeObjectData[ResultHash]);
+				int32_t	 ExitCode	  = ResultObject["e"sv].AsInt32();
+				IoHash	 StdOutHash	  = ResultObject["so"sv].AsBinaryAttachment();
+				IoHash	 StdErrHash	  = ResultObject["se"sv].AsBinaryAttachment();
+				IoHash	 OutputHash	  = ResultObject["o"sv].AsObjectAttachment();
+
+				std::string StdOut = std::string((const char*)TreeBinaryData[StdOutHash].GetData(), TreeBinaryData[StdOutHash].GetSize());
+				std::string StdErr = std::string((const char*)TreeBinaryData[StdErrHash].GetData(), TreeBinaryData[StdErrHash].GetSize());
+
+				if (ExitCode != 0)
+				{
+					return {.Error{.ErrorCode = ExitCode, .Reason = "Task completed with errors"},
+							.Bytes			= Bytes,
+							.ElapsedSeconds = ElapsedSeconds,
+							.StdOut			= std::move(StdOut),
+							.StdErr			= std::move(StdErr)};
+				}
+
+				CbObject OutputObject = LoadCompactBinaryObject(TreeObjectData[OutputHash]);
+
+				// Get build.output
+				IoHash	 BuildOutputId;
+				IoBuffer BuildOutput;
+				for (auto& It : OutputObject["f"sv])
+				{
+					const CbObjectView FileObject = It.AsObjectView();
+					if (FileObject["n"sv].AsString() == "Build.output"sv)
+					{
+						BuildOutputId = FileObject["h"sv].AsBinaryAttachment();
+						BuildOutput	  = TreeBinaryData[BuildOutputId];
+						break;
+					}
+				}
+
+				if (BuildOutput.GetSize() == 0)
+				{
+					return {.Error{.ErrorCode = ExitCode, .Reason = "Build.output file not found in task results"},
+							.Bytes			= Bytes,
+							.ElapsedSeconds = ElapsedSeconds};
+				}
+
+				// Get Output directory node
+				IoBuffer OutputDirectoryTree;
+				for (auto& It : OutputObject["d"sv])
+				{
+					const CbObjectView DirectoryObject = It.AsObjectView();
+					if (DirectoryObject["n"sv].AsString() == "Outputs"sv)
+					{
+						OutputDirectoryTree = TreeObjectData[DirectoryObject["h"sv].AsObjectAttachment()];
+						break;
+					}
+				}
+
+				if (OutputDirectoryTree.GetSize() == 0)
+				{
+					return {.Error{.ErrorCode = ExitCode, .Reason = "Outputs directory not found in task results"},
+							.Bytes			= Bytes,
+							.ElapsedSeconds = ElapsedSeconds};
+				}
+
+				// load build.output as CbObject
+
+				// Move Outputs from Horde to CbPackage
+
+				std::unordered_map<IoHash, IoHash> CidToCompressedId;
+				CbPackage						   OutputPackage;
+				CbObject						   OutputDirectoryTreeObject = LoadCompactBinaryObject(OutputDirectoryTree);
+				int64_t							   TotalAttachmentBytes		 = 0;
+				int64_t							   TotalRawAttachmentBytes	 = 0;
+
+				for (auto& It : OutputDirectoryTreeObject["f"sv])
+				{
+					CbObjectView FileObject = It.AsObjectView();
+					// Name is the uncompressed hash
+					IoHash DecompressedId = IoHash::FromHexString(FileObject["n"sv].AsString());
+					// Hash is the compressed data hash, and how it is stored in Horde
+					IoHash CompressedId = FileObject["h"sv].AsBinaryAttachment();
+
+					if (!TreeBinaryData.contains(CompressedId))
+					{
+						Log().warn("Object attachment chunk not retrieved from Horde {}", CompressedId.ToHexString());
+						return {.Error{.ErrorCode = -1, .Reason = "Object attachment chunk not retrieved from Horde"},
+								.Bytes			= Bytes,
+								.ElapsedSeconds = ElapsedSeconds};
+					}
+					CidToCompressedId[DecompressedId] = CompressedId;
+				}
+
+				// Iterate attachments, verify all chunks exist, and add to CbPackage
+				bool	 AnyErrors		   = false;
+				CbObject BuildOutputObject = LoadCompactBinaryObject(BuildOutput);
+				BuildOutputObject.IterateAttachments([&](CbFieldView Field) {
+					const IoHash DecompressedId = Field.AsHash();
+					if (!CidToCompressedId.contains(DecompressedId))
+					{
+						Log().warn("Attachment not found {}", DecompressedId.ToHexString());
+						AnyErrors = true;
+						return;
+					}
+					const IoHash& CompressedId = CidToCompressedId.at(DecompressedId);
+
+					if (!TreeBinaryData.contains(CompressedId))
+					{
+						Log().warn("Missing output {} compressed {} uncompressed",
+								   CompressedId.ToHexString(),
+								   DecompressedId.ToHexString());
+						AnyErrors = true;
+						return;
+					}
+
+					CompressedBuffer AttachmentBuffer = CompressedBuffer::FromCompressed(SharedBuffer(TreeBinaryData[CompressedId]));
+
+					if (!AttachmentBuffer)
+					{
+						Log().warn("Invalid output encountered (not valid CompressedBuffer format) {} compressed {} uncompressed",
+								   CompressedId.ToHexString(),
+								   DecompressedId.ToHexString());
+						AnyErrors = true;
+						return;
+					}
+
+					TotalAttachmentBytes += AttachmentBuffer.GetCompressedSize();
+					TotalRawAttachmentBytes += AttachmentBuffer.GetRawSize();
+
+					CbAttachment Attachment(AttachmentBuffer);
+					OutputPackage.AddAttachment(Attachment);
+				});
+
+				if (AnyErrors)
+				{
+					return {.Error{.ErrorCode = -1, .Reason = "Failed to get result object attachment data"},
+							.Bytes			= Bytes,
+							.ElapsedSeconds = ElapsedSeconds};
+				}
+
+				OutputPackage.SetObject(BuildOutputObject);
+
+				return {.OutputPackage			 = std::move(OutputPackage),
+						.TotalAttachmentBytes	 = TotalAttachmentBytes,
+						.TotalRawAttachmentBytes = TotalRawAttachmentBytes,
+						.Bytes					 = Bytes,
+						.ElapsedSeconds			 = ElapsedSeconds,
+						.StdOut					 = std::move(StdOut),
+						.StdErr					 = std::move(StdErr),
+						.Success				 = true};
+			}
+			catch (std::exception& Err)
+			{
+				return {.Error{.ErrorCode = -1, .Reason = Err.what()}};
+			}
+		}
+
+		[[nodiscard]] bool ProcessApplyKey(const UpstreamApplyRecord& ApplyRecord, UpstreamData& Data)
+		{
+			std::string								ExecutablePath;
+			std::map<std::string, std::string>		Environment;
+			std::set<std::filesystem::path>			InputFiles;
+			std::map<std::filesystem::path, IoHash> InputFileHashes;
+
+			ExecutablePath = ApplyRecord.WorkerDescriptor["path"sv].AsString();
+			if (ExecutablePath.empty())
+			{
+				Log().warn("process apply upstream FAILED, '{}', path missing from worker descriptor",
+						   ApplyRecord.WorkerDescriptor.GetHash());
+				return false;
+			}
+
+			for (auto& It : ApplyRecord.WorkerDescriptor["executables"sv])
+			{
+				CbObjectView FileEntry = It.AsObjectView();
+				if (!ProcessFileEntry(FileEntry, InputFiles, InputFileHashes, Data.Blobs))
+				{
+					return false;
+				}
+			}
+
+			for (auto& It : ApplyRecord.WorkerDescriptor["files"sv])
+			{
+				CbObjectView FileEntry = It.AsObjectView();
+				if (!ProcessFileEntry(FileEntry, InputFiles, InputFileHashes, Data.Blobs))
+				{
+					return false;
+				}
+			}
+
+			for (auto& It : ApplyRecord.WorkerDescriptor["environment"sv])
+			{
+				std::string_view Env   = It.AsString();
+				auto			 Index = Env.find('=');
+				if (Index < 0)
+				{
+					Log().warn("process apply upstream FAILED, environment '{}' malformed", Env);
+					return false;
+				}
+
+				Environment[std::string(Env.substr(0, Index))] = Env.substr(Index + 1);
+			}
+
+			{
+				static const std::filesystem::path BuildActionPath = "Build.action"sv;
+				static const std::filesystem::path InputPath	   = "Inputs"sv;
+				const IoHash					   ActionId		   = ApplyRecord.Action.GetHash();
+
+				InputFiles.insert(BuildActionPath);
+				InputFileHashes[BuildActionPath] = ActionId;
+				Data.Blobs[ActionId]			 = IoBufferBuilder::MakeCloneFromMemory(ApplyRecord.Action.GetBuffer().GetData(),
+																			ApplyRecord.Action.GetBuffer().GetSize());
+
+				bool AnyErrors = false;
+				ApplyRecord.Action.IterateAttachments([&](CbFieldView Field) {
+					const IoHash				Cid		   = Field.AsHash();
+					const std::filesystem::path FilePath   = {InputPath / Cid.ToHexString()};
+					IoBuffer					DataBuffer = m_CidStore.FindChunkByCid(Cid);
+
+					if (!DataBuffer)
+					{
+						Log().warn("process apply upstream FAILED, input CID chunk '{}' missing", Cid);
+						AnyErrors = true;
+						return;
+					}
+
+					if (InputFiles.contains(FilePath))
+					{
+						return;
+					}
+
+					const IoHash CompressedId = IoHash::HashBuffer(DataBuffer.GetData(), DataBuffer.GetSize());
+
+					InputFiles.insert(FilePath);
+					InputFileHashes[FilePath] = CompressedId;
+					Data.Blobs[CompressedId]  = std::move(DataBuffer);
+				});
+
+				if (AnyErrors)
+				{
+					return false;
+				}
+			}
+
+			const UpstreamDirectory RootDirectory = BuildDirectoryTree(InputFiles);
+
+			CbObject	 Sandbox	  = BuildMerkleTreeDirectory(RootDirectory, InputFileHashes, Data.Blobs, Data.Objects);
+			const IoHash SandboxHash  = Sandbox.GetHash();
+			Data.Objects[SandboxHash] = std::move(Sandbox);
+
+			{
+				using namespace fmt::literals;
+				std::string_view HostPlatform = ApplyRecord.WorkerDescriptor["host"sv].AsString();
+				if (HostPlatform.empty())
+				{
+					Log().warn("process apply upstream FAILED, 'host' platform not provided");
+					return false;
+				}
+
+				int32_t LogicalCores = ApplyRecord.WorkerDescriptor["cores"sv].AsInt32();
+				int64_t Memory		 = ApplyRecord.WorkerDescriptor["memory"sv].AsInt64();
+				bool	Exclusive	 = ApplyRecord.WorkerDescriptor["exclusive"sv].AsBool();
+
+				// TODO: Remove override when Horde accepts the UE style Host Platforms (Win64, Linux, Mac)
+				std::string Condition;
+				if (HostPlatform == "Win64" || HostPlatform == "Windows")
+				{
+					Condition = "OSFamily == 'Windows' && Pool == 'Win-RemoteExec'";
+				}
+				else if (HostPlatform == "Mac")
+				{
+					Condition = "OSFamily == 'MacOS'";
+				}
+				else
+				{
+					Condition = "OSFamily == '{}'"_format(HostPlatform);
+				}
+
+				std::map<std::string_view, int64_t> Resources;
+				if (LogicalCores > 0)
+				{
+					Resources["LogicalCores"sv] = LogicalCores;
+				}
+				if (Memory > 0)
+				{
+					Resources["RAM"sv] = std::max(Memory / 1024 / 1024 / 1024, 1LL);
+				}
+
+				CbObject	 Requirements	 = BuildRequirements(Condition, Resources, Exclusive);
+				const IoHash RequirementsId	 = Requirements.GetHash();
+				Data.Objects[RequirementsId] = std::move(Requirements);
+				Data.RequirementsId			 = RequirementsId;
+			}
+
+			CbObject Task = BuildTask(ExecutablePath,
+									  {"-Build=build.action"},
+									  Environment,
+									  {},
+									  SandboxHash,
+									  Data.RequirementsId,
+									  {"Build.output", "Outputs"});
+
+			const IoHash TaskId	 = Task.GetHash();
+			Data.Objects[TaskId] = std::move(Task);
+			Data.TaskId			 = TaskId;
+
+			return true;
+		}
+
+		[[nodiscard]] bool ProcessFileEntry(const CbObjectView&						 FileEntry,
+											std::set<std::filesystem::path>&		 InputFiles,
+											std::map<std::filesystem::path, IoHash>& InputFileHashes,
+											std::map<IoHash, IoBuffer>&				 Blobs)
+		{
+			const std::filesystem::path FilePath   = FileEntry["name"sv].AsString();
+			const IoHash				ChunkId	   = FileEntry["hash"sv].AsHash();
+			const uint64_t				Size	   = FileEntry["size"sv].AsUInt64();
+			IoBuffer					DataBuffer = m_CasStore.FindChunk(ChunkId);
+
+			if (!DataBuffer)
+			{
+				Log().warn("process apply upstream FAILED, worker CAS chunk '{}' missing", ChunkId);
+				return false;
+			}
+
+			if (DataBuffer.Size() != Size)
+			{
+				Log().warn("process apply upstream FAILED, worker CAS chunk '{}' size: {}, action spec expected {}",
+						   ChunkId,
+						   DataBuffer.Size(),
+						   Size);
+				return false;
+			}
+
+			if (InputFiles.contains(FilePath))
+			{
+				Log().warn("process apply upstream FAILED, worker CAS chunk '{}' size: {} duplicate filename {}", ChunkId, Size, FilePath);
+				return false;
+			}
+
+			InputFiles.insert(FilePath);
+			InputFileHashes[FilePath] = ChunkId;
+			Blobs[ChunkId]			  = std::move(DataBuffer);
+			return true;
+		}
+
+		[[nodiscard]] UpstreamDirectory BuildDirectoryTree(const std::set<std::filesystem::path>& InputFiles)
+		{
+			static const std::filesystem::path					RootPath;
+			std::map<std::filesystem::path, UpstreamDirectory*> AllDirectories;
+			UpstreamDirectory									RootDirectory = {.Path = RootPath};
+
+			AllDirectories[RootPath] = &RootDirectory;
+
+			// Build tree from flat list
+			for (const auto& Path : InputFiles)
+			{
+				if (Path.has_parent_path())
+				{
+					if (!AllDirectories.contains(Path.parent_path()))
+					{
+						std::stack<std::string> PathSplit;
+						{
+							std::filesystem::path ParentPath = Path.parent_path();
+							PathSplit.push(ParentPath.filename().string());
+							while (ParentPath.has_parent_path())
+							{
+								ParentPath = ParentPath.parent_path();
+								PathSplit.push(ParentPath.filename().string());
+							}
+						}
+						UpstreamDirectory* ParentPtr = &RootDirectory;
+						while (!PathSplit.empty())
+						{
+							if (!ParentPtr->Directories.contains(PathSplit.top()))
+							{
+								std::filesystem::path NewParentPath		= {ParentPtr->Path / PathSplit.top()};
+								ParentPtr->Directories[PathSplit.top()] = {.Path = NewParentPath};
+								AllDirectories[NewParentPath]			= &ParentPtr->Directories[PathSplit.top()];
+							}
+							ParentPtr = &ParentPtr->Directories[PathSplit.top()];
+							PathSplit.pop();
+						}
+					}
+
+					AllDirectories[Path.parent_path()]->Files.insert(Path.filename().string());
+				}
+				else
+				{
+					RootDirectory.Files.insert(Path.filename().string());
+				}
+			}
+
+			return RootDirectory;
+		}
+
+		[[nodiscard]] CbObject BuildMerkleTreeDirectory(const UpstreamDirectory&					   RootDirectory,
+														const std::map<std::filesystem::path, IoHash>& InputFileHashes,
+														const std::map<IoHash, IoBuffer>&			   Blobs,
+														std::map<IoHash, CbObject>&					   Objects)
+		{
+			CbObjectWriter DirectoryTreeWriter;
+
+			if (!RootDirectory.Files.empty())
+			{
+				DirectoryTreeWriter.BeginArray("f"sv);
+				for (const auto& File : RootDirectory.Files)
+				{
+					const std::filesystem::path FilePath = {RootDirectory.Path / File};
+					const IoHash&				FileHash = InputFileHashes.at(FilePath);
+					const uint64_t				FileSize = Blobs.at(FileHash).Size();
+					DirectoryTreeWriter.BeginObject();
+					DirectoryTreeWriter.AddString("n"sv, File);
+					DirectoryTreeWriter.AddBinaryAttachment("h"sv, FileHash);
+					DirectoryTreeWriter.AddInteger("s"sv, FileSize);  // Size
+					// DirectoryTreeWriter.AddInteger("a"sv, 0);		  // Attributes Currently unneeded
+					DirectoryTreeWriter.EndObject();
+				}
+				DirectoryTreeWriter.EndArray();
+			}
+
+			if (!RootDirectory.Directories.empty())
+			{
+				DirectoryTreeWriter.BeginArray("d"sv);
+				for (const auto& Item : RootDirectory.Directories)
+				{
+					CbObject	 Directory	   = BuildMerkleTreeDirectory(Item.second, InputFileHashes, Blobs, Objects);
+					const IoHash DirectoryHash = Directory.GetHash();
+					Objects[DirectoryHash]	   = std::move(Directory);
+
+					DirectoryTreeWriter.BeginObject();
+					DirectoryTreeWriter.AddString("n"sv, Item.first);
+					DirectoryTreeWriter.AddObjectAttachment("h"sv, DirectoryHash);
+					DirectoryTreeWriter.EndObject();
+				}
+				DirectoryTreeWriter.EndArray();
+			}
+
+			return std::move(DirectoryTreeWriter.Save());
+		}
+
+		[[nodiscard]] CbObject BuildRequirements(const std::string_view						Condition,
+												 const std::map<std::string_view, int64_t>& Resources,
+												 const bool									Exclusive)
+		{
+			CbObjectWriter Writer;
+			Writer.AddString("c", Condition);
+			if (!Resources.empty())
+			{
+				Writer.BeginArray("r");
+				for (const auto& Resource : Resources)
+				{
+					Writer.BeginArray();
+					Writer.AddString(Resource.first);
+					Writer.AddInteger(Resource.second);
+					Writer.EndArray();
+				}
+				Writer.EndArray();
+			}
+			Writer.AddBool("e", Exclusive);
+			return std::move(Writer.Save());
+		}
+
+		[[nodiscard]] CbObject BuildTask(const std::string_view					   Executable,
+										 const std::vector<std::string>&		   Arguments,
+										 const std::map<std::string, std::string>& Environment,
+										 const std::string_view					   WorkingDirectory,
+										 const IoHash&							   SandboxHash,
+										 const IoHash&							   RequirementsId,
+										 const std::set<std::string>&			   Outputs)
+		{
+			CbObjectWriter TaskWriter;
+			TaskWriter.AddString("e"sv, Executable);
+
+			if (!Arguments.empty())
+			{
+				TaskWriter.BeginArray("a"sv);
+				for (const auto& Argument : Arguments)
+				{
+					TaskWriter.AddString(Argument);
+				}
+				TaskWriter.EndArray();
+			}
+
+			if (!Environment.empty())
+			{
+				TaskWriter.BeginArray("v"sv);
+				for (const auto& Env : Environment)
+				{
+					TaskWriter.BeginArray();
+					TaskWriter.AddString(Env.first);
+					TaskWriter.AddString(Env.second);
+					TaskWriter.EndArray();
+				}
+				TaskWriter.EndArray();
+			}
+
+			if (!WorkingDirectory.empty())
+			{
+				TaskWriter.AddString("s"sv, WorkingDirectory);
+			}
+
+			TaskWriter.AddObjectAttachment("s"sv, SandboxHash);
+			TaskWriter.AddObjectAttachment("r"sv, RequirementsId);
+
+			// Outputs
+			if (!Outputs.empty())
+			{
+				TaskWriter.BeginArray("o"sv);
+				for (const auto& Output : Outputs)
+				{
+					TaskWriter.AddString(Output);
+				}
+				TaskWriter.EndArray();
+			}
+
+			return std::move(TaskWriter.Save());
+		}
+	};
+}  // namespace detail
+
+//////////////////////////////////////////////////////////////////////////
+
+struct UpstreamApplyStats
+{
+	static constexpr uint64_t MaxSampleCount = 1000ull;
+
+	UpstreamApplyStats(bool Enabled) : m_Enabled(Enabled) {}
+
+	void Add(spdlog::logger&											Logger,
+			 UpstreamApplyEndpoint&										Endpoint,
+			 const PostUpstreamApplyResult&								Result,
+			 const std::vector<std::unique_ptr<UpstreamApplyEndpoint>>& Endpoints)
+	{
+		UpstreamApplyEndpointStats& Stats = Endpoint.Stats();
+
+		if (Result.Error)
+		{
+			Stats.ErrorCount++;
+		}
+		else if (Result.Success)
+		{
+			Stats.PostCount++;
+			Stats.UpBytes.fetch_add(double(Result.Bytes) / 1024.0 / 1024.0);
+			Stats.SecondsUp.fetch_add(Result.ElapsedSeconds);
+		}
+
+		if (m_Enabled && m_SampleCount++ % MaxSampleCount)
+		{
+			Dump(Logger, Endpoints);
+		}
+	}
+
+	void Add(spdlog::logger&											Logger,
+			 UpstreamApplyEndpoint&										Endpoint,
+			 const GetUpstreamApplyUpdatesResult&						Result,
+			 const std::vector<std::unique_ptr<UpstreamApplyEndpoint>>& Endpoints)
+	{
+		UpstreamApplyEndpointStats& Stats = Endpoint.Stats();
+
+		if (Result.Error)
+		{
+			Stats.ErrorCount++;
+		}
+		else if (Result.Success)
+		{
+			Stats.UpdateCount++;
+			Stats.DownBytes.fetch_add(double(Result.Bytes) / 1024.0 / 1024.0);
+			Stats.SecondsDown.fetch_add(Result.ElapsedSeconds);
+			if (!Result.Completed.empty())
+			{
+				uint64_t Completed = 0;
+				for (auto& It : Result.Completed)
+				{
+					Completed += It.second.size();
+				}
+				Stats.CompleteCount.fetch_add(Completed);
+			}
+		}
+
+		if (m_Enabled && m_SampleCount++ % MaxSampleCount)
+		{
+			Dump(Logger, Endpoints);
+		}
+	}
+
+	void Dump(spdlog::logger& Logger, const std::vector<std::unique_ptr<UpstreamApplyEndpoint>>& Endpoints)
+	{
+		for (auto& Ep : Endpoints)
+		{
+			// These stats will not be totally correct as the numbers are not captured atomically
+
+			UpstreamApplyEndpointStats& Stats		  = Ep->Stats();
+			const uint64_t				PostCount	  = Stats.PostCount;
+			const uint64_t				CompleteCount = Stats.CompleteCount;
+			const uint64_t				UpdateCount	  = Stats.UpdateCount;
+			const double				DownBytes	  = Stats.DownBytes;
+			const double				SecondsDown	  = Stats.SecondsDown;
+			const double				UpBytes		  = Stats.UpBytes;
+			const double				SecondsUp	  = Stats.SecondsUp;
+
+			const double UpSpeed	  = UpBytes > 0 ? UpBytes / SecondsUp : 0.0;
+			const double DownSpeed	  = DownBytes > 0 ? DownBytes / SecondsDown : 0.0;
+			const double CompleteRate = CompleteCount > 0 ? (double(PostCount) / double(CompleteCount)) : 0.0;
+
+			Logger.debug("STATS - '{}', Complete rate: {:.2f}%, DOWN: '{:.2f} MiB {:.2f} MiB/s', UP: '{:.2f} MiB {:.2f} MiB/s'",
+						 Ep->DisplayName(),
+						 CompleteRate,
+						 DownBytes,
+						 DownSpeed,
+						 UpBytes,
+						 UpSpeed);
+		}
+	}
+
+	bool				 m_Enabled;
+	std::atomic_uint64_t m_SampleCount = {};
+};
+
+//////////////////////////////////////////////////////////////////////////
+
+class DefaultUpstreamApply final : public UpstreamApply
+{
+public:
+	DefaultUpstreamApply(const UpstreamApplyOptions& Options, CasStore& CasStore, CidStore& CidStore)
+	: m_Log(logging::Get("upstream-apply"))
+	, m_Options(Options)
+	, m_CasStore(CasStore)
+	, m_CidStore(CidStore)
+	, m_Stats(Options.StatsEnabled)
+	{
+	}
+
+	virtual ~DefaultUpstreamApply() { Shutdown(); }
+
+	virtual bool Initialize() override
+	{
+		for (auto& Endpoint : m_Endpoints)
+		{
+			const UpstreamEndpointHealth Health = Endpoint->Initialize();
+			if (Health.Ok)
+			{
+				Log().info("initialize endpoint '{}' OK", Endpoint->DisplayName());
+			}
+			else
+			{
+				Log().warn("initialize endpoint '{}' FAILED, reason '{}'", Endpoint->DisplayName(), Health.Reason);
+			}
+		}
+
+		m_RunState.IsRunning = !m_Endpoints.empty();
+
+		if (m_RunState.IsRunning)
+		{
+			for (uint32_t Idx = 0; Idx < m_Options.ThreadCount; Idx++)
+			{
+				m_UpstreamThreads.emplace_back(&DefaultUpstreamApply::ProcessUpstreamQueue, this);
+			}
+
+			m_UpstreamUpdatesThread = std::thread(&DefaultUpstreamApply::ProcessUpstreamUpdates, this);
+
+			m_EndpointMonitorThread = std::thread(&DefaultUpstreamApply::MonitorEndpoints, this);
+		}
+
+		return m_RunState.IsRunning;
+	}
+
+	virtual void RegisterEndpoint(std::unique_ptr<UpstreamApplyEndpoint> Endpoint) override
+	{
+		m_Endpoints.emplace_back(std::move(Endpoint));
+	}
+
+	virtual EnqueueResult EnqueueUpstream(UpstreamApplyRecord ApplyRecord) override
+	{
+		if (m_RunState.IsRunning)
+		{
+			const IoHash   WorkerId		  = ApplyRecord.WorkerDescriptor.GetHash();
+			const IoHash   ActionId		  = ApplyRecord.Action.GetHash();
+			const uint32_t TimeoutSeconds = ApplyRecord.WorkerDescriptor["timeout"sv].AsInt32(300);
+
+			{
+				std::scoped_lock Lock(m_ApplyTasksMutex);
+				if (auto Status = FindStatus(WorkerId, ActionId); Status != nullptr)
+				{
+					// Already in progress
+					return {.ApplyId = ActionId, .Success = true};
+				}
+
+				std::chrono::steady_clock::time_point ExpireTime =
+					TimeoutSeconds > 0 ? std::chrono::steady_clock::now() + std::chrono::seconds(TimeoutSeconds)
+									   : std::chrono::steady_clock::time_point::max();
+
+				m_ApplyTasks[WorkerId][ActionId] = {.State = UpstreamApplyState::Queued, .Result{}, .ExpireTime = std::move(ExpireTime)};
+			}
+
+			if (!m_UpstreamThreads.empty())
+			{
+				m_UpstreamQueue.Enqueue(std::move(ApplyRecord));
+			}
+			else
+			{
+				ProcessApplyRecord(std::move(ApplyRecord));
+			}
+
+			return {.ApplyId = ActionId, .Success = true};
+		}
+
+		return {};
+	}
+
+	virtual StatusResult GetStatus(const IoHash& WorkerId, const IoHash& ActionId) override
+	{
+		if (m_RunState.IsRunning)
+		{
+			std::scoped_lock Lock(m_ApplyTasksMutex);
+			if (auto Status = FindStatus(WorkerId, ActionId); Status != nullptr)
+			{
+				return {.Status = *Status, .Success = true};
+			}
+		}
+
+		return {};
+	}
+
+	virtual void GetStatus(CbObjectWriter& Status) override
+	{
+		Status << "worker_threads" << m_Options.ThreadCount;
+		Status << "queue_count" << m_UpstreamQueue.Size();
+
+		Status.BeginArray("endpoints");
+		for (const auto& Ep : m_Endpoints)
+		{
+			Status.BeginObject();
+			Status << "name" << Ep->DisplayName();
+			Status << "health" << (Ep->IsHealthy() ? "ok"sv : "inactive"sv);
+
+			UpstreamApplyEndpointStats& Stats		  = Ep->Stats();
+			const uint64_t				PostCount	  = Stats.PostCount;
+			const uint64_t				CompleteCount = Stats.CompleteCount;
+			const uint64_t				UpdateCount	  = Stats.UpdateCount;
+			const double				CompleteRate  = CompleteCount > 0 ? (double(PostCount) / double(CompleteCount)) : 0.0;
+
+			Status << "post_count" << PostCount;
+			Status << "complete_count" << PostCount;
+			Status << "update_count" << Stats.UpdateCount;
+
+			Status << "complete_ratio" << CompleteRate;
+			Status << "downloaded_mb" << Stats.DownBytes;
+			Status << "uploaded_mb" << Stats.UpBytes;
+			Status << "error_count" << Stats.ErrorCount;
+
+			Status.EndObject();
+		}
+		Status.EndArray();
+	}
+
+private:
+	// The caller is responsible for locking if required
+	UpstreamApplyStatus* FindStatus(const IoHash& WorkerId, const IoHash& ActionId)
+	{
+		if (auto It = m_ApplyTasks.find(WorkerId); It != m_ApplyTasks.end())
+		{
+			if (auto It2 = It->second.find(ActionId); It2 != It->second.end())
+			{
+				return &It2->second;
+			}
+		}
+		return nullptr;
+	}
+
+	void ProcessApplyRecord(UpstreamApplyRecord ApplyRecord)
+	{
+		const IoHash WorkerId = ApplyRecord.WorkerDescriptor.GetHash();
+		const IoHash ActionId = ApplyRecord.Action.GetHash();
+		try
+		{
+			for (auto& Endpoint : m_Endpoints)
+			{
+				if (Endpoint->IsHealthy())
+				{
+					PostUpstreamApplyResult Result = Endpoint->PostApply(std::move(ApplyRecord));
+					{
+						std::scoped_lock Lock(m_ApplyTasksMutex);
+						if (auto Status = FindStatus(WorkerId, ActionId); Status != nullptr)
+						{
+							if (Result.Success)
+							{
+								Status->State = UpstreamApplyState::Executing;
+							}
+							else
+							{
+								Status->State  = UpstreamApplyState::Complete;
+								Status->Result = {.Error		  = std::move(Result.Error),
+												  .Bytes		  = Result.Bytes,
+												  .ElapsedSeconds = Result.ElapsedSeconds};
+							}
+						}
+					}
+					m_Stats.Add(m_Log, *Endpoint, Result, m_Endpoints);
+					return;
+				}
+			}
+
+			{
+				std::scoped_lock Lock(m_ApplyTasksMutex);
+				if (auto Status = FindStatus(WorkerId, ActionId); Status != nullptr)
+				{
+					Status->State  = UpstreamApplyState::Complete;
+					Status->Result = {.Error{.ErrorCode = -1, .Reason = "No available endpoint"}};
+				}
+				Log().warn("process upstream apply ({}/{}) FAILED 'No available endpoint'", WorkerId, ActionId);
+			}
+		}
+		catch (std::exception& e)
+		{
+			std::scoped_lock Lock(m_ApplyTasksMutex);
+			if (auto Status = FindStatus(WorkerId, ActionId); Status != nullptr)
+			{
+				Status->State  = UpstreamApplyState::Complete;
+				Status->Result = {.Error{.ErrorCode = -1, .Reason = e.what()}};
+			}
+			Log().warn("process upstream apply ({}/{}) FAILED '{}'", WorkerId, ActionId, e.what());
+		}
+	}
+
+	void ProcessUpstreamQueue()
+	{
+		for (;;)
+		{
+			UpstreamApplyRecord ApplyRecord;
+			if (m_UpstreamQueue.WaitAndDequeue(ApplyRecord))
+			{
+				ProcessApplyRecord(std::move(ApplyRecord));
+			}
+
+			if (!m_RunState.IsRunning)
+			{
+				break;
+			}
+		}
+	}
+
+	void ProcessApplyUpdates()
+	{
+		for (auto& Endpoint : m_Endpoints)
+		{
+			if (Endpoint->IsHealthy())
+			{
+				GetUpstreamApplyUpdatesResult Result = Endpoint->GetUpdates();
+				m_Stats.Add(m_Log, *Endpoint, Result, m_Endpoints);
+
+				if (!Result.Success)
+				{
+					Log().warn("process upstream apply updates FAILED '{}'", Result.Error.Reason);
+				}
+
+				if (!Result.Completed.empty())
+				{
+					for (auto& It : Result.Completed)
+					{
+						for (auto& It2 : It.second)
+						{
+							std::scoped_lock Lock(m_ApplyTasksMutex);
+							if (auto Status = FindStatus(It.first, It2.first); Status != nullptr)
+							{
+								Status->State  = UpstreamApplyState::Complete;
+								Status->Result = std::move(It2.second);
+							}
+						}
+					}
+				}
+			}
+		}
+	}
+
+	void ProcessUpstreamUpdates()
+	{
+		const auto& UpdateSleep = std::chrono::seconds(m_Options.UpdatesInterval);
+		for (;;)
+		{
+			std::this_thread::sleep_for(UpdateSleep);
+
+			if (!m_RunState.IsRunning)
+			{
+				break;
+			}
+
+			ProcessApplyUpdates();
+
+			// Remove any expired tasks, regardless of state
+			{
+				std::scoped_lock Lock(m_ApplyTasksMutex);
+				for (auto& WorkerIt : m_ApplyTasks)
+				{
+					const auto Count = std::erase_if(WorkerIt.second, [](const auto& Item) {
+						return Item.second.ExpireTime < std::chrono::steady_clock::now();
+					});
+					if (Count > 0)
+					{
+						Log().debug("Removed '{}' expired tasks", Count);
+					}
+				}
+				const auto Count = std::erase_if(m_ApplyTasks, [](const auto& Item) { return Item.second.empty(); });
+				if (Count > 0)
+				{
+					Log().debug("Removed '{}' empty task lists", Count);
+				}
+			}
+		}
+	}
+
+	void MonitorEndpoints()
+	{
+		for (;;)
+		{
+			{
+				std::unique_lock Lock(m_RunState.Mutex);
+				if (m_RunState.ExitSignal.wait_for(Lock, m_Options.HealthCheckInterval, [this]() { return !m_RunState.IsRunning.load(); }))
+				{
+					break;
+				}
+			}
+
+			for (auto& Endpoint : m_Endpoints)
+			{
+				if (!Endpoint->IsHealthy())
+				{
+					if (const UpstreamEndpointHealth Health = Endpoint->CheckHealth(); Health.Ok)
+					{
+						Log().warn("health check endpoint '{}' OK", Endpoint->DisplayName(), Health.Reason);
+					}
+					else
+					{
+						Log().warn("health check endpoint '{}' FAILED, reason '{}'", Endpoint->DisplayName(), Health.Reason);
+					}
+				}
+			}
+		}
+	}
+
+	void Shutdown()
+	{
+		if (m_RunState.Stop())
+		{
+			m_UpstreamQueue.CompleteAdding();
+			for (std::thread& Thread : m_UpstreamThreads)
+			{
+				Thread.join();
+			}
+
+			m_EndpointMonitorThread.join();
+			m_UpstreamUpdatesThread.join();
+			m_UpstreamThreads.clear();
+			m_Endpoints.clear();
+		}
+	}
+
+	spdlog::logger& Log() { return m_Log; }
+
+	using UpstreamApplyQueue = BlockingQueue<UpstreamApplyRecord>;
+
+	struct RunState
+	{
+		std::mutex				Mutex;
+		std::condition_variable ExitSignal;
+		std::atomic_bool		IsRunning{false};
+
+		bool Stop()
+		{
+			bool Stopped = false;
+			{
+				std::scoped_lock Lock(Mutex);
+				Stopped = IsRunning.exchange(false);
+			}
+			if (Stopped)
+			{
+				ExitSignal.notify_all();
+			}
+			return Stopped;
+		}
+	};
+
+	spdlog::logger&										m_Log;
+	UpstreamApplyOptions								m_Options;
+	CasStore&											m_CasStore;
+	CidStore&											m_CidStore;
+	UpstreamApplyQueue									m_UpstreamQueue;
+	UpstreamApplyStats									m_Stats;
+	UpstreamApplyTasks									m_ApplyTasks;
+	std::mutex											m_ApplyTasksMutex;
+	std::vector<std::unique_ptr<UpstreamApplyEndpoint>> m_Endpoints;
+	std::vector<std::thread>							m_UpstreamThreads;
+	std::thread											m_UpstreamUpdatesThread;
+	std::thread											m_EndpointMonitorThread;
+	RunState											m_RunState;
+};
+
+//////////////////////////////////////////////////////////////////////////
+
+std::unique_ptr<UpstreamApply>
+MakeUpstreamApply(const UpstreamApplyOptions& Options, CasStore& CasStore, CidStore& CidStore)
+{
+	return std::make_unique<DefaultUpstreamApply>(Options, CasStore, CidStore);
+}
+
+std::unique_ptr<UpstreamApplyEndpoint>
+MakeHordeUpstreamEndpoint(const CloudCacheClientOptions& Options, CasStore& CasStore, CidStore& CidStore)
+{
+	return std::make_unique<detail::HordeUpstreamApplyEndpoint>(Options, CasStore, CidStore);
+}
+
+}  // namespace zen
diff --git a/zenserver/upstream/upstreamapply.h b/zenserver/upstream/upstreamapply.h
new file mode 100644
index 000000000..98f193c02
--- /dev/null
+++ b/zenserver/upstream/upstreamapply.h
@@ -0,0 +1,172 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <zencore/compactbinarypackage.h>
+#include <zencore/iobuffer.h>
+#include <zencore/iohash.h>
+#include <zencore/zencore.h>
+
+#include <atomic>
+#include <chrono>
+#include <memory>
+#include <unordered_map>
+#include <unordered_set>
+
+namespace zen {
+
+class CbObjectWriter;
+class CasStore;
+class CidStore;
+class ZenCacheStore;
+struct CloudCacheClientOptions;
+
+enum class UpstreamApplyState : int32_t
+{
+	Queued	  = 0,
+	Executing = 1,
+	Complete  = 2,
+};
+
+struct UpstreamApplyRecord
+{
+	CbObject WorkerDescriptor;
+	CbObject Action;
+};
+
+struct UpstreamApplyOptions
+{
+	std::chrono::seconds HealthCheckInterval{5};
+	std::chrono::seconds UpdatesInterval{5};
+	uint32_t			 ThreadCount  = 4;
+	bool				 StatsEnabled = false;
+};
+
+struct UpstreamApplyError
+{
+	int32_t		ErrorCode{};
+	std::string Reason{};
+
+	explicit operator bool() const { return ErrorCode != 0; }
+};
+
+struct PostUpstreamApplyResult
+{
+	UpstreamApplyError Error{};
+	int64_t			   Bytes{};
+	double			   ElapsedSeconds{};
+	bool			   Success = false;
+};
+
+struct GetUpstreamApplyResult
+{
+	CbPackage		   OutputPackage{};
+	int64_t			   TotalAttachmentBytes{};
+	int64_t			   TotalRawAttachmentBytes{};
+	UpstreamApplyError Error{};
+	int64_t			   Bytes{};
+	double			   ElapsedSeconds{};
+	std::string		   StdOut{};
+	std::string		   StdErr{};
+	bool			   Success = false;
+};
+
+using UpstreamApplyCompleted = std::unordered_map<IoHash, std::unordered_map<IoHash, GetUpstreamApplyResult>>;
+
+struct GetUpstreamApplyUpdatesResult
+{
+	UpstreamApplyError	   Error{};
+	int64_t				   Bytes{};
+	double				   ElapsedSeconds{};
+	UpstreamApplyCompleted Completed{};
+	bool				   Success = false;
+};
+
+struct UpstreamApplyStatus
+{
+	UpstreamApplyState					  State{};
+	GetUpstreamApplyResult				  Result{};
+	std::chrono::steady_clock::time_point ExpireTime{};
+};
+
+using UpstreamApplyTasks = std::unordered_map<IoHash, std::unordered_map<IoHash, UpstreamApplyStatus>>;
+
+struct UpstreamEndpointHealth
+{
+	std::string Reason;
+	bool		Ok = false;
+};
+
+struct UpstreamApplyEndpointStats
+{
+	std::atomic_uint64_t PostCount{};
+	std::atomic_uint64_t CompleteCount{};
+	std::atomic_uint64_t UpdateCount{};
+	std::atomic_uint64_t ErrorCount{};
+	std::atomic<double>	 UpBytes{};
+	std::atomic<double>	 DownBytes{};
+	std::atomic<double>	 SecondsUp{};
+	std::atomic<double>	 SecondsDown{};
+};
+
+/**
+ * The upstream apply endpont is responsible for handling remote execution.
+ */
+class UpstreamApplyEndpoint
+{
+public:
+	virtual ~UpstreamApplyEndpoint() = default;
+
+	virtual UpstreamEndpointHealth Initialize() = 0;
+
+	virtual bool IsHealthy() const = 0;
+
+	virtual UpstreamEndpointHealth CheckHealth() = 0;
+
+	virtual std::string_view DisplayName() const = 0;
+
+	virtual PostUpstreamApplyResult PostApply(const UpstreamApplyRecord& ApplyRecord) = 0;
+
+	virtual GetUpstreamApplyUpdatesResult GetUpdates() = 0;
+
+	virtual UpstreamApplyEndpointStats& Stats() = 0;
+};
+
+/**
+ * Manages one or more upstream cache endpoints.
+ */
+class UpstreamApply
+{
+public:
+	virtual ~UpstreamApply() = default;
+
+	virtual bool Initialize() = 0;
+
+	virtual void RegisterEndpoint(std::unique_ptr<UpstreamApplyEndpoint> Endpoint) = 0;
+
+	struct EnqueueResult
+	{
+		IoHash ApplyId{};
+		bool   Success = false;
+	};
+
+	struct StatusResult
+	{
+		UpstreamApplyStatus Status{};
+		bool				Success = false;
+	};
+
+	virtual EnqueueResult EnqueueUpstream(UpstreamApplyRecord ApplyRecord) = 0;
+
+	virtual StatusResult GetStatus(const IoHash& WorkerId, const IoHash& ActionId) = 0;
+
+	virtual void GetStatus(CbObjectWriter& CbO) = 0;
+};
+
+std::unique_ptr<UpstreamApply> MakeUpstreamApply(const UpstreamApplyOptions& Options, CasStore& CasStore, CidStore& CidStore);
+
+std::unique_ptr<UpstreamApplyEndpoint> MakeHordeUpstreamEndpoint(const CloudCacheClientOptions& Options,
+																 CasStore&						CasStore,
+																 CidStore&						CidStore);
+
+}  // namespace zen
diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp
index 168449d05..e2dc09872 100644
--- a/zenserver/upstream/upstreamcache.cpp
+++ b/zenserver/upstream/upstreamcache.cpp
@@ -40,13 +40,15 @@ namespace detail {
 		: m_Log(zen::logging::Get("upstream"))
 		, m_UseLegacyDdc(Options.UseLegacyDdc)
 		{
-			using namespace fmt::literals;
-			m_DisplayName = "Jupiter - '{}'"_format(Options.ServiceUrl);
-			m_Client	  = new CloudCacheClient(Options);
+			m_Info.Name = "Horde"sv;
+			m_Info.Url	= Options.ServiceUrl;
+			m_Client	= new CloudCacheClient(Options);
 		}
 
 		virtual ~JupiterUpstreamEndpoint() = default;
 
+		virtual const UpstreamEndpointInfo& GetEndpointInfo() const { return m_Info; }
+
 		virtual UpstreamEndpointHealth Initialize() override { return CheckHealth(); }
 
 		virtual bool IsHealthy() const override { return m_HealthOk.load(); }
@@ -58,7 +60,7 @@ namespace detail {
 				CloudCacheSession	   Session(m_Client);
 				const CloudCacheResult Result = Session.Authenticate();
 
-				m_HealthOk = Result.ErrorCode == 0;
+				m_HealthOk = Result.Success && Result.ErrorCode == 0;
 
 				return {.Reason = std::move(Result.Reason), .Ok = Result.Success};
 			}
@@ -68,9 +70,7 @@ namespace detail {
 			}
 		}
 
-		virtual std::string_view DisplayName() const override { return m_DisplayName; }
-
-		virtual GetUpstreamCacheResult GetCacheRecord(UpstreamCacheKey CacheKey, ZenContentType Type) override
+		virtual GetUpstreamCacheResult GetCacheRecord(CacheKey CacheKey, ZenContentType Type) override
 		{
 			try
 			{
@@ -144,12 +144,69 @@ namespace detail {
 			}
 		}
 
-		virtual GetUpstreamCacheResult GetCachePayload(UpstreamPayloadKey PayloadKey) override
+		virtual GetUpstreamCacheResult GetCacheRecords(std::span<CacheKey>		  CacheKeys,
+													   std::span<size_t>		  KeyIndex,
+													   const CacheRecordPolicy&	  Policy,
+													   OnCacheRecordGetComplete&& OnComplete) override
+		{
+			ZEN_UNUSED(Policy);
+
+			CloudCacheSession	   Session(m_Client);
+			GetUpstreamCacheResult Result;
+
+			for (size_t Index : KeyIndex)
+			{
+				const CacheKey& CacheKey = CacheKeys[Index];
+				CbPackage		Package;
+				CbObject		Record;
+
+				if (!Result.Error)
+				{
+					CloudCacheResult RefResult = Session.GetRef(CacheKey.Bucket, CacheKey.Hash, ZenContentType::kCbObject);
+					AppendResult(RefResult, Result);
+
+					if (RefResult.ErrorCode == 0)
+					{
+						const CbValidateError ValidationResult = ValidateCompactBinary(RefResult.Response, CbValidateMode::All);
+						if (ValidationResult == CbValidateError::None)
+						{
+							Record = LoadCompactBinaryObject(RefResult.Response);
+							Record.IterateAttachments([this, &Session, &Result, &Package](CbFieldView AttachmentHash) {
+								CloudCacheResult BlobResult = Session.GetCompressedBlob(AttachmentHash.AsHash());
+								AppendResult(BlobResult, Result);
+
+								if (BlobResult.ErrorCode == 0)
+								{
+									if (CompressedBuffer Chunk = CompressedBuffer::FromCompressed(SharedBuffer(BlobResult.Response)))
+									{
+										Package.AddAttachment(CbAttachment(Chunk));
+									}
+								}
+								else
+								{
+									m_HealthOk = false;
+								}
+							});
+						}
+					}
+					else
+					{
+						m_HealthOk = false;
+					}
+				}
+
+				OnComplete({.CacheKey = CacheKey, .KeyIndex = Index, .Record = Record, .Package = Package});
+			}
+
+			return Result;
+		}
+
+		virtual GetUpstreamCacheResult GetCachePayload(const CacheKey&, const IoHash& PayloadId) override
 		{
 			try
 			{
 				CloudCacheSession	   Session(m_Client);
-				const CloudCacheResult Result = Session.GetCompressedBlob(PayloadKey.PayloadId);
+				const CloudCacheResult Result = Session.GetCompressedBlob(PayloadId);
 
 				if (Result.ErrorCode == 0)
 				{
@@ -171,12 +228,41 @@ namespace detail {
 			}
 		}
 
+		virtual GetUpstreamCacheResult GetCachePayloads(std::span<CacheChunkRequest> CacheChunkRequests,
+														std::span<size_t>			 RequestIndex,
+														OnCachePayloadGetComplete&&	 OnComplete) override final
+		{
+			CloudCacheSession	   Session(m_Client);
+			GetUpstreamCacheResult Result;
+
+			for (size_t Index : RequestIndex)
+			{
+				const CacheChunkRequest& Request = CacheChunkRequests[Index];
+				IoBuffer				 Payload;
+
+				if (!Result.Error)
+				{
+					const CloudCacheResult BlobResult = Session.GetCompressedBlob(Request.ChunkId);
+					Payload							  = BlobResult.Response;
+
+					AppendResult(BlobResult, Result);
+					m_HealthOk = BlobResult.ErrorCode == 0;
+				}
+
+				OnComplete({.Request = Request, .RequestIndex = Index, .Payload = Payload});
+			}
+
+			return Result;
+		}
+
 		virtual PutUpstreamCacheResult PutCacheRecord(const UpstreamCacheRecord& CacheRecord,
 													  IoBuffer					 RecordValue,
 													  std::span<IoBuffer const>	 Payloads) override
 		{
+			using namespace fmt::literals;
+
 			ZEN_ASSERT(CacheRecord.PayloadIds.size() == Payloads.size());
-			const uint32_t MaxAttempts = 3;
+			const int32_t MaxAttempts = 3;
 
 			try
 			{
@@ -200,125 +286,132 @@ namespace detail {
 						}
 					}
 
-					return {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = Result.Success};
+					m_HealthOk = Result.ErrorCode == 0;
+
+					return {.Reason			= std::move(Result.Reason),
+							.Bytes			= Result.Bytes,
+							.ElapsedSeconds = Result.ElapsedSeconds,
+							.Success		= Result.Success};
 				}
 				else
 				{
-					bool	Success				= false;
 					int64_t TotalBytes			= 0ull;
 					double	TotalElapsedSeconds = 0.0;
 
-					for (size_t Idx = 0, Count = Payloads.size(); Idx < Count; Idx++)
-					{
-						Success = false;
-						for (int32_t Attempt = 0; Attempt < MaxAttempts; Attempt++)
+					const auto PutBlobs = [&](std::span<IoHash> PayloadIds, std::string& OutReason) -> bool {
+						for (const IoHash& PayloadId : PayloadIds)
 						{
-							if (CloudCacheResult Result = Session.PutCompressedBlob(CacheRecord.PayloadIds[Idx], Payloads[Idx]);
-								Result.Success)
+							const auto It = std::find(std::begin(CacheRecord.PayloadIds), std::end(CacheRecord.PayloadIds), PayloadId);
+
+							if (It == std::end(CacheRecord.PayloadIds))
 							{
-								TotalBytes += Result.Bytes;
-								TotalElapsedSeconds += Result.ElapsedSeconds;
-								Success = true;
-								break;
+								OutReason = "payload '{}' MISSING from local cache"_format(PayloadId);
+								return false;
 							}
-						}
 
-						if (!Success)
-						{
-							return {.Reason			= "Failed to upload payload",
-									.Bytes			= TotalBytes,
-									.ElapsedSeconds = TotalElapsedSeconds,
-									.Success		= false};
+							const size_t Idx = std::distance(std::begin(CacheRecord.PayloadIds), It);
+
+							CloudCacheResult BlobResult;
+							for (int32_t Attempt = 0; Attempt < MaxAttempts && !BlobResult.Success; Attempt++)
+							{
+								BlobResult = Session.PutCompressedBlob(CacheRecord.PayloadIds[Idx], Payloads[Idx]);
+							}
+
+							m_HealthOk = BlobResult.ErrorCode == 0;
+
+							if (!BlobResult.Success)
+							{
+								OutReason = "upload payload '{}' FAILED, reason '{}'"_format(PayloadId, BlobResult.Reason);
+								return false;
+							}
+
+							TotalBytes += BlobResult.Bytes;
+							TotalElapsedSeconds += BlobResult.ElapsedSeconds;
 						}
+
+						return true;
+					};
+
+					PutRefResult RefResult;
+					for (int32_t Attempt = 0; Attempt < MaxAttempts && !RefResult.Success; Attempt++)
+					{
+						RefResult =
+							Session.PutRef(CacheRecord.CacheKey.Bucket, CacheRecord.CacheKey.Hash, RecordValue, ZenContentType::kCbObject);
 					}
 
-					Success = false;
-					for (int32_t Attempt = 0; Attempt < MaxAttempts; Attempt++)
+					m_HealthOk = RefResult.ErrorCode == 0;
+
+					if (!RefResult.Success)
 					{
-						if (PutRefResult Result = Session.PutRef(CacheRecord.CacheKey.Bucket,
-																 CacheRecord.CacheKey.Hash,
-																 RecordValue,
-																 ZenContentType::kCbObject);
-							Result.Success)
-						{
-							TotalBytes += Result.Bytes;
-							TotalElapsedSeconds += Result.ElapsedSeconds;
-							Success = true;
+						return {.Reason	 = "upload cache record '{}/{}' FAILED, reason '{}'"_format(CacheRecord.CacheKey.Bucket,
+																									CacheRecord.CacheKey.Hash,
+																									RefResult.Reason),
+								.Success = false};
+					}
 
-							if (!Result.Needs.empty())
-							{
-								for (const IoHash& NeededHash : Result.Needs)
-								{
-									Success = false;
+					TotalBytes += RefResult.Bytes;
+					TotalElapsedSeconds += RefResult.ElapsedSeconds;
 
-									if (auto It =
-											std::find(std::begin(CacheRecord.PayloadIds), std::end(CacheRecord.PayloadIds), NeededHash);
-										It != std::end(CacheRecord.PayloadIds))
-									{
-										const size_t Idx = It - std::begin(CacheRecord.PayloadIds);
-
-										if (CloudCacheResult BlobResult =
-												Session.PutCompressedBlob(CacheRecord.PayloadIds[Idx], Payloads[Idx]);
-											BlobResult.Success)
-										{
-											TotalBytes += BlobResult.Bytes;
-											TotalElapsedSeconds += BlobResult.ElapsedSeconds;
-											Success = true;
-										}
-										else
-										{
-											ZEN_WARN("upload missing payload '{}/{}/{}' FAILED",
-													 CacheRecord.CacheKey.Bucket,
-													 CacheRecord.CacheKey.Hash,
-													 NeededHash);
-										}
-									}
-									else
-									{
-										ZEN_WARN("needed payload '{}/{}/{}' MISSING",
-												 CacheRecord.CacheKey.Bucket,
-												 CacheRecord.CacheKey.Hash,
-												 NeededHash);
-									}
-								}
+					std::string Reason;
+					if (!PutBlobs(RefResult.Needs, Reason))
+					{
+						return {.Reason = std::move(Reason), .Success = false};
+					}
 
-								const IoHash RefHash = IoHash::HashBuffer(RecordValue);
+					const IoHash	  RefHash		 = IoHash::HashBuffer(RecordValue);
+					FinalizeRefResult FinalizeResult = Session.FinalizeRef(CacheRecord.CacheKey.Bucket, CacheRecord.CacheKey.Hash, RefHash);
+					m_HealthOk						 = FinalizeResult.ErrorCode == 0;
 
-								if (FinalizeRefResult FinalizeResult =
-										Session.FinalizeRef(CacheRecord.CacheKey.Bucket, CacheRecord.CacheKey.Hash, RefHash);
-									FinalizeResult.Success)
-								{
-									TotalBytes += FinalizeResult.Bytes;
-									TotalElapsedSeconds += FinalizeResult.ElapsedSeconds;
-									Success = true;
+					if (!FinalizeResult.Success)
+					{
+						return {.Reason	 = "finalize cache record '{}/{}' FAILED, reason '{}'"_format(CacheRecord.CacheKey.Bucket,
+																									  CacheRecord.CacheKey.Hash,
+																									  FinalizeResult.Reason),
+								.Success = false};
+					}
 
-									for (const IoHash& MissingHash : FinalizeResult.Needs)
-									{
-										ZEN_WARN("finalize '{}/{}' FAILED, missing '{}'",
-												 CacheRecord.CacheKey.Bucket,
-												 CacheRecord.CacheKey.Hash,
-												 MissingHash);
-									}
-								}
-								else
-								{
-									ZEN_WARN("finalize '{}/{}' FAILED", CacheRecord.CacheKey.Bucket, CacheRecord.CacheKey.Hash);
-									Success = false;
-								}
-							}
+					if (!FinalizeResult.Needs.empty())
+					{
+						if (!PutBlobs(FinalizeResult.Needs, Reason))
+						{
+							return {.Reason = std::move(Reason), .Success = false};
+						}
 
-							if (Success)
+						FinalizeResult = Session.FinalizeRef(CacheRecord.CacheKey.Bucket, CacheRecord.CacheKey.Hash, RefHash);
+						m_HealthOk	   = FinalizeResult.ErrorCode == 0;
+
+						if (!FinalizeResult.Success)
+						{
+							return {.Reason	 = "finalize '{}/{}' FAILED, reason '{}'"_format(CacheRecord.CacheKey.Bucket,
+																							 CacheRecord.CacheKey.Hash,
+																							 FinalizeResult.Reason),
+									.Success = false};
+						}
+
+						if (!FinalizeResult.Needs.empty())
+						{
+							ExtendableStringBuilder<256> Sb;
+							for (const IoHash& MissingHash : FinalizeResult.Needs)
 							{
-								break;
+								Sb << MissingHash.ToHexString() << ",";
 							}
+
+							return {.Reason	 = "finalize '{}/{}' FAILED, still needs payload(s) '{}'"_format(CacheRecord.CacheKey.Bucket,
+																											 CacheRecord.CacheKey.Hash,
+																											 Sb.ToString()),
+									.Success = false};
 						}
 					}
 
-					return {.Bytes = TotalBytes, .ElapsedSeconds = TotalElapsedSeconds, .Success = Success};
+					TotalBytes += FinalizeResult.Bytes;
+					TotalElapsedSeconds += FinalizeResult.ElapsedSeconds;
+
+					return {.Bytes = TotalBytes, .ElapsedSeconds = TotalElapsedSeconds, .Success = true};
 				}
 			}
 			catch (std::exception& Err)
 			{
+				m_HealthOk = false;
 				return {.Reason = std::string(Err.what()), .Success = false};
 			}
 		}
@@ -326,9 +419,22 @@ namespace detail {
 		virtual UpstreamEndpointStats& Stats() override { return m_Stats; }
 
 	private:
+		static void AppendResult(const CloudCacheResult& Result, GetUpstreamCacheResult& Out)
+		{
+			Out.Success &= Result.Success;
+			Out.Bytes += Result.Bytes;
+			Out.ElapsedSeconds += Result.ElapsedSeconds;
+
+			if (Result.ErrorCode)
+			{
+				Out.Error = {.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)};
+			}
+		};
+
 		spdlog::logger& Log() { return m_Log; }
 
 		spdlog::logger&			 m_Log;
+		UpstreamEndpointInfo	 m_Info;
 		bool					 m_UseLegacyDdc;
 		std::string				 m_DisplayName;
 		RefPtr<CloudCacheClient> m_Client;
@@ -349,9 +455,13 @@ namespace detail {
 		};
 
 	public:
-		ZenUpstreamEndpoint(std::span<std::string const> Urls) : m_Log(zen::logging::Get("upstream")), m_DisplayName("ZEN")
+		ZenUpstreamEndpoint(const ZenStructuredCacheClientOptions& Options)
+		: m_Log(zen::logging::Get("upstream"))
+		, m_Info({.Name = std::string("Zen")})
+		, m_ConnectTimeout(Options.ConnectTimeout)
+		, m_Timeout(Options.Timeout)
 		{
-			for (const auto& Url : Urls)
+			for (const auto& Url : Options.Urls)
 			{
 				m_Endpoints.push_back({.Url = Url});
 			}
@@ -359,6 +469,8 @@ namespace detail {
 
 		~ZenUpstreamEndpoint() = default;
 
+		virtual const UpstreamEndpointInfo& GetEndpointInfo() const { return m_Info; }
+
 		virtual UpstreamEndpointHealth Initialize() override
 		{
 			using namespace fmt::literals;
@@ -366,9 +478,8 @@ namespace detail {
 			const ZenEndpoint& Ep = GetEndpoint();
 			if (Ep.Ok)
 			{
-				m_ServiceUrl  = Ep.Url;
-				m_DisplayName = "ZEN - {}"_format(m_ServiceUrl);
-				m_Client	  = new ZenStructuredCacheClient(m_ServiceUrl);
+				m_Info.Url = Ep.Url;
+				m_Client   = new ZenStructuredCacheClient({.Url = m_Info.Url, .ConnectTimeout = m_ConnectTimeout, .Timeout = m_Timeout});
 
 				m_HealthOk = true;
 				return {.Ok = true};
@@ -391,9 +502,9 @@ namespace detail {
 					const ZenEndpoint& Ep = GetEndpoint();
 					if (Ep.Ok)
 					{
-						m_ServiceUrl  = Ep.Url;
-						m_DisplayName = "ZEN - {}"_format(m_ServiceUrl);
-						m_Client	  = new ZenStructuredCacheClient(m_ServiceUrl);
+						m_Info.Url = Ep.Url;
+						m_Client =
+							new ZenStructuredCacheClient({.Url = m_Info.Url, .ConnectTimeout = m_ConnectTimeout, .Timeout = m_Timeout});
 
 						m_HealthOk = true;
 						return {.Ok = true};
@@ -420,9 +531,7 @@ namespace detail {
 			}
 		}
 
-		virtual std::string_view DisplayName() const override { return m_DisplayName; }
-
-		virtual GetUpstreamCacheResult GetCacheRecord(UpstreamCacheKey CacheKey, ZenContentType Type) override
+		virtual GetUpstreamCacheResult GetCacheRecord(CacheKey CacheKey, ZenContentType Type) override
 		{
 			try
 			{
@@ -449,13 +558,80 @@ namespace detail {
 			}
 		}
 
-		virtual GetUpstreamCacheResult GetCachePayload(UpstreamPayloadKey PayloadKey) override
+		virtual GetUpstreamCacheResult GetCacheRecords(std::span<CacheKey>		  CacheKeys,
+													   std::span<size_t>		  KeyIndex,
+													   const CacheRecordPolicy&	  Policy,
+													   OnCacheRecordGetComplete&& OnComplete) override
+		{
+			std::vector<size_t> IndexMap;
+			IndexMap.reserve(KeyIndex.size());
+
+			CbObjectWriter BatchRequest;
+			BatchRequest << "Method"sv
+						 << "GetCacheRecords";
+
+			BatchRequest.BeginObject("Params"sv);
+			{
+				BatchRequest.BeginArray("CacheKeys"sv);
+				for (size_t Index : KeyIndex)
+				{
+					const CacheKey& Key = CacheKeys[Index];
+					IndexMap.push_back(Index);
+
+					BatchRequest.BeginObject();
+					BatchRequest << "Bucket"sv << Key.Bucket;
+					BatchRequest << "Hash"sv << Key.Hash;
+					BatchRequest.EndObject();
+				}
+				BatchRequest.EndArray();
+
+				BatchRequest.BeginObject("Policy"sv);
+				CacheRecordPolicy::Save(Policy, BatchRequest);
+				BatchRequest.EndObject();
+			}
+			BatchRequest.EndObject();
+
+			CbPackage	   BatchResponse;
+			ZenCacheResult Result;
+
+			{
+				ZenStructuredCacheSession Session(*m_Client);
+				Result = Session.InvokeRpc(BatchRequest.Save());
+			}
+
+			if (Result.Success)
+			{
+				if (BatchResponse.TryLoad(Result.Response))
+				{
+					for (size_t LocalIndex = 0; CbFieldView Record : BatchResponse.GetObject()["Result"sv])
+					{
+						const size_t Index = IndexMap[LocalIndex++];
+						OnComplete(
+							{.CacheKey = CacheKeys[Index], .KeyIndex = Index, .Record = Record.AsObjectView(), .Package = BatchResponse});
+					}
+
+					return {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = true};
+				}
+			}
+			else if (Result.ErrorCode)
+			{
+				m_HealthOk = false;
+			}
+
+			for (size_t Index : KeyIndex)
+			{
+				OnComplete({.CacheKey = CacheKeys[Index], .KeyIndex = Index, .Record = CbObjectView(), .Package = CbPackage()});
+			}
+
+			return {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}};
+		}
+
+		virtual GetUpstreamCacheResult GetCachePayload(const CacheKey& CacheKey, const IoHash& PayloadId) override
 		{
 			try
 			{
 				ZenStructuredCacheSession Session(*m_Client);
-				const ZenCacheResult	  Result =
-					Session.GetCachePayload(PayloadKey.CacheKey.Bucket, PayloadKey.CacheKey.Hash, PayloadKey.PayloadId);
+				const ZenCacheResult	  Result = Session.GetCachePayload(CacheKey.Bucket, CacheKey.Hash, PayloadId);
 
 				if (Result.ErrorCode == 0)
 				{
@@ -477,12 +653,96 @@ namespace detail {
 			}
 		}
 
+		virtual GetUpstreamCacheResult GetCachePayloads(std::span<CacheChunkRequest> CacheChunkRequests,
+														std::span<size_t>			 RequestIndex,
+														OnCachePayloadGetComplete&&	 OnComplete) override final
+		{
+			std::vector<size_t> IndexMap;
+			IndexMap.reserve(RequestIndex.size());
+
+			CbObjectWriter BatchRequest;
+			BatchRequest << "Method"sv
+						 << "GetCachePayloads";
+
+			BatchRequest.BeginObject("Params"sv);
+			{
+				BatchRequest.BeginArray("ChunkRequests"sv);
+				{
+					for (size_t Index : RequestIndex)
+					{
+						const CacheChunkRequest& Request = CacheChunkRequests[Index];
+						IndexMap.push_back(Index);
+
+						BatchRequest.BeginObject();
+						{
+							BatchRequest.BeginObject("Key"sv);
+							BatchRequest << "Bucket"sv << Request.Key.Bucket;
+							BatchRequest << "Hash"sv << Request.Key.Hash;
+							BatchRequest.EndObject();
+
+							BatchRequest.AddObjectId("PayloadId"sv, Request.PayloadId);
+							BatchRequest << "ChunkId"sv << Request.ChunkId;
+							BatchRequest << "RawOffset"sv << Request.RawOffset;
+							BatchRequest << "RawSize"sv << Request.RawSize;
+							BatchRequest << "Policy"sv << static_cast<uint32_t>(Request.Policy);
+						}
+						BatchRequest.EndObject();
+					}
+				}
+				BatchRequest.EndArray();
+			}
+			BatchRequest.EndObject();
+
+			CbPackage	   BatchResponse;
+			ZenCacheResult Result;
+
+			{
+				ZenStructuredCacheSession Session(*m_Client);
+				Result = Session.InvokeRpc(BatchRequest.Save());
+			}
+
+			if (Result.Success)
+			{
+				if (BatchResponse.TryLoad(Result.Response))
+				{
+					for (size_t LocalIndex = 0; CbFieldView AttachmentHash : BatchResponse.GetObject()["Result"sv])
+					{
+						const size_t Index = IndexMap[LocalIndex++];
+						IoBuffer	 Payload;
+
+						if (const CbAttachment* Attachment = BatchResponse.FindAttachment(AttachmentHash.AsHash()))
+						{
+							if (const CompressedBuffer& Compressed = Attachment->AsCompressedBinary())
+							{
+								Payload = Compressed.GetCompressed().Flatten().AsIoBuffer();
+							}
+						}
+
+						OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Payload = std::move(Payload)});
+					}
+
+					return {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = true};
+				}
+			}
+			else if (Result.ErrorCode)
+			{
+				m_HealthOk = false;
+			}
+
+			for (size_t Index : RequestIndex)
+			{
+				OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Payload = IoBuffer()});
+			}
+
+			return {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}};
+		}
+
 		virtual PutUpstreamCacheResult PutCacheRecord(const UpstreamCacheRecord& CacheRecord,
 													  IoBuffer					 RecordValue,
 													  std::span<IoBuffer const>	 Payloads) override
 		{
 			ZEN_ASSERT(CacheRecord.PayloadIds.size() == Payloads.size());
-			const uint32_t MaxAttempts = 3;
+			const int32_t MaxAttempts = 3;
 
 			try
 			{
@@ -565,12 +825,15 @@ namespace detail {
 					TotalElapsedSeconds += Result.ElapsedSeconds;
 				}
 
-				return {.Bytes = TotalBytes, .ElapsedSeconds = TotalElapsedSeconds, .Success = Result.Success};
+				return {.Reason			= std::move(Result.Reason),
+						.Bytes			= TotalBytes,
+						.ElapsedSeconds = TotalElapsedSeconds,
+						.Success		= Result.Success};
 			}
-			catch (std::exception& e)
+			catch (std::exception& Err)
 			{
 				m_HealthOk = false;
-				return {.Reason = std::string(e.what()), .Success = false};
+				return {.Reason = std::string(Err.what()), .Success = false};
 			}
 		}
 
@@ -581,7 +844,7 @@ namespace detail {
 		{
 			for (ZenEndpoint& Ep : m_Endpoints)
 			{
-				ZenStructuredCacheClient  Client(Ep.Url);
+				ZenStructuredCacheClient  Client({.Url = Ep.Url, .ConnectTimeout = std::chrono::milliseconds(1000)});
 				ZenStructuredCacheSession Session(Client);
 				const int32_t			  SampleCount = 2;
 
@@ -602,7 +865,7 @@ namespace detail {
 
 			for (const auto& Ep : m_Endpoints)
 			{
-				ZEN_INFO("ping ZEN endpoint '{}' latency '{:.3}s' {}", Ep.Url, Ep.Latency, Ep.Ok ? "OK" : Ep.Reason);
+				ZEN_INFO("ping 'Zen' endpoint '{}' latency '{:.3}s' {}", Ep.Url, Ep.Latency, Ep.Ok ? "OK" : Ep.Reason);
 			}
 
 			return m_Endpoints.front();
@@ -611,9 +874,10 @@ namespace detail {
 		spdlog::logger& Log() { return m_Log; }
 
 		spdlog::logger&					 m_Log;
-		std::string						 m_ServiceUrl;
+		UpstreamEndpointInfo			 m_Info;
 		std::vector<ZenEndpoint>		 m_Endpoints;
-		std::string						 m_DisplayName;
+		std::chrono::milliseconds		 m_ConnectTimeout;
+		std::chrono::milliseconds		 m_Timeout;
 		RefPtr<ZenStructuredCacheClient> m_Client;
 		UpstreamEndpointStats			 m_Stats;
 		std::atomic_bool				 m_HealthOk{false};
@@ -700,7 +964,7 @@ struct UpstreamStats
 			const double   HitRate	  = TotalCount > 0 ? (double(HitCount) / double(TotalCount)) : 0.0;
 
 			Logger.debug("STATS - '{}', Hit rate: {:.2f}%, DOWN: '{:.2f} MiB {:.2f} MiB/s', UP: '{:.2f} MiB {:.2f} MiB/s'",
-						 Ep->DisplayName(),
+						 Ep->GetEndpointInfo().Name,
 						 HitRate,
 						 DownBytes,
 						 DownSpeed,
@@ -734,13 +998,15 @@ public:
 		for (auto& Endpoint : m_Endpoints)
 		{
 			const UpstreamEndpointHealth Health = Endpoint->Initialize();
+			const UpstreamEndpointInfo&	 Info	= Endpoint->GetEndpointInfo();
+
 			if (Health.Ok)
 			{
-				ZEN_INFO("initialize endpoint '{}' OK", Endpoint->DisplayName());
+				ZEN_INFO("'{}' endpoint '{}' OK", Info.Name, Info.Url);
 			}
 			else
 			{
-				ZEN_WARN("initialize endpoint '{}' FAILED, reason '{}'", Endpoint->DisplayName(), Health.Reason);
+				ZEN_WARN("'{}' endpoint '{}' FAILED, reason '{}'", Info.Name, Info.Url, Health.Reason);
 			}
 		}
 
@@ -761,7 +1027,7 @@ public:
 
 	virtual void RegisterEndpoint(std::unique_ptr<UpstreamEndpoint> Endpoint) override { m_Endpoints.emplace_back(std::move(Endpoint)); }
 
-	virtual GetUpstreamCacheResult GetCacheRecord(UpstreamCacheKey CacheKey, ZenContentType Type) override
+	virtual GetUpstreamCacheResult GetCacheRecord(CacheKey CacheKey, ZenContentType Type) override
 	{
 		if (m_Options.ReadUpstream)
 		{
@@ -776,6 +1042,14 @@ public:
 					{
 						return Result;
 					}
+
+					if (Result.Error)
+					{
+						ZEN_ERROR("get cache record FAILED, endpoint '{}', reason '{}', error code '{}'",
+								  Endpoint->GetEndpointInfo().Url,
+								  Result.Error.Reason,
+								  Result.Error.ErrorCode);
+					}
 				}
 			}
 		}
@@ -783,7 +1057,99 @@ public:
 		return {};
 	}
 
-	virtual GetUpstreamCacheResult GetCachePayload(UpstreamPayloadKey PayloadKey) override
+	virtual void GetCacheRecords(std::span<CacheKey>		CacheKeys,
+								 std::span<size_t>			KeyIndex,
+								 const CacheRecordPolicy&	Policy,
+								 OnCacheRecordGetComplete&& OnComplete) override final
+	{
+		std::vector<size_t> MissingKeys(KeyIndex.begin(), KeyIndex.end());
+
+		if (m_Options.ReadUpstream)
+		{
+			for (auto& Endpoint : m_Endpoints)
+			{
+				if (Endpoint->IsHealthy() && !MissingKeys.empty())
+				{
+					std::vector<size_t> Missing;
+
+					auto Result = Endpoint->GetCacheRecords(CacheKeys, MissingKeys, Policy, [&](CacheRecordGetCompleteParams&& Params) {
+						if (Params.Record)
+						{
+							OnComplete(std::forward<CacheRecordGetCompleteParams>(Params));
+						}
+						else
+						{
+							Missing.push_back(Params.KeyIndex);
+						}
+					});
+
+					if (Result.Error)
+					{
+						ZEN_ERROR("get cache record(s) (rpc) FAILED, endpoint '{}', reason '{}', error code '{}'",
+								  Endpoint->GetEndpointInfo().Url,
+								  Result.Error.Reason,
+								  Result.Error.ErrorCode);
+					}
+
+					m_Stats.Add(m_Log, *Endpoint, Result, m_Endpoints);
+					MissingKeys = std::move(Missing);
+				}
+			}
+		}
+
+		for (size_t Index : MissingKeys)
+		{
+			OnComplete({.CacheKey = CacheKeys[Index], .KeyIndex = Index, .Record = CbObjectView(), .Package = CbPackage()});
+		}
+	}
+
+	virtual void GetCachePayloads(std::span<CacheChunkRequest> CacheChunkRequests,
+								  std::span<size_t>			   RequestIndex,
+								  OnCachePayloadGetComplete&&  OnComplete) override final
+	{
+		std::vector<size_t> MissingPayloads(RequestIndex.begin(), RequestIndex.end());
+
+		if (m_Options.ReadUpstream)
+		{
+			for (auto& Endpoint : m_Endpoints)
+			{
+				if (Endpoint->IsHealthy() && !MissingPayloads.empty())
+				{
+					std::vector<size_t> Missing;
+
+					auto Result =
+						Endpoint->GetCachePayloads(CacheChunkRequests, MissingPayloads, [&](CachePayloadGetCompleteParams&& Params) {
+							if (Params.Payload)
+							{
+								OnComplete(std::forward<CachePayloadGetCompleteParams>(Params));
+							}
+							else
+							{
+								Missing.push_back(Params.RequestIndex);
+							}
+						});
+
+					if (Result.Error)
+					{
+						ZEN_ERROR("get cache payloads(s) (rpc) FAILED, endpoint '{}', reason '{}', error code '{}'",
+								  Endpoint->GetEndpointInfo().Url,
+								  Result.Error.Reason,
+								  Result.Error.ErrorCode);
+					}
+
+					m_Stats.Add(m_Log, *Endpoint, Result, m_Endpoints);
+					MissingPayloads = std::move(Missing);
+				}
+			}
+		}
+
+		for (size_t Index : MissingPayloads)
+		{
+			OnComplete({.Request = CacheChunkRequests[Index], .RequestIndex = Index, .Payload = IoBuffer()});
+		}
+	}
+
+	virtual GetUpstreamCacheResult GetCachePayload(const CacheKey& CacheKey, const IoHash& PayloadId) override
 	{
 		if (m_Options.ReadUpstream)
 		{
@@ -791,13 +1157,21 @@ public:
 			{
 				if (Endpoint->IsHealthy())
 				{
-					const GetUpstreamCacheResult Result = Endpoint->GetCachePayload(PayloadKey);
+					const GetUpstreamCacheResult Result = Endpoint->GetCachePayload(CacheKey, PayloadId);
 					m_Stats.Add(m_Log, *Endpoint, Result, m_Endpoints);
 
 					if (Result.Success)
 					{
 						return Result;
 					}
+
+					if (Result.Error)
+					{
+						ZEN_ERROR("get cache payload FAILED, endpoint '{}', reason '{}', error code '{}'",
+								  Endpoint->GetEndpointInfo().Url,
+								  Result.Error.Reason,
+								  Result.Error.ErrorCode);
+					}
 				}
 			}
 		}
@@ -834,8 +1208,10 @@ public:
 		Status.BeginArray("endpoints");
 		for (const auto& Ep : m_Endpoints)
 		{
+			const UpstreamEndpointInfo& Info = Ep->GetEndpointInfo();
 			Status.BeginObject();
-			Status << "name" << Ep->DisplayName();
+			Status << "name" << Info.Name;
+			Status << "url" << Info.Url;
 			Status << "health" << (Ep->IsHealthy() ? "ok"sv : "inactive"sv);
 
 			UpstreamEndpointStats& Stats	  = Ep->Stats();
@@ -890,6 +1266,15 @@ private:
 			{
 				const PutUpstreamCacheResult Result = Endpoint->PutCacheRecord(CacheRecord, CacheValue.Value, std::span(Payloads));
 				m_Stats.Add(m_Log, *Endpoint, Result, m_Endpoints);
+
+				if (!Result.Success)
+				{
+					ZEN_WARN("upload cache record '{}/{}' FAILED, endpoint '{}', reason '{}'",
+							 CacheRecord.CacheKey.Bucket,
+							 CacheRecord.CacheKey.Hash,
+							 Endpoint->GetEndpointInfo().Url,
+							 Result.Reason);
+				}
 			}
 		}
 	}
@@ -905,9 +1290,12 @@ private:
 				{
 					ProcessCacheRecord(std::move(CacheRecord));
 				}
-				catch (std::exception& e)
+				catch (std::exception& Err)
 				{
-					ZEN_WARN("process upstream ({}/{}) FAILED '{}'", CacheRecord.CacheKey.Bucket, CacheRecord.CacheKey.Hash, e.what());
+					ZEN_ERROR("upload cache record '{}/{}' FAILED, reason '{}'",
+							  CacheRecord.CacheKey.Bucket,
+							  CacheRecord.CacheKey.Hash,
+							  Err.what());
 				}
 			}
 
@@ -930,20 +1318,28 @@ private:
 				}
 			}
 
-			for (auto& Endpoint : m_Endpoints)
+			try
 			{
-				if (!Endpoint->IsHealthy())
+				for (auto& Endpoint : m_Endpoints)
 				{
-					if (const UpstreamEndpointHealth Health = Endpoint->CheckHealth(); Health.Ok)
-					{
-						ZEN_INFO("health check endpoint '{}' OK", Endpoint->DisplayName(), Health.Reason);
-					}
-					else
+					if (!Endpoint->IsHealthy())
 					{
-						ZEN_WARN("health check endpoint '{}' FAILED, reason '{}'", Endpoint->DisplayName(), Health.Reason);
+						const UpstreamEndpointInfo& Info = Endpoint->GetEndpointInfo();
+						if (const UpstreamEndpointHealth Health = Endpoint->CheckHealth(); Health.Ok)
+						{
+							ZEN_INFO("health check endpoint '{} - {}' OK", Info.Name, Info.Url, Health.Reason);
+						}
+						else
+						{
+							ZEN_WARN("health check endpoint '{} - {}' FAILED, reason '{}'", Info.Name, Info.Url, Health.Reason);
+						}
 					}
 				}
 			}
+			catch (std::exception& Err)
+			{
+				ZEN_ERROR("check endpoint(s) health FAILED, reason '{}'", Err.what());
+			}
 		}
 	}
 
@@ -1015,9 +1411,9 @@ MakeJupiterUpstreamEndpoint(const CloudCacheClientOptions& Options)
 }
 
 std::unique_ptr<UpstreamEndpoint>
-MakeZenUpstreamEndpoint(std::span<std::string const> Urls)
+MakeZenUpstreamEndpoint(const ZenStructuredCacheClientOptions& Options)
 {
-	return std::make_unique<detail::ZenUpstreamEndpoint>(Urls);
+	return std::make_unique<detail::ZenUpstreamEndpoint>(Options);
 }
 
 }  // namespace zen
diff --git a/zenserver/upstream/upstreamcache.h b/zenserver/upstream/upstreamcache.h
index edc995da6..12287198d 100644
--- a/zenserver/upstream/upstreamcache.h
+++ b/zenserver/upstream/upstreamcache.h
@@ -5,34 +5,27 @@
 #include <zencore/iobuffer.h>
 #include <zencore/iohash.h>
 #include <zencore/zencore.h>
+#include <zenutil/cache/cache.h>
 
 #include <atomic>
 #include <chrono>
+#include <functional>
 #include <memory>
 
 namespace zen {
 
+class CbObjectView;
+class CbPackage;
 class CbObjectWriter;
 class CidStore;
 class ZenCacheStore;
 struct CloudCacheClientOptions;
-
-struct UpstreamCacheKey
-{
-	std::string Bucket;
-	IoHash		Hash;
-};
-
-struct UpstreamPayloadKey
-{
-	UpstreamCacheKey CacheKey;
-	IoHash			 PayloadId;
-};
+struct ZenStructuredCacheClientOptions;
 
 struct UpstreamCacheRecord
 {
 	ZenContentType		Type = ZenContentType::kBinary;
-	UpstreamCacheKey	CacheKey;
+	CacheKey			CacheKey;
 	std::vector<IoHash> PayloadIds;
 };
 
@@ -88,6 +81,31 @@ struct UpstreamEndpointStats
 	std::atomic<double>	 SecondsDown{};
 };
 
+struct CacheRecordGetCompleteParams
+{
+	const CacheKey&		CacheKey;
+	size_t				KeyIndex = ~size_t(0);
+	const CbObjectView& Record;
+	const CbPackage&	Package;
+};
+
+using OnCacheRecordGetComplete = std::function<void(CacheRecordGetCompleteParams&&)>;
+
+struct CachePayloadGetCompleteParams
+{
+	const CacheChunkRequest& Request;
+	size_t					 RequestIndex{~size_t(0)};
+	IoBuffer				 Payload;
+};
+
+using OnCachePayloadGetComplete = std::function<void(CachePayloadGetCompleteParams&&)>;
+
+struct UpstreamEndpointInfo
+{
+	std::string Name;
+	std::string Url;
+};
+
 /**
  * The upstream endpont is responsible for handling upload/downloading of cache records.
  */
@@ -96,17 +114,26 @@ class UpstreamEndpoint
 public:
 	virtual ~UpstreamEndpoint() = default;
 
+	virtual const UpstreamEndpointInfo& GetEndpointInfo() const = 0;
+
 	virtual UpstreamEndpointHealth Initialize() = 0;
 
 	virtual bool IsHealthy() const = 0;
 
 	virtual UpstreamEndpointHealth CheckHealth() = 0;
 
-	virtual std::string_view DisplayName() const = 0;
+	virtual GetUpstreamCacheResult GetCacheRecord(CacheKey CacheKey, ZenContentType Type) = 0;
+
+	virtual GetUpstreamCacheResult GetCacheRecords(std::span<CacheKey>		  CacheKeys,
+												   std::span<size_t>		  KeyIndex,
+												   const CacheRecordPolicy&	  Policy,
+												   OnCacheRecordGetComplete&& OnComplete) = 0;
 
-	virtual GetUpstreamCacheResult GetCacheRecord(UpstreamCacheKey CacheKey, ZenContentType Type) = 0;
+	virtual GetUpstreamCacheResult GetCachePayload(const CacheKey& CacheKey, const IoHash& PayloadId) = 0;
 
-	virtual GetUpstreamCacheResult GetCachePayload(UpstreamPayloadKey PayloadKey) = 0;
+	virtual GetUpstreamCacheResult GetCachePayloads(std::span<CacheChunkRequest> CacheChunkRequests,
+													std::span<size_t>			 RequestIndex,
+													OnCachePayloadGetComplete&&	 OnComplete) = 0;
 
 	virtual PutUpstreamCacheResult PutCacheRecord(const UpstreamCacheRecord& CacheRecord,
 												  IoBuffer					 RecordValue,
@@ -127,9 +154,18 @@ public:
 
 	virtual void RegisterEndpoint(std::unique_ptr<UpstreamEndpoint> Endpoint) = 0;
 
-	virtual GetUpstreamCacheResult GetCacheRecord(UpstreamCacheKey CacheKey, ZenContentType Type) = 0;
+	virtual GetUpstreamCacheResult GetCacheRecord(CacheKey CacheKey, ZenContentType Type) = 0;
+
+	virtual void GetCacheRecords(std::span<CacheKey>		CacheKeys,
+								 std::span<size_t>			KeyIndex,
+								 const CacheRecordPolicy&	RecordPolicy,
+								 OnCacheRecordGetComplete&& OnComplete) = 0;
+
+	virtual GetUpstreamCacheResult GetCachePayload(const CacheKey& CacheKey, const IoHash& PayloadId) = 0;
 
-	virtual GetUpstreamCacheResult GetCachePayload(UpstreamPayloadKey PayloadKey) = 0;
+	virtual void GetCachePayloads(std::span<CacheChunkRequest> CacheChunkRequests,
+								  std::span<size_t>			   RequestIndex,
+								  OnCachePayloadGetComplete&&  OnComplete) = 0;
 
 	struct EnqueueResult
 	{
@@ -145,6 +181,6 @@ std::unique_ptr<UpstreamCache> MakeUpstreamCache(const UpstreamCacheOptions& Opt
 
 std::unique_ptr<UpstreamEndpoint> MakeJupiterUpstreamEndpoint(const CloudCacheClientOptions& Options);
 
-std::unique_ptr<UpstreamEndpoint> MakeZenUpstreamEndpoint(std::span<std::string const> Urls);
+std::unique_ptr<UpstreamEndpoint> MakeZenUpstreamEndpoint(const ZenStructuredCacheClientOptions& Options);
 
 }  // namespace zen
diff --git a/zenserver/upstream/zen.cpp b/zenserver/upstream/zen.cpp
index 14333f45a..cd7f48334 100644
--- a/zenserver/upstream/zen.cpp
+++ b/zenserver/upstream/zen.cpp
@@ -311,8 +311,17 @@ namespace detail {
 		ZenCacheSessionState(ZenStructuredCacheClient& Client) : OwnerClient(Client) {}
 		~ZenCacheSessionState() {}
 
-		void Reset() {}
+		void Reset(std::chrono::milliseconds ConnectTimeout, std::chrono::milliseconds Timeout)
+		{
+			Session.SetBody({});
+			Session.SetHeader({});
+			Session.SetConnectTimeout(ConnectTimeout);
+			Session.SetTimeout(Timeout);
+		}
+
+		cpr::Session& GetSession() { return Session; }
 
+	private:
 		ZenStructuredCacheClient& OwnerClient;
 		cpr::Session			  Session;
 	};
@@ -321,9 +330,11 @@ namespace detail {
 
 //////////////////////////////////////////////////////////////////////////
 
-ZenStructuredCacheClient::ZenStructuredCacheClient(std::string_view ServiceUrl)
+ZenStructuredCacheClient::ZenStructuredCacheClient(const ZenStructuredCacheClientOptions& Options)
 : m_Log(logging::Get(std::string_view("zenclient")))
-, m_ServiceUrl(ServiceUrl)
+, m_ServiceUrl(Options.Url)
+, m_ConnectTimeout(Options.ConnectTimeout)
+, m_Timeout(Options.Timeout)
 {
 }
 
@@ -347,7 +358,7 @@ ZenStructuredCacheClient::AllocSessionState()
 		State = new detail::ZenCacheSessionState(*this);
 	}
 
-	State->Reset();
+	State->Reset(m_ConnectTimeout, m_Timeout);
 
 	return State;
 }
@@ -381,7 +392,7 @@ ZenStructuredCacheSession::CheckHealth()
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_Client.ServiceUrl() << "/health/check";
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	cpr::Response Response = Session.Get();
 
@@ -399,7 +410,7 @@ ZenStructuredCacheSession::GetCacheRecord(std::string_view BucketId, const IoHas
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_Client.ServiceUrl() << "/z$/" << BucketId << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetHeader(cpr::Header{{"Accept",
@@ -427,7 +438,7 @@ ZenStructuredCacheSession::GetCachePayload(std::string_view BucketId, const IoHa
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_Client.ServiceUrl() << "/z$/" << BucketId << "/" << Key.ToHexString() << "/" << PayloadId.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetHeader(cpr::Header{{"Accept", "application/x-ue-comp"}});
@@ -452,7 +463,7 @@ ZenStructuredCacheSession::PutCacheRecord(std::string_view BucketId, const IoHas
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_Client.ServiceUrl() << "/z$/" << BucketId << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetHeader(cpr::Header{{"Content-Type",
@@ -480,7 +491,7 @@ ZenStructuredCacheSession::PutCachePayload(std::string_view BucketId, const IoHa
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_Client.ServiceUrl() << "/z$/" << BucketId << "/" << Key.ToHexString() << "/" << PayloadId.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetHeader(cpr::Header{{"Content-Type", "application/x-ue-comp"}});
@@ -499,4 +510,33 @@ ZenStructuredCacheSession::PutCachePayload(std::string_view BucketId, const IoHa
 			.Success		= (Response.status_code == 200 || Response.status_code == 201)};
 }
 
+ZenCacheResult
+ZenStructuredCacheSession::InvokeRpc(const CbObjectView& Request)
+{
+	ExtendableStringBuilder<256> Uri;
+	Uri << m_Client.ServiceUrl() << "/z$/$rpc";
+
+	BinaryWriter Body;
+	Request.CopyTo(Body);
+
+	cpr::Session& Session = m_SessionState->GetSession();
+
+	Session.SetOption(cpr::Url{Uri.c_str()});
+	Session.SetHeader(cpr::Header{{"Content-Type", "application/x-ue-cb"}, {"Accept", "application/x-ue-cbpkg"}});
+	Session.SetBody(cpr::Body{reinterpret_cast<const char*>(Body.GetData()), Body.GetSize()});
+
+	cpr::Response Response = Session.Post();
+	ZEN_DEBUG("POST {}", Response);
+
+	if (Response.error)
+	{
+		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = std::move(Response.error.message)};
+	}
+
+	const bool	   Success = Response.status_code == 200;
+	const IoBuffer Buffer  = Success ? IoBufferBuilder::MakeCloneFromMemory(Response.text.data(), Response.text.size()) : IoBuffer();
+
+	return {.Response = std::move(Buffer), .Bytes = Response.uploaded_bytes, .ElapsedSeconds = Response.elapsed, .Success = Success};
+}
+
 }  // namespace zen
diff --git a/zenserver/upstream/zen.h b/zenserver/upstream/zen.h
index 7f55294ce..df975df1f 100644
--- a/zenserver/upstream/zen.h
+++ b/zenserver/upstream/zen.h
@@ -26,6 +26,8 @@ class logger;
 namespace zen {
 
 class CbObjectWriter;
+class CbObjectView;
+class CbPackage;
 class ZenStructuredCacheClient;
 
 /** Zen mesh tracker
@@ -97,6 +99,14 @@ struct ZenCacheResult
 	bool		Success = false;
 };
 
+struct ZenStructuredCacheClientOptions
+{
+	std::string_view			 Url;
+	std::span<std::string const> Urls;
+	std::chrono::milliseconds	 ConnectTimeout{};
+	std::chrono::milliseconds	 Timeout{};
+};
+
 /** Zen Structured Cache session
  *
  * This provides a context in which cache queries can be performed
@@ -114,6 +124,7 @@ public:
 	ZenCacheResult GetCachePayload(std::string_view BucketId, const IoHash& Key, const IoHash& PayloadId);
 	ZenCacheResult PutCacheRecord(std::string_view BucketId, const IoHash& Key, IoBuffer Value, ZenContentType Type);
 	ZenCacheResult PutCachePayload(std::string_view BucketId, const IoHash& Key, const IoHash& PayloadId, IoBuffer Payload);
+	ZenCacheResult InvokeRpc(const CbObjectView& Request);
 
 private:
 	inline spdlog::logger& Log() { return m_Log; }
@@ -131,7 +142,7 @@ private:
 class ZenStructuredCacheClient : public RefCounted
 {
 public:
-	ZenStructuredCacheClient(std::string_view ServiceUrl);
+	ZenStructuredCacheClient(const ZenStructuredCacheClientOptions& Options);
 	~ZenStructuredCacheClient();
 
 	std::string_view ServiceUrl() const { return m_ServiceUrl; }
@@ -139,8 +150,10 @@ public:
 	inline spdlog::logger& Log() { return m_Log; }
 
 private:
-	spdlog::logger& m_Log;
-	std::string		m_ServiceUrl;
+	spdlog::logger&			  m_Log;
+	std::string				  m_ServiceUrl;
+	std::chrono::milliseconds m_ConnectTimeout;
+	std::chrono::milliseconds m_Timeout;
 
 	RwLock									 m_SessionStateLock;
 	std::list<detail::ZenCacheSessionState*> m_SessionStateCache;